diff --git a/packages/subagent/subagent-codex/src/run.ts b/packages/subagent/subagent-codex/src/run.ts index c5eb7c1e61..78f9a6fd9c 100644 --- a/packages/subagent/subagent-codex/src/run.ts +++ b/packages/subagent/subagent-codex/src/run.ts @@ -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. } diff --git a/packages/subagent/subagent-codex/src/wire.ts b/packages/subagent/subagent-codex/src/wire.ts index 28777b7d7a..a777b05331 100644 --- a/packages/subagent/subagent-codex/src/wire.ts +++ b/packages/subagent/subagent-codex/src/wire.ts @@ -152,7 +152,10 @@ export class CodexAppServerWire { private threadId: string | undefined private turnId: string | undefined private pendingTurnId: string | undefined - private turnCompleted: PromiseWithResolvers | 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[1] + readonly decision: Parameters[2] + readonly reason: string + } | undefined private stderrTail = '' private closed = false @@ -247,7 +257,10 @@ export class CodexAppServerWire { texts: readonly string[], signal: AbortSignal, ): Promise { - const completion = Promise.withResolvers() + 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[1], + decision: Parameters[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(), + }) } } diff --git a/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts b/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts index 36c7185952..1c63e11592 100644 --- a/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts +++ b/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts @@ -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() 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