From 9e91e206d4ad1662a843956cf73892aa4f3bf94a Mon Sep 17 00:00:00 2001 From: pku-xht Date: Wed, 5 Aug 2026 07:36:54 +0800 Subject: [PATCH] refactor(subagent-codex): centralize cancellation settlement --- packages/subagent/subagent-codex/src/run.ts | 2 +- packages/subagent/subagent-codex/src/wire.ts | 4 -- .../tests/subagent-codex.spec.ts | 44 +++++-------------- 3 files changed, 13 insertions(+), 37 deletions(-) diff --git a/packages/subagent/subagent-codex/src/run.ts b/packages/subagent/subagent-codex/src/run.ts index ecf71fb2ba..c3ebf4ba19 100644 --- a/packages/subagent/subagent-codex/src/run.ts +++ b/packages/subagent/subagent-codex/src/run.ts @@ -179,7 +179,7 @@ export async function startCodexRun( const collectOutput = (): ContentBlock[] => wire.collectOutput() const result: Promise = settleRunResult({ attempt: () => Promise.race([ - wire.runTurn(texts, runAbort.signal, () => runAbort.signal.aborted), + wire.runTurn(texts, runAbort.signal), processFailure, ]), collectOutput, diff --git a/packages/subagent/subagent-codex/src/wire.ts b/packages/subagent/subagent-codex/src/wire.ts index ca24fdcadf..51be212841 100644 --- a/packages/subagent/subagent-codex/src/wire.ts +++ b/packages/subagent/subagent-codex/src/wire.ts @@ -168,13 +168,11 @@ export class CodexAppServerWire { * 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 subagent result. */ async runTurn( texts: readonly string[], signal: AbortSignal, - cancelled: () => boolean, ): Promise { const completion = Promise.withResolvers() this.turnCompleted = completion @@ -187,8 +185,6 @@ export class CodexAppServerWire { 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 (isContextWindowExceeded(terminal)) { diff --git a/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts b/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts index 37c8649f8b..de89aa4854 100644 --- a/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts +++ b/packages/subagent/subagent-codex/tests/subagent-codex.spec.ts @@ -393,7 +393,6 @@ describe('CodexAppServerWire', () => { const result = wire.runTurn( ['first', 'second'], new AbortController().signal, - () => false, ) const turnStart = await child.peer.nextMethod('turn/start') expect(turnStart.params).toEqual({ @@ -437,7 +436,7 @@ describe('CodexAppServerWire', () => { it('uses the last nullable-phase answer when no explicit final exists', async () => { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) child.peer.send( @@ -454,7 +453,7 @@ describe('CodexAppServerWire', () => { it('maps only an explicit context-window failure to max-tokens', async () => { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) child.peer.send( @@ -494,7 +493,7 @@ describe('CodexAppServerWire', () => { } { const { child, wire } = await initializeWire() - const pending = wire.runTurn(['task'], new AbortController().signal, () => false) + const pending = wire.runTurn(['task'], new AbortController().signal) const frame = await child.peer.nextMethod('turn/start') child.peer.respond(frame, { turn: { id: '' } }) await expect(pending).rejects.toThrow('turn/start turn id') @@ -542,7 +541,7 @@ describe('CodexAppServerWire', () => { ] for (const scenario of scenarios) { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) child.peer.send(...scenario.frames) @@ -553,7 +552,7 @@ describe('CodexAppServerWire', () => { it('fails closed when terminal notification params are not an object', async () => { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) child.peer.send({ method: 'turn/completed', params: null }) @@ -563,7 +562,7 @@ describe('CodexAppServerWire', () => { it('keeps an unsupported request authoritative over an early terminal in the same chunk', async () => { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.send( { id: turnStart.id, result: { turn: { id: 'turn-1' } } }, @@ -575,28 +574,9 @@ describe('CodexAppServerWire', () => { wire.close() }) - it('gives local cancellation precedence over a remote completed turn', async () => { - const { child, wire } = await initializeWire() - let cancelled = false - const result = wire.runTurn( - ['task'], - new AbortController().signal, - () => cancelled, - ) - const turnStart = await child.peer.nextMethod('turn/start') - child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) - cancelled = true - child.peer.send(agentMessage('late', 'final_answer'), turnCompleted('completed')) - await expect(result).resolves.toEqual({ - output: [{ type: 'text', text: 'late' }], - stopReason: 'aborted', - }) - wire.close() - }) - it('answers all five unattended request classes without granting authority', async () => { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.send({ @@ -699,7 +679,7 @@ describe('CodexAppServerWire', () => { }, ]) { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) await nextTask() @@ -713,7 +693,7 @@ describe('CodexAppServerWire', () => { it('rejects conflicting early turn identities before accepting output', async () => { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.send({ method: 'turn/started', @@ -738,7 +718,7 @@ describe('CodexAppServerWire', () => { } { const { child, wire } = await initializeWire() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) await child.peer.nextMethod('turn/start') child.peer.send( { @@ -755,7 +735,7 @@ describe('CodexAppServerWire', () => { it('interrupts only an active open turn and contains remote interrupt failure', async () => { const { child, wire } = await initializeWire() wire.interrupt() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) await nextTask() @@ -790,7 +770,7 @@ describe('CodexAppServerWire', () => { ) await nextTask() - const result = wire.runTurn(['task'], new AbortController().signal, () => false) + const result = wire.runTurn(['task'], new AbortController().signal) const turnStart = await child.peer.nextMethod('turn/start') child.peer.respond(turnStart, { turn: { id: 'turn-1' } }) await nextTask()