fix(subagent): preserve Codex diagnostic ordering

2026-08-15 19:58:08 +08:00
1 parent d1e9dcae7a
commit 6ed3cab9b5
@@ -158,7 +158,17 @@ export async function startCodexRun(
    const bytes = typeof chunk === 'string' ? Buffer.from(chunk) : chunk
    wire.observeStderr(bytes.toString())
    try {
-      writeSync(process.stderr.fd, bytes)
+      let offset = 0
+      while (offset < bytes.byteLength) {
+        const written = writeSync(
+          process.stderr.fd,
+          bytes,
+          offset,
+          bytes.byteLength - offset,
+        )
+        if (written <= 0) throw new Error('subagent-codex: host stderr made no write progress')
+        offset += written
+      }
    } catch {
      // Host stderr is an observation sink, not a child-run failure authority.
    }
@@ -152,7 +152,10 @@ export class CodexAppServerWire {
  private threadId: string | undefined
  private turnId: string | undefined
  private pendingTurnId: string | undefined
-  private turnCompleted: PromiseWithResolvers<JsonObject> | undefined
+  private turnCompleted: PromiseWithResolvers<{
+    readonly params: JsonObject
+    readonly order: number
+  }> | undefined
  private readonly earlyTurnNotifications: Array<{
    readonly method: string
    readonly params: JsonObject
@@ -163,6 +166,13 @@ export class CodexAppServerWire {
  private diagnostic: string | undefined
  private diagnosticOrder = 0
  private observationOrder = 0
+  private pendingDiagnostic: {
+    readonly turnId: string
+    readonly order: number
+    readonly request: Parameters<typeof unattendedDiagnostic>[1]
+    readonly decision: Parameters<typeof unattendedDiagnostic>[2]
+    readonly reason: string
+  } | undefined
  private stderrTail = ''
  private closed = false

@@ -247,7 +257,10 @@ export class CodexAppServerWire {
    texts: readonly string[],
    signal: AbortSignal,
  ): Promise<SubagentResult> {
-    const completion = Promise.withResolvers<JsonObject>()
+    const completion = Promise.withResolvers<{
+      readonly params: JsonObject
+      readonly order: number
+    }>()
    this.turnCompleted = completion
    const threadId = this.threadId as string
    const response = object(await this.guarded(this.transport.request('turn/start', {
@@ -258,7 +271,7 @@ export class CodexAppServerWire {
    this.commitTurnId(string(turn.id, 'turn/start turn id'))

    const completed = await this.guarded(completion.promise, signal)
-    const terminal = object(completed.turn, 'turn/completed turn')
+    const terminal = object(completed.params.turn, 'turn/completed turn')
    const status = terminal.status
    if (isContextWindowExceeded(terminal)) {
      return { output: this.collectOutput(), stopReason: 'max-tokens' }
@@ -270,6 +283,7 @@ export class CodexAppServerWire {
          'sandbox execution',
          'failed',
          'Codex reported a sandbox failure',
+          completed.order,
        )
      }
      const detail = status === 'failed'
@@ -383,6 +397,16 @@ export class CodexAppServerWire {
      throw new Error('subagent-codex: turn/start response did not match the active turn')
    }
    this.turnId = id
+    const pendingDiagnostic = this.pendingDiagnostic
+    this.pendingDiagnostic = undefined
+    if (pendingDiagnostic?.turnId === id) {
+      this.recordDiagnostic(
+        pendingDiagnostic.request,
+        pendingDiagnostic.decision,
+        pendingDiagnostic.reason,
+        pendingDiagnostic.order,
+      )
+    }
    const notifications = this.earlyTurnNotifications.splice(0)
    for (const notification of notifications) {
      this.handleNotification(
@@ -393,19 +417,43 @@ export class CodexAppServerWire {
    }
  }

-  private validateRunIds(params: JsonObject, nullableTurn = false): void {
+  private validateRunIds(
+    params: JsonObject,
+    nullableTurn = false,
+  ): string | undefined {
    if (params.threadId !== this.threadId) {
      throw new Error('subagent-codex: app-server request referenced another thread')
    }
-    if (nullableTurn && params.turnId === null) return
+    if (nullableTurn && params.turnId === null) return undefined
    const id = string(params.turnId, 'server request turn id')
    if (this.turnId === undefined) {
      this.observePendingTurnId(id)
-      return
+      return id
    }
    if (id !== this.turnId) {
      throw new Error('subagent-codex: app-server request referenced another turn')
    }
+    return undefined
+  }
+
+  private recordRequestDiagnostic(
+    provisionalTurnId: string | undefined,
+    request: Parameters<typeof unattendedDiagnostic>[1],
+    decision: Parameters<typeof unattendedDiagnostic>[2],
+    reason: string,
+  ): void {
+    const order = this.nextObservationOrder()
+    if (provisionalTurnId !== undefined) {
+      this.pendingDiagnostic = {
+        turnId: provisionalTurnId,
+        order,
+        request,
+        decision,
+        reason,
+      }
+      return
+    }
+    this.recordDiagnostic(request, decision, reason, order)
  }

  private recordDiagnostic(
@@ -455,46 +503,48 @@ export class CodexAppServerWire {
    try {
      switch (method) {
        case 'item/commandExecution/requestApproval':
-          this.validateRunIds(params)
-          {
-            const decision = unattendedDecision(params)
-            this.recordDiagnostic(
-              'command approval',
-              decision === 'cancel' ? 'cancelled' : 'declined',
-              'the provider does not grant interactive approval',
-            )
-            return Promise.resolve({ decision })
-          }
+        {
+          const provisionalTurnId = this.validateRunIds(params)
+          const decision = unattendedDecision(params)
+          this.recordRequestDiagnostic(
+            provisionalTurnId,
+            'command approval',
+            decision === 'cancel' ? 'cancelled' : 'declined',
+            'the provider does not grant interactive approval',
+          )
+          return Promise.resolve({ decision })
+        }
        case 'item/fileChange/requestApproval':
-          this.validateRunIds(params)
-          {
-            const decision = unattendedDecision(params)
-            this.recordDiagnostic(
-              'file approval',
-              decision === 'cancel' ? 'cancelled' : 'declined',
-              'the provider does not grant interactive approval',
-            )
-            return Promise.resolve({ decision })
-          }
+        {
+          const provisionalTurnId = this.validateRunIds(params)
+          const decision = unattendedDecision(params)
+          this.recordRequestDiagnostic(
+            provisionalTurnId,
+            'file approval',
+            decision === 'cancel' ? 'cancelled' : 'declined',
+            'the provider does not grant interactive approval',
+          )
+          return Promise.resolve({ decision })
+        }
        case 'item/permissions/requestApproval':
-          this.validateRunIds(params)
-          this.recordDiagnostic(
+          this.recordRequestDiagnostic(
+            this.validateRunIds(params),
            'permission grant',
            'denied',
            'the provider grants no additional turn permissions',
          )
          return Promise.resolve({ permissions: {}, scope: 'turn' })
        case 'item/tool/requestUserInput':
-          this.validateRunIds(params)
-          this.recordDiagnostic(
+          this.recordRequestDiagnostic(
+            this.validateRunIds(params),
            'user input',
            'empty response',
            'the provider does not collect interactive answers',
          )
          return Promise.resolve({ answers: {} })
        case 'mcpServer/elicitation/request':
-          this.validateRunIds(params, true)
-          this.recordDiagnostic(
+          this.recordRequestDiagnostic(
+            this.validateRunIds(params, true),
            'MCP elicitation',
            'declined',
            'the provider does not collect interactive MCP input',
@@ -575,6 +625,9 @@ export class CodexAppServerWire {
    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)
+    turnCompleted.resolve({
+      params,
+      order: order ?? this.nextObservationOrder(),
+    })
  }
 }
@@ -30,6 +30,8 @@ const { hostStderrWrite } = vi.hoisted(() => ({
  hostStderrWrite: {
    capture: false,
    failNext: false,
+    zeroNext: false,
+    maxBytesPerWrite: undefined as number | undefined,
    chunks: [] as Buffer[],
  },
 }))
@@ -38,8 +40,17 @@ vi.mock('node:fs', async (importOriginal) => {
  const actual = await importOriginal<typeof import('node:fs')>()
  return {
    ...actual,
-    writeSync(fd: number, value: string | Uint8Array): number {
+    writeSync(
+      fd: number,
+      value: string | Uint8Array,
+      offset?: number | null,
+      length?: number | null,
+    ): number {
      if (fd === 2 && hostStderrWrite.capture) {
+        if (hostStderrWrite.zeroNext) {
+          hostStderrWrite.zeroNext = false
+          return 0
+        }
        if (hostStderrWrite.failNext) {
          hostStderrWrite.failNext = false
          throw Object.assign(new Error('host stderr broke'), { code: 'EIO' })
@@ -47,12 +58,26 @@ vi.mock('node:fs', async (importOriginal) => {
        const bytes = typeof value === 'string'
          ? Buffer.from(value)
          : Buffer.from(value.buffer, value.byteOffset, value.byteLength)
-        hostStderrWrite.chunks.push(bytes)
-        return bytes.byteLength
+        const start = typeof value === 'string' ? 0 : offset ?? 0
+        const requested = typeof value === 'string'
+          ? bytes.byteLength
+          : length ?? bytes.byteLength - start
+        const written = Math.min(
+          requested,
+          hostStderrWrite.maxBytesPerWrite ?? requested,
+        )
+        hostStderrWrite.chunks.push(Buffer.from(bytes.subarray(start, start + written)))
+        return written
      }
      return typeof value === 'string'
        ? actual.writeSync(fd, value, null, 'utf8')
-        : actual.writeSync(fd, value, 0, value.byteLength, null)
+        : actual.writeSync(
+          fd,
+          value,
+          offset ?? 0,
+          length ?? value.byteLength - (offset ?? 0),
+          null,
+        )
    },
  }
 })
@@ -698,12 +723,13 @@ describe('CodexAppServerWire', () => {
    expect(await child.peer.nextResponse('command')).toMatchObject({
      result: { decision: 'cancel' },
    })
-    expect(wire.collectDiagnostic()).toBe(
-      'Codex unattended decision (mode: never; request: command approval; decision: cancelled): the provider does not grant interactive approval',
-    )
+    expect(wire.collectDiagnostic()).toBeUndefined()

    child.peer.respond(turnStart, { turn: { id: 'turn-1' } })
    await nextTask()
+    expect(wire.collectDiagnostic()).toBe(
+      'Codex unattended decision (mode: never; request: command approval; decision: cancelled): the provider does not grant interactive approval',
+    )
    const requests = [
      {
        id: 'command-decline',
@@ -952,6 +978,25 @@ describe('CodexAppServerWire', () => {
    wire.close()
  })

+  it('keeps a newer stderr fact after replaying an older early terminal', async () => {
+    hostStderrWrite.capture = true
+    hostStderrWrite.chunks.length = 0
+    const { child, wire } = await initializeWire()
+    const result = wire.runTurn(['task'], new AbortController().signal)
+    const turnStart = await child.peer.nextMethod('turn/start')
+    child.peer.send(turnCompleted('failed', 'turn-1', 'thread-1', {
+      message: 'sandbox failure',
+      codexErrorInfo: 'sandboxError',
+    }))
+    await nextTask()
+    wire.observeStderr('approval policy is Never; reject command')
+    child.peer.respond(turnStart, { turn: { id: 'turn-1' } })
+    await expect(result).rejects.toThrow('sandboxError')
+    expect(wire.collectDiagnostic()).toContain('request: command execution')
+    wire.close()
+    hostStderrWrite.capture = false
+  })
+
  it('fails the run on unknown requests or wrong request association', async () => {
    for (const serverRequest of [
      {
@@ -1032,6 +1077,26 @@ describe('CodexAppServerWire', () => {
    wire.close()
  })

+  it('does not retain a diagnostic from a mismatched provisional request', async () => {
+    const { child, wire } = await initializeWire()
+    const result = wire.runTurn(['task'], new AbortController().signal)
+    const turnStart = await child.peer.nextMethod('turn/start')
+    child.peer.send({
+      id: 'provisional-approval',
+      method: 'item/commandExecution/requestApproval',
+      params: {
+        threadId: 'thread-1',
+        turnId: 'turn-early',
+        availableDecisions: ['cancel'],
+      },
+    })
+    await child.peer.nextResponse('provisional-approval')
+    child.peer.respond(turnStart, { turn: { id: 'turn-response' } })
+    await expect(result).rejects.toThrow('did not match the active turn')
+    expect(wire.collectDiagnostic()).toBeUndefined()
+    wire.close()
+  })
+
  it('rejects conflicting early notifications and requests before turn/start', async () => {
    {
      const { child, wire } = await initializeWire()
@@ -1325,6 +1390,7 @@ describe('run lifecycle and quiescence', () => {
  it('forwards stderr while extracting only a fixed safe permission signature', async () => {
    const child = fakeChild()
    hostStderrWrite.capture = true
+    hostStderrWrite.maxBytesPerWrite = 3
    hostStderrWrite.chunks.length = 0
    const { run, turnStart } = await publishRun(child)
    child.peer.respond(turnStart, { turn: { id: 'turn-1' } })
@@ -1341,9 +1407,10 @@ describe('run lifecycle and quiescence', () => {
      stopReason: 'error',
    })
    expect(Buffer.concat(hostStderrWrite.chunks).toString()).toContain('SECRET_TOKEN')
-    expect(hostStderrWrite.chunks).toHaveLength(3)
+    expect(hostStderrWrite.chunks.length).toBeGreaterThan(3)
    await run.dispose()
    expect(child.stderr.listenerCount('data')).toBe(0)
+    hostStderrWrite.maxBytesPerWrite = undefined
    hostStderrWrite.capture = false
  })

@@ -1353,11 +1420,35 @@ describe('run lifecycle and quiescence', () => {
    hostStderrWrite.failNext = true
    const { run, turnStart } = await publishRun(child)
    child.peer.respond(turnStart, { turn: { id: 'turn-1' } })
-    child.stderr.write('forwarding failure')
-    child.peer.send(agentMessage('answer', 'final_answer'), turnCompleted('completed'))
+    child.stderr.write('approval policy is Never; reject command')
+    child.peer.send(turnCompleted('failed', 'turn-1', 'thread-1', {
+      message: 'fixture terminal failure',
+      codexErrorInfo: 'badRequest',
+    }))
    await expect(run.result).resolves.toEqual({
-      output: [{ type: 'text', text: 'answer' }],
-      stopReason: 'completed',
+      output: [],
+      diagnostic: 'Codex unattended decision (mode: never; request: command execution; decision: denied): Codex rejected an escalation because the selected policy never asks for approval',
+      stopReason: 'error',
+    })
+    await run.dispose()
+    hostStderrWrite.capture = false
+  })
+
+  it('contains a zero-progress host stderr write without losing the diagnostic', async () => {
+    const child = fakeChild()
+    hostStderrWrite.capture = true
+    hostStderrWrite.zeroNext = true
+    const { run, turnStart } = await publishRun(child)
+    child.peer.respond(turnStart, { turn: { id: 'turn-1' } })
+    child.stderr.write('approval policy is Never; reject command')
+    child.peer.send(turnCompleted('failed', 'turn-1', 'thread-1', {
+      message: 'fixture terminal failure',
+      codexErrorInfo: 'badRequest',
+    }))
+    await expect(run.result).resolves.toEqual({
+      output: [],
+      diagnostic: 'Codex unattended decision (mode: never; request: command execution; decision: denied): Codex rejected an escalation because the selected policy never asks for approval',
+      stopReason: 'error',
    })
    await run.dispose()
    hostStderrWrite.capture = false