import Foundation

// Network.framework carries the TCP listener on Darwin; on Linux the TCP path is a BSD-socket
// mirror of the AF_UNIX path below (CLOUD_RUNTIME §3.1's approval-server seam) — same accept
// source, same `ByteConn` pipeline, so the HTTP/JSON-RPC dispatch is byte-identical. On
// Windows the TCP path rides SwiftNIO instead (docs/WINDOWS_PORT.md §4.3 — Winsock SOCKETs
// aren't POSIX fds, so the BSD mirror can't serve there), and the AF_UNIX transport doesn't
// exist at all: the control plane is gateway TCP end-to-end (§5).
#if canImport(Network)
import Network
#endif
#if canImport(Glibc)
import Glibc
#elseif canImport(Musl)
import Musl
#endif
#if os(Windows)
import NIOCore
import NIOPosix
#endif

// The libc socket surface differs in spelling across platforms; alias the few call sites the
// server needs so both bind paths read the same. `bind`/`close` are qualified through helpers
// because actor methods (`bind(host:)`) and connection methods (`close()`) shadow them.
// Three C libraries: Darwin (Apple), Glibc (Linux distro builds), Musl (the static-linux SDK —
// the runner image's fully-static amd64 build). Windows has none of them and needs none: every
// caller of these helpers is a BSD-socket or AF_UNIX path that `#if !os(Windows)` already
// excludes there, so the whole ladder is skipped rather than falling into the Glibc arm.
#if os(Windows)
// (no POSIX socket helpers — the Windows transport is NIO, see `performNIOBind`)
#elseif canImport(Darwin)
private let sockStreamType: Int32 = SOCK_STREAM
private func posixBind(_ fd: Int32, _ addr: UnsafePointer<sockaddr>, _ len: socklen_t) -> Int32 {
    Darwin.bind(fd, addr, len)
}
private func posixCloseFD(_ fd: Int32) { _ = Darwin.close(fd) }
private func writeNoSigpipe(_ fd: Int32, _ buf: UnsafeRawPointer, _ count: Int) -> Int {
    write(fd, buf, count)  // SO_NOSIGPIPE is set on the fd at init
}
#elseif canImport(Musl)
private let sockStreamType: Int32 = Int32(SOCK_STREAM)
private func posixBind(_ fd: Int32, _ addr: UnsafePointer<sockaddr>, _ len: socklen_t) -> Int32 {
    Musl.bind(fd, addr, len)
}
private func posixCloseFD(_ fd: Int32) { _ = Musl.close(fd) }
private func writeNoSigpipe(_ fd: Int32, _ buf: UnsafeRawPointer, _ count: Int) -> Int {
    // Linux has no SO_NOSIGPIPE; MSG_NOSIGNAL per send is the localized equivalent.
    send(fd, buf, count, Int32(MSG_NOSIGNAL))
}
#else
private let sockStreamType = Int32(SOCK_STREAM.rawValue)
private func posixBind(_ fd: Int32, _ addr: UnsafePointer<sockaddr>, _ len: socklen_t) -> Int32 {
    Glibc.bind(fd, addr, len)
}
private func posixCloseFD(_ fd: Int32) { _ = Glibc.close(fd) }
private func writeNoSigpipe(_ fd: Int32, _ buf: UnsafeRawPointer, _ count: Int) -> Int {
    // Linux has no SO_NOSIGPIPE; MSG_NOSIGNAL per send is the localized equivalent.
    send(fd, buf, count, Int32(MSG_NOSIGNAL))
}
#endif

/// Localhost HTTP MCP server hosted inside the app (ADAPTERS §1.2). Claude Code
/// is pointed at it via `--mcp-config` + `--permission-prompt-tool`; it exposes a
/// single tool `approve` whose call SUSPENDS until a human resolves the approval.
///
/// One server can serve all sessions: requests are authenticated by a per-session
/// bearer token, which keys the registered handler.
public actor MCPApprovalServer {
    public struct ApprovalCall: Sendable {
        public let toolName: String
        public let input: JSONValue
        public let toolUseID: String?
    }

    public enum Reply: Sendable {
        case allow(updatedInput: JSONValue)
        case deny(message: String)

        /// The confirmed wire shape: a JSON-stringified `{behavior,…}` object
        /// placed inside an MCP text content block — `behavior`/`message`/
        /// `updatedInput`, NOT `decision`/`reason` (ADAPTERS §1.2 ✅).
        public func wireJSON() -> String {
            switch self {
            case .allow(let updatedInput):
                return JSONValue.object([
                    "behavior": .string("allow"), "updatedInput": updatedInput,
                ]).canonicalString()
            case .deny(let message):
                return JSONValue.object([
                    "behavior": .string("deny"), "message": .string(message),
                ]).canonicalString()
            }
        }
    }

    /// A `check_conflict` tool call: the agent's task and the files/areas it expects to
    /// touch. The handler suspends (like `approve`) until Nucleic decides + the user
    /// answers, then returns a `ConflictReply` the agent reads to know how to proceed.
    public struct ConflictCall: Sendable {
        public let task: String
        public let files: [String]
    }

    public struct ConflictReply: Sendable {
        public let resolution: ConflictResolution

        public init(resolution: ConflictResolution) { self.resolution = resolution }

        /// Text-block JSON the agent reads. `conflict` is true unless there was no
        /// overlap; `action` + `message` tell it exactly what to do.
        public func wireJSON() -> String {
            switch resolution {
            case .noConflict:
                return JSONValue.object(["conflict": .bool(false)]).canonicalString()
            case .deferred:
                return conflictJSON(
                    "deferred",
                    "This task was added to the Nucleic to-do list because it conflicts with "
                        + "another active agent. Stop now and do NOT work on it.")
            case .cancelled:
                return conflictJSON(
                    "cancelled",
                    "The user cancelled this task because it conflicts with another active "
                        + "agent. Stop now and do NOT work on it.")
            case .proceed:
                return conflictJSON(
                    "proceed",
                    "This task conflicts with another active agent, but the user chose to "
                        + "proceed. You may continue.")
            }
        }

        private func conflictJSON(_ action: String, _ message: String) -> String {
            JSONValue.object([
                "conflict": .bool(true),
                "action": .string(action),
                "message": .string(message),
            ]).canonicalString()
        }
    }

    /// A `host_exec` tool call: the agent asks to run a shell command on the *host* machine,
    /// outside the sandbox container (e.g. to build/run a macOS binary the Linux container
    /// can't). The handler suspends (like `approve`) until the user approves and the host
    /// command finishes, then returns its output. Always gated; never auto-approved.
    public struct HostExecCall: Sendable {
        public let command: String
        /// The agent's stated justification for why this cannot run in the container. Required by
        /// the tool schema and enforced (non-empty, non-boilerplate) at the choke point; surfaced
        /// on the approval card as an *unverified claim* beside the command-derived summary.
        public let reason: String
        /// Optional shell override: `"zsh"` forces `/bin/zsh -lc` for this one call even when the
        /// host shell is nash — the per-call escape hatch for a zsh-ism nash can't parse
        /// (docs/NASH.md §7.7.6), mirroring the macOS-VM wire op's `shell` field.
        public let shell: String?
        public init(command: String, reason: String = "", shell: String? = nil) {
            self.command = command
            self.reason = reason
            self.shell = shell
        }
    }

    public enum HostExecReply: Sendable {
        case denied(message: String)
        case ran(exitCode: Int32, stdout: String, stderr: String)

        /// Text-block JSON the agent reads back from the tool result.
        public func wireJSON() -> String {
            switch self {
            case .denied(let message):
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            case .ran(let exitCode, let stdout, let stderr):
                return JSONValue.object([
                    "exit_code": .number(Double(exitCode)),
                    "stdout": .string(stdout),
                    "stderr": .string(stderr),
                ]).canonicalString()
            }
        }
    }

    /// A `mac_vm_exec` tool call: the agent asks to run a shell command inside its session's isolated
    /// **macOS guest VM** (Apple Virtualization framework) — for Mac-only work like `xcodebuild`, an
    /// `xcrun simctl` simulator run, or `codesign`. Unlike `host_exec` (which escapes onto the one
    /// shared host), each session gets its own VM with its own Mac toolchain, so parallel Mac builds
    /// don't thrash the one shared host. The handler suspends (like `approve`) until the user
    /// approves and the in-VM command finishes, then returns its output. Always gated; never
    /// auto-approved.
    public struct MacVMExecCall: Sendable {
        public let command: String
        /// The agent's stated justification for why this needs the macOS VM (rather than the Linux
        /// container). Required by the schema and enforced at the choke point; surfaced on the approval
        /// card as an *unverified claim*.
        public let reason: String
        public init(command: String, reason: String = "") {
            self.command = command
            self.reason = reason
        }
    }

    public enum MacVMExecReply: Sendable {
        case denied(message: String)
        case ran(exitCode: Int32, stdout: String, stderr: String)

        /// Text-block JSON the agent reads back from the tool result (same shape as `host_exec`).
        public func wireJSON() -> String {
            switch self {
            case .denied(let message):
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            case .ran(let exitCode, let stdout, let stderr):
                return JSONValue.object([
                    "exit_code": .number(Double(exitCode)),
                    "stdout": .string(stdout),
                    "stderr": .string(stderr),
                ]).canonicalString()
            }
        }
    }

    /// A `mac_vm_computer` tool call: one **computer-use action** against the session's macOS VM —
    /// the agent driving the guest's GUI like its own Mac (Anthropic computer-use shape). The handler
    /// performs the action in the guest over SSH and returns a fresh screenshot so the agent can see
    /// the result. Runs inside the isolated, disposable VM, so — unlike `host_exec`/`mac_vm_exec` — it
    /// is NOT per-action approval-gated (a per-click prompt would make a navigation loop unusable);
    /// exposure is the opt-in boundary.
    public struct MacVMComputerCall: Sendable {
        /// The action verb: `screenshot`, `left_click`, `right_click`, `double_click`, `mouse_move`,
        /// `left_click_drag`, `type`, `key`, `scroll`, `cursor_position`, `launch_app`, `open_browser`,
        /// `wait`; plus the AX-semantic actions when the guest runs the native agent
        /// (docs/MACOS_VM_NATIVE_AGENT.md): `ax_dump`, `ax_element_at`, `ax_press`, `ax_set_value`,
        /// `ax_focus`.
        public let action: String
        /// Target point (screen pixels; the screenshot is 1:1 so the agent targets what it sees).
        public let x: Int?
        public let y: Int?
        /// Payload for `type` (literal text), `key` (a chord like `cmd+s`, `return`, `tab`),
        /// `launch_app` (an application name), `open_browser` (a short browser name like `chrome` or
        /// `firefox`; the page to start on goes in `url`), or `ax_press` (an optional AX action name to
        /// perform instead of the default `AXPress`, e.g. `AXShowMenu`).
        public let text: String?
        /// For the `ax_*` element actions: the element handle from the last `ax_dump`/`ax_element_at`.
        public let ref: String?
        /// For `ax_set_value`: the value to write into the element.
        public let value: String?
        /// For `open_browser`: the page to start on (e.g. `https://example.com`, `localhost:3000`).
        /// Omitted opens the browser's own start page.
        public let url: String?
        /// For `scroll`: `up`/`down`/`left`/`right` and a line amount (delivered as paced wheel notches;
        /// `x`/`y`, when given, aim the gesture — a wheel routes to whatever the pointer is over).
        public let scrollDirection: String?
        public let scrollAmount: Int?
        /// For `wait`: milliseconds to pause before the screenshot (e.g. let a window open).
        public let durationMs: Int?

        public init(
            action: String, x: Int? = nil, y: Int? = nil, text: String? = nil,
            ref: String? = nil, value: String? = nil, url: String? = nil,
            scrollDirection: String? = nil, scrollAmount: Int? = nil, durationMs: Int? = nil
        ) {
            self.action = action
            self.x = x
            self.y = y
            self.text = text
            self.ref = ref
            self.value = value
            self.url = url
            self.scrollDirection = scrollDirection
            self.scrollAmount = scrollAmount
            self.durationMs = durationMs
        }
    }

    public enum MacVMComputerReply: Sendable {
        case denied(message: String)
        /// The action ran; `text` is a short summary (e.g. cursor position / screen size), and
        /// `imageBase64` is the post-action screenshot as base64 JPEG (nil only if capture failed).
        case ok(text: String, imageBase64: String?)

        /// Denial JSON for the text-only failure path.
        public func deniedJSON() -> String {
            if case .denied(let message) = self {
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            }
            return JSONValue.object([:]).canonicalString()
        }
    }

    /// A `mac_vm_computer_batch` / `linux_vm_computer_batch` tool call: an **ordered sequence** of
    /// computer-use actions run against the session's VM in ONE turn. The agent that can already see
    /// every target (e.g. every window's close button in the last screenshot) plans the whole click
    /// sequence up front and fires it here, instead of burning one turn per click on the single-action
    /// `*_computer` tool. The handler boots/reuses the VM once and performs each ``MacVMComputerCall``
    /// step in order over the same GUI channel; like the single-action tool it runs inside the isolated,
    /// disposable VM and is NOT per-action approval-gated. Each step carries the exact same fields as a
    /// single `*_computer` action (including the semantic `ax_*` verbs), so anything valid there is
    /// valid as a step.
    public struct MacVMComputerBatchCall: Sendable {
        /// The actions to run, in order. An empty list is a no-op (the VM still boots/reuses, but no
        /// action runs and the log comes back empty).
        public let steps: [MacVMComputerCall]
        /// Stop the run at the first step that fails, rather than pressing on (default true). When a
        /// step fails and this is true, the remaining steps are reported as skipped.
        public let stopOnError: Bool
        /// Return a screenshot after EVERY step (default false — only a single final screenshot is
        /// attached, to keep the reply small). Turn on when the agent needs to see intermediate state.
        public let screenshotEachStep: Bool
        /// Milliseconds to pause after each step before the next, to let the UI settle (e.g. a window
        /// finishing its close animation). Applied between steps and after the last one; nil/0 = none.
        public let stepDelayMs: Int?

        public init(
            steps: [MacVMComputerCall], stopOnError: Bool = true,
            screenshotEachStep: Bool = false, stepDelayMs: Int? = nil
        ) {
            self.steps = steps
            self.stopOnError = stopOnError
            self.screenshotEachStep = screenshotEachStep
            self.stepDelayMs = stepDelayMs
        }
    }

    public enum MacVMComputerBatchReply: Sendable {
        case denied(message: String)
        /// The batch ran; `steps` is one outcome per executed step, in order. `stoppedAtError` is true
        /// when `stop_on_error` cut the run short — steps after the failing one were not attempted.
        case ran(steps: [StepOutcome], stoppedAtError: Bool)

        /// Outcome of a single executed step: its 0-based `index`, the `action` verb, a short `summary`,
        /// an optional post-step screenshot, and whether it `failed`.
        public struct StepOutcome: Sendable {
            public let index: Int
            public let action: String
            public let summary: String
            public let imageBase64: String?
            public let failed: Bool
            public init(
                index: Int, action: String, summary: String,
                imageBase64: String?, failed: Bool
            ) {
                self.index = index
                self.action = action
                self.summary = summary
                self.imageBase64 = imageBase64
                self.failed = failed
            }
        }

        /// Denial JSON for the text-only failure path (mirrors ``MacVMComputerReply/deniedJSON()``).
        public func deniedJSON() -> String {
            if case .denied(let message) = self {
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            }
            return JSONValue.object([:]).canonicalString()
        }
    }

    /// A `mac_vm_clear_notifications` tool call: while driving the session's macOS VM via
    /// `mac_vm_computer`, a banner/alert has slid in and is obstructing part of the screen. This
    /// dismisses the on-screen notifications by killing the guest's NotificationCenter (which relaunches
    /// clean). No parameters — it always runs the same fixed `killall NotificationCenter`. Like the
    /// computer-use / lifecycle tools it is NOT approval-gated: the blast radius is the agent's own
    /// disposable VM. Replies with the same `{exit_code, stdout, stderr}` shape as `mac_vm_exec`.
    public struct MacVMClearNotificationsCall: Sendable {
        public init() {}
    }

    /// A `mac_vm_request_operator` tool call: the agent — while driving the session's macOS VM via
    /// `mac_vm_computer` — hits a step only a human can clear (a CAPTCHA, a login / 2FA prompt, an
    /// interactive OS dialog) and asks the *user* to take the wheel. Modeled on `host_exec`: the agent
    /// must justify the request (`reason`) and say exactly what it needs done (`instructions`). The
    /// handler suspends (like `host_exec`) until the user finishes operating the VM or declines, then
    /// returns the outcome. Always gated; never auto-approved.
    public struct MacVMOperatorCall: Sendable {
        /// The agent's stated justification for why a human is required (rather than the agent driving
        /// the VM itself). Required by the schema and enforced at the choke point; surfaced on the
        /// request card as an *unverified claim*.
        public let reason: String
        /// What, concretely, the agent needs the user to do in the VM (e.g. "solve the CAPTCHA in the
        /// Safari window, then click Continue"). Required; shown to the user in the walkthrough popup.
        public let instructions: String
        public init(reason: String = "", instructions: String = "") {
            self.reason = reason
            self.instructions = instructions
        }
    }

    public enum MacVMOperatorReply: Sendable {
        case denied(message: String)
        /// The user finished (or gave up on) the hand-off. `completed` is whether they say they did
        /// what was asked; `note` is an optional free-text message back to the agent.
        case resolved(completed: Bool, note: String)

        /// Text-block JSON the agent reads back from the tool result.
        public func wireJSON() -> String {
            switch self {
            case .denied(let message):
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            case .resolved(let completed, let note):
                return JSONValue.object([
                    "completed": .bool(completed), "note": .string(note),
                ]).canonicalString()
            }
        }
    }

    /// A `linux_container` tool call: the agent manages its OWN throwaway Linux container — the
    /// lightest rung of isolation below a VM. One tool, multiple `op`s: `create` brings up a fresh
    /// container from an image (defaulting to the agent's own sandbox image) and returns its `name`;
    /// `exec` runs a shell `command` inside a named container; `stop`/`remove` tear it down; `list`
    /// enumerates the containers this session created. All operations are authorized by session
    /// ownership: the handler only resolves names from the calling session's private container map,
    /// so none of these operations can target a host resource or another session's container.
    public struct ContainerToolCall: Sendable {
        /// One of: `create`, `exec`, `stop`, `remove`, `list`.
        public let op: String
        /// The container to act on (required for `exec`/`stop`/`remove`; ignored for `list`). For
        /// `create` it is an optional caller-chosen label (namespaced per session); nil → auto-named.
        public let name: String?
        /// For `create`: the OCI image reference to run; nil → the agent's own sandbox image.
        public let image: String?
        /// For `exec`: the shell command to run inside the container.
        public let command: String?
        /// For `exec`: the working directory inside the container; nil → the image default.
        public let workdir: String?
        /// Legacy caller-supplied context. Kept wire-compatible with older clients; agent-owned
        /// container operations do not require a justification or a separate approval.
        public let reason: String
        public init(
            op: String, name: String? = nil, image: String? = nil, command: String? = nil,
            workdir: String? = nil, reason: String = ""
        ) {
            self.op = op
            self.name = name
            self.image = image
            self.command = command
            self.workdir = workdir
            self.reason = reason
        }
    }

    /// A Nucleic-managed Orchestra subagent request. Unlike Claude's native `Task` tool, this
    /// creates an ordinary Nucleic session so the app owns the worker lifecycle, transcript,
    /// sandbox/control routing, and model selection.
    public struct OrchestraSubagentCall: Sendable {
        public let task: String
        public let prompt: String
        /// Optional batch this worker joins, so the supervisor can wait on the group with
        /// `nucleic_supervise`'s `batch` argument.
        public let batch: String?
        public init(task: String, prompt: String, batch: String? = nil) {
            self.task = task
            self.prompt = prompt
            self.batch = batch
        }
    }

    /// A supervisor's `nucleic_delegate_plan` call: the plan prose plus every delegated task, grouped
    /// into batches. Parsed from the tool's input schema, which is what makes the orchestrator's plan a
    /// *structured* output on every provider (no provider-specific JSON-schema support needed).
    public struct OrchestraPlanCall: Sendable {
        public let plan: String
        public let batches: [OrchestraPlanBatch]
        public init(plan: String, batches: [OrchestraPlanBatch]) {
            self.plan = plan
            self.batches = batches
        }
    }

    /// A `nucleic_ask_user` call: one or more multiple-choice questions for the human. Mirrors Claude
    /// Code's `AskUserQuestion` payload so the same UI renders it, and the raw `input` is carried
    /// through verbatim for that UI (and for folding the answers back in).
    public struct AskUserCall: Sendable {
        public let questions: [AskUserQuestion.Question]
        /// The original tool input, passed to the approval card unchanged.
        public let input: JSONValue
        public init(questions: [AskUserQuestion.Question], input: JSONValue) {
            self.questions = questions
            self.input = input
        }
    }

    /// A supervisor's `nucleic_reply_to_worker` call: the answer to a worker waiting in
    /// `nucleic_ask_supervisor`, addressed by the `worker_id` from the spawn ack.
    public struct OrchestraReplyToWorkerCall: Sendable {
        public let workerID: String
        public let reply: String
        public init(workerID: String, reply: String) {
            self.workerID = workerID
            self.reply = reply
        }
    }

    /// A worker's `nucleic_ask_supervisor` call. The worker's own identity is known from the run's
    /// token, so the tool carries only the question text.
    public struct OrchestraAskSupervisorCall: Sendable {
        public let question: String
        public init(question: String) {
            self.question = question
        }
    }

    public enum ContainerToolReply: Sendable {
        case denied(message: String)
        /// A `create`/`stop`/`remove`/`list` op succeeded. `name` echoes the affected container (empty
        /// for `list`); `names` lists the session's containers (for `list`); `message` is a short note.
        case ok(name: String, names: [String], message: String)
        /// An `exec` op ran to completion inside the container.
        case ran(exitCode: Int32, stdout: String, stderr: String)

        /// Text-block JSON the agent reads back from the tool result.
        public func wireJSON() -> String {
            switch self {
            case .denied(let message):
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            case .ok(let name, let names, let message):
                return JSONValue.object([
                    "ok": .bool(true),
                    "name": .string(name),
                    "names": .array(names.map { .string($0) }),
                    "message": .string(message),
                ]).canonicalString()
            case .ran(let exitCode, let stdout, let stderr):
                return JSONValue.object([
                    "exit_code": .number(Double(exitCode)),
                    "stdout": .string(stdout),
                    "stderr": .string(stderr),
                ]).canonicalString()
            }
        }
    }

    /// A `mac_vm_control` / `linux_vm_control` tool call: the agent manages the lifecycle of the
    /// isolated VM it booted for this session — `status` (a read-only query), `stop` (power off, keep
    /// the disk so the next command reboots), `suspend`/`resume` (freeze in RAM / thaw), `restart`
    /// (stop → reboot, reclaiming RAM), or `kill` (stop and delete the disk clone). One tool, one `op`.
    /// Not approval-gated: the blast radius is the agent's own disposable VM (exposure is the opt-in
    /// boundary, like the computer-use tools), and these ops only change that VM's power state.
    public struct VMControlCall: Sendable {
        /// One of: `status`, `stop`, `suspend`, `resume`, `restart`, `kill`.
        public let op: String
        public init(op: String) { self.op = op }
    }

    public enum VMControlReply: Sendable {
        case denied(message: String)
        /// The op completed. `state` is the VM's resulting lifecycle state — one of `running`,
        /// `suspended`, `stopped`, `removed`, or `absent`; `message` is a short human note.
        case ok(state: String, message: String)

        /// Text-block JSON the agent reads back from the tool result.
        public func wireJSON() -> String {
            switch self {
            case .denied(let message):
                return JSONValue.object([
                    "denied": .bool(true), "message": .string(message),
                ]).canonicalString()
            case .ok(let state, let message):
                return JSONValue.object([
                    "ok": .bool(true),
                    "state": .string(state),
                    "message": .string(message),
                ]).canonicalString()
            }
        }
    }

    /// One codex `PreToolUse` hook call — the Codex edit gate (docs/LOCKING.md §4.6,
    /// ``CodexToolGate``). Codex runs the hook **synchronously before the tool executes**, so unlike
    /// the report routes below this one is a *gate*: the handler suspends for as long as the
    /// arbitration takes (a contended file parks it in the lock queue) and its verdict decides
    /// whether the tool runs at all.
    public struct PreToolUseCall: Sendable {
        /// codex's normalized tool name — `apply_patch` for a patch, `Bash` for a shell command.
        public let toolName: String
        /// The tool's arguments as codex passes them to the hook (`tool_input`).
        public let input: JSONValue
        /// The tool call's id (`tool_use_id`) — the same attribution key an `Edit` carries into
        /// `arbitrate`, so lock notes nest under the right transcript row.
        public let toolUseID: String?
        /// The turn's working directory, used to resolve a patch's relative paths.
        public let cwd: String?
        public init(toolName: String, input: JSONValue, toolUseID: String?, cwd: String?) {
            self.toolName = toolName
            self.input = input
            self.toolUseID = toolUseID
            self.cwd = cwd
        }
    }

    /// The gate's verdict for a ``PreToolUseCall``. `deny` carries the message the model sees
    /// ("Command blocked by PreToolUse hook: …"); codex honors no other decision, which is why
    /// there is no `ask` case.
    public enum PreToolUseDecision: Sendable, Equatable {
        case allow
        case deny(reason: String)

        /// The JSON body the hook program reads (``CodexToolGate/hookScriptSource``).
        public func wireJSON() -> String {
            switch self {
            case .allow:
                return JSONValue.object(["decision": .string("allow")]).canonicalString()
            case .deny(let reason):
                return JSONValue.object([
                    "decision": .string("deny"), "reason": .string(reason),
                ]).canonicalString()
            }
        }
    }

    /// A ground-truth report from the in-container `git` interceptor shim: the real git the
    /// agent ran (argv + cwd + exit code) in a Nucleic Control container. Delivered over a
    /// dedicated `POST /git-event` route (not an MCP tool), so the shim needs no MCP handshake.
    public struct GitReportCall: Sendable {
        public let argv: [String]
        public let cwd: String
        public let exitCode: Int
        public let subcommand: String?
        public let sessionID: String?
        public init(argv: [String], cwd: String, exitCode: Int, subcommand: String?, sessionID: String?) {
            self.argv = argv
            self.cwd = cwd
            self.exitCode = exitCode
            self.subcommand = subcommand
            self.sessionID = sessionID
        }
    }

    /// A ground-truth report from the in-container `gh` interceptor shim — the agentic GitHub
    /// action the agent ran (argv + cwd + exit code). Same shape and `POST /gh-event` transport as
    /// `GitReportCall`; the host classifies it via `GhCommandSummary`.
    public struct GhReportCall: Sendable {
        public let argv: [String]
        public let cwd: String
        public let exitCode: Int
        public let subcommand: String?
        public let sessionID: String?
        public init(argv: [String], cwd: String, exitCode: Int, subcommand: String?, sessionID: String?) {
            self.argv = argv
            self.cwd = cwd
            self.exitCode = exitCode
            self.subcommand = subcommand
            self.sessionID = sessionID
        }
    }

    /// A ground-truth report of one ordinary (non-git) command the in-container interceptor
    /// observed — from the output-capturing shim (`source == .shim`, carrying argv + a capped
    /// stdout/stderr head) or the bash command tracer (`source == .tracer`, a raw command line +
    /// metadata only). Delivered over `POST /command-event`. One HTTP request may carry many
    /// (the tracer batches a whole Bash tool call), so the route fans out to one call each.
    public struct CommandReportCall: Sendable {
        /// How the command was observed: the output-capturing shim, the bash `BASH_ENV` tracer,
        /// or (nash) the agent shell's own exec gate (`POST /shell-event`).
        public enum Source: String, Sendable { case shim, tracer, nash }
        /// The resolved command name (shim), or nil when only a raw command line is available
        /// (tracer) — the consumer derives it by tokenizing `commandLine`.
        public let command: String?
        /// Full argv (shim), or empty when only `commandLine` is available (tracer).
        public let argv: [String]
        /// The raw command line (tracer), or nil (shim).
        public let commandLine: String?
        public let cwd: String
        public let exitCode: Int
        /// Wall-clock duration in milliseconds, when the reporter measured it.
        public let durationMs: Int?
        /// Capped head of stdout / stderr (shim only; empty for the tracer).
        public let stdout: String
        public let stderr: String
        /// True when stdout/stderr exceeded the capture cap (shim).
        public let truncated: Bool
        public let source: Source
        public let sessionID: String?

        public init(
            command: String?, argv: [String], commandLine: String?, cwd: String, exitCode: Int,
            durationMs: Int?, stdout: String, stderr: String, truncated: Bool, source: Source,
            sessionID: String?
        ) {
            self.command = command
            self.argv = argv
            self.commandLine = commandLine
            self.cwd = cwd
            self.exitCode = exitCode
            self.durationMs = durationMs
            self.stdout = stdout
            self.stderr = stderr
            self.truncated = truncated
            self.source = source
            self.sessionID = sessionID
        }
    }

    /// One event decoded from a nash `shell-batch` (`POST /shell-event`; docs/NASH.md §6.1). nash
    /// is the agent's own shell, so its exec gate reports every simple command it runs — structured
    /// argv, cwd, exit code, duration — plus derived `cd` / names-only `export` events, compat
    /// `fallback` warnings, and a `dropped` back-pressure counter. Delivered as a batch; the route
    /// fans it out to one call each. Exec events converge onto the same consumer path as the git/gh
    /// shims and the (retired) bash tracer.
    public struct ShellReportCall: Sendable, Equatable {
        public enum Kind: Sendable, Equatable {
            /// A simple command that ran (external, builtin, or function). `pipeline` is present
            /// only when it ran as a stage of `a | b | c`, and shares its id with that pipeline's
            /// `pipe` events so the two can be joined (docs/NASH.md §5.1).
            case exec(
                argv: [String], cwd: String, exitCode: Int, durationMs: Int?,
                pipeline: NashPipelineSlot?)
            /// A command nash has been running for a while that has **not returned** (docs/NASH.md
            /// §5.1). Purely an "it is still in flight" announcement: there is no exit code or
            /// duration yet, and the matching ``exec`` arrives when the command finally returns,
            /// carrying the same ``ShellReportCall/execToken``. Only commands that outlive nash's
            /// announce threshold are reported this way — a fast command is reported once, on exit.
            ///
            /// `elapsedMs` is how long it had already been running when nash announced it, which is
            /// what lets a row show elapsed time from when the command *started* rather than from
            /// when the host heard about it.
            case execStart(argv: [String], cwd: String, elapsedMs: Int, pipeline: NashPipelineSlot?)
            /// A successful `cd`.
            case cd(from: String, to: String)
            /// A successful environment mutation — **names only**, never values (docs/NASH.md §5.4).
            case export(op: String, names: [String])
            /// Data crossing a redirection operator (docs/NASH.md §5.2): `op` (`>`,`>>`,`<`,`2>`,
            /// `&>`,`<<`,`<<<`), target path (nil for heredoc/here-string), total `bytes`, and a
            /// bounded, redacted base64 `previewB64` (empty for special files like `/dev/null`).
            case redirect(
                op: String, fd: Int, target: String?, bytes: Int, truncated: Bool,
                hash: String, previewB64: String)
            /// Command-substitution `$(…)` output (docs/NASH.md §5.3): total `bytes` + bounded,
            /// redacted base64 preview.
            case cmdsub(bytes: Int, truncated: Bool, hash: String, previewB64: String)
            /// Data crossing one pipe link `a | b` (docs/NASH.md §5.3): the pipeline id, the
            /// producing/consuming stage indices and their source text, total `bytes`, and a
            /// bounded, redacted preview. `fromText`/`toText` are rendered from the AST at spawn,
            /// so they name a compound stage correctly and arrive with the link itself; they are
            /// empty only for a batch from a nash that predates them.
            case pipe(
                pipelineID: Int, fromIndex: Int, toIndex: Int, fromText: String, toText: String,
                bytes: Int, truncated: Bool, hash: String, previewB64: String)
            /// nash re-execed the real bash because it could not handle `input` (docs/NASH.md §4.1).
            case fallback(reason: String, input: String)
            /// An operator lever nash was asked to act on (docs/NASH.md §4.3). `action` is
            /// `ignored` when the lever arrived through the environment — which the observed
            /// process writes, so nash refuses it — and `honored` when the trusted policy file set
            /// it. The `ignored` rows are the interesting ones: they are an agent's attempt to turn
            /// off the layer watching it, and they exist so that attempt shows up as a row instead
            /// of as a shell that quietly stopped reporting.
            case policy(lever: String, value: String, action: String, note: String)
            /// nash dropped `count` events under back-pressure (best-effort; docs/NASH.md §5.4).
            case dropped(count: Int)
        }
        public let kind: Kind
        public let sessionID: String?
        /// Emitter attribution from the batch envelope. Nil is backward-compatible and means the
        /// session's own agent unless the ingestion site (for example a VM spool drain) overrides it.
        public let environment: NashEnvironment?
        /// The emitting nash process (`shellId` on the batch envelope). Scopes that process's
        /// pipeline counter, which restarts at 1 in every shell — see ``NashPipelineKey``.
        public let shellID: String?
        /// nash's gate token for this command, carried by the ``Kind/execStart`` that announced it
        /// as still running and by the ``Kind/exec`` that later reports how it ended. Together with
        /// ``shellID`` it identifies one command run, which is what lets the in-flight row become
        /// the finished row instead of the two appearing as separate commands. Nil for every other
        /// kind, and for a batch from a nash that predates the announcement.
        public let execToken: Int?
        public init(
            kind: Kind, sessionID: String?, environment: NashEnvironment? = nil,
            shellID: String? = nil, execToken: Int? = nil
        ) {
            self.kind = kind
            self.sessionID = sessionID
            self.environment = environment
            self.shellID = shellID
            self.execToken = execToken
        }

        public func attributed(to environment: NashEnvironment) -> ShellReportCall {
            ShellReportCall(
                kind: kind, sessionID: sessionID, environment: environment, shellID: shellID,
                execToken: execToken)
        }
    }

    public typealias Handler = @Sendable (ApprovalCall) async -> Reply
    public typealias ConflictHandler = @Sendable (ConflictCall) async -> ConflictReply
    public typealias HostExecHandler = @Sendable (HostExecCall) async -> HostExecReply
    public typealias MacVMExecHandler = @Sendable (MacVMExecCall) async -> MacVMExecReply
    public typealias MacVMComputerHandler = @Sendable (MacVMComputerCall) async -> MacVMComputerReply
    /// Runs an ordered batch of computer-use actions against the VM in one call (see
    /// ``MacVMComputerBatchCall``). Shared by the macOS and Linux batch tools.
    public typealias MacVMComputerBatchHandler = @Sendable (MacVMComputerBatchCall) async -> MacVMComputerBatchReply
    /// Clears the guest's on-screen notifications. Reuses ``MacVMExecReply`` — the fixed `killall
    /// NotificationCenter` returns the same `{exit_code, stdout, stderr}` shape as `mac_vm_exec`.
    public typealias MacVMClearNotificationsHandler = @Sendable (MacVMClearNotificationsCall) async -> MacVMExecReply
    public typealias MacVMOperatorHandler = @Sendable (MacVMOperatorCall) async -> MacVMOperatorReply
    public typealias ContainerToolHandler = @Sendable (ContainerToolCall) async -> ContainerToolReply
    public typealias OrchestraSubagentHandler = @Sendable (OrchestraSubagentCall) async -> OrchestraSubagentResult
    /// Supervisor-side: `nucleic_delegate_plan` — spawns a worker for every task in the plan.
    public typealias OrchestraPlanServerHandler = @Sendable (OrchestraPlanCall) async -> OrchestraPlanResult
    /// `nucleic_ask_user` — surfaces the questions to the human and blocks for their answer.
    public typealias AskUserServerHandler = @Sendable (AskUserCall) async -> AskUserResult
    /// Supervisor-side: `nucleic_supervise` — blocks until worker events arrive (or the named batch
    /// finishes), then drains them. The argument is the optional batch to wait on.
    public typealias OrchestraSuperviseServerHandler = @Sendable (String?) async -> OrchestraSuperviseResult
    /// Supervisor-side: `nucleic_reply_to_worker` — answers a worker's pending question.
    public typealias OrchestraReplyToWorkerServerHandler = @Sendable (OrchestraReplyToWorkerCall) async -> OrchestraReplyResult
    /// Worker-side: `nucleic_ask_supervisor` — routes a question to the supervisor and blocks for the reply.
    public typealias OrchestraAskSupervisorServerHandler = @Sendable (OrchestraAskSupervisorCall) async -> OrchestraAskSupervisorResult
    /// Mesh messaging: `nucleic_send_message` — publish an app-level message onto the mesh
    /// (host-mediated; provenance is stamped host-side). Returns the tool's result text.
    public typealias MeshSendMessageHandler = @Sendable (MeshSendRequest) async -> String
    /// Mesh messaging: `nucleic_subscribe_topic` — subscribe this session to a broadcast topic.
    public typealias MeshSubscribeTopicHandler = @Sendable (String) async -> String
    /// Mesh messaging: `nucleic_wait_for_message` — block until a message arrives for this
    /// session (optionally on a topic) or the timeout lapses. `(topic, timeoutSeconds)`.
    public typealias MeshWaitForMessageHandler = @Sendable (String?, Int?) async -> String
    /// Parsed `nucleic_monitor` arguments — arm a long-lived watch whose output re-invokes the
    /// session as fresh turns (the working replacement for the blocked provider-native `Monitor`).
    public struct MonitorCall: Sendable {
        /// Shell command/script; each stdout line is an event. Exit ends the watch. Empty when the
        /// `ws` source is used instead — exactly one of `command` / `ws` must be set.
        public let command: String
        /// WebSocket URL to open (mutually exclusive with `command`): each inbound text frame is an
        /// event, a binary frame is reported as a placeholder line, and socket close ends the watch.
        public let wsURL: String?
        /// Optional WebSocket subprotocols to offer on the handshake.
        public let wsProtocols: [String]
        /// Short human label shown in each delivered event and the transcript note.
        public let description: String
        /// Kill the watch after this deadline (ignored when `persistent`). Clamped host-side.
        public let timeoutMs: Int?
        /// Run for the session's lifetime (no timeout).
        public let persistent: Bool
        public init(
            command: String, wsURL: String? = nil, wsProtocols: [String] = [],
            description: String, timeoutMs: Int?, persistent: Bool
        ) {
            self.command = command
            self.wsURL = wsURL
            self.wsProtocols = wsProtocols
            self.description = description
            self.timeoutMs = timeoutMs
            self.persistent = persistent
        }
    }
    /// `nucleic_monitor` — arm a background watch. Returns the agent-facing result text
    /// (armed confirmation, or a refusal). Events are delivered out-of-band via the host sink.
    public typealias MonitorHandler = @Sendable (MonitorCall) async -> String
    public typealias VMControlHandler = @Sendable (VMControlCall) async -> VMControlReply
    /// Synchronous + non-gated: it just records the observed op and returns. No user approval,
    /// so a git report never blocks the agent's git call.
    public typealias GitReportHandler = @Sendable (GitReportCall) -> Void
    /// As `GitReportHandler`, for the `gh` interceptor's `POST /gh-event`.
    public typealias GhReportHandler = @Sendable (GhReportCall) -> Void
    /// As ``GitReportHandler``, for the non-git command interceptor (`POST /command-event`).
    public typealias CommandReportHandler = @Sendable (CommandReportCall) -> Void
    /// The Codex edit gate (`POST /pre-tool-use`). **Blocking, unlike the report handlers**: it
    /// arbitrates the call's file locks and the reply decides whether codex runs the tool, so it
    /// suspends exactly as long as an `Edit`'s `arbitrate` would.
    public typealias PreToolUseHandler = @Sendable (PreToolUseCall) async -> PreToolUseDecision
    /// As ``GitReportHandler``, for the nash agent shell's exec gate (`POST /shell-event`).
    ///
    /// Takes the **whole batch** rather than one call at a time (docs/NASH_STREAM_PERF_PLAN.md §P1):
    /// nash posts up to 200 events per body, and a build storm posts thousands per minute across
    /// every concurrent agent. Delivering per event made each one its own hop to the main actor —
    /// and, because those hops were unstructured `Task`s, let a batch's rows land out of the order
    /// nash emitted them (an `exec` overtaking its own `exec-start`). One call per batch fixes both.
    public typealias ShellReportHandler = @Sendable ([ShellReportCall]) -> Void

    /// The dedicated POST path the `git` interceptor shim reports to — bearer-token gated like
    /// the MCP route, but a plain JSON POST rather than JSON-RPC.
    public static let gitEventPath = "/git-event"
    /// The dedicated POST path the `gh` interceptor shim reports to (sibling of `gitEventPath`).
    public static let ghEventPath = "/gh-event"
    /// The dedicated POST path the non-git command shims + bash tracer report to.
    public static let commandEventPath = CommandInterceptor.eventPath
    /// The dedicated POST path the nash agent shell reports its exec gate to (docs/NASH.md §6.1).
    public static let shellEventPath = "/shell-event"
    /// The dedicated POST path codex's `PreToolUse` hook consults before each tool call
    /// (docs/LOCKING.md §4.6). A *gate*, not a report: the response decides whether the tool runs.
    public static let preToolUsePath = CodexToolGate.eventPath

    public static let toolName = "approve"
    /// Fully-qualified name passed to `--permission-prompt-tool`.
    public static let qualifiedToolName = "mcp__nucleic__approve"
    public static let conflictToolName = "check_conflict"
    /// Fully-qualified name — pre-allowed so the agent calls it without an approval round-trip.
    public static let qualifiedConflictToolName = "mcp__nucleic__check_conflict"
    public static let hostExecToolName = "host_exec"
    /// Fully-qualified name — pre-allowed via `--allowedTools` so the call reaches our handler
    /// directly (the handler is the sole gate); never routed through Claude's permission path,
    /// so `auto` mode can't auto-approve it. Only advertised when a handler is registered.
    public static let qualifiedHostExecToolName = "mcp__nucleic__host_exec"
    /// Appended to EVERY macOS-VM tool description. Concurrent macOS guests are hard-capped by macOS
    /// itself (``MacVMSettings/maxConcurrentVMs``), so one agent parking an idle VM starves every
    /// other agent's boot — that cap is why this notice rides on the tools that *use* the VM, not
    /// just on `mac_vm_control`. Linux VMs carry no such cap and deliberately omit it.
    public static let macVMFreeWhenIdleNotice =
        "MANDATORY — concurrent macOS VMs are hard-capped across ALL agents, and one you leave "
        + "running blocks every other agent (and your own later work) from booting one. The moment "
        + "you are no longer ACTIVELY using this VM you MUST free it with `mac_vm_control` — even "
        + "mid-turn, between two steps of the same task, before you read files, think, edit code, or "
        + "run container work: `suspend` when you'll need it again (it saves all state to disk and "
        + "frees the slot), `stop` when the in-VM state is disposable, `kill` when you're done with "
        + "it entirely. Never leave an idle macOS VM running."
    public static let macVMExecToolName = "mac_vm_exec"
    /// Fully-qualified name — pre-allowed like `host_exec` so the call reaches our handler directly
    /// (the handler is the sole gate). Only advertised when a handler is registered.
    public static let qualifiedMacVMExecToolName = "mcp__nucleic__mac_vm_exec"
    public static let macVMComputerToolName = "mac_vm_computer"
    /// Fully-qualified name — pre-allowed so the computer-use action reaches our handler directly
    /// (no per-action prompt). Only advertised when a handler is registered.
    public static let qualifiedMacVMComputerToolName = "mcp__nucleic__mac_vm_computer"
    public static let macVMComputerBatchToolName = "mac_vm_computer_batch"
    /// Fully-qualified name — pre-allowed like `mac_vm_computer` so the batch of actions reaches our
    /// handler directly (no per-action prompt). Only advertised when a handler is registered.
    public static let qualifiedMacVMComputerBatchToolName = "mcp__nucleic__mac_vm_computer_batch"
    public static let macVMClearNotificationsToolName = "mac_vm_clear_notifications"
    /// Fully-qualified name — pre-allowed like `mac_vm_computer` so the call reaches our handler directly
    /// (no per-call prompt; it only touches the agent's own disposable VM). Only advertised when a
    /// handler is registered.
    public static let qualifiedMacVMClearNotificationsToolName = "mcp__nucleic__mac_vm_clear_notifications"
    public static let macVMOperatorToolName = "mac_vm_request_operator"
    /// Fully-qualified name — pre-allowed like `host_exec` so the operator-help request reaches our
    /// handler directly (the handler is the sole gate). Only advertised when a handler is registered.
    public static let qualifiedMacVMOperatorToolName = "mcp__nucleic__mac_vm_request_operator"
    public static let linuxVMExecToolName = "linux_vm_exec"
    /// Fully-qualified name — pre-allowed like `mac_vm_exec` so the call reaches our handler directly
    /// (the handler is the sole gate). Only advertised when a handler is registered.
    public static let qualifiedLinuxVMExecToolName = "mcp__nucleic__linux_vm_exec"
    public static let linuxVMComputerToolName = "linux_vm_computer"
    /// Fully-qualified name — pre-allowed so the computer-use action reaches our handler directly
    /// (no per-action prompt). Only advertised when a handler is registered.
    public static let qualifiedLinuxVMComputerToolName = "mcp__nucleic__linux_vm_computer"
    public static let linuxVMComputerBatchToolName = "linux_vm_computer_batch"
    /// Fully-qualified name — the Linux sibling of ``qualifiedMacVMComputerBatchToolName``.
    public static let qualifiedLinuxVMComputerBatchToolName = "mcp__nucleic__linux_vm_computer_batch"
    public static let linuxContainerToolName = "linux_container"
    /// Fully-qualified name — pre-allowed so the container lifecycle op reaches our handler directly
    /// (the handler is the sole gate). Only advertised when a handler is registered.
    public static let qualifiedLinuxContainerToolName = "mcp__nucleic__linux_container"
    public static let macVMControlToolName = "mac_vm_control"
    /// Fully-qualified name — pre-allowed so the macOS-VM lifecycle op reaches our handler directly (no
    /// per-op prompt; the VM is the agent's own disposable one). Only advertised when a handler is
    /// registered.
    public static let qualifiedMacVMControlToolName = "mcp__nucleic__mac_vm_control"
    public static let linuxVMControlToolName = "linux_vm_control"
    /// Fully-qualified name — the Linux-VM sibling of ``qualifiedMacVMControlToolName``.
    public static let qualifiedLinuxVMControlToolName = "mcp__nucleic__linux_vm_control"
    public static let orchestraSubagentToolName = "nucleic_subagent"
    /// Fully-qualified name — pre-allowed in Orchestra so spawning goes through Nucleic's handler.
    public static let qualifiedOrchestraSubagentToolName = "mcp__nucleic__nucleic_subagent"
    /// Submit a whole batched delegation plan in one call; Nucleic spawns a worker per task.
    public static let orchestraPlanToolName = "nucleic_delegate_plan"
    public static let qualifiedOrchestraPlanToolName = "mcp__nucleic__nucleic_delegate_plan"
    /// Provider-neutral counterpart to Claude Code's built-in `AskUserQuestion`: ask the human a
    /// multiple-choice question and block for their answer. Available on every provider, so an
    /// orchestrator can clarify intent regardless of which model is driving.
    public static let askUserToolName = "nucleic_ask_user"
    public static let qualifiedAskUserToolName = "mcp__nucleic__nucleic_ask_user"
    /// Supervisor drive loop: block until worker events arrive, then drain them.
    public static let orchestraSuperviseToolName = "nucleic_supervise"
    public static let qualifiedOrchestraSuperviseToolName = "mcp__nucleic__nucleic_supervise"
    /// Supervisor answers a worker's `nucleic_ask_supervisor`.
    public static let orchestraReplyToWorkerToolName = "nucleic_reply_to_worker"
    public static let qualifiedOrchestraReplyToWorkerToolName = "mcp__nucleic__nucleic_reply_to_worker"
    /// Worker asks its supervisor and blocks for the reply.
    public static let orchestraAskSupervisorToolName = "nucleic_ask_supervisor"
    public static let qualifiedOrchestraAskSupervisorToolName = "mcp__nucleic__nucleic_ask_supervisor"
    /// Mesh messaging (mesh casting "messages" channel): send a message to a session/topic.
    public static let meshSendMessageToolName = "nucleic_send_message"
    public static let qualifiedMeshSendMessageToolName = "mcp__nucleic__nucleic_send_message"
    /// Mesh messaging: subscribe this session to a broadcast topic.
    public static let meshSubscribeTopicToolName = "nucleic_subscribe_topic"
    public static let qualifiedMeshSubscribeTopicToolName = "mcp__nucleic__nucleic_subscribe_topic"
    /// Mesh messaging: block for the next inbound message.
    public static let meshWaitForMessageToolName = "nucleic_wait_for_message"
    public static let qualifiedMeshWaitForMessageToolName = "mcp__nucleic__nucleic_wait_for_message"
    /// Monitor: arm a background watch whose output re-invokes the session as new turns (the
    /// working replacement for the provider-native `Monitor`, which Nucleic blocks).
    public static let monitorToolName = "nucleic_monitor"
    public static let qualifiedMonitorToolName = "mcp__nucleic__nucleic_monitor"

    #if canImport(Network)
    private var listener: NWListener?
    #elseif os(Windows)
    /// The live NIO server channel (the Windows mirror of `listener` — docs/WINDOWS_PORT.md
    /// §4.3). Bound only to the WSL-facing interface, never 0.0.0.0 (§5).
    private var nioServerChannel: Channel?
    #else
    /// The listening TCP socket fd + its accept source (the Linux BSD-socket mirror of
    /// `listener`; same shape as the AF_UNIX pair below).
    private var tcpListenFD: Int32?
    private var tcpAcceptSource: DispatchSourceRead?
    #endif
    /// In-flight TCP bind, so concurrent `start(host:)` callers (the default once sessions share one
    /// control-container server via ``ApprovalServerRegistry``) join the SAME bind and all receive
    /// the real bound port. Without this, a second caller entering while the first is mid-bind saw a
    /// half-initialized listener and returned `port == 0`, which the agent's MCP config baked into an
    /// unreachable `:0` endpoint — stalling that session's first gated tool. Cleared when the bind
    /// settles (success or failure), so a failed bind can be retried.
    private var tcpStartTask: Task<UInt16, Error>?
    /// The interface the live TCP listener was bound to (`"127.0.0.1"` / `"0.0.0.0"`), nil when no
    /// TCP listener is bound. Lets ``start(host:)`` detect a run that needs a *wider* bind than the
    /// cached one (a loopback bind can't serve a containerized child over the VM gateway) and rebind
    /// instead of returning a port the child can never reach. Internal (not private) for tests.
    private(set) var boundTCPHost: String?
    private var handlers: [String: Handler] = [:]
    private var conflictHandlers: [String: ConflictHandler] = [:]
    private var hostExecHandlers: [String: HostExecHandler] = [:]
    private var macVMExecHandlers: [String: MacVMExecHandler] = [:]
    private var macVMComputerHandlers: [String: MacVMComputerHandler] = [:]
    private var macVMComputerBatchHandlers: [String: MacVMComputerBatchHandler] = [:]
    private var macVMClearNotificationsHandlers: [String: MacVMClearNotificationsHandler] = [:]
    private var macVMOperatorHandlers: [String: MacVMOperatorHandler] = [:]
    private var linuxVMExecHandlers: [String: MacVMExecHandler] = [:]
    private var linuxVMComputerHandlers: [String: MacVMComputerHandler] = [:]
    private var linuxVMComputerBatchHandlers: [String: MacVMComputerBatchHandler] = [:]
    private var linuxContainerHandlers: [String: ContainerToolHandler] = [:]
    private var orchestraSubagentHandlers: [String: OrchestraSubagentHandler] = [:]
    private var orchestraPlanHandlers: [String: OrchestraPlanServerHandler] = [:]
    private var askUserHandlers: [String: AskUserServerHandler] = [:]
    private var orchestraSuperviseHandlers: [String: OrchestraSuperviseServerHandler] = [:]
    private var orchestraReplyToWorkerHandlers: [String: OrchestraReplyToWorkerServerHandler] = [:]
    private var orchestraAskSupervisorHandlers: [String: OrchestraAskSupervisorServerHandler] = [:]
    private var meshSendMessageHandlers: [String: MeshSendMessageHandler] = [:]
    private var meshSubscribeTopicHandlers: [String: MeshSubscribeTopicHandler] = [:]
    private var meshWaitForMessageHandlers: [String: MeshWaitForMessageHandler] = [:]
    private var monitorHandlers: [String: MonitorHandler] = [:]
    private var macVMControlHandlers: [String: VMControlHandler] = [:]
    private var linuxVMControlHandlers: [String: VMControlHandler] = [:]
    private var gitReportHandlers: [String: GitReportHandler] = [:]
    private var ghReportHandlers: [String: GhReportHandler] = [:]
    private var commandReportHandlers: [String: CommandReportHandler] = [:]
    private var shellReportHandlers: [String: ShellReportHandler] = [:]
    private var preToolUseHandlers: [String: PreToolUseHandler] = [:]
    /// Per-tool-call "Skip" waiters: when the user taps **Skip** on a still-running host/VM exec card
    /// in the transcript, its in-flight handler must return *early* with a synthetic "skipped" reply so
    /// the agent gives up on the (apparently stuck) call and moves on — without killing the whole turn.
    /// Keyed by the call's fingerprint (normalized tool name + command); each entry holds the on-skip
    /// closures for the call(s) in flight with that fingerprint (usually one — more only if the agent
    /// fired two identical commands at once, in which case Skip trips both). Registered by
    /// ``runSkippable(name:command:work:skipped:)`` for the run's duration and torn down when it
    /// settles; ``skipToolCall(name:command:)`` fires them. The abandoned underlying process (the
    /// SSH/host command) keeps running in its disposable VM/host — Skip abandons the *wait*, not the OS
    /// process, which is the right thing for a hang the user just wants the agent to stop blocking on.
    private var skipHandlers: [String: [UUID: @Sendable () -> Void]] = [:]
    private var connectionTasks: [Int: Task<Void, Never>] = [:]
    private var nextConnectionID = 0
    /// In-flight `handle(_:)` tasks — one per request being served — indexed by id and by the
    /// request's bearer token. Handlers can suspend for a long time (an `approve` awaiting a human,
    /// a `nucleic_subagent` spawn awaiting a whole worker session), so when their session's run ends
    /// — the connection closes or the token unregisters — the task must be CANCELLED, not merely
    /// awaited: nobody can receive the response anymore, and a parked subagent spawn would otherwise
    /// go on to create and run a full worker for a dead supervisor. Entries remove themselves when
    /// the task finishes.
    private var handlerTasks: [UUID: Task<Void, Never>] = [:]
    private var handlerTaskIDsByToken: [String: Set<UUID>] = [:]
    /// How often an SSE-streamed `tools/call` response emits a keepalive comment while its handler
    /// is still running. A gated tool call can block for many minutes (a human answering `approve`,
    /// a `nucleic_subagent` spawn awaiting a whole worker session); the spawned CLI's HTTP client
    /// (Node/undici) aborts a request whose response stays byte-silent past its ~300s
    /// headers/body timeout — a layer *below* the MCP SDK, so the pinned `MCP_TOOL_TIMEOUT` never
    /// governs it. Emitting a keepalive well under that ceiling keeps the client's timeout (and
    /// every relay's idle clock) from ever firing on a legitimately long call. Set via ``init`` so a
    /// test can shorten it. See ``respondStreamingToolCall(_:on:)``.
    private let sseKeepaliveInterval: Duration
    public private(set) var port: UInt16 = 0
    /// The bound unix-socket path when started via ``start(unixSocketPath:)`` (nil for a TCP
    /// bind). Recorded so ``stop()`` can unlink the socket file it created.
    public private(set) var boundUnixSocketPath: String?
    #if !os(Windows)
    /// The listening AF_UNIX socket fd + its accept source, for the unix-socket transport.
    /// (No unix-socket transport on Windows — the control plane is gateway TCP, §5.)
    private var unixListenFD: Int32?
    private var unixAcceptSource: DispatchSourceRead?
    /// `(dev, ino)` of the bound unix socket node, captured at bind time. Compared against the live
    /// file on each ``start(unixSocketPath:)`` to detect the socket being unlinked (a `/tmp`-style
    /// reaper, a container teardown) or replaced by a different inode — the desync that used to
    /// permanently brick a shared control server. Nil when no unix socket is bound.
    private var boundUnixSocketInode: (dev: dev_t, ino: ino_t)?
    #endif

    /// `sseKeepaliveInterval` defaults to the production cadence; a test passes a short value to see
    /// several keepalive frames within its window (see ``sseKeepaliveInterval``).
    public init(sseKeepaliveInterval: Duration = .seconds(20)) {
        self.sseKeepaliveInterval = sseKeepaliveInterval
    }

    /// Returns the bound ephemeral port. `host` is the local interface to bind to;
    /// default loopback for host runs, `0.0.0.0` for sandbox runs so the containerized
    /// `claude` can reach the approval server over the VM gateway (bearer-token gated).
    @discardableResult
    public func start(host: String = "127.0.0.1") async throws -> UInt16 {
        #if canImport(Network)
        // Self-heal (the TCP mirror of `start(unixSocketPath:)`). The shared control-container
        // server is long-lived — on the default (non-vsock) transport it serves EVERY session in
        // the container over this one TCP bind — so it must not trust its first bind forever.
        // Keep the cached listener only while it is still `.ready` AND already serves the
        // requested interface (a wildcard bind serves loopback too; the reverse does not — a
        // loopback bind is unreachable from a containerized child over the VM gateway). A
        // listener that died post-`.ready` (network-stack transition, external cancel) would
        // otherwise keep publishing its stale port, which every new session baked into an
        // unreachable MCP config — surfacing as the CLI's opaque "Available MCP tools: none" and
        // poisoning ALL new sessions sharing the server until an app restart.
        if let existing = listener {
            let servesHost = boundTCPHost == host || boundTCPHost == "0.0.0.0"
            if existing.state == .ready, port != 0, servesHost { return port }
            invalidateTCPListener(existing)  // dead or too-narrow bind — rebind below
        }
        if let task = tcpStartTask { return try await task.value }  // bind in flight → join it
        let task = Task { [self] in try await bind(host: host) }
        tcpStartTask = task
        defer { tcpStartTask = nil }  // settle the single-flight (success or throw); allow a retry
        return try await task.value
        #elseif os(Windows)
        // Windows (docs/WINDOWS_PORT.md §4.3): same self-heal contract over a NIO listener.
        // A set channel is live only while `isActive`; a channel that died post-bind (network
        // stack reset) rebinds here instead of publishing its stale port.
        if let existing = nioServerChannel {
            let servesHost = boundTCPHost == host || boundTCPHost == "0.0.0.0"
            if existing.isActive, port != 0, servesHost { return port }
            invalidateNIOListener(existing)  // dead or too-narrow bind — rebind below
        }
        if let task = tcpStartTask { return try await task.value }  // bind in flight → join it
        let task = Task { [self] in try await performNIOBind(host: host) }
        tcpStartTask = task
        defer { tcpStartTask = nil }  // settle the single-flight (success or throw); allow a retry
        return try await task.value
        #else
        // Linux (runner): same self-heal contract over a BSD listener — the fd is only ever
        // closed by stop()/teardownTCPListener(), so a set fd is live; rebind when the cached
        // bind is missing or narrower than the run needs.
        if tcpListenFD != nil {
            let servesHost = boundTCPHost == host || boundTCPHost == "0.0.0.0"
            if port != 0, servesHost { return port }
            teardownTCPListener()  // too-narrow bind — rebind below
        }
        if let task = tcpStartTask { return try await task.value }  // bind in flight → join it
        let task = Task { [self] in try performTCPBind(host: host) }
        tcpStartTask = task
        defer { tcpStartTask = nil }  // settle the single-flight (success or throw); allow a retry
        return try await task.value
        #endif
    }

    #if canImport(Network)
    /// Drop the cached TCP bind because its listener is no longer usable — it died after `.ready`
    /// (`.failed` on a network-stack transition, an out-from-under cancel) or it's bound to a
    /// narrower interface than the run needs. Clears `listener`/`port`/`boundTCPHost` so the next
    /// ``start(host:)`` rebinds fresh instead of returning the stale port. Identity-checked so a
    /// listener that was already replaced (``stop()``, an earlier self-heal) can't clobber its
    /// successor's live bind. Established connections are independent of the listener and unaffected.
    private func invalidateTCPListener(_ dead: NWListener) {
        guard listener === dead else { return }
        listener = nil
        port = 0
        boundTCPHost = nil
        dead.cancel()
    }
    #elseif os(Windows)
    /// Windows counterpart of `invalidateTCPListener`: close the dead/too-narrow NIO server
    /// channel and clear the cached bind so the next ``start(host:)`` rebinds. Identity-checked
    /// like the Darwin path so a replaced listener can't clobber its successor.
    private func invalidateNIOListener(_ dead: Channel) {
        guard nioServerChannel === dead else { return }
        nioServerChannel = nil
        port = 0
        boundTCPHost = nil
        dead.close(promise: nil)
    }

    /// One shared event-loop group for the approval servers' listeners + connections; sized
    /// small — the control plane is a handful of long-lived connections per container.
    private static let nioGroup = MultiThreadedEventLoopGroup(numberOfThreads: 2)

    /// NIO TCP bind (the Windows mirror of `bind(host:)`/`performTCPBind(host:)`): every
    /// accepted child channel becomes an ``NIOByteConn`` driving the same transport-agnostic
    /// `adopt` → `serve` HTTP/JSON-RPC pipeline. Publishes `nioServerChannel`/`port`/
    /// `boundTCPHost` only after the bind resolves, so no caller observes a half-bound state.
    private func performNIOBind(host: String) async throws -> UInt16 {
        let bootstrap = ServerBootstrap(group: Self.nioGroup)
            .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
            .childChannelInitializer { channel in
                let conn = NIOByteConn(channel: channel)
                return channel.pipeline.addHandler(NIOByteConnHandler(conn: conn)).map {
                    [weak self] in
                    guard let self else {
                        channel.close(promise: nil)
                        return
                    }
                    Task { await self.adopt(conn) }
                }
            }
        let channel = try await bootstrap.bind(host: host, port: 0).get()
        guard let boundPort = channel.localAddress?.port, boundPort > 0 else {
            channel.close(promise: nil)
            throw UnixSocketError("NIO bind(\(host):0) resolved with no local port")
        }
        nioServerChannel = channel
        port = UInt16(boundPort)
        boundTCPHost = host
        return UInt16(boundPort)
    }
    #else
    /// Linux counterpart of `invalidateTCPListener`: cancel the accept source (its cancel handler
    /// closes the listening fd) and clear the cached bind so the next ``start(host:)`` rebinds.
    private func teardownTCPListener() {
        tcpAcceptSource?.cancel()
        tcpAcceptSource = nil
        tcpListenFD = nil
        port = 0
        boundTCPHost = nil
    }

    /// BSD-socket TCP bind for Linux — the same primitives (and the same accept-source → adopt
    /// pipeline) as ``performUnixBind(path:)``, pointed at `host`:ephemeral. Returns the bound
    /// port. Synchronous, like the unix bind, so the actor never exposes a half-bound listener.
    private func performTCPBind(host: String) throws -> UInt16 {
        let fd = socket(AF_INET, sockStreamType, 0)
        guard fd >= 0 else {
            throw UnixSocketError("socket() failed: \(String(cString: strerror(errno)))")
        }
        _ = fcntl(fd, F_SETFD, FD_CLOEXEC)
        var one: Int32 = 1
        setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, socklen_t(MemoryLayout<Int32>.size))

        let raw = inet_addr(host)
        guard raw != in_addr_t(0xFFFF_FFFF) else {
            posixCloseFD(fd)
            throw UnixSocketError("bad bind host: \(host)")
        }
        var addr = sockaddr_in()
        addr.sin_family = sa_family_t(AF_INET)
        addr.sin_port = UInt16(0).bigEndian  // ephemeral
        addr.sin_addr = in_addr(s_addr: raw)
        let bindRC = withUnsafePointer(to: &addr) { ptr in
            ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) {
                posixBind(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size))
            }
        }
        guard bindRC == 0 else {
            let err = String(cString: strerror(errno))
            posixCloseFD(fd)
            throw UnixSocketError("bind(\(host):0) failed: \(err)")
        }
        guard listen(fd, SOMAXCONN) == 0 else {
            let err = String(cString: strerror(errno))
            posixCloseFD(fd)
            throw UnixSocketError("listen(\(host)) failed: \(err)")
        }
        var bound = sockaddr_in()
        var len = socklen_t(MemoryLayout<sockaddr_in>.size)
        let nameRC = withUnsafeMutablePointer(to: &bound) { ptr in
            ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { getsockname(fd, $0, &len) }
        }
        guard nameRC == 0 else {
            let err = String(cString: strerror(errno))
            posixCloseFD(fd)
            throw UnixSocketError("getsockname failed: \(err)")
        }
        let boundPort = UInt16(bigEndian: bound.sin_port)

        let source = DispatchSource.makeReadSource(
            fileDescriptor: fd, queue: .global(qos: .userInitiated))
        source.setEventHandler { [weak self] in
            let client = accept(fd, nil, nil)
            guard client >= 0 else { return }
            _ = fcntl(client, F_SETFD, FD_CLOEXEC)
            guard let self else {
                posixCloseFD(client)
                return
            }
            Task { await self.adopt(UnixSocketByteConn(fd: client)) }
        }
        source.setCancelHandler { posixCloseFD(fd) }
        source.resume()

        tcpListenFD = fd
        tcpAcceptSource = source
        port = boundPort
        boundTCPHost = host
        return boundPort
    }
    #endif

    /// Bind the server to a **unix domain socket** instead of a TCP port — the transport for a
    /// sandboxed agent, whose connection is relayed in over vsock (the framework's
    /// `UnixSocketConfiguration`). No IP listener means macOS raises no incoming-connection /
    /// local-network prompts. The HTTP/JSON-RPC dispatch is byte-for-byte identical to the TCP
    /// path; only the listening socket differs.
    ///
    /// A real `AF_UNIX` listener (not `NWListener` with a `.unix` endpoint, which Network.framework
    /// matches in-process and never materializes on disk) so the framework's host-side relay — an
    /// ordinary socket client — can connect to it. Any stale socket file at `path` (e.g. left by a
    /// crash) is removed first, since `bind` fails if the path already exists. Returns the bound path.
    @discardableResult
    public func start(unixSocketPath path: String) async throws -> String {
        #if os(Windows)
        // No vsock-relayed unix socket on Windows — the control plane is gateway TCP
        // (docs/WINDOWS_PORT.md §5); backends take the `start(host:)` path there.
        throw UnixSocketError("the unix-socket control transport does not exist on Windows")
        #else
        // Self-heal (the core fix). A control container's approval server is long-lived and shared by
        // every session in the box, reached through this ONE socket; a single desync used to brick the
        // whole container — every new session's `claude` failing MCP init ("Available MCP tools: none")
        // and every running session stalling on its next gated call — until an app restart, because the
        // cached server trusted its first bind forever. So RE-VALIDATE on every call: keep the existing
        // listener ONLY when it's still a live listening socket AND the file at `path` is still the exact
        // socket node we bound. If the file was unlinked (a `/tmp`-style reaper, a container teardown) or
        // replaced by a different inode, or the fd died, drop the orphaned listener and rebind so this
        // call converges on a working socket. No `await` runs before the rebind, so concurrent callers on
        // this actor can't interleave a half-bound state.
        if unixListenFD != nil {
            if boundUnixSocketPath == path, unixListenerIsHealthy(path: path) {
                return path
            }
            teardownUnixListener()  // stale / desynced — rebind below
        }
        try performUnixBind(path: path)
        return path
        #endif
    }

    #if !os(Windows)
    /// Bind + listen a fresh AF_UNIX socket at `path` and wire its accept source, publishing
    /// `unixListenFD` / `unixAcceptSource` / `boundUnixSocketPath` / `boundUnixSocketInode`. The
    /// single bind site for both the first `start(unixSocketPath:)` and every self-heal rebind, so
    /// they can't drift. Synchronous (no suspension) so the actor never exposes a half-bound listener.
    private func performUnixBind(path: String) throws {
        let fm = FileManager.default
        try? fm.createDirectory(
            atPath: (path as NSString).deletingLastPathComponent,
            withIntermediateDirectories: true)
        try? fm.removeItem(atPath: path)  // a bind fails if the path already exists

        let fd = socket(AF_UNIX, sockStreamType, 0)
        guard fd >= 0 else {
            throw UnixSocketError("socket() failed: \(String(cString: strerror(errno)))")
        }
        _ = fcntl(fd, F_SETFD, FD_CLOEXEC)

        var addr = sockaddr_un()
        addr.sun_family = sa_family_t(AF_UNIX)
        let pathBytes = Array(path.utf8)
        let capacity = MemoryLayout.size(ofValue: addr.sun_path)
        guard pathBytes.count < capacity else {
            close(fd)
            throw UnixSocketError("socket path too long (\(pathBytes.count) ≥ \(capacity)): \(path)")
        }
        withUnsafeMutablePointer(to: &addr.sun_path) { raw in
            raw.withMemoryRebound(to: CChar.self, capacity: capacity) { dst in
                for (i, byte) in pathBytes.enumerated() { dst[i] = CChar(bitPattern: byte) }
                dst[pathBytes.count] = 0
            }
        }
        let bindRC = withUnsafePointer(to: &addr) { ptr in
            ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) {
                posixBind(fd, $0, socklen_t(MemoryLayout<sockaddr_un>.size))
            }
        }
        guard bindRC == 0 else {
            let err = String(cString: strerror(errno))
            close(fd)
            throw UnixSocketError("bind(\(path)) failed: \(err)")
        }
        // Only the host-side relay connects here; lock the socket to the owner.
        chmod(path, 0o600)
        // Backlog sized for the shared control container: EVERY session in the box reaches us
        // through this one socket, and the traffic is connection-churny by design — each SSE-
        // streamed tools/call closes its connection, and every intercepted git/gh/command event is
        // a short-lived POST — so parallel sessions easily burst past the old backlog of 16.
        // Overflow makes the relay's connect(2) fail ECONNREFUSED, surfacing to the agent as MCP
        // errors during the burst. SOMAXCONN (128) absorbs bursts; the kernel clamps it anyway.
        guard listen(fd, SOMAXCONN) == 0 else {
            let err = String(cString: strerror(errno))
            close(fd)
            try? fm.removeItem(atPath: path)
            throw UnixSocketError("listen(\(path)) failed: \(err)")
        }

        // Accept off a background source so the actor never blocks; each accepted fd drives the same
        // HTTP/JSON-RPC serve loop as a TCP connection.
        let source = DispatchSource.makeReadSource(
            fileDescriptor: fd, queue: .global(qos: .userInitiated))
        source.setEventHandler { [weak self] in
            let client = accept(fd, nil, nil)
            guard client >= 0 else { return }
            _ = fcntl(client, F_SETFD, FD_CLOEXEC)
            guard let self else {
                close(client)
                return
            }
            Task { await self.adopt(UnixSocketByteConn(fd: client)) }
        }
        source.setCancelHandler { close(fd) }
        source.resume()

        unixListenFD = fd
        unixAcceptSource = source
        boundUnixSocketPath = path
        boundUnixSocketInode = Self.inode(ofPath: path)
    }

    /// Whether a unix-socket control endpoint is currently bound and listening at `path`. The
    /// containerized spawn path checks this (mirroring the TCP path's `port != 0` guard) so a
    /// desynced control socket fails the run loudly and retryably instead of surfacing only as the
    /// CLI's opaque "Available MCP tools: none" — and poisoning every session sharing the container.
    public func isUnixSocketListening(atPath path: String) -> Bool {
        boundUnixSocketPath == path && unixListenerIsHealthy(path: path)
    }
    #else
    /// Windows: no unix-socket transport exists, so nothing is ever listening at a path.
    public func isUnixSocketListening(atPath path: String) -> Bool { false }
    #endif

    #if !os(Windows)

    /// Whether the current unix listener is still usable for `path`: we hold the listening fd AND the
    /// file at `path` is still the very socket node we bound (not unlinked, not replaced). The fd
    /// itself is only ever closed by ``stop()`` / ``teardownUnixListener()``, both of which also clear
    /// `unixListenFD`, so a set fd is always live — the failure modes that actually occur are the file
    /// being reaped (stat fails) or replaced by a fresh inode (identity mismatch), which is exactly
    /// what this checks. Cheap enough to run on every `start(unixSocketPath:)` so a shared server
    /// converges on a live socket instead of trusting a first bind a reaper or teardown may have voided.
    private func unixListenerIsHealthy(path: String) -> Bool {
        guard unixListenFD != nil, let bound = boundUnixSocketInode else { return false }
        guard let current = Self.inode(ofPath: path) else { return false }  // unlinked (reaped/torn down)
        return current == bound  // false → the path now points at a different socket node
    }

    /// Tear down the current unix listener without binding a replacement — cancels the accept source
    /// (its cancel handler closes the fd) and clears the bound-socket bookkeeping. Leaves any file at
    /// the path for ``performUnixBind(path:)`` to remove, so a rebind starts clean.
    private func teardownUnixListener() {
        unixAcceptSource?.cancel()
        unixAcceptSource = nil
        unixListenFD = nil
        boundUnixSocketInode = nil
    }

    /// `(st_dev, st_ino)` identity of the filesystem node at `path`, or nil if it doesn't exist.
    /// Used to detect a socket file unlinked or replaced out from under a live listener.
    private static func inode(ofPath path: String) -> (dev: dev_t, ino: ino_t)? {
        var st = stat()
        guard stat(path, &st) == 0 else { return nil }
        return (st.st_dev, st.st_ino)
    }
    #endif

    #if canImport(Network)
    /// Bind a TCP listener and resume once it is `.ready` (or throw on failure). The connection-
    /// serving pipeline (``adopt(_:)`` → ``serve(_:)``) is transport-agnostic, so this drives the
    /// exact same HTTP/JSON-RPC dispatch as the unix-socket path. Returns the bound ephemeral port.
    ///
    /// `self.listener` and `self.port` are published only AFTER `.ready` — never before the await —
    /// so a concurrent caller can't observe a half-bound listener with a 0 port, and a `.failed`
    /// bind leaves both unset (no stale listener pinning a permanent port 0). The connection handler
    /// is wired before `start`; it captures `self` directly, so it works whether or not
    /// `self.listener` is assigned yet.
    private func bind(host: String) async throws -> UInt16 {
        let parameters = NWParameters.tcp
        parameters.requiredLocalEndpoint = NWEndpoint.hostPort(
            host: NWEndpoint.Host(host), port: .any)
        let listener = try NWListener(using: parameters)

        listener.newConnectionHandler = { [weak self] connection in
            guard let self else {
                connection.cancel()
                return
            }
            Task { await self.adopt(NWByteConn(connection)) }
        }

        let resumeOnce = OnceBox()
        // The handler stays installed for the listener's whole life: first it settles the bind
        // (resuming the continuation exactly once), then it keeps watching so a post-`.ready`
        // death — `.failed` on a network-stack transition, an out-from-under `.cancelled` —
        // invalidates the cached bind and the next `start(host:)` rebinds, instead of the dead
        // listener's stale port poisoning every new session that shares this server.
        do {
            try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
                listener.stateUpdateHandler = { [weak self, weak listener] state in
                    switch state {
                    case .ready:
                        if resumeOnce.claim() { cont.resume() }
                    case .waiting(let error):
                        // Only a bind-time failure; post-`.ready` waiting may recover — leave it.
                        if resumeOnce.claim() { cont.resume(throwing: error) }
                    case .failed(let error):
                        if resumeOnce.claim() {
                            cont.resume(throwing: error)
                        } else if let self, let listener {
                            Task { await self.invalidateTCPListener(listener) }
                        }
                    case .cancelled:
                        if resumeOnce.claim() {
                            cont.resume(throwing: CancellationError())
                        } else if let self, let listener {
                            Task { await self.invalidateTCPListener(listener) }
                        }
                    default:
                        break
                    }
                }
                listener.start(queue: .global(qos: .userInitiated))
            }
        } catch {
            // A `.waiting` bind failure leaves the listener alive and able to become ready later —
            // cancel it so an unpublished listener can't linger holding a socket nobody knows about.
            listener.cancel()
            throw error
        }
        // Ready and bound: publish atomically (no suspension between these and the return).
        self.listener = listener
        self.port = listener.port?.rawValue ?? 0
        self.boundTCPHost = host
        return self.port
    }
    #endif

    public func stop() {
        tcpStartTask?.cancel()
        tcpStartTask = nil
        #if canImport(Network)
        listener?.cancel()
        listener = nil
        #elseif os(Windows)
        nioServerChannel?.close(promise: nil)
        nioServerChannel = nil
        #else
        tcpAcceptSource?.cancel()  // its cancel handler closes the listening fd
        tcpAcceptSource = nil
        tcpListenFD = nil
        #endif
        boundTCPHost = nil
        #if !os(Windows)
        unixAcceptSource?.cancel()  // its cancel handler closes the listening fd
        unixAcceptSource = nil
        unixListenFD = nil
        boundUnixSocketInode = nil
        #endif
        if let path = boundUnixSocketPath {
            try? FileManager.default.removeItem(atPath: path)
            boundUnixSocketPath = nil
        }
        for task in connectionTasks.values { task.cancel() }
        connectionTasks.removeAll()
        for task in handlerTasks.values { task.cancel() }
        handlerTasks.removeAll()
        handlerTaskIDsByToken.removeAll()
        handlers.removeAll()
        conflictHandlers.removeAll()
        preToolUseHandlers.removeAll()
        hostExecHandlers.removeAll()
        macVMExecHandlers.removeAll()
        macVMComputerHandlers.removeAll()
        macVMOperatorHandlers.removeAll()
        orchestraSubagentHandlers.removeAll()
        gitReportHandlers.removeAll()
        ghReportHandlers.removeAll()
        commandReportHandlers.removeAll()
        shellReportHandlers.removeAll()
        // Any Skip waiters whose handler was cancelled mid-flight (so its own teardown never ran) go
        // with the reset — the calls they belonged to can no longer complete.
        skipHandlers.removeAll()
    }

    /// TEST-ONLY seam: cancel the live TCP listener *out from under* the cached bind — simulating
    /// the post-`.ready` death (a network-stack `.failed`, an external cancel) that the state
    /// watcher and ``start(host:)``'s re-validation exist to heal. Does NOT clear the cached
    /// `listener`/`port`, so tests can observe the watcher doing that itself. Internal for
    /// `@testable` access; never called from production code.
    func debugCancelTCPListenerOutFromUnder() {
        #if canImport(Network)
        listener?.cancel()
        #elseif os(Windows)
        nioServerChannel?.close(promise: nil)
        #else
        tcpAcceptSource?.cancel()
        #endif
    }

    public func register(token: String, handler: @escaping Handler) {
        handlers[token] = handler
    }

    /// Register the per-session `check_conflict` handler under the same bearer token.
    public func registerConflict(token: String, handler: @escaping ConflictHandler) {
        conflictHandlers[token] = handler
    }

    /// Register the per-session `host_exec` handler. Registering it both advertises the tool
    /// in `tools/list` for this token and routes calls to the host runner (HOST_EXEC).
    public func registerHostExec(token: String, handler: @escaping HostExecHandler) {
        hostExecHandlers[token] = handler
    }

    /// Register the per-session `mac_vm_exec` handler. Registering it both advertises the tool in
    /// `tools/list` for this token and routes calls to the macOS-VM runner (see docs/MACOS_VM.md).
    public func registerMacVMExec(token: String, handler: @escaping MacVMExecHandler) {
        macVMExecHandlers[token] = handler
    }

    /// Register the per-session `mac_vm_computer` handler. Registering it both advertises the tool in
    /// `tools/list` for this token and routes computer-use actions to the macOS-VM GUI driver.
    public func registerMacVMComputer(token: String, handler: @escaping MacVMComputerHandler) {
        macVMComputerHandlers[token] = handler
    }

    /// Register the per-session `mac_vm_computer_batch` handler — advertises the tool and routes an
    /// ordered batch of computer-use actions to the macOS-VM GUI driver in one call.
    public func registerMacVMComputerBatch(
        token: String, handler: @escaping MacVMComputerBatchHandler
    ) {
        macVMComputerBatchHandlers[token] = handler
    }

    /// Register the per-session `mac_vm_clear_notifications` handler — advertises the tool and routes the
    /// "dismiss on-screen notifications" action (a fixed `killall NotificationCenter`) to the VM runner.
    public func registerMacVMClearNotifications(
        token: String, handler: @escaping MacVMClearNotificationsHandler
    ) {
        macVMClearNotificationsHandlers[token] = handler
    }

    /// Register the per-session `mac_vm_request_operator` handler. Registering it both advertises the
    /// tool in `tools/list` for this token and routes operator-help requests to the assist coordinator.
    public func registerMacVMOperator(token: String, handler: @escaping MacVMOperatorHandler) {
        macVMOperatorHandlers[token] = handler
    }

    /// Register the per-session `linux_vm_exec` handler. Registering it both advertises the tool in
    /// `tools/list` for this token and routes calls to the Linux-VM runner (see docs/LINUX_VM.md).
    public func registerLinuxVMExec(token: String, handler: @escaping MacVMExecHandler) {
        linuxVMExecHandlers[token] = handler
    }

    /// Register the per-session `linux_vm_computer` handler — advertises the tool and routes
    /// computer-use actions to the Linux-VM GUI driver.
    public func registerLinuxVMComputer(token: String, handler: @escaping MacVMComputerHandler) {
        linuxVMComputerHandlers[token] = handler
    }

    /// Register the per-session `linux_vm_computer_batch` handler — the Linux sibling of
    /// ``registerMacVMComputerBatch(token:handler:)``.
    public func registerLinuxVMComputerBatch(
        token: String, handler: @escaping MacVMComputerBatchHandler
    ) {
        linuxVMComputerBatchHandlers[token] = handler
    }

    /// Register the per-session `linux_container` handler — advertises the tool and routes the
    /// agent's container lifecycle ops (create/exec/stop/remove/list) to the container runner.
    public func registerLinuxContainer(token: String, handler: @escaping ContainerToolHandler) {
        linuxContainerHandlers[token] = handler
    }

    /// Register the per-session Nucleic-managed Orchestra subagent handler. Registering it advertises
    /// `nucleic_subagent` and routes worker spawning back into the app instead of Claude's native Task.
    public func registerOrchestraSubagent(token: String, handler: @escaping OrchestraSubagentHandler) {
        orchestraSubagentHandlers[token] = handler
    }

    /// Register the supervisor-side `nucleic_delegate_plan` handler — advertises the tool and routes a
    /// submitted plan (batched task list) into Nucleic, which spawns a worker per task.
    public func registerOrchestraPlan(token: String, handler: @escaping OrchestraPlanServerHandler) {
        orchestraPlanHandlers[token] = handler
    }

    /// Register the `nucleic_ask_user` handler — advertises the tool and, on call, surfaces the
    /// questions to the human and blocks until they answer. Provider-neutral: registered for any run
    /// that can reach the user, so a non-Claude orchestrator has the same clarify-before-you-plan
    /// affordance as Claude's built-in `AskUserQuestion`.
    public func registerAskUser(token: String, handler: @escaping AskUserServerHandler) {
        askUserHandlers[token] = handler
    }

    /// Register the supervisor-side `nucleic_supervise` handler — advertises the tool and, on call,
    /// blocks until the supervisor's workers report events, then returns the drained batch.
    public func registerOrchestraSupervise(
        token: String, handler: @escaping OrchestraSuperviseServerHandler
    ) {
        orchestraSuperviseHandlers[token] = handler
    }

    /// Register the supervisor-side `nucleic_reply_to_worker` handler — advertises the tool and routes
    /// the supervisor's answer to a worker waiting in `nucleic_ask_supervisor`.
    public func registerOrchestraReplyToWorker(
        token: String, handler: @escaping OrchestraReplyToWorkerServerHandler
    ) {
        orchestraReplyToWorkerHandlers[token] = handler
    }

    /// Register the worker-side `nucleic_ask_supervisor` handler — advertises the tool and, on call,
    /// routes the worker's question to its supervisor and blocks until the supervisor replies (or the
    /// supervisor's run ends). Registered only for machine-spawned worker sessions.
    public func registerOrchestraAskSupervisor(
        token: String, handler: @escaping OrchestraAskSupervisorServerHandler
    ) {
        orchestraAskSupervisorHandlers[token] = handler
    }

    /// Register the mesh-messaging handlers — advertises `nucleic_send_message` /
    /// `nucleic_subscribe_topic` and routes them to the host's mediated publish path
    /// (mesh casting "messages" channel). Registered only when agent messaging is enabled.
    public func registerMeshMessaging(
        token: String,
        send: @escaping MeshSendMessageHandler,
        subscribe: @escaping MeshSubscribeTopicHandler,
        waitForMessage: @escaping MeshWaitForMessageHandler
    ) {
        meshSendMessageHandlers[token] = send
        meshSubscribeTopicHandlers[token] = subscribe
        meshWaitForMessageHandlers[token] = waitForMessage
    }

    /// Register the per-session `nucleic_monitor` handler — advertises the tool and routes an
    /// arm request to the backend, which owns the watch across turns and delivers its events
    /// out-of-band by re-invoking the session (see `MonitorEventSink`).
    public func registerMonitor(token: String, handler: @escaping MonitorHandler) {
        monitorHandlers[token] = handler
    }

    /// Register the per-session `mac_vm_control` handler — advertises the tool and routes the agent's
    /// macOS-VM lifecycle ops (status/stop/suspend/resume/restart/kill) to the VM manager.
    public func registerMacVMControl(token: String, handler: @escaping VMControlHandler) {
        macVMControlHandlers[token] = handler
    }

    /// Register the per-session `linux_vm_control` handler — the Linux-VM sibling of
    /// ``registerMacVMControl(token:handler:)``.
    public func registerLinuxVMControl(token: String, handler: @escaping VMControlHandler) {
        linuxVMControlHandlers[token] = handler
    }

    /// The `tools/list` entry for a VM lifecycle tool (`mac_vm_control` / `linux_vm_control`) — the two
    /// differ only by name, the guest-OS word in the prose, and whether the guest is capped, so they
    /// share one schema builder. `sibling` is the exec tool that boots the VM, named so the agent
    /// knows what it's controlling. `capped` is true only for macOS: macOS itself limits concurrent
    /// guests (``MacVMSettings/maxConcurrentVMs``), so an idle macOS VM starves other agents and MUST
    /// be freed. Linux VMs have no such ceiling, so their entry asks only for ordinary RAM courtesy —
    /// claiming a cap there would be false and would push needless suspend/resume churn.
    private static func vmControlToolSchema(
        name: String, osLabel: String, sibling: String, capped: Bool
    ) -> JSONValue {
        let lifecycleDuty = capped
            ? macVMFreeWhenIdleNotice + " It's also torn down for you at session end."
            : "\(osLabel) VMs are NOT capped the way macOS VMs are, so there is no slot to race for "
                + "— you may leave this one running across a task. Still, `suspend` or `stop` one "
                + "you're done with to give the host its RAM back. It's also torn down for you at "
                + "session end."
        return .object([
            "name": .string(name),
            "description": .string(
                "Manage the lifecycle of THIS session's isolated \(osLabel) VM — the one `\(sibling)` "
                    + "boots. Use it to free host resources or reset the VM without ending the session. "
                    + "One `op`: `status` reports whether the VM is running / suspended / stopped / "
                    + "absent (a free, read-only query); `stop` powers it off but keeps its disk, so the "
                    + "next command reboots it (frees its RAM); `suspend` saves the VM's full state to "
                    + "disk and frees its RAM AND its VM slot (so another VM can boot in its place), and "
                    + "`resume` restores it exactly where it left off — the next command also "
                    + "auto-resumes; `restart` stops and reboots it in place (reclaims RAM a long-running "
                    + "VM holds); `kill` powers it off AND deletes its disk clone (the next command "
                    + "boots a fresh one — you lose any state you built up inside it). These ops act only "
                    + "on your own disposable VM, so they need NO approval. Returns {ok, state, message}. "
                    + lifecycleDuty),
            "inputSchema": .object([
                "type": .string("object"),
                "properties": .object([
                    "op": .object([
                        "type": .string("string"),
                        "description": .string(
                            "One of: status, stop, suspend, resume, restart, kill."),
                    ])
                ]),
                "required": .array([.string("op")]),
            ]),
        ])
    }

    /// Register the per-session `git` interceptor report handler under the same bearer token.
    /// Called for the `POST /git-event` route the in-container shim uses.
    public func registerGitReport(token: String, handler: @escaping GitReportHandler) {
        gitReportHandlers[token] = handler
    }

    /// Register the per-session `gh` interceptor report handler under the same bearer token.
    /// Called for the `POST /gh-event` route the in-container shim uses.
    public func registerGhReport(token: String, handler: @escaping GhReportHandler) {
        ghReportHandlers[token] = handler
    }

    /// Register the per-session non-git command report handler (`POST /command-event`).
    public func registerCommandReport(token: String, handler: @escaping CommandReportHandler) {
        commandReportHandlers[token] = handler
    }

    /// Register the per-session nash shell-event handler (`POST /shell-event`).
    public func registerShellReport(token: String, handler: @escaping ShellReportHandler) {
        shellReportHandlers[token] = handler
    }

    /// Register the per-session Codex edit gate (`POST /pre-tool-use`, docs/LOCKING.md §4.6).
    /// MUST be registered for every codex run in a Nucleic Control container: the container's
    /// managed config points every codex at this route, and the hook fails **closed** on a token it
    /// can't authenticate — an unregistered token would deny the run's tool calls rather than let
    /// them through unlocked.
    public func registerPreToolUse(token: String, handler: @escaping PreToolUseHandler) {
        preToolUseHandlers[token] = handler
    }

    public func unregister(token: String) {
        // The token's run is over: unwind its suspended calls first (a parked `nucleic_subagent`
        // spawn, a dangling `approve`) so nothing keeps working on behalf of the dead run. The
        // connection may well stay open — on a shared control-container server it carries other
        // sessions' traffic — so token unregistration, not connection close, is this signal.
        cancelHandlerTasks(token: token)
        handlers.removeValue(forKey: token)
        conflictHandlers.removeValue(forKey: token)
        preToolUseHandlers.removeValue(forKey: token)
        hostExecHandlers.removeValue(forKey: token)
        macVMExecHandlers.removeValue(forKey: token)
        macVMComputerHandlers.removeValue(forKey: token)
        macVMClearNotificationsHandlers.removeValue(forKey: token)
        macVMOperatorHandlers.removeValue(forKey: token)
        linuxVMExecHandlers.removeValue(forKey: token)
        linuxVMComputerHandlers.removeValue(forKey: token)
        linuxContainerHandlers.removeValue(forKey: token)
        orchestraSubagentHandlers.removeValue(forKey: token)
        orchestraPlanHandlers.removeValue(forKey: token)
        askUserHandlers.removeValue(forKey: token)
        orchestraSuperviseHandlers.removeValue(forKey: token)
        orchestraReplyToWorkerHandlers.removeValue(forKey: token)
        orchestraAskSupervisorHandlers.removeValue(forKey: token)
        meshSendMessageHandlers.removeValue(forKey: token)
        meshSubscribeTopicHandlers.removeValue(forKey: token)
        meshWaitForMessageHandlers.removeValue(forKey: token)
        monitorHandlers.removeValue(forKey: token)
        macVMControlHandlers.removeValue(forKey: token)
        linuxVMControlHandlers.removeValue(forKey: token)
        gitReportHandlers.removeValue(forKey: token)
        ghReportHandlers.removeValue(forKey: token)
        commandReportHandlers.removeValue(forKey: token)
        shellReportHandlers.removeValue(forKey: token)
    }

    /// The MCP server entry for `--mcp-config` (inline JSON). `host` is the address the
    /// child reaches the server at — loopback for host runs, the VM gateway IP for a
    /// containerized child (which can't see the host's `127.0.0.1`).
    public nonisolated func mcpConfigJSON(
        host: String = "127.0.0.1", port: UInt16, token: String
    ) -> String {
        JSONValue.object([
            "mcpServers": .object([
                "nucleic": .object([
                    "type": .string("http"),
                    "url": .string("http://\(host):\(port)/mcp"),
                    "headers": .object(["Authorization": .string("Bearer \(token)")]),
                ])
            ])
        ]).canonicalString()
    }

    // MARK: - Connection handling

    private func adopt(_ conn: any ByteConn) {
        let id = nextConnectionID
        nextConnectionID += 1
        let task = Task {
            await self.serve(conn)
            self.forget(id)
        }
        connectionTasks[id] = task
    }

    private func forget(_ id: Int) {
        connectionTasks.removeValue(forKey: id)
    }

    private func serve(_ conn: any ByteConn) async {
        defer { conn.close() }
        var buffer = Data()
        // Requests are read serially (HTTP framing on one socket is inherently ordered), but each is
        // *handled* on its own task so a long-running call — most importantly a `nucleic_subagent`
        // spawn that blocks until its worker finishes — doesn't stall later requests pipelined on the
        // same keep-alive connection. That is what lets an Orchestra supervisor's parallel subagent
        // calls actually run in parallel instead of one-at-a-time. Responses must still leave in
        // request order (HTTP/1.1 has no response IDs), so each handler awaits the previous one's
        // send before writing its own; `previous` retains the in-flight chain for draining on close.
        var previous: Task<Void, Never>? = nil
        // Ids of every handler task spawned on this connection. Finished tasks removed themselves
        // from `handlerTasks`, so cancelling by id at teardown only reaches the still-running ones.
        var spawned: [UUID] = []
        while !Task.isCancelled {
            guard let request = await nextRequest(on: conn, buffer: &buffer) else { break }
            let close = request.headers["connection"]?.lowercased() == "close"
            let prior = previous
            let taskID = UUID()
            let token = Self.bearerToken(of: request)
            // A gated `tools/call` from an SSE-capable client is answered as a keepalive'd event
            // stream (``respondStreamingToolCall``) so a handler that blocks for minutes never goes
            // byte-silent long enough for the client to abort the request. Everything else keeps the
            // buffered single-response path. Decided here on the actor (cheap: header-gated before any
            // parse) so the streaming branch inherits the same per-request task registration, ordering
            // (`await prior?.value`), and cancellation as the buffered one.
            let streaming = streamableToolCall(from: request)
            let task = Task {
                if streaming {
                    await prior?.value
                    await self.respondStreamingToolCall(request, on: conn)
                } else {
                    let response = await self.handle(request)
                    await prior?.value
                    try? await self.sendResponse(response, on: conn)
                }
                self.handlerTaskFinished(taskID, token: token)
            }
            previous = task
            spawned.append(taskID)
            // No suspension since the Task was created (this actor still holds `serve`), so the
            // task can't have reached `handlerTaskFinished` yet — registration never races removal.
            registerHandlerTask(task, id: taskID, token: token)
            if close { break }
        }
        // The connection is dead (client hung up, `Connection: close`, or the server is stopping):
        // no response can be delivered anymore, so CANCEL the still-running handlers instead of
        // letting them keep working for a peer that's gone — this is what unwinds an in-flight
        // `nucleic_subagent` spawn when its supervisor is killed. Cancellation is cooperative;
        // handlers that finish anyway just flush into the closed socket harmlessly.
        for id in spawned { handlerTasks[id]?.cancel() }
        // Let outstanding handlers unwind and flush (in order) before `defer` closes the socket.
        await previous?.value
    }

    /// The request's bearer token (empty when absent/malformed) — the key handler tasks are
    /// indexed under so `unregister(token:)` can cancel a dead session's in-flight calls.
    private static func bearerToken(of request: HTTPRequest) -> String {
        let authorization = request.headers["authorization"] ?? ""
        return authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
    }

    private func registerHandlerTask(_ task: Task<Void, Never>, id: UUID, token: String) {
        handlerTasks[id] = task
        if !token.isEmpty { handlerTaskIDsByToken[token, default: []].insert(id) }
    }

    private func handlerTaskFinished(_ id: UUID, token: String) {
        handlerTasks.removeValue(forKey: id)
        if !token.isEmpty, var ids = handlerTaskIDsByToken[token] {
            ids.remove(id)
            handlerTaskIDsByToken[token] = ids.isEmpty ? nil : ids
        }
    }

    /// Cancel every in-flight handler task authenticated by `token` — the session's run has ended
    /// (teardown/shutdown unregisters its token), so its suspended calls must unwind now rather
    /// than survive their supervisor. Entries clean themselves up as the cancelled tasks finish.
    private func cancelHandlerTasks(token: String) {
        for id in handlerTaskIDsByToken[token] ?? [] { handlerTasks[id]?.cancel() }
    }

    private struct HTTPRequest: Sendable {
        let method: String
        let path: String
        let headers: [String: String]  // lowercased keys
        let body: Data
    }

    private struct HTTPResponse: Sendable {
        let status: Int
        let statusText: String
        let contentType: String?
        let body: Data
    }

    /// Hard ceiling on a single request's bytes (headers + body). The only clients are our own
    /// in-container shims/tracer over loopback/VM-gateway, but a compromised agent shares the
    /// per-session token, so cap the buffer to bound memory: a request that grows past this drops
    /// the connection rather than letting an attacker stream gigabytes into `Data`. Generous enough
    /// for a full command-event batch (capped at the source too) plus the largest MCP payload.
    static let maxRequestBytes = 4 * 1024 * 1024

    private func nextRequest(on conn: any ByteConn, buffer: inout Data) async -> HTTPRequest? {
        while true {
            if let request = Self.parseRequest(from: &buffer) { return request }
            // Bound memory: an over-long request (or a lying Content-Length) drops the connection.
            if buffer.count > Self.maxRequestBytes { return nil }
            do {
                guard let chunk = try await conn.receive(maxLength: 1 << 16) else { return nil }
                buffer.append(chunk)
            } catch {
                return nil
            }
        }
    }

    private static func parseRequest(from buffer: inout Data) -> HTTPRequest? {
        guard let headerEnd = buffer.range(of: Data("\r\n\r\n".utf8)) else { return nil }
        let headerData = buffer.subdata(in: buffer.startIndex..<headerEnd.lowerBound)
        guard let headerText = String(data: headerData, encoding: .utf8) else { return nil }
        let lines = headerText.components(separatedBy: "\r\n")
        guard let requestLine = lines.first else { return nil }
        let parts = requestLine.split(separator: " ")
        guard parts.count >= 2 else { return nil }

        var headers: [String: String] = [:]
        for line in lines.dropFirst() {
            guard let colon = line.firstIndex(of: ":") else { continue }
            let key = line[..<colon].trimmingCharacters(in: .whitespaces).lowercased()
            let value = line[line.index(after: colon)...].trimmingCharacters(in: .whitespaces)
            headers[key] = value
        }

        let contentLength = headers["content-length"].flatMap(Int.init) ?? 0
        let bodyStart = headerEnd.upperBound
        guard buffer.distance(from: bodyStart, to: buffer.endIndex) >= contentLength else {
            return nil
        }
        let body = buffer.subdata(in: bodyStart..<buffer.index(bodyStart, offsetBy: contentLength))
        buffer.removeSubrange(buffer.startIndex..<buffer.index(bodyStart, offsetBy: contentLength))

        return HTTPRequest(
            method: String(parts[0]), path: String(parts[1]), headers: headers, body: body)
    }

    private func sendResponse(_ response: HTTPResponse, on conn: any ByteConn) async throws {
        var head = "HTTP/1.1 \(response.status) \(response.statusText)\r\n"
        if let contentType = response.contentType {
            head += "Content-Type: \(contentType)\r\n"
        }
        head += "Content-Length: \(response.body.count)\r\nConnection: keep-alive\r\n\r\n"
        var payload = Data(head.utf8)
        payload.append(response.body)
        try await conn.send(payload)
    }

    // MARK: - SSE streaming for long-running tool calls

    /// Response head for an SSE-streamed `tools/call` (see ``respondStreamingToolCall(_:on:)``).
    /// No `Content-Length` — the body streams — and `Connection: close`, since the stream carries a
    /// single JSON-RPC result and ends the connection right after it.
    private static let sseResponseHead = Data(
        ("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n"
            + "Cache-Control: no-cache\r\nConnection: close\r\n\r\n").utf8)
    /// One SSE keepalive frame — a `:`-prefixed comment the client ignores as an MCP message but
    /// which still puts bytes on the wire, resetting the client's body timeout and every relay's
    /// idle clock between here and the CLI.
    private static let sseKeepaliveFrame = Data(": keepalive\n\n".utf8)

    private enum StreamOutcome: Sendable {
        case handled(HTTPResponse)
        case keepaliveEnded
    }

    /// Whether `request` is a gated `tools/call` from an SSE-capable client — the calls that can
    /// block for minutes and so must be answered as a keepalive'd stream rather than a byte-silent
    /// buffered response. Cheap-gates on method + `Accept` before parsing the body, and requires a
    /// registered tool for the token so an unauthorized request falls to the buffered path (which
    /// returns the proper 401) instead of a 200 stream. A body that doesn't parse also falls back —
    /// the buffered path emits the correct `-32700`.
    private func streamableToolCall(from request: HTTPRequest) -> Bool {
        guard request.method == "POST" else { return false }
        guard (request.headers["accept"] ?? "").contains("text/event-stream") else { return false }
        guard hasRegisteredMCPTool(Self.bearerToken(of: request)) else { return false }
        guard let message = try? JSONValue(parsing: request.body) else { return false }
        return message["method"]?.stringValue == "tools/call"
    }

    /// Answer a `tools/call` as an MCP Streamable-HTTP SSE response: flush the head immediately (so
    /// the client's headers timeout is satisfied at once), emit a keepalive comment every
    /// ``sseKeepaliveInterval`` while the handler runs, then deliver the JSON-RPC result as a single
    /// `data:` event and close. Runs inside the caller's registered per-request task; the handler is
    /// awaited through a structured task group so cancellation (token unregister / connection close)
    /// still unwinds a parked spawn, and the keepalive child is cancelled and fully drained before
    /// the result is written so the two never interleave on the socket.
    private func respondStreamingToolCall(_ request: HTTPRequest, on conn: any ByteConn) async {
        try? await conn.send(Self.sseResponseHead)
        try? await conn.send(Self.sseKeepaliveFrame)

        let interval = sseKeepaliveInterval
        let response: HTTPResponse = await withTaskGroup(of: StreamOutcome.self) { group in
            group.addTask { .handled(await self.handle(request)) }
            group.addTask {
                while !Task.isCancelled {
                    do { try await Task.sleep(for: interval) } catch { break }  // cancelled → stop
                    if Task.isCancelled { break }
                    try? await conn.send(Self.sseKeepaliveFrame)
                }
                return .keepaliveEnded
            }
            var result: HTTPResponse?
            for await outcome in group {
                if case .handled(let r) = outcome {
                    result = r
                    group.cancelAll()  // stop keepalives; keep draining so the child fully returns
                }
            }
            // Unreachable in practice (`handle` always returns), but keep the stream well-formed.
            return result
                ?? HTTPResponse(
                    status: 200, statusText: "OK", contentType: "application/json",
                    body: Data("{}".utf8))
        }

        var event = Data("data: ".utf8)
        event.append(response.body)  // compact single-line JSON-RPC — one `data:` line is valid SSE
        event.append(Data("\n\n".utf8))
        try? await conn.send(event)
        conn.close()  // single-response stream: close so the connection/thread frees immediately
    }

    // MARK: - MCP / JSON-RPC dispatch

    /// Whether `token` has any agent-callable MCP tool registered (ignoring the passive git/gh/
    /// command report routes, which aren't tools). Gates the `/mcp` endpoint: Claude registers
    /// `approve`; Codex/Grok register `check_conflict` (and the capability tools) but not `approve`,
    /// so authorizing on `approve` alone would lock them out. Each tool is still individually gated
    /// on its own handler in `tools/list` + `tools/call`, so a token only sees what it registered.
    private func hasRegisteredMCPTool(_ token: String) -> Bool {
        handlers[token] != nil || conflictHandlers[token] != nil || hostExecHandlers[token] != nil
            || macVMExecHandlers[token] != nil || macVMComputerHandlers[token] != nil
            || macVMClearNotificationsHandlers[token] != nil
            || macVMOperatorHandlers[token] != nil || linuxVMExecHandlers[token] != nil
            || linuxVMComputerHandlers[token] != nil || linuxContainerHandlers[token] != nil
            || orchestraSubagentHandlers[token] != nil
            || orchestraPlanHandlers[token] != nil
            || askUserHandlers[token] != nil
            || orchestraSuperviseHandlers[token] != nil
            || orchestraReplyToWorkerHandlers[token] != nil
            || orchestraAskSupervisorHandlers[token] != nil
            || macVMControlHandlers[token] != nil
            || linuxVMControlHandlers[token] != nil
    }

    private func handle(_ request: HTTPRequest) async -> HTTPResponse {
        guard request.method == "POST" else {
            // Streamable-HTTP clients may probe GET (server-push channel) — we
            // don't offer one, 405 is the spec-sanctioned answer.
            return HTTPResponse(
                status: 405, statusText: "Method Not Allowed", contentType: nil, body: Data())
        }

        // The `git` / `gh` interceptor shims post here (not JSON-RPC). Route before MCP dispatch.
        if request.path == Self.gitEventPath {
            return handleGitEvent(request)
        }
        if request.path == Self.ghEventPath {
            return handleGhEvent(request)
        }
        // The non-git command shims + bash tracer post here (not JSON-RPC).
        if request.path == Self.commandEventPath {
            return handleCommandEvent(request)
        }
        // The nash agent shell posts its exec gate here (not JSON-RPC).
        if request.path == Self.shellEventPath {
            return handleShellEvent(request)
        }
        // Codex's PreToolUse hook consults the edit gate here (not JSON-RPC). Awaited, not
        // fire-and-forget: its reply is what allows or blocks the agent's edit.
        if request.path == Self.preToolUsePath {
            return await handlePreToolUse(request)
        }

        let authorization = request.headers["authorization"] ?? ""
        let token = authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
        guard hasRegisteredMCPTool(token) else {
            return HTTPResponse(
                status: 401, statusText: "Unauthorized", contentType: "application/json",
                body: Data(#"{"error":"invalid bearer token"}"#.utf8))
        }

        guard let message = try? JSONValue(parsing: request.body) else {
            return rpcError(id: .null, code: -32700, message: "Parse error")
        }

        let id = message["id"] ?? .null
        let method = message["method"]?.stringValue ?? ""

        // Notifications (no id) are acknowledged with 202 and no body.
        if message["id"] == nil {
            return HTTPResponse(status: 202, statusText: "Accepted", contentType: nil, body: Data())
        }

        switch method {
        case "initialize":
            let requested = message["params"]?["protocolVersion"]?.stringValue ?? "2025-06-18"
            return rpcResult(
                id: id,
                result: .object([
                    "protocolVersion": .string(requested),
                    "capabilities": .object(["tools": .object(["listChanged": .bool(false)])]),
                    "serverInfo": .object([
                        "name": .string("nucleic-approval"), "version": .string("0.1.0"),
                    ]),
                ]))

        case "ping":
            return rpcResult(id: id, result: .object([:]))

        case "tools/list":
            var tools: [JSONValue] = []
            // `approve` — only for tokens whose backend drives approvals through us (Claude). Codex
            // and Grok use their own native approvals and never register this handler, so it stays
            // hidden from them rather than advertising a tool they can't use.
            if handlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.toolName),
                        "description": .string(
                            "Ask the Nucleic user to approve or deny a gated tool call."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "tool_name": .object(["type": .string("string")]),
                                "input": .object(["type": .string("object")]),
                                "tool_use_id": .object(["type": .string("string")]),
                            ]),
                            "required": .array([.string("tool_name"), .string("input")]),
                        ]),
                    ]))
            }
            // `check_conflict` — any backend that registered the conflict handler (Claude, Codex,
            // Grok), so every agent can proactively avoid colliding with another active session.
            if conflictHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.conflictToolName),
                        "description": .string(
                            "Before starting a distinct task that edits files, check whether it "
                                + "conflicts with another currently-active Nucleic agent. Pass a "
                                + "one-line task description and the repo-relative files/areas you "
                                + "expect to change. If the result says the task was deferred or "
                                + "cancelled, STOP and do not edit; if it says proceed (or there is "
                                + "no conflict), continue."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "task": .object(["type": .string("string")]),
                                "files": .object([
                                    "type": .string("array"),
                                    "items": .object(["type": .string("string")]),
                                ]),
                            ]),
                            "required": .array([.string("task")]),
                        ]),
                    ]))
            }
            // Only advertise host_exec when this session opted into host build/run.
            if hostExecHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.hostExecToolName),
                        "description": .string(
                            "Run a command on the HOST machine (macOS), OUTSIDE this Linux sandbox "
                                + "— a LAST-RESORT escape hatch, NOT a general-purpose tool. It exists "
                                + "solely for work that genuinely cannot run inside the container: "
                                + "compiling or running a macOS-only target (a Swift package, a "
                                + "SwiftUI/AppKit app, an xcodebuild/codesign/xcrun step). Before "
                                + "calling it you must have confirmed the task is impossible in the "
                                + "container — which already builds and tests JavaScript, Python, and "
                                + "C/C++ and handles all ordinary shell, file, and search work. Do NOT "
                                + "use it for convenience, for anything that would run in the "
                                + "container, to install software, to fetch from the network, or to "
                                + "run git/gh (host git/gh is blocked and bypasses the shared-trunk "
                                + "version control — run git inside the container). Pass a single shell "
                                + "`command`; it runs in this session's worktree on the host (combine "
                                + "build + run into one command, e.g. \"swift build && swift run "
                                + "MyTool\"). The host shell is bash-compatible — write the command "
                                + "as bash, not zsh (no setopt/zsh-only globs); pass shell:\"zsh\" "
                                + "only if a command genuinely needs zsh. Also pass `reason`: a "
                                + "specific justification for why "
                                + "this cannot run in the container (a container-runnable command is "
                                + "rejected, and an empty/vague reason is rejected). Returns "
                                + "{exit_code, stdout, stderr}. Every call needs explicit user "
                                + "approval and is NEVER auto-approved, so call it only when host "
                                + "execution is genuinely unavoidable."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "command": .object(["type": .string("string")]),
                                "reason": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Why this command cannot run inside the sandbox container. "
                                            + "If the command names an executable that also exists on "
                                            + "Linux, do not just restate the command — the host need "
                                            + "is not self-evident, so explain what specifically makes "
                                            + "THIS invocation require macOS and why the container "
                                            + "cannot do it (e.g. it links an Xcode-only framework, or "
                                            + "it is a codesign/notarization step). Required; empty or "
                                            + "vague reasons are rejected."),
                                ]),
                                "shell": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Optional: \"zsh\" runs this one command under /bin/zsh "
                                            + "instead of the default bash-compatible host shell — "
                                            + "only for a command that genuinely needs zsh syntax."),
                                ]),
                            ]),
                            "required": .array([.string("command"), .string("reason")]),
                        ]),
                    ]))
            }
            // Only advertise mac_vm_exec when this session opted into the macOS VM.
            if macVMExecHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.macVMExecToolName),
                        "description": .string(
                            "Run a command inside this session's isolated macOS VM (Apple "
                                + "Virtualization) — a real, disposable Mac separate from both this Linux "
                                + "sandbox AND the shared host. Use it for macOS-only work that needs a "
                                + "full Mac: `xcodebuild`, `xcrun simctl` (booting/driving iOS/other "
                                + "simulators), running a built .app, `codesign`/notarization, or any "
                                + "true end-to-end test of a Mac/iOS build. PREFER this over `host_exec` "
                                + "for such work: each agent gets its OWN VM with its own Mac "
                                + "toolchain, so parallel Mac builds don't thrash the ONE shared Mac the "
                                + "way host_exec does (host_exec escapes onto that single shared host). Do "
                                + "NOT use it for work the Linux container already does (JavaScript, "
                                + "Python, C/C++, ordinary shell/file/git) — that belongs in the "
                                + "container. The command runs in the VM, already IN the session's repo "
                                + "— its working directory is the repo at its original absolute host "
                                + "path, so do NOT `cd` there yourself and do NOT look for it under "
                                + "\"/Volumes/My Shared Files\" (a legacy automount; globbing that path "
                                + "in zsh aborts the command). The command must be non-interactive and "
                                + "exit on its own: a build/test that never returns (a launched server, a "
                                + "foregrounded app, or a long build piped through `| tail`) is "
                                + "terminated after a timeout and returns its partial output. Pass a "
                                + "single shell `command` (combine build + run, e.g. \"xcodebuild "
                                + "-scheme App test\"), plus `reason`: why this needs the macOS VM "
                                + "specifically. Returns {exit_code, stdout, stderr}. Every call needs "
                                + "explicit user approval and is NEVER auto-approved. "
                                + Self.macVMFreeWhenIdleNotice),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "command": .object(["type": .string("string")]),
                                "reason": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Why this command needs the macOS VM (a Mac-only toolchain: "
                                            + "Xcode, the simulators, codesign) rather than the Linux "
                                            + "container. Required; empty or vague reasons are rejected."),
                                ]),
                            ]),
                            "required": .array([.string("command"), .string("reason")]),
                        ]),
                    ]))
            }
            // Only advertise mac_vm_clear_notifications when this session opted into VM computer-use.
            if macVMClearNotificationsHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.macVMClearNotificationsToolName),
                        "description": .string(
                            "Dismiss the macOS VM's on-screen notifications when a banner or alert is "
                                + "obstructing part of the screen (e.g. it's covering a control you need to "
                                + "click, or hiding what you're trying to read in a `mac_vm_computer` "
                                + "screenshot). Kills the guest's NotificationCenter — the on-screen "
                                + "notifications clear immediately and it relaunches clean. Takes no "
                                + "arguments. Acts only on your own disposable VM, so it needs NO approval. "
                                + "Returns {exit_code, stdout, stderr}. After calling it, take a fresh "
                                + "`mac_vm_computer` screenshot to confirm the view is clear. "
                                + Self.macVMFreeWhenIdleNotice),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([:]),
                        ]),
                    ]))
            }
            // Only advertise mac_vm_computer when this session opted into VM computer-use.
            if macVMComputerHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.macVMComputerToolName),
                        "description": .string(
                            "Drive the GUI of this session's isolated macOS VM like your own Mac — a "
                                + "computer-use loop for a real, disposable Mac. Use it to SEE and "
                                + "operate the desktop when a command isn't enough: visually debug a "
                                + "macOS/iOS app you built (launch it, look at its window, click through "
                                + "it), drive Xcode or the Simulator UI, or verify how something looks. "
                                + "Each call performs ONE `action` and returns a fresh screenshot of the "
                                + "VM screen, so work in a loop: screenshot → decide → act → screenshot. "
                                + "The screenshot is the full screen at its native pixels, so target "
                                + "`x`,`y` exactly as you see them. Pixel actions: `screenshot` (just "
                                + "look); `left_click`/`right_click`/`double_click` at `x`,`y`; "
                                + "`mouse_move` to `x`,`y`; `left_click_drag` to `x`,`y` (from the "
                                + "current cursor); `type` the literal `text`; `key` a chord in `text` "
                                + "(e.g. \"cmd+s\", \"return\", \"tab\", \"cmd+shift+4\"); `scroll` with "
                                + "`scroll_direction`+`scroll_amount`, plus `x`,`y` to aim it at a "
                                + "specific pane (a scroll goes to whatever the pointer is over, so "
                                + "aim it when the window has more than one scrollable area); "
                                + "`launch_app` the app named in "
                                + "`text` (e.g. \"Safari\", \"Xcode\"); `open_browser` a browser named "
                                + "in `text` (\"chrome\", \"firefox\", \"safari\", \"edge\"; default "
                                + "safari), optionally straight to the page in `url` (e.g. "
                                + "\"localhost:3000\" — scheme-less loopback gets http, anything else "
                                + "https) — a predictable shortcut that finds and opens it, then waits "
                                + "for its window; `cursor_position`; `wait` "
                                + "`duration_ms` (let a window open). SEMANTIC actions (preferred when "
                                + "available — they act on controls by identity, not coordinates, and "
                                + "keep working even when screenshots come back blank): `ax_dump` "
                                + "returns the frontmost app's accessibility tree (every control's "
                                + "role/title/value/frame/actions, each with a `ref`); `ax_press` a "
                                + "`ref` (optionally a specific AX action name in `text`); "
                                + "`ax_set_value` writes `value` into a `ref`'d field; `ax_focus` a "
                                + "`ref`; `ax_element_at` resolves the element under `x`,`y`. Convention: "
                                + "ax_dump → act on a ref → ax_dump again (refs go stale after the UI "
                                + "changes). The ax_* actions need the VM's native agent; if the reply "
                                + "says it's unavailable, use screenshots + pixel actions. If screenshots "
                                + "come back blank or unavailable, rely on `ax_dump` — it is "
                                + "framebuffer-independent. This runs in the VM sandbox, so actions are "
                                + "NOT individually approval-gated. Treat this VM as entirely your "
                                + "own sandbox — you have free rein to install tools, sign in to apps, "
                                + "change settings, edit files, or otherwise reconfigure it however "
                                + "gets the job done; it's a disposable clone reset at session end, so "
                                + "don't hold back or ask permission for changes inside it. Prefer "
                                + "`mac_vm_exec` for headless "
                                + "build/test commands; use this when you need to look at or operate the "
                                + "screen. " + Self.macVMFreeWhenIdleNotice),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "action": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "One of: screenshot, left_click, right_click, double_click, "
                                            + "mouse_move, left_click_drag, type, key, scroll, "
                                            + "cursor_position, launch_app, open_browser, wait, "
                                            + "ax_dump, ax_element_at, ax_press, ax_set_value, "
                                            + "ax_focus."),
                                ]),
                                "x": .object(["type": .string("integer")]),
                                "y": .object(["type": .string("integer")]),
                                "text": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Text to type, a key/chord for `key`, an app name for "
                                            + "`launch_app`, a browser name for `open_browser` "
                                            + "(chrome/firefox/safari/edge), or an AX action name for "
                                            + "`ax_press` (default AXPress)."),
                                ]),
                                "url": .object([
                                    "type": .string("string"),
                                    "description": .string(Self.openBrowserURLDescription),
                                ]),
                                "ref": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Element handle from the last ax_dump/ax_element_at (for "
                                            + "ax_press/ax_set_value/ax_focus)."),
                                ]),
                                "value": .object([
                                    "type": .string("string"),
                                    "description": .string("The value to write, for ax_set_value."),
                                ]),
                                "scroll_direction": .object(["type": .string("string")]),
                                "scroll_amount": .object(["type": .string("integer")]),
                                "duration_ms": .object(["type": .string("integer")]),
                            ]),
                            "required": .array([.string("action")]),
                        ]),
                    ]))
            }
            // Batch sibling of mac_vm_computer — advertised on the same opt-in.
            if macVMComputerBatchHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.macVMComputerBatchToolName),
                        "description": .string(
                            "Run an ORDERED SEQUENCE of macOS-VM computer-use actions in ONE call — the "
                                + "batch form of `mac_vm_computer`. Reach for it when you can already see "
                                + "every target in the last screenshot and want to fire the whole plan at "
                                + "once instead of spending a turn per action: e.g. close five windows by "
                                + "clicking each close button, or click → type → press return to fill a "
                                + "field. Pass `steps`: an array where each element has the SAME fields as "
                                + "a single `mac_vm_computer` action (`action` plus its `x`,`y`,`text`,"
                                + "`url`,`ref`,`value`,`scroll_direction`,`scroll_amount`,"
                                + "`duration_ms`), "
                                + "including the semantic `ax_*` verbs. The steps run top-to-bottom against "
                                + "this session's VM, which is booted/reused once. By default the run stops "
                                + "at the first failing step (`stop_on_error`, default true) and only a "
                                + "single screenshot of the FINAL state is returned; set "
                                + "`screenshot_each_step` true to get a screenshot after every step, and "
                                + "`step_delay_ms` to pause between steps so the UI can settle (e.g. a "
                                + "closing window's animation). The reply lists each step's outcome in "
                                + "order. Like `mac_vm_computer` this runs in the VM sandbox and is NOT "
                                + "per-action approval-gated. Prefer the single-action `mac_vm_computer` "
                                + "when you need to SEE the result before deciding the next action; use "
                                + "this when the plan is already known. "
                                + Self.macVMFreeWhenIdleNotice),
                        "inputSchema": Self.computerBatchInputSchema(
                            actionEnumDescription:
                                "One of: screenshot, left_click, right_click, double_click, mouse_move, "
                                + "left_click_drag, type, key, scroll, cursor_position, launch_app, "
                                + "open_browser, wait, ax_dump, ax_element_at, ax_press, ax_set_value, "
                                + "ax_focus.",
                            textDescription:
                                "Text to type, a key/chord for `key`, an app name for `launch_app`, a "
                                + "browser name for `open_browser` (chrome/firefox/safari/edge), or an AX "
                                + "action name for `ax_press` (default AXPress).",
                            includeAX: true),
                    ]))
            }
            // Only advertise mac_vm_request_operator when this session opted into VM computer-use.
            if macVMOperatorHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.macVMOperatorToolName),
                        "description": .string(
                            "Ask the Nucleic USER to take the wheel and operate this session's macOS VM "
                                + "by hand — a LAST-RESORT hand-off for the rare step you cannot do "
                                + "yourself with `mac_vm_computer`, because it genuinely requires a human: "
                                + "solving a CAPTCHA, completing a login / two-factor prompt, accepting a "
                                + "one-time interactive OS dialog. Do NOT use it for ordinary GUI work you "
                                + "can drive yourself (clicking, typing, launching apps) — that is what "
                                + "`mac_vm_computer` is for; use this ONLY when a real person is required. "
                                + "When you call it, Nucleic shows the user a request card, and if they "
                                + "accept, opens an interactive viewer of the VM they can click and type "
                                + "into, with your `instructions` shown alongside. The call BLOCKS until "
                                + "the user finishes or declines. Pass `instructions`: exactly what the "
                                + "user should do in the VM, specific and self-contained (name the window "
                                + "/ page and the concrete action). Pass `reason`: why a human is required "
                                + "rather than you doing it. Returns {completed, note}: `completed` is "
                                + "whether the user says they did it, `note` is any message they left — "
                                + "read both, then continue (re-screenshot the VM to confirm state). Every "
                                + "call needs explicit user consent and is NEVER auto-approved. "
                                + Self.macVMFreeWhenIdleNotice),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "instructions": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "What the user should do in the VM, concretely and "
                                            + "self-contained (e.g. \"solve the CAPTCHA in the Safari "
                                            + "window, then click Continue\"). Required."),
                                ]),
                                "reason": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Why this step needs a human rather than you driving the VM "
                                            + "yourself. Required; empty or vague reasons are rejected."),
                                ]),
                            ]),
                            "required": .array([.string("instructions"), .string("reason")]),
                        ]),
                    ]))
            }
            // Only advertise linux_vm_exec when this session opted into the Linux VM.
            if linuxVMExecHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.linuxVMExecToolName),
                        "description": .string(
                            "Run a command inside this session's isolated Linux VM (Apple "
                                + "Virtualization) — a real, disposable Linux machine with its own kernel, "
                                + "separate from both this Linux container AND any `linux_container` you "
                                + "create. Use it ONLY for Linux dev work a container CANNOT do: things that "
                                + "need a real kernel (custom kernel modules, systemd, nested "
                                + "virtualization, low-level networking) or a full desktop/GUI. For ordinary "
                                + "build/test/scripting, a `linux_container` covers it — the "
                                + "VM is heavier. Prefer this over a macOS VM whenever the choice is open "
                                + "(it is lighter and boots faster). The command runs over the VM's vsock "
                                + "agent; the session's repo is shared into the VM (the tool mounts it and "
                                + "runs there by default). Pass a single shell `command`, plus `reason`: why "
                                + "a container can't do this and it needs the Linux VM. Returns {exit_code, "
                                + "stdout, stderr}. Every call needs explicit user approval and is NEVER "
                                + "auto-approved."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "command": .object(["type": .string("string")]),
                                "reason": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Why this needs the Linux VM (a real kernel, systemd, or a "
                                            + "desktop/GUI) rather than the container. Required; empty or "
                                            + "vague reasons are rejected."),
                                ]),
                            ]),
                            "required": .array([.string("command"), .string("reason")]),
                        ]),
                    ]))
            }
            // Only advertise linux_vm_computer when this session opted into Linux VM computer-use.
            if linuxVMComputerHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.linuxVMComputerToolName),
                        "description": .string(
                            "Drive the GUI of this session's isolated Linux VM like your own Linux "
                                + "desktop — a computer-use loop for a real, disposable machine. Use it to "
                                + "SEE and operate a Linux desktop app or GUI you can't test headlessly. "
                                + "Each call performs ONE `action` and returns a fresh screenshot, so work "
                                + "in a loop: screenshot → decide → act → screenshot. The screenshot is the "
                                + "full screen at native pixels, so target `x`,`y` exactly as you see them. "
                                + "Actions: `screenshot`; `left_click`/`right_click`/`double_click` at "
                                + "`x`,`y`; `mouse_move` to `x`,`y`; `left_click_drag` to `x`,`y`; `type` "
                                + "the literal `text`; `key` a chord in `text` (e.g. \"ctrl+s\", \"return\", "
                                + "\"tab\"); `scroll` with `scroll_direction`+`scroll_amount`, plus "
                                + "`x`,`y` to aim it at a specific pane (a scroll goes to whatever the "
                                + "pointer is over); `launch_app` "
                                + "the binary named in `text`; `open_browser` a browser named in `text` "
                                + "(\"firefox\" or \"chrome\"; default firefox — both Firefox and "
                                + "Chromium ship in the base), optionally straight to the page in `url` (e.g. "
                                + "\"localhost:3000\" — scheme-less loopback gets http, anything else "
                                + "https) — a predictable shortcut that finds and opens it in the desktop "
                                + "session, then waits for its window; `cursor_position`; `wait` "
                                + "`duration_ms`. The "
                                + "semantic `ax_*` actions are macOS-only and unavailable here — use pixel "
                                + "actions. This runs in the VM sandbox, so actions are NOT individually "
                                + "approval-gated. Treat this VM as entirely your own sandbox — you have "
                                + "free rein to install tools, sign in to apps, change settings, edit "
                                + "files, or otherwise reconfigure it however gets the job done; it's a "
                                + "disposable clone reset at session end, so don't hold back or ask "
                                + "permission for changes inside it."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "action": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "One of: screenshot, left_click, right_click, double_click, "
                                            + "mouse_move, left_click_drag, type, key, scroll, "
                                            + "cursor_position, launch_app, open_browser, wait."),
                                ]),
                                "x": .object(["type": .string("integer")]),
                                "y": .object(["type": .string("integer")]),
                                "text": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Text to type, a key/chord for `key`, a binary name for "
                                            + "`launch_app`, or a browser name for `open_browser` "
                                            + "(firefox/chrome)."),
                                ]),
                                "url": .object([
                                    "type": .string("string"),
                                    "description": .string(Self.openBrowserURLDescription),
                                ]),
                                "scroll_direction": .object(["type": .string("string")]),
                                "scroll_amount": .object(["type": .string("integer")]),
                                "duration_ms": .object(["type": .string("integer")]),
                            ]),
                            "required": .array([.string("action")]),
                        ]),
                    ]))
            }
            // Batch sibling of linux_vm_computer — advertised on the same opt-in.
            if linuxVMComputerBatchHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.linuxVMComputerBatchToolName),
                        "description": .string(
                            "Run an ORDERED SEQUENCE of Linux-VM computer-use actions in ONE call — the "
                                + "batch form of `linux_vm_computer`. Reach for it when you can already see "
                                + "every target in the last screenshot and want to fire the whole plan at "
                                + "once instead of spending a turn per action (e.g. close several windows, "
                                + "or click → type → press return to fill a field). Pass `steps`: an array "
                                + "where each element has the SAME fields as a single `linux_vm_computer` "
                                + "action (`action` plus `x`,`y`,`text`,`url`,`scroll_direction`,"
                                + "`scroll_amount`,`duration_ms`). The steps run top-to-bottom against "
                                + "this session's VM, "
                                + "which is booted/reused once. By default the run stops at the first "
                                + "failing step (`stop_on_error`, default true) and only a single "
                                + "screenshot of the FINAL state is returned; set `screenshot_each_step` "
                                + "true for a screenshot after every step, and `step_delay_ms` to pause "
                                + "between steps so the UI can settle. The reply lists each step's outcome "
                                + "in order. Runs in the VM sandbox, NOT per-action approval-gated. The "
                                + "semantic `ax_*` actions are macOS-only and unavailable here — use pixel "
                                + "actions. Prefer the single-action `linux_vm_computer` when you need to "
                                + "SEE the result before deciding the next action; use this when the plan "
                                + "is already known."),
                        "inputSchema": Self.computerBatchInputSchema(
                            actionEnumDescription:
                                "One of: screenshot, left_click, right_click, double_click, mouse_move, "
                                + "left_click_drag, type, key, scroll, cursor_position, launch_app, "
                                + "open_browser, wait.",
                            textDescription:
                                "Text to type, a key/chord for `key`, a binary name for `launch_app`, or "
                                + "a browser name for `open_browser` (firefox/chrome).",
                            includeAX: false),
                    ]))
            }
            // Only advertise linux_container when the agent may create its own containers.
            if linuxContainerHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.linuxContainerToolName),
                        "description": .string(
                            "Spin up and manage your OWN throwaway Linux containers — the LIGHTEST "
                                + "isolation available, far cheaper than any VM. This is where HEAVY "
                                + "processes belong: run a build, a full test suite, a long compile, a "
                                + "big install or a long-running server HERE, not in your session "
                                + "container — your own agent process runs in that one and shares its "
                                + "CPU and memory, so heavy work there can crash the session mid-task. "
                                + "A container you create mounts your worktree at the SAME absolute "
                                + "path and starts in it, so the build sees the files you just edited "
                                + "and its artifacts land in the same tree — nothing to copy in or out, "
                                + "and one container serves every build in the task. Also reach for it "
                                + "for work you want kept separate from your main sandbox: trying a "
                                + "risky install, testing a build in a fresh checkout, or running something "
                                + "you don't want touching your working tree. PREFER a container over a "
                                + "Linux VM for anything that doesn't need a real kernel, systemd, or a GUI "
                                + "— it covers almost all ordinary build/test/scripting. Managed lifecycle "
                                + "via `op`: `create` starts a container (optional `image`, defaults to your "
                                + "sandbox image; optional `name` label) and returns its `name`; `exec` runs "
                                + "a `command` (optional `workdir`) in the named container; `stop` and "
                                + "`remove` tear it down; `list` returns the containers you created. "
                                + "Every operation is limited to containers created by this session, so "
                                + "no separate approval is required — and `exec` runs as root, so this "
                                + "is also how you get unattended root: a `sudo` in your main sandbox "
                                + "is classified as destructive and stops for user approval, while "
                                + "installing packages or editing system files in a container you "
                                + "created does not. Returns {ok, name, names, message} "
                                + "or, for `exec`, "
                                + "{exit_code, stdout, stderr}."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "op": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "One of: create, exec, stop, remove, list."),
                                ]),
                                "name": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "The container to act on (exec/stop/remove), or an optional "
                                            + "label for create."),
                                ]),
                                "image": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "For create: the OCI image; omit for your own sandbox image."),
                                ]),
                                "command": .object([
                                    "type": .string("string"),
                                    "description": .string("For exec: the shell command to run."),
                                ]),
                                "workdir": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "For exec: the working directory inside the container."),
                                ]),
                                "reason": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Optional legacy context for the container operation."),
                                ]),
                            ]),
                            "required": .array([.string("op")]),
                        ]),
                    ]))
            }
            // Advertise nucleic_subagent whenever a subagent handler is registered — for an Orchestra
            // supervisor (non-blocking, driven via nucleic_supervise) or for a Control-project blocking
            // subagent run. The two are told apart by whether a `supervise` handler is registered: only
            // the supervisor has one, so its absence means this is the blocking (Task-drop-in) shape,
            // which gets a matching description.
            if orchestraSubagentHandlers[token] != nil {
                let isSupervisor = orchestraSuperviseHandlers[token] != nil
                let description = isSupervisor
                    ? ("Spawn ONE Nucleic-managed worker subagent for Orchestra — the single-task "
                        + "sibling of `nucleic_delegate_plan` (prefer the plan tool to dispatch a "
                        + "whole decomposition at once; use this for a one-off follow-up worker). "
                        + "Use either instead of native Task/Agent tools: Nucleic creates an "
                        + "ordinary observed session for the worker, applies the user's configured "
                        + "worker model, and routes it through Nucleic Control/sandboxing, tracking "
                        + "its transcript and approvals. NON-BLOCKING: it returns as soon as the "
                        + "worker starts, with a `worker_id` to track it — it does NOT wait "
                        + "for the worker to finish. Pass `task`: a short label, and `prompt`: "
                        + "the complete self-contained instructions the worker should execute; "
                        + "optionally `batch` to group it with siblings you want to wait on "
                        + "together. For independent subtasks, issue multiple calls in one message "
                        + "to fan out (bounded by the user's worker cap). After dispatching, call "
                        + "`nucleic_supervise` to receive worker questions and results, and "
                        + "`nucleic_reply_to_worker` to answer questions. Returns "
                        + "{ok, worker_id, status:\"running\", title, model} or "
                        + "{denied:true, message}.")
                    : ("Spawn a Nucleic-managed worker subagent to carry out a delegated subtask. "
                        + "Use this instead of the native Task/Agent tools (which are disabled): "
                        + "Nucleic runs the worker as an ordinary observed session — shown in the "
                        + "Subagents panel with its own transcript and approvals — using the "
                        + "configured worker model. BLOCKING: the call runs the worker to completion "
                        + "and returns its final report inline, so use it exactly like a Task "
                        + "subagent. Pass `task`: a short label, and `prompt`: the complete, "
                        + "self-contained instructions the worker should execute. For independent "
                        + "subtasks, issue multiple calls in one message to run them in parallel. "
                        + "Returns {ok, worker_id, status:\"completed\", title, model, output} or "
                        + "{denied:true, message}.")
                tools.append(
                    .object([
                        "name": .string(Self.orchestraSubagentToolName),
                        "description": .string(description),
                        // Claude Code executes an MCP tool call concurrently with its siblings only
                        // when the tool advertises `readOnlyHint` (its MCP adapter maps
                        // isConcurrencySafe directly to that annotation; there is no separate
                        // concurrency hint). Without it, several nucleic_subagent calls in one
                        // assistant block run strictly one-at-a-time — serializing the whole
                        // fan-out. Spawning is not literally read-only, but the hint is safe here:
                        // the tool is pre-allowed (--allowedTools), so it never rides the
                        // permission path this hint could otherwise relax, and real concurrency
                        // is bounded app-side by the Settings > Agents worker cap
                        // (`defaultOrchestraMaxConcurrentWorkers`).
                        "annotations": .object(["readOnlyHint": .bool(true)]),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "task": .object([
                                    "type": .string("string"),
                                    "description": .string("Short label for the worker session."),
                                ]),
                                "prompt": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Complete, self-contained instructions for the worker."),
                                ]),
                                "batch": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Optional group name. Workers sharing a batch can be awaited "
                                            + "together with `nucleic_supervise` {batch}, so you act "
                                            + "on the group's results while other batches run."),
                                ]),
                            ]),
                            "required": .array([.string("task"), .string("prompt")]),
                        ]),
                    ]))
            }
            // Plan submission — the supervisor's primary dispatch path. Advertised alongside
            // nucleic_subagent for a supervisor run only (a blocking-subagent run has no plan handler).
            // readOnlyHint for the same reason the spawn tool carries it: it is pre-allowed, and it may
            // be issued beside other calls in one block.
            if orchestraPlanHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.orchestraPlanToolName),
                        "description": .string(
                            "Submit your delegation plan and dispatch it in one call — the primary way "
                                + "to fan work out in Orchestra. Pass `plan` (your plan in prose, shown "
                                + "to the user) and `batches`: an ordered list of {name, purpose, "
                                + "tasks:[{task, prompt}]}. Nucleic spawns a worker session for EVERY "
                                + "task immediately (the user's worker cap queues the excess, so "
                                + "nothing is dropped) and returns {ok, total_spawned, batches:[{batch, "
                                + "spawned, workers:[{worker_id, task}], rejected}]}. NON-BLOCKING. "
                                + "A batch is a group whose results you can act on WITHOUT the rest — "
                                + "typically an explore/research batch that shapes the work you "
                                + "dispatch next. Wait for one with `nucleic_supervise` {batch:\"name\"}: "
                                + "you are woken when that whole group has reported, while the other "
                                + "batches keep running, and you may then submit another plan (or call "
                                + "`nucleic_subagent`) based on what it found. Each `prompt` must be "
                                + "complete and self-contained — the worker sees only that text. "
                                + "Returns {denied:true, message} if the plan carries no usable tasks."),
                        "annotations": .object(["readOnlyHint": .bool(true)]),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "plan": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Your plan for accomplishing the user's goal, in prose: the "
                                            + "approach, how it decomposes, and what each batch is for."),
                                ]),
                                "batches": .object([
                                    "type": .string("array"),
                                    "description": .string(
                                        "Ordered batches of delegated tasks. Use one batch when every "
                                            + "task is independent; use several when an earlier group's "
                                            + "results should shape what you dispatch next."),
                                    "items": .object([
                                        "type": .string("object"),
                                        "properties": .object([
                                            "name": .object([
                                                "type": .string("string"),
                                                "description": .string(
                                                    "Short batch name you'll pass to "
                                                        + "`nucleic_supervise` (e.g. \"explore\")."),
                                            ]),
                                            "purpose": .object([
                                                "type": .string("string"),
                                                "description": .string(
                                                    "What this batch is for (explore, implement, "
                                                        + "verify, …). Shown to the user."),
                                            ]),
                                            "tasks": .object([
                                                "type": .string("array"),
                                                "description": .string(
                                                    "The delegated tasks in this batch."),
                                                "items": .object([
                                                    "type": .string("object"),
                                                    "properties": .object([
                                                        "task": .object([
                                                            "type": .string("string"),
                                                            "description": .string(
                                                                "Short label for the worker session."),
                                                        ]),
                                                        "prompt": .object([
                                                            "type": .string("string"),
                                                            "description": .string(
                                                                "Complete, self-contained instructions "
                                                                    + "for this worker."),
                                                        ]),
                                                    ]),
                                                    "required": .array([
                                                        .string("task"), .string("prompt"),
                                                    ]),
                                                ]),
                                            ]),
                                        ]),
                                        "required": .array([.string("name"), .string("tasks")]),
                                    ]),
                                ]),
                            ]),
                            "required": .array([.string("plan"), .string("batches")]),
                        ]),
                    ]))
            }
            // Ask the human — provider-neutral AskUserQuestion. Advertised whenever the run can reach a
            // user (i.e. not a worker, which must route questions to its supervisor instead).
            if askUserHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.askUserToolName),
                        "description": .string(
                            "Ask the USER a multiple-choice question and BLOCK until they answer. Use "
                                + "it when a decision is genuinely theirs — an ambiguous requirement, a "
                                + "product or scope choice, a tradeoff you shouldn't silently pick — "
                                + "and especially to clarify intent BEFORE committing to a plan. Pass "
                                + "`questions`: 1-4 items of {question, header (≤12 chars), multiSelect, "
                                + "options:[{label, description}]} with 2-4 options each; the user can "
                                + "always add their own answer, so don't include an \"Other\" option. "
                                + "Put your recommended option first and mark it \"(Recommended)\". "
                                + "Returns {ok:true, answers:{\"<question>\": \"<label>\" | [labels]}, "
                                + "answered:[{question, labels}]}, or {ok:false, declined:true} if the "
                                + "user dismissed it — then proceed on your best judgment. Reserve this "
                                + "for choices you cannot settle from the request, the code, or a sane "
                                + "default; don't ask what you can verify yourself."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "questions": .object([
                                    "type": .string("array"),
                                    "description": .string("The questions to ask (1-4)."),
                                    "items": .object([
                                        "type": .string("object"),
                                        "properties": .object([
                                            "question": .object([
                                                "type": .string("string"),
                                                "description": .string(
                                                    "The complete question, ending in a question mark."),
                                            ]),
                                            "header": .object([
                                                "type": .string("string"),
                                                "description": .string(
                                                    "Very short label shown as a chip (max 12 chars)."),
                                            ]),
                                            "multiSelect": .object([
                                                "type": .string("boolean"),
                                                "description": .string(
                                                    "True to let the user pick several options."),
                                            ]),
                                            "options": .object([
                                                "type": .string("array"),
                                                "description": .string(
                                                    "The choices (2-4). Each {label, description}."),
                                                "items": .object([
                                                    "type": .string("object"),
                                                    "properties": .object([
                                                        "label": .object([
                                                            "type": .string("string"),
                                                            "description": .string(
                                                                "Concise choice text (1-5 words)."),
                                                        ]),
                                                        "description": .object([
                                                            "type": .string("string"),
                                                            "description": .string(
                                                                "What choosing this means / its "
                                                                    + "tradeoffs."),
                                                        ]),
                                                    ]),
                                                    "required": .array([.string("label")]),
                                                ]),
                                            ]),
                                        ]),
                                        "required": .array([
                                            .string("question"), .string("options"),
                                        ]),
                                    ]),
                                ]),
                            ]),
                            "required": .array([.string("questions")]),
                        ]),
                    ]))
            }
            // Supervisor drive loop — advertised alongside nucleic_subagent. No readOnlyHint: this is
            // the single serializing wait the supervisor blocks on, not a concurrent sibling call.
            if orchestraSuperviseHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.orchestraSuperviseToolName),
                        "description": .string(
                            "Drive your dispatched Orchestra workers. BLOCKS until there is something "
                                + "to act on, then drains every pending event: {events:[…], "
                                + "remaining_running, batches:[{batch, running, completed, failed, "
                                + "complete}]}. Each event has a `kind`: \"question\" ({worker_id, "
                                + "task, batch, question}) — a worker is paused awaiting your answer, "
                                + "reply with `nucleic_reply_to_worker`; \"completed\" ({worker_id, "
                                + "task, batch, model, output}) — the worker's final result; \"failed\" "
                                + "({worker_id, task, batch, message}). Pass `batch` to wait for one "
                                + "GROUP to finish instead of the next single event: you are woken once "
                                + "that whole batch has reported (also returning "
                                + "{awaited_batch, awaited_batch_complete}) while other batches keep "
                                + "running — so you can act on it, dispatch follow-on work, and come "
                                + "back. A worker question always cuts the wait short, since it blocks "
                                + "that worker. With no argument, returns as soon as any event lands. "
                                + "Loop until `remaining_running` is 0 and every worker has reported."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "batch": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Optional batch name to wait for (from your plan). Omit to "
                                            + "return on the next event from any worker."),
                                ])
                            ]),
                        ]),
                    ]))
            }
            // Supervisor answers a worker's nucleic_ask_supervisor. No readOnlyHint (mutates channel state).
            if orchestraReplyToWorkerHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.orchestraReplyToWorkerToolName),
                        "description": .string(
                            "Answer a worker that is paused in `nucleic_ask_supervisor` (surfaced as a "
                                + "\"question\" event from `nucleic_supervise`). Pass `worker_id` (from "
                                + "the question event) and `reply` (your answer/decision — relay the "
                                + "user's choice here if you escalated via AskUserQuestion). The worker "
                                + "resumes with your reply. Returns {ok:true} or {ok:false, message} if "
                                + "no worker with that id is currently waiting."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "worker_id": .object([
                                    "type": .string("string"),
                                    "description": .string("The waiting worker's id."),
                                ]),
                                "reply": .object([
                                    "type": .string("string"),
                                    "description": .string("Your answer to the worker's question."),
                                ]),
                            ]),
                            "required": .array([.string("worker_id"), .string("reply")]),
                        ]),
                    ]))
            }
            // Worker-only: ask the supervisor. Advertised only for machine-spawned worker sessions.
            if orchestraAskSupervisorHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.orchestraAskSupervisorToolName),
                        "description": .string(
                            "Ask your supervisor for a decision, missing context, or clarification, or "
                                + "flag a blocker or a discovery that may change the plan. BLOCKS until "
                                + "the supervisor replies (they may relay to the human), then returns "
                                + "{ok:true, answer}. Use this instead of guessing on anything that "
                                + "would meaningfully change your output; do NOT use AskUserQuestion — "
                                + "you cannot reach the user directly. If the supervisor's run has "
                                + "ended, returns {ok:false, supervisor_unavailable:true, answer} — "
                                + "proceed autonomously on your best judgment. Pass `question`."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "question": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "What you need from the supervisor."),
                                ]),
                            ]),
                            "required": .array([.string("question")]),
                        ]),
                    ]))
            }
            // Mesh messaging — advertised only when the user enabled agent messaging in Settings.
            if meshSendMessageHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.meshSendMessageToolName),
                        "description": .string(
                            "Send a short app-level message to another Nucleic session or device "
                                + "on this user's mesh (their other Macs/runners), or broadcast it "
                                + "on a topic. Use it to coordinate with agent sessions running "
                                + "elsewhere — hand off findings, signal readiness, request "
                                + "artifacts. Pass exactly one destination: `topic` (delivered to "
                                + "sessions subscribed via nucleic_subscribe_topic), or `to` (a "
                                + "host's name or id) with optional `session` (a session id on "
                                + "that host). `body` is markdown text (max 16 KiB). Delivery is "
                                + "asynchronous and best-effort: the message lands as a user turn "
                                + "in the target session's transcript, attributed to this session. "
                                + "Your identity is attached by Nucleic automatically and cannot "
                                + "be spoofed via arguments."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "to": .object([
                                    "type": .string("string"),
                                    "description": .string("Destination host name or id."),
                                ]),
                                "session": .object([
                                    "type": .string("string"),
                                    "description": .string("Target session id on that host."),
                                ]),
                                "topic": .object([
                                    "type": .string("string"),
                                    "description": .string("Broadcast topic (instead of to/session)."),
                                ]),
                                "body": .object(["type": .string("string")]),
                            ]),
                            "required": .array([.string("body")]),
                        ]),
                    ]))
            }
            if meshSubscribeTopicHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.meshSubscribeTopicToolName),
                        "description": .string(
                            "Subscribe this session to a mesh message topic. Messages other "
                                + "sessions/devices broadcast on that topic (nucleic_send_message "
                                + "with `topic`) are then delivered into this session as user "
                                + "turns. The subscription lasts for this session's lifetime. "
                                + "Pass `topic`."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "topic": .object(["type": .string("string")]),
                            ]),
                            "required": .array([.string("topic")]),
                        ]),
                    ]))
            }
            if meshWaitForMessageHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.meshWaitForMessageToolName),
                        "description": .string(
                            "BLOCK until a mesh message arrives for this session, then return it. "
                                + "Use this to coordinate with agent sessions elsewhere on the mesh "
                                + "— wait for another session to hand off a result or signal "
                                + "readiness before you continue. Pass an optional `topic` (waits "
                                + "for the next message broadcast on it — and subscribes you to it) "
                                + "and an optional `timeout_seconds` (default 300, max 3600). "
                                + "Returns {from, topic, body} on a message, {timedOut:true} if "
                                + "nothing arrived, or {unavailable:true} if the session ended. Do "
                                + "NOT poll in a tight loop; one call blocks efficiently."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "topic": .object(["type": .string("string")]),
                                "timeout_seconds": .object(["type": .string("integer")]),
                            ]),
                        ]),
                    ]))
            }
            // Monitor — the working replacement for the provider-native `Monitor` tool, which
            // Nucleic blocks (its single-shot turn model has no loop to re-invoke the agent when a
            // monitor fires). Advertised whenever the host wired an event sink for this session.
            if monitorHandlers[token] != nil {
                tools.append(
                    .object([
                        "name": .string(Self.monitorToolName),
                        "description": .string(
                            "Start a background monitor that watches a long-running source and "
                                + "re-invokes YOU whenever it produces output — use this instead of "
                                + "the native `Monitor` tool (which does not work here). Arm it and "
                                + "end your turn: you do NOT block and you do NOT need to wait. Each "
                                + "batch of events is delivered to you as a new turn in this chat "
                                + "(labelled with `description`), so you are woken up exactly when "
                                + "the event fires. Pass EXACTLY ONE source: `command` — a shell "
                                + "command/script run in the SAME environment as your Bash tool, "
                                + "where each stdout line is one event and the watch ends when the "
                                + "command exits (use an unbounded command like `tail -f` or a poll "
                                + "loop to get an event per occurrence); OR `ws` — a WebSocket URL "
                                + "to open, where each inbound text frame is one event and the watch "
                                + "ends when the socket closes (no shell, no polling: the server "
                                + "pushes, you get woken). For `command`, make the filter selective "
                                + "— emit only the success AND failure lines you'd act on (e.g. "
                                + "`grep -E --line-buffered 'PASSED|FAILED|Error'`), and every pipe "
                                + "stage must flush per line (`grep --line-buffered`, `awk` "
                                + "`fflush()`; never `| head`); prefer `ws` over "
                                + "`command: 'websocat …'` when a WebSocket feed exists. Pass "
                                + "`persistent: true` for a session-length watch, or `timeout_ms` to "
                                + "auto-stop a bounded one (default 300000, max 3600000). "
                                + "`timeout_ms` is the EARLIEST the watch may auto-stop, not a hard "
                                + "kill: past it, the watch is reaped only once the source also goes "
                                + "idle (no output and near-zero CPU/disk I/O), so a build that "
                                + "simply takes longer than the timeout keeps running to completion. "
                                + "The watch runs outside your turn and survives until the source "
                                + "ends, it is reaped as idle, or the session ends. Returns an armed "
                                + "confirmation with a monitor id."),
                        "inputSchema": .object([
                            "type": .string("object"),
                            "properties": .object([
                                "command": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Shell command/script. Each stdout line is an event; exit "
                                            + "ends the watch. Mutually exclusive with `ws`."),
                                ]),
                                "ws": .object([
                                    "type": .string("object"),
                                    "description": .string(
                                        "WebSocket source (mutually exclusive with `command`): each "
                                            + "text frame is an event; close ends the watch."),
                                    "properties": .object([
                                        "url": .object([
                                            "type": .string("string"),
                                            "description": .string("ws:// or wss:// URL to open."),
                                        ]),
                                        "protocols": .object([
                                            "type": .string("array"),
                                            "items": .object(["type": .string("string")]),
                                            "description": .string(
                                                "Optional subprotocols to offer on the handshake."),
                                        ]),
                                    ]),
                                    "required": .array([.string("url")]),
                                ]),
                                "description": .object([
                                    "type": .string("string"),
                                    "description": .string(
                                        "Short label for what you're watching (shown on every "
                                            + "delivered event)."),
                                ]),
                                "timeout_ms": .object([
                                    "type": .string("integer"),
                                    "description": .string(
                                        "Earliest auto-stop, in ms (default 300000, max 3600000). "
                                            + "After this the watch stops only once the source is "
                                            + "also idle; work still in progress is never cut off. "
                                            + "Ignored when persistent."),
                                ]),
                                "persistent": .object([
                                    "type": .string("boolean"),
                                    "description": .string(
                                        "Run for the whole session (no timeout)."),
                                ]),
                            ]),
                            "required": .array([.string("description")]),
                        ]),
                    ]))
            }
            // Only advertise mac_vm_control when this session can use the macOS VM (exec or computer).
            if macVMControlHandlers[token] != nil {
                tools.append(Self.vmControlToolSchema(
                    name: Self.macVMControlToolName, osLabel: "macOS",
                    sibling: Self.macVMExecToolName, capped: true))
            }
            // Only advertise linux_vm_control when this session can use the Linux VM.
            if linuxVMControlHandlers[token] != nil {
                tools.append(Self.vmControlToolSchema(
                    name: Self.linuxVMControlToolName, osLabel: "Linux",
                    sibling: Self.linuxVMExecToolName, capped: false))
            }
            return rpcResult(id: id, result: .object(["tools": .array(tools)]))

        case "tools/call":
            let params = message["params"]
            let toolName = params?["name"]?.stringValue
            let arguments = params?["arguments"]
            switch toolName {
            case Self.toolName:
                // Only reachable for tokens that registered the approve handler (Claude); a backend
                // with native approvals never advertises this tool, but fail safe if one calls it.
                guard let handler = handlers[token] else {
                    return toolResult(
                        id: id,
                        text: Reply.deny(message: "Approvals are handled natively for this agent.")
                            .wireJSON())
                }
                let call = ApprovalCall(
                    toolName: arguments?["tool_name"]?.stringValue ?? "unknown",
                    input: arguments?["input"] ?? .object([:]),
                    toolUseID: arguments?["tool_use_id"]?.stringValue)
                // This await is the bridge: it suspends the HTTP response until the
                // human answers (ApprovalCoordinator), then returns the mapped reply.
                let reply = await handler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.conflictToolName:
                let files = (arguments?["files"]?.arrayValue ?? []).compactMap { $0.stringValue }
                let call = ConflictCall(
                    task: arguments?["task"]?.stringValue ?? "", files: files)
                // No handler registered (shouldn't happen) → fail open: no conflict.
                guard let conflictHandler = conflictHandlers[token] else {
                    return toolResult(id: id, text: ConflictReply(resolution: .noConflict).wireJSON())
                }
                let reply = await conflictHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.hostExecToolName:
                let call = HostExecCall(
                    command: arguments?["command"]?.stringValue ?? "",
                    reason: arguments?["reason"]?.stringValue ?? "",
                    shell: arguments?["shell"]?.stringValue)
                // No handler (tool shouldn't be visible) → refuse rather than run anything.
                guard let hostExecHandler = hostExecHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: HostExecReply.denied(message: "Host execution is not enabled.")
                            .wireJSON())
                }
                // Suspends until the user approves and the host command finishes (HOST_EXEC) — or
                // until the user taps Skip on the running card, which abandons the wait and hands the
                // agent a "skipped" reply so it stops blocking on a hung command.
                let reply = await runSkippable(
                    name: toolName ?? Self.hostExecToolName, command: call.command,
                    work: { await hostExecHandler(call) },
                    skipped: HostExecReply.denied(message: Self.skippedReplyMessage))
                return toolResult(id: id, text: reply.wireJSON())

            case Self.macVMExecToolName:
                let call = MacVMExecCall(
                    command: arguments?["command"]?.stringValue ?? "",
                    reason: arguments?["reason"]?.stringValue ?? "")
                guard let macVMExecHandler = macVMExecHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMExecReply.denied(message: "The macOS VM is not enabled.")
                            .wireJSON())
                }
                // Suspends until the user approves and the in-VM command finishes — or until the user
                // taps Skip on the running card, abandoning the wait with a "skipped" reply.
                let reply = await runSkippable(
                    name: toolName ?? Self.macVMExecToolName, command: call.command,
                    work: { await macVMExecHandler(call) },
                    skipped: MacVMExecReply.denied(message: Self.skippedReplyMessage))
                return toolResult(id: id, text: reply.wireJSON())

            case Self.macVMClearNotificationsToolName:
                guard let macVMClearNotificationsHandler = macVMClearNotificationsHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMExecReply.denied(message: "The macOS VM is not enabled.")
                            .wireJSON())
                }
                let clearReply = await macVMClearNotificationsHandler(MacVMClearNotificationsCall())
                return toolResult(id: id, text: clearReply.wireJSON())

            case Self.macVMComputerToolName:
                let call = MacVMComputerCall(
                    action: arguments?["action"]?.stringValue ?? "screenshot",
                    x: arguments?["x"]?.intValue,
                    y: arguments?["y"]?.intValue,
                    text: arguments?["text"]?.stringValue,
                    ref: arguments?["ref"]?.stringValue,
                    value: arguments?["value"]?.stringValue,
                    url: arguments?["url"]?.stringValue,
                    scrollDirection: arguments?["scroll_direction"]?.stringValue,
                    scrollAmount: arguments?["scroll_amount"]?.intValue,
                    durationMs: arguments?["duration_ms"]?.intValue)
                guard let macVMComputerHandler = macVMComputerHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMComputerReply.denied(message: "VM computer-use is not enabled.")
                            .deniedJSON())
                }
                // Runs the action in the VM and returns the post-action screenshot as image content.
                switch await macVMComputerHandler(call) {
                case .denied(let message):
                    return toolResult(
                        id: id, text: MacVMComputerReply.denied(message: message).deniedJSON())
                case .ok(let text, let image):
                    if let image {
                        return toolResult(id: id, text: text, imageBase64: image)
                    }
                    return toolResult(id: id, text: text)
                }

            case Self.macVMComputerBatchToolName:
                let call = Self.parseComputerBatchCall(arguments)
                guard let handler = macVMComputerBatchHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMComputerBatchReply.denied(message: "VM computer-use is not enabled.")
                            .deniedJSON())
                }
                return batchToolResult(id: id, reply: await handler(call))

            case Self.macVMOperatorToolName:
                let call = MacVMOperatorCall(
                    reason: arguments?["reason"]?.stringValue ?? "",
                    instructions: arguments?["instructions"]?.stringValue ?? "")
                guard let macVMOperatorHandler = macVMOperatorHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMOperatorReply.denied(message: "The macOS VM is not enabled.")
                            .wireJSON())
                }
                // Suspends until the user finishes operating the VM (or declines).
                let reply = await macVMOperatorHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.linuxVMExecToolName:
                let call = MacVMExecCall(
                    command: arguments?["command"]?.stringValue ?? "",
                    reason: arguments?["reason"]?.stringValue ?? "")
                guard let linuxVMExecHandler = linuxVMExecHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMExecReply.denied(message: "The Linux VM is not enabled.")
                            .wireJSON())
                }
                // Skippable like the other exec tools: Skip abandons a hung in-VM command.
                let reply = await runSkippable(
                    name: toolName ?? Self.linuxVMExecToolName, command: call.command,
                    work: { await linuxVMExecHandler(call) },
                    skipped: MacVMExecReply.denied(message: Self.skippedReplyMessage))
                return toolResult(id: id, text: reply.wireJSON())

            case Self.linuxVMComputerToolName:
                let call = MacVMComputerCall(
                    action: arguments?["action"]?.stringValue ?? "screenshot",
                    x: arguments?["x"]?.intValue,
                    y: arguments?["y"]?.intValue,
                    text: arguments?["text"]?.stringValue,
                    ref: arguments?["ref"]?.stringValue,
                    value: arguments?["value"]?.stringValue,
                    url: arguments?["url"]?.stringValue,
                    scrollDirection: arguments?["scroll_direction"]?.stringValue,
                    scrollAmount: arguments?["scroll_amount"]?.intValue,
                    durationMs: arguments?["duration_ms"]?.intValue)
                guard let linuxVMComputerHandler = linuxVMComputerHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMComputerReply.denied(message: "VM computer-use is not enabled.")
                            .deniedJSON())
                }
                switch await linuxVMComputerHandler(call) {
                case .denied(let message):
                    return toolResult(
                        id: id, text: MacVMComputerReply.denied(message: message).deniedJSON())
                case .ok(let text, let image):
                    if let image {
                        return toolResult(id: id, text: text, imageBase64: image)
                    }
                    return toolResult(id: id, text: text)
                }

            case Self.linuxVMComputerBatchToolName:
                let call = Self.parseComputerBatchCall(arguments)
                guard let handler = linuxVMComputerBatchHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: MacVMComputerBatchReply.denied(message: "VM computer-use is not enabled.")
                            .deniedJSON())
                }
                return batchToolResult(id: id, reply: await handler(call))

            case Self.linuxContainerToolName:
                let call = ContainerToolCall(
                    op: arguments?["op"]?.stringValue ?? "",
                    name: arguments?["name"]?.stringValue,
                    image: arguments?["image"]?.stringValue,
                    command: arguments?["command"]?.stringValue,
                    workdir: arguments?["workdir"]?.stringValue,
                    reason: arguments?["reason"]?.stringValue ?? "")
                guard let linuxContainerHandler = linuxContainerHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: ContainerToolReply.denied(
                            message: "Agent-managed containers are not enabled.").wireJSON())
                }
                let reply = await linuxContainerHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.orchestraSubagentToolName:
                let call = OrchestraSubagentCall(
                    task: arguments?["task"]?.stringValue ?? "",
                    prompt: arguments?["prompt"]?.stringValue ?? "",
                    batch: Self.trimmedNonEmpty(arguments?["batch"]?.stringValue))
                guard let orchestraSubagentHandler = orchestraSubagentHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: OrchestraSubagentResult.denied(
                            message: "Nucleic-managed Orchestra subagents are not enabled.")
                            .wireJSON())
                }
                let reply = await orchestraSubagentHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.orchestraPlanToolName:
                let call = Self.parsePlanCall(arguments)
                guard let planHandler = orchestraPlanHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: OrchestraPlanResult.denied(
                            message: "Orchestra plan delegation is not enabled for this session.")
                            .wireJSON())
                }
                let reply = await planHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.askUserToolName:
                let input = arguments ?? .object([:])
                guard let askUserHandler = askUserHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: AskUserResult.unavailable(
                            message: "Asking the user isn't available in this session; proceed on your "
                                + "best judgment.").wireJSON())
                }
                guard let questions = AskUserQuestion.questions(from: input),
                    questions.contains(where: { !$0.question.isEmpty })
                else {
                    return toolResult(
                        id: id,
                        text: AskUserResult.unavailable(
                            message: "No questions to ask: pass `questions` as a non-empty array of "
                                + "{question, options:[{label}]}.").wireJSON())
                }
                // Suspends until the human answers; cancelled by `unregister` if the run ends first.
                let reply = await askUserHandler(AskUserCall(questions: questions, input: input))
                return toolResult(id: id, text: reply.wireJSON())

            case Self.orchestraSuperviseToolName:
                let awaitedBatch = Self.trimmedNonEmpty(arguments?["batch"]?.stringValue)
                guard let superviseHandler = orchestraSuperviseHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: OrchestraSuperviseResult(events: [], remainingRunning: 0).wireJSON())
                }
                // Suspends (like `approve`) until the supervisor's workers report events — or until the
                // awaited batch finishes; the handler task is cancelled by `unregister` if the
                // supervisor's run ends first.
                let reply = await superviseHandler(awaitedBatch)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.orchestraReplyToWorkerToolName:
                let call = OrchestraReplyToWorkerCall(
                    workerID: arguments?["worker_id"]?.stringValue ?? "",
                    reply: arguments?["reply"]?.stringValue ?? "")
                guard let replyHandler = orchestraReplyToWorkerHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: OrchestraReplyResult.unknown(
                            message: "Orchestra supervision is not enabled for this session.")
                            .wireJSON())
                }
                let reply = await replyHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.orchestraAskSupervisorToolName:
                let call = OrchestraAskSupervisorCall(
                    question: arguments?["question"]?.stringValue ?? "")
                guard let askHandler = orchestraAskSupervisorHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: OrchestraAskSupervisorResult.unavailable(
                            message: "No supervisor is available; proceed autonomously.")
                            .wireJSON())
                }
                // Suspends until the supervisor replies via `nucleic_reply_to_worker`, or this
                // worker's own run ends (its token is unregistered → this handler task cancelled →
                // the AppStore side resumes it with an `.unavailable` answer).
                let reply = await askHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.meshSendMessageToolName:
                guard let sendHandler = meshSendMessageHandlers[token] else {
                    return toolResult(
                        id: id, text: "Mesh messaging is not enabled on this Mac.")
                }
                let request = MeshSendRequest(
                    to: arguments?["to"]?.stringValue,
                    sessionID: arguments?["session"]?.stringValue,
                    topic: arguments?["topic"]?.stringValue,
                    body: arguments?["body"]?.stringValue ?? "")
                return toolResult(id: id, text: await sendHandler(request))

            case Self.meshSubscribeTopicToolName:
                guard let subscribeHandler = meshSubscribeTopicHandlers[token] else {
                    return toolResult(
                        id: id, text: "Mesh messaging is not enabled on this Mac.")
                }
                let topic = arguments?["topic"]?.stringValue ?? ""
                return toolResult(id: id, text: await subscribeHandler(topic))

            case Self.meshWaitForMessageToolName:
                guard let waitHandler = meshWaitForMessageHandlers[token] else {
                    return toolResult(id: id, text: "Mesh messaging is not enabled on this Mac.")
                }
                let topic = arguments?["topic"]?.stringValue
                let timeout = arguments?["timeout_seconds"]?.intValue
                return toolResult(id: id, text: await waitHandler(topic, timeout))

            case Self.monitorToolName:
                guard let monitorHandler = monitorHandlers[token] else {
                    return toolResult(id: id, text: "Monitoring is not available in this session.")
                }
                let ws = arguments?["ws"]
                let call = MonitorCall(
                    command: arguments?["command"]?.stringValue ?? "",
                    wsURL: ws?["url"]?.stringValue,
                    wsProtocols: ws?["protocols"]?.arrayValue?.compactMap(\.stringValue) ?? [],
                    description: arguments?["description"]?.stringValue ?? "",
                    timeoutMs: arguments?["timeout_ms"]?.intValue,
                    persistent: arguments?["persistent"]?.boolValue ?? false)
                return toolResult(id: id, text: await monitorHandler(call))

            case Self.macVMControlToolName:
                let call = VMControlCall(op: arguments?["op"]?.stringValue ?? "")
                guard let macVMControlHandler = macVMControlHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: VMControlReply.denied(message: "The macOS VM is not enabled.")
                            .wireJSON())
                }
                let reply = await macVMControlHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            case Self.linuxVMControlToolName:
                let call = VMControlCall(op: arguments?["op"]?.stringValue ?? "")
                guard let linuxVMControlHandler = linuxVMControlHandlers[token] else {
                    return toolResult(
                        id: id,
                        text: VMControlReply.denied(message: "The Linux VM is not enabled.")
                            .wireJSON())
                }
                let reply = await linuxVMControlHandler(call)
                return toolResult(id: id, text: reply.wireJSON())

            default:
                return rpcError(id: id, code: -32602, message: "Unknown tool")
            }

        default:
            return rpcError(id: id, code: -32601, message: "Method not found: \(method)")
        }
    }

    /// Handle a `POST /git-event` from the in-container `git` interceptor shim: bearer-token
    /// gated like the MCP route, but a plain JSON body `{argv, cwd, exitCode, subcommand,
    /// sessionId}`. Fire-and-forget — the registered handler returns immediately (it offloads
    /// any real work), so a report never blocks the agent's git call. Always 202s a known
    /// token (even on a malformed body) so the shim never retries or stalls; 401s an unknown one.
    private func handleGitEvent(_ request: HTTPRequest) -> HTTPResponse {
        let authorization = request.headers["authorization"] ?? ""
        let token = authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
        guard let handler = gitReportHandlers[token] else {
            return HTTPResponse(
                status: 401, statusText: "Unauthorized", contentType: "application/json",
                body: Data(#"{"error":"invalid bearer token"}"#.utf8))
        }
        if let message = try? JSONValue(parsing: request.body) {
            let argv = (message["argv"]?.arrayValue ?? []).compactMap { $0.stringValue }
            handler(GitReportCall(
                argv: argv,
                cwd: message["cwd"]?.stringValue ?? "",
                exitCode: message["exitCode"]?.intValue ?? 0,
                subcommand: message["subcommand"]?.stringValue,
                sessionID: message["sessionId"]?.stringValue))
        }
        return HTTPResponse(status: 202, statusText: "Accepted", contentType: nil, body: Data())
    }

    /// Handle a `POST /gh-event` from the in-container `gh` interceptor shim — the exact twin of
    /// `handleGitEvent`: bearer-token gated, plain JSON `{argv, cwd, exitCode, subcommand,
    /// sessionId}`, fire-and-forget (202 on any known-token body, 401 on an unknown token).
    private func handleGhEvent(_ request: HTTPRequest) -> HTTPResponse {
        let authorization = request.headers["authorization"] ?? ""
        let token = authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
        guard let handler = ghReportHandlers[token] else {
            return HTTPResponse(
                status: 401, statusText: "Unauthorized", contentType: "application/json",
                body: Data(#"{"error":"invalid bearer token"}"#.utf8))
        }
        if let message = try? JSONValue(parsing: request.body) {
            let argv = (message["argv"]?.arrayValue ?? []).compactMap { $0.stringValue }
            handler(GhReportCall(
                argv: argv,
                cwd: message["cwd"]?.stringValue ?? "",
                exitCode: message["exitCode"]?.intValue ?? 0,
                subcommand: message["subcommand"]?.stringValue,
                sessionID: message["sessionId"]?.stringValue))
        }
        return HTTPResponse(status: 202, statusText: "Accepted", contentType: nil, body: Data())
    }

    /// Handle a `POST /command-event` from the non-git command interceptor: bearer-token gated
    /// like `/git-event`. Two body shapes, both fanned out to one `CommandReportCall` per command:
    ///
    ///   * shim — `{type:"command", command, argv, cwd, exitCode, durationMs, stdout, stderr,
    ///     truncated, sessionId}` — one rich, output-carrying report.
    ///   * tracer — `{type:"command-batch", sessionId, events:[{line, cwd, exitCode, durationMs}]}`
    ///     — a batch of metadata-only reports for a whole Bash tool call.
    ///
    /// Fire-and-forget: handlers return immediately. Always 202s a known token (even on a malformed
    /// body) so the shim/tracer never retry or stall; 401s an unknown one.
    private func handleCommandEvent(_ request: HTTPRequest) -> HTTPResponse {
        let authorization = request.headers["authorization"] ?? ""
        let token = authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
        guard let handler = commandReportHandlers[token] else {
            return HTTPResponse(
                status: 401, statusText: "Unauthorized", contentType: "application/json",
                body: Data(#"{"error":"invalid bearer token"}"#.utf8))
        }
        if let message = try? JSONValue(parsing: request.body) {
            for call in Self.parseCommandReports(message) { handler(call) }
        }
        return HTTPResponse(status: 202, statusText: "Accepted", contentType: nil, body: Data())
    }

    /// Decode a `/command-event` body into zero or more `CommandReportCall`s. A `command-batch`
    /// (tracer) yields one per `events[]` entry; anything else is treated as a single shim report.
    static func parseCommandReports(_ message: JSONValue) -> [CommandReportCall] {
        let sessionID = message["sessionId"]?.stringValue
        if message["type"]?.stringValue == "command-batch" {
            return (message["events"]?.arrayValue ?? []).map { event in
                CommandReportCall(
                    command: nil, argv: [], commandLine: event["line"]?.stringValue ?? "",
                    cwd: event["cwd"]?.stringValue ?? "",
                    exitCode: event["exitCode"]?.intValue ?? 0,
                    durationMs: event["durationMs"]?.intValue,
                    stdout: "", stderr: "", truncated: false, source: .tracer, sessionID: sessionID)
            }
        }
        let argv = (message["argv"]?.arrayValue ?? []).compactMap { $0.stringValue }
        return [CommandReportCall(
            command: message["command"]?.stringValue, argv: argv,
            commandLine: message["commandLine"]?.stringValue,
            cwd: message["cwd"]?.stringValue ?? "",
            exitCode: message["exitCode"]?.intValue ?? 0,
            durationMs: message["durationMs"]?.intValue,
            stdout: message["stdout"]?.stringValue ?? "",
            stderr: message["stderr"]?.stringValue ?? "",
            truncated: message["truncated"]?.boolValue ?? false,
            source: .shim, sessionID: sessionID)]
    }

    /// Handle a `POST /shell-event` from the nash agent shell (docs/NASH.md §6.1): bearer-token
    /// gated like `/git-event`. Body is a `shell-batch` (`{type, source:"nash", sessionId, shellId,
    /// events:[…]}`), decoded to one ``ShellReportCall`` per event and delivered to the handler as
    /// **one batch** — the ingest cost is per delivery, not per event, so a 200-event body must not
    /// become 200 hops (docs/NASH_STREAM_PERF_PLAN.md §P1). Fire-and-forget: the handler returns
    /// immediately. Always 202s a known token (even on a malformed body) so nash never retries or
    /// stalls; 401s an unknown one (which nash treats as "spool instead").
    private func handleShellEvent(_ request: HTTPRequest) -> HTTPResponse {
        let authorization = request.headers["authorization"] ?? ""
        let token = authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
        guard let handler = shellReportHandlers[token] else {
            return HTTPResponse(
                status: 401, statusText: "Unauthorized", contentType: "application/json",
                body: Data(#"{"error":"invalid bearer token"}"#.utf8))
        }
        if let message = try? JSONValue(parsing: request.body) {
            let calls = Self.parseShellBatch(message)
            if !calls.isEmpty { handler(calls) }
        }
        return HTTPResponse(status: 202, statusText: "Accepted", contentType: nil, body: Data())
    }

    /// Handle a `POST /pre-tool-use` from codex's `PreToolUse` hook (docs/LOCKING.md §4.6): the
    /// Codex edit gate. Bearer-token gated like `/git-event`, but **awaited** — codex is blocked on
    /// this response, and the registered handler acquires the call's file locks before replying, so
    /// a contended file legitimately parks the request for as long as the holder needs.
    ///
    /// A body that doesn't parse is *allowed* (`decision: allow`): there is no path to lock, so the
    /// only alternative would be blocking a tool call over our own parse failure. An unknown token
    /// still 401s, which the hook treats as fail-closed — the run is either over or was never wired,
    /// and letting an unauthenticated codex edit unlocked is the outcome this whole route exists to
    /// prevent.
    private func handlePreToolUse(_ request: HTTPRequest) async -> HTTPResponse {
        let token = Self.bearerToken(of: request)
        guard let handler = preToolUseHandlers[token] else {
            return HTTPResponse(
                status: 401, statusText: "Unauthorized", contentType: "application/json",
                body: Data(#"{"error":"invalid bearer token"}"#.utf8))
        }
        func reply(_ decision: PreToolUseDecision) -> HTTPResponse {
            HTTPResponse(
                status: 200, statusText: "OK", contentType: "application/json",
                body: Data(decision.wireJSON().utf8))
        }
        guard let message = try? JSONValue(parsing: request.body),
            let toolName = message["tool_name"]?.stringValue
        else {
            return reply(.allow)
        }
        let decision = await handler(
            PreToolUseCall(
                toolName: toolName,
                input: message["tool_input"] ?? .object([:]),
                toolUseID: message["tool_use_id"]?.stringValue,
                cwd: message["cwd"]?.stringValue))
        return reply(decision)
    }

    /// Most stages any one pipeline is credited with. Real pipelines are a handful of stages; this
    /// exists only so a corrupt or hostile batch can't hand the viewer a stage count it will try to
    /// build a row for.
    static let maxPipelineStages = 256

    /// A pipe link's stage index, forced into `0...maxPipelineStages`.
    static func clampStageIndex(_ raw: Int?) -> Int {
        min(max(raw ?? 0, 0), maxPipelineStages)
    }

    /// The pipeline slot on an `exec` / `exec-start` event, or nil when the command didn't run as
    /// a stage of one — absent for a standalone command, which is the overwhelming majority, so a
    /// missing object means "not in a pipeline", not a malformed event.
    ///
    /// A stage index must fall inside its own pipeline. These bound loops and arithmetic in the
    /// viewer, so a nonsensical slot is dropped here rather than trusted downstream — a negative
    /// `len` reached a SwiftUI range and trapped. Dropping the slot costs only the stage's
    /// pipeline attribution.
    static func pipelineSlot(_ slot: JSONValue?) -> NashPipelineSlot? {
        guard let slot, let id = slot["id"]?.intValue, let index = slot["index"]?.intValue,
            let length = slot["len"]?.intValue,
            length > 0, length <= maxPipelineStages, index >= 0, index < length
        else { return nil }
        return NashPipelineSlot(id: id, index: index, length: length)
    }

    /// Decode a nash `shell-batch` body into zero or more ``ShellReportCall``s — one per `events[]`
    /// entry, skipping any event whose `kind` is unknown or whose required fields are absent.
    static func parseShellBatch(_ message: JSONValue) -> [ShellReportCall] {
        guard message["type"]?.stringValue == "shell-batch" else { return [] }
        let sessionID = message["sessionId"]?.stringValue
        let environment = NashEnvironment.fromWire(
            kind: message["environmentKind"]?.stringValue,
            id: message["environmentId"]?.stringValue,
            label: message["environmentLabel"]?.stringValue)
        let shellID = message["shellId"]?.stringValue
        return (message["events"]?.arrayValue ?? []).compactMap { event in
            let kind: ShellReportCall.Kind
            switch event["kind"]?.stringValue {
            case "exec":
                kind = .exec(
                    argv: (event["argv"]?.arrayValue ?? []).compactMap { $0.stringValue },
                    cwd: event["cwd"]?.stringValue ?? "",
                    exitCode: event["exitCode"]?.intValue ?? 0,
                    durationMs: event["durationMs"]?.intValue,
                    pipeline: pipelineSlot(event["pipeline"]))
            case "exec-start":
                // The in-flight half of an exec (docs/NASH.md §5.1). A command with no argv says
                // nothing a viewer could render as a running row, and the `exec` that follows will
                // report it anyway, so drop it rather than spin an empty row.
                let argv = (event["argv"]?.arrayValue ?? []).compactMap { $0.stringValue }
                guard !argv.isEmpty else { return nil }
                kind = .execStart(
                    argv: argv, cwd: event["cwd"]?.stringValue ?? "",
                    // Back-dates the row's clock, so it is clamped to a day: a garbage value would
                    // otherwise stamp a live command with a timestamp from an arbitrary past.
                    elapsedMs: min(max(event["elapsedMs"]?.intValue ?? 0, 0), 86_400_000),
                    pipeline: pipelineSlot(event["pipeline"]))
            case "cd":
                kind = .cd(
                    from: event["from"]?.stringValue ?? "",
                    to: event["to"]?.stringValue ?? "")
            case "export":
                kind = .export(
                    op: event["op"]?.stringValue ?? "export",
                    names: (event["names"]?.arrayValue ?? []).compactMap { $0.stringValue })
            case "redirect":
                kind = .redirect(
                    op: event["op"]?.stringValue ?? "",
                    fd: event["fd"]?.intValue ?? 1,
                    target: event["target"]?.stringValue,
                    bytes: event["bytes"]?.intValue ?? 0,
                    truncated: event["truncated"]?.boolValue ?? false,
                    hash: event["hash"]?.stringValue ?? "",
                    previewB64: event["previewB64"]?.stringValue ?? "")
            case "cmdsub":
                kind = .cmdsub(
                    bytes: event["bytes"]?.intValue ?? 0,
                    truncated: event["truncated"]?.boolValue ?? false,
                    hash: event["hash"]?.stringValue ?? "",
                    previewB64: event["previewB64"]?.stringValue ?? "")
            case "pipe":
                kind = .pipe(
                    pipelineID: event["pipelineId"]?.intValue ?? 0,
                    // Clamped for the same reason as the exec slot: these indices size a view's
                    // stage list and get `+ 1`'d for display, so unbounded wire values are an
                    // out-of-memory and an overflow waiting to happen.
                    fromIndex: Self.clampStageIndex(event["fromIndex"]?.intValue),
                    toIndex: Self.clampStageIndex(event["toIndex"]?.intValue),
                    fromText: event["fromText"]?.stringValue ?? "",
                    toText: event["toText"]?.stringValue ?? "",
                    bytes: event["bytes"]?.intValue ?? 0,
                    truncated: event["truncated"]?.boolValue ?? false,
                    hash: event["hash"]?.stringValue ?? "",
                    previewB64: event["previewB64"]?.stringValue ?? "")
            case "fallback":
                kind = .fallback(
                    reason: event["reason"]?.stringValue ?? "",
                    input: event["input"]?.stringValue ?? "")
            case "policy":
                // A lever with no name says nothing a viewer could act on; drop it rather than
                // render an empty row.
                guard let lever = event["lever"]?.stringValue, !lever.isEmpty else { return nil }
                kind = .policy(
                    lever: lever,
                    value: event["value"]?.stringValue ?? "",
                    action: event["action"]?.stringValue ?? "ignored",
                    note: event["note"]?.stringValue ?? "")
            case "dropped":
                kind = .dropped(count: event["count"]?.intValue ?? 0)
            default:
                return nil
            }
            return ShellReportCall(
                kind: kind, sessionID: sessionID, environment: environment, shellID: shellID,
                execToken: event["token"]?.intValue)
        }
    }

    /// A successful `tools/call` result wrapping `text` in a single MCP text content block.
    // MARK: - Per-call Skip (abandon a stuck host/VM exec)

    /// The fully-qualified names of the tools whose calls carry a **Skip** affordance — the
    /// long-running host/VM exec tools, where a hung command (a stuck build, a wedged SSH) is the
    /// thing the user needs the agent to stop waiting on. The transcript shows Skip only for these,
    /// and only ``runSkippable(name:command:work:skipped:)`` wraps their dispatch, so the UI and the
    /// mechanism agree on exactly which calls are skippable.
    public static let skippableToolNames: Set<String> = [
        qualifiedHostExecToolName, qualifiedMacVMExecToolName, qualifiedLinuxVMExecToolName,
    ]

    /// Whether `toolName` (fully-qualified, as it appears on a transcript ``ToolCall``) is a skippable
    /// host/VM exec call. Used by the UI to decide whether to offer the Skip button.
    public static func isSkippable(toolName: String) -> Bool { skippableToolNames.contains(toolName) }

    /// The message the agent reads back when the user skips a call: it explains the call was abandoned
    /// as apparently stuck (not that it failed on its own or was policy-denied), notes the process may
    /// still be running, and steers the agent away from blindly re-running the same command.
    static let skippedReplyMessage =
        "The user skipped this call because it appeared to be stuck. The command may still be running "
        + "in the background; its result was abandoned. Do not automatically retry the same command — "
        + "reconsider the approach, or check the VM/host state, before trying anything similar again."

    /// Strip the `mcp__nucleic__` prefix so a fully-qualified tool name (what the transcript
    /// ``ToolCall`` carries, e.g. `mcp__nucleic__mac_vm_exec`) and the bare `tools/call` name the
    /// dispatch sees (e.g. `mac_vm_exec`) map to the SAME fingerprint.
    static func normalizedSkipName(_ name: String) -> String {
        let prefix = "mcp__nucleic__"
        return name.hasPrefix(prefix) ? String(name.dropFirst(prefix.count)) : name
    }

    /// The fingerprint keying a call for Skip: its normalized tool name + its exact command. Both
    /// sides (the dispatch that runs the call and the UI-driven ``skipToolCall``) derive it the same
    /// way, so a Skip tap finds the right in-flight handler.
    private func skipFingerprint(name: String, command: String) -> String {
        Self.normalizedSkipName(name) + "\u{1}" + command
    }

    /// Fire the on-skip closures for every in-flight call matching this tool name + command, causing
    /// each to return its synthetic "skipped" reply. A no-op if nothing matches (the call already
    /// finished, or never registered) — Skip is always safe to send. Called from the app's control
    /// path (``ClaudeCodeBackend/skipToolCall(_:)``) when the user taps Skip on a running exec card.
    public func skipToolCall(name: String, command: String) {
        let fingerprint = skipFingerprint(name: name, command: command)
        // Snapshot the closures: each `onSkip` unregisters its own entry, mutating `skipHandlers`.
        for (_, onSkip) in skipHandlers[fingerprint] ?? [:] { onSkip() }
    }

    private func registerSkip(fingerprint: String, id: UUID, _ onSkip: @escaping @Sendable () -> Void) {
        skipHandlers[fingerprint, default: [:]][id] = onSkip
    }

    private func unregisterSkip(fingerprint: String, id: UUID) {
        skipHandlers[fingerprint]?[id] = nil
        if skipHandlers[fingerprint]?.isEmpty == true { skipHandlers[fingerprint] = nil }
    }

    /// Run a skippable exec handler, returning its reply — but if the user taps **Skip** on this
    /// call while it's in flight, return `skipped` instead and let the real work orphan. This is the
    /// choke point that makes a stuck host/VM command abandonable: the handler (`work`) runs in an
    /// unstructured task so we can hand back the synthetic reply the moment Skip fires, rather than
    /// being pinned to the `withTaskGroup` barrier that would wait for the hung command to finish.
    /// Whichever of {work completes, Skip fires} happens first wins; the loser's later resume is a
    /// no-op (``ResumeOnce``). The abandoned `work` keeps running in its disposable VM/host until its
    /// command finally exits, then quietly discards its reply.
    private func runSkippable<R: Sendable>(
        name: String,
        command: String,
        work: @escaping @Sendable () async -> R,
        skipped: R
    ) async -> R {
        let fingerprint = skipFingerprint(name: name, command: command)
        let id = UUID()
        return await withCheckedContinuation { (continuation: CheckedContinuation<R, Never>) in
            let box = ResumeOnce(continuation)
            // Register the Skip closure BEFORE any suspension point, so a Skip that races in can't slip
            // between dispatch and registration (both run on this actor; skipToolCall can only interleave
            // at an await). It resumes with the synthetic reply and tears its own entry down.
            registerSkip(fingerprint: fingerprint, id: id) { [weak self] in
                box.resume(skipped)
                Task { await self?.unregisterSkip(fingerprint: fingerprint, id: id) }
            }
            // The real handler, unstructured so it can outlive an early Skip return.
            Task { [weak self] in
                let reply = await work()
                box.resume(reply)
                await self?.unregisterSkip(fingerprint: fingerprint, id: id)
            }
        }
    }

    /// The `url` field's schema description — identical on macOS and Linux, single-action and batch, so
    /// the agent reads the same contract wherever it meets `open_browser`.
    static let openBrowserURLDescription =
        "For `open_browser`: the page to start on (e.g. \"https://example.com\", \"localhost:3000\", a "
        + "file:// path). Opens it directly instead of making you drive the address bar; if the browser "
        + "is already running it lands in a new tab. A scheme-less URL gets http:// for localhost and "
        + "https:// otherwise. Omit to open the browser's own start page."

    /// The JSON Schema for a `*_computer_batch` tool: a `steps` array whose element schema mirrors the
    /// single-action `*_computer` tool, plus the batch-run options. `actionEnumDescription`/
    /// `textDescription` differ between macOS and Linux; `includeAX` adds the semantic-action fields
    /// (`ref`, `value`) that only macOS supports.
    private static func computerBatchInputSchema(
        actionEnumDescription: String, textDescription: String, includeAX: Bool
    ) -> JSONValue {
        var stepProps: [String: JSONValue] = [
            "action": .object([
                "type": .string("string"),
                "description": .string(actionEnumDescription),
            ]),
            "x": .object(["type": .string("integer")]),
            "y": .object(["type": .string("integer")]),
            "text": .object([
                "type": .string("string"),
                "description": .string(textDescription),
            ]),
            "url": .object([
                "type": .string("string"),
                "description": .string(openBrowserURLDescription),
            ]),
            "scroll_direction": .object(["type": .string("string")]),
            "scroll_amount": .object(["type": .string("integer")]),
            "duration_ms": .object(["type": .string("integer")]),
        ]
        if includeAX {
            stepProps["ref"] = .object([
                "type": .string("string"),
                "description": .string(
                    "Element handle from the last ax_dump/ax_element_at (for "
                        + "ax_press/ax_set_value/ax_focus)."),
            ])
            stepProps["value"] = .object([
                "type": .string("string"),
                "description": .string("The value to write, for ax_set_value."),
            ])
        }
        return .object([
            "type": .string("object"),
            "properties": .object([
                "steps": .object([
                    "type": .string("array"),
                    "description": .string(
                        "The actions to run, in order. Each element is a single computer-use action "
                            + "with the same fields as the non-batch tool."),
                    "items": .object([
                        "type": .string("object"),
                        "properties": .object(stepProps),
                        "required": .array([.string("action")]),
                    ]),
                ]),
                "stop_on_error": .object([
                    "type": .string("boolean"),
                    "description": .string(
                        "Stop at the first failing step instead of continuing (default true)."),
                ]),
                "screenshot_each_step": .object([
                    "type": .string("boolean"),
                    "description": .string(
                        "Return a screenshot after every step, not just the final one (default "
                            + "false)."),
                ]),
                "step_delay_ms": .object([
                    "type": .string("integer"),
                    "description": .string(
                        "Milliseconds to pause after each step so the UI can settle (optional)."),
                ]),
            ]),
            "required": .array([.string("steps")]),
        ])
    }

    /// Parse a `*_computer_batch` call's arguments into a ``MacVMComputerBatchCall`` — each element of
    /// `steps` becomes a ``MacVMComputerCall`` with the same field mapping the single-action tool uses
    /// (a missing `action` defaults to `screenshot`, matching the single-action parse).
    private static func parseComputerBatchCall(
        _ arguments: JSONValue?
    ) -> MacVMComputerBatchCall {
        let steps = (arguments?["steps"]?.arrayValue ?? []).map { element -> MacVMComputerCall in
            MacVMComputerCall(
                action: element["action"]?.stringValue ?? "screenshot",
                x: element["x"]?.intValue,
                y: element["y"]?.intValue,
                text: element["text"]?.stringValue,
                ref: element["ref"]?.stringValue,
                value: element["value"]?.stringValue,
                url: element["url"]?.stringValue,
                scrollDirection: element["scroll_direction"]?.stringValue,
                scrollAmount: element["scroll_amount"]?.intValue,
                durationMs: element["duration_ms"]?.intValue)
        }
        return MacVMComputerBatchCall(
            steps: steps,
            stopOnError: arguments?["stop_on_error"]?.boolValue ?? true,
            screenshotEachStep: arguments?["screenshot_each_step"]?.boolValue ?? false,
            stepDelayMs: arguments?["step_delay_ms"]?.intValue)
    }

    /// A trimmed string argument, or nil when absent/blank — so `{"batch": ""}` (or a lone space) reads
    /// as "no batch" rather than creating a batch literally named the empty string.
    static func trimmedNonEmpty(_ raw: String?) -> String? {
        guard let trimmed = raw?.trimmingCharacters(in: .whitespacesAndNewlines), !trimmed.isEmpty
        else { return nil }
        return trimmed
    }

    /// Parse a `nucleic_delegate_plan` call into an ``OrchestraPlanCall``. Tolerant by design — a model
    /// that omits a batch `name` still gets its tasks dispatched (the batch is named positionally,
    /// `batch-1`, `batch-2`, …) rather than having the plan rejected — while tasks with no usable
    /// `prompt` are dropped here, since a worker cannot be given nothing to do. An empty `batches` list
    /// yields no batches, which the handler reports as `denied`.
    static func parsePlanCall(_ arguments: JSONValue?) -> OrchestraPlanCall {
        let rawBatches = arguments?["batches"]?.arrayValue ?? []
        var batches: [OrchestraPlanBatch] = []
        for (index, raw) in rawBatches.enumerated() {
            let tasks = (raw["tasks"]?.arrayValue ?? []).compactMap { element -> OrchestraPlanTask? in
                guard let prompt = trimmedNonEmpty(element["prompt"]?.stringValue) else { return nil }
                let label = trimmedNonEmpty(element["task"]?.stringValue) ?? "Delegated task"
                return OrchestraPlanTask(task: label, prompt: prompt)
            }
            guard !tasks.isEmpty else { continue }
            batches.append(
                OrchestraPlanBatch(
                    name: trimmedNonEmpty(raw["name"]?.stringValue) ?? "batch-\(index + 1)",
                    purpose: trimmedNonEmpty(raw["purpose"]?.stringValue),
                    tasks: tasks))
        }
        return OrchestraPlanCall(
            plan: arguments?["plan"]?.stringValue ?? "", batches: batches)
    }

    private func toolResult(id: JSONValue, text: String) -> HTTPResponse {
        rpcResult(
            id: id,
            result: .object([
                "content": .array([
                    .object(["type": .string("text"), "text": .string(text)])
                ]),
                "isError": .bool(false),
            ]))
    }

    /// As ``toolResult(id:text:)`` but also attaches an image content block, so a tool can return a
    /// **screenshot the model actually sees** (standard MCP image content; the Claude CLI forwards an
    /// image tool result into the model's context). Used by `mac_vm_computer` to close the see→act loop.
    /// `mimeType` is the base64 payload's type (e.g. `image/jpeg`).
    private func toolResult(
        id: JSONValue, text: String, imageBase64: String, mimeType: String = "image/jpeg"
    ) -> HTTPResponse {
        rpcResult(
            id: id,
            result: .object([
                "content": .array([
                    .object(["type": .string("text"), "text": .string(text)]),
                    .object([
                        "type": .string("image"),
                        "data": .string(imageBase64),
                        "mimeType": .string(mimeType),
                    ]),
                ]),
                "isError": .bool(false),
            ]))
    }

    /// Render a `*_computer_batch` reply into a tool result: a text log of every step in order,
    /// followed by the attached screenshot(s). Each step contributes a text line, and any step that
    /// carries an image (the final one, or all of them under `screenshot_each_step`) contributes an
    /// image block right after its line, so the model reads "step → picture" in order.
    private func batchToolResult(
        id: JSONValue, reply: MCPApprovalServer.MacVMComputerBatchReply
    ) -> HTTPResponse {
        guard case .ran(let steps, let stoppedAtError) = reply else {
            return toolResult(id: id, text: reply.deniedJSON())
        }
        var content: [JSONValue] = []
        var lines: [String] = []
        for step in steps {
            let status = step.failed ? "FAILED" : "ok"
            lines.append("step \(step.index) [\(step.action)] \(status): \(step.summary)")
        }
        if stoppedAtError {
            lines.append("(stopped at first error; remaining steps not attempted)")
        }
        content.append(
            .object([
                "type": .string("text"),
                "text": .string(lines.joined(separator: "\n")),
            ]))
        // Attach screenshots after the summary. Label each so the model can tell which step a picture
        // belongs to when several are present (screenshot_each_step).
        for step in steps {
            guard let image = step.imageBase64 else { continue }
            content.append(
                .object([
                    "type": .string("text"),
                    "text": .string("screenshot after step \(step.index) [\(step.action)]:"),
                ]))
            content.append(
                .object([
                    "type": .string("image"),
                    "data": .string(image),
                    "mimeType": .string("image/jpeg"),
                ]))
        }
        return rpcResult(
            id: id,
            result: .object([
                "content": .array(content),
                "isError": .bool(false),
            ]))
    }

    private func rpcResult(id: JSONValue, result: JSONValue) -> HTTPResponse {
        let body = JSONValue.object(["jsonrpc": .string("2.0"), "id": id, "result": result])
        return HTTPResponse(
            status: 200, statusText: "OK", contentType: "application/json",
            body: (try? body.encodedData()) ?? Data())
    }

    private func rpcError(id: JSONValue, code: Int, message: String) -> HTTPResponse {
        let body = JSONValue.object([
            "jsonrpc": .string("2.0"), "id": id,
            "error": .object(["code": .number(Double(code)), "message": .string(message)]),
        ])
        return HTTPResponse(
            status: 200, statusText: "OK", contentType: "application/json",
            body: (try? body.encodedData()) ?? Data())
    }
}

// MARK: - Transport abstraction

/// A bidirectional byte stream the HTTP/JSON-RPC serve loop drives, so the same dispatch runs over
/// either a Network.framework TCP connection (host loopback) or a raw unix-domain socket (the
/// vsock-relayed control plane). `receive` returns nil at EOF; `close` is idempotent.
private protocol ByteConn: Sendable {
    func receive(maxLength: Int) async throws -> Data?
    func send(_ data: Data) async throws
    func close()
}

#if canImport(Network)
/// ``ByteConn`` over a Network.framework connection (the TCP transport). Started on creation so the
/// serve loop can receive immediately.
private final class NWByteConn: ByteConn, @unchecked Sendable {
    private let connection: NWConnection
    init(_ connection: NWConnection) {
        self.connection = connection
        connection.start(queue: .global(qos: .userInitiated))
    }
    func receive(maxLength: Int) async throws -> Data? {
        try await withCheckedThrowingContinuation { cont in
            connection.receive(minimumIncompleteLength: 1, maximumLength: maxLength) {
                data, _, isComplete, error in
                if let error {
                    cont.resume(throwing: error)
                } else if let data, !data.isEmpty {
                    cont.resume(returning: data)
                } else if isComplete {
                    cont.resume(returning: nil)
                } else {
                    cont.resume(returning: Data())
                }
            }
        }
    }
    func send(_ data: Data) async throws {
        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
            connection.send(
                content: data,
                completion: .contentProcessed { error in
                    if let error { cont.resume(throwing: error) } else { cont.resume() }
                })
        }
    }
    func close() { connection.cancel() }
}
#endif

/// ``ByteConn`` over an accepted `AF_UNIX` stream socket (the vsock-relayed transport). Reads/writes
/// are blocking POSIX calls offloaded to a background queue, so the actor never blocks and a turn
/// suspended on an approval holds no thread (no read is in flight between request and response). The
/// control plane's live connection count is small (one MCP connection plus short-lived interceptor
/// posts per container), so a thread parked on `read` per live connection is acceptable.
///
/// **fd lifetime (load-bearing).** `close()` only `shutdown(2)`s the socket; the descriptor itself
/// is released exactly once, in `deinit`, and every offloaded read/write block captures `self` — so
/// the fd *number* cannot be recycled while any block that could still pass it to the kernel is in
/// flight. A straggler `receive`/`send` after `close()` sees the shut-down socket (instant EOF /
/// EPIPE), never a stranger's descriptor. The previous revision `close(2)`d in `close()`: the serve
/// loop's next `receive` — dispatched concurrently with an SSE handler's mid-loop `close()` (see
/// `respondStreamingToolCall`) — could then issue a blocking `read(2)` on a number the kernel had
/// already handed to a NEW fd (another session's MCP socket, a container's stdio channel…), parking
/// a global-queue thread forever on — and stealing bytes from — an unrelated stream. Those zombie
/// threads/reads accumulated until every container session's stdio stalled at once (the "produced
/// no output within 60s / stdio transport stalled" lockup).
#if !os(Windows)
private final class UnixSocketByteConn: ByteConn, @unchecked Sendable {
    private let fd: Int32
    private let lock = NSLock()
    private var closed = false
    init(fd: Int32) {
        self.fd = fd
        // A client that dies mid-call (a killed supervisor's CLI) closes its end while a response
        // may still be in flight; without this, that write raises SIGPIPE and kills the WHOLE app
        // rather than surfacing as a plain EPIPE from `write` (handled in `send`). Linux has no
        // SO_NOSIGPIPE; there `writeNoSigpipe` passes MSG_NOSIGNAL on every send instead.
        #if canImport(Darwin)
        var one: Int32 = 1
        setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, socklen_t(MemoryLayout<Int32>.size))
        #endif
    }

    deinit {
        // The sole close(2) of the descriptor. Every offloaded block strongly captures `self`, so
        // deinit is ordered after the last kernel call that could reference this fd number.
        posixCloseFD(fd)
    }

    private var isClosed: Bool {
        lock.lock()
        defer { lock.unlock() }
        return closed
    }

    func receive(maxLength: Int) async throws -> Data? {
        // Fast EOF once closed (the serve loop re-enters receive after an SSE handler's mid-loop
        // close). Racing closes are safe regardless: the fd stays valid until deinit, and shutdown
        // makes a late read return 0.
        guard !isClosed else { return nil }
        return try await withTaskCancellationHandler {
            try await withCheckedThrowingContinuation { cont in
                DispatchQueue.global(qos: .userInitiated).async {
                    var buffer = [UInt8](repeating: 0, count: maxLength)
                    while true {
                        let n = buffer.withUnsafeMutableBytes {
                            read(self.fd, $0.baseAddress, maxLength)
                        }
                        if n > 0 {
                            cont.resume(returning: Data(buffer[0..<n]))
                        } else if n == 0 {
                            cont.resume(returning: nil)
                        } else if errno == EINTR {
                            continue
                        } else {
                            cont.resume(throwing: POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO))
                        }
                        return
                    }
                }
            }
        } onCancel: {
            // Task cancellation (server stop / connection teardown) can't interrupt a blocking
            // read(2); shutting the socket down wakes the parked read with EOF so the serve loop
            // actually exits instead of stranding the thread until the peer hangs up.
            self.close()
        }
    }

    func send(_ data: Data) async throws {
        guard !isClosed else { throw POSIXError(.EPIPE) }
        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
            DispatchQueue.global(qos: .userInitiated).async {
                let result: Result<Void, Error> = data.withUnsafeBytes { raw in
                    guard let base = raw.baseAddress else { return .success(()) }
                    var offset = 0
                    while offset < raw.count {
                        let n = writeNoSigpipe(self.fd, base + offset, raw.count - offset)
                        if n > 0 {
                            offset += n
                        } else if n < 0 && errno == EINTR {
                            continue
                        } else {
                            return .failure(POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO))
                        }
                    }
                    return .success(())
                }
                cont.resume(with: result)
            }
        }
    }

    func close() {
        lock.lock()
        defer { lock.unlock() }
        guard !closed else { return }
        closed = true
        // Wakes any parked read with EOF and fails later writes with EPIPE/ENOTCONN. Deliberately
        // NO close(2) here — deinit releases the descriptor (see the class comment).
        shutdown(fd, Int32(SHUT_RDWR))  // Glibc spells the constant as Int; Int32 on both platforms
    }
}
#endif

#if os(Windows)
/// ``ByteConn`` over an accepted NIO child channel (the Windows TCP transport —
/// docs/WINDOWS_PORT.md §4.3). ``NIOByteConnHandler`` feeds inbound bytes from the event
/// loop; `receive` drains the buffer or parks exactly one waiter; channel-inactive is EOF.
/// The serve loop issues one receive at a time, so a single waiter slot suffices.
private final class NIOByteConn: ByteConn, @unchecked Sendable {
    private let channel: Channel
    private let lock = NSLock()
    private var buffered = Data()
    private var eof = false
    private var pendingError: Error?
    private var waiter: CheckedContinuation<Data?, Error>?

    init(channel: Channel) {
        self.channel = channel
    }

    func receive(maxLength: Int) async throws -> Data? {
        try await withCheckedThrowingContinuation { cont in
            lock.lock()
            if !buffered.isEmpty {
                let chunk = Data(buffered.prefix(maxLength))
                buffered.removeFirst(chunk.count)
                lock.unlock()
                cont.resume(returning: chunk)
            } else if let error = pendingError {
                pendingError = nil
                lock.unlock()
                cont.resume(throwing: error)
            } else if eof {
                lock.unlock()
                cont.resume(returning: nil)
            } else {
                waiter = cont
                lock.unlock()
            }
        }
    }

    func send(_ data: Data) async throws {
        var buffer = channel.allocator.buffer(capacity: data.count)
        buffer.writeBytes(data)
        try await channel.writeAndFlush(buffer).get()
    }

    func close() {
        channel.close(promise: nil)
    }

    // Fed by the handler on the channel's event loop.

    func push(_ data: Data) {
        lock.lock()
        if let cont = waiter {
            waiter = nil
            lock.unlock()
            cont.resume(returning: data)
            return
        }
        buffered.append(data)
        lock.unlock()
    }

    func pushEOF() {
        lock.lock()
        eof = true
        let cont = waiter
        waiter = nil
        lock.unlock()
        cont?.resume(returning: nil)
    }

    func pushError(_ error: Error) {
        lock.lock()
        let cont = waiter
        waiter = nil
        if cont == nil { pendingError = error }
        lock.unlock()
        cont?.resume(throwing: error)
    }
}

/// Pipeline handler pairing one accepted channel with its ``NIOByteConn``.
private final class NIOByteConnHandler: ChannelInboundHandler, @unchecked Sendable {
    typealias InboundIn = ByteBuffer
    private let conn: NIOByteConn

    init(conn: NIOByteConn) {
        self.conn = conn
    }

    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
        var buffer = unwrapInboundIn(data)
        if let bytes = buffer.readBytes(length: buffer.readableBytes) {
            conn.push(Data(bytes))
        }
    }

    func channelInactive(context: ChannelHandlerContext) {
        conn.pushEOF()
        context.fireChannelInactive()
    }

    func errorCaught(context: ChannelHandlerContext, error: Error) {
        conn.pushError(error)
        context.close(promise: nil)
    }
}
#endif

/// Failure binding/listening the control-plane socket (unix on Darwin/Linux, NIO TCP on Windows).
private struct UnixSocketError: Error, CustomStringConvertible {
    let description: String
    init(_ description: String) { self.description = description }
}

/// A `CheckedContinuation` that can be resumed from two racing paths — the real handler finishing vs.
/// the user tapping Skip — where exactly one resume must reach the continuation and any later ones are
/// dropped. Used by ``MCPApprovalServer/runSkippable(name:command:work:skipped:)``; the lock makes the
/// claim atomic across the handler's task and the actor-driven Skip closure.
private final class ResumeOnce<R: Sendable>: @unchecked Sendable {
    private let lock = NSLock()
    private var continuation: CheckedContinuation<R, Never>?

    init(_ continuation: CheckedContinuation<R, Never>) { self.continuation = continuation }

    func resume(_ value: R) {
        lock.lock()
        let cont = continuation
        continuation = nil
        lock.unlock()
        cont?.resume(returning: value)
    }
}

#if canImport(Network)
/// Resume-once guard for NWListener state callbacks (ready may be followed by
/// failed; the continuation must fire exactly once).
private final class OnceBox: @unchecked Sendable {
    private let lock = NSLock()
    private var claimed = false

    func claim() -> Bool {
        lock.lock()
        defer { lock.unlock() }
        if claimed { return false }
        claimed = true
        return true
    }
}
#endif
