feat(subagent): add Codex product provider

packages/subagent/subagent-codex/src/index.ts

@@ -0,0 +1,89 @@
+/**
+ * Fixed Codex one-shot subagent provider. Every accepted run starts a fresh
+ * official `codex app-server --stdio` process in the delegating Session's
+ * workspace and publishes only after an ephemeral thread exists.
+ *
+ * @module @deepseek-ai/dsh-subagent-codex
+ */
+
+import type { Context } from 'cordis'
+import z from 'schemastery'
+import {
+  assertPositiveFinite,
+  NO_START_CAPABILITIES,
+  resolveChildCwd,
+  type ResolvedSubagentStartRequest,
+  type SubagentCapabilities,
+  type SubagentProvider,
+} from '@deepseek-ai/dsh-subagent'
+import {
+  DEFAULT_DISPOSE_GRACE_MS,
+  startCodexRun,
+  type CodexRunSpec,
+} from './run.ts'
+
+export const name = 'subagent-codex'
+export const inject = ['subagents', 'subprocess']
+
+/** Deployment-owned environment and process-release bound. */
+export interface Config {
+  /**
+   * Explicit environment entries layered over the subprocess seam's
+   * credential-scrubbed parent environment.
+   */
+  env?: Record<string, string>
+  /** Grace in milliseconds for app-server process-tree termination. */
+  disposeGraceMs?: number
+}
+
+export const Config: z<Config> = z.object({
+  env: z.dict(z.string()).default({}),
+  disposeGraceMs: z.number().default(DEFAULT_DISPOSE_GRACE_MS),
+})
+
+type ResolvedConfig = Required<Config>
+
+class CodexProvider implements SubagentProvider {
+  readonly name = 'codex'
+  readonly capabilities: SubagentCapabilities = NO_START_CAPABILITIES
+  readonly inheritsParentContext = false
+
+  constructor(
+    private readonly ctx: Context,
+    private readonly config: ResolvedConfig,
+  ) {}
+
+  start(request: ResolvedSubagentStartRequest) {
+    const spec: CodexRunSpec = {
+      cwd: resolveChildCwd(
+        'subagent-codex',
+        undefined,
+        request.parent.session.header.cwd,
+      ),
+      env: this.config.env,
+      disposeGraceMs: this.config.disposeGraceMs,
+      spawn: spawnSpec => this.ctx.subprocess.spawn(spawnSpec),
+      onError: (error, stopReason) => {
+        this.ctx.logger.warn(
+          `subagent-codex: child run failed (${stopReason}): ${error.message}`,
+        )
+      },
+    }
+    return startCodexRun(request, spec)
+  }
+}
+
+/**
+ * Register the fixed `codex` provider.
+ * @param ctx - context carrying shared subagent and subprocess services.
+ * @param config - explicit child environment and disposal grace.
+ */
+export function apply(ctx: Context, config: Config): void {
+  const resolved = config as ResolvedConfig
+  assertPositiveFinite(
+    'subagent-codex',
+    'disposeGraceMs',
+    resolved.disposeGraceMs,
+  )
+  ctx.subagents.registerProvider(new CodexProvider(ctx, resolved))
+}

packages/subagent/subagent-codex/src/invariant.ts

@@ -0,0 +1,30 @@
+/**
+ * Package-owned invariant companion for `@deepseek-ai/dsh-subagent-codex`.
+ * @module @deepseek-ai/dsh-subagent-codex/invariant
+ */
+
+/* jscpd:ignore-start */
+import type { Context } from 'cordis'
+import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
+
+const PACKAGE_NAME = '@deepseek-ai/dsh-subagent-codex'
+
+/** Cordis companion plugin name. */
+export const name = 'subagent-codex-invariant'
+/** Service required before the companion can reserve package ownership. */
+export const inject = ['invariants']
+
+/**
+ * No runtime invariant: lifecycle pairing belongs to the shared subagent
+ * service and process-tree ownership belongs to the subprocess service.
+ */
+const install: InvariantInstaller = () => {}
+
+/**
+ * Register this package's invariant companion.
+ * @param ctx - plugin context carrying the invariant registry.
+ * @returns the installed registration's disposer.
+ */
+export const apply = (ctx: Context): Promise<() => void> =>
+  Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
+/* jscpd:ignore-end */

packages/subagent/subagent-codex/src/run.ts

@@ -0,0 +1,209 @@
+/**
+ * One-shot Codex child lifecycle: spawn the real app-server through the
+ * subprocess seam, publish only after initialization and ephemeral thread
+ * creation, flatten post-publication failures, and dispose to whole-tree
+ * quiescence.
+ *
+ * @module @deepseek-ai/dsh-subagent-codex/run
+ */
+
+import { randomUUID } from 'node:crypto'
+import type { ContentBlock } from '@deepseek-ai/dsh-llm'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import {
+  settleRunResult,
+  subprocessRunHandle,
+  type SubagentResult,
+  type SubagentRun,
+  type SubagentStartRequest,
+  type SubagentStopReason,
+} from '@deepseek-ai/dsh-subagent'
+import type { SubprocessHandle, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { CodexAppServerWire } from './wire.ts'
+
+/** Default POSIX grace between subprocess termination tiers. */
+export const DEFAULT_DISPOSE_GRACE_MS = 3_000
+
+/** Fully resolved inputs for one Codex app-server run. */
+export interface CodexRunSpec {
+  /** Parent Session workspace, also supplied to `thread/start`. */
+  readonly cwd: string
+  /** Explicit deployment/test environment layered after the shared scrub. */
+  readonly env: Record<string, string>
+  /** Subprocess termination grace and final tree-exit bound. */
+  readonly disposeGraceMs: number
+  /** Shared subprocess service spawn operation. */
+  readonly spawn: (spec: SubprocessSpawnSpec) => SubprocessHandle
+  /** Diagnostic sink for a post-publication error flattened into a result. */
+  readonly onError?: (error: Error, stopReason: SubagentStopReason) => void
+}
+
+function thrown(value: unknown): Error {
+  /* v8 ignore next -- typed subprocess/wire failures reject with Error. */
+  return value instanceof Error ? value : new Error(String(value))
+}
+
+/**
+ * Validate and preserve the one-shot task before crossing the process seam.
+ * @param prompt - task content accepted from the shared subagent service.
+ * @returns the exact non-empty text block sequence.
+ */
+export function textTask(prompt: readonly ContentBlock[]): string[] {
+  if (prompt.length === 0) {
+    throw new Error('subagent-codex: the one-shot task must contain only text blocks')
+  }
+  const texts: string[] = []
+  for (const block of prompt) {
+    if (block.type !== 'text') {
+      throw new Error('subagent-codex: the one-shot task must contain only text blocks')
+    }
+    texts.push(block.text)
+  }
+  if (texts.every(text => text.trim().length === 0)) {
+    throw new Error('subagent-codex: the one-shot task must not be empty')
+  }
+  return texts
+}
+
+async function treeExitsWithin(child: SubprocessHandle, ms: number): Promise<boolean> {
+  const controller = new AbortController()
+  const timer = setTimeout(() => { controller.abort() }, ms)
+  try {
+    return await child.waitForExit(controller.signal)
+  } finally {
+    clearTimeout(timer)
+  }
+}
+
+/**
+ * Close the private wire, terminate the managed process tree, and wait for the
+ * subprocess owner to prove it is gone.
+ * @param wire - private app-server protocol connection.
+ * @param child - shared-service handle that owns the process tree.
+ * @param graceMs - termination grace used to bound final exit observation.
+ */
+export async function disposeCodexChild(
+  wire: CodexAppServerWire,
+  child: SubprocessHandle,
+  graceMs: number,
+): Promise<void> {
+  wire.close()
+  if (child.pid <= 0) {
+    await child.done.catch(() => {})
+    return
+  }
+  try {
+    child.stdin?.end()
+  } catch {
+    // A concurrently closed stdin does not change tree ownership below.
+  }
+  child.terminate()
+  if (!(await treeExitsWithin(child, graceMs * 2))) {
+    throw new Error('subagent-codex: app-server process tree did not exit within its dispose window')
+  }
+  await child.done
+}
+
+/**
+ * Start the real `codex app-server --stdio` child and publish its one-shot run.
+ * @param request - resolved shared subagent request.
+ * @param spec - workspace, environment, process seam, and diagnostic policy.
+ * @returns the published run after initialization and ephemeral thread creation.
+ */
+export async function startCodexRun(
+  request: SubagentStartRequest,
+  spec: CodexRunSpec,
+): Promise<SubagentRun> {
+  const texts = textTask(request.prompt)
+  if (request.signal.aborted) {
+    throw new Error('subagent-codex: request was aborted before app-server startup')
+  }
+
+  const child = spec.spawn({
+    argv: ['codex', 'app-server', '--stdio'],
+    cwd: spec.cwd,
+    stdio: { stdin: 'pipe', stdout: 'pipe', stderr: 'inherit' },
+    graceMs: spec.disposeGraceMs,
+    env: spec.env,
+  })
+  if (child.stdin === undefined || child.stdout === undefined) {
+    child.terminate()
+    await child.waitForExit()
+    throw new Error('subagent-codex: subprocess implementation dropped a piped protocol stream')
+  }
+
+  const wire = new CodexAppServerWire(child.stdout, child.stdin)
+  const disposeProcess = (): Promise<void> =>
+    disposeCodexChild(wire, child, spec.disposeGraceMs)
+
+  const processFailure: Promise<never> = child.done.then(
+    outcome => Promise.reject(new Error(
+      'subagent-codex: app-server exited before the run settled '
+      + `(code ${String(outcome.exitCode)}, signal ${String(outcome.signal)})`,
+    )),
+    (error: unknown) => Promise.reject(thrown(error)),
+  )
+  // A normal post-result dispose also closes the process. Keep that expected
+  // late rejection observed after the result race has already settled.
+  processFailure.catch(() => {})
+
+  const flags = { cancelled: false }
+  const runAbort = new AbortController()
+  let settleCancellation!: () => void
+  const cancellation = new Promise<void>((resolve) => { settleCancellation = resolve })
+  const requestCancel = (): void => {
+    if (flags.cancelled) return
+    flags.cancelled = true
+    runAbort.abort(new Error('subagent-codex: run cancelled locally'))
+    settleCancellation()
+    wire.interrupt()
+  }
+  const onAbort = (): void => { requestCancel() }
+  request.signal.addEventListener('abort', onAbort, { once: true })
+
+  try {
+    wire.start()
+    await Promise.race([wire.initialize(request.signal), processFailure])
+    await Promise.race([wire.startThread(spec.cwd, request.signal), processFailure])
+  } catch (error: unknown) {
+    request.signal.removeEventListener('abort', onAbort)
+    try {
+      await disposeProcess()
+    } catch (disposeError: unknown) {
+      throw new AggregateError(
+        [thrown(error), thrown(disposeError)],
+        'subagent-codex: startup failed and app-server cleanup also failed',
+      )
+    }
+    if (flags.cancelled) {
+      throw new Error('subagent-codex: request was aborted before app-server startup')
+    }
+    throw thrown(error)
+  }
+
+  const collectOutput = (): ContentBlock[] => wire.collectOutput()
+  const result: Promise<SubagentResult> = settleRunResult({
+    attempt: () => Promise.race([
+      wire.runTurn(texts, runAbort.signal, () => flags.cancelled),
+      processFailure,
+      cancellation.then((): SubagentResult => ({
+        output: collectOutput(),
+        stopReason: 'aborted',
+      })),
+    ]),
+    collectOutput,
+    cancelled: () => flags.cancelled,
+    onError: spec.onError,
+    signal: request.signal,
+    onAbort,
+  })
+
+  return subprocessRunHandle({
+    id: SessionId(randomUUID()),
+    result,
+    signal: request.signal,
+    onAbort,
+    requestCancel,
+    teardown: disposeProcess,
+  })
+}

packages/subagent/subagent-codex/src/wire.ts

@@ -0,0 +1,366 @@
+/**
+ * Minimal Codex app-server 0.146.0 protocol adapter. The shared JSON-RPC
+ * transport owns framing and request correlation; this module owns only the
+ * product methods, current thread/turn association, unattended approval
+ * responses, and terminal-answer selection.
+ *
+ * @module @deepseek-ai/dsh-subagent-codex/wire
+ */
+
+import type { Readable, Writable } from 'node:stream'
+import type { ContentBlock } from '@deepseek-ai/dsh-llm'
+import type { SubagentResult } from '@deepseek-ai/dsh-subagent'
+import { JsonRpcLineTransport } from '@deepseek-ai/dsh-sdk-protocol'
+
+type JsonObject = Record<string, unknown>
+
+interface Deferred<T> {
+  readonly promise: Promise<T>
+  readonly resolve: (value: T) => void
+}
+
+function deferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>((settle) => { resolve = settle })
+  return { promise, resolve }
+}
+
+function object(value: unknown, label: string): JsonObject {
+  if (value === null || typeof value !== 'object' || Array.isArray(value)) {
+    throw new Error(`subagent-codex: app-server returned invalid ${label}`)
+  }
+  return value as JsonObject
+}
+
+function string(value: unknown, label: string): string {
+  if (typeof value !== 'string' || value.length === 0) {
+    throw new Error(`subagent-codex: app-server returned invalid ${label}`)
+  }
+  return value
+}
+
+function thrown(value: unknown): Error {
+  /* v8 ignore next -- typed protocol and stream failures reject with Error. */
+  return value instanceof Error ? value : new Error(String(value))
+}
+
+function abortError(signal: AbortSignal): Error {
+  return signal.reason instanceof Error
+    ? signal.reason
+    : new Error(`subagent-codex: app-server request aborted: ${String(signal.reason)}`)
+}
+
+async function raceAbort<T>(pending: Promise<T>, signal: AbortSignal): Promise<T> {
+  if (signal.aborted) {
+    void pending.catch(() => {})
+    throw abortError(signal)
+  }
+  let rejectAbort!: (error: Error) => void
+  const aborted = new Promise<never>((_resolve, reject) => { rejectAbort = reject })
+  const onAbort = (): void => { rejectAbort(abortError(signal)) }
+  signal.addEventListener('abort', onAbort, { once: true })
+  try {
+    return await Promise.race([pending, aborted])
+  } finally {
+    signal.removeEventListener('abort', onAbort)
+  }
+}
+
+/**
+ * One app-server connection and its single ephemeral thread/turn.
+ *
+ * The class deliberately exposes no generic request surface. Supporting
+ * another product method must first become part of the provider contract.
+ */
+export class CodexAppServerWire {
+  private readonly transport: JsonRpcLineTransport
+  private readonly fatal = deferred<Error>()
+  private threadId: string | undefined
+  private turnId: string | undefined
+  private pendingTurnId: string | undefined
+  private turnCompleted: Deferred<JsonObject> | undefined
+  private readonly earlyTurnNotifications: Array<{
+    readonly method: string
+    readonly params: JsonObject
+  }> = []
+  private readonly finalAnswers: string[] = []
+  private readonly unphasedAnswers: string[] = []
+  private started = false
+  private closed = false
+
+  constructor(
+    private readonly input: Readable,
+    output: Writable,
+  ) {
+    this.transport = new JsonRpcLineTransport(input, output)
+    this.transport.onRequest((method, params) => this.handleServerRequest(method, params))
+    this.transport.onNotification((method, params) => {
+      try {
+        this.handleNotification(method, params)
+      } catch (error: unknown) {
+        this.fail(thrown(error))
+      }
+    })
+  }
+
+  /** Start reading app-server frames. */
+  start(): void {
+    if (this.started) return
+    this.started = true
+    this.input.on('error', this.onInputError)
+    this.input.on('end', this.onInputEnd)
+    this.transport.start()
+  }
+
+  /**
+   * Perform the required app-server initialize/initialized handshake.
+   * @param signal - unpublished-start cancellation.
+   */
+  async initialize(signal: AbortSignal): Promise<void> {
+    const response = object(await this.guarded(this.transport.request('initialize', {
+      clientInfo: {
+        name: 'deepseek-harness',
+        title: 'DeepSeek Harness',
+        version: '0.0.1',
+      },
+      capabilities: {
+        experimentalApi: false,
+        requestAttestation: false,
+      },
+    }, signal), signal), 'initialize response')
+    string(response.userAgent, 'initialize userAgent')
+    this.transport.notify('initialized')
+    await this.guarded(this.transport.flush(), signal)
+  }
+
+  /**
+   * Create the run's private ephemeral thread and retain its identity.
+   * @param cwd - parent Session workspace.
+   * @param signal - unpublished-start cancellation.
+   * @returns the app-server thread id.
+   */
+  async startThread(cwd: string, signal: AbortSignal): Promise<string> {
+    const response = object(await this.guarded(this.transport.request('thread/start', {
+      cwd,
+      ephemeral: true,
+    }, signal), signal), 'thread/start response')
+    const thread = object(response.thread, 'thread/start thread')
+    const id = string(thread.id, 'thread/start thread id')
+    if (thread.ephemeral !== true) {
+      throw new Error('subagent-codex: app-server did not create an ephemeral thread')
+    }
+    this.threadId = id
+    return id
+  }
+
+  /**
+   * Submit the one text-only task and wait for this thread/turn's authoritative
+   * terminal notification.
+   * @param texts - already validated task text blocks.
+   * @param signal - local cancellation for the published run.
+   * @param cancelled - whether local cancellation has already won.
+   * @returns the shared three-state subagent result.
+   */
+  async runTurn(
+    texts: readonly string[],
+    signal: AbortSignal,
+    cancelled: () => boolean,
+  ): Promise<SubagentResult> {
+    if (this.threadId === undefined) {
+      throw new Error('subagent-codex: cannot start a turn before thread/start')
+    }
+    if (this.turnCompleted !== undefined) {
+      throw new Error('subagent-codex: this one-shot wire already started its turn')
+    }
+    const completion = deferred<JsonObject>()
+    this.turnCompleted = completion
+    const response = object(await this.guarded(this.transport.request('turn/start', {
+      threadId: this.threadId,
+      input: texts.map(text => ({ type: 'text', text, text_elements: [] })),
+    }, signal), signal), 'turn/start response')
+    const turn = object(response.turn, 'turn/start turn')
+    this.commitTurnId(string(turn.id, 'turn/start turn id'))
+
+    const completed = await this.guarded(completion.promise, signal)
+    if (cancelled()) return { output: this.collectOutput(), stopReason: 'aborted' }
+
+    const terminal = object(completed.turn, 'turn/completed turn')
+    const status = terminal.status
+    if (status !== 'completed') {
+      const detail = status === 'failed'
+        ? `: ${JSON.stringify(terminal.error)}`
+        : ''
+      throw new Error(`subagent-codex: Codex turn ended with status ${String(status)}${detail}`)
+    }
+    const output = this.collectOutput()
+    if (output.length === 0) {
+      throw new Error('subagent-codex: Codex completed without a final answer')
+    }
+    return { output, stopReason: 'completed' }
+  }
+
+  /**
+   * Best-effort remote cancellation. Local settlement and process teardown
+   * remain authoritative when the child no longer accepts protocol requests.
+   */
+  interrupt(): void {
+    if (this.threadId === undefined || this.turnId === undefined || this.closed) return
+    void this.transport.request('turn/interrupt', {
+      threadId: this.threadId,
+      turnId: this.turnId,
+    }).catch(() => {})
+  }
+
+  /**
+   * The best non-commentary answer observed so far, preserving exact bytes.
+   * @returns the selected final or nullable-phase text block, if any.
+   */
+  collectOutput(): ContentBlock[] {
+    const selected = this.finalAnswers.length > 0
+      ? this.finalAnswers.at(-1)
+      : this.unphasedAnswers.at(-1)
+    return selected !== undefined && selected.trim().length > 0
+      ? [{ type: 'text', text: selected }]
+      : []
+  }
+
+  /** Detach JSON-RPC listeners and reject outstanding requests. Idempotent. */
+  close(): void {
+    if (this.closed) return
+    this.closed = true
+    this.input.off('error', this.onInputError)
+    this.input.off('end', this.onInputEnd)
+    this.transport.close()
+  }
+
+  private async guarded<T>(pending: Promise<T>, signal: AbortSignal): Promise<T> {
+    const withFatal = Promise.race([
+      pending,
+      this.fatal.promise.then((error): Promise<never> => Promise.reject(error)),
+    ])
+    return raceAbort(withFatal, signal)
+  }
+
+  private fail(error: Error): void {
+    this.fatal.resolve(error)
+  }
+
+  private readonly onInputError = (error: Error): void => {
+    this.fail(error)
+  }
+
+  private readonly onInputEnd = (): void => {
+    this.fail(new Error('subagent-codex: app-server protocol stream closed'))
+  }
+
+  private observePendingTurnId(id: string): void {
+    if (this.turnCompleted === undefined) {
+      throw new Error('subagent-codex: app-server referenced a turn before turn/start')
+    }
+    if (this.pendingTurnId !== undefined && this.pendingTurnId !== id) {
+      throw new Error('subagent-codex: app-server referenced conflicting turns')
+    }
+    this.pendingTurnId = id
+  }
+
+  private commitTurnId(id: string): void {
+    if (this.pendingTurnId !== undefined && this.pendingTurnId !== id) {
+      throw new Error('subagent-codex: turn/start response did not match the active turn')
+    }
+    this.turnId = id
+    const notifications = this.earlyTurnNotifications.splice(0)
+    for (const notification of notifications) {
+      this.handleNotification(notification.method, notification.params)
+    }
+  }
+
+  private validateRunIds(params: JsonObject, nullableTurn = false): void {
+    if (params.threadId !== this.threadId) {
+      throw new Error('subagent-codex: app-server request referenced another thread')
+    }
+    if (nullableTurn && params.turnId === null) return
+    const id = string(params.turnId, 'server request turn id')
+    if (this.turnId === undefined) {
+      this.observePendingTurnId(id)
+      return
+    }
+    if (id !== this.turnId) {
+      throw new Error('subagent-codex: app-server request referenced another turn')
+    }
+  }
+
+  private handleServerRequest(method: string, params: JsonObject): Promise<unknown> {
+    try {
+      switch (method) {
+        case 'item/commandExecution/requestApproval':
+        case 'item/fileChange/requestApproval':
+          this.validateRunIds(params)
+          return Promise.resolve({ decision: 'decline' })
+        case 'item/permissions/requestApproval':
+          this.validateRunIds(params)
+          return Promise.resolve({ permissions: {}, scope: 'turn' })
+        case 'mcpServer/elicitation/request':
+          this.validateRunIds(params, true)
+          return Promise.resolve({ action: 'decline', content: null, _meta: null })
+        default:
+          throw new Error(`subagent-codex: unsupported app-server request ${JSON.stringify(method)}`)
+      }
+    } catch (error: unknown) {
+      const normalized = thrown(error)
+      this.fail(normalized)
+      return Promise.reject(normalized)
+    }
+  }
+
+  private handleNotification(method: string, params: JsonObject): void {
+    if (method === 'turn/started') {
+      if (params.threadId !== this.threadId) return
+      const turn = object(params.turn, 'turn/started turn')
+      if (this.turnCompleted !== undefined && this.turnId === undefined) {
+        this.observePendingTurnId(string(turn.id, 'turn/started turn id'))
+      }
+      return
+    }
+    if (method === 'item/completed') {
+      if (params.threadId !== this.threadId) return
+      const id = string(params.turnId, 'item/completed turn id')
+      if (this.turnId === undefined) {
+        if (this.turnCompleted !== undefined) {
+          this.observePendingTurnId(id)
+          this.earlyTurnNotifications.push({ method, params })
+        }
+        return
+      }
+      if (id !== this.turnId) return
+      const item = object(params.item, 'item/completed item')
+      if (item.type !== 'agentMessage') return
+      const text = typeof item.text === 'string'
+        ? item.text
+        : (() => { throw new Error('subagent-codex: app-server returned an invalid agent message') })()
+      if (item.phase === 'final_answer') {
+        this.finalAnswers.push(text)
+      } else if (item.phase === null) {
+        this.unphasedAnswers.push(text)
+      } else if (item.phase !== 'commentary') {
+        throw new Error(`subagent-codex: app-server returned an unknown agent message phase ${JSON.stringify(item.phase)}`)
+      }
+      return
+    }
+    if (method !== 'turn/completed') return
+    if (params.threadId !== this.threadId) return
+    const turn = object(params.turn, 'turn/completed turn')
+    const id = string(turn.id, 'turn/completed turn id')
+    const turnCompleted = this.turnCompleted
+    if (turnCompleted === undefined) return
+    if (this.turnId === undefined) {
+      this.observePendingTurnId(id)
+      this.earlyTurnNotifications.push({ method, params })
+      return
+    }
+    if (id !== this.turnId) return
+    if (!['completed', 'interrupted', 'failed'].includes(String(turn.status))) {
+      throw new Error(`subagent-codex: app-server returned invalid terminal turn status ${String(turn.status)}`)
+    }
+    turnCompleted.resolve(params)
+  }
+}