Merge pull request #743 from rowboatlabs/fix/turn-lock-straggler-append

fix(x): hold the per-turn lock until straggler sync-tool appends land

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<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", () => {

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<void> {
+        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}`);