Merge pull request #727 from rowboatlabs/fix/turn-commit-validation-gate

fix(x): validate turn event batches before they become durable

apps/x/packages/core/docs/turn-runtime-design.md

@@ -164,10 +164,18 @@ naturally complete independently. No behavior may rely on physical completion
 order.
 
 Durable appends are serialized through a single internal queue per invocation:
-the persist → reduce → stream ritual runs to completion for one batch of
+the reduce → persist → stream ritual runs to completion for one batch of
 events before the next begins, so file order, in-memory order, and stream
 order are identical by construction even while executions overlap.
 
+The reduce step comes first as a validation gate: the batch is reduced
+against the in-memory history before anything is written, so an illegal
+append (for example, a misbehaving tool reporting progress after its
+terminal result) rejects in memory for its caller only and never becomes
+durable. A persisted illegal event would make every future read of the
+file fail and, through context references (section 6.6), block the whole
+session chain behind it.
+
 ## 5. Storage design
 
 ### 5.1 File location
@@ -1656,6 +1664,11 @@ but the first turn implementation does not enforce it.
 The reducer validates `context` structurally but treats it as opaque; it
 never resolves references (section 6.6).
 
+The runtime also uses the reducer as its append gate: every batch is
+reduced against the existing history before it is persisted (section 4.5),
+so a history that violates these invariants cannot become durable through
+the loop.
+
 ## 25. Historical and live UI behavior
 
 ### 25.1 Historical load

apps/x/packages/core/src/runtime/turns/runtime.test.ts

@@ -2143,6 +2143,91 @@ describe("concurrent sync tool execution (10.5)", () => {
             "S",
         ]);
     });
+
+    it("a misbehaving tool's late progress rejects in memory and never poisons the log", async () => {
+        // fast stashes its reportProgress callback and finishes immediately;
+        // slow keeps the invocation alive, waits until fast's result is
+        // durable, then fires the stashed callback — tool_progress after a
+        // terminal tool_result, which the reducer forbids. The commit gate
+        // must reject that append before it becomes durable: a persisted
+        // illegal event would fail every future read of the file.
+        let lateReport: ToolExecutionContext["reportProgress"] | undefined;
+        const busRef: { current?: FakeTurnEventBus } = {};
+        const fastResultDurable = () =>
+            busRef.current?.events.some(
+                (e) =>
+                    e.event.type === "tool_result" && e.event.toolCallId === "F",
+            ) ?? false;
+        let lateOutcome: { settled: "resolved" } | { settled: "rejected"; error: unknown } | undefined;
+        const tools: RuntimeTool[] = [
+            syncTool(slowDescriptor, async () => {
+                while (!fastResultDurable()) {
+                    await new Promise((resolve) => setTimeout(resolve, 1));
+                }
+                try {
+                    await lateReport?.({ note: "too late" });
+                    lateOutcome = { settled: "resolved" };
+                } catch (error) {
+                    lateOutcome = { settled: "rejected", error };
+                }
+                return { output: "slow-done", isError: false };
+            }),
+            syncTool(fastDescriptor, async (_input, ctx) => {
+                lateReport = ctx.reportProgress;
+                return { output: "fast-done", isError: false };
+            }),
+        ];
+        const { runtime, repo, turnEventBus } = makeRuntime({
+            agent,
+            tools,
+            models: [
+                respond(
+                    completedResp(
+                        assistantCalls(
+                            toolCallPart("F", "fast"),
+                            toolCallPart("S", "slow"),
+                        ),
+                    ),
+                ),
+                respond(completedResp(assistantText("done"))),
+            ],
+        });
+        busRef.current = turnEventBus;
+        const turnId = await newTurn(runtime);
+        const { outcome, events } = await advanceAndSettle(runtime, turnId);
+
+        // The late append rejected for its caller only; the turn carried on.
+        expect(lateOutcome).toMatchObject({ settled: "rejected" });
+        expect(
+            String((lateOutcome as { error: unknown }).error),
+        ).toMatch(/tool progress after terminal result/);
+        expect(outcome?.status).toBe("completed");
+
+        // Nothing illegal became durable, streamed, or published: the file
+        // replays cleanly and holds exactly the legal history.
+        const log = await persisted(repo, turnId);
+        expect(() => reduceTurn(log)).not.toThrow();
+        expect(typesOf(log)).toEqual([
+            "turn_created",
+            "model_call_requested",
+            "model_call_completed",
+            "tool_invocation_requested",
+            "tool_invocation_requested",
+            "tool_result",
+            "tool_result",
+            "model_call_requested",
+            "model_call_completed",
+            "turn_completed",
+        ]);
+        expect(typesOf(events)).not.toContain("tool_progress");
+        expect(
+            turnEventBus.events.filter((e) => e.event.type === "tool_progress"),
+        ).toHaveLength(0);
+
+        // The turn stays readable and re-advanceable.
+        const again = await advanceAndSettle(runtime, turnId);
+        expect(again.outcome?.status).toBe("completed");
+    });
 });
 
 describe("mid-turn tool extension", () => {

apps/x/packages/core/src/runtime/turns/runtime.ts

@@ -390,11 +390,11 @@ class TurnAdvance {
         return this.clock.now();
     }
 
-    // Durable barrier: persist, re-reduce (the reducer doubles as a runtime
-    // assertion that the appended history is legal), then stream. Commits are
-    // serialized through an internal queue so concurrently executing tools
-    // can never interleave the persist/reduce/stream ritual — file order,
-    // in-memory order, and stream order stay identical by construction.
+    // Durable barrier: reduce (the reducer gates the append — see commit),
+    // persist, then stream. Commits are serialized through an internal queue
+    // so concurrently executing tools can never interleave the
+    // reduce/persist/stream ritual — file order, in-memory order, and stream
+    // order stay identical by construction.
     private appendChain: Promise<void> = Promise.resolve();
 
     private append(...batch: TEvent[]): Promise<void> {
@@ -429,12 +429,18 @@ class TurnAdvance {
     }
 
     private async commit(batch: TEvent[]): Promise<void> {
+        // Gate before the write: an illegal batch (e.g. a misbehaving tool
+        // reporting progress after its result) must reject in memory, never
+        // become durable — a persisted illegal event makes the file fail
+        // every future read and, through context references, blocks the
+        // whole session chain.
+        const next = reduceTurn([...this.events, ...batch]);
         await this.turnRepo.append(this.turnId, batch);
         // this.events holds the full file history (read at advance start), so
         // its length is the absolute 1-based line offset of each new event.
         const base = this.events.length;
         this.events.push(...batch);
-        this.state = reduceTurn(this.events);
+        this.state = next;
         this.appended = true;
         for (const [i, event] of batch.entries()) {
             this.stream.push(event);