448 lines
23 KiB
Swift
448 lines
23 KiB
Swift
import Foundation
|
|
|
|
/// `AgentBackend` over `codex exec --json` — the **unattended** Codex path (ADAPTERS §3), for
|
|
/// `ApprovalPolicy.fixed` runs. `codex` has no *native* approval channel here, so its own
|
|
/// command/file execution is fail-closed: `interactiveApprovals = false`, run with
|
|
/// `--ask-for-approval never` inside a sandbox. Its JSONL surface is a distinct snake_case schema
|
|
/// decoded by `CodexExecDecoder`.
|
|
///
|
|
/// A containerized run still exposes Nucleic's own MCP platform tools (`host_exec`, the VM tools,
|
|
/// `linux_container`, `check_conflict`, …) at parity with the app-server backend — those are gated
|
|
/// through Nucleic's `ApprovalCoordinator` (Nucleic is the interactive channel) and resolved via
|
|
/// `respond`, independent of Codex's absent native seam. Nested containers are unsupported, so
|
|
/// `linux_container` is how a run gets a fresh sandbox.
|
|
///
|
|
/// Like the other backends each turn is a single-shot process: `start` runs `codex exec`;
|
|
/// `resume` runs `codex exec resume <sessionId>`. Validated against codex-cli 0.141.0.
|
|
public actor CodexExecBackend: AgentBackend {
|
|
public static let id = BackendID.codexExec
|
|
|
|
public nonisolated let capabilities = BackendCapabilities(
|
|
interactiveApprovals: false,
|
|
allowAlwaysScopes: [],
|
|
canModifyToolInput: false,
|
|
partialMessageStreaming: false,
|
|
emitsThinking: true,
|
|
emitsFileChangeEvents: true,
|
|
nativeResume: true,
|
|
sandboxModes: [.readOnly, .workspaceWrite, .fullAccess],
|
|
followUpWhileRunning: false)
|
|
|
|
public struct Configuration: Sendable {
|
|
public var executable: String
|
|
public var now: @Sendable () -> Date
|
|
public init(
|
|
executable: String = "codex",
|
|
now: @escaping @Sendable () -> Date = { Date() }
|
|
) {
|
|
self.executable = executable
|
|
self.now = now
|
|
}
|
|
}
|
|
|
|
private let configuration: Configuration
|
|
private let processHost: ProcessHost
|
|
/// Codex's *native* command/file approvals don't exist here (`--ask-for-approval never`), but the
|
|
/// Nucleic MCP platform tools below (`host_exec`, `linux_container`, the VM tools) are gated
|
|
/// through this coordinator — Nucleic is the interactive channel — and resolved via `respond`.
|
|
public let approvals: ApprovalCoordinator
|
|
/// Optional sandbox orchestrator: execs `codex exec` inside a vsock control container when a run
|
|
/// carries one (isolation for unattended/autoship runs). `nil` → always host.
|
|
private let containerManager: ContainerManager?
|
|
/// Provider-neutral Nucleic platform-tool runtime (shared with Claude): serves the MCP platform
|
|
/// tools (`host_exec`, the VM tools, `linux_container`) plus Nucleic-owned tools for a
|
|
/// containerized run, at parity with the app-server backend. Nested containers aren't supported,
|
|
/// so `linux_container` is how a run gets a *fresh* sandbox.
|
|
private let platformToolRuntime: ClaudeCodeBackend
|
|
private var activeContainerName: String?
|
|
/// Conflict coordinator + approval-server registry — for the git/gh/command interceptor only.
|
|
/// `codex exec` runs `--ask-for-approval never`, so there is no interactive seam to *acquire*
|
|
/// locks on edits; it joins the lock system on the **observe/release** side (merges → release +
|
|
/// autoship + the activity feed), not the acquire side.
|
|
private let conflictCoordinator: ConflictCoordinator?
|
|
private let approvalServerRegistry: ApprovalServerRegistry?
|
|
private var interceptorServer: MCPApprovalServer?
|
|
private var interceptorToken: String?
|
|
|
|
private var handle: (any ProcessHandle)?
|
|
private var continuation: AsyncThrowingStream<AgentEvent, Error>.Continuation?
|
|
private var sessionID: SessionID?
|
|
private var provisionalSeq: UInt64 = 0
|
|
private var interruptRequested = false
|
|
private var terminating = false
|
|
private var sawRunFinished = false
|
|
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, resumeSessionID: nil)
|
|
}
|
|
|
|
public nonisolated func resume(_ resume: ResumeSpec) -> AsyncThrowingStream<AgentEvent, Error> {
|
|
return makeStream(run: Self.runSpec(resuming: resume), resumeSessionID: resume.backendSessionID)
|
|
}
|
|
|
|
/// Rebuild every per-turn capability on resume. `codex exec` is single-shot, so every
|
|
/// conversational follow-up arrives here as a fresh run — this must carry the SAME platform-tool
|
|
/// opt-ins, orchestra/mesh handlers, auto-approve, and appended guidance as the initial `start`,
|
|
/// or a resumed turn would silently lose `host_exec`/VM/`linux_container`/subagent/mesh tools (and
|
|
/// the system-prompt guidance announcing them). Mirrors `CodexAppServerBackend.runSpec(resuming:)`.
|
|
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,
|
|
orchestraSuperviseHandler: resume.orchestraSuperviseHandler,
|
|
orchestraReplyToWorkerHandler: resume.orchestraReplyToWorkerHandler,
|
|
orchestraAskSupervisorHandler: resume.orchestraAskSupervisorHandler,
|
|
meshMessaging: resume.meshMessaging,
|
|
appendSystemPrompt: resume.appendSystemPrompt)
|
|
}
|
|
|
|
public func send(_ input: AgentInput) async throws {
|
|
throw BackendError.unsupported("codex exec is single-shot; follow-ups resume a new run")
|
|
}
|
|
|
|
public func respond(to approvalID: ApprovalID, _ decision: Decision, by responder: String) async throws {
|
|
// Codex's *native* command/file approvals don't exist here (`--ask-for-approval never`), but
|
|
// Nucleic's own MCP platform tools (`host_exec`, `linux_container`, the VM tools) are gated
|
|
// through the shared ApprovalCoordinator — Nucleic is the interactive channel — so resolve
|
|
// against it, exactly as the app-server backend does.
|
|
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
|
|
// `codex exec` is single-shot; SIGINT ends it, escalating to a group SIGKILL if it ignores
|
|
// the interrupt so Stop always takes effect.
|
|
handle?.interruptThenKill()
|
|
}
|
|
|
|
/// 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
|
|
guard let handle else { return }
|
|
handle.closeStdin()
|
|
await handle.terminate()
|
|
self.handle = nil
|
|
}
|
|
|
|
// MARK: - Run loop
|
|
|
|
private nonisolated func makeStream(run: RunSpec, resumeSessionID: String?)
|
|
-> AsyncThrowingStream<AgentEvent, Error>
|
|
{
|
|
AsyncThrowingStream { continuation in
|
|
let task = Task {
|
|
await self.runLoop(run: run, resumeSessionID: resumeSessionID, continuation: continuation)
|
|
}
|
|
continuation.onTermination = { _ in task.cancel() }
|
|
}
|
|
}
|
|
|
|
private func runLoop(
|
|
run: RunSpec,
|
|
resumeSessionID: String?,
|
|
continuation: AsyncThrowingStream<AgentEvent, Error>.Continuation
|
|
) async {
|
|
self.continuation = continuation
|
|
self.sessionID = run.sessionID
|
|
self.sawRunFinished = false
|
|
self.interruptRequested = false
|
|
|
|
let stderrTail = CodexExecStderrTail()
|
|
do {
|
|
// Nucleic's environment guidance (build tooling, the `linux_container`/VM tool ladder,
|
|
// nvrsion) rides in as a leading section of the prompt: `codex exec` has no
|
|
// developer-instructions channel like the app-server's `thread/start`, and it is single-shot
|
|
// per turn, so prepending each turn matches how the app-server re-sends it every turn.
|
|
let prompt = Self.execPrompt(run.prompt.plainText ?? "", guidance: run.appendSystemPrompt)
|
|
// A Nucleic Control container already *is* the isolation boundary, and Codex's own
|
|
// OS-level sandbox (bwrap/landlock) can't nest inside the container's capability set —
|
|
// it fails before the command runs ("bwrap: Unexpected capabilities…"). So when we exec
|
|
// Codex inside a control container, disable its inner sandbox (danger-full-access); the
|
|
// container is the sandbox, and a run that wants a *fresh* box uses the `linux_container`
|
|
// MCP tool. Host runs keep the requested/default sandbox.
|
|
let runsInControlContainer = containerManager != nil && run.container?.controlSocketHostPath != nil
|
|
let sandboxMode: SandboxMode = runsInControlContainer ? .fullAccess : (run.sandbox ?? .workspaceWrite)
|
|
// Global `codex` args (before the `exec` subcommand): disable external MCP servers unless
|
|
// this model may use them. The per-container Nucleic MCP wiring (`mcpArgs`) is added in the
|
|
// container branch below so `codex exec` reaches Nucleic's platform tools.
|
|
let externalMCPArgs = ExternalAgentIntegrationSettings.allows(.mcp, model: run.model)
|
|
? [] : ["-c", "mcp_servers={}"]
|
|
// The `exec` subcommand and its flags (everything after any global `-c` overrides).
|
|
var execArgs: [String] = ["exec"]
|
|
if let resumeSessionID { execArgs += ["resume", resumeSessionID] }
|
|
execArgs += [
|
|
"--json",
|
|
"--sandbox", sandboxMode.rawValue,
|
|
"--ask-for-approval", "never",
|
|
"--skip-git-repo-check",
|
|
"-C", run.worktree,
|
|
]
|
|
if let model = run.model { execArgs += ["-m", model] }
|
|
if !prompt.isEmpty { execArgs.append(prompt) }
|
|
|
|
// Spawn `codex exec` — inside the shared control container (stdio over vsock) when this
|
|
// run carries a vsock control socket, else on the host. Gated, so default is unchanged.
|
|
// 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, let controlSock = cspec.controlSocketHostPath {
|
|
let (name, _) = try await containerManager.ensureRunning(cspec)
|
|
activeContainerName = name
|
|
var env = cspec.env.merging(run.extraEnv) { _, new in new }
|
|
// Extra global `codex` args that point it at Nucleic's per-container MCP endpoint (set
|
|
// once the control server is up), so `codex exec` reaches Nucleic's platform tools.
|
|
var mcpArgs: [String] = []
|
|
if let registry = approvalServerRegistry {
|
|
let server = await registry.server(for: cspec.name)
|
|
try await server.start(unixSocketPath: controlSock)
|
|
let token = UUID().uuidString
|
|
interceptorServer = server
|
|
interceptorToken = token
|
|
// Git/gh/command interceptor (observe/release only — no acquire seam): the
|
|
// in-container shims POST observed ops to the shared per-container server →
|
|
// conflict coordinator. Gated as before on the interceptor + coordinator.
|
|
if cspec.installGitInterceptor, conflictCoordinator != nil {
|
|
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] call in
|
|
Task { await self?.handleShellReport(call) }
|
|
}
|
|
env.merge(
|
|
CommandInterceptor.hookEnv(
|
|
host: "127.0.0.1", port: ContainerSpec.controlBridgePort,
|
|
token: token, sessionID: run.sessionID,
|
|
commandTracing: ContainerServiceSettings.commandTracingEnabled)
|
|
) { _, new in new }
|
|
}
|
|
// Nucleic platform + owned MCP tools (`host_exec`, the VM tools, `linux_container`,
|
|
// `check_conflict`, …) at full parity with the app-server backend. Nested containers
|
|
// are unsupported, so `linux_container` is how this run gets a *fresh* sandbox. The
|
|
// tools are gated by Nucleic's approval system, which resolves through `respond`
|
|
// even though codex exec has no native approval channel.
|
|
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)
|
|
// Point Codex at our MCP server: streamable HTTP over the in-guest control bridge,
|
|
// authenticated by the per-session bearer token read from the env var — so the token
|
|
// stays out of any config file on disk.
|
|
env["NUCLEIC_MCP_TOKEN"] = token
|
|
mcpArgs = [
|
|
"-c",
|
|
"mcp_servers.nucleic.url=\"http://127.0.0.1:\(ContainerSpec.controlBridgePort)/mcp\"",
|
|
"-c", "mcp_servers.nucleic.bearer_token_env_var=\"NUCLEIC_MCP_TOKEN\"",
|
|
]
|
|
}
|
|
handle = try await containerManager.exec(
|
|
name: name, workdir: cspec.workdir, env: env,
|
|
argv: [configuration.executable] + externalMCPArgs + mcpArgs + execArgs,
|
|
uid: cspec.runAsUID, gid: cspec.runAsGID)
|
|
handle.closeStdin() // single-shot; no stdin (matches the host .closed mode)
|
|
} else {
|
|
let spec = ProcessSpec(
|
|
executable: configuration.executable, args: externalMCPArgs + execArgs,
|
|
cwd: run.worktree,
|
|
env: run.extraEnv.merging(hostManagedGitEnv) { _, new in new }, stdinMode: .closed)
|
|
handle = try await processHost.launch(spec)
|
|
}
|
|
self.handle = handle
|
|
|
|
let stderrTask = Task {
|
|
for try await line in handle.stderrLines {
|
|
stderrTail.append(String(decoding: line, as: UTF8.self))
|
|
}
|
|
}
|
|
|
|
let decoder = CodexExecDecoder()
|
|
for try await line in handle.stdoutLines {
|
|
for decoded in decoder.decode(line: line) {
|
|
if case .sessionStarted(let started) = decoded.kind {
|
|
backendSessionID = started.backendSessionID
|
|
}
|
|
emit(decoded.kind, nativeType: decoded.nativeType)
|
|
}
|
|
}
|
|
|
|
let exitCode = await handle.wait()
|
|
stderrTask.cancel()
|
|
if !sawRunFinished {
|
|
if exitCode != 0 && !interruptRequested && !terminating {
|
|
emit(
|
|
.error(
|
|
AgentError(
|
|
recoverable: false,
|
|
message: BackendDiagnostics.abnormalExitMessage(
|
|
tool: "codex exec", exitCode: exitCode,
|
|
stderrTail: stderrTail.joined(),
|
|
containerized: false))),
|
|
nativeType: nil)
|
|
}
|
|
let outcome: RunFinished.Outcome =
|
|
interruptRequested ? .interrupted : (terminating || exitCode == 0) ? .completed : .errored
|
|
sawRunFinished = true
|
|
emit(.runFinished(RunFinished(outcome: outcome)), nativeType: nil)
|
|
}
|
|
await teardownRunState()
|
|
self.handle = nil
|
|
continuation.finish()
|
|
} catch {
|
|
if interruptRequested || terminating {
|
|
if !sawRunFinished {
|
|
emit(
|
|
.runFinished(
|
|
RunFinished(outcome: interruptRequested ? .interrupted : .completed)),
|
|
nativeType: nil)
|
|
}
|
|
await teardownRunState()
|
|
self.handle = nil
|
|
continuation.finish()
|
|
} else {
|
|
await teardownRunState()
|
|
self.handle = nil
|
|
continuation.finish(throwing: error)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Compose the positional prompt for `codex exec`: Nucleic's environment guidance (when present)
|
|
/// as a leading section, then the user's turn. Codex exec has no developer-instructions channel,
|
|
/// so the guidance rides in the prompt. Guidance-only (empty user text) stays empty so a bare
|
|
/// re-attach never starts a spurious turn.
|
|
static func execPrompt(_ userText: String, guidance: String?) -> String {
|
|
guard !userText.isEmpty, let guidance,
|
|
!guidance.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
|
|
else { return userText }
|
|
return guidance + "\n\n" + userText
|
|
}
|
|
|
|
private func emit(_ kind: AgentEvent.Kind, nativeType: String?) {
|
|
if case .runFinished = kind { sawRunFinished = true }
|
|
provisionalSeq += 1
|
|
continuation?.yield(
|
|
AgentEvent(
|
|
sessionID: sessionID ?? SessionID(rawValue: "unknown"),
|
|
seq: provisionalSeq, at: configuration.now(),
|
|
backend: .codexExec, nativeType: nativeType, kind: kind))
|
|
}
|
|
|
|
/// Release the run's interceptor token + container hold at run end. Idempotent.
|
|
private func teardownRunState() async {
|
|
if let token = interceptorToken {
|
|
await interceptorServer?.unregister(token: token)
|
|
interceptorToken = nil
|
|
interceptorServer = nil
|
|
}
|
|
if let name = activeContainerName {
|
|
await containerManager?.finished(name: name)
|
|
activeContainerName = nil
|
|
}
|
|
}
|
|
|
|
// MARK: - Interceptor reporting (Nucleic Control containers; observe/release)
|
|
|
|
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)
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
private func handleCommandReport(_ call: MCPApprovalServer.CommandReportCall) async {
|
|
guard let sessionID, let conflictCoordinator else { return }
|
|
await conflictCoordinator.observeCommand(sessionID: sessionID, call: call)
|
|
}
|
|
|
|
private func handleShellReport(_ call: MCPApprovalServer.ShellReportCall) async {
|
|
guard let sessionID, let conflictCoordinator else { return }
|
|
await conflictCoordinator.observeShellEvent(sessionID: sessionID, call: call)
|
|
}
|
|
}
|
|
|
|
private final class CodexExecStderrTail: @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")
|
|
}
|
|
}
|