hatchery-node 0.2.64

//! Node restream g4a bridge (A2).
//!
//! Tunnels the local g4a `NodeServer` through a [`crate::restream_session::NodeSession`].
//! Core speaks the OperatorLink wire (`{"op_id","g4a"}`); C2 only ever sees opaque
//! Noise ciphertext. The node never learns what sits behind C2.

use crate::restream_session::{NodeSession, SendPrio, UpstreamKeys};
use crate::uplink_journal::UplinkJournal;
use gate4agent_node_protocol::{
    ClientRole, NodeEventEnvelope, NodeFailureCode, NodeId, NodeIncarnationId, NodeRequest,
    NodeResponse, ServerFrame,
};
use gate4agent_node_wire::{LocalNodeClient, NodeClientError};
use hatchery_core::{
    BrowserClientOp, BrowserOpReply, BrowserStreamEvent, C2NodeEvent, C2NodeFailure,
    C2NodeResponse, ChannelKind, NodeCursor, RoutedNodeEvent, RoutedNodeRequest,
    RoutedNodeResponse,
};
use serde_json::{json, Value};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, watch};

const HANDSHAKE: Duration = Duration::from_secs(10);
const CONTROL_CHANNEL: u32 = 1;
/// Separate Noise session for TerminalFrame bulk (UDP data plane).
const PTY_DATA_CHANNEL: u32 = 3;

/// Run until `shutdown` is true: local g4a operator client ↔ restream session.
pub async fn run(
    local_endpoint: impl Into<String>,
    node_id: impl Into<String>,
    local_token: impl Into<String>,
    uplink: SocketAddr,
    keys: UpstreamKeys,
    journal: Option<Arc<UplinkJournal>>,
    browser_profile_root: Option<PathBuf>,
    calls: Option<crate::uplink_calls::NodeCalls>,
    mut shutdown: watch::Receiver<bool>,
) -> Result<(), String> {
    let local_endpoint = local_endpoint.into();
    let node_id_str = node_id.into();
    let local_token = local_token.into();
    let node_id = NodeId::new(node_id_str.clone()).map_err(|e| e.to_string())?;
    let browser_profile_root = browser_profile_root.unwrap_or_else(|| {
        std::env::temp_dir().join("hatchery-browser-profiles")
    });

    let mut client =
        LocalNodeClient::connect(&local_endpoint, &node_id, ClientRole::Operator, &local_token)
            .await
            .map_err(|e| e.to_string())?;

    let mut session = NodeSession::connect(
        uplink,
        &node_id_str,
        CONTROL_CHANNEL,
        ChannelKind::G4aControl,
        keys,
        journal,
        HANDSHAKE,
    )
    .await?;

    let hello = client.hello().clone();
    let incarnation_id = hello.incarnation_id;
    let connection_id = hello.connection_id;
    tracing::info!(%node_id_str, connection_id, "restream g4a bridge up");
    // Warm the machine_env cache off the request queue so the first Snapshot
    // after reconnect does not hold Resize / attach for ~700ms.
    let _ = tokio::task::spawn_blocking(crate::collect_machine_environment);
    // Presence only: mail/task use a dedicated TCP (spawn_mail_uplink). Passing the
    // bridge uplink here used to open ch2 on the shared socket and could stall PTY.
    if let Some(c) = &calls {
        c.set_uplink(Some(session.uplink())).await;
    }
    // PTY bulk channel (3) is opened inside run_loop so control is not blocked on its handshake.
    let result = run_loop(&mut client, &mut session, uplink, keys, &node_id_str, &node_id, incarnation_id, &local_endpoint, &local_token, browser_profile_root, &mut shutdown).await;
    if let Some(c) = &calls {
        c.set_uplink(None).await;
    }
    result
}

#[allow(clippy::too_many_arguments)]
async fn run_loop(
    client: &mut LocalNodeClient,
    session: &mut NodeSession,
    uplink_addr: SocketAddr,
    keys: crate::restream_session::UpstreamKeys,
    node_id_str: &str,
    node_id: &NodeId,
    incarnation_id: NodeIncarnationId,
    local_endpoint: &str,
    local_token: &str,
    browser_profile_root: PathBuf,
    shutdown: &mut watch::Receiver<bool>,
) -> Result<(), String> {

    // Requests run on their own local connection in a worker task. A request that
    // waits on the node (input settle, spawn readiness, snapshot) used to hold this
    // loop, and with it every PTY frame; now this connection only carries events.
    let worker_client =
        LocalNodeClient::connect(&local_endpoint, &node_id, ClientRole::Operator, &local_token)
            .await
            .map_err(|e| e.to_string())?;
    let (req_tx, req_rx) = mpsc::unbounded_channel::<Vec<u8>>();
    let (out_tx, mut out_rx) = mpsc::unbounded_channel::<Vec<u8>>();
    let (browser_ev_tx, mut browser_ev_rx) = mpsc::unbounded_channel::<BrowserStreamEvent>();
    // Re-bind live dig2 sessions onto this bridge so frames keep flowing after
    // noise decrypt / early-eof reconnects (otherwise HQ stays on a white canvas).
    #[cfg(feature = "d2b")]
    {
        crate::browser_capture::shared_map()
            .bind_bridge(browser_ev_tx.clone())
            .await;
    }
    let (worker_stop_tx, worker_stop) = watch::channel(false);
    let mut worker = tokio::spawn(request_worker(
        worker_client,
        node_id.clone(),
        req_rx,
        out_tx,
        browser_profile_root.clone(),
        browser_ev_tx,
        worker_stop,
    ));
    let mut last_pty_input: Option<std::time::Instant> = None;
    let uplink_watch = session.uplink();
    // PTY data plane: ch3 (G4aPty) on its OWN TCP (dial + OPEN + Noise never touch
    // the control socket), terminal bytes over UDP inside WG. A UDP probe/ack confirms
    // the datagram path; no ack → ch3 reopens TCP-only and UDP is retried with backoff.
    // While ch3 is down, TerminalFrame rides the control TCP. HATCHERY_PTY_DATA=0 disables.
    let pty_data_on = !matches!(
        std::env::var("HATCHERY_PTY_DATA").ok().as_deref(),
        Some("0") | Some("false") | Some("FALSE") | Some("no")
    );
    let mut pty = PtyPlane::new(uplink_addr, node_id_str.to_owned(), keys);
    let mut pty_open = if pty_data_on { pty.spawn_open(Duration::ZERO) } else { tokio::spawn(std::future::pending()) };
    let mut pty_session: Option<crate::restream_session::NodeSession> = None;
    let mut pty_tick = tokio::time::interval(PTY_PROBE_EVERY);
    pty_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let result = loop {
        if *shutdown.borrow() {
            break Ok(());
        }
        tokio::select! {
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() { break Ok(()); }
            }
            done = &mut worker => {
                break match done { Ok(r) => r.and(Err("request worker ended".into())), Err(e) => Err(e.to_string()) };
            }
            opened = &mut pty_open, if pty_data_on && pty_session.is_none() => {
                match opened {
                    Ok(Ok(mut s)) => {
                        let want = pty.want_udp();
                        s.set_udp(want);
                        pty.on_open(s.udp());
                        tracing::info!(channel = PTY_DATA_CHANNEL, transport = if s.udp() { "udp" } else { "tcp-ch3" }, udp_wanted = want, "PTY data channel open on dedicated TCP");
                        pty_session = Some(s);
                    }
                    Ok(Err(e)) => {
                        let d = pty.fail_delay();
                        tracing::warn!(error = %e, retry_in_ms = d.as_millis() as u64, "PTY data channel open failed; TerminalFrame on control TCP");
                        pty_open = pty.spawn_open(d);
                    }
                    Err(e) => {
                        let d = pty.fail_delay();
                        tracing::warn!(error = %e, "PTY data channel task join failed");
                        pty_open = pty.spawn_open(d);
                    }
                }
            }
            down = pty_recv(&mut pty_session) => {
                match down {
                    Ok(bytes) => pty.on_down(&bytes),
                    Err(e) => {
                        // Nudge from core (desync after a lost datagram) or ch3 TCP closed.
                        let was_udp = pty_session.as_ref().is_some_and(|s| s.udp());
                        pty_session = None;
                        pty.on_desync(was_udp);
                        let d = pty.fail_delay();
                        tracing::warn!(error = %e, was_udp, retry_in_ms = d.as_millis() as u64, "PTY data channel lost; reopening (frames on control TCP meanwhile)");
                        pty_open = pty.spawn_open(d);
                    }
                }
            }
            _ = pty_tick.tick(), if pty_data_on => {
                if let Some(s) = pty_session.as_mut() {
                    match pty.tick(s.udp(), s.uplink().udp_sent()) {
                        PtyTick::Idle => {}
                        PtyTick::Probe(body) => {
                            if let Err(e) = s.send_prio(&body, SendPrio::Pty).await {
                                tracing::warn!(error = %e, "PTY probe send failed; reopening ch3");
                                pty_session = None;
                                pty.on_desync(true);
                                let d = pty.fail_delay();
                                    pty_open = pty.spawn_open(d);
                            }
                        }
                        PtyTick::FallBackToTcp => {
                            pty_session = None;
                            pty_open = pty.spawn_open(Duration::ZERO);
                        }
                    }
                }
            }
            frame = client.recv() => {
                match frame {
                    Ok(ServerFrame::Event(envelope)) => {
                        let is_pty = matches!(envelope.event, gate4agent_node_protocol::NodeEvent::TerminalFrame { .. });
                        if is_pty {
                            if let Some(at) = last_pty_input.take() {
                                tracing::debug!(target: "pty_lat", bridge_us = at.elapsed().as_micros() as u64, "bridge: pty input received to echo frame forwarded");
                            }
                        }
                        if envelope.event.requires_harness_mcp_proxy_capability() { continue; }
                        let plain = event_bytes(&node_id, incarnation_id, envelope)?;
                        if is_pty {
                            if let Some(ps) = pty_session.as_mut() {
                                if let Err(e) = ps.send_prio(&plain, SendPrio::Pty).await {
                                    tracing::warn!(error = %e, "PTY data channel send failed; falling back to control");
                                    let was_udp = ps.udp();
                                    pty_session = None;
                                    pty.on_desync(was_udp);
                                    let d = pty.fail_delay();
                                    pty_open = pty.spawn_open(d);
                                    if let Err(e) = session.send_prio(&plain, SendPrio::Pty).await { break Err(e); }
                                }
                            } else if let Err(e) = session.send_prio(&plain, SendPrio::Pty).await {
                                break Err(e);
                            }
                        } else if let Err(e) = session.send_prio(&plain, SendPrio::Control).await {
                            break Err(e);
                        }
                    }
                    Ok(_) => break Err("local g4a sent an unexpected frame on the event connection".into()),
                    Err(err) => break Err(err.to_string()),
                }
            }
            out = out_rx.recv() => {
                let Some(out) = out else { break Err("request worker closed".into()) };
                if let Err(e) = session.send_prio(&out, SendPrio::Control).await { break Err(e); }
            }
            plain = session.recv(Duration::from_secs(120)) => {
                match plain {
                    Ok(bytes) => {
                        if last_pty_input.is_none() && is_pty_input(&bytes) { last_pty_input = Some(std::time::Instant::now()); }
                        if req_tx.send(bytes).is_err() { break Err("request worker closed".into()); }
                    }
                    Err(err) if err.contains("timeout") => continue,
                    Err(err) => break Err(err),
                }
            }
            _ = uplink_watch.closed() => break Err("uplink write stalled; reconnecting".into()),
            ev = browser_ev_rx.recv() => {
                let Some(mut ev) = ev else { continue };
                // Coalesce consecutive Frames before encrypt (post-encrypt drop = nonce gap).
                let mut flush: Vec<BrowserStreamEvent> = Vec::new();
                while matches!(ev, BrowserStreamEvent::Frame { .. }) {
                    match browser_ev_rx.try_recv() {
                        Ok(more @ BrowserStreamEvent::Frame { .. }) => ev = more,
                        Ok(more) => { flush.push(ev); ev = more; break; }
                        Err(_) => break,
                    }
                }
                flush.push(ev);
                let mut send_err: Option<String> = None;
                for ev in flush {
                    let kind = match &ev {
                        BrowserStreamEvent::Frame { bytes, session, .. } => format!("frame:{session}:{}", bytes.len()),
                        BrowserStreamEvent::Started { session, .. } => format!("started:{session}"),
                        BrowserStreamEvent::Stopped { session, .. } => format!("stopped:{session}"),
                        BrowserStreamEvent::Error { session, .. } => format!("error:{session}"),
                        _ => "other".into(),
                    };
                    let plain = match serde_json::to_vec(&json!({ "browser": ev })) {
                        Ok(p) => p,
                        Err(e) => { send_err = Some(e.to_string()); break; }
                    };
                    tracing::debug!(%kind, plain_len = plain.len(), "bridge browser → uplink");
                    if let Err(e) = session.send_prio(&plain, SendPrio::Browser).await {
                        tracing::warn!(%kind, error = %e, "bridge browser send_prio failed");
                        send_err = Some(e);
                        break;
                    }
                }
                if let Some(e) = send_err { break Err(e); }
            }
        }
    };
    // The worker releases the controller lease it holds before it exits.
    let _ = worker_stop_tx.send(true);
    let _ = tokio::time::timeout(Duration::from_secs(3), worker).await;
    result
}

const PTY_PROBE_EVERY: Duration = Duration::from_secs(5);
/// Unanswered UDP probes in a row before ch3 falls back to TCP.
const PTY_PROBE_MISSES: u32 = 3;
const PTY_STATS_EVERY: Duration = Duration::from_secs(60);

enum PtyTick {
    Idle,
    Probe(Vec<u8>),
    FallBackToTcp,
}

/// State of the ch3 PTY data plane: dedicated-TCP open backoff and UDP health.
struct PtyPlane {
    addr: SocketAddr,
    node_id: String,
    keys: UpstreamKeys,
    open_backoff: Duration,
    /// UDP disabled until this instant (after lost probes / repeated desync).
    udp_off_until: Option<std::time::Instant>,
    udp_backoff: Duration,
    probe_seq: u64,
    outstanding: bool,
    misses: u32,
    confirmed: bool,
    udp_desyncs: u32,
    last_stats: std::time::Instant,
    acks: u64,
}

impl PtyPlane {
    fn new(addr: SocketAddr, node_id: String, keys: UpstreamKeys) -> Self {
        Self {
            addr,
            node_id,
            keys,
            open_backoff: Duration::from_secs(1),
            udp_off_until: None,
            udp_backoff: Duration::from_secs(30),
            probe_seq: 0,
            outstanding: false,
            misses: 0,
            confirmed: false,
            udp_desyncs: 0,
            last_stats: std::time::Instant::now(),
            acks: 0,
        }
    }

    fn spawn_open(&self, delay: Duration) -> tokio::task::JoinHandle<Result<NodeSession, String>> {
        let (addr, nid, keys) = (self.addr, self.node_id.clone(), self.keys);
        tokio::spawn(async move {
            if !delay.is_zero() {
                tokio::time::sleep(delay).await;
            }
            // Own TCP: OPEN/WINDOW/Noise of ch3 never queue behind or ahead of control.
            NodeSession::connect(addr, &nid, PTY_DATA_CHANNEL, ChannelKind::G4aPty, keys, None, HANDSHAKE).await
        })
    }

    fn want_udp(&mut self) -> bool {
        match self.udp_off_until {
            Some(t) if std::time::Instant::now() < t => false,
            Some(_) => {
                self.udp_off_until = None;
                true
            }
            None => true,
        }
    }

    fn on_open(&mut self, _udp: bool) {
        self.open_backoff = Duration::from_secs(1);
        self.outstanding = false;
        self.misses = 0;
        self.confirmed = false;
    }

    fn fail_delay(&mut self) -> Duration {
        let d = self.open_backoff;
        self.open_backoff = (self.open_backoff * 2).min(Duration::from_secs(30));
        d
    }

    fn disable_udp(&mut self, why: &str) {
        let d = self.udp_backoff;
        self.udp_backoff = (self.udp_backoff * 2).min(Duration::from_secs(300));
        self.udp_off_until = Some(std::time::Instant::now() + d);
        tracing::warn!(reason = why, retry_udp_in_s = d.as_secs(), "PTY UDP data plane off; ch3 uses TCP fallback");
    }

    fn on_desync(&mut self, was_udp: bool) {
        if was_udp {
            self.udp_desyncs += 1;
            if self.udp_desyncs >= 3 {
                self.udp_desyncs = 0;
                self.disable_udp("repeated ch3 desync over UDP");
            }
        }
    }

    fn on_down(&mut self, bytes: &[u8]) {
        let Ok(v) = serde_json::from_slice::<Value>(bytes) else { return };
        if let Some(seq) = v.get("udp_ack").and_then(Value::as_u64) {
            if seq <= self.probe_seq {
                self.outstanding = false;
                self.misses = 0;
                self.acks += 1;
                self.udp_backoff = Duration::from_secs(30);
                self.udp_desyncs = 0;
                if !self.confirmed {
                    self.confirmed = true;
                    tracing::info!(seq, "PTY transport=udp confirmed (core acked UDP probe)");
                }
            }
        }
    }

    fn tick(&mut self, udp: bool, udp_sent: u64) -> PtyTick {
        if self.last_stats.elapsed() >= PTY_STATS_EVERY {
            self.last_stats = std::time::Instant::now();
            tracing::info!(transport = if udp { "udp" } else { "tcp-ch3" }, udp_datagrams = udp_sent, udp_acks = self.acks, confirmed = self.confirmed, "PTY data plane stats");
        }
        if !udp {
            // Cooldown over: reopen ch3 so want_udp() can flip back on.
            if self.udp_off_until.is_some_and(|t| std::time::Instant::now() >= t) {
                self.udp_off_until = None;
                return PtyTick::FallBackToTcp; // reuse: drop session, spawn_open(0)
            }
            return PtyTick::Idle;
        }
        if self.outstanding {
            self.misses += 1;
            if self.misses >= PTY_PROBE_MISSES {
                self.disable_udp("UDP probes unanswered");
                return PtyTick::FallBackToTcp;
            }
        }
        self.probe_seq += 1;
        self.outstanding = true;
        PtyTick::Probe(serde_json::to_vec(&json!({ "udp_probe": self.probe_seq })).unwrap_or_default())
    }
}

async fn pty_recv(s: &mut Option<NodeSession>) -> Result<Vec<u8>, String> {
    match s.as_mut() {
        Some(s) => loop {
            match s.recv(Duration::from_secs(3600)).await {
                Err(e) if e.contains("timeout") => continue,
                other => return other,
            }
        },
        None => std::future::pending().await,
    }
}

fn is_pty_input(plain: &[u8]) -> bool {
    // Cheap probe (timing only); the worker does the real parsing.
    let s = String::from_utf8_lossy(&plain[..plain.len().min(512)]);
    s.contains("TerminalBytes") || s.contains("terminal_bytes")
}

type Sink = mpsc::UnboundedSender<Vec<u8>>;

/// All node requests, in arrival order, on a dedicated local connection.
/// PTY input is pipelined (sent at once, answered when its reply arrives);
/// anything else first settles in-flight input, then runs request/reply.
async fn request_worker(
    mut client: LocalNodeClient,
    node_id: NodeId,
    mut rx: mpsc::UnboundedReceiver<Vec<u8>>,
    out: Sink,
    browser_profile_root: PathBuf,
    browser_ev_tx: mpsc::UnboundedSender<BrowserStreamEvent>,
    mut stop: watch::Receiver<bool>,
) -> Result<(), String> {
    #[cfg(feature = "d2b")]
    let browsers = crate::browser_capture::shared_map();
    let hello = client.hello().clone();
    let mut incarnation_id = hello.incarnation_id;
    let connection_id = hello.connection_id;
    let mut controller_owned = hello.controller.as_ref().is_some_and(|c| c.connection_id == connection_id);
    let mut inflight: std::collections::HashMap<u64, (u64, NodeRequest)> = std::collections::HashMap::new();
    let mut retry: Vec<(u64, NodeRequest)> = Vec::new();
    // Events on this connection: only the controller-scoped harness MCP proxy
    // events are forwarded (the event connection does not receive them).
    let fwd = |out: &Sink, incarnation_id: NodeIncarnationId, envelope: NodeEventEnvelope| -> Result<(), String> {
        if envelope.event.requires_harness_mcp_proxy_capability() {
            out.send(event_bytes(&node_id, incarnation_id, envelope)?).map_err(|_| "bridge closed".to_string())?;
        }
        Ok(())
    };
    loop {
        tokio::select! {
            _ = stop.changed() => {
                let _ = release_controller(&mut client, &mut controller_owned).await;
                return Ok(());
            }
            frame = client.recv() => {
                match frame.map_err(|e| e.to_string())? {
                    ServerFrame::Event(envelope) => fwd(&out, incarnation_id, envelope)?,
                    ServerFrame::Reply(reply) => {
                        on_pipelined_reply(&out, &node_id, &mut incarnation_id, &mut controller_owned, &mut inflight, &mut retry, reply)?;
                    }
                    _ => return Err("local g4a sent an unexpected idle frame".into()),
                }
            }
            plain = rx.recv() => {
                let Some(bytes) = plain else {
                    let _ = release_controller(&mut client, &mut controller_owned).await;
                    return Ok(());
                };
                if controller_owned && retry.is_empty() {
                    if let Some((op_id, request)) = pipelinable(&bytes, &node_id, incarnation_id) {
                        let request_id = client.send(request.clone()).await.map_err(|e| e.to_string())?;
                        inflight.insert(request_id, (op_id, request));
                        continue;
                    }
                }
                while !inflight.is_empty() {
                    match client.recv().await.map_err(|e| e.to_string())? {
                        ServerFrame::Event(envelope) => fwd(&out, incarnation_id, envelope)?,
                        ServerFrame::Reply(reply) => on_pipelined_reply(&out, &node_id, &mut incarnation_id, &mut controller_owned, &mut inflight, &mut retry, reply)?,
                        _ => return Err("local g4a sent an unexpected idle frame".into()),
                    }
                }
                let mut queue: Vec<Vec<u8>> = Vec::new();
                for (op_id, request) in std::mem::take(&mut retry) {
                    // Controller lapsed under pipelined input: reacquire and replay in order.
                    let routed = RoutedNodeRequest {
                        route: hatchery_core::NodeRoute {
                            node_id: hatchery_core::NodeId::new(node_id.as_str().to_owned()).map_err(|e| e.to_string())?,
                            expected_incarnation_id: NodeIncarnationId::from_bytes([0; 16]),
                        },
                        request,
                    };
                    queue.push(serde_json::to_vec(&json!({ "op_id": op_id, "g4a": routed })).map_err(|e| e.to_string())?);
                }
                queue.push(bytes);
                // Drain anything already waiting so attach/view can jump ahead of Snapshot.
                while let Ok(more) = rx.try_recv() {
                    queue.push(more);
                }
                prioritize_attach(&mut queue);
                for plain in queue {
                    let started = std::time::Instant::now();
                    let kind: String = serde_json::from_slice::<Value>(&plain).ok().and_then(|v| v.pointer("/g4a/request").and_then(|r| r.as_object().and_then(|o| o.keys().next().cloned()).or_else(|| r.as_str().map(str::to_owned)))).unwrap_or_else(|| "browser".into());
                    handle_control(&out, &mut client, &node_id, &mut incarnation_id, connection_id, &mut controller_owned, &plain, &browser_profile_root, browser_ev_tx.clone(), #[cfg(feature = "d2b")] &browsers).await?;
                    while let Some(envelope) = client.take_event() {
                        fwd(&out, incarnation_id, envelope)?;
                    }
                    if started.elapsed() > Duration::from_millis(500) {
                        tracing::warn!(request = %kind, elapsed_ms = started.elapsed().as_millis() as u64, "bridge: slow node request held the request queue");
                    }
                }
            }
        }
    }
}

/// Rank for bridge request ordering. Lower = sooner.
/// Attach/view (Resize, TerminalBytes, Spawn) must beat Snapshot so the provisional
/// draft timer is claimed before grace expires under a slow machine_env probe.
fn request_prio(plain: &[u8]) -> u8 {
    let Ok(v) = serde_json::from_slice::<Value>(plain) else {
        return 50;
    };
    if v.get("browser").is_some() {
        return 80;
    }
    let Some(req) = v.pointer("/g4a/request") else {
        return 50;
    };
    let key = req
        .as_object()
        .and_then(|o| o.keys().next().cloned())
        .or_else(|| req.as_str().map(str::to_owned))
        .unwrap_or_default();
    match key.as_str() {
        "Resize" | "resize" => 0,
        "TerminalBytes" | "terminal_bytes" | "TerminalControl" | "terminal_control" | "Interrupt" | "interrupt" => 1,
        "Spawn" | "spawn" | "SpawnWithOptions" | "spawn_with_options" | "SpawnManagedWorktree" | "spawn_managed_worktree"
        | "Resume" | "resume" | "ResumeSessionRecord" | "resume_session_record" | "Stop" | "stop" => 2,
        "Input" | "input" | "Paste" | "paste" | "SubmitPrompt" | "submit_prompt" => 3,
        "Snapshot" | "snapshot" | "Resync" | "resync" => 90,
        _ => 40,
    }
}

fn prioritize_attach(queue: &mut Vec<Vec<u8>>) {
    if queue.len() < 2 {
        return;
    }
    // Stable: equal prio keeps arrival order.
    let mut indexed: Vec<(usize, u8, Vec<u8>)> = queue
        .drain(..)
        .enumerate()
        .map(|(i, b)| (i, request_prio(&b), b))
        .collect();
    indexed.sort_by(|a, b| a.1.cmp(&b.1).then(a.0.cmp(&b.0)));
    *queue = indexed.into_iter().map(|(_, _, b)| b).collect();
}

/// PTY input that may be pipelined (no ordering dependency on replies).
fn pipelinable(plain: &[u8], node_id: &NodeId, incarnation_id: NodeIncarnationId) -> Option<(u64, NodeRequest)> {
    let v: Value = serde_json::from_slice(plain).ok()?;
    let op_id = v.get("op_id")?.as_u64()?;
    let routed: RoutedNodeRequest = serde_json::from_value(v.get("g4a")?.clone()).ok()?;
    if routed.route.node_id.as_str() != node_id.as_str() { return None; }
    let expected = routed.route.expected_incarnation_id;
    if expected != NodeIncarnationId::from_bytes([0; 16]) && expected != incarnation_id { return None; }
    match routed.request {
        NodeRequest::TerminalBytes { .. } | NodeRequest::TerminalControl { .. } | NodeRequest::Interrupt { .. } => Some((op_id, routed.request)),
        _ => None,
    }
}

fn on_pipelined_reply(
    out: &Sink,
    node_id: &NodeId,
    incarnation_id: &mut NodeIncarnationId,
    controller_owned: &mut bool,
    inflight: &mut std::collections::HashMap<u64, (u64, NodeRequest)>,
    retry: &mut Vec<(u64, NodeRequest)>,
    reply: gate4agent_node_protocol::ResponseEnvelope,
) -> Result<(), String> {
    let Some((op_id, request)) = inflight.remove(&reply.request_id) else {
        return Err(format!("unexpected local reply id {}", reply.request_id));
    };
    if let Err(f) = &reply.result {
        if f.code == NodeFailureCode::ControllerRequired {
            *controller_owned = false;
            retry.push((op_id, request));
            return Ok(());
        }
    }
    if !retry.is_empty() {
        // Keep order: everything after a lapsed one replays behind it.
        retry.push((op_id, request));
        return Ok(());
    }
    let mapped = reply
        .result
        .map(|r| C2NodeResponse::from_node_response_with_control_detail(&r))
        .map_err(|f| C2NodeFailure::from(&f));
    let plain = serde_json::to_vec(&json!({ "op_id": op_id, "g4a": RoutedNodeResponse { node_id: node_id.clone(), incarnation_id: *incarnation_id, response: mapped } })).map_err(|e| e.to_string())?;
    out.send(plain).map_err(|_| "bridge closed".to_string())
}

async fn handle_control(
    session: &Sink,
    client: &mut LocalNodeClient,
    node_id: &NodeId,
    incarnation_id: &mut NodeIncarnationId,
    connection_id: u64,
    controller_owned: &mut bool,
    plain: &[u8],
    browser_profile_root: &std::path::Path,
    browser_ev_tx: mpsc::UnboundedSender<BrowserStreamEvent>,
    #[cfg(feature = "d2b")] browsers: &std::sync::Arc<crate::browser_capture::BrowserSessionMap>,
) -> Result<(), String> {
    let v: Value = serde_json::from_slice(plain).map_err(|e| e.to_string())?;
    let Some(op_id) = v.get("op_id").and_then(Value::as_u64) else {
        return Ok(());
    };
    if let Some(browser) = v.get("browser") {
        let op: BrowserClientOp = serde_json::from_value(browser.clone()).map_err(|e| e.to_string())?;
        let reply = handle_browser_op(
            &node_id_str_from(node_id),
            op,
            browser_profile_root,
            browser_ev_tx,
            #[cfg(feature = "d2b")]
            browsers,
        )
        .await;
        let plain = serde_json::to_vec(&json!({ "op_id": op_id, "browser": reply })).map_err(|e| e.to_string())?;
        return session.send(plain).map_err(|_| "bridge closed".to_string());
    }
    let g4a = v.get("g4a").cloned().unwrap_or(Value::Null);
    let routed: RoutedNodeRequest = serde_json::from_value(g4a).map_err(|e| e.to_string())?;
    if routed.route.node_id.as_str() != node_id.as_str() {
        return send_reply(
            session,
            op_id,
            &failure_response(node_id, *incarnation_id, NodeFailureCode::InvalidRequest, "route node id does not match bridge"),
        )
        .await;
    }
    // All-zero expected incarnation is the core cursor-bootstrap wildcard.
    let expected = routed.route.expected_incarnation_id;
    if expected != NodeIncarnationId::from_bytes([0; 16]) && expected != *incarnation_id {
        return send_reply(
            session,
            op_id,
            &failure_response(
                node_id,
                *incarnation_id,
                NodeFailureCode::InvalidRequest,
                "node incarnation changed",
            ),
        )
        .await;
    }

    let request = routed.request;
    let controller_gated = !is_read_only(&request);
    if controller_gated && !*controller_owned {
        match acquire_controller(client, connection_id).await {
            Ok(true) => *controller_owned = true,
            Ok(false) => {
                return send_reply(
                    session,
                    op_id,
                    &failure_response(
                        node_id,
                        *incarnation_id,
                        NodeFailureCode::ControllerBusy,
                        "node controller lease is unavailable",
                    ),
                )
                .await;
            }
            Err(err) => return Err(err.to_string()),
        }
    }

    let deadline = request_deadline(&request);
    // The node controller lease is short-lived and the node clears it on
    // expiry, but `controller_owned` stays set on this long-lived bridge. So a
    // controller-gated request can come back `ControllerRequired` even though
    // we believe we hold it (lease lapsed, or the local client reconnected
    // under a new connection id). Re-acquire once and retry so live PTY input
    // and fresh spawns keep working on a held stream.
    let mut remaining_retries = 1u8;
    let response = loop {
        match tokio::time::timeout(deadline, client.request(request.clone())).await {
            Ok(Ok(response)) => break Ok(response),
            Ok(Err(NodeClientError::Node(failure))) => {
                if controller_gated
                    && failure.code == NodeFailureCode::ControllerRequired
                    && remaining_retries > 0
                {
                    remaining_retries -= 1;
                    *controller_owned = false;
                    match acquire_controller(client, connection_id).await {
                        Ok(true) => {
                            *controller_owned = true;
                            continue;
                        }
                        Ok(false) => break Err(failure),
                        Err(err) => return Err(err.to_string()),
                    }
                }
                break Err(failure);
            }
            Ok(Err(err)) => return Err(err.to_string()),
            Err(_) => {
                return send_reply(
                    session,
                    op_id,
                    &failure_response(
                        node_id,
                        *incarnation_id,
                        NodeFailureCode::HostDirectoryReadTimedOut,
                        "node request deadline exceeded",
                    ),
                )
                .await;
            }
        }
    };

    *incarnation_id = client.hello().incarnation_id;

    let mut mapped = response
        .map(|r| C2NodeResponse::from_node_response_with_control_detail(&r))
        .map_err(|f| C2NodeFailure::from(&f));
    if let Ok(C2NodeResponse::Snapshot { snapshot, .. }) = &mut mapped {
        snapshot.machine = Some(crate::collect_machine_environment());
    }
    send_reply(
        session,
        op_id,
        &RoutedNodeResponse {
            node_id: node_id.clone(),
            incarnation_id: *incarnation_id,
            response: mapped,
        },
    )
    .await
}

fn failure_response(
    node_id: &NodeId,
    incarnation_id: NodeIncarnationId,
    code: NodeFailureCode,
    message: &str,
) -> RoutedNodeResponse {
    RoutedNodeResponse {
        node_id: node_id.clone(),
        incarnation_id,
        response: Err(C2NodeFailure {
            code,
            message: message.into(),
        }),
    }
}

async fn send_reply(
    session: &Sink,
    op_id: u64,
    response: &RoutedNodeResponse,
) -> Result<(), String> {
    let plain =
        serde_json::to_vec(&json!({ "op_id": op_id, "g4a": response })).map_err(|e| e.to_string())?;
    session.send(plain).map_err(|_| "bridge closed".to_string())
}

fn event_bytes(
    node_id: &NodeId,
    incarnation_id: NodeIncarnationId,
    envelope: NodeEventEnvelope,
) -> Result<Vec<u8>, String> {
    let event = RoutedNodeEvent {
        node_id: node_id.clone(),
        cursor: NodeCursor {
            incarnation_id,
            sequence: envelope.sequence,
        },
        event: C2NodeEvent::from_node_event_with_control_detail(&envelope.event),
    };
    serde_json::to_vec(&json!({ "g4a": event })).map_err(|e| e.to_string())
}

async fn acquire_controller(
    client: &mut LocalNodeClient,
    connection_id: u64,
) -> Result<bool, NodeClientError> {
    match client
        .request(NodeRequest::AcquireController {
            lease_ms: gate4agent_node_protocol::MAX_CONTROLLER_LEASE_MS,
        })
        .await?
    {
        NodeResponse::Controller { controller } => Ok(controller
            .as_ref()
            .is_some_and(|state| state.connection_id == connection_id)),
        _ => Err(NodeClientError::Protocol(
            "controller acquisition returned a different response".into(),
        )),
    }
}

async fn release_controller(
    client: &mut LocalNodeClient,
    controller_owned: &mut bool,
) -> Result<(), NodeClientError> {
    if !*controller_owned {
        return Ok(());
    }
    let _ = client.request(NodeRequest::ReleaseController).await?;
    *controller_owned = false;
    Ok(())
}

fn is_read_only(request: &NodeRequest) -> bool {
    matches!(
        request,
        NodeRequest::Snapshot
            | NodeRequest::Resync { .. }
            | NodeRequest::BrowseHostDirectories { .. }
            | NodeRequest::InspectWorkspace { .. }
            | NodeRequest::ReadWorkspaceFile { .. }
            | NodeRequest::ReadGitHistory { .. }
            | NodeRequest::ReadGitDiff { .. }
            | NodeRequest::CatalogNativeSessions { .. }
            | NodeRequest::PageNativeSessions { .. }
            | NodeRequest::PreviewNativeSession { .. }
            | NodeRequest::PreviewSessionRecord { .. }
    )
}

fn request_deadline(request: &NodeRequest) -> Duration {
    match request {
        NodeRequest::CreateStandaloneWorkspace { .. }
        | NodeRequest::CreateWorktree { .. }
        | NodeRequest::RemoveWorktree { .. }
        | NodeRequest::CleanupManagedWorktree { .. } => Duration::from_secs(240),
        NodeRequest::Spawn { .. }
        | NodeRequest::Resume { .. }
        | NodeRequest::Stop { .. }
        | NodeRequest::SpawnSpec { .. }
        | NodeRequest::SpawnSpecWithHarnessMcp { .. }
        | NodeRequest::SpawnManagedWorktree { .. } => Duration::from_secs(60),
        NodeRequest::InspectWorkspace { .. } => Duration::from_secs(15),
        _ => Duration::from_secs(30),
    }
}


fn node_id_str_from(node_id: &NodeId) -> String {
    node_id.as_str().to_owned()
}

async fn handle_browser_op(
    node_id: &str,
    op: BrowserClientOp,
    browser_profile_root: &std::path::Path,
    _browser_ev_tx: mpsc::UnboundedSender<BrowserStreamEvent>,
    #[cfg(feature = "d2b")] browsers: &std::sync::Arc<crate::browser_capture::BrowserSessionMap>,
) -> BrowserOpReply {
    // Session-local servers are independent of dig2browser; handle them first.
    #[cfg(feature = "local-server")]
    {
        if let Some(reply) = handle_local_server_op(node_id, &op).await {
            return reply;
        }
    }
    #[cfg(not(feature = "d2b"))]
    {
        let _ = (node_id, op, browser_profile_root, _browser_ev_tx);
        return BrowserOpReply {
            ok: false,
            session: String::new(),
            message: "dig2browser feature (d2b) is not enabled in this node build".into(),
            body: None,
        };
    }
    #[cfg(feature = "d2b")]
    {
        match op {
            BrowserClientOp::Start {
                session,
                url,
                profile_dir,
            } => {
                let dir = match profile_dir {
                    Some(p) if !p.is_empty() => PathBuf::from(p),
                    // One persistent node profile for every session when configured.
                    _ => crate::browser_capture::shared::profile()
                        .unwrap_or_else(|| browser_profile_root.join(&session)),
                };
                match browsers
                    .start(node_id.to_owned(), session.clone(), url, dir, _browser_ev_tx)
                    .await
                {
                    Ok(()) => BrowserOpReply {
                        ok: true,
                        session,
                        message: String::new(),
                        body: None,
                    },
                    Err(message) => BrowserOpReply {
                        ok: false,
                        session,
                        message,
                        body: None,
                    },
                }
            }
            BrowserClientOp::Input { session, bytes } => match browsers.input(&session, &bytes).await
            {
                Ok(()) => BrowserOpReply {
                    ok: true,
                    session,
                    message: String::new(),
                    body: None,
                },
                Err(message) => BrowserOpReply {
                    ok: false,
                    session,
                    message,
                    body: None,
                },
            },
            BrowserClientOp::Stop { session } => {
                #[cfg(feature = "local-server")]
                {
                    crate::local_server::LocalServerPool::global()
                        .stop_session(&session)
                        .await;
                }
                match browsers.stop(&session).await {
                    Ok(()) => BrowserOpReply {
                        ok: true,
                        session,
                        message: String::new(),
                        body: None,
                    },
                    Err(message) => BrowserOpReply {
                        ok: false,
                        session,
                        message,
                        body: None,
                    },
                }
            }
            BrowserClientOp::LocalServerStart { session, .. }
            | BrowserClientOp::LocalServerStop { session, .. }
            | BrowserClientOp::LocalServerList { session } => BrowserOpReply {
                ok: false,
                session,
                message: "local_server op was not handled".into(),
                body: None,
            },
        }
    }
}

#[cfg(feature = "local-server")]
async fn handle_local_server_op(node_id: &str, op: &BrowserClientOp) -> Option<BrowserOpReply> {
    let _ = node_id;
    let (session, door_op, payload) = match op {
        BrowserClientOp::LocalServerList { session } => (
            session.clone(),
            "local_server.list",
            serde_json::json!({ "session": session }),
        ),
        BrowserClientOp::LocalServerStop { session, id } => (
            session.clone(),
            "local_server.stop",
            serde_json::json!({ "session": session, "id": id }),
        ),
        BrowserClientOp::LocalServerStart {
            session,
            kind,
            root,
            cwd,
            bin,
            open,
            spa,
        } => {
            let mut payload = serde_json::json!({
                "session": session,
                "kind": kind,
                "open": open,
            });
            if let Some(r) = root {
                payload["root"] = serde_json::json!(r);
            }
            if let Some(c) = cwd {
                payload["cwd"] = serde_json::json!(c);
            }
            if let Some(b) = bin {
                payload["bin"] = serde_json::json!(b);
            }
            if let Some(s) = spa {
                payload["spa"] = serde_json::json!(s);
            }
            (session.clone(), "local_server.start", payload)
        }
        _ => return None,
    };
    match crate::local_server::handle_op(door_op, &payload).await {
        Ok(body) => Some(BrowserOpReply {
            ok: true,
            session,
            message: String::new(),
            body: Some(body),
        }),
        Err(message) => Some(BrowserOpReply {
            ok: false,
            session,
            message,
            body: None,
        }),
    }
}