From 395953ae1d0f257b2891db35c8811b132f54dba9 Mon Sep 17 00:00:00 2001 From: Paul Bakaus Date: Sun, 12 Jul 2026 20:31:30 -0700 Subject: [PATCH] Publish app-server output before turn completion Validate and transactionally publish complete structured agent messages as soon as they arrive while retaining turn-completion serialization for subsequent phases.\n\nAI-assisted: OpenAI Codex. --- .../scripts/live/codex-app-server-client.mjs | 18 +++++- .../scripts/live/codex-worker-supervisor.mjs | 58 ++++++++++++------- tests/live-codex-app-server-client.test.mjs | 4 ++ tests/live-codex-worker-supervisor.test.mjs | 6 +- 4 files changed, 60 insertions(+), 26 deletions(-) diff --git a/skill/scripts/live/codex-app-server-client.mjs b/skill/scripts/live/codex-app-server-client.mjs index 35fcbdb73..de16395be 100644 --- a/skill/scripts/live/codex-app-server-client.mjs +++ b/skill/scripts/live/codex-app-server-client.mjs @@ -371,7 +371,7 @@ export class CodexAppServerClient { } } - async startTurn({ threadId, input, timeoutMs = this.turnTimeoutMs, onStarted, ...params }) { + async startTurn({ threadId, input, timeoutMs = this.turnTimeoutMs, onStarted, onAgentMessage, ...params }) { this._requireDedicatedThread(threadId); const normalizedInput = typeof input === 'string' ? [{ type: 'text', text: input }] @@ -385,6 +385,8 @@ export class CodexAppServerClient { let started = null; let completed = null; const agentMessages = []; + const agentMessageCallbacks = []; + let firstAgentMessageAt = null; const buffered = []; let completionResolve; let completionReject; @@ -408,7 +410,16 @@ export class CodexAppServerClient { if (notification.method === 'item/completed' && notification.params?.item?.type === 'agentMessage' && typeof notification.params.item.text === 'string') { - agentMessages.push(notification.params.item.text); + const message = notification.params.item.text; + agentMessages.push(message); + if (firstAgentMessageAt == null) firstAgentMessageAt = notification.receivedAt ?? this.clock(); + if (typeof onAgentMessage === 'function') { + agentMessageCallbacks.push(Promise.resolve().then(() => onAgentMessage(message, { + threadId, + turnId, + notification, + }))); + } } if (notification.method === 'turn/completed') { completed = notification; @@ -436,6 +447,7 @@ export class CodexAppServerClient { if (typeof onStarted === 'function') onStarted(turnId, result.turn); for (const notification of buffered.splice(0)) consider(notification); await completionPromise; + await Promise.all(agentMessageCallbacks); const completedAt = completed?.receivedAt ?? this.clock(); return { threadId, @@ -448,6 +460,8 @@ export class CodexAppServerClient { agentMessages, message: agentMessages.at(-1) || null, requestedAt, + firstAgentMessageAt, + firstAgentMessageMs: firstAgentMessageAt == null ? null : firstAgentMessageAt - requestedAt, completedAt, durationMs: completedAt - requestedAt, }; diff --git a/skill/scripts/live/codex-worker-supervisor.mjs b/skill/scripts/live/codex-worker-supervisor.mjs index c4506fc59..a2f54e196 100644 --- a/skill/scripts/live/codex-worker-supervisor.mjs +++ b/skill/scripts/live/codex-worker-supervisor.mjs @@ -217,35 +217,48 @@ export class CodexLiveWorkerSupervisor { screenshotPath: event.screenshotPath, cwd: this.cwd, }); + let publishedFromMessage = false; + let earlyCandidateError = null; + const publishCandidate = async (answer) => { + if (publishedFromMessage || this.isCanceled(event.id)) return; + try { + await this.publishPhase(this.base, this.token, { + eventId: event.id, + phase: phase === 'final' ? 'remaining_variants_validating' : 'first_variant_validating', + durationMs: Date.now() - phaseStartedAt, + }); + applyCodexWorkerOutput({ + output: answer, + prepared, + phase, + expectedVariants: Number(event.count || arrivedVariants), + cwd: this.cwd, + maxBytes: this.config.maxArtifactBytes, + }); + if (this.isCanceled(event.id)) return; + const published = publishCodexWorkerPhase({ event, prepared, arrivedVariants, cwd: this.cwd }); + await this.publishCheckpoint(this.base, this.token, { + event, + published, + scaffold: event.scaffold, + arrivedVariants, + }); + publishedFromMessage = true; + } catch (error) { + earlyCandidateError = error; + } + }; const result = await this.runTurnWithReconnect({ input, outputSchema: CODEX_WORKER_OUTPUT_SCHEMA, + onAgentMessage: publishCandidate, }); if (this.isCanceled(event.id)) return; - await this.publishPhase(this.base, this.token, { - eventId: event.id, - phase: phase === 'final' ? 'remaining_variants_validating' : 'first_variant_validating', - durationMs: Date.now() - phaseStartedAt, - }); - applyCodexWorkerOutput({ - output: result.answer, - prepared, - phase, - expectedVariants: Number(event.count || arrivedVariants), - cwd: this.cwd, - maxBytes: this.config.maxArtifactBytes, - }); - if (this.isCanceled(event.id)) return; - const published = publishCodexWorkerPhase({ event, prepared, arrivedVariants, cwd: this.cwd }); - await this.publishCheckpoint(this.base, this.token, { - event, - published, - scaffold: event.scaffold, - arrivedVariants, - }); + if (!publishedFromMessage) await publishCandidate(result.answer); + if (!publishedFromMessage) throw earlyCandidateError || supervisorError('worker_output_not_published'); } - async runTurnWithReconnect({ input, outputSchema }) { + async runTurnWithReconnect({ input, outputSchema, onAgentMessage }) { let firstError; for (let attempt = 0; attempt < 2; attempt += 1) { try { @@ -259,6 +272,7 @@ export class CodexLiveWorkerSupervisor { approvalPolicy: 'never', sandboxPolicy: { type: 'readOnly' }, outputSchema, + onAgentMessage, onStarted: (turnId) => { if (!this.active) return; this.active.turnId = turnId; diff --git a/tests/live-codex-app-server-client.test.mjs b/tests/live-codex-app-server-client.test.mjs index 73bb082e6..d9729674a 100644 --- a/tests/live-codex-app-server-client.test.mjs +++ b/tests/live-codex-app-server-client.test.mjs @@ -304,19 +304,23 @@ describe('dedicated Codex worker threads', () => { await client.startDedicatedThread({ serviceName: 'impeccable_live_worker' }); let startedTurnId = null; + const deliveredMessages = []; const result = await client.startTurn({ threadId: 'worker', input: 'Reply exactly', model: 'gpt-5.3-codex-spark', effort: 'low', onStarted: (turnId) => { startedTurnId = turnId; }, + onAgentMessage: async (message) => { deliveredMessages.push(message); }, }); assert.equal(startedTurnId, 'turn-1'); assert.equal(result.turnId, 'turn-1'); assert.equal(result.status, 'completed'); assert.deepEqual(result.agentMessages, ['first fragment', 'final answer']); + assert.deepEqual(deliveredMessages, ['first fragment', 'final answer']); assert.equal(result.message, 'final answer'); + assert.equal(result.firstAgentMessageMs >= 0, true); assert.equal(result.started.method, 'turn/started'); assert.equal(result.durationMs > 0, true); await client.close(); diff --git a/tests/live-codex-worker-supervisor.test.mjs b/tests/live-codex-worker-supervisor.test.mjs index bbec2eaa0..4f29d1fac 100644 --- a/tests/live-codex-worker-supervisor.test.mjs +++ b/tests/live-codex-worker-supervisor.test.mjs @@ -235,12 +235,14 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { const final = '

Original

One

Two

Three

'; const client = fakeClient(); let turn = 0; - client.startTurn = async ({ input, onStarted }) => { + client.startTurn = async ({ input, onStarted, onAgentMessage }) => { turn += 1; onStarted?.(`turn-${turn}`); const prompt = input.find((item) => item.type === 'text').text; const artifactPath = JSON.parse(prompt.match(/Return exactly one file whose path is ("[^"]+")/)[1]); - return { message: JSON.stringify({ files: [{ path: artifactPath, content: turn === 1 ? first : final }] }) }; + const message = JSON.stringify({ files: [{ path: artifactPath, content: turn === 1 ? first : final }] }); + await onAgentMessage?.(message); + return { message }; }; const replies = []; const checkpoints = [];