hatchery-c2-client 0.2.3

use hatchery_c2_protocol::{
    c2_auth_transcript, c2_bound_auth_transcript, provider_id_is_legacy, C2AuthDirection, C2ClientAuthentication, C2ClientFrame,
    C2ClientHello, C2Hello, C2NodeEvent, C2NodeResponse, C2RelayFailure,
    C2RelayFailureCode,
    C2RequestEnvelope, C2RequestId, C2ServerFrame, C2Topology, CapabilityId,
    ClientCompatibilityOffer, NegotiatedC2ControlCompatibility, NodeRequest, NodeRoute,
    RoutedNodeEvent, RoutedNodeRequest, RoutedNodeResponse,
    BUILD_STAMP, C2_COMPATIBILITY_METADATA_CAPABILITY,
    C2_OPAQUE_UNIX_PATH_CAPABILITY, C2_REPOSITORY_PATH_CAPABILITY,
    C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY,
    C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY,
    C2_HISTORY_CONTEXT_PACK_CAPABILITY,
    C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY,
    C2_NATIVE_SESSION_CATALOG_CAPABILITY,
    C2_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY,
    C2_NATIVE_SESSION_INDEX_CAPABILITY, C2_NATIVE_SESSION_PREVIEW_CAPABILITY,
    C2_HOST_DIRECTORY_BROWSE_CAPABILITY,
    C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY,
    C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY,
    C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY,
    C2_CONTROL_DETAIL_CAPABILITY, C2_SESSION_TASK_CORRELATION_CAPABILITY,
    C2_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY,
    C2_HARNESS_MCP_READ_PROXY_CAPABILITY,
    C2_PROVIDER_ID_OPEN_CAPABILITY,
    C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY,
    C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY,
    C2_PROVIDER_CONTRACT_MANIFEST_CAPABILITY, C2_PROVIDER_RUNTIME_STATUS_CAPABILITY,
    C2_SPAWN_PROFILE_REVISION_CAPABILITY,
    C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY,
    C2_TERMINAL_FRAME_EVENTS_CAPABILITY,
    C2_AGENT_STREAM_EVENTS_CAPABILITY,
    C2_ACP_CONTROL_CAPABILITY,
    C2_GIT_READ_CAPABILITY, C2_WORKSPACE_FILE_READ_CAPABILITY,
    C2_WORKSPACE_FILE_WRITE_CAPABILITY,
    C2_WORKSPACE_ENTRY_CREATE_CAPABILITY,
    C2_WORKTREE_SELECTION_CAPABILITY,
    C2_AUTH_NONCE_BYTES, MAX_C2_AUTH_FRAME_BYTES, MAX_C2_CLIENT_FRAME_BYTES, MAX_C2_HELLO_FRAME_BYTES,
    MAX_C2_SERVER_FRAME_BYTES,
};
use gate4agent_node_protocol::{
    read_json_frame_limited_body_timeout, write_json_frame_limited, FrameError,
};
use gate4agent_node_wire::{
    connect_local_stream, local_hmac_sha256, proofs_match, random_nonce,
};
use std::collections::BTreeMap;
use std::io;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use thiserror::Error;
use tokio::sync::{mpsc, oneshot, watch};
use tokio::time::{timeout, timeout_at, Instant};

const AUTH_DEADLINE: Duration = Duration::from_secs(5);
const HELLO_DEADLINE: Duration = Duration::from_secs(10);
const FRAME_BODY_DEADLINE: Duration = Duration::from_secs(5);
const RELAY_REPLY_HEADROOM: Duration = Duration::from_secs(5);
const CONTROLLER_ACQUIRE_RELAY_DEADLINE: Duration = Duration::from_secs(5);
const WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE: Duration = Duration::from_secs(10);
const NATIVE_SESSION_RELAY_DEADLINE: Duration = Duration::from_secs(35);
const COMMAND_CAPACITY: usize = 64;
const INBOUND_CAPACITY: usize = 2;
const WRITER_CAPACITY: usize = 64;
/// Buffer for the primary node-event stream (control, terminal, and
/// negotiated agent-stream text deltas). Sized so
/// normal live-session bursts buffer instead of tearing the connection
/// down; teardown (`RegularEventBackpressure`) stays the last resort for a
/// consumer that is genuinely stuck, not merely bursty.
pub(crate) const EVENT_CAPACITY: usize = 1024;
pub(crate) const HARNESS_MCP_EVENT_CAPACITY: usize = 128;
const OPAQUE_UNIX_PATH_NOT_NEGOTIATED: &str =
    "opaque Unix paths require negotiated C2 capability";
const REPOSITORY_PATH_NOT_NEGOTIATED: &str =
    "tagged repository paths require negotiated C2 capability";
const WORKSPACE_FILE_READ_NOT_NEGOTIATED: &str =
    "workspace file reads require negotiated C2 capability";
const WORKSPACE_ENTRY_CREATE_NOT_NEGOTIATED: &str =
    "workspace entry creation requires negotiated C2 capability";
const OPEN_PROVIDER_ID_NOT_NEGOTIATED: &str =
    "open provider IDs require negotiated C2 capability";
const SPAWN_SPEC_NOT_NEGOTIATED: &str =
    "spawn spec defaults/overrides require negotiated C2 capability";
const SPAWN_PROFILE_REVISION_NOT_NEGOTIATED: &str =
    "spawn profile revisions require negotiated C2 capability";
const TERMINAL_FRAME_EVENTS_NOT_NEGOTIATED: &str =
    "terminal frame events require negotiated C2 capability";
const AGENT_STREAM_EVENTS_NOT_NEGOTIATED: &str =
    "agent stream events require negotiated C2 capability";
const ACP_CONTROL_NOT_NEGOTIATED: &str =
    "ACP control verbs require negotiated C2 capability";
const WORKTREE_SELECTION_NOT_NEGOTIATED: &str =
    "worktree selection requires negotiated C2 capability";
const MANAGED_WORKTREE_LIFECYCLE_NOT_NEGOTIATED: &str =
    "managed worktree lifecycle requires negotiated C2 capability";
const MANAGED_WORKTREE_SPAWN_V2_NOT_NEGOTIATED: &str =
    "managed worktree V2 spawn requires negotiated C2 capability";
const CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED: &str =
    "child environment profiles require negotiated C2 capability";
const SESSION_BUNDLE_MATERIALIZATION_NOT_NEGOTIATED: &str =
    "session bundle materialization requires negotiated C2 capability";
const HISTORY_CONTEXT_PACK_NOT_NEGOTIATED: &str =
    "history context packs require negotiated C2 capability";
const SESSION_RECORD_CONTEXT_EXPORT_NOT_NEGOTIATED: &str =
    "session-record context export requires negotiated C2 capability";
const HOST_DIRECTORY_BROWSE_NOT_NEGOTIATED: &str =
    "host directory browsing requires negotiated C2 capability";
const STANDALONE_WORKSPACE_LIFECYCLE_NOT_NEGOTIATED: &str =
    "standalone workspace lifecycle requires negotiated C2 capability";
const PROVIDER_SESSION_REFERENCE_INDEX_NOT_NEGOTIATED: &str =
    "provider session reference index requires negotiated C2 capability";
const INVALID_HISTORY_CONTEXT_PACK_REQUEST: &str =
    "invalid history context pack request";
const SESSION_TASK_CORRELATION_NOT_NEGOTIATED: &str =
    "session task correlation requires negotiated C2 capability";
const CONTROL_DETAIL_NOT_NEGOTIATED: &str =
    "control detail requires negotiated C2 capability";

#[derive(Clone, Copy, Default)]
struct NegotiatedPathCapabilities {
    opaque_host_paths: bool,
    repository_paths: bool,
    workspace_file_read: bool,
    workspace_file_write: bool,
    workspace_entry_create: bool,
    git_read: bool,
    host_directory_browse: bool,
    standalone_workspace_lifecycle: bool,
    provider_session_reference_index: bool,
    provider_ids_open: bool,
    spawn_spec_defaults_overrides: bool,
    spawn_profile_revision: bool,
    worktree_selection: bool,
    managed_worktree_lifecycle: bool,
    managed_worktree_spawn_v2: bool,
    child_environment_profile: bool,
    session_bundle_materialization: bool,
    history_context_pack: bool,
    session_record_context_export: bool,
    native_session_catalog: bool,
    native_session_catalog_paging: bool,
    native_session_index: bool,
    native_session_preview: bool,
    terminal_frame_events: bool,
    agent_stream_events: bool,
    agent_progress_snapshot: bool,
    session_task_correlation: bool,
    control_detail: bool,
    delivery_bundle_v2_stage_commit: bool,
    harness_mcp_read_proxy: bool,
    acp_control: bool,
}

#[derive(Clone)]
pub struct C2ControlHandle {
    commands: mpsc::Sender<ControlCommand>,
    hello: Arc<C2Hello>,
    topology: watch::Receiver<Arc<C2Topology>>,
    terminal_frame_events: bool,
}

impl C2ControlHandle {
    pub fn hello(&self) -> &C2Hello { &self.hello }

    pub fn current_topology(&self) -> Arc<C2Topology> { Arc::clone(&*self.topology.borrow()) }

    pub fn subscribe_topology(&self) -> watch::Receiver<Arc<C2Topology>> {
        self.topology.clone()
    }

    pub fn terminal_frame_events_enabled(&self) -> bool { self.terminal_frame_events }

    pub async fn request(
        &self,
        route: NodeRoute,
        request: NodeRequest,
    ) -> Result<RoutedNodeResponse, C2ControlError> {
        reject_unnegotiated_outbound_path(
            &request,
            negotiated_path_capabilities(self.hello.compatibility.as_ref()),
        )?;
        let deadline = control_request_deadline(&request);
        let (reply_tx, reply_rx) = oneshot::channel();
        timeout(deadline, async {
            self.commands.send(ControlCommand { route, request, reply: reply_tx })
                .await.map_err(|_| C2ControlError::Closed)?;
            reply_rx.await.map_err(|_| C2ControlError::Closed)?
        }).await.map_err(|_| C2ControlError::Closed)?
    }

    /// Validates and synchronously enqueues one typed request. The returned
    /// non-cloneable waiter cannot enqueue or replay the request again.
    pub fn start_request(
        &self,
        route: NodeRoute,
        request: NodeRequest,
    ) -> Result<C2PendingRequest, C2ControlError> {
        start_control_request(
            &self.commands,
            negotiated_path_capabilities(self.hello.compatibility.as_ref()),
            route,
            request,
        )
    }
}

pub struct C2PendingRequest {
    reply: oneshot::Receiver<Result<RoutedNodeResponse, C2ControlError>>,
    deadline: Instant,
}

impl C2PendingRequest {
    pub async fn finish(self) -> Result<RoutedNodeResponse, C2ControlError> {
        timeout_at(self.deadline, self.reply)
            .await
            .map_err(|_| C2ControlError::Closed)?
            .map_err(|_| C2ControlError::Closed)?
    }
}

fn start_control_request(
    commands: &mpsc::Sender<ControlCommand>,
    capabilities: NegotiatedPathCapabilities,
    route: NodeRoute,
    request: NodeRequest,
) -> Result<C2PendingRequest, C2ControlError> {
        reject_unnegotiated_outbound_path(
            &request,
            capabilities,
        )?;
        let deadline_at = Instant::now() + control_request_deadline(&request);
        let (reply_tx, reply_rx) = oneshot::channel();
        commands.try_send(ControlCommand { route, request, reply: reply_tx })
            .map_err(|error| match error {
                mpsc::error::TrySendError::Full(_) => C2ControlError::QueueFull,
                mpsc::error::TrySendError::Closed(_) => C2ControlError::Closed,
            })?;
        Ok(C2PendingRequest {
            reply: reply_rx,
            deadline: deadline_at,
        })
}

fn control_request_deadline(request: &NodeRequest) -> Duration {
    let relay_deadline = match request {
        NodeRequest::Snapshot
        | NodeRequest::Resync { .. }
        | NodeRequest::ArmHarnessMcpReservation { .. }
        | NodeRequest::SpawnSpecWithHarnessMcp { .. }
        | NodeRequest::ActivateHarnessMcpReservation { .. }
        | NodeRequest::AbortHarnessMcpReservation { .. }
        | NodeRequest::PutHarnessMcpReplyChunk { .. }
        | NodeRequest::RejectHarnessMcpCall { .. }
        | NodeRequest::BeginDeliveryStage { .. }
        | NodeRequest::PutDeliveryBlobChunk { .. }
        | NodeRequest::CommitDeliveryStage { .. }
        | NodeRequest::AbortDeliveryStage { .. }
        | NodeRequest::BrowseHostDirectories { .. }
        | NodeRequest::InspectWorkspace { .. }
        | NodeRequest::ReadWorkspaceFile { .. }
        | NodeRequest::WriteWorkspaceFile { .. }
        | NodeRequest::ReadGitHistory { .. }
        | NodeRequest::ReadGitDiff { .. }
        | NodeRequest::AcquireController { .. }
        | NodeRequest::ReleaseController
        | NodeRequest::RenameSessionRecord { .. }
        | NodeRequest::SetSessionTask { .. }
        | NodeRequest::ForgetSessionRecord { .. } => Duration::from_secs(5),
        NodeRequest::CreateWorkspaceFile { .. }
        | NodeRequest::CreateWorkspaceDirectory { .. } => {
            CONTROLLER_ACQUIRE_RELAY_DEADLINE
                .saturating_add(WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE)
        }
        NodeRequest::CatalogNativeSessions { .. }
        | NodeRequest::PageNativeSessions { .. } => NATIVE_SESSION_RELAY_DEADLINE,
        NodeRequest::PreviewNativeSession { .. }
        | NodeRequest::IndexNativeSession { .. }
        | NodeRequest::PreviewSessionRecord { .. } => NATIVE_SESSION_RELAY_DEADLINE,
        NodeRequest::ExportContextPackForSessionRecord { .. } => Duration::from_secs(45),
        NodeRequest::CreateWorktree { .. }
        | NodeRequest::CreateStandaloneWorkspace { .. }
        | 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()) + RELAY_REPLY_HEADROOM,
        NodeRequest::SpawnManagedWorktree { request } =>
            Duration::from_millis(request.spawn_spec.deadline_ms.get())
                + RELAY_REPLY_HEADROOM,
        NodeRequest::SpawnManagedWorktreeV2 { request } =>
            Duration::from_millis(request.spawn_spec.deadline_ms.get())
                + RELAY_REPLY_HEADROOM,
        NodeRequest::ResumeSessionRecord { .. } => Duration::from_secs(35),
        _ => Duration::from_secs(10),
    };
    relay_deadline + RELAY_REPLY_HEADROOM
}

pub struct C2EventReceiver {
    events: mpsc::Receiver<RoutedNodeEvent>,
    harness_mcp_events: mpsc::Receiver<RoutedNodeEvent>,
}

impl C2EventReceiver {
    pub async fn recv(&mut self) -> Option<RoutedNodeEvent> {
        loop {
            if self.harness_mcp_events.is_closed() && self.harness_mcp_events.is_empty() {
                return self.events.recv().await;
            }
            if self.events.is_closed() && self.events.is_empty() {
                return self.harness_mcp_events.recv().await;
            }
            tokio::select! {
                biased;
                event = self.harness_mcp_events.recv() => {
                    if event.is_some() { return event; }
                }
                event = self.events.recv() => {
                    if event.is_some() { return event; }
                }
            }
        }
    }

    /// Splits into two independent mutable borrows, one per channel, so a
    /// caller can race both in a single `tokio::select!` (the reconnect
    /// bridge's own `pump_one_connection`). Two separate `&mut self` async
    /// accessor methods (one per channel) cannot be used together in one
    /// `select!`: the future each returns borrows the whole receiver for
    /// its lifetime, and the borrow checker rejects two such futures
    /// coexisting even though the methods would only touch disjoint
    /// fields. Splitting once, up front, hands out two genuinely disjoint
    /// borrows instead.
    pub(crate) fn split_mut(
        &mut self,
    ) -> (&mut mpsc::Receiver<RoutedNodeEvent>, &mut mpsc::Receiver<RoutedNodeEvent>) {
        (&mut self.events, &mut self.harness_mcp_events)
    }
}

struct EventDelivery {
    regular: mpsc::Sender<RoutedNodeEvent>,
    harness_mcp: mpsc::Sender<RoutedNodeEvent>,
}

impl From<mpsc::Sender<RoutedNodeEvent>> for EventDelivery {
    fn from(regular: mpsc::Sender<RoutedNodeEvent>) -> Self {
        Self { harness_mcp: regular.clone(), regular }
    }
}

struct ControlCommand {
    route: NodeRoute,
    request: NodeRequest,
    reply: oneshot::Sender<Result<RoutedNodeResponse, C2ControlError>>,
}

/// Recovers a `BuildStampMismatch` relay rejection's [`C2ControlError::
/// BuildStampMismatch`] shape from the frame `gate4agent-c2`'s own
/// pre-handshake refusal actually sends. `C2RelayFailureCode::
/// BuildStampMismatch` itself carries nothing -- both stamps travel only in
/// `failure.message` (`"build stamp mismatch: local=<s> remote=<s>"`,
/// produced by this exact tree's own `gate4agent-c2`), so the peer's stamp
/// is recovered by parsing that self-authored, fixed-format text rather
/// than inventing a second wire shape for the same two values. Any other
/// code, or a `BuildStampMismatch` whose message does not parse (never
/// happens against a peer built from this tree, but never trusted blindly
/// either), falls back to the generic [`C2ControlError::Relay`].
fn map_relay_rejection(failure: C2RelayFailure) -> C2ControlError {
    if failure.code != C2RelayFailureCode::BuildStampMismatch {
        return C2ControlError::Relay(failure);
    }
    match failure.message.split_whitespace().find_map(|token| token.strip_prefix("remote=")) {
        Some(remote) => C2ControlError::BuildStampMismatch {
            local: BUILD_STAMP.to_owned(),
            remote: remote.to_owned(),
        },
        None => C2ControlError::Relay(failure),
    }
}

enum OwnerInput {
    Frame(C2ServerFrame),
    Closed,
}

pub async fn connect_local(
    endpoint: &str,
    token: &str,
) -> Result<(C2ControlHandle, C2EventReceiver), C2ControlError> {
    validate_endpoint(endpoint)?;
    validate_token(token)?;
    let pipe = connect_local_stream(endpoint).await?;
    connect_control_stream(pipe, token).await
}

/// Dials the C2 control frame protocol over TCP.
///
/// This is the HQ path after the kernel WireGuard tunnel is up: `address`
/// is `c2_tunnel_ip:control_port`, not a public endpoint and not a node.
/// The unix validator ([`validate_endpoint`]) does not accept this form.
pub async fn connect_tunnel_tcp(
    address: std::net::SocketAddr,
    token: &str,
) -> Result<(C2ControlHandle, C2EventReceiver), C2ControlError> {
    validate_tunnel_control_target(address)?;
    validate_token(token)?;
    let stream = tokio::net::TcpStream::connect(address).await?;
    let _ = stream.set_nodelay(true);
    connect_control_stream(stream, token).await
}

async fn connect_control_stream<S>(
    mut pipe: S,
    token: &str,
) -> Result<(C2ControlHandle, C2EventReceiver), C2ControlError>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
    let client_nonce = random_nonce().map_err(C2ControlError::Authentication)?;
    let compatibility_offer = client_compatibility_offer()?;
    timeout(AUTH_DEADLINE, write_json_frame_limited(
        &mut pipe,
        &C2ClientFrame::Hello(C2ClientHello::negotiating(
            client_nonce,
            compatibility_offer.clone(),
        )),
        MAX_C2_AUTH_FRAME_BYTES,
    )).await.map_err(|_| C2ControlError::AuthenticationTimedOut)??;
    let challenge = timeout(AUTH_DEADLINE, read_server_frame(&mut pipe, MAX_C2_AUTH_FRAME_BYTES))
        .await.map_err(|_| C2ControlError::AuthenticationTimedOut)??;
    let challenge = match challenge {
        C2ServerFrame::Challenge(challenge) => challenge,
        C2ServerFrame::Rejected(failure) => return Err(map_relay_rejection(failure)),
        _ => return Err(C2ControlError::Protocol(
            "C2 did not return an authentication challenge".to_owned(),
        )),
    };
    if challenge.build_stamp != BUILD_STAMP {
        return Err(C2ControlError::BuildStampMismatch {
            local: BUILD_STAMP.to_owned(),
            remote: challenge.build_stamp.clone(),
        });
    }
    let selected = challenge.compatibility.as_ref().ok_or_else(|| {
        C2ControlError::Protocol(
            "C2 omitted the required authenticated compatibility selection".to_owned(),
        )
    })?;
    validate_selected_compatibility(
        &compatibility_offer,
        Some(selected),
    )?;
    let expected_server = c2_proof(
        token,
        C2AuthDirection::Server,
        &client_nonce,
        &challenge.server_nonce,
        Some((&compatibility_offer, selected)),
    )?;
    if !proofs_match(&challenge.server_proof, &expected_server) {
        return Err(C2ControlError::Authentication("C2 server proof mismatch".to_owned()));
    }
    let client_proof = c2_proof(
        token,
        C2AuthDirection::Client,
        &client_nonce,
        &challenge.server_nonce,
        Some((&compatibility_offer, selected)),
    )?;
    timeout(AUTH_DEADLINE, write_json_frame_limited(
        &mut pipe,
        &C2ClientFrame::Authenticate(C2ClientAuthentication { client_proof }),
        MAX_C2_AUTH_FRAME_BYTES,
    )).await.map_err(|_| C2ControlError::AuthenticationTimedOut)??;
    let hello = timeout(HELLO_DEADLINE, read_server_frame(&mut pipe, MAX_C2_HELLO_FRAME_BYTES))
        .await.map_err(|_| C2ControlError::AuthenticationTimedOut)??;
    let hello = match hello {
        C2ServerFrame::Hello(hello) if hello.build_stamp == BUILD_STAMP => hello,
        C2ServerFrame::Rejected(failure) =>
            return Err(map_relay_rejection(failure)),
        C2ServerFrame::Hello(hello) =>
            return Err(C2ControlError::BuildStampMismatch {
                local: BUILD_STAMP.to_owned(),
                remote: hello.build_stamp.clone(),
            }),
        _ => return Err(C2ControlError::Protocol("C2 did not return hello".to_owned())),
    };
    if hello.compatibility.is_none() {
        return Err(C2ControlError::Protocol(
            "C2 omitted the authenticated compatibility selection from hello".to_owned(),
        ));
    }
    validate_selected_compatibility(&compatibility_offer, hello.compatibility.as_ref())?;
    if hello.compatibility != challenge.compatibility {
        return Err(C2ControlError::Protocol(
            "C2 compatibility selection changed after authentication".to_owned(),
        ));
    }
    let path_capabilities = negotiated_path_capabilities(hello.compatibility.as_ref());
    if !path_capabilities.provider_ids_open && status_has_open_provider_id(&hello.status) {
        return Err(C2ControlError::Protocol(
            OPEN_PROVIDER_ID_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if !path_capabilities.managed_worktree_lifecycle
        && status_has_managed_worktree(&hello.status)
    {
        return Err(C2ControlError::Protocol(
            MANAGED_WORKTREE_LIFECYCLE_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if !path_capabilities.worktree_selection
        && status_has_managed_worktree(&hello.status)
    {
        return Err(C2ControlError::Protocol(
            WORKTREE_SELECTION_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if !path_capabilities.child_environment_profile
        && status_has_child_environment_profile(&hello.status)
    {
        return Err(C2ControlError::Protocol(
            CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if !path_capabilities.session_bundle_materialization
        && status_has_session_bundle_materialization(&hello.status)
    {
        return Err(C2ControlError::Protocol(
            SESSION_BUNDLE_MATERIALIZATION_NOT_NEGOTIATED.to_owned(),
        ));
    }

    let (reader, writer) = tokio::io::split(pipe);
    let (commands_tx, commands_rx) = mpsc::channel(COMMAND_CAPACITY);
    let (events_tx, events_rx) = mpsc::channel(EVENT_CAPACITY);
    let (harness_mcp_events_tx, harness_mcp_events_rx) =
        mpsc::channel(HARNESS_MCP_EVENT_CAPACITY);
    let initial_topology = Arc::new(C2Topology::from_status(&hello.status));
    let (topology_tx, topology_rx) = watch::channel(initial_topology);
    let (writer_tx, writer_rx) = mpsc::channel(WRITER_CAPACITY);
    let (owner_tx, owner_rx) = mpsc::channel(INBOUND_CAPACITY);
    let reader_task = tokio::spawn(control_reader(reader, owner_tx.clone()));
    let writer_task = tokio::spawn(control_writer(writer, writer_rx, owner_tx));
    tokio::spawn(async move {
        control_owner(
            commands_rx,
        EventDelivery { regular: events_tx, harness_mcp: harness_mcp_events_tx },
            topology_tx,
            writer_tx,
            owner_rx,
            path_capabilities,
        ).await;
        reader_task.abort();
        writer_task.abort();
    });
    Ok((C2ControlHandle {
        commands: commands_tx,
        hello: Arc::new(hello),
        topology: topology_rx,
        terminal_frame_events: path_capabilities.terminal_frame_events,
    }, C2EventReceiver {
        events: events_rx,
        harness_mcp_events: harness_mcp_events_rx,
    }))
}

pub(crate) fn client_compatibility_offer() -> Result<ClientCompatibilityOffer, C2ControlError> {
    Ok(ClientCompatibilityOffer {
        build_stamp: BUILD_STAMP.to_owned(),
        capabilities: vec![
            CapabilityId::new(C2_COMPATIBILITY_METADATA_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_OPAQUE_UNIX_PATH_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_REPOSITORY_PATH_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_WORKSPACE_FILE_READ_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_WORKSPACE_FILE_WRITE_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_WORKSPACE_ENTRY_CREATE_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_GIT_READ_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_PROVIDER_CONTRACT_MANIFEST_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_PROVIDER_RUNTIME_STATUS_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_PROVIDER_ID_OPEN_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_SPAWN_PROFILE_REVISION_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_TERMINAL_FRAME_EVENTS_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_AGENT_STREAM_EVENTS_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_WORKTREE_SELECTION_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_HISTORY_CONTEXT_PACK_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_NATIVE_SESSION_CATALOG_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_NATIVE_SESSION_INDEX_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_NATIVE_SESSION_PREVIEW_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_HOST_DIRECTORY_BROWSE_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_SESSION_TASK_CORRELATION_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_CONTROL_DETAIL_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_HARNESS_MCP_READ_PROXY_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
            CapabilityId::new(C2_ACP_CONTROL_CAPABILITY)
                .map_err(|error| C2ControlError::Protocol(error.to_string()))?,
        ],
        state_schema: None,
    })
}

fn validate_selected_compatibility(
    offer: &ClientCompatibilityOffer,
    selected: Option<&NegotiatedC2ControlCompatibility>,
) -> Result<(), C2ControlError> {
    let Some(selected) = selected else {
        return Err(C2ControlError::Protocol(
            "C2 omitted the required authenticated compatibility selection".to_owned(),
        ));
    };
    if selected.build_stamp != offer.build_stamp {
        return Err(C2ControlError::BuildStampMismatch {
            local: offer.build_stamp.clone(),
            remote: selected.build_stamp.clone(),
        });
    }
    if selected
        .capabilities
        .iter()
        .any(|capability| !offer.capabilities.contains(capability))
    {
        return Err(C2ControlError::Protocol(
            "C2 selected a capability outside the client offer".to_owned(),
        ));
    }
    if !selected
        .capabilities
        .iter()
        .any(|capability| capability.as_str() == C2_COMPATIBILITY_METADATA_CAPABILITY)
    {
        return Err(C2ControlError::Protocol(
            "C2 omitted the required compatibility metadata capability".to_owned(),
        ));
    }
    Ok(())
}

fn negotiated_path_capabilities(
    selected: Option<&NegotiatedC2ControlCompatibility>,
) -> NegotiatedPathCapabilities {
    let selected_has = |expected| {
        selected.is_some_and(|selected| {
            selected
                .capabilities
                .iter()
                .any(|capability| capability.as_str() == expected)
        })
    };
    NegotiatedPathCapabilities {
        opaque_host_paths: selected_has(C2_OPAQUE_UNIX_PATH_CAPABILITY),
        repository_paths: selected_has(C2_REPOSITORY_PATH_CAPABILITY),
        workspace_file_read: selected_has(C2_WORKSPACE_FILE_READ_CAPABILITY),
        workspace_file_write: selected_has(C2_WORKSPACE_FILE_WRITE_CAPABILITY),
        workspace_entry_create: selected_has(C2_WORKSPACE_ENTRY_CREATE_CAPABILITY),
        git_read: selected_has(C2_GIT_READ_CAPABILITY),
        host_directory_browse: selected_has(C2_HOST_DIRECTORY_BROWSE_CAPABILITY),
        standalone_workspace_lifecycle:
            selected_has(C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY),
        provider_session_reference_index:
            selected_has(C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY),
        provider_ids_open: selected_has(C2_PROVIDER_ID_OPEN_CAPABILITY),
        spawn_spec_defaults_overrides:
            selected_has(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY),
        spawn_profile_revision: selected_has(C2_SPAWN_PROFILE_REVISION_CAPABILITY),
        worktree_selection: selected_has(C2_WORKTREE_SELECTION_CAPABILITY),
        managed_worktree_lifecycle:
            selected_has(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY),
        managed_worktree_spawn_v2:
            selected_has(C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY),
        child_environment_profile:
            selected_has(C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY),
        session_bundle_materialization:
            selected_has(C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY),
        history_context_pack: selected_has(C2_HISTORY_CONTEXT_PACK_CAPABILITY),
        session_record_context_export:
            selected_has(C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY),
        native_session_catalog: selected_has(C2_NATIVE_SESSION_CATALOG_CAPABILITY),
        native_session_catalog_paging:
            selected_has(C2_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY),
        native_session_index: selected_has(C2_NATIVE_SESSION_INDEX_CAPABILITY),
        native_session_preview: selected_has(C2_NATIVE_SESSION_PREVIEW_CAPABILITY),
        terminal_frame_events: selected_has(C2_TERMINAL_FRAME_EVENTS_CAPABILITY),
        agent_stream_events: selected_has(C2_AGENT_STREAM_EVENTS_CAPABILITY),
        agent_progress_snapshot: selected_has(C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY),
        session_task_correlation: selected_has(C2_SESSION_TASK_CORRELATION_CAPABILITY),
        control_detail: selected_has(C2_CONTROL_DETAIL_CAPABILITY),
        delivery_bundle_v2_stage_commit:
            selected_has(C2_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY),
        harness_mcp_read_proxy: selected_has(C2_HARNESS_MCP_READ_PROXY_CAPABILITY),
        acp_control: selected_has(C2_ACP_CONTROL_CAPABILITY),
    }
}

/// Names the capability a refused request required. Every capability with a
/// dedicated arm here names itself by its own message; a capability that
/// falls to the final arm is one `NodeRequest::required_capability()`
/// returned but this build's `NegotiatedPathCapabilities` has no dedicated
/// field for yet -- it must refuse under ITS OWN name, never borrow a
/// sibling's. That exact misattribution once sent a live investigation into
/// the wrong crate: a refused `acp-control-v1` request was reported as a
/// refused workspace file read, because the fallback arm always named
/// `C2_WORKSPACE_FILE_READ_CAPABILITY` regardless of what was actually
/// missing.
fn unnegotiated_capability_refusal(capability: Option<&str>) -> String {
    match capability {
        Some(C2_ACP_CONTROL_CAPABILITY) => ACP_CONTROL_NOT_NEGOTIATED.to_owned(),
        Some(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY) => {
            SPAWN_SPEC_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY) => {
            MANAGED_WORKTREE_LIFECYCLE_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY) => {
            MANAGED_WORKTREE_SPAWN_V2_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_HISTORY_CONTEXT_PACK_CAPABILITY) => {
            HISTORY_CONTEXT_PACK_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY) => {
            SESSION_RECORD_CONTEXT_EXPORT_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_HOST_DIRECTORY_BROWSE_CAPABILITY) => {
            HOST_DIRECTORY_BROWSE_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY) => {
            STANDALONE_WORKSPACE_LIFECYCLE_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY) => {
            PROVIDER_SESSION_REFERENCE_INDEX_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_WORKSPACE_ENTRY_CREATE_CAPABILITY) => {
            WORKSPACE_ENTRY_CREATE_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_NATIVE_SESSION_INDEX_CAPABILITY) => {
            "native session index capability was not negotiated".to_owned()
        }
        Some(C2_SESSION_TASK_CORRELATION_CAPABILITY) => {
            SESSION_TASK_CORRELATION_NOT_NEGOTIATED.to_owned()
        }
        Some(C2_CONTROL_DETAIL_CAPABILITY) => CONTROL_DETAIL_NOT_NEGOTIATED.to_owned(),
        Some(C2_HARNESS_MCP_READ_PROXY_CAPABILITY) => {
            "harness MCP read proxy capability was not negotiated".to_owned()
        }
        Some(C2_WORKSPACE_FILE_READ_CAPABILITY) => {
            WORKSPACE_FILE_READ_NOT_NEGOTIATED.to_owned()
        }
        Some(unrecognized) => format!(
            "C2 control capability \"{unrecognized}\" is required but was not negotiated",
        ),
        // `required_capability_available` is only `false` when
        // `required_capability()` returned `Some(..)`; this arm is
        // unreachable in practice and carries no capability to name.
        None => "a required C2 control capability was not negotiated".to_owned(),
    }
}

fn reject_unnegotiated_outbound_path(
    request: &NodeRequest,
    capabilities: NegotiatedPathCapabilities,
) -> Result<(), C2ControlError> {
    let now_unix_ms = current_unix_ms()?;
    if !request.harness_mcp_contract_is_valid_at(now_unix_ms) {
        return Err(C2ControlError::Protocol(
            "invalid harness MCP proxy request".to_owned(),
        ));
    }
    if !request.history_context_pack_contract_is_valid() {
        return Err(C2ControlError::Protocol(
            INVALID_HISTORY_CONTEXT_PACK_REQUEST.to_owned(),
        ));
    }
    if !request.native_session_catalog_contract_is_valid() {
        return Err(C2ControlError::Protocol(
            "invalid native session catalog request".to_owned(),
        ));
    }
    if !request.native_session_preview_contract_is_valid() {
        return Err(C2ControlError::Protocol(
            "invalid native session preview request".to_owned(),
        ));
    }
    if matches!(request, NodeRequest::IndexNativeSession { selection, .. }
        if selection.route.scope
            != gate4agent_node_protocol::NativeSessionCatalogScope::Workspace)
    {
        return Err(C2ControlError::Protocol(
            "external native sessions must be registered as workspaces before indexing"
                .to_owned(),
        ));
    }
    if !capabilities.opaque_host_paths && node_request_has_unix_bytes(request) {
        return Err(C2ControlError::Protocol(
            OPAQUE_UNIX_PATH_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if !capabilities.provider_ids_open
        && (matches!(request, NodeRequest::Spawn { provider, .. } if !provider_id_is_legacy(provider))
            || matches!(request, NodeRequest::IndexProviderSession { provider, .. } if !provider_id_is_legacy(provider))
            || matches!(request, NodeRequest::CatalogNativeSessions { route, .. } if !provider_id_is_legacy(&route.provider))
            || matches!(request, NodeRequest::PageNativeSessions { route, .. } if !provider_id_is_legacy(&route.provider))
            || matches!(request, NodeRequest::PreviewNativeSession { selection, .. } if !provider_id_is_legacy(&selection.route.provider))
            || matches!(request, NodeRequest::IndexNativeSession { selection, .. } if !provider_id_is_legacy(&selection.route.provider))
            || spawn_spec_requires_open_provider_capability(request)
            || matches!(request, NodeRequest::ForgetContextPack { .. }))
    {
        return Err(C2ControlError::Protocol(
            OPEN_PROVIDER_ID_NOT_NEGOTIATED.to_owned(),
        ));
    }
    let required_capability_available = match request.required_capability() {
        None => true,
        Some(C2_WORKSPACE_FILE_READ_CAPABILITY) => capabilities.workspace_file_read,
        Some(C2_WORKSPACE_FILE_WRITE_CAPABILITY) => capabilities.workspace_file_write,
        Some(C2_WORKSPACE_ENTRY_CREATE_CAPABILITY) => capabilities.workspace_entry_create,
        Some(C2_GIT_READ_CAPABILITY) => capabilities.git_read,
        Some(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY) => {
            capabilities.spawn_spec_defaults_overrides
        }
        Some(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY) => {
            capabilities.managed_worktree_lifecycle
        }
        Some(C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY) => {
            capabilities.managed_worktree_spawn_v2
        }
        Some(C2_HISTORY_CONTEXT_PACK_CAPABILITY) => capabilities.history_context_pack,
        Some(C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY) => {
            capabilities.session_record_context_export
        }
        Some(C2_NATIVE_SESSION_CATALOG_CAPABILITY) => capabilities.native_session_catalog,
        Some(C2_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY) => {
            capabilities.native_session_catalog_paging
        }
        Some(C2_NATIVE_SESSION_INDEX_CAPABILITY) => capabilities.native_session_index,
        Some(C2_NATIVE_SESSION_PREVIEW_CAPABILITY) => capabilities.native_session_preview,
        Some(C2_HOST_DIRECTORY_BROWSE_CAPABILITY) => capabilities.host_directory_browse,
        Some(C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY) => {
            capabilities.standalone_workspace_lifecycle
        }
        Some(C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY) => {
            capabilities.provider_session_reference_index
        }
        Some(C2_SESSION_TASK_CORRELATION_CAPABILITY) => {
            capabilities.session_task_correlation
        }
        Some(C2_CONTROL_DETAIL_CAPABILITY) => capabilities.control_detail,
        Some(C2_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY) => {
            capabilities.delivery_bundle_v2_stage_commit
        }
        Some(C2_HARNESS_MCP_READ_PROXY_CAPABILITY) => capabilities.harness_mcp_read_proxy,
        Some(C2_ACP_CONTROL_CAPABILITY) => capabilities.acp_control,
        Some(_) => false,
    };
    if !required_capability_available {
        return Err(C2ControlError::Protocol(unnegotiated_capability_refusal(
            request.required_capability(),
        )));
    }
    if request.requires_spawn_spec_defaults_overrides_capability()
        && !capabilities.spawn_spec_defaults_overrides
    {
        return Err(C2ControlError::Protocol(
            SPAWN_SPEC_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if request.requires_spawn_profile_revision_capability()
        && !capabilities.spawn_profile_revision
    {
        return Err(C2ControlError::Protocol(
            SPAWN_PROFILE_REVISION_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if request.requires_worktree_selection_capability()
        && !capabilities.worktree_selection
    {
        return Err(C2ControlError::Protocol(
            WORKTREE_SELECTION_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if c2_request_requires_child_environment_profile_capability(request)
        && !capabilities.child_environment_profile
    {
        return Err(C2ControlError::Protocol(
            CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if request.requires_session_bundle_materialization_capability()
        && !capabilities.session_bundle_materialization
    {
        return Err(C2ControlError::Protocol(
            SESSION_BUNDLE_MATERIALIZATION_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if request.requires_history_context_pack_capability()
        && !capabilities.history_context_pack
    {
        return Err(C2ControlError::Protocol(
            HISTORY_CONTEXT_PACK_NOT_NEGOTIATED.to_owned(),
        ));
    }
    if !capabilities.repository_paths && node_request_has_unix_repository_path(request) {
        return Err(C2ControlError::Protocol(
            REPOSITORY_PATH_NOT_NEGOTIATED.to_owned(),
        ));
    }
    Ok(())
}

fn current_unix_ms() -> Result<u64, C2ControlError> {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| C2ControlError::Protocol("system clock precedes Unix epoch".to_owned()))?
        .as_millis()
        .try_into()
        .map_err(|_| C2ControlError::Protocol("system clock exceeds protocol range".to_owned()))
}

fn c2_request_requires_child_environment_profile_capability(request: &NodeRequest) -> bool {
    let spec = match request {
        NodeRequest::SpawnSpec { spec } => spec,
        NodeRequest::SpawnManagedWorktree { request } => &request.spawn_spec,
        NodeRequest::SpawnManagedWorktreeV2 { request } => &request.spawn_spec,
        _ => return false,
    };
    !matches!(
        &spec.overrides.environment_profile_id,
        gate4agent_node_protocol::SpawnOverride::Clear
    )
}

fn spawn_spec_requires_open_provider_capability(request: &NodeRequest) -> bool {
    let spec = match request {
        NodeRequest::SpawnSpec { spec } => spec,
        NodeRequest::SpawnManagedWorktree { request } => &request.spawn_spec,
        NodeRequest::SpawnManagedWorktreeV2 { request } => &request.spawn_spec,
        _ => return false,
    };
    match &spec.overrides.provider {
        gate4agent_node_protocol::SpawnOverride::Set { value } => {
            !provider_id_is_legacy(value)
        }
        gate4agent_node_protocol::SpawnOverride::Inherit
        | gate4agent_node_protocol::SpawnOverride::Clear => true,
    }
}

fn node_request_has_unix_repository_path(request: &NodeRequest) -> bool {
    match request {
        NodeRequest::ReadWorkspaceFile { path, .. }
        | NodeRequest::WriteWorkspaceFile { path, .. }
        | NodeRequest::CreateWorkspaceFile { path, .. }
        | NodeRequest::CreateWorkspaceDirectory { path, .. } => path.as_unix_bytes().is_some(),
        NodeRequest::ReadGitDiff { request, .. } => request
            .path
            .as_ref()
            .is_some_and(|path| path.as_unix_bytes().is_some()),
        NodeRequest::Snapshot
        | NodeRequest::Resync { .. }
        | NodeRequest::ArmHarnessMcpReservation { .. }
        | NodeRequest::SpawnSpecWithHarnessMcp { .. }
        | NodeRequest::ActivateHarnessMcpReservation { .. }
        | NodeRequest::AbortHarnessMcpReservation { .. }
        | NodeRequest::PutHarnessMcpReplyChunk { .. }
        | NodeRequest::RejectHarnessMcpCall { .. }
        | NodeRequest::BeginDeliveryStage { .. }
        | NodeRequest::PutDeliveryBlobChunk { .. }
        | NodeRequest::CommitDeliveryStage { .. }
        | NodeRequest::AbortDeliveryStage { .. }
        | NodeRequest::BrowseHostDirectories { .. }
        | NodeRequest::InspectWorkspace { .. }
        | NodeRequest::ReadGitHistory { .. }
        | NodeRequest::AcquireController { .. }
        | NodeRequest::ReleaseController
        | NodeRequest::RegisterWorkspace { .. }
        | NodeRequest::CreateStandaloneWorkspace { .. }
        | NodeRequest::UnregisterWorkspace { .. }
        | NodeRequest::CreateWorktree { .. }
        | NodeRequest::RemoveWorktree { .. }
        | NodeRequest::Spawn { .. }
        | NodeRequest::SpawnSpec { .. }
        | NodeRequest::SpawnManagedWorktree { .. }
        | NodeRequest::SpawnManagedWorktreeV2 { .. }
        | NodeRequest::CleanupManagedWorktree { .. }
        | NodeRequest::Resume { .. }
        | NodeRequest::RenameSessionRecord { .. }
        | NodeRequest::SetSessionTask { .. }
        | NodeRequest::IndexProviderSession { .. }
        | NodeRequest::IndexNativeSession { .. }
        | NodeRequest::ResumeSessionRecord { .. }
        | NodeRequest::ForgetSessionRecord { .. }
        | NodeRequest::CatalogNativeSessions { .. }
        | NodeRequest::PageNativeSessions { .. }
        | NodeRequest::PreviewNativeSession { .. }
        | NodeRequest::PreviewSessionRecord { .. }
        | NodeRequest::DiscoverHistory { .. }
        | NodeRequest::LoadHistory { .. }
        | NodeRequest::ExportContextPackForSessionRecord { .. }
        | NodeRequest::ExportContextPack { .. }
        | NodeRequest::ForgetContextPack { .. }
        | NodeRequest::ResolveDurableContextPack { .. }
        | NodeRequest::ReadContextPack { .. }
        | NodeRequest::Prompt { .. }
        | NodeRequest::Paste { .. }
        | NodeRequest::Input { .. }
        | NodeRequest::TerminalBytes { .. }
        | NodeRequest::TerminalControl { .. }
        | NodeRequest::Resize { .. }
        | NodeRequest::Interrupt { .. }
        | NodeRequest::Stop { .. }
        | NodeRequest::Remove { .. }
        | NodeRequest::ResolveInteraction { .. }
        | NodeRequest::SetSessionMode { .. }
        | NodeRequest::SetSessionConfigOption { .. }
        | NodeRequest::SetSessionModel { .. }
        | NodeRequest::Shutdown => false,
    }
}

fn node_request_has_unix_bytes(request: &NodeRequest) -> bool {
    match request {
        NodeRequest::RegisterWorkspace { root, .. }
        | NodeRequest::CreateStandaloneWorkspace { root, .. } => {
            root.as_unix_bytes().is_some()
        }
        NodeRequest::BrowseHostDirectories { directory, after } => {
            directory.as_ref().is_some_and(|path| path.as_unix_bytes().is_some())
                || after.as_ref().is_some_and(|path| path.as_unix_bytes().is_some())
        }
        NodeRequest::CreateWorktree { target_root, .. }
        | NodeRequest::RemoveWorktree { target_root, .. } => {
            target_root.as_unix_bytes().is_some()
        }
        NodeRequest::Snapshot
        | NodeRequest::Resync { .. }
        | NodeRequest::ArmHarnessMcpReservation { .. }
        | NodeRequest::SpawnSpecWithHarnessMcp { .. }
        | NodeRequest::ActivateHarnessMcpReservation { .. }
        | NodeRequest::AbortHarnessMcpReservation { .. }
        | NodeRequest::PutHarnessMcpReplyChunk { .. }
        | NodeRequest::RejectHarnessMcpCall { .. }
        | NodeRequest::InspectWorkspace { .. }
        | NodeRequest::ReadWorkspaceFile { .. }
        | NodeRequest::WriteWorkspaceFile { .. }
        | NodeRequest::CreateWorkspaceFile { .. }
        | NodeRequest::CreateWorkspaceDirectory { .. }
        | NodeRequest::ReadGitHistory { .. }
        | NodeRequest::ReadGitDiff { .. }
        | NodeRequest::BeginDeliveryStage { .. }
        | NodeRequest::PutDeliveryBlobChunk { .. }
        | NodeRequest::CommitDeliveryStage { .. }
        | NodeRequest::AbortDeliveryStage { .. }
        | NodeRequest::AcquireController { .. }
        | NodeRequest::ReleaseController
        | NodeRequest::UnregisterWorkspace { .. }
        | NodeRequest::Spawn { .. }
        | NodeRequest::SpawnSpec { .. }
        | NodeRequest::SpawnManagedWorktree { .. }
        | NodeRequest::SpawnManagedWorktreeV2 { .. }
        | NodeRequest::CleanupManagedWorktree { .. }
        | NodeRequest::Resume { .. }
        | NodeRequest::RenameSessionRecord { .. }
        | NodeRequest::SetSessionTask { .. }
        | NodeRequest::IndexProviderSession { .. }
        | NodeRequest::IndexNativeSession { .. }
        | NodeRequest::ResumeSessionRecord { .. }
        | NodeRequest::ForgetSessionRecord { .. }
        | NodeRequest::CatalogNativeSessions { .. }
        | NodeRequest::PageNativeSessions { .. }
        | NodeRequest::PreviewNativeSession { .. }
        | NodeRequest::PreviewSessionRecord { .. }
        | NodeRequest::DiscoverHistory { .. }
        | NodeRequest::LoadHistory { .. }
        | NodeRequest::ExportContextPackForSessionRecord { .. }
        | NodeRequest::ExportContextPack { .. }
        | NodeRequest::ForgetContextPack { .. }
        | NodeRequest::ResolveDurableContextPack { .. }
        | NodeRequest::ReadContextPack { .. }
        | NodeRequest::Prompt { .. }
        | NodeRequest::Paste { .. }
        | NodeRequest::Input { .. }
        | NodeRequest::TerminalBytes { .. }
        | NodeRequest::TerminalControl { .. }
        | NodeRequest::Resize { .. }
        | NodeRequest::Interrupt { .. }
        | NodeRequest::Stop { .. }
        | NodeRequest::Remove { .. }
        | NodeRequest::ResolveInteraction { .. }
        | NodeRequest::SetSessionMode { .. }
        | NodeRequest::SetSessionConfigOption { .. }
        | NodeRequest::SetSessionModel { .. }
        | NodeRequest::Shutdown => false,
    }
}

fn routed_response_has_unix_bytes(response: &RoutedNodeResponse) -> bool {
    response
        .response
        .as_ref()
        .is_ok_and(c2_node_response_has_unix_bytes)
}

fn routed_response_has_terminal_frame_event(response: &RoutedNodeResponse) -> bool {
    response
        .response
        .as_ref()
        .is_ok_and(c2_node_response_has_terminal_frame_event)
}

fn routed_response_has_agent_stream_event(response: &RoutedNodeResponse) -> bool {
    response
        .response
        .as_ref()
        .is_ok_and(c2_node_response_has_agent_stream_event)
}

fn routed_response_has_agent_progress(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| match response {
        C2NodeResponse::Snapshot { snapshot, .. }
        | C2NodeResponse::Resync { snapshot, .. } => !snapshot.agent_progress.is_empty(),
        _ => false,
    })
}

fn routed_response_requires_history_context_pack(response: &RoutedNodeResponse) -> bool {
    match &response.response {
        Ok(response) => response.requires_history_context_pack_capability(),
        Err(failure) => matches!(
            failure.code,
            gate4agent_node_protocol::NodeFailureCode::UnknownContextPack
                | gate4agent_node_protocol::NodeFailureCode::ContextPackBusy
                | gate4agent_node_protocol::NodeFailureCode::ContextPackMaterializationFailed
        ),
    }
}

fn routed_response_requires_native_session_catalog(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(
        C2NodeResponse::requires_native_session_catalog_capability,
    )
}

fn routed_response_requires_native_session_catalog_paging(response: &RoutedNodeResponse) -> bool {
    match &response.response {
        Ok(response) => response.requires_native_session_catalog_paging_capability(),
        Err(failure) => failure.requires_native_session_catalog_paging_capability(),
    }
}

fn routed_response_requires_native_session_preview(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(
        C2NodeResponse::requires_native_session_preview_capability,
    )
}

fn routed_response_requires_native_session_index(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(
        C2NodeResponse::requires_native_session_index_capability,
    )
}

fn routed_response_requires_harness_mcp_proxy(response: &RoutedNodeResponse) -> bool {
    match &response.response {
        Ok(response) => response.requires_harness_mcp_proxy_capability(),
        Err(failure) => failure.requires_harness_mcp_proxy_capability(),
    }
}

fn validate_harness_mcp_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else { return Ok(()); };
    use C2NodeResponse as Response;
    use NodeRequest as Request;
    let valid = match (expected, &routed.response) {
        (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,
        (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
                }),
        (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,
        (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,
        (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,
        (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,
        (request, Err(_)) if request.required_capability()
            == Some(C2_HARNESS_MCP_READ_PROXY_CAPABILITY) => true,
        (request, Ok(_)) if request.required_capability()
            == Some(C2_HARNESS_MCP_READ_PROXY_CAPABILITY) => false,
        (_, Ok(response)) if response.requires_harness_mcp_proxy_capability() => false,
        _ => true,
    };
    if valid { Ok(()) } else { Err("C2 harness MCP response does not match the routed request") }
}

fn validate_provider_session_index_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else {
        return Ok(());
    };
    match (expected, &routed.response) {
        (
            NodeRequest::IndexProviderSession {
                workspace_id,
                provider,
                ..
            },
            Ok(C2NodeResponse::ProviderSessionIndexed { record }),
        ) if &record.workspace_id == workspace_id && &record.provider == provider => Ok(()),
        (NodeRequest::IndexProviderSession { .. }, Err(_)) => Ok(()),
        (NodeRequest::IndexProviderSession { .. }, Ok(_)) => {
            Err("C2 provider session index response does not match the routed request")
        }
        (_, Ok(C2NodeResponse::ProviderSessionIndexed { .. })) => {
            Err("C2 returned an unexpected provider session index response")
        }
        _ => Ok(()),
    }
}

fn validate_session_record_context_export_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else {
        return Ok(());
    };
    match (expected, &routed.response) {
        (
            NodeRequest::ExportContextPackForSessionRecord { record_id, session },
            Ok(C2NodeResponse::ContextPackForSessionRecordExported {
                record_id: echoed_record_id,
                session: echoed_session,
                context,
            }),
        ) if record_id == echoed_record_id && session == echoed_session && context.is_valid() => {
            Ok(())
        }
        (NodeRequest::ExportContextPackForSessionRecord { .. }, Err(_)) => Ok(()),
        (NodeRequest::ExportContextPackForSessionRecord { .. }, Ok(_)) => Err(
            "C2 session-record context export response does not match the routed request",
        ),
        (_, Ok(C2NodeResponse::ContextPackForSessionRecordExported { .. })) => Err(
            "C2 returned an unexpected session-record context export response",
        ),
        _ => Ok(()),
    }
}

fn validate_native_session_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else {
        return Ok(());
    };
    match (expected, &routed.response) {
        (
            NodeRequest::CatalogNativeSessions { route, .. },
            Ok(response @ C2NodeResponse::NativeSessionsCataloged {
                route: echoed_route,
                ..
            }),
        ) if echoed_route == route && response.native_session_catalog_contract_is_valid() => {
            Ok(())
        }
        (
            NodeRequest::PageNativeSessions {
                route,
                window,
                catalog_revision,
                ..
            },
            Ok(response @ C2NodeResponse::NativeSessionsPaged {
                route: echoed_route,
                page,
            }),
        ) if echoed_route == route
            && page.window == *window
            && page.revision == *catalog_revision
            && response.native_session_catalog_contract_is_valid() => Ok(()),
        (
            NodeRequest::PreviewNativeSession { selection, .. },
            Ok(response @ C2NodeResponse::NativeSessionPreviewed {
                selection: echoed_selection,
                ..
            }),
        ) if echoed_selection == selection
            && response.native_session_preview_contract_is_valid() => Ok(()),
        (
            NodeRequest::IndexNativeSession { selection, .. },
            Ok(response @ C2NodeResponse::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
            && response.native_session_index_contract_is_valid() => Ok(()),
        (
            NodeRequest::CatalogNativeSessions { .. }
            | NodeRequest::PageNativeSessions { .. }
            | NodeRequest::PreviewNativeSession { .. }
            | NodeRequest::IndexNativeSession { .. },
            Err(_),
        ) => Ok(()),
        (
            NodeRequest::CatalogNativeSessions { .. }
            | NodeRequest::PageNativeSessions { .. }
            | NodeRequest::PreviewNativeSession { .. }
            | NodeRequest::IndexNativeSession { .. },
            Ok(_),
        ) => Err("C2 native session response does not match the routed request"),
        (
            _,
            Ok(
                C2NodeResponse::NativeSessionsCataloged { .. }
                | C2NodeResponse::NativeSessionsPaged { .. }
                | C2NodeResponse::NativeSessionPreviewed { .. }
                | C2NodeResponse::NativeSessionIndexed { .. },
            ),
        ) => Err("C2 returned an unexpected native session response"),
        _ => Ok(()),
    }
}

fn validate_workspace_content_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else {
        return Ok(());
    };
    match (expected, &routed.response) {
        (NodeRequest::ReadWorkspaceFile { workspace_id, path }, Ok(C2NodeResponse::WorkspaceFileRead { file }))
            if &file.workspace_id == workspace_id && &file.path == path => Ok(()),
        (NodeRequest::WriteWorkspaceFile { workspace_id, path, text, .. }, Ok(C2NodeResponse::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(()),
        (NodeRequest::CreateWorkspaceFile { workspace_id, path }, Ok(C2NodeResponse::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(()),
        (
            NodeRequest::CreateWorkspaceDirectory { workspace_id, path },
            Ok(C2NodeResponse::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(()),
        (NodeRequest::ReadGitHistory { workspace_id, .. }, Ok(C2NodeResponse::GitHistoryRead { workspace_id: actual, .. }))
            if actual == workspace_id => Ok(()),
        (NodeRequest::ReadGitDiff { workspace_id, request }, Ok(C2NodeResponse::GitDiffRead { workspace_id: actual, diff }))
            if actual == workspace_id && diff.mode == request.mode && diff.path == request.path => Ok(()),
        (NodeRequest::ReadWorkspaceFile { .. }
            | NodeRequest::WriteWorkspaceFile { .. }
            | NodeRequest::CreateWorkspaceFile { .. }
            | NodeRequest::CreateWorkspaceDirectory { .. }
            | NodeRequest::ReadGitHistory { .. }
            | NodeRequest::ReadGitDiff { .. }, Err(_)) => Ok(()),
        (NodeRequest::ReadWorkspaceFile { .. }
            | NodeRequest::WriteWorkspaceFile { .. }
            | NodeRequest::CreateWorkspaceFile { .. }
            | NodeRequest::CreateWorkspaceDirectory { .. }
            | NodeRequest::ReadGitHistory { .. }
            | NodeRequest::ReadGitDiff { .. }, Ok(_)) => {
                Err("C2 workspace content response does not match the routed request")
            }
        (_, Ok(C2NodeResponse::WorkspaceFileRead { .. }
            | C2NodeResponse::WorkspaceFileWritten { .. }
            | C2NodeResponse::WorkspaceFileCreated { .. }
            | C2NodeResponse::WorkspaceDirectoryCreated { .. }
            | C2NodeResponse::GitHistoryRead { .. }
            | C2NodeResponse::GitDiffRead { .. })) => {
                Err("C2 returned an unexpected workspace content response")
            }
        _ => Ok(()),
    }
}

fn validate_session_task_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else { return Ok(()); };
    match (expected, &routed.response) {
        (NodeRequest::SetSessionTask { .. }, Err(_)) => Ok(()),
        (
            NodeRequest::SetSessionTask {
                record_id,
                expected_revision,
                target,
            },
            Ok(C2NodeResponse::SessionRecordUpdated { record }),
        ) if c2_session_task_record_matches(record, record_id, *expected_revision, target) => Ok(()),
        (NodeRequest::SetSessionTask { .. }, Ok(_)) => {
            Err("C2 session task response does not match the routed request")
        }
        _ => Ok(()),
    }
}

fn validate_managed_worktree_spawn_v2_response(
    expected: &NodeRequest,
    response: &Result<RoutedNodeResponse, C2RelayFailure>,
) -> Result<(), &'static str> {
    let Ok(routed) = response else {
        return Ok(());
    };
    match (expected, &routed.response) {
        (
            NodeRequest::SpawnManagedWorktreeV2 { request },
            Ok(C2NodeResponse::ManagedWorktreeSpawnAccepted { receipt }),
        ) if receipt.spawn.incarnation_id == routed.incarnation_id
            && receipt.spawn.target.node_id == request.spawn_spec.target.node_id
            && receipt.spawn.target.workspace_id == request.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
            && receipt.spawn.profile_id == request.spawn_spec.profile_id
            && receipt.spawn.profile_revision == request.spawn_spec.expected_profile_revision
            && receipt.spawn.idempotency_key == request.spawn_spec.idempotency_key
            && receipt.lease.source_workspace_id == request.spawn_spec.target.workspace_id
            && receipt.lease.profile_id == request.worktree_profile_id
            && receipt.lease.profile_revision == request.expected_profile_revision => Ok(()),
        (NodeRequest::SpawnManagedWorktreeV2 { .. }, Err(_)) => Ok(()),
        (NodeRequest::SpawnManagedWorktreeV2 { .. }, Ok(_)) => {
            Err("C2 managed worktree V2 response does not match the routed request")
        }
        _ => Ok(()),
    }
}

fn c2_session_task_record_matches(
    record: &hatchery_c2_protocol::C2ManagedSessionRecord,
    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 c2_node_response_has_terminal_frame_event(response: &C2NodeResponse) -> bool {
    match response {
        C2NodeResponse::Resync { events, .. } => events
            .iter()
            .any(|event| c2_node_event_is_terminal_frame(&event.event)),
        C2NodeResponse::Snapshot { .. }
        | C2NodeResponse::Armed { .. }
        | C2NodeResponse::Spawned { .. }
        | C2NodeResponse::Activated { .. }
        | C2NodeResponse::Aborted { .. }
        | C2NodeResponse::ReplyChunkAccepted { .. }
        | C2NodeResponse::CallRejected { .. }
        | C2NodeResponse::DeliveryStageBegun { .. }
        | C2NodeResponse::DeliveryBlobChunkAccepted { .. }
        | C2NodeResponse::DeliveryCommitted { .. }
        | C2NodeResponse::DeliveryStageAborted { .. }
        | C2NodeResponse::WorkspaceInspected { .. }
        | C2NodeResponse::HostDirectoriesBrowsed { .. }
        | C2NodeResponse::WorkspaceFileRead { .. }
        | C2NodeResponse::WorkspaceFileWritten { .. }
        | C2NodeResponse::WorkspaceFileCreated { .. }
        | C2NodeResponse::WorkspaceDirectoryCreated { .. }
        | C2NodeResponse::GitHistoryRead { .. }
        | C2NodeResponse::GitDiffRead { .. }
        | C2NodeResponse::Controller { .. }
        | C2NodeResponse::SpawnAccepted { .. }
        | C2NodeResponse::SpawnSpecAccepted { .. }
        | C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        | C2NodeResponse::ManagedWorktreeCleanup { .. }
        | C2NodeResponse::SessionRecordUpdated { .. }
        | C2NodeResponse::ProviderSessionIndexed { .. }
        | C2NodeResponse::NativeSessionIndexed { .. }
        | C2NodeResponse::SessionRecordResumed { .. }
        | C2NodeResponse::SessionRecordForgotten { .. }
        | C2NodeResponse::NativeSessionsCataloged { .. }
        | C2NodeResponse::NativeSessionsPaged { .. }
        | C2NodeResponse::NativeSessionPreviewed { .. }
        | C2NodeResponse::SessionRecordPreviewed { .. }
        | C2NodeResponse::HistoryDiscovered { .. }
        | C2NodeResponse::HistoryLoaded { .. }
        | C2NodeResponse::ContextPackForSessionRecordExported { .. }
        | C2NodeResponse::ContextPackExported { .. }
        | C2NodeResponse::ContextPackForgotten { .. }
        | C2NodeResponse::DurableContextPackResolved { .. }
        | C2NodeResponse::ContextPackBytesRead { .. }
        | C2NodeResponse::WorkspaceRegistered { .. }
        | C2NodeResponse::StandaloneWorkspaceCreated { .. }
        | C2NodeResponse::WorkspaceUnregistered { .. }
        | C2NodeResponse::WorktreeCreated { .. }
        | C2NodeResponse::WorktreeRemoved { .. }
        | C2NodeResponse::Accepted
        | C2NodeResponse::ShuttingDown => false,
    }
}

fn c2_node_event_is_terminal_frame(event: &C2NodeEvent) -> bool {
    match event {
        C2NodeEvent::TerminalFrame { .. } => true,
        C2NodeEvent::HarnessMcpReadCall { .. }
        | C2NodeEvent::Control { .. }
        | C2NodeEvent::SessionRecordHistorySummarized { .. }
        | C2NodeEvent::ControllerChanged { .. }
        | C2NodeEvent::WorkspaceAdded { .. }
        | C2NodeEvent::WorkspaceRemoved { .. }
        | C2NodeEvent::SessionRecordUpserted { .. }
        | C2NodeEvent::SessionRecordRemoved { .. }
        | C2NodeEvent::ManagedWorktreeUpserted { .. }
        | C2NodeEvent::ManagedWorktreeRemoved { .. }
        | C2NodeEvent::AgentStream { .. }
        | C2NodeEvent::ResyncRequired { .. } => false,
    }
}

fn c2_node_response_has_agent_stream_event(response: &C2NodeResponse) -> bool {
    match response {
        C2NodeResponse::Resync { events, .. } => events
            .iter()
            .any(|event| c2_node_event_is_agent_stream(&event.event)),
        C2NodeResponse::Snapshot { .. }
        | C2NodeResponse::Armed { .. }
        | C2NodeResponse::Spawned { .. }
        | C2NodeResponse::Activated { .. }
        | C2NodeResponse::Aborted { .. }
        | C2NodeResponse::ReplyChunkAccepted { .. }
        | C2NodeResponse::CallRejected { .. }
        | C2NodeResponse::DeliveryStageBegun { .. }
        | C2NodeResponse::DeliveryBlobChunkAccepted { .. }
        | C2NodeResponse::DeliveryCommitted { .. }
        | C2NodeResponse::DeliveryStageAborted { .. }
        | C2NodeResponse::WorkspaceInspected { .. }
        | C2NodeResponse::HostDirectoriesBrowsed { .. }
        | C2NodeResponse::WorkspaceFileRead { .. }
        | C2NodeResponse::WorkspaceFileWritten { .. }
        | C2NodeResponse::WorkspaceFileCreated { .. }
        | C2NodeResponse::WorkspaceDirectoryCreated { .. }
        | C2NodeResponse::GitHistoryRead { .. }
        | C2NodeResponse::GitDiffRead { .. }
        | C2NodeResponse::Controller { .. }
        | C2NodeResponse::SpawnAccepted { .. }
        | C2NodeResponse::SpawnSpecAccepted { .. }
        | C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        | C2NodeResponse::ManagedWorktreeCleanup { .. }
        | C2NodeResponse::SessionRecordUpdated { .. }
        | C2NodeResponse::ProviderSessionIndexed { .. }
        | C2NodeResponse::NativeSessionIndexed { .. }
        | C2NodeResponse::SessionRecordResumed { .. }
        | C2NodeResponse::SessionRecordForgotten { .. }
        | C2NodeResponse::NativeSessionsCataloged { .. }
        | C2NodeResponse::NativeSessionsPaged { .. }
        | C2NodeResponse::NativeSessionPreviewed { .. }
        | C2NodeResponse::SessionRecordPreviewed { .. }
        | C2NodeResponse::HistoryDiscovered { .. }
        | C2NodeResponse::HistoryLoaded { .. }
        | C2NodeResponse::ContextPackForSessionRecordExported { .. }
        | C2NodeResponse::ContextPackExported { .. }
        | C2NodeResponse::ContextPackForgotten { .. }
        | C2NodeResponse::DurableContextPackResolved { .. }
        | C2NodeResponse::ContextPackBytesRead { .. }
        | C2NodeResponse::WorkspaceRegistered { .. }
        | C2NodeResponse::StandaloneWorkspaceCreated { .. }
        | C2NodeResponse::WorkspaceUnregistered { .. }
        | C2NodeResponse::WorktreeCreated { .. }
        | C2NodeResponse::WorktreeRemoved { .. }
        | C2NodeResponse::Accepted
        | C2NodeResponse::ShuttingDown => false,
    }
}

fn c2_node_event_is_agent_stream(event: &C2NodeEvent) -> bool {
    match event {
        C2NodeEvent::AgentStream { .. } => true,
        C2NodeEvent::HarnessMcpReadCall { .. }
        | C2NodeEvent::Control { .. }
        | C2NodeEvent::SessionRecordHistorySummarized { .. }
        | C2NodeEvent::ControllerChanged { .. }
        | C2NodeEvent::WorkspaceAdded { .. }
        | C2NodeEvent::WorkspaceRemoved { .. }
        | C2NodeEvent::SessionRecordUpserted { .. }
        | C2NodeEvent::SessionRecordRemoved { .. }
        | C2NodeEvent::ManagedWorktreeUpserted { .. }
        | C2NodeEvent::ManagedWorktreeRemoved { .. }
        | C2NodeEvent::TerminalFrame { .. }
        | C2NodeEvent::ResyncRequired { .. } => false,
    }
}

/// Coarse label for a dropped/tearing-down `C2NodeEvent` -- used only by
/// `control_owner`'s event-delivery warns, mirroring `hatchery-harness-
/// service`'s own `agent_stream_event_kind_label` (same "name every drop"
/// reasoning): a log line that fires once per backpressure episode still
/// needs to say WHAT kind of event was in flight when it fired.
fn c2_node_event_kind_label(event: &C2NodeEvent) -> &'static str {
    match event {
        C2NodeEvent::HarnessMcpReadCall { .. } => "harness-mcp-read-call",
        C2NodeEvent::Control { .. } => "control",
        C2NodeEvent::SessionRecordHistorySummarized { .. } => "session-record-history-summarized",
        C2NodeEvent::TerminalFrame { .. } => "terminal-frame",
        C2NodeEvent::AgentStream { .. } => "agent-stream",
        C2NodeEvent::ControllerChanged { .. } => "controller-changed",
        C2NodeEvent::WorkspaceAdded { .. } => "workspace-added",
        C2NodeEvent::WorkspaceRemoved { .. } => "workspace-removed",
        C2NodeEvent::SessionRecordUpserted { .. } => "session-record-upserted",
        C2NodeEvent::SessionRecordRemoved { .. } => "session-record-removed",
        C2NodeEvent::ManagedWorktreeUpserted { .. } => "managed-worktree-upserted",
        C2NodeEvent::ManagedWorktreeRemoved { .. } => "managed-worktree-removed",
        C2NodeEvent::ResyncRequired { .. } => "resync-required",
    }
}

/// Whether THIS full-channel drop should be logged: `true` exactly once per
/// contiguous backlog episode (the first drop since `logged` was last
/// reset), flipping `*logged` so every later drop in the same episode
/// returns `false` -- see `reset_channel_full_drop_episode` for the
/// counterpart that ends an episode. Split out as a pure function, used by
/// both `control_owner`'s `harness_mcp` and terminal-frame delivery arms, so
/// the "once per episode, not per event" contract is directly testable
/// without needing a tracing subscriber to observe log call counts.
fn should_log_channel_full_drop(logged: &mut bool) -> bool {
    if *logged {
        return false;
    }
    *logged = true;
    true
}

/// Ends a full-channel drop episode: a send that succeeds again means the
/// backlog cleared, so the next drop (if any) is a fresh episode and earns
/// its own log line.
fn reset_channel_full_drop_episode(logged: &mut bool) {
    *logged = false;
}

fn c2_node_response_has_unix_bytes(response: &C2NodeResponse) -> bool {
    match response {
        C2NodeResponse::Snapshot { snapshot, .. } => node_snapshot_has_unix_bytes(snapshot),
        C2NodeResponse::Resync { snapshot, events, .. } => {
            node_snapshot_has_unix_bytes(snapshot)
                || events
                    .iter()
                    .any(|envelope| c2_node_event_has_unix_bytes(&envelope.event))
        }
        C2NodeResponse::WorkspaceInspected { inspection } => inspection
            .git
            .worktrees
            .iter()
            .any(|worktree| worktree.path.as_unix_bytes().is_some()),
        C2NodeResponse::HostDirectoriesBrowsed { listing } => {
            listing.directory.as_ref().is_some_and(|path| path.as_unix_bytes().is_some())
                || listing.parent.as_ref().is_some_and(|path| path.as_unix_bytes().is_some())
                || listing.entries.iter().any(|entry| entry.path.as_unix_bytes().is_some())
                || listing.next_after.as_ref().is_some_and(|path| path.as_unix_bytes().is_some())
        }
        C2NodeResponse::WorkspaceFileRead { .. }
        | C2NodeResponse::WorkspaceFileWritten { .. }
        | C2NodeResponse::WorkspaceFileCreated { .. }
        | C2NodeResponse::WorkspaceDirectoryCreated { .. }
        | C2NodeResponse::GitHistoryRead { .. }
        | C2NodeResponse::GitDiffRead { .. }
        | C2NodeResponse::DeliveryStageBegun { .. }
        | C2NodeResponse::DeliveryBlobChunkAccepted { .. }
        | C2NodeResponse::DeliveryCommitted { .. }
        | C2NodeResponse::DeliveryStageAborted { .. } => false,
        C2NodeResponse::WorkspaceRegistered { workspace }
        | C2NodeResponse::StandaloneWorkspaceCreated { workspace } => {
            workspace.canonical_root.as_unix_bytes().is_some()
        }
        C2NodeResponse::WorktreeCreated { worktree, workspace } => {
            worktree.path.as_unix_bytes().is_some()
                || workspace.canonical_root.as_unix_bytes().is_some()
        }
        C2NodeResponse::WorktreeRemoved { target_root, .. } => {
            target_root.as_unix_bytes().is_some()
        }
        C2NodeResponse::Armed { .. }
        | C2NodeResponse::Spawned { .. }
        | C2NodeResponse::Activated { .. }
        | C2NodeResponse::Aborted { .. }
        | C2NodeResponse::ReplyChunkAccepted { .. }
        | C2NodeResponse::CallRejected { .. }
        | C2NodeResponse::Controller { .. }
        | C2NodeResponse::SpawnAccepted { .. }
        | C2NodeResponse::SpawnSpecAccepted { .. }
        | C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        | C2NodeResponse::ManagedWorktreeCleanup { .. }
        | C2NodeResponse::SessionRecordUpdated { .. }
        | C2NodeResponse::ProviderSessionIndexed { .. }
        | C2NodeResponse::NativeSessionIndexed { .. }
        | C2NodeResponse::SessionRecordResumed { .. }
        | C2NodeResponse::SessionRecordForgotten { .. }
        | C2NodeResponse::NativeSessionsCataloged { .. }
        | C2NodeResponse::NativeSessionsPaged { .. }
        | C2NodeResponse::NativeSessionPreviewed { .. }
        | C2NodeResponse::SessionRecordPreviewed { .. }
        | C2NodeResponse::HistoryDiscovered { .. }
        | C2NodeResponse::HistoryLoaded { .. }
        | C2NodeResponse::ContextPackForSessionRecordExported { .. }
        | C2NodeResponse::ContextPackExported { .. }
        | C2NodeResponse::ContextPackForgotten { .. }
        | C2NodeResponse::DurableContextPackResolved { .. }
        | C2NodeResponse::ContextPackBytesRead { .. }
        | C2NodeResponse::WorkspaceUnregistered { .. }
        | C2NodeResponse::Accepted
        | C2NodeResponse::ShuttingDown => false,
    }
}

fn routed_response_has_unix_repository_path(response: &RoutedNodeResponse) -> bool {
    response
        .response
        .as_ref()
        .is_ok_and(c2_node_response_has_unix_repository_path)
}

fn c2_node_response_has_unix_repository_path(response: &C2NodeResponse) -> bool {
    match response {
        C2NodeResponse::WorkspaceInspected { inspection } => {
            inspection.entries.iter().any(|entry| {
                entry.relative_path.as_unix_bytes().is_some()
            }) || inspection.git.status.iter().any(|entry| {
                entry.path.as_unix_bytes().is_some()
                    || entry.previous_path.as_ref().is_some_and(|path| {
                        path.as_unix_bytes().is_some()
                    })
            })
        }
        C2NodeResponse::WorkspaceFileRead { file }
        | C2NodeResponse::WorkspaceFileWritten { file }
        | C2NodeResponse::WorkspaceFileCreated { file } => file.path.as_unix_bytes().is_some(),
        C2NodeResponse::WorkspaceDirectoryCreated { entry, .. } => {
            entry.relative_path.as_unix_bytes().is_some()
        }
        C2NodeResponse::GitDiffRead { diff, .. } => diff
            .path
            .as_ref()
            .is_some_and(|path| path.as_unix_bytes().is_some()),
        C2NodeResponse::DurableContextPackResolved { .. }
        | C2NodeResponse::ContextPackBytesRead { .. } => false,
        C2NodeResponse::Snapshot { .. }
        | C2NodeResponse::Resync { .. }
        | C2NodeResponse::Armed { .. }
        | C2NodeResponse::Spawned { .. }
        | C2NodeResponse::Activated { .. }
        | C2NodeResponse::Aborted { .. }
        | C2NodeResponse::ReplyChunkAccepted { .. }
        | C2NodeResponse::CallRejected { .. }
        | C2NodeResponse::DeliveryStageBegun { .. }
        | C2NodeResponse::DeliveryBlobChunkAccepted { .. }
        | C2NodeResponse::DeliveryCommitted { .. }
        | C2NodeResponse::DeliveryStageAborted { .. }
        | C2NodeResponse::HostDirectoriesBrowsed { .. }
        | C2NodeResponse::GitHistoryRead { .. }
        | C2NodeResponse::Controller { .. }
        | C2NodeResponse::SpawnAccepted { .. }
        | C2NodeResponse::SpawnSpecAccepted { .. }
        | C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        | C2NodeResponse::ManagedWorktreeCleanup { .. }
        | C2NodeResponse::SessionRecordUpdated { .. }
        | C2NodeResponse::ProviderSessionIndexed { .. }
        | C2NodeResponse::NativeSessionIndexed { .. }
        | C2NodeResponse::SessionRecordResumed { .. }
        | C2NodeResponse::SessionRecordForgotten { .. }
        | C2NodeResponse::NativeSessionsCataloged { .. }
        | C2NodeResponse::NativeSessionsPaged { .. }
        | C2NodeResponse::NativeSessionPreviewed { .. }
        | C2NodeResponse::SessionRecordPreviewed { .. }
        | C2NodeResponse::HistoryDiscovered { .. }
        | C2NodeResponse::HistoryLoaded { .. }
        | C2NodeResponse::ContextPackForSessionRecordExported { .. }
        | C2NodeResponse::ContextPackExported { .. }
        | C2NodeResponse::ContextPackForgotten { .. }
        | C2NodeResponse::WorkspaceRegistered { .. }
        | C2NodeResponse::StandaloneWorkspaceCreated { .. }
        | C2NodeResponse::WorkspaceUnregistered { .. }
        | C2NodeResponse::WorktreeCreated { .. }
        | C2NodeResponse::WorktreeRemoved { .. }
        | C2NodeResponse::Accepted
        | C2NodeResponse::ShuttingDown => false,
    }
}

fn routed_response_requires_workspace_file_read(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        matches!(response, C2NodeResponse::WorkspaceFileRead { .. })
    })
}

fn routed_response_requires_workspace_file_write(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        matches!(response, C2NodeResponse::WorkspaceFileWritten { .. })
    })
}

fn routed_response_requires_workspace_entry_create(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        response.requires_workspace_entry_create_capability()
    })
}

fn routed_response_requires_git_read(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        matches!(response, C2NodeResponse::GitHistoryRead { .. } | C2NodeResponse::GitDiffRead { .. })
    })
}

fn routed_response_requires_host_directory_browse(response: &RoutedNodeResponse) -> bool {
    match &response.response {
        Ok(response) => response.requires_host_directory_browse_capability(),
        Err(failure) => failure.requires_host_directory_browse_capability(),
    }
}

fn routed_response_requires_standalone_workspace_lifecycle(
    response: &RoutedNodeResponse,
) -> bool {
    matches!(
        &response.response,
        Ok(C2NodeResponse::StandaloneWorkspaceCreated { .. })
    )
}

fn routed_response_requires_provider_session_reference_index(
    response: &RoutedNodeResponse,
) -> bool {
    matches!(
        &response.response,
        Ok(C2NodeResponse::ProviderSessionIndexed { .. })
    )
}

fn routed_response_requires_spawn_spec(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        matches!(response,
            C2NodeResponse::SpawnSpecAccepted { .. }
            | C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        )
    })
}

fn routed_response_requires_spawn_profile_revision(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        matches!(response,
            C2NodeResponse::SpawnSpecAccepted { .. }
            | C2NodeResponse::Spawned { .. }
            | C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        )
    })
}

fn routed_response_requires_worktree_selection(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(|response| {
        matches!(response, C2NodeResponse::SpawnSpecAccepted { receipt }
            if receipt.target.worktree_id.is_some())
            || c2_node_response_has_managed_worktree(response)
    })
}

fn routed_response_requires_managed_worktree(response: &RoutedNodeResponse) -> bool {
    response
        .response
        .as_ref()
        .is_ok_and(c2_node_response_has_managed_worktree)
}

fn c2_node_response_has_managed_worktree(response: &C2NodeResponse) -> bool {
    match response {
        C2NodeResponse::Snapshot { snapshot, .. } => c2_snapshot_has_managed_worktree(snapshot),
        C2NodeResponse::Resync { snapshot, events, .. } => {
            c2_snapshot_has_managed_worktree(snapshot)
                || events
                    .iter()
                    .any(|event| c2_node_event_is_managed_worktree(&event.event))
        }
        C2NodeResponse::WorkspaceRegistered { workspace }
        | C2NodeResponse::StandaloneWorkspaceCreated { workspace }
        | C2NodeResponse::WorktreeCreated { workspace, .. } => {
            c2_workspace_has_managed_worktree_metadata(workspace)
        }
        C2NodeResponse::ManagedWorktreeSpawnAccepted { .. }
        | C2NodeResponse::ManagedWorktreeCleanup { .. } => true,
        _ => false,
    }
}

fn c2_node_event_is_managed_worktree(event: &C2NodeEvent) -> bool {
    match event {
        C2NodeEvent::WorkspaceAdded { workspace } => {
            c2_workspace_has_managed_worktree_metadata(workspace)
        }
        C2NodeEvent::ManagedWorktreeUpserted { .. }
        | C2NodeEvent::ManagedWorktreeRemoved { .. } => true,
        _ => false,
    }
}

fn c2_workspace_has_managed_worktree_metadata(
    workspace: &hatchery_c2_protocol::C2WorkspaceSnapshot,
) -> bool {
    workspace.worktree_service_mode.is_some()
        || workspace.managed_worktree_profiles.is_some()
}

fn c2_snapshot_has_managed_worktree(
    snapshot: &hatchery_c2_protocol::C2NodeSnapshot,
) -> bool {
    !snapshot.managed_worktrees.is_empty()
        || snapshot
            .workspaces
            .iter()
            .any(c2_workspace_has_managed_worktree_metadata)
}

fn node_snapshot_has_unix_bytes(snapshot: &hatchery_c2_protocol::C2NodeSnapshot) -> bool {
    snapshot
        .workspaces
        .iter()
        .any(|workspace| workspace.canonical_root.as_unix_bytes().is_some())
}

fn routed_event_has_unix_bytes(event: &RoutedNodeEvent) -> bool {
    c2_node_event_has_unix_bytes(&event.event)
}

fn c2_node_event_has_unix_bytes(event: &C2NodeEvent) -> bool {
    match event {
        C2NodeEvent::WorkspaceAdded { workspace } => {
            workspace.canonical_root.as_unix_bytes().is_some()
        }
        C2NodeEvent::HarnessMcpReadCall { .. }
        | C2NodeEvent::Control { .. }
        | C2NodeEvent::SessionRecordHistorySummarized { .. }
        | C2NodeEvent::TerminalFrame { .. }
        | C2NodeEvent::ControllerChanged { .. }
        | C2NodeEvent::WorkspaceRemoved { .. }
        | C2NodeEvent::SessionRecordUpserted { .. }
        | C2NodeEvent::SessionRecordRemoved { .. }
        | C2NodeEvent::ManagedWorktreeUpserted { .. }
        | C2NodeEvent::ManagedWorktreeRemoved { .. }
        | C2NodeEvent::AgentStream { .. }
        | C2NodeEvent::ResyncRequired { .. } => false,
    }
}

fn provider_text_is_legacy(value: &str) -> bool {
    hatchery_c2_protocol::AgentId::new(value)
        .is_ok_and(|provider| provider_id_is_legacy(&provider))
}

fn status_has_open_provider_id(status: &hatchery_c2_protocol::StatusResponse) -> bool {
    status.nodes.values().filter_map(|node| node.inventory.as_ref()).any(|inventory| {
        inventory.enabled_providers.iter().any(|provider| !provider_id_is_legacy(provider))
            || inventory.provider_runtime_statuses.iter()
                .any(|runtime| !provider_id_is_legacy(runtime.provider()))
            || inventory.provider_contracts.iter()
                .any(|contract| !provider_id_is_legacy(&contract.provider))
            || inventory.provider_adapter_contracts.iter()
                .any(|contract| !provider_id_is_legacy(&contract.provider))
            || inventory.workspaces.values().any(|workspace| {
                workspace.sessions.iter()
                    .any(|session| !provider_text_is_legacy(&session.agent_id))
            })
            || inventory.managed_sessions.iter()
                .any(|record| !provider_id_is_legacy(&record.provider))
    })
}

fn status_has_managed_worktree(status: &hatchery_c2_protocol::StatusResponse) -> bool {
    status.nodes.values().any(|node| {
        node.inventory
            .as_ref()
            .is_some_and(|inventory| {
                !inventory.managed_worktrees.is_empty()
                    || inventory.workspaces.values().any(|workspace| {
                        workspace.worktree_service_mode.is_some()
                            || workspace.managed_worktree_profiles.is_some()
                    })
            })
    })
}

fn status_has_child_environment_profile(
    status: &hatchery_c2_protocol::StatusResponse,
) -> bool {
    status.nodes.values().any(|node| {
        node.inventory.as_ref().is_some_and(|inventory| {
            inventory.managed_sessions.iter().any(|record| {
                record.environment_profile.is_some()
            })
        })
    })
}

fn status_has_session_bundle_materialization(
    status: &hatchery_c2_protocol::StatusResponse,
) -> bool {
    status.nodes.values().any(|node| {
        node.inventory.as_ref().is_some_and(|inventory| {
            inventory.managed_sessions.iter().any(|record| record.bundle.is_some())
        })
    })
}

fn topology_has_open_provider_id(topology: &C2Topology) -> bool {
    topology.nodes.iter().any(|node| {
        node.provider_contracts.iter()
            .any(|contract| !provider_id_is_legacy(&contract.provider))
            || node.provider_adapter_contracts.iter()
                .any(|contract| !provider_id_is_legacy(&contract.provider))
            || node.provider_runtime_statuses.iter()
                .any(|runtime| !provider_id_is_legacy(runtime.provider()))
    })
}

fn workspace_has_open_provider_id(
    workspace: &hatchery_c2_protocol::C2WorkspaceSnapshot,
) -> bool {
    workspace.sessions.iter()
        .any(|session| !provider_id_is_legacy(&session.agent_id))
}

fn snapshot_has_open_provider_id(
    snapshot: &hatchery_c2_protocol::C2NodeSnapshot,
) -> bool {
    snapshot.enabled_providers.iter().any(|provider| !provider_id_is_legacy(provider))
        || snapshot.provider_runtime_statuses.iter()
            .any(|runtime| !provider_id_is_legacy(runtime.provider()))
        || snapshot.workspaces.iter().any(workspace_has_open_provider_id)
        || snapshot.session_records.iter()
            .any(c2_managed_record_has_open_provider_id)
}

fn context_receipt_has_open_provider_id(
    context: &gate4agent_node_protocol::ResolvedContextPackReceipt,
) -> bool {
    !provider_id_is_legacy(&context.lineage.source_provider)
}

fn c2_managed_record_has_open_provider_id(
    record: &hatchery_c2_protocol::C2ManagedSessionRecord,
) -> bool {
    !provider_id_is_legacy(&record.provider)
        || record
            .context
            .as_ref()
            .is_some_and(context_receipt_has_open_provider_id)
}

fn c2_event_has_open_provider_id(event: &C2NodeEvent) -> bool {
    match event {
        C2NodeEvent::WorkspaceAdded { workspace } => workspace_has_open_provider_id(workspace),
        C2NodeEvent::SessionRecordUpserted { record } => {
            c2_managed_record_has_open_provider_id(record)
        }
        C2NodeEvent::HarnessMcpReadCall { .. }
        | C2NodeEvent::Control { .. }
        | C2NodeEvent::SessionRecordHistorySummarized { .. }
        | C2NodeEvent::TerminalFrame { .. }
        | C2NodeEvent::ControllerChanged { .. }
        | C2NodeEvent::WorkspaceRemoved { .. }
        | C2NodeEvent::SessionRecordRemoved { .. }
        | C2NodeEvent::ManagedWorktreeUpserted { .. }
        | C2NodeEvent::ManagedWorktreeRemoved { .. }
        | C2NodeEvent::AgentStream { .. }
        | C2NodeEvent::ResyncRequired { .. } => false,
    }
}

fn routed_event_has_open_provider_id(event: &RoutedNodeEvent) -> bool {
    c2_event_has_open_provider_id(&event.event)
}

fn c2_node_response_has_open_provider_id(response: &C2NodeResponse) -> bool {
    match response {
        C2NodeResponse::Snapshot { snapshot, .. } => snapshot_has_open_provider_id(snapshot),
        C2NodeResponse::Resync { snapshot, events, .. } => {
            snapshot_has_open_provider_id(snapshot)
                || events.iter().any(|event| c2_event_has_open_provider_id(&event.event))
        }
        C2NodeResponse::SessionRecordUpdated { record }
        | C2NodeResponse::ProviderSessionIndexed { record }
        | C2NodeResponse::NativeSessionIndexed { record, .. }
        | C2NodeResponse::SessionRecordResumed { record, .. } => {
            c2_managed_record_has_open_provider_id(record)
        }
        C2NodeResponse::SpawnSpecAccepted { receipt }
        | C2NodeResponse::Spawned { receipt, .. } => {
            !provider_id_is_legacy(&receipt.provider)
                || receipt
                    .context
                    .as_ref()
                    .is_some_and(context_receipt_has_open_provider_id)
        }
        C2NodeResponse::ManagedWorktreeSpawnAccepted { receipt } => {
            !provider_id_is_legacy(&receipt.spawn.provider)
                || receipt
                    .spawn
                    .context
                    .as_ref()
                    .is_some_and(context_receipt_has_open_provider_id)
        }
        C2NodeResponse::ContextPackExported { context }
        | C2NodeResponse::ContextPackForSessionRecordExported { context, .. }
        | C2NodeResponse::DurableContextPackResolved { context } => {
            context_receipt_has_open_provider_id(context)
        }
        C2NodeResponse::NativeSessionsCataloged { route, .. }
        | C2NodeResponse::NativeSessionsPaged { route, .. } => {
            !provider_id_is_legacy(&route.provider)
        }
        C2NodeResponse::NativeSessionPreviewed { selection, .. } => {
            !provider_id_is_legacy(&selection.route.provider)
        }
        C2NodeResponse::WorkspaceRegistered { workspace }
        | C2NodeResponse::StandaloneWorkspaceCreated { workspace }
        | C2NodeResponse::WorktreeCreated { workspace, .. } => {
            workspace_has_open_provider_id(workspace)
        }
        C2NodeResponse::Armed { .. }
        | C2NodeResponse::Activated { .. }
        | C2NodeResponse::Aborted { .. }
        | C2NodeResponse::ReplyChunkAccepted { .. }
        | C2NodeResponse::CallRejected { .. }
        | C2NodeResponse::WorkspaceInspected { .. }
        | C2NodeResponse::DeliveryStageBegun { .. }
        | C2NodeResponse::DeliveryBlobChunkAccepted { .. }
        | C2NodeResponse::DeliveryCommitted { .. }
        | C2NodeResponse::DeliveryStageAborted { .. }
        | C2NodeResponse::HostDirectoriesBrowsed { .. }
        | C2NodeResponse::WorkspaceFileRead { .. }
        | C2NodeResponse::WorkspaceFileWritten { .. }
        | C2NodeResponse::WorkspaceFileCreated { .. }
        | C2NodeResponse::WorkspaceDirectoryCreated { .. }
        | C2NodeResponse::GitHistoryRead { .. }
        | C2NodeResponse::GitDiffRead { .. }
        | C2NodeResponse::Controller { .. }
        | C2NodeResponse::SpawnAccepted { .. }
        | C2NodeResponse::ManagedWorktreeCleanup { .. }
        | C2NodeResponse::SessionRecordForgotten { .. }
        | C2NodeResponse::SessionRecordPreviewed { .. }
        | C2NodeResponse::HistoryDiscovered { .. }
        | C2NodeResponse::HistoryLoaded { .. }
        | C2NodeResponse::ContextPackForgotten { .. }
        | C2NodeResponse::ContextPackBytesRead { .. }
        | C2NodeResponse::WorkspaceUnregistered { .. }
        | C2NodeResponse::WorktreeRemoved { .. }
        | C2NodeResponse::Accepted
        | C2NodeResponse::ShuttingDown => false,
    }
}

fn routed_response_has_open_provider_id(response: &RoutedNodeResponse) -> bool {
    response.response.as_ref().is_ok_and(c2_node_response_has_open_provider_id)
}

async fn read_server_frame<S>(
    pipe: &mut S,
    limit: usize,
) -> Result<C2ServerFrame, C2ControlError>
where
    S: tokio::io::AsyncRead + Unpin,
{
    Ok(read_json_frame_limited_body_timeout(pipe, limit, FRAME_BODY_DEADLINE).await?)
}

async fn control_reader<R>(mut reader: R, owner: mpsc::Sender<OwnerInput>)
where
    R: tokio::io::AsyncRead + Unpin,
{
    loop {
        match read_json_frame_limited_body_timeout(
            &mut reader,
            MAX_C2_SERVER_FRAME_BYTES,
            FRAME_BODY_DEADLINE,
        ).await {
            Ok(frame) => if owner.send(OwnerInput::Frame(frame)).await.is_err() { return; },
            Err(_) => { let _ = owner.send(OwnerInput::Closed).await; return; }
        }
    }
}

async fn control_writer<W>(
    mut writer: W,
    mut frames: mpsc::Receiver<C2ClientFrame>,
    owner: mpsc::Sender<OwnerInput>,
) where
    W: tokio::io::AsyncWrite + Unpin,
{
    while let Some(frame) = frames.recv().await {
        if !matches!(timeout(FRAME_BODY_DEADLINE, write_json_frame_limited(
            &mut writer,
            &frame,
            MAX_C2_CLIENT_FRAME_BYTES,
        )).await, Ok(Ok(()))) {
            break;
        }
    }
    let _ = owner.send(OwnerInput::Closed).await;
}

async fn control_owner<E>(
    mut commands: mpsc::Receiver<ControlCommand>,
    events: E,
    topology: watch::Sender<Arc<C2Topology>>,
    writer: mpsc::Sender<C2ClientFrame>,
    mut incoming: mpsc::Receiver<OwnerInput>,
    path_capabilities: NegotiatedPathCapabilities,
) where
    E: Into<EventDelivery>,
{
    let events = events.into();
    let mut next_request_id = 1_u64;
    let mut pending = BTreeMap::new();
    let mut harness_mcp_receiver_gone = false;
    // Set the moment a `harness_mcp`/`regular` (terminal-frame) delivery
    // first drops an event for a full channel, cleared the moment a send
    // succeeds again -- so a sustained burst of drops during one backlog
    // episode logs exactly once (at the episode's start) instead of once
    // per dropped event, mirroring `harness_mcp_receiver_gone`'s own
    // one-shot-per-episode discipline for the channel-closed case right
    // next to it.
    let mut harness_mcp_channel_full_logged = false;
    let mut terminal_frame_channel_full_logged = false;
    let mut regular_event_channel_full_logged = false;
    let mut loss_reason = C2ConnectionLossReason::Shutdown;
    loop {
        tokio::select! {
            command = commands.recv() => {
                let Some(command) = command else { break; };
                if let Err(error) = reject_unnegotiated_outbound_path(
                    &command.request,
                    path_capabilities,
                ) {
                    let _ = command.reply.send(Err(error));
                    continue;
                }
                let request_id = C2RequestId(next_request_id);
                let Some(next) = next_request_id.checked_add(1) else {
                    let _ = command.reply.send(Err(C2ControlError::RequestIdExhausted));
                    break;
                };
                next_request_id = next;
                let expected_request = command.request.clone();
                let frame = C2ClientFrame::Request(C2RequestEnvelope {
                    request_id,
                    request: RoutedNodeRequest {
                        route: command.route.clone(),
                        request: command.request,
                    },
                });
                pending.insert(
                    request_id,
                    (command.route, expected_request, command.reply),
                );
                if writer.send(frame).await.is_err() {
                    loss_reason = C2ConnectionLossReason::PipeClosed;
                    break;
                }
            }
            input = incoming.recv() => {
                match input {
                    Some(OwnerInput::Frame(C2ServerFrame::Reply(reply))) => {
                        loss_reason = C2ConnectionLossReason::Protocol;
                        let Some((expected_route, expected_request, waiter)) =
                            pending.remove(&reply.request_id)
                        else {
                            break;
                        };
                        if reply.result.as_ref().is_ok_and(|routed| {
                            routed.node_id != expected_route.node_id
                                || routed.incarnation_id
                                    != expected_route.expected_incarnation_id
                        }) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "C2 reply route or node incarnation does not match the request"
                                    .to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_provider_session_index_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_session_record_context_export_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_native_session_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_workspace_content_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_session_task_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_managed_worktree_spawn_v2_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if let Err(message) = validate_harness_mcp_response(
                            &expected_request,
                            &reply.result,
                        ) {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                message.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.terminal_frame_events
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_has_terminal_frame_event)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                TERMINAL_FRAME_EVENTS_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.agent_stream_events
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_has_agent_stream_event)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                AGENT_STREAM_EVENTS_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.agent_progress_snapshot
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_has_agent_progress)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "agent progress snapshots require negotiated C2 capability"
                                    .to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.provider_ids_open
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_has_open_provider_id)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                OPEN_PROVIDER_ID_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.opaque_host_paths
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_has_unix_bytes)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                OPAQUE_UNIX_PATH_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.repository_paths
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_has_unix_repository_path)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                REPOSITORY_PATH_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.workspace_file_read
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_workspace_file_read)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                WORKSPACE_FILE_READ_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.workspace_file_write
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_workspace_file_write)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "workspace file writes require negotiated C2 capability".to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.workspace_entry_create
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_workspace_entry_create)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                WORKSPACE_ENTRY_CREATE_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.git_read
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_git_read)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "git reads require negotiated C2 capability".to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.host_directory_browse
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_host_directory_browse)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                HOST_DIRECTORY_BROWSE_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.standalone_workspace_lifecycle
                            && reply.result.as_ref().is_ok_and(
                                routed_response_requires_standalone_workspace_lifecycle,
                            )
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                STANDALONE_WORKSPACE_LIFECYCLE_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.provider_session_reference_index
                            && matches!(
                                expected_request,
                                NodeRequest::IndexProviderSession { .. }
                            )
                            && reply.result.as_ref().is_ok_and(
                                routed_response_requires_provider_session_reference_index,
                            )
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                PROVIDER_SESSION_REFERENCE_INDEX_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.spawn_spec_defaults_overrides
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_spawn_spec)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                SPAWN_SPEC_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.spawn_profile_revision
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_spawn_profile_revision)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                SPAWN_PROFILE_REVISION_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.worktree_selection
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_worktree_selection)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                WORKTREE_SELECTION_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.managed_worktree_lifecycle
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_managed_worktree)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                MANAGED_WORKTREE_LIFECYCLE_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.child_environment_profile
                            && reply.result.as_ref().is_ok_and(|routed| {
                                routed.response.as_ref().is_ok_and(
                                    C2NodeResponse::requires_child_environment_profile_capability,
                                )
                            })
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.session_bundle_materialization
                            && reply.result.as_ref().is_ok_and(|routed| {
                                routed.response.as_ref().is_ok_and(
                                    C2NodeResponse::requires_session_bundle_materialization_capability,
                                )
                            })
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                SESSION_BUNDLE_MATERIALIZATION_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.history_context_pack
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_history_context_pack)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                HISTORY_CONTEXT_PACK_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.session_record_context_export
                            && reply.result.as_ref().is_ok_and(|routed| {
                                routed.response.as_ref().is_ok_and(
                                    C2NodeResponse::requires_session_record_context_export_capability,
                                )
                            })
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                SESSION_RECORD_CONTEXT_EXPORT_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.control_detail
                            && reply.result.as_ref().is_ok_and(|routed| {
                                routed.response.as_ref().is_ok_and(
                                    C2NodeResponse::requires_control_detail_capability,
                                )
                            })
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                CONTROL_DETAIL_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.session_task_correlation
                            && reply.result.as_ref().is_ok_and(|routed| {
                                routed.response.as_ref().is_ok_and(
                                    C2NodeResponse::requires_session_task_correlation_capability,
                                )
                            })
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                SESSION_TASK_CORRELATION_NOT_NEGOTIATED.to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.native_session_catalog
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_native_session_catalog)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "native session catalog capability was not negotiated".to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.native_session_catalog_paging
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_native_session_catalog_paging)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "native session catalog paging capability was not negotiated".to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.native_session_preview
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_native_session_preview)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "native session preview capability was not negotiated".to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.native_session_index
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_native_session_index)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "native session index capability was not negotiated".to_owned(),
                            )));
                            break;
                        }
                        if !path_capabilities.harness_mcp_read_proxy
                            && reply
                                .result
                                .as_ref()
                                .is_ok_and(routed_response_requires_harness_mcp_proxy)
                        {
                            let _ = waiter.send(Err(C2ControlError::Protocol(
                                "harness MCP read proxy capability was not negotiated".to_owned(),
                            )));
                            break;
                        }
                        let _ = waiter.send(reply.result.map_err(C2ControlError::Relay));
                    }
                    Some(OwnerInput::Frame(C2ServerFrame::Event(event))) => {
                        loss_reason = C2ConnectionLossReason::Protocol;
                        if event.event.requires_harness_mcp_proxy_capability()
                            && (!path_capabilities.harness_mcp_read_proxy
                                || !event.event.harness_mcp_contract_is_valid_at(
                                    match current_unix_ms() { Ok(value) => value, Err(_) => break },
                                ))
                        {
                            break;
                        }
                        if !path_capabilities.terminal_frame_events
                            && c2_node_event_is_terminal_frame(&event.event)
                        {
                            tracing::warn!(
                                kind = c2_node_event_kind_label(&event.event),
                                sequence = event.cursor.sequence,
                                cursor = ?event.cursor,
                                reason = "terminal frame events capability not negotiated",
                                "dropping event and tearing down the C2 control connection",
                            );
                            break;
                        }
                        if !path_capabilities.agent_stream_events
                            && c2_node_event_is_agent_stream(&event.event)
                        {
                            tracing::warn!(
                                kind = c2_node_event_kind_label(&event.event),
                                sequence = event.cursor.sequence,
                                cursor = ?event.cursor,
                                reason = "agent stream events capability not negotiated",
                                "dropping event and tearing down the C2 control connection",
                            );
                            break;
                        }
                        if !path_capabilities.provider_ids_open
                            && routed_event_has_open_provider_id(&event)
                        {
                            break;
                        }
                        if !path_capabilities.opaque_host_paths
                            && routed_event_has_unix_bytes(&event)
                        {
                            break;
                        }
                        if !path_capabilities.managed_worktree_lifecycle
                            && c2_node_event_is_managed_worktree(&event.event)
                        {
                            break;
                        }
                        if !path_capabilities.worktree_selection
                            && c2_node_event_is_managed_worktree(&event.event)
                        {
                            break;
                        }
                        if !path_capabilities.child_environment_profile
                            && event
                                .event
                                .requires_child_environment_profile_capability()
                        {
                            break;
                        }
                        if !path_capabilities.session_bundle_materialization
                            && event
                                .event
                                .requires_session_bundle_materialization_capability()
                        {
                            break;
                        }
                        if !path_capabilities.history_context_pack
                            && event.event.requires_history_context_pack_capability()
                        {
                            break;
                        }
                        if !path_capabilities.session_task_correlation
                            && event.event.requires_session_task_correlation_capability()
                        {
                            break;
                        }
                        if !path_capabilities.control_detail
                            && event.event.requires_control_detail_capability()
                        {
                            break;
                        }
                        if event.event.requires_harness_mcp_proxy_capability() {
                            // A full or gone harness-MCP receiver must never
                            // tear down the whole C2 connection -- every
                            // other in-flight request (Arm, Activate,
                            // Abort, ...) would otherwise be answered
                            // `Closed` seconds later with no log line
                            // anywhere naming why.
                            let sequence = event.cursor.sequence;
                            let cursor = event.cursor;
                            match events.harness_mcp.try_send(event) {
                                Ok(()) => {
                                    reset_channel_full_drop_episode(&mut harness_mcp_channel_full_logged);
                                }
                                Err(mpsc::error::TrySendError::Full(_)) => {
                                    if should_log_channel_full_drop(&mut harness_mcp_channel_full_logged) {
                                        tracing::warn!(
                                            kind = "harness-mcp-read-call",
                                            sequence,
                                            cursor = ?cursor,
                                            reason = "harness_mcp event channel full",
                                            "dropped harness MCP event; C2 control connection stays open",
                                        );
                                    }
                                }
                                Err(mpsc::error::TrySendError::Closed(_)) => {
                                    if !harness_mcp_receiver_gone {
                                        harness_mcp_receiver_gone = true;
                                        tracing::warn!(
                                            kind = "harness-mcp-read-call",
                                            sequence,
                                            cursor = ?cursor,
                                            reason = "harness_mcp event receiver dropped",
                                            "no harness MCP event subscriber; further harness MCP events on this connection are dropped silently",
                                        );
                                    }
                                }
                            }
                        } else if c2_node_event_is_terminal_frame(&event.event) {
                            let kind = c2_node_event_kind_label(&event.event);
                            let sequence = event.cursor.sequence;
                            let cursor = event.cursor;
                            match events.regular.try_send(event) {
                                Ok(()) => {
                                    reset_channel_full_drop_episode(&mut terminal_frame_channel_full_logged);
                                }
                                Err(mpsc::error::TrySendError::Full(_)) => {
                                    if should_log_channel_full_drop(&mut terminal_frame_channel_full_logged) {
                                        tracing::warn!(
                                            kind,
                                            sequence,
                                            cursor = ?cursor,
                                            reason = "regular event channel full",
                                            "dropped terminal frame event; C2 control connection stays open",
                                        );
                                    }
                                }
                                Err(mpsc::error::TrySendError::Closed(_)) => {
                                    tracing::warn!(
                                        kind,
                                        sequence,
                                        cursor = ?cursor,
                                        reason = "regular event receiver dropped",
                                        "dropping event and tearing down the C2 control connection",
                                    );
                                    loss_reason = C2ConnectionLossReason::RegularEventBackpressure;
                                    break;
                                }
                            }
                        } else {
                            let kind = c2_node_event_kind_label(&event.event);
                            let sequence = event.cursor.sequence;
                            let cursor = event.cursor;
                            match events.regular.try_send(event) {
                                Ok(()) => {
                                    reset_channel_full_drop_episode(&mut regular_event_channel_full_logged);
                                }
                                Err(mpsc::error::TrySendError::Full(_)) => {
                                    if should_log_channel_full_drop(&mut regular_event_channel_full_logged) {
                                        tracing::warn!(
                                            kind,
                                            sequence,
                                            cursor = ?cursor,
                                            reason = "regular event channel full",
                                            "dropped regular event; C2 control connection stays open",
                                        );
                                    }
                                }
                                Err(mpsc::error::TrySendError::Closed(_)) => {
                                    tracing::warn!(
                                        kind,
                                        sequence,
                                        cursor = ?cursor,
                                        reason = "regular event receiver dropped",
                                        "dropping event and tearing down the C2 control connection",
                                    );
                                    loss_reason = C2ConnectionLossReason::RegularEventBackpressure;
                                    break;
                                }
                            }
                        }
                    }
                    Some(OwnerInput::Frame(C2ServerFrame::Topology(next))) => {
                        loss_reason = C2ConnectionLossReason::Protocol;
                        if !path_capabilities.provider_ids_open
                            && topology_has_open_provider_id(&next)
                        {
                            break;
                        }
                        if topology.borrow().as_ref() != &next {
                            topology.send_replace(Arc::new(next));
                        }
                    }
                    Some(OwnerInput::Frame(C2ServerFrame::Challenge(_) | C2ServerFrame::Hello(_) | C2ServerFrame::Rejected(_))) => {
                        loss_reason = C2ConnectionLossReason::Protocol;
                        break;
                    }
                    Some(OwnerInput::Closed) | None => {
                        loss_reason = C2ConnectionLossReason::PipeClosed;
                        break;
                    }
                }
            }
        }
    }
    if !pending.is_empty() {
        tracing::warn!(
            reason = %loss_reason,
            pending = pending.len(),
            "C2 control connection lost: failing every request still pending",
        );
    }
    for (_, (_, _, waiter)) in pending {
        let _ = waiter.send(Err(C2ControlError::ConnectionLost { reason: loss_reason }));
    }
}

pub(crate) fn c2_proof(
    token: &str,
    direction: C2AuthDirection,
    client_nonce: &[u8; C2_AUTH_NONCE_BYTES],
    server_nonce: &[u8; C2_AUTH_NONCE_BYTES],
    compatibility: Option<(&ClientCompatibilityOffer, &NegotiatedC2ControlCompatibility)>,
) -> Result<[u8; 32], C2ControlError> {
    let transcript = match compatibility {
        Some((offer, selected)) => c2_bound_auth_transcript(
            direction,
            client_nonce,
            server_nonce,
            offer,
            selected,
        ).map_err(|error| C2ControlError::Authentication(error.to_string()))?,
        None => c2_auth_transcript(direction, client_nonce, server_nonce),
    };
    local_hmac_sha256(token.as_bytes(), &transcript)
        .map_err(C2ControlError::Authentication)
}

#[cfg(windows)]
fn validate_endpoint(endpoint: &str) -> Result<(), C2ControlError> {
    if !endpoint.starts_with(r"\\.\pipe\") || endpoint.len() <= r"\\.\pipe\".len() || endpoint.len() > 1024 {
        return Err(C2ControlError::InvalidEndpoint);
    }
    Ok(())
}

#[cfg(unix)]
fn validate_endpoint(endpoint: &str) -> Result<(), C2ControlError> {
    use std::os::unix::ffi::OsStrExt;
    use std::path::Path;

    let path = Path::new(endpoint);
    if endpoint.is_empty() || path.as_os_str().as_bytes().len() > 103 || !path.is_absolute()
        || path.file_name().is_none()
    {
        return Err(C2ControlError::InvalidEndpoint);
    }
    Ok(())
}

/// Tunnel control target for HQ. Not a substitute for [`validate_endpoint`]:
/// a raw `host:port` string is still [`C2ControlError::InvalidEndpoint`] there.
fn validate_tunnel_control_target(address: std::net::SocketAddr) -> Result<(), C2ControlError> {
    if address.port() == 0 || address.ip().is_unspecified() || address.ip().is_multicast() {
        return Err(C2ControlError::InvalidTunnelControl);
    }
    Ok(())
}

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

#[derive(Debug, Error)]
pub enum C2ControlError {
    #[cfg_attr(windows, error("C2 control endpoint is not a bounded local named pipe"))]
    #[cfg_attr(unix, error("C2 control endpoint is not a bounded absolute local endpoint"))]
    InvalidEndpoint,
    #[error("C2 tunnel control address must be a concrete host and a nonzero port")]
    InvalidTunnelControl,
    #[error("C2 token must contain 1..=4096 visible ASCII bytes without whitespace")]
    InvalidToken,
    #[error("C2 control I/O failed: {0}")]
    Io(#[from] io::Error),
    #[error(transparent)]
    Frame(#[from] FrameError),
    #[error("C2 authentication timed out")]
    AuthenticationTimedOut,
    #[error("C2 authentication failed: {0}")]
    Authentication(String),
    #[error("C2 control protocol failed: {0}")]
    Protocol(String),
    #[error("build stamp mismatch: local={local} remote={remote}")]
    BuildStampMismatch { local: String, remote: String },
    #[error("C2 relay rejected request: {0:?}")]
    Relay(C2RelayFailure),
    #[error("C2 control connection closed")]
    Closed,
    /// The `control_owner` loop tore the connection down for a NAMED
    /// reason (as opposed to `Closed`, which also covers a local caller
    /// dropping its own handle/queue -- never emitted by `control_owner`
    /// itself). Every request pending at teardown gets this, once, with
    /// the reason logged alongside the pending count.
    #[error("C2 control connection lost: {reason}")]
    ConnectionLost { reason: C2ConnectionLossReason },
    #[error("C2 control request queue is full")]
    QueueFull,
    #[error("C2 request ID space exhausted")]
    RequestIdExhausted,
}

/// Why `control_owner` tore its connection down, attached to every request
/// still pending at that moment (`C2ControlError::ConnectionLost`). Logged
/// once at the teardown point instead of being silently swallowed.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum C2ConnectionLossReason {
    /// The reader or writer half of the physical pipe ended (I/O error,
    /// EOF, or a downstream send failure into the writer task).
    PipeClosed,
    /// The relay sent a frame or reply that violates the negotiated
    /// protocol: an unmatched request id, a route/incarnation mismatch, an
    /// unnegotiated-capability leak, an invalid topology frame, or a
    /// repeated handshake frame after authentication.
    Protocol,
    /// The `regular` event consumer is gone: its channel closed because
    /// the receiver was dropped. A merely full (but alive) `regular`
    /// channel never reaches this reason -- the offending event is
    /// dropped and the connection stays open instead.
    RegularEventBackpressure,
    /// The local caller side shut the connection down deliberately (every
    /// `C2ControlHandle` clone dropped) or the request id space was
    /// exhausted.
    Shutdown,
}

impl std::fmt::Display for C2ConnectionLossReason {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(match self {
            Self::PipeClosed => "pipe closed",
            Self::Protocol => "protocol violation",
            Self::RegularEventBackpressure => "regular event consumer fell behind",
            Self::Shutdown => "local caller shut the connection down",
        })
    }
}

#[cfg(all(test, windows))]
mod tests {
    use super::*;

    #[test]
    fn harness_mcp_c2_client_capability_and_correlation_are_exact() {
        let reservation_id = gate4agent_node_protocol::HarnessMcpReservationId::new(
            format!("hmcpres_{}", "a".repeat(24)),
        ).unwrap();
        let activation_digest = gate4agent_node_protocol::HarnessMcpActivationDigest::new(
            format!("sha256:{}", "b".repeat(64)),
        ).unwrap();
        let request = NodeRequest::AbortHarnessMcpReservation {
            reservation_id: reservation_id.clone(),
            activation_digest: activation_digest.clone(),
        };
        assert!(reject_unnegotiated_outbound_path(
            &request,
            NegotiatedPathCapabilities::default(),
        ).is_err());
        let capabilities = NegotiatedPathCapabilities {
            harness_mcp_read_proxy: true,
            spawn_spec_defaults_overrides: true,
            ..NegotiatedPathCapabilities::default()
        };
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_ok());

        let routed = RoutedNodeResponse {
            node_id: gate4agent_node_protocol::NodeId::new("node-a").unwrap(),
            incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([1; 16]),
            response: Ok(C2NodeResponse::Aborted {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
            }),
        };
        assert!(validate_harness_mcp_response(&request, &Ok(routed.clone())).is_ok());
        let mut mismatch = routed;
        mismatch.response = Ok(C2NodeResponse::Aborted {
            reservation_id: gate4agent_node_protocol::HarnessMcpReservationId::new(
                format!("hmcpres_{}", "c".repeat(24)),
            ).unwrap(),
            activation_digest,
        });
        assert!(validate_harness_mcp_response(&request, &Ok(mismatch)).is_err());
    }

    #[test]
    fn acp_control_verbs_are_admitted_only_when_negotiated_and_refuse_by_own_name() {
        let session = gate4agent_node_protocol::SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(3),
                generation: SessionGeneration(1),
            },
        };
        let requests = [
            NodeRequest::ResolveInteraction {
                session: session.clone(),
                correlation_id: "acp-correlation-1".to_owned(),
                response: ProviderInteractionResponse::ApproveOnce,
            },
            NodeRequest::SetSessionMode {
                session: session.clone(),
                mode_id: "plan".to_owned(),
            },
            NodeRequest::SetSessionConfigOption {
                session: session.clone(),
                option_id: "reasoning-effort".to_owned(),
                value_json: "\"high\"".to_owned(),
            },
            NodeRequest::SetSessionModel {
                session: session.clone(),
                model_id: "claude-opus".to_owned(),
            },
        ];
        for request in &requests {
            assert_eq!(
                request.required_capability(),
                Some(C2_ACP_CONTROL_CAPABILITY),
            );
            let refusal = reject_unnegotiated_outbound_path(
                request,
                NegotiatedPathCapabilities::default(),
            )
            .unwrap_err();
            assert!(
                matches!(
                    &refusal,
                    C2ControlError::Protocol(message)
                        if message == ACP_CONTROL_NOT_NEGOTIATED
                ),
                "unexpected refusal for {request:?}: {refusal:?}",
            );
            let capabilities = NegotiatedPathCapabilities {
                acp_control: true,
                ..NegotiatedPathCapabilities::default()
            };
            assert!(reject_unnegotiated_outbound_path(request, capabilities).is_ok());
        }
    }

    #[test]
    fn unrecognized_required_capability_refuses_naming_itself_not_workspace_file_read() {
        // Every capability `NodeRequest::required_capability()` can return
        // today has a dedicated arm now, so this reaches straight for the
        // fallback that stands in for tomorrow's capability -- exactly the
        // shape of the defect that shipped: a capability string with no
        // dedicated arm yet must never borrow another capability's message.
        let message = unnegotiated_capability_refusal(Some("future-control-v9"));
        assert_ne!(message, WORKSPACE_FILE_READ_NOT_NEGOTIATED);
        assert!(
            message.contains("future-control-v9"),
            "refusal must name the unrecognised capability, got: {message}",
        );

        // The one capability the fallback must never again be mistaken for.
        let workspace_file_read_message =
            unnegotiated_capability_refusal(Some(C2_WORKSPACE_FILE_READ_CAPABILITY));
        assert_eq!(workspace_file_read_message, WORKSPACE_FILE_READ_NOT_NEGOTIATED);
        assert_ne!(workspace_file_read_message, message);
    }

    use hatchery_c2_protocol::{
        ArchitectureId, C2GitSnapshot, C2WorkspaceInspection, C2WorkspaceSnapshot,
        C2ServerChallenge, HostDescriptor, HostDirectoryEntry, HostDirectoryListing,
        NodeCursor, NodeId, OpaqueHostPath,
        OperatingSystemId, PathEncoding, PathSemantics, PathStyle, RepositoryPath,
        ResolvedSpawnReceipt, SpawnDeadlineMs, SpawnFieldProvenance,
        SpawnIdempotencyKey, SpawnOverrides, SpawnProfileId, SpawnProfileRevision,
        SpawnPromptMetadata, SpawnRequiredCapabilities, SpawnResolutionProvenance,
        SpawnSpec, SpawnTarget, WorkspaceFileContent, WorkspaceFileRead,
        WorktreeServiceMode,
    };
    use gate4agent_node_protocol::{
        GitStatusEntry, NodeIncarnationId, SessionMode, WorkspaceEntry, WorkspaceEntryKind,
        WorkspaceId,
    };
    use gate4agent_types::{
        AgentInstanceId, ProviderInteractionResponse, ProviderSessionIdentity,
        ProviderSessionKey, PtyScreenState, SessionGeneration, TerminalFrame, TerminalSize,
    };
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};
    use tokio::net::windows::named_pipe::ServerOptions;

    fn unique_control_endpoint() -> String {
        static NEXT: AtomicU64 = AtomicU64::new(1);
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        format!(
            r"\\.\pipe\hatchery-c2-client-strict-{}-{now}-{}",
            std::process::id(),
            NEXT.fetch_add(1, Ordering::Relaxed),
        )
    }

    fn routed_event(sequence: u64) -> RoutedNodeEvent {
        RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                sequence,
            },
            event: hatchery_c2_protocol::C2NodeEvent::ResyncRequired {
                oldest_available_sequence: sequence,
            },
        }
    }

    fn event(sequence: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Event(routed_event(sequence)))
    }

    fn ordinary_state_event(sequence: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Event(RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                sequence,
            },
            event: C2NodeEvent::WorkspaceRemoved {
                workspace_id: WorkspaceId::new("retired").unwrap(),
            },
        }))
    }

    fn terminal_event(sequence: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Event(RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                sequence,
            },
            event: terminal_node_event(sequence),
        }))
    }

    fn harness_mcp_event(call_index: usize, deadline_unix_ms: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Event(RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                sequence: 41,
            },
            event: C2NodeEvent::HarnessMcpReadCall {
                reservation_id: gate4agent_node_protocol::HarnessMcpReservationId::new(
                    format!("hmcpres_{:024x}", 1),
                ).unwrap(),
                activation_digest: gate4agent_node_protocol::HarnessMcpActivationDigest::new(
                    format!("sha256:{}", "b".repeat(64)),
                ).unwrap(),
                record_id: gate4agent_node_protocol::SessionRecordId::new(
                    "session-001",
                ).unwrap(),
                session: gate4agent_node_protocol::SessionAddress {
                    workspace_id: WorkspaceId::new("primary").unwrap(),
                    session: gate4agent_node_protocol::SessionKey {
                        instance_id: AgentInstanceId(7),
                        generation: SessionGeneration(2),
                    },
                },
                call_id: gate4agent_node_protocol::HarnessMcpCallId::new(
                    format!("hmcpcall_{call_index:024x}"),
                ).unwrap(),
                request: gate4agent_node_protocol::HarnessMcpOpaquePayloadV1 {
                    content_type:
                        gate4agent_node_protocol::HarnessMcpContentTypeV1::HarnessReadRequestJsonV1,
                    body: br#"{"kind":"context-get"}"#.to_vec(),
                },
                deadline_unix_ms,
            },
        }))
    }

    fn terminal_node_event(sequence: u64) -> C2NodeEvent {
        C2NodeEvent::TerminalFrame {
            address: gate4agent_node_protocol::SessionAddress {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                session: gate4agent_node_protocol::SessionKey {
                    instance_id: AgentInstanceId(7),
                    generation: SessionGeneration(2),
                },
            },
            frame: TerminalFrame {
                sequence,
                size: TerminalSize { rows: 24, columns: 80 },
                cursor_row: 1,
                cursor_column: 2,
                contents: "ready".to_owned(),
                formatted: b"ready".to_vec(),
                scrollback_formatted: Vec::new(),
                alternate_screen: false,
                mouse_protocol_enabled: false,
                mouse_protocol_encoding: Default::default(),
                produced_at_unix_ms: 0,
                screen_state: PtyScreenState::default(),
                bracketed_paste: None,
            },
        }
    }

    fn route() -> NodeRoute {
        NodeRoute {
            node_id: NodeId::new("node-a").unwrap(),
            expected_incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
        }
    }

    #[tokio::test]
    async fn linearized_start_request_is_bounded_closed_and_exactly_correlated() {
        let (commands, mut receiver) = mpsc::channel(1);
        let first = start_control_request(
            &commands,
            all_path_capabilities(),
            route(),
            NodeRequest::Snapshot,
        ).unwrap();
        let full = start_control_request(
            &commands,
            all_path_capabilities(),
            route(),
            NodeRequest::Snapshot,
        );
        assert!(matches!(full, Err(C2ControlError::QueueFull)));

        let queued = receiver.recv().await.unwrap();
        assert_eq!(queued.route, route());
        assert!(matches!(queued.request, NodeRequest::Snapshot));
        let exact = RoutedNodeResponse {
            node_id: route().node_id,
            incarnation_id: route().expected_incarnation_id,
            response: Ok(C2NodeResponse::Accepted),
        };
        queued.reply.send(Ok(exact.clone())).unwrap();
        assert_eq!(first.finish().await.unwrap(), exact);

        let (closed_commands, closed_receiver) = mpsc::channel(1);
        drop(closed_receiver);
        let closed = start_control_request(
            &closed_commands,
            all_path_capabilities(),
            route(),
            NodeRequest::Snapshot,
        );
        assert!(matches!(closed, Err(C2ControlError::Closed)));
    }

    fn spawn_spec(node_id: &str) -> SpawnSpec {
        SpawnSpec {
            target: SpawnTarget {
                node_id: NodeId::new(node_id).unwrap(),
                workspace_id: WorkspaceId::new("primary").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("spawn-1").unwrap(),
            required_capabilities: SpawnRequiredCapabilities::default(),
        }
    }

    fn spawn_receipt() -> ResolvedSpawnReceipt {
        ResolvedSpawnReceipt {
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            session: gate4agent_node_protocol::SessionAddress {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                session: gate4agent_node_protocol::SessionKey {
                    instance_id: AgentInstanceId(7),
                    generation: SessionGeneration(2),
                },
            },
            target: spawn_spec("node-a").target,
            profile_id: SpawnProfileId::new("default").unwrap(),
            profile_revision: SpawnProfileRevision::new("r1").unwrap(),
            provider: hatchery_c2_protocol::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: SpawnDeadlineMs::new(5_000).unwrap(),
            idempotency_key: SpawnIdempotencyKey::new("spawn-1").unwrap(),
            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,
        }
    }

    fn no_path_capabilities() -> NegotiatedPathCapabilities {
        NegotiatedPathCapabilities {
            opaque_host_paths: false,
            repository_paths: false,
            workspace_file_read: false,
            host_directory_browse: false,
            standalone_workspace_lifecycle: false,
            provider_session_reference_index: false,
            provider_ids_open: false,
            spawn_spec_defaults_overrides: false,
            worktree_selection: false,
            managed_worktree_lifecycle: false,
            managed_worktree_spawn_v2: false,
            child_environment_profile: false,
            session_bundle_materialization: false,
            history_context_pack: false,
            native_session_catalog: false,
            native_session_catalog_paging: false,
            native_session_index: false,
            native_session_preview: false,
            terminal_frame_events: false,
            ..Default::default()
        }
    }

    fn all_path_capabilities() -> NegotiatedPathCapabilities {
        NegotiatedPathCapabilities {
            opaque_host_paths: true,
            repository_paths: true,
            workspace_file_read: true,
            host_directory_browse: true,
            standalone_workspace_lifecycle: true,
            provider_session_reference_index: true,
            provider_ids_open: true,
            spawn_spec_defaults_overrides: true,
            spawn_profile_revision: true,
            worktree_selection: true,
            managed_worktree_lifecycle: true,
            managed_worktree_spawn_v2: true,
            child_environment_profile: true,
            session_bundle_materialization: true,
            history_context_pack: true,
            session_record_context_export: true,
            native_session_catalog: true,
            native_session_catalog_paging: true,
            native_session_index: true,
            native_session_preview: true,
            terminal_frame_events: true,
            agent_progress_snapshot: true,
            ..Default::default()
        }
    }

    #[test]
    fn session_record_context_export_requires_both_capabilities_and_exact_echo() {
        let session = gate4agent_node_protocol::SessionAddress {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            session: gate4agent_node_protocol::SessionKey {
                instance_id: AgentInstanceId(41),
                generation: SessionGeneration(3),
            },
        };
        let record_id = gate4agent_node_protocol::SessionRecordId::new(
            "record-context-41",
        )
        .unwrap();
        let request = NodeRequest::ExportContextPackForSessionRecord {
            record_id: record_id.clone(),
            session: session.clone(),
        };
        let mut capabilities = NegotiatedPathCapabilities {
            session_record_context_export: true,
            ..NegotiatedPathCapabilities::default()
        };
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_err());
        capabilities.history_context_pack = true;
        capabilities.session_record_context_export = false;
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_err());
        capabilities.session_record_context_export = true;
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_ok());

        let context = gate4agent_node_protocol::ResolvedContextPackReceipt {
            id: gate4agent_node_protocol::SpawnContextId::new("context-record-41").unwrap(),
            digest: gate4agent_node_protocol::SpawnContextDigest::new(format!(
                "sha256:{}",
                "a".repeat(64),
            ))
            .unwrap(),
            lineage: gate4agent_node_protocol::ContextPackLineageReceipt {
                source_node_id: NodeId::new("node-a").unwrap(),
                source_session: session.clone(),
                source_provider: gate4agent_types::AgentId::new("claude").unwrap(),
            },
            source_message_count: 2,
            retained_message_count: 2,
            byte_len: 64,
            truncated: false,
        };
        let response = Ok(RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::ContextPackForSessionRecordExported {
                record_id: record_id.clone(),
                session: session.clone(),
                context: context.clone(),
            }),
        });
        assert!(validate_session_record_context_export_response(&request, &response).is_ok());
        let mismatch = Ok(RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::ContextPackForSessionRecordExported {
                record_id: gate4agent_node_protocol::SessionRecordId::new("record-other")
                    .unwrap(),
                session,
                context,
            }),
        });
        assert!(validate_session_record_context_export_response(&request, &mismatch).is_err());
    }

    #[test]
    fn native_session_response_correlation_is_fail_closed() {
        let node_id = NodeId::new("node-a").unwrap();
        let incarnation_id = NodeIncarnationId::from_bytes([7; 16]);
        let route = gate4agent_node_protocol::NativeSessionCatalogRoute::workspace(
            WorkspaceId::new("primary").unwrap(),
            gate4agent_types::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 routed = |response| {
            Ok(RoutedNodeResponse {
                node_id: node_id.clone(),
                incarnation_id,
                response: Ok(response),
            })
        };

        let catalog = NodeRequest::CatalogNativeSessions {
            route: route.clone(),
            limit: 10,
        };
        assert!(validate_native_session_response(
            &catalog,
            &routed(C2NodeResponse::NativeSessionsCataloged {
                route: route.clone(),
                entries: Vec::new(),
                summary: None,
            }),
        )
        .is_ok());
        let wrong_route = gate4agent_node_protocol::NativeSessionCatalogRoute::workspace(
            WorkspaceId::new("other").unwrap(),
            gate4agent_types::AgentId::new("codex").unwrap(),
        );
        assert!(validate_native_session_response(
            &catalog,
            &routed(C2NodeResponse::NativeSessionsCataloged {
                route: wrong_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,
        };
        assert!(validate_native_session_response(
            &page,
            &routed(C2NodeResponse::NativeSessionsPaged {
                route: route.clone(),
                page: gate4agent_node_protocol::NativeSessionCatalogPage {
                    window: gate4agent_node_protocol::NativeSessionCatalogWindow::Recent,
                    revision: 8,
                    entries: Vec::new(),
                    next_after_selection_id: None,
                    remaining_count: 0,
                    has_more: false,
                },
            }),
        )
        .is_err());

        let preview = NodeRequest::PreviewNativeSession {
            selection: selection.clone(),
            message_limit: 10,
        };
        let mut wrong_selection = selection.clone();
        wrong_selection.catalog_revision = 8;
        assert!(validate_native_session_response(
            &preview,
            &routed(C2NodeResponse::NativeSessionPreviewed {
                selection: wrong_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(),
                },
            }),
        )
        .is_err());

        let index = NodeRequest::IndexNativeSession {
            selection: selection.clone(),
            display_name: "Indexed".to_owned(),
        };
        let record = hatchery_c2_protocol::C2ManagedSessionRecord {
            record_id: gate4agent_node_protocol::SessionRecordId::new("record-7").unwrap(),
            display_name: "Indexed".to_owned(),
            provider: gate4agent_types::AgentId::new("codex").unwrap(),
            mode: SessionMode::Pty,
            state: gate4agent_node_protocol::ManagedSessionState::Dormant,
            workspace_id: WorkspaceId::new("primary").unwrap(),
            active_session: None,
            environment_profile: None,
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            provider_identity_present: true,
            created_at_unix_ms: 1,
            updated_at_unix_ms: 1,
        };
        assert!(validate_native_session_response(
            &index,
            &routed(C2NodeResponse::NativeSessionIndexed {
                selection: selection.clone(),
                record: record.clone(),
            }),
        )
        .is_ok());
        assert!(validate_native_session_response(
            &index,
            &routed(C2NodeResponse::ProviderSessionIndexed {
                record: record.clone(),
            }),
        )
        .is_err());
        let mut wrong_echo = selection.clone();
        wrong_echo.catalog_revision = 8;
        assert!(validate_native_session_response(
            &index,
            &routed(C2NodeResponse::NativeSessionIndexed {
                selection: wrong_echo,
                record: record.clone(),
            }),
        )
        .is_err());
        let mut wrong_provider = record.clone();
        wrong_provider.provider = gate4agent_types::AgentId::new("claude").unwrap();
        assert!(validate_native_session_response(
            &index,
            &routed(C2NodeResponse::NativeSessionIndexed {
                selection: selection.clone(),
                record: wrong_provider,
            }),
        )
        .is_err());
        let mut wrong_workspace = record.clone();
        wrong_workspace.workspace_id = WorkspaceId::new("other").unwrap();
        assert!(validate_native_session_response(
            &index,
            &routed(C2NodeResponse::NativeSessionIndexed {
                selection: selection.clone(),
                record: wrong_workspace,
            }),
        )
        .is_err());

        let provider_index = NodeRequest::IndexProviderSession {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            provider: gate4agent_types::AgentId::new("codex").unwrap(),
            identity: gate4agent_types::ProviderSessionIdentity {
                key: gate4agent_types::ProviderSessionKey::SessionId,
                id: "provider-session-7".to_owned(),
                transcript_path: None,
            },
            display_name: "Indexed".to_owned(),
        };
        assert!(validate_provider_session_index_response(
            &provider_index,
            &routed(C2NodeResponse::ProviderSessionIndexed {
                record: record.clone(),
            }),
        )
        .is_ok());
        assert!(validate_provider_session_index_response(
            &provider_index,
            &routed(C2NodeResponse::NativeSessionIndexed {
                selection,
                record: record.clone(),
            }),
        )
        .is_err());
        assert!(validate_provider_session_index_response(
            &catalog,
            &routed(C2NodeResponse::ProviderSessionIndexed { record }),
        )
        .is_err());
    }

    fn reply(request_id: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Reply(hatchery_c2_protocol::C2ReplyEnvelope {
            request_id: C2RequestId(request_id),
            result: Ok(RoutedNodeResponse {
                node_id: NodeId::new("node-a").unwrap(),
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                response: Ok(hatchery_c2_protocol::C2NodeResponse::Accepted),
            }),
        }))
    }

    fn terminal_resync_reply(request_id: u64, sequence: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Reply(hatchery_c2_protocol::C2ReplyEnvelope {
            request_id: C2RequestId(request_id),
            result: Ok(RoutedNodeResponse {
                node_id: NodeId::new("node-a").unwrap(),
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                response: Ok(C2NodeResponse::Resync {
                    event_sequence: sequence,
                    oldest_available_sequence: 1,
                    snapshot: hatchery_c2_protocol::C2NodeSnapshot {
                        node_id: NodeId::new("node-a").unwrap(),
                        enabled_providers: Vec::new(),
                        provider_runtime_statuses: Default::default(),
                        workspaces: Vec::new(),
                        session_records: Vec::new(),
                        agent_progress: Vec::new(),
                        managed_worktrees: Vec::new(),
                        launch_inventory: None,
                    },
                    events: vec![hatchery_c2_protocol::C2NodeEventEnvelope {
                        sequence,
                        event: terminal_node_event(sequence),
                    }],
                }),
            }),
        }))
    }

    fn unix_path() -> OpaqueHostPath {
        OpaqueHostPath::unix_bytes(vec![b'/', b's', b'r', b'v', b'/', 0xff]).unwrap()
    }

    fn unix_path_reply(request_id: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Reply(hatchery_c2_protocol::C2ReplyEnvelope {
            request_id: C2RequestId(request_id),
            result: Ok(RoutedNodeResponse {
                node_id: NodeId::new("node-a").unwrap(),
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                response: Ok(C2NodeResponse::WorktreeRemoved {
                    target_root: unix_path(),
                    workspace_id: None,
                }),
            }),
        }))
    }

    fn repository_path(value: &str) -> RepositoryPath {
        RepositoryPath::utf8(value.to_owned()).unwrap()
    }

    fn unix_repository_path(value: &[u8]) -> RepositoryPath {
        RepositoryPath::unix_bytes(value.to_vec()).unwrap()
    }

    fn repository_inspection_response(
        entry_path: RepositoryPath,
        status_path: RepositoryPath,
        previous_path: Option<RepositoryPath>,
    ) -> C2NodeResponse {
        C2NodeResponse::WorkspaceInspected {
            inspection: C2WorkspaceInspection {
                workspace_id: WorkspaceId::new("foreign").unwrap(),
                entries: vec![WorkspaceEntry {
                    relative_path: entry_path,
                    kind: WorkspaceEntryKind::File,
                }],
                tree_truncated: false,
                git: C2GitSnapshot {
                    is_repository: true,
                    branch: Some("main".to_owned()),
                    status: vec![GitStatusEntry {
                        index_status: " ".to_owned(),
                        worktree_status: "M".to_owned(),
                        path: status_path,
                        previous_path,
                    }],
                    recent_commits: Vec::new(),
                    worktrees: Vec::new(),
                    managed_worktree: None,
                    truncated: false,
                    diagnostic_present: false,
                },
                truncation: None,
            },
        }
    }

    fn unix_repository_path_reply(request_id: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Reply(hatchery_c2_protocol::C2ReplyEnvelope {
            request_id: C2RequestId(request_id),
            result: Ok(RoutedNodeResponse {
                node_id: NodeId::new("node-a").unwrap(),
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                response: Ok(repository_inspection_response(
                    unix_repository_path(b"src/\xff"),
                    repository_path("src/main.rs"),
                    None,
                )),
            }),
        }))
    }

    fn unix_path_event(sequence: u64) -> OwnerInput {
        OwnerInput::Frame(C2ServerFrame::Event(RoutedNodeEvent {
            node_id: NodeId::new("node-a").unwrap(),
            cursor: NodeCursor {
                incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                sequence,
            },
            event: C2NodeEvent::WorkspaceAdded {
                workspace: C2WorkspaceSnapshot {
                    workspace_id: WorkspaceId::new("foreign").unwrap(),
                    canonical_root: unix_path(),
                    sessions: Vec::new(),
                    worktree_service_mode: None,
                    managed_worktree_profiles: None,
                },
            },
        }))
    }

    #[test]
    fn c2_control_client_offer_is_exact_v2_with_authenticated_opt_ins() {
        let offer = client_compatibility_offer().unwrap();

        assert_eq!(offer.build_stamp, BUILD_STAMP);
        assert_eq!(
            offer.capabilities,
            vec![
                CapabilityId::new(C2_COMPATIBILITY_METADATA_CAPABILITY).unwrap(),
                CapabilityId::new(C2_OPAQUE_UNIX_PATH_CAPABILITY).unwrap(),
                CapabilityId::new(C2_REPOSITORY_PATH_CAPABILITY).unwrap(),
                CapabilityId::new(C2_WORKSPACE_FILE_READ_CAPABILITY).unwrap(),
                CapabilityId::new(C2_WORKSPACE_FILE_WRITE_CAPABILITY).unwrap(),
                CapabilityId::new(C2_WORKSPACE_ENTRY_CREATE_CAPABILITY).unwrap(),
                CapabilityId::new(C2_GIT_READ_CAPABILITY).unwrap(),
                CapabilityId::new(C2_PROVIDER_CONTRACT_MANIFEST_CAPABILITY).unwrap(),
                CapabilityId::new(C2_PROVIDER_RUNTIME_STATUS_CAPABILITY).unwrap(),
                CapabilityId::new(C2_PROVIDER_ID_OPEN_CAPABILITY).unwrap(),
                CapabilityId::new(C2_SPAWN_SPEC_DEFAULTS_OVERRIDES_CAPABILITY).unwrap(),
                CapabilityId::new(C2_SPAWN_PROFILE_REVISION_CAPABILITY).unwrap(),
                CapabilityId::new(C2_TERMINAL_FRAME_EVENTS_CAPABILITY).unwrap(),
                CapabilityId::new(C2_AGENT_STREAM_EVENTS_CAPABILITY).unwrap(),
                CapabilityId::new(C2_WORKTREE_SELECTION_CAPABILITY).unwrap(),
                CapabilityId::new(C2_MANAGED_WORKTREE_LIFECYCLE_CAPABILITY).unwrap(),
                CapabilityId::new(C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY).unwrap(),
                CapabilityId::new(C2_CHILD_ENVIRONMENT_PROFILE_CAPABILITY).unwrap(),
                CapabilityId::new(C2_SESSION_BUNDLE_MATERIALIZATION_CAPABILITY).unwrap(),
                CapabilityId::new(C2_HISTORY_CONTEXT_PACK_CAPABILITY).unwrap(),
                CapabilityId::new(C2_SESSION_RECORD_CONTEXT_EXPORT_CAPABILITY).unwrap(),
                CapabilityId::new(C2_NATIVE_SESSION_CATALOG_CAPABILITY).unwrap(),
                CapabilityId::new(C2_NATIVE_SESSION_CATALOG_PAGING_CAPABILITY).unwrap(),
                CapabilityId::new(C2_NATIVE_SESSION_INDEX_CAPABILITY).unwrap(),
                CapabilityId::new(C2_NATIVE_SESSION_PREVIEW_CAPABILITY).unwrap(),
                CapabilityId::new(C2_HOST_DIRECTORY_BROWSE_CAPABILITY).unwrap(),
                CapabilityId::new(C2_STANDALONE_WORKSPACE_LIFECYCLE_CAPABILITY).unwrap(),
                CapabilityId::new(C2_PROVIDER_SESSION_REFERENCE_INDEX_CAPABILITY).unwrap(),
                CapabilityId::new(C2_AGENT_PROGRESS_SNAPSHOT_CAPABILITY).unwrap(),
                CapabilityId::new(C2_SESSION_TASK_CORRELATION_CAPABILITY).unwrap(),
                CapabilityId::new(C2_CONTROL_DETAIL_CAPABILITY).unwrap(),
                CapabilityId::new(C2_DELIVERY_BUNDLE_V2_STAGE_COMMIT_CAPABILITY).unwrap(),
                CapabilityId::new(C2_HARNESS_MCP_READ_PROXY_CAPABILITY).unwrap(),
                CapabilityId::new(C2_ACP_CONTROL_CAPABILITY).unwrap(),
            ],
        );
        assert_eq!(offer.state_schema, None);
        assert!(matches!(
            validate_selected_compatibility(&offer, None),
            Err(C2ControlError::Protocol(_)),
        ));
    }

    #[tokio::test]
    async fn host_directory_browse_preserves_route_and_requires_negotiated_capability() {
        let request = NodeRequest::BrowseHostDirectories {
            directory: Some(OpaqueHostPath::utf8(r"C:\Users".to_owned()).unwrap()),
            after: Some(OpaqueHostPath::utf8(r"C:\Users\Public".to_owned()).unwrap()),
        };
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            all_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: request.clone(),
            reply: reply_tx,
        }).await.unwrap();
        let Some(C2ClientFrame::Request(envelope)) = writer_rx.recv().await else {
            panic!("host directory browse was not written");
        };
        assert_eq!(envelope.request.route, route());
        assert_eq!(envelope.request.request, request);
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: envelope.request_id,
                result: Ok(RoutedNodeResponse {
                    node_id: route().node_id,
                    incarnation_id: route().expected_incarnation_id,
                    response: Ok(C2NodeResponse::HostDirectoriesBrowsed {
                        listing: HostDirectoryListing {
                            directory: None,
                            parent: None,
                            entries: vec![HostDirectoryEntry {
                                path: OpaqueHostPath::utf8(r"C:\Users".to_owned()).unwrap(),
                                display_name: "Users".to_owned(),
                                is_link: false,
                            }],
                            next_after: None,
                            incomplete: false,
                        },
                    }),
                }),
            },
        ))).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse {
                response: Ok(C2NodeResponse::HostDirectoriesBrowsed { .. }),
                ..
            })
        ));
        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();

        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            no_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        let Some(C2ClientFrame::Request(envelope)) = writer_rx.recv().await else {
            panic!("snapshot request was not written");
        };
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: envelope.request_id,
                result: Ok(RoutedNodeResponse {
                    node_id: route().node_id,
                    incarnation_id: route().expected_incarnation_id,
                    response: Ok(C2NodeResponse::HostDirectoriesBrowsed {
                        listing: HostDirectoryListing {
                            directory: None,
                            parent: None,
                            entries: Vec::new(),
                            next_after: None,
                            incomplete: false,
                        },
                    }),
                }),
            },
        ))).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == HOST_DIRECTORY_BROWSE_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();

        let mut capabilities = all_path_capabilities();
        capabilities.host_directory_browse = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(
                &NodeRequest::BrowseHostDirectories {
                    directory: None,
                    after: None,
                },
                capabilities,
            ),
            Err(C2ControlError::Protocol(ref message))
                if message == HOST_DIRECTORY_BROWSE_NOT_NEGOTIATED
        ));
        assert!(node_request_has_unix_bytes(&NodeRequest::BrowseHostDirectories {
            directory: None,
            after: Some(OpaqueHostPath::unix_bytes(b"/srv/\xff".to_vec()).unwrap()),
        }));
    }

    #[tokio::test]
    async fn standalone_workspace_roundtrip_preserves_route_snapshot_and_fails_closed() {
        let request = NodeRequest::CreateStandaloneWorkspace {
            workspace_id: WorkspaceId::new("standalone").unwrap(),
            root: OpaqueHostPath::utf8(r"C:\standalone".to_owned()).unwrap(),
            initial_branch: Some("main".to_owned()),
        };
        assert_eq!(control_request_deadline(&request), Duration::from_secs(245));
        let mut legacy = all_path_capabilities();
        legacy.standalone_workspace_lifecycle = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, legacy),
            Err(C2ControlError::Protocol(ref message))
                if message == STANDALONE_WORKSPACE_LIFECYCLE_NOT_NEGOTIATED
        ));

        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            all_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: request.clone(),
            reply: reply_tx,
        }).await.unwrap();
        let Some(C2ClientFrame::Request(envelope)) = writer_rx.recv().await else {
            panic!("standalone workspace request was not written");
        };
        assert_eq!(envelope.request.route, route());
        assert_eq!(envelope.request.request, request);
        let workspace = C2WorkspaceSnapshot {
            workspace_id: WorkspaceId::new("standalone").unwrap(),
            canonical_root: OpaqueHostPath::utf8(r"C:\standalone".to_owned()).unwrap(),
            sessions: Vec::new(),
            worktree_service_mode: Some(WorktreeServiceMode::Manual),
            managed_worktree_profiles: None,
        };
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: envelope.request_id,
                result: Ok(RoutedNodeResponse {
                    node_id: route().node_id,
                    incarnation_id: route().expected_incarnation_id,
                    response: Ok(C2NodeResponse::StandaloneWorkspaceCreated {
                        workspace: workspace.clone(),
                    }),
                }),
            },
        ))).await.unwrap();
        let response = timeout(Duration::from_secs(1), reply_rx)
            .await.unwrap().unwrap().unwrap();
        assert_eq!(response.node_id, route().node_id);
        assert_eq!(response.incarnation_id, route().expected_incarnation_id);
        assert!(matches!(
            response.response,
            Ok(C2NodeResponse::StandaloneWorkspaceCreated {
                workspace: observed,
            }) if observed == workspace
        ));
        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();

        assert!(routed_response_requires_standalone_workspace_lifecycle(
            &RoutedNodeResponse {
                node_id: route().node_id,
                incarnation_id: route().expected_incarnation_id,
                response: Ok(C2NodeResponse::StandaloneWorkspaceCreated { workspace }),
            },
        ));
    }

    #[tokio::test]
    async fn control_owner_fails_closed_on_unnegotiated_terminal_frame_event() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, mut events_rx) = mpsc::channel(1);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            no_path_capabilities(),
        ));

        incoming_tx.send(terminal_event(8)).await.unwrap();
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();

        assert!(commands_tx.is_closed());
        assert!(matches!(
            events_rx.try_recv(),
            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected),
        ));
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_terminal_frame_resync_reply_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, mut events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            no_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Resync { after_sequence: 7 },
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));

        incoming_tx.send(terminal_resync_reply(1, 8)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == TERMINAL_FRAME_EVENTS_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();

        assert!(commands_tx.is_closed());
        assert!(matches!(
            events_rx.try_recv(),
            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected),
        ));
    }

    #[tokio::test]
    async fn saturated_terminal_event_channel_does_not_block_following_reply() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, mut events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            all_path_capabilities(),
        ));

        incoming_tx.send(terminal_event(8)).await.unwrap();
        incoming_tx.send(terminal_event(9)).await.unwrap();
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(1)).await.unwrap();

        let response = timeout(Duration::from_secs(1), reply_rx)
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert!(matches!(response.response, Ok(C2NodeResponse::Accepted)));
        assert!(matches!(
            events_rx.try_recv(),
            Ok(RoutedNodeEvent {
                event: C2NodeEvent::TerminalFrame { .. },
                ..
            })
        ));

        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn harness_mcp_burst_uses_dedicated_128_event_capacity_without_cursor_change() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(EVENT_CAPACITY);
        let (harness_tx, mut harness_rx) = mpsc::channel(HARNESS_MCP_EVENT_CAPACITY);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(HARNESS_MCP_EVENT_CAPACITY + 1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.harness_mcp_read_proxy = true;
        let deadline_unix_ms = current_unix_ms().unwrap() + 3_000;
        for call_index in 0..=HARNESS_MCP_EVENT_CAPACITY {
            incoming_tx.send(harness_mcp_event(call_index, deadline_unix_ms)).await.unwrap();
        }
        // The overflow (129th) event must not tear the connection down --
        // it is dropped with a warning and the connection stays open. This
        // test ends the connection by closing the reader input instead, so
        // the 128-capacity assertion below still exercises a genuinely
        // full (not closed) harness-MCP receiver.
        drop(incoming_tx);
        let owner = tokio::spawn(control_owner(
            commands_rx,
            EventDelivery { regular: events_tx, harness_mcp: harness_tx },
            topology_tx,
            writer_tx,
            incoming_rx,
            capabilities,
        ));

        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.is_closed());
        let mut delivered = Vec::new();
        while let Ok(event) = harness_rx.try_recv() {
            delivered.push(event);
        }
        assert_eq!(delivered.len(), HARNESS_MCP_EVENT_CAPACITY);
        assert!(delivered.iter().all(|event| {
            event.cursor.sequence == 41
                && matches!(event.event, C2NodeEvent::HarnessMcpReadCall { .. })
        }));
    }

    /// Item 3's direct proof: a sustained full-channel backlog -- a "gap
    /// episode" of dropped events -- logs exactly once at its start, not
    /// once per dropped event, and a fresh episode after recovery earns its
    /// own log line again. `control_owner`'s `harness_mcp`/terminal-frame
    /// arms both gate their warn through this exact function.
    #[test]
    fn channel_full_drop_logs_once_per_gap_episode_then_resets() {
        let mut logged = false;
        assert!(
            should_log_channel_full_drop(&mut logged),
            "the first drop of a new episode must log",
        );
        assert!(
            !should_log_channel_full_drop(&mut logged),
            "a second drop in the same still-unreported episode must not log again",
        );
        assert!(
            !should_log_channel_full_drop(&mut logged),
            "nor a third, no matter how long the episode runs",
        );

        reset_channel_full_drop_episode(&mut logged);

        assert!(
            should_log_channel_full_drop(&mut logged),
            "a drop in a NEW episode, after the channel recovered, logs again",
        );
    }

    #[tokio::test]
    async fn harness_mcp_full_channel_drops_the_event_but_a_pending_request_still_completes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(EVENT_CAPACITY);
        // Capacity 1, never drained: the very first harness-MCP event
        // fills it, so the second one exercises the `Full` branch.
        let (harness_tx, _harness_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(4);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.harness_mcp_read_proxy = true;
        let deadline_unix_ms = current_unix_ms().unwrap() + 3_000;
        let owner = tokio::spawn(control_owner(
            commands_rx,
            EventDelivery { regular: events_tx, harness_mcp: harness_tx },
            topology_tx,
            writer_tx,
            incoming_rx,
            capabilities,
        ));

        incoming_tx.send(harness_mcp_event(1, deadline_unix_ms)).await.unwrap();
        incoming_tx.send(harness_mcp_event(2, deadline_unix_ms)).await.unwrap();
        tokio::task::yield_now().await;

        // A request enqueued (and answered) AFTER the channel-full event
        // still gets served: the owner loop never broke.
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(1)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse { response: Ok(C2NodeResponse::Accepted), .. })
        ));

        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn harness_mcp_dropped_receiver_stops_that_class_but_a_pending_request_still_completes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(EVENT_CAPACITY);
        let (harness_tx, harness_rx) = mpsc::channel(HARNESS_MCP_EVENT_CAPACITY);
        // Nobody ever subscribes to harness-MCP events on this connection.
        drop(harness_rx);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(4);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.harness_mcp_read_proxy = true;
        let deadline_unix_ms = current_unix_ms().unwrap() + 3_000;
        let owner = tokio::spawn(control_owner(
            commands_rx,
            EventDelivery { regular: events_tx, harness_mcp: harness_tx },
            topology_tx,
            writer_tx,
            incoming_rx,
            capabilities,
        ));

        incoming_tx.send(harness_mcp_event(1, deadline_unix_ms)).await.unwrap();

        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(1)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse { response: Ok(C2NodeResponse::Accepted), .. })
        ));

        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn genuine_pipe_loss_answers_every_pending_request_with_the_named_reason() {
        let (commands_tx, commands_rx) = mpsc::channel(4);
        let (events_tx, _events_rx) = mpsc::channel(EVENT_CAPACITY);
        let (writer_tx, mut writer_rx) = mpsc::channel(4);
        let (incoming_tx, incoming_rx) = mpsc::channel(4);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, all_path_capabilities(),
        ));

        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));

        // The reader half signals a genuine pipe loss -- no reply for the
        // pending Snapshot request will ever arrive.
        incoming_tx.send(OwnerInput::Closed).await.unwrap();
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();

        assert!(matches!(
            reply_rx.await.unwrap(),
            Err(C2ControlError::ConnectionLost { reason: C2ConnectionLossReason::PipeClosed }),
        ));
    }

    #[tokio::test]
    async fn harness_mcp_mixed_pending_burst_preserves_durable_event_without_cursor_pollution_or_replay() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, events_rx) = mpsc::channel(EVENT_CAPACITY);
        let (harness_tx, harness_rx) = mpsc::channel(HARNESS_MCP_EVENT_CAPACITY);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(2);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.harness_mcp_read_proxy = true;
        let deadline_unix_ms = current_unix_ms().unwrap() + 3_000;
        incoming_tx.send(harness_mcp_event(1, deadline_unix_ms)).await.unwrap();
        incoming_tx.send(ordinary_state_event(42)).await.unwrap();
        let owner = tokio::spawn(control_owner(
            commands_rx,
            EventDelivery { regular: events_tx, harness_mcp: harness_tx },
            topology_tx,
            writer_tx,
            incoming_rx,
            capabilities,
        ));
        let mut receiver = C2EventReceiver {
            events: events_rx,
            harness_mcp_events: harness_rx,
        };
        timeout(Duration::from_secs(1), async {
            while receiver.events.len() != 1 || receiver.harness_mcp_events.len() != 1 {
                tokio::task::yield_now().await;
            }
        }).await.unwrap();

        let transient = receiver.recv().await.unwrap();
        assert_eq!(transient.cursor.sequence, 41);
        assert!(matches!(transient.event, C2NodeEvent::HarnessMcpReadCall { .. }));
        let durable = receiver.recv().await.unwrap();
        assert_eq!(durable.cursor.sequence, 42);
        assert!(matches!(
            durable.event,
            C2NodeEvent::WorkspaceRemoved { ref workspace_id }
                if workspace_id.as_str() == "retired"
        ));
        assert!(matches!(
            receiver.harness_mcp_events.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
        ));
        assert!(matches!(
            receiver.events.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
        ));

        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    /// `split_mut` (the Stage 1 accessor the reconnect bridge's
    /// `pump_one_connection` needs to race both channels in one `select!`)
    /// must hand out two receivers that each drain only their own class,
    /// regardless of enqueue order -- proving the two channels are
    /// genuinely independent streams, not merely `recv()`'s internal
    /// biased-merge order.
    #[tokio::test]
    async fn split_mut_drains_each_channel_independently_of_enqueue_order() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, events_rx) = mpsc::channel(EVENT_CAPACITY);
        let (harness_tx, harness_rx) = mpsc::channel(HARNESS_MCP_EVENT_CAPACITY);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(2);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.harness_mcp_read_proxy = true;
        let deadline_unix_ms = current_unix_ms().unwrap() + 3_000;
        // The regular event is enqueued FIRST -- if the two split
        // receivers were secretly still merged, a naive FIFO reading of the
        // harness_mcp side first would see nothing ready and this test
        // would hang (caught by the outer `timeout`) or, worse, silently
        // return the regular event.
        incoming_tx.send(ordinary_state_event(42)).await.unwrap();
        incoming_tx.send(harness_mcp_event(1, deadline_unix_ms)).await.unwrap();
        let owner = tokio::spawn(control_owner(
            commands_rx,
            EventDelivery { regular: events_tx, harness_mcp: harness_tx },
            topology_tx,
            writer_tx,
            incoming_rx,
            capabilities,
        ));
        let mut receiver = C2EventReceiver {
            events: events_rx,
            harness_mcp_events: harness_rx,
        };
        timeout(Duration::from_secs(1), async {
            while receiver.events.len() != 1 || receiver.harness_mcp_events.len() != 1 {
                tokio::task::yield_now().await;
            }
        }).await.unwrap();

        let (regular_events, harness_mcp_events) = receiver.split_mut();

        let mcp_event = timeout(Duration::from_secs(1), harness_mcp_events.recv())
            .await
            .expect("the harness_mcp receiver must not block on the regular channel")
            .unwrap();
        assert!(matches!(mcp_event.event, C2NodeEvent::HarnessMcpReadCall { .. }));

        let regular_event = timeout(Duration::from_secs(1), regular_events.recv())
            .await
            .expect("the regular receiver must not block on the harness_mcp channel")
            .unwrap();
        assert_eq!(regular_event.cursor.sequence, 42);
        assert!(matches!(
            regular_event.event,
            C2NodeEvent::WorkspaceRemoved { ref workspace_id }
                if workspace_id.as_str() == "retired"
        ));

        drop(commands_tx);
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[test]
    fn c2_repository_path_gate_covers_entries_status_and_previous_path() {
        let utf8 = || repository_path("src/main.rs");
        let tagged = || unix_repository_path(b"src/\xff");
        for response in [
            repository_inspection_response(tagged(), utf8(), None),
            repository_inspection_response(utf8(), tagged(), None),
            repository_inspection_response(utf8(), utf8(), Some(tagged())),
        ] {
            assert!(c2_node_response_has_unix_repository_path(&response));
        }
        assert!(!c2_node_response_has_unix_repository_path(
            &repository_inspection_response(utf8(), utf8(), Some(utf8())),
        ));
    }

    #[test]
    fn c2_control_client_rejects_selection_outside_offer() {
        let offer = client_compatibility_offer().unwrap();
        let selected = NegotiatedC2ControlCompatibility {
            build_stamp: format!("{}-tampered", offer.build_stamp),
            capabilities: offer.capabilities.clone(),
            host: HostDescriptor {
                operating_system: OperatingSystemId::new("windows").unwrap(),
                architecture: ArchitectureId::new("x86_64").unwrap(),
            },
            path_semantics: PathSemantics {
                style: PathStyle::Windows,
                encoding: PathEncoding::Utf8,
            },
        };

        assert!(matches!(
            validate_selected_compatibility(&offer, Some(&selected)),
            Err(C2ControlError::BuildStampMismatch { .. }),
        ));
    }

    #[test]
    fn c2_control_client_requires_authenticated_compatibility_metadata() {
        let offer = client_compatibility_offer().unwrap();
        let selected = NegotiatedC2ControlCompatibility {
            build_stamp: offer.build_stamp.clone(),
            capabilities: vec![
                CapabilityId::new(C2_WORKSPACE_FILE_READ_CAPABILITY).unwrap(),
            ],
            host: HostDescriptor {
                operating_system: OperatingSystemId::new("windows").unwrap(),
                architecture: ArchitectureId::new("x86_64").unwrap(),
            },
            path_semantics: PathSemantics {
                style: PathStyle::Windows,
                encoding: PathEncoding::Utf8,
            },
        };

        assert!(matches!(
            validate_selected_compatibility(&offer, Some(&selected)),
            Err(C2ControlError::Protocol(_)),
        ));
    }

    #[tokio::test]
    async fn negotiating_c2_client_rejects_hmac_valid_legacy_challenge_before_authenticate() {
        let endpoint = unique_control_endpoint();
        let token = "strict-negotiation-token";
        let mut server = ServerOptions::new()
            .first_pipe_instance(true)
            .create(&endpoint)
            .unwrap();
        let server_task = tokio::spawn({
            let token = token.to_owned();
            async move {
                server.connect().await.unwrap();
                let frame = read_json_frame_limited_body_timeout::<_, C2ClientFrame>(
                    &mut server,
                    MAX_C2_AUTH_FRAME_BYTES,
                    Duration::from_secs(1),
                )
                .await
                .unwrap();
                let C2ClientFrame::Hello(hello) = frame else {
                    panic!("client did not send hello");
                };
                assert!(hello.compatibility.is_some());
                let server_nonce = [7; C2_AUTH_NONCE_BYTES];
                let server_proof = c2_proof(
                    &token,
                    C2AuthDirection::Server,
                    &hello.client_nonce,
                    &server_nonce,
                    None,
                )
                .unwrap();
                write_json_frame_limited(
                    &mut server,
                    &C2ServerFrame::Challenge(C2ServerChallenge {
                        build_stamp: BUILD_STAMP.to_owned(),
                        server_nonce,
                        server_proof,
                        compatibility: None,
                    }),
                    MAX_C2_AUTH_FRAME_BYTES,
                )
                .await
                .unwrap();
                timeout(
                    Duration::from_secs(1),
                    read_json_frame_limited_body_timeout::<_, C2ClientFrame>(
                        &mut server,
                        MAX_C2_AUTH_FRAME_BYTES,
                        Duration::from_secs(1),
                    ),
                )
                .await
            }
        });

        let error = match connect_local(&endpoint, token).await {
            Ok(_) => panic!("negotiated client accepted a legacy challenge"),
            Err(error) => error,
        };
        assert!(matches!(
            error,
            C2ControlError::Protocol(ref message)
                if message.contains("omitted the required authenticated compatibility selection")
        ));
        match server_task.await.unwrap() {
            Ok(Ok(frame)) => panic!("client sent a post-challenge frame: {frame:?}"),
            Ok(Err(_)) | Err(_) => {}
        }
    }

    #[test]
    fn c2_control_bound_proof_rejects_tampered_selection() {
        let offer = client_compatibility_offer().unwrap();
        let selected = NegotiatedC2ControlCompatibility {
            build_stamp: offer.build_stamp.clone(),
            capabilities: offer.capabilities.clone(),
            host: HostDescriptor {
                operating_system: OperatingSystemId::new("windows").unwrap(),
                architecture: ArchitectureId::new("x86_64").unwrap(),
            },
            path_semantics: PathSemantics {
                style: PathStyle::Windows,
                encoding: PathEncoding::Utf8,
            },
        };
        let expected = c2_proof(
            "test-token",
            C2AuthDirection::Server,
            &[1; C2_AUTH_NONCE_BYTES],
            &[2; C2_AUTH_NONCE_BYTES],
            Some((&offer, &selected)),
        ).unwrap();
        let mut tampered = selected;
        tampered.path_semantics.style = PathStyle::Posix;
        let received = c2_proof(
            "test-token",
            C2AuthDirection::Server,
            &[1; C2_AUTH_NONCE_BYTES],
            &[2; C2_AUTH_NONCE_BYTES],
            Some((&offer, &tampered)),
        ).unwrap();

        assert!(!proofs_match(&received, &expected));
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_unix_path_before_write() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (_incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let request = NodeRequest::RegisterWorkspace {
            workspace_id: WorkspaceId::new("foreign").unwrap(),
            root: unix_path(),
        };
        assert!(reject_unnegotiated_outbound_path(&request, all_path_capabilities()).is_ok());
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx
            .send(ControlCommand { route: route(), request, reply: reply_tx })
            .await
            .unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == OPAQUE_UNIX_PATH_NOT_NEGOTIATED
        ));
        assert!(writer_rx.try_recv().is_err());
        drop(commands_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_open_provider_spawn_before_write() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (_incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Spawn {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                provider: hatchery_c2_protocol::AgentId::new("third-party-agent").unwrap(),
                mode: SessionMode::Pty,
                terminal_size: TerminalSize { rows: 40, columns: 120 },
                initial_prompt: None,
            },
            reply: reply_tx,
        }).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == OPEN_PROVIDER_ID_NOT_NEGOTIATED
        ));
        assert!(writer_rx.try_recv().is_err());
        drop(commands_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[test]
    fn spawn_profile_revision_is_advertised_and_rejected_before_relay_on_all_spawn_paths() {
        let offer = client_compatibility_offer().unwrap();
        assert!(offer.capabilities.iter().any(|capability| {
            capability.as_str() == C2_SPAWN_PROFILE_REVISION_CAPABILITY
        }));

        let spec = spawn_spec("node-a");
        let reservation_id = gate4agent_node_protocol::HarnessMcpReservationId::new(
            format!("hmcpres_{}", "a".repeat(24)),
        ).unwrap();
        let activation_digest = gate4agent_node_protocol::HarnessMcpActivationDigest::new(
            format!("sha256:{}", "b".repeat(64)),
        ).unwrap();
        let deadline_unix_ms = current_unix_ms().unwrap().saturating_add(60_000);
        let requests = [
            NodeRequest::SpawnSpec { spec: spec.clone() },
            NodeRequest::SpawnManagedWorktree {
                request: gate4agent_node_protocol::ManagedWorktreeSpawnRequest {
                    spawn_spec: spec.clone(),
                    worktree_profile_id:
                        gate4agent_node_protocol::WorktreeProfileId::new("review").unwrap(),
                },
            },
            NodeRequest::ArmHarnessMcpReservation {
                reservation_id: reservation_id.clone(),
                activation_digest: activation_digest.clone(),
                spawn_spec: spec.clone(),
                launch: gate4agent_node_protocol::HarnessMcpLaunchV1 {
                    server_name: "fixture-harness".to_owned(),
                    args: vec!["--session-proxy".to_owned()],
                    endpoint_env: "FIXTURE_SESSION_ENDPOINT".to_owned(),
                    token_env: "FIXTURE_SESSION_TOKEN".to_owned(),
                    program_env: "FIXTURE_MCP_PROGRAM".to_owned(),
                    trace: None,
                    scrub_env: Vec::new(),
                },
                expires_at_unix_ms: deadline_unix_ms,
            },
            NodeRequest::SpawnSpecWithHarnessMcp {
                reservation_id,
                activation_digest,
                spec,
                deadline_unix_ms,
            },
        ];
        let mut capabilities = all_path_capabilities();
        capabilities.harness_mcp_read_proxy = true;
        capabilities.spawn_profile_revision = false;
        for request in &requests {
            assert!(matches!(
                reject_unnegotiated_outbound_path(request, capabilities),
                Err(C2ControlError::Protocol(ref message))
                    if message == SPAWN_PROFILE_REVISION_NOT_NEGOTIATED
            ));
        }

        let response = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: gate4agent_node_protocol::NodeIncarnationId::from_bytes([1; 16]),
            response: Ok(C2NodeResponse::SpawnSpecAccepted {
                receipt: spawn_receipt(),
            }),
        };
        assert!(routed_response_requires_spawn_profile_revision(&response));
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_spawn_spec_before_write() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (_incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.spawn_spec_defaults_overrides = false;
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx,
            capabilities,
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::SpawnSpec { spec: spawn_spec("node-a") },
            reply: reply_tx,
        }).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == SPAWN_SPEC_NOT_NEGOTIATED
        ));
        assert!(writer_rx.try_recv().is_err());
        drop(commands_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_spawn_spec_reply_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.spawn_spec_defaults_overrides = false;
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, capabilities,
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: C2RequestId(1),
                result: Ok(RoutedNodeResponse {
                    node_id: NodeId::new("node-a").unwrap(),
                    incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                    response: Ok(C2NodeResponse::SpawnSpecAccepted {
                        receipt: spawn_receipt(),
                    }),
                }),
            },
        ))).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == SPAWN_SPEC_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: oneshot::channel().0,
        }).await.is_err());
    }

    #[tokio::test]
    async fn control_owner_conditionally_gates_worktree_spawn_before_write() {
        let (commands_tx, commands_rx) = mpsc::channel(2);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(2);
        let (_incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.worktree_selection = false;
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, capabilities,
        ));

        let (legacy_reply_tx, _legacy_reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::SpawnSpec { spec: spawn_spec("node-a") },
            reply: legacy_reply_tx,
        }).await.unwrap();
        assert!(matches!(
            writer_rx.recv().await,
            Some(C2ClientFrame::Request(_)),
        ));

        let mut spec = spawn_spec("node-a");
        spec.target.worktree_id = Some(WorkspaceId::new("review-tree").unwrap());
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::SpawnSpec { spec },
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKTREE_SELECTION_NOT_NEGOTIATED
        ));
        assert!(writer_rx.try_recv().is_err());
        drop(commands_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_worktree_receipt_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.worktree_selection = false;
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, capabilities,
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        let mut receipt = spawn_receipt();
        receipt.target.worktree_id = Some(WorkspaceId::new("review-tree").unwrap());
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: C2RequestId(1),
                result: Ok(RoutedNodeResponse {
                    node_id: NodeId::new("node-a").unwrap(),
                    incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                    response: Ok(C2NodeResponse::SpawnSpecAccepted { receipt }),
                }),
            },
        ))).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKTREE_SELECTION_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_environment_receipt_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let mut capabilities = all_path_capabilities();
        capabilities.child_environment_profile = false;
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, capabilities,
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        let mut receipt = spawn_receipt();
        receipt.environment_profile = Some(
            hatchery_c2_protocol::ResolvedEnvironmentProfileReceipt {
                profile_id: hatchery_c2_protocol::SpawnEnvironmentProfileId::new(
                    "local-default",
                )
                .unwrap(),
                profile_revision:
                    hatchery_c2_protocol::SpawnEnvironmentProfileRevision::new(
                        "local-default.r1",
                    )
                    .unwrap(),
                    network_allowlist: None,
                    browser_profile_id: None,
            },
        );
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: C2RequestId(1),
                result: Ok(RoutedNodeResponse {
                    node_id: NodeId::new("node-a").unwrap(),
                    incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                    response: Ok(C2NodeResponse::SpawnSpecAccepted { receipt }),
                }),
            },
        ))).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_open_provider_reply_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Reply(
            hatchery_c2_protocol::C2ReplyEnvelope {
                request_id: C2RequestId(1),
                result: Ok(RoutedNodeResponse {
                    node_id: NodeId::new("node-a").unwrap(),
                    incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
                    response: Ok(C2NodeResponse::Snapshot {
                        event_sequence: 1,
                        controller: None,
                        snapshot: hatchery_c2_protocol::C2NodeSnapshot {
                            node_id: NodeId::new("node-a").unwrap(),
                            enabled_providers: vec![
                                hatchery_c2_protocol::AgentId::new("third-party-agent").unwrap(),
                            ],
                            provider_runtime_statuses: Default::default(),
                            workspaces: Vec::new(),
                            agent_progress: Vec::new(),
                            session_records: Vec::new(),
                            managed_worktrees: Vec::new(),
                            launch_inventory: None,
                        },
                    }),
                }),
            },
        ))).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == OPEN_PROVIDER_ID_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.is_closed());
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_workspace_file_read_before_write_and_stays_healthy() {
        let (commands_tx, commands_rx) = mpsc::channel(2);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let (read_tx, read_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::ReadWorkspaceFile {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                path: repository_path("src/lib.rs"),
            },
            reply: read_tx,
        }).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), read_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKSPACE_FILE_READ_NOT_NEGOTIATED
        ));
        assert!(writer_rx.try_recv().is_err());

        let (snapshot_tx, snapshot_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: snapshot_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(1)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), snapshot_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse { response: Ok(C2NodeResponse::Accepted), .. })
        ));

        drop(commands_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_workspace_entry_create_before_write_and_stays_healthy() {
        let (commands_tx, commands_rx) = mpsc::channel(2);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            no_path_capabilities(),
        ));
        let request = NodeRequest::CreateWorkspaceDirectory {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            path: repository_path("src/new"),
        };
        assert_eq!(control_request_deadline(&request), Duration::from_secs(20));
        assert!(
            control_request_deadline(&request)
                > CONTROLLER_ACQUIRE_RELAY_DEADLINE
                    .saturating_add(WORKSPACE_ENTRY_CREATE_RELAY_DEADLINE),
        );
        let (create_tx, create_rx) = oneshot::channel();
        commands_tx
            .send(ControlCommand {
                route: route(),
                request,
                reply: create_tx,
            })
            .await
            .unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), create_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKSPACE_ENTRY_CREATE_NOT_NEGOTIATED
        ));
        assert!(writer_rx.try_recv().is_err());

        let (snapshot_tx, snapshot_rx) = oneshot::channel();
        commands_tx
            .send(ControlCommand {
                route: route(),
                request: NodeRequest::Snapshot,
                reply: snapshot_tx,
            })
            .await
            .unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(1)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), snapshot_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse {
                response: Ok(C2NodeResponse::Accepted),
                ..
            })
        ));

        drop(commands_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[test]
    fn workspace_entry_create_requires_both_create_and_repository_capabilities() {
        let request = NodeRequest::CreateWorkspaceFile {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            path: unix_repository_path(b"src/\xff"),
        };
        let mut capabilities = all_path_capabilities();
        capabilities.workspace_entry_create = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKSPACE_ENTRY_CREATE_NOT_NEGOTIATED
        ));
        capabilities.workspace_entry_create = true;
        capabilities.repository_paths = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == REPOSITORY_PATH_NOT_NEGOTIATED
        ));
        capabilities.repository_paths = true;
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_ok());
    }

    #[test]
    fn provider_session_reference_index_request_is_c2_capability_gated() {
        let request = NodeRequest::IndexProviderSession {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            provider: hatchery_c2_protocol::AgentId::new("claude").unwrap(),
            identity: ProviderSessionIdentity {
                key: ProviderSessionKey::SessionId,
                id: "provider-session-42".to_owned(),
                transcript_path: None,
            },
            display_name: "release shepherd".to_owned(),
        };
        let mut capabilities = all_path_capabilities();
        capabilities.provider_session_reference_index = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == PROVIDER_SESSION_REFERENCE_INDEX_NOT_NEGOTIATED
        ));
        assert!(reject_unnegotiated_outbound_path(
            &request,
            all_path_capabilities(),
        )
        .is_ok());
    }

    #[test]
    fn tagged_workspace_file_read_requires_both_file_and_repository_capabilities() {
        let request = NodeRequest::ReadWorkspaceFile {
            workspace_id: WorkspaceId::new("primary").unwrap(),
            path: unix_repository_path(b"src/\xff"),
        };
        let missing_repository = NegotiatedPathCapabilities {
            opaque_host_paths: true,
            repository_paths: false,
            workspace_file_read: true,
            host_directory_browse: true,
            standalone_workspace_lifecycle: true,
            provider_session_reference_index: true,
            provider_ids_open: true,
            spawn_spec_defaults_overrides: true,
            worktree_selection: true,
            managed_worktree_lifecycle: true,
            child_environment_profile: true,
            session_bundle_materialization: true,
            history_context_pack: true,
            native_session_catalog: true,
            native_session_catalog_paging: true,
            native_session_index: true,
            native_session_preview: true,
            terminal_frame_events: true,
            ..Default::default()
        };
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, missing_repository),
            Err(C2ControlError::Protocol(ref message))
                if message == REPOSITORY_PATH_NOT_NEGOTIATED
        ));
        let missing_file_read = NegotiatedPathCapabilities {
            opaque_host_paths: true,
            repository_paths: true,
            workspace_file_read: false,
            host_directory_browse: true,
            standalone_workspace_lifecycle: true,
            provider_session_reference_index: true,
            provider_ids_open: true,
            spawn_spec_defaults_overrides: true,
            worktree_selection: true,
            managed_worktree_lifecycle: true,
            child_environment_profile: true,
            session_bundle_materialization: true,
            history_context_pack: true,
            native_session_catalog: true,
            native_session_catalog_paging: true,
            native_session_index: true,
            native_session_preview: true,
            terminal_frame_events: true,
            ..Default::default()
        };
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, missing_file_read),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKSPACE_FILE_READ_NOT_NEGOTIATED
        ));
        assert!(reject_unnegotiated_outbound_path(
            &request,
            all_path_capabilities(),
        ).is_ok());
    }

    #[test]
    fn workspace_file_read_response_is_gated_and_path_checked() {
        let response = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::WorkspaceFileRead {
                file: WorkspaceFileRead {
                    workspace_id: WorkspaceId::new("primary").unwrap(),
                    path: unix_repository_path(b"src/\xff"),
                    content: WorkspaceFileContent::NonUtf8 { byte_len: 3 },
                    revision: None,
                },
            }),
        };
        assert!(routed_response_requires_workspace_file_read(&response));
        assert!(routed_response_has_unix_repository_path(&response));
        assert!(!routed_response_has_unix_bytes(&response));
    }

    #[test]
    fn workspace_entry_create_response_correlation_rejects_content_and_kind_mismatches() {
        let workspace_id = WorkspaceId::new("primary").unwrap();
        let file_path = repository_path("src/new.rs");
        let request = NodeRequest::CreateWorkspaceFile {
            workspace_id: workspace_id.clone(),
            path: file_path.clone(),
        };
        let response = Ok(RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::WorkspaceFileCreated {
                file: WorkspaceFileRead {
                    workspace_id: workspace_id.clone(),
                    path: file_path,
                    content: WorkspaceFileContent::Utf8 {
                        text: String::new(),
                        byte_len: 0,
                    },
                    revision: Some(
                        gate4agent_node_protocol::WorkspaceFileRevision::new(
                            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
                                .to_owned(),
                        )
                        .unwrap(),
                    ),
                },
            }),
        });
        assert!(validate_workspace_content_response(&request, &response).is_ok());

        let wrong_content = Ok(RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::WorkspaceFileCreated {
                file: WorkspaceFileRead {
                    workspace_id: workspace_id.clone(),
                    path: repository_path("src/new.rs"),
                    content: WorkspaceFileContent::Utf8 {
                        text: "unexpected".to_owned(),
                        byte_len: 10,
                    },
                    revision: None,
                },
            }),
        });
        assert!(validate_workspace_content_response(&request, &wrong_content).is_err());

        let directory_path = repository_path("src/new");
        let directory_request = NodeRequest::CreateWorkspaceDirectory {
            workspace_id: workspace_id.clone(),
            path: directory_path.clone(),
        };
        let wrong_kind = Ok(RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::WorkspaceDirectoryCreated {
                workspace_id,
                entry: WorkspaceEntry {
                    relative_path: directory_path,
                    kind: WorkspaceEntryKind::File,
                },
            }),
        });
        assert!(validate_workspace_content_response(&directory_request, &wrong_kind).is_err());
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_unix_path_reply_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx
            .send(ControlCommand {
                route: route(),
                request: NodeRequest::Snapshot,
                reply: reply_tx,
            })
            .await
            .unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(unix_path_reply(1)).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == OPAQUE_UNIX_PATH_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.is_closed());
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_tagged_repository_path_reply_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx
            .send(ControlCommand {
                route: route(),
                request: NodeRequest::InspectWorkspace {
                    workspace_id: WorkspaceId::new("foreign").unwrap(),
                },
                reply: reply_tx,
            })
            .await
            .unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(unix_repository_path_reply(1)).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message == REPOSITORY_PATH_NOT_NEGOTIATED
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.is_closed());
    }

    #[tokio::test]
    async fn control_owner_rejects_unnegotiated_unix_path_event_and_closes() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, mut events_rx) = mpsc::channel(1);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        incoming_tx.send(unix_path_event(1)).await.unwrap();

        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(events_rx.try_recv().is_err());
        assert!(commands_tx.is_closed());
    }

    #[tokio::test]
    async fn late_reply_after_timed_waiter_does_not_desynchronize_following_request() {
        let (commands_tx, commands_rx) = mpsc::channel(2);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(2);
        let (incoming_tx, incoming_rx) = mpsc::channel(2);
        let (topology_tx, _topology_rx) = watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));

        let (expired_tx, expired_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: expired_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        drop(expired_rx);
        incoming_tx.send(reply(1)).await.unwrap();

        let (live_tx, live_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: live_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(2)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), live_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse { response: Ok(hatchery_c2_protocol::C2NodeResponse::Accepted), .. })
        ));

        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn dropped_event_receiver_closes_control_owner() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, events_rx) = mpsc::channel(EVENT_CAPACITY);
        drop(events_rx);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) = watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));

        incoming_tx.send(event(1)).await.unwrap();
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.is_closed());
    }

    #[tokio::test]
    async fn full_regular_channel_drops_event_and_still_serves_following_reply() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(EVENT_CAPACITY);
        let mut seq = 1u64;
        while events_tx.try_send(routed_event(seq)).is_ok() {
            seq += 1;
        }
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(2);
        let (topology_tx, _topology_rx) = watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));

        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        // The `regular` channel is full but its receiver is alive -- this
        // event must be dropped, not tear the connection down.
        incoming_tx.send(event(3)).await.unwrap();
        incoming_tx.send(reply(1)).await.unwrap();

        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse {
                response: Ok(hatchery_c2_protocol::C2NodeResponse::Accepted),
                ..
            })
        ));
        assert!(!commands_tx.is_closed());

        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn active_event_consumer_drains_regular_burst_and_owner_serves_following_reply() {
        let burst_len = EVENT_CAPACITY + 2;
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, mut events_rx) =
            mpsc::channel::<RoutedNodeEvent>(EVENT_CAPACITY);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(burst_len + 1);
        let (topology_tx, _topology_rx) = watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));

        for sequence in 1..=burst_len as u64 {
            incoming_tx.send(event(sequence)).await.unwrap();
        }
        let collector = tokio::spawn(async move {
            let mut sequences = Vec::with_capacity(burst_len);
            while sequences.len() < burst_len {
                let event = events_rx.recv().await.expect("event stream must stay open");
                sequences.push(event.cursor.sequence);
            }
            sequences
        });
        tokio::task::yield_now().await;
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));

        let sequences = timeout(Duration::from_secs(1), collector)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(sequences, (1..=burst_len as u64).collect::<Vec<_>>());

        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx.send(ControlCommand {
            route: route(),
            request: NodeRequest::Snapshot,
            reply: reply_tx,
        }).await.unwrap();
        assert!(matches!(writer_rx.recv().await, Some(C2ClientFrame::Request(_))));
        incoming_tx.send(reply(1)).await.unwrap();
        assert!(matches!(
            timeout(Duration::from_secs(1), reply_rx).await.unwrap().unwrap(),
            Ok(RoutedNodeResponse {
                response: Ok(C2NodeResponse::Accepted),
                ..
            })
        ));

        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn c2_client_topology_replaces_old_incarnation_status() {
        use hatchery_c2_protocol::{C2TopologyNode, NodeTransportState};

        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, mut events_rx) = mpsc::channel(1);
        let (writer_tx, _writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let old_runtime_status = hatchery_c2_protocol::ProviderRuntimeStatuses::new([
            hatchery_c2_protocol::ProviderRuntimeStatus::raw_passthrough(
                hatchery_c2_protocol::AgentId::new("claude").unwrap(),
                Some(hatchery_c2_protocol::ProviderRuntimeVersion::new("1.0.0").unwrap()),
            ),
        ])
        .unwrap();
        let offline = Arc::new(C2Topology { nodes: vec![C2TopologyNode {
            node_id: NodeId::new("node-a").unwrap(),
            endpoint: r"\\.\pipe\node-a".to_owned(),
            relay_route: hatchery_c2_protocol::C2RelayRoute::LocalIpc,
            transport: NodeTransportState::Offline,
            current_incarnation_id: Some(NodeIncarnationId::from_bytes([8; 16])),
            provider_contracts: Vec::new(),
            provider_adapter_contracts: Vec::new(),
            provider_runtime_statuses: old_runtime_status,
        }] });
        let (topology_tx, mut topology_rx) = watch::channel(offline);
        let owner = tokio::spawn(control_owner(
            commands_rx, events_tx, topology_tx, writer_tx, incoming_rx, no_path_capabilities(),
        ));
        let incarnation_id = NodeIncarnationId::from_bytes([9; 16]);
        incoming_tx.send(OwnerInput::Frame(C2ServerFrame::Topology(C2Topology {
            nodes: vec![C2TopologyNode {
                node_id: NodeId::new("node-a").unwrap(),
                endpoint: r"\\.\pipe\node-a".to_owned(),
                relay_route: hatchery_c2_protocol::C2RelayRoute::LocalIpc,
                transport: NodeTransportState::Online,
                current_incarnation_id: Some(incarnation_id),
                provider_contracts: Vec::new(),
                provider_adapter_contracts: Vec::new(),
                provider_runtime_statuses: hatchery_c2_protocol::ProviderRuntimeStatuses::new([
                    hatchery_c2_protocol::ProviderRuntimeStatus::raw_passthrough(
                        hatchery_c2_protocol::AgentId::new("codex").unwrap(),
                        Some(hatchery_c2_protocol::ProviderRuntimeVersion::new("2.0.0").unwrap()),
                    ),
                ])
                .unwrap(),
            }],
        }))).await.unwrap();

        timeout(Duration::from_secs(1), topology_rx.changed()).await.unwrap().unwrap();
        assert_eq!(topology_rx.borrow().nodes[0].current_incarnation_id, Some(incarnation_id));
        let topology = topology_rx.borrow().clone();
        let statuses = &topology.nodes[0].provider_runtime_statuses;
        assert_eq!(statuses.as_slice().len(), 1);
        assert_eq!(
            statuses.as_slice()[0].provider(),
            &hatchery_c2_protocol::AgentId::new("codex").unwrap(),
        );
        assert_eq!(statuses.as_slice()[0].version().unwrap().as_str(), "2.0.0");
        assert!(events_rx.try_recv().is_err());
        drop(incoming_tx);
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
        assert!(commands_tx.is_closed());
    }

    #[test]
    fn child_environment_profile_set_and_inherit_require_capability_before_write() {
        let mut explicit = spawn_spec("node-a");
        explicit.overrides.environment_profile_id =
            gate4agent_node_protocol::SpawnOverride::Set {
                value: hatchery_c2_protocol::SpawnEnvironmentProfileId::new(
                    "local-default",
                )
                .unwrap(),
            };
        let mut capabilities = all_path_capabilities();
        capabilities.child_environment_profile = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(
                &NodeRequest::SpawnSpec { spec: explicit },
                capabilities,
            ),
            Err(C2ControlError::Protocol(ref message))
                if message == CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED
        ));

        let inherited = spawn_spec("node-a");
        assert!(matches!(
            reject_unnegotiated_outbound_path(
                &NodeRequest::SpawnSpec {
                    spec: inherited.clone(),
                },
                capabilities,
            ),
            Err(C2ControlError::Protocol(ref message))
                if message == CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED
        ));
        let managed = NodeRequest::SpawnManagedWorktree {
            request: hatchery_c2_protocol::ManagedWorktreeSpawnRequest {
                spawn_spec: inherited,
                worktree_profile_id:
                    hatchery_c2_protocol::WorktreeProfileId::new("review").unwrap(),
            },
        };
        assert!(matches!(
            reject_unnegotiated_outbound_path(&managed, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == CHILD_ENVIRONMENT_PROFILE_NOT_NEGOTIATED
        ));

        let mut cleared = spawn_spec("node-a");
        cleared.overrides.environment_profile_id =
            gate4agent_node_protocol::SpawnOverride::Clear;
        assert!(reject_unnegotiated_outbound_path(
            &NodeRequest::SpawnSpec { spec: cleared },
            capabilities,
        )
        .is_ok());
    }

    #[test]
    fn session_bundle_inherit_and_set_require_capability_but_clear_does_not() {
        let mut capabilities = all_path_capabilities();
        capabilities.session_bundle_materialization = false;

        let inherited = spawn_spec("node-a");
        assert!(matches!(
            reject_unnegotiated_outbound_path(
                &NodeRequest::SpawnSpec {
                    spec: inherited.clone(),
                },
                capabilities,
            ),
            Err(C2ControlError::Protocol(ref message))
                if message == SESSION_BUNDLE_MATERIALIZATION_NOT_NEGOTIATED
        ));

        let mut explicit = inherited.clone();
        explicit.overrides.bundle_id = gate4agent_node_protocol::SpawnOverride::Set {
            value: hatchery_c2_protocol::SpawnBundleId::new("review-bundle").unwrap(),
        };
        assert!(matches!(
            reject_unnegotiated_outbound_path(
                &NodeRequest::SpawnSpec { spec: explicit },
                capabilities,
            ),
            Err(C2ControlError::Protocol(ref message))
                if message == SESSION_BUNDLE_MATERIALIZATION_NOT_NEGOTIATED
        ));

        let mut cleared = inherited;
        cleared.overrides.bundle_id = gate4agent_node_protocol::SpawnOverride::Clear;
        assert!(reject_unnegotiated_outbound_path(
            &NodeRequest::SpawnSpec { spec: cleared },
            capabilities,
        )
        .is_ok());
    }

    #[test]
    fn managed_worktree_partial_capabilities_fail_closed() {
        let request = NodeRequest::SpawnManagedWorktree {
            request: hatchery_c2_protocol::ManagedWorktreeSpawnRequest {
                spawn_spec: spawn_spec("node-a"),
                worktree_profile_id:
                    hatchery_c2_protocol::WorktreeProfileId::new("review").unwrap(),
            },
        };

        let mut capabilities = all_path_capabilities();
        capabilities.managed_worktree_lifecycle = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == MANAGED_WORKTREE_LIFECYCLE_NOT_NEGOTIATED
        ));

        capabilities.managed_worktree_lifecycle = true;
        capabilities.worktree_selection = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == WORKTREE_SELECTION_NOT_NEGOTIATED
        ));

        capabilities.worktree_selection = true;
        capabilities.spawn_spec_defaults_overrides = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == SPAWN_SPEC_NOT_NEGOTIATED
        ));

        let cleanup = NodeRequest::CleanupManagedWorktree {
            lease_id: hatchery_c2_protocol::ManagedWorktreeLeaseId::new("lease-a").unwrap(),
        };
        capabilities.spawn_spec_defaults_overrides = false;
        capabilities.worktree_selection = true;
        assert!(reject_unnegotiated_outbound_path(&cleanup, capabilities).is_ok());
        capabilities.worktree_selection = false;
        assert!(reject_unnegotiated_outbound_path(&cleanup, capabilities).is_err());

        let event = C2NodeEvent::ManagedWorktreeRemoved {
            lease_id: hatchery_c2_protocol::ManagedWorktreeLeaseId::new("lease-a").unwrap(),
        };
        assert!(c2_node_event_is_managed_worktree(&event));

        let workspace = C2WorkspaceSnapshot {
            workspace_id: WorkspaceId::new("repo").unwrap(),
            canonical_root: OpaqueHostPath::utf8(r"C:\repo".to_owned()).unwrap(),
            sessions: Vec::new(),
            worktree_service_mode: Some(WorktreeServiceMode::Manual),
            managed_worktree_profiles: None,
        };
        assert!(c2_workspace_has_managed_worktree_metadata(&workspace));
        assert!(routed_response_requires_managed_worktree(&RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([1; 16]),
            response: Ok(C2NodeResponse::WorkspaceRegistered {
                workspace: workspace.clone(),
            }),
        }));
        assert!(c2_node_event_is_managed_worktree(
            &C2NodeEvent::WorkspaceAdded { workspace },
        ));

        let legacy = C2WorkspaceSnapshot {
            workspace_id: WorkspaceId::new("legacy").unwrap(),
            canonical_root: OpaqueHostPath::utf8(r"C:\legacy".to_owned()).unwrap(),
            sessions: Vec::new(),
            worktree_service_mode: None,
            managed_worktree_profiles: None,
        };
        assert!(!c2_workspace_has_managed_worktree_metadata(&legacy));
        let profile_only = C2WorkspaceSnapshot {
            managed_worktree_profiles: Some(
                hatchery_c2_protocol::WorktreeProfileInventory {
                    profiles: Vec::new(),
                },
            ),
            ..legacy
        };
        assert!(c2_workspace_has_managed_worktree_metadata(&profile_only));
    }

    #[test]
    fn managed_worktree_v2_is_capability_gated_and_receipt_correlated() {
        let offer = client_compatibility_offer().unwrap();
        assert!(offer.capabilities.iter().any(|capability| {
            capability.as_str() == C2_MANAGED_WORKTREE_SPAWN_V2_CAPABILITY
        }));
        let request = NodeRequest::SpawnManagedWorktreeV2 {
            request: hatchery_c2_protocol::ManagedWorktreeSpawnRequestV2 {
                spawn_spec: spawn_spec("node-a"),
                worktree_profile_id:
                    hatchery_c2_protocol::WorktreeProfileId::new("review").unwrap(),
                expected_profile_revision:
                    hatchery_c2_protocol::WorktreeProfileRevision::new("wr1").unwrap(),
            },
        };
        let mut capabilities = all_path_capabilities();
        capabilities.managed_worktree_spawn_v2 = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == MANAGED_WORKTREE_SPAWN_V2_NOT_NEGOTIATED
        ));

        let managed_workspace_id = WorkspaceId::new("managed-v2").unwrap();
        let mut spawn = spawn_receipt();
        spawn.profile_revision = SpawnProfileRevision::new("default.r1").unwrap();
        spawn.target.worktree_id = Some(managed_workspace_id.clone());
        spawn.session.workspace_id = managed_workspace_id.clone();
        let receipt = hatchery_c2_protocol::ManagedWorktreeSpawnReceipt {
            spawn,
            lease: hatchery_c2_protocol::ManagedWorktreeLeaseSnapshot {
                lease_id:
                    hatchery_c2_protocol::ManagedWorktreeLeaseId::new("lease-v2").unwrap(),
                source_workspace_id: WorkspaceId::new("primary").unwrap(),
                workspace_id: managed_workspace_id,
                profile_id:
                    hatchery_c2_protocol::WorktreeProfileId::new("review").unwrap(),
                profile_revision:
                    hatchery_c2_protocol::WorktreeProfileRevision::new("wr1").unwrap(),
                retention: hatchery_c2_protocol::ManagedWorktreeRetention::RemoveWhenReleased,
                state: hatchery_c2_protocol::ManagedWorktreeLeaseState::InUse,
                active_session_count: 1,
                managed_record_count: 1,
                cleanup_failure: None,
                created_at_unix_ms: 1,
                updated_at_unix_ms: 2,
            },
        };
        let routed = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: receipt.spawn.incarnation_id,
            response: Ok(C2NodeResponse::ManagedWorktreeSpawnAccepted {
                receipt: receipt.clone(),
            }),
        };
        assert!(validate_managed_worktree_spawn_v2_response(
            &request,
            &Ok(routed),
        )
        .is_ok());

        let mut mismatched = receipt;
        mismatched.lease.profile_revision =
            hatchery_c2_protocol::WorktreeProfileRevision::new("wr2").unwrap();
        let routed = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: mismatched.spawn.incarnation_id,
            response: Ok(C2NodeResponse::ManagedWorktreeSpawnAccepted {
                receipt: mismatched,
            }),
        };
        assert!(validate_managed_worktree_spawn_v2_response(
            &request,
            &Ok(routed),
        )
        .is_err());
    }

    #[test]
    fn history_context_pack_requests_and_failures_are_capability_bound() {
        let request = NodeRequest::DiscoverHistory {
            session: spawn_receipt().session,
            limit: 1,
        };
        let mut capabilities = all_path_capabilities();
        capabilities.history_context_pack = false;
        assert!(matches!(
            reject_unnegotiated_outbound_path(&request, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == HISTORY_CONTEXT_PACK_NOT_NEGOTIATED
        ));
        capabilities.history_context_pack = true;
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_ok());

        let invalid = NodeRequest::DiscoverHistory {
            session: spawn_receipt().session,
            limit: 0,
        };
        assert!(matches!(
            reject_unnegotiated_outbound_path(&invalid, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == INVALID_HISTORY_CONTEXT_PACK_REQUEST
        ));

        capabilities.provider_ids_open = false;
        let forget = NodeRequest::ForgetContextPack {
            context_id: gate4agent_node_protocol::SpawnContextId::new("context-a").unwrap(),
        };
        assert!(matches!(
            reject_unnegotiated_outbound_path(&forget, capabilities),
            Err(C2ControlError::Protocol(ref message))
                if message == OPEN_PROVIDER_ID_NOT_NEGOTIATED
        ));
        capabilities.provider_ids_open = true;
        assert!(reject_unnegotiated_outbound_path(&forget, capabilities).is_ok());

        let failure = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Err(hatchery_c2_protocol::C2NodeFailure {
                code: gate4agent_node_protocol::NodeFailureCode::UnknownContextPack,
                message: "context pack unavailable".to_owned(),
            }),
        };
        assert!(routed_response_requires_history_context_pack(&failure));
        let response = RoutedNodeResponse {
            node_id: NodeId::new("node-a").unwrap(),
            incarnation_id: NodeIncarnationId::from_bytes([7; 16]),
            response: Ok(C2NodeResponse::HistoryLoaded {
                session: spawn_receipt().session,
                session_id: "session-1".to_owned(),
                message_count: 2,
                completed_turn_count: None,
            }),
        };
        assert!(routed_response_requires_history_context_pack(&response));
    }

    #[test]
    fn native_session_catalog_request_is_capability_bound_and_read_only() {
        let route = gate4agent_node_protocol::NativeSessionCatalogRoute::workspace(
            gate4agent_node_protocol::WorkspaceId::new("primary").unwrap(),
            hatchery_c2_protocol::AgentId::new("codex").unwrap(),
        );
        let request = NodeRequest::CatalogNativeSessions {
            route: route.clone(),
            limit: 8,
        };
        let mut capabilities = all_path_capabilities();
        capabilities.native_session_catalog = false;
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_err());
        capabilities.native_session_catalog = true;
        assert!(reject_unnegotiated_outbound_path(&request, capabilities).is_ok());
        assert_eq!(control_request_deadline(&request), Duration::from_secs(40));

        let page = NodeRequest::PageNativeSessions {
            route,
            window: hatchery_c2_protocol::NativeSessionCatalogWindow::Older,
            catalog_revision: 7,
            recent_cutoff_unix_ms: 8,
            after_selection_id: Some("hist_selection_1".to_owned()),
            limit: 8,
        };
        capabilities.native_session_catalog_paging = false;
        assert!(reject_unnegotiated_outbound_path(&page, capabilities).is_err());
        capabilities.native_session_catalog_paging = true;
        assert!(reject_unnegotiated_outbound_path(&page, capabilities).is_ok());
        assert_eq!(control_request_deadline(&page), Duration::from_secs(40));
    }

    #[tokio::test]
    async fn control_owner_rejects_reply_route_or_incarnation_mismatch() {
        let (commands_tx, commands_rx) = mpsc::channel(1);
        let (events_tx, _events_rx) = mpsc::channel(1);
        let (writer_tx, mut writer_rx) = mpsc::channel(1);
        let (incoming_tx, incoming_rx) = mpsc::channel(1);
        let (topology_tx, _topology_rx) =
            watch::channel(Arc::new(C2Topology { nodes: Vec::new() }));
        let owner = tokio::spawn(control_owner(
            commands_rx,
            events_tx,
            topology_tx,
            writer_tx,
            incoming_rx,
            all_path_capabilities(),
        ));
        let (reply_tx, reply_rx) = oneshot::channel();
        commands_tx
            .send(ControlCommand {
                route: route(),
                request: NodeRequest::Snapshot,
                reply: reply_tx,
            })
            .await
            .unwrap();
        let C2ClientFrame::Request(request) = writer_rx.recv().await.unwrap() else {
            panic!("owner did not emit a request");
        };
        incoming_tx
            .send(OwnerInput::Frame(C2ServerFrame::Reply(
                hatchery_c2_protocol::C2ReplyEnvelope {
                    request_id: request.request_id,
                    result: Ok(RoutedNodeResponse {
                        node_id: request.request.route.node_id,
                        incarnation_id: NodeIncarnationId::from_bytes([8; 16]),
                        response: Ok(C2NodeResponse::Accepted),
                    }),
                },
            )))
            .await
            .unwrap();
        assert!(matches!(
            reply_rx.await.unwrap(),
            Err(C2ControlError::Protocol(ref message))
                if message.contains("route or node incarnation")
        ));
        timeout(Duration::from_secs(1), owner).await.unwrap().unwrap();
    }
}

#[cfg(all(test, unix))]
mod unix_endpoint_tests {
    use super::*;

    #[test]
    fn unix_endpoint_remains_valid() {
        assert!(validate_endpoint("/tmp/gate4agent-c2.sock").is_ok());
    }

    #[test]
    fn raw_public_tcp_endpoint_is_still_rejected_by_the_unix_validator() {
        for endpoint in ["203.0.113.8:51820", "10.88.1.2:27441", "tcp://10.88.1.2:27441"] {
            assert!(
                matches!(validate_endpoint(endpoint), Err(C2ControlError::InvalidEndpoint)),
                "{endpoint} must stay InvalidEndpoint",
            );
        }
    }
}