From db63d08168505e2a15026ded3076fdbbaebb1288 Mon Sep 17 00:00:00 2001 From: Paul Bakaus Date: Mon, 13 Jul 2026 10:08:27 -0700 Subject: [PATCH] Detach canceled Live generation tails AI-assisted: Codex --- scripts/benchmark-live.mjs | 3 + .../scripts/live/codex-worker-supervisor.mjs | 90 ++++++++++++++----- tests/live-codex-worker-supervisor.test.mjs | 46 +++++++++- 3 files changed, 118 insertions(+), 21 deletions(-) diff --git a/scripts/benchmark-live.mjs b/scripts/benchmark-live.mjs index 070d72d6a..214540849 100644 --- a/scripts/benchmark-live.mjs +++ b/scripts/benchmark-live.mjs @@ -204,6 +204,9 @@ try { if (args.action) await selectAction(session.page, String(args.action)); await resetBrowserTimingProbe(session.page, `${iteration}-followup`); const nextFirstVariant = waitForFirstVariant(session.page); + // Keep teardown from surfacing this background waiter as an unhandled + // rejection if a preceding follow-up action is the real failure. + void nextFirstVariant.catch(() => {}); await clickGo(session.page); await waitForBrowserGeneratePost(session.page); run.acceptToNextGoDispatchMs = roundMs(performance.now() - acceptStartedAt); diff --git a/skill/scripts/live/codex-worker-supervisor.mjs b/skill/scripts/live/codex-worker-supervisor.mjs index 2380d1882..874fe5453 100644 --- a/skill/scripts/live/codex-worker-supervisor.mjs +++ b/skill/scripts/live/codex-worker-supervisor.mjs @@ -74,6 +74,7 @@ export class CodexLiveWorkerSupervisor { this.canceled = new Set(); this.queuedGenerationIds = new Set(); this.thread = null; + this.threadReady = Promise.resolve(null); this.model = null; this.liveSpec = ''; } @@ -104,16 +105,9 @@ export class CodexLiveWorkerSupervisor { } } if (!this.thread) { - this.thread = await this.client.startDedicatedThread({ - model: this.model.model || this.model.id, - cwd: this.cwd, - approvalPolicy: 'never', - sandbox: 'read-only', - ephemeral: false, - serviceName: 'impeccable_live_codex_worker', - baseInstructions: buildCodexWorkerInstructions(this.liveSpec), - }); + this.thread = await this.startWorkerThread(); } + this.threadReady = Promise.resolve(this.thread); this.writeState('ready'); return this.status(); } @@ -134,10 +128,12 @@ export class CodexLiveWorkerSupervisor { } if (event.type === 'accept' || event.type === 'discard') { this.canceled.add(event.id); + const replaceBusyThread = this.active?.eventId === event.id; // Cancellation fences publication synchronously. Do not make the // deterministic Accept/Discard path wait on a slow app-server // interrupt round trip before it can update source and reply. void this.cancelActive(event.type, event.id); + if (replaceBusyThread) this.rotateWorkerThread(event.type); const handled = await this.handleAccept(event, this.base, this.token); if (event.type === 'accept' && handled?._acceptResult?.carbonize === true) { await this.postCleanup(this.base, this.token, { @@ -173,12 +169,14 @@ export class CodexLiveWorkerSupervisor { } async processGeneration(event) { + if (this.isCanceled(event.id)) return; + await this.threadReady; if (this.isCanceled(event.id)) return; if (!event.scaffold?.file) event.scaffold = runDeterministicScaffold(event, { cwd: this.cwd, scriptsDir: this.scriptsDir, }); - this.active = { eventId: event.id, turnId: null }; + this.active = { eventId: event.id, turnId: null, threadId: this.thread.id }; this.writeState('working', { eventId: event.id }); try { const expectedVariants = Number(event.count || 1); @@ -202,11 +200,52 @@ export class CodexLiveWorkerSupervisor { file: event.scaffold.file, }); } finally { - this.active = null; - this.writeState('ready'); + if (this.active?.eventId === event.id) { + this.active = null; + this.writeState('ready'); + } } } + startWorkerThread() { + return this.client.startDedicatedThread({ + model: this.model.model || this.model.id, + cwd: this.cwd, + approvalPolicy: 'never', + sandbox: 'read-only', + ephemeral: false, + serviceName: 'impeccable_live_codex_worker', + baseInstructions: buildCodexWorkerInstructions(this.liveSpec), + }); + } + + rotateWorkerThread(reason) { + const priorThread = this.thread; + const drainingQueue = this.queue; + this.queue = Promise.resolve(); + this.thread = null; + this.threadReady = this.startWorkerThread().then((thread) => { + this.thread = thread; + this.writeState('ready', { + rotatedAt: new Date().toISOString(), + rotationReason: reason, + }); + return thread; + }); + void this.threadReady.catch((error) => { + this.writeState('error', { error: error.message, rotationReason: reason }); + this.log(`replacement worker thread failed: ${error.message}`); + }); + if (priorThread) { + void drainingQueue.finally(async () => { + await this.client.archiveThread(priorThread.id).catch((error) => { + this.log(`retired worker thread archive failed: ${error.message}`); + }); + }); + } + return this.threadReady; + } + async runGenerationPhase(event, phase, arrivedVariants) { if (this.isCanceled(event.id)) return; const phaseStartedAt = Date.now(); @@ -277,22 +316,25 @@ export class CodexLiveWorkerSupervisor { if (publicationPromise === pendingPublication) publicationPromise = null; } }; + if (this.isCanceled(event.id)) return; const result = await this.runTurnWithReconnect({ input, outputSchema: CODEX_WORKER_OUTPUT_SCHEMA, onAgentMessage: publishCandidate, + eventId: event.id, }); if (this.isCanceled(event.id)) return; if (!publishedFromMessage) await publishCandidate(result.answer); if (!publishedFromMessage) throw earlyCandidateError || supervisorError('worker_output_not_published'); } - async runTurnWithReconnect({ input, outputSchema, onAgentMessage }) { + async runTurnWithReconnect({ input, outputSchema, onAgentMessage, eventId = this.active?.eventId }) { let firstError; for (let attempt = 0; attempt < 2; attempt += 1) { try { + const threadId = this.thread.id; const turn = await this.client.startTurn({ - threadId: this.thread.id, + threadId, input, cwd: this.cwd, model: this.model.model || this.model.id, @@ -303,16 +345,16 @@ export class CodexLiveWorkerSupervisor { outputSchema, onAgentMessage, onStarted: (turnId) => { - if (!this.active) return; - this.active.turnId = turnId; - if (this.isCanceled(this.active.eventId)) { - this.client.interruptTurn(this.thread.id, turnId).catch(() => {}); + if (this.active?.eventId === eventId) this.active.turnId = turnId; + if (eventId && this.isCanceled(eventId)) { + this.client.interruptTurn(threadId, turnId).catch(() => {}); } }, }); return { ...turn, answer: turn.message }; } catch (error) { if (!firstError) firstError = error; + if (eventId && this.isCanceled(eventId)) throw error; if (attempt > 0 || error.code === 'TURN_INTERRUPTED') throw error; this.log(`app-server turn failed; reconnecting once: ${error.message}`); await this.reconnect(); @@ -339,8 +381,9 @@ export class CodexLiveWorkerSupervisor { if (!this.active) return; if (eventId && this.active.eventId !== eventId) return; this.canceled.add(this.active.eventId); - if (this.active.turnId) { - await this.client.interruptTurn(this.thread.id, this.active.turnId).catch(() => {}); + const threadId = this.active.threadId || this.thread?.id; + if (threadId && this.active.turnId) { + await this.client.interruptTurn(threadId, this.active.turnId).catch(() => {}); } this.log(`interrupted ${this.active.eventId}: ${reason}`); } @@ -363,6 +406,13 @@ export class CodexLiveWorkerSupervisor { async shutdown({ archive = false } = {}) { this.running = false; await this.cancelActive('shutdown'); + await Promise.race([ + this.threadReady.catch(() => null), + new Promise((resolve) => { + const timer = setTimeout(resolve, 1_000); + timer.unref?.(); + }), + ]); let archived = false; if (archive && this.thread) { try { diff --git a/tests/live-codex-worker-supervisor.test.mjs b/tests/live-codex-worker-supervisor.test.mjs index de180f19b..bc0b92c34 100644 --- a/tests/live-codex-worker-supervisor.test.mjs +++ b/tests/live-codex-worker-supervisor.test.mjs @@ -101,6 +101,7 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { client, }); supervisor.thread = { id: 'live-worker-thread' }; + supervisor.model = client.models[0]; supervisor.active = { eventId: 'generation-1', turnId: 'turn-1' }; await supervisor.cancelActive('accept', 'generation-1'); assert.deepEqual(client.calls.interruptTurn, [{ threadId: 'live-worker-thread', turnId: 'turn-1' }]); @@ -137,7 +138,8 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { }, }); supervisor.thread = { id: 'live-worker-thread' }; - supervisor.active = { eventId: 'generation-1', turnId: 'turn-1' }; + supervisor.model = client.models[0]; + supervisor.active = { eventId: 'generation-1', turnId: 'turn-1', threadId: 'live-worker-thread' }; await supervisor.run(); @@ -145,6 +147,46 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { assert.equal(client.calls.interruptTurn.length >= 1, true); }); + it('rotates a busy thread so the next generation does not wait for the canceled tail', async () => { + const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-tail-rotation-')); + const client = fakeClient(); + client.startDedicatedThread = async (params) => { + client.calls.startDedicatedThread.push(params); + await new Promise((resolve) => setImmediate(resolve)); + return { id: 'replacement-live-thread' }; + }; + const supervisor = createSupervisor({ cwd, statePath: path.join(cwd, 'state.json'), client }); + supervisor.model = client.models[0]; + supervisor.thread = { id: 'draining-live-thread' }; + let releaseDrainingQueue; + let drainingQueueFinished = false; + supervisor.queue = new Promise((resolve) => { + releaseDrainingQueue = () => { + drainingQueueFinished = true; + resolve(); + }; + }); + let observedThread = null; + supervisor.runGenerationPhase = async () => { + observedThread = supervisor.thread.id; + }; + supervisor.reply = async () => {}; + + supervisor.rotateWorkerThread('accept'); + await supervisor.processGeneration({ + type: 'generate', + id: 'next-generation', + count: 1, + scaffold: { file: 'src/App.jsx' }, + }); + + assert.equal(observedThread, 'replacement-live-thread'); + assert.equal(drainingQueueFinished, false, 'the next generation must not join the canceled tail queue'); + releaseDrainingQueue(); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(client.calls.archiveThread, [{ threadId: 'draining-live-thread' }]); + }); + it('interrupts a canceled turn whose id arrives after Accept', async () => { const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-late-turn-')); const client = fakeClient(); @@ -260,6 +302,8 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { scriptsDir: path.join(cwd, 'skill/scripts'), reply: async (_base, _token, value) => { replies.push(value); }, }); + supervisor.thread = { id: 'live-worker-thread' }; + supervisor.threadReady = Promise.resolve(supervisor.thread); supervisor.runGenerationPhase = async (_event, phase, arrivedVariants) => { phases.push({ phase, arrivedVariants }); };