import Foundation

/// `AgentBackend` over the Codex CLI's `app-server` (ADAPTERS §2; BACKEND_PROTOCOL §7.2).
/// Spawns `codex app-server`, speaks JSON-RPC 2.0 over its stdio (`JSONRPCConnection`, with the
/// `"jsonrpc"` header omitted — codex's confirmed dialect),
/// normalizes `thread/`·`turn/`·`item/` notifications into `AgentEvent`s
/// (`CodexAppServerDecoder`), and answers interactive approvals — which the server sends
/// as JSON-RPC server→client requests — through the shared `ApprovalCoordinator`, exactly
/// the seam Claude's MCP approval bridge suspends on (ADAPTERS §5).
///
/// **Turn model (v0).** Like `ClaudeCodeBackend`, each turn is a single-shot run: `start`
/// spawns the process and opens a thread; each follow-up `resume` spawns a fresh process
/// and `thread/resume`s the same thread by id (the documented resume mechanism, ADAPTERS
/// §2.6). A persistent connection with `turn/steer` follow-ups is a later optimization —
/// hence `followUpWhileRunning = false`.
///
/// Wire shapes below are source-confirmed against `openai/codex`
/// (`codex-rs/app-server-protocol`, commit `7a19b14`); byte-level confirmation against a
/// live `codex` awaits a fixture capture (no `codex` on the dev machine — M3).
public actor CodexAppServerBackend: AgentBackend {
    public static let id = BackendID.codex

    public nonisolated let capabilities = BackendCapabilities(
        interactiveApprovals: true,
        allowAlwaysScopes: [.session],          // Codex's only "always" is acceptForSession
        canModifyToolInput: false,              // no arbitrary command rewrite (ADAPTERS §2.4)
        partialMessageStreaming: true,          // item/*/delta notifications
        emitsThinking: true,                    // reasoning items
        emitsFileChangeEvents: true,            // native + authoritative fileChange items
        nativeResume: true,                     // thread/resume
        sandboxModes: [.readOnly, .workspaceWrite, .fullAccess],
        followUpWhileRunning: false)

    public struct Configuration: Sendable {
        /// Executable name (resolved via PATH) or absolute path.
        public var executable: String
        public var clientName: String
        public var clientVersion: String
        /// Interactive approval policy passed to `thread/start`.
        public var approvalPolicy: String
        public var now: @Sendable () -> Date

        public init(
            executable: String = "codex",
            clientName: String = "Nucleic",
            clientVersion: String = "0.1",
            approvalPolicy: String = "on-request",
            now: @escaping @Sendable () -> Date = { Date() }
        ) {
            self.executable = executable
            self.clientName = clientName
            self.clientVersion = clientVersion
            self.approvalPolicy = approvalPolicy
            self.now = now
        }
    }

    private let configuration: Configuration
    private let processHost: ProcessHost
    public let approvals: ApprovalCoordinator
    /// Optional sandbox orchestrator. When set and a run carries a vsock control container, the
    /// agent execs inside it (stdio over vsock) — the per-family control container isolation. `nil` →
    /// always host. Codex's approvals are native (JSON-RPC, no MCP), but a control container still
    /// runs the git/gh/command interceptor, whose reports reach the shared per-container server.
    private let containerManager: ContainerManager?
    /// Provider-neutral Nucleic platform-tool handlers shared with Claude.
    private let platformToolRuntime: ClaudeCodeBackend
    /// Name of the container the current run execs in (for the teardown `finished` callback).
    private var activeContainerName: String?
    /// Conflict coordinator (locks + autoship). Joins Codex to the lock/arbitration system: the
    /// container's `PreToolUse` hook (``CodexToolGate``, docs/LOCKING.md §4.6) acquires locks on the
    /// paths each `apply_patch`/`Bash` call writes *before* it runs — the gate that actually fires,
    /// since codex auto-approves patches inside its writable roots — with the file-change approval
    /// below as a second gate for runs whose policy does ask. The interceptor reports merges/commands
    /// for release + the activity feed. `nil` → not joined.
    private let conflictCoordinator: ConflictCoordinator?
    /// Vends the shared, token-multiplexed Nucleic MCP server for a control container. Native
    /// command/file approvals still arrive over app-server JSON-RPC; Nucleic-owned tools use MCP.
    private let approvalServerRegistry: ApprovalServerRegistry?
    /// The interceptor server + token for the current containerized run (for teardown unregister).
    private var interceptorServer: MCPApprovalServer?
    private var interceptorToken: String?
    /// Edited paths announced by each `fileChange` item (id → paths), captured from the item
    /// lifecycle so the file-change approval (which carries only the item id) can arbitrate locks.
    private var fileChangePathsByItem: [String: [String]] = [:]

    private var handle: (any ProcessHandle)?
    private var rpc: JSONRPCConnection?
    private var continuation: AsyncThrowingStream<AgentEvent, Error>.Continuation?
    private var sessionID: SessionID?
    /// The directory the agent's shell runs in (the session container's workdir when sandboxed,
    /// else the host worktree), used to resolve a raw-shell write's relative target into a lockable
    /// path before the command runs (docs/NASH.md §5.2).
    private var agentShellWorkdir: String?
    private var provisionalSeq: UInt64 = 0
    private var interruptRequested = false
    /// Nucleic-owned auto policy for the active turn; provider auto-review is never used.
    private var autoApprove = false
    private var terminating = false
    private var sawRunFinished = false
    /// The open thread's id (kept across turns so `resume` reattaches the same thread).
    private var currentThreadID: String?
    /// The in-flight turn's id, needed for `turn/interrupt`.
    private var currentTurnID: String?
    public private(set) var backendSessionID: String?

    public init(
        configuration: Configuration = Configuration(),
        processHost: ProcessHost = ProcessHost(),
        approvals: ApprovalCoordinator = ApprovalCoordinator(),
        containerManager: ContainerManager? = nil,
        macVMManager: MacVMManager? = nil,
        conflictCoordinator: ConflictCoordinator? = nil,
        approvalServerRegistry: ApprovalServerRegistry? = nil
    ) {
        self.configuration = configuration
        self.processHost = processHost
        self.approvals = approvals
        self.containerManager = containerManager
        self.platformToolRuntime = ClaudeCodeBackend(
            processHost: processHost, approvals: approvals,
            containerManager: containerManager, macVMManager: macVMManager,
            conflictCoordinator: conflictCoordinator,
            approvalServerRegistry: approvalServerRegistry)
        self.conflictCoordinator = conflictCoordinator
        self.approvalServerRegistry = approvalServerRegistry
    }

    // MARK: - AgentBackend

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

    public nonisolated func resume(_ resume: ResumeSpec) -> AsyncThrowingStream<AgentEvent, Error> {
        let run = Self.runSpec(resuming: resume)
        return makeStream(run: run, resumeThreadID: resume.backendSessionID)
    }

    /// Rebuild every per-turn capability on resume; Codex app-server is a fresh process each turn.
    nonisolated static func runSpec(resuming resume: ResumeSpec) -> RunSpec {
        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,
            orchestraPlanHandler: resume.orchestraPlanHandler,
            orchestraSuperviseHandler: resume.orchestraSuperviseHandler,
            orchestraReplyToWorkerHandler: resume.orchestraReplyToWorkerHandler,
            orchestraAskSupervisorHandler: resume.orchestraAskSupervisorHandler,
            meshMessaging: resume.meshMessaging,
            appendSystemPrompt: resume.appendSystemPrompt)
    }

    /// Always keep Codex's reviewer on the client path. Nucleic applies both interactive and
    /// auto policy itself so the same UI and risk rules govern every provider.
    nonisolated static func approvalsReviewer(autoApprove _: Bool) -> String {
        "user"
    }

    /// Configuration that must be applied identically when a thread is first opened and whenever
    /// it is resumed in a fresh app-server process. In particular, a containerized resume must
    /// override a legacy/persisted workspace sandbox or Codex will try to start bwrap inside the
    /// already-sandboxed Control container before a command can run.
    nonisolated static func threadRuntimeParams(
        for run: RunSpec, isHostRun: Bool, approvalPolicy: String
    ) -> [String: JSONValue] {
        let containerized = !isHostRun
        var params: [String: JSONValue] = [
            "approvalPolicy": .string(approvalPolicy),
            "approvalsReviewer": .string(Self.approvalsReviewer(autoApprove: run.autoApprove)),
            "sandbox": .string(CodexRuntimeEnvironment.sandboxMode(
                containerized: containerized, requested: run.sandbox).rawValue),
        ]
        if let instructions = CodexRuntimeEnvironment.developerInstructions(
            for: run, containerized: containerized)
        {
            params["developerInstructions"] = .string(instructions)
        }
        return params
    }

    public func send(_ input: AgentInput) async throws {
        // Conversational follow-ups arrive via resume() (single-shot turn model); there is
        // no mid-stream send in v0. Non-conversational streaming would use turn/steer.
        throw BackendError.unsupported("Codex follow-ups resume a new turn; send() is unused in v0")
    }

    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
        // Platform tools are served by a provider-neutral helper. Reap any host command it launched
        // too, so Stop has the same whole-turn semantics as it does for Claude.
        await platformToolRuntime.interruptPlatformToolProcesses()
        await interruptCurrentTurn()
    }

    /// Host commands run through the nested Nucleic tool runtime, so a hang alert (and its Kill) is
    /// resolved there.
    public func resolveProcessStall(stallID: String, kill: Bool) async {
        await platformToolRuntime.resolveProcessStall(stallID: stallID, kill: kill)
    }

    public func shutdown() async {
        terminating = true
        await approvals.cancelOutstanding(reason: "Session terminated")
        await teardownTurn()
    }

    // MARK: - Run loop

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

    private func runLoop(
        run: RunSpec,
        resumeThreadID: 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.autoApprove = run.autoApprove

        let stderrTail = CodexStderrTail()
        do {
            // 1. Spawn `codex app-server` — inside the shared control container (stdio over vsock)
            // when this run carries a vsock control socket, else on the host. Gated on that signal,
            // so default behavior is unchanged. (Requires `codex` + its auth in the sandbox image.)
            // 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 — after the agent process ends). A
            // containerized run gets the key via `provision` (cspec.env). No-op when unset.
            let isHostRun = !(containerManager != nil && run.container?.controlSocketHostPath != 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() }

            let handle: any ProcessHandle
            if let containerManager, let cspec = run.container, cspec.controlSocketHostPath != nil {
                // The shared control server is resolved BEFORE the container comes up:
                // ContainerControlPlane binds it first on Windows (gateway TCP baked into the
                // create env) and after create on macOS (the relayed socket) — see §5.
                var controlServer: MCPApprovalServer?
                if let registry = approvalServerRegistry {
                    controlServer = await registry.server(for: cspec.name)
                }
                let ready = try await ContainerControlPlane.bringUp(
                    spec: cspec, manager: containerManager, server: controlServer)
                let name = ready.containerName
                activeContainerName = name
                var env = cspec.env.merging(run.extraEnv) { _, new in new }
                let externalMCPArgs = ExternalAgentIntegrationSettings.allows(.mcp, model: run.model)
                    ? []
                    : ["-c", "mcp_servers={}"]
                // Extra `codex` args (before the `app-server` subcommand) that wire it to our MCP
                // endpoint when the per-container control server is up.
                var mcpArgs: [String] = []
                // Git/gh/command interceptor: Codex's in-container shims POST observed ops to the
                // shared per-container approval server (over the relayed socket, via the in-guest
                // bridge), which forwards them to the conflict coordinator (release + activity feed),
                // exactly as Claude/Grok. Codex's file/command approvals stay native; the same server
                // also exposes Nucleic's MCP tools (starting with `check_conflict`) at `/mcp`.
                if let server = controlServer {
                    let token = UUID().uuidString
                    interceptorServer = server
                    interceptorToken = token
                    await server.registerGitReport(token: token) { [weak self] call in
                        Task { await self?.handleGitReport(call) }
                    }
                    await server.registerGhReport(token: token) { [weak self] call in
                        Task { await self?.handleGhReport(call) }
                    }
                    await server.registerCommandReport(token: token) { [weak self] call in
                        Task { await self?.handleCommandReport(call) }
                    }
                    await server.registerShellReport(token: token) { [weak self] calls in
                        Task { await self?.handleShellReport(calls) }
                    }
                    // The edit gate (docs/LOCKING.md §4.6). Codex has no client-side choke point
                    // for its internal `apply_patch` — it applies a patch without asking whenever
                    // the write stays in its writable roots, which under Nucleic is always — so the
                    // `item/fileChange/requestApproval` arbitration below never fired and Codex
                    // edits landed unlocked. The container's root-owned managed config points every
                    // codex here instead, and this handler is where the edit acquires its locks.
                    // Registered whenever there's a control server (not only when a coordinator is
                    // attached): the hook fails closed on a token it can't authenticate, so an
                    // unregistered token would deny the run's tool calls.
                    await server.registerPreToolUse(token: token) { [weak self] call in
                        guard let self else {
                            return .deny(reason: "This Nucleic session has ended; stop editing.")
                        }
                        return await self.gatePreToolUse(call)
                    }
                    await platformToolRuntime.configurePlatformToolRuntime(for: run) {
                        [weak self] kind, nativeType in
                        Task { await self?.emit(kind, nativeType: nativeType) }
                    }
                    await platformToolRuntime.registerNucleicTools(
                        on: server, token: token, run: run)
                    env.merge(
                        CommandInterceptor.hookEnv(
                            host: "127.0.0.1", port: ContainerSpec.controlBridgePort,
                            token: token, sessionID: run.sessionID,
                            commandTracing: ContainerServiceSettings.commandTracingEnabled)
                    ) { _, new in new }
                    // Point Codex at our MCP server (streamable HTTP over the in-guest control
                    // bridge — the same endpoint Claude uses — bearer token via env var so it stays
                    // off disk) with the server's tools pre-approved, and strip Codex's own review
                    // layer: Nucleic already gates every one of these tools. See CodexGateOwnership.
                    env[CodexGateOwnership.tokenEnvVar] = token
                    mcpArgs =
                        CodexGateOwnership.launchArgs
                        + CodexGateOwnership.nucleicMCPArgs(port: ContainerSpec.controlBridgePort)
                }
                handle = try await containerManager.exec(
                    name: name, workdir: cspec.workdir, env: env,
                    argv: [configuration.executable] + externalMCPArgs + mcpArgs + ["app-server"],
                    uid: cspec.runAsUID, gid: cspec.runAsGID)
            } else {
                let externalMCPArgs = ExternalAgentIntegrationSettings.allows(.mcp, model: run.model)
                    ? [] : ["-c", "mcp_servers={}"]
                let spec = ProcessSpec(
                    executable: configuration.executable,
                    args: externalMCPArgs + ["app-server"],
                    cwd: run.worktree,
                    env: run.extraEnv.merging(hostManagedGitEnv) { _, new in new },
                    stdinMode: .pipe)
                handle = try await processHost.launch(spec)
            }
            self.handle = handle

            let stderrTask = Task {
                for try await line in handle.stderrLines {
                    let text = String(decoding: line, as: UTF8.self)
                    stderrTail.append(text)
                    if ProcessInfo.processInfo.environment["NUCLEIC_DEBUG_STDERR"] != nil {
                        FileHandle.standardError.write(Data(("codex-stderr: " + text + "\n").utf8))
                    }
                }
            }

            // 2. JSON-RPC client + inbound (approval) request handler.
            let rpc = JSONRPCConnection(handle: handle)
            self.rpc = rpc
            await rpc.start { [weak self] request in
                await self?.handleInboundRequest(request)
            }

            // 3. Handshake: initialize (request) → initialized (notification, REQUIRED first).
            // `capabilities` and `clientInfo.title` are required-nullable in the schema.
            _ = try await rpc.request("initialize", params: .object([
                "clientInfo": .object([
                    "name": .string(configuration.clientName),
                    "title": .null,
                    "version": .string(configuration.clientVersion),
                ]),
                "capabilities": .null,
            ]))
            try await rpc.notify("initialized")

            // 4. Open the thread — fresh (`thread/start`) or reattached (`thread/resume` by id,
            //    the preferred resume path; codex loads the thread from its on-disk rollout).
            let threadResult: JSONValue
            if let resumeThreadID {
                // All environment settings are overrides accepted by thread/resume. Reapply them
                // every turn: besides honoring current settings, this repairs older threads whose
                // persisted workspace-write policy would otherwise attempt nested bwrap here.
                var resumeParams = Self.threadRuntimeParams(
                    for: run, isHostRun: isHostRun,
                    approvalPolicy: configuration.approvalPolicy)
                resumeParams["threadId"] = .string(resumeThreadID)
                threadResult = try await rpc.request("thread/resume", params: .object(resumeParams))
            } else {
                var startParams = Self.threadRuntimeParams(
                    for: run, isHostRun: isHostRun,
                    approvalPolicy: configuration.approvalPolicy)
                startParams["cwd"] = .string(run.worktree)
                if let model = run.model { startParams["model"] = .string(model) }
                threadResult = try await rpc.request("thread/start", params: .object(startParams))
            }
            let thread = threadResult["thread"] ?? threadResult
            let threadID = thread["id"]?.stringValue ?? resumeThreadID ?? ""
            currentThreadID = threadID
            backendSessionID = threadID

            // Authoritative sessionStarted from the thread response (the started *notification*
            // carries only the provider, not the resolved model id).
            emit(
                .sessionStarted(
                    SessionStarted(
                        backendSessionID: threadID,
                        model: threadResult["model"]?.stringValue ?? run.model ?? "",
                        cwd: threadResult["cwd"]?.stringValue ?? thread["cwd"]?.stringValue
                            ?? run.worktree,
                        toolNames: [],
                        nativeTranscriptPath: thread["path"]?.stringValue)),
                nativeType: "thread/start")

            // Populate account usage after the session's authoritative opening event.
            // Notifications only describe changes, so a new app-server process may otherwise
            // leave the quota card empty until Codex pushes an update.
            if let snapshot = try? await rpc.request("account/rateLimits/read"),
               let usage = CodexUsage.fromRateLimitsNotification(snapshot, now: configuration.now()) {
                emit(.codexUsage(usage), nativeType: "account/rateLimits/read")
            }

            let decoder = CodexAppServerDecoder()

            // 5. Start the turn (skip for a re-attach-only resume with no prompt).
            if run.prompt.parts.isEmpty {
                finishRun(outcome: .completed)
                stderrTask.cancel()
                await teardownTurn()
                continuation.finish()
                return
            }
            var turnParams: [String: JSONValue] = [
                "threadId": .string(threadID),
                "input": Self.userInput(for: run.prompt),
            ]
            // `effort` is a per-turn override; only forward a value codex accepts as a
            // `ReasoningEffort` so an unrecognized string can't fail the turn. GPT-5.6 accepts
            // `max`; `ultra` is the Sol-only option Nucleic labels “Pro”.
            if let effort = Self.acceptedReasoningEffort(
                run.effort, model: run.model,
                codexPro: CodexAuthFile.currentSubscriptionPlanType() == "pro") {
                turnParams["effort"] = .string(effort)
            }
            let turnResult = try await rpc.request("turn/start", params: .object(turnParams))
            currentTurnID = (turnResult["turn"] ?? turnResult)["id"]?.stringValue

            // 6. Drain notifications until the turn reaches a terminal state.
            var sawTerminal = false
            var outcome: RunFinished.Outcome = .completed
            for await note in rpc.notifications {
                // Codex's recoverable "Reconnecting…" row hides the decisive detail one level
                // deeper: reqwest reports Linux getaddrinfo's "failed to lookup address
                // information: Try again" when the sandbox gateway DNS proxy wedges. Queue a
                // manager-wide network recovery now; it waits for every turn (including this one)
                // to settle, re-probes DNS, and only then replaces shared vmnet. Generic provider
                // disconnects deliberately do not trigger this path.
                if Self.reportsSandboxDNSFailure(method: note.method, params: note.params),
                   let containerManager, let name = activeContainerName
                {
                    Task.detached {
                        await containerManager.requestNetworkRecovery(
                            name: name, hostname: "chatgpt.com")
                    }
                }
                // Capture each fileChange item's edited paths (id → paths) so the file-change
                // approval — which references the item by id and carries no paths — can acquire locks.
                if let (id, paths) = CodexAppServerDecoder.fileChangeItemPaths(
                    method: note.method, params: note.params)
                {
                    fileChangePathsByItem[id] = paths
                }
                for decoded in decoder.decode(method: note.method, params: note.params) {
                    emit(decoded.kind, nativeType: decoded.nativeType)
                }
                switch note.method {
                case "turn/completed":
                    let status = (note.params["turn"] ?? note.params)["status"]?.stringValue
                    outcome = Self.outcome(forTurnStatus: status, interrupted: interruptRequested)
                    sawTerminal = true
                case "turn/failed":
                    outcome = .errored
                    sawTerminal = true
                default:
                    break
                }
                if sawTerminal { break }
            }

            // 7. The notification stream ran dry without a terminal turn event → the process
            //    exited mid-turn. Classify like Claude does (ADAPTERS §1.5).
            if !sawTerminal {
                if interruptRequested {
                    outcome = .interrupted
                } else if terminating {
                    outcome = .completed
                } else {
                    outcome = .errored
                    emit(
                        .error(
                            AgentError(
                                recoverable: false,
                                message: "codex app-server exited mid-turn: \(stderrTail.joined())")),
                        nativeType: nil)
                }
            }

            finishRun(outcome: outcome)
            stderrTask.cancel()
            await approvals.cancelOutstanding(reason: "Run ended")
            await teardownTurn()
            continuation.finish()
        } catch {
            // A deliberate stop (interrupt/shutdown) can abort an in-flight request — settle
            // the run cleanly. Any other error is a genuine failure; throw so SessionController
            // synthesizes the terminal error + runFinished (matching ClaudeCodeBackend §1.5).
            if interruptRequested || terminating {
                if !sawRunFinished {
                    finishRun(outcome: interruptRequested ? .interrupted : .completed)
                }
                await approvals.cancelOutstanding(reason: "Run interrupted")
                await teardownTurn()
                continuation.finish()
            } else {
                await approvals.cancelOutstanding(reason: "Run failed")
                await teardownTurn()
                continuation.finish(throwing: error)
            }
        }
    }

    /// Normalize a UI effort to a Codex-supported wire value. Kept pure for contract tests.
    nonisolated static func acceptedReasoningEffort(
        _ effort: String?, model: String?, codexPro: Bool
    ) -> String? {
        guard let effort = effort?.lowercased() else { return nil }
        if effort == "ultra" {
            return model == "gpt-5.6-sol" && codexPro ? effort : nil
        }
        if effort == "max" {
            return model?.hasPrefix("gpt-5.6-") == true ? effort : nil
        }
        guard ["minimal", "low", "medium", "high", "xhigh"].contains(effort) else { return nil }
        return effort
    }

    /// True only for the resolver failures emitted by Codex's response-stream client inside the
    /// Linux sandbox. Kept pure for regression tests: an ordinary 5xx, transport disconnect, or
    /// "error sending request" must retain Codex's own retry behavior without cycling networking.
    nonisolated static func reportsSandboxDNSFailure(method: String, params: JSONValue) -> Bool {
        guard method == "error" else { return false }
        let detail = params["error"]?["additionalDetails"]?.stringValue
            ?? params["additionalDetails"]?.stringValue
            ?? ""
        let normalized = detail.lowercased()
        return normalized.contains("failed to lookup address information")
            || normalized.contains("temporary failure in name resolution")
            || normalized.contains("eai_again")
    }

    private func finishRun(outcome: RunFinished.Outcome) {
        guard !sawRunFinished else { return }
        sawRunFinished = true
        emit(.runFinished(RunFinished(outcome: outcome)), nativeType: nil)
    }

    /// Tear down the just-finished turn's process + JSON-RPC connection. `currentThreadID`
    /// and `backendSessionID` persist so the next `resume` reattaches the same thread.
    private func teardownTurn() async {
        if let rpc { await rpc.close() }
        self.rpc = nil
        if let handle {
            handle.closeStdin()
            await handle.terminate()
        }
        self.handle = nil
        if let token = interceptorToken {
            // Unregister only this run's token; the shared per-container server keeps serving the
            // other sessions in the box.
            await interceptorServer?.unregister(token: token)
            interceptorToken = nil
            interceptorServer = nil
        }
        if let name = activeContainerName {
            await containerManager?.finished(name: name)
            activeContainerName = nil
        }
        currentTurnID = nil
        fileChangePathsByItem.removeAll()
    }

    // MARK: - Interceptor reporting (Nucleic Control containers)

    /// Forward a ground-truth git op reported by the in-container `git` shim to the conflict
    /// coordinator (lock release on merge + autoship), the same seam Claude/Grok use. Best-effort.
    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 (PR/release/API) reported by the in-container `gh` shim.
    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)
    }

    // MARK: - Edit gate (codex PreToolUse hook — LOCKING §4.6)

    /// Arbitrate one codex tool call before it runs — reached from the container's `PreToolUse` hook
    /// rather than an approval request, because codex only *asks* about a patch its approval policy
    /// doesn't already allow. The contract lives in ``CodexToolGate/arbitrate(_:sessionID:worktreeRoot:coordinator:)``,
    /// shared with `codex exec`.
    func gatePreToolUse(_ call: MCPApprovalServer.PreToolUseCall) async
        -> MCPApprovalServer.PreToolUseDecision
    {
        await CodexToolGate.arbitrate(
            call, sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            worktreeRoot: agentShellWorkdir, coordinator: conflictCoordinator)
    }

    /// Forward a non-git command the interceptor reported (shim or bash tracer) for the command feed.
    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`) for the feeds (docs/NASH.md §9).
    private func handleShellReport(_ calls: [MCPApprovalServer.ShellReportCall]) async {
        guard let sessionID, let conflictCoordinator else { return }
        await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
    }

    private func interruptCurrentTurn() async {
        guard let rpc, let threadID = currentThreadID, let turnID = currentTurnID else {
            handle?.interruptThenKill()  // no live turn id → signal + escalate so Stop always lands
            return
        }
        // Arm the hard-stop BEFORE awaiting the cooperative request. A wedged app-server can accept
        // the JSON-RPC write yet never answer it; arming afterward leaves `interrupt()` suspended and
        // means the Stop button never reaches its advertised force-stop path.
        handle?.killGroupAfter(graceNanos: 8 * 1_000_000_000)
        // Cooperative: the server replies `{}` then emits turn/completed(interrupted) (§2.6).
        _ = try? await rpc.request("turn/interrupt", params: .object([
            "threadId": .string(threadID),
            "turnId": .string(turnID),
        ]))
    }

    // MARK: - Approval bridge (server→client requests, ADAPTERS §2.4)

    private func handleInboundRequest(_ request: JSONRPCConnection.InboundRequest) async {
        guard let rpc else { return }
        switch request.method {
        case "item/commandExecution/requestApproval", "item/fileChange/requestApproval":
            await handleReviewApproval(request, rpc: rpc, legacy: false)
        case "execCommandApproval", "applyPatchApproval":
            // Deprecated v1 surface — answered defensively with the v1 `ReviewDecision` enum.
            await handleReviewApproval(request, rpc: rpc, legacy: true)
        case "item/permissions/requestApproval":
            await handlePermissionsApproval(request, rpc: rpc)
        default:
            // Unknown server request — answer with an error so the server never hangs on us.
            try? await rpc.replyError(
                to: request.id, code: -32601, message: "Unsupported request \(request.method)")
        }
    }

    private func handleReviewApproval(
        _ request: JSONRPCConnection.InboundRequest, rpc: JSONRPCConnection, legacy: Bool
    ) async {
        let fields = Self.reviewApprovalFields(request)
        func reply(for decision: Decision) -> JSONValue {
            legacy
                ? CodexDecisionMapping.reviewReplyV1(for: decision)
                : CodexDecisionMapping.reviewReply(for: decision)
        }

        // Raw-shell writes take real locks instead of being refused (shared with Claude/Grok;
        // docs/NASH.md §5.2). Every workspace file the command authors is arbitrated exactly like a
        // fileChange's paths, so the write is conflict-checked before it happens and lands in the
        // trunk after; only a target too dynamic to name is still denied. Best-effort at this layer
        // — Codex only asks when its sandbox/approval policy requires it; the residue-capture
        // backstop covers the rest.
        if fields.toolName == "Bash", let command = fields.input["command"]?.stringValue {
            let clearance: ShellWriteClearance
            if let conflictCoordinator {
                clearance = await conflictCoordinator.arbitrateShellWrites(
                    sessionID: sessionID ?? SessionID(rawValue: "unknown"), command: command,
                    workdir: agentShellWorkdir, toolUseID: request.params["itemId"]?.stringValue)
            } else if let reason = WorktreeMutationGuard.denialReason(command: command) {
                clearance = .deny(reason)
            } else {
                clearance = .clear
            }
            if case .deny(let reason) = clearance {
                try? await rpc.reply(to: request.id, result: reply(for: .deny(reason: reason)))
                return
            }
        }

        // Auto-allow Nucleic's own MCP tools: our `nucleic` server already gates each tool through
        // Nucleic's UI/handlers, so Codex must never double-prompt for them (Nucleic still owns the
        // gate — no permission is bypassed). A no-op for Codex builds that don't route MCP tool
        // approvals through this channel; belt-and-suspenders per RUNTIME §5.
        if Self.concernsNucleicMCP(request) {
            try? await rpc.reply(to: request.id, result: reply(for: .allow()))
            return
        }

        // Conflict pre-check (Nucleic Control): a file-change edit acquires locks on its paths before
        // the patch applies. **Secondary gate.** Codex sends this approval only when its policy
        // doesn't already allow the write, which under Nucleic's config (on-request + a container's
        // danger-full-access) is never — the `PreToolUse` hook (``CodexToolGate``) is the gate that
        // actually fires. This path still matters for a run whose policy does ask, and double
        // arbitration is free (a path this session already holds re-grants instantly). Runs ahead of
        // the always-rule cache so an "always allow" can't slip a conflicting edit past. The approval
        // carries no paths, so use the ones captured from its fileChange item. On deferred/cancelled
        // (or an update-on-grant re-ground) deny, so Codex doesn't apply; its natural retry re-grants.
        if let conflictCoordinator,
            request.method == "item/fileChange/requestApproval"
                || request.method == "applyPatchApproval",
            let itemId = request.params["itemId"]?.stringValue,
            let paths = fileChangePathsByItem[itemId], !paths.isEmpty
        {
            let outcome = await conflictCoordinator.arbitrate(
                sessionID: sessionID ?? SessionID(rawValue: "unknown"), task: "", files: paths,
                toolUseID: itemId)
            switch outcome.resolution {
            case .deferred, .cancelled:
                try? await rpc.reply(to: request.id, result: reply(for: .deny(reason: nil)))
                return
            case .proceed, .noConflict:
                if outcome.refreshNotice != nil {
                    try? await rpc.reply(to: request.id, result: reply(for: .deny(reason: nil)))
                    return
                }
            }
        }

        // Always-rule short-circuit: auto-answer without surfacing (RUNTIME §5).
        if let cached = await approvals.cachedDecision(toolName: fields.toolName, input: fields.input)
        {
            try? await rpc.reply(to: request.id, result: reply(for: cached.decision))
            return
        }

        if NucleicApprovalPolicy.shouldAutoApprove(
            autoApprove: autoApprove, risk: fields.risk)
        {
            try? await rpc.reply(to: request.id, result: reply(for: .allow()))
            return
        }

        let approval = ApprovalRequest(
            id: .generate(),
            sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            toolCallID: request.params["itemId"]?.stringValue,
            toolName: fields.toolName,
            input: fields.input,
            title: fields.title,
            risk: fields.risk,
            createdAt: configuration.now())

        emit(.approvalRequested(approval), nativeType: request.method)
        let resolved = await approvals.waitForDecision(approval)
        emit(.approvalResolved(resolved), nativeType: nil)

        try? await rpc.reply(to: request.id, result: reply(for: resolved.decision))
        if case .cancelRun = resolved.decision {
            interruptRequested = true
            await interruptCurrentTurn()
        }
    }

    private func handlePermissionsApproval(
        _ request: JSONRPCConnection.InboundRequest, rpc: JSONRPCConnection
    ) async {
        let requested = request.params["permissions"] ?? .object([:])
        // Auto-allow permission grants that concern Nucleic's own `nucleic` MCP server — Nucleic
        // already gates those tools itself, so re-prompting here would be a redundant second gate.
        if Self.concernsNucleicMCP(request) {
            try? await rpc.reply(
                to: request.id,
                result: CodexDecisionMapping.permissionsReply(
                    for: .allow(), requestedPermissions: requested))
            return
        }
        let approval = ApprovalRequest(
            id: .generate(),
            sessionID: sessionID ?? SessionID(rawValue: "unknown"),
            toolCallID: request.params["itemId"]?.stringValue,
            toolName: "permissions",
            input: requested,
            title: "Grant permissions",
            risk: .unknown,
            createdAt: configuration.now())

        emit(.approvalRequested(approval), nativeType: request.method)
        let resolved = await approvals.waitForDecision(approval)
        emit(.approvalResolved(resolved), nativeType: nil)

        try? await rpc.reply(
            to: request.id,
            result: CodexDecisionMapping.permissionsReply(
                for: resolved.decision, requestedPermissions: requested))
        if case .cancelRun = resolved.decision {
            interruptRequested = true
            await interruptCurrentTurn()
        }
    }

    /// Whether an inbound approval request concerns Nucleic's own MCP server (named `nucleic` in the
    /// `-c mcp_servers.nucleic…` config we launch codex with). Those tools are already gated by
    /// Nucleic's own handlers/UI, so such an approval is auto-allowed rather than double-prompted.
    /// Checks the server/tool identity across the field names Codex may use; unknown shapes → false.
    static func concernsNucleicMCP(_ request: JSONRPCConnection.InboundRequest) -> Bool {
        let p = request.params
        for key in ["server", "serverName", "mcpServer"] where p[key]?.stringValue == "nucleic" {
            return true
        }
        for key in ["tool", "toolName", "name"] {
            if let v = p[key]?.stringValue, v == "nucleic" || v.hasPrefix("mcp__nucleic__") {
                return true
            }
        }
        // Nested shapes: a guardian permission request identifies the call under
        // `permissions[].mcpToolCall.{toolName,connectorName}` (and snake_case in the core payloads)
        // rather than at the top level, so the checks above missed it entirely and Nucleic's own
        // tools fell through to a generic "Grant permissions" prompt — or, with no channel able to
        // answer, to a denial. See CodexGateOwnership for the launch-side half of this.
        return containsNucleicMCPIdentity(p, depth: 0)
    }

    /// Keys Codex may carry the *server*/connector identity under, across its camelCase app-server
    /// protocol and the snake_case guardian payloads. Matched only against an exact `nucleic`.
    private static let nucleicServerKeys: Set<String> = [
        "server", "serverName", "server_name", "mcpServer", "mcp_server",
        "connector", "connectorName", "connector_name", "connectorId", "connector_id",
    ]

    /// Keys carrying the tool's own name. Matched only against the `mcp__nucleic__` prefix — a bare
    /// `nucleic` is deliberately NOT accepted here, because keys this generic (`name`) collide with
    /// unrelated payload fields, and this project is itself named "nucleic"; a bare match could
    /// auto-allow an unrelated command approval.
    private static let nucleicToolKeys: Set<String> = [
        "tool", "toolName", "tool_name", "name", "toolTitle", "tool_title",
    ]

    /// Recursively look for the `nucleic` MCP server's identity anywhere in an approval payload.
    /// Depth-bounded so a pathological payload can't spin.
    private static func containsNucleicMCPIdentity(_ value: JSONValue, depth: Int) -> Bool {
        guard depth <= 6 else { return false }
        switch value {
        case .object(let fields):
            for (key, field) in fields {
                if let name = field.stringValue {
                    if nucleicServerKeys.contains(key), name == "nucleic" { return true }
                    if nucleicToolKeys.contains(key), name.hasPrefix("mcp__nucleic__") { return true }
                }
                if containsNucleicMCPIdentity(field, depth: depth + 1) { return true }
            }
            return false
        case .array(let items):
            return items.contains { containsNucleicMCPIdentity($0, depth: depth + 1) }
        default:
            return false
        }
    }

    /// Map an approval request's params onto a normalized `ApprovalRequest`. A command
    /// approval becomes a `Bash{command}` call (shared risk classifier + command-summary UI);
    /// a file-change approval carries no path in its params (it references the already-announced
    /// `fileChange` item by id), so it surfaces as a `.write` with an explanatory title.
    static func reviewApprovalFields(_ request: JSONRPCConnection.InboundRequest)
        -> (toolName: String, input: JSONValue, title: String, risk: Risk)
    {
        let p = request.params
        let isFileChange =
            request.method == "item/fileChange/requestApproval"
            || request.method == "applyPatchApproval"
        if isFileChange {
            let reason = p["reason"]?.stringValue
            let input = JSONValue.object(reason.map { ["reason": .string($0)] } ?? [:])
            return ("Edit", input, reason ?? "Apply file changes", .write)
        }
        let command = p["command"]?.stringValue ?? ""
        let input = JSONValue.object(["command": .string(command)])
        return (
            "Bash", input,
            RiskClassifier.title(toolName: "Bash", input: input),
            RiskClassifier.classify(toolName: "Bash", input: input)
        )
    }

    // MARK: - Helpers

    /// `UserInput[]` for a turn (ADAPTERS §2.3). Context parts render as labeled text. The
    /// `text` variant carries a required `text_elements` array (snake_case in the schema).
    static func userInput(for input: AgentInput) -> JSONValue {
        func textInput(_ text: String) -> JSONValue {
            .object(["type": .string("text"), "text": .string(text), "text_elements": .array([])])
        }
        var items: [JSONValue] = []
        for part in input.parts {
            switch part {
            case .text(let text):
                items.append(textInput(text))
            case .context(let label, let body):
                items.append(textInput("[\(label)]\n\(body)"))
            case .attachment(let attachment):
                // Defensive: attachments are materialized + folded into text before reaching a
                // backend (`AppStore.materializeRemoteAttachments`); reference by name if one leaks.
                items.append(textInput("[Attached file: \(attachment.filename)]"))
            }
        }
        return .array(items)
    }

    static func outcome(forTurnStatus status: String?, interrupted: Bool) -> RunFinished.Outcome {
        if interrupted { return .interrupted }
        switch status {
        case "completed": return .completed
        case "interrupted": return .interrupted
        case "failed": return .errored
        default: return .completed
        }
    }

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

/// Ring of recent stderr lines for the synthesized error message on an abnormal exit.
private final class CodexStderrTail: @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")
    }
}
