bamboo-subagent 2026.10.21

//! WebSocket transport (design §6): full-duplex parent↔child link.
//!
//! - Child side: [`WsServer`] accepts a connection and drives a [`ChildExecutor`] — `Run` starts a
//!   task whose events stream out as `ChildFrame::EventBatch`, then a `ChildFrame::Terminal`; `Cancel`
//!   trips the run's token (out-of-band, never queued behind events).
//! - Parent side: [`ChildClient`] sends [`ParentFrame`]s and reads [`ChildFrame`]s.

use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::{Arc, Mutex};

use futures_util::stream::{SplitSink, SplitStream};
use futures_util::{SinkExt, StreamExt};
use rustls::pki_types::pem::{Error as PemError, PemObject};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{mpsc, oneshot};
use tokio_rustls::TlsAcceptor;
use tokio_tungstenite::tungstenite::handshake::server::{
    Callback, ErrorResponse, Request, Response,
};
use tokio_tungstenite::tungstenite::http::StatusCode;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{accept_async, accept_hdr_async, connect_async, MaybeTlsStream};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::executor::{
    ChildExecutor, EventSink, ExecutorControl, HostBridge, HostRequestKind, SteerInbox,
    SteerMessage,
};
use crate::poison::PoisonRecover;
use crate::proto::{
    ActorEventBatch, ActorEventBatcher, ActorEventQos, ChildFrame, ParentFrame, RunSpec,
    TerminalStatus,
};

/// Direct actor links use separate bounded data/control queues. A slow parent
/// can lose lossy batches (visible as a sequence gap), but it cannot grow the
/// worker heap without bound or strand approvals behind token traffic.
const DIRECT_EVENT_QUEUE_CAPACITY: usize = 64;
const DIRECT_CONTROL_QUEUE_CAPACITY: usize = 32;
const ACTOR_EVENT_BATCH_LATENCY: std::time::Duration = std::time::Duration::from_millis(20);
const DIRECT_RUN_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);

#[cfg(test)]
mod history_delivery_tests;

#[cfg(test)]
mod tls_tests;

/// Every connection/run owns its spawned helpers. A cancelled owner must not
/// detach children merely because Tokio's plain JoinHandle was dropped.
struct OwnedTask<T>(Option<tokio::task::JoinHandle<T>>);

impl<T> OwnedTask<T> {
    fn new(task: tokio::task::JoinHandle<T>) -> Self {
        Self(Some(task))
    }

    async fn join(&mut self) -> Result<T, tokio::task::JoinError> {
        let result = self.0.as_mut().expect("owned task present").await;
        self.0.take();
        result
    }

    fn abort(&self) {
        if let Some(task) = &self.0 {
            task.abort();
        }
    }
}

impl<T> Drop for OwnedTask<T> {
    fn drop(&mut self) {
        self.abort();
    }
}

struct ActiveRun {
    cancel: CancellationToken,
    task: OwnedTask<()>,
}

impl ActiveRun {
    async fn stop(mut self, terminal_tx: Option<&mpsc::Sender<ChildFrame>>) {
        self.cancel.cancel();
        if tokio::time::timeout(DIRECT_RUN_DRAIN_TIMEOUT, self.task.join())
            .await
            .is_err()
        {
            self.task.abort();
            let _ = tokio::time::timeout(DIRECT_RUN_DRAIN_TIMEOUT, self.task.join()).await;
            if let Some(terminal_tx) = terminal_tx {
                let _ = tokio::time::timeout(
                    DIRECT_RUN_DRAIN_TIMEOUT,
                    terminal_tx.send(ChildFrame::Terminal {
                        status: crate::proto::TerminalStatus::Cancelled,
                        result: None,
                        error: None,
                        final_event_watermark: None,
                        transcript: Vec::new(),
                    }),
                )
                .await;
            }
        }
    }
}

impl Drop for ActiveRun {
    fn drop(&mut self) {
        self.cancel.cancel();
    }
}

/// Pending host-callback replies, keyed by request id, shared between the
/// per-run pump (which inserts) and the connection read loop (which resolves
/// them on [`ParentFrame::ApprovalReply`]).
type PendingReplies = Arc<Mutex<HashMap<String, oneshot::Sender<serde_json::Value>>>>;

struct PendingReplyOwner {
    pending: PendingReplies,
    ids: Vec<String>,
}

impl Drop for PendingReplyOwner {
    fn drop(&mut self) {
        let mut pending = self.pending.lock().recover_poison();
        for id in &self.ids {
            pending.remove(id);
        }
    }
}

#[derive(Debug)]
pub enum TransportError {
    Io(std::io::Error),
    Ws(Box<tokio_tungstenite::tungstenite::Error>),
    Decode(serde_json::Error),
    Protocol(String),
    /// TLS configuration / handshake failure (bad cert/key, provider, etc.).
    Tls(String),
}

impl std::fmt::Display for TransportError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io(error) => write!(formatter, "io: {error}"),
            Self::Ws(error) => write!(formatter, "ws: {error}"),
            Self::Decode(error) => write!(formatter, "decode: {error}"),
            Self::Protocol(message) => write!(formatter, "protocol: {message}"),
            Self::Tls(message) => write!(formatter, "tls: {message}"),
        }
    }
}

impl std::error::Error for TransportError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Ws(error) => Some(error.as_ref()),
            Self::Decode(error) => Some(error),
            Self::Protocol(_) | Self::Tls(_) => None,
        }
    }
}

impl From<std::io::Error> for TransportError {
    fn from(error: std::io::Error) -> Self {
        Self::Io(error)
    }
}

impl From<tokio_tungstenite::tungstenite::Error> for TransportError {
    fn from(error: tokio_tungstenite::tungstenite::Error) -> Self {
        Self::Ws(Box::new(error))
    }
}

impl From<serde_json::Error> for TransportError {
    fn from(error: serde_json::Error) -> Self {
        Self::Decode(error)
    }
}

pub type TransportResult<T> = Result<T, TransportError>;

// ---- child side --------------------------------------------------------------

/// A WebSocket server an actor runs to receive work.
///
/// Three knobs, all optional and independent (remote-actor-plan §3.2 / §3.4):
/// - `listener`/`addr`: where it binds (`bind_loopback` for local default,
///   `bind`/`bind_tls` for a remotely-reachable worker on `0.0.0.0:PORT`).
/// - `tls`: when `Some`, every accepted connection is wrapped in TLS before the
///   WS upgrade and the advertised endpoint is `wss://`.
/// - `expected_token`: when `Some`, the WS upgrade is rejected (401) unless the
///   client presents `Authorization: Bearer <token>`. None ⇒ accept any
///   (loopback default, zero regression).
pub struct WsServer {
    listener: TcpListener,
    addr: SocketAddr,
    tls: Option<TlsAcceptor>,
    expected_token: Option<String>,
}

impl WsServer {
    /// Bind an arbitrary address with NO TLS and NO token. Pass `0.0.0.0:PORT`
    /// for a remotely-reachable worker/broker; [`bind_loopback`](Self::bind_loopback)
    /// is the local default. Use [`bind_with_token`](Self::bind_with_token) /
    /// [`bind_tls`](Self::bind_tls) to require auth.
    pub async fn bind(addr: SocketAddr) -> TransportResult<Self> {
        Self::bind_with_token(addr, None).await
    }

    /// Bind plaintext but optionally require a bearer token on the WS upgrade.
    ///
    /// A token WITHOUT TLS is insecure (the bearer crosses the wire in the
    /// clear) — acceptable only on a trusted/loopback link (e.g. the
    /// deterministic `remote_connect_e2e` test). For real remote exposure use
    /// [`bind_tls`](Self::bind_tls).
    pub async fn bind_with_token(
        addr: SocketAddr,
        expected_token: Option<String>,
    ) -> TransportResult<Self> {
        let listener = TcpListener::bind(addr).await?;
        let addr = listener.local_addr()?;
        Ok(Self {
            listener,
            addr,
            tls: None,
            expected_token,
        })
    }

    /// Bind with TLS termination (`wss://`) and an optional bearer token.
    ///
    /// Builds a `tokio_rustls::TlsAcceptor` from `cert_file`/`key_file` (PEM)
    /// against an explicit `ring` provider — the same approach as the server's
    /// HTTP TLS face (`bamboo-server`'s `build_rustls_config`) so the workspace
    /// resolves a single crypto provider and never hits the rustls 0.23
    /// no-default-provider panic.
    ///
    /// **Fail-fast:** a missing/unreadable/unparseable cert or key returns
    /// `Err` — it never silently downgrades to plaintext.
    pub async fn bind_tls(
        addr: SocketAddr,
        cert_file: &Path,
        key_file: &Path,
        expected_token: Option<String>,
    ) -> TransportResult<Self> {
        let server_config =
            build_server_config(cert_file, key_file).map_err(TransportError::Tls)?;
        let acceptor = TlsAcceptor::from(Arc::new(server_config));
        let listener = TcpListener::bind(addr).await?;
        let addr = listener.local_addr()?;
        Ok(Self {
            listener,
            addr,
            tls: Some(acceptor),
            expected_token,
        })
    }

    /// Bind `127.0.0.1:0` (ephemeral port) — the local default (no TLS, no token).
    pub async fn bind_loopback() -> TransportResult<Self> {
        Self::bind((std::net::Ipv4Addr::LOCALHOST, 0).into()).await
    }

    pub fn local_addr(&self) -> SocketAddr {
        self.addr
    }

    /// The reachable endpoint to advertise: `wss://` when TLS is active,
    /// `ws://` otherwise.
    pub fn ws_endpoint(&self) -> String {
        let scheme = if self.tls.is_some() { "wss" } else { "ws" };
        format!("{scheme}://{}", self.addr)
    }

    /// Serve exactly one connection (owned child / demo), then return.
    pub async fn serve_one<E: ChildExecutor + ?Sized>(
        self,
        executor: Arc<E>,
    ) -> TransportResult<()> {
        let (stream, _) = self.listener.accept().await?;
        accept_and_handle(stream, &self.tls, &self.expected_token, executor).await
    }

    /// Serve exactly one connection, but give up if no client connects within
    /// `accept_timeout` (orphan defense: a spawned worker whose parent died
    /// before connecting must not linger forever). An ACTIVE connection is
    /// never cut short — the timeout only guards the accept.
    pub async fn serve_one_with_accept_timeout<E: ChildExecutor + ?Sized>(
        self,
        executor: Arc<E>,
        accept_timeout: std::time::Duration,
    ) -> TransportResult<()> {
        let (stream, _) = tokio::time::timeout(accept_timeout, self.listener.accept())
            .await
            .map_err(|_| {
                TransportError::Protocol(format!(
                    "no connection within {accept_timeout:?}; exiting"
                ))
            })??;
        accept_and_handle(stream, &self.tls, &self.expected_token, executor).await
    }

    /// Serve connection-after-connection for a **reusable** actor, one at a time.
    ///
    /// Unlike [`serve`], connections are handled serially (the next accept waits
    /// for the current run to finish) so a pooled worker never runs two
    /// assignments at once — preserving per-run context isolation. Unlike
    /// [`serve_one`], the worker survives a connection close and is ready for the
    /// next assignment. Exits when no new connection arrives within `idle_timeout`
    /// (so an unreused worker reclaims itself instead of lingering forever). An
    /// active run is never cut short — the timeout only guards the idle accept.
    pub async fn serve_reusable_with_idle_timeout<E: ChildExecutor + ?Sized>(
        self,
        executor: Arc<E>,
        idle_timeout: std::time::Duration,
    ) -> TransportResult<()> {
        loop {
            let accept = tokio::time::timeout(idle_timeout, self.listener.accept()).await;
            let (stream, _) = match accept {
                Ok(res) => res?,
                Err(_) => return Ok(()), // idle: reclaim self, clean exit
            };
            // Handle this assignment to completion before accepting the next.
            let _ =
                accept_and_handle(stream, &self.tls, &self.expected_token, executor.clone()).await;
        }
    }

    /// Serve connections forever (long-running service agent).
    pub async fn serve<E: ChildExecutor + ?Sized>(self, executor: Arc<E>) -> TransportResult<()> {
        let mut connections = tokio::task::JoinSet::new();
        loop {
            let (stream, _) = tokio::select! {
                connection = connections.join_next(), if !connections.is_empty() => {
                    let _ = connection;
                    continue;
                }
                accepted = self.listener.accept() => accepted?,
            };
            let exec = executor.clone();
            // The TLS handshake + auth callback run inside the spawned task (not
            // on the accept loop) so a slow/hostile client cannot stall the
            // server from accepting other connections. Clones are cheap:
            // `TlsAcceptor` is `Arc`-backed, the token is a short `String`.
            let tls = self.tls.clone();
            let token = self.expected_token.clone();
            connections.spawn(async move {
                let _ = accept_and_handle(stream, &tls, &token, exec).await;
            });
        }
    }
}

/// Apply TLS (if configured) + the bearer-auth WS upgrade to a freshly-accepted
/// `TcpStream`, then drive the connection. Plaintext + no-token reduces to the
/// historical `accept_async(stream)` path (zero regression).
async fn accept_and_handle<E: ChildExecutor + ?Sized>(
    stream: TcpStream,
    tls: &Option<TlsAcceptor>,
    expected_token: &Option<String>,
    executor: Arc<E>,
) -> TransportResult<()> {
    match tls {
        Some(acceptor) => {
            // Terminate TLS before the WS upgrade so the bearer + frames ride an
            // encrypted link.
            let tls_stream = acceptor
                .accept(stream)
                .await
                .map_err(|e| TransportError::Tls(format!("accept handshake: {e}")))?;
            let ws = ws_upgrade(tls_stream, expected_token).await?;
            handle_conn(ws, executor).await
        }
        None => {
            let ws = ws_upgrade(stream, expected_token).await?;
            handle_conn(ws, executor).await
        }
    }
}

/// Perform the WS upgrade over an arbitrary stream, validating the bearer token
/// at the upgrade layer when one is expected. When `expected_token` is `None`
/// this is exactly `accept_async` (any client accepted) — zero regression.
async fn ws_upgrade<S>(
    stream: S,
    expected_token: &Option<String>,
) -> TransportResult<tokio_tungstenite::WebSocketStream<S>>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    match expected_token {
        None => Ok(accept_async(stream).await?),
        Some(token) => {
            let callback = BearerCallback {
                expected: token.clone(),
            };
            Ok(accept_hdr_async(stream, callback).await?)
        }
    }
}

/// WS-upgrade callback that enforces `Authorization: Bearer <token>`. Rejecting
/// here means the WebSocket never establishes — the wire protocol
/// (`ParentFrame`/`ChildFrame`) stays byte-identical; auth lives entirely at the
/// HTTP upgrade (remote-actor-plan §6.3). The token is never logged.
struct BearerCallback {
    expected: String,
}

impl Callback for BearerCallback {
    fn on_request(self, request: &Request, response: Response) -> Result<Response, ErrorResponse> {
        let presented = request
            .headers()
            .get(tokio_tungstenite::tungstenite::http::header::AUTHORIZATION)
            .and_then(|v| v.to_str().ok());
        let ok = matches!(presented, Some(h) if bearer_matches(h, &self.expected));
        if ok {
            Ok(response)
        } else {
            let err = ErrorResponse::new(Some("unauthorized: bad or missing bearer token".into()));
            let (mut parts, body) = err.into_parts();
            parts.status = StatusCode::UNAUTHORIZED;
            Err(ErrorResponse::from_parts(parts, body))
        }
    }
}

/// Check that an `Authorization` header is `Bearer <expected>`, comparing the
/// token in constant time so a network attacker cannot recover it byte-by-byte
/// via response timing (mirrors `bamboo-server`'s `constant_time_eq` convention
/// for credential comparison). The `Bearer ` prefix is non-secret, so the
/// `strip_prefix` short-circuit is fine.
fn bearer_matches(header: &str, expected: &str) -> bool {
    match header.strip_prefix("Bearer ") {
        Some(t) => constant_time_eq(t.as_bytes(), expected.as_bytes()),
        None => false,
    }
}

/// Constant-time byte comparison: returns early only on a length mismatch
/// (length is not secret), then folds all bytes so the time does not depend on
/// where a mismatch occurs.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Build a rustls [`rustls::ServerConfig`] from PEM cert/key files against an
/// explicit `ring` provider (mirrors `bamboo-server`'s `build_rustls_config`).
///
/// `pub` (not crate-private) so a sibling network-facing crate — e.g.
/// `bamboo-broker`'s own WS listener (#48) — can build a `TlsAcceptor` from
/// the exact same, already-hardened cert/key-loading logic instead of
/// duplicating it. Keeps the workspace's "single crypto provider" invariant
/// (see this function's own `ring` provider construction below) enforced in
/// one place.
pub fn build_server_config(
    cert_file: &Path,
    key_file: &Path,
) -> Result<rustls::ServerConfig, String> {
    use std::fs::File;

    let cert_path = cert_file.display();
    let key_path = key_file.display();

    let cf = File::open(cert_file).map_err(|e| format!("open cert_file '{cert_path}': {e}"))?;
    let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_reader_iter(cf)
        .collect::<Result<_, _>>()
        .map_err(|e| format!("parse cert_file '{cert_path}': {e}"))?;
    if certs.is_empty() {
        return Err(format!(
            "no certificates in cert_file '{cert_path}' (expected PEM CERTIFICATE blocks)"
        ));
    }

    let kf = File::open(key_file).map_err(|e| format!("open key_file '{key_path}': {e}"))?;
    let key = match PrivateKeyDer::from_pem_reader(kf) {
        Ok(k) => k,
        Err(PemError::NoItemsFound) => {
            return Err(format!(
                "no private key in key_file '{key_path}' (expected PKCS#8/RSA/SEC1)"
            ))
        }
        Err(e) => return Err(format!("parse key_file '{key_path}': {e}")),
    };

    let provider = Arc::new(rustls::crypto::ring::default_provider());
    rustls::ServerConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .map_err(|e| format!("protocol versions: {e}"))?
        .with_no_client_auth()
        .with_single_cert(certs, key)
        .map_err(|e| {
            format!("rustls rejected cert/key (cert '{cert_path}', key '{key_path}'): {e}")
        })
}

/// Build a rustls [`rustls::ClientConfig`] that trusts exactly the certificate(s)
/// in `cert_file` (PEM) as roots — for a parent connecting over `wss://` to a
/// worker with a self-signed cert under single-direction TLS (remote-actor-plan
/// §7 Q2: "P1 先单向"). The worker's cert must carry a SAN matching the endpoint
/// host (e.g. `IP:127.0.0.1`) for rustls to accept it. Uses the explicit `ring`
/// provider to match the rest of the workspace. For CA-signed worker certs use
/// [`ChildClient::connect_with_auth`] (default webpki roots) instead.
pub fn client_config_trusting_cert(cert_file: &Path) -> Result<rustls::ClientConfig, String> {
    use std::fs::File;

    let cert_path = cert_file.display();
    let cf = File::open(cert_file).map_err(|e| format!("open cert_file '{cert_path}': {e}"))?;
    let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_reader_iter(cf)
        .collect::<Result<_, _>>()
        .map_err(|e| format!("parse cert_file '{cert_path}': {e}"))?;
    if certs.is_empty() {
        return Err(format!(
            "no certificates in cert_file '{cert_path}' (expected PEM CERTIFICATE blocks)"
        ));
    }

    let mut roots = rustls::RootCertStore::empty();
    for cert in certs {
        roots
            .add(cert)
            .map_err(|e| format!("add trust anchor from '{cert_path}': {e}"))?;
    }

    client_config_with_roots(roots)
}

fn standard_webpki_roots() -> rustls::RootCertStore {
    let mut roots = rustls::RootCertStore::empty();
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
    roots
}

fn client_config_with_roots(roots: rustls::RootCertStore) -> Result<rustls::ClientConfig, String> {
    let provider = Arc::new(rustls::crypto::ring::default_provider());
    let cfg = rustls::ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .map_err(|e| format!("protocol versions: {e}"))?
        .with_root_certificates(roots)
        .with_no_client_auth();
    Ok(cfg)
}

/// Drive an already-upgraded WS connection (generic over the underlying stream
/// so it serves both plaintext `TcpStream` and `TlsStream<TcpStream>`).
async fn handle_conn<S, E>(
    ws: tokio_tungstenite::WebSocketStream<S>,
    executor: Arc<E>,
) -> TransportResult<()>
where
    S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
    E: ChildExecutor + ?Sized,
{
    let (ws_tx, mut ws_rx) = ws.split();
    // One writer task owns the sink. Event traffic is bounded and independent
    // from control traffic, so 200 active actors cannot create an unbounded
    // per-connection token queue or starve an approval/cancellation response.
    let (control_tx, control_rx) = mpsc::channel::<ChildFrame>(DIRECT_CONTROL_QUEUE_CAPACITY);
    let (event_tx, event_rx) = mpsc::channel::<ChildFrame>(DIRECT_EVENT_QUEUE_CAPACITY);
    let mut writer = OwnedTask::new(tokio::spawn(writer_task(ws_tx, control_rx, event_rx)));

    // Host-callback replies for gated-tool approvals the active run proxies back
    // to the host over this WS. Shared with each run's pump.
    let pending: PendingReplies = Arc::new(Mutex::new(HashMap::new()));

    let mut active_run: Option<ActiveRun> = None;
    let mut active_steer: Option<mpsc::UnboundedSender<SteerMessage>> = None;
    let result = loop {
        let msg = tokio::select! {
            _ = writer.join() => break Ok(()),
            message = ws_rx.next() => match message {
                Some(Ok(message)) => message,
                Some(Err(error)) => break Err(TransportError::from(error)),
                None => break Ok(()),
            },
        };
        match msg {
            Message::Text(t) => match ParentFrame::from_text(t.as_str()) {
                // Phase 2: the host's decision on a proxied gated-tool approval.
                // Routed through the `pending` correlation map (body =
                // {"approved": bool}); a worker that proxied an `ApprovalRequest`
                // awaits its oneshot here.
                Ok(ParentFrame::ApprovalReply { id, approved }) => {
                    if let Some(reply) = pending.lock().recover_poison().remove(&id) {
                        let _ = reply.send(serde_json::json!({ "approved": approved }));
                    }
                }
                Ok(ParentFrame::OwnedTreeReply { id, page }) => {
                    if let Some(reply) = pending.lock().recover_poison().remove(&id) {
                        let _ = reply.send(serde_json::json!({ "page": page }));
                    }
                }
                Ok(ParentFrame::SubAgentReply { id, result }) => {
                    if let Some(reply) = pending.lock().recover_poison().remove(&id) {
                        let _ = reply.send(result);
                    }
                }
                Ok(ParentFrame::Run(spec)) => {
                    // A new run supersedes any active run: cancel the previous
                    // task *before* starting the new one so the old run stops
                    // emitting events into the shared `out_tx`. Without this the
                    // old task is orphaned and its output interleaves with the
                    // new run's. (The old steer sender is dropped implicitly by
                    // the reassignment of `active_steer` below.)
                    if let Some(prev) = active_run.take() {
                        prev.stop(Some(&event_tx)).await;
                    }
                    let cancel = CancellationToken::new();
                    let (steer_tx, steer_rx) = SteerInbox::channel();
                    active_steer = Some(steer_tx);
                    active_run = Some(start_run(
                        executor.clone(),
                        spec,
                        steer_rx,
                        cancel,
                        control_tx.clone(),
                        event_tx.clone(),
                        pending.clone(),
                    ));
                }
                Ok(ParentFrame::Cancel) => {
                    if let Some(run) = active_run.take() {
                        run.stop(Some(&event_tx)).await;
                    }
                    active_steer = None;
                }
                Ok(ParentFrame::Message { text }) => {
                    // In-band steering: hand to the active run's inbox; the
                    // executor admits it at its next safe point.
                    if let Some(steer) = &active_steer {
                        let _ = steer.send(SteerMessage::Text(text));
                    }
                }
                Ok(ParentFrame::InitialInputRelease { release }) => {
                    if let Some(steer) = &active_steer {
                        let _ = steer.send(SteerMessage::InitialInputRelease(release));
                    }
                }
                Ok(ParentFrame::SessionMessage { delivery }) => {
                    if let Some(steer) = &active_steer {
                        let _ = steer.send(SteerMessage::SessionMessage(Box::new(delivery)));
                    }
                }
                Err(_) => { /* ignore malformed frame */ }
            },
            Message::Close(_) => break Ok(()),
            _ => {}
        }
    };

    // Parent disconnected / closed: cancel any still-active run so its spawned
    // task ends and stops feeding the writer. Without this `writer.await` blocks
    // until the orphaned run finishes on its own.
    if let Some(run) = active_run {
        run.stop(None).await;
    }
    drop(active_steer);
    pending.lock().recover_poison().clear();
    drop(control_tx);
    drop(event_tx);
    if writer.0.is_some() {
        let _ = tokio::time::timeout(DIRECT_RUN_DRAIN_TIMEOUT, writer.join()).await;
    }
    result
}

async fn writer_task<S>(
    mut ws_tx: SplitSink<tokio_tungstenite::WebSocketStream<S>, Message>,
    mut control_rx: mpsc::Receiver<ChildFrame>,
    mut event_rx: mpsc::Receiver<ChildFrame>,
) where
    S: AsyncRead + AsyncWrite + Unpin,
{
    let mut control_open = true;
    let mut events_open = true;
    while control_open || events_open {
        let frame = tokio::select! {
            biased;
            frame = control_rx.recv(), if control_open => match frame {
                Some(frame) => Some(frame),
                None => {
                    control_open = false;
                    None
                }
            },
            frame = event_rx.recv(), if events_open => match frame {
                Some(frame) => Some(frame),
                None => {
                    events_open = false;
                    None
                }
            },
        };
        let Some(frame) = frame else {
            continue;
        };
        if ws_tx.send(Message::text(frame.to_text())).await.is_err() {
            break;
        }
    }
    let _ = ws_tx.close().await;
}

async fn forward_direct_events(
    mut ev_rx: mpsc::Receiver<serde_json::Value>,
    mut batcher: ActorEventBatcher,
    legacy_event_wire: bool,
    event_fwd: mpsc::Sender<ChildFrame>,
) -> Result<Option<crate::ActorEventWatermark>, ()> {
    let mut flush = tokio::time::interval(ACTOR_EVENT_BATCH_LATENCY);
    flush.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    flush.tick().await;
    let mut open = true;
    while open {
        tokio::select! {
            event = ev_rx.recv() => match event {
                Some(event) => {
                    if legacy_event_wire {
                        if event_fwd.send(ChildFrame::Event { event }).await.is_err() {
                            return Err(());
                        }
                        continue;
                    }
                    for batch in batcher.push(event) {
                        if !send_direct_event_batch(&event_fwd, batch).await {
                            return Err(());
                        }
                    }
                }
                None => open = false,
            },
            _ = flush.tick(), if !legacy_event_wire && batcher.has_pending() => {
                if let Some(batch) = batcher.flush() {
                    if !send_direct_event_batch(&event_fwd, batch).await {
                        return Err(());
                    }
                }
            }
        }
    }
    if !legacy_event_wire {
        if let Some(batch) = batcher.flush() {
            if !send_direct_event_batch(&event_fwd, batch).await {
                return Err(());
            }
        }
    }
    Ok(batcher.final_watermark())
}

fn start_run<E: ChildExecutor + ?Sized>(
    executor: Arc<E>,
    spec: RunSpec,
    steer: SteerInbox,
    cancel: CancellationToken,
    control_tx: mpsc::Sender<ChildFrame>,
    event_tx: mpsc::Sender<ChildFrame>,
    pending: PendingReplies,
) -> ActiveRun {
    let (sink, ev_rx, mut control_rx) = EventSink::channel_with_control();
    let legacy_event_wire = spec.execution_epoch == 0;
    let batcher = ActorEventBatcher::for_run(&spec, None, None)
        .with_durable_events(executor.requires_contiguous_events());
    let event_fwd = event_tx.clone();
    // All event batches share one ordered data lane. Durable batches wait for
    // capacity, including the complete trace of a strict history executor.
    // Ordinary lossy data uses `try_send`, making overload observable as a
    // sequence gap. Approval/admission controls remain independent.
    let mut fwd = OwnedTask::new(tokio::spawn(forward_direct_events(
        ev_rx,
        batcher,
        legacy_event_wire,
        event_fwd,
    )));
    let out_control = control_tx.clone();
    let mut control = OwnedTask::new(tokio::spawn(async move {
        while let Some(control) = control_rx.recv().await {
            let frame = match control {
                ExecutorControl::InitialInputReleaseRequest(request) => ChildFrame::Event {
                    event: serde_json::to_value(crate::proto::InitialInputControl::Request {
                        request,
                    })
                    .expect("typed initial control"),
                },
                ExecutorControl::SessionMessageAdmitted(confirmation) => {
                    ChildFrame::SessionMessageAdmitted { confirmation }
                }
            };
            if out_control.send(frame).await.is_err() {
                break;
            }
        }
    }));

    // Host-callback pump: each gated-tool approval the executor proxies back
    // becomes an `ApprovalRequest` on the wire; its reply (an `ApprovalReply`
    // resolved in `handle_conn`) is delivered to the awaiting oneshot via
    // `pending`.
    let (bridge, mut req_rx) = HostBridge::channel();
    let sink = sink.with_host_bridge(bridge);
    let out_req = control_tx.clone();
    let pending_for_pump = pending.clone();
    let pump = OwnedTask::new(tokio::spawn(async move {
        let mut registrations = PendingReplyOwner {
            pending: pending_for_pump.clone(),
            ids: Vec::new(),
        };
        while let Some(req) = req_rx.recv().await {
            // Random + unguessable: the worker's pending map and the host-side
            // pending-approval registry both key on this exact string, so a
            // sequential `sa-{n}` would let an external POST guess a live
            // request_id. A v4 uuid removes that. Keep the `sa-` prefix.
            let id = format!("sa-{}", Uuid::new_v4());
            pending_for_pump
                .lock()
                .recover_poison()
                .insert(id.clone(), req.reply);
            registrations.ids.push(id.clone());
            let frame = match req.kind {
                HostRequestKind::Approval => ChildFrame::ApprovalRequest { id, body: req.body },
                HostRequestKind::OwnedTree => ChildFrame::OwnedTreeRequest {
                    id,
                    cursor: req
                        .body
                        .get("cursor")
                        .and_then(serde_json::Value::as_str)
                        .map(str::to_owned),
                },
                HostRequestKind::SubAgent => ChildFrame::SubAgentRequest {
                    id,
                    tool_call_id: req
                        .body
                        .get("tool_call_id")
                        .and_then(serde_json::Value::as_str)
                        .unwrap_or_default()
                        .to_owned(),
                    args: req.body.get("args").cloned().unwrap_or_default(),
                },
            };
            if out_req.send(frame).await.is_err() {
                break;
            }
        }
    }));

    let run_cancel = cancel.clone();
    let task = OwnedTask::new(tokio::spawn(async move {
        use futures_util::FutureExt;
        let outcome = std::panic::AssertUnwindSafe(executor.run(spec, sink, steer, run_cancel))
            .catch_unwind()
            .await
            .unwrap_or_else(|_| crate::executor::ChildOutcome::error("actor executor panicked"));
        // A failed/panicked forwarder cannot certify complete history or send
        // successful Terminal. The connection close remains retryable.
        let Ok(Ok(final_event_watermark)) = fwd.join().await else {
            pump.abort();
            return;
        };
        let _ = control.join().await; // flush durable-admission confirmations first
        pump.abort(); // no more host callbacks after the run ends
                      // Terminal follows every accepted event batch on the bounded event
                      // lane. Approval/admission control remains independently prioritized.
        let _ = event_tx
            .send(ChildFrame::Terminal {
                status: outcome.status,
                final_event_watermark,
                result: outcome.result,
                error: outcome.error,
                transcript: outcome.transcript,
            })
            .await;
    }));
    ActiveRun { cancel, task }
}

async fn send_direct_event_batch(
    event_tx: &mpsc::Sender<ChildFrame>,
    batch: ActorEventBatch,
) -> bool {
    let frame = ChildFrame::EventBatch { batch };
    if matches!(
        &frame,
        ChildFrame::EventBatch {
            batch: ActorEventBatch {
                qos: ActorEventQos::Durable,
                ..
            }
        }
    ) {
        return event_tx.send(frame).await.is_ok();
    }
    match event_tx.try_send(frame) {
        Ok(()) | Err(mpsc::error::TrySendError::Full(_)) => true,
        Err(mpsc::error::TrySendError::Closed(_)) => false,
    }
}

// ---- parent side -------------------------------------------------------------

type ClientStream = tokio_tungstenite::WebSocketStream<MaybeTlsStream<TcpStream>>;

/// Parent-side connection to a child actor.
pub struct ChildClient {
    tx: SplitSink<ClientStream, Message>,
    rx: SplitStream<ClientStream>,
}

impl ChildClient {
    /// Connect with no bearer token (loopback / trusted-link default). Identical
    /// to `connect_with_auth(endpoint, None)` — zero regression.
    pub async fn connect(endpoint: &str) -> TransportResult<Self> {
        Self::connect_with_auth(endpoint, None).await
    }

    /// Connect to a worker endpoint, optionally presenting a bearer token on the
    /// WS upgrade (`Authorization: Bearer <token>`). The token is never logged.
    ///
    /// A `wss://` endpoint is negotiated against the standard webpki CA roots
    /// (the current Mozilla root set, with the explicit ring provider) — i.e. a worker serving a
    /// CA-signed cert (Let's Encrypt etc.) works with no extra configuration. To
    /// trust a self-signed worker cert (the typical P1 single-direction-TLS
    /// case), use [`connect_with_auth_tls`](Self::connect_with_auth_tls) with a
    /// client config built by [`client_config_trusting_cert`].
    pub async fn connect_with_auth(endpoint: &str, token: Option<&str>) -> TransportResult<Self> {
        Self::connect_with_auth_tls(endpoint, token, None).await
    }

    /// Like [`connect_with_auth`](Self::connect_with_auth), but for `wss://` lets
    /// the caller supply a custom rustls [`rustls::ClientConfig`] (e.g. one that
    /// pins/trusts a self-signed worker cert). `None` uses the default webpki-root
    /// trust. The config is ignored for plaintext `ws://`.
    pub async fn connect_with_auth_tls(
        endpoint: &str,
        token: Option<&str>,
        tls_config: Option<rustls::ClientConfig>,
    ) -> TransportResult<Self> {
        use tokio_tungstenite::tungstenite::client::IntoClientRequest;
        let mut request = endpoint.into_client_request()?;
        if let Some(token) = token {
            let value = format!("Bearer {token}")
                .parse()
                .map_err(|e| TransportError::Protocol(format!("bad bearer header: {e}")))?;
            request.headers_mut().insert(
                tokio_tungstenite::tungstenite::http::header::AUTHORIZATION,
                value,
            );
        }
        // Do not let the WebSocket dependency choose its obsolete bundled
        // roots or fall back to the machine's OS trust store. A caller's pinned
        // config is retained verbatim; plaintext ws needs no TLS config.
        let tls_config = match tls_config {
            None if request.uri().scheme_str() == Some("wss") => Some(
                client_config_with_roots(standard_webpki_roots()).map_err(TransportError::Tls)?,
            ),
            other => other,
        };
        let (ws, _resp) = match tls_config {
            Some(cfg) => {
                let connector = tokio_tungstenite::Connector::Rustls(std::sync::Arc::new(cfg));
                tokio_tungstenite::connect_async_tls_with_config(
                    request,
                    None,
                    false,
                    Some(connector),
                )
                .await?
            }
            None => connect_async(request).await?,
        };
        let (tx, rx) = ws.split();
        Ok(Self { tx, rx })
    }

    pub async fn send(&mut self, frame: ParentFrame) -> TransportResult<()> {
        self.tx.send(Message::text(frame.to_text())).await?;
        Ok(())
    }

    /// Next child frame, or `None` once the connection closes.
    pub async fn next_frame(&mut self) -> TransportResult<Option<ChildFrame>> {
        while let Some(msg) = self.rx.next().await {
            match msg? {
                Message::Text(t) => return Ok(Some(ChildFrame::from_text(t.as_str())?)),
                Message::Close(_) => return Ok(None),
                _ => continue,
            }
        }
        Ok(None)
    }

    pub async fn close(mut self) -> TransportResult<()> {
        self.tx.close().await?;
        Ok(())
    }
}

/// The parent→child control link, abstracted over transport so the actor runner
/// drives a child identically whether it is reached by a direct WS connection
/// ([`ChildClient`]) or over the mailbox bus
/// (`bamboo_broker::BrokerChildLink`). This trait is the seam where the two
/// sub-agent families (PULL direct / PUSH broker) collapse into one drive path.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DurableChildDeliveryReceipt {
    /// Stable identity of the broker Maildir that owns these MsgIds.
    pub broker_identity: String,
    /// Host-owned mailbox where the broker stored these Event/Outcome MsgIds.
    /// Recovery must reconnect to this exact mailbox before confirming ACKs.
    pub parent_mailbox: String,
    /// Exact broker Run message id that correlated these frames.
    pub correlation_id: String,
    /// Exact Event and Outcome mailbox message ids, in delivery order.
    pub message_ids: Vec<String>,
    /// Worker terminal accepted by the Host frame pump, before SDK final save.
    pub terminal_status: TerminalStatus,
}

#[async_trait::async_trait]
pub trait ChildLink: Send {
    /// Send a parent→child frame (Run / Cancel / steer / approval reply).
    async fn send(&mut self, frame: ParentFrame) -> TransportResult<()>;
    /// Next child→parent frame, or `None` once the run is terminal / the link closes.
    async fn next_frame(&mut self) -> TransportResult<Option<ChildFrame>>;

    /// Called only after the Host frame pump has validated and consumed the
    /// terminal frame. A surfaced Outcome alone is not checkpoint authority.
    fn accept_durable_terminal(&mut self, _status: TerminalStatus) {}

    /// Whether this link holds a broker Outcome receipt for a fully processed
    /// terminal frame. Direct WebSocket links have no mailbox receipt.
    fn has_pending_durable_terminal(&self) -> bool {
        false
    }

    fn durable_delivery_receipt(&self) -> Option<DurableChildDeliveryReceipt> {
        None
    }

    /// The authenticated parent mailbox subscribed by this broker link.
    /// Recovery must match it to the Host checkpoint before ACKing frames.
    fn broker_parent_mailbox(&self) -> Option<&str> {
        None
    }

    /// Confirm the exact durable broker frames only after the Host has saved
    /// the logical Child transcript and terminal status. A failed confirmation
    /// leaves remaining mailbox receipts unconfirmed for replay.
    async fn acknowledge_durable_frames(&mut self) -> TransportResult<()> {
        Ok(())
    }

    /// Delete a previously Host-committed Run's frames on a newly connected
    /// link. Unsupported transports must fail closed instead of claiming a
    /// receipt was processed.
    async fn acknowledge_recovered_durable_frames(
        &mut self,
        _broker_identity: &str,
        _message_ids: &[String],
    ) -> TransportResult<()> {
        Err(TransportError::Protocol(
            "recovered durable ACK is unavailable on this transport".into(),
        ))
    }
}

#[async_trait::async_trait]
impl ChildLink for ChildClient {
    async fn send(&mut self, frame: ParentFrame) -> TransportResult<()> {
        ChildClient::send(self, frame).await
    }
    async fn next_frame(&mut self) -> TransportResult<Option<ChildFrame>> {
        ChildClient::next_frame(self).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::executor::EchoExecutor;
    use crate::proto::{ChildFrame, TerminalStatus};

    struct RunDropFlag(Arc<std::sync::atomic::AtomicBool>);

    impl Drop for RunDropFlag {
        fn drop(&mut self) {
            self.0.store(true, std::sync::atomic::Ordering::SeqCst);
        }
    }

    struct UncooperativeExecutor(Arc<std::sync::atomic::AtomicBool>);

    #[async_trait::async_trait]
    impl ChildExecutor for UncooperativeExecutor {
        async fn run(
            &self,
            _spec: RunSpec,
            events: EventSink,
            _steer: SteerInbox,
            _cancel: CancellationToken,
        ) -> crate::ChildOutcome {
            let _dropped = RunDropFlag(self.0.clone());
            events.emit(serde_json::json!({"type":"ready"})).await;
            std::future::pending().await
        }
    }

    #[tokio::test]
    async fn disconnect_and_server_abort_reclaim_uncooperative_execution() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::time::Duration;
        for abort_server in [false, true] {
            let server = WsServer::bind_loopback().await.unwrap();
            let endpoint = server.ws_endpoint();
            let dropped = Arc::new(AtomicBool::new(false));
            let executor = Arc::new(UncooperativeExecutor(dropped.clone()));
            let task = tokio::spawn(async move { server.serve(executor).await });
            let mut client = ChildClient::connect(&endpoint).await.unwrap();
            let spec = serde_json::from_value(serde_json::json!({"assignment":"hang"})).unwrap();
            client.send(ParentFrame::Run(spec)).await.unwrap();
            tokio::time::timeout(Duration::from_secs(2), client.next_frame())
                .await
                .unwrap()
                .unwrap();
            if abort_server {
                task.abort();
                let _ = task.await;
            } else {
                client.close().await.unwrap();
                tokio::time::timeout(Duration::from_secs(8), async {
                    while !dropped.load(Ordering::SeqCst) {
                        tokio::task::yield_now().await;
                    }
                })
                .await
                .expect("disconnect must abort an executor that ignores cancellation");
                task.abort();
                let _ = task.await;
            }
            tokio::time::timeout(Duration::from_secs(2), async {
                while !dropped.load(Ordering::SeqCst) {
                    tokio::task::yield_now().await;
                }
            })
            .await
            .expect("server cancellation must own and release connection/run tasks");
        }
    }

    #[tokio::test]
    async fn explicit_cancel_bounds_uncooperative_execution_and_emits_terminal() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::time::Duration;
        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let dropped = Arc::new(AtomicBool::new(false));
        let executor = Arc::new(UncooperativeExecutor(dropped.clone()));
        let task = tokio::spawn(async move { server.serve_one(executor).await });
        let mut client = ChildClient::connect(&endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(
                serde_json::from_value(serde_json::json!({"assignment":"hang"})).unwrap(),
            ))
            .await
            .unwrap();
        client.next_frame().await.unwrap();
        client.send(ParentFrame::Cancel).await.unwrap();
        let frame = tokio::time::timeout(Duration::from_secs(8), client.next_frame())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert!(matches!(
            frame,
            ChildFrame::Terminal {
                status: TerminalStatus::Cancelled,
                ..
            }
        ));
        assert!(dropped.load(Ordering::SeqCst));
        client.close().await.unwrap();
        tokio::time::timeout(Duration::from_secs(2), task)
            .await
            .unwrap()
            .unwrap()
            .unwrap();
    }

    #[tokio::test]
    async fn executor_panic_becomes_terminal_error() {
        struct Panics;
        #[async_trait::async_trait]
        impl ChildExecutor for Panics {
            async fn run(
                &self,
                _spec: RunSpec,
                _events: EventSink,
                _steer: SteerInbox,
                _cancel: CancellationToken,
            ) -> crate::ChildOutcome {
                panic!("intentional executor failure");
            }
        }
        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let task = tokio::spawn(async move { server.serve_one(Arc::new(Panics)).await });
        let mut client = ChildClient::connect(&endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(
                serde_json::from_value(serde_json::json!({"assignment":"panic"})).unwrap(),
            ))
            .await
            .unwrap();
        let frame = tokio::time::timeout(std::time::Duration::from_secs(2), client.next_frame())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert!(matches!(
            frame,
            ChildFrame::Terminal {
                status: TerminalStatus::Error,
                ..
            }
        ));
        client.close().await.unwrap();
        task.await.unwrap().unwrap();
    }

    #[test]
    fn transport_error_conversions_preserve_diagnostics_and_sources() {
        let io_source = std::io::Error::other("socket unavailable");
        let io_error = TransportError::from(io_source);
        assert_eq!(io_error.to_string(), "io: socket unavailable");
        assert!(
            std::error::Error::source(&io_error)
                .and_then(|source| source.downcast_ref::<std::io::Error>())
                .is_some(),
            "the concrete IO error type must remain discoverable"
        );

        let decode_source = serde_json::from_str::<serde_json::Value>("{")
            .expect_err("the fixture must be invalid JSON");
        let decode_message = decode_source.to_string();
        let decode_error = TransportError::from(decode_source);
        assert_eq!(
            decode_error.to_string(),
            format!("decode: {decode_message}")
        );
        assert!(
            std::error::Error::source(&decode_error)
                .and_then(|source| source.downcast_ref::<serde_json::Error>())
                .is_some(),
            "the concrete JSON error type must remain discoverable"
        );

        let websocket_source = tokio_tungstenite::tungstenite::Error::ConnectionClosed;
        let websocket_message = websocket_source.to_string();
        let websocket_error = TransportError::from(websocket_source);

        assert_eq!(
            websocket_error.to_string(),
            format!("ws: {websocket_message}")
        );
        let reported_source = std::error::Error::source(&websocket_error)
            .expect("the boxed WebSocket error remains the transport error source");
        assert_eq!(reported_source.to_string(), websocket_message);
        assert!(
            reported_source
                .downcast_ref::<tokio_tungstenite::tungstenite::Error>()
                .is_some(),
            "the concrete WebSocket error type must remain discoverable"
        );
        assert!(matches!(
            websocket_error,
            TransportError::Ws(inner)
                if matches!(*inner, tokio_tungstenite::tungstenite::Error::ConnectionClosed)
        ));

        let protocol_error = TransportError::Protocol("unexpected frame".to_string());
        assert_eq!(protocol_error.to_string(), "protocol: unexpected frame");
        assert!(std::error::Error::source(&protocol_error).is_none());

        let tls_error = TransportError::Tls("bad certificate".to_string());
        assert_eq!(tls_error.to_string(), "tls: bad certificate");
        assert!(std::error::Error::source(&tls_error).is_none());
    }

    /// In-process server + client over a real loopback socket (no subprocess).
    #[tokio::test]
    async fn loopback_run_streams_events_then_terminal() {
        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let srv = tokio::spawn(async move { server.serve_one(Arc::new(EchoExecutor)).await });

        let mut client = ChildClient::connect(&endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "one two".into(),
                logical_session: None,
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 1,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();

        let mut events = Vec::new();
        let mut saw_batch = false;
        let mut terminal = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            match frame {
                ChildFrame::Event { event } => events.push(event),
                ChildFrame::EventBatch { batch } => {
                    saw_batch = true;
                    events.extend(batch.events);
                }
                ChildFrame::ApprovalRequest { .. }
                | ChildFrame::OwnedTreeRequest { .. }
                | ChildFrame::SubAgentRequest { .. } => {}
                ChildFrame::SessionMessageAdmitted { confirmation } => {
                    panic!(
                        "echo run emitted unexpected SessionInbox confirmation: {confirmation:?}"
                    )
                }
                ChildFrame::Terminal { status, result, .. } => {
                    terminal = Some((status, result));
                    break;
                }
            }
        }

        let (status, result) = terminal.expect("terminal frame");
        assert_eq!(status, TerminalStatus::Completed);
        assert_eq!(result.as_deref(), Some("echo: one two"));
        assert!(events.iter().any(|e| e["content"] == "one "));
        assert!(
            saw_batch,
            "a non-zero execution epoch must select event batches"
        );

        let _ = client.close().await;
        let _ = srv.await;
    }

    /// `bind_with_token` gates the WS upgrade: the right bearer connects + runs,
    /// a wrong / missing bearer is rejected at the upgrade (no WS, no Terminal).
    #[tokio::test]
    async fn bearer_token_gates_the_upgrade() {
        let server = WsServer::bind_with_token(
            (std::net::Ipv4Addr::LOCALHOST, 0).into(),
            Some("T-secret".into()),
        )
        .await
        .unwrap();
        let endpoint = server.ws_endpoint();
        // ws:// (no TLS) even though a token is set — token alone never upgrades scheme.
        assert!(endpoint.starts_with("ws://"));
        let srv = tokio::spawn(async move { server.serve(Arc::new(EchoExecutor)).await });

        // Correct token: connects and runs to completion.
        let mut client = ChildClient::connect_with_auth(&endpoint, Some("T-secret"))
            .await
            .expect("correct bearer should connect");
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "go".into(),
                logical_session: None,
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();
        let mut terminal = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            if let ChildFrame::Terminal { status, .. } = frame {
                terminal = Some(status);
                break;
            }
        }
        assert_eq!(terminal, Some(TerminalStatus::Completed));
        let _ = client.close().await;

        // Wrong token: the upgrade is rejected — connect itself errors.
        let bad = ChildClient::connect_with_auth(&endpoint, Some("WRONG")).await;
        assert!(bad.is_err(), "wrong bearer must be rejected at the upgrade");

        // Missing token: same rejection.
        let missing = ChildClient::connect(&endpoint).await;
        assert!(
            missing.is_err(),
            "missing bearer must be rejected when a token is required"
        );

        srv.abort();
    }

    /// `bind_tls` fails fast on a missing cert/key (never falls back to plaintext).
    #[tokio::test]
    async fn bind_tls_fails_fast_on_missing_cert() {
        let result = WsServer::bind_tls(
            (std::net::Ipv4Addr::LOCALHOST, 0).into(),
            Path::new("/nonexistent/bamboo-subagent/cert.pem"),
            Path::new("/nonexistent/bamboo-subagent/key.pem"),
            None,
        )
        .await;
        match result {
            Err(TransportError::Tls(m)) if m.contains("cert_file") => {}
            Err(other) => panic!("expected a TLS error naming cert_file, got {other:?}"),
            Ok(_) => panic!("missing cert must fail"),
        }
    }

    async fn run_once(
        endpoint: &str,
        assignment: &str,
        logical_session_id: &str,
    ) -> (TerminalStatus, Option<String>) {
        let mut client = ChildClient::connect(endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: assignment.into(),
                logical_session: Some(crate::proto::LogicalSessionIdentity {
                    creation: None,
                    session_id: logical_session_id.to_string(),
                    parent_session_id: Some("logical-parent".to_string()),
                    root_session_id: "logical-root".to_string(),
                }),
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();
        let mut terminal = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            if let ChildFrame::Terminal { status, result, .. } = frame {
                terminal = Some((status, result));
                break;
            }
        }
        let _ = client.close().await;
        terminal.expect("terminal frame")
    }

    /// A reusable worker serves connection-after-connection (the pool's reuse
    /// path), then reclaims itself once left idle past the timeout.
    #[tokio::test]
    async fn reusable_server_serves_sequential_connections_then_idles_out() {
        use crate::executor::{ChildExecutor, ChildOutcome, SteerInbox};
        use std::time::Duration;

        struct IdentityRecordingEcho {
            seen: mpsc::UnboundedSender<crate::proto::LogicalSessionIdentity>,
        }

        #[async_trait::async_trait]
        impl ChildExecutor for IdentityRecordingEcho {
            async fn run(
                &self,
                spec: RunSpec,
                events: EventSink,
                steer: SteerInbox,
                cancel: CancellationToken,
            ) -> ChildOutcome {
                self.seen
                    .send(
                        spec.logical_session
                            .clone()
                            .expect("warm run must carry logical identity"),
                    )
                    .unwrap();
                EchoExecutor.run(spec, events, steer, cancel).await
            }
        }

        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let (seen_tx, mut seen_rx) = mpsc::unbounded_channel();
        let srv = tokio::spawn(async move {
            server
                .serve_reusable_with_idle_timeout(
                    Arc::new(IdentityRecordingEcho { seen: seen_tx }),
                    Duration::from_millis(400),
                )
                .await
        });

        // Two SEQUENTIAL assignments land on the SAME warm worker.
        let (s1, r1) = run_once(&endpoint, "first", "logical-warm-first").await;
        assert_eq!(s1, TerminalStatus::Completed);
        assert_eq!(r1.as_deref(), Some("echo: first"));

        let (s2, r2) = run_once(&endpoint, "second", "logical-warm-second").await;
        assert_eq!(s2, TerminalStatus::Completed);
        assert_eq!(r2.as_deref(), Some("echo: second"));
        assert_eq!(
            [
                seen_rx.recv().await.unwrap().session_id,
                seen_rx.recv().await.unwrap().session_id,
            ],
            [
                "logical-warm-first".to_string(),
                "logical-warm-second".to_string(),
            ],
            "one warm worker must not reuse its mailbox or previous logical Session id"
        );

        // No further connection: the worker idles out and exits cleanly.
        let exited = tokio::time::timeout(Duration::from_secs(5), srv).await;
        assert!(
            matches!(exited, Ok(Ok(Ok(())))),
            "reusable server should idle out cleanly, got {exited:?}"
        );
    }

    /// Mid-run steering: a `Message` frame reaches the active run's SteerInbox.
    #[tokio::test]
    async fn message_frame_routes_to_active_steer_inbox() {
        use crate::executor::{ChildExecutor, ChildOutcome, SteerInbox};

        /// Completes only after receiving one steering message, echoing it back.
        struct SteerEcho;
        #[async_trait::async_trait]
        impl ChildExecutor for SteerEcho {
            async fn run(
                &self,
                _spec: RunSpec,
                events: EventSink,
                mut steer: SteerInbox,
                _cancel: CancellationToken,
            ) -> ChildOutcome {
                let steered = steer.recv().await.unwrap_or_default();
                events
                    .emit(serde_json::json!({"type": "token", "content": steered.clone()}))
                    .await;
                ChildOutcome::completed(format!("steered: {steered}"))
            }
        }

        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let srv = tokio::spawn(async move { server.serve_one(Arc::new(SteerEcho)).await });

        let mut client = ChildClient::connect(&endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "start".into(),
                logical_session: None,
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();
        // mid-run in-band message
        client
            .send(ParentFrame::Message {
                text: "change course".into(),
            })
            .await
            .unwrap();

        let mut terminal = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            if let ChildFrame::Terminal { status, result, .. } = frame {
                terminal = Some((status, result));
                break;
            }
        }
        let (status, result) = terminal.expect("terminal");
        assert_eq!(status, TerminalStatus::Completed);
        assert_eq!(result.as_deref(), Some("steered: change course"));

        let _ = client.close().await;
        let _ = srv.await;
    }

    #[tokio::test]
    async fn typed_session_message_and_confirmation_cross_direct_ws() {
        use crate::executor::{ChildExecutor, ChildOutcome, SteerInbox, SteerMessage};
        use crate::proto::{SessionMessageAdmissionConfirmation, SessionMessageDelivery};
        use bamboo_domain::{SessionMessageEnvelope, SessionMessageId};

        struct TypedSteer;
        #[async_trait::async_trait]
        impl ChildExecutor for TypedSteer {
            async fn run(
                &self,
                spec: RunSpec,
                events: EventSink,
                mut steer: SteerInbox,
                _cancel: CancellationToken,
            ) -> ChildOutcome {
                assert_eq!(
                    spec.logical_session,
                    Some(crate::proto::LogicalSessionIdentity {
                        creation: None,
                        session_id: "logical-child".to_string(),
                        parent_session_id: Some("logical-parent".to_string()),
                        root_session_id: "logical-root".to_string(),
                    })
                );
                let SteerMessage::SessionMessage(delivery) =
                    steer.recv_message().await.expect("typed steer")
                else {
                    panic!("expected typed SessionInbox delivery");
                };
                events
                    .confirm_session_message(SessionMessageAdmissionConfirmation {
                        target_session_id: delivery.target_session_id,
                        envelope_id: delivery.envelope.id.as_str().to_string(),
                        canonical_claim_generation: delivery.canonical_claim_generation,
                        activation_run_id: delivery.activation_run_id,
                    })
                    .await;
                ChildOutcome::completed("confirmed")
            }
        }

        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let srv = tokio::spawn(async move { server.serve_one(Arc::new(TypedSteer)).await });
        let mut client = ChildClient::connect(&endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "wait".into(),
                logical_session: Some(crate::proto::LogicalSessionIdentity {
                    creation: None,
                    session_id: "logical-child".to_string(),
                    parent_session_id: Some("logical-parent".to_string()),
                    root_session_id: "logical-root".to_string(),
                }),
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();
        let mut envelope = SessionMessageEnvelope::user_input("logical-child", "hello");
        envelope.id = SessionMessageId::parse("typed-direct-id").unwrap();
        client
            .send(ParentFrame::SessionMessage {
                delivery: SessionMessageDelivery {
                    target_session_id: "logical-child".to_string(),
                    envelope,
                    canonical_claim_generation: 41,
                    activation_run_id: "activation-run-1".to_string(),
                    activation_policy:
                        bamboo_domain::SessionActivationPolicy::InterruptSpecificWait,
                },
            })
            .await
            .unwrap();

        match client.next_frame().await.unwrap() {
            Some(ChildFrame::SessionMessageAdmitted { confirmation }) => {
                assert_eq!(confirmation.target_session_id, "logical-child");
                assert_eq!(confirmation.envelope_id, "typed-direct-id");
                assert_eq!(confirmation.canonical_claim_generation, 41);
                assert_eq!(confirmation.activation_run_id, "activation-run-1");
            }
            other => panic!("expected admission confirmation, got {other:?}"),
        }
        assert!(matches!(
            client.next_frame().await.unwrap(),
            Some(ChildFrame::Terminal { .. })
        ));
        let _ = client.close().await;
        let _ = srv.await;
    }

    /// Phase 2: a worker that hits a gated tool proxies an approval to the host
    /// over the per-child WS and blocks for the decision — the FULL cross-process
    /// round-trip (worker `approval_call` → `ChildFrame::ApprovalRequest` →
    /// host `ParentFrame::ApprovalReply` → the awaiting call resolves).
    #[tokio::test]
    async fn approval_request_round_trips_to_host_and_back() {
        use crate::executor::{ChildExecutor, ChildOutcome, SteerInbox};

        /// Proxies one approval to the host and reports the decision.
        struct ApprovalProber;
        #[async_trait::async_trait]
        impl ChildExecutor for ApprovalProber {
            async fn run(
                &self,
                _spec: RunSpec,
                events: EventSink,
                _steer: SteerInbox,
                _cancel: CancellationToken,
            ) -> ChildOutcome {
                let Some(host) = events.host() else {
                    return ChildOutcome::error("no host bridge wired");
                };
                match host
                    .approval_call(serde_json::json!({
                        "tool_name": "Write",
                        "permission_type": "WriteFile",
                        "resource": "/tmp/x",
                        "question": "approve?",
                    }))
                    .await
                {
                    Ok(decision) => ChildOutcome::completed(format!(
                        "approved: {}",
                        decision["approved"].as_bool().unwrap_or(false)
                    )),
                    Err(e) => ChildOutcome::error(format!("bridge: {e}")),
                }
            }
        }

        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let srv = tokio::spawn(async move { server.serve_one(Arc::new(ApprovalProber)).await });

        let mut client = ChildClient::connect(&endpoint).await.unwrap();
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "go".into(),
                logical_session: None,
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();

        let mut terminal = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            match frame {
                ChildFrame::ApprovalRequest { id, body } => {
                    assert_eq!(body["resource"], "/tmp/x");
                    assert_eq!(body["tool_name"], "Write");
                    client
                        .send(ParentFrame::ApprovalReply { id, approved: true })
                        .await
                        .unwrap();
                }
                ChildFrame::Terminal { status, result, .. } => {
                    terminal = Some((status, result));
                    break;
                }
                _ => {}
            }
        }
        let (status, result) = terminal.expect("terminal");
        assert_eq!(status, TerminalStatus::Completed);
        assert_eq!(result.as_deref(), Some("approved: true"));

        let _ = client.close().await;
        let _ = srv.await;
    }

    /// Service-agent mode: two concurrent connections to one `serve()` must be
    /// fully isolated — each client sees only its own run's events/terminal,
    /// and per-connection cancel state (the round-2 fix) must not cross talk.
    #[tokio::test(flavor = "multi_thread")]
    async fn service_agent_concurrent_no_crosstalk() {
        use crate::executor::EchoExecutor;

        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let srv = tokio::spawn(async move { server.serve(Arc::new(EchoExecutor)).await });

        // Client A holds its run open (300ms sleep) while client B runs to
        // completion — guaranteed overlap.
        let endpoint_a = endpoint.clone();
        let a = tokio::spawn(async move {
            let mut client = ChildClient::connect(&endpoint_a).await.unwrap();
            client
                .send(ParentFrame::Run(RunSpec {
                    assignment: "__sleep_ms:300 alpha only".into(),
                    logical_session: None,
                    project_id: None,
                    reasoning_effort: None,
                    permission_policy: None,
                    messages: Vec::new(),
                    activation_run_id: None,
                    execution_epoch: 0,
                    initial_session_messages: Vec::new(),
                    secrets: Default::default(),
                }))
                .await
                .unwrap();
            collect_stream(client).await
        });
        let endpoint_b = endpoint.clone();
        let b = tokio::spawn(async move {
            // Give A a head start so its run is live when B's traffic flows.
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            let mut client = ChildClient::connect(&endpoint_b).await.unwrap();
            client
                .send(ParentFrame::Run(RunSpec {
                    assignment: "beta only".into(),
                    logical_session: None,
                    project_id: None,
                    reasoning_effort: None,
                    permission_policy: None,
                    messages: Vec::new(),
                    activation_run_id: None,
                    execution_epoch: 0,
                    initial_session_messages: Vec::new(),
                    secrets: Default::default(),
                }))
                .await
                .unwrap();
            collect_stream(client).await
        });

        let (tokens_a, result_a) = a.await.unwrap();
        let (tokens_b, result_b) = b.await.unwrap();

        // Each stream carries exactly its own run.
        assert_eq!(result_a.as_deref(), Some("echo: alpha only"));
        assert_eq!(result_b.as_deref(), Some("echo: beta only"));
        assert!(
            tokens_a.iter().all(|t| !t.contains("beta")),
            "client A saw B's tokens: {tokens_a:?}"
        );
        assert!(
            tokens_b.iter().all(|t| !t.contains("alpha")),
            "client B saw A's tokens: {tokens_b:?}"
        );

        srv.abort();
    }

    /// Drain a client until Terminal; returns (token contents, result).
    async fn collect_stream(mut client: ChildClient) -> (Vec<String>, Option<String>) {
        let mut tokens = Vec::new();
        let mut result = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            match frame {
                ChildFrame::Event { event } => {
                    if let Some(t) = event["content"].as_str() {
                        tokens.push(t.to_string());
                    }
                }
                ChildFrame::EventBatch { batch } => {
                    tokens.extend(
                        batch
                            .events
                            .into_iter()
                            .filter_map(|event| event["content"].as_str().map(ToString::to_string)),
                    );
                }
                ChildFrame::ApprovalRequest { .. }
                | ChildFrame::OwnedTreeRequest { .. }
                | ChildFrame::SubAgentRequest { .. } => {}
                ChildFrame::SessionMessageAdmitted { confirmation } => {
                    panic!(
                        "echo run emitted unexpected SessionInbox confirmation: {confirmation:?}"
                    )
                }
                ChildFrame::Terminal {
                    status, result: r, ..
                } => {
                    assert_eq!(status, TerminalStatus::Completed);
                    result = r;
                    break;
                }
            }
        }
        let _ = client.close().await;
        (tokens, result)
    }

    /// Orphan defense: with no client, the accept-timeout variant returns
    /// instead of hanging forever.
    #[tokio::test]
    async fn accept_timeout_fires_when_nobody_connects() {
        let server = WsServer::bind_loopback().await.unwrap();
        let result = server
            .serve_one_with_accept_timeout(
                Arc::new(EchoExecutor),
                std::time::Duration::from_millis(50),
            )
            .await;
        assert!(matches!(result, Err(TransportError::Protocol(_))));
    }

    /// Defect A: a second `Run` must cancel the previously active run so the old
    /// task ends promptly — its `Terminal(Cancelled)` arrives *before* any
    /// explicit `Cancel` is sent, proving the supersession, not orphaning.
    #[tokio::test]
    async fn second_run_cancels_previous() {
        use crate::executor::{ChildExecutor, ChildOutcome, SteerInbox};
        use std::time::Duration;

        /// Emits one `token` then parks on its cancel token; always ends
        /// `Cancelled`. Forces a run that can only finish via cancellation.
        struct WaitForCancel;
        #[async_trait::async_trait]
        impl ChildExecutor for WaitForCancel {
            async fn run(
                &self,
                _spec: RunSpec,
                events: EventSink,
                _steer: SteerInbox,
                cancel: CancellationToken,
            ) -> ChildOutcome {
                events
                    .emit(serde_json::json!({"type": "token", "content": "go"}))
                    .await;
                cancel.cancelled().await;
                ChildOutcome::cancelled()
            }
        }

        let server = WsServer::bind_loopback().await.unwrap();
        let endpoint = server.ws_endpoint();
        let srv = tokio::spawn(async move { server.serve_one(Arc::new(WaitForCancel)).await });

        let mut client = ChildClient::connect(&endpoint).await.unwrap();

        // First run: goes live, emits its token, then parks on cancellation.
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "first".into(),
                logical_session: None,
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();
        loop {
            match client.next_frame().await.unwrap() {
                Some(ChildFrame::Event { event }) if event["content"] == "go" => break,
                Some(ChildFrame::EventBatch { batch })
                    if batch.events.iter().any(|event| event["content"] == "go") =>
                {
                    break;
                }
                Some(_) => continue,
                None => panic!("connection closed before first token"),
            }
        }

        // Second run: must supersede (cancel) the first — defect A fix.
        client
            .send(ParentFrame::Run(RunSpec {
                assignment: "second".into(),
                logical_session: None,
                project_id: None,
                reasoning_effort: None,
                permission_policy: None,
                messages: Vec::new(),
                activation_run_id: None,
                execution_epoch: 0,
                initial_session_messages: Vec::new(),
                secrets: Default::default(),
            }))
            .await
            .unwrap();

        // The first run's Terminal must now arrive with no explicit Cancel —
        // driven solely by the new Run frame. Bound it so a regression fails
        // fast instead of hanging on the parked first run.
        let mut first_terminal: Option<TerminalStatus> = None;
        let drain = async {
            while let Some(frame) = client.next_frame().await.unwrap() {
                if let ChildFrame::Terminal { status, .. } = frame {
                    first_terminal = Some(status);
                    break;
                }
            }
        };
        tokio::time::timeout(Duration::from_secs(2), drain)
            .await
            .expect("first run's terminal never arrived — new Run did not cancel it");
        assert_eq!(
            first_terminal.expect("first terminal status"),
            TerminalStatus::Cancelled,
            "previous run should be cancelled by the new Run frame"
        );

        // The second run is still parked; only an explicit Cancel ends it.
        client.send(ParentFrame::Cancel).await.unwrap();

        let mut second_terminal: Option<TerminalStatus> = None;
        while let Some(frame) = client.next_frame().await.unwrap() {
            if let ChildFrame::Terminal { status, .. } = frame {
                second_terminal = Some(status);
                break;
            }
        }
        assert_eq!(
            second_terminal.expect("second terminal status"),
            TerminalStatus::Cancelled
        );

        let _ = client.close().await;
        let _ = srv.await;
    }
}