hatchery_c2/

runtime.rs

use crate::protocol::{
    C2ErrorCategory, C2NodeEvent, C2NodeFailure, C2NodeResponse, C2RelayFailure, C2RelayFailureCode, GapKind, HealthResponse, NodeCursor, NodeFreshness, NodeGap, NodeId,
    NodeIncarnationId, NodeRequest, ResolvedSpawnReceipt, RoutedNodeEvent, RoutedNodeResponse,
    ManagedWorktreeLeaseState, ManagedWorktreeSpawnRequest, ManagedWorktreeSpawnRequestV2,
    SpawnOverride, SpawnSpec,
    NodeTransportState, ObservedNode, ProviderAdapterContractSupport, ProviderContractSupport,
    ReadyResponse, SanitizedError, SlimNodeInventory, StatusResponse,
    C2_API_VERSION, C2_PROVIDER_CONTRACT_MANIFEST_CAPABILITY,
    MAX_C2_GAPS_PER_NODE, MAX_C2_NODES,
};
#[cfg(windows)]
use crate::protocol::MAX_C2_ENDPOINT_BYTES;
use gate4agent_node_protocol::{
    ClientRole, FrameError, NegotiatedNodeCompatibility, NodeEvent, NodeEventEnvelope, NodeFailureCode,
    NodeResponse, NodeSnapshot, ServerFrame,
};
use gate4agent_node_wire::{read_call_home_announce, LocalNodeClient, NodeClientError};
use std::collections::{BTreeMap, BTreeSet};
use std::io;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
#[cfg(unix)]
use std::path::{Path, PathBuf};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use thiserror::Error;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{mpsc, oneshot, watch, Semaphore};
use tokio::task::{JoinHandle, JoinSet};
use tokio::time::{sleep, timeout, Instant};

const MANAGED_RESUME_SETTLE_DEADLINE: Duration = Duration::from_secs(30);
const NODE_REQUEST_IO_HEADROOM: Duration = Duration::from_secs(5);
const NATIVE_SESSION_REQUEST_DEADLINE: Duration = Duration::from_secs(35);
const WORKSPACE_ENTRY_CREATE_NODE_SEMANTIC_DEADLINE: Duration = Duration::from_secs(5);
const WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE: Duration =
    WORKSPACE_ENTRY_CREATE_NODE_SEMANTIC_DEADLINE.saturating_add(NODE_REQUEST_IO_HEADROOM);

const HEADER_LIMIT_BYTES: usize = 16 * 1024;
const OPERATOR_BODY_LIMIT_BYTES: usize = 16 * 1024;
const MAX_HTTP_CONNECTIONS: usize = 16;
const RESPONSE_BODY_LIMIT_BYTES: usize = 8 * 1024 * 1024;

mod control;

#[cfg(windows)]
pub const DEFAULT_C2_CONTROL_ENDPOINT: &str = r"\\.\pipe\gate4agent-c2";
#[cfg(unix)]
pub const DEFAULT_C2_CONTROL_ENDPOINT: &str = "gate4agent-c2.sock";

#[cfg(windows)]
pub fn default_c2_control_endpoint() -> Result<String, C2ConfigError> {
    Ok(DEFAULT_C2_CONTROL_ENDPOINT.to_owned())
}

#[cfg(unix)]
pub fn default_c2_control_endpoint() -> Result<String, C2ConfigError> {
    let root = unix_runtime_root()?;
    let directory = root.join("gate4agent");
    let endpoint = directory.join(DEFAULT_C2_CONTROL_ENDPOINT);
    validate_unix_endpoint(&endpoint).map_err(|_| C2ConfigError::InvalidControlEndpoint)?;
    Ok(endpoint.to_string_lossy().into_owned())
}

#[cfg(unix)]
fn unix_runtime_root() -> Result<PathBuf, C2ConfigError> {
    if let Some(root) = std::env::var_os("XDG_RUNTIME_DIR").filter(|value| !value.is_empty()) {
        let root = PathBuf::from(root);
        if root.is_absolute() { return Ok(root); }
    }
    let home = std::env::var_os("HOME").filter(|value| !value.is_empty())
        .ok_or_else(|| C2ConfigError::RuntimeEndpoint(
            "neither absolute XDG_RUNTIME_DIR nor HOME is available".to_owned(),
        ))?;
    let home = PathBuf::from(home);
    if !home.is_absolute() {
        return Err(C2ConfigError::RuntimeEndpoint("HOME is not absolute".to_owned()));
    }
    Ok(home.join(".gate4agent").join("run"))
}

#[derive(Clone)]
pub struct C2NodeConfig {
    pub node_id: NodeId,
    pub endpoint: String,
    route: C2NodeRoute,
    token: String,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum C2NodeRoute {
    Local,
    SshForwardedLoopback(SocketAddr),
    /// This relay does not dial the node; the node dials the relay.
    ///
    /// For a node the relay has no way to reach -- behind NAT, in a
    /// container with no published port, on a machine with no listener the
    /// relay is allowed to open. The protocol roles are unchanged: the
    /// node is still the server, this relay is still the client asking it
    /// for things. Only the direction of the TCP connection moves, which
    /// is why the whole of this variant's implementation is "wait for a
    /// socket instead of opening one" and then the identical handshake.
    CallHome,
}

impl C2NodeConfig {
    pub fn new(node_id: NodeId, endpoint: impl Into<String>, token: impl Into<String>) -> Result<Self, C2ConfigError> {
        let endpoint = endpoint.into();
        let token = token.into();
        let (endpoint, route) = parse_node_endpoint(&endpoint)
            .ok_or_else(|| C2ConfigError::InvalidEndpoint(node_id.clone()))?;
        validate_token(&token)?;
        Ok(Self { node_id, endpoint, route, token })
    }

    fn transport_label(&self) -> &'static str {
        match self.route {
            C2NodeRoute::Local => local_transport_label(),
            C2NodeRoute::SshForwardedLoopback(_) => "ssh-forwarded-loopback",
            C2NodeRoute::CallHome => "call-home",
        }
    }
}

#[derive(Clone, Copy, Debug)]
pub struct C2Timings {
    pub poll_interval: Duration,
    pub fresh_for: Duration,
    pub attempt_deadline: Duration,
    pub transient_backoffs: [Duration; 5],
    pub parked_backoff: Duration,
    pub http_io_deadline: Duration,
}

impl Default for C2Timings {
    fn default() -> Self {
        Self {
            poll_interval: Duration::from_millis(250),
            fresh_for: Duration::from_secs(10),
            attempt_deadline: Duration::from_secs(5),
            transient_backoffs: [
                Duration::from_millis(500), Duration::from_secs(1), Duration::from_secs(2),
                Duration::from_secs(4), Duration::from_secs(8),
            ],
            parked_backoff: Duration::from_secs(30),
            http_io_deadline: Duration::from_secs(3),
        }
    }
}

#[derive(Clone)]
pub struct C2Config {
    pub api_listen: SocketAddr,
    pub control_endpoint: String,
    /// Optional TCP control accept bound to the C2 tunnel address.
    /// The unix `control_endpoint` listener is always started as well.
    pub control_tunnel_listen: Option<SocketAddr>,
    api_token: String,
    pub nodes: Vec<C2NodeConfig>,
    /// Where nodes that cannot be dialled come to announce themselves.
    /// `None` unless at least one node is configured `accept`.
    pub node_listen: Option<SocketAddr>,
    pub timings: C2Timings,
    /// Node enrollment (one-time code, neutral status, runtime node add). `None` keeps the
    /// fixed `--node` set and token authentication only.
    pub enrollment: Option<crate::enroll_service::EnrollConfig>,
}

impl C2Config {
    /// Turn on enrollment. Requires a node-listen address (checked by `validate_call_home`).
    pub fn with_enrollment(mut self, enrollment: crate::enroll_service::EnrollConfig) -> Self {
        self.enrollment = Some(enrollment);
        self
    }

    pub fn new(api_listen: SocketAddr, api_token: impl Into<String>, nodes: Vec<C2NodeConfig>) -> Result<Self, C2ConfigError> {
        let api_token = api_token.into();
        if !api_listen.ip().is_loopback() { return Err(C2ConfigError::NonLoopback(api_listen)); }
        validate_token(&api_token)?;
        if nodes.len() > MAX_C2_NODES { return Err(C2ConfigError::NodeCount(nodes.len())); }
        let mut ids = BTreeSet::new();
        let mut endpoints = BTreeSet::new();
        if nodes.iter().any(|node| !ids.insert(node.node_id.clone())) { return Err(C2ConfigError::DuplicateNode); }
        // Call-home nodes are exempt: they name no endpoint, so every one
        // of them carries the same `accept` and the uniqueness this check
        // enforces is `node_id`'s job for them (already done above). The
        // check exists to stop two dialled nodes pointing at one socket,
        // which cannot happen to a node that is never dialled.
        if nodes
            .iter()
            .filter(|node| node.route != C2NodeRoute::CallHome)
            .any(|node| !endpoints.insert(endpoint_key(&node.endpoint)))
        {
            return Err(C2ConfigError::DuplicateEndpoint);
        }
        let control_endpoint = default_c2_control_endpoint()?;
        if nodes.iter().any(|node| endpoints_equal(&node.endpoint, &control_endpoint)) {
            return Err(C2ConfigError::ControlEndpointConflict);
        }
        Ok(Self {
            api_listen,
            control_endpoint,
            control_tunnel_listen: None,
            api_token,
            nodes,
            node_listen: None,
            timings: C2Timings::default(),
            enrollment: None,
        })
    }

    /// Binds the address nodes call in on.
    ///
    /// Loopback, or one concrete C2 WireGuard tunnel address (same rule as
    /// [`Self::with_control_tunnel_listen`]). The node wire authenticates both
    /// ends but encrypts nothing, so `0.0.0.0` and other unspecified binds are
    /// refused: off-tunnel plaintext would expose terminal contents. When the
    /// listen address is the C2 tunnel IP, kernel WireGuard encrypts the path
    /// and either side may dial.
    pub fn with_node_listen(mut self, node_listen: SocketAddr) -> Result<Self, C2ConfigError> {
        // Concrete non-unspecified + nonzero port. Loopback stays valid; WG
        // tunnel addresses (e.g. 10.88.1.2:PORT) are valid for kit peers.
        if node_listen.port() == 0 || node_listen.ip().is_unspecified() {
            return Err(C2ConfigError::NonLoopbackNodeListen(node_listen));
        }
        if node_listen == self.api_listen {
            return Err(C2ConfigError::NodeListenConflict);
        }
        if self.control_tunnel_listen == Some(node_listen) {
            return Err(C2ConfigError::NonLoopbackNodeListen(node_listen));
        }
        self.node_listen = Some(node_listen);
        Ok(self)
    }

    /// Every node that waits to be called needs somewhere to call, so a
    /// config that asks for one without the other is refused rather than
    /// started into a relay that can never reach that node. Checked at
    /// startup, not at connect time, because the failure is total and
    /// permanent -- there is nothing to retry.
    pub fn validate_call_home(&self) -> Result<(), C2ConfigError> {
        if self.nodes.is_empty() && self.enrollment.is_none() {
            return Err(C2ConfigError::NodeCount(0));
        }
        let waiting = self.nodes.iter().find(|node| node.route == C2NodeRoute::CallHome);
        match (waiting, self.node_listen) {
            (Some(node), None) => Err(C2ConfigError::CallHomeWithoutListener(node.node_id.clone())),
            (None, None) if self.enrollment.is_some() => Err(C2ConfigError::EnrollmentWithoutListener),
            _ => Ok(()),
        }
    }

    pub fn with_timings(mut self, timings: C2Timings) -> Self {
        self.timings = timings;
        self
    }

    pub fn with_control_endpoint(mut self, endpoint: impl Into<String>) -> Result<Self, C2ConfigError> {
        let endpoint = endpoint.into();
        validate_control_endpoint(&endpoint)?;
        if self.nodes.iter().any(|node| endpoints_equal(&node.endpoint, &endpoint)) {
            return Err(C2ConfigError::ControlEndpointConflict);
        }
        self.control_endpoint = endpoint;
        Ok(self)
    }

    /// TCP door for an HQ WireGuard client. Bound to one concrete address
    /// (the address on the C2 tunnel), never `0.0.0.0` or port 0. Does not
    /// replace the unix control listener.
    pub fn with_control_tunnel_listen(mut self, listen: SocketAddr) -> Result<Self, C2ConfigError> {
        if listen.port() == 0 || listen.ip().is_unspecified() {
            return Err(C2ConfigError::InvalidControlTunnel(listen));
        }
        if listen == self.api_listen || self.node_listen == Some(listen) {
            return Err(C2ConfigError::InvalidControlTunnel(listen));
        }
        self.control_tunnel_listen = Some(listen);
        Ok(self)
    }
}

fn validate_control_endpoint(endpoint: &str) -> Result<(), C2ConfigError> {
    if !valid_local_endpoint(endpoint) {
        return Err(C2ConfigError::InvalidControlEndpoint);
    }
    Ok(())
}

fn parse_node_endpoint(endpoint: &str) -> Option<(String, C2NodeRoute)> {
    // The one assignment that names no address, because there is nothing
    // to address: this node will arrive on the call-home listener under
    // its own name. Spelled as a word rather than an empty value so a
    // config reader can tell "waits to be called" from "somebody forgot to
    // fill this in".
    if endpoint == "accept" {
        return Some((endpoint.to_owned(), C2NodeRoute::CallHome));
    }
    if let Some(authority) = endpoint.strip_prefix("tcp://") {
        let address = authority.parse::<SocketAddr>().ok()?;
        let is_exact_loopback = match address.ip() {
            std::net::IpAddr::V4(ip) => ip == std::net::Ipv4Addr::LOCALHOST,
            std::net::IpAddr::V6(ip) => ip == std::net::Ipv6Addr::LOCALHOST,
        };
        if !is_exact_loopback || address.port() == 0 {
            return None;
        }
        return Some((format!("tcp://{address}"), C2NodeRoute::SshForwardedLoopback(address)));
    }
    if endpoint.contains("://") || !valid_local_endpoint(endpoint) {
        return None;
    }
    Some((endpoint.to_owned(), C2NodeRoute::Local))
}

#[cfg(windows)]
fn valid_local_endpoint(endpoint: &str) -> bool {
    endpoint.starts_with(r"\\.\pipe\") && endpoint.len() > r"\\.\pipe\".len()
        && endpoint.len() <= MAX_C2_ENDPOINT_BYTES
}

#[cfg(unix)]
fn valid_local_endpoint(endpoint: &str) -> bool {
    validate_unix_endpoint(Path::new(endpoint)).is_ok()
}

#[cfg(unix)]
fn validate_unix_endpoint(endpoint: &Path) -> Result<(), ()> {
    const MAX_UNIX_ENDPOINT_BYTES: usize = 103;
    use std::os::unix::ffi::OsStrExt;

    if !endpoint.is_absolute() || endpoint.file_name().is_none()
        || endpoint.as_os_str().as_bytes().len() > MAX_UNIX_ENDPOINT_BYTES
    {
        return Err(());
    }
    Ok(())
}

#[cfg(windows)]
fn endpoint_key(endpoint: &str) -> String {
    if endpoint.starts_with("tcp://") { endpoint.to_owned() } else { endpoint.to_ascii_lowercase() }
}

#[cfg(unix)]
fn endpoint_key(endpoint: &str) -> String { endpoint.to_owned() }

fn endpoints_equal(left: &str, right: &str) -> bool { endpoint_key(left) == endpoint_key(right) }

fn validate_token(token: &str) -> Result<(), C2ConfigError> {
    if token.is_empty() || token.len() > 4096 || !token.bytes().all(|byte| matches!(byte, 0x21..=0x7e)) {
        return Err(C2ConfigError::InvalidToken);
    }
    Ok(())
}

#[derive(Debug, Error)]
pub enum C2ConfigError {
    #[error("C2 tokens must contain 1..=4096 visible ASCII bytes without whitespace")]
    InvalidToken,
    #[cfg_attr(windows, error("node '{0}' requires a bounded Windows named pipe or exact loopback TCP endpoint"))]
    #[cfg_attr(unix, error("node '{0}' requires a bounded local socket or exact loopback TCP endpoint"))]
    InvalidEndpoint(NodeId),
    #[error("C2 API listen address must be loopback: {0}")]
    NonLoopback(SocketAddr),
    #[error("C2 requires 1..=64 configured nodes; received {0}")]
    NodeCount(usize),
    #[error("C2 node IDs must be unique")]
    DuplicateNode,
    #[error("C2 node endpoints must be unique")]
    DuplicateEndpoint,
    #[cfg_attr(windows, error("C2 control endpoint must be a bounded local Windows named pipe"))]
    #[cfg_attr(unix, error("C2 control endpoint must be a bounded local endpoint"))]
    InvalidControlEndpoint,
    #[error("enrollment needs a --node-listen address for the nodes to call home on")]
    EnrollmentWithoutListener,
    #[error("C2 control endpoint must not equal a configured node endpoint")]
    ControlEndpointConflict,
    #[error("C2 control tunnel listen must be a concrete non-unspecified address with a nonzero port, not shared with another listener: {0}")]
    InvalidControlTunnel(SocketAddr),
    #[error("C2 node call-home listen must be loopback or a concrete C2 tunnel address with a nonzero port (not 0.0.0.0, not shared with another listener): {0}")]
    NonLoopbackNodeListen(SocketAddr),
    #[error("C2 node call-home listen address must not equal the API listen address")]
    NodeListenConflict,
    #[error("node '{0}' waits to be called but no --node-listen address was configured")]
    CallHomeWithoutListener(NodeId),
    #[cfg(unix)]
    #[error("C2 default runtime endpoint is unavailable: {0}")]
    RuntimeEndpoint(String),
}

type RelayResult = Result<RoutedNodeResponse, C2RelayFailure>;

enum RelayCommand {
    Request {
        operator_connection_id: u64,
        expected_incarnation_id: NodeIncarnationId,
        request: NodeRequest,
        reply: oneshot::Sender<RelayResult>,
    },
}

/// Relay endpoints by node. Nodes can be added while C2 runs (enrollment), so readers take a
/// snapshot per operation instead of holding a fixed map.
#[derive(Clone)]
struct RelayRegistry {
    map: Arc<std::sync::RwLock<Arc<BTreeMap<NodeId, RelayEndpoint>>>>,
    gate: Option<Arc<dyn crate::enroll_service::CapabilityGate>>,
}

impl RelayRegistry {
    fn new(initial: BTreeMap<NodeId, RelayEndpoint>) -> Self {
        Self { map: Arc::new(std::sync::RwLock::new(Arc::new(initial))), gate: None }
    }
    /// May the operator drive and observe this node? Token nodes are not governed by policy.
    fn allows(&self, node_id: &NodeId, capability: &str) -> bool {
        self.gate.as_ref().and_then(|gate| gate.allows(node_id.as_str(), capability)).unwrap_or(true)
    }
    /// Heartbeat-only nodes show liveness but no inventory.
    fn redact(&self, status: &mut StatusResponse) {
        for (id, node) in status.nodes.iter_mut() {
            if !self.allows(id, "attach") {
                node.inventory = None;
            }
        }
    }
    fn snapshot(&self) -> Arc<BTreeMap<NodeId, RelayEndpoint>> {
        Arc::clone(&self.map.read().expect("relay registry lock"))
    }
    fn insert(&self, node_id: NodeId, endpoint: RelayEndpoint) {
        let mut guard = self.map.write().expect("relay registry lock");
        let mut next = (**guard).clone();
        next.insert(node_id, endpoint);
        *guard = Arc::new(next);
    }
}

#[derive(Clone)]
struct RelayEndpoint {
    commands: mpsc::Sender<RelayCommand>,
    releases: mpsc::Sender<oneshot::Sender<()>>,
    force_disconnect: watch::Sender<u64>,
}

#[derive(Clone)]
struct OperatorHub {
    sink: Arc<Mutex<Option<OperatorEventSink>>>,
    gate: Arc<Mutex<Option<Arc<dyn crate::enroll_service::CapabilityGate>>>>,
}

#[derive(Clone)]
struct OperatorEventSink {
    connection_id: u64,
    outbound: mpsc::Sender<control::QueuedFrame>,
    budget: Arc<AtomicUsize>,
    disconnect: watch::Sender<bool>,
}

impl OperatorHub {
    fn new() -> Self { Self { sink: Arc::new(Mutex::new(None)), gate: Arc::new(Mutex::new(None)) } }

    fn set_gate(&self, gate: Arc<dyn crate::enroll_service::CapabilityGate>) {
        *self.gate.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(gate);
    }

    fn attach(&self, sink: OperatorEventSink) {
        *self.sink.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(sink);
    }

    fn detach(&self, connection_id: u64) {
        let mut sink = self.sink.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if sink.as_ref().is_some_and(|current| current.connection_id == connection_id) { *sink = None; }
    }

    fn is_active(&self, connection_id: u64) -> bool {
        self.sink.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
            .as_ref().is_some_and(|current| current.connection_id == connection_id)
    }

    fn has_active_operator(&self) -> bool {
        self.sink.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).is_some()
    }

    fn publish(&self, event: RoutedNodeEvent) {
        // Events of a node whose status does not hold `attach` never reach the operator.
        let gate = self.gate.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).clone();
        if gate.is_some_and(|gate| gate.allows(event.node_id.as_str(), "attach") == Some(false)) {
            return;
        }
        let sink = self.sink.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(sink) = sink.as_ref() {
            if control::queue_operator_event(&sink.outbound, &sink.budget, event).is_err() {
                let _ = sink.disconnect.send(true);
            }
        }
    }
}

fn relay_failure(
    code: C2RelayFailureCode,
    message: &'static str,
    current_incarnation_id: Option<NodeIncarnationId>,
) -> C2RelayFailure {
    C2RelayFailure { code, message: message.to_owned(), current_incarnation_id }
}

#[derive(Debug, Error)]
pub enum C2Error {
    #[error("C2 API failed: {0}")]
    Api(#[from] io::Error),
    #[error("C2 task failed: {0}")]
    Task(#[from] tokio::task::JoinError),
}

#[derive(Clone)]
pub struct C2ShutdownHandle {
    shutdown: watch::Sender<bool>,
}

impl C2ShutdownHandle {
    pub fn shutdown(&self) { let _ = self.shutdown.send(true); }
}

pub struct C2Running {
    api_addr: SocketAddr,
    shutdown: C2ShutdownHandle,
    task: Option<JoinHandle<Result<(), C2Error>>>,
}

impl C2Running {
    pub async fn start(config: C2Config) -> Result<Self, C2Error> {
        prepare_default_control_parent(&config.control_endpoint)?;
        let listener = TcpListener::bind(config.api_listen).await?;
        let api_addr = listener.local_addr()?;
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let shutdown = C2ShutdownHandle { shutdown: shutdown_tx };
        let task = tokio::spawn(run_bound(config, listener, shutdown_rx));
        Ok(Self { api_addr, shutdown, task: Some(task) })
    }

    pub fn api_addr(&self) -> SocketAddr { self.api_addr }
    pub fn shutdown_handle(&self) -> C2ShutdownHandle { self.shutdown.clone() }
    pub async fn wait(mut self) -> Result<(), C2Error> {
        self.task.take().expect("C2 task is present").await?
    }
}

#[cfg(windows)]
fn prepare_default_control_parent(_endpoint: &str) -> io::Result<()> { Ok(()) }

#[cfg(unix)]
fn prepare_default_control_parent(endpoint: &str) -> io::Result<()> {
    let default = default_c2_control_endpoint()
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error.to_string()))?;
    if endpoint != default { return Ok(()); }
    let parent = Path::new(endpoint).parent().ok_or_else(|| {
        io::Error::new(io::ErrorKind::InvalidInput, "default C2 endpoint has no parent")
    })?;
    std::fs::create_dir_all(parent)?;
    std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))
}

impl Drop for C2Running {
    fn drop(&mut self) {
        self.shutdown.shutdown();
        if let Some(task) = self.task.take() { task.abort(); }
    }
}

/// A socket handed from the call-home listener to a node's relay worker, with the
/// per-connection session token when the node authenticated by signature.
type CallHomeSocket = (TcpStream, Option<String>);
type CallHomeRoutes = Arc<Mutex<BTreeMap<NodeId, mpsc::Sender<CallHomeSocket>>>>;

struct Spawner {
    timings: C2Timings,
    relay_senders: RelayRegistry,
    call_home_routes: CallHomeRoutes,
    ingress: mpsc::Sender<Attempt>,
    status: watch::Receiver<Arc<StatusResponse>>,
    hub: OperatorHub,
}

impl Spawner {
    /// Create the relay worker for a runtime-enrolled node, once.
    fn spawn_enrolled(&self, node_id: &NodeId, shutdown: &watch::Receiver<bool>, tasks: &mut JoinSet<io::Result<()>>) {
        if self.relay_senders.snapshot().contains_key(node_id) {
            return;
        }
        // The placeholder token is never used on the wire: signed call-home hands each
        // connection its own session token.
        let placeholder = "0".repeat(64);
        let Ok(node) = C2NodeConfig::new(node_id.clone(), "accept", placeholder) else { return };
        let (commands_tx, commands_rx) = mpsc::channel(8);
        let (releases_tx, releases_rx) = mpsc::channel(1);
        let (force_tx, force_rx) = watch::channel(0_u64);
        let (call_home_tx, call_home_rx) = mpsc::channel(1);
        self.call_home_routes.lock().expect("routes lock").insert(node_id.clone(), call_home_tx);
        self.relay_senders.insert(node_id.clone(), RelayEndpoint { commands: commands_tx, releases: releases_tx, force_disconnect: force_tx });
        tasks.spawn(node_relay_worker(
            node,
            self.timings,
            commands_rx,
            releases_rx,
            force_rx,
            Some(call_home_rx),
            self.ingress.clone(),
            self.status.clone(),
            self.hub.clone(),
            shutdown.clone(),
        ));
        tracing::info!(%node_id, "relay worker created for enrolled node");
    }
}

async fn run_bound(config: C2Config, listener: TcpListener, mut shutdown: watch::Receiver<bool>) -> Result<(), C2Error> {
    let now = unix_ms();
    let nodes = config.nodes.iter().map(|node| {
        (node.node_id.clone(), initial_observed_node(node))
    }).collect();
    let initial = Arc::new(StatusResponse { api_version: C2_API_VERSION, ready: config.nodes.is_empty(), observed_at_unix_ms: now, nodes });
    let (status_tx, status_rx) = watch::channel(initial);
    let (ingress_tx, ingress_rx) = mpsc::channel(config.nodes.len().saturating_mul(2).max(16));
    let mut tasks = JoinSet::new();
    let hub = OperatorHub::new();
    let mut relay_senders = BTreeMap::new();
    let mut relay_receivers = Vec::new();
    // Where the call-home listener hands a freshly announced socket. One
    // slot per waiting node and no more: a node has one live connection to
    // this relay, so a second socket arriving while the first is still
    // being served means the node reconnected and the relay has not
    // noticed yet -- queueing more of those would only serve them stale.
    let mut static_routes: BTreeMap<NodeId, mpsc::Sender<CallHomeSocket>> = BTreeMap::new();
    for node in config.nodes.clone() {
        let (commands_tx, commands_rx) = mpsc::channel(8);
        let (releases_tx, releases_rx) = mpsc::channel(1);
        let (force_tx, force_rx) = watch::channel(0_u64);
        relay_senders.insert(node.node_id.clone(), RelayEndpoint {
            commands: commands_tx,
            releases: releases_tx,
            force_disconnect: force_tx,
        });
        let call_home_rx = if node.route == C2NodeRoute::CallHome {
            let (call_home_tx, call_home_rx) = mpsc::channel(1);
            static_routes.insert(node.node_id.clone(), call_home_tx);
            Some(call_home_rx)
        } else {
            None
        };
        relay_receivers.push((node, commands_rx, releases_rx, force_rx, call_home_rx));
    }
    let static_ids: BTreeSet<NodeId> = static_routes.keys().cloned().collect();
    let call_home_routes: CallHomeRoutes = Arc::new(Mutex::new(static_routes));
    let mut relay_senders = RelayRegistry::new(relay_senders);
    // Enrollment: registry, listener, runtime node adds.
    let mut enroll_events = None;
    let mut enroll_service = None;
    if let Some(enrollment) = &config.enrollment {
        let (service, events) = crate::enroll_service::EnrollService::start(enrollment)
            .map_err(|error| C2Error::Api(io::Error::other(error.to_string())))?;
        let enroll_listener = TcpListener::bind(enrollment.listen).await?;
        tasks.spawn(crate::enroll_service::serve_enroll(enroll_listener, Arc::clone(&service), shutdown.clone()));
        relay_senders.gate = Some(Arc::clone(&service) as Arc<dyn crate::enroll_service::CapabilityGate>);
        hub.set_gate(Arc::clone(&service) as Arc<dyn crate::enroll_service::CapabilityGate>);
        enroll_events = Some(events);
        enroll_service = Some(service);
    }
    if let Some(node_listen) = config.node_listen {
        tasks.spawn(accept_call_home(node_listen, Arc::clone(&call_home_routes), static_ids, enroll_service.clone(), shutdown.clone()));
    }
    tasks.spawn(inventory_owner(config.timings.fresh_for, ingress_rx, status_tx, shutdown.clone()));
    for (node, commands, releases, force_disconnect, call_home) in relay_receivers {
        tasks.spawn(node_relay_worker(node, config.timings, commands, releases, force_disconnect, call_home, ingress_tx.clone(), status_rx.clone(), hub.clone(), shutdown.clone()));
    }
    let spawner = Spawner {
        timings: config.timings,
        relay_senders: relay_senders.clone(),
        call_home_routes: Arc::clone(&call_home_routes),
        ingress: ingress_tx.clone(),
        status: status_rx.clone(),
        hub: hub.clone(),
    };
    drop(ingress_tx);
    tasks.spawn(http_server(listener, config.api_token.clone(), config.timings.http_io_deadline, status_rx.clone(), enroll_service.clone(), shutdown.clone()));
    let slots = control::ControlSlots::new();
    let tunnel_listen = config.control_tunnel_listen;
    let tunnel_token = config.api_token.clone();
    tasks.spawn(control::run(
        config.control_endpoint,
        config.api_token,
        relay_senders.clone(),
        status_rx.clone(),
        hub.clone(),
        shutdown.clone(),
        slots.clone(),
    ));
    if let Some(listen) = tunnel_listen {
        tasks.spawn(control::run_tcp(
            listen,
            tunnel_token,
            relay_senders.clone(),
            status_rx,
            hub,
            shutdown.clone(),
            slots,
        ));
    }
    loop {
        tokio::select! {
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() { break; }
            }
            event = async {
                match enroll_events.as_mut() {
                    Some(events) => events.recv().await,
                    None => std::future::pending().await,
                }
            } => {
                match event {
                    Some(crate::enroll_service::EnrollEvent::Enrolled(id)) => {
                        if let Ok(node_id) = NodeId::new(id) { spawner.spawn_enrolled(&node_id, &shutdown, &mut tasks); }
                    }
                    Some(crate::enroll_service::EnrollEvent::StatusChanged(id, _)) => {
                        // A node that no longer holds the heartbeat capability is dropped.
                        if let Ok(node_id) = NodeId::new(id) {
                            if !relay_senders.allows(&node_id, "heartbeat") {
                                if let Some(endpoint) = relay_senders.snapshot().get(&node_id) {
                                    endpoint.force_disconnect.send_modify(|generation| *generation = generation.wrapping_add(1));
                                }
                            }
                        }
                    }
                    Some(crate::enroll_service::EnrollEvent::PolicyChanged) => {
                        if let Some(service) = &enroll_service {
                            for id in service.nodes_without("heartbeat") {
                                if let Ok(node_id) = NodeId::new(id) {
                                    if let Some(endpoint) = relay_senders.snapshot().get(&node_id) {
                                        endpoint.force_disconnect.send_modify(|generation| *generation = generation.wrapping_add(1));
                                    }
                                }
                            }
                        }
                    }
                    None => enroll_events = None,
                }
            }
            result = tasks.join_next() => {
                match result {
                    Some(Ok(Ok(()))) if *shutdown.borrow() => break,
                    Some(Ok(Ok(()))) => return Err(C2Error::Api(io::Error::new(io::ErrorKind::Other, "C2 task exited unexpectedly"))),
                    Some(Ok(Err(error))) => return Err(C2Error::Api(error)),
                    Some(Err(error)) => return Err(C2Error::Task(error)),
                    None => break,
                }
            }
        }
    }
    tasks.shutdown().await;
    Ok(())
}

fn initial_enrolled_node() -> ObservedNode {
    ObservedNode {
        endpoint: "call-home".to_owned(), transport_label: "call-home".to_owned(),
        transport: NodeTransportState::Offline, freshness: NodeFreshness::Unavailable,
        cursor: None, inventory: None, last_attempt_unix_ms: None, last_success_unix_ms: None,
        consecutive_failures: 0, last_error: None, gaps: Vec::new(), gaps_truncated: 0,
    }
}

fn initial_observed_node(node: &C2NodeConfig) -> ObservedNode {
    ObservedNode {
        endpoint: node.endpoint.clone(), transport_label: node.transport_label().to_owned(),
        transport: NodeTransportState::Offline, freshness: NodeFreshness::Unavailable,
        cursor: None, inventory: None, last_attempt_unix_ms: None, last_success_unix_ms: None,
        consecutive_failures: 0, last_error: None, gaps: Vec::new(), gaps_truncated: 0,
    }
}

#[cfg(windows)]
fn local_transport_label() -> &'static str { "windows-named-pipe" }

#[cfg(unix)]
fn local_transport_label() -> &'static str { "unix-domain-socket" }

#[derive(Clone, Default)]
struct ProviderContractManifest {
    provider_contracts: Vec<ProviderContractSupport>,
    provider_adapter_contracts: Vec<ProviderAdapterContractSupport>,
}

impl ProviderContractManifest {
    fn from_compatibility(compatibility: Option<&NegotiatedNodeCompatibility>) -> Self {
        let Some(compatibility) = compatibility.filter(|compatibility| {
            compatibility.capabilities.iter().any(|capability| {
                capability.as_str() == C2_PROVIDER_CONTRACT_MANIFEST_CAPABILITY
            })
        }) else {
            return Self::default();
        };
        Self {
            provider_contracts: compatibility.provider_contracts.clone(),
            provider_adapter_contracts: compatibility.provider_adapter_contracts.clone(),
        }
    }
}

enum AttemptResult {
    Connected {
        cursor: NodeCursor,
        snapshot: NodeSnapshot,
        gaps: Vec<GapKind>,
        provider_contract_manifest: ProviderContractManifest,
    },
    Success { cursor: NodeCursor, snapshot: NodeSnapshot, gaps: Vec<GapKind> },
    Cursor {
        cursor: NodeCursor,
        gaps: Vec<GapKind>,
        managed_worktree_events: Vec<NodeEvent>,
    },
    Failure { error: SanitizedError, hard: bool },
}

struct Attempt { node_id: NodeId, at_unix_ms: u64, result: AttemptResult }

/// Accepts node connections and hands each one to the relay worker for
/// whichever node it says it is.
///
/// This is the only place in the relay that learns a node's identity from
/// the wire rather than from configuration, and it is careful about what
/// that means: the announced name SELECTS a worker and nothing more. The
/// worker then runs the same mutual challenge-response it runs on a
/// dialled connection, against the token configured for that node, so a
/// caller that announced a name it cannot prove gets no further than the
/// next frame.
///
/// A socket for an unknown node, an unreadable preface, or a node whose
/// slot is already full is dropped with a line saying which -- never
/// silently, because "the node is calling but nothing happens" is exactly
/// the failure that is impossible to diagnose from the node's end.
async fn accept_call_home(
    listen: SocketAddr,
    routes: CallHomeRoutes,
    static_ids: BTreeSet<NodeId>,
    enroll: Option<Arc<crate::enroll_service::EnrollService>>,
    mut shutdown: watch::Receiver<bool>,
) -> io::Result<()> {
    let listener = TcpListener::bind(listen).await?;
    tracing::info!(%listen, waiting_nodes = routes.lock().expect("routes lock").len(), "call-home listener bound");
    loop {
        if *shutdown.borrow() {
            return Ok(());
        }
        let (stream, peer) = tokio::select! {
            accepted = listener.accept() => accepted?,
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() { return Ok(()); }
                continue;
            }
        };
        let routes = Arc::clone(&routes);
        let static_ids = static_ids.clone();
        let enroll = enroll.clone();
        // Off the accept loop: the preface has its own deadline, and one
        // peer that connects and then says nothing must not stop every
        // other node from being accepted meanwhile.
        tokio::spawn(async move {
            let mut stream = stream;
            let _ = stream.set_nodelay(true);
            let node_id = match read_call_home_announce(&mut stream).await {
                Ok(node_id) => node_id,
                Err(error) => {
                    tracing::warn!(%peer, cause = %error, "call-home connection rejected");
                    return;
                }
            };
            // Token nodes (configured with --node) keep the original path. Enrolled nodes
            // prove their identity by signing a challenge; the exchange yields a
            // per-connection session token for the unchanged wire handshake.
            let mut session_token = None;
            if !static_ids.contains(&node_id) {
                let Some(enroll) = enroll.as_ref() else {
                    tracing::warn!(%peer, %node_id, "call-home connection named a node this relay does not wait for");
                    return;
                };
                match enroll.signed_call_home(&mut stream, node_id.as_str()).await {
                    Ok(token) => session_token = Some(token),
                    Err(cause) => {
                        tracing::warn!(%peer, %node_id, %cause, "signed call-home refused");
                        return;
                    }
                }
            }
            let sender = routes.lock().expect("routes lock").get(&node_id).cloned();
            let Some(sender) = sender else {
                tracing::warn!(
                    %peer,
                    %node_id,
                    "call-home connection named a node this relay does not wait for",
                );
                return;
            };
            match sender.try_send((stream, session_token)) {
                Ok(()) => tracing::info!(%peer, %node_id, "node called home"),
                Err(mpsc::error::TrySendError::Full(_)) => tracing::warn!(
                    %peer,
                    %node_id,
                    "node called home while its previous connection is still being taken up",
                ),
                Err(mpsc::error::TrySendError::Closed(_)) => tracing::warn!(
                    %peer,
                    %node_id,
                    "node called home but its relay worker is gone",
                ),
            }
        });
    }
}

/// Blocks until this node calls in, or shutdown -- with no deadline of its
/// own on purpose.
///
/// A dialled connection either answers within `attempt_deadline` or has
/// failed, and treating a slow answer as a failure is right there. Waiting
/// for a node to call is the opposite: a node that has not called yet has
/// not failed at anything, and it may be minutes from starting. Putting
/// that wait under the attempt deadline would turn an idle relay into a
/// stream of "connection deadline exceeded" warnings about nodes that are
/// merely not running. The handshake that FOLLOWS is deadlined normally --
/// a peer that connects and then stalls is a real failure.
async fn await_call_home(
    receiver: Option<&mut mpsc::Receiver<CallHomeSocket>>,
    shutdown: &mut watch::Receiver<bool>,
) -> Option<CallHomeSocket> {
    let receiver = receiver?;
    loop {
        if *shutdown.borrow() {
            return None;
        }
        tokio::select! {
            stream = receiver.recv() => return stream,
            changed = shutdown.changed() => {
                if changed.is_err() || *shutdown.borrow() { return None; }
            }
        }
    }
}

async fn node_relay_worker(
    node: C2NodeConfig,
    timings: C2Timings,
    mut commands: mpsc::Receiver<RelayCommand>,
    mut releases: mpsc::Receiver<oneshot::Sender<()>>,
    mut force_disconnect: watch::Receiver<u64>,
    mut call_home: Option<mpsc::Receiver<CallHomeSocket>>,
    ingress: mpsc::Sender<Attempt>,
    status: watch::Receiver<Arc<StatusResponse>>,
    hub: OperatorHub,
    mut shutdown: watch::Receiver<bool>,
) -> io::Result<()> {
    let mut failures = 0_usize;
    loop {
        if *shutdown.borrow() { return Ok(()); }
        let previous = status.borrow().nodes.get(&node.node_id).and_then(|item| item.cursor);
        // A waiting node's socket is collected BEFORE the deadline starts,
        // because waiting for a node to call is not an attempt that can
        // time out -- see `await_call_home`. Once it is here, the
        // handshake over it is deadlined exactly like a dialled one.
        let adopted = if node.route == C2NodeRoute::CallHome {
            match await_call_home(call_home.as_mut(), &mut shutdown).await {
                Some((stream, token)) => Some((stream, token)),
                None => return Ok(()),
            }
        } else {
            None
        };
        let (adopted, session_token) = match adopted {
            Some((stream, token)) => (Some(stream), token),
            None => (None, None),
        };
        let attempt_node = match session_token {
            Some(token) => C2NodeConfig { token, ..node.clone() },
            None => node.clone(),
        };
        let connected = timeout(timings.attempt_deadline, connect_operator(&attempt_node, adopted)).await;
        let mut client = match connected {
            Ok(Ok(client)) => client,
            Ok(Err(error)) => {
                failures = failures.saturating_add(1);
                let (error, hard) = sanitize_node_error(&error);
                tracing::warn!(
                    node_id = %node.node_id,
                    cause = %error.message,
                    category = ?error.category,
                    hard,
                    "node connection attempt failed",
                );
                ingress_attempt(&ingress, &node.node_id, AttemptResult::Failure { error, hard }).await?;
                reject_disconnected_commands(&mut commands, previous);
                acknowledge_disconnected_releases(&mut releases);
                relay_backoff(&mut shutdown, timings, failures, hard).await?;
                continue;
            }
            Err(_) => {
                failures = failures.saturating_add(1);
                tracing::warn!(
                    node_id = %node.node_id,
                    cause = "node connection deadline exceeded",
                    "node connection attempt failed",
                );
                ingress_attempt(&ingress, &node.node_id, AttemptResult::Failure {
                    error: SanitizedError { category: C2ErrorCategory::Timeout, message: "node connection deadline exceeded".to_owned() },
                    hard: false,
                }).await?;
                reject_disconnected_commands(&mut commands, previous);
                acknowledge_disconnected_releases(&mut releases);
                relay_backoff(&mut shutdown, timings, failures, false).await?;
                continue;
            }
        };
        failures = 0;
        let hello = client.hello().clone();
        let provider_contract_manifest =
            ProviderContractManifest::from_compatibility(hello.compatibility.as_ref());
        let incarnation_id = hello.incarnation_id;
        let connection_id = hello.connection_id;
        tracing::info!(
            node_id = %node.node_id,
            connection_id,
            incarnation_id = ?incarnation_id,
            "node attached to relay",
        );
        let mut controller_owned = hello.controller.as_ref().is_some_and(|controller| controller.connection_id == connection_id);
        let mut cursor = NodeCursor { incarnation_id, sequence: hello.event_sequence };
        let mut snapshot = hello.snapshot;
        let mut gaps = Vec::new();
        let mut did_resync = false;
        if let Some(previous) = previous {
            if previous.incarnation_id != incarnation_id {
                gaps.push(GapKind::IncarnationChanged);
            } else if cursor.sequence < previous.sequence {
                gaps.push(GapKind::CursorRegression);
            } else if cursor.sequence > previous.sequence {
                match bounded_node_request(&mut client, NodeRequest::Resync { after_sequence: previous.sequence }).await {
                    Ok(NodeResponse::Resync {
                        event_sequence,
                        oldest_available_sequence,
                        snapshot: resync_snapshot,
                        events,
                    }) => {
                        let resync_gaps = validate_resync(
                            previous.sequence,
                            cursor.sequence,
                            event_sequence,
                            oldest_available_sequence,
                            &events,
                        );
                        if resync_gaps.is_empty() {
                            publish_recovered_events(&node.node_id, incarnation_id, &events, &hub);
                        } else {
                            hub.publish(RoutedNodeEvent {
                                node_id: node.node_id.clone(),
                                cursor: NodeCursor { incarnation_id, sequence: event_sequence },
                                event: C2NodeEvent::ResyncRequired {
                                    oldest_available_sequence,
                                },
                            });
                        }
                        gaps.extend(resync_gaps);
                        cursor.sequence = event_sequence;
                        snapshot = resync_snapshot;
                        did_resync = true;
                    }
                    Ok(_) => gaps.push(GapKind::NonContiguousEvents),
                    Err(error) => {
                        ingress_attempt(&ingress, &node.node_id, relay_failure_attempt(&error)).await?;
                        reject_disconnected_commands(&mut commands, Some(cursor));
                        continue;
                    }
                }
            }
        }
        ingress_attempt(&ingress, &node.node_id, AttemptResult::Connected {
            cursor,
            snapshot,
            gaps,
            provider_contract_manifest,
        }).await?;
        if let Err(error) = drain_pending_events(&mut client, &node.node_id, &mut cursor, &hub, &ingress, did_resync, None).await {
            ingress_attempt(&ingress, &node.node_id, relay_failure_attempt(&error)).await?;
            reject_disconnected_commands(&mut commands, Some(cursor));
            continue;
        }

        let cadence = timings.poll_interval.max(Duration::from_millis(1)).min(Duration::from_millis(250));
        let mut snapshot_tick = tokio::time::interval(cadence);
        snapshot_tick.reset();
        let mut lease_tick = tokio::time::interval(Duration::from_secs(30));
        lease_tick.reset();
        let disconnect_error = loop {
            tokio::select! {
                command = commands.recv() => {
                    let Some(command) = command else { return Ok(()); };
                    tokio::select! {
                        result = handle_relay_command(
                            &mut client, command, &node.node_id, incarnation_id, connection_id,
                            &mut controller_owned, &hub, &mut cursor, &ingress,
                        ) => match result {
                            Ok(()) => {}
                            Err(error) => break error,
                        },
                        changed = force_disconnect.changed() => {
                            let _ = changed;
                            break NodeClientError::Io(io::Error::new(io::ErrorKind::ConnectionAborted, "node relay cleanup forced reconnect"));
                        }
                    }
                }
                release = releases.recv() => {
                    let Some(reply) = release else { return Ok(()); };
                    let result = release_controller(&mut client, &mut controller_owned).await;
                    let _ = reply.send(());
                    if let Err(error) = result { break error; }
                }
                changed = force_disconnect.changed() => {
                    let _ = changed;
                    break NodeClientError::Io(io::Error::new(io::ErrorKind::ConnectionAborted, "node relay cleanup forced reconnect"));
                }
                frame = client.recv() => {
                    match frame {
                        Ok(ServerFrame::Event(envelope)) => {
                            if let Err(error) = handle_live_node_event(
                                &mut client,
                                &node.node_id,
                                envelope,
                                &mut cursor,
                                &hub,
                                &ingress,
                            ).await {
                                break error;
                            }
                        }
                        Ok(ServerFrame::Reply(_)
                            | ServerFrame::Challenge(_)
                            | ServerFrame::Hello(_)) => {
                            break NodeClientError::Protocol(
                                "node sent an unexpected idle frame".to_owned(),
                            );
                        }
                        Err(error) => break error,
                    }
                }
                _ = snapshot_tick.tick() => {
                    match bounded_node_request(&mut client, NodeRequest::Snapshot).await {
                        Ok(NodeResponse::Snapshot { event_sequence, snapshot, .. }) => {
                            if let Err(error) = drain_pending_events(&mut client, &node.node_id, &mut cursor, &hub, &ingress, false, None).await { break error; }
                            if event_sequence >= cursor.sequence {
                                cursor.sequence = event_sequence;
                                ingress_attempt(&ingress, &node.node_id, AttemptResult::Success { cursor, snapshot, gaps: Vec::new() }).await?;
                            }
                        }
                        Ok(_) => break NodeClientError::Protocol("snapshot returned a different response".to_owned()),
                        Err(error) => break error,
                    }
                }
                _ = lease_tick.tick() => {
                    if controller_owned {
                        if hub.has_active_operator() {
                            match acquire_controller(&mut client, connection_id).await {
                                Ok(owned) => controller_owned = owned,
                                Err(error) => break error,
                            }
                        } else if let Err(error) = release_controller(&mut client, &mut controller_owned).await {
                            break error;
                        }
                        if let Err(error) = drain_pending_events(&mut client, &node.node_id, &mut cursor, &hub, &ingress, false, None).await { break error; }
                    }
                }
                changed = shutdown.changed() => {
                    if changed.is_err() || *shutdown.borrow() {
                        let _ = release_controller(&mut client, &mut controller_owned).await;
                        return Ok(());
                    }
                }
            }
        };
        let (error, hard) = sanitize_node_error(&disconnect_error);
        tracing::warn!(
            node_id = %node.node_id,
            connection_id,
            cause = %error.message,
            category = ?error.category,
            hard,
            "node dropped from relay",
        );
        ingress_attempt(&ingress, &node.node_id, AttemptResult::Failure { error, hard }).await?;
        reject_disconnected_commands(&mut commands, Some(cursor));
        failures = failures.saturating_add(1);
        relay_backoff(&mut shutdown, timings, failures, hard).await?;
    }
}

/// Produces an authenticated operator client for `node`, whichever end
/// opened the socket.
///
/// All three arms end in the same handshake. The first two open a
/// connection and hand it straight to it; the third is handed one that a
/// node opened and the listener already matched to this node's name. That
/// symmetry is the point of the change: `LocalNodeClient::adopt` is what
/// `connect`/`connect_loopback` both do after their own `connect` call, so
/// a call-home node is authenticated by exactly the same code, against
/// exactly the same per-node token, as one this relay dialled.
async fn connect_operator(
    node: &C2NodeConfig,
    adopted: Option<TcpStream>,
) -> Result<LocalNodeClient, NodeClientError> {
    match node.route {
        C2NodeRoute::Local => {
            LocalNodeClient::connect(&node.endpoint, &node.node_id, ClientRole::Operator, &node.token).await
        }
        C2NodeRoute::SshForwardedLoopback(endpoint) => {
            LocalNodeClient::connect_loopback(endpoint, &node.node_id, ClientRole::Operator, &node.token).await
        }
        C2NodeRoute::CallHome => {
            // `node_relay_worker` collects the socket before calling this
            // and returns rather than calling without one, so `None` here
            // is a caller bug, not a runtime condition -- reported instead
            // of panicking because a relay worker is not worth aborting a
            // whole C2 over.
            let stream = adopted.ok_or_else(|| {
                NodeClientError::Protocol(
                    "call-home node reached the connect step with no adopted stream".to_owned(),
                )
            })?;
            LocalNodeClient::adopt(stream, &node.node_id, ClientRole::Operator, &node.token).await
        }
    }
}

async fn bounded_node_request(
    client: &mut LocalNodeClient,
    request: NodeRequest,
) -> Result<NodeResponse, NodeClientError> {
    bounded_node_request_with_deadline(client, request, None).await
}

async fn bounded_node_request_with_deadline(
    client: &mut LocalNodeClient,
    request: NodeRequest,
    relay_deadline: Option<Instant>,
) -> Result<NodeResponse, NodeClientError> {
    let deadline = request_budget(&request, relay_deadline, Instant::now());
    if deadline.is_zero() {
        return Err(NodeClientError::Frame(FrameError::PrefixTimedOut));
    }
    match timeout(deadline, client.request(request)).await {
        Ok(result) => result,
        Err(_) => Err(NodeClientError::Frame(FrameError::PrefixTimedOut)),
    }
}

fn relay_request_deadline(request: &NodeRequest, now: Instant) -> Option<Instant> {
    matches!(request, NodeRequest::ResumeSessionRecord { .. })
        .then(|| now + node_request_deadline(request))
}

fn request_budget(request: &NodeRequest, relay_deadline: Option<Instant>, now: Instant) -> Duration {
    relay_deadline
        .map(|deadline| deadline.saturating_duration_since(now))
        .unwrap_or_else(|| node_request_deadline(request))
}

/// Relay bound for a workspace inspection. The node's own inspection
/// budget is 8s by default and 11s at most; this must clear that maximum
/// plus the round trip, or the relay kills a request the node was still
/// legitimately working on -- and killing it drops the node, not just the
/// request. See `node_request_deadline`'s own `InspectWorkspace` arm.
const WORKSPACE_INSPECTION_RELAY_DEADLINE: Duration = Duration::from_secs(15);

fn node_request_deadline(request: &NodeRequest) -> Duration {
    match request {
        NodeRequest::Snapshot
        | NodeRequest::Resync { .. }
        | NodeRequest::ArmHarnessMcpReservation { .. }
        | NodeRequest::ActivateHarnessMcpReservation { .. }
        | NodeRequest::AbortHarnessMcpReservation { .. }
        | NodeRequest::PutHarnessMcpReplyChunk { .. }
        | NodeRequest::RejectHarnessMcpCall { .. }
        | NodeRequest::BrowseHostDirectories { .. }
        | NodeRequest::ReadWorkspaceFile { .. }
        | NodeRequest::WriteWorkspaceFile { .. }
        | NodeRequest::ReadGitHistory { .. }
        | NodeRequest::ReadGitDiff { .. }
        | NodeRequest::AcquireController { .. }
        | NodeRequest::ReleaseController
        | NodeRequest::RenameSessionRecord { .. }
        | NodeRequest::SetSessionTask { .. }
        | NodeRequest::ForgetSessionRecord { .. } => Duration::from_secs(5),
        // The node gives its own workspace inspection 8s by default and up
        // to 11s, and says so when it uses them
        // (`git_time_budget_exceeded`, elapsed over eight seconds on a
        // real repository here). Bounding it from out here at five made
        // that inner budget unreachable: any workspace big enough to spend
        // its own allowance timed out at the relay every single time.
        //
        // And a relay timeout is not a slow answer, it is a dead node --
        // the node is dropped from the relay, its controller lease is
        // released, and every read behind it starts answering
        // "unavailable" until it reattaches, which it then does, and the
        // cycle repeats. One number two seconds too small took the whole
        // stack down in a loop: no inventory, so a spawned session could
        // never be shown, while its process ran perfectly well.
        //
        // The outer bound must therefore clear the node's own MAXIMUM, not
        // its default, with room for the round trip on top.
        NodeRequest::InspectWorkspace { .. } => WORKSPACE_INSPECTION_RELAY_DEADLINE,
        NodeRequest::CreateWorkspaceFile { .. }
        | NodeRequest::CreateWorkspaceDirectory { .. } => {
            WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE
        }
        NodeRequest::CatalogNativeSessions { .. }
        | NodeRequest::PageNativeSessions { .. }
        | NodeRequest::PreviewNativeSession { .. }
        | NodeRequest::IndexNativeSession { .. }
        | NodeRequest::PreviewSessionRecord { .. } => NATIVE_SESSION_REQUEST_DEADLINE,
        NodeRequest::CreateStandaloneWorkspace { .. }
        | NodeRequest::CreateWorktree { .. }
        | NodeRequest::RemoveWorktree { .. }
        | NodeRequest::CleanupManagedWorktree { .. } => Duration::from_secs(240),
        NodeRequest::Spawn { .. }
        | NodeRequest::Resume { .. }
        | NodeRequest::Stop { .. } => Duration::from_secs(15),
        NodeRequest::SpawnSpec { spec } =>
            Duration::from_millis(spec.deadline_ms.get()) + NODE_REQUEST_IO_HEADROOM,
        NodeRequest::SpawnSpecWithHarnessMcp { deadline_unix_ms, .. } =>
            Duration::from_millis(deadline_unix_ms.saturating_sub(unix_ms()))
                + NODE_REQUEST_IO_HEADROOM,
        NodeRequest::SpawnManagedWorktree { request } =>
            Duration::from_millis(request.spawn_spec.deadline_ms.get())
                + NODE_REQUEST_IO_HEADROOM,
        NodeRequest::ResumeSessionRecord { .. } =>
            MANAGED_RESUME_SETTLE_DEADLINE + NODE_REQUEST_IO_HEADROOM,
        _ => Duration::from_secs(10),
    }
}

fn validate_resync(
    previous: u64,
    hello: u64,
    current: u64,
    oldest_available_sequence: u64,
    events: &[NodeEventEnvelope],
) -> Vec<GapKind> {
    let mut gaps = validate_events(
        previous,
        current,
        oldest_available_sequence,
        events,
    );
    if current < hello && !gaps.contains(&GapKind::CursorRegression) {
        gaps.push(GapKind::CursorRegression);
    }
    gaps
}

async fn ingress_attempt(
    ingress: &mpsc::Sender<Attempt>,
    node_id: &NodeId,
    result: AttemptResult,
) -> io::Result<()> {
    ingress.send(Attempt { node_id: node_id.clone(), at_unix_ms: unix_ms(), result })
        .await.map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "inventory owner closed"))
}

async fn relay_backoff(
    shutdown: &mut watch::Receiver<bool>,
    timings: C2Timings,
    failures: usize,
    hard: bool,
) -> io::Result<()> {
    let delay = if hard || failures >= timings.transient_backoffs.len() {
        timings.parked_backoff
    } else {
        timings.transient_backoffs[failures.saturating_sub(1)]
    };
    tokio::select! {
        _ = sleep(delay) => Ok(()),
        _ = shutdown.changed() => {
            Ok(())
        }
    }
}

fn relay_failure_attempt(error: &NodeClientError) -> AttemptResult {
    let (error, hard) = sanitize_node_error(error);
    AttemptResult::Failure { error, hard }
}

async fn handle_relay_command(
    client: &mut LocalNodeClient,
    command: RelayCommand,
    node_id: &NodeId,
    incarnation_id: NodeIncarnationId,
    connection_id: u64,
    controller_owned: &mut bool,
    hub: &OperatorHub,
    cursor: &mut NodeCursor,
    ingress: &mpsc::Sender<Attempt>,
) -> Result<(), NodeClientError> {
    match command {
        RelayCommand::Request { operator_connection_id, expected_incarnation_id, request, reply } => {
            let relay_deadline = relay_request_deadline(&request, Instant::now());
            let expected_spawn = expected_spawn_request(&request);
            let expected_provider_session_index = matches!(
                &request,
                NodeRequest::IndexProviderSession { .. }
            ).then(|| request.clone());
            let expected_native_session = matches!(
                &request,
                NodeRequest::CatalogNativeSessions { .. }
                    | NodeRequest::PageNativeSessions { .. }
                    | NodeRequest::PreviewNativeSession { .. }
                    | NodeRequest::IndexNativeSession { .. }
            ).then(|| request.clone());
            let expected_workspace_content = matches!(
                &request,
                NodeRequest::ReadWorkspaceFile { .. }
                    | NodeRequest::WriteWorkspaceFile { .. }
                    | NodeRequest::CreateWorkspaceFile { .. }
                    | NodeRequest::CreateWorkspaceDirectory { .. }
                    | NodeRequest::ReadGitHistory { .. }
                    | NodeRequest::ReadGitDiff { .. }
            ).then(|| request.clone());
            let expected_session_task = matches!(
                &request,
                NodeRequest::SetSessionTask { .. }
            ).then(|| request.clone());
            let expected_harness_mcp = (request.required_capability()
                == Some(gate4agent_node_protocol::NODE_HARNESS_MCP_READ_PROXY_CAPABILITY))
                .then(|| request.clone());
            if !hub.is_active(operator_connection_id) {
                let _ = reply.send(Err(relay_failure(C2RelayFailureCode::ClientLagged, "C2 operator connection is no longer active", Some(incarnation_id))));
                return Ok(());
            }
            if expected_incarnation_id != incarnation_id {
                let _ = reply.send(Err(relay_failure(C2RelayFailureCode::StaleNodeIncarnation, "node incarnation changed", Some(incarnation_id))));
                return Ok(());
            }
            if matches!(&request, NodeRequest::IndexNativeSession { selection, .. }
                if selection.route.scope
                    != gate4agent_node_protocol::NativeSessionCatalogScope::Workspace)
            {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::RequestForbidden,
                    "external native sessions must be registered as workspaces before indexing",
                    Some(incarnation_id),
                )));
                return Ok(());
            }
            if !is_read_only_request(&request) && !*controller_owned {
                match acquire_controller_with_deadline(client, connection_id, relay_deadline).await {
                    Ok(owned) if owned => *controller_owned = true,
                    Ok(_) => {
                        let _ = reply.send(Err(relay_failure(C2RelayFailureCode::RelayBusy, "node controller lease is unavailable", Some(incarnation_id))));
                        return Ok(());
                    }
                    Err(error) if relay_node_failure(&error).is_some() => {
                        let failure = relay_node_failure(&error)
                            .expect("guarded node request failure");
                        let _ = reply.send(Ok(RoutedNodeResponse {
                            node_id: node_id.clone(),
                            incarnation_id,
                            response: Err(failure),
                        }));
                        return Ok(());
                    }
                    Err(error) => {
                        let _ = reply.send(Err(relay_failure(C2RelayFailureCode::NodeOffline, "node relay disconnected", Some(incarnation_id))));
                        return Err(error);
                    }
                }
            }
            let response = match bounded_node_request_with_deadline(client, request, relay_deadline).await {
                Ok(response) => Ok(response),
                Err(NodeClientError::Node(failure)) => Err(failure),
                Err(error @ NodeClientError::UnsupportedCapability(_)) => {
                    let failure = relay_node_failure(&error)
                        .expect("unsupported capability is a routed node failure");
                    let _ = reply.send(Ok(RoutedNodeResponse {
                        node_id: node_id.clone(),
                        incarnation_id,
                        response: Err(failure),
                    }));
                    return Ok(());
                }
                Err(error) => {
                    let _ = reply.send(Err(relay_failure(C2RelayFailureCode::NodeOffline, "node relay disconnected", Some(incarnation_id))));
                    return Err(error);
                }
            };
            if let Err(message) = validate_spawn_spec_response(
                expected_spawn.as_ref(),
                &response,
                incarnation_id,
            ) {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay returned an invalid spawn receipt",
                    Some(incarnation_id),
                )));
                return Err(NodeClientError::Protocol(message.to_owned()));
            }
            if let Err(message) = validate_provider_session_index_response(
                expected_provider_session_index.as_ref(),
                &response,
            ) {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay returned an invalid provider session index response",
                    Some(incarnation_id),
                )));
                return Err(NodeClientError::Protocol(message.to_owned()));
            }
            if let Err(message) = validate_native_session_response(
                expected_native_session.as_ref(),
                &response,
            ) {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay returned an invalid native session response",
                    Some(incarnation_id),
                )));
                return Err(NodeClientError::Protocol(message.to_owned()));
            }
            if let Err(message) = validate_workspace_content_response(
                expected_workspace_content.as_ref(),
                &response,
            ) {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay returned an invalid workspace content response",
                    Some(incarnation_id),
                )));
                return Err(NodeClientError::Protocol(message.to_owned()));
            }
            if let Err(message) = validate_session_task_response(
                expected_session_task.as_ref(),
                &response,
            ) {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay returned an invalid session task response",
                    Some(incarnation_id),
                )));
                return Err(NodeClientError::Protocol(message.to_owned()));
            }
            if let Err(message) = validate_harness_mcp_response(
                expected_harness_mcp.as_ref(),
                &response,
            ) {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay returned an invalid harness MCP response",
                    Some(incarnation_id),
                )));
                return Err(NodeClientError::Protocol(message.to_owned()));
            }
            drain_pending_events(client, node_id, cursor, hub, ingress, false, relay_deadline).await?;
            update_inventory_from_response(node_id, cursor, &response, ingress).await
                .map_err(NodeClientError::Io)?;
            let response = response
                .map(|response| C2NodeResponse::from_node_response_with_control_detail(&response))
                .map_err(|failure| C2NodeFailure::from(&failure));
            let _ = reply.send(Ok(RoutedNodeResponse { node_id: node_id.clone(), incarnation_id, response }));
        }
    }
    Ok(())
}

fn validate_session_task_response(
    expected: Option<&NodeRequest>,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
) -> Result<(), &'static str> {
    match (expected, response) {
        (Some(_), Err(_)) | (None, Err(_)) => Ok(()),
        (
            Some(NodeRequest::SetSessionTask {
                record_id,
                expected_revision,
                target,
            }),
            Ok(NodeResponse::SessionRecordUpdated { record }),
        ) if session_task_record_matches(record, record_id, *expected_revision, target) => Ok(()),
        (Some(NodeRequest::SetSessionTask { .. }), Ok(_)) => {
            Err("session task response does not match the routed request")
        }
        (None, Ok(_)) => Ok(()),
        (Some(_), Ok(_)) => Ok(()),
    }
}

fn validate_harness_mcp_response(
    expected: Option<&NodeRequest>,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
) -> Result<(), &'static str> {
    use NodeRequest as Request;
    use NodeResponse as Response;
    let valid = match (expected, response) {
        (Some(Request::ArmHarnessMcpReservation { reservation_id, activation_digest, expires_at_unix_ms, .. }),
            Ok(Response::Armed { reservation_id: echoed_id, activation_digest: echoed_digest, expires_at_unix_ms: echoed_expiry })) =>
            reservation_id == echoed_id && activation_digest == echoed_digest
                && expires_at_unix_ms == echoed_expiry,
        (Some(Request::SpawnSpecWithHarnessMcp { reservation_id, activation_digest, .. }),
            Ok(Response::Spawned { reservation_id: echoed_id, activation_digest: echoed_digest, receipt })) =>
            reservation_id == echoed_id && activation_digest == echoed_digest
                && receipt.harness_mcp_proxy.as_ref().is_some_and(|proxy| {
                    &proxy.reservation_id == reservation_id
                        && &proxy.activation_digest == activation_digest
                }),
        (Some(Request::ActivateHarnessMcpReservation { reservation_id, activation_digest, record_id, session }),
            Ok(Response::Activated { reservation_id: echoed_id, activation_digest: echoed_digest, record_id: echoed_record, session: echoed_session })) =>
            reservation_id == echoed_id && activation_digest == echoed_digest
                && record_id == echoed_record && session == echoed_session,
        (Some(Request::AbortHarnessMcpReservation { reservation_id, activation_digest }),
            Ok(Response::Aborted { reservation_id: echoed_id, activation_digest: echoed_digest })) =>
            reservation_id == echoed_id && activation_digest == echoed_digest,
        (Some(Request::PutHarnessMcpReplyChunk { reservation_id, activation_digest, record_id, session, call_id, offset, final_chunk, chunk_hex }),
            Ok(Response::ReplyChunkAccepted { reservation_id: echoed_id, activation_digest: echoed_digest, record_id: echoed_record, session: echoed_session, call_id: echoed_call, next_offset, completed })) =>
            reservation_id == echoed_id && activation_digest == echoed_digest
                && record_id == echoed_record && session == echoed_session && call_id == echoed_call
                && offset.checked_add(u32::try_from(chunk_hex.raw_len()).unwrap_or(u32::MAX))
                    == Some(*next_offset) && completed == final_chunk,
        (Some(Request::RejectHarnessMcpCall { reservation_id, activation_digest, record_id, session, call_id, .. }),
            Ok(Response::CallRejected { reservation_id: echoed_id, activation_digest: echoed_digest, record_id: echoed_record, session: echoed_session, call_id: echoed_call })) =>
            reservation_id == echoed_id && activation_digest == echoed_digest
                && record_id == echoed_record && session == echoed_session && call_id == echoed_call,
        (Some(_), Err(_)) | (None, Err(_)) => true,
        (Some(_), Ok(_)) => false,
        (None, Ok(response)) => !response.requires_harness_mcp_proxy_capability(),
    };
    if valid { Ok(()) } else { Err("harness MCP response does not match the routed request") }
}

fn session_task_record_matches(
    record: &gate4agent_node_protocol::ManagedSessionRecord,
    record_id: &gate4agent_node_protocol::SessionRecordId,
    expected_revision: u64,
    target: &gate4agent_node_protocol::SessionTaskTargetV1,
) -> bool {
    if &record.record_id != record_id { return false; }
    let next_revision = expected_revision.checked_add(1);
    match target {
        gate4agent_node_protocol::SessionTaskTargetV1::New => record.task_binding.as_ref()
            .is_some_and(|binding| Some(binding.revision) == next_revision && binding.task_id.is_some()),
        gate4agent_node_protocol::SessionTaskTargetV1::Existing { task_id } => record.task_binding.as_ref()
            .is_some_and(|binding| (binding.revision == expected_revision || Some(binding.revision) == next_revision)
                && binding.task_id.as_ref() == Some(task_id)),
        gate4agent_node_protocol::SessionTaskTargetV1::Clear => match &record.task_binding {
            None => expected_revision == 0,
            Some(binding) => binding.task_id.is_none()
                && (binding.revision == expected_revision || Some(binding.revision) == next_revision),
        },
    }
}

fn validate_workspace_content_response(
    expected: Option<&NodeRequest>,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
) -> Result<(), &'static str> {
    match (expected, response) {
        (Some(_), Err(_)) | (None, Err(_)) => Ok(()),
        (
            Some(NodeRequest::ReadWorkspaceFile { workspace_id, path }),
            Ok(NodeResponse::WorkspaceFileRead { file }),
        ) if &file.workspace_id == workspace_id && &file.path == path => Ok(()),
        (
            Some(NodeRequest::WriteWorkspaceFile { workspace_id, path, text, .. }),
            Ok(NodeResponse::WorkspaceFileWritten { file }),
        ) if &file.workspace_id == workspace_id
            && &file.path == path
            && matches!(
                &file.content,
                gate4agent_node_protocol::WorkspaceFileContent::Utf8 {
                    text: written,
                    byte_len,
                } if written == text
                    && u32::try_from(text.len()).ok() == Some(*byte_len)
            ) => Ok(()),
        (
            Some(NodeRequest::CreateWorkspaceFile { workspace_id, path }),
            Ok(NodeResponse::WorkspaceFileCreated { file }),
        ) if &file.workspace_id == workspace_id
            && &file.path == path
            && file.revision.is_some()
            && matches!(
                &file.content,
                gate4agent_node_protocol::WorkspaceFileContent::Utf8 {
                    text,
                    byte_len: 0,
                } if text.is_empty()
            ) => Ok(()),
        (
            Some(NodeRequest::CreateWorkspaceDirectory { workspace_id, path }),
            Ok(NodeResponse::WorkspaceDirectoryCreated {
                workspace_id: actual_workspace_id,
                entry,
            }),
        ) if actual_workspace_id == workspace_id
            && &entry.relative_path == path
            && entry.kind == gate4agent_node_protocol::WorkspaceEntryKind::Directory => Ok(()),
        (
            Some(NodeRequest::ReadGitHistory { workspace_id, .. }),
            Ok(NodeResponse::GitHistoryRead { workspace_id: actual, .. }),
        ) if actual == workspace_id => Ok(()),
        (
            Some(NodeRequest::ReadGitDiff { workspace_id, request }),
            Ok(NodeResponse::GitDiffRead { workspace_id: actual, diff }),
        ) if actual == workspace_id && diff.mode == request.mode && diff.path == request.path => Ok(()),
        (Some(_), Ok(_)) => Err("workspace content response does not match routed request"),
        (
            None,
            Ok(NodeResponse::WorkspaceFileRead { .. }
                | NodeResponse::WorkspaceFileWritten { .. }
                | NodeResponse::WorkspaceFileCreated { .. }
                | NodeResponse::WorkspaceDirectoryCreated { .. }
                | NodeResponse::GitHistoryRead { .. }
                | NodeResponse::GitDiffRead { .. }),
        ) => Err("unexpected workspace content response"),
        (None, Ok(_)) => Ok(()),
    }
}

enum ExpectedSpawnRequest {
    Spec(SpawnSpec),
    ManagedV1(ManagedWorktreeSpawnRequest),
    ManagedV2(ManagedWorktreeSpawnRequestV2),
}

fn expected_spawn_request(request: &NodeRequest) -> Option<ExpectedSpawnRequest> {
    match request {
        NodeRequest::SpawnSpec { spec } => Some(ExpectedSpawnRequest::Spec(spec.clone())),
        NodeRequest::SpawnManagedWorktree { request } => {
            Some(ExpectedSpawnRequest::ManagedV1(request.clone()))
        }
        NodeRequest::SpawnManagedWorktreeV2 { request } => {
            Some(ExpectedSpawnRequest::ManagedV2(request.clone()))
        }
        _ => None,
    }
}

fn validate_provider_session_index_response(
    expected: Option<&NodeRequest>,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
) -> Result<(), &'static str> {
    match (expected, response) {
        (
            Some(NodeRequest::IndexProviderSession {
                workspace_id,
                provider,
                identity,
                ..
            }),
            Ok(NodeResponse::ProviderSessionIndexed { record }),
        ) if &record.workspace_id == workspace_id
            && &record.provider == provider
            && record.provider_session.as_ref() == Some(identity) => Ok(()),
        (Some(_), Ok(NodeResponse::ProviderSessionIndexed { .. })) => {
            Err("provider session index response does not match routed request")
        }
        (Some(_), Ok(_)) => Err("provider session index request returned a different response"),
        (None, Ok(NodeResponse::ProviderSessionIndexed { .. })) => {
            Err("unexpected provider session index response for a different node request")
        }
        (Some(_), Err(_)) | (None, _) => Ok(()),
    }
}

fn validate_native_session_response(
    expected: Option<&NodeRequest>,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
) -> Result<(), &'static str> {
    match (expected, response) {
        (
            Some(NodeRequest::CatalogNativeSessions { route, .. }),
            Ok(NodeResponse::NativeSessionsCataloged {
                route: echoed_route,
                ..
            }),
        ) if echoed_route == route => Ok(()),
        (
            Some(NodeRequest::PageNativeSessions {
                route,
                window,
                catalog_revision,
                ..
            }),
            Ok(NodeResponse::NativeSessionsPaged {
                route: echoed_route,
                page,
            }),
        ) if echoed_route == route
            && page.window == *window
            && page.revision == *catalog_revision => Ok(()),
        (
            Some(NodeRequest::PreviewNativeSession { selection, .. }),
            Ok(NodeResponse::NativeSessionPreviewed {
                selection: echoed_selection,
                ..
            }),
        ) if echoed_selection == selection => Ok(()),
        (
            Some(NodeRequest::IndexNativeSession { selection, .. }),
            Ok(NodeResponse::NativeSessionIndexed {
                selection: echoed_selection,
                record,
            }),
        ) if echoed_selection == selection
            && selection.route.scope
                == gate4agent_node_protocol::NativeSessionCatalogScope::Workspace
            && selection.route.workspace_id.as_ref() == Some(&record.workspace_id)
            && selection.route.provider == record.provider => Ok(()),
        (Some(_), Err(_)) => Ok(()),
        (Some(_), Ok(_)) => Err("native session response does not match routed request"),
        (
            None,
            Ok(
                NodeResponse::NativeSessionsCataloged { .. }
                | NodeResponse::NativeSessionsPaged { .. }
                | NodeResponse::NativeSessionPreviewed { .. }
                | NodeResponse::NativeSessionIndexed { .. },
            ),
        ) => Err("unexpected native session response for a different node request"),
        (None, _) => Ok(()),
    }
}

fn validate_spawn_spec_response(
    expected: Option<&ExpectedSpawnRequest>,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
    relay_incarnation_id: NodeIncarnationId,
) -> Result<(), &'static str> {
    match (expected, response) {
        (Some(ExpectedSpawnRequest::Spec(spec)), Ok(NodeResponse::SpawnSpecAccepted { receipt })) => {
            validate_spawn_receipt(spec, receipt, relay_incarnation_id)
        }
        (
            Some(ExpectedSpawnRequest::ManagedV1(request)),
            Ok(NodeResponse::ManagedWorktreeSpawnAccepted { receipt }),
        ) => validate_managed_spawn_receipt(
            &request.spawn_spec,
            &request.worktree_profile_id,
            receipt,
            relay_incarnation_id,
        ),
        (
            Some(ExpectedSpawnRequest::ManagedV2(request)),
            Ok(NodeResponse::ManagedWorktreeSpawnAccepted { receipt }),
        ) if receipt.lease.profile_revision == request.expected_profile_revision => {
            validate_managed_spawn_receipt(
                &request.spawn_spec,
                &request.worktree_profile_id,
                receipt,
                relay_incarnation_id,
            )
        }
        (
            Some(ExpectedSpawnRequest::ManagedV2(_)),
            Ok(NodeResponse::ManagedWorktreeSpawnAccepted { .. }),
        ) => Err("managed spawn receipt profile revision does not match routed request"),
        (Some(_), Ok(_)) => return Err("spawn spec request returned a different response"),
        (Some(_), Err(_)) => return Ok(()),
        (None, Ok(NodeResponse::SpawnSpecAccepted { .. }
            | NodeResponse::ManagedWorktreeSpawnAccepted { .. })) => {
            return Err("unexpected spawn receipt for a different node request");
        }
        (None, _) => return Ok(()),
    }
}

fn validate_managed_spawn_receipt(
    spawn_spec: &SpawnSpec,
    worktree_profile_id: &gate4agent_node_protocol::WorktreeProfileId,
    receipt: &gate4agent_node_protocol::ManagedWorktreeSpawnReceipt,
    relay_incarnation_id: NodeIncarnationId,
) -> Result<(), &'static str> {
    if receipt.lease.source_workspace_id != spawn_spec.target.workspace_id
        || &receipt.lease.profile_id != worktree_profile_id
        || receipt.lease.state != ManagedWorktreeLeaseState::InUse
        || receipt.lease.cleanup_failure.is_some()
        || receipt.lease.active_session_count != 1
        || receipt.spawn.target.node_id != spawn_spec.target.node_id
        || receipt.spawn.target.workspace_id != spawn_spec.target.workspace_id
        || receipt.spawn.target.worktree_id.as_ref() != Some(&receipt.lease.workspace_id)
        || receipt.spawn.session.workspace_id != receipt.lease.workspace_id
    {
        return Err("managed spawn receipt does not match routed request");
    }
    let mut resolved_spec = spawn_spec.clone();
    resolved_spec.target.worktree_id = Some(receipt.lease.workspace_id.clone());
    validate_spawn_receipt(&resolved_spec, &receipt.spawn, relay_incarnation_id)
}

fn validate_spawn_receipt(
    spec: &SpawnSpec,
    receipt: &ResolvedSpawnReceipt,
    relay_incarnation_id: NodeIncarnationId,
) -> Result<(), &'static str> {
    if receipt.incarnation_id != relay_incarnation_id
        || !receipt.context_binding_is_valid()
        || receipt.target != spec.target
        || receipt.profile_id != spec.profile_id
        || receipt.idempotency_key != spec.idempotency_key
        || receipt.deadline_ms != spec.deadline_ms
        || receipt.required_capabilities != spec.required_capabilities
        || receipt.bundle.as_ref().is_some_and(|bundle| {
            receipt.bundle_id.as_ref() != Some(&bundle.id)
        })
        || &receipt.session.workspace_id
            != spec
                .target
                .worktree_id
                .as_ref()
                .unwrap_or(&spec.target.workspace_id)
    {
        return Err("spawn receipt does not match routed request");
    }
    if !required_override_matches(&spec.overrides.provider, &receipt.provider)
        || !required_override_matches(&spec.overrides.mode, &receipt.mode)
        || !required_override_matches(&spec.overrides.terminal_size, &receipt.terminal_size)
        || !optional_override_matches(&spec.overrides.bundle_id, &receipt.bundle_id)
        || !optional_override_matches(&spec.overrides.context_id, &receipt.context_id)
        || !environment_profile_override_matches(
            &spec.overrides.environment_profile_id,
            receipt.environment_profile.as_ref(),
        )
    {
        return Err("spawn receipt contradicts explicit overrides");
    }
    match &spec.overrides.prompt {
        SpawnOverride::Inherit => {}
        SpawnOverride::Set { value }
            if receipt.prompt.present
                && receipt.prompt.byte_len == u32::try_from(value.byte_len()).unwrap_or(0) => {}
        SpawnOverride::Clear if !receipt.prompt.present && receipt.prompt.byte_len == 0 => {}
        SpawnOverride::Set { .. } | SpawnOverride::Clear => {
            return Err("spawn receipt contradicts explicit prompt override");
        }
    }
    Ok(())
}

fn environment_profile_override_matches(
    expected: &SpawnOverride<gate4agent_node_protocol::SpawnEnvironmentProfileId>,
    actual: Option<&gate4agent_node_protocol::ResolvedEnvironmentProfileReceipt>,
) -> bool {
    match expected {
        SpawnOverride::Inherit => true,
        SpawnOverride::Set { value } => {
            actual.is_some_and(|receipt| &receipt.profile_id == value)
        }
        SpawnOverride::Clear => actual.is_none(),
    }
}

fn required_override_matches<T: Eq>(override_value: &SpawnOverride<T>, actual: &T) -> bool {
    match override_value {
        SpawnOverride::Inherit => true,
        SpawnOverride::Set { value } => value == actual,
        SpawnOverride::Clear => false,
    }
}

fn optional_override_matches<T: Eq>(
    override_value: &SpawnOverride<T>,
    actual: &Option<T>,
) -> bool {
    match override_value {
        SpawnOverride::Inherit => true,
        SpawnOverride::Set { value } => actual.as_ref() == Some(value),
        SpawnOverride::Clear => actual.is_none(),
    }
}

fn relay_node_failure(error: &NodeClientError) -> Option<C2NodeFailure> {
    match error {
        NodeClientError::Node(failure) => Some(C2NodeFailure::from(failure)),
        NodeClientError::UnsupportedCapability(_) => Some(C2NodeFailure {
            code: NodeFailureCode::UnsupportedCapability,
            message: "required capability unavailable".to_owned(),
        }),
        NodeClientError::Io(_)
        | NodeClientError::Frame(_)
        | NodeClientError::Protocol(_)
        | NodeClientError::BuildStampMismatch { .. }
        | NodeClientError::AuthenticationTimedOut
        | NodeClientError::Authentication(_)
        | NodeClientError::RequestIdExhausted => None,
    }
}

fn is_read_only_request(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 { .. }
    )
}

async fn acquire_controller(
    client: &mut LocalNodeClient,
    connection_id: u64,
) -> Result<bool, NodeClientError> {
    acquire_controller_with_deadline(client, connection_id, None).await
}

async fn acquire_controller_with_deadline(
    client: &mut LocalNodeClient,
    connection_id: u64,
    relay_deadline: Option<Instant>,
) -> Result<bool, NodeClientError> {
    match bounded_node_request_with_deadline(
        client,
        NodeRequest::AcquireController { lease_ms: gate4agent_node_protocol::MAX_CONTROLLER_LEASE_MS },
        relay_deadline,
    ).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".to_owned())),
    }
}

async fn release_controller(
    client: &mut LocalNodeClient,
    controller_owned: &mut bool,
) -> Result<(), NodeClientError> {
    if !*controller_owned { return Ok(()); }
    match bounded_node_request(client, NodeRequest::ReleaseController).await? {
        NodeResponse::Controller { .. } => { *controller_owned = false; Ok(()) }
        _ => Err(NodeClientError::Protocol("controller release returned a different response".to_owned())),
    }
}

async fn update_inventory_from_response(
    node_id: &NodeId,
    cursor: &mut NodeCursor,
    response: &Result<NodeResponse, gate4agent_node_protocol::NodeFailure>,
    ingress: &mpsc::Sender<Attempt>,
) -> io::Result<()> {
    match response {
        Ok(NodeResponse::Snapshot { event_sequence, snapshot, .. })
        | Ok(NodeResponse::Resync { event_sequence, snapshot, .. }) => {
            if *event_sequence < cursor.sequence { return Ok(()); }
            cursor.sequence = *event_sequence;
            ingress_attempt(ingress, node_id, AttemptResult::Success { cursor: *cursor, snapshot: snapshot.clone(), gaps: Vec::new() }).await
        }
        _ => Ok(()),
    }
}

fn publish_recovered_events(
    node_id: &NodeId,
    incarnation_id: NodeIncarnationId,
    events: &[NodeEventEnvelope],
    hub: &OperatorHub,
) {
    for envelope in events {
        if let Some(event) = routed_recovered_node_event(node_id, incarnation_id, envelope) {
            hub.publish(event);
        }
    }
}

fn routed_recovered_node_event(
    node_id: &NodeId,
    incarnation_id: NodeIncarnationId,
    envelope: &NodeEventEnvelope,
) -> Option<RoutedNodeEvent> {
    (!matches!(&envelope.event, NodeEvent::HarnessMcpReadCall { .. })).then(|| {
        RoutedNodeEvent {
            node_id: node_id.clone(),
            cursor: NodeCursor { incarnation_id, sequence: envelope.sequence },
            event: C2NodeEvent::from_node_event_with_control_detail(&envelope.event),
        }
    })
}

fn routed_transient_node_event(
    node_id: &NodeId,
    cursor: NodeCursor,
    envelope: &NodeEventEnvelope,
) -> Option<RoutedNodeEvent> {
    matches!(&envelope.event, NodeEvent::HarnessMcpReadCall { .. }).then(|| RoutedNodeEvent {
        node_id: node_id.clone(),
        cursor,
        event: C2NodeEvent::from(&envelope.event),
    })
}

/// `NodeEvent::AgentStream` chunks are published unconditionally -- never
/// gated on `cursor` contiguity, and never the cause of a
/// `CursorRegression`/`NonContiguousEvents` gap -- but, unlike
/// `HarnessMcpReadCall` (always `sequence: 0`, see
/// `gate4agent-node/src/server.rs::publish_transient`), an
/// `AgentStreamChunkV1` envelope draws its `sequence` from the
/// SAME counter every durable `NodeEvent` shares
/// (`gate4agent-node/src/server.rs::publish_agent_stream_chunk`'s own doc,
/// there purely so the node's per-connection discard watermark can compare
/// it against everything else). So a chunk really does consume a slot in
/// the durable sequence space, and `routed_transient_node_event`'s
/// "leave `cursor` untouched" rule -- correct for a call that never had a
/// slot to begin with -- would otherwise make the very next durable event
/// look one short of contiguous, forcing a resync round trip on every
/// single provider turn that streams so much as one chunk.
///
/// This folds the chunk's sequence into `cursor` with `max` instead: on the
/// ordinary, non-bursty path (`Control(N)` then its own `AgentStream(N+1)`,
/// each one delivered as they are produced) that keeps `cursor` moving
/// exactly as it did before this function existed, so the very next
/// `Control(N+2)` still finds `cursor == N+1` and stays contiguous. On the
/// bursty path -- the one actually measured live: 19 chunks published
/// `source_sequence` 22-40 in a 17ms window, `envelope_sequence` interleaved
/// 1:1 with `Control`, only 2 of the 19 ever reached the harness -- the
/// node's connection loop can drain its durable channel far enough ahead of
/// this dedicated one that a lower-sequence chunk arrives at this relay
/// AFTER `cursor` has already been carried past it by a resync recovering
/// the higher-sequence `Control` envelopes around it. `max` refuses to let
/// that late chunk rewind `cursor` -- it is still published (chunks promise
/// no resync, so there is nothing to recover if it were dropped instead:
/// `gate4agent-node-protocol::AgentStreamChunkV1`'s own doc, "no
/// `ObservationV1` resync promise") -- it simply stops mattering to the
/// durable cursor's own contiguity bookkeeping once something newer has
/// already passed it by. Ordering and true loss within the agent-stream
/// channel itself remain the node's own broadcast `Lagged` warn to name,
/// not this cursor's to adjudicate.
fn route_agent_stream_event(
    node_id: &NodeId,
    cursor: &mut NodeCursor,
    envelope: &NodeEventEnvelope,
) -> Option<RoutedNodeEvent> {
    if !matches!(&envelope.event, NodeEvent::AgentStream { .. }) {
        return None;
    }
    let routed = RoutedNodeEvent {
        node_id: node_id.clone(),
        cursor: *cursor,
        event: C2NodeEvent::from(&envelope.event),
    };
    cursor.sequence = cursor.sequence.max(envelope.sequence);
    Some(routed)
}

async fn drain_pending_events(
    client: &mut LocalNodeClient,
    node_id: &NodeId,
    cursor: &mut NodeCursor,
    hub: &OperatorHub,
    ingress: &mpsc::Sender<Attempt>,
    skip_replayed: bool,
    relay_deadline: Option<Instant>,
) -> Result<(), NodeClientError> {
    let mut skip_replayed = skip_replayed;
    for repair_pass in 0..=1 {
        let mut gaps = Vec::new();
        let mut managed_worktree_events = Vec::new();
        let mut changed = false;
        let mut repair = false;
        while let Some(envelope) = client.take_event() {
            if let Some(event) = routed_transient_node_event(node_id, *cursor, &envelope) {
                hub.publish(event);
                continue;
            }
            if let Some(event) = route_agent_stream_event(node_id, cursor, &envelope) {
                hub.publish(event);
                continue;
            }
            if skip_replayed && envelope.sequence <= cursor.sequence { continue; }
            let resync_required = matches!(&envelope.event, gate4agent_node_protocol::NodeEvent::ResyncRequired { .. });
            if resync_required || envelope.sequence != cursor.sequence.saturating_add(1) {
                gaps.push(if envelope.sequence <= cursor.sequence && !resync_required {
                    GapKind::CursorRegression
                } else if resync_required {
                    GapKind::HistoryEvicted
                } else {
                    GapKind::NonContiguousEvents
                });
                repair = true;
                continue;
            }
            let event_cursor = NodeCursor { incarnation_id: cursor.incarnation_id, sequence: envelope.sequence };
            if matches!(
                &envelope.event,
                NodeEvent::ManagedWorktreeUpserted { .. }
                    | NodeEvent::ManagedWorktreeRemoved { .. }
            ) {
                managed_worktree_events.push(envelope.event.clone());
            }
            hub.publish(RoutedNodeEvent {
                node_id: node_id.clone(),
                cursor: event_cursor,
                event: C2NodeEvent::from_node_event_with_control_detail(&envelope.event),
            });
            cursor.sequence = envelope.sequence;
            changed = true;
        }
        if !repair {
            if changed || !gaps.is_empty() {
                ingress_attempt(ingress, node_id, AttemptResult::Cursor {
                    cursor: *cursor,
                    gaps,
                    managed_worktree_events,
                }).await
                    .map_err(NodeClientError::Io)?;
            }
            return Ok(());
        }
        if repair_pass == 1 {
            ingress_attempt(ingress, node_id, AttemptResult::Cursor {
                cursor: *cursor,
                gaps,
                managed_worktree_events,
            }).await
                .map_err(NodeClientError::Io)?;
            return Err(NodeClientError::Protocol("node event stream remained noncontiguous after resync".to_owned()));
        }
        let after_sequence = cursor.sequence;
        let response = bounded_node_request_with_deadline(
            client,
            NodeRequest::Resync { after_sequence },
            relay_deadline,
        ).await?;
        let NodeResponse::Resync {
            event_sequence,
            oldest_available_sequence,
            snapshot,
            events,
        } = response else {
            return Err(NodeClientError::Protocol("event repair resync returned a different response".to_owned()));
        };
        let repair_gaps = validate_events(
            after_sequence,
            event_sequence,
            oldest_available_sequence,
            &events,
        );
        let contiguous = repair_gaps.is_empty();
        gaps = repair_gaps;
        if contiguous {
            for envelope in events.iter().filter(|event| event.sequence > after_sequence) {
                if let Some(event) = routed_recovered_node_event(
                    node_id,
                    cursor.incarnation_id,
                    envelope,
                ) {
                    hub.publish(event);
                }
            }
        } else {
            hub.publish(RoutedNodeEvent {
                node_id: node_id.clone(),
                cursor: NodeCursor { incarnation_id: cursor.incarnation_id, sequence: event_sequence },
                event: C2NodeEvent::ResyncRequired {
                    oldest_available_sequence,
                },
            });
        }
        cursor.sequence = event_sequence;
        ingress_attempt(ingress, node_id, AttemptResult::Success { cursor: *cursor, snapshot, gaps }).await
            .map_err(NodeClientError::Io)?;
        skip_replayed = true;
    }
    Ok(())
}

async fn handle_live_node_event(
    client: &mut LocalNodeClient,
    node_id: &NodeId,
    envelope: NodeEventEnvelope,
    cursor: &mut NodeCursor,
    hub: &OperatorHub,
    ingress: &mpsc::Sender<Attempt>,
) -> Result<(), NodeClientError> {
    if let Some(event) = routed_transient_node_event(node_id, *cursor, &envelope) {
        hub.publish(event);
        return Ok(());
    }
    if let Some(event) = route_agent_stream_event(node_id, cursor, &envelope) {
        hub.publish(event);
        return Ok(());
    }
    if live_event_gap(cursor.sequence, &envelope).is_some() {
        let after_sequence = cursor.sequence;
        let response = bounded_node_request(
            client,
            NodeRequest::Resync { after_sequence },
        ).await?;
        let NodeResponse::Resync {
            event_sequence,
            oldest_available_sequence,
            snapshot,
            events,
        } = response else {
            return Err(NodeClientError::Protocol(
                "live event repair resync returned a different response".to_owned(),
            ));
        };
        let repair_gaps = validate_events(
            after_sequence,
            event_sequence,
            oldest_available_sequence,
            &events,
        );
        if repair_gaps.is_empty() {
            publish_recovered_events(node_id, cursor.incarnation_id, &events, hub);
        } else {
            hub.publish(RoutedNodeEvent {
                node_id: node_id.clone(),
                cursor: NodeCursor {
                    incarnation_id: cursor.incarnation_id,
                    sequence: event_sequence,
                },
                event: C2NodeEvent::ResyncRequired {
                    oldest_available_sequence,
                },
            });
        }
        cursor.sequence = event_sequence;
        ingress_attempt(
            ingress,
            node_id,
            AttemptResult::Success {
                cursor: *cursor,
                snapshot,
                gaps: repair_gaps,
            },
        ).await.map_err(NodeClientError::Io)?;
        return drain_pending_events(
            client,
            node_id,
            cursor,
            hub,
            ingress,
            true,
            None,
        ).await;
    }

    cursor.sequence = envelope.sequence;
    let managed_worktree_events = matches!(
        &envelope.event,
        NodeEvent::ManagedWorktreeUpserted { .. }
            | NodeEvent::ManagedWorktreeRemoved { .. }
    )
    .then(|| vec![envelope.event.clone()])
    .unwrap_or_default();
    hub.publish(RoutedNodeEvent {
        node_id: node_id.clone(),
        cursor: *cursor,
        event: C2NodeEvent::from_node_event_with_control_detail(&envelope.event),
    });
    ingress_attempt(
        ingress,
        node_id,
        AttemptResult::Cursor {
            cursor: *cursor,
            gaps: Vec::new(),
            managed_worktree_events,
        },
    ).await.map_err(NodeClientError::Io)
}

fn live_event_gap(previous: u64, envelope: &NodeEventEnvelope) -> Option<GapKind> {
    if matches!(
        &envelope.event,
        gate4agent_node_protocol::NodeEvent::ResyncRequired { .. }
    ) {
        return Some(GapKind::HistoryEvicted);
    }
    if envelope.sequence <= previous {
        return Some(GapKind::CursorRegression);
    }
    (envelope.sequence != previous.saturating_add(1))
        .then_some(GapKind::NonContiguousEvents)
}

fn reject_disconnected_commands(commands: &mut mpsc::Receiver<RelayCommand>, cursor: Option<NodeCursor>) {
    while let Ok(command) = commands.try_recv() {
        match command {
            RelayCommand::Request { reply, .. } => {
                let _ = reply.send(Err(relay_failure(
                    C2RelayFailureCode::NodeOffline,
                    "node relay disconnected before request dispatch",
                    cursor.map(|value| value.incarnation_id),
                )));
            }
        }
    }
}

fn acknowledge_disconnected_releases(releases: &mut mpsc::Receiver<oneshot::Sender<()>>) {
    while let Ok(reply) = releases.try_recv() { let _ = reply.send(()); }
}

fn validate_events(
    previous: u64,
    current: u64,
    oldest_available_sequence: u64,
    events: &[NodeEventEnvelope],
) -> Vec<GapKind> {
    if current < previous { return vec![GapKind::CursorRegression]; }
    let max_floor = current.checked_add(1).unwrap_or(u64::MAX);
    let minimum_event_sequence = previous
        .checked_add(1)
        .unwrap_or(u64::MAX)
        .max(oldest_available_sequence);
    if oldest_available_sequence == 0
        || oldest_available_sequence > max_floor
        || (current == previous && !events.is_empty())
        || events.iter().any(|event| {
            event.sequence < minimum_event_sequence || event.sequence > current
        })
        || events.windows(2).any(|pair| pair[0].sequence >= pair[1].sequence)
    {
        return vec![GapKind::NonContiguousEvents];
    }
    if previous.saturating_add(1) < oldest_available_sequence {
        return vec![GapKind::HistoryEvicted];
    }
    Vec::new()
}

fn sanitize_node_error(error: &NodeClientError) -> (SanitizedError, bool) {
    let (category, message, hard) = match error {
        NodeClientError::Protocol(message) if message.contains("identity mismatch") =>
            (C2ErrorCategory::Identity, "node identity mismatch", true),
        NodeClientError::Protocol(message) if message.contains("access-token proof") || message.contains("access denied") =>
            (C2ErrorCategory::Authentication, "node authentication failed", true),
        NodeClientError::Protocol(_) | NodeClientError::Frame(FrameError::Json(_) | FrameError::InvalidLength { .. }) =>
            (C2ErrorCategory::Protocol, "node protocol failed", true),
        NodeClientError::BuildStampMismatch { .. } =>
            (C2ErrorCategory::Protocol, "node build stamp mismatch", true),
        NodeClientError::UnsupportedCapability(_) =>
            (C2ErrorCategory::Protocol, "node capability unavailable", true),
        NodeClientError::Node(failure) if failure.code == NodeFailureCode::Unauthorized =>
            (C2ErrorCategory::Authentication, "node request authentication failed", true),
        NodeClientError::Node(_) =>
            (C2ErrorCategory::Protocol, "node rejected observer request", true),
        NodeClientError::Frame(FrameError::BodyTimedOut { .. } | FrameError::PrefixTimedOut)
            | NodeClientError::AuthenticationTimedOut =>
            (C2ErrorCategory::Timeout, "node observation deadline exceeded", false),
        NodeClientError::Authentication(_) | NodeClientError::RequestIdExhausted =>
            (C2ErrorCategory::Internal, "node client failed internally", true),
        NodeClientError::Io(_) | NodeClientError::Frame(FrameError::Io(_)) =>
            (C2ErrorCategory::Transport, "node transport unavailable", false),
    };
    (SanitizedError { category, message: message.to_owned() }, hard)
}

async fn inventory_owner(
    fresh_for: Duration, mut ingress: mpsc::Receiver<Attempt>,
    status: watch::Sender<Arc<StatusResponse>>, mut shutdown: watch::Receiver<bool>,
) -> io::Result<()> {
    let mut current = (**status.borrow()).clone();
    let mut attempted = BTreeSet::new();
    loop {
        tokio::select! {
            attempt = ingress.recv() => {
                let Some(attempt) = attempt else { return Ok(()); };
                attempted.insert(attempt.node_id.clone());
                let node = current.nodes.entry(attempt.node_id.clone()).or_insert_with(initial_enrolled_node);
                node.last_attempt_unix_ms = Some(attempt.at_unix_ms);
                match attempt.result {
                    AttemptResult::Connected {
                        cursor,
                        snapshot,
                        gaps,
                        provider_contract_manifest,
                    } => {
                        let previous = node.cursor;
                        let mut inventory = SlimNodeInventory::from_snapshot(&snapshot);
                        inventory.provider_contracts =
                            provider_contract_manifest.provider_contracts;
                        inventory.provider_adapter_contracts =
                            provider_contract_manifest.provider_adapter_contracts;
                        node.transport = NodeTransportState::Online;
                        node.freshness = NodeFreshness::Fresh;
                        node.cursor = Some(cursor);
                        node.inventory = Some(inventory);
                        node.last_success_unix_ms = Some(attempt.at_unix_ms);
                        node.consecutive_failures = 0;
                        node.last_error = None;
                        for kind in gaps {
                            if node.gaps.len() == MAX_C2_GAPS_PER_NODE { node.gaps.remove(0); node.gaps_truncated += 1; }
                            node.gaps.push(NodeGap { kind, detected_at_unix_ms: attempt.at_unix_ms, previous, observed: cursor });
                        }
                    }
                    AttemptResult::Success { cursor, snapshot, gaps } => {
                        let previous = node.cursor;
                        let provider_contract_manifest = node.inventory.as_ref().map(|inventory| {
                            ProviderContractManifest {
                                provider_contracts: inventory.provider_contracts.clone(),
                                provider_adapter_contracts: inventory.provider_adapter_contracts.clone(),
                            }
                        }).unwrap_or_default();
                        let mut inventory = SlimNodeInventory::from_snapshot(&snapshot);
                        inventory.provider_contracts =
                            provider_contract_manifest.provider_contracts;
                        inventory.provider_adapter_contracts =
                            provider_contract_manifest.provider_adapter_contracts;
                        node.transport = NodeTransportState::Online;
                        node.freshness = NodeFreshness::Fresh;
                        node.cursor = Some(cursor);
                        node.inventory = Some(inventory);
                        node.last_success_unix_ms = Some(attempt.at_unix_ms);
                        node.consecutive_failures = 0;
                        node.last_error = None;
                        for kind in gaps {
                            if node.gaps.len() == MAX_C2_GAPS_PER_NODE { node.gaps.remove(0); node.gaps_truncated += 1; }
                            node.gaps.push(NodeGap { kind, detected_at_unix_ms: attempt.at_unix_ms, previous, observed: cursor });
                        }
                    }
                    AttemptResult::Cursor {
                        cursor,
                        gaps,
                        managed_worktree_events,
                    } => {
                        let previous = node.cursor;
                        let incarnation_changed = previous.is_some_and(|previous| {
                            previous.incarnation_id != cursor.incarnation_id
                        });
                        apply_managed_worktree_cursor(
                            node.inventory.as_mut(),
                            incarnation_changed,
                            &managed_worktree_events,
                        );
                        node.transport = NodeTransportState::Online;
                        node.freshness = NodeFreshness::Fresh;
                        node.cursor = Some(cursor);
                        node.last_success_unix_ms = Some(attempt.at_unix_ms);
                        node.consecutive_failures = 0;
                        node.last_error = None;
                        for kind in gaps {
                            if node.gaps.len() == MAX_C2_GAPS_PER_NODE { node.gaps.remove(0); node.gaps_truncated += 1; }
                            node.gaps.push(NodeGap { kind, detected_at_unix_ms: attempt.at_unix_ms, previous, observed: cursor });
                        }
                    }
                    AttemptResult::Failure { error, hard } => {
                        node.consecutive_failures = node.consecutive_failures.saturating_add(1);
                        node.transport = if hard || node.consecutive_failures >= 5 { NodeTransportState::Parked } else { NodeTransportState::Offline };
                        node.last_error = Some(error);
                    }
                }
                current.ready = current.nodes.keys().all(|id| attempted.contains(id));
                refresh_freshness(&mut current, fresh_for);
                current.observed_at_unix_ms = unix_ms();
                status.send_replace(Arc::new(current.clone()));
            }
            _ = sleep(Duration::from_millis(250)) => {
                refresh_freshness(&mut current, fresh_for);
                current.observed_at_unix_ms = unix_ms();
                status.send_replace(Arc::new(current.clone()));
            }
            changed = shutdown.changed() => if changed.is_err() || *shutdown.borrow() { return Ok(()); },
        }
    }
}

fn apply_managed_worktree_cursor(
    inventory: Option<&mut SlimNodeInventory>,
    incarnation_changed: bool,
    events: &[NodeEvent],
) {
    let Some(inventory) = inventory else { return; };
    if incarnation_changed {
        inventory.provider_runtime_statuses.clear();
        inventory.managed_worktrees.clear();
        inventory.managed_worktree_count = 0;
        inventory.managed_worktrees_truncated = false;
        return;
    }
    for event in events {
        match event {
            NodeEvent::ManagedWorktreeUpserted { lease } => {
                inventory.managed_worktrees.retain(|existing| {
                    existing.lease_id != lease.lease_id
                        && existing.workspace_id != lease.workspace_id
                });
                inventory.managed_worktrees.push(lease.clone());
                inventory.managed_worktrees.sort_by(|left, right| {
                    left.lease_id.cmp(&right.lease_id)
                });
                inventory.managed_worktree_count = inventory.managed_worktrees.len();
                inventory.managed_worktrees.truncate(
                    crate::protocol::MAX_C2_MANAGED_WORKTREES_PER_NODE,
                );
                inventory.managed_worktrees_truncated =
                    inventory.managed_worktrees.len() < inventory.managed_worktree_count;
            }
            NodeEvent::ManagedWorktreeRemoved { lease_id } => {
                let before = inventory.managed_worktrees.len();
                inventory
                    .managed_worktrees
                    .retain(|lease| &lease.lease_id != lease_id);
                if inventory.managed_worktrees.len() < before
                    || inventory.managed_worktrees_truncated
                {
                    inventory.managed_worktree_count =
                        inventory.managed_worktree_count.saturating_sub(1);
                }
                inventory.managed_worktrees_truncated =
                    inventory.managed_worktrees.len() < inventory.managed_worktree_count;
            }
            _ => {}
        }
    }
}

fn refresh_freshness(status: &mut StatusResponse, fresh_for: Duration) {
    let now = unix_ms();
    let fresh_ms = fresh_for.as_millis().min(u64::MAX as u128) as u64;
    for node in status.nodes.values_mut() {
        node.freshness = match node.last_success_unix_ms {
            None => NodeFreshness::Unavailable,
            Some(last) if now.saturating_sub(last) <= fresh_ms => NodeFreshness::Fresh,
            Some(_) => NodeFreshness::Stale,
        };
    }
}

async fn http_server(
    listener: TcpListener, token: String, io_deadline: Duration,
    status: watch::Receiver<Arc<StatusResponse>>,
    enroll: Option<Arc<crate::enroll_service::EnrollService>>,
    mut shutdown: watch::Receiver<bool>,
) -> io::Result<()> {
    let permits = Arc::new(Semaphore::new(MAX_HTTP_CONNECTIONS));
    let mut connections = JoinSet::new();
    loop {
        tokio::select! {
            accepted = listener.accept() => {
                let (stream, _) = accepted?;
                let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else { drop(stream); continue; };
                let token = token.clone();
                let status = status.clone();
                let enroll = enroll.clone();
                connections.spawn(async move { let _permit = permit; let _ = serve_http(stream, &token, io_deadline, &status, enroll.as_deref()).await; });
            }
            changed = shutdown.changed() => if changed.is_err() || *shutdown.borrow() { break; },
        }
        while let Some(result) = connections.try_join_next() { result.map_err(io::Error::other)?; }
    }
    connections.shutdown().await;
    Ok(())
}

async fn serve_http(mut stream: TcpStream, token: &str, deadline: Duration, status: &watch::Receiver<Arc<StatusResponse>>, enroll: Option<&crate::enroll_service::EnrollService>) -> io::Result<()> {
    let request = match timeout(deadline, read_request(&mut stream)).await {
        Ok(Ok(request)) => request,
        Ok(Err(ReadError::TooLarge)) => return write_response(&mut stream, deadline, Response::plain(413, "Payload Too Large")).await,
        Ok(Err(ReadError::Io(error))) => return Err(error),
        _ => return Ok(()),
    };
    let response = route(request, token, status.borrow().as_ref(), enroll);
    write_response(&mut stream, deadline, response).await
}

struct Request { method: String, path: String, authorization: Option<String>, body: Vec<u8> }
enum ReadError { Closed, Invalid, TooLarge, Io(io::Error) }

async fn read_request(stream: &mut TcpStream) -> Result<Request, ReadError> {
    let mut bytes = Vec::with_capacity(1024);
    let mut chunk = [0_u8; 1024];
    let header_end = loop {
        let count = stream.read(&mut chunk).await.map_err(ReadError::Io)?;
        if count == 0 { return Err(ReadError::Closed); }
        if bytes.len().saturating_add(count) > HEADER_LIMIT_BYTES + OPERATOR_BODY_LIMIT_BYTES { return Err(ReadError::TooLarge); }
        bytes.extend_from_slice(&chunk[..count]);
        if let Some(position) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { break position + 4; }
    };
    let text = std::str::from_utf8(&bytes[..header_end]).map_err(|_| ReadError::Invalid)?;
    let mut lines = text.split("\r\n");
    let mut first = lines.next().ok_or(ReadError::Invalid)?.split_whitespace();
    let method = first.next().ok_or(ReadError::Invalid)?;
    let path = first.next().ok_or(ReadError::Invalid)?;
    let version = first.next().ok_or(ReadError::Invalid)?;
    if first.next().is_some() || !version.starts_with("HTTP/1.") || !path.starts_with('/') { return Err(ReadError::Invalid); }
    let mut authorization = None;
    let mut content_length = 0_usize;
    for line in lines {
        if line.is_empty() { break; }
        let (name, value) = line.split_once(':').ok_or(ReadError::Invalid)?;
        if name.eq_ignore_ascii_case("authorization") {
            if authorization.is_some() { return Err(ReadError::Invalid); }
            authorization = Some(value.trim().to_owned());
        } else if name.eq_ignore_ascii_case("content-length") {
            content_length = value.trim().parse().map_err(|_| ReadError::Invalid)?;
        }
    }
    if content_length > OPERATOR_BODY_LIMIT_BYTES { return Err(ReadError::TooLarge); }
    let mut body = bytes[header_end..].to_vec();
    while body.len() < content_length {
        let count = stream.read(&mut chunk).await.map_err(ReadError::Io)?;
        if count == 0 { return Err(ReadError::Closed); }
        body.extend_from_slice(&chunk[..count]);
    }
    body.truncate(content_length);
    Ok(Request { method: method.to_owned(), path: path.to_owned(), authorization, body })
}

fn route(request: Request, token: &str, status: &StatusResponse, enroll: Option<&crate::enroll_service::EnrollService>) -> Response {
    let path = request.path.split_once('?').map_or(request.path.as_str(), |pair| pair.0);
    // Operator enrollment endpoints: bearer-authenticated, loopback API only.
    if path == "/v1/enroll" || path.starts_with("/v1/enroll/") {
        let Some(service) = enroll else { return Response::plain(404, "Not Found"); };
        if !authorized(request.authorization.as_deref(), token) { return Response::plain(401, "Unauthorized").authenticate(); }
        let reply = service.handle_operator(&request.method, path, &request.body);
        return Response::raw_json(reply.status, reply.body);
    }
    if request.method != "GET" { return Response::plain(405, "Method Not Allowed").allow_get(); }
    match path {
        "/health" => Response::json(200, &HealthResponse { ok: true, service: "gate4agent-c2".to_owned(), api_version: C2_API_VERSION, pid: std::process::id(), version: env!("CARGO_PKG_VERSION").to_owned() }),
        "/ready" => {
            let online_nodes = status.nodes.values().filter(|node| node.transport == NodeTransportState::Online).count();
            let offline_nodes = status.nodes.values().filter(|node| node.transport == NodeTransportState::Offline).count();
            let parked_nodes = status.nodes.values().filter(|node| node.transport == NodeTransportState::Parked).count();
            let body = ReadyResponse { ready: status.ready, api_version: C2_API_VERSION, configured_nodes: status.nodes.len(), attempted_nodes: status.nodes.values().filter(|node| node.last_attempt_unix_ms.is_some()).count(), online_nodes, offline_nodes, parked_nodes };
            Response::json(if status.ready { 200 } else { 503 }, &body)
        }
        "/status" => {
            if !authorized(request.authorization.as_deref(), token) { Response::plain(401, "Unauthorized").authenticate() }
            else {
                // Nodes whose status does not hold `attach` show liveness only.
                let mut shown = status.clone();
                if let Some(service) = enroll {
                    for (id, node) in shown.nodes.iter_mut() {
                        if service.allows(id.as_str(), "attach") == Some(false) { node.inventory = None; }
                    }
                }
                Response::json(200, &shown)
            }
        }
        _ => Response::plain(404, "Not Found"),
    }
}

fn authorized(header: Option<&str>, token: &str) -> bool {
    let Some((scheme, candidate)) = header.and_then(|value| value.split_once(' ')) else { return false; };
    scheme.eq_ignore_ascii_case("bearer") && constant_time_eq(candidate.as_bytes(), token.as_bytes())
}

fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
    if left.len() != right.len() { return false; }
    left.iter().zip(right).fold(0_u8, |difference, (left, right)| difference | (left ^ right)) == 0
}

struct Response { status: u16, reason: &'static str, content_type: &'static str, body: Vec<u8>, headers: Vec<(&'static str, &'static str)> }
impl Response {
    fn plain(status: u16, reason: &'static str) -> Self { Self { status, reason, content_type: "text/plain; charset=utf-8", body: reason.as_bytes().to_vec(), headers: Vec::new() } }
    fn json<T: serde::Serialize>(status: u16, value: &T) -> Self {
        let body = serde_json::to_vec(value).expect("C2 DTO must serialize");
        if body.len() > RESPONSE_BODY_LIMIT_BYTES { return Self::plain(503, "Service Unavailable"); }
        let reason = if status == 200 { "OK" } else { "Service Unavailable" };
        Self { status, reason, content_type: "application/json", body, headers: Vec::new() }
    }
    fn raw_json(status: u16, body: Vec<u8>) -> Self {
        let reason = match status { 200 => "OK", 400 => "Bad Request", 404 => "Not Found", 409 => "Conflict", 429 => "Too Many Requests", _ => "Service Unavailable" };
        Self { status, reason, content_type: "application/json", body, headers: Vec::new() }
    }
    fn allow_get(mut self) -> Self { self.headers.push(("Allow", "GET")); self }
    fn authenticate(mut self) -> Self { self.headers.push(("WWW-Authenticate", "Bearer")); self }
}

async fn write_response(stream: &mut TcpStream, deadline: Duration, response: Response) -> io::Result<()> {
    let mut headers = format!("HTTP/1.1 {} {}\r\nContent-Type: {}\r\nContent-Length: {}\r\nConnection: close\r\n", response.status, response.reason, response.content_type, response.body.len());
    for (name, value) in response.headers { headers.push_str(name); headers.push_str(": "); headers.push_str(value); headers.push_str("\r\n"); }
    headers.push_str("\r\n");
    timeout(deadline, async { stream.write_all(headers.as_bytes()).await?; stream.write_all(&response.body).await?; stream.shutdown().await }).await
        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "C2 HTTP write timed out"))?
}

fn unix_ms() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis().min(u64::MAX as u128) as u64
}

#[cfg(test)]
mod endpoint_tests {
    use super::*;
    use crate::protocol::{C2RelayRoute, C2Topology};

    fn node(endpoint: &str) -> Result<C2NodeConfig, C2ConfigError> {
        C2NodeConfig::new(NodeId::new("remote-node").unwrap(), endpoint, "safe-token")
    }

    #[cfg(windows)]
    const LOCAL_ENDPOINT: &str = r"\\.\pipe\relay-route-fact";
    #[cfg(unix)]
    const LOCAL_ENDPOINT: &str = "/tmp/gate4agent-relay-route-fact.sock";

    fn projected_route(node: &C2NodeConfig, transport: NodeTransportState) -> C2RelayRoute {
        let mut observed = initial_observed_node(node);
        observed.transport = transport;
        let status = StatusResponse {
            api_version: C2_API_VERSION,
            ready: false,
            observed_at_unix_ms: 1,
            nodes: BTreeMap::from([(node.node_id.clone(), observed)]),
        };
        C2Topology::from_status(&status).nodes[0].relay_route
    }

    #[test]
    fn relay_route_fact_projects_exactly_and_survives_offline_and_parked() {
        let local = node(LOCAL_ENDPOINT).unwrap();
        let ssh = node("tcp://127.0.0.1:48100").unwrap();

        for transport in [
            NodeTransportState::Online,
            NodeTransportState::Offline,
            NodeTransportState::Parked,
        ] {
            assert_eq!(projected_route(&local, transport), C2RelayRoute::LocalIpc);
            assert_eq!(
                projected_route(&ssh, transport),
                C2RelayRoute::SshForwardedLoopback,
            );
        }
    }

    /// `accept` is the assignment that names no address, and the config
    /// has to treat it as a different KIND of route rather than as a
    /// malformed endpoint: it is how a node that cannot be dialled is
    /// declared.
    #[test]
    fn a_node_that_waits_to_be_called_is_declared_by_name_not_by_address() {
        let waiting = node("accept").unwrap();
        assert_eq!(waiting.route, C2NodeRoute::CallHome);
        assert_eq!(waiting.transport_label(), "call-home");

        // Every waiting node carries the same `accept`, so the
        // duplicate-endpoint check must not see them as two nodes fighting
        // over one socket. Their uniqueness is `node_id`'s job, and that
        // check still applies.
        let second = C2NodeConfig::new(
            NodeId::new("second-waiting-node").unwrap(),
            "accept",
            "safe-token",
        )
        .unwrap();
        let config = C2Config::new(
            "127.0.0.1:0".parse().unwrap(),
            "safe-token",
            vec![waiting.clone(), second],
        )
        .expect("two waiting nodes are not a duplicate endpoint");
        assert_eq!(config.nodes.len(), 2);
    }

    /// A node that waits for a call the relay never listens for is a
    /// deployment that can never work, so it is refused at startup instead
    /// of becoming a node that is silently offline forever.
    #[test]
    fn waiting_for_a_call_requires_somewhere_to_be_called() {
        let waiting = node("accept").unwrap();
        let config =
            C2Config::new("127.0.0.1:0".parse().unwrap(), "safe-token", vec![waiting]).unwrap();
        assert!(matches!(
            config.validate_call_home(),
            Err(C2ConfigError::CallHomeWithoutListener(_)),
        ));
        let config = config.with_node_listen("127.0.0.1:48200".parse().unwrap()).unwrap();
        assert!(config.validate_call_home().is_ok());
    }

    /// Call-home may bind loopback or a concrete C2 WireGuard tunnel address.
    /// Unspecified (`0.0.0.0`) and port 0 stay refused so plaintext frames do
    /// not land on every interface.
    #[test]
    fn the_call_home_listener_refuses_unspecified_and_accepts_tunnel() {
        let waiting = node("accept").unwrap();
        let config =
            C2Config::new("127.0.0.1:9000".parse().unwrap(), "safe-token", vec![waiting]).unwrap();
        for refused in ["0.0.0.0:48200", "127.0.0.1:0", "[::]:48200"] {
            assert!(
                matches!(
                    config.clone().with_node_listen(refused.parse().unwrap()),
                    Err(C2ConfigError::NonLoopbackNodeListen(_)),
                ),
                "accepted {refused}",
            );
        }
        let tunnel: SocketAddr = "10.88.1.2:27442".parse().unwrap();
        let accepted = config.clone().with_node_listen(tunnel).unwrap();
        assert_eq!(accepted.node_listen, Some(tunnel));
        // Sharing the API's own address would mean HTTP and node frames
        // arriving on one socket; neither parser would survive the other.
        assert!(matches!(
            config.with_node_listen("127.0.0.1:9000".parse().unwrap()),
            Err(C2ConfigError::NodeListenConflict),
        ));
    }

    #[test]
    fn ssh_forwarded_loopback_route_is_strict_canonical_and_control_stays_local() {
        let ipv4 = node("tcp://127.0.0.1:48100").unwrap();
        assert_eq!(ipv4.endpoint, "tcp://127.0.0.1:48100");
        assert_eq!(ipv4.route, C2NodeRoute::SshForwardedLoopback("127.0.0.1:48100".parse().unwrap()));
        assert_eq!(ipv4.transport_label(), "ssh-forwarded-loopback");

        let ipv6 = node("tcp://[0:0:0:0:0:0:0:1]:48100").unwrap();
        assert_eq!(ipv6.endpoint, "tcp://[::1]:48100");
        assert_eq!(ipv6.transport_label(), "ssh-forwarded-loopback");

        for invalid in [
            "tcp://localhost:48100",
            "tcp://127.0.0.2:48100",
            "tcp://0.0.0.0:48100",
            "tcp://127.0.0.1:0",
            "tcp://user@127.0.0.1:48100",
            "tcp://127.0.0.1:48100/path",
            "TCP://127.0.0.1:48100",
        ] {
            assert!(matches!(node(invalid), Err(C2ConfigError::InvalidEndpoint(_))), "accepted {invalid}");
        }

        assert!(matches!(
            C2Config::new(
                "127.0.0.1:0".parse().unwrap(),
                "safe-token",
                vec![ipv4.clone(), C2NodeConfig::new(
                    NodeId::new("duplicate-route").unwrap(),
                    "tcp://127.0.0.1:48100",
                    "safe-token",
                ).unwrap()],
            ),
            Err(C2ConfigError::DuplicateEndpoint)
        ));
        assert!(matches!(
            C2Config::new("127.0.0.1:0".parse().unwrap(), "safe-token", vec![ipv4])
                .unwrap()
                .with_control_endpoint("tcp://127.0.0.1:48101"),
            Err(C2ConfigError::InvalidControlEndpoint)
        ));
    }
}

#[cfg(all(test, windows))]
mod tests {
    use super::*;
    use gate4agent_node_protocol::{
        AgentId, AgentStreamChunkKindV1, AgentStreamChunkV1, CapabilityId, SessionAddress,
        SessionKey, SessionMode, SessionRecordId,
        HarnessMcpActivationDigest, HarnessMcpCallId, HarnessMcpContentTypeV1,
        HarnessMcpOpaquePayloadV1, HarnessMcpReservationId,
        SpawnDeadlineMs, SpawnFieldProvenance, SpawnIdempotencyKey, SpawnOverrides,
        SpawnProfileId, SpawnProfileRevision, SpawnPrompt, SpawnPromptMetadata,
        SpawnRequiredCapabilities, SpawnResolutionProvenance, SpawnTarget, WorkspaceId,
        ManagedWorktreeCleanupFailure, ManagedWorktreeLeaseId,
        ManagedWorktreeLeaseSnapshot, ManagedWorktreeRetention, ManagedWorktreeSpawnReceipt,
        WorktreeProfileId, WorktreeProfileRevision,
    };
    use gate4agent_types::{AgentInstanceId, SessionGeneration, TerminalSize};
    use std::collections::BTreeMap;

    fn agent(value: &str) -> gate4agent_node_protocol::AgentId {
        gate4agent_node_protocol::AgentId::new(value).unwrap()
    }

    fn managed_lease(
        lease_id: &str,
        workspace_id: &str,
        state: ManagedWorktreeLeaseState,
    ) -> ManagedWorktreeLeaseSnapshot {
        let in_use = state == ManagedWorktreeLeaseState::InUse;
        ManagedWorktreeLeaseSnapshot {
            lease_id: ManagedWorktreeLeaseId::new(lease_id).unwrap(),
            source_workspace_id: WorkspaceId::new("repo").unwrap(),
            workspace_id: WorkspaceId::new(workspace_id).unwrap(),
            profile_id: WorktreeProfileId::new("review").unwrap(),
            profile_revision: WorktreeProfileRevision::new("review.r1").unwrap(),
            retention: ManagedWorktreeRetention::RemoveWhenReleased,
            state,
            active_session_count: u16::from(in_use),
            managed_record_count: u16::from(in_use),
            cleanup_failure: None::<ManagedWorktreeCleanupFailure>,
            created_at_unix_ms: 1,
            updated_at_unix_ms: 2,
        }
    }

    #[test]
    fn spawn_spec_receipt_correlation_rejects_mismatches_before_forwarding() {
        let incarnation_id = NodeIncarnationId::from_bytes([7; 16]);
        let terminal_size = TerminalSize { rows: 24, columns: 80 };
        let prompt = SpawnPrompt::new("hi").unwrap();
        let required_capabilities = SpawnRequiredCapabilities::new([
            CapabilityId::new("raw-pty-lifecycle").unwrap(),
        ]).unwrap();
        let spec = SpawnSpec {
            target: SpawnTarget {
                node_id: NodeId::new("node-a").unwrap(),
                workspace_id: WorkspaceId::new("repo").unwrap(),
                worktree_id: None,
            },
            profile_id: SpawnProfileId::new("default").unwrap(),
            expected_profile_revision: SpawnProfileRevision::new("r1").unwrap(),
            overrides: SpawnOverrides {
                provider: SpawnOverride::Set { value: AgentId::new("codex").unwrap() },
                mode: SpawnOverride::Set { value: SessionMode::Pty },
                terminal_size: SpawnOverride::Set { value: terminal_size },
                prompt: SpawnOverride::Set { value: prompt.clone() },
                bundle_id: SpawnOverride::Clear,
                context_id: SpawnOverride::Clear,
                environment_profile_id: SpawnOverride::Clear,
                approval_level: None,
                network_allowlist: None,
                browser_profile_id: None,
            },
            deadline_ms: SpawnDeadlineMs::new(5_000).unwrap(),
            idempotency_key: SpawnIdempotencyKey::new("spawn-1").unwrap(),
            required_capabilities: required_capabilities.clone(),
        };
        let receipt = ResolvedSpawnReceipt {
            incarnation_id,
            session: SessionAddress {
                workspace_id: spec.target.workspace_id.clone(),
                session: SessionKey {
                    instance_id: AgentInstanceId(7),
                    generation: SessionGeneration(1),
                },
            },
            target: spec.target.clone(),
            profile_id: spec.profile_id.clone(),
            profile_revision: SpawnProfileRevision::new("r1").unwrap(),
            provider: AgentId::new("codex").unwrap(),
            mode: SessionMode::Pty,
            terminal_size,
            prompt: SpawnPromptMetadata::from_prompt(Some(&prompt)),
            bundle_id: None,
            bundle: None,
            context_id: None,
            context: None,
            environment_profile: None,
            deadline_ms: spec.deadline_ms,
            idempotency_key: spec.idempotency_key.clone(),
            required_capabilities,
            provenance: SpawnResolutionProvenance {
                provider: SpawnFieldProvenance::Override,
                mode: SpawnFieldProvenance::Override,
                terminal_size: SpawnFieldProvenance::Override,
                prompt: SpawnFieldProvenance::Override,
                bundle_id: SpawnFieldProvenance::Cleared,
                context_id: SpawnFieldProvenance::Cleared,
                environment_profile_id: SpawnFieldProvenance::Cleared,
            },
            harness_mcp_proxy: None,
        };
        let response = |receipt| Ok(NodeResponse::SpawnSpecAccepted { receipt });
        let expected = ExpectedSpawnRequest::Spec(spec.clone());
        assert!(validate_spawn_spec_response(
            Some(&expected),
            &response(receipt.clone()),
            incarnation_id,
        ).is_ok());

        let mut mismatches = Vec::new();
        let mut changed = receipt.clone();
        changed.incarnation_id = NodeIncarnationId::from_bytes([8; 16]);
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.target.node_id = NodeId::new("node-b").unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.profile_id = SpawnProfileId::new("other").unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.idempotency_key = SpawnIdempotencyKey::new("spawn-2").unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.deadline_ms = SpawnDeadlineMs::new(4_999).unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.required_capabilities = SpawnRequiredCapabilities::default();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.session.workspace_id = WorkspaceId::new("other").unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.provider = AgentId::new("claude").unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.prompt = SpawnPromptMetadata { present: false, byte_len: 0 };
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.environment_profile = Some(
            gate4agent_node_protocol::ResolvedEnvironmentProfileReceipt {
                profile_id:
                    gate4agent_node_protocol::SpawnEnvironmentProfileId::new(
                        "local-default",
                    )
                    .unwrap(),
                profile_revision:
                    gate4agent_node_protocol::SpawnEnvironmentProfileRevision::new(
                        "local-default.r1",
                    )
                    .unwrap(),
                    network_allowlist: None,
                    browser_profile_id: None,
            },
        );
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.bundle = Some(gate4agent_node_protocol::ResolvedBundleReceipt {
            id: gate4agent_node_protocol::SpawnBundleId::new("unexpected-bundle")
                .unwrap(),
            revision: gate4agent_node_protocol::SpawnBundleRevision::new(
                "unexpected-bundle.r1",
            )
            .unwrap(),
            digest: gate4agent_node_protocol::SpawnBundleDigest::new(format!(
                "sha256:{}",
                "a".repeat(64),
            ))
            .unwrap(),
        });
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.context = Some(gate4agent_node_protocol::ResolvedContextPackReceipt {
            id: gate4agent_node_protocol::SpawnContextId::new("unexpected-context")
                .unwrap(),
            digest: gate4agent_node_protocol::SpawnContextDigest::new(format!(
                "sha256:{}",
                "b".repeat(64),
            ))
            .unwrap(),
            lineage: gate4agent_node_protocol::ContextPackLineageReceipt {
                source_node_id: NodeId::new("node-a").unwrap(),
                source_session: receipt.session.clone(),
                source_provider: AgentId::new("codex").unwrap(),
            },
            source_message_count: 1,
            retained_message_count: 1,
            byte_len: 16,
            truncated: false,
        });
        mismatches.push(changed);

        for mismatch in mismatches {
            assert!(validate_spawn_spec_response(
                Some(&expected),
                &response(mismatch),
                incarnation_id,
            ).is_err());
        }

        let mut environment_spec = spec;
        environment_spec.overrides.environment_profile_id = SpawnOverride::Set {
            value: gate4agent_node_protocol::SpawnEnvironmentProfileId::new(
                "local-default",
            )
            .unwrap(),
        };
        let environment_expected = ExpectedSpawnRequest::Spec(environment_spec);
        let mut environment_receipt = receipt;
        environment_receipt.environment_profile = Some(
            gate4agent_node_protocol::ResolvedEnvironmentProfileReceipt {
                profile_id:
                    gate4agent_node_protocol::SpawnEnvironmentProfileId::new(
                        "local-default",
                    )
                    .unwrap(),
                profile_revision:
                    gate4agent_node_protocol::SpawnEnvironmentProfileRevision::new(
                        "local-default.r1",
                    )
                    .unwrap(),
                    network_allowlist: None,
                    browser_profile_id: None,
            },
        );
        assert!(validate_spawn_spec_response(
            Some(&environment_expected),
            &response(environment_receipt.clone()),
            incarnation_id,
        )
        .is_ok());
        environment_receipt.environment_profile.as_mut().unwrap().profile_id =
            gate4agent_node_protocol::SpawnEnvironmentProfileId::new("other")
                .unwrap();
        assert!(validate_spawn_spec_response(
            Some(&environment_expected),
            &response(environment_receipt),
            incarnation_id,
        )
        .is_err());
    }

    #[test]
    fn provider_session_index_correlation_accepts_exact_identity_and_rejects_mismatches() {
        let identity = gate4agent_types::ProviderSessionIdentity {
            key: gate4agent_types::ProviderSessionKey::SessionId,
            id: "native-session-1".to_owned(),
            transcript_path: Some(r"C:\provider\sessions\native-session-1.jsonl".to_owned()),
        };
        let expected = NodeRequest::IndexProviderSession {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            provider: AgentId::new("codex").unwrap(),
            identity: identity.clone(),
            display_name: "release shepherd".to_owned(),
        };
        let NodeRequest::IndexProviderSession { workspace_id, provider, .. } = &expected else {
            unreachable!();
        };
        let record = gate4agent_node_protocol::ManagedSessionRecord {
            record_id: SessionRecordId::new("session-001").unwrap(),
            display_name: "release shepherd".to_owned(),
            provider: provider.clone(),
            mode: SessionMode::Pty,
            state: gate4agent_node_protocol::ManagedSessionState::Dormant,
            workspace_id: workspace_id.clone(),
            canonical_root: gate4agent_node_protocol::OpaqueHostPath::utf8(
                r"C:\repo".to_owned(),
            ).unwrap(),
            provider_session: Some(identity),
            active_session: None,
            environment_profile: None,
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            created_at_unix_ms: 1,
            updated_at_unix_ms: 2,
            last_error: None,
        };
        let response = |record| Ok(NodeResponse::ProviderSessionIndexed { record });

        assert!(validate_provider_session_index_response(
            Some(&expected),
            &response(record.clone()),
        ).is_ok());
        assert!(validate_provider_session_index_response(
            None,
            &response(record.clone()),
        ).is_err());

        let mut identity_mismatch = record.clone();
        identity_mismatch.provider_session.as_mut().unwrap().id =
            "native-session-2".to_owned();
        assert!(validate_provider_session_index_response(
            Some(&expected),
            &response(identity_mismatch),
        ).is_err());

        let mut workspace_mismatch = record.clone();
        workspace_mismatch.workspace_id = WorkspaceId::new("other").unwrap();
        assert!(validate_provider_session_index_response(
            Some(&expected),
            &response(workspace_mismatch),
        ).is_err());

        let mut provider_mismatch = record;
        provider_mismatch.provider = AgentId::new("claude").unwrap();
        assert!(validate_provider_session_index_response(
            Some(&expected),
            &response(provider_mismatch),
        ).is_err());
    }

    #[test]
    fn native_session_route_correlation_rejects_mismatches_before_projection() {
        let route = gate4agent_node_protocol::NativeSessionCatalogRoute::workspace(
            WorkspaceId::new("primary").unwrap(),
            AgentId::new("codex").unwrap(),
        );
        let selection = gate4agent_node_protocol::NativeSessionSelection {
            route: route.clone(),
            catalog_revision: 7,
            recent_cutoff_unix_ms: 70,
            selection_id: "selection-7".to_owned(),
        };
        let catalog = NodeRequest::CatalogNativeSessions {
            route: route.clone(),
            limit: 10,
        };
        let catalog_response = Ok(NodeResponse::NativeSessionsCataloged {
            route: route.clone(),
            entries: Vec::new(),
            summary: None,
        });
        assert!(validate_native_session_response(Some(&catalog), &catalog_response).is_ok());
        let mismatched_route = gate4agent_node_protocol::NativeSessionCatalogRoute::workspace(
            WorkspaceId::new("other").unwrap(),
            AgentId::new("codex").unwrap(),
        );
        assert!(validate_native_session_response(
            Some(&catalog),
            &Ok(NodeResponse::NativeSessionsCataloged {
                route: mismatched_route,
                entries: Vec::new(),
                summary: None,
            }),
        )
        .is_err());

        let page = NodeRequest::PageNativeSessions {
            route: route.clone(),
            window: gate4agent_node_protocol::NativeSessionCatalogWindow::Recent,
            catalog_revision: 7,
            recent_cutoff_unix_ms: 70,
            after_selection_id: None,
            limit: 10,
        };
        let page_response = |revision| {
            Ok(NodeResponse::NativeSessionsPaged {
                route: route.clone(),
                page: gate4agent_node_protocol::NativeSessionCatalogPage {
                    window: gate4agent_node_protocol::NativeSessionCatalogWindow::Recent,
                    revision,
                    entries: Vec::new(),
                    next_after_selection_id: None,
                    remaining_count: 0,
                    has_more: false,
                },
            })
        };
        assert!(validate_native_session_response(Some(&page), &page_response(7)).is_ok());
        assert!(validate_native_session_response(Some(&page), &page_response(8)).is_err());

        let preview = NodeRequest::PreviewNativeSession {
            selection: selection.clone(),
            message_limit: 10,
        };
        let preview_response = |selection| {
            Ok(NodeResponse::NativeSessionPreviewed {
                selection,
                preview: gate4agent_node_protocol::SessionRecordPreview {
                    title: None,
                    modified_at_unix_ms: None,
                    model: None,
                    message_count: 0,
                    message_count_exact: true,
                    completed_turn_count: None,
                    total_tokens: None,
                    truncated: false,
                    messages: Vec::new(),
                },
            })
        };
        assert!(validate_native_session_response(
            Some(&preview),
            &preview_response(selection.clone()),
        )
        .is_ok());
        let mut mismatched_selection = selection.clone();
        mismatched_selection.catalog_revision = 8;
        assert!(validate_native_session_response(
            Some(&preview),
            &preview_response(mismatched_selection),
        )
        .is_err());

        let index = NodeRequest::IndexNativeSession {
            selection: selection.clone(),
            display_name: "Indexed".to_owned(),
        };
        let record = gate4agent_node_protocol::ManagedSessionRecord {
            record_id: SessionRecordId::new("session-007").unwrap(),
            display_name: "Indexed".to_owned(),
            provider: route.provider.clone(),
            mode: SessionMode::Pty,
            state: gate4agent_node_protocol::ManagedSessionState::Dormant,
            workspace_id: route.workspace_id.clone().unwrap(),
            canonical_root: gate4agent_node_protocol::OpaqueHostPath::utf8(
                r"C:\repo".to_owned(),
            )
            .unwrap(),
            provider_session: None,
            active_session: None,
            environment_profile: None,
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            created_at_unix_ms: 1,
            updated_at_unix_ms: 2,
            last_error: None,
        };
        assert!(validate_native_session_response(
            Some(&index),
            &Ok(NodeResponse::NativeSessionIndexed {
                selection: selection.clone(),
                record: record.clone(),
            }),
        )
        .is_ok());
        assert!(validate_native_session_response(
            Some(&index),
            &Ok(NodeResponse::ProviderSessionIndexed {
                record: record.clone(),
            }),
        )
        .is_err());
        let mut wrong_echo = selection.clone();
        wrong_echo.catalog_revision = 8;
        assert!(validate_native_session_response(
            Some(&index),
            &Ok(NodeResponse::NativeSessionIndexed {
                selection: wrong_echo,
                record: record.clone(),
            }),
        )
        .is_err());
        let mut wrong_provider = record.clone();
        wrong_provider.provider = AgentId::new("claude").unwrap();
        assert!(validate_native_session_response(
            Some(&index),
            &Ok(NodeResponse::NativeSessionIndexed {
                selection: selection.clone(),
                record: wrong_provider.clone(),
            }),
        )
        .is_err());
        let mut wrong_workspace = record.clone();
        wrong_workspace.workspace_id = WorkspaceId::new("other").unwrap();
        assert!(validate_native_session_response(
            Some(&index),
            &Ok(NodeResponse::NativeSessionIndexed {
                selection: selection.clone(),
                record: wrong_workspace,
            }),
        )
        .is_err());

        let external = NodeRequest::IndexNativeSession {
            selection: gate4agent_node_protocol::NativeSessionSelection {
                route: gate4agent_node_protocol::NativeSessionCatalogRoute::unregistered(
                    AgentId::new("codex").unwrap(),
                ),
                ..selection
            },
            display_name: "External".to_owned(),
        };
        let NodeRequest::IndexNativeSession {
            selection: external_selection,
            ..
        } = &external else {
            unreachable!();
        };
        assert!(validate_native_session_response(
            Some(&external),
            &Ok(NodeResponse::NativeSessionIndexed {
                selection: external_selection.clone(),
                record: gate4agent_node_protocol::ManagedSessionRecord {
                    provider: AgentId::new("codex").unwrap(),
                    ..wrong_provider
                },
            }),
        )
        .is_err());
    }

    #[test]
    fn managed_spawn_receipt_correlation_covers_legacy_and_v2_requests() {
        let incarnation_id = NodeIncarnationId::from_bytes([7; 16]);
        let spec = SpawnSpec {
            target: SpawnTarget {
                node_id: NodeId::new("node-a").unwrap(),
                workspace_id: WorkspaceId::new("repo").unwrap(),
                worktree_id: None,
            },
            profile_id: SpawnProfileId::new("default").unwrap(),
            expected_profile_revision:
                SpawnProfileRevision::new("default.r1").unwrap(),
            overrides: SpawnOverrides::default(),
            deadline_ms: SpawnDeadlineMs::new(5_000).unwrap(),
            idempotency_key: SpawnIdempotencyKey::new("managed-1").unwrap(),
            required_capabilities: SpawnRequiredCapabilities::default(),
        };
        let managed = ManagedWorktreeSpawnRequest {
            spawn_spec: spec.clone(),
            worktree_profile_id: WorktreeProfileId::new("review").unwrap(),
        };
        let workspace_id = WorkspaceId::new("managed-a").unwrap();
        let spawn = ResolvedSpawnReceipt {
            incarnation_id,
            session: SessionAddress {
                workspace_id: workspace_id.clone(),
                session: SessionKey {
                    instance_id: AgentInstanceId(8),
                    generation: SessionGeneration(1),
                },
            },
            target: SpawnTarget {
                node_id: spec.target.node_id.clone(),
                workspace_id: spec.target.workspace_id.clone(),
                worktree_id: Some(workspace_id.clone()),
            },
            profile_id: spec.profile_id.clone(),
            profile_revision: SpawnProfileRevision::new("default.r1").unwrap(),
            provider: AgentId::new("claude").unwrap(),
            mode: SessionMode::Pty,
            terminal_size: TerminalSize { rows: 24, columns: 80 },
            prompt: SpawnPromptMetadata { present: false, byte_len: 0 },
            bundle_id: None,
            bundle: None,
            context_id: None,
            context: None,
            environment_profile: None,
            deadline_ms: spec.deadline_ms,
            idempotency_key: spec.idempotency_key.clone(),
            required_capabilities: SpawnRequiredCapabilities::default(),
            provenance: SpawnResolutionProvenance {
                provider: SpawnFieldProvenance::Profile,
                mode: SpawnFieldProvenance::Profile,
                terminal_size: SpawnFieldProvenance::Profile,
                prompt: SpawnFieldProvenance::Profile,
                bundle_id: SpawnFieldProvenance::Profile,
                context_id: SpawnFieldProvenance::Profile,
                environment_profile_id: SpawnFieldProvenance::Profile,
            },
            harness_mcp_proxy: None,
        };
        let receipt = ManagedWorktreeSpawnReceipt {
            spawn,
            lease: managed_lease(
                "lease-a",
                "managed-a",
                ManagedWorktreeLeaseState::InUse,
            ),
        };
        let expected = ExpectedSpawnRequest::ManagedV1(managed.clone());
        let response = |receipt| Ok(NodeResponse::ManagedWorktreeSpawnAccepted { receipt });
        assert!(validate_spawn_spec_response(
            Some(&expected),
            &response(receipt.clone()),
            incarnation_id,
        )
        .is_ok());

        let mut mismatches = Vec::new();
        let mut changed = receipt.clone();
        changed.lease.state = ManagedWorktreeLeaseState::Ready;
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.lease.cleanup_failure = Some(ManagedWorktreeCleanupFailure::Busy);
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.lease.active_session_count = 0;
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.lease.profile_id = WorktreeProfileId::new("other").unwrap();
        mismatches.push(changed);
        let mut changed = receipt.clone();
        changed.lease.source_workspace_id = WorkspaceId::new("other").unwrap();
        mismatches.push(changed);

        for mismatch in mismatches {
            assert!(validate_spawn_spec_response(
                Some(&expected),
                &response(mismatch),
                incarnation_id,
            )
            .is_err());
        }

        let mut legacy_revision = receipt.clone();
        legacy_revision.lease.profile_revision =
            WorktreeProfileRevision::new("review.r2").unwrap();
        assert!(validate_spawn_spec_response(
            Some(&expected),
            &response(legacy_revision),
            incarnation_id,
        )
        .is_ok());

        let v2_request = ManagedWorktreeSpawnRequestV2 {
            spawn_spec: managed.spawn_spec,
            worktree_profile_id: managed.worktree_profile_id,
            expected_profile_revision: WorktreeProfileRevision::new("review.r1").unwrap(),
        };
        let routed = NodeRequest::SpawnManagedWorktreeV2 {
            request: v2_request.clone(),
        };
        let captured_expected = expected_spawn_request(&routed).unwrap();
        let ExpectedSpawnRequest::ManagedV2(captured_request) = &captured_expected else {
            panic!("V2 managed spawn request was not captured separately");
        };
        assert_eq!(captured_request, &v2_request);
        assert!(validate_spawn_spec_response(
            Some(&captured_expected),
            &response(receipt.clone()),
            incarnation_id,
        )
        .is_ok());

        let mut wrong_revision = receipt.clone();
        wrong_revision.lease.profile_revision =
            WorktreeProfileRevision::new("review.r2").unwrap();
        assert!(validate_spawn_spec_response(
            Some(&ExpectedSpawnRequest::ManagedV2(v2_request.clone())),
            &response(wrong_revision),
            incarnation_id,
        )
        .is_err());
        assert!(validate_spawn_spec_response(
            Some(&ExpectedSpawnRequest::ManagedV2(v2_request)),
            &Ok(NodeResponse::SpawnSpecAccepted {
                receipt: receipt.spawn,
            }),
            incarnation_id,
        )
        .is_err());
    }

    #[test]
    fn windows_runtime_default_control_endpoint_is_exact_valid_and_distinct_from_a_node() {
        assert_eq!(DEFAULT_C2_CONTROL_ENDPOINT, r"\\.\pipe\gate4agent-c2");
        validate_control_endpoint(DEFAULT_C2_CONTROL_ENDPOINT).unwrap();
        let node = C2NodeConfig::new(
            NodeId::new("node-a").unwrap(),
            r"\\.\pipe\gate4agent-node",
            "safe-token",
        )
        .unwrap();
        let config = C2Config::new(
            "127.0.0.1:0".parse().unwrap(),
            "safe-token",
            vec![node],
        )
        .unwrap();
        assert_eq!(config.control_endpoint, DEFAULT_C2_CONTROL_ENDPOINT);
        assert!(!config.nodes[0]
            .endpoint
            .eq_ignore_ascii_case(&config.control_endpoint));

        let conflicting_node = C2NodeConfig::new(
            NodeId::new("node-b").unwrap(),
            DEFAULT_C2_CONTROL_ENDPOINT,
            "safe-token",
        )
        .unwrap();
        assert!(matches!(
            C2Config::new(
                "127.0.0.1:0".parse().unwrap(),
                "safe-token",
                vec![conflicting_node],
            ),
            Err(C2ConfigError::ControlEndpointConflict)
        ));
    }

    #[test]
    fn terminal_only_sequence_holes_do_not_mark_c2_partial() {
        use gate4agent_node_protocol::{NodeEvent, WorkspaceId};
        let event = |sequence| NodeEventEnvelope { sequence, event: NodeEvent::WorkspaceRemoved { workspace_id: WorkspaceId::new("work").unwrap() } };
        assert_eq!(validate_events(4, 7, 1, &[event(6)]), Vec::<GapKind>::new());
        assert_eq!(validate_events(4, 7, 1, &[]), Vec::<GapKind>::new());
        assert_eq!(validate_events(4, 7, 1, &[event(7), event(6)]), vec![GapKind::NonContiguousEvents]);
        assert_eq!(validate_events(7, 6, 1, &[]), vec![GapKind::CursorRegression]);
        assert_eq!(validate_resync(4, 6, 4, 1, &[]), vec![GapKind::CursorRegression]);
    }

    #[test]
    fn real_durable_eviction_marks_c2_history_evicted() {
        use gate4agent_node_protocol::{NodeEvent, WorkspaceId};
        let event = |sequence| NodeEventEnvelope {
            sequence,
            event: NodeEvent::WorkspaceRemoved {
                workspace_id: WorkspaceId::new("work").unwrap(),
            },
        };
        assert_eq!(
            validate_events(4, 7, 6, &[event(6)]),
            vec![GapKind::HistoryEvicted],
        );
        assert_eq!(validate_events(5, 7, 6, &[event(6)]), Vec::<GapKind>::new());
    }

    #[test]
    fn live_event_gap_preserves_cursor_and_resync_rules() {
        use gate4agent_node_protocol::{NodeEvent, WorkspaceId};
        let event = |sequence, event| NodeEventEnvelope { sequence, event };
        let removed = || NodeEvent::WorkspaceRemoved {
            workspace_id: WorkspaceId::new("work").unwrap(),
        };

        assert_eq!(live_event_gap(4, &event(5, removed())), None);
        assert_eq!(
            live_event_gap(4, &event(4, removed())),
            Some(GapKind::CursorRegression),
        );
        assert_eq!(
            live_event_gap(4, &event(7, removed())),
            Some(GapKind::NonContiguousEvents),
        );
        assert_eq!(
            live_event_gap(4, &event(5, NodeEvent::ResyncRequired {
                oldest_available_sequence: 3,
            })),
            Some(GapKind::HistoryEvicted),
        );
    }

    #[test]
    fn harness_mcp_transient_live_and_pending_projection_preserves_cursor_without_replay() {
        let node_id = NodeId::new("node-a").unwrap();
        let cursor = NodeCursor {
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            sequence: 41,
        };
        let envelope = NodeEventEnvelope {
            sequence: 0,
            event: NodeEvent::HarnessMcpReadCall {
                reservation_id: HarnessMcpReservationId::new(
                    format!("hmcpres_{}", "a".repeat(24)),
                ).unwrap(),
                activation_digest: HarnessMcpActivationDigest::new(
                    format!("sha256:{}", "b".repeat(64)),
                ).unwrap(),
                record_id: SessionRecordId::new("session-001").unwrap(),
                session: SessionAddress {
                    workspace_id: WorkspaceId::new("repo").unwrap(),
                    session: SessionKey {
                        instance_id: AgentInstanceId(8),
                        generation: SessionGeneration(1),
                    },
                },
                call_id: HarnessMcpCallId::new(
                    format!("hmcpcall_{}", "c".repeat(24)),
                ).unwrap(),
                request: HarnessMcpOpaquePayloadV1 {
                    content_type: HarnessMcpContentTypeV1::HarnessReadRequestJsonV1,
                    body: br#"{"kind":"context-get"}"#.to_vec(),
                },
                deadline_unix_ms: u64::MAX,
            },
        };

        let live = routed_transient_node_event(&node_id, cursor, &envelope).unwrap();
        let pending = routed_transient_node_event(&node_id, cursor, &envelope).unwrap();
        assert_eq!(live, pending);
        assert_eq!(live.cursor, cursor);
        assert!(matches!(live.event, C2NodeEvent::HarnessMcpReadCall { .. }));
        assert!(routed_recovered_node_event(
            &node_id,
            cursor.incarnation_id,
            &envelope,
        ).is_none());
        assert_eq!(cursor.sequence, 41);
    }

    fn agent_stream_test_address() -> SessionAddress {
        SessionAddress {
            workspace_id: WorkspaceId::new("repo").unwrap(),
            session: SessionKey {
                instance_id: AgentInstanceId(8),
                generation: SessionGeneration(1),
            },
        }
    }

    fn agent_stream_test_envelope(sequence: u64, source_sequence: u64) -> NodeEventEnvelope {
        NodeEventEnvelope {
            sequence,
            event: NodeEvent::AgentStream {
                address: agent_stream_test_address(),
                chunk: AgentStreamChunkV1 {
                    source_sequence,
                    kind: AgentStreamChunkKindV1::Text {
                        text: format!("chunk-{source_sequence}"),
                        is_delta: true,
                    },
                },
            },
        }
    }

    #[test]
    fn agent_stream_chunk_publishes_unconditionally_and_only_advances_cursor_forward() {
        let node_id = NodeId::new("node-a").unwrap();
        let mut cursor = NodeCursor {
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            sequence: 674,
        };

        // A non-`AgentStream` envelope is not this function's to route.
        let control = NodeEventEnvelope {
            sequence: 675,
            event: NodeEvent::WorkspaceRemoved {
                workspace_id: WorkspaceId::new("work").unwrap(),
            },
        };
        assert!(route_agent_stream_event(&node_id, &mut cursor, &control).is_none());
        assert_eq!(cursor.sequence, 674);

        // A chunk one past the cursor is published, and its OWN cursor
        // field on the wire reads the cursor as it stood before this
        // chunk (mirroring `HarnessMcpReadCall`'s convention exactly) --
        // then the live cursor moves forward to make room for it.
        let forward = agent_stream_test_envelope(675, 22);
        let routed = route_agent_stream_event(&node_id, &mut cursor, &forward).unwrap();
        assert_eq!(routed.cursor.sequence, 674);
        assert!(matches!(
            &routed.event,
            C2NodeEvent::AgentStream { chunk, .. } if chunk.source_sequence == 22
        ));
        assert_eq!(cursor.sequence, 675);

        // A chunk that arrives (or, after a durable resync already carried
        // the cursor past it, is only now examined) BEHIND the cursor is
        // still published -- chunks promise no resync, so there is nothing
        // to recover if this one were dropped instead -- but it must never
        // rewind the cursor a genuinely later `Control` envelope already
        // advanced past.
        cursor.sequence = 711;
        let late = agent_stream_test_envelope(677, 23);
        let routed = route_agent_stream_event(&node_id, &mut cursor, &late).unwrap();
        assert_eq!(routed.cursor.sequence, 711);
        assert!(matches!(
            &routed.event,
            C2NodeEvent::AgentStream { chunk, .. } if chunk.source_sequence == 23
        ));
        assert_eq!(cursor.sequence, 711, "a late chunk must never rewind the cursor");
    }

    #[test]
    fn agent_stream_burst_survives_a_durable_cursor_that_already_ran_past_it() {
        // Reproduces the measured live defect's mechanism at its worst: the
        // node's connection loop can drain its durable channel to
        // exhaustion across several ticks before this dedicated channel
        // gets any budget at all, so a whole burst's `Control` envelopes
        // can reach this relay, and carry `cursor` forward, before a
        // single one of the SAME burst's interleaved `AgentStream` chunks
        // is even looked at. `route_agent_stream_event` must publish every
        // one of them regardless.
        let node_id = NodeId::new("node-a").unwrap();
        let mut cursor = NodeCursor {
            incarnation_id: NodeIncarnationId::from_bytes([9; 16]),
            sequence: 0,
        };
        let chunk_count = 200_u64;

        for turn in 1..=chunk_count {
            // Durable (`Control`) processing advancing the cursor is the
            // pre-existing, unchanged path (`live_event_gap` and its
            // resync repair, exercised by this module's other tests) --
            // simulated here only by its end effect on `cursor`, since
            // this test is about the `AgentStream` side's own contract.
            cursor.sequence = turn * 2;
        }

        let mut recovered = Vec::new();
        for turn in 1..=chunk_count {
            let envelope = agent_stream_test_envelope(turn * 2 - 1, turn);
            let routed = route_agent_stream_event(&node_id, &mut cursor, &envelope)
                .expect("every chunk in the burst must be published, however far the durable cursor already ran past its sequence");
            recovered.push(routed);
        }

        assert_eq!(recovered.len(), chunk_count as usize);
        for (turn, routed) in (1..=chunk_count).zip(recovered.iter()) {
            assert!(matches!(
                &routed.event,
                C2NodeEvent::AgentStream { chunk, .. } if chunk.source_sequence == turn
            ));
        }
        assert_eq!(
            cursor.sequence,
            chunk_count * 2,
            "a whole burst of stale chunks must never rewind the cursor the durable side already advanced",
        );
    }

    #[test]
    fn config_rejects_header_injection_duplicate_nodes_and_non_loopback() {
        let id = NodeId::new("node-a").unwrap();
        assert!(matches!(C2NodeConfig::new(id.clone(), r"\\.\pipe\a", "bad\r\ntoken"), Err(C2ConfigError::InvalidToken)));
        let node = C2NodeConfig::new(id, r"\\.\pipe\a", "safe-token").unwrap();
        assert!(matches!(C2Config::new("0.0.0.0:0".parse().unwrap(), "safe", vec![node.clone()]), Err(C2ConfigError::NonLoopback(_))));
        assert!(matches!(C2Config::new("127.0.0.1:0".parse().unwrap(), "safe", vec![node.clone(), node]), Err(C2ConfigError::DuplicateNode)));
        let first = C2NodeConfig::new(NodeId::new("node-a").unwrap(), r"\\.\pipe\same", "safe").unwrap();
        let second = C2NodeConfig::new(NodeId::new("node-b").unwrap(), r"\\.\pipe\same", "safe").unwrap();
        assert!(matches!(C2Config::new("127.0.0.1:0".parse().unwrap(), "safe", vec![first, second]), Err(C2ConfigError::DuplicateEndpoint)));
        let oversized = format!(r"\\.\pipe\{}", "x".repeat(MAX_C2_ENDPOINT_BYTES));
        assert!(matches!(C2NodeConfig::new(NodeId::new("node-c").unwrap(), oversized, "safe"), Err(C2ConfigError::InvalidEndpoint(_))));
    }

    #[test]
    fn durable_session_mutations_require_controller_and_use_bounded_deadlines() {
        let record_id = SessionRecordId::new("session-001").unwrap();
        let rename = NodeRequest::RenameSessionRecord {
            record_id: record_id.clone(),
            display_name: "release shepherd".to_owned(),
        };
        let resume = NodeRequest::ResumeSessionRecord {
            record_id: record_id.clone(),
            terminal_size: TerminalSize { rows: 40, columns: 120 },
            initial_prompt: None,
        };
        let forget = NodeRequest::ForgetSessionRecord { record_id };

        assert!(!is_read_only_request(&rename));
        assert!(!is_read_only_request(&resume));
        assert!(!is_read_only_request(&forget));
        assert_eq!(node_request_deadline(&rename), Duration::from_secs(5));
        assert_eq!(node_request_deadline(&resume), Duration::from_secs(35));
        assert!(node_request_deadline(&resume) > MANAGED_RESUME_SETTLE_DEADLINE);
        assert_eq!(
            node_request_deadline(&resume) - MANAGED_RESUME_SETTLE_DEADLINE,
            NODE_REQUEST_IO_HEADROOM,
        );
        let started = Instant::now();
        let relay_deadline = relay_request_deadline(&resume, started).unwrap();
        assert_eq!(
            request_budget(&resume, Some(relay_deadline), started),
            Duration::from_secs(35),
        );
        assert_eq!(
            request_budget(
                &resume,
                Some(relay_deadline),
                started + Duration::from_secs(5),
            ),
            MANAGED_RESUME_SETTLE_DEADLINE,
        );
        assert_eq!(
            request_budget(
                &resume,
                Some(relay_deadline),
                started + Duration::from_secs(35),
            ),
            Duration::ZERO,
        );
        assert_eq!(node_request_deadline(&forget), Duration::from_secs(5));
    }

    #[test]
    fn workspace_file_reads_are_read_only_with_five_second_deadline() {
        let request = NodeRequest::ReadWorkspaceFile {
            workspace_id: gate4agent_node_protocol::WorkspaceId::new("primary").unwrap(),
            path: gate4agent_node_protocol::RepositoryPath::utf8(
                "src/lib.rs".to_owned(),
            ).unwrap(),
        };

        assert!(is_read_only_request(&request));
        assert_eq!(node_request_deadline(&request), Duration::from_secs(5));
        assert!(relay_request_deadline(&request, Instant::now()).is_none());

    }

    #[test]
    fn workspace_entry_create_relay_has_headroom_and_preserves_semantic_timeout() {
        let workspace_id = gate4agent_node_protocol::WorkspaceId::new("primary").unwrap();
        let file_path = gate4agent_node_protocol::RepositoryPath::utf8(
            "src/new.rs".to_owned(),
        ).unwrap();
        let create_file = NodeRequest::CreateWorkspaceFile {
            workspace_id: workspace_id.clone(),
            path: file_path.clone(),
        };
        assert!(!is_read_only_request(&create_file));
        assert_eq!(
            node_request_deadline(&create_file),
            WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE,
        );
        assert!(
            node_request_deadline(&create_file)
                > WORKSPACE_ENTRY_CREATE_NODE_SEMANTIC_DEADLINE,
        );
        assert!(relay_request_deadline(&create_file, Instant::now()).is_none());

        let propagated = relay_node_failure(&NodeClientError::Node(
            gate4agent_node_protocol::NodeFailure {
                code: NodeFailureCode::RepositoryEntryCreateTimedOut,
                message: "private node timeout detail".to_owned(),
            },
        )).unwrap();
        assert_eq!(
            propagated.code,
            NodeFailureCode::RepositoryEntryCreateTimedOut,
        );
        assert_eq!(propagated.message, "repository entry creation timed out");

        let created_file = gate4agent_node_protocol::WorkspaceFileRead {
            workspace_id: workspace_id.clone(),
            path: file_path,
            content: gate4agent_node_protocol::WorkspaceFileContent::Utf8 {
                text: String::new(),
                byte_len: 0,
            },
            revision: Some(
                gate4agent_node_protocol::WorkspaceFileRevision::new(
                    "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
                        .to_owned(),
                )
                .unwrap(),
            ),
        };
        assert!(validate_workspace_content_response(
            Some(&create_file),
            &Ok(NodeResponse::WorkspaceFileCreated {
                file: created_file.clone(),
            }),
        ).is_ok());
        let mut wrong_content = created_file;
        wrong_content.content = gate4agent_node_protocol::WorkspaceFileContent::Utf8 {
            text: "unexpected".to_owned(),
            byte_len: 10,
        };
        assert!(validate_workspace_content_response(
            Some(&create_file),
            &Ok(NodeResponse::WorkspaceFileCreated { file: wrong_content }),
        ).is_err());

        let directory_path = gate4agent_node_protocol::RepositoryPath::utf8(
            "src/new".to_owned(),
        ).unwrap();
        let create_directory = NodeRequest::CreateWorkspaceDirectory {
            workspace_id: workspace_id.clone(),
            path: directory_path.clone(),
        };
        assert!(!is_read_only_request(&create_directory));
        assert_eq!(
            node_request_deadline(&create_directory),
            WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE,
        );
        assert!(validate_workspace_content_response(
            Some(&create_directory),
            &Ok(NodeResponse::WorkspaceDirectoryCreated {
                workspace_id,
                entry: gate4agent_node_protocol::WorkspaceEntry {
                    relative_path: directory_path,
                    kind: gate4agent_node_protocol::WorkspaceEntryKind::Directory,
                },
            }),
        ).is_ok());
        assert!(validate_workspace_content_response(
            Some(&create_directory),
            &Ok(NodeResponse::Accepted),
        ).is_err());
    }

    #[test]
    fn native_session_catalog_is_lease_free_read_only() {
        let route = gate4agent_node_protocol::NativeSessionCatalogRoute::workspace(
            gate4agent_node_protocol::WorkspaceId::new("primary").unwrap(),
            gate4agent_types::AgentId::new("codex").unwrap(),
        );
        let request = NodeRequest::CatalogNativeSessions {
            route: route.clone(),
            limit: 8,
        };
        assert!(is_read_only_request(&request));
        assert_eq!(
            node_request_deadline(&request),
            NATIVE_SESSION_REQUEST_DEADLINE,
        );
        assert!(relay_request_deadline(&request, Instant::now()).is_none());

        let page = NodeRequest::PageNativeSessions {
            route: route.clone(),
            window: gate4agent_types::NativeSessionCatalogWindow::Older,
            catalog_revision: 7,
            recent_cutoff_unix_ms: 8,
            after_selection_id: Some("hist_selection_1".to_owned()),
            limit: 8,
        };
        assert!(is_read_only_request(&page));
        assert_eq!(node_request_deadline(&page), NATIVE_SESSION_REQUEST_DEADLINE);
        assert!(relay_request_deadline(&page, Instant::now()).is_none());

        let preview = NodeRequest::PreviewNativeSession {
            selection: gate4agent_node_protocol::NativeSessionSelection {
                route,
                catalog_revision: 7,
                recent_cutoff_unix_ms: 8,
                selection_id: "hist_selection_1".to_owned(),
            },
            message_limit: 12,
        };
        assert!(is_read_only_request(&preview));
        assert_eq!(
            node_request_deadline(&preview),
            NATIVE_SESSION_REQUEST_DEADLINE,
        );
        assert!(relay_request_deadline(&preview, Instant::now()).is_none());

        let record_preview = NodeRequest::PreviewSessionRecord {
            record_id: gate4agent_node_protocol::SessionRecordId::new("record-1").unwrap(),
            message_limit: 12,
        };
        assert!(is_read_only_request(&record_preview));
        assert_eq!(
            node_request_deadline(&record_preview),
            NATIVE_SESSION_REQUEST_DEADLINE,
        );
        assert!(relay_request_deadline(&record_preview, Instant::now()).is_none());
    }

    #[test]
    fn standalone_workspace_creation_is_controller_mutation_with_worktree_deadline() {
        let request = NodeRequest::CreateStandaloneWorkspace {
            workspace_id: gate4agent_node_protocol::WorkspaceId::new("standalone").unwrap(),
            root: gate4agent_node_protocol::OpaqueHostPath::utf8(
                r"C:\standalone".to_owned(),
            ).unwrap(),
            initial_branch: Some("main".to_owned()),
        };

        assert!(!is_read_only_request(&request));
        assert_eq!(node_request_deadline(&request), Duration::from_secs(240));
        assert!(relay_request_deadline(&request, Instant::now()).is_none());
    }

    #[test]
    fn unsupported_node_capability_is_correlated_without_offline_classification() {
        let error = NodeClientError::UnsupportedCapability(
            "workspace-file-read-v1-private-detail".to_owned(),
        );
        let failure = relay_node_failure(&error)
            .expect("unsupported node capability must remain an in-band node failure");

        assert_eq!(failure.code, NodeFailureCode::UnsupportedCapability);
        assert_eq!(failure.message, "required capability unavailable");
        assert!(!failure.message.contains("private-detail"));
        assert!(relay_node_failure(&NodeClientError::Io(io::Error::new(
            io::ErrorKind::BrokenPipe,
            "transport closed",
        ))).is_none());
    }

    async fn raw_request(request: Vec<u8>, status: StatusResponse) -> Vec<u8> {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let (_status_tx, status_rx) = watch::channel(Arc::new(status));
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            serve_http(stream, "api-token", Duration::from_secs(1), &status_rx).await.unwrap();
        });
        let mut stream = TcpStream::connect(address).await.unwrap();
        stream.write_all(&request).await.unwrap();
        let mut response = Vec::new();
        stream.read_to_end(&mut response).await.unwrap();
        server.await.unwrap();
        response
    }

    fn empty_status(ready: bool) -> StatusResponse {
        StatusResponse { api_version: C2_API_VERSION, ready, observed_at_unix_ms: 0, nodes: BTreeMap::new() }
    }

    #[tokio::test]
    async fn http_api_enforces_initializing_auth_method_path_and_header_bounds() {
        let initializing = raw_request(b"GET /ready HTTP/1.1\r\nHost: localhost\r\n\r\n".to_vec(), empty_status(false)).await;
        assert!(initializing.starts_with(b"HTTP/1.1 503 Service Unavailable\r\n"));
        assert!(String::from_utf8_lossy(&initializing).contains("\"ready\":false"));

        let unauthorized = raw_request(b"GET /status HTTP/1.1\r\nHost: localhost\r\n\r\n".to_vec(), empty_status(true)).await;
        assert!(unauthorized.starts_with(b"HTTP/1.1 401 Unauthorized\r\n"));
        let method = raw_request(b"POST /health HTTP/1.1\r\nHost: localhost\r\n\r\n".to_vec(), empty_status(true)).await;
        assert!(method.starts_with(b"HTTP/1.1 405 Method Not Allowed\r\n"));
        let missing = raw_request(b"GET /missing HTTP/1.1\r\nHost: localhost\r\n\r\n".to_vec(), empty_status(true)).await;
        assert!(missing.starts_with(b"HTTP/1.1 404 Not Found\r\n"));

        let mut oversized = b"GET /health HTTP/1.1\r\nX-Fill: ".to_vec();
        oversized.extend(std::iter::repeat(b'x').take(HEADER_LIMIT_BYTES));
        oversized.extend_from_slice(b"\r\n\r\n");
        let rejected = raw_request(oversized, empty_status(true)).await;
        assert!(rejected.starts_with(b"HTTP/1.1 413 Payload Too Large\r\n"));
    }

    #[tokio::test]
    async fn inventory_state_transitions_offline_to_stale_parked_and_recovers() {
        let node_id = NodeId::new("node-a").unwrap();
        let mut nodes = BTreeMap::new();
        nodes.insert(node_id.clone(), ObservedNode {
            endpoint: r"\\.\pipe\a".to_owned(), transport_label: "windows-named-pipe".to_owned(),
            transport: NodeTransportState::Offline, freshness: NodeFreshness::Unavailable,
            cursor: None, inventory: None, last_attempt_unix_ms: None, last_success_unix_ms: None,
            consecutive_failures: 0, last_error: None, gaps: Vec::new(), gaps_truncated: 0,
        });
        let initial = Arc::new(StatusResponse { api_version: C2_API_VERSION, ready: false, observed_at_unix_ms: unix_ms(), nodes });
        let (status_tx, mut status_rx) = watch::channel(initial);
        let (ingress_tx, ingress_rx) = mpsc::channel(4);
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let owner = tokio::spawn(inventory_owner(1, Duration::from_millis(10), ingress_rx, status_tx, shutdown_rx));
        let snapshot = NodeSnapshot {
            node_id: node_id.clone(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: crate::protocol::ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            launch_inventory: None,
            agent_progress: Vec::new(),
        };
        let cursor = NodeCursor { incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([1; 16]), sequence: 0 };
        let old_manifest = ProviderContractManifest {
            provider_contracts: vec![crate::protocol::ProviderContractSupport {
                provider: agent("codex"),
                revision: crate::protocol::ProviderContractRevision::new("old-contract").unwrap(),
            }],
            provider_adapter_contracts: vec![crate::protocol::ProviderAdapterContractSupport {
                provider: agent("codex"),
                family: crate::protocol::AdapterFamily::PtySemantic,
                adapter_id: crate::protocol::AdapterId::new("codex").unwrap(),
                revision: crate::protocol::AdapterContractRevision::new("old-adapter").unwrap(),
            }],
        };
        ingress_tx.send(Attempt { node_id: node_id.clone(), at_unix_ms: unix_ms(), result: AttemptResult::Connected {
            cursor,
            snapshot: snapshot.clone(),
            gaps: Vec::new(),
            provider_contract_manifest: old_manifest,
        } }).await.unwrap();
        status_rx.changed().await.unwrap();
        assert_eq!(status_rx.borrow().nodes[&node_id].freshness, NodeFreshness::Fresh);
        assert_eq!(
            status_rx.borrow().nodes[&node_id].inventory.as_ref().unwrap()
                .provider_contracts[0].revision.as_str(),
            "old-contract",
        );

        let failure = || AttemptResult::Failure { error: SanitizedError { category: C2ErrorCategory::Transport, message: "node transport unavailable".to_owned() }, hard: false };
        ingress_tx.send(Attempt { node_id: node_id.clone(), at_unix_ms: unix_ms(), result: failure() }).await.unwrap();
        status_rx.changed().await.unwrap();
        assert_eq!(status_rx.borrow().nodes[&node_id].transport, NodeTransportState::Offline);
        timeout(Duration::from_secs(1), async {
            loop {
                status_rx.changed().await.unwrap();
                if status_rx.borrow().nodes[&node_id].freshness == NodeFreshness::Stale { break; }
            }
        }).await.unwrap();
        for _ in 0..4 {
            ingress_tx.send(Attempt { node_id: node_id.clone(), at_unix_ms: unix_ms(), result: failure() }).await.unwrap();
            status_rx.changed().await.unwrap();
        }
        assert_eq!(status_rx.borrow().nodes[&node_id].transport, NodeTransportState::Parked);
        let replacement_manifest = ProviderContractManifest {
            provider_contracts: vec![crate::protocol::ProviderContractSupport {
                provider: agent("claude"),
                revision: crate::protocol::ProviderContractRevision::new("new-contract").unwrap(),
            }],
            provider_adapter_contracts: Vec::new(),
        };
        let replacement_cursor = NodeCursor {
            incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([2; 16]),
            sequence: 0,
        };
        ingress_tx.send(Attempt { node_id: node_id.clone(), at_unix_ms: unix_ms(), result: AttemptResult::Connected {
            cursor: replacement_cursor,
            snapshot,
            gaps: Vec::new(),
            provider_contract_manifest: replacement_manifest,
        } }).await.unwrap();
        status_rx.changed().await.unwrap();
        assert_eq!(status_rx.borrow().nodes[&node_id].transport, NodeTransportState::Online);
        assert_eq!(status_rx.borrow().nodes[&node_id].freshness, NodeFreshness::Fresh);
        let recovered_inventory = status_rx.borrow().nodes[&node_id].inventory.as_ref().unwrap().clone();
        assert_eq!(recovered_inventory.provider_contracts.len(), 1);
        assert_eq!(recovered_inventory.provider_contracts[0].provider, agent("claude"));
        assert_eq!(recovered_inventory.provider_contracts[0].revision.as_str(), "new-contract");
        assert!(recovered_inventory.provider_adapter_contracts.is_empty());
        ingress_tx.send(Attempt {
            node_id: node_id.clone(),
            at_unix_ms: unix_ms(),
            result: AttemptResult::Connected {
                cursor: NodeCursor {
                    incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([3; 16]),
                    sequence: 0,
                },
                snapshot: NodeSnapshot {
                    node_id: node_id.clone(),
                    enabled_providers: Vec::new(),
                    provider_runtime_statuses: crate::protocol::ProviderRuntimeStatuses::default(),
                    workspaces: Vec::new(),
                    session_records: Vec::new(),
                    managed_worktrees: Vec::new(),
                    launch_inventory: None,
                    agent_progress: Vec::new(),
                },
                gaps: Vec::new(),
                provider_contract_manifest: ProviderContractManifest::default(),
            },
        }).await.unwrap();
        status_rx.changed().await.unwrap();
        let unpublished = status_rx.borrow().nodes[&node_id].inventory.as_ref().unwrap().clone();
        assert!(unpublished.provider_contracts.is_empty());
        assert!(unpublished.provider_adapter_contracts.is_empty());
        shutdown_tx.send(true).unwrap();
        owner.await.unwrap().unwrap();
    }

    fn runtime_statuses(
        provider: gate4agent_node_protocol::AgentId,
        version: &str,
    ) -> crate::protocol::ProviderRuntimeStatuses {
        crate::protocol::ProviderRuntimeStatuses::new([
            crate::protocol::ProviderRuntimeStatus::raw_passthrough(
                provider,
                Some(crate::protocol::ProviderRuntimeVersion::new(version).unwrap()),
            ),
        ])
        .unwrap()
    }

    async fn runtime_inventory_owner() -> (
        NodeId,
        mpsc::Sender<Attempt>,
        watch::Receiver<Arc<StatusResponse>>,
        watch::Sender<bool>,
        tokio::task::JoinHandle<io::Result<()>>,
    ) {
        let node_id = NodeId::new("runtime-node").unwrap();
        let nodes = BTreeMap::from([(
            node_id.clone(),
            ObservedNode {
                endpoint: r"\\.\pipe\runtime-node".to_owned(),
                transport_label: "windows-named-pipe".to_owned(),
                transport: NodeTransportState::Offline,
                freshness: NodeFreshness::Unavailable,
                cursor: None,
                inventory: None,
                last_attempt_unix_ms: None,
                last_success_unix_ms: None,
                consecutive_failures: 0,
                last_error: None,
                gaps: Vec::new(),
                gaps_truncated: 0,
            },
        )]);
        let initial = Arc::new(StatusResponse {
            api_version: C2_API_VERSION,
            ready: false,
            observed_at_unix_ms: unix_ms(),
            nodes,
        });
        let (status_tx, status_rx) = watch::channel(initial);
        let (ingress_tx, ingress_rx) = mpsc::channel(4);
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let owner = tokio::spawn(inventory_owner(
            1,
            Duration::from_secs(1),
            ingress_rx,
            status_tx,
            shutdown_rx,
        ));
        (node_id, ingress_tx, status_rx, shutdown_tx, owner)
    }

    #[tokio::test]
    async fn incarnation_change_replaces_runtime_status() {
        let (node_id, ingress, mut status, shutdown, owner) = runtime_inventory_owner().await;
        for (incarnation, provider, version) in [
            (1, agent("claude"), "1.0.0"),
            (2, agent("codex"), "2.0.0"),
        ] {
            ingress
                .send(Attempt {
                    node_id: node_id.clone(),
                    at_unix_ms: unix_ms(),
                    result: AttemptResult::Connected {
                        cursor: NodeCursor {
                            incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([
                                incarnation;
                                16
                            ]),
                            sequence: 0,
                        },
                        snapshot: NodeSnapshot {
                            node_id: node_id.clone(),
                            enabled_providers: vec![provider.clone()],
                            provider_runtime_statuses: runtime_statuses(provider, version),
                            workspaces: Vec::new(),
                            session_records: Vec::new(),
                            managed_worktrees: Vec::new(),
                            launch_inventory: None,
                            agent_progress: Vec::new(),
                        },
                        gaps: Vec::new(),
                        provider_contract_manifest: ProviderContractManifest::default(),
                    },
                })
                .await
                .unwrap();
            status.changed().await.unwrap();
        }
        let current_status = status.borrow();
        let statuses = &current_status.nodes[&node_id]
            .inventory
            .as_ref()
            .unwrap()
            .provider_runtime_statuses;
        assert_eq!(statuses.as_slice().len(), 1);
        assert_eq!(
            statuses.as_slice()[0].provider(),
            &agent("codex"),
        );
        assert_eq!(statuses.as_slice()[0].version().unwrap().as_str(), "2.0.0");
        shutdown.send(true).unwrap();
        owner.await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn incarnation_change_without_snapshot_clears_dynamic_inventory() {
        let (node_id, ingress, mut status, shutdown, owner) = runtime_inventory_owner().await;
        ingress
            .send(Attempt {
                node_id: node_id.clone(),
                at_unix_ms: unix_ms(),
                result: AttemptResult::Connected {
                    cursor: NodeCursor {
                        incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([
                            3; 16
                        ]),
                        sequence: 0,
                    },
                    snapshot: NodeSnapshot {
                        node_id: node_id.clone(),
                        enabled_providers: vec![agent("claude")],
                        provider_runtime_statuses: runtime_statuses(
                            agent("claude"),
                            "3.0.0",
                        ),
                        workspaces: Vec::new(),
                        session_records: Vec::new(),
                        managed_worktrees: Vec::new(),
                        launch_inventory: None,
                        agent_progress: Vec::new(),
                    },
                    gaps: Vec::new(),
                    provider_contract_manifest: ProviderContractManifest::default(),
                },
            })
            .await
            .unwrap();
        status.changed().await.unwrap();
        ingress
            .send(Attempt {
                node_id: node_id.clone(),
                at_unix_ms: unix_ms(),
                result: AttemptResult::Cursor {
                    cursor: NodeCursor {
                        incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([
                            4; 16
                        ]),
                        sequence: 0,
                    },
                    gaps: vec![GapKind::IncarnationChanged],
                    managed_worktree_events: Vec::new(),
                },
            })
            .await
            .unwrap();
        status.changed().await.unwrap();
        assert!(status.borrow().nodes[&node_id]
            .inventory
            .as_ref()
            .unwrap()
            .provider_runtime_statuses
            .is_empty());
        shutdown.send(true).unwrap();
        owner.await.unwrap().unwrap();
    }

    #[test]
    fn managed_worktree_inventory_events_are_exact_bounded_and_incarnation_fenced() {
        let mut inventory = SlimNodeInventory::from_snapshot(&NodeSnapshot {
            node_id: NodeId::new("node-a").unwrap(),
            enabled_providers: Vec::new(),
            provider_runtime_statuses: crate::protocol::ProviderRuntimeStatuses::default(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: vec![managed_lease(
                "lease-a",
                "managed-a",
                ManagedWorktreeLeaseState::Ready,
            )],
            launch_inventory: None,
            agent_progress: Vec::new(),
        });
        apply_managed_worktree_cursor(
            Some(&mut inventory),
            false,
            &[
                NodeEvent::ManagedWorktreeUpserted {
                    lease: managed_lease(
                        "lease-b",
                        "managed-b",
                        ManagedWorktreeLeaseState::Ready,
                    ),
                },
                NodeEvent::ManagedWorktreeUpserted {
                    lease: managed_lease(
                        "lease-a",
                        "managed-a",
                        ManagedWorktreeLeaseState::InUse,
                    ),
                },
            ],
        );
        assert_eq!(inventory.managed_worktree_count, 2);
        assert_eq!(inventory.managed_worktrees[0].lease_id.as_str(), "lease-a");
        assert_eq!(
            inventory.managed_worktrees[0].state,
            ManagedWorktreeLeaseState::InUse,
        );

        apply_managed_worktree_cursor(
            Some(&mut inventory),
            false,
            &[NodeEvent::ManagedWorktreeRemoved {
                lease_id: ManagedWorktreeLeaseId::new("lease-a").unwrap(),
            }],
        );
        assert_eq!(inventory.managed_worktree_count, 1);
        assert_eq!(inventory.managed_worktrees[0].lease_id.as_str(), "lease-b");

        apply_managed_worktree_cursor(
            Some(&mut inventory),
            true,
            &[NodeEvent::ManagedWorktreeUpserted {
                lease: managed_lease(
                    "lease-c",
                    "managed-c",
                    ManagedWorktreeLeaseState::Ready,
                ),
            }],
        );
        assert!(inventory.managed_worktrees.is_empty());
        assert_eq!(inventory.managed_worktree_count, 0);
        assert!(!inventory.managed_worktrees_truncated);
    }
}

#[cfg(test)]
mod relay_deadline_tests {
    use super::*;

    /// The defect this pins, stated as an invariant rather than a number:
    /// the relay's bound on a request must never sit below the budget the
    /// node itself is allowed to spend answering it. It sat at five
    /// seconds against the node's eight-to-eleven, so every inspection of
    /// a workspace large enough to use its own allowance timed out at the
    /// relay -- and a relay timeout drops the NODE, not the request, which
    /// took every read behind it down in a loop.
    #[test]
    fn workspace_inspection_relay_bound_clears_the_nodes_own_maximum() {
        // The node's own ceiling (`WORKSPACE_INSPECTION_TIME_BUDGET_MS_MAX`
        // in the node crate) restated here rather than imported: these are
        // separate crates by design, and the point of the test is that the
        // two numbers must be compared by a human when either moves.
        const NODE_INSPECTION_MAX: Duration = Duration::from_millis(11_000);
        assert!(
            WORKSPACE_INSPECTION_RELAY_DEADLINE > NODE_INSPECTION_MAX,
            "relay bound {:?} must exceed the node's own inspection maximum {:?}",
            WORKSPACE_INSPECTION_RELAY_DEADLINE,
            NODE_INSPECTION_MAX,
        );
    }
}

#[cfg(test)]
mod control_tunnel_listen_tests {
    use super::*;

    fn config() -> C2Config {
        let node = C2NodeConfig::new(
            NodeId::new("node-a").unwrap(),
            if cfg!(unix) { "/tmp/gate4agent-node-a.sock" } else { r"\\.\pipe\gate4agent-node-a" },
            "token-token",
        ).unwrap();
        C2Config::new("127.0.0.1:18321".parse().unwrap(), "api-token", vec![node]).unwrap()
    }

    #[test]
    fn control_tunnel_rejects_unspecified_and_accepts_tunnel_address() {
        let bad: SocketAddr = "0.0.0.0:27441".parse().unwrap();
        assert!(matches!(
            config().with_control_tunnel_listen(bad),
            Err(C2ConfigError::InvalidControlTunnel(_))
        ));
        let ok: SocketAddr = "10.88.1.2:27441".parse().unwrap();
        let config = config().with_control_tunnel_listen(ok).unwrap();
        assert_eq!(config.control_tunnel_listen, Some(ok));
        assert!(!config.control_endpoint.is_empty());
    }
}