capsule-runtime 0.9.0

//! The tool bridge — a loopback tool server for the `transport: process` inference path
//! **only**. Nothing here is used by the primary `transport: http` driver path; it exists solely
//! to give process-transport capsules the same artifact tool calling the HTTP path gets.
//!
//! # Why this exists
//!
//! With `transport: http`, murmur owns the agent loop: it sends the model `messages + tool
//! schemas`, the model returns a structured tool-call request, murmur executes the WASM tool,
//! and appends the result. The tool boundary sits between murmur and the model.
//!
//! With `transport: process` the runtime drives a harness CLI, which is a self-contained agent
//! that runs its *own* loop and executes its *own* tools on the host — the tool boundary sits inside the
//! subprocess, out of murmur's reach. So a plain process capsule could only do inference;
//! declared tool artifacts would be invisible to it.
//!
//! This bridge relocates the tool boundary back out to murmur. It is a tiny loopback HTTP
//! server that advertises the capsule's tool **schemas only** (no logic) over the protocol
//! harnesses speak to externally-hosted tool servers. When the model calls a tool, the request
//! comes here, and murmur executes it through the *same* `CapsuleStoreState` dispatch the HTTP
//! path uses — under the capsule's declared capabilities and sandbox. The model gets native,
//! structured tool calling; murmur keeps ownership of execution, capabilities, and the trace.
//!
//! # Scope / invariants (process transport only)
//!
//! - Bound to loopback with a per-run bearer token. The process driver is handed the URL, the
//!   token and the bare tool names, and is what writes them into whatever configuration its
//!   harness reads: the runtime builds no harness configuration and no harness tool name.
//! - Only the capsule's declared tools are advertised.
//! - Every call crosses the session's decision point first, over the gate channel, because the
//!   hook runtime and the trace writer belong to the task driving the harness and this bridge
//!   holds only the store. The decision point runs the required-field check on every call and
//!   then, only for a capsule with a policy hook or a `read_only` path, the manifest's
//!   `capabilities.filesystem.read_only` check and an `on-tool-call` policy hook.
//! - Request/response is plain JSON, so the transport is a minimal manual HTTP/1.1 handler
//!   mirroring `identity.rs`, not a full server stack.
//! - Tool calls run concurrently, up to [`MAX_IN_FLIGHT_CALLS`] at once: a shell command that
//!   runs for minutes does not hold up the harness's other calls.
//! - An image a tool attaches to its result is returned as an MCP `image` content item after the
//!   result's text, under the `context.images` checks, and stored nowhere: the harness holds the
//!   conversation.
//! - Each dispatched call that named an attachment or whose result was fenced sends one
//!   [`CallReport`] to the task driving the harness, for the `tool_call` record and the A2A
//!   `artifact` frame it writes.

use std::{
    sync::atomic::{AtomicBool, AtomicUsize, Ordering},
    time::{Duration, Instant},
};

use base64::Engine as _;
use futures_util::stream::{FuturesUnordered, StreamExt};
use serde_json::{json, Value};
use tokio::{
    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
    net::{tcp::OwnedReadHalf, TcpListener, TcpStream},
    sync::{mpsc, oneshot},
};
use uuid::Uuid;

use crate::{
    bindings::host::murmur::tool::run::{Attachment, Status, ToolInput},
    detached::AbandonSignal,
    images::CheckedImage,
    runtime::CapsuleStoreState,
    trace::AttachmentRecord,
};

/// How many `tools/call` dispatches the bridge runs at once.
///
/// Above what a harness fans out — one message's worth of parallel calls plus a few sub-agents
/// sharing the bridge — and small beside what each call can hold: a bridged shell call is one
/// host process under the scope's `cgroup_pids_max` and one thread of tokio's blocking pool. A
/// call beyond it is refused with [`at_capacity_text`] rather than queued, because a queued call
/// is a call that hangs behind a long one.
pub(super) const MAX_IN_FLIGHT_CALLS: usize = 16;

/// How many connections the bridge holds open at once. Further connections wait in the kernel's
/// accept backlog. At most [`MAX_IN_FLIGHT_CALLS`] of them can be long-running; every other one
/// is answered at once or dropped after [`REQUEST_READ_TIMEOUT`], so a waiting connection waits
/// about one read timeout, never for a tool call.
const MAX_OPEN_CONNECTIONS: usize = 64;

/// How long a connection has to deliver its whole request — request line, headers and body. A
/// connection that sends nothing in that time is closed without a response.
const REQUEST_READ_TIMEOUT: Duration = Duration::from_secs(10);

/// How many bytes a connection may send after its request while its call runs. They are read
/// and discarded so a close behind them can be seen; past this, the connection is no longer read.
const TRAILING_BYTES_CAP: usize = 64 * 1024;

/// The failing tool result's text for a call refused at [`MAX_IN_FLIGHT_CALLS`].
fn at_capacity_text(limit: usize) -> String {
    format!(
        "Refused: {limit} tool calls were already running, the most this session runs at once, \
         so nothing ran for this call. Call it again once one of the running calls has returned."
    )
}

/// Something the bridge saw that belongs in the trace as a `harness_note`.
///
/// The bridge holds no trace writer, so these go to the task driving the harness, which writes
/// [`BridgeNote::text`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum BridgeNote {
    /// The harness closed a call's connection before its response was written.
    Abandoned {
        tool_name: String,
        elapsed_ms: u64,
        /// Whether the decision point had answered. Undecided means nothing was dispatched.
        decided: bool,
    },
    /// A call was refused because [`MAX_IN_FLIGHT_CALLS`] calls were already running.
    AtCapacity { tool_name: String, limit: usize },
}

impl BridgeNote {
    /// The `harness_note` text.
    pub(super) fn text(&self) -> String {
        match self {
            BridgeNote::Abandoned {
                tool_name,
                decided: false,
                ..
            } => format!(
                "The harness closed the connection for its call to {tool_name} before the call \
                 was decided, so nothing ran."
            ),
            BridgeNote::Abandoned {
                tool_name,
                elapsed_ms,
                decided: true,
            } => format!(
                "The harness closed the connection for its call to {tool_name} {elapsed_ms} ms \
                 into the call. A shell command still running is moved to the background, or, \
                 with no task to deliver its completion to, runs to the end with its result \
                 discarded; anything else still running finishes unobserved."
            ),
            BridgeNote::AtCapacity { tool_name, limit } => format!(
                "The call to {tool_name} was refused because {limit} tool calls were already \
                 running."
            ),
        }
    }
}

/// One admitted `tools/call`, counted in `in_flight` until it is dropped — whether its response
/// was written or its future was.
struct InFlightSlot<'a>(&'a AtomicUsize);

impl<'a> InFlightSlot<'a> {
    fn claim(in_flight: &'a AtomicUsize, limit: usize) -> Option<Self> {
        in_flight
            .try_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
                (n < limit).then_some(n + 1)
            })
            .ok()
            .map(|_| Self(in_flight))
    }
}

impl Drop for InFlightSlot<'_> {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::SeqCst);
    }
}

/// What reading one request produced.
enum Incoming {
    /// The connection failed or ended before a whole request arrived. Nothing to answer.
    Gone,
    /// A request answered with a bare HTTP status.
    Reject(&'static str),
    /// An authorized JSON body.
    Rpc(Value),
}

/// The name this bridge registers itself under. Handed to the process driver as
/// `bridge.server-name`; how — and whether — a harness uses it to namespace tool names is the
/// driver's business, not the runtime's.
pub(super) const BRIDGE_SERVER_NAME: &str = "murmur";

/// Path the bridge listens on. Arbitrary; the driver is handed the full URL.
const BRIDGE_PATH: &str = "/bridge";

/// What a bridged tool call is refused with when the session's decision point cannot be reached
/// at all.
///
/// Fail closed: the alternative is a tool running because the thing that would have refused it
/// went away.
const GATE_UNREACHABLE: &str =
    "Refused: the session stopped answering policy checks, so this tool call was not dispatched.";

/// One bridged tool call, asking the session's decision point whether it may run.
///
/// The task driving the harness owns the hook runtime and the trace writer, and answers this
/// between two batches of harness output; the call is parked on `reply` in the meantime. Each call
/// carries its own `reply`, so a verdict reaches only the call that asked for it however many are
/// parked. A `reply` that is already closed belongs to a call whose harness went away, and is
/// skipped undecided.
pub(super) struct GateRequest {
    /// The tool's bare name, as the harness asked for it.
    pub(super) tool_name: String,
    /// The call's arguments, serialised exactly as dispatch is about to receive them.
    pub(super) input_json: String,
    /// The refusal text, or `None` to let the call run.
    pub(super) reply: oneshot::Sender<Option<String>>,
}

/// Put one call to the session's decision point and wait for its verdict.
///
/// Fail closed on both arms — a send onto a closed channel and a dropped reply alike — so a
/// session that has stopped answering cannot let a tool through ungated.
async fn ask_gate(
    gate: &mpsc::UnboundedSender<GateRequest>,
    tool_name: &str,
    input_json: &str,
) -> Option<String> {
    let (reply, verdict) = oneshot::channel();
    if gate
        .send(GateRequest {
            tool_name: tool_name.to_string(),
            input_json: input_json.to_string(),
            reply,
        })
        .is_err()
    {
        return Some(GATE_UNREACHABLE.to_string());
    }
    verdict
        .await
        .unwrap_or_else(|_| Some(GATE_UNREACHABLE.to_string()))
}

/// A tool-server result carrying `text` as a failure.
fn error_result(text: String) -> Value {
    json!({
        "content": [{ "type": "text", "text": text }],
        "isError": true
    })
}

/// What one bridged call's attachments came to and which fence its result went back under, for
/// the `tool_call` record and the A2A `artifact` frame the task driving the harness writes from the
/// harness's own report of the result.
///
/// Both are written from the harness's `tool-result` event, which carries nothing tying it to this
/// bridge's request, so the report names the call the way the event sink can match it: by the bare
/// tool name and the arguments. Sent only for a dispatched call that named at least one attachment
/// or whose result was fenced, before its answer is written.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct CallReport {
    /// The tool's bare name, as the harness asked for it.
    pub(super) tool_name: String,
    /// The call's arguments, serialised exactly as dispatch received them.
    pub(super) input_json: String,
    /// The record's `attachments`, accepted and refused, in the order the tool named them.
    pub(super) trace: Vec<AttachmentRecord>,
    /// Every image the answer carries, in order.
    pub(super) images: Vec<SentImage>,
    /// The source the result was fenced under, as [`crate::types::DispatchOutcome::fence_source`]
    /// names it; `None` for an operator-pinned skill, whose result goes back unfenced.
    pub(super) fence_source: Option<String>,
}

/// One image a bridged answer carries as an MCP `image` content item.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct SentImage {
    /// The item's `data`: standard, padded base64 of the bytes that were checked.
    pub(super) base64: String,
    pub(super) media_type: String,
    pub(super) bytes: u64,
    /// Lowercase-hex sha256 of the bytes.
    pub(super) sha256: String,
}

impl SentImage {
    fn encode(image: &CheckedImage) -> Self {
        Self {
            base64: base64::engine::general_purpose::STANDARD.encode(&image.bytes),
            media_type: image.media_type.clone(),
            bytes: image.len(),
            sha256: image.sha256.clone(),
        }
    }

    /// What a trace record says in place of [`Self::base64`] wherever a harness's report of the
    /// result repeats it: `[image: <media_type>, <n> bytes, sha256 <hex>]`.
    pub(super) fn placeholder(&self) -> String {
        format!(
            "[image: {}, {} bytes, sha256 {}]",
            self.media_type, self.bytes, self.sha256
        )
    }
}

/// A tool-server result: `text` as one text item, then one `image` item per entry of `images`.
fn tool_result(text: String, is_error: bool, images: &[SentImage]) -> Value {
    let mut content = vec![json!({ "type": "text", "text": text })];
    content.extend(images.iter().map(
        |image| json!({ "type": "image", "data": image.base64, "mimeType": image.media_type }),
    ));
    json!({
        "content": content,
        "isError": is_error
    })
}

/// Everything the process runner needs to hand a freshly-bound bridge to the driver.
pub(super) struct BridgeHandle {
    pub(super) listener: TcpListener,
    /// e.g. `http://127.0.0.1:52344/bridge`
    pub(super) url: String,
    /// Per-run bearer token every request must present.
    pub(super) token: String,
    /// Session id echoed back on `initialize`.
    pub(super) session_id: String,
    /// The capsule's tool names, bare. The driver names them the way its harness needs.
    pub(super) tool_names: Vec<String>,
    /// MCP-shaped tool schemas advertised on `tools/list`.
    mcp_tools: Vec<Value>,
    /// The in-flight bound, [`MAX_IN_FLIGHT_CALLS`]. A field only so this module's tests can
    /// reach it with a handful of calls.
    max_in_flight_calls: usize,
}

/// Convert murmur's tool inventory (`{name, parameters, description?}`) into the tool-server
/// schema shape (`{name, description, inputSchema}`) `tools/list` answers with.
fn inventory_to_mcp_tools(inventory: &[Value]) -> Vec<Value> {
    inventory
        .iter()
        .filter_map(|t| {
            let name = t.get("name").and_then(Value::as_str)?;
            let schema = t
                .get("parameters")
                .cloned()
                .unwrap_or_else(|| json!({"type": "object"}));
            let mut tool = json!({ "name": name, "inputSchema": schema });
            if let Some(desc) = t.get("description").and_then(Value::as_str) {
                tool["description"] = Value::String(desc.to_string());
            }
            Some(tool)
        })
        .collect()
}

/// Bind the bridge on loopback (ephemeral port) and precompute what the driver is told about it.
/// Returns `None` when the capsule declares no tools — the harness then runs with none.
pub(super) async fn bind_bridge(bind_addr: &str, inventory: &[Value]) -> Option<BridgeHandle> {
    let mcp_tools = inventory_to_mcp_tools(inventory);
    if mcp_tools.is_empty() {
        return None;
    }

    // Port 0 = let the OS pick a free ephemeral port.
    let listener = TcpListener::bind(format!("{bind_addr}:0")).await.ok()?;
    let port = listener.local_addr().ok()?.port();
    let host = if bind_addr.is_empty() {
        "127.0.0.1"
    } else {
        bind_addr
    };

    let tool_names = mcp_tools
        .iter()
        .filter_map(|t| t.get("name").and_then(Value::as_str))
        .map(str::to_string)
        .collect();

    Some(BridgeHandle {
        listener,
        url: format!("http://{host}:{port}{BRIDGE_PATH}"),
        token: Uuid::new_v4().simple().to_string(),
        session_id: Uuid::new_v4().simple().to_string(),
        tool_names,
        mcp_tools,
        max_in_flight_calls: MAX_IN_FLIGHT_CALLS,
    })
}

impl BridgeHandle {
    /// Accept-and-serve loop. Never returns on its own; the process runner drops this future once
    /// the run ends, which drops every connection still being served and every call still in
    /// flight with it.
    ///
    /// Connections are served concurrently, on the caller's task: every accepted connection's
    /// handler joins one set polled beside the accept, and nothing is spawned, which is what lets
    /// `store` stay a borrow rather than something `'static`. A shell command that runs for
    /// minutes holds one handler while `initialize`, `tools/list` and other calls are answered.
    ///
    /// Two fixed bounds:
    ///
    /// - At most [`MAX_IN_FLIGHT_CALLS`] `tools/call` dispatches at once, counted from the
    ///   decision point until the response is written or the call is dropped. A call beyond that
    ///   is answered at once with a failing tool result saying nothing ran, and never reaches the
    ///   decision point or dispatch. Nothing else is refused by it.
    /// - At most [`MAX_OPEN_CONNECTIONS`] connections, each given [`REQUEST_READ_TIMEOUT`] to
    ///   deliver its request. While that many are open, no connection is accepted and new ones
    ///   wait in the kernel's backlog.
    ///
    /// A connection the harness closes while its call runs abandons the call: the call's future is
    /// dropped, a shell command still in the foreground is demoted to the background, and a
    /// [`BridgeNote::Abandoned`] goes to `notes`.
    ///
    /// `on_request` is called once per accepted connection, before it is read — a refused one
    /// included. A harness working through a long tool call writes nothing to stdout, so this is
    /// the other half of the evidence that it is still alive — see the runner's inactivity clock.
    ///
    /// `gate` is where every admitted `tools/call` asks whether it may run. The answering side
    /// runs the required-field check on each call and then, only for a capsule with a policy hook
    /// or a `read_only` path, the manifest's `read_only` check and the policy hook.
    ///
    /// `reports` gets one [`CallReport`] per dispatched call whose tool named an attachment or
    /// whose result was fenced, sent before that call's answer is written. A call refused at the
    /// decision point or at the in-flight bound, and one whose dispatch failed, sends none.
    pub(super) async fn serve(
        &self,
        store: &CapsuleStoreState,
        gate: &mpsc::UnboundedSender<GateRequest>,
        notes: &mpsc::UnboundedSender<BridgeNote>,
        reports: &mpsc::UnboundedSender<CallReport>,
        on_request: &dyn Fn(),
    ) {
        let in_flight = AtomicUsize::new(0);
        let mut open = FuturesUnordered::new();
        loop {
            tokio::select! {
                accepted = self.listener.accept(), if open.len() < MAX_OPEN_CONNECTIONS => {
                    let Ok((stream, _)) = accepted else {
                        return;
                    };
                    on_request();
                    open.push(self.handle_connection(
                        stream, store, gate, notes, reports, &in_flight,
                    ));
                }
                Some(()) = open.next(), if !open.is_empty() => {}
            }
        }
    }

    /// Manual HTTP/1.1 request handler (mirrors `identity.rs`), routing the tool-server
    /// protocol: `initialize`, `notifications/initialized`, `tools/list`, `tools/call`.
    async fn handle_connection(
        &self,
        stream: TcpStream,
        store: &CapsuleStoreState,
        gate: &mpsc::UnboundedSender<GateRequest>,
        notes: &mpsc::UnboundedSender<BridgeNote>,
        reports: &mpsc::UnboundedSender<CallReport>,
        in_flight: &AtomicUsize,
    ) {
        let (reader_half, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader_half);

        let request = match tokio::time::timeout(
            REQUEST_READ_TIMEOUT,
            self.read_request(&mut reader),
        )
        .await
        {
            Ok(Incoming::Rpc(request)) => request,
            Ok(Incoming::Reject(status)) => {
                write_response(&mut writer, status, None, None).await;
                return;
            }
            Ok(Incoming::Gone) | Err(_) => return,
        };

        let rpc_method = request.get("method").and_then(Value::as_str).unwrap_or("");
        let id = request.get("id").cloned();

        // Notifications carry no id and expect no body.
        if id.is_none() {
            write_response(&mut writer, "202 Accepted", None, None).await;
            return;
        }
        let id = id.unwrap();

        match rpc_method {
            "initialize" => {
                let protocol = request
                    .get("params")
                    .and_then(|p| p.get("protocolVersion"))
                    .and_then(Value::as_str)
                    .unwrap_or("2024-11-05");
                let result = json!({
                    "protocolVersion": protocol,
                    "capabilities": { "tools": {} },
                    "serverInfo": { "name": "murmur-bridge", "version": env!("CARGO_PKG_VERSION") }
                });
                let session_header = format!("Mcp-Session-Id: {}", self.session_id);
                write_json_rpc(&mut writer, &id, result, Some(&session_header)).await;
            }
            "tools/list" => {
                write_json_rpc(&mut writer, &id, json!({ "tools": self.mcp_tools }), None).await;
            }
            "tools/call" => {
                let (tool_name, input_json) = call_target(request.get("params"));
                let Some(_slot) = InFlightSlot::claim(in_flight, self.max_in_flight_calls) else {
                    let limit = self.max_in_flight_calls;
                    let _ = notes.send(BridgeNote::AtCapacity { tool_name, limit });
                    write_json_rpc(
                        &mut writer,
                        &id,
                        error_result(at_capacity_text(limit)),
                        None,
                    )
                    .await;
                    return;
                };

                let abandon = AbandonSignal::new();
                let decided = AtomicBool::new(false);
                let started = Instant::now();
                let dispatch = self.dispatch_tool_call(
                    &tool_name,
                    input_json,
                    store,
                    gate,
                    abandon.clone(),
                    &decided,
                );
                // A client that half-closes its write side after sending its request reads here
                // as one that went away, and its call is abandoned: a client must keep the
                // connection open until the response arrives.
                let result = tokio::select! {
                    result = dispatch => Some(result),
                    () = peer_closed(&mut reader) => None,
                };
                match result {
                    Some((result, report)) => {
                        // Sent before the answer is written: the harness reports the call's
                        // result only after reading the answer, so the report always reaches the
                        // driving task ahead of that result.
                        if let Some(report) = report {
                            let _ = reports.send(report);
                        }
                        write_json_rpc(&mut writer, &id, result, None).await;
                    }
                    None => {
                        // The dispatch future is already dropped. A shell command it left on the
                        // blocking pool reads this at its next poll and demotes.
                        abandon.raise();
                        let _ = notes.send(BridgeNote::Abandoned {
                            tool_name,
                            elapsed_ms: started
                                .elapsed()
                                .as_millis()
                                .try_into()
                                .unwrap_or(u64::MAX),
                            decided: decided.load(Ordering::SeqCst),
                        });
                    }
                }
            }
            _ => {
                let err = json!({
                    "jsonrpc": "2.0", "id": id,
                    "error": { "code": -32601, "message": "method not found" }
                });
                write_response(
                    &mut writer,
                    "200 OK",
                    Some("application/json"),
                    Some(&err.to_string()),
                )
                .await;
            }
        }
    }

    /// Read one request: request line, headers and body. The caller bounds the whole read by
    /// [`REQUEST_READ_TIMEOUT`].
    async fn read_request(&self, reader: &mut BufReader<OwnedReadHalf>) -> Incoming {
        // Request line: "<METHOD> <PATH> HTTP/1.1"
        let mut request_line = String::new();
        if reader.read_line(&mut request_line).await.is_err() {
            return Incoming::Gone;
        }
        let mut parts = request_line.split_whitespace();
        let method = parts.next().unwrap_or("").to_string();

        // Headers
        let mut content_length = 0usize;
        let mut authorized = false;
        loop {
            let mut line = String::new();
            match reader.read_line(&mut line).await {
                Ok(0) | Err(_) => break,
                Ok(_) => {}
            }
            if line.trim().is_empty() {
                break;
            }
            let lower = line.to_ascii_lowercase();
            let lower = lower.trim_end();
            if let Some(rest) = lower.strip_prefix("content-length:") {
                content_length = rest.trim().parse().unwrap_or(0);
            } else if let Some(rest) = lower.strip_prefix("authorization:") {
                authorized = rest.trim() == format!("bearer {}", self.token);
            }
        }

        // A client may open a GET stream for server->client messages; this bridge needs none.
        if method == "GET" {
            return Incoming::Reject("405 Method Not Allowed");
        }

        if !authorized {
            return Incoming::Reject("401 Unauthorized");
        }

        let mut body = vec![0u8; content_length];
        if content_length > 0 && reader.read_exact(&mut body).await.is_err() {
            return Incoming::Gone;
        }
        match serde_json::from_slice(&body) {
            Ok(request) => Incoming::Rpc(request),
            Err(_) => Incoming::Reject("400 Bad Request"),
        }
    }

    /// Execute one `tools/call` through murmur's WASM tool dispatch — the same executor the
    /// HTTP transport uses — and shape the outcome as a tool-server result. Tool execution
    /// stays entirely in murmur's sandbox under the capsule's declared capabilities; the harness
    /// never runs anything itself.
    ///
    /// The session's decision point comes first, over `gate`, and `decided` is set once it has
    /// answered. A refusal is returned as the failing tool result the harness reports back to its
    /// model, carrying the refusal's own text unaltered — its no-retry sentence is what stops the
    /// model asking again.
    ///
    /// `abandon` is raised by the caller if the harness goes away; see
    /// [`CapsuleStoreState::dispatch_bridged_tool_async`].
    ///
    /// Returns the result with the [`CallReport`] of a dispatched call whose tool named an
    /// attachment or whose result was fenced.
    async fn dispatch_tool_call(
        &self,
        name: &str,
        input_json: String,
        store: &CapsuleStoreState,
        gate: &mpsc::UnboundedSender<GateRequest>,
        abandon: AbandonSignal,
        decided: &AtomicBool,
    ) -> (Value, Option<CallReport>) {
        // A refusal means nothing ran, and its trace line is written on the answering side. The
        // harness still reports the failure as a tool result of its own, which the event sink
        // records as a `tool_call` with `status: error`: this request carries no id tying it to
        // that result, so suppressing the pair would mean guessing.
        let refusal = ask_gate(gate, name, &input_json).await;
        decided.store(true, Ordering::SeqCst);
        if let Some(refusal) = refusal {
            return (error_result(refusal), None);
        }

        match store
            .dispatch_bridged_tool_async(
                name,
                ToolInput {
                    data: Some(input_json.clone()),
                    log_path: None,
                },
                abandon,
            )
            .await
        {
            Ok(mut outcome) => {
                let is_error = !matches!(outcome.result.status, Status::Passed);
                let attachments = std::mem::take(&mut outcome.result.attachments);
                let fence_source = outcome.fence_source.take();
                // `outcome.fatal` is deliberately not acted on here: this bridge is a tool server
                // for an external harness process and owns no murmur session to end. The
                // failure still reaches the caller in full — `result.data` carries the same named
                // text (`RuntimeError`'s Display) that the agent loop would end the session with.
                let text = outcome
                    .result
                    .data
                    .or(outcome.result.summary)
                    .unwrap_or_else(|| "tool returned no data".to_string());
                answer(
                    store,
                    name,
                    input_json,
                    text,
                    is_error,
                    &attachments,
                    fence_source,
                )
            }
            Err(err) => (error_result(format!("tool '{name}' failed: {err}")), None),
        }
    }
}

/// The result of a dispatched call whose tool returned `text`, and its [`CallReport`] when the
/// tool named any `attachments` or the result was fenced under `fence_source`.
///
/// Each attachment adds one line after the fence's closing marker: `[image attached: …]` for one
/// that passes the `context.images` checks, the refusal note for one that does not. Each accepted
/// one also becomes an `image` item after the text item, carried whatever `is_error` says. With no
/// attachment the result is the text item alone. Nothing is stored.
///
/// `fence_source` reaches the report only: the result is the same whatever it says.
fn answer(
    store: &CapsuleStoreState,
    name: &str,
    input_json: String,
    mut text: String,
    is_error: bool,
    attachments: &[Attachment],
    fence_source: Option<String>,
) -> (Value, Option<CallReport>) {
    let checked = crate::images::check_bridged_attachments(
        attachments
            .iter()
            .map(|attachment| (attachment.media_type.as_str(), attachment.path.as_str())),
        &store.accessible_root,
        &store.image_limits,
    );
    for line in &checked.lines {
        text.push('\n');
        text.push_str(line);
    }
    let images: Vec<SentImage> = checked.images.iter().map(SentImage::encode).collect();
    let result = tool_result(text, is_error, &images);
    let report = (!checked.trace.is_empty() || fence_source.is_some()).then(|| CallReport {
        tool_name: name.to_string(),
        input_json,
        trace: checked.trace,
        images,
        fence_source,
    });
    (result, report)
}

/// A `tools/call`'s tool name and its arguments serialised for dispatch.
///
/// The arguments object is forwarded verbatim as the tool input payload, exactly like the HTTP
/// path forwards `tool_use.input`.
fn call_target(params: Option<&Value>) -> (String, String) {
    let name = params
        .and_then(|p| p.get("name"))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    let arguments = params
        .and_then(|p| p.get("arguments"))
        .cloned()
        .unwrap_or_else(|| json!({}));
    let input_json = serde_json::to_string(&arguments).unwrap_or_else(|_| "{}".to_string());
    (name, input_json)
}

/// Resolves when the peer has gone away: a read on the request's read half reports end of stream
/// or fails.
///
/// Bytes sent after the request are read and discarded, up to [`TRAILING_BYTES_CAP`]; past that
/// the connection is no longer watched and this never resolves.
async fn peer_closed(reader: &mut BufReader<OwnedReadHalf>) {
    let mut scratch = [0u8; 1024];
    let mut discarded = 0usize;
    while discarded <= TRAILING_BYTES_CAP {
        match reader.read(&mut scratch).await {
            Ok(0) | Err(_) => return,
            Ok(read) => discarded += read,
        }
    }
    std::future::pending().await
}

/// Write a JSON-RPC 2.0 success response as an `application/json` HTTP reply.
async fn write_json_rpc(
    writer: &mut tokio::net::tcp::OwnedWriteHalf,
    id: &Value,
    result: Value,
    extra_header: Option<&str>,
) {
    let payload = json!({ "jsonrpc": "2.0", "id": id, "result": result }).to_string();
    write_response_with_header(
        writer,
        "200 OK",
        Some("application/json"),
        Some(&payload),
        extra_header,
    )
    .await;
}

async fn write_response(
    writer: &mut tokio::net::tcp::OwnedWriteHalf,
    status: &str,
    content_type: Option<&str>,
    body: Option<&str>,
) {
    write_response_with_header(writer, status, content_type, body, None).await;
}

/// Build and write a minimal HTTP/1.1 response with `connection: close` (so the client opens a
/// fresh connection per request — the simplest correct behaviour for this short-lived server).
async fn write_response_with_header(
    writer: &mut tokio::net::tcp::OwnedWriteHalf,
    status: &str,
    content_type: Option<&str>,
    body: Option<&str>,
    extra_header: Option<&str>,
) {
    let body = body.unwrap_or("");
    let mut response = format!("HTTP/1.1 {status}\r\n");
    if let Some(ct) = content_type {
        response.push_str(&format!("content-type: {ct}\r\n"));
    }
    if let Some(h) = extra_header {
        response.push_str(h);
        response.push_str("\r\n");
    }
    response.push_str(&format!("content-length: {}\r\n", body.len()));
    response.push_str("connection: close\r\n\r\n");
    response.push_str(body);
    let _ = writer.write_all(response.as_bytes()).await;
    let _ = writer.shutdown().await;
}

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

    #[test]
    fn inventory_maps_to_mcp_tool_shape() {
        // murmur inventory uses `parameters`; the tool-server protocol expects `inputSchema`.
        let inventory = vec![json!({
            "name": "murmur-tool-editor",
            "description": "edits files",
            "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}
        })];
        let mcp = inventory_to_mcp_tools(&inventory);
        assert_eq!(mcp.len(), 1);
        assert_eq!(mcp[0]["name"], "murmur-tool-editor");
        assert_eq!(mcp[0]["description"], "edits files");
        assert_eq!(mcp[0]["inputSchema"]["type"], "object");
        assert!(mcp[0].get("parameters").is_none());
    }

    fn test_store(dir: &tempfile::TempDir) -> CapsuleStoreState {
        crate::runtime::build_test_state(
            Arc::new(murmur_artifact::LocalRegistry::new(
                dir.path().join("registry"),
            )),
            dir.path().to_path_buf(),
            dir.path().join("murmur.lock"),
        )
    }

    const PNG: &[u8] = &[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A, 0, 0];

    fn png(index: u8) -> Vec<u8> {
        let mut bytes = PNG.to_vec();
        bytes.push(index);
        bytes
    }

    fn attachment(media_type: &str, path: &str) -> Attachment {
        Attachment {
            media_type: media_type.to_string(),
            path: path.to_string(),
        }
    }

    fn decoded(item: &Value) -> Vec<u8> {
        base64::engine::general_purpose::STANDARD
            .decode(item["data"].as_str().expect("a data string"))
            .expect("standard base64")
    }

    /// An attachment that passes the checks becomes the `[image attached: …]` line and one that
    /// does not its refusal note, and the bridge stores nothing: no `blobs/` and no conversation
    /// record appear anywhere under the session.
    #[test]
    fn claude_bridge_attachment_lines_check_and_store_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        std::fs::write(store.accessible_root.host_path("shot.png"), PNG).unwrap();
        std::fs::write(store.accessible_root.host_path("note.png"), b"not an image").unwrap();
        let (result, _) = answer(
            &store,
            "t",
            "{}".to_string(),
            "fenced result".to_string(),
            false,
            &[
                attachment("image/png", "shot.png"),
                attachment("image/png", "note.png"),
            ],
            None,
        );
        let text = result["content"][0]["text"].as_str().unwrap();
        let lines: Vec<&str> = text.lines().collect();
        assert_eq!(lines[0], "fenced result");
        assert_eq!(lines[1], "[image attached: shot.png, image/png, 10 bytes]");
        assert!(
            lines[2].starts_with("attachment 'note.png' was not attached: ")
                && lines[2].ends_with("(content_mismatch)"),
            "{}",
            lines[2]
        );
        let written: Vec<_> = walk(dir.path())
            .into_iter()
            .filter(|path| {
                path.components()
                    .any(|c| c.as_os_str() == "blobs" || c.as_os_str() == "conversations")
            })
            .collect();
        assert!(written.is_empty(), "the bridge stored {written:?}");
    }

    /// An accepted PNG is the text item and then one MCP image item holding exactly its bytes, and
    /// the report carries the same base64 with the trace entry.
    #[test]
    fn an_accepted_png_follows_the_text_as_an_image_item() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        std::fs::write(store.accessible_root.host_path("a.png"), PNG).unwrap();
        let (result, report) = answer(
            &store,
            "png-tool",
            r#"{"a":1}"#.to_string(),
            "fenced".to_string(),
            false,
            &[attachment("image/png", "a.png")],
            None,
        );
        let content = result["content"].as_array().unwrap();
        assert_eq!(content.len(), 2, "{result}");
        assert_eq!(content[0]["type"], "text");
        assert!(content[0]["text"]
            .as_str()
            .unwrap()
            .ends_with("[image attached: a.png, image/png, 10 bytes]"));
        let keys: Vec<&String> = content[1].as_object().unwrap().keys().collect();
        let mut sorted = keys.clone();
        sorted.sort();
        assert_eq!(sorted, ["data", "mimeType", "type"]);
        assert_eq!(content[1]["type"], "image");
        assert_eq!(content[1]["mimeType"], "image/png");
        assert_eq!(decoded(&content[1]), PNG);
        assert_eq!(result["isError"], false);

        let report = report.expect("a call with an attachment is reported");
        assert_eq!(report.tool_name, "png-tool");
        assert_eq!(report.input_json, r#"{"a":1}"#);
        let sha = murmur_artifact::sha256_hex(PNG);
        assert_eq!(
            report.trace,
            [AttachmentRecord {
                path: "a.png".to_string(),
                media_type: "image/png".to_string(),
                bytes: Some(10),
                sha256: Some(sha.clone()),
                refused: None,
            }]
        );
        assert_eq!(report.images.len(), 1);
        assert_eq!(report.images[0].base64, content[1]["data"]);
        assert_eq!(report.images[0].sha256, sha);
        assert_eq!(
            report.images[0].placeholder(),
            format!("[image: image/png, 10 bytes, sha256 {sha}]")
        );
    }

    #[test]
    fn two_attachments_are_two_image_items_in_order() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        std::fs::write(store.accessible_root.host_path("one.png"), png(1)).unwrap();
        std::fs::write(store.accessible_root.host_path("two.png"), png(2)).unwrap();
        let (result, _) = answer(
            &store,
            "t",
            "{}".to_string(),
            "fenced".to_string(),
            false,
            &[
                attachment("image/png", "one.png"),
                attachment("image/png", "two.png"),
            ],
            None,
        );
        let content = result["content"].as_array().unwrap();
        assert_eq!(content.len(), 3);
        assert_eq!(decoded(&content[1]), png(1));
        assert_eq!(decoded(&content[2]), png(2));
    }

    #[test]
    fn a_failed_call_still_carries_its_image() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        std::fs::write(store.accessible_root.host_path("a.png"), PNG).unwrap();
        let (result, report) = answer(
            &store,
            "t",
            "{}".to_string(),
            "it failed".to_string(),
            true,
            &[attachment("image/png", "a.png")],
            None,
        );
        assert_eq!(result["isError"], true);
        assert_eq!(result["content"][1]["type"], "image");
        assert!(report.is_some());
    }

    /// A tool that attaches nothing and was not fenced gets the text item alone, byte for byte,
    /// and no report.
    #[test]
    fn without_attachments_the_result_is_the_text_item_alone() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        for is_error in [false, true] {
            let (result, report) = answer(
                &store,
                "t",
                "{}".to_string(),
                "fenced result".to_string(),
                is_error,
                &[],
                None,
            );
            assert_eq!(
                result.to_string(),
                json!({
                    "content": [{ "type": "text", "text": "fenced result" }],
                    "isError": is_error
                })
                .to_string()
            );
            assert!(report.is_none());
        }
    }

    /// A fenced result with no attachment is reported with its fence source and nothing else,
    /// and the harness's answer is the one an unfenced call gets.
    #[test]
    fn a_fenced_call_is_reported_without_changing_its_answer() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        let text = "<untrusted-content source=tool:t>\ndone\n</untrusted-content>".to_string();
        for is_error in [false, true] {
            let call = |fence_source: Option<String>| {
                answer(
                    &store,
                    "t",
                    r#"{"a":1}"#.to_string(),
                    text.clone(),
                    is_error,
                    &[],
                    fence_source,
                )
            };
            let (unlabelled, none) = call(None);
            let (result, report) = call(Some("tool:t".to_string()));
            assert_eq!(result, unlabelled);
            assert!(none.is_none());
            assert_eq!(
                report,
                Some(CallReport {
                    tool_name: "t".to_string(),
                    input_json: r#"{"a":1}"#.to_string(),
                    trace: Vec::new(),
                    images: Vec::new(),
                    fence_source: Some("tool:t".to_string()),
                })
            );
        }
    }

    /// A file over `context.images.max_bytes` is refused `too_large` and returned as no item; a
    /// refusal alone is still reported, as the trace's refused entry.
    #[test]
    fn a_file_over_max_bytes_is_refused_and_not_encoded() {
        let dir = tempfile::tempdir().unwrap();
        let mut store = test_store(&dir);
        store.image_limits.max_bytes = 1024;
        let mut big = PNG.to_vec();
        big.resize(2048, 0);
        std::fs::write(store.accessible_root.host_path("big.png"), &big).unwrap();
        let (result, report) = answer(
            &store,
            "t",
            "{}".to_string(),
            "fenced".to_string(),
            false,
            &[attachment("image/png", "big.png")],
            None,
        );
        assert_eq!(result["content"].as_array().unwrap().len(), 1);
        let text = result["content"][0]["text"].as_str().unwrap();
        assert!(
            text.contains("2048 bytes is over context.images.max_bytes of 1024 bytes")
                && text.ends_with("(too_large)"),
            "{text}"
        );
        let report = report.expect("a refused attachment is reported");
        assert!(report.images.is_empty());
        assert_eq!(report.trace[0].refused, Some("too_large"));
    }

    /// `context.images.max_per_turn` bounds the image items of one call.
    #[test]
    fn the_image_past_max_per_turn_in_one_call_is_refused_with_turn_limit() {
        let dir = tempfile::tempdir().unwrap();
        let store = test_store(&dir);
        let mut attachments = Vec::new();
        for index in 0..6u8 {
            let name = format!("{index}.png");
            std::fs::write(store.accessible_root.host_path(&name), png(index)).unwrap();
            attachments.push(attachment("image/png", &name));
        }
        let (result, report) = answer(
            &store,
            "t",
            "{}".to_string(),
            "fenced".to_string(),
            false,
            &attachments,
            None,
        );
        let content = result["content"].as_array().unwrap();
        assert_eq!(content.len(), 6, "one text item and five images");
        for (index, item) in content[1..].iter().enumerate() {
            assert_eq!(decoded(item), png(index as u8));
        }
        let text = content[0]["text"].as_str().unwrap();
        let last = text.lines().last().unwrap();
        assert!(
            last.starts_with("attachment '5.png' was not attached: ")
                && last.ends_with("(turn_limit)"),
            "{last}"
        );
        let report = report.unwrap();
        assert_eq!(report.images.len(), 5);
        assert_eq!(report.trace.len(), 6);
        assert_eq!(report.trace[5].refused, Some("turn_limit"));
    }

    fn walk(root: &std::path::Path) -> Vec<std::path::PathBuf> {
        let mut found = Vec::new();
        let mut stack = vec![root.to_path_buf()];
        while let Some(dir) = stack.pop() {
            for entry in std::fs::read_dir(&dir).into_iter().flatten().flatten() {
                let path = entry.path();
                if path.is_dir() {
                    stack.push(path.clone());
                }
                found.push(path);
            }
        }
        found
    }

    #[test]
    fn inventory_without_schema_defaults_to_object() {
        let inventory = vec![json!({ "name": "t" })];
        let mcp = inventory_to_mcp_tools(&inventory);
        assert_eq!(mcp[0]["inputSchema"]["type"], "object");
    }

    use std::{net::SocketAddr, sync::Arc};

    use tokio::sync::mpsc::error::TryRecvError;

    /// A call must be answered inside this for the bridge to count as not holding it up.
    const PROMPT: Duration = Duration::from_secs(1);

    /// A bridge serving one `echo-tool`, with a store behind it that no test dispatches into: the
    /// test answers the decision point itself, always with a refusal.
    struct Fixture {
        _dir: tempfile::TempDir,
        store: CapsuleStoreState,
        handle: BridgeHandle,
        addr: SocketAddr,
    }

    impl Fixture {
        async fn new() -> Self {
            let dir = tempfile::tempdir().unwrap();
            let store = crate::runtime::build_test_state(
                Arc::new(murmur_artifact::LocalRegistry::new(
                    dir.path().join("registry"),
                )),
                dir.path().to_path_buf(),
                dir.path().join("murmur.lock"),
            );
            let handle = bind_bridge(
                "127.0.0.1",
                &[json!({"name": "echo-tool", "parameters": {"type": "object"}})],
            )
            .await
            .expect("bridge binds");
            let addr = handle.listener.local_addr().unwrap();
            Self {
                _dir: dir,
                store,
                handle,
                addr,
            }
        }

        fn client(&self) -> Client {
            Client {
                addr: self.addr,
                token: self.handle.token.clone(),
            }
        }
    }

    /// A raw HTTP/1.1 client that speaks to the bridge the way a harness does: one request per
    /// connection, read to EOF.
    #[derive(Clone)]
    struct Client {
        addr: SocketAddr,
        token: String,
    }

    impl Client {
        fn request(&self, body: &Value) -> String {
            let body = body.to_string();
            format!(
                "POST {BRIDGE_PATH} HTTP/1.1\r\nhost: {}\r\nauthorization: Bearer {}\r\n\
                 content-type: application/json\r\ncontent-length: {}\r\n\r\n{body}",
                self.addr,
                self.token,
                body.len()
            )
        }

        /// Open a connection and send one JSON-RPC request on it, leaving it open.
        async fn send(&self, body: Value) -> TcpStream {
            let mut stream = TcpStream::connect(self.addr).await.unwrap();
            stream
                .write_all(self.request(&body).as_bytes())
                .await
                .unwrap();
            stream
        }

        async fn post(&self, body: Value) -> Value {
            let mut raw = Vec::new();
            let mut stream = self.send(body).await;
            stream.read_to_end(&mut raw).await.unwrap();
            response_body(raw)
        }

        /// A `tools/call` on a plain thread of its own, so the test can go on while it is parked.
        /// The answer arrives on the returned receiver.
        fn call_in_background(&self, msg: &str) -> oneshot::Receiver<Value> {
            use std::io::{Read, Write};

            let (answer, answered) = oneshot::channel();
            let request = self.request(&tool_call(msg));
            let addr = self.addr;
            std::thread::spawn(move || {
                let mut stream = std::net::TcpStream::connect(addr).unwrap();
                stream.write_all(request.as_bytes()).unwrap();
                let mut raw = Vec::new();
                stream.read_to_end(&mut raw).unwrap();
                let _ = answer.send(response_body(raw));
            });
            answered
        }
    }

    /// The JSON body of a whole HTTP response.
    fn response_body(raw: Vec<u8>) -> Value {
        let raw = String::from_utf8(raw).unwrap();
        let (_, body) = raw.split_once("\r\n\r\n").expect("a whole response");
        serde_json::from_str(body).unwrap_or_else(|_| panic!("a JSON body, got {raw:?}"))
    }

    fn tool_call(msg: &str) -> Value {
        json!({
            "jsonrpc": "2.0", "id": 1, "method": "tools/call",
            "params": {"name": "echo-tool", "arguments": {"msg": msg}}
        })
    }

    fn text_of(response: &Value) -> &str {
        response["result"]["content"][0]["text"]
            .as_str()
            .unwrap_or("")
    }

    async fn next_gate(gate: &mut mpsc::UnboundedReceiver<GateRequest>) -> GateRequest {
        tokio::time::timeout(PROMPT, gate.recv())
            .await
            .expect("the call reached the decision point in time")
            .expect("the gate channel is open")
    }

    async fn within<T>(what: &str, future: impl std::future::Future<Output = T>) -> T {
        tokio::time::timeout(PROMPT, future)
            .await
            .unwrap_or_else(|_| panic!("{what} was not answered within {PROMPT:?}"))
    }

    #[tokio::test]
    async fn a_parked_call_does_not_block_the_next_connection() {
        let fixture = Fixture::new().await;
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, _note_rx) = mpsc::unbounded_channel();
        let (report_tx, _report_rx) = mpsc::unbounded_channel();
        let accepted = AtomicUsize::new(0);
        let on_request = || {
            accepted.fetch_add(1, Ordering::SeqCst);
        };
        let serve =
            fixture
                .handle
                .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &on_request);

        let body = async {
            let parked = client.call_in_background("A");
            let a = next_gate(&mut gate_rx).await;

            let initialized = within(
                "initialize",
                client.post(json!({
                    "jsonrpc": "2.0", "id": 2, "method": "initialize",
                    "params": {"protocolVersion": "2025-03-26"}
                })),
            )
            .await;
            assert_eq!(initialized["result"]["protocolVersion"], "2025-03-26");

            let listed = within(
                "tools/list",
                client.post(json!({"jsonrpc": "2.0", "id": 3, "method": "tools/list"})),
            )
            .await;
            assert_eq!(listed["result"]["tools"][0]["name"], "echo-tool");

            let second = client.call_in_background("B");
            let b = next_gate(&mut gate_rx).await;
            assert!(b.input_json.contains("\"B\""));

            let _ = a.reply.send(Some("refused".to_string()));
            let _ = b.reply.send(Some("refused".to_string()));
            within("call A", parked).await.unwrap();
            within("call B", second).await.unwrap();
        };
        tokio::select! {
            () = serve => panic!("serve returned"),
            () = body => {}
        }
        assert_eq!(
            accepted.load(Ordering::SeqCst),
            4,
            "one per connection opened"
        );
    }

    #[tokio::test]
    async fn each_parked_call_gets_its_own_verdict() {
        let fixture = Fixture::new().await;
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, _note_rx) = mpsc::unbounded_channel();
        let (report_tx, _report_rx) = mpsc::unbounded_channel();
        let serve = fixture
            .handle
            .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &|| {});

        let body = async {
            let call_a = client.call_in_background("A");
            let a = next_gate(&mut gate_rx).await;
            let call_b = client.call_in_background("B");
            let b = next_gate(&mut gate_rx).await;
            assert!(a.input_json.contains("\"A\"") && b.input_json.contains("\"B\""));

            // Answered in the reverse of arrival order.
            let _ = b.reply.send(Some("refused-B".to_string()));
            let _ = a.reply.send(Some("refused-A".to_string()));

            let answer_a = within("call A", call_a).await.unwrap();
            let answer_b = within("call B", call_b).await.unwrap();
            assert_eq!(text_of(&answer_a), "refused-A");
            assert_eq!(text_of(&answer_b), "refused-B");
            assert_eq!(answer_a["result"]["isError"], true);
            assert_eq!(answer_b["result"]["isError"], true);
        };
        tokio::select! {
            () = serve => panic!("serve returned"),
            () = body => {}
        }
    }

    #[tokio::test]
    async fn a_call_above_the_bound_is_refused_at_once() {
        let mut fixture = Fixture::new().await;
        fixture.handle.max_in_flight_calls = 2;
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, mut note_rx) = mpsc::unbounded_channel();
        let (report_tx, _report_rx) = mpsc::unbounded_channel();
        let serve = fixture
            .handle
            .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &|| {});

        let body = async {
            let call_a = client.call_in_background("A");
            let a = next_gate(&mut gate_rx).await;
            let call_b = client.call_in_background("B");
            let b = next_gate(&mut gate_rx).await;

            let refused = within("the third call", client.post(tool_call("C"))).await;
            assert_eq!(refused["result"]["isError"], true);
            let text = text_of(&refused);
            assert!(text.contains('2'), "names the limit: {text}");
            assert!(text.contains("nothing ran"), "says nothing ran: {text}");

            tokio::time::sleep(Duration::from_millis(100)).await;
            assert!(
                matches!(gate_rx.try_recv(), Err(TryRecvError::Empty)),
                "a call refused at the bound never reaches the decision point"
            );
            assert_eq!(
                note_rx.try_recv().ok(),
                Some(BridgeNote::AtCapacity {
                    tool_name: "echo-tool".to_string(),
                    limit: 2
                })
            );

            let listed = within(
                "tools/list",
                client.post(json!({"jsonrpc": "2.0", "id": 3, "method": "tools/list"})),
            )
            .await;
            assert_eq!(listed["result"]["tools"][0]["name"], "echo-tool");

            let _ = a.reply.send(Some("refused-A".to_string()));
            within("call A", call_a).await.unwrap();

            let call_d = client.call_in_background("D");
            let d = next_gate(&mut gate_rx).await;
            assert!(
                d.input_json.contains("\"D\""),
                "a freed slot admits the next call"
            );

            let _ = b.reply.send(Some("refused".to_string()));
            let _ = d.reply.send(Some("refused".to_string()));
            within("call B", call_b).await.unwrap();
            within("call D", call_d).await.unwrap();
        };
        tokio::select! {
            () = serve => panic!("serve returned"),
            () = body => {}
        }
    }

    #[tokio::test]
    async fn a_closed_connection_abandons_its_call() {
        let fixture = Fixture::new().await;
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, mut note_rx) = mpsc::unbounded_channel();
        let (report_tx, _report_rx) = mpsc::unbounded_channel();
        let serve = fixture
            .handle
            .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &|| {});

        let body = async {
            let connection = client.send(tool_call("A")).await;
            let mut a = next_gate(&mut gate_rx).await;
            drop(connection);

            within("the closed reply", a.reply.closed()).await;
            let note = within("the abandonment note", note_rx.recv())
                .await
                .expect("the note channel is open");
            match note {
                BridgeNote::Abandoned {
                    tool_name, decided, ..
                } => {
                    assert_eq!(tool_name, "echo-tool");
                    assert!(!decided, "the decision point had not answered");
                }
                other => panic!("expected an abandonment note, got {other:?}"),
            }
        };
        tokio::select! {
            () = serve => panic!("serve returned"),
            () = body => {}
        }
    }

    #[tokio::test]
    async fn dropping_serve_cancels_parked_calls() {
        let fixture = Fixture::new().await;
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, _note_rx) = mpsc::unbounded_channel();
        let (report_tx, _report_rx) = mpsc::unbounded_channel();
        let mut serve =
            Box::pin(
                fixture
                    .handle
                    .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &|| {}),
            );

        let _connection = client.send(tool_call("A")).await;
        let a = tokio::select! {
            () = &mut serve => panic!("serve returned"),
            a = next_gate(&mut gate_rx) => a,
        };
        assert!(!a.reply.is_closed());
        drop(serve);
        assert!(
            a.reply.is_closed(),
            "dropping serve drops the call waiting on this verdict"
        );
    }

    /// A call refused by the decision point, one refused at the in-flight bound, and one whose
    /// dispatch fails each answer with the one text item, and none sends a report.
    #[tokio::test]
    async fn refused_and_failed_calls_carry_no_image_and_send_no_report() {
        let mut fixture = Fixture::new().await;
        fixture.handle.max_in_flight_calls = 1;
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, _note_rx) = mpsc::unbounded_channel();
        let (report_tx, mut report_rx) = mpsc::unbounded_channel::<CallReport>();
        let serve = fixture
            .handle
            .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &|| {});

        let body = async {
            let refused = client.call_in_background("A");
            let a = next_gate(&mut gate_rx).await;
            let at_capacity = within("the second call", client.post(tool_call("B"))).await;
            let _ = a.reply.send(Some("refused-A".to_string()));
            let refused = within("call A", refused).await.unwrap();

            // No `echo-tool` is installed in the fixture's store, so its dispatch fails.
            let failed = client.call_in_background("C");
            let c = next_gate(&mut gate_rx).await;
            let _ = c.reply.send(None);
            let failed = within("call C", failed).await.unwrap();

            for answer in [&refused, &at_capacity, &failed] {
                let content = answer["result"]["content"].as_array().unwrap();
                assert_eq!(content.len(), 1, "{answer}");
                assert_eq!(content[0]["type"], "text");
                assert_eq!(answer["result"]["isError"], true, "{answer}");
            }
            assert!(
                matches!(report_rx.try_recv(), Err(TryRecvError::Empty)),
                "no call named an attachment"
            );
        };
        tokio::select! {
            () = serve => panic!("serve returned"),
            () = body => {}
        }
    }

    /// A call the decision point lets through and a tool answers sends the harness the fenced
    /// result, and reports it under the bare name and arguments with the source it was fenced
    /// under. `session-status` is the runtime's own tool, so the fixture's store can answer it.
    #[tokio::test]
    async fn a_dispatched_call_reports_the_fence_its_answer_carries() {
        let mut fixture = Fixture::new().await;
        fixture.store.session_status = Some(Arc::new(crate::session_status::SessionStatus::new(
            Instant::now(),
            true,
            None,
            None,
            0,
            None,
            None,
        )));
        let client = fixture.client();
        let (gate_tx, mut gate_rx) = mpsc::unbounded_channel();
        let (note_tx, _note_rx) = mpsc::unbounded_channel();
        let (report_tx, mut report_rx) = mpsc::unbounded_channel::<CallReport>();
        let serve = fixture
            .handle
            .serve(&fixture.store, &gate_tx, &note_tx, &report_tx, &|| {});

        let body = async {
            let (answered, answer) = oneshot::channel();
            let call = client.clone();
            tokio::spawn(async move {
                let response = call
                    .post(json!({
                        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
                        "params": {"name": "session-status", "arguments": {}}
                    }))
                    .await;
                let _ = answered.send(response);
            });
            let gate = next_gate(&mut gate_rx).await;
            let _ = gate.reply.send(None);
            let response = within("the call", answer).await.unwrap();

            let text = text_of(&response);
            assert!(
                text.starts_with("<untrusted-content source=tool:session-status>\n"),
                "{text}"
            );
            assert_eq!(response["result"]["isError"], false, "{response}");
            let report = report_rx.try_recv().expect("the call is reported");
            assert_eq!(report.tool_name, "session-status");
            assert_eq!(report.input_json, "{}");
            assert_eq!(report.fence_source.as_deref(), Some("tool:session-status"));
            assert!(report.trace.is_empty() && report.images.is_empty());
        };
        tokio::select! {
            () = serve => panic!("serve returned"),
            () = body => {}
        }
    }

    #[tokio::test]
    async fn bind_bridge_is_none_without_tools() {
        // No tools declared → no bridge, so the harness runs with no tools.
        assert!(bind_bridge("127.0.0.1", &[]).await.is_none());
    }

    /// The driver is handed bare names and builds whatever its harness needs from them: the
    /// runtime never spells a harness's tool naming.
    #[tokio::test]
    async fn bind_bridge_names_tools_bare() {
        let inventory = vec![json!({"name": "editor", "parameters": {"type": "object"}})];
        let handle = bind_bridge("127.0.0.1", &inventory)
            .await
            .expect("bridge should bind when tools are declared");
        assert_eq!(handle.tool_names, vec!["editor"]);
        assert!(handle.url.starts_with("http://127.0.0.1:"));
        assert!(!handle.token.is_empty());
    }
}