mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-21 21:31:12 +02:00
fix(x): hold the per-turn lock until straggler sync-tool appends land
A sync-tool batch runs under Promise.all, which rejects the moment one tool's result append fails (a transient fs fault, or errorMessage() throwing on a nullish thrown value). That rejection propagated out through withLock and freed the per-turn lock while sibling tools were still executing. A straggler's later append then ran with no lock held and validated against a stale in-memory history rather than the file — so if the caller re-advanced in between (writing the recovery indeterminate result for the not-yet-settled tool), the straggler appended a second tool_result for the same call. Duplicate results make reduceTurn throw on every future read, permanently corrupting the turn file and, through context references, every session behind it. - executeAllowedTools now awaits Promise.allSettled and rethrows the first fault only after every tool has finished appending, so a fault never orphans a sibling. - advance() drains the append queue (settleAppends) before withLock releases the turn, closing the fire-and-forget reportProgress path and any future append-path rejection. - errorMessage() uses optional chaining so a `throw null`/`throw undefined` yields a normal isError tool result instead of a TypeError that escapes as an infrastructure rejection (the same fault that triggers the lock-escape mid-batch). Adds two regression tests (both fail without the fix): a nullish throw records an isError result and completes the turn; a faulted batch does not settle until the slow sibling's append is durable. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
2e7950182a
commit
1dc042adc8
2 changed files with 181 additions and 6 deletions
|
|
@ -2604,6 +2604,147 @@ describe("sync tool result append failures (21.4)", () => {
|
||||||
result: { isError: true },
|
result: { isError: true },
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("a sync tool that throws a nullish value is a tool error, not an infrastructure rejection", async () => {
|
||||||
|
// Regression: errorMessage() dereferenced the thrown value directly, so
|
||||||
|
// `throw null` threw a TypeError *inside* the catch handler — promoting
|
||||||
|
// a modeled tool failure to an infrastructure rejection (and, mid-batch,
|
||||||
|
// orphaning sibling appends behind the released lock).
|
||||||
|
const executed: string[] = [];
|
||||||
|
const { runtime, repo } = makeRuntime({
|
||||||
|
tools: [
|
||||||
|
syncTool(echoDescriptor, async () => {
|
||||||
|
executed.push("echo");
|
||||||
|
const boom: unknown = null;
|
||||||
|
throw boom;
|
||||||
|
}),
|
||||||
|
...defaultTools.slice(1),
|
||||||
|
],
|
||||||
|
models: [
|
||||||
|
respond(
|
||||||
|
completedResp(assistantCalls(toolCallPart("A", "echo"))),
|
||||||
|
),
|
||||||
|
respond(completedResp(assistantText("done"))),
|
||||||
|
],
|
||||||
|
});
|
||||||
|
const turnId = await newTurn(runtime);
|
||||||
|
const { outcome, error } = await advanceAndSettle(runtime, turnId);
|
||||||
|
|
||||||
|
// The throw is conversational: the turn is not derailed.
|
||||||
|
expect(error).toBeUndefined();
|
||||||
|
expect(outcome?.status).toBe("completed");
|
||||||
|
expect(executed).toEqual(["echo"]);
|
||||||
|
const log = await persisted(repo, turnId);
|
||||||
|
const result = log.find((e) => e.type === "tool_result");
|
||||||
|
expect(result).toMatchObject({
|
||||||
|
source: "sync",
|
||||||
|
result: { output: "null", isError: true },
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not settle (and free the per-turn lock) until a straggling sibling append lands", async () => {
|
||||||
|
// Regression: when one tool's result append failed, the batch rejected
|
||||||
|
// immediately (Promise.all), settling the invocation and freeing the
|
||||||
|
// per-turn lock while a slower sibling was still executing. The
|
||||||
|
// sibling's later append then ran unlocked and against a stale gate
|
||||||
|
// view — racing a recovery advance, it could write a DUPLICATE result
|
||||||
|
// that the reducer rejects, permanently corrupting the turn file (and
|
||||||
|
// every session referencing it). The invocation must hold the lock
|
||||||
|
// until every sibling has finished appending, so here the invocation
|
||||||
|
// must not settle until slow's result is durable.
|
||||||
|
const slowDescriptor: z.infer<typeof ToolDescriptor> = {
|
||||||
|
toolId: "tool.slow",
|
||||||
|
name: "slow",
|
||||||
|
description: "Slow tool",
|
||||||
|
inputSchema: {},
|
||||||
|
execution: "sync",
|
||||||
|
requiresHuman: false,
|
||||||
|
};
|
||||||
|
const agent: z.infer<typeof ResolvedAgent> = {
|
||||||
|
...defaultAgent,
|
||||||
|
tools: [echoDescriptor, slowDescriptor],
|
||||||
|
};
|
||||||
|
let releaseSlow!: () => void;
|
||||||
|
const slowGate = new Promise<void>((resolve) => {
|
||||||
|
releaseSlow = resolve;
|
||||||
|
});
|
||||||
|
const repo = new FailingAppendRepo();
|
||||||
|
// Fail only echo's success append (the trigger); slow commits normally.
|
||||||
|
repo.failWhen = (events) =>
|
||||||
|
events.some(
|
||||||
|
(e) =>
|
||||||
|
e.type === "tool_result" &&
|
||||||
|
e.source === "sync" &&
|
||||||
|
e.toolCallId === "E" &&
|
||||||
|
e.result.isError === false,
|
||||||
|
);
|
||||||
|
const { runtime } = makeRuntime({
|
||||||
|
repo,
|
||||||
|
agent,
|
||||||
|
tools: [
|
||||||
|
syncTool(echoDescriptor, async () => ({
|
||||||
|
output: "e-done",
|
||||||
|
isError: false,
|
||||||
|
})),
|
||||||
|
syncTool(slowDescriptor, async () => {
|
||||||
|
await slowGate;
|
||||||
|
return { output: "s-done", isError: false };
|
||||||
|
}),
|
||||||
|
],
|
||||||
|
models: [
|
||||||
|
respond(
|
||||||
|
completedResp(
|
||||||
|
assistantCalls(
|
||||||
|
toolCallPart("E", "echo"),
|
||||||
|
toolCallPart("S", "slow"),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
respond(completedResp(assistantText("done"))),
|
||||||
|
],
|
||||||
|
});
|
||||||
|
const turnId = await newTurn(runtime);
|
||||||
|
|
||||||
|
// Advance 1: echo appends-and-fails while slow is still gated on
|
||||||
|
// slowGate. Under the old code the batch rejects here and the
|
||||||
|
// invocation settles; under the fix it stays pending, holding the lock.
|
||||||
|
let settled = false;
|
||||||
|
const first = settle(runtime.advanceTurn(turnId)).then((r) => {
|
||||||
|
settled = true;
|
||||||
|
return r;
|
||||||
|
});
|
||||||
|
// Yield a macrotask: all of the invocation's microtask work (including
|
||||||
|
// an early short-circuit) has drained. slow is still gated.
|
||||||
|
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
||||||
|
expect(settled).toBe(false);
|
||||||
|
|
||||||
|
// Only now let slow finish. The invocation can surface its error.
|
||||||
|
releaseSlow();
|
||||||
|
const r1 = await first;
|
||||||
|
expect(r1.outcome).toBeUndefined();
|
||||||
|
expect(String(r1.error)).toContain("disk full");
|
||||||
|
|
||||||
|
// slow's real result was durably appended before the invocation
|
||||||
|
// settled, so the log is intact — no window for a duplicate.
|
||||||
|
const log = await repo.read(turnId);
|
||||||
|
expect(() => reduceTurn(log)).not.toThrow();
|
||||||
|
const sResult = log.find(
|
||||||
|
(e) => e.type === "tool_result" && e.toolCallId === "S",
|
||||||
|
);
|
||||||
|
expect(sResult).toMatchObject({
|
||||||
|
source: "sync",
|
||||||
|
result: { output: "s-done", isError: false },
|
||||||
|
});
|
||||||
|
|
||||||
|
// Recovery then closes echo honestly and completes; one result per call.
|
||||||
|
repo.failWhen = undefined;
|
||||||
|
const r2 = await advanceAndSettle(runtime, turnId);
|
||||||
|
expect(r2.outcome?.status).toBe("completed");
|
||||||
|
const finalLog = await repo.read(turnId);
|
||||||
|
expect(
|
||||||
|
finalLog.filter((e) => e.type === "tool_result"),
|
||||||
|
).toHaveLength(2);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
describe("mid-turn tool extension", () => {
|
describe("mid-turn tool extension", () => {
|
||||||
|
|
|
||||||
|
|
@ -301,6 +301,10 @@ export class TurnRuntime implements ITurnRuntime {
|
||||||
try {
|
try {
|
||||||
return await run.run(input);
|
return await run.run(input);
|
||||||
} finally {
|
} finally {
|
||||||
|
// Drain any queued commits before withLock releases the turn, so
|
||||||
|
// no append (straggler tool, un-awaited progress) can run after
|
||||||
|
// the lock is gone — the invariant fs-repo relies on.
|
||||||
|
await run.settleAppends();
|
||||||
if (externalSignal) {
|
if (externalSignal) {
|
||||||
externalSignal.removeEventListener("abort", forwardAbort);
|
externalSignal.removeEventListener("abort", forwardAbort);
|
||||||
}
|
}
|
||||||
|
|
@ -463,6 +467,16 @@ class TurnAdvance {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Wait for every queued commit to run. advance() awaits this before it
|
||||||
|
// releases the per-turn lock, so an append can never land unlocked — not
|
||||||
|
// even from a fire-and-forget path (a tool that calls reportProgress
|
||||||
|
// without awaiting) or a sibling still in flight when the batch faulted.
|
||||||
|
// appendChain swallows rejections, so this never throws; any caller that
|
||||||
|
// needed to observe an append error already saw it from its own append().
|
||||||
|
async settleAppends(): Promise<void> {
|
||||||
|
await this.appendChain;
|
||||||
|
}
|
||||||
|
|
||||||
// §18 step 8: repeatedly advance deterministic work. Each phase either
|
// §18 step 8: repeatedly advance deterministic work. Each phase either
|
||||||
// appends durable facts and lets the loop continue, or produces the
|
// appends durable facts and lets the loop continue, or produces the
|
||||||
// invocation's outcome.
|
// invocation's outcome.
|
||||||
|
|
@ -898,11 +912,25 @@ class TurnAdvance {
|
||||||
}
|
}
|
||||||
started.push({ tc, tool: tool as SyncRuntimeTool });
|
started.push({ tc, tool: tool as SyncRuntimeTool });
|
||||||
}
|
}
|
||||||
// Each task settles its own call (result or error), so Promise.all
|
// Each task normally settles its own call (result or error). The one
|
||||||
// never rejects and one slow tool never blocks its siblings.
|
// way a task rejects is an append failure (repo fault, or a gate
|
||||||
await Promise.all(
|
// rejection) — and that MUST NOT short-circuit the batch: with
|
||||||
|
// Promise.all, the first rejection resolves this invocation and frees
|
||||||
|
// the per-turn lock while sibling tools are still executing, so a
|
||||||
|
// straggler's later append runs unlocked and against a stale gate
|
||||||
|
// view, which can write a duplicate tool_result and corrupt the turn
|
||||||
|
// file. allSettled holds the batch (and thus the lock) until every
|
||||||
|
// tool has finished appending; only then do we surface the fault as an
|
||||||
|
// infrastructure rejection.
|
||||||
|
const settled = await Promise.allSettled(
|
||||||
started.map(({ tc, tool }) => this.executeSyncTool(tc, tool)),
|
started.map(({ tc, tool }) => this.executeSyncTool(tc, tool)),
|
||||||
);
|
);
|
||||||
|
const rejected = settled.find(
|
||||||
|
(r): r is PromiseRejectedResult => r.status === "rejected",
|
||||||
|
);
|
||||||
|
if (rejected) {
|
||||||
|
throw rejected.reason;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private async executeSyncTool(
|
private async executeSyncTool(
|
||||||
|
|
@ -1337,9 +1365,15 @@ function errorMessage(error: unknown): string {
|
||||||
// (AI SDK APICallError; RetryError wraps the last one as lastError).
|
// (AI SDK APICallError; RetryError wraps the last one as lastError).
|
||||||
// "Failed after 3 attempts" alone is undebuggable — persist the payload,
|
// "Failed after 3 attempts" alone is undebuggable — persist the payload,
|
||||||
// bounded so request events stay reference-sized.
|
// bounded so request events stay reference-sized.
|
||||||
const source = (error as { lastError?: unknown }).lastError ?? error;
|
// Optional chaining throughout: the thrown value may be null/undefined (a
|
||||||
const statusCode = (source as { statusCode?: unknown }).statusCode;
|
// tool doing `throw null`, a provider rejecting with undefined). A bare
|
||||||
const responseBody = (source as { responseBody?: unknown }).responseBody;
|
// property access here would throw a TypeError inside a catch handler,
|
||||||
|
// turning a modeled tool/model failure into an infrastructure rejection —
|
||||||
|
// and, mid sync-tool batch, orphaning sibling appends (see the lock-drain
|
||||||
|
// in advance()).
|
||||||
|
const source = (error as { lastError?: unknown } | null)?.lastError ?? error;
|
||||||
|
const statusCode = (source as { statusCode?: unknown } | null)?.statusCode;
|
||||||
|
const responseBody = (source as { responseBody?: unknown } | null)?.responseBody;
|
||||||
const details: string[] = [];
|
const details: string[] = [];
|
||||||
if (typeof statusCode === "number") {
|
if (typeof statusCode === "number") {
|
||||||
details.push(`status ${statusCode}`);
|
details.push(`status ${statusCode}`);
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue