diff --git a/scripts/lib/live-provider-benchmark.mjs b/scripts/lib/live-provider-benchmark.mjs index ac74b352d..d6dcab112 100644 --- a/scripts/lib/live-provider-benchmark.mjs +++ b/scripts/lib/live-provider-benchmark.mjs @@ -163,17 +163,22 @@ export function resolveProviderSelection(providerNames, modelOverrides = {}) { }); } -export function createProviderLiveAgent({ provider, model, strategy, liveSpec, onRecord = () => {} }) { +/** + * `requestImpl` overrides the per-lane model call. It exists so the lane + * orchestration (which lane wins, what happens when one fails) is testable + * without a provider key or a network round trip; production passes nothing. + */ +export function createProviderLiveAgent({ provider, model, strategy, liveSpec, onRecord = () => {}, requestImpl = null }) { const strategyConfig = STRATEGIES[strategy]; if (!strategyConfig) throw new Error(`unknown strategy ${JSON.stringify(strategy)}`); - const languageModel = providerModel(provider, model); + const languageModel = requestImpl ? null : providerModel(provider, model); const system = strategyConfig.promptMode === 'full-live-context' ? `${COMPACT_CONTRACT}\n\nFULL LIVE CONTEXT:\n${liveSpec}` : COMPACT_CONTRACT; const pendingParallel = new Map(); const pendingFirst = new Map(); - const request = async ({ event, phase, lane = null, firstVariant = null }) => { + const request = requestImpl || (async ({ event, phase, lane = null, firstVariant = null }) => { const startedAt = performance.now(); const expectedCount = Number(event.count); const payload = benchmarkPayload(event, { phase, lane, firstVariant }); @@ -235,7 +240,7 @@ export function createProviderLiveAgent({ provider, model, strategy, liveSpec, o } } throw lastError; - }; + }); if (strategy === 'atomic-full') { return { @@ -253,15 +258,38 @@ export function createProviderLiveAgent({ provider, model, strategy, liveSpec, o const laneEvent = { ...event, count: 1 }; return request({ event: laneEvent, phase: 'parallel-lane', lane }).then((output) => ({ lane, output })); }); - const first = await Promise.race(calls); + // Promise.any, not race: race settles on the first *settlement*, so one + // lane failing fast rejected the whole first-variant step while a slower + // lane was still on its way to succeeding. Only a total wipeout is fatal. + let first; + try { + first = await Promise.any(calls); + } catch (error) { + const reasons = (error?.errors || [error]).map((e) => e?.message || String(e)); + throw new Error(`every parallel lane failed for ${event.id}: ${reasons.join('; ')}`); + } pendingParallel.set(event.id, { calls, first }); return first.output; }, async generateRemainingVariants(event) { const pending = pendingParallel.get(event.id); if (!pending) throw new Error(`parallel generation state missing for ${event.id}`); - const settled = await Promise.all(pending.calls); + // allSettled, not all: a lane that rejects after another already won the + // race must not throw its raw error from here. Collect every outcome and + // report the failures together, so the result does not depend on which + // lane happened to settle first. + const outcomes = await Promise.allSettled(pending.calls); pendingParallel.delete(event.id); + const failures = outcomes + .filter((outcome) => outcome.status === 'rejected') + .map((outcome) => outcome.reason?.message || String(outcome.reason)); + if (failures.length > 0) { + throw new Error( + `${failures.length} of ${pending.calls.length} parallel lanes failed for ${event.id}, ` + + `so the ${event.count}-variant set cannot be assembled: ${failures.join('; ')}`, + ); + } + const settled = outcomes.map((outcome) => outcome.value); const ordered = [pending.first, ...settled.filter((item) => item !== pending.first)]; const variants = ordered.map((item) => item.output.variants[0]); const scopedCss = ordered.map((item, index) => remapSingleVariantCss(item.output.scopedCss, index + 1)).join('\n'); diff --git a/scripts/test-suites.mjs b/scripts/test-suites.mjs index 1733af752..9106deb80 100644 --- a/scripts/test-suites.mjs +++ b/scripts/test-suites.mjs @@ -143,6 +143,7 @@ export const SUITES = { 'tests/live-insert-ui.test.mjs', 'tests/live-manual-edits-buffer.test.mjs', 'tests/live-poll.test.mjs', + 'tests/live-poll-lanes.test.mjs', 'tests/live-poll-stream.test.mjs', 'tests/live-provider-benchmark.test.mjs', 'tests/live-recovery-commands.test.mjs', diff --git a/skill/reference/live.md b/skill/reference/live.md index 560e2e685..4a7970703 100644 --- a/skill/reference/live.md +++ b/skill/reference/live.md @@ -508,6 +508,10 @@ Event: `{id, variantId, _acceptResult, _completionAck}`. The poll script already - `_acceptResult.handled: true` and `carbonize: false`: nothing to do. Poll again. - `_acceptResult.handled: true` and `carbonize: true`: post-accept cleanup is required, but it must not stall Codex's control lane. See "Required after accept (carbonize)" below. The `event._acceptResult.todo` field, `_completionAck.requiresComplete`, and stderr banner all point at this required follow-up; none are decorative. - `_acceptResult.handled: false, mode: "fallback"`: the session lived in a generated file and the script refused to persist there. You've already written the accepted variant into true source during Handle fallback Step 3; just clean up the temporary wrapper in the served file if any, and poll again. +- `_acceptResult.handled: false, mode: "error"`: the operation genuinely failed. **Do not hand-edit the file**; the source was not touched and editing it yourself would either double-apply or race whoever holds it. + - `error: "source_locked"`: a generation publish holds the file. Run the same `live-accept.mjs` command again; it is idempotent and will succeed once the publisher releases. Do not poll past it. + - `error: "accept_receipt_conflict"`: this session already resolved as `priorOperation` (on `priorVariantId` for an accept), so the request contradicts durable truth. Do not edit. Run `live-status.mjs` and tell the user what the session actually resolved to. + - anything else: report the error briefly and run `live-status.mjs` before continuing. - `_acceptResult.handled: false` without `mode`: manual cleanup: read file, find markers, edit. ### Required after accept (carbonize) diff --git a/skill/scripts/live-accept.mjs b/skill/scripts/live-accept.mjs index 08883c90d..a3d59f253 100644 --- a/skill/scripts/live-accept.mjs +++ b/skill/scripts/live-accept.mjs @@ -41,6 +41,20 @@ const ACCEPT_LOCK_WAIT_MS = 1_000; // value arriving over HTTP. const VARIANT_NUM_PATTERN = /^[0-9]{1,3}$/; +/** + * A thrown accept/discard is a real failure, not a manual handoff. + * + * live/completion.mjs only classifies a result as `error` when it carries + * `mode: 'error'`; anything else unhandled falls through to `agent_done` with a + * successful ack, and reference/live.md then tells the agent to finish the edit + * by hand. That is right for the documented fallback paths and wrong here: a + * `source_locked` contention needs a retry (hand-editing races the publisher + * holding the lock), and a crash needs surfacing, not a hand-applied guess. + */ +function operationFailure(err, extra = {}) { + return { handled: false, mode: 'error', error: err.message, ...extra }; +} + // --------------------------------------------------------------------------- // CLI // --------------------------------------------------------------------------- @@ -100,7 +114,13 @@ Output (JSON): console.log(JSON.stringify(sameOperation ? { ...priorReceipt.result, handled: true, alreadyApplied: true } : { + // mode: 'error' is what marks this a real failure rather than a manual + // handoff. Without it, live/completion.mjs classifies the reply as + // agent_done and reference/live.md tells the agent to "read file, find + // markers, edit" by hand — which would apply a second, conflicting + // accept on top of the one the receipt already recorded. handled: false, + mode: 'error', error: 'accept_receipt_conflict', priorOperation: priorReceipt.operation, priorVariantId: priorReceipt.variantId ?? null, @@ -154,7 +174,7 @@ Output (JSON): { waitMs: ACCEPT_LOCK_WAIT_MS }, ); } catch (err) { - result = { handled: false, error: err.message }; + result = operationFailure(err); } emitResult({ ...result, @@ -175,7 +195,7 @@ Output (JSON): { waitMs: ACCEPT_LOCK_WAIT_MS }, ); } catch (err) { - result = { handled: false, error: err.message }; + result = operationFailure(err); } if (result.handled !== false) { removeSourceArtifactSession(id, process.cwd()); @@ -211,7 +231,7 @@ Output (JSON): { waitMs: ACCEPT_LOCK_WAIT_MS }, ); } catch (err) { - result = { handled: false, error: err.message }; + result = operationFailure(err); } emitResult({ ...result, @@ -260,7 +280,7 @@ Output (JSON): { waitMs: ACCEPT_LOCK_WAIT_MS }, ); } catch (err) { - result = { handled: false, error: err.message }; + result = operationFailure(err); } emitResult({ ...result, @@ -336,7 +356,7 @@ Output (JSON): try { result = handleDiscard(id, lines, targetFile); } catch (err) { - emitResult({ handled: false, file: relFile, error: err.message }); + emitResult(operationFailure(err, { file: relFile })); return; } emitResult({ handled: true, file: relFile, carbonize: false, ...result }); @@ -345,7 +365,7 @@ Output (JSON): try { result = handleAccept(id, variantNum, lines, targetFile, paramValues); } catch (err) { - emitResult({ handled: false, file: relFile, error: err.message }); + emitResult(operationFailure(err, { file: relFile })); return; } const acceptedOriginalText = result.acceptedOriginalText || ''; diff --git a/tests/live-poll-lanes.test.mjs b/tests/live-poll-lanes.test.mjs new file mode 100644 index 000000000..b775de131 --- /dev/null +++ b/tests/live-poll-lanes.test.mjs @@ -0,0 +1,99 @@ +/** + * Tests for live/poll-lanes.mjs — which pending event a poll gets next. + * Run with: node --test tests/live-poll-lanes.test.mjs + */ + +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; + +import { eventPriority, selectAvailablePendingEvent } from '../skill/scripts/live/poll-lanes.mjs'; + +const entry = (type, seq, leaseUntil = 0, id = type + seq) => ({ event: { id, type }, leaseUntil, seq }); + +describe('poll lane priority', () => { + it('puts terminal user actions ahead of generation', () => { + for (const type of ['accept', 'discard', 'exit']) { + assert.ok( + eventPriority({ type }) < eventPriority({ type: 'generate' }), + `${type} must outrank generate`, + ); + } + }); + + it('ranks unknown event types last rather than first', () => { + assert.ok(eventPriority({ type: 'something-new' }) > eventPriority({ type: 'generate' })); + assert.ok(eventPriority({}) > eventPriority({ type: 'generate' })); + }); + + // This is what makes the browser's optimistic Accept safe. The browser returns + // to PICKING as soon as /events durably journals the accept, before the source + // write happens, so the user can pick and hit Go while the accept is still + // queued. If that generate were leased first, its preflight would wrap source + // that still contains the previous session's variant markers. + it('delivers a queued accept before a generate the user queued afterwards', () => { + const selected = selectAvailablePendingEvent([ + entry('accept', 1), + entry('generate', 2), + ]); + assert.equal(selected.event.type, 'accept'); + }); + + it('delivers the accept first even when the generate was queued earlier', () => { + const selected = selectAvailablePendingEvent([ + entry('generate', 1), + entry('accept', 2), + ]); + assert.equal( + selected.event.type, + 'accept', + 'priority must beat arrival order, or a slow poller preflights against stale source', + ); + }); + + it('breaks ties within one lane by arrival order', () => { + const selected = selectAvailablePendingEvent([ + entry('generate', 7), + entry('generate', 3), + ]); + assert.equal(selected.seq, 3); + }); +}); + +describe('poll lane availability', () => { + it('skips an entry whose lease is still held', () => { + const now = 1_000_000; + const selected = selectAvailablePendingEvent([ + entry('accept', 1, now + 30_000), + entry('generate', 2), + ], { now }); + assert.equal(selected.event.type, 'generate', 'a leased accept must not be handed out twice'); + }); + + it('re-offers an entry once its lease has expired', () => { + const now = 1_000_000; + const selected = selectAvailablePendingEvent([entry('accept', 1, now - 1)], { now }); + assert.equal(selected.event.type, 'accept'); + }); + + it('returns null when everything is leased', () => { + const now = 1_000_000; + assert.equal(selectAvailablePendingEvent([entry('accept', 1, now + 5_000)], { now }), null); + }); + + it('returns null for an empty queue', () => { + assert.equal(selectAvailablePendingEvent([]), null); + }); + + it('restricts delivery to the requested types', () => { + const entries = [entry('accept', 1), entry('generate', 2)]; + assert.equal(selectAvailablePendingEvent(entries, { types: ['generate'] }).event.type, 'generate'); + assert.equal(selectAvailablePendingEvent(entries, { types: new Set(['generate']) }).event.type, 'generate'); + assert.equal(selectAvailablePendingEvent(entries, { types: ['steer'] }), null); + }); + + it('ignores an empty or absent type filter instead of starving the queue', () => { + const entries = [entry('generate', 1)]; + assert.equal(selectAvailablePendingEvent(entries, { types: null }).event.type, 'generate'); + assert.equal(selectAvailablePendingEvent(entries, {}).event.type, 'generate'); + }); +}); diff --git a/tests/live-provider-benchmark.test.mjs b/tests/live-provider-benchmark.test.mjs index 354b4eda5..6d70f7320 100644 --- a/tests/live-provider-benchmark.test.mjs +++ b/tests/live-provider-benchmark.test.mjs @@ -3,6 +3,7 @@ import { describe, it } from 'node:test'; import { STRATEGIES, + createProviderLiveAgent, assembleProgressiveOutput, applyRuntimeSourceScore, estimateCostUsd, @@ -113,3 +114,67 @@ describe('cross-provider Live benchmark', () => { assert.equal(summary.estimatedCostUsd, 0.3); }); }); + +describe('parallel-compact lane orchestration', () => { + const laneOutput = (lane) => ({ + scopedCss: `@scope ([data-impeccable-variant="1"]) { .${lane} { color: var(--color-ink); } }`, + variants: [{ innerHtml: VARIANT, params: [] }], + }); + + // requestImpl stands in for the model call so lane timing/failure is exact. + const agentWith = (behaviour) => createProviderLiveAgent({ + provider: 'anthropic', + model: 'test-model', + strategy: 'parallel-compact', + liveSpec: '', + requestImpl: ({ lane }) => behaviour(lane), + }); + + const after = (ms, value) => new Promise((resolve) => setTimeout(() => resolve(value), ms)); + const failAfter = (ms, message) => new Promise((_, reject) => setTimeout(() => reject(new Error(message)), ms)); + + it('returns a slower lane rather than rejecting on the lane that fails first', async () => { + // Promise.race settles on the first *settlement*, so the fast failure below + // used to reject the whole first-variant step while two lanes were still on + // their way to succeeding. + const agent = agentWith((lane) => ( + lane === 'hierarchy' ? failAfter(2, 'hierarchy lane failed') : after(30, laneOutput(lane)) + )); + const first = await agent.generateFirstVariant({ id: 'par1', count: 3 }); + assert.ok(first.variants?.[0], 'a successful lane must still produce variant 1'); + }); + + it('fails with every reason when all lanes fail', async () => { + const agent = agentWith((lane) => failAfter(2, `${lane} lane failed`)); + await assert.rejects( + () => agent.generateFirstVariant({ id: 'par2', count: 3 }), + (err) => { + assert.match(err.message, /every parallel lane failed for par2/); + for (const lane of ['hierarchy', 'layout', 'density']) { + assert.match(err.message, new RegExp(`${lane} lane failed`), `${lane} reason must be reported`); + } + return true; + }, + ); + }); + + it('reports a late lane failure as a lane failure, not a raw rejection', async () => { + // The tail step used to Promise.all the same lane promises, so a lane that + // rejected after another won the race surfaced its bare error from here. + const agent = agentWith((lane) => ( + lane === 'density' ? failAfter(40, 'density lane failed') : after(2, laneOutput(lane)) + )); + await agent.generateFirstVariant({ id: 'par3', count: 3 }); + await assert.rejects( + () => agent.generateRemainingVariants({ id: 'par3', count: 3 }), + /1 of 3 parallel lanes failed for par3.*density lane failed/s, + ); + }); + + it('assembles all three lanes when every lane succeeds', async () => { + const agent = agentWith((lane) => after(lane === 'layout' ? 1 : 10, laneOutput(lane))); + await agent.generateFirstVariant({ id: 'par4', count: 3 }); + const output = await agent.generateRemainingVariants({ id: 'par4', count: 3 }); + assert.equal(output.variants.length, 3); + }); +});