diff --git a/apps/x/packages/core/src/runtime/turns/runtime.test.ts b/apps/x/packages/core/src/runtime/turns/runtime.test.ts index 6deb6b64..8478867e 100644 --- a/apps/x/packages/core/src/runtime/turns/runtime.test.ts +++ b/apps/x/packages/core/src/runtime/turns/runtime.test.ts @@ -2604,6 +2604,147 @@ describe("sync tool result append failures (21.4)", () => { 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 = { + toolId: "tool.slow", + name: "slow", + description: "Slow tool", + inputSchema: {}, + execution: "sync", + requiresHuman: false, + }; + const agent: z.infer = { + ...defaultAgent, + tools: [echoDescriptor, slowDescriptor], + }; + let releaseSlow!: () => void; + const slowGate = new Promise((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((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", () => { diff --git a/apps/x/packages/core/src/runtime/turns/runtime.ts b/apps/x/packages/core/src/runtime/turns/runtime.ts index bca8a857..271f68ba 100644 --- a/apps/x/packages/core/src/runtime/turns/runtime.ts +++ b/apps/x/packages/core/src/runtime/turns/runtime.ts @@ -301,6 +301,10 @@ export class TurnRuntime implements ITurnRuntime { try { return await run.run(input); } 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) { 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 { + await this.appendChain; + } + // §18 step 8: repeatedly advance deterministic work. Each phase either // appends durable facts and lets the loop continue, or produces the // invocation's outcome. @@ -898,11 +912,25 @@ class TurnAdvance { } started.push({ tc, tool: tool as SyncRuntimeTool }); } - // Each task settles its own call (result or error), so Promise.all - // never rejects and one slow tool never blocks its siblings. - await Promise.all( + // Each task normally settles its own call (result or error). The one + // way a task rejects is an append failure (repo fault, or a gate + // 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)), ); + const rejected = settled.find( + (r): r is PromiseRejectedResult => r.status === "rejected", + ); + if (rejected) { + throw rejected.reason; + } } private async executeSyncTool( @@ -1337,9 +1365,15 @@ function errorMessage(error: unknown): string { // (AI SDK APICallError; RetryError wraps the last one as lastError). // "Failed after 3 attempts" alone is undebuggable — persist the payload, // bounded so request events stay reference-sized. - const source = (error as { lastError?: unknown }).lastError ?? error; - const statusCode = (source as { statusCode?: unknown }).statusCode; - const responseBody = (source as { responseBody?: unknown }).responseBody; + // Optional chaining throughout: the thrown value may be null/undefined (a + // tool doing `throw null`, a provider rejecting with undefined). A bare + // 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[] = []; if (typeof statusCode === "number") { details.push(`status ${statusCode}`);