diff --git a/skill/scripts/live-browser.js b/skill/scripts/live-browser.js index 0103ed94e..7f8c7ec00 100644 --- a/skill/scripts/live-browser.js +++ b/skill/scripts/live-browser.js @@ -2539,6 +2539,10 @@ if (generationPhase === 'scaffolding') return 'Finding the source...'; if (generationPhase === 'source_ready') return 'Source ready. Generating...'; if (generationPhase === 'scaffold_fallback') return 'Agent is locating the source...'; + if (generationPhase === 'first_variant_generating') return 'Designing the first variant...'; + if (generationPhase === 'first_variant_validating') return 'Checking the first variant...'; + if (generationPhase === 'remaining_variants_generating') return 'Exploring two more directions...'; + if (generationPhase === 'remaining_variants_validating') return 'Checking the remaining variants...'; return 'Generating ' + expectedVariants + ' variants...'; } diff --git a/skill/scripts/live-server.mjs b/skill/scripts/live-server.mjs index dc93065eb..9b3f0cbb0 100644 --- a/skill/scripts/live-server.mjs +++ b/skill/scripts/live-server.mjs @@ -805,6 +805,15 @@ function createRequestHandler({ detectScript, liveScriptParts }) { res.end(JSON.stringify({ error })); return; } + if (msg.type === 'agent_phase') { + recordAgentPhase(msg.id, msg.phase, { + ...(Number.isFinite(msg.durationMs) ? { durationMs: msg.durationMs } : {}), + owner: typeof msg.owner === 'string' ? msg.owner : undefined, + }); + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ ok: true })); + return; + } if (state.sessionStore && msg.id) { try { state.sessionStore.appendEvent(msg); diff --git a/skill/scripts/live/codex-worker-supervisor.mjs b/skill/scripts/live/codex-worker-supervisor.mjs index 04aff5ad9..c4506fc59 100644 --- a/skill/scripts/live/codex-worker-supervisor.mjs +++ b/skill/scripts/live/codex-worker-supervisor.mjs @@ -46,6 +46,7 @@ export class CodexLiveWorkerSupervisor { handleAccept = augmentEventWithAcceptHandling, reply = postReply, publishCheckpoint = postVariantCheckpoint, + publishPhase = postAgentPhase, postCleanup = postCarbonizeCleanup, log = () => {}, }) { @@ -60,6 +61,7 @@ export class CodexLiveWorkerSupervisor { this.handleAccept = handleAccept; this.reply = reply; this.publishCheckpoint = publishCheckpoint; + this.publishPhase = publishPhase; this.postCleanup = postCleanup; this.log = log; this.running = false; @@ -187,6 +189,11 @@ export class CodexLiveWorkerSupervisor { async runGenerationPhase(event, phase, arrivedVariants) { if (this.isCanceled(event.id)) return; + const phaseStartedAt = Date.now(); + await this.publishPhase(this.base, this.token, { + eventId: event.id, + phase: phase === 'final' ? 'remaining_variants_generating' : 'first_variant_generating', + }); const prepared = prepareCodexWorkerPhase({ id: event.id, sourceFile: event.scaffold.file, @@ -215,6 +222,11 @@ export class CodexLiveWorkerSupervisor { outputSchema: CODEX_WORKER_OUTPUT_SCHEMA, }); 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, @@ -388,6 +400,26 @@ export async function postVariantCheckpoint(base, token, { if (!response.ok) throw supervisorError(`checkpoint_${response.status}`); } +export async function postAgentPhase(base, token, { + eventId, + phase, + durationMs, +}) { + const response = await fetch(`${base}/events`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + token, + type: 'agent_phase', + id: eventId, + phase, + owner: CODEX_WORKER_OWNER, + ...(Number.isFinite(durationMs) ? { durationMs } : {}), + }), + }); + if (!response.ok) throw supervisorError(`agent_phase_${response.status}`); +} + export async function postCarbonizeCleanup(base, token, { sessionId, file, diff --git a/skill/scripts/live/event-validation.mjs b/skill/scripts/live/event-validation.mjs index b2f0ee18b..122c29836 100644 --- a/skill/scripts/live/event-validation.mjs +++ b/skill/scripts/live/event-validation.mjs @@ -118,6 +118,15 @@ export function validateEvent(msg) { return 'checkpoint: paramValues must be an object'; } return null; + case 'agent_phase': + if (!isValidId(msg.id)) return 'agent_phase: missing or malformed id'; + if (typeof msg.phase !== 'string' || !/^[a-z][a-z0-9_]{1,63}$/.test(msg.phase)) { + return 'agent_phase: missing or malformed phase'; + } + if (msg.durationMs !== undefined && (!Number.isFinite(msg.durationMs) || msg.durationMs < 0)) { + return 'agent_phase: durationMs must be a non-negative number'; + } + return null; case 'exit': return null; case 'prefetch': diff --git a/tests/live-codex-worker-supervisor.test.mjs b/tests/live-codex-worker-supervisor.test.mjs index 77193705a..bbec2eaa0 100644 --- a/tests/live-codex-worker-supervisor.test.mjs +++ b/tests/live-codex-worker-supervisor.test.mjs @@ -244,6 +244,7 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { }; const replies = []; const checkpoints = []; + const phases = []; const supervisor = new CodexLiveWorkerSupervisor({ cwd, base: 'http://localhost:1', @@ -254,6 +255,7 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { scriptsDir: path.join(cwd, 'skill/scripts'), reply: async (_base, _token, value) => { replies.push(value); }, publishCheckpoint: async (_base, _token, value) => { checkpoints.push(value); }, + publishPhase: async (_base, _token, value) => { phases.push(value); }, }); supervisor.thread = { id: 'live-worker-thread' }; supervisor.model = client.models[0]; @@ -268,6 +270,12 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => { assert.equal(checkpoints.length, 2); assert.deepEqual(checkpoints.map((item) => item.arrivedVariants), [1, 3]); + assert.deepEqual(phases.map((item) => item.phase), [ + 'first_variant_generating', + 'first_variant_validating', + 'remaining_variants_generating', + 'remaining_variants_validating', + ]); assert.equal(replies.at(-1).type, 'done'); assert.equal((readFileSync(path.join(cwd, 'src/App.jsx'), 'utf-8').match(/data-impeccable-variant="1"/g) || []).length, 2, 'selector and variant 1 remain once each'); const snapshot = createLiveSessionStore({ cwd, sessionId }).getSnapshot(sessionId, { includeCompleted: true }); diff --git a/tests/live-event-validation.test.mjs b/tests/live-event-validation.test.mjs index 8282f5137..d2e110cf3 100644 --- a/tests/live-event-validation.test.mjs +++ b/tests/live-event-validation.test.mjs @@ -97,3 +97,16 @@ describe('validateEvent — replace generate (regression)', () => { ); }); }); + +describe('validateEvent — worker progress', () => { + it('accepts bounded agent phases and rejects malformed telemetry', () => { + assert.equal(validateEvent({ + type: 'agent_phase', + id: VALID_ID, + phase: 'first_variant_generating', + durationMs: 123, + }), null); + assert.match(validateEvent({ type: 'agent_phase', id: VALID_ID, phase: 'Not valid' }), /phase/); + assert.match(validateEvent({ type: 'agent_phase', id: VALID_ID, phase: 'valid', durationMs: -1 }), /durationMs/); + }); +}); diff --git a/tests/live-server.test.mjs b/tests/live-server.test.mjs index c8e8ce9e8..76b7d2262 100644 --- a/tests/live-server.test.mjs +++ b/tests/live-server.test.mjs @@ -2345,6 +2345,37 @@ colors: {} ); }); + it('journals and streams dedicated worker progress without leasing it as work', async () => { + await drainPolls(server); + const controller = new AbortController(); + const sseRes = await fetch( + `http://localhost:${server.port}/events?token=${server.token}`, + { signal: controller.signal }, + ); + const reader = sseRes.body.getReader(); + await reader.read(); + const progress = await fetch(`http://localhost:${server.port}/events`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + token: server.token, + type: 'agent_phase', + id: 'a1b2c3e1', + phase: 'first_variant_generating', + owner: 'impeccable-live-codex-worker-v1', + }), + }); + assert.equal(progress.status, 200); + const message = new TextDecoder().decode((await reader.read()).value); + controller.abort(); + assert.match(message, /"type":"agent_phase"/); + assert.match(message, /"phase":"first_variant_generating"/); + const polled = await fetch(`http://localhost:${server.port}/poll?token=${server.token}&timeout=50`).then(r => r.json()); + assert.equal(polled.type, 'timeout'); + const snapshot = JSON.parse(readFileSync(join(getLiveSessionsDir(server.cwd), 'a1b2c3e1.snapshot.json'), 'utf-8')); + assert.ok(snapshot.generationTimings.first_variant_generating?.at); + }); + it('streams Svelte component checkpoints as progressive preview updates', async () => { const controller = new AbortController(); const sseRes = await fetch(