fix(harness): check processing and close usage on every length continuation path (#1852)
Review round fixes for the length continuation in Part A. Before each continuation the runner now reads processing.status as well as state. A halted, invalid or unreadable processing state starts no further attempt; the message is left unread with no reply and no failure record, as for a halt during any turn. A failed state or processing read keeps the empty length classification, so the sender gets no RESULT. The chief notice and the trail event name how many attempts hit the limit. usage.end for the message id now runs in a finally block, so it closes once even when an attempt throws. The library citation retry no longer starts continuations. They would run outside the message's usage bracket, and the retry's own bracket is refused (#1853). An empty length retry fails visibly instead, with no fixed RESULT. Tests cover a halt between attempts, a failed continuation check on the plain and library paths, a throwing attempt, a library continuation that succeeds, and the retry path.
This commit is contained in:
jason.woltje committed
1 parent
3c689efa31
commit
3e03e123fe
3 files changed
+135
-25
No files matched your search
@@ -613,13 +613,20 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
|
||||
return true;
|
||||
};
|
||||
const continuationRequest = "Your last turn hit the output limit while thinking, before any answer or tool call. Now answer or call a tool, and keep your reasoning short.";
|
||||
// State checks the broker's stop switch before another process can start.
|
||||
// Before another process starts: the state read checks the session, and
|
||||
// processing.status the instance halt, which state and usage.end ignore.
|
||||
// A halted, invalid or unreadable processing state starts nothing. A failed
|
||||
// read keeps the empty-length classification, so no reply follows it.
|
||||
// The surrounding usage bracket captures every attempt in the transcript.
|
||||
const continueLength = async (initial, counter) => {
|
||||
let turn = initial;
|
||||
while (turn.emptyLength && counter.count < 2 && !stopping) {
|
||||
try { await client.call("state"); }
|
||||
catch (error) { return { ok: false, brokerError: error, reason: "could not read the current state" }; }
|
||||
let processing;
|
||||
try {
|
||||
await client.call("state");
|
||||
processing = await client.call("processing.status");
|
||||
} catch (error) { return { ...turn, brokerError: error }; }
|
||||
if (processing?.halted !== false || processing.invalid) return { ...turn, halted: true };
|
||||
if (stopping) break;
|
||||
++counter.count;
|
||||
++n;
|
||||
@@ -628,8 +635,10 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
|
||||
}
|
||||
return turn;
|
||||
};
|
||||
const recordLengthFailure = async m => {
|
||||
const detail = "turn failed: output limit reached while thinking before any answer or tool call after two continuations";
|
||||
// `on` names the attempts that hit the limit, for the trail and the notice.
|
||||
const attemptsText = count => count === 0 ? "the turn" : `the turn and ${count} continuation${count === 1 ? "" : "s"}`;
|
||||
const recordLengthFailure = async (m, on) => {
|
||||
const detail = `turn failed: output limit reached while thinking before any answer or tool call on ${on}`;
|
||||
log(`runner: message ${m.id} ${detail}`);
|
||||
// The existing routine event is visible to the chief via trail and to the
|
||||
// Console on this message's trail; it does not send an automatic reply.
|
||||
@@ -637,7 +646,7 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
|
||||
if (m.from_role === "chief") { log(`runner: chief notice for ${m.id} skipped: chief is the sender`); return; }
|
||||
try {
|
||||
await client.call("message.send", { to: "chief", class: "INFO", in_reply_to: m.id,
|
||||
body: `${LENGTH_FAILURE_NOTICE_PREFIX}Turn for message ${m.id} from ${m.from_role} failed: the output limit was reached while thinking, before any answer or tool call, on the turn and 2 continuations. No answer was sent.` });
|
||||
body: `${LENGTH_FAILURE_NOTICE_PREFIX}Turn for message ${m.id} from ${m.from_role} failed: the output limit was reached while thinking, before any answer or tool call, on ${on}. No answer was sent.` });
|
||||
} catch (e) { log(`runner: chief notice for ${m.id} failed: ${e.code ?? e.message}`); }
|
||||
};
|
||||
for (const e of session.landing ?? []) {
|
||||
@@ -697,21 +706,31 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
|
||||
try { await client.call("usage.begin", { turn: m.id }); }
|
||||
catch (e) { log(`runner: usage baseline unavailable: ${e.code ?? "failed"}`); }
|
||||
}
|
||||
let turn = unread
|
||||
? { ok: false, reason: unread }
|
||||
: await runTurn({ session, request: `${block}\n${turnRequest(m)}`, n, env: turnEnv, onChild: (c) => (child = c) });
|
||||
unslop.read();
|
||||
let turn;
|
||||
const continuationCount = { count: 0 };
|
||||
if (!turn.markerFailed && !stopping) turn = await continueLength(turn, continuationCount);
|
||||
// One transcript delta covers the original attempt and all continuations,
|
||||
// even if the runner is stopped or a state check fails.
|
||||
try { await client.call("usage.end", { turn: m.id }); }
|
||||
catch (e) { log(`runner: usage outcome unavailable: ${e.code ?? "failed"}`); }
|
||||
// even if the runner is stopped, a check fails or an attempt throws.
|
||||
try {
|
||||
turn = unread
|
||||
? { ok: false, reason: unread }
|
||||
: await runTurn({ session, request: `${block}\n${turnRequest(m)}`, n, env: turnEnv, onChild: (c) => (child = c) });
|
||||
unslop.read();
|
||||
if (!turn.markerFailed && !stopping) turn = await continueLength(turn, continuationCount);
|
||||
} finally {
|
||||
try { await client.call("usage.end", { turn: m.id }); }
|
||||
catch (e) { log(`runner: usage outcome unavailable: ${e.code ?? "failed"}`); }
|
||||
}
|
||||
if (failedMarker(turn)) break loop;
|
||||
if (stopping) break loop;
|
||||
if (turn.brokerError && ended(turn.brokerError)) break loop;
|
||||
if (turn.halted) {
|
||||
// As for a halt during any turn: no reply, no failure record, message left unread.
|
||||
log(`runner: message ${m.id} left unread: processing is halted or its state is invalid`);
|
||||
continue;
|
||||
}
|
||||
if (turn.emptyLength) {
|
||||
try { await recordLengthFailure(m); await client.call("message.read", { id: m.id }); }
|
||||
if (turn.brokerError) log(`runner: continuation check for ${m.id} failed: ${turn.brokerError.code ?? turn.brokerError.message}`);
|
||||
try { await recordLengthFailure(m, attemptsText(continuationCount.count)); await client.call("message.read", { id: m.id }); }
|
||||
catch (e) { log(`runner: failure record for ${m.id} failed: ${e.code ?? e.message}`); if (ended(e, false)) break loop; }
|
||||
continue;
|
||||
}
|
||||
@@ -764,12 +783,11 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
|
||||
catch (e) { log(`runner: usage outcome unavailable: ${e.code ?? "failed"}`); }
|
||||
if (failedMarker(retry)) break loop;
|
||||
if (stopping) break loop;
|
||||
retry = await continueLength(retry, continuationCount);
|
||||
if (failedMarker(retry)) break loop;
|
||||
if (stopping) break loop;
|
||||
if (retry.brokerError && ended(retry.brokerError)) break loop;
|
||||
// No continuation here: it would run outside the m.id usage bracket,
|
||||
// and the :retry bracket is refused (#1853). An empty length retry
|
||||
// fails visibly like an exhausted turn.
|
||||
if (retry.emptyLength) {
|
||||
await recordLengthFailure(m);
|
||||
await recordLengthFailure(m, "the citation retry");
|
||||
await client.call("message.read", { id: m.id });
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -169,13 +169,39 @@ for (const library of [false, true]) {
|
||||
});
|
||||
}
|
||||
|
||||
test('a library guard retry that exhausts empty length continuations sends no fixed RESULT', async t => {
|
||||
// Sol review r1 (#1852): a continuation after the retry would run outside every usage bracket,
|
||||
// so an empty length retry fails visibly at once.
|
||||
test('an empty length library guard retry gets no continuation and sends no fixed RESULT', async t => {
|
||||
const f = await setup(t, () => [
|
||||
{text:'uncited'}, ...Array(3).fill({marker:{stopReason:'length',hasText:false,hasToolCall:false}}),
|
||||
]);
|
||||
const usage = []; const request = f.broker.request.bind(f.broker);
|
||||
f.broker.request = (cap, req) => { if (req.verb.startsWith('usage.')) usage.push({verb:req.verb,turn:req.args.turn,attempts:existsSync(join(f.runDir,'fixture-count')) ? f.count() : 0}); return request(cap, req); };
|
||||
const sent = await f.call(f.human,'message.send',{to:'coder',body:'Fixture question.',class:'REQUEST'});
|
||||
await until(() => f.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'",sent.id));
|
||||
assert.equal(f.count(),4);
|
||||
assert.equal(f.count(),2);
|
||||
assert.deepEqual(usage.map(u => [u.verb, u.turn, u.attempts]), [
|
||||
['usage.begin', sent.id, 0], ['usage.end', sent.id, 1], ['usage.begin', `${sent.id}:retry`, 1], ['usage.end', `${sent.id}:retry`, 2]]);
|
||||
assert.equal(f.store.get("SELECT count(*) n FROM messages WHERE from_role='coder' AND to_role='human'").n,0);
|
||||
const events = f.store.all("SELECT body FROM events WHERE subject=? AND kind='action.allowed'",sent.id);
|
||||
assert.equal(events.length,1); assert.match(events[0].body,/on the citation retry/);
|
||||
});
|
||||
|
||||
test('a failed library continuation check sends no fixed RESULT', async t => {
|
||||
const f = await setup(t, () => [{marker:{stopReason:'length',hasText:false,hasToolCall:false}}, {text:'must not run'}]);
|
||||
const request = f.broker.request.bind(f.broker); let reads = 0;
|
||||
f.broker.request = (cap, req) => { if (req.verb === 'state' && ++reads === 2) throw Object.assign(new Error('state-unavailable'), {code:'state-unavailable'}); return request(cap, req); };
|
||||
const sent = await f.call(f.human,'message.send',{to:'coder',body:'Fixture question.',class:'REQUEST'});
|
||||
await until(() => f.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'",sent.id));
|
||||
assert.equal(f.count(),1);
|
||||
assert.equal(f.store.get("SELECT count(*) n FROM messages WHERE from_role='coder' AND to_role='human'").n,0);
|
||||
assert.equal(f.store.get("SELECT count(*) n FROM events WHERE subject=? AND kind='action.allowed'",sent.id).n,1);
|
||||
});
|
||||
|
||||
test('a library continuation that succeeds sends the cited RESULT once', async t => {
|
||||
const f = await setup(t, commit => [{marker:{stopReason:'length',hasText:false,hasToolCall:false}}, {text:citedBody([fixtureCite(commit)])}]);
|
||||
const reply = await f.ask();
|
||||
assert.equal(reply.class,'RESULT'); assert.equal(reply.body,citedBody([fixtureCite(f.commit)]));
|
||||
assert.equal(f.count(),2);
|
||||
assert.equal(f.store.get("SELECT count(*) n FROM messages WHERE from_role='coder'").n,1);
|
||||
});
|
||||
@@ -102,13 +102,13 @@ test("stateBlock names the launch state with its time and event, and the open ta
|
||||
});
|
||||
|
||||
// `roles` replaces demo's role entries by name (row 156 round 2 gives the coder git.merge.protected).
|
||||
async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null, limits = null, roles = null, tester = false } = {}) {
|
||||
async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null, limits = null, roles = null, tester = false, processing = null } = {}) {
|
||||
const dir = scratch(t);
|
||||
const store = new Store(dir);
|
||||
let demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo;
|
||||
if (tester) demo = { ...demo, vars: { ...demo.vars, "bus.tester": true } };
|
||||
if (roles) demo = { ...demo, roles: { ...demo.roles, ...roles } };
|
||||
const broker = new Broker({ store, businesses: { demo }, organization, usage, limits });
|
||||
const broker = new Broker({ store, businesses: { demo }, organization, usage, limits, ...(processing ? { processing } : {}) });
|
||||
const brokerSocket = join(dir, "bus.sock");
|
||||
const server = await serve({ broker, path: brokerSocket });
|
||||
t.after(async () => {
|
||||
@@ -1081,7 +1081,7 @@ test('empty length continues in the same session, records usage and relays the a
|
||||
const sent = ctx.call(ctx.pm, 'message.send', {to:'coder',body:'work',class:'REQUEST'});
|
||||
await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
|
||||
const rows = ctx.store.all("SELECT * FROM messages WHERE from_role='coder' AND to_role='pm'");
|
||||
assert.equal(rows.length,1); assert.equal(rows[0].class,'RESULT'); assert.equal(rows[0].body,'recovered');
|
||||
assert.equal(rows.length,1); assert.equal(rows[0].class,'RESULT'); assert.equal(rows[0].body,'recovered'); assert.equal(rows[0].in_reply_to,sent.id);
|
||||
assert.equal(Number(readFileSync(join(ctx.runDir,'fixture-count'),'utf8')),2);
|
||||
const requests = readFileSync(join(ctx.runDir,'fixture-requests.jsonl'),'utf8').trim().split('\n').map(JSON.parse);
|
||||
assert.match(requests[1], /output limit.*thinking/i); assert.match(requests[1], /answer or call a tool/i);
|
||||
@@ -1161,3 +1161,69 @@ test('failure notice recognition excludes bus notices and non-agent senders', as
|
||||
assert.equal(isLengthFailureNotice({...notice,class:'REQUEST'}),false);
|
||||
assert.equal(isLengthFailureNotice({...notice,body:'ordinary information'}),false);
|
||||
});
|
||||
|
||||
// Sol review r1 (#1852): the continuation checks, every attempt inside the m.id usage bracket.
|
||||
const lengthAdapter = (ctx, steps) => {
|
||||
const fixture = new URL('./fixtures/citation-adapter.mjs', import.meta.url).pathname;
|
||||
writeFileSync(join(ctx.dir, 'adapter.sh'), `exec '${process.execPath}' '${fixture}'\n`);
|
||||
writeFileSync(join(ctx.runDir, 'fixture-steps.json'), JSON.stringify(steps));
|
||||
};
|
||||
const emptyLength = {marker:{stopReason:'length',hasText:false,hasToolCall:false}};
|
||||
const attemptsOf = ctx => existsSync(join(ctx.runDir, 'fixture-count')) ? Number(readFileSync(join(ctx.runDir, 'fixture-count'), 'utf8')) : 0;
|
||||
// The second state read is the first continuation's check; the first is the message's own read.
|
||||
const onSecondState = (ctx, act) => {
|
||||
let reads = 0;
|
||||
const request = ctx.broker.request.bind(ctx.broker);
|
||||
ctx.broker.request = (cap, req) => { if (req.verb === 'state' && ++reads === 2) act(); return request(cap, req); };
|
||||
};
|
||||
|
||||
test('a processing halt after an empty length turn starts no continuation, closes usage and leaves the message unread', async t => {
|
||||
let halted = false; const calls = []; let ctx;
|
||||
const processing = { read: () => ({ version: 1, halted, generation: '00000000-0000-4000-8000-000000000000' }) };
|
||||
ctx = await setup(t, { processing, usage: { collect: async (_s, phase, turn) => { calls.push({ phase, turn, attempts: attemptsOf(ctx) }); return { recorded: true }; } } });
|
||||
lengthAdapter(ctx, [emptyLength, { text: 'must not run' }]);
|
||||
onSecondState(ctx, () => { halted = true; });
|
||||
const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
|
||||
await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
|
||||
const sent = ctx.call(ctx.pm, 'message.send', { to: 'coder', body: 'work', class: 'REQUEST' });
|
||||
await until(() => /left unread: processing is halted/.test(r.stderr()));
|
||||
r.child.kill('SIGTERM'); await r.exited;
|
||||
assert.equal(attemptsOf(ctx), 1);
|
||||
assert.deepEqual(calls, [{ phase: 'begin', turn: sent.id, attempts: 0 }, { phase: 'end', turn: sent.id, attempts: 1 }]);
|
||||
assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='coder'").n, 0);
|
||||
assert.equal(ctx.store.get("SELECT count(*) n FROM deliveries WHERE message=? AND op='read'", sent.id).n, 0);
|
||||
assert.equal(ctx.store.get("SELECT count(*) n FROM events WHERE subject=? AND kind='action.allowed'", sent.id).n, 0);
|
||||
});
|
||||
|
||||
test('a failed continuation check after an empty length turn sends no RESULT and records the failure', async t => {
|
||||
const calls = []; let ctx;
|
||||
ctx = await setup(t, { usage: { collect: async (_s, phase, turn) => { calls.push({ phase, turn }); return { recorded: true }; } } });
|
||||
lengthAdapter(ctx, [emptyLength, { text: 'must not run' }]);
|
||||
onSecondState(ctx, () => { throw new BusError('state-unavailable'); });
|
||||
const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
|
||||
await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
|
||||
const sent = ctx.call(ctx.pm, 'message.send', { to: 'coder', body: 'work', class: 'REQUEST' });
|
||||
await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
|
||||
assert.equal(attemptsOf(ctx), 1);
|
||||
assert.deepEqual(calls, [{ phase: 'begin', turn: sent.id }, { phase: 'end', turn: sent.id }]);
|
||||
assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='coder' AND to_role='pm'").n, 0);
|
||||
const notice = ctx.store.all("SELECT * FROM messages WHERE from_role='coder' AND to_role='chief'");
|
||||
assert.equal(notice.length, 1); assert.match(notice[0].body, /on the turn\. No answer was sent\.$/);
|
||||
assert.equal(ctx.store.get("SELECT count(*) n FROM events WHERE subject=? AND kind='action.allowed'", sent.id).n, 1);
|
||||
assert.match(r.stderr(), /continuation check for .* failed: state-unavailable/);
|
||||
});
|
||||
|
||||
test('an attempt that throws still closes the message usage bracket once', async t => {
|
||||
const calls = []; let ctx;
|
||||
ctx = await setup(t, { usage: { collect: async (_s, phase, turn) => { calls.push({ phase, turn }); return { recorded: true }; } } });
|
||||
lengthAdapter(ctx, [emptyLength, { text: 'must not run' }]);
|
||||
// runTurn opens turns/0002.stderr with "wx": an existing file throws before the adapter starts.
|
||||
onSecondState(ctx, () => writeFileSync(join(ctx.runDir, 'turns', '0002.stderr'), 'preexisting'));
|
||||
const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
|
||||
await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
|
||||
const sent = ctx.call(ctx.pm, 'message.send', { to: 'coder', body: 'work', class: 'REQUEST' });
|
||||
const exit = await r.exited;
|
||||
assert.notEqual(exit.code, 0);
|
||||
assert.deepEqual(calls, [{ phase: 'begin', turn: sent.id }, { phase: 'end', turn: sent.id }]);
|
||||
assert.equal(attemptsOf(ctx), 1);
|
||||
});
|
||||
Reference in new issue
Block a user