diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index c29d8304..35dbefa0 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/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 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 ef8ea96c..394ed55a 100644 --- a/apps/x/packages/core/src/runtime/turns/runtime.test.ts +++ b/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", () => { diff --git a/apps/x/packages/core/src/runtime/turns/runtime.ts b/apps/x/packages/core/src/runtime/turns/runtime.ts index a7dfd5ce..0344347a 100644 --- a/apps/x/packages/core/src/runtime/turns/runtime.ts +++ b/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 = Promise.resolve(); private append(...batch: TEvent[]): Promise { @@ -429,12 +429,18 @@ class TurnAdvance { } private async commit(batch: TEvent[]): Promise { + // 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);