Classify failed accepts as errors, and fix parallel lane race/all misuse

Two of the three new findings, plus the bug that chasing them exposed in my own
earlier fix. The third is mitigated rather than broken; details below.

Failed accepts reported success:
live/completion.mjs only classifies a result as `error` when it carries
`mode: 'error'`. Everything else unhandled falls through to `agent_done` with an
ok ack, which is deliberate for the documented fallback paths (two tests pin it)
but wrong for a real failure. So `accept_receipt_conflict` reported success, and
reference/live.md's `handled: false` without `mode` bullet told the agent to
"read file, find markers, edit" — hand-applying a second accept on top of the one
the receipt already recorded.

The same hole swallowed `source_locked`, which is mine: the earlier commit made
lock contention return clean JSON so the agent could retry, but the classifier
turned that failure into agent_done/ok, so the accept was dequeued and silently
lost. Mark genuine failures with `mode: 'error'` through one `operationFailure`
helper, and give live.md a `mode: "error"` bullet with per-error guidance: retry
the same command on `source_locked`, never hand-edit, and on a receipt conflict
report what the session actually resolved to. The deliberate fallback and
markers-not-found handoffs stay untouched.

parallel-compact lane orchestration:
`Promise.race` settles on the first *settlement*, so one lane failing fast
rejected the whole first-variant step while two lanes were still on their way to
succeeding. `Promise.any` now takes the first success and only a total wipeout is
fatal, reporting every lane's reason. The tail step's `Promise.all` surfaced a
raw lane error non-deterministically; `Promise.allSettled` now reports how many
lanes failed and why. Added a `requestImpl` seam so lane orchestration is
testable without a provider key.

Not a defect: the browser releasing Accept before the source write. That is the
intended optimistic design, and it is safe because poll-lanes ranks accept at
priority 0 against generate at 2, so a queued accept is always leased before a
generate the user queues afterwards, even if the generate arrived first. Its
source write lands inside the poll script before the next generate preflights.
That invariant is load-bearing and had no tests at all; poll-lanes.mjs now has a
suite covering it plus lease and type filtering.

Prepared with AI assistance under maintainer direction.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Paul Bakaus
2026-07-17 16:10:57 -07:00
co-authored by Claude
parent 529184bbe4
commit 6cbb7ce8d1
6 changed files with 229 additions and 12 deletions
+34 -6
View File
@@ -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');
+1
View File
@@ -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',
+4
View File
@@ -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)
+26 -6
View File
@@ -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 || '';
+99
View File
@@ -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');
});
});
+65
View File
@@ -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);
});
});