fix(subagent): close Codex provider review findings

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

@@ -116,13 +116,11 @@ export async function startCodexRun(
     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 wire = new CodexAppServerWire(
+    child.stdout as NonNullable<SubprocessHandle['stdout']>,
+    child.stdin as NonNullable<SubprocessHandle['stdin']>,
+  )
   const disposeProcess = (): Promise<void> =>
     disposeCodexChild(wire, child, spec.disposeGraceMs)
 
@@ -137,15 +135,10 @@ export async function startCodexRun(
   // 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
+    if (runAbort.signal.aborted) return
     runAbort.abort(new Error('subagent-codex: run cancelled locally'))
-    settleCancellation()
     wire.interrupt()
   }
   const onAbort = (): void => { requestCancel() }
@@ -165,7 +158,7 @@ export async function startCodexRun(
         'subagent-codex: startup failed and app-server cleanup also failed',
       )
     }
-    if (flags.cancelled) {
+    if (runAbort.signal.aborted) {
       throw new Error('subagent-codex: request was aborted before app-server startup')
     }
     throw thrown(error)
@@ -174,15 +167,11 @@ export async function startCodexRun(
   const collectOutput = (): ContentBlock[] => wire.collectOutput()
   const result: Promise<SubagentResult> = settleRunResult({
     attempt: () => Promise.race([
-      wire.runTurn(texts, runAbort.signal, () => flags.cancelled),
+      wire.runTurn(texts, runAbort.signal, () => runAbort.signal.aborted),
       processFailure,
-      cancellation.then((): SubagentResult => ({
-        output: collectOutput(),
-        stopReason: 'aborted',
-      })),
     ]),
     collectOutput,
-    cancelled: () => flags.cancelled,
+    cancelled: () => runAbort.signal.aborted,
     onError: spec.onError,
     signal: request.signal,
     onAbort,

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

@@ -83,9 +83,8 @@ export class CodexAppServerWire {
     readonly method: string
     readonly params: JsonObject
   }> = []
-  private readonly finalAnswers: string[] = []
-  private readonly unphasedAnswers: string[] = []
-  private started = false
+  private lastFinalAnswer: string | undefined
+  private lastUnphasedAnswer: string | undefined
   private closed = false
 
   constructor(
@@ -101,14 +100,16 @@ export class CodexAppServerWire {
         this.fail(thrown(error))
       }
     })
+    this.input.on('error', this.onInputError)
+    this.input.on('end', this.onInputEnd)
+    // Pipe errors can race protocol closure and process teardown. Retain both
+    // error listeners for the lifetime of their per-run streams so no late
+    // EPIPE or read failure becomes an unhandled EventEmitter error.
+    output.on('error', this.onOutputError)
   }
 
   /** 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()
   }
 
@@ -166,16 +167,11 @@ export class CodexAppServerWire {
     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 threadId = this.threadId as string
     const response = object(await this.guarded(this.transport.request('turn/start', {
-      threadId: this.threadId,
+      threadId,
       input: texts.map(text => ({ type: 'text', text, text_elements: [] })),
     }, signal), signal), 'turn/start response')
     const turn = object(response.turn, 'turn/start turn')
@@ -216,9 +212,7 @@ export class CodexAppServerWire {
    * @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)
+    const selected = this.lastFinalAnswer ?? this.lastUnphasedAnswer
     return selected !== undefined && selected.trim().length > 0
       ? [{ type: 'text', text: selected }]
       : []
@@ -228,7 +222,6 @@ export class CodexAppServerWire {
   close(): void {
     if (this.closed) return
     this.closed = true
-    this.input.off('error', this.onInputError)
     this.input.off('end', this.onInputEnd)
     this.transport.close()
   }
@@ -249,6 +242,10 @@ export class CodexAppServerWire {
     this.fail(error)
   }
 
+  private readonly onOutputError = (error: Error): void => {
+    this.fail(error)
+  }
+
   private readonly onInputEnd = (): void => {
     this.fail(new Error('subagent-codex: app-server protocol stream closed'))
   }
@@ -338,9 +335,9 @@ export class CodexAppServerWire {
         ? item.text
         : (() => { throw new Error('subagent-codex: app-server returned an invalid agent message') })()
       if (item.phase === 'final_answer') {
-        this.finalAnswers.push(text)
+        this.lastFinalAnswer = text
       } else if (item.phase === null) {
-        this.unphasedAnswers.push(text)
+        this.lastUnphasedAnswer = text
       } else if (item.phase !== 'commentary') {
         throw new Error(`subagent-codex: app-server returned an unknown agent message phase ${JSON.stringify(item.phase)}`)
       }

packages/subagent/subagent-codex/tests/subagent-codex.spec.ts

@@ -90,8 +90,6 @@ class ProtocolPeer {
 
 interface FakeChildOptions {
   readonly pid?: number
-  readonly stdin?: boolean
-  readonly stdout?: boolean
   readonly exitOnTerminate?: boolean
   readonly waitForExitResult?: boolean
   readonly doneError?: Error
@@ -161,8 +159,8 @@ function fakeChild(options: FakeChildOptions = {}): FakeChild {
   })
   const handle: SubprocessHandle = {
     pid: options.pid ?? 1234,
-    stdin: options.stdin === false ? undefined : toChild,
-    stdout: options.stdout === false ? undefined : fromChild,
+    stdin: toChild,
+    stdout: fromChild,
     stderr: undefined,
     collected: {},
     done,
@@ -334,7 +332,6 @@ describe('CodexAppServerWire', () => {
     const wire = new CodexAppServerWire(child.handle.stdout!, child.handle.stdin!)
     expect(wire.collectOutput()).toEqual([])
     wire.start()
-    wire.start()
 
     const initializing = wire.initialize(new AbortController().signal)
     const initialize = await child.peer.nextMethod('initialize')
@@ -451,27 +448,6 @@ describe('CodexAppServerWire', () => {
     }
   })
 
-  it('rejects a turn before thread publication and a second one-shot turn', async () => {
-    const child = fakeChild()
-    const wire = new CodexAppServerWire(child.handle.stdout!, child.handle.stdin!)
-    await expect(wire.runTurn(['task'], new AbortController().signal, () => false))
-      .rejects.toThrow('before thread/start')
-    const initialized = await initializeWire()
-    const first = initialized.wire.runTurn(
-      ['task'],
-      new AbortController().signal,
-      () => false,
-    )
-    await initialized.child.peer.nextMethod('turn/start')
-    await expect(initialized.wire.runTurn(
-      ['again'],
-      new AbortController().signal,
-      () => false,
-    )).rejects.toThrow('already started')
-    initialized.wire.close()
-    await expect(first).rejects.toThrow('transport closed')
-  })
-
   it('fails closed for empty output, malformed messages, phases, and terminal status', async () => {
     const scenarios: Array<{
       readonly frames: JsonObject[]
@@ -757,6 +733,17 @@ describe('CodexAppServerWire', () => {
       await expect(pending).rejects.toThrow('stdout broke')
       wire.close()
     }
+    {
+      const child = fakeChild()
+      const wire = new CodexAppServerWire(child.handle.stdout!, child.handle.stdin!)
+      wire.start()
+      const pending = wire.initialize(new AbortController().signal)
+      await child.peer.nextMethod('initialize')
+      child.toChild.emit('error', new Error('stdin broke'))
+      await expect(pending).rejects.toThrow('stdin broke')
+      wire.close()
+      child.toChild.emit('error', new Error('late stdin close'))
+    }
   })
 })
 
@@ -918,16 +905,6 @@ describe('run lifecycle and quiescence', () => {
     )
   })
 
-  it('rejects a missing protocol stream after reaping the unpublished child', async () => {
-    for (const options of [{ stdin: false }, { stdout: false }]) {
-      const child = fakeChild(options)
-      await expect(startCodexRun(request(), runSpec(child)))
-        .rejects.toThrow('dropped a piped protocol stream')
-      expect(child.terminate).toHaveBeenCalledTimes(1)
-      expect(child.waitForExit).toHaveBeenCalledTimes(1)
-    }
-  })
-
   it('keeps overlapping runs isolated', async () => {
     const first = fakeChild()
     const second = fakeChild()