import Foundation

/// AgentBackend over the Claude Code CLI: spawns
/// `claude -p --output-format stream-json --input-format stream-json --verbose`,
/// normalizes stdout into `AgentEvent`s, and bridges interactive approvals
/// through an in-process MCP server + `--permission-prompt-tool`
/// (ADAPTERS §1; BACKEND_PROTOCOL §7.1).
public actor ClaudeCodeBackend: AgentBackend {
    public static let id = BackendID.claudeCode

    public nonisolated let capabilities = BackendCapabilities(
        interactiveApprovals: true,
        allowAlwaysScopes: [.session, .toolName, .toolNameWithPattern],
        canModifyToolInput: true,
        partialMessageStreaming: true,
        emitsThinking: true,
        emitsFileChangeEvents: false,  // synthesized from Edit/Write tool calls
        nativeResume: true,
        sandboxModes: [],
        followUpWhileRunning: false)

    public struct Configuration: Sendable {
        /// Executable name (resolved via PATH) or absolute path. Tests point this
        /// at the fake-claude stub (OBSERVABILITY B.4).
        public var executable: String
        public var includePartialMessages: Bool
        /// Pre-allowed via --allowedTools so cheap read-only tools never
        /// round-trip through the approval server (ADAPTERS §1.1).
        public var allowedTools: [String]
        /// Single-shot runs close stdin right after the prompt so the CLI exits
        /// at end of turn; interactive sessions keep it open for follow-ups.
        public var closeStdinAfterPrompt: Bool
        /// Pass `--strict-mcp-config` so the child uses ONLY our `--mcp-config`
        /// (the approval server) and ignores the host's global MCP servers/plugins.
        /// The M0 live capture showed the spawned CLI otherwise inherits the user's
        /// Cloudflare/Stripe/etc. plugin tools — undesirable for a hermetic Nucleic
        /// session (M0 finding 7). Default on.
        public var strictMCPConfig: Bool
        /// Optional curated settings (inline JSON or a file path) passed via
        /// `--settings`, to further isolate the child from user-level config.
        public var settings: String?
        /// Directory for the optional raw native-stream capture
        /// (OBSERVABILITY A.5); nil = off.
        public var captureDirectory: URL?
        /// Wall-clock ceiling for a single `host_exec` command (``runOnHost``). Unlike the
        /// human-in-the-loop MCP timeouts (pinned effectively infinite so an approval can suspend as
        /// long as a person needs), a host build/test involves no human, so a hang — a deadlocked
        /// `swift test`, a wedged build — must NOT wedge the tool call forever. When it elapses the
        /// command's whole process tree is reaped (``ProcessHandle/terminateTree``) and the agent gets
        /// a real "timed out" result instead of the session hanging until someone kills processes on
        /// the host. Generous by default (30 min) so a legitimately long clean build/test never trips
        /// it; a run that needs longer can raise it.
        public var hostCommandTimeoutNanos: UInt64
        public var now: @Sendable () -> Date

        public init(
            executable: String = "claude",
            includePartialMessages: Bool = false,
            allowedTools: [String] = ["Read", "Glob", "Grep"],
            closeStdinAfterPrompt: Bool = true,
            strictMCPConfig: Bool = true,
            settings: String? = nil,
            captureDirectory: URL? = nil,
            hostCommandTimeoutNanos: UInt64 = 30 * 60 * 1_000_000_000,
            now: @escaping @Sendable () -> Date = { Date() }
        ) {
            self.executable = executable
            self.includePartialMessages = includePartialMessages
            self.allowedTools = allowedTools
            self.closeStdinAfterPrompt = closeStdinAfterPrompt
            self.strictMCPConfig = strictMCPConfig
            self.settings = settings
            self.captureDirectory = captureDirectory
            self.hostCommandTimeoutNanos = hostCommandTimeoutNanos
            self.now = now
        }
    }

    /// MCP timeouts for the spawned CLI. Two knobs with deliberately different values:
    ///
    /// * `MCP_TOOL_TIMEOUT` — per *tool call*. Our gated calls (`approve`, `AskUserQuestion`,
    ///   `check_conflict`, `host_exec`) SUSPEND until a human responds, which may be many minutes
    ///   if they step away. The CLI's default would abandon the suspended call and let the model
    ///   proceed *without* an answer (observed as questions/approvals "timing out"), so this stays
    ///   pinned to Node's max delay (~24.8 days, the largest `setTimeout` accepts before it wraps).
    ///   The unbounded per-call wait does NOT strand server-side work when the run dies first:
    ///   `teardownRun()`'s token unregister (and any connection close) cancels the approval
    ///   server's in-flight handler tasks — most importantly parked/running `nucleic_subagent`
    ///   spawns, which would otherwise hold Orchestra worker slots for a dead supervisor.
    ///
    /// * `MCP_TIMEOUT` — the initial server *connection/handshake*, which involves no human. Pinning
    ///   it as high as the tool timeout meant an unreachable approval server would wedge startup for
    ///   ~24.8 days instead of failing fast, so it's bounded to a generous-but-finite 60s. The
    ///   approval server is in-process and already listening before the child execs, so a real
    ///   connection completes in milliseconds; 60s only ever bounds a genuine failure.
    ///
    /// An explicit `RunSpec.extraEnv` still wins.
    static let mcpTimeoutEnv: [String: String] = [
        "MCP_TIMEOUT": "60000",
        "MCP_TOOL_TIMEOUT": "2147483647",
    ]

    /// `CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS` — at the end of a `-p` turn whose stdin is closed, the
    /// CLI blocks waiting for any still-running background subagents/tasks so their results fold into
    /// the turn's final output (which is exactly how a fan-out's work gets *received and acted upon*),
    /// but only up to this ceiling. The CLI default is 600000 ms = 10 minutes (`claude` 2.1.185); once
    /// it elapses the run prints "Background tasks still running … terminating", KILLS those tasks, and
    /// exits — their work is abandoned, never folded in, so the parent never sees it.
    ///
    /// Nucleic closes stdin on every turn (``Configuration/closeStdinAfterPrompt``), which is the
    /// `inputClosed` state that arms that wind-down, and its Orchestra fan-out parents routinely spawn
    /// subagents that run far longer than 10 minutes. So we pin this to `0` (wait indefinitely) — the
    /// value the CLI's own message recommends. A runaway subagent is instead bounded by the user's Stop
    /// (SIGINT→SIGKILL) and the container idle timeout, mirroring how `MCP_TOOL_TIMEOUT` above is pinned
    /// to Node's max for the human-in-the-loop calls that suspend just as long. An explicit
    /// ``RunSpec/extraEnv`` still wins (merged after, at both spawn sites).
    static let printBackgroundWaitEnv: [String: String] = [
        "CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS": "0",
    ]

    /// `CLAUDE_CODE_MCP_TOOL_IDLE_TIMEOUT` — separate from the wall-clock `MCP_TOOL_TIMEOUT` above, the
    /// CLI (v2.1.187+) also aborts an MCP tool call that produces NO response and NO progress
    /// notification for an *idle* window: 5 minutes for HTTP/SSE/WebSocket servers, 30 minutes for
    /// stdio servers. Our approval server is a streamable-HTTP MCP server, so its window is 5 minutes.
    ///
    /// The gated calls it serves (`approve`, `AskUserQuestion`, …) are exactly the case this kills:
    /// they suspend awaiting a human and emit no progress in the meantime, so a human who takes longer
    /// than 5 minutes to answer would see the call aborted even though `MCP_TOOL_TIMEOUT` is pinned to
    /// Node's max. We pin the idle timeout to `0` (disable the check) for the same reason we pin that
    /// wall-clock timeout high — a human-in-the-loop call legitimately suspends indefinitely, bounded
    /// instead by the user's Stop and the container idle timeout. An explicit ``RunSpec/extraEnv`` still
    /// wins (merged after, at both spawn sites).
    static let mcpToolIdleTimeoutEnv: [String: String] = [
        "CLAUDE_CODE_MCP_TOOL_IDLE_TIMEOUT": "0",
    ]

    /// `BASH_DEFAULT_TIMEOUT_MS` / `BASH_MAX_TIMEOUT_MS` — the Bash tool's per-call default and hard
    /// ceiling. Unlike the human-in-the-loop MCP calls above (pinned effectively infinite because a
    /// person legitimately suspends them), a Bash command runs unattended, so a hang — a wedged build,
    /// a command waiting on input that never comes — must be reaped rather than left to run until the
    /// container idle timeout. The CLI's own defaults (2 min default / 10 min max) apply only when
    /// these are unset; we pin BOTH to 270000 ms (4m30s) so every Bash call defaults to that timeout
    /// AND cannot exceed it (a per-call `timeout` argument is clamped to the max). A command that
    /// genuinely needs longer should run through `host_exec` (bounded instead by
    /// ``Configuration/hostCommandTimeoutNanos``). An explicit ``RunSpec/extraEnv`` still wins.
    static let bashTimeoutEnv: [String: String] = [
        "BASH_DEFAULT_TIMEOUT_MS": "270000",
        "BASH_MAX_TIMEOUT_MS": "270000",
    ]

    /// The full environment pinned onto every spawned `claude` (host spawn or container exec): the MCP
    /// timeouts (connection, tool wall-clock, and tool idle), the background-wait ceiling, and the Bash
    /// tool timeout. Both spawn sites start from this and then merge ``RunSpec/extraEnv`` (and,
    /// host-side, the Managed Git env) on top, so a per-run override wins.
    static var pinnedChildEnv: [String: String] {
        mcpTimeoutEnv
            .merging(printBackgroundWaitEnv) { _, new in new }
            .merging(mcpToolIdleTimeoutEnv) { _, new in new }
            .merging(bashTimeoutEnv) { _, new in new }
    }

    /// How long a *containerized* run may produce NO stdout before it's declared a stalled spawn.
    /// The first event (`system/init`) lands within seconds of launch — after the CLI finishes MCP
    /// init against the approval server, before any model call — so a containerized run silent this
    /// long is almost always hanging in MCP connect against a wedged control plane (a hop that
    /// accepts connections it never answers). The watchdog probes that path and recovers the
    /// container when the probe confirms it dead. Generous enough never to false-fire on a slow
    /// cold container start; host runs are exempt (a host pipe either works at once or the spawn
    /// throws). See ``reportStalledSpawnIfSilent(stderrTail:)``.
    static let firstOutputTimeoutNanos: UInt64 = 60 * 1_000_000_000

    private let configuration: Configuration
    private let processHost: ProcessHost
    /// The per-backend approval server — used for host runs and per-session sandbox containers.
    /// Shared-control-container runs use the registry's container-scoped server instead (resolved
    /// per run into ``runServer``).
    private let approvalServer: MCPApprovalServer
    /// Vends the shared, token-multiplexed server for a Nucleic Control container, so every session
    /// in that shared box reaches us at the container's single relayed control socket. `nil` (tests,
    /// host-only setups) → always use ``approvalServer``.
    private let approvalServerRegistry: ApprovalServerRegistry?
    /// The approval server the current run uses (the shared one for a control container, else
    /// ``approvalServer``). Set at run start; teardown/shutdown unregister this run's token from it.
    private var runServer: MCPApprovalServer?
    /// The server to act on for this run's token lifecycle (the resolved run server, or the
    /// per-backend default before/after a run).
    private var activeServer: MCPApprovalServer { runServer ?? approvalServer }
    public let approvals: ApprovalCoordinator
    /// Optional sandbox orchestrator. `nil` → always spawn on the host (default). When a
    /// `RunSpec.container` is present and this is set, the run executes inside the container.
    private let containerManager: ContainerManager?
    /// Optional macOS-VM orchestrator. When set and `RunSpec.allowMacVMExec` is on, the `mac_vm_exec`
    /// tool is exposed so the agent runs Mac-only work in this session's own isolated macOS VM (see
    /// docs/MACOS_VM.md). `nil` (tests / no service) → the tool is never advertised.
    private let macVMManager: MacVMManager?
    /// Optional conflict coordinator. When set, the `check_conflict` MCP tool is exposed and
    /// the agent is instructed to consult it before editing (internal project management).
    private let conflictCoordinator: ConflictCoordinator?
    /// Optional provider for Nucleic's mediated Claude subscription access token, refreshed on
    /// demand. A HOST-run turn (no container) would otherwise let `claude` read its OWN credential
    /// store (the macOS Keychain / `~/.claude`), which is NOT the login the user signed into through
    /// Nucleic — so it fails with "OAuth session expired and could not be refreshed" when that
    /// external login is stale. This supplies the mediated token as `CLAUDE_CODE_OAUTH_TOKEN`. `nil`
    /// (tests / no broker) or a nil result → claude keeps its own auth. Container runs are unaffected
    /// (the broker seeds their token via the container spec).
    private let oauthTokenProvider: (@Sendable () async -> String?)?

    private var handle: (any ProcessHandle)?
    /// The per-session Claude token proxy (opt-in) — a loopback TCP listener for a host run, or the
    /// host end of the relayed unix socket for a container run. Started lazily and reused across this
    /// session's turns; torn down in ``shutdown()``. `nil` until first started / when disabled.
    private var tokenProxy: ClaudeTokenProxy?
    /// Memoized host-run base URL, so repeated turns reuse the one loopback listener.
    private var hostTokenProxyBaseURL: String?
    /// In-flight host `host_exec` command handles, keyed by a monotonic id. Tracked so Stop
    /// (``interrupt``) and session teardown (``shutdown``) can reap a running host build/test — and
    /// its whole process tree — instead of leaving it orphaned on the shared host (see ``runOnHost``).
    /// Only host commands are tracked: VM/SSH exec runs in a disposable guest, where abandoning the
    /// wait (Skip) is the right behavior and the guest is torn down wholesale.
    private var inFlightHostHandles: [UInt64: any ProcessHandle] = [:]
    private var hostCommandSeq: UInt64 = 0
    /// Host command handles for which a "looks hung" alert is currently live in the transcript, keyed
    /// by the alert's stall id. Lets the user's Kill button (routed through ``resolveProcessStall``)
    /// find and tree-kill exactly that command. An entry exists only between the alert firing and its
    /// resolution (kill/dismiss/exit/resume).
    private var stalledHostHandles: [String: any ProcessHandle] = [:]
    /// Name of the container the current run executes in (for the `finished` callback).
    private var activeContainerName: String?
    private var continuation: AsyncThrowingStream<AgentEvent, Error>.Continuation?
    private var sessionID: SessionID?
    private var serverToken: String?
    /// The live run's loopback control endpoint (host, port, bearer token), captured when the
    /// approval server starts. `runOnHost`/`spawnMonitorProcess` read it to wire nash's host
    /// observation env (docs/NASH.md §7.7.3) — those sites can't see the spawn block's locals, and
    /// the triple is stable for the run's lifetime. Loopback-TCP (host) runs only; containerized
    /// runs leave it `nil` and use ``hostShellSocketPath`` instead.
    private var hostControlEndpoint: (host: String, port: UInt16, token: String)?
    /// A containerized run's control socket **host path** — the same relayed unix socket the guest
    /// bridge forwards to, but reachable directly from the host filesystem. A host spawn under nash
    /// (`host_exec` escaping the sandbox — exactly NASH.md §1 goal 5's case) posts its shell events
    /// here via `NUCLEIC_SHELL_SOCKET`, since a containerized run has no loopback TCP listener.
    private var hostShellSocketPath: String?
    private var provisionalSeq: UInt64 = 0
    private var interruptRequested = false
    private var terminating = false
    private var sawRunFinished = false
    /// True once the agent's first stdout line reaches the decode loop. The containerized spawn
    /// watchdog disarms on this; while it's false past the deadline the run is a stalled spawn.
    private var sawFirstOutput = false
    /// Set when the spawn watchdog fired (no first output in time), so the post-loop exit handling
    /// settles the run as a stalled-spawn error and doesn't also report the watchdog's own SIGKILL
    /// as an abnormal/OOM exit.
    private var spawnStalled = false
    /// Set once this run's stderr showed the `claude -p` background-wait cutoff (see
    /// ``printBackgroundWaitEnv``), so the visible note is emitted at most once per run.
    private var reportedBackgroundWaitCutoff = false
    /// Whether the current run is in auto-approve mode (set per run). Edit tools are routed
    /// through us even in auto mode for a conflict pre-check; on no-conflict they auto-allow.
    private var autoApprove = false
    /// When set, native Claude Task/Agent spawns are blocked so subagent creation goes through
    /// Nucleic's managed `nucleic_subagent` tool, where model selection and lifecycle are enforced and
    /// each worker is an observed session. True for an Orchestra supervisor (non-blocking spawner) and
    /// for a Control-project run with the blocking subagent spawner.
    private var blockNativeSubagentSpawns = false
    /// Set once the user picks "Allow for this session" on a `host_exec` prompt, so further host
    /// commands this session run without re-prompting (HOST_EXEC). Auto-approve never sets this —
    /// only an explicit user choice does. Persists across turns (actor lives for the session).
    private var hostExecSessionGranted = false
    /// As ``hostExecSessionGranted``, for `mac_vm_exec` — set when the user picks "Allow for this
    /// session" on a macOS-VM prompt, so further in-VM commands run without re-prompting.
    private var macVMExecSessionGranted = false
    /// Whether this session opted into computer use (`run.allowMacVMComputer`). Drives whether the
    /// session's VM boots **computer-use-capable** (main queue + a host-side surface for SIP-free
    /// capture/input) — set for BOTH the exec and computer specs so the VM boots consistently
    /// regardless of which tool touches it first.
    private var macVMComputerAllowed = false
    /// Name of the macOS VM the current session booted for `mac_vm_exec` (for the `finished` callback).
    private var activeMacVMName: String?
    /// As ``macVMExecSessionGranted``, for `linux_vm_exec`.
    private var linuxVMExecSessionGranted = false
    /// Whether this session opted into Linux-VM computer use (`run.allowLinuxVMComputer`) — set for
    /// both the Linux exec and computer specs so the Linux VM boots computer-use-capable consistently.
    private var linuxVMComputerAllowed = false
    /// Name of the Linux VM the current session booted (for the `finished` callback).
    private var activeLinuxVMName: String?
    /// Background poller that keeps the Nash Viewer current while a VM is running (docs/NASH.md
    /// §6.2). Started the first time this session activates a VM; cancelled on ``shutdown()``.
    private var vmShellDrainTask: Task<Void, Never>?
    /// How often the poller drains a running VM's nash spool. nash flushes to the spool every 500 ms,
    /// so a 1 s poll keeps the viewer within about a second of the guest without making the drain
    /// chatty (an empty spool answers immediately).
    static let vmShellDrainIntervalNanos: UInt64 = 1_000_000_000
    /// The throwaway containers this session created via `linux_container`, keyed by logical name,
    /// so `list` can enumerate them and `exec` can re-ensure one that idle-stopped between calls.
    private var agentContainerSpecs: [String: ContainerSpec] = [:]
    /// Event sink used when this actor services platform tools for another provider.
    private var platformToolEventSink: (@Sendable (AgentEvent.Kind, String?) -> Void)?
    public private(set) var backendSessionID: String?

    // MARK: - Monitor state (nucleic_monitor)

    /// The maximum number of `nucleic_monitor` watches a session may run at once — a backstop so a
    /// runaway loop of arm calls can't spawn unbounded background processes.
    static let maxConcurrentMonitors = 8
    /// Per-monitor cap on delivered events before the watch is auto-stopped, mirroring the native
    /// tool's "monitors that produce too many events are automatically stopped" guard. Each event is
    /// a full agent turn here, so an over-chatty monitor is both expensive and useless.
    static let maxMonitorEvents = 200
    /// How long lines are coalesced before one delivery, so a burst of output becomes a single turn
    /// rather than one turn per line (delivery is far costlier here than the native in-session notice).
    static let monitorCoalesceNanos: UInt64 = 400_000_000
    /// Once a bounded monitor is past its `timeout_ms` age, how long the watched process must ALSO be
    /// output-silent and resource-quiet (under ``ProcessStallConfig``'s CPU and disk-I/O thresholds)
    /// before the watch is reaped. The age alone never stops a monitor: a build that simply takes longer
    /// than the bound is still working, and killing its tree there is the bug this guards against.
    /// Only used when the handle's pid is host-samplable; see ``beginMonitorStallWatch(id:ageNanos:)``.
    static let monitorStallIdleNanos: UInt64 = 60 * 1_000_000_000
    /// How often a `host_exec` command that outlived ``Configuration/hostCommandTimeoutNanos`` is
    /// re-examined. Past the ceiling the command is spared for as long as it keeps working, so the
    /// stall verdict is polled on this cadence rather than granting it another whole ceiling.
    static let hostStallRecheckNanos: UInt64 = 5 * 60 * 1_000_000_000
    /// How long an administrative monitor notice waits for a real event to carry it before it is
    /// delivered on its own. Long enough that a live watch's next event usually collects it for free;
    /// short enough that an agent waiting on a watch that already ended isn't left guessing.
    static let monitorNoticeParkNanos: UInt64 = 5 * 60 * 1_000_000_000

    /// How to spawn a monitor's command so it runs in the SAME environment as the agent's Bash tool:
    /// inside the session's container when sandboxed, else on the host worktree. Captured at run
    /// start (mirrors how `host_exec` captures its `workdir`) and reused for every monitor armed this
    /// session — the container is session-lifetime, so its name stays valid across turns.
    private struct MonitorSpawnContext: Sendable {
        /// Logical container name, or nil for a host run.
        let containerName: String?
        /// The session container's spec (nil for a host run), so a monitor can hold its own activity
        /// ref and keep the box alive past the arming turn (it would otherwise idle-stop and kill the
        /// watch).
        let cspec: ContainerSpec?
        let workdir: String
        let env: [String: String]
        let uid: Int?
        let gid: Int?
    }
    /// A monitor's event source: a spawned process (its stdout lines) or a WebSocket (its inbound
    /// text frames). The WebSocket case is Darwin-only — NucleicCore also builds for the Linux
    /// runner, where `URLSessionWebSocketTask` rides libcurl's absent experimental WS (the mesh uses
    /// SwiftNIO there instead), so a `ws` monitor is refused on Linux with a clear fallback.
    private enum MonitorSource {
        case process(any ProcessHandle)
        #if canImport(Darwin)
        case webSocket(URLSessionWebSocketTask)
        #endif
    }
    /// One armed monitor's live state. Held across turns (the watch outlives the ephemeral CLI turn
    /// that armed it) and torn down only on stop/exit/timeout or session ``shutdown()``.
    private final class MonitorRun {
        let source: MonitorSource
        /// Whether the source is a WebSocket (for end-note wording).
        let isWebSocket: Bool
        let description: String
        /// Container whose activity ref this monitor holds (nil for a host / WebSocket run), balanced
        /// once on end.
        let pinnedContainerName: String?
        var task: Task<Void, Never>?
        /// The bounded watch's age timer, and after it lapses the follow-on idle re-check (WebSocket
        /// sources). Cancelled exactly once, when the monitor is claimed by an end path.
        var deadlineTask: Task<Void, Never>?
        /// Armed only once a bounded process monitor is past its age, to decide whether it is genuinely
        /// wedged or merely slow. Nil until then, and for persistent / WebSocket monitors.
        var stallMonitor: ProcessStallMonitor?
        var pending: [String] = []
        var flushScheduled = false
        var delivered = 0
        var stopped = false
        var containerReleased = false
        /// When this monitor last produced an event. The liveness signal for a WebSocket source, which
        /// has no process to sample; a process source uses its handle's own activity clock instead.
        var lastEventNanos: UInt64 = DispatchTime.now().uptimeNanoseconds
        init(
            source: MonitorSource, isWebSocket: Bool, description: String,
            pinnedContainerName: String?
        ) {
            self.source = source
            self.isWebSocket = isWebSocket
            self.description = description
            self.pinnedContainerName = pinnedContainerName
        }
    }
    /// The latest run's event sink — re-supplied each turn (the CLI doesn't persist `--allowedTools`
    /// across `--resume`, so every turn re-registers). A monitor armed in an earlier turn delivers
    /// through whatever sink the current turn installed; all sinks target the same session.
    private var monitorEventSink: MonitorEventSink?
    /// How to spawn monitor commands for this session; set at run start when the tool is registered.
    private var monitorSpawnContext: MonitorSpawnContext?
    /// The directory the agent's Bash tool runs in — the session container's workdir when
    /// sandboxed, else the host worktree. Set at run start and used to resolve a shell command's
    /// relative write targets into lockable paths before the command runs (docs/NASH.md §5.2).
    private var agentShellWorkdir: String?
    /// Armed monitors by id. Survives `teardownRun()`; cleared only on ``shutdown()``.
    private var activeMonitors: [String: MonitorRun] = [:]
    private var monitorSeq: UInt64 = 0
    /// Administrative notices (a watch ended / was stopped) parked until a real monitor event can carry
    /// them, so they don't each buy an agent turn to say nothing. See ``deliverMonitorNotice(_:label:)``.
    private var pendingMonitorNotices: [String] = []
    /// The watch label the parked notices are delivered under if the backstop has to send them alone.
    private var pendingMonitorNoticeLabel: String?
    /// Backstop that delivers parked notices when no real event turns up to carry them.
    private var pendingNoticeTask: Task<Void, Never>?

    public init(
        configuration: Configuration = Configuration(),
        processHost: ProcessHost = ProcessHost(),
        approvalServer: MCPApprovalServer = MCPApprovalServer(),
        approvals: ApprovalCoordinator = ApprovalCoordinator(),
        containerManager: ContainerManager? = nil,
        macVMManager: MacVMManager? = nil,
        conflictCoordinator: ConflictCoordinator? = nil,
        approvalServerRegistry: ApprovalServerRegistry? = nil,
        oauthTokenProvider: (@Sendable () async -> String?)? = nil
    ) {
        self.configuration = configuration
        self.processHost = processHost
        self.approvalServer = approvalServer
        self.approvals = approvals
        self.containerManager = containerManager
        self.macVMManager = macVMManager
        self.conflictCoordinator = conflictCoordinator
        self.approvalServerRegistry = approvalServerRegistry
        self.oauthTokenProvider = oauthTokenProvider
    }

    /// Configure this actor as the provider-neutral Nucleic platform-tool runtime.
    func configurePlatformToolRuntime(
        for run: RunSpec,
        eventSink: @escaping @Sendable (AgentEvent.Kind, String?) -> Void
    ) {
        sessionID = run.sessionID
        autoApprove = run.autoApprove
        platformToolEventSink = eventSink
        // The agent's Bash tool runs inside the session container when sandboxed, else on the host
        // worktree — the same `workdir` the host-exec bridge captures. The container's workdir is
        // the canonicalized host worktree path, so resolving a shell write target against it yields
        // a path the arbiter can strip back to repo-relative.
        agentShellWorkdir = run.container?.workdir ?? run.worktree
    }

    /// Register the host, VM, computer-use, lifecycle, and disposable-container MCP tools.
    func registerPlatformTools(on server: MCPApprovalServer, token: String, run: RunSpec) async {
        // Host-exec bridge (HOST_EXEC): when this run opts in, expose the `host_exec`
        // tool. Its handler is the sole gate — every call surfaces an approval (never
        // auto-approved) and, once granted, runs on the host outside the container.
        if run.allowHostExec {
            let workdir = run.container?.workdir ?? run.worktree
            await server.registerHostExec(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleHostExecCall(call, workdir: workdir)
            }
        }
        // macOS-VM bridge: when this run opts in AND a VM manager is wired, expose `mac_vm_exec`.
        // Its handler is the sole gate (every call surfaces an approval, never auto-approved) and,
        // once granted, boots/reuses this session's isolated macOS VM and runs the command through
        // its vsock agent — the Mac-native, per-agent alternative to shared-host `host_exec`.
        if run.allowMacVMExec, macVMManager != nil {
            let hostWorktree = run.worktree
            await server.registerMacVMExec(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleMacVMExecCall(call, hostWorktree: hostWorktree)
            }
        }
        // macOS-VM computer-use bridge: when opted in, expose `mac_vm_computer` so the agent can
        // see and drive the VM's GUI. Actions run in the VM sandbox and are NOT per-action gated.
        macVMComputerAllowed = run.allowMacVMComputer
        if run.allowMacVMComputer, macVMManager != nil {
            let hostWorktree = run.worktree
            await server.registerMacVMComputer(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleMacVMComputerCall(call, hostWorktree: hostWorktree)
            }
            // Batch sibling: run an ordered sequence of computer-use actions in one call.
            await server.registerMacVMComputerBatch(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleMacVMComputerBatchCall(call, hostWorktree: hostWorktree)
            }
            // macOS-VM operator-assist bridge: rides on computer-use (the interactive host-side
            // surface only exists for a computer-use VM). Its handler is the sole gate — every
            // `mac_vm_request_operator` call surfaces an approval (never auto-approved) and, once
            // the user accepts, hands them an interactive viewer to operate the VM by hand.
            await server.registerMacVMOperator(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleMacVMOperatorCall(call, hostWorktree: hostWorktree)
            }
            // macOS-VM notification-clearing bridge: rides on computer-use (only meaningful when the
            // agent is looking at the VM's screen). Not gated — a fixed `killall NotificationCenter`
            // against the agent's own disposable VM.
            await server.registerMacVMClearNotifications(token: token) { [weak self] _ in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleMacVMClearNotificationsCall(hostWorktree: hostWorktree)
            }
        }
        // macOS-VM lifecycle bridge: expose `mac_vm_control` whenever this session can use the
        // macOS VM at all (exec or computer-use), so the agent can stop/suspend/resume/restart/kill
        // the VM it booted. Not gated — it only manages the agent's own disposable VM.
        if (run.allowMacVMExec || run.allowMacVMComputer), macVMManager != nil {
            await server.registerMacVMControl(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleMacVMControlCall(call)
            }
        }
        // Linux-VM bridge: exposes `linux_vm_exec` — this session's own isolated Linux VM (the
        // lighter, faster-booting sibling of the macOS VM), for Linux dev work a container can't do.
        if run.allowLinuxVMExec, macVMManager != nil {
            let hostWorktree = run.worktree
            await server.registerLinuxVMExec(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleLinuxVMExecCall(call, hostWorktree: hostWorktree)
            }
        }
        // Linux-VM computer-use bridge: exposes `linux_vm_computer` (GUI over the Linux VM).
        linuxVMComputerAllowed = run.allowLinuxVMComputer
        if run.allowLinuxVMComputer, macVMManager != nil {
            let hostWorktree = run.worktree
            await server.registerLinuxVMComputer(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleLinuxVMComputerCall(call, hostWorktree: hostWorktree)
            }
            // Batch sibling: run an ordered sequence of computer-use actions in one call.
            await server.registerLinuxVMComputerBatch(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleLinuxVMComputerBatchCall(call, hostWorktree: hostWorktree)
            }
        }
        // Linux-VM lifecycle bridge: the Linux sibling of `mac_vm_control` above.
        if (run.allowLinuxVMExec || run.allowLinuxVMComputer), macVMManager != nil {
            await server.registerLinuxVMControl(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleLinuxVMControlCall(call)
            }
        }
        // Agent-container bridge: exposes `linux_container` so the agent can spin up its own
        // throwaway Linux containers (the lightest rung). Needs a container manager to run them.
        if run.allowAgentContainers, containerManager != nil {
            let sessionImage = run.container?.image ?? ProjectSandbox.defaultImage
            let hostWorktree = run.worktree
            await server.registerLinuxContainer(token: token) { [weak self] call in
                guard let self else { return .denied(message: "Session terminated") }
                return await self.handleLinuxContainerCall(
                    call, defaultImage: sessionImage, hostWorktree: hostWorktree)
            }
        }
    }

    /// Register every agent-callable Nucleic tool whose semantics are provider-neutral.
    func registerNucleicTools(on server: MCPApprovalServer, token: String, run: RunSpec) async {
        if let conflictCoordinator {
            let sid = run.sessionID
            await server.registerConflict(token: token) { call in
                let outcome = await conflictCoordinator.arbitrate(
                    sessionID: sid, task: call.task, files: call.files)
                return MCPApprovalServer.ConflictReply(resolution: outcome.resolution)
            }
        }
        await registerPlatformTools(on: server, token: token, run: run)
        // Orchestra worker bridge: the fan-out tool creates Nucleic sessions instead of opaque
        // provider-native background tasks, so Nucleic enforces the configured worker model and
        // observes the worker lifecycle/transcript like any other chat.
        if run.orchestraActive, let spawner = run.orchestraSubagentSpawner {
            await server.registerOrchestraSubagent(token: token) { call in
                let request = OrchestraSubagentRequest(
                    task: call.task, prompt: call.prompt, batch: call.batch)
                return await spawner(request)
            }
        }
        // Plan delegation: `nucleic_delegate_plan` takes the supervisor's whole batched task list in one
        // call and spawns a worker per task. The supervisor's primary dispatch path; the single-spawn
        // tool above remains for one-off follow-up workers.
        if run.orchestraActive, let planHandler = run.orchestraPlanHandler {
            await server.registerOrchestraPlan(token: token) { call in
                let request = OrchestraPlanRequest(plan: call.plan, batches: call.batches)
                return await planHandler(request)
            }
        }
        // Managed-subagent bridge (non-Orchestra, Nucleic Control projects): the same
        // `nucleic_subagent` tool in its BLOCKING shape — the call runs a worker session to
        // completion and returns its final reply inline, a drop-in for the native Task/Agent tool
        // (which is denied while this is active). Registered on the same handler slot as the
        // Orchestra spawner; the two are mutually exclusive per run, and the absence of a
        // `supervise` handler is what makes the tool advertise its blocking description.
        else if let blockingSpawner = run.blockingSubagentSpawner {
            await server.registerOrchestraSubagent(token: token) { call in
                let request = OrchestraSubagentRequest(task: call.task, prompt: call.prompt)
                return await blockingSpawner(request)
            }
        }
        // Supervisor drive loop: `nucleic_supervise` (block for worker events) and
        // `nucleic_reply_to_worker` (answer a worker's question). Present for an active supervisor
        // alongside the spawner.
        if run.orchestraActive, let supervise = run.orchestraSuperviseHandler {
            await server.registerOrchestraSupervise(token: token) { batch in await supervise(batch) }
        }
        // Ask-the-user bridge: `nucleic_ask_user`, the provider-neutral counterpart to Claude Code's
        // built-in `AskUserQuestion`. Registered for every run EXCEPT a machine-spawned worker — a
        // worker has no user to reach and must route questions to its supervisor
        // (`nucleic_ask_supervisor`), whose handler's presence is this run's worker marker. Handled
        // here rather than app-side because the whole exchange is an approval round-trip this actor
        // already owns, which is what makes it work identically on Claude, Codex, and Grok.
        if run.orchestraAskSupervisorHandler == nil {
            await server.registerAskUser(token: token) { [weak self] call in
                guard let self else {
                    return .unavailable(message: "The session ended before the user could answer.")
                }
                return await self.handleAskUserCall(call)
            }
        }
        if run.orchestraActive, let replyToWorker = run.orchestraReplyToWorkerHandler {
            await server.registerOrchestraReplyToWorker(token: token) { call in
                let request = OrchestraReplyRequest(
                    workerID: SessionID(rawValue: call.workerID), reply: call.reply)
                return await replyToWorker(request)
            }
        }
        // Worker-side bridge: `nucleic_ask_supervisor`. Present ONLY for a machine-spawned worker
        // (the app injects the handler iff this run is a worker), so it's the run's worker marker.
        if let askSupervisor = run.orchestraAskSupervisorHandler {
            await server.registerOrchestraAskSupervisor(token: token) { call in
                await askSupervisor(call.question)
            }
        }
        // Mesh messaging bridge: `nucleic_send_message` + `nucleic_subscribe_topic` (mesh
        // casting "messages" channel). Present only when the user enabled agent messaging in
        // Settings; the host stamps sender provenance, so tool args can't forge an identity.
        if let mesh = run.meshMessaging {
            await server.registerMeshMessaging(
                token: token,
                send: { request in await mesh.send(request) },
                subscribe: { topic in await mesh.subscribe(topic) },
                waitForMessage: { topic, timeout in await mesh.waitForMessage(topic, timeout) })
        }
        // Monitor bridge: `nucleic_monitor` — arm a background watch whose output re-invokes this
        // session as fresh turns (the working replacement for the blocked provider-native `Monitor`).
        // Present whenever the host wired an event sink. Capture how to spawn the watch in the same
        // environment as the agent's Bash tool (its container when sandboxed, else the host worktree)
        // — `activeContainerName` is already resolved (the container was brought up before this).
        if let sink = run.monitorEventSink {
            monitorEventSink = sink
            monitorSpawnContext = MonitorSpawnContext(
                containerName: activeContainerName,
                cspec: run.container,
                workdir: run.container?.workdir ?? run.worktree,
                env: run.container?.env ?? [:],
                uid: run.container?.runAsUID,
                gid: run.container?.runAsGID)
            await server.registerMonitor(token: token) { [weak self] call in
                guard let self else { return "Session terminated." }
                return await self.handleMonitorCall(call)
            }
        }
    }

    // MARK: - AgentBackend

    public nonisolated func start(_ run: RunSpec) -> AsyncThrowingStream<AgentEvent, Error> {
        makeStream(run: run, resumeArgs: [])
    }

    public nonisolated func resume(_ resume: ResumeSpec) -> AsyncThrowingStream<AgentEvent, Error> {
        var args = ["--resume", resume.backendSessionID]
        if resume.fork { args.append("--fork-session") }
        let run = RunSpec(
            sessionID: resume.sessionID,
            worktree: resume.worktree,
            prompt: resume.prompt ?? AgentInput(parts: []),
            model: resume.model,
            effort: resume.effort,
            autoApprove: resume.autoApprove,
            container: resume.container,
            allowHostExec: resume.allowHostExec,
            allowMacVMExec: resume.allowMacVMExec,
            allowMacVMComputer: resume.allowMacVMComputer,
            allowLinuxVMExec: resume.allowLinuxVMExec,
            allowLinuxVMComputer: resume.allowLinuxVMComputer,
            allowAgentContainers: resume.allowAgentContainers,
            orchestraActive: resume.orchestraActive,
            orchestraSubagentSpawner: resume.orchestraSubagentSpawner,
            blockingSubagentSpawner: resume.blockingSubagentSpawner,
            orchestraSuperviseHandler: resume.orchestraSuperviseHandler,
            orchestraReplyToWorkerHandler: resume.orchestraReplyToWorkerHandler,
            orchestraAskSupervisorHandler: resume.orchestraAskSupervisorHandler,
            meshMessaging: resume.meshMessaging,
            monitorEventSink: resume.monitorEventSink,
            appendSystemPrompt: resume.appendSystemPrompt)
        return makeStream(run: run, resumeArgs: args)
    }

    public func send(_ input: AgentInput) async throws {
        guard let handle else { throw BackendError.notRunning }
        try handle.writeLine(Self.userMessageLine(for: input))
    }

    public func respond(to approvalID: ApprovalID, _ decision: Decision, by responder: String) async throws {
        let outcome = await approvals.resolve(approvalID, decision, by: responder)
        if outcome == .unknownID {
            throw BackendError.protocolViolation("No outstanding approval \(approvalID)")
        }
    }

    public func interrupt() async {
        interruptRequested = true
        // SIGINT lets `claude` end the turn cleanly (it emits an interrupted result); the escalation
        // SIGKILLs the process group if it ignores the interrupt, so Stop always takes effect even
        // on a wedged agent. The session stays resumable — a follow-up turn re-spawns the process.
        handle?.interruptThenKill()
        interruptPlatformToolProcesses()
    }

    /// Stop the processes owned by this instance's provider-neutral platform-tool runtime.
    ///
    /// Claude uses those handlers directly, while Codex and every ACP agent own a helper
    /// `ClaudeCodeBackend` solely to serve the same Nucleic tools. Keeping this operation separate
    /// from ``interrupt`` is important for those helpers: setting their agent-level
    /// `interruptRequested` flag would poison the next tool call even though no Claude run exists.
    /// The concrete provider backends call this method from their own `interrupt()` implementations,
    /// making the UI's Stop action reap `host_exec` process trees consistently for every agent type.
    func interruptPlatformToolProcesses() {
        // A leftover `swift test`/build would otherwise outlive the stopped turn and keep holding the
        // shared trunk's build lock. Detached so Stop stays non-blocking; each tree reap self-bounds.
        // `runOnHost`'s own wait observes the kill and returns a real terminated result.
        let handles = Array(inFlightHostHandles.values)
        inFlightHostHandles.removeAll()
        for handle in handles { Task { await handle.terminateTree() } }
    }

    public func skipToolCall(_ call: ToolCall) async {
        // Only the host/VM exec tools are skippable, and each carries a `command`. Route the Skip to
        // this run's MCP server, which matches it against the in-flight handler by tool name + command
        // and makes that handler return early with a "skipped" reply — abandoning a hung command
        // without disturbing the rest of the turn. A non-skippable call or one already finished is a
        // harmless no-op (nothing matches).
        guard MCPApprovalServer.isSkippable(toolName: call.name),
              let command = call.input["command"]?.stringValue
        else { return }
        await activeServer.skipToolCall(name: call.name, command: command)
    }

    public func shutdown() async {
        terminating = true
        await approvals.cancelOutstanding(reason: "Session terminated")
        // Kill any armed `nucleic_monitor` watches (their processes are owned by this actor, not the
        // CLI, so they'd otherwise outlive the session as orphans).
        await teardownMonitors()
        // Stop the VM nash-spool poller (docs/NASH.md §6.2). Before the `handle` guard below, which
        // returns early for a session with no live CLI process — the task must not outlive either.
        vmShellDrainTask?.cancel()
        vmShellDrainTask = nil
        // Release this backend's OWN approval server — its loopback TCP listener (host runs) or
        // its per-session control socket (per-session containers), which would otherwise linger
        // on disk after the session ends. Never the registry's shared server (`runServer` on a
        // control-container run): other sessions in that container are still using it, and a
        // future run of this backend re-`start()`s idempotently either way.
        await approvalServer.stop()
        // Reap any in-flight host command (and its process tree) before returning — a running
        // `host_exec` build/test must not outlive the session as an orphan holding the shared trunk's
        // SwiftPM build lock. Done before the CLI-handle guard below so it runs even for a session
        // that has no live CLI process (e.g. one torn down between turns).
        await reapInFlightHostCommands()
        // Tear down the token proxy (its loopback TCP listener, or the relayed unix socket) if this
        // session started one.
        #if canImport(Network)
        await tokenProxy?.stop()
        tokenProxy = nil
        hostTokenProxyBaseURL = nil
        #endif
        guard let handle else { return }
        handle.closeStdin()
        await handle.terminate()
        self.handle = nil
        if let serverToken {
            await activeServer.unregister(token: serverToken)
        }
    }

    /// Start (once) the session's host token proxy and return its `ANTHROPIC_BASE_URL`. Reused across
    /// this session's turns. Returns nil if the platform has no loopback listener or the bind fails —
    /// the caller then leaves `claude` on the static token alone (no regression, just no mid-turn
    /// rotation).
    private func ensureHostTokenProxyBaseURL(
        tokenProvider: @escaping @Sendable () async -> String?
    ) async -> String? {
        #if canImport(Network)
        if let base = hostTokenProxyBaseURL { return base }
        let proxy = ClaudeTokenProxy(tokenProvider: tokenProvider)
        do {
            let port = try await proxy.start(host: "127.0.0.1")
            let base = "http://127.0.0.1:\(port)"
            tokenProxy = proxy
            hostTokenProxyBaseURL = base
            return base
        } catch {
            claudeAuthLog.error("Claude token proxy failed to start: \(String(describing: error), privacy: .public)")
            return nil
        }
        #else
        return nil
        #endif
    }

    /// Start (once) the session's token proxy on the relayed unix socket for a container run. Returns
    /// whether it's listening — the caller sets `ANTHROPIC_BASE_URL` only then, so a bind failure
    /// leaves `claude` on the static token rather than pointing it at a dead endpoint.
    private func startContainerTokenProxy(
        socketPath: String, tokenProvider: @escaping @Sendable () async -> String?
    ) async -> Bool {
        #if canImport(Network)
        if tokenProxy != nil { return true }  // reused across this session's turns
        let proxy = ClaudeTokenProxy(tokenProvider: tokenProvider)
        do {
            try await proxy.start(unixSocketPath: socketPath)
            tokenProxy = proxy
            return true
        } catch {
            claudeAuthLog.error("Claude token proxy (container) failed to start: \(String(describing: error), privacy: .public)")
            return false
        }
        #else
        return false
        #endif
    }

    // MARK: - Run loop

    private nonisolated func makeStream(run: RunSpec, resumeArgs: [String])
        -> AsyncThrowingStream<AgentEvent, Error>
    {
        AsyncThrowingStream { continuation in
            let task = Task {
                await self.runLoop(run: run, resumeArgs: resumeArgs, continuation: continuation)
            }
            continuation.onTermination = { _ in task.cancel() }
        }
    }

    private func runLoop(
        run: RunSpec,
        resumeArgs: [String],
        continuation: AsyncThrowingStream<AgentEvent, Error>.Continuation
    ) async {
        self.continuation = continuation
        self.sessionID = run.sessionID
        self.agentShellWorkdir = run.container?.workdir ?? run.worktree
        self.sawRunFinished = false
        self.interruptRequested = false
        self.sawFirstOutput = false
        self.spawnStalled = false
        self.reportedBackgroundWaitCutoff = false
        self.autoApprove = run.autoApprove
        self.blockNativeSubagentSpawns =
            (run.orchestraActive && run.orchestraSubagentSpawner != nil)
            || run.blockingSubagentSpawner != nil

        do {
            // 0. Resolve the approval server for this run. Sessions that share a Nucleic Control
            // container share ONE token-multiplexed server (so they can share that container's
            // single relayed control socket); host runs and per-session sandboxes keep the
            // per-backend server. The bearer token disambiguates sessions on the shared server.
            let server: MCPApprovalServer
            if let cspec = run.container,
                ContainerManager.allSharedControlContainerNames.contains(cspec.name),
                let registry = approvalServerRegistry
            {
                server = await registry.server(for: cspec.name)
            } else {
                server = approvalServer
            }
            runServer = server

            // 0b. Sandbox: if this run is containerized, bring the container + its control
            // plane up in the platform's required order (``ContainerControlPlane`` —
            // relay-then-bind on macOS, gateway-TCP bind-then-create on Windows). The
            // container's init also launches the in-guest control bridge the child reaches
            // us through.
            let sandbox = (run.container != nil) ? containerManager : nil
            var containerControl: ContainerControlPlane.Ready?
            if let sandbox, let cspec = run.container {
                let ready = try await ContainerControlPlane.bringUp(
                    spec: cspec, manager: sandbox, server: server)
                activeContainerName = ready.containerName
                containerControl = ready
            }

            // 1. Approval bridge: per-session bearer token → handler.
            let token = UUID().uuidString
            serverToken = token
            // Transport: a containerized child rides the per-platform control plane
            // ``ContainerControlPlane`` bound above (relayed unix socket on macOS — no IP
            // listener at all, so macOS raises no incoming-connection / local-network
            // prompts; interface-scoped gateway TCP on Windows). The agent + interceptor
            // shims reach us via the in-guest loopback bridge either way, so the control
            // endpoint is `127.0.0.1:<bridge port>` inside the guest. A host (non-container)
            // run keeps the loopback TCP listener.
            let port: UInt16
            let controlHost: String
            if let containerControl {
                controlHost = "127.0.0.1"
                port = ContainerSpec.controlBridgePort
                // Host spawns (host_exec) under nash post shell events straight to the relayed
                // socket from the host filesystem (NASH.md §7.7.3); nil on Windows (§9.4).
                hostShellSocketPath = containerControl.hostShellSocketPath
            } else {
                port = try await server.start(host: "127.0.0.1")
                controlHost = "127.0.0.1"
                // A bound listener always reports a non-zero ephemeral port; a 0 here means the
                // approval server didn't actually bind, so the agent could never reach us and would
                // silently stall on its first gated tool. Fail the run loudly instead.
                guard port != 0 else {
                    throw BackendError.spawnFailed(
                        "approval server did not bind a port (control endpoint unreachable)")
                }
                // Host runs post shell events straight to this loopback listener (NASH.md §6.2's
                // "simplest surface"); park the triple where `runOnHost`/monitors can reach it.
                hostControlEndpoint = (controlHost, port, token)
            }
            await server.register(token: token) { [weak self] call in
                guard let self else { return .deny(message: "Session terminated") }
                return await self.handleApprovalCall(call)
            }
            await registerNucleicTools(on: server, token: token, run: run)
            // Git-interceptor bridge (Nucleic Control containers): the in-container `git` shim
            // reports each mutating git op to `POST /git-event`. Fire-and-forget — record it and
            // return immediately so a report never blocks the agent's git call.
            //
            // A host run on a narOS host takes the same routes (docs/NAROS.md N3): on the
            // Covalence runner image the container IS the sandbox (`RunSpec.container = nil`),
            // no shims are seeded, but `/bin/sh` is nash by the base's divert — so the shell
            // route is that surface's whole interception layer, and the git/gh convergence in
            // `AppStore.observeShellEvent` (nash exec events whose argv head is git/gh feed
            // `observeGitOp`/`observeGhOp`, docs/NASH.md §9.4) restores git/gh observation the
            // shims provide elsewhere. The git/gh/command routes are registered alongside for
            // uniformity; without shims they simply stay idle.
            // Third disjunct (docs/NASH.md §7.7.3): a host run on a *Mac* with the host shell set
            // to nash — `hostIsNarOS` is false on macOS by construction, so without it a Mac host
            // run would register no `/shell-event` route and nash's posts would 401.
            if run.container?.installGitInterceptor == true
                || (run.container == nil && CommandInterceptor.hostIsNarOS)
                || (run.container == nil && HostShellSettings.nashEnabled)
            {
                await server.registerGitReport(token: token) { [weak self] call in
                    Task { await self?.handleGitReport(call) }
                }
                // The sibling `gh` shim reports agentic GitHub actions (PR/release/API) to
                // `POST /gh-event`. Same fire-and-forget contract as the git report.
                await server.registerGhReport(token: token) { [weak self] call in
                    Task { await self?.handleGhReport(call) }
                }
                // The non-git command interceptor (shims + bash tracer) reports to `/command-event`
                // under the same gate/token — every command the agent's shell ran, for the GUI feed
                // and logs. Also fire-and-forget so a report never blocks the agent.
                await server.registerCommandReport(token: token) { [weak self] call in
                    Task { await self?.handleCommandReport(call) }
                }
                // The nash agent shell's exec gate reports to `/shell-event` under the same
                // gate/token (docs/NASH.md §6) — ground-truth exec/cd/export/fallback events for
                // the feeds. Same fire-and-forget contract.
                await server.registerShellReport(token: token) { [weak self] calls in
                    Task { await self?.handleShellReport(calls) }
                }
            } else if run.container != nil {
                // Non-Control sandbox runs do not install the git/gh/command shims, but their
                // default `linux_container` children can still reuse this control socket for nash.
                await server.registerShellReport(token: token) { [weak self] calls in
                    Task { await self?.handleShellReport(calls) }
                }
            }
            let mcpConfig = server.mcpConfigJSON(host: controlHost, port: port, token: token)

            // 2. Spawn.
            var args = [
                "-p",
                "--output-format", "stream-json",
                "--input-format", "stream-json",
                "--verbose",
            ]
            if configuration.includePartialMessages {
                args.append("--include-partial-messages")
            }
            args += ["--permission-prompt-tool", MCPApprovalServer.qualifiedToolName]
            // Pin interactive runs to Claude's client-review path. Provider-native auto mode can
            // execute tools before Nucleic sees them, so Nucleic reproduces Auto with the shared
            // risk policy in handleApprovalCall instead.
            if case .interactive = run.approvalPolicy {
                // Always route provider approvals through Nucleic. Auto mode is evaluated by
                // NucleicApprovalPolicy below, never by the provider's private classifier.
                args += ["--permission-mode", "default"]
            }
            args += ["--mcp-config", mcpConfig]
            // Hermetic child: only our approval server, none of the host's global
            // MCP servers/plugins (M0 finding 7).
            // Only an explicit MCP opt-in lifts that isolation. Skills and plugins are discovered
            // from the selectively copied config home and do not broaden the global MCP surface.
            let allowsExternalMCP = ExternalAgentIntegrationSettings.allows(.mcp, model: run.model)
            if configuration.strictMCPConfig && !allowsExternalMCP {
                args.append("--strict-mcp-config")
            }
            if let settings = configuration.settings {
                args += ["--settings", settings]
            }
            // Pre-allow `check_conflict` so an agent that consults it directly never
            // round-trips through the approval prompt (it's a query, not a gated action).
            var allowedTools = configuration.allowedTools
            if conflictCoordinator != nil {
                allowedTools.append(MCPApprovalServer.qualifiedConflictToolName)
            }
            // Pre-allow `host_exec` so the call reaches our handler directly instead of
            // Claude's permission path — our handler is the sole gate, so `auto` mode can
            // never auto-approve it (HOST_EXEC).
            if run.allowHostExec {
                allowedTools.append(MCPApprovalServer.qualifiedHostExecToolName)
            }
            // Pre-allow `mac_vm_exec` for the same reason (our handler is the sole gate).
            if run.allowMacVMExec, macVMManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedMacVMExecToolName)
            }
            // Pre-allow `mac_vm_computer` so computer-use actions reach our handler with no per-action
            // prompt (the VM sandbox is the safety boundary).
            if run.allowMacVMComputer, macVMManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedMacVMComputerToolName)
                // Pre-allow the batch sibling on the same terms (VM sandbox is the boundary).
                allowedTools.append(MCPApprovalServer.qualifiedMacVMComputerBatchToolName)
                // Pre-allow `mac_vm_request_operator` too (rides on computer-use); our handler gates it.
                allowedTools.append(MCPApprovalServer.qualifiedMacVMOperatorToolName)
                // Pre-allow `mac_vm_clear_notifications` (rides on computer-use; not gated — it only
                // touches the agent's own disposable VM).
                allowedTools.append(MCPApprovalServer.qualifiedMacVMClearNotificationsToolName)
            }
            // Pre-allow the Linux-VM tools for the same reason (our handler is the sole gate).
            if run.allowLinuxVMExec, macVMManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedLinuxVMExecToolName)
            }
            if run.allowLinuxVMComputer, macVMManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedLinuxVMComputerToolName)
                // Pre-allow the batch sibling on the same terms (VM sandbox is the boundary).
                allowedTools.append(MCPApprovalServer.qualifiedLinuxVMComputerBatchToolName)
            }
            // Pre-allow the VM lifecycle tools so their ops reach our handler directly (no per-op
            // prompt — they only manage the agent's own disposable VM). Exposed alongside the VM.
            if (run.allowMacVMExec || run.allowMacVMComputer), macVMManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedMacVMControlToolName)
            }
            if (run.allowLinuxVMExec || run.allowLinuxVMComputer), macVMManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedLinuxVMControlToolName)
            }
            // Pre-allow `linux_container` so lifecycle ops reach our handler directly (the sole gate).
            if run.allowAgentContainers, containerManager != nil {
                allowedTools.append(MCPApprovalServer.qualifiedLinuxContainerToolName)
            }
            // Orchestra: pre-allow only Nucleic's managed subagent tool. Native Task/Agent are
            // deliberately not pre-allowed here, so worker spawning flows through Nucleic where the
            // app can enforce the configured worker model and observe the worker lifecycle.
            if run.orchestraActive, run.orchestraSubagentSpawner != nil {
                allowedTools.append(MCPApprovalServer.qualifiedOrchestraSubagentToolName)
            }
            // Blocking managed subagent (non-Orchestra Control run): pre-allow the same tool so its
            // calls reach our handler directly, and so native Task/Agent (deliberately not pre-allowed)
            // still routes through the permission path where we deny it toward `nucleic_subagent`.
            else if run.blockingSubagentSpawner != nil {
                allowedTools.append(MCPApprovalServer.qualifiedOrchestraSubagentToolName)
            }
            // Plan delegation — pre-allowed on the same terms as the spawner it dispatches through.
            if run.orchestraActive, run.orchestraPlanHandler != nil {
                allowedTools.append(MCPApprovalServer.qualifiedOrchestraPlanToolName)
            }
            // `nucleic_ask_user` — pre-allowed so the call reaches our handler, which raises the
            // question card itself. Gating it on the permission path would put an approval prompt in
            // front of an approval prompt. Present for every non-worker run (see registerNucleicTools).
            if run.orchestraAskSupervisorHandler == nil {
                allowedTools.append(MCPApprovalServer.qualifiedAskUserToolName)
            }
            // Supervisor drive/reply tools — pre-allowed so they reach our handlers directly, like the
            // spawner (they suspend/mutate the worker channel, never the permission path).
            if run.orchestraActive, run.orchestraSuperviseHandler != nil {
                allowedTools.append(MCPApprovalServer.qualifiedOrchestraSuperviseToolName)
            }
            if run.orchestraActive, run.orchestraReplyToWorkerHandler != nil {
                allowedTools.append(MCPApprovalServer.qualifiedOrchestraReplyToWorkerToolName)
            }
            // Worker-side ask tool — pre-allowed for a machine-spawned worker.
            if run.orchestraAskSupervisorHandler != nil {
                allowedTools.append(MCPApprovalServer.qualifiedOrchestraAskSupervisorToolName)
            }
            // Mesh messaging tools — pre-allowed (sending text within the user's own mesh is
            // benign; the Settings toggle is the gate, and the body size is capped host-side).
            if run.meshMessaging != nil {
                allowedTools.append(MCPApprovalServer.qualifiedMeshSendMessageToolName)
                allowedTools.append(MCPApprovalServer.qualifiedMeshSubscribeTopicToolName)
                allowedTools.append(MCPApprovalServer.qualifiedMeshWaitForMessageToolName)
            }
            // Pre-allow `nucleic_monitor` so an arm request reaches our handler directly (it only
            // starts a benign, agent-chosen watch; the handler is the sole gate). Re-applied each
            // turn since the CLI doesn't persist `--allowedTools` across `--resume`.
            if run.monitorEventSink != nil {
                allowedTools.append(MCPApprovalServer.qualifiedMonitorToolName)
            }
            if !allowedTools.isEmpty {
                args += ["--allowedTools", allowedTools.joined(separator: ",")]
            }
            if let model = run.model { args += ["--model", model] }
            if let effort = run.effort { args += ["--effort", effort] }
            if let appendSystemPrompt = run.appendSystemPrompt {
                args += ["--append-system-prompt", appendSystemPrompt]
            }
            args += resumeArgs
            args += run.extraArgs

            // Host-run (unsandboxed) agent: pin its git (every remote) to the Managed Git key for the
            // run's lifetime — shredded when this `do` block exits, i.e. after the decode loop below
            // ends (the agent process exited). A containerized run gets the key via `provision`
            // (cspec.env) instead. No-op when no Managed Git credential is configured.
            let isHostRun = !(sandbox != nil && run.container != nil && activeContainerName != nil)
            var hostManagedGitEnv: [String: String] = [:]
            let hostManagedGitCleanup: @Sendable () -> Void
            if isHostRun {
                let (e, c) = GitHubCredentialProvisioner.provisionHost()
                hostManagedGitEnv = e
                hostManagedGitEnv.removeValue(forKey: "GH_CONFIG_DIR")
                hostManagedGitCleanup = c
            } else {
                hostManagedGitCleanup = {}
            }
            defer { hostManagedGitCleanup() }

            // Spawn on the host, or exec the same `claude` invocation inside the container VM.
            let handle: any ProcessHandle
            if let sandbox, let cspec = run.container, let name = activeContainerName {
                var env = Self.pinnedChildEnv
                    .merging(cspec.env) { _, new in new }
                    .merging(run.extraEnv) { _, new in new }
                // Tell the in-container `git` shim where to report (the host approval server,
                // reached over the in-guest control bridge) and under which per-session bearer
                // token. The env is inherited by every git subprocess the agent spawns.
                if cspec.installGitInterceptor {
                    // Point the in-container git/gh/command interceptor at the host control endpoint
                    // (the loopback control bridge, forwarding to the relayed socket). Shared helper
                    // so every containerized backend wires the interceptor identically.
                    env.merge(
                        CommandInterceptor.hookEnv(
                            host: controlHost, port: port, token: token, sessionID: run.sessionID,
                            commandTracing: ContainerServiceSettings.commandTracingEnabled)
                    ) { _, new in new }
                }
                // Claude token proxy (opt-in): start the host end of the relayed proxy socket and
                // point the guest `claude` at it via `ANTHROPIC_BASE_URL`, so every request is
                // re-bearered with a broker-fresh token — fixing mid-turn expiry while the refresh
                // token stays broker-only. `CLAUDE_CODE_OAUTH_TOKEN` (in cspec.env) still lets
                // `claude` start; the proxy supersedes it per request. On any failure we leave the
                // static token alone (no base-URL override), so the turn is never worse off.
                if ContainerServiceSettings.claudeTokenProxyEnabled,
                   let proxySocket = cspec.proxySocketHostPath,
                   let oauthTokenProvider,
                   await startContainerTokenProxy(
                       socketPath: proxySocket, tokenProvider: oauthTokenProvider) {
                    env["ANTHROPIC_BASE_URL"] = "http://127.0.0.1:\(ContainerSpec.proxyBridgePort)"
                }
                // Exec inside the running container VM; the returned handle streams the guest
                // process's stdio over vsock and satisfies the same `ProcessHandle` contract as a
                // host spawn, so the decode loop below is identical in-container and on the host.
                handle = try await sandbox.exec(
                    name: name, workdir: cspec.workdir, env: env,
                    argv: [configuration.executable] + args,
                    uid: cspec.runAsUID, gid: cspec.runAsGID)
            } else {
                var hostEnv = Self.pinnedChildEnv
                    .merging(run.extraEnv) { _, new in new }
                    .merging(hostManagedGitEnv) { _, new in new }
                // narOS host (the Covalence runner container — docs/NAROS.md N3): the spawned
                // agent's `/bin/sh` IS nash (forced by the base image's divert), so point the
                // shell observation env at this run's loopback approval server — exec/data-flow
                // events reach the feed exactly as a containerized run's do (docs/NASH.md §6.2,
                // §7.3). The `legacyShell` lever reaches nash through the trusted policy file
                // rather than this env, which no longer carries a kill switch (§4.3). The git/gh
                // hook vars ride along inertly (no shims on this surface; real git ignores them)
                // and the bash tracer stays off — its trace script is container-seeded and absent
                // here. On every other host (macOS, non-narOS Linux) this branch is skipped and
                // the env unchanged.
                if CommandInterceptor.hostIsNarOS {
                    hostEnv.merge(
                        CommandInterceptor.hookEnv(
                            host: controlHost, port: port, token: token,
                            sessionID: run.sessionID, commandTracing: false)
                    ) { _, new in new }
                }
                // Host-run `claude` reads its OWN credential store, not Nucleic's mediated login, so a
                // stale external login fails auth ("OAuth session expired and could not be refreshed").
                // Inject the token the user signed into through Nucleic (refreshed on demand) so a
                // host-run turn authenticates with the mediated credential. Skipped when there's no
                // Nucleic login (nil) — claude then falls back to its own auth / surfaces the login row.
                if let oauthTokenProvider, let token = await oauthTokenProvider(),
                   !token.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty {
                    hostEnv["CLAUDE_CODE_OAUTH_TOKEN"] = token
                    // Token proxy (opt-in): a static `CLAUDE_CODE_OAUTH_TOKEN` can't rotate for the
                    // life of the process, so a turn outliving the ~1h token 401s mid-turn. When
                    // enabled, also point `claude` at the loopback proxy, which overwrites the bearer
                    // with a broker-fresh token on every request — the static token above still lets
                    // `claude` start and is simply superseded per request.
                    if ContainerServiceSettings.claudeTokenProxyEnabled,
                       let base = await ensureHostTokenProxyBaseURL(tokenProvider: oauthTokenProvider) {
                        hostEnv["ANTHROPIC_BASE_URL"] = base
                    }
                }
                let spec = ProcessSpec(
                    executable: configuration.executable,
                    args: args,
                    cwd: run.worktree,
                    env: hostEnv,
                    stdinMode: .pipe)
                handle = try await processHost.launch(spec)
            }
            self.handle = handle

            // 3. Initial prompt over stdin (NDJSON user message).
            if !run.prompt.parts.isEmpty {
                try handle.writeLine(Self.userMessageLine(for: run.prompt))
            }
            if configuration.closeStdinAfterPrompt {
                handle.closeStdin()
            }

            // 4. stderr tail for diagnostics.
            let stderrTail = StderrTail()
            let stderrTask = Task { [weak self] in
                for try await line in handle.stderrLines {
                    let text = String(decoding: line, as: UTF8.self)
                    stderrTail.append(text)
                    // Turn the otherwise-silent background-wait cutoff (see `printBackgroundWaitEnv`)
                    // into a visible note. Unreachable while we pin the ceiling to 0, but a finite
                    // override (RunSpec.extraEnv) or a future CLI default would otherwise drop the
                    // abandoned subagents' work without a trace.
                    if Self.isBackgroundWaitCutoff(text) {
                        await self?.reportBackgroundWaitCutoff()
                    }
                    if ProcessInfo.processInfo.environment["NUCLEIC_DEBUG_STDERR"] != nil {
                        FileHandle.standardError.write(Data(("claude-stderr: " + text + "\n").utf8))
                    }
                }
            }

            // 5. Optional raw native capture (OBSERVABILITY A.5).
            let capture = try makeCaptureFile()

            // 5b. Spawn watchdog (containerized runs only). The first event lands within seconds of
            // launch, so if NO stdout reaches the decode loop within `firstOutputTimeoutNanos`, the
            // agent started but its stdio transport stalled — surface a retryable error and kill the
            // wedged process so the run settles, instead of spinning "Working…" forever. Disarmed by
            // the first decoded line below; cancelled unconditionally after the loop.
            let watchdog: Task<Void, Never>? = (run.container != nil)
                ? Task { [weak self] in
                    try? await Task.sleep(nanoseconds: Self.firstOutputTimeoutNanos)
                    await self?.reportStalledSpawnIfSilent(stderrTail: stderrTail)
                }
                : nil

            // 6. Decode loop — the single consuming task of the native stream.
            let decoder = ClaudeStreamDecoder()
            // The exact host directory a `host_exec` command runs in for this run — the same
            // `workdir` the host-exec handler uses. Recorded onto each host_exec tool call below so
            // the transcript's in-chat host card resolves `$PWD` to precisely the directory the
            // command ran in, identically to the approval card. Nil when host_exec isn't exposed.
            let hostExecWorkdir = run.allowHostExec ? (run.container?.workdir ?? run.worktree) : nil
            for try await line in handle.stdoutLines {
                if !sawFirstOutput {
                    sawFirstOutput = true
                    watchdog?.cancel()  // output flows → the spawn isn't stalled
                }
                capture?.append(line: line)
                for decoded in decoder.decode(line: line) {
                    if case .sessionStarted(let started) = decoded.kind {
                        backendSessionID = started.backendSessionID
                    }
                    var kind = Self.recordingHostExecCwd(decoded.kind, workdir: hostExecWorkdir)
                    if case .runFinished(let finished) = kind {
                        sawRunFinished = true
                        // A SIGINT (manual Stop) makes the CLI emit an `is_error`
                        // result, which the decoder maps to `.errored`. That's a
                        // deliberate stop, not a failure — remap it so the session
                        // settles as interrupted ("Run stopped"), not errored.
                        let outcome: RunFinished.Outcome =
                            (interruptRequested && finished.outcome == .errored)
                            ? .interrupted : finished.outcome
                        // An armed `nucleic_monitor` means the turn ended to WAIT for that watch
                        // to fire, not because the agent is done or needs you. Tag it so the host
                        // marks the turn "Monitoring…" and suppresses the turn-end notification —
                        // otherwise every monitor tick's re-invoked turn would re-alert.
                        kind = .runFinished(
                            RunFinished(
                                outcome: outcome,
                                finalText: finished.finalText,
                                totalUsage: finished.totalUsage,
                                durationMs: finished.durationMs,
                                waitingOnBackground: !activeMonitors.isEmpty))
                    }
                    emit(kind, nativeType: decoded.nativeType)
                }
            }

            // 7. Exit handling: if the CLI died without a result event, synthesize
            //    the run outcome (ADAPTERS §1.5).
            watchdog?.cancel()
            let exitCode = await handle.wait()
            stderrTask.cancel()
            capture?.close()
            if !sawRunFinished {
                // A nonzero exit is only an error if we didn't ask the process to stop
                // (interrupt or session shutdown both end it deliberately). A stalled spawn already
                // emitted its own error and SIGKILLed the process, so don't also report that kill as
                // an abnormal/OOM exit.
                if exitCode != 0 && !interruptRequested && !terminating && !spawnStalled {
                    let message: String
                    // A SIGKILL inside the sandbox is ambiguous (real OOM vs. the whole container
                    // going down vs. another process) — so probe the container now and report what
                    // actually happened, rather than assuming "out of memory".
                    if BackendDiagnostics.isSIGKILL(exitCode),
                        let name = activeContainerName, let containerManager
                    {
                        let diagnosis = await containerManager.diagnoseContainerKill(name: name)
                        message = BackendDiagnostics.containerKillMessage(
                            tool: "claude", diagnosis: diagnosis, stderrTail: stderrTail.joined())
                    } else {
                        message = BackendDiagnostics.abnormalExitMessage(
                            tool: "claude", exitCode: exitCode, stderrTail: stderrTail.joined(),
                            containerized: run.container != nil)
                    }
                    emit(.error(AgentError(recoverable: false, message: message)), nativeType: nil)
                }
                let outcome: RunFinished.Outcome =
                    spawnStalled ? .errored
                    : interruptRequested ? .interrupted
                    : (terminating || exitCode == 0) ? .completed : .errored
                // Mirror the CLI-emitted path: a clean synthesized end with a monitor still armed
                // is a background wait, not a completion (see the `.runFinished` remap above).
                emit(
                    .runFinished(
                        RunFinished(outcome: outcome, waitingOnBackground: !activeMonitors.isEmpty)),
                    nativeType: nil)
            }
            await approvals.cancelOutstanding(reason: "Run ended")
            // Tear down this run's per-instance state BEFORE finishing the stream: the
            // consumer (SessionController) starts the next turn the moment the stream
            // closes, and start/resume reuse this same actor. Cleaning up first means a
            // follow-up turn can never overlap and clobber the previous run's handle,
            // bearer token, or container reference.
            await teardownRun()
            continuation.finish()
        } catch {
            // A SIGINT interrupt (Stop) or a shutdown can abort an in-flight stdout
            // read mid-turn — that's a deliberate stop, not a run failure. Finish the
            // stream cleanly with the right terminal outcome so a conversational
            // session settles to `awaitingInput` (resumable) instead of being thrown
            // as an error, which `SessionController.consume` would turn into a terminal
            // `.error` and lock the composer.
            if interruptRequested || terminating || spawnStalled {
                if !sawRunFinished {
                    let outcome: RunFinished.Outcome =
                        spawnStalled ? .errored
                        : interruptRequested ? .interrupted : .completed
                    emit(.runFinished(RunFinished(outcome: outcome)), nativeType: nil)
                }
                await approvals.cancelOutstanding(reason: "Run interrupted")
                await teardownRun()
                continuation.finish()
            } else if Self.isStdioStartAbort(error) {
                // The container aborted the launch because an agent stdio stream never connected from
                // the guest (vendored containerization patch #6 — atomic stdio-or-abort start). The
                // run never began, so this is transient and safe to retry. Surface it as a
                // *recoverable* error and settle the session as resumable — exactly like a spawn stall
                // — rather than a hard failure that locks the composer.
                emit(
                    .error(AgentError(recoverable: true, message: Self.stdioStartAbortMessage)),
                    nativeType: nil)
                if !sawRunFinished {
                    emit(.runFinished(RunFinished(outcome: .errored)), nativeType: nil)
                }
                await approvals.cancelOutstanding(reason: "Run failed to start")
                await teardownRun()
                continuation.finish()
            } else {
                // A genuine failure still has to release any approval waiter blocked on this
                // run, or its MCP serve task (and the connection it holds) leaks until the
                // whole session shuts down. Deny-resolving it returns the agent's blocked
                // call; the UI side is dismissed independently by the consumer at run end.
                await approvals.cancelOutstanding(reason: "Run failed")
                await teardownRun()
                continuation.finish(throwing: error)
            }
        }
    }

    /// Release everything the just-finished run held: the approval-server bearer token,
    /// the sandbox container (so its idle timer can arm), and the process handle.
    private func teardownRun() async {
        if let serverToken {
            // Unregister only this run's token; a shared control-container server keeps serving the
            // other sessions in the box. The server itself is never stopped here.
            await activeServer.unregister(token: serverToken)
            self.serverToken = nil
        }
        runServer = nil
        if let name = activeContainerName {
            await containerManager?.finished(name: name)
            activeContainerName = nil
        }
        self.handle = nil
    }

    /// Spawn-watchdog handler (containerized runs). Fires `firstOutputTimeoutNanos` after launch; if
    /// the agent STILL hasn't produced any stdout, the spawn has stalled. The CLI's first event only
    /// comes after it finishes MCP init against the approval server, so total silence this long
    /// almost always means the container's CONTROL PLANE is wedged (the bridge is down, or a relay
    /// hop accepts connections it never answers) and the CLI is hanging in MCP connect. Probe that
    /// path end-to-end and say which hop is dead rather than guessing; when the probe reports a
    /// dead control plane, QUEUE a recovery (`requestControlPlaneRecovery`) — which waits for the
    /// container to go idle, re-probes, and only then restarts — so a genuinely bricked box heals
    /// without an app restart, while a false alarm under load never SIGKILLs the other sessions
    /// sharing the container. Convert the dead end into an actionable, *recoverable* error (the
    /// session stays resumable) and SIGKILL the wedged process group so its stdio closes, the
    /// decode loop ends, and the run settles. No-op if output has since arrived or the run is
    /// already interrupting/terminating/finished.
    private func reportStalledSpawnIfSilent(stderrTail: StderrTail) async {
        guard !sawFirstOutput, !sawRunFinished, !interruptRequested, !terminating,
            let handle
        else { return }
        spawnStalled = true
        var message =
            "The agent started inside the container but produced no output within "
            + "\(Self.firstOutputTimeoutNanos / 1_000_000_000)s."
        let tail = stderrTail.joined()
        if !tail.isEmpty { message += "\n\nLast container stderr:\n\(tail)" }
        var controlPlaneDead = false
        if let name = activeContainerName, let containerManager {
            let diagnosis = await containerManager.diagnoseContainerKill(name: name)
            if !diagnosis.containerRunning {
                message += "\n\n(The app's container runtime no longer tracks this container — it "
                    + "was stopped, restarted, or removed. Run the turn again to start a fresh one.)"
            } else if diagnosis.probeUnanswered {
                controlPlaneDead = true
                message += "\n\n(The container's VM did not answer a status probe even after a "
                    + "retry — it likely crashed or is severely wedged. A recovery restart has "
                    + "been queued for when the container is idle; run the turn again.)"
            } else {
                switch await containerManager.probeControlPlane(name: name) {
                case .responsive:
                    message += "\n\n(The container and its control plane are healthy — the agent "
                        + "process itself never wrote anything. This may be transient; run the "
                        + "turn again.)"
                case .bridgeUnreachable(let detail):
                    controlPlaneDead = true
                    message += "\n\n(The container's control bridge is not accepting connections "
                        + "(\(detail)), so the agent could never reach the approval server and "
                        + "hung before its first event. A recovery restart has been queued for "
                        + "when the container is idle — no other session is interrupted. Run the "
                        + "turn again.)"
                case .unresponsive:
                    controlPlaneDead = true
                    message += "\n\n(The container's control plane accepts connections but never "
                        + "answers — a wedged relay. The agent hung waiting on it before its "
                        + "first event. A recovery restart has been queued for when the container "
                        + "is idle — no other session is interrupted. Run the turn again.)"
                case .probeFailed(let why):
                    message += "\n\n(The container is still running, but the control-plane probe "
                        + "couldn't run: \(why). Run the turn again.)"
                }
                if let kills = diagnosis.oomKills, kills > 0 {
                    message += " (Kernel OOM kills inside this container so far: \(kills).)"
                }
            }
        } else {
            message += " This is usually transient; run the turn again."
        }
        emit(.error(AgentError(recoverable: true, message: message)), nativeType: nil)
        // Force the wedged process group down, then finish the line streams host-side so the decode
        // loop ends (EOF) and the run settles even if the guest's wait/stdio RPC is itself stalled.
        handle.sendSignal(SIGKILL)
        handle.forceCloseStreams()
        // Queue recovery off-actor so the run settles immediately. `requestControlPlaneRecovery`
        // NEVER kills live work: it dedups concurrent requests, waits for the container to go idle
        // (this run's own busy ref releases when the run settles moments from now), re-probes with
        // retries, and only restarts a container that is idle AND still dead — a false alarm under
        // load costs nothing and touches no other session.
        if controlPlaneDead, let name = activeContainerName, let containerManager {
            Task.detached { await containerManager.requestControlPlaneRecovery(name: name) }
        }
    }

    /// Whether a stderr line is the `claude -p` background-wait cutoff warning — the CLI announcing it
    /// gave up waiting on still-running background subagents/tasks and is terminating them (so their
    /// work is lost). Matched on the two stable fragments of the CLI's message ("Background tasks still
    /// running …; terminating.") rather than the whole sentence, so a reword of the surrounding text
    /// (e.g. the env-var hint) doesn't silently disable detection. See ``printBackgroundWaitEnv``.
    static func isBackgroundWaitCutoff(_ line: String) -> Bool {
        line.contains("Background tasks still running") && line.contains("terminating")
    }

    /// User-facing copy for a launch aborted because the agent's stdio channel never established (the
    /// vendored containerization atomic stdio-or-abort start). The run never began, so it's retryable.
    static let stdioStartAbortMessage =
        "The container couldn't establish the agent's I/O channel, so the turn didn't start. "
        + "This is usually transient — run the turn again."

    /// Whether `error` is the container's atomic stdio-or-abort launch failure (vendored
    /// containerization patch #6). Matched on the stable sentinel phrase the framework throws
    /// ("stream never connected from the guest") — keep the two in lockstep if either is reworded.
    static func isStdioStartAbort(_ error: Error) -> Bool {
        String(describing: error).contains("stream never connected from the guest")
    }

    /// Surface the `claude -p` background-wait cutoff as a *recoverable* error in the transcript: the
    /// CLI stopped waiting on still-running background subagents/tasks and killed them, so their work
    /// never made it into this turn. Recoverable (the session stays resumable) and lifecycle-neutral —
    /// like ``reportStalledSpawnIfSilent(stderrTail:)``, the sibling out-of-band surfacing helper.
    /// Should be unreachable while we pin the ceiling to 0 (wait indefinitely); it exists so a finite
    /// ``RunSpec/extraEnv`` override — or a future CLI default — can't drop the work without a trace.
    /// Emitted at most once per run.
    private func reportBackgroundWaitCutoff() {
        guard !reportedBackgroundWaitCutoff else { return }
        reportedBackgroundWaitCutoff = true
        emit(
            .error(
                AgentError(
                    recoverable: true,
                    message:
                        "A background subagent or task was still running when this turn ended, and "
                        + "Claude stopped waiting for it — so its work was not incorporated into this "
                        + "turn. Re-run the turn (or raise CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS) to let "
                        + "long-running subagents finish.")),
            nativeType: nil)
    }

    // MARK: - Approval bridge

    private func handleApprovalCall(_ call: MCPApprovalServer.ApprovalCall) async
        -> MCPApprovalServer.Reply
    {
        // Provider-native asynchronous tools feed a synthetic task notification back into the
        // still-open `claude -p` process. If the model has already delivered what reads as its final
        // answer, that late notification starts another model pass and makes the session appear to
        // restart by itself. Nucleic has an explicit, lifecycle-aware cross-turn mechanism
        // (`nucleic_monitor`) for work that genuinely needs to wake a later turn; keep the provider's
        // implicit callbacks out of the single-shot turn altogether. This must run before cached
        // always-rules / Auto, so neither can re-enable the incompatible path.
        if let denial = Self.nativeBackgroundToolDenial(
            toolName: call.toolName, input: call.input)
        {
            return .deny(message: denial)
        }

        // Block the provider-native `Monitor` tool. It can't work under Nucleic's single-shot turn
        // model: each turn is one foreground `claude -p` process that exits at end of turn, so there
        // is no harness loop to re-invoke the agent when a monitor fires — and because Nucleic pins
        // the print background-wait ceiling to "wait indefinitely", a *persistent* native monitor
        // would hang the turn forever at wind-down. Steer to `nucleic_monitor`, whose watch the host
        // owns across turns and delivers as fresh turns. Fall back to foreground-wait guidance if the
        // replacement isn't wired for this session.
        if call.toolName == "Monitor" {
            if monitorEventSink != nil {
                return .deny(message:
                    "The native Monitor tool doesn't work here: this environment runs each turn as a "
                        + "single foreground process, so nothing re-invokes you when a monitor fires "
                        + "(and a persistent one would hang the turn). Use Nucleic's `nucleic_monitor` "
                        + "tool instead — pass the same `command` plus a short `description`; it runs "
                        + "the watch outside your turn and delivers each batch of output to you as a "
                        + "new turn in this chat, waking you exactly as you expect. Arm it and end "
                        + "your turn; you don't need to wait.")
            }
            return .deny(message:
                "The native Monitor tool doesn't work here: this environment runs each turn as a "
                    + "single foreground process, so nothing re-invokes you when a monitor fires. If "
                    + "you need to wait for an external event, wait for it *within this turn* — e.g. a "
                    + "foreground shell loop that sleeps and re-checks. Don't use background Bash "
                    + "or end your turn expecting to be resumed automatically.")
        }

        if blockNativeSubagentSpawns, call.toolName == "Task" || call.toolName == "Agent" {
            return .deny(message:
                "Subagent spawning must go through Nucleic's `nucleic_subagent` tool so Nucleic can "
                    + "run the worker as an observed session (shown in the Subagents panel) and enforce "
                    + "the configured worker model. Call `nucleic_subagent` with a short task label and "
                    + "a complete, self-contained worker prompt — it works just like the Task tool.")
        }

        // Raw-shell writes take real locks instead of being refused (docs/NASH.md §5.2). A shell
        // command that authors a workspace file (`echo … > f`, `cat > f`, `sed -i`, `tee`) used to
        // be denied outright, because it emits no `fileChange` and so skipped the conflict lock and
        // the per-edit trunk commit. Now the paths it writes are named up front and arbitrated
        // exactly like an `Edit`'s — contention queues the command, a defer/cancel denies it — and
        // nash reports the write afterwards so it lands in the trunk like any other edit. Only a
        // target too dynamic to name (a `patch`, a glob, an unexpanded `$VAR`) is still refused.
        // Runs ahead of the always-rule cache and the auto-approve blanket so neither an "allow
        // always" rule nor auto mode can slip an *unlocked* mutation through.
        if call.toolName == "Bash", let command = call.input["command"]?.stringValue {
            // Without a coordinator (tests / early startup) there is no lock to take, so the
            // unlockable-target refusal is all that's left — it is a rule about the command, not
            // about contention, and must not fail open with the arbitration.
            let clearance: ShellWriteClearance
            if let conflictCoordinator {
                clearance = await conflictCoordinator.arbitrateShellWrites(
                    sessionID: sessionID ?? SessionID(rawValue: "unknown"), command: command,
                    workdir: agentShellWorkdir, toolUseID: call.toolUseID)
            } else if let reason = WorktreeMutationGuard.denialReason(command: command) {
                clearance = .deny(reason)
            } else {
                clearance = .clear
            }
            if case .deny(let reason) = clearance { return .deny(message: reason) }
        }

        // Conflict pre-check (internal project management): edit-class tools are routed here
        // even in auto mode. Before the edit lands, ask Nucleic whether the target file
        // overlaps another active agent's work. Runs ahead of the always-rule cache so an
        // "always allow" rule can't slip a conflicting edit past.
        let editPaths = RiskClassifier.editedPaths(toolName: call.toolName, input: call.input)
        if let conflictCoordinator, !editPaths.isEmpty {
            // Arbitrate EVERY edit-class call so each touched file acquires its own lock — not
            // just the first edit of the turn. (A prior per-turn `conflictOverridden` latch
            // suppressed arbitration after the first grant, leaving every later file unlocked
            // and free for a parallel session to edit, which surfaced as autoship merge
            // conflicts.) Re-arbitrating is cheap: a file this session already holds re-grants
            // instantly via the LockManager fast-path, and a contended file correctly blocks.
            let outcome = await conflictCoordinator.arbitrate(
                sessionID: sessionID ?? SessionID(rawValue: "unknown"), task: "", files: editPaths,
                toolUseID: call.toolUseID)
            switch outcome.resolution {
            case .deferred:
                return .deny(message:
                    "This edit conflicts with another active Nucleic agent and was added to the "
                        + "to-do list. Stop now and do not edit — tell the user it was deferred.")
            case .cancelled:
                return .deny(message:
                    "This edit conflicts with another active Nucleic agent and the user cancelled "
                        + "it. Stop now and do not edit.")
            case .proceed, .noConflict:
                // Lock granted (or nothing to lock). If update-on-grant pulled a parent change into a
                // file this edit targets, block THIS edit once with the re-ground notice (the scoped
                // diff) so the agent re-issues against fresh content; the lock is held, so the retry
                // re-grants instantly (no merge) and proceeds (LOCKING §4.5). Otherwise let it through.
                if let notice = outcome.refreshNotice { return .deny(message: notice) }
                break
            }
        }

        // `AskUserQuestion` and `ExitPlanMode` collect deliberate user decisions rather than
        // merely gating an action. The former needs choices folded into `updatedInput`; the latter
        // must let the user accept, reject, or return revision feedback. Neither may be answered by
        // an old "allow always" rule or by Auto mode.
        let requiresExplicitUserDecision = NucleicApprovalPolicy.requiresExplicitUserDecision(
            toolName: call.toolName)

        // Always-rule short-circuit: auto-answer without surfacing (RUNTIME §5).
        if !requiresExplicitUserDecision,
            let cached = await approvals.cachedDecision(toolName: call.toolName, input: call.input)
        {
            return ClaudeDecisionMapping.reply(for: cached.decision, originalInput: call.input)
        }

        // Nucleic-managed Auto uses the same risk policy as Codex and ACP. Destructive,
        // unknown, and user-input-collection calls still surface explicitly.
        if NucleicApprovalPolicy.shouldAutoApprove(
            autoApprove: autoApprove,
            risk: RiskClassifier.classify(toolName: call.toolName, input: call.input),
            requiresExplicitUserDecision: requiresExplicitUserDecision)
        {
            return ClaudeDecisionMapping.reply(for: .allow(), originalInput: call.input)
        }

        let request = ApprovalRequest(
            id: .generate(),
            sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            toolCallID: call.toolUseID,
            toolName: call.toolName,
            input: call.input,
            title: RiskClassifier.title(toolName: call.toolName, input: call.input),
            risk: RiskClassifier.classify(toolName: call.toolName, input: call.input),
            createdAt: configuration.now())

        emit(.approvalRequested(request), nativeType: "mcp/tools/call")
        let resolved = await approvals.waitForDecision(request)
        emit(.approvalResolved(resolved), nativeType: nil)

        if case .cancelRun = resolved.decision {
            interruptRequested = true
            handle?.sendSignal(SIGINT)
        }
        return ClaudeDecisionMapping.reply(for: resolved.decision, originalInput: call.input)
    }

    /// Reject provider-native tools whose completion is delivered as an unsolicited message to the
    /// model. Claude keeps `-p` alive for those tasks after an apparent final answer; a late callback
    /// then triggers another inference pass, which users experience as a session spontaneously
    /// restarting. Foreground Bash remains available, as does Nucleic's explicit `nucleic_monitor`
    /// bridge for a deliberate cross-turn wake.
    static func nativeBackgroundToolDenial(toolName: String, input: JSONValue) -> String? {
        let isBackgroundBash =
            toolName == "Bash" && input["run_in_background"]?.boolValue == true
        let isScheduledCallback = toolName == "ScheduleWakeup" || toolName == "CronCreate"
        guard isBackgroundBash || isScheduledCallback else { return nil }

        return "This provider-native background callback is disabled in Nucleic because its late "
            + "completion message can re-invoke the model after it has already answered, making the "
            + "session appear to restart itself. Keep the work in the foreground if this turn must "
            + "wait for it. If the work should outlive this turn, use `nucleic_monitor`; it is "
            + "Nucleic's lifecycle-aware cross-turn wake mechanism."
    }

    // MARK: - Ask-the-user bridge (nucleic_ask_user)

    /// Put the agent's multiple-choice questions in front of the human and block until they answer —
    /// Nucleic's provider-neutral `AskUserQuestion`.
    ///
    /// It deliberately raises an approval whose `toolName` is `AskUserQuestion`, because that is what
    /// every one of Nucleic's question surfaces keys on: the Mac `AskUserQuestionBar` wizard, the iOS
    /// card, the "Question(s) Asked" status, and the settled transcript card. So a Codex or Grok
    /// orchestrator's question renders exactly like Claude's built-in one, with no new UI.
    ///
    /// Unlike a permission gate this can never be auto-answered: the handler goes straight to
    /// `waitForDecision` without consulting the always-rule cache or auto-approve policy (which
    /// `NucleicApprovalPolicy.requiresExplicitUserDecision` excludes `AskUserQuestion` from anyway), so
    /// a question always reaches a person. The answers ride back in the decision's `updatedInput` — the
    /// same channel the Claude CLI expects — and are read out of it here.
    private func handleAskUserCall(_ call: MCPApprovalServer.AskUserCall) async -> AskUserResult {
        let request = ApprovalRequest(
            id: .generate(),
            sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            toolCallID: nil,
            toolName: AskUserQuestion.toolName,
            input: call.input,
            title: call.questions.first?.question ?? "The agent has a question",
            // Answering a question changes nothing on its own; the risk of the *work* is gated where
            // that work happens.
            risk: .readOnly,
            createdAt: configuration.now())

        emit(.approvalRequested(request), nativeType: "mcp/tools/call")
        let resolved = await approvals.waitForDecision(request)
        emit(.approvalResolved(resolved), nativeType: nil)

        switch resolved.decision {
        case .deny(let reason):
            return .declined(
                message: reason ?? "The user dismissed the question; proceed on your best judgment.")
        case .cancelRun:
            interruptRequested = true
            handle?.sendSignal(SIGINT)
            return .declined(message: "The user cancelled the run.")
        case .allow(let updatedInput):
            guard let updatedInput else {
                return .declined(
                    message: "The user allowed the question without choosing an answer; proceed on "
                        + "your best judgment.")
            }
            let answers = Self.askUserAnswers(from: updatedInput, questions: call.questions)
            return answers.isEmpty
                ? .declined(
                    message: "The user answered without selecting an option; proceed on your best "
                        + "judgment.")
                : .answered(answers: answers)
        case .allowAlways:
            // No surface offers "always" for a question, but if one ever did there is nothing to read.
            return .declined(
                message: "No answer was recorded for the question; proceed on your best judgment.")
        }
    }

    /// Read the user's picks out of an answered `AskUserQuestion` payload, in the order the questions
    /// were asked. Mirrors `AskUserQuestion.updatedInput`: `answers` is keyed by the exact question
    /// text, with a string for a single-select and an array for a `multiSelect`. Questions the user
    /// left blank are omitted.
    static func askUserAnswers(
        from updatedInput: JSONValue, questions: [AskUserQuestion.Question]
    ) -> [(question: String, labels: [String])] {
        guard let answers = updatedInput["answers"] else { return [] }
        return questions.compactMap { question in
            guard let recorded = answers[question.question] else { return nil }
            let labels: [String]
            if let array = recorded.arrayValue {
                labels = array.compactMap { $0.stringValue }.filter { !$0.isEmpty }
            } else if let single = recorded.stringValue, !single.isEmpty {
                labels = [single]
            } else {
                labels = []
            }
            return labels.isEmpty ? nil : (question: question.question, labels: labels)
        }
    }

    // MARK: - Monitor bridge (nucleic_monitor)

    /// Arm a `nucleic_monitor` watch: spawn `command` in the same environment as the agent's Bash
    /// tool (the session container, else the host worktree), stream its stdout, and — since the
    /// process is owned by this actor rather than the ephemeral CLI turn — keep it running across
    /// turns. Each batch of output is delivered back to the session as a fresh turn via
    /// ``monitorEventSink``, which is the cross-turn re-invocation the native `Monitor` can't do.
    /// Returns immediately with an armed confirmation; it never blocks the turn.
    private func handleMonitorCall(_ call: MCPApprovalServer.MonitorCall) async -> String {
        guard monitorEventSink != nil, let ctx = monitorSpawnContext else {
            return "Monitoring isn't available in this session."
        }
        // Exactly one source: a shell `command` OR a `ws` WebSocket URL.
        let command = call.command.trimmingCharacters(in: .whitespacesAndNewlines)
        let wsURL = call.wsURL?.trimmingCharacters(in: .whitespacesAndNewlines)
        let hasCommand = !command.isEmpty
        let hasWS = !(wsURL?.isEmpty ?? true)
        guard hasCommand != hasWS else {
            return "Pass exactly one source: `command` (a shell command) OR `ws` (a WebSocket URL)."
        }
        guard activeMonitors.count < Self.maxConcurrentMonitors else {
            return "Too many monitors are already running (max \(Self.maxConcurrentMonitors)). "
                + "Let one finish before arming another."
        }
        if hasWS {
            return await armWebSocketMonitor(
                urlString: wsURL!, protocols: call.wsProtocols, description: call.description,
                persistent: call.persistent, timeoutMs: call.timeoutMs)
        }
        let description: String = {
            let d = call.description.trimmingCharacters(in: .whitespacesAndNewlines)
            return d.isEmpty ? Self.hostCommandLabel(command) : d
        }()
        // Hold an activity ref on the session container so it doesn't idle-stop after the arming turn
        // ends and kill the watch's exec'd process. Balanced once when the monitor stops. Host runs
        // need no pin (the process is a direct child of this app).
        var pinnedContainerName: String? = nil
        if let cspec = ctx.cspec, let cm = containerManager,
            let result = try? await cm.ensureRunning(cspec)
        {
            pinnedContainerName = result.name
        }
        let handle: any ProcessHandle
        do {
            handle = try await spawnMonitorProcess(command: command, ctx: ctx)
        } catch {
            if let name = pinnedContainerName { await containerManager?.finished(name: name) }
            return "Couldn't start the monitor: \(error)"
        }
        monitorSeq += 1
        let id = "mon-\(monitorSeq)"
        let run = MonitorRun(
            source: .process(handle), isWebSocket: false, description: description,
            pinnedContainerName: pinnedContainerName)
        run.task = Task { [weak self] in await self?.pumpMonitor(id: id, handle: handle) }
        activeMonitors[id] = run
        armMonitorDeadline(id: id, persistent: call.persistent, timeoutMs: call.timeoutMs)
        let bound = call.persistent
            ? "It runs until the command exits or the session ends"
            : "It runs until the command exits, the session ends, or — at the earliest once past the "
                + "timeout — the command goes idle with near-zero CPU and disk I/O (a command that is "
                + "still working past the timeout keeps running)"
        return "Monitor armed (\(id)): watching \"\(description)\". Each batch of output arrives as a "
            + "new turn in this chat — you don't need to wait, just end your turn and you'll be "
            + "re-invoked when it fires. \(bound)."
    }

    /// Arm a WebSocket-source monitor: open the socket and stream each inbound text frame as an
    /// event (binary frames become a placeholder line), delivered through the same coalesce/cap
    /// path as a command monitor. Darwin-only — see ``MonitorSource``; the Linux runner gets a
    /// clear fallback to a `command` source.
    private func armWebSocketMonitor(
        urlString: String, protocols: [String], description rawDescription: String,
        persistent: Bool, timeoutMs: Int?
    ) async -> String {
        #if canImport(Darwin)
        guard let url = URL(string: urlString), let scheme = url.scheme?.lowercased(),
            scheme == "ws" || scheme == "wss"
        else {
            return "Invalid WebSocket URL — pass a ws:// or wss:// URL in `ws.url`."
        }
        let description: String = {
            let d = rawDescription.trimmingCharacters(in: .whitespacesAndNewlines)
            return d.isEmpty ? (url.host ?? "websocket") : d
        }()
        let task = protocols.isEmpty
            ? URLSession.shared.webSocketTask(with: url)
            : URLSession.shared.webSocketTask(with: url, protocols: protocols)
        task.resume()
        monitorSeq += 1
        let id = "mon-\(monitorSeq)"
        let run = MonitorRun(
            source: .webSocket(task), isWebSocket: true, description: description,
            pinnedContainerName: nil)
        run.task = Task { [weak self] in await self?.pumpMonitorWebSocket(id: id) }
        activeMonitors[id] = run
        armMonitorDeadline(id: id, persistent: persistent, timeoutMs: timeoutMs)
        let bound = persistent
            ? "It runs until the socket closes or the session ends"
            : "It runs until the socket closes, the session ends, or — at the earliest once past the "
                + "timeout — the socket goes quiet for that same window (a socket still pushing frames "
                + "past the timeout keeps running)"
        return "Monitor armed (\(id)): watching WebSocket \"\(description)\". Each text frame arrives "
            + "as a new turn in this chat — you don't need to wait, just end your turn and you'll be "
            + "re-invoked when it fires. \(bound)."
        #else
        return "WebSocket monitors aren't supported on this host. Use a `command` source instead "
            + "(e.g. `websocat \(urlString)` if available, or a poll loop)."
        #endif
    }

    /// Balance a monitor's container activity ref exactly once, re-arming that container's idle timer.
    private func releaseMonitorContainer(_ run: MonitorRun) async {
        guard let name = run.pinnedContainerName, !run.containerReleased else { return }
        run.containerReleased = true
        await containerManager?.finished(name: name)
    }

    /// Spawn a monitor command as a login-shell child, in the session container (over the same vsock
    /// exec path the agent uses) or on the host worktree. The returned handle streams stdout the pump
    /// consumes; the process is a child of this app (not the ephemeral CLI), so it survives the turn.
    private func spawnMonitorProcess(command: String, ctx: MonitorSpawnContext) async throws
        -> any ProcessHandle
    {
        if let name = ctx.containerName, let containerManager {
            return try await containerManager.exec(
                name: name, workdir: ctx.workdir, env: ctx.env,
                argv: CommandInterceptor.agentShellArgv(command), uid: ctx.uid, gid: ctx.gid)
        }
        // The host twin of the container branch's `agentShellArgv` (docs/NASH.md §7.7.1): same
        // helper as `runOnHost`, so the two monitor arms stay symmetric. Observation env rides
        // along only when nash is actually picked (zsh would ignore it).
        let argv = CommandInterceptor.hostShellArgv(command)
        var env = ctx.env
        if argv[0] != "/bin/zsh", let sid = sessionID {
            if let endpoint = hostControlEndpoint {
                env.merge(
                    CommandInterceptor.hookEnv(
                        host: endpoint.host, port: endpoint.port, token: endpoint.token,
                        sessionID: sid, commandTracing: false,
                        shellCapture: HostShellSettings.shellCapture, environment: .host)
                ) { _, new in new }
            } else if let socket = hostShellSocketPath, let token = serverToken {
                env["NUCLEIC_SHELL_SOCKET"] = socket
                env["NUCLEIC_HOOK_TOKEN"] = token
                env["NUCLEIC_SESSION_ID"] = sid.rawValue
                env["NUCLEIC_SHELL_CAPTURE"] = HostShellSettings.shellCapture
                env["NUCLEIC_SHELL_ENVIRONMENT_KIND"] = NashEnvironment.host.kind.rawValue
                env["NUCLEIC_SHELL_ENVIRONMENT_ID"] = NashEnvironment.host.id
                env["NUCLEIC_SHELL_ENVIRONMENT_LABEL"] = NashEnvironment.host.label
            }
        }
        let spec = ProcessSpec(
            executable: argv[0],
            args: Array(argv.dropFirst()),
            cwd: ctx.workdir,
            env: env,
            stdinMode: .closed,
            removeFromEnvironment: ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "CODEX_API_KEY"])
        return try await processHost.launch(spec)
    }

    /// Arm a bounded monitor's age timer. The age NEVER stops the watch by itself — when it lapses,
    /// ``beginMonitorStallWatch(id:ageNanos:)`` starts looking for evidence the source is actually
    /// wedged, and only that evidence reaps it. A persistent monitor arms nothing.
    ///
    /// The timer is owned by the ``MonitorRun`` (not the pump) so every end path cancels it through the
    /// same claim, and a cancelled timer can never race a natural exit into a second closing note.
    private func armMonitorDeadline(id: String, persistent: Bool, timeoutMs: Int?) {
        guard !persistent, let run = activeMonitors[id] else { return }
        let ageNanos = UInt64(min(max(timeoutMs ?? 300_000, 1_000), 3_600_000)) * 1_000_000
        run.deadlineTask = Task { [weak self] in
            try? await Task.sleep(nanoseconds: ageNanos)
            // `try?` swallows the cancellation error, so a cancelled timer would otherwise fall
            // straight through and stop a monitor that already ended on its own.
            if Task.isCancelled { return }
            await self?.beginMonitorStallWatch(id: id, ageNanos: ageNanos)
        }
    }

    /// A bounded monitor is now past its age. Decide whether it is wedged or merely slow, and reap it
    /// only in the first case — "past an age" alone is not evidence of a hang.
    ///
    /// For a process source this hands the verdict to ``ProcessStallMonitor``, which requires the tree
    /// to be output-silent AND under both the CPU and disk-I/O thresholds continuously before it fires;
    /// a single busy sample resets that window, so a slow build keeps running. When the pid is
    /// host-samplable the confirmation window is ``monitorStallIdleNanos``; when it isn't (container /
    /// VM handles, whose guest pid can't be sampled from here) output-idle is the only signal, so the
    /// bar is the full age window of complete silence instead of a short one.
    ///
    /// A WebSocket source has no process to sample: its liveness is frame arrival, so it re-checks idle
    /// time against the same window.
    private func beginMonitorStallWatch(id: String, ageNanos: UInt64) {
        guard let run = activeMonitors[id], !run.stopped else { return }
        switch run.source {
        case .process(let handle):
            let samplable = handle.supportsHostCPUSampling
            let cpuSampler: (@Sendable () -> UInt64)?
            let ioSampler: (@Sendable () -> UInt64)?
            if samplable {
                let pid = handle.processID
                cpuSampler = { hostProcessTreeCPUNanos(of: pid) }
                ioSampler = { hostProcessTreeIOBytes(of: pid) }
            } else {
                cpuSampler = nil
                ioSampler = nil
            }
            let monitor = ProcessStallMonitor(
                handle: handle,
                config: ProcessStallConfig(
                    idleThresholdNanos: samplable ? Self.monitorStallIdleNanos : ageNanos),
                cpuSampler: cpuSampler,
                ioSampler: ioSampler,
                onStall: { [weak self] _ in await self?.stopMonitor(id: id, reason: .stalled) },
                // Recovery needs no action: the monitor re-arms itself and fires again if the
                // command goes quiet for good.
                onResolve: { _ in })
            run.stallMonitor = monitor
            monitor.start()
        #if canImport(Darwin)
        case .webSocket:
            run.deadlineTask = Task { [weak self] in
                await self?.watchWebSocketIdle(id: id, windowNanos: ageNanos)
            }
        #endif
        }
    }

    #if canImport(Darwin)
    /// Reap a past-its-age WebSocket monitor once — and only once — the socket has been silent for a
    /// full window. A feed still pushing frames is live work and is left alone; each frame pushes the
    /// decision out by another window.
    private func watchWebSocketIdle(id: String, windowNanos: UInt64) async {
        while true {
            guard let run = activeMonitors[id], !run.stopped else { return }
            let idle = DispatchTime.now().uptimeNanoseconds &- run.lastEventNanos
            if idle >= windowNanos {
                await stopMonitor(id: id, reason: .stalled)
                return
            }
            try? await Task.sleep(nanoseconds: windowNanos &- idle)
            if Task.isCancelled { return }
        }
    }
    #endif

    /// Consume a monitor's stdout until the command exits, coalescing lines into batched deliveries.
    /// Emits a closing note when the watch ends so the agent isn't left expecting more events from a
    /// monitor that's already done. The bounded-watch deadline is owned by the run, not this pump.
    private func pumpMonitor(id: String, handle: any ProcessHandle) async {
        do {
            for try await line in handle.stdoutLines {
                // Over the event cap → the monitor already stopped itself; stop pumping.
                let keepGoing = await appendMonitorLine(
                    id: id, text: String(decoding: line, as: UTF8.self))
                if !keepGoing { return }
            }
        } catch {
            // Stream ended abnormally (process died / container torn down) — fall through to finish.
        }
        await finishMonitor(id: id, reason: .exited)
    }

    #if canImport(Darwin)
    /// Consume a monitor's WebSocket frames until the socket closes (or the bounded timeout fires),
    /// delivering each text frame through the same coalesce/cap path as stdout lines. Binary frames
    /// become a placeholder line. The task is read from the monitor's state (never captured across
    /// the concurrency boundary — `URLSessionWebSocketTask` isn't `Sendable`).
    private func pumpMonitorWebSocket(id: String) async {
        guard let run = activeMonitors[id], case .webSocket(let task) = run.source else { return }
        while true {
            do {
                let message = try await task.receive()
                let text: String
                switch message {
                case .string(let string): text = string
                case .data(let data): text = "[binary frame, \(data.count) bytes]"
                @unknown default: text = "[unknown frame]"
                }
                let keepGoing = await appendMonitorLine(id: id, text: text)
                if !keepGoing { return }
            } catch {
                break  // socket closed or errored — end the watch
            }
        }
        await finishMonitor(id: id, reason: .exited)
    }
    #endif

    /// Kill a monitor's underlying source — tree-kill a process, or cancel a WebSocket.
    private func terminateMonitorSource(_ run: MonitorRun) async {
        switch run.source {
        case .process(let handle): await handle.terminateTree()
        #if canImport(Darwin)
        case .webSocket(let task): task.cancel(with: .goingAway, reason: nil)
        #endif
        }
    }

    /// Append one stdout line to a monitor's pending batch and schedule a coalesced flush. Enforces
    /// the per-monitor event cap (auto-stops an over-chatty watch, like the native tool). Returns
    /// `false` once the monitor has stopped, so the pump loop ends.
    private func appendMonitorLine(id: String, text: String) async -> Bool {
        guard let run = activeMonitors[id], !run.stopped else { return false }
        run.delivered += 1
        run.lastEventNanos = DispatchTime.now().uptimeNanoseconds
        if run.delivered > Self.maxMonitorEvents {
            guard let run = claimMonitor(id: id) else { return false }
            await terminateMonitorSource(run)
            await releaseMonitorContainer(run)
            await deliverMonitorNotice(
                "[monitor \(id) (\(run.description)) stopped: it produced more than "
                    + "\(Self.maxMonitorEvents) events. If you still need this signal, re-arm "
                    + "nucleic_monitor with a tighter filter so it emits only the lines you'd act on.]",
                label: run.description)
            return false
        }
        run.pending.append(text)
        if !run.flushScheduled {
            run.flushScheduled = true
            Task { [weak self] in
                try? await Task.sleep(nanoseconds: Self.monitorCoalesceNanos)
                await self?.flushMonitor(id: id)
            }
        }
        return true
    }

    /// Deliver a monitor's accumulated lines as one turn, resetting the coalesce window.
    private func flushMonitor(id: String) async {
        guard let run = activeMonitors[id], !run.stopped else { return }
        run.flushScheduled = false
        let lines = run.pending
        run.pending.removeAll()
        guard !lines.isEmpty else { return }
        await deliverMonitorTurn(label: run.description, lines: lines)
    }

    /// Send one monitor wake-up: the agent is re-invoked with `lines`, preceded by any administrative
    /// notices that were parked waiting for a ride (see ``deliverMonitorNotice(_:label:)``). Every
    /// monitor delivery goes through here, so a parked notice always leaves on the next real event
    /// instead of buying a turn of its own.
    private func deliverMonitorTurn(label: String, lines: [String]) async {
        guard let sink = monitorEventSink else { return }
        let parked = takePendingMonitorNotices()
        let all = parked + lines
        guard !all.isEmpty else { return }
        await sink(label, all.joined(separator: "\n"))
    }

    /// Whether an administrative monitor notice can be parked to ride out with a later event, instead
    /// of waking the agent by itself.
    ///
    /// Every monitor delivery re-invokes the agent as a whole turn, so a notice sent on its own buys a
    /// full turn to say "nothing happened" — the "just the old monitor's teardown notices, nothing to
    /// act on" turn, at the cost of a real inference pass. Parking is safe exactly while something else
    /// is going to wake the agent anyway: another armed watch. With trailing output in hand the wake is
    /// already happening, so the notice rides along with it and nothing needs parking. When neither
    /// holds, the notice IS the signal the agent is parked waiting for and must go now.
    static func monitorNoticeCanBeParked(hasTrailingOutput: Bool, otherMonitorsArmed: Bool) -> Bool {
        !hasTrailingOutput && otherMonitorsArmed
    }

    /// Deliver an administrative notice — a watch ended, or was stopped for being over-chatty — with
    /// any trailing output it should be carried out with. These notices carry no work of their own:
    /// they exist so the agent doesn't sit waiting on a watch that is already dead, so they are parked
    /// rather than spent on a turn whenever ``monitorNoticeCanBeParked(hasTrailingOutput:otherMonitorsArmed:)``
    /// allows. ``monitorNoticeParkNanos`` bounds how long a parked notice waits for a ride.
    private func deliverMonitorNotice(_ note: String, label: String, trailing: [String] = []) async {
        guard monitorEventSink != nil else { return }
        guard
            !Self.monitorNoticeCanBeParked(
                hasTrailingOutput: !trailing.isEmpty, otherMonitorsArmed: !activeMonitors.isEmpty)
        else { return parkMonitorNotice(note, label: label) }
        await deliverMonitorTurn(label: label, lines: trailing + [note])
    }

    /// Park a notice until a real event carries it, or the backstop timer gives up waiting for one.
    private func parkMonitorNotice(_ note: String, label: String) {
        pendingMonitorNotices.append(note)
        pendingMonitorNoticeLabel = label
        guard pendingNoticeTask == nil else { return }
        pendingNoticeTask = Task { [weak self] in
            try? await Task.sleep(nanoseconds: Self.monitorNoticeParkNanos)
            if Task.isCancelled { return }
            await self?.flushPendingMonitorNotices()
        }
    }

    /// Take the parked notices, atomically, and disarm the backstop — the caller now owns delivering
    /// them. No suspension between the read and the clear, so two concurrent deliveries can't both
    /// carry the same notice.
    private func takePendingMonitorNotices() -> [String] {
        pendingNoticeTask?.cancel()
        pendingNoticeTask = nil
        let notices = pendingMonitorNotices
        pendingMonitorNotices.removeAll()
        return notices
    }

    /// No real event came along to carry the parked notices, so spend the turn on them now rather than
    /// leave the agent waiting indefinitely on a watch that already ended.
    private func flushPendingMonitorNotices() async {
        pendingNoticeTask = nil
        let label = pendingMonitorNoticeLabel ?? "monitor"
        await deliverMonitorTurn(label: label, lines: [])
    }

    /// Reason a monitor ended, for the closing note wording.
    private enum MonitorEndReason: Equatable {
        /// The source ended on its own — the command exited, or the WebSocket closed.
        case exited
        /// Past its age AND confirmed quiet: reaped as wedged (see ``beginMonitorStallWatch(id:ageNanos:)``).
        case stalled
        case shutdown
    }

    /// Claim a monitor for ending, atomically. Returns the run to exactly ONE caller — the first to
    /// claim it — and nil to every other; the winner owns the teardown and the single closing note.
    ///
    /// This is the whole fix for the double-end race: it removes the run from ``activeMonitors``, marks
    /// it stopped, and cancels its timers with NO suspension point in between, so a deadline firing
    /// concurrently with a natural exit can no longer slip through a half-finished teardown and deliver
    /// a second, contradictory note ("reached its timeout" on a command that had already exited).
    private func claimMonitor(id: String) -> MonitorRun? {
        guard let run = activeMonitors[id], !run.stopped else { return nil }
        activeMonitors[id] = nil
        run.stopped = true
        run.task?.cancel()
        run.deadlineTask?.cancel()
        run.stallMonitor?.cancel()
        return run
    }

    /// Stop an armed monitor early (confirmed stall, or teardown). Kills the process tree, cancels the
    /// pump, balances the container ref, and — except on session shutdown — delivers a closing note.
    private func stopMonitor(id: String, reason: MonitorEndReason) async {
        guard let run = claimMonitor(id: id) else { return }
        await terminateMonitorSource(run)
        await releaseMonitorContainer(run)
        if reason != .shutdown { await deliverMonitorEnd(run: run, id: id, reason: reason) }
    }

    /// The pump saw the command exit on its own: claim the monitor, flush any tail, balance the
    /// container ref, and note the end. Claiming BEFORE the flush is what keeps the flush's suspension
    /// from being a window in which the deadline can end the same monitor a second time.
    private func finishMonitor(id: String, reason: MonitorEndReason) async {
        guard let run = claimMonitor(id: id) else { return }
        // Carry the final partial batch out WITH the closing note, in one turn — sending the tail and
        // then "…and it ended" as two wake-ups costs two full turns to say one thing.
        let lines = run.pending
        run.pending.removeAll()
        await releaseMonitorContainer(run)
        await deliverMonitorEnd(run: run, id: id, reason: reason, trailing: lines)
    }

    /// Tell the session a monitor stopped, so the agent doesn't keep waiting on a dead watch. Called
    /// exactly once per monitor, by whichever end path claimed it. `trailing` is the watch's last
    /// unflushed output, which rides out in the same turn as the note.
    private func deliverMonitorEnd(
        run: MonitorRun, id: String, reason: MonitorEndReason, trailing: [String] = []
    ) async {
        guard monitorEventSink != nil else { return }
        let why: String
        if reason == .stalled {
            why = run.isWebSocket
                ? "it passed its timeout and the socket then went silent for that same window"
                : "it passed its timeout and the command then went idle — no output and near-zero "
                    + "CPU and disk I/O — so it was treated as wedged and its process tree was killed"
        } else if run.isWebSocket {
            why = "the WebSocket closed"
        } else {
            why = "the watched command exited"
        }
        // Name the watch inline: a parked notice can be carried out by a DIFFERENT monitor's delivery,
        // whose label heads the turn, so the note has to identify itself.
        let note = "[monitor \(id) (\(run.description)) ended: \(why).]"
        await deliverMonitorNotice(note, label: run.description, trailing: trailing)
    }

    /// Kill every armed monitor (session ``shutdown()``). Silent — the session is going away, so a
    /// closing note would have nowhere to land. Still balances each container ref.
    private func teardownMonitors() async {
        // Parked notices die with the session too — there is no turn left to carry them.
        pendingNoticeTask?.cancel()
        pendingNoticeTask = nil
        pendingMonitorNotices.removeAll()
        for id in Array(activeMonitors.keys) {  // snapshot: each claim mutates `activeMonitors`
            guard let run = claimMonitor(id: id) else { continue }
            await terminateMonitorSource(run)
            await releaseMonitorContainer(run)
        }
    }

    // MARK: - Host exec bridge (HOST_EXEC)

    /// Gate + run a `host_exec` call. This path NEVER consults the auto-approve blanket or
    /// the always-rule cache: it surfaces an explicit approval every time, unless the user
    /// previously chose "Allow for this session" (which only an explicit choice can set).
    private func handleHostExecCall(_ call: MCPApprovalServer.HostExecCall, workdir: String) async
        -> MCPApprovalServer.HostExecReply
    {
        // Programmatic resistance (HostExecPolicy): host_exec escapes the sandbox onto the shared
        // macOS host, so it is a last resort for the macOS-only toolchain — not a general shell.
        // Sort the command before doing anything else.
        let decision = HostExecPolicy.classify(call.command)

        // Reading a `.env` secrets file is forbidden outright — denied here, before anything else,
        // and no session grant can ever wave it through.
        if case .secretFileBlocked = decision {
            return .denied(message: HostExecPolicy.secretFileBlockedDenial())
        }

        // A command that plainly runs in the container is refused right here, with a pointer back
        // to the container — it never reaches an approval prompt.
        if case .containerRunnable(let program) = decision {
            return .denied(message: HostExecPolicy.containerRunnableDenial(program: program))
        }

        // Every accepted call must justify why the container can't do it. Empty/vague is refused
        // (the agent retries with a real reason), keeping the escape hatch deliberate.
        guard HostExecPolicy.isAcceptableReason(call.reason) else {
            return .denied(message: HostExecPolicy.missingReasonMessage())
        }

        // "Allow for this session" only ever blankets genuine host-only builds. An ambiguous
        // command (make, bash -c "…", a bare script path) can't be shown to need the host, and a
        // read-only inspection command — where a `-exec`/`-o`/`!cmd` escape could hide inside an
        // otherwise-benign tool — is shown in full each time too; both ALWAYS re-prompt, so a prior
        // session grant can never wave either through unseen.
        let forceFreshPrompt: Bool
        switch decision {
        case .ambiguous, .readOnly:
            forceFreshPrompt = true
        case .hostOnly, .containerRunnable, .secretFileBlocked:
            forceFreshPrompt = false
        }

        if forceFreshPrompt || !hostExecSessionGranted {
            let request = ApprovalRequest(
                id: .generate(),
                sessionID: sessionID ?? SessionID(rawValue: "unknown"),
                toolCallID: nil,
                toolName: MCPApprovalServer.qualifiedHostExecToolName,
                input: .object([
                    "command": .string(call.command),
                    "reason": .string(call.reason),
                    // The host cwd this command runs in, so the approval card can resolve `$PWD`.
                    "cwd": .string(workdir),
                ]),
                title: "Run on host: \(Self.oneLine(call.command))",
                risk: .hostExec,
                createdAt: configuration.now())

            emit(.approvalRequested(request), nativeType: "mcp/tools/call")
            let resolved = await approvals.waitForDecision(request)
            emit(.approvalResolved(resolved), nativeType: nil)

            switch resolved.decision {
            case .deny(let reason):
                return .denied(message: reason ?? "The user denied host execution.")
            case .cancelRun:
                interruptRequested = true
                handle?.sendSignal(SIGINT)
                return .denied(message: "The user cancelled the run.")
            case .allowAlways:
                // "Allow for this session" — but only a host-only build earns the blanket; an
                // ambiguous command stays gated per-call (see forceFreshPrompt above).
                if !forceFreshPrompt { hostExecSessionGranted = true }
            case .allow:
                break
            }
        }

        // Host-command concurrency gate (HOST_EXEC §concurrency): before running on the shared host,
        // claim a run slot so two sessions don't fire *similar* commands (e.g. two builds) at once —
        // which thrash the host toolchain, and for nvrsion sessions sharing one trunk collide on the
        // same build directory. On a detected conflict the user gets an in-chat prompt naming the
        // other session with three choices: "Run anyway" overrides a false positive, "Wait" queues
        // behind the running twin and auto-runs once it's clear, "Cancel" skips. The slot is held
        // only for the command's duration. Fail-open: with no coordinator (host-only/test setups),
        // just run, exactly as before.
        guard let coordinator = conflictCoordinator, let sid = sessionID else {
            return await runOnHost(command: call.command, workdir: workdir, shell: call.shell)
        }
        var override = false
        while true {
            switch await coordinator.beginHostCommand(
                sessionID: sid, command: call.command, override: override)
            {
            case .clear:
                let reply = await runOnHost(
                    command: call.command, workdir: workdir, shell: call.shell)
                await coordinator.endHostCommand(sessionID: sid)
                return reply
            case .conflict(let conflicts):
                switch await surfaceHostCommandConflict(command: call.command, conflicts: conflicts) {
                case .runAnyway:
                    override = true  // user dismissed it as a false positive — re-claim, forcing past it
                    continue
                case .wait:
                    // FIFO wait: queue behind the running twin(s), poll until it's our turn, then run.
                    // A nil reply means the run was interrupted (Stop) before the slot freed.
                    if let reply = await waitForHostCommandSlot(
                        sessionID: sid, command: call.command, workdir: workdir,
                        shell: call.shell, conflicts: conflicts, coordinator: coordinator)
                    {
                        return reply
                    }
                    return .denied(message: "The run was cancelled while waiting for the host to free up.")
                case .skip:
                    return .denied(message: Self.hostConflictDenial(conflicts))
                }
            }
        }
    }

    /// Drop the one-time "VM opened" transcript note when a guest becomes usable *anew* for this
    /// session, so the chat gets a clickable "Open VM Monitor" affordance (the flagged note the
    /// transcript renders with a button). Fires on a cold open — it wasn't live (a fresh clone, or a
    /// reboot from a stopped / idle-stopped clone) — and on a thaw from suspension — it was live but
    /// paused, which `ensureRunning` (or an explicit `mac_vm_control resume`) transparently resumes.
    /// Both flags are the VM's state *before* the boot, since a suspended VM still reads as running
    /// (it stays in the live registry, just paused) and would otherwise slip past. A plain reuse of an
    /// already-running, unpaused VM adds nothing. `label` names the guest OS in the copy.
    /// Returns the "state could not be restored" notice when this boot was a failed-restore cold start,
    /// so a caller building a tool reply can repeat it there (the transcript note this emits is for the
    /// human; the agent needs it in the reply it actually reads). `nil` on a healthy boot/resume.
    @discardableResult
    private func announceVMOpened(
        wasRunning: Bool, wasPaused: Bool, label: String,
        manager: MacVMManager? = nil, name: String? = nil
    ) async -> String? {
        // A resume whose saved state VZ refused cold-boots the guest instead (see
        // `MacVMEngine.performBoot`). Every boot/reuse path funnels through here, so this is the one
        // place that catches it for exec, computer-use, AND the explicit control op — and it must be
        // read even when the note below is skipped, so the loss can't be swallowed by an early return.
        var lost: String?
        if let manager, let name { lost = await manager.takeRestoreFailure(name: name) }
        if let lost {
            emit(
                .note(NoteEvent(
                    text: "The \(label) VM cold-booted. \(lost)", icon: "exclamationmark.triangle")),
                nativeType: nil)
        }
        guard !wasRunning || wasPaused else { return lost }
        // "Resumed" would be a lie when the restore failed — the guest came back empty.
        let verb = lost != nil ? "Rebooted" : (wasPaused ? "Resumed" : "Opened")
        emit(
            .note(NoteEvent(
                text: "\(verb) the \(label) VM. Open the VM Monitor to watch its screen.",
                icon: "display",
                vmMonitor: true)),
            nativeType: nil)
        return lost
    }

    /// Narrate a `mac_vm_control` / `linux_vm_control` / `linux_container` lifecycle transition into the
    /// transcript, so a suspend / shutdown / restart / kill (VMs) or a create / stop / remove
    /// (containers) shows up in chat the way the "Booting the … VM …" note does for a cold start.
    /// Progress-style (trailing "…") and emitted just before the (possibly slow) engine op runs; the
    /// tool reply still carries the settled-state message the agent reads. Callers gate this on the
    /// resource actually being in a state the op will change, so a no-op isn't narrated.
    private func announceLifecycleTransition(_ text: String, icon: String) {
        // `narratesTool` marks this as the pretty stand-in for the lifecycle tool's card, so the
        // transcript drops that redundant card and shows only this line (TranscriptProjection).
        emit(.note(NoteEvent(text: text, icon: icon, narratesTool: true)), nativeType: nil)
    }

    /// Format an error thrown while booting/reusing a session VM into an agent-facing message. When the
    /// boot was refused because the configured **maximum number of VMs is already running**
    /// (``MacVMError/concurrencyLimit``), surface that error's own clear, VM-type-neutral text so the
    /// agent is told the cap is reached (and to wait or free a VM) rather than reading it as an opaque
    /// failure. Any other error keeps the `<kind> VM error:` prefix so its origin stays obvious.
    private static func vmBootErrorMessage(_ error: Error, kind: String) -> String {
        if case MacVMError.concurrencyLimit = error {
            return String(describing: error)
        }
        return "\(kind) VM error: \(String(describing: error))"
    }

    /// Gate + run a `mac_vm_exec` call: it boots or reuses **this session's own** isolated macOS VM and
    /// runs the command through its vsock agent. `hostWorktree` is shared into the VM so the build sees the
    /// repo, mounted directly at its original absolute host path. Because each session gets a distinct VM,
    /// there is no shared-host concurrency gate — parallel agents don't collide the way they do on
    /// `host_exec`. For the same reason the VM is disposable and can't escape onto the shared host,
    /// **Auto mode approves every call without prompting**; interactive sessions surface an explicit
    /// approval unless the user already chose "Allow for this session".
    private func handleMacVMExecCall(
        _ call: MCPApprovalServer.MacVMExecCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMExecReply {
        guard let manager = macVMManager else {
            return .denied(message: "The macOS VM service is not available.")
        }
        // Every accepted call must justify why it needs the macOS VM; empty/vague is refused (reuse the
        // host_exec reason gate — same "deliberate escape hatch" intent).
        guard HostExecPolicy.isAcceptableReason(call.reason) else {
            return .denied(message: HostExecPolicy.missingReasonMessage())
        }

        // Auto mode waves every macOS-VM command through: the VM is isolated and disposable and can't
        // reach the shared host, so there's nothing a per-command prompt would protect.
        if !autoApprove, !macVMExecSessionGranted {
            let request = ApprovalRequest(
                id: .generate(),
                sessionID: sessionID ?? SessionID(rawValue: "unknown"),
                toolCallID: nil,
                toolName: MCPApprovalServer.qualifiedMacVMExecToolName,
                input: .object([
                    "command": .string(call.command),
                    "reason": .string(call.reason),
                ]),
                title: "Run in macOS VM: \(Self.oneLine(call.command))",
                risk: .hostExec,
                createdAt: configuration.now())
            emit(.approvalRequested(request), nativeType: "mcp/tools/call")
            let resolved = await approvals.waitForDecision(request)
            emit(.approvalResolved(resolved), nativeType: nil)
            switch resolved.decision {
            case .deny(let reason):
                return .denied(message: reason ?? "The user denied running in the macOS VM.")
            case .cancelRun:
                interruptRequested = true
                handle?.sendSignal(SIGINT)
                return .denied(message: "The user cancelled the run.")
            case .allowAlways:
                macVMExecSessionGranted = true
            case .allow:
                break
            }
        }

        guard let sid = sessionID else {
            return .denied(message: "No active session for the macOS VM.")
        }
        let vmName = MacVMManager.vmName(for: sid)
        let shareName = "workspace"
        // Boot mounts the custom-tagged share directly here and fails readiness if attachment fails.
        let guestWorkdir = hostWorktree
        // Honor the user's configured per-session VM resources + SSH account (Settings ▸ Virtual
        // Machines), not the bare `MacVMSpec` defaults.
        let spec = MacVMSpec(
            name: vmName,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: shareName, readOnly: false)],
            workdir: guestWorkdir,
            sshUser: MacVMSettings.sshUser,
            computerUse: macVMComputerAllowed)
        do {
            // Narrate the wait so a cold start doesn't read as a hang (mirrors the host-command wait
            // notes): a fresh clone's first boot can take a couple of minutes — see
            // MacVMEngine.performBoot. Only announce the boot when the VM isn't already up; a reuse is
            // quick and needs no note.
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            if !alreadyRunning {
                emit(
                    .note(NoteEvent(
                        text: "Booting the macOS VM (a first boot can take a couple of minutes)…",
                        icon: "hourglass")),
                    nativeType: nil)
            }
            // First call boots the VM (a fresh clone can take a couple of minutes); later calls reuse it.
            let (name, _) = try await manager.ensureRunning(spec)
            markActiveVM(name: name, isLinux: false)
            defer { Task { await manager.finished(name: name) } }  // re-arm the idle timer after the turn
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: "macOS",
                manager: manager, name: name)
            emit(
                .note(NoteEvent(
                    text: "Running in the macOS VM: \(Self.oneLine(call.command))",
                    icon: "play.fill")),
                nativeType: nil)
            // Bound the wait with the same timeout host_exec uses: a guest command that never exits (a
            // foregrounded server, a stdin-blocked process, a long build piped through `| tail`) must not
            // wedge the tool call forever — the exec vsock channel has no timeout of its own. On timeout
            // the guest process is signalled and the agent gets its partial output plus a note.
            let result = try await manager.run(
                name: name, command: call.command, workdir: guestWorkdir,
                timeoutNanos: configuration.hostCommandTimeoutNanos)
            // Pull nash's spooled shell-event batches out of the guest and into the feed (NASH.md M5).
            await drainVMShellEvents(vmName: name, environment: .macVM(name: name))
            return .ran(exitCode: result.exitCode, stdout: result.stdout, stderr: result.stderr)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: "macOS"))
        }
    }

    /// Perform one `mac_vm_computer` computer-use action against this session's macOS VM and return the
    /// post-action screenshot (as an MCP image the model sees) plus a short summary. Boots/reuses the
    /// SAME VM `mac_vm_exec` uses. Unlike `host_exec`/`mac_vm_exec`, actions are NOT individually
    /// approval-gated — a per-click prompt would make a navigation loop unusable, and the blast radius
    /// is the agent's own disposable VM. Exposure (the tool being advertised at all) is the opt-in gate.
    private func handleMacVMComputerCall(
        _ call: MCPApprovalServer.MacVMComputerCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMComputerReply {
        guard let manager = macVMManager else {
            return .denied(message: "The macOS VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the macOS VM.")
        }
        let vmName = MacVMManager.vmName(for: sid)
        let shareName = "workspace"
        let spec = MacVMSpec(
            name: vmName,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: shareName, readOnly: false)],
            workdir: hostWorktree,  // the re-created identical host path (see MacVMEngine+HostPaths)
            sshUser: MacVMSettings.sshUser,
            computerUse: macVMComputerAllowed)
        do {
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            let (name, _) = try await manager.ensureRunning(spec)
            markActiveVM(name: name, isLinux: false)
            defer { Task { await manager.finished(name: name) } }
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: "macOS",
                manager: manager, name: name)
            let (image, summary) = try await manager.performComputerAction(
                name: name, action: call.action, x: call.x, y: call.y, text: call.text,
                ref: call.ref, value: call.value, url: call.url,
                scrollDirection: call.scrollDirection, scrollAmount: call.scrollAmount,
                durationMs: call.durationMs)
            return .ok(text: summary, imageBase64: image)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: "macOS"))
        }
    }

    /// Handle a `mac_vm_computer_batch` call: boot/reuse this session's macOS VM once, then run every
    /// step in order. Delegates to ``performComputerBatch`` after building the macOS VM spec.
    private func handleMacVMComputerBatchCall(
        _ call: MCPApprovalServer.MacVMComputerBatchCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMComputerBatchReply {
        guard let manager = macVMManager else {
            return .denied(message: "The macOS VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the macOS VM.")
        }
        let vmName = MacVMManager.vmName(for: sid)
        let spec = MacVMSpec(
            name: vmName,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: "workspace", readOnly: false)],
            workdir: hostWorktree,  // the re-created identical host path (see MacVMEngine+HostPaths)
            sshUser: MacVMSettings.sshUser,
            computerUse: macVMComputerAllowed)
        return await performComputerBatch(
            call, manager: manager, spec: spec, vmName: vmName, kind: "macOS", isLinux: false)
    }

    /// Run a computer-use batch against an already-specced VM: boot/reuse it once, perform each step in
    /// order over the same GUI channel, and collect a per-step outcome. Shared by the macOS and Linux
    /// batch handlers — only the spec, VM name, `kind` label, and active-name setter differ. On a step
    /// error it records the failure and, when `stop_on_error` is set, stops without attempting the rest.
    /// When `screenshot_each_step` is off, only the last captured screenshot is kept (the final state),
    /// so the reply carries a single image; otherwise every step keeps its own.
    private func performComputerBatch(
        _ call: MCPApprovalServer.MacVMComputerBatchCall,
        manager: MacVMManager, spec: MacVMSpec, vmName: String, kind: String, isLinux: Bool
    ) async -> MCPApprovalServer.MacVMComputerBatchReply {
        do {
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            let (name, _) = try await manager.ensureRunning(spec)
            markActiveVM(name: name, isLinux: isLinux)
            defer { Task { await manager.finished(name: name) } }
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: kind,
                manager: manager, name: name)

            // Capture the screenshot each step returns; post-filter below keeps only what was asked for.
            typealias Raw = (index: Int, action: String, summary: String, image: String?, failed: Bool)
            var raw: [Raw] = []
            var stoppedAtError = false
            for (index, step) in call.steps.enumerated() {
                do {
                    let (image, summary) = try await manager.performComputerAction(
                        name: name, action: step.action, x: step.x, y: step.y, text: step.text,
                        ref: step.ref, value: step.value, url: step.url,
                        scrollDirection: step.scrollDirection, scrollAmount: step.scrollAmount,
                        durationMs: step.durationMs)
                    raw.append((index, step.action, summary, image, false))
                } catch {
                    raw.append((
                        index, step.action, Self.vmBootErrorMessage(error, kind: kind), nil, true))
                    if call.stopOnError {
                        stoppedAtError = true
                        break
                    }
                }
                if let delay = call.stepDelayMs, delay > 0 {
                    try? await Task.sleep(nanoseconds: UInt64(delay) * 1_000_000)
                }
            }

            // Only the FINAL screenshot is kept unless the agent asked for one per step — the last step
            // that actually produced an image is the current on-screen state.
            let lastImageIndex = raw.last(where: { $0.image != nil })?.index
            let steps = raw.map { r in
                let keep = call.screenshotEachStep || r.index == lastImageIndex
                return MCPApprovalServer.MacVMComputerBatchReply.StepOutcome(
                    index: r.index, action: r.action, summary: r.summary,
                    imageBase64: keep ? r.image : nil, failed: r.failed)
            }
            return .ran(steps: steps, stoppedAtError: stoppedAtError)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: kind))
        }
    }

    /// Handle a `mac_vm_clear_notifications` call: dismiss the macOS VM's on-screen notifications by
    /// killing the guest's NotificationCenter (which relaunches clean). Not approval-gated — like the
    /// computer-use actions it only touches the agent's own disposable VM. Boots/reuses the SAME VM
    /// `mac_vm_exec`/`mac_vm_computer` use, then runs the fixed command over the exec channel.
    private func handleMacVMClearNotificationsCall(
        hostWorktree: String
    ) async -> MCPApprovalServer.MacVMExecReply {
        guard let manager = macVMManager else {
            return .denied(message: "The macOS VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the macOS VM.")
        }
        let vmName = MacVMManager.vmName(for: sid)
        let shareName = "workspace"
        let spec = MacVMSpec(
            name: vmName,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: shareName, readOnly: false)],
            workdir: hostWorktree,  // the re-created identical host path (see MacVMEngine+HostPaths)
            sshUser: MacVMSettings.sshUser,
            computerUse: macVMComputerAllowed)
        do {
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            let (name, _) = try await manager.ensureRunning(spec)
            markActiveVM(name: name, isLinux: false)
            defer { Task { await manager.finished(name: name) } }
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: "macOS",
                manager: manager, name: name)
            emit(
                .note(NoteEvent(
                    text: "Clearing notifications in the macOS VM…", icon: "bell.slash")),
                nativeType: nil)
            // `killall` exits non-zero when NotificationCenter isn't running (nothing to clear) — that's
            // a success for our purposes, so swallow it and report exit 0 with an empty banner list.
            // Not background work: this rides the computer-use flow (its effect is on the watched
            // screen), so it must not flip the monitor's "Working in the background…" overlay on.
            let result = try await manager.run(
                name: name, command: "killall NotificationCenter", workdir: nil,
                asBackgroundWork: false)
            return .ran(exitCode: result.exitCode, stdout: result.stdout, stderr: result.stderr)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: "macOS"))
        }
    }

    /// Gate a `mac_vm_request_operator` call: the agent asks the *user* to operate this session's macOS
    /// VM by hand for a step only a human can clear (CAPTCHA / login / interactive dialog). Modeled on
    /// `handleHostExecCall` — surface an approval carrying the agent's (unverified) `reason` +
    /// `instructions`, and suspend until the user finishes or declines. The VM is booted/kept alive for
    /// the whole hand-off so the interactive host-side surface exists when the user opens the viewer; the
    /// in-app card + walkthrough popup drive the viewer and resolve this approval. Never auto-approved.
    private func handleMacVMOperatorCall(
        _ call: MCPApprovalServer.MacVMOperatorCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMOperatorReply {
        guard let manager = macVMManager else {
            return .denied(message: "The macOS VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the macOS VM.")
        }
        // The user needs to know what to do — an empty instruction set is refused (the agent retries
        // with real instructions), and the reason must justify why a human is required (reuse the
        // host_exec reason gate — same "deliberate, well-explained" intent).
        let instructions = call.instructions.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !instructions.isEmpty else {
            return .denied(
                message: "Provide `instructions`: exactly what the user should do in the VM.")
        }
        guard HostExecPolicy.isAcceptableReason(call.reason) else {
            return .denied(message: HostExecPolicy.missingReasonMessage())
        }

        // Boot/reuse this session's VM with the computer-use surface so the viewer the user opens is
        // the real, interactive guest. `ensureRunning` marks the turn busy (disarming the idle timer);
        // the matching `finished` runs only when this handler returns — i.e. after the user is done — so
        // the VM stays alive for the whole hand-off.
        let vmName = MacVMManager.vmName(for: sid)
        let shareName = "workspace"
        let spec = MacVMSpec(
            name: vmName,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: shareName, readOnly: false)],
            workdir: hostWorktree,  // the re-created identical host path (see MacVMEngine+HostPaths)
            sshUser: MacVMSettings.sshUser,
            computerUse: macVMComputerAllowed)
        let runningName: String
        do {
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            let (name, _) = try await manager.ensureRunning(spec)
            runningName = name
            markActiveVM(name: name, isLinux: false)
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: "macOS",
                manager: manager, name: name)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: "macOS"))
        }
        defer { Task { await manager.finished(name: runningName) } }

        let request = ApprovalRequest(
            id: .generate(),
            sessionID: sid,
            toolCallID: nil,
            toolName: MCPApprovalServer.qualifiedMacVMOperatorToolName,
            input: .object([
                "reason": .string(call.reason),
                "instructions": .string(instructions),
                // The VM the viewer should open, so the in-app card can present the right guest.
                "vm_name": .string(runningName),
            ]),
            title: "VM help needed: \(Self.oneLine(instructions))",
            // Reuse the host-exec risk so the request is always surfaced, never auto-approved, and
            // offers no "Allow Always" — each hand-off is a one-off. The card is chosen by toolName.
            risk: .hostExec,
            createdAt: configuration.now())

        emit(.approvalRequested(request), nativeType: "mcp/tools/call")
        let resolved = await approvals.waitForDecision(request)
        emit(.approvalResolved(resolved), nativeType: nil)

        switch resolved.decision {
        case .deny(let reason):
            return .denied(message: reason ?? "The user declined to operate the VM.")
        case .cancelRun:
            interruptRequested = true
            handle?.sendSignal(SIGINT)
            return .denied(message: "The user cancelled the run.")
        case .allow(let updatedInput):
            // The walkthrough popup resolves with {completed, note}; default to completed when absent
            // (a plain Allow with no payload reads as "the user handled it").
            let completed = updatedInput?["completed"]?.boolValue ?? true
            let note = updatedInput?["note"]?.stringValue ?? ""
            return .resolved(completed: completed, note: note)
        case .allowAlways:
            return .resolved(completed: true, note: "")
        }
    }

    /// Gate + run a `linux_vm_exec` call: boots/reuses this session's isolated **Linux** VM (the
    /// lighter, faster-booting sibling of the macOS VM) and runs the command inside it over the vsock
    /// agent. The repo is shared in over virtiofs; because the Linux base has no automount, the handler
    /// mounts the share on demand and runs from it. Gated like `mac_vm_exec` — approved per call (or per
    /// session), never auto-approved through Claude's path; Auto mode waves it through (the VM is
    /// isolated and disposable).
    private func handleLinuxVMExecCall(
        _ call: MCPApprovalServer.MacVMExecCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMExecReply {
        guard let manager = macVMManager else {
            return .denied(message: "The Linux VM service is not available.")
        }
        guard HostExecPolicy.isAcceptableReason(call.reason) else {
            return .denied(message: HostExecPolicy.missingReasonMessage())
        }
        if !autoApprove, !linuxVMExecSessionGranted {
            let request = ApprovalRequest(
                id: .generate(),
                sessionID: sessionID ?? SessionID(rawValue: "unknown"),
                toolCallID: nil,
                toolName: MCPApprovalServer.qualifiedLinuxVMExecToolName,
                input: .object([
                    "command": .string(call.command),
                    "reason": .string(call.reason),
                ]),
                title: "Run in Linux VM: \(Self.oneLine(call.command))",
                risk: .hostExec,
                createdAt: configuration.now())
            emit(.approvalRequested(request), nativeType: "mcp/tools/call")
            let resolved = await approvals.waitForDecision(request)
            emit(.approvalResolved(resolved), nativeType: nil)
            switch resolved.decision {
            case .deny(let reason):
                return .denied(message: reason ?? "The user denied running in the Linux VM.")
            case .cancelRun:
                interruptRequested = true
                handle?.sendSignal(SIGINT)
                return .denied(message: "The user cancelled the run.")
            case .allowAlways:
                linuxVMExecSessionGranted = true
            case .allow:
                break
            }
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the Linux VM.")
        }
        let vmName = MacVMManager.linuxVMName(for: sid)
        let shareName = "workspace"
        // The Linux base does NOT automount virtiofs (its fstab carries only root — see
        // guest/linux-base/bootstrap/init), so mount the session share on demand and run from it.
        // `nucleicshare` is the fixed virtiofs tag both sides agree on (MacVMEngine+LinuxConfig).
        let guestMountRoot = "/mnt/nucleic"
        let sharePath = "\(guestMountRoot)/\(shareName)"
        // Run from the re-created **identical host path** — host-path re-creation bind-mounts the share
        // there at boot (see MacVMEngine+HostPaths), so artifacts that hardcode it resolve — falling
        // back to the share path then the mount root, since that re-creation is best-effort.
        let guestWorkdir = hostWorktree
        let spec = MacVMSpec(
            name: vmName,
            os: .linux,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: shareName, readOnly: false)],
            workdir: guestWorkdir,
            sshUser: MacVMSettings.sshUser,
            computerUse: linuxVMComputerAllowed)
        do {
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            if !alreadyRunning {
                emit(
                    .note(NoteEvent(
                        text: "Booting the Linux VM (a first boot can take a couple of minutes)…",
                        icon: "hourglass")),
                    nativeType: nil)
            }
            let (name, _) = try await manager.ensureRunning(spec)
            markActiveVM(name: name, isLinux: true)
            defer { Task { await manager.finished(name: name) } }
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: "Linux",
                manager: manager, name: name)
            emit(
                .note(NoteEvent(
                    text: "Running in the Linux VM: \(Self.oneLine(call.command))",
                    icon: "play.fill")),
                nativeType: nil)
            let mountPrep =
                "mkdir -p \(guestMountRoot); mountpoint -q \(guestMountRoot) 2>/dev/null || "
                + "mount -t virtiofs nucleicshare \(guestMountRoot) 2>/dev/null || true; "
            let wrapped = mountPrep
                + MacVMSpec.workdirPrefix([guestWorkdir, sharePath, guestMountRoot])
                + call.command
            let result = try await manager.run(name: name, command: wrapped, workdir: nil)
            // Pull nash's spooled shell-event batches out of the guest and into the feed (NASH.md M4).
            await drainVMShellEvents(vmName: name, environment: .linuxVM(name: name))
            return .ran(exitCode: result.exitCode, stdout: result.stdout, stderr: result.stderr)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: "Linux"))
        }
    }

    /// After a `linux_vm_exec`/`mac_vm_exec`, poll the guest for nash's spooled shell-event batches and
    /// forward them through the shared `ConflictCoordinator` — the same seam ``handleShellReport(_:)``
    /// uses for the container surface (docs/NASH.md §6.2, §7.4, §7.5). nash inside a VM spools rather
    /// than POSTing, so this host poll is what lands VM shell/data-flow events in the feed. Guest-OS
    /// agnostic: the drain op speaks the same wire on both the Linux and macOS agents. Best-effort:
    /// no-ops without a manager/coordinator/session, and each drained batch reuses `parseShellBatch`
    /// unchanged (the batch's own `sessionId` may be absent, so the backend's session id is
    /// authoritative — as in `handleShellReport`).
    private func drainVMShellEvents(vmName: String, environment: NashEnvironment) async {
        guard let manager = macVMManager, let sessionID, let conflictCoordinator else { return }
        let batches = await manager.drainShellEvents(name: vmName)
        for batch in batches {
            let calls = MCPApprovalServer.parseShellBatch(batch).map { $0.attributed(to: environment) }
            await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
        }
    }

    /// Record which VM this session just activated and make sure the background drain poller is
    /// running. Every site that boots/reuses a VM goes through here, so the poller starts on the
    /// first VM touch of the session regardless of which tool (exec, computer, computer-batch) got
    /// there first — including the GUI paths, whose emissions have no exec behind them to drain them.
    private func markActiveVM(name: String, isLinux: Bool) {
        if isLinux { activeLinuxVMName = name } else { activeMacVMName = name }
        startVMShellDrainPoller()
    }

    /// Start the background poller that drains nash's spool from this session's running VMs
    /// (docs/NASH.md §6.2). Idempotent — the first VM activation starts it and it then runs for the
    /// life of the session.
    ///
    /// The post-exec drains in the `*_vm_exec` handlers stay as the correctness anchor (they
    /// guarantee a drain the instant a command finishes); this poller exists for everything those
    /// can't reach. nash spools continuously, but a drain only happened when an exec *returned* — so
    /// a long build's events stayed invisible until it finished, and anything emitted with no exec
    /// behind it (boot, readiness, provisioning, `*_vm_computer` GUI actions, background processes,
    /// the tail of a turn) sat in the spool until some later exec happened to sweep it up, which also
    /// mis-attributed it to whichever exec did the sweeping.
    private func startVMShellDrainPoller() {
        guard vmShellDrainTask == nil else { return }
        vmShellDrainTask = Task { [weak self] in
            while !Task.isCancelled {
                try? await Task.sleep(nanoseconds: ClaudeCodeBackend.vmShellDrainIntervalNanos)
                if Task.isCancelled { return }
                guard let self else { return }
                await self.drainActiveVMShellEvents()
            }
        }
    }

    /// Drain whichever of this session's VMs are actually running, attributing each guest's batches
    /// to its own environment. Skips a **paused** VM: a suspended guest has no agent to answer, and
    /// the drain deliberately does not touch the manager's idle bookkeeping, so polling never keeps a
    /// VM awake or fights the idle-suspend timer.
    private func drainActiveVMShellEvents() async {
        guard let manager = macVMManager else { return }
        var targets: [(name: String, environment: NashEnvironment)] = []
        if let name = activeMacVMName { targets.append((name, .macVM(name: name))) }
        if let name = activeLinuxVMName { targets.append((name, .linuxVM(name: name))) }
        for target in targets {
            // A suspended VM reads as running, so both checks are needed to tell it from a live one.
            guard await manager.isRunning(name: target.name) else { continue }
            let paused = await manager.isPaused(name: target.name)
            guard !paused else { continue }
            await drainVMShellEvents(vmName: target.name, environment: target.environment)
        }
    }

    /// Perform one `linux_vm_computer` action against this session's Linux VM and return the
    /// post-action screenshot. Boots/reuses the SAME VM `linux_vm_exec` uses. Not per-action gated —
    /// exposure is the opt-in boundary (the blast radius is the agent's own disposable VM).
    private func handleLinuxVMComputerCall(
        _ call: MCPApprovalServer.MacVMComputerCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMComputerReply {
        guard let manager = macVMManager else {
            return .denied(message: "The Linux VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the Linux VM.")
        }
        let vmName = MacVMManager.linuxVMName(for: sid)
        let shareName = "workspace"
        let spec = MacVMSpec(
            name: vmName,
            os: .linux,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: shareName, readOnly: false)],
            workdir: hostWorktree,  // the re-created identical host path (see MacVMEngine+HostPaths)
            sshUser: MacVMSettings.sshUser,
            computerUse: linuxVMComputerAllowed)
        do {
            let alreadyRunning = await manager.isRunning(name: vmName)
            // Read paused state before the boot: `ensureRunning` thaws a suspended VM, and a suspended
            // VM already reads as running, so capture it now to tell a resume from a plain reuse.
            let wasPaused = await manager.isPaused(name: vmName)
            let (name, _) = try await manager.ensureRunning(spec)
            markActiveVM(name: name, isLinux: true)
            defer { Task { await manager.finished(name: name) } }
            await announceVMOpened(
                wasRunning: alreadyRunning, wasPaused: wasPaused, label: "Linux",
                manager: manager, name: name)
            let (image, summary) = try await manager.performComputerAction(
                name: name, action: call.action, x: call.x, y: call.y, text: call.text,
                ref: call.ref, value: call.value, url: call.url,
                scrollDirection: call.scrollDirection, scrollAmount: call.scrollAmount,
                durationMs: call.durationMs)
            return .ok(text: summary, imageBase64: image)
        } catch {
            return .denied(message: Self.vmBootErrorMessage(error, kind: "Linux"))
        }
    }

    /// Handle a `linux_vm_computer_batch` call: the Linux sibling of ``handleMacVMComputerBatchCall`` —
    /// build the Linux VM spec, then run every step in order via ``performComputerBatch``.
    private func handleLinuxVMComputerBatchCall(
        _ call: MCPApprovalServer.MacVMComputerBatchCall, hostWorktree: String
    ) async -> MCPApprovalServer.MacVMComputerBatchReply {
        guard let manager = macVMManager else {
            return .denied(message: "The Linux VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the Linux VM.")
        }
        let vmName = MacVMManager.linuxVMName(for: sid)
        let spec = MacVMSpec(
            name: vmName,
            os: .linux,
            cpus: MacVMSettings.vmCPUs,
            memoryGiB: MacVMSettings.vmMemoryGiB,
            mounts: [MacVMSpec.Mount(host: hostWorktree, name: "workspace", readOnly: false)],
            workdir: hostWorktree,  // the re-created identical host path (see MacVMEngine+HostPaths)
            sshUser: MacVMSettings.sshUser,
            computerUse: linuxVMComputerAllowed)
        return await performComputerBatch(
            call, manager: manager, spec: spec, vmName: vmName, kind: "Linux", isLinux: true)
    }

    /// Handle a `mac_vm_control` call: apply one lifecycle op to this session's macOS VM.
    private func handleMacVMControlCall(
        _ call: MCPApprovalServer.VMControlCall
    ) async -> MCPApprovalServer.VMControlReply {
        guard let manager = macVMManager else {
            return .denied(message: "The macOS VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the macOS VM.")
        }
        return await performVMControl(
            op: call.op, vmName: MacVMManager.vmName(for: sid), label: "macOS", manager: manager)
    }

    /// Handle a `linux_vm_control` call: the Linux-VM sibling of ``handleMacVMControlCall(_:)``.
    private func handleLinuxVMControlCall(
        _ call: MCPApprovalServer.VMControlCall
    ) async -> MCPApprovalServer.VMControlReply {
        guard let manager = macVMManager else {
            return .denied(message: "The Linux VM service is not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for the Linux VM.")
        }
        return await performVMControl(
            op: call.op, vmName: MacVMManager.linuxVMName(for: sid), label: "Linux", manager: manager)
    }

    /// Shared core for `mac_vm_control` / `linux_vm_control`: apply one lifecycle `op` to the named
    /// per-session VM. 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. `label` is the guest-OS word ("macOS" / "Linux") used in the notes and messages.
    private func performVMControl(
        op rawOp: String, vmName: String, label: String, manager: MacVMManager
    ) async -> MCPApprovalServer.VMControlReply {
        let op = rawOp.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
        switch op {
        case "status":
            // Check suspension FIRST: a guest suspended **to disk** freed its slot and so reads as
            // not-running (`isRunning` means "occupies a live slot"), but its saved state makes it
            // fully resumable — reporting it "stopped" would tell the agent its state is already gone.
            if await manager.isPaused(name: vmName) {
                return .ok(
                    state: "suspended",
                    message: "The \(label) VM is suspended; `resume` — or the next command — "
                        + "restores it with all its state.")
            }
            if await manager.isRunning(name: vmName) {
                return .ok(state: "running", message: "The \(label) VM is running.")
            }
            let exists = await manager.exists(name: vmName)
            return .ok(
                state: exists ? "stopped" : "absent",
                message: exists
                    ? "The \(label) VM is stopped; its disk is kept, so the next command reboots it."
                    : "No \(label) VM exists yet for this session.")

        case "stop":
            // Narrate an actual shutdown; stopping an already-down VM changes nothing.
            if await manager.isRunning(name: vmName) {
                announceLifecycleTransition("Shutting down the \(label) VM…", icon: "power")
            }
            await manager.stop(name: vmName)
            return .ok(
                state: "stopped",
                message: "Stopped the \(label) VM (freed its RAM; the next command reboots it).")

        case "suspend":
            // Narrate only when there's a live, un-paused VM to freeze (a suspended VM still reads as
            // running, so exclude the already-paused case).
            if await manager.isRunning(name: vmName), await !manager.isPaused(name: vmName) {
                announceLifecycleTransition("Suspending the \(label) VM…", icon: "pause.circle")
            }
            let paused = await manager.suspend(name: vmName)
            return paused
                ? .ok(
                    state: "suspended",
                    message: "Suspended the \(label) VM (state saved to disk, its RAM and VM slot "
                        + "freed; `resume` — or the next command — restores it where it left off, "
                        + "and tells you if the state could not be restored).")
                : .denied(message: "The \(label) VM isn't running, so there's nothing to suspend.")

        case "resume":
            // Narrate only when there's a suspended VM to thaw.
            if await manager.isPaused(name: vmName) {
                announceLifecycleTransition("Resuming the \(label) VM…", icon: "play.circle")
            }
            let resumed = await manager.resume(name: vmName)
            guard resumed else { return .denied(message: "The \(label) VM isn't suspended.") }
            // An explicit thaw re-opens the guest's live screen — surface the monitor link, same as the
            // transparent resume a reusing exec/computer call triggers. This also consumes (once) any
            // failed-restore notice, which we repeat in the reply: a disk-restore VZ refused silently
            // cold-boots the guest, and saying only "Resumed" would confirm a state promise we just
            // failed to keep. Still `.ok` — the VM *is* running, just empty.
            let lost = await announceVMOpened(
                wasRunning: true, wasPaused: true, label: label, manager: manager, name: vmName)
            if let lost {
                return .ok(state: "running", message: "Rebooted the \(label) VM. \(lost)")
            }
            return .ok(state: "running", message: "Resumed the \(label) VM.")

        case "restart":
            guard await manager.exists(name: vmName) else {
                return .denied(
                    message: "No \(label) VM exists yet to restart — run a command to boot one first.")
            }
            announceLifecycleTransition("Restarting the \(label) VM…", icon: "arrow.clockwise")
            await manager.restart(name: vmName)
            return .ok(state: "running", message: "Restarted the \(label) VM.")

        case "kill", "remove":
            // Narrate only when there's a clone to delete.
            if await manager.exists(name: vmName) {
                announceLifecycleTransition("Killing the \(label) VM…", icon: "trash")
            }
            let removed = await manager.remove(name: vmName)
            return .ok(
                state: "removed",
                message: removed
                    ? "Killed the \(label) VM and deleted its disk clone (the next command boots a "
                        + "fresh one)."
                    : "No \(label) VM clone remained to kill.")

        default:
            return .denied(
                message: "Unknown op '\(rawOp)'. Use status, stop, suspend, resume, restart, or kill.")
        }
    }

    /// Handle a `linux_container` call: the agent manages its own throwaway Linux containers. Every
    /// operation is authorized by ownership rather than a host-style approval: names are resolved only
    /// through this backend session's private `agentContainerSpecs` map, so the agent cannot address a
    /// host resource, its enclosing sandbox, or another session's container. Containers run under a
    /// session-scoped `nucleic-scr-<short>-…` name; the idle timer stops them and launch-time disk GC
    /// reaps their clones, so a forgotten container costs nothing durable.
    private func handleLinuxContainerCall(
        _ call: MCPApprovalServer.ContainerToolCall, defaultImage: String, hostWorktree: String
    ) async -> MCPApprovalServer.ContainerToolReply {
        guard let manager = containerManager else {
            return .denied(message: "Agent-managed containers are not available.")
        }
        guard let sid = sessionID else {
            return .denied(message: "No active session for containers.")
        }
        let op = call.op.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()

        // The map is the ownership boundary. Even a guessed physical name is rejected unless this
        // session created it, which makes create/exec/stop/remove safe without host approval.
        if op == "list" {
            let names = agentContainerSpecs.keys.sorted()
            return .ok(
                name: "", names: names,
                message: names.isEmpty
                    ? "You have not created any containers this session."
                    : "\(names.count) container(s).")
        }
        guard ["create", "exec", "stop", "remove"].contains(op) else {
            return .denied(
                message: "Unknown op '\(call.op)'. Use create, exec, stop, remove, or list.")
        }
        switch op {
        case "create":
            let logical = agentContainerName(for: sid, label: call.name)
            if agentContainerSpecs[logical] != nil {
                return .denied(
                    message: "A container named \(logical) already exists — exec into it, or remove "
                        + "it first.")
            }
            let image = call.image.flatMap { $0.isEmpty ? nil : $0 } ?? defaultImage
            // Narrate the (possibly slow — an image pull can take a while) create so it reads in chat
            // the way "Booting the … VM…" does for a cold VM start; the settled reply the agent reads
            // still carries the container's name.
            announceLifecycleTransition(
                "Creating the Linux container from \(image)…", icon: "shippingbox")
            // The session runs in a linked git worktree, whose `.git` is a file pointing at the
            // repo's common git dir (`<repo>/.git`) — a SIBLING of the worktree, not underneath it.
            // Mounting only the worktree subtree leaves git unable to follow that pointer, so every
            // in-container `git` fails with `fatal: not a git repository: <repo>/.git/worktrees/…`.
            // Also mount the common dir at its identical host path (when it lies outside the
            // worktree) so the worktree's absolute pointers resolve; the per-worktree git dir lives
            // under it, so the one mount covers both. Writable, to match the worktree mount (so the
            // agent can commit/merge from the throwaway container).
            var mounts = [ContainerSpec.Mount(
                host: hostWorktree, container: hostWorktree, readOnly: false)]
            if let gitMount = GitDirResolver.commonDirMount(forWorktree: hostWorktree) {
                mounts.append(gitMount)
            }
            // A host-run session normally serves MCP over loopback TCP, which an isolated scratch
            // container cannot reach. Lazily add a sibling unix listener and relay that into the
            // default narOS image, just as a containerized parent session does. Observation stays
            // best-effort: failure to add this auxiliary listener must never prevent container use.
            if image == ProjectSandbox.defaultImage, hostShellSocketPath == nil,
                let server = runServer
            {
                let path = ApprovalServerRegistry.controlSocketPath(
                    for: "scratch-\(sid.short.lowercased())\(ContainerManager.channelSuffix)")
                if (try? await server.start(unixSocketPath: path)) != nil {
                    hostShellSocketPath = path
                }
            }
            let spec = ContainerSpec(
                name: logical,
                image: image,
                mounts: mounts,
                workdir: hostWorktree,
                env: [:],
                idleTimeout: TimeInterval(ProjectSandbox.defaultIdleTimeoutSeconds),
                claudeHomeStaging: "",
                claudeHomeWritable: "",
                // Default narOS scratch containers can reuse the session's live control socket.
                // Custom images keep their existing no-control-plane degradation behavior.
                controlSocketHostPath: image == ProjectSandbox.defaultImage
                    ? hostShellSocketPath : nil)
            do {
                _ = try await manager.ensureRunning(spec)
                await manager.finished(name: logical)  // arm the idle timer; the agent execs later
                agentContainerSpecs[logical] = spec
                return .ok(
                    name: logical, names: agentContainerSpecs.keys.sorted(),
                    message: "Created container \(logical) from \(image).")
            } catch {
                return .denied(message: "Container error: \(String(describing: error))")
            }

        case "exec":
            guard let logical = call.name.flatMap({ $0.isEmpty ? nil : $0 }),
                let spec = agentContainerSpecs[logical]
            else {
                return .denied(
                    message: "Unknown container. Create one first, or check `list` for the name.")
            }
            guard let command = call.command.flatMap({ $0.isEmpty ? nil : $0 }) else {
                return .denied(message: "Provide a `command` to exec.")
            }
            do {
                // Narrate a cold restart — an idle-stopped container has to be started again before
                // the command runs — the way "Booting the … VM…" does for a VM exec. A warm reuse is
                // instant and needs no note. Plain note (not `narratesTool`): unlike the lifecycle
                // ops, an `exec` keeps its tool card (it shows the command + output), so this line
                // sits alongside it rather than replacing it.
                if await !manager.isRunning(name: logical) {
                    emit(
                        .note(NoteEvent(
                            text: "Starting the Linux container \(logical)…", icon: "hourglass")),
                        nativeType: nil)
                }
                _ = try await manager.ensureRunning(spec)  // restart it if it idle-stopped
                defer { Task { await manager.finished(name: logical) } }
                var env: [String: String] = [:]
                if spec.controlSocketHostPath != nil, let server = runServer,
                    let token = serverToken
                {
                    await server.registerShellReport(token: token) { [weak self] calls in
                        Task { await self?.handleShellReport(calls) }
                    }
                    env = CommandInterceptor.hookEnv(
                        host: "127.0.0.1", port: ContainerSpec.controlBridgePort,
                        token: token, sessionID: sid, commandTracing: false,
                        environment: .container(name: logical))
                }
                let handle = try await manager.exec(
                    name: logical, workdir: call.workdir ?? spec.workdir, env: env,
                    argv: CommandInterceptor.agentShellArgv(command))
                let (code, out, err) = await drainContainerExec(handle)
                return .ran(exitCode: code, stdout: out, stderr: err)
            } catch {
                return .denied(message: "Container error: \(String(describing: error))")
            }

        case "stop":
            guard let logical = call.name.flatMap({ $0.isEmpty ? nil : $0 }),
                agentContainerSpecs[logical] != nil
            else {
                return .denied(message: "Unknown container. Check `list` for the name.")
            }
            announceLifecycleTransition("Stopping the Linux container \(logical)…", icon: "stop.circle")
            await manager.stopAgentContainer(name: logical)
            return .ok(
                name: logical, names: agentContainerSpecs.keys.sorted(),
                message: "Stopped \(logical) (its disk is kept; exec restarts it).")

        default:  // "remove"
            guard let logical = call.name.flatMap({ $0.isEmpty ? nil : $0 }),
                agentContainerSpecs[logical] != nil
            else {
                return .denied(message: "Unknown container. Check `list` for the name.")
            }
            announceLifecycleTransition("Removing the Linux container \(logical)…", icon: "trash")
            let removed = await manager.removeAgentContainer(name: logical)
            agentContainerSpecs[logical] = nil
            return .ok(
                name: logical, names: agentContainerSpecs.keys.sorted(),
                message: removed ? "Removed \(logical)." : "Removed \(logical) (was not running).")
        }
    }

    /// Drain a container `exec` handle to completion: collect stdout/stderr concurrently and await the
    /// exit code. The line streams strip the trailing newline, so re-add one per line. `nonisolated`
    /// so the two `async let` collectors run off-actor (and don't serialize on this actor).
    private nonisolated func drainContainerExec(
        _ handle: any ProcessHandle
    ) async -> (exitCode: Int32, stdout: String, stderr: String) {
        func collect(_ stream: AsyncThrowingStream<Data, Error>) async -> String {
            var acc = Data()
            do {
                for try await line in stream {
                    acc.append(line)
                    acc.append(0x0a)
                }
            } catch {}
            return String(decoding: acc, as: UTF8.self)
        }
        async let out = collect(handle.stdoutLines)
        async let err = collect(handle.stderrLines)
        let code = await handle.wait()
        return (code, await out, await err)
    }

    /// A session-scoped logical name for an agent-created container. `label` (if given) is sanitized to
    /// `[a-z0-9-]`; otherwise a `scratch-N` name is generated. Carries the build-channel suffix so
    /// launch-time disk GC (`ContainerManager.ownsContainer`) reaps only this channel's clones.
    private func agentContainerName(for sid: SessionID, label: String?) -> String {
        let base = "nucleic-scr-\(sid.short.lowercased())"
        let suffix = ContainerManager.channelSuffix
        if let label, !label.isEmpty {
            let cleaned = label.lowercased().map { $0.isLetter || $0.isNumber ? $0 : Character("-") }
            let safe = String(cleaned.prefix(24))
            let trimmed = safe.trimmingCharacters(in: CharacterSet(charactersIn: "-"))
            if !trimmed.isEmpty { return "\(base)-\(trimmed)\(suffix)" }
        }
        var n = 1
        while agentContainerSpecs["\(base)-scratch\(n)\(suffix)"] != nil { n += 1 }
        return "\(base)-scratch\(n)\(suffix)"
    }

    /// How the user answered a host-command conflict prompt (HOST_EXEC §concurrency).
    private enum HostConflictResolution {
        /// "Run anyway" — a false positive; override the detector and claim the slot now.
        case runAnyway
        /// "Wait" — join the FIFO queue and run once the conflicting command finishes.
        case wait
        /// "Cancel" — don't run; the agent is told to wait-and-retry itself.
        case skip
    }

    /// After the user chooses **Wait** on a host-command conflict (HOST_EXEC §concurrency): join the
    /// FIFO wait queue and poll until this session reaches the head and no similar command is still
    /// running, then run on the host and release the slot. Returns the host reply once it runs, or
    /// `nil` if the run was interrupted (Stop) before the slot freed — the caller turns that into a
    /// cancellation result. The queue keeps our place across polls, so waiters proceed in the order
    /// they chose Wait.
    ///
    /// Narrates the wait into the transcript so a parked run doesn't read as a hang: one muted note
    /// when we start waiting (naming the conflicting session), and another when the slot frees and the
    /// command actually starts. Mirrors the lock-event notes.
    private func waitForHostCommandSlot(
        sessionID sid: SessionID, command: String, workdir: String, shell: String? = nil,
        conflicts: [HostCommandConflict], coordinator: ConflictCoordinator
    ) async -> MCPApprovalServer.HostExecReply? {
        await coordinator.enqueueHostCommandWait(sessionID: sid, command: command)
        emit(
            .note(NoteEvent(
                text: "Waiting for \(Self.conflictSessionNames(conflicts)) to finish a similar "
                    + "host command, then running: \(Self.oneLine(command))",
                icon: "hourglass")),
            nativeType: nil)
        while true {
            // Bail the moment the user stops the run — don't leave a phantom in the queue.
            if interruptRequested {
                await coordinator.dequeueHostCommandWait(sessionID: sid)
                return nil
            }
            switch await coordinator.claimQueuedHostCommand(sessionID: sid, command: command) {
            case .clear:
                emit(
                    .note(NoteEvent(
                        text: "The host is free — now running: \(Self.oneLine(command))",
                        icon: "play.fill")),
                    nativeType: nil)
                let reply = await runOnHost(command: command, workdir: workdir, shell: shell)
                await coordinator.endHostCommand(sessionID: sid)
                return reply
            case .conflict:
                // Still blocked — the twin is running (or an earlier waiter is ahead). Sleep, then
                // re-poll; the wait queue holds our FIFO place across the gap. A cancelled sleep
                // (Task teardown) just loops back to the interrupt check above.
                try? await Task.sleep(nanoseconds: Self.hostCommandWaitPollInterval)
            }
        }
    }

    /// How often a queued host command re-checks whether its slot has freed (HOST_EXEC §concurrency).
    /// One second trades negligible latency for a near-idle poll — a build the waiter is parked behind
    /// runs for far longer than this.
    private static let hostCommandWaitPollInterval: UInt64 = 1_000_000_000

    /// Surface the in-chat override prompt when a host command would collide with a *similar* command
    /// already running in another session (HOST_EXEC §concurrency). The approval carries a
    /// `host_command_conflict` payload (which session, its command, why) that the GUI renders as a
    /// dedicated card with **Run anyway** / **Wait** / **Cancel**. Deliberately one-shot — a false
    /// positive now isn't remembered, so the next overlap is re-evaluated on its own merits. "Wait"
    /// and "Run anyway" both arrive as non-deny `.allow`; they're told apart by the
    /// `HostCommandConflictSignal.waitInputKey` flag the Wait button sets in `updatedInput`.
    private func surfaceHostCommandConflict(command: String, conflicts: [HostCommandConflict]) async
        -> HostConflictResolution
    {
        let request = ApprovalRequest(
            id: .generate(),
            sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            toolCallID: nil,
            toolName: MCPApprovalServer.qualifiedHostExecToolName,
            input: .object([
                "command": .string(command),
                "host_command_conflict": .object([
                    "with": .array(conflicts.map { c in
                        .object([
                            "session": .string(c.sessionTitle),
                            "command": .string(c.command),
                            "reason": .string(c.reason),
                        ])
                    })
                ]),
            ]),
            title: "Possible conflict with \(Self.conflictSessionNames(conflicts)): "
                + "a similar host command is already running",
            risk: .hostExec,
            createdAt: configuration.now())

        emit(.approvalRequested(request), nativeType: "mcp/tools/call")
        let resolved = await approvals.waitForDecision(request)
        emit(.approvalResolved(resolved), nativeType: nil)

        switch resolved.decision {
        case .allow(let updatedInput):
            // "Wait" and "Run anyway" both come back as `.allow`; the Wait button flags itself in
            // `updatedInput`, so a present-and-true flag means queue, otherwise override now.
            if updatedInput?[HostCommandConflictSignal.waitInputKey]?.boolValue == true {
                return .wait
            }
            return .runAnyway
        case .allowAlways:
            return .runAnyway  // a blanket allow → override the detector
        case .deny:
            return .skip
        case .cancelRun:
            interruptRequested = true
            handle?.sendSignal(SIGINT)
            return .skip
        }
    }

    /// Name the conflicting session(s) for the prompt title / denial message: "session 'A'",
    /// "sessions 'A' and 'B'", or "sessions 'A', 'B', and 1 more".
    private static func conflictSessionNames(_ conflicts: [HostCommandConflict]) -> String {
        let titles = conflicts.map { "'\($0.sessionTitle)'" }
        switch titles.count {
        case 0: return "another session"
        case 1: return "session \(titles[0])"
        case 2: return "sessions \(titles[0]) and \(titles[1])"
        default:
            return "sessions \(titles.prefix(2).joined(separator: ", ")), and \(titles.count - 2) more"
        }
    }

    /// The tool-result text when the user skips a conflicting host command: name the session(s) and
    /// frame it as "wait and retry," not a hard failure, so the agent coordinates rather than gives up.
    private static func hostConflictDenial(_ conflicts: [HostCommandConflict]) -> String {
        let detail = conflicts.first.map { " (\($0.sessionTitle) is running `\(oneLine($0.command))`)" } ?? ""
        return "Not run: a similar host command is already in progress in "
            + "\(conflictSessionNames(conflicts))\(detail). To avoid two host commands colliding "
            + "(e.g. simultaneous builds sharing the host, or a shared trunk's build directory), the "
            + "user chose not to run both at once. Wait for the other to finish and retry, or run a "
            + "non-overlapping command."
    }

    /// Run `command` on the host shell in the session's worktree, capturing stdout/stderr (each
    /// capped) and the exit code. Runs OUTSIDE the container — this is the deliberate sandbox
    /// escape, reached only after explicit approval. The shell is `hostShellArgv`'s pick
    /// (docs/NASH.md §7.7.6): bundled nash `-c` when `nucleic.host.shell` = nash, else the
    /// historical `/bin/zsh -lc`; `shell: "zsh"` is the per-call escape hatch.
    private func runOnHost(command: String, workdir: String, shell: String? = nil) async
        -> MCPApprovalServer.HostExecReply
    {
        // Pin the git/gh the agent runs on the host (`git push`, `gh pr create`, a signed `git
        // commit`, …) to the app's Managed Git credential — the same hermetic env Nucleic's own
        // host-side git uses (`provisionHost`: GIT_SSH_COMMAND on the configured key, IdentitiesOnly /
        // IdentityAgent=none / -F /dev/null, plus the SSH-signing config). Without it, the login shell
        // below inherits the user's ssh-agent and `~/.ssh` (e.g. 1Password), so the agent's push
        // authenticates with the user's *personal* key — never the Managed Git key — and prompts for
        // authorization. GIT_SSH_COMMAND pins git's ssh to the Managed Git key for *every* remote, not
        // just github.com; it governs only git, so a standalone `ssh somehost` the agent runs is
        // unaffected. No-op when no Managed Git credential is configured. GH_CONFIG_DIR is dropped: it
        // carries only `git_protocol: ssh` (no gh auth), and forcing it would break gh's *API* auth when
        // no token is set (gh resolves auth from GH_TOKEN or the host gh login instead); SSH is pinned.
        let (gitEnv, gitCleanup) = GitHubCredentialProvisioner.provisionHost()
        defer { gitCleanup() }
        var env = gitEnv
        env.removeValue(forKey: "GH_CONFIG_DIR")
        let argv = CommandInterceptor.hostShellArgv(command, forceZsh: shell == "zsh")
        // nash on the host (docs/NASH.md §7.7): wire its observation at this run's own control
        // endpoint so host_exec exec events reach the feed — direct loopback POST on a host run
        // (§6.2's simplest surface), or the run's control socket host path on a containerized run.
        // Capture defaults to `meta` (§7.7.4: the host filesystem is the user's). Skipped entirely
        // on the zsh path, where the env would be inert anyway.
        if argv[0] != "/bin/zsh", let sid = sessionID {
            if let endpoint = hostControlEndpoint {
                env.merge(
                    CommandInterceptor.hookEnv(
                        host: endpoint.host, port: endpoint.port, token: endpoint.token,
                        sessionID: sid, commandTracing: false,
                        shellCapture: HostShellSettings.shellCapture, environment: .host)
                ) { _, new in new }
            } else if let socket = hostShellSocketPath, let token = serverToken {
                env["NUCLEIC_SHELL_SOCKET"] = socket
                env["NUCLEIC_HOOK_TOKEN"] = token
                env["NUCLEIC_SESSION_ID"] = sid.rawValue
                env["NUCLEIC_SHELL_CAPTURE"] = HostShellSettings.shellCapture
                env["NUCLEIC_SHELL_ENVIRONMENT_KIND"] = NashEnvironment.host.kind.rawValue
                env["NUCLEIC_SHELL_ENVIRONMENT_ID"] = NashEnvironment.host.id
                env["NUCLEIC_SHELL_ENVIRONMENT_LABEL"] = NashEnvironment.host.label
            }
        }
        let spec = ProcessSpec(
            executable: argv[0],
            args: Array(argv.dropFirst()),
            cwd: workdir,
            env: env,
            stdinMode: .closed,
            // No Nucleic subprocess may use a credential Nucleic didn't mediate. Strip any ambient
            // Claude/Codex API key the app happened to inherit so it can't reach a host command
            // (the login shell may still re-export one from the user's own dotfiles — that's the
            // user's host, outside Nucleic's control).
            removeFromEnvironment: ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "CODEX_API_KEY"])
        do {
            let handle = try await processHost.launch(spec)
            // Track it so Stop / session teardown can reap this command (and its process tree).
            hostCommandSeq += 1
            let commandID = hostCommandSeq
            inFlightHostHandles[commandID] = handle
            defer { inFlightHostHandles[commandID] = nil }

            // Watch for a hang: if this command goes silent with near-zero CPU, raise an in-transcript
            // alert with a Kill button (``reportHostStall``) so the user isn't stuck waiting out the
            // 30-minute wall-clock timeout below. The monitor never touches the process — it only
            // observes and reports; the Kill button (``resolveProcessStall``) does the tree-kill.
            let stallID = UUID().uuidString
            let stallLabel = Self.hostCommandLabel(command)
            let cpuSampler: (@Sendable () -> UInt64)?
            let ioSampler: (@Sendable () -> UInt64)?
            if handle.supportsHostCPUSampling {
                let pid = handle.processID
                cpuSampler = { hostProcessTreeCPUNanos(of: pid) }
                ioSampler = { hostProcessTreeIOBytes(of: pid) }
            } else {
                cpuSampler = nil
                ioSampler = nil
            }
            // When the pipeline's terminal stage is a non-follow `tail`/`head`, it buffers its whole
            // input until EOF, so the captured stdout is silent by construction — "no output" is then
            // a meaningless liveness signal, and the stall verdict must rest on CPU/I/O alone.
            let outputIdleCounts = !Self.terminalStageBuffersOutput(command)
            let stallMonitor = ProcessStallMonitor(
                handle: handle,
                cpuSampler: cpuSampler,
                ioSampler: ioSampler,
                outputIdleCounts: outputIdleCounts,
                onStall: { [weak self] idleSeconds in
                    await self?.reportHostStall(
                        stallID: stallID, label: stallLabel, handle: handle, idleSeconds: idleSeconds)
                },
                onResolve: { [weak self] resolution in
                    await self?.resolveHostStallInternal(stallID: stallID, resolution: resolution.rawValue)
                })
            stallMonitor.start()
            defer { stallMonitor.cancel() }

            // Drain both pipes concurrently so a full stderr buffer can't deadlock stdout.
            async let out = Self.collect(handle.stdoutLines)
            async let err = Self.collect(handle.stderrLines)

            // Bound the wait: a hung host command (a deadlocked `swift test`, a wedged build) must not
            // wedge the tool call forever — the MCP tool timeouts are pinned effectively infinite for
            // human-in-the-loop calls, so nothing else caps this. On timeout, reap the WHOLE process
            // tree (the login shell forks make/swift/xctest/…; killing just the shell orphans them and
            // they keep the shared trunk's `.build` lock) and hand the agent a real result.
            //
            // But the wall clock alone is NOT evidence of a hang, and a long build that is still
            // compiling must be allowed to finish. So the ceiling only reaps a command that `stallMonitor`
            // has independently confirmed wedged — output-silent AND under both the CPU and disk-I/O
            // thresholds, sustained. A command that is merely slow gets another window, re-checked on the
            // same terms, for as long as it keeps doing work; the user still sees the stall alert (with
            // its Kill button) and Stop, which are the interactive ways out.
            let timeout = configuration.hostCommandTimeoutNanos
            var waited: UInt64 = 0
            var window = timeout
            while !(await handle.waitForExit(within: window)) {
                waited &+= window
                // Past the ceiling, re-examine on a shorter cadence rather than granting another full
                // ceiling — a command that wedges just after the 30m mark shouldn't hold the tool call
                // for another 30. Each pass leaves one background waiter parked on `wait()` (harmless,
                // and it resolves on exit), so the cadence is minutes, not seconds; the user already
                // gets the interactive stall alert within ~6m of real silence either way.
                window = Self.hostStallRecheckNanos
                guard stallMonitor.isStalled else { continue }  // still working — let it run
                await handle.terminateTree()
                let (stdout, stderr) = await (out, err)
                let minutes = waited / 1_000_000_000 / 60
                let note = "… host command ran \(minutes)m and then went idle — no output and "
                    + "near-zero CPU and disk I/O — so it was treated as wedged and terminated (its "
                    + "process tree was killed). Re-run with a narrower scope, or increase the host "
                    + "command timeout if it legitimately needs longer."
                return .ran(
                    exitCode: -1, stdout: stdout,
                    stderr: stderr.isEmpty ? note : stderr + "\n" + note)
            }
            let code = await handle.wait()
            let (stdout, stderr) = await (out, err)
            return .ran(exitCode: code, stdout: stdout, stderr: stderr)
        } catch {
            return .ran(
                exitCode: -1, stdout: "",
                stderr: "Failed to launch host command: \(error)")
        }
    }

    /// A short, human-friendly label for a host command line — the first meaningful token(s), so the
    /// stall alert reads "…“swift test”…" rather than the whole `zsh -lc` script. Best-effort.
    static func hostCommandLabel(_ command: String) -> String {
        let trimmed = command.trimmingCharacters(in: .whitespacesAndNewlines)
        // Take the first line, then cap length so a giant one-liner doesn't blow out the note.
        let firstLine = trimmed.split(whereSeparator: \.isNewline).first.map(String.init) ?? trimmed
        return firstLine.count > 60 ? String(firstLine.prefix(60)) + "…" : firstLine
    }

    /// Whether the pipeline's *terminal* stage withholds its output until upstream EOF — a non-follow
    /// `tail`, which can't know its "last N" lines until the stream ends and so buffers everything
    /// until then. When it does, the captured stdout stays silent for the whole run no matter how
    /// alive the command is, so ``ProcessStallMonitor`` must not count "no output" against it (agents
    /// routinely wrap a long, quiet command in `| tail -6` for token thrift, which otherwise reads as
    /// a hang). A following `tail -f`/`-F` streams and is *not* buffering; and `head`, which emits its
    /// first N lines as they arrive, streams too — so only non-follow `tail` qualifies. Uses the same
    /// parsed, ordered ``HostCommandSummary`` invocations the approval UI relies on, so it sees the
    /// real last program of the pipeline.
    static func terminalStageBuffersOutput(_ command: String) -> Bool {
        guard let last = HostCommandSummary.summary(for: command)?.invocations.last,
            last.program == "tail"
        else { return false }
        // `-f`/`-F`/`--follow`/`--retry` make tail stream; anything else buffers to EOF.
        return !last.flags.contains { flag in
            let name = flag.name.drop(while: { $0 == "-" })
            if flag.name.hasPrefix("--") {
                return name == "follow" || name == "retry"
            }
            // A short flag (possibly bundled, e.g. `-fn`) that includes `f`/`F` is a follow.
            return name.contains("f") || name.contains("F")
        }
    }

    /// The stall monitor observed this host command go silent + idle-CPU: raise the in-transcript
    /// alert (a distinct note the UI renders with a Kill button) and remember the handle so the Kill
    /// button can find it. Runs on the actor (the monitor hops here).
    private func reportHostStall(
        stallID: String, label: String, handle: any ProcessHandle, idleSeconds: Int
    ) {
        guard stalledHostHandles[stallID] == nil else { return }  // already alerted (re-arm no-op)
        stalledHostHandles[stallID] = handle
        let pidCount = 1 + hostDescendantPIDs(of: handle.processID).count
        let text = "Host command “\(label)” looks hung — no output for \(idleSeconds)s and near-zero "
            + "CPU. It may be wedged; kill it, or keep waiting."
        emit(
            .note(NoteEvent(
                text: text,
                icon: "exclamationmark.triangle",
                processStall: ProcessStallNote(
                    id: stallID, label: label, pidCount: pidCount, idleSeconds: idleSeconds))),
            nativeType: nil)
    }

    /// The stall self-resolved (the command produced output again, or exited before the user acted):
    /// close out the alert with a follow-up note. Only fires when an alert was actually live.
    private func resolveHostStallInternal(stallID: String, resolution: String) {
        guard stalledHostHandles[stallID] != nil else { return }
        stalledHostHandles[stallID] = nil
        let text: String = resolution == StallResolution.resumed.rawValue
            ? "The host command started producing output again — no longer looks hung."
            : "The host command exited on its own."
        emit(
            .note(NoteEvent(
                text: text,
                icon: "checkmark.circle",
                processStall: ProcessStallNote(
                    id: stallID, label: "", pidCount: 0, idleSeconds: 0, resolution: resolution))),
            nativeType: nil)
    }

    /// The user resolved a live stall alert from the transcript: `kill` tears down the command's whole
    /// process tree (its `runOnHost` `waitForExit` then returns a terminated result), otherwise the
    /// alert is just dismissed and the command keeps running. Emits the closing resolution note.
    public func resolveProcessStall(stallID: String, kill: Bool) async {
        guard let handle = stalledHostHandles[stallID] else { return }  // already resolved
        stalledHostHandles[stallID] = nil
        if kill {
            // Detached, self-bounding SIGTERM→SIGKILL over the whole tree — same reap Stop uses.
            Task { await handle.terminateTree() }
            emit(
                .note(NoteEvent(
                    text: "Killed the hung host command's process tree.",
                    icon: "xmark.octagon",
                    processStall: ProcessStallNote(
                        id: stallID, label: "", pidCount: 0, idleSeconds: 0, resolution: "killed"))),
                nativeType: nil)
        } else {
            emit(
                .note(NoteEvent(
                    text: "Dismissed the hang alert — the host command is still running.",
                    icon: "bell.slash",
                    processStall: ProcessStallNote(
                        id: stallID, label: "", pidCount: 0, idleSeconds: 0, resolution: "dismissed"))),
                nativeType: nil)
        }
    }

    /// Reap every in-flight host `host_exec` command and its process tree. Called on Stop
    /// (``interrupt``) and session teardown (``shutdown``) so a running host build/test doesn't
    /// outlive the turn/session as an orphan holding the shared trunk's SwiftPM build lock. Each
    /// ``runOnHost`` also self-heals its own map entry via `defer`; clearing here first makes the
    /// reap idempotent against that. Reaps run concurrently and each self-bounds (SIGTERM→SIGKILL
    /// with a grace), so awaiting this stays quick even with several commands in flight.
    private func reapInFlightHostCommands() async {
        let handles = Array(inFlightHostHandles.values)
        inFlightHostHandles.removeAll()
        guard !handles.isEmpty else { return }
        await withTaskGroup(of: Void.self) { group in
            for handle in handles {
                group.addTask { await handle.terminateTree() }
            }
        }
    }

    /// Collect a line stream into one string, capped so a runaway build log can't blow up
    /// the tool result the agent reads back.
    private static func collect(_ stream: AsyncThrowingStream<Data, Error>, cap: Int = 64 * 1024)
        async -> String
    {
        var data = Data()
        var truncated = false
        do {
            for try await line in stream {
                if data.count < cap {
                    data.append(line)
                    data.append(0x0A)
                } else {
                    truncated = true
                }
            }
        } catch {}
        var text = String(decoding: data, as: UTF8.self)
        if truncated { text += "\n…(output truncated)" }
        return text
    }

    /// Records the run's host `cwd` onto a `host_exec` tool call event, so the transcript's in-chat
    /// host card resolves `$PWD`-style substitutions to the exact directory the command ran in — the
    /// same value the approval card uses (the agent's raw tool call carries only `command`/`reason`).
    /// A no-op for every other event/tool, when host_exec isn't exposed (`workdir == nil`), or when a
    /// `cwd` is already present — so it's idempotent across the started/completed pair for one call.
    static func recordingHostExecCwd(_ kind: AgentEvent.Kind, workdir: String?) -> AgentEvent.Kind {
        guard let workdir else { return kind }
        func enrich(_ call: ToolCall) -> ToolCall {
            guard call.name == MCPApprovalServer.qualifiedHostExecToolName,
                  var fields = call.input.objectValue, fields["cwd"] == nil
            else { return call }
            fields["cwd"] = .string(workdir)
            return ToolCall(
                toolCallID: call.toolCallID, name: call.name,
                input: .object(fields), parentToolCallID: call.parentToolCallID)
        }
        switch kind {
        case .toolCallStarted(let call): return .toolCallStarted(enrich(call))
        case .toolCallCompleted(let call): return .toolCallCompleted(enrich(call))
        default: return kind
        }
    }

    private static func oneLine(_ command: String, max: Int = 80) -> String {
        let flat = command.replacingOccurrences(of: "\n", with: " ")
            .trimmingCharacters(in: .whitespaces)
        return flat.count <= max ? flat : String(flat.prefix(max - 1)) + "…"
    }

    // MARK: - Helpers

    private func emit(_ kind: AgentEvent.Kind, nativeType: String?) {
        if let platformToolEventSink {
            platformToolEventSink(kind, nativeType)
            return
        }
        provisionalSeq += 1
        let event = AgentEvent(
            sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            seq: provisionalSeq,
            at: configuration.now(),
            backend: .claudeCode,
            nativeType: nativeType,
            kind: kind)
        continuation?.yield(event)
    }

    /// Forward a ground-truth git op reported by the in-container interceptor shim (hopped onto
    /// this actor from the fire-and-forget `/git-event` handler) to the host via the shared
    /// `ConflictCoordinator` — the same seam `check_conflict` uses. `AppStore` owns the
    /// semantics: releasing file locks the moment a merge lands ([[git-event]] → the lock
    /// system) and recording an agent-performed merge as an autoship ship. Best-effort: dropped
    /// when there's no coordinator (pipeline tests).
    private func handleGitReport(_ call: MCPApprovalServer.GitReportCall) async {
        guard let sessionID, let conflictCoordinator else { return }
        await conflictCoordinator.observeGitOp(
            sessionID: sessionID, argv: call.argv, exitCode: call.exitCode)
    }

    /// Forward a ground-truth `gh` action reported by the in-container interceptor shim to the host
    /// via the shared `ConflictCoordinator` — the same seam as `handleGitReport`. `AppStore` records
    /// the agentic GitHub action (open/merge a PR, cut a release, …) in the Control activity feed.
    /// Best-effort: dropped when there's no coordinator (pipeline tests).
    private func handleGhReport(_ call: MCPApprovalServer.GhReportCall) async {
        guard let sessionID, let conflictCoordinator else { return }
        await conflictCoordinator.observeGhOp(
            sessionID: sessionID, argv: call.argv, exitCode: call.exitCode)
    }

    /// Forward a non-git command the interceptor reported (shim or bash tracer) to the host via
    /// the shared `ConflictCoordinator` — the same seam `handleGitReport` uses. `AppStore` owns the
    /// semantics: classify, gate to Nucleic Control sessions, and record into the command-activity
    /// feed + structured log. Best-effort: dropped when there's no coordinator (pipeline tests).
    private func handleCommandReport(_ call: MCPApprovalServer.CommandReportCall) async {
        guard let sessionID, let conflictCoordinator else { return }
        await conflictCoordinator.observeCommand(sessionID: sessionID, call: call)
    }

    /// Forward a nash agent-shell event (`/shell-event`) to the host via the shared
    /// `ConflictCoordinator` — the same seam as `handleCommandReport`. `AppStore` owns the
    /// semantics: converge exec events onto the command/git/gh feeds and log shell state changes
    /// (docs/NASH.md §9). Best-effort: dropped when there's no coordinator (pipeline tests).
    private func handleShellReport(_ calls: [MCPApprovalServer.ShellReportCall]) async {
        guard let sessionID, let conflictCoordinator else { return }
        await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
    }

    /// NDJSON user-message envelope (ADAPTERS §1.3 — pinned by the M0 live capture).
    static func userMessageLine(for input: AgentInput) -> Data {
        var blocks: [JSONValue] = []
        for part in input.parts {
            switch part {
            case .text(let text):
                blocks.append(.object(["type": .string("text"), "text": .string(text)]))
            case .context(let label, let body):
                blocks.append(
                    .object(["type": .string("text"), "text": .string("[\(label)]\n\(body)")]))
            case .attachment(let attachment):
                // Attachments are normally materialized to disk and folded into the text before a
                // turn reaches any backend (`AppStore.materializeRemoteAttachments`), so this is a
                // defensive fallback: reference the file by name rather than drop it silently.
                blocks.append(.object([
                    "type": .string("text"),
                    "text": .string("[Attached file: \(attachment.filename)]")]))
            }
        }
        let line = JSONValue.object([
            "type": .string("user"),
            "message": .object(["role": .string("user"), "content": .array(blocks)]),
        ])
        return (try? line.encodedData()) ?? Data()
    }

    private func makeCaptureFile() throws -> CaptureFile? {
        guard let directory = configuration.captureDirectory else { return nil }
        try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
        return try CaptureFile(url: directory.appendingPathComponent("native.ndjson"))
    }
}

/// Ring of recent stderr lines, kept for the synthesized error message when the
/// CLI exits abnormally.
private final class StderrTail: @unchecked Sendable {
    private let lock = NSLock()
    private var lines: [String] = []

    func append(_ line: String) {
        lock.lock()
        defer { lock.unlock() }
        lines.append(line)
        if lines.count > 50 { lines.removeFirst(lines.count - 50) }
    }

    func joined() -> String {
        lock.lock()
        defer { lock.unlock() }
        return lines.suffix(10).joined(separator: "\n")
    }
}

private final class CaptureFile: @unchecked Sendable {
    private let handle: FileHandle
    private let lock = NSLock()

    init(url: URL) throws {
        FileManager.default.createFile(atPath: url.path, contents: nil)
        handle = try FileHandle(forWritingTo: url)
    }

    func append(line: Data) {
        lock.lock()
        defer { lock.unlock() }
        var framed = line
        framed.append(0x0A)
        try? handle.write(contentsOf: framed)
    }

    func close() {
        lock.lock()
        defer { lock.unlock() }
        try? handle.close()
    }
}
