From cda17c2d40300e929b12ade0ab6b0b3a9d765048 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 2 Jul 2026 09:32:27 +0530 Subject: [PATCH] feat(x/core): turn runtime with event-sourced execution loop (stage 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The TurnRuntime per turn-runtime-design.md: constructor-injected, container-ignorant, no per-turn state — every advanceTurn reconstructs from the JSONL log, so normal execution and crash recovery share one path. - runtime.ts: createTurn/advanceTurn/getTurn; durable-barrier appends before side effects; sequential sync tools with progress; async tool exposure + durable suspension; permission pipeline (checker → optional classifier batch → human/deny fallback, fail-closed on checker errors); cancellation with synthetic results; recovery that re-issues interrupted model calls and continues after interrupted sync tools; model-call-limit exhaustion with machine-readable code. - fs-repo.ts: date-partitioned append-only JSONL, strict line validation, path-traversal rejection, per-turn in-process locking. - stream.ts: hot execution stream — buffers before the consumer attaches, outcome independent of consumption, close-drops-events. - context-resolver.ts: chain-walking materialization of context references with cycle detection. - Seam interfaces for stage-4 bridges: model/tool registries, permission checker/classifier, agent resolver, lifecycle bus, clock. - shared reducer: track classification failures durably so failed classifications are never re-run. 67 new tests: all nine §26 end-to-end scenarios (mocked deps, recovery boundaries seeded directly) plus repo/stream/resolver suites. Co-Authored-By: Claude Fable 5 --- .../packages/core/src/turns/agent-resolver.ts | 14 + apps/x/packages/core/src/turns/api.ts | 112 ++ apps/x/packages/core/src/turns/bus.ts | 18 + apps/x/packages/core/src/turns/clock.ts | 11 + .../core/src/turns/context-resolver.test.ts | 257 +++ .../core/src/turns/context-resolver.ts | 55 + .../x/packages/core/src/turns/fs-repo.test.ts | 205 +++ apps/x/packages/core/src/turns/fs-repo.ts | 128 ++ .../core/src/turns/in-memory-turn-repo.ts | 62 + apps/x/packages/core/src/turns/index.ts | 14 + apps/x/packages/core/src/turns/keyed-mutex.ts | 21 + .../packages/core/src/turns/model-registry.ts | 46 + apps/x/packages/core/src/turns/permission.ts | 52 + apps/x/packages/core/src/turns/repo.ts | 15 + .../x/packages/core/src/turns/runtime.test.ts | 1495 +++++++++++++++++ apps/x/packages/core/src/turns/runtime.ts | 928 ++++++++++ apps/x/packages/core/src/turns/stream.test.ts | 89 + apps/x/packages/core/src/turns/stream.ts | 94 ++ .../packages/core/src/turns/tool-registry.ts | 35 + apps/x/packages/shared/src/turns.test.ts | 1 + apps/x/packages/shared/src/turns.ts | 5 +- 21 files changed, 3656 insertions(+), 1 deletion(-) create mode 100644 apps/x/packages/core/src/turns/agent-resolver.ts create mode 100644 apps/x/packages/core/src/turns/api.ts create mode 100644 apps/x/packages/core/src/turns/bus.ts create mode 100644 apps/x/packages/core/src/turns/clock.ts create mode 100644 apps/x/packages/core/src/turns/context-resolver.test.ts create mode 100644 apps/x/packages/core/src/turns/context-resolver.ts create mode 100644 apps/x/packages/core/src/turns/fs-repo.test.ts create mode 100644 apps/x/packages/core/src/turns/fs-repo.ts create mode 100644 apps/x/packages/core/src/turns/in-memory-turn-repo.ts create mode 100644 apps/x/packages/core/src/turns/index.ts create mode 100644 apps/x/packages/core/src/turns/keyed-mutex.ts create mode 100644 apps/x/packages/core/src/turns/model-registry.ts create mode 100644 apps/x/packages/core/src/turns/permission.ts create mode 100644 apps/x/packages/core/src/turns/repo.ts create mode 100644 apps/x/packages/core/src/turns/runtime.test.ts create mode 100644 apps/x/packages/core/src/turns/runtime.ts create mode 100644 apps/x/packages/core/src/turns/stream.test.ts create mode 100644 apps/x/packages/core/src/turns/stream.ts create mode 100644 apps/x/packages/core/src/turns/tool-registry.ts diff --git a/apps/x/packages/core/src/turns/agent-resolver.ts b/apps/x/packages/core/src/turns/agent-resolver.ts new file mode 100644 index 00000000..5ea1760d --- /dev/null +++ b/apps/x/packages/core/src/turns/agent-resolver.ts @@ -0,0 +1,14 @@ +import type { z } from "zod"; +import type { RequestedAgent, ResolvedAgent } from "@x/shared/dist/turns.js"; + +// Absorbs agent assembly: built-in/dynamic agent selection, user-defined +// agent loading, system-prompt augmentation, model precedence +// (override > agent > application default), tool attachment and filtering. +// The result is the immutable execution snapshot persisted in turn_created; +// the resolved system prompt is final byte-for-byte. Resolution failure +// rejects createTurn without creating a turn file. +export interface IAgentResolver { + resolve( + agent: z.infer, + ): Promise>; +} diff --git a/apps/x/packages/core/src/turns/api.ts b/apps/x/packages/core/src/turns/api.ts new file mode 100644 index 00000000..55a60e2c --- /dev/null +++ b/apps/x/packages/core/src/turns/api.ts @@ -0,0 +1,112 @@ +import type { z } from "zod"; +import type { AssistantMessage, UserMessage } from "@x/shared/dist/message.js"; +import type { + JsonValue, + RequestedAgent, + ToolResultData, + TurnContext, + TurnEvent, + TurnStreamEvent, + TurnSuspended, + TurnUsage, +} from "@x/shared/dist/turns.js"; + +export interface CreateTurnInput { + agent: z.infer; + sessionId?: string | null; + context: z.infer; + input: z.infer; + config: { + autoPermission?: boolean; + humanAvailable: boolean; + maxModelCalls?: number; + }; +} + +// Exactly one external input per advanceTurn invocation. +export type TurnExternalInput = + | { + type: "permission_decision"; + toolCallId: string; + decision: "allow" | "deny"; + metadata?: JsonValue; + } + | { + type: "async_tool_progress"; + toolCallId: string; + progress: JsonValue; + } + | { + type: "async_tool_result"; + toolCallId: string; + result: z.infer; + } + | { + type: "cancel"; + reason?: string; + }; + +export type TurnOutcome = + | { + status: "completed"; + output: z.infer; + finishReason: string; + usage: z.infer; + } + | { + status: "suspended"; + pendingPermissions: z.infer["pendingPermissions"]; + pendingAsyncTools: z.infer["pendingAsyncTools"]; + usage: z.infer; + } + | { + status: "failed"; + error: string; + // Mirrors turn_failed.code (e.g. MODEL_CALL_LIMIT_ERROR_CODE). + code?: string; + usage: z.infer; + } + | { + status: "cancelled"; + reason?: string; + usage: z.infer; + }; + +export interface TurnExecution { + events: AsyncIterable; + outcome: Promise; +} + +export interface Turn { + turnId: string; + events: Array>; +} + +export interface ITurnRuntime { + createTurn(input: CreateTurnInput): Promise; + advanceTurn( + turnId: string, + input?: TurnExternalInput, + options?: { signal?: AbortSignal }, + ): TurnExecution; + getTurn(turnId: string): Promise; +} + +// An external input that does not match the turn's current durable pending +// state. Rejects the execution; nothing is appended. +export class TurnInputError extends Error { + constructor(message: string) { + super(message); + this.name = "TurnInputError"; + } +} + +// A live runtime dependency is missing or incompatible with the persisted +// snapshot. Rejects the execution; the turn is left unchanged so the caller +// can fix its environment and retry. +export class TurnDependencyError extends Error { + constructor(message: string) { + super(message); + this.name = "TurnDependencyError"; + } +} diff --git a/apps/x/packages/core/src/turns/bus.ts b/apps/x/packages/core/src/turns/bus.ts new file mode 100644 index 00000000..c998e6a1 --- /dev/null +++ b/apps/x/packages/core/src/turns/bus.ts @@ -0,0 +1,18 @@ +// Ephemeral process-lifecycle events (turn-runtime-design.md §17). Never +// persisted, never replayed; if the process crashes the information +// disappears, which accurately reflects that no execution is known active. +export interface TurnProcessingStart { + type: "turn-processing-start"; + turnId: string; +} + +export interface TurnProcessingEnd { + type: "turn-processing-end"; + turnId: string; +} + +export type TurnLifecycleEvent = TurnProcessingStart | TurnProcessingEnd; + +export interface ITurnLifecycleBus { + publish(event: TurnLifecycleEvent): void; +} diff --git a/apps/x/packages/core/src/turns/clock.ts b/apps/x/packages/core/src/turns/clock.ts new file mode 100644 index 00000000..a88a3122 --- /dev/null +++ b/apps/x/packages/core/src/turns/clock.ts @@ -0,0 +1,11 @@ +// Deterministic timestamp seam. No IClock existed in the codebase before the +// turn runtime; production uses SystemClock, tests inject fixed clocks. +export interface IClock { + now(): string; // ISO-8601 timestamp +} + +export class SystemClock implements IClock { + now(): string { + return new Date().toISOString(); + } +} diff --git a/apps/x/packages/core/src/turns/context-resolver.test.ts b/apps/x/packages/core/src/turns/context-resolver.test.ts new file mode 100644 index 00000000..39911786 --- /dev/null +++ b/apps/x/packages/core/src/turns/context-resolver.test.ts @@ -0,0 +1,257 @@ +import { describe, expect, it } from "vitest"; +import type { z } from "zod"; +import { + MODEL_CALL_LIMIT_ERROR_CODE, + type TurnContext, + TurnCorruptionError, + type TurnEvent, +} from "@x/shared/dist/turns.js"; +import { TurnRepoContextResolver } from "./context-resolver.js"; +import { InMemoryTurnRepo } from "./in-memory-turn-repo.js"; + +type TEvent = z.infer; + +function user(text: string) { + return { role: "user" as const, content: text }; +} + +function assistant(text: string) { + return { role: "assistant" as const, content: text }; +} + +// A minimal completed turn: input → one model call → text response. +function completedTurnLog( + turnId: string, + context: z.infer, + inputText: string, + responseText: string, +): TEvent[] { + const ts = "2026-07-02T10:00:00Z"; + return [ + { + type: "turn_created", + schemaVersion: 1, + turnId, + ts, + sessionId: "sess-1", + agent: { + requested: { agentId: "copilot" }, + resolved: { + agentId: "copilot", + systemPrompt: "SYS", + model: { provider: "fake", model: "m" }, + tools: [], + }, + }, + context, + input: user(inputText), + config: { + autoPermission: false, + humanAvailable: true, + maxModelCalls: 20, + }, + }, + { + type: "model_call_requested", + turnId, + ts, + modelCallIndex: 0, + request: { + systemPrompt: "SYS", + ...(Array.isArray(context) ? {} : { contextRef: context }), + messages: Array.isArray(context) + ? [...context, user(inputText)] + : [user(inputText)], + tools: [], + parameters: {}, + }, + }, + { + type: "model_call_completed", + turnId, + ts, + modelCallIndex: 0, + message: assistant(responseText), + finishReason: "stop", + usage: {}, + }, + { + type: "turn_completed", + turnId, + ts, + output: assistant(responseText), + finishReason: "stop", + usage: {}, + }, + ]; +} + +// A turn that failed at the model-call limit after one tool round trip; its +// transcript is structurally complete including the synthetic closure. +function limitFailedTurnLog(turnId: string): TEvent[] { + const ts = "2026-07-02T10:00:00Z"; + const echo = { + toolId: "tool.echo", + name: "echo", + description: "Echo", + inputSchema: {}, + execution: "sync" as const, + requiresHuman: false, + }; + const call = { + role: "assistant" as const, + content: [ + { + type: "tool-call" as const, + toolCallId: "tc1", + toolName: "echo", + arguments: {}, + }, + ], + }; + return [ + { + type: "turn_created", + schemaVersion: 1, + turnId, + ts, + sessionId: "sess-1", + agent: { + requested: { agentId: "copilot" }, + resolved: { + agentId: "copilot", + systemPrompt: "SYS", + model: { provider: "fake", model: "m" }, + tools: [echo], + }, + }, + context: [], + input: user("do it"), + config: { + autoPermission: false, + humanAvailable: true, + maxModelCalls: 1, + }, + }, + { + type: "model_call_requested", + turnId, + ts, + modelCallIndex: 0, + request: { + systemPrompt: "SYS", + messages: [user("do it")], + tools: [echo], + parameters: {}, + }, + }, + { + type: "model_call_completed", + turnId, + ts, + modelCallIndex: 0, + message: call, + finishReason: "tool-calls", + usage: {}, + }, + { + type: "tool_invocation_requested", + turnId, + ts, + toolCallId: "tc1", + toolId: "tool.echo", + toolName: "echo", + execution: "sync", + input: {}, + }, + { + type: "tool_result", + turnId, + ts, + toolCallId: "tc1", + toolName: "echo", + source: "sync", + result: { output: "echoed", isError: false }, + }, + { + type: "turn_failed", + turnId, + ts, + error: "Model call limit of 1 reached before the turn completed.", + code: MODEL_CALL_LIMIT_ERROR_CODE, + usage: {}, + }, + ]; +} + +const T1 = "2026-07-02T10-00-00Z-0000001-000"; +const T2 = "2026-07-02T10-00-00Z-0000002-000"; +const T3 = "2026-07-02T10-00-00Z-0000003-000"; + +describe("TurnRepoContextResolver", () => { + it("passes inline context through unchanged", async () => { + const repo = new InMemoryTurnRepo(); + const resolver = new TurnRepoContextResolver({ turnRepo: repo }); + const inline = [user("a"), assistant("b")]; + expect(await resolver.resolve(inline)).toEqual(inline); + }); + + it("resolves a single reference to the referenced turn's transcript", async () => { + const repo = new InMemoryTurnRepo(); + repo.seed(completedTurnLog(T1, [], "first question", "first answer")); + const resolver = new TurnRepoContextResolver({ turnRepo: repo }); + expect(await resolver.resolve({ previousTurnId: T1 })).toEqual([ + user("first question"), + assistant("first answer"), + ]); + }); + + it("resolves a chain of references down to the inline base", async () => { + const repo = new InMemoryTurnRepo(); + repo.seed(completedTurnLog(T1, [user("preamble")], "q1", "a1")); + repo.seed(completedTurnLog(T2, { previousTurnId: T1 }, "q2", "a2")); + repo.seed(completedTurnLog(T3, { previousTurnId: T2 }, "q3", "a3")); + const resolver = new TurnRepoContextResolver({ turnRepo: repo }); + expect(await resolver.resolve({ previousTurnId: T3 })).toEqual([ + user("preamble"), + user("q1"), + assistant("a1"), + user("q2"), + assistant("a2"), + user("q3"), + assistant("a3"), + ]); + }); + + it("includes failed turns' transcripts with synthetic closures", async () => { + const repo = new InMemoryTurnRepo(); + repo.seed(limitFailedTurnLog(T1)); + const resolver = new TurnRepoContextResolver({ turnRepo: repo }); + const resolved = await resolver.resolve({ previousTurnId: T1 }); + expect(resolved.map((m) => m.role)).toEqual(["user", "assistant", "tool"]); + expect(resolved[2]).toMatchObject({ + role: "tool", + toolCallId: "tc1", + content: "echoed", + }); + }); + + it("rejects references to missing turns as infrastructure errors", async () => { + const repo = new InMemoryTurnRepo(); + const resolver = new TurnRepoContextResolver({ turnRepo: repo }); + await expect( + resolver.resolve({ previousTurnId: T1 }), + ).rejects.toThrowError(/turn not found/); + }); + + it("rejects cyclic reference chains as corruption", async () => { + const repo = new InMemoryTurnRepo(); + // Two turns referencing each other (only constructable by corruption). + repo.seed(completedTurnLog(T1, { previousTurnId: T2 }, "q1", "a1")); + repo.seed(completedTurnLog(T2, { previousTurnId: T1 }, "q2", "a2")); + const resolver = new TurnRepoContextResolver({ turnRepo: repo }); + await expect( + resolver.resolve({ previousTurnId: T1 }), + ).rejects.toThrowError(TurnCorruptionError); + }); +}); diff --git a/apps/x/packages/core/src/turns/context-resolver.ts b/apps/x/packages/core/src/turns/context-resolver.ts new file mode 100644 index 00000000..0dce3181 --- /dev/null +++ b/apps/x/packages/core/src/turns/context-resolver.ts @@ -0,0 +1,55 @@ +import type { z } from "zod"; +import { + type ConversationMessage, + type TurnContext, + TurnCorruptionError, + reduceTurn, + turnTranscript, +} from "@x/shared/dist/turns.js"; +import type { ITurnRepo } from "./repo.js"; + +// Materializes a turn's context (turn-runtime-design.md §6.6). Inline +// contexts pass through; references resolve to the referenced turn's full +// transcript by walking the chain down to its inline base. Resolution always +// reads durable state, so normal execution and crash recovery share one +// path. A missing or corrupt referenced turn is an infrastructure error. +export interface IContextResolver { + resolve( + context: z.infer, + ): Promise>>; +} + +export class TurnRepoContextResolver implements IContextResolver { + private readonly turnRepo: ITurnRepo; + + constructor({ turnRepo }: { turnRepo: ITurnRepo }) { + this.turnRepo = turnRepo; + } + + async resolve( + context: z.infer, + ): Promise>> { + // Walk the reference chain back to the inline base, then concatenate + // transcripts oldest-first. Iterative to bound stack depth; a visited + // set catches cyclic (corrupt) chains. + const segments: Array>> = []; + const visited = new Set(); + let current = context; + while (!Array.isArray(current)) { + const turnId = current.previousTurnId; + if (visited.has(turnId)) { + throw new TurnCorruptionError( + `cyclic context reference chain at turn ${turnId}`, + ); + } + visited.add(turnId); + const events = await this.turnRepo.read(turnId); + const state = reduceTurn(events); + segments.push(turnTranscript(state)); + current = state.definition.context; + } + segments.push(current); + segments.reverse(); + return segments.flat(); + } +} diff --git a/apps/x/packages/core/src/turns/fs-repo.test.ts b/apps/x/packages/core/src/turns/fs-repo.test.ts new file mode 100644 index 00000000..8e44b5b1 --- /dev/null +++ b/apps/x/packages/core/src/turns/fs-repo.test.ts @@ -0,0 +1,205 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import type { z } from "zod"; +import { + TurnCorruptionError, + TurnCreated, + TurnEvent, +} from "@x/shared/dist/turns.js"; +import { FSTurnRepo } from "./fs-repo.js"; + +const TURN_ID = "2026-07-02T10-00-00Z-0000001-000"; + +function created(turnId = TURN_ID): z.infer { + return { + type: "turn_created", + schemaVersion: 1, + turnId, + ts: "2026-07-02T10:00:00Z", + sessionId: null, + agent: { + requested: { agentId: "copilot" }, + resolved: { + agentId: "copilot", + systemPrompt: "SYS", + model: { provider: "fake", model: "m" }, + tools: [], + }, + }, + context: [], + input: { role: "user", content: "hello" }, + config: { autoPermission: false, humanAvailable: true, maxModelCalls: 20 }, + }; +} + +function requested(turnId = TURN_ID): z.infer { + return { + type: "model_call_requested", + turnId, + ts: "2026-07-02T10:00:01Z", + modelCallIndex: 0, + request: { + systemPrompt: "SYS", + messages: [{ role: "user", content: "hello" }], + tools: [], + parameters: {}, + }, + }; +} + +function failed(turnId = TURN_ID): z.infer { + return { + type: "model_call_failed", + turnId, + ts: "2026-07-02T10:00:02Z", + modelCallIndex: 0, + error: "boom", + }; +} + +describe("FSTurnRepo", () => { + let root: string; + let repo: FSTurnRepo; + + beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "turn-repo-")); + repo = new FSTurnRepo({ turnsRootDir: root }); + }); + + afterEach(async () => { + await fs.rm(root, { recursive: true, force: true }); + }); + + it("writes to a deterministic date-partitioned path", async () => { + await repo.create(created()); + const file = path.join(root, "2026", "07", "02", `${TURN_ID}.jsonl`); + const raw = await fs.readFile(file, "utf8"); + expect(raw.endsWith("\n")).toBe(true); + expect(JSON.parse(raw.trim()).type).toBe("turn_created"); + }); + + it("rejects malformed and path-like turn ids", async () => { + for (const bad of [ + "../../../etc/passwd", + "2026-07-02T10/evil", + "2026-07-02T10-00-00Z-0000001-000.jsonl", + "not-a-turn-id", + "", + ]) { + await expect(repo.read(bad)).rejects.toThrowError(/invalid turn id/); + } + }); + + it("create fails if the turn already exists", async () => { + await repo.create(created()); + await expect(repo.create(created())).rejects.toThrowError(); + }); + + it("appends preserve order and read validates every line", async () => { + await repo.create(created()); + await repo.append(TURN_ID, [requested()]); + await repo.append(TURN_ID, [failed()]); + const events = await repo.read(TURN_ID); + expect(events.map((e) => e.type)).toEqual([ + "turn_created", + "model_call_requested", + "model_call_failed", + ]); + }); + + it("append validates events and turn id match before writing", async () => { + await repo.create(created()); + await expect( + repo.append(TURN_ID, [requested("2026-07-02T10-00-00Z-0000002-000")]), + ).rejects.toThrowError(/does not match/); + await expect( + repo.append(TURN_ID, [{ type: "wat" } as unknown as z.infer]), + ).rejects.toThrowError(); + // Nothing was written by the failed appends. + expect((await repo.read(TURN_ID)).length).toBe(1); + }); + + it("append never creates a missing turn file", async () => { + await expect(repo.append(TURN_ID, [requested()])).rejects.toThrowError( + /turn not found/, + ); + }); + + it("reading a missing turn reports not found", async () => { + await expect(repo.read(TURN_ID)).rejects.toThrowError(/turn not found/); + }); + + async function writeRaw(content: string): Promise { + const file = path.join(root, "2026", "07", "02", `${TURN_ID}.jsonl`); + await fs.mkdir(path.dirname(file), { recursive: true }); + await fs.writeFile(file, content); + } + + it("rejects an empty file as corrupt", async () => { + await writeRaw(""); + await expect(repo.read(TURN_ID)).rejects.toThrowError(TurnCorruptionError); + }); + + it("rejects malformed first, middle, and final lines", async () => { + const good = JSON.stringify(created()); + const req = JSON.stringify(requested()); + for (const content of [ + `not json\n${req}\n`, + `${good}\nnot json\n${req}\n`, + `${good}\n{"type":"wat"}\n`, + ]) { + await writeRaw(content); + await expect(repo.read(TURN_ID)).rejects.toThrowError( + TurnCorruptionError, + ); + } + }); + + it("rejects a torn final line (no trailing newline)", async () => { + const good = JSON.stringify(created()); + const torn = JSON.stringify(requested()).slice(0, 20); + await writeRaw(`${good}\n${torn}`); + await expect(repo.read(TURN_ID)).rejects.toThrowError( + /does not end with a complete line/, + ); + }); + + it("rejects unsupported schema versions on read", async () => { + const v2 = { ...created(), schemaVersion: 2 }; + await writeRaw(`${JSON.stringify(v2)}\n`); + await expect(repo.read(TURN_ID)).rejects.toThrowError(TurnCorruptionError); + }); + + it("rejects events whose turnId does not match the file", async () => { + const other = created("2026-07-02T10-00-00Z-0000002-000"); + await writeRaw(`${JSON.stringify(other)}\n`); + await expect(repo.read(TURN_ID)).rejects.toThrowError(/does not match file/); + }); + + it("withLock serializes work per turn", async () => { + const order: string[] = []; + await Promise.all([ + repo.withLock(TURN_ID, async () => { + order.push("a-start"); + await new Promise((r) => setTimeout(r, 20)); + order.push("a-end"); + }), + repo.withLock(TURN_ID, async () => { + order.push("b-start"); + order.push("b-end"); + }), + ]); + expect(order).toEqual(["a-start", "a-end", "b-start", "b-end"]); + }); + + it("withLock releases after failures", async () => { + await expect( + repo.withLock(TURN_ID, async () => { + throw new Error("first fails"); + }), + ).rejects.toThrowError("first fails"); + await expect(repo.withLock(TURN_ID, async () => "ok")).resolves.toBe("ok"); + }); +}); diff --git a/apps/x/packages/core/src/turns/fs-repo.ts b/apps/x/packages/core/src/turns/fs-repo.ts new file mode 100644 index 00000000..57d19ca8 --- /dev/null +++ b/apps/x/packages/core/src/turns/fs-repo.ts @@ -0,0 +1,128 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { z } from "zod"; +import { + TurnCorruptionError, + TurnCreated, + TurnEvent, +} from "@x/shared/dist/turns.js"; +import { KeyedMutex } from "./keyed-mutex.js"; +import type { ITurnRepo } from "./repo.js"; + +// Turn IDs come from IMonotonicallyIncreasingIdGenerator and look like +// 2025-11-11T04-36-29Z-0001234-000. The repo validates the format before +// deriving a path and rejects anything path-like. +const TURN_ID_PATTERN = /^(\d{4})-(\d{2})-(\d{2})T[A-Za-z0-9-]+$/; + +export class FSTurnRepo implements ITurnRepo { + private readonly rootDir: string; + private readonly mutex = new KeyedMutex(); + + constructor({ turnsRootDir }: { turnsRootDir: string }) { + this.rootDir = turnsRootDir; + } + + private filePath(turnId: string): string { + const match = TURN_ID_PATTERN.exec(turnId); + if (!match) { + throw new Error(`invalid turn id: ${turnId}`); + } + const [, year, month, day] = match; + return path.join(this.rootDir, year, month, day, `${turnId}.jsonl`); + } + + private serialize( + turnId: string, + events: Array>, + ): string { + let payload = ""; + for (const event of events) { + const parsed = TurnEvent.parse(event); + if (parsed.turnId !== turnId) { + throw new Error( + `event turnId ${parsed.turnId} does not match ${turnId}`, + ); + } + payload += `${JSON.stringify(parsed)}\n`; + } + return payload; + } + + async create(event: z.infer): Promise { + const parsed = TurnCreated.parse(event); + const file = this.filePath(parsed.turnId); + await fs.mkdir(path.dirname(file), { recursive: true }); + // "wx" fails if the file already exists. + await fs.writeFile(file, this.serialize(parsed.turnId, [parsed]), { + flag: "wx", + }); + } + + async read(turnId: string): Promise>> { + const file = this.filePath(turnId); + let raw: string; + try { + raw = await fs.readFile(file, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + throw new Error(`turn not found: ${turnId}`); + } + throw error; + } + if (raw.length === 0) { + throw new TurnCorruptionError(`turn file is empty: ${turnId}`); + } + const lines = raw.split("\n"); + // A well-formed file ends with a newline, leaving one trailing empty + // segment. Anything else (including a torn final line) is corrupt. + const trailing = lines.pop(); + if (trailing !== "") { + throw new TurnCorruptionError( + `turn file does not end with a complete line: ${turnId}`, + ); + } + const events: Array> = []; + for (const [index, line] of lines.entries()) { + let parsed: z.infer; + try { + parsed = TurnEvent.parse(JSON.parse(line)); + } catch (error) { + throw new TurnCorruptionError( + `malformed turn event at ${turnId}:${index + 1}: ${String( + error instanceof Error ? error.message : error, + )}`, + ); + } + if (parsed.turnId !== turnId) { + throw new TurnCorruptionError( + `event turnId ${parsed.turnId} does not match file ${turnId}`, + ); + } + events.push(parsed); + } + return events; + } + + async append( + turnId: string, + events: Array>, + ): Promise { + if (events.length === 0) { + return; + } + const payload = this.serialize(turnId, events); + const file = this.filePath(turnId); + // Appends must never create a file: a turn exists only via create(). + // All writers hold the per-turn lock, so check-then-append is safe. + try { + await fs.access(file); + } catch { + throw new Error(`turn not found: ${turnId}`); + } + await fs.appendFile(file, payload); + } + + async withLock(turnId: string, fn: () => Promise): Promise { + return this.mutex.run(turnId, fn); + } +} diff --git a/apps/x/packages/core/src/turns/in-memory-turn-repo.ts b/apps/x/packages/core/src/turns/in-memory-turn-repo.ts new file mode 100644 index 00000000..3dfd176d --- /dev/null +++ b/apps/x/packages/core/src/turns/in-memory-turn-repo.ts @@ -0,0 +1,62 @@ +import { z } from "zod"; +import { TurnCreated, TurnEvent } from "@x/shared/dist/turns.js"; +import { KeyedMutex } from "./keyed-mutex.js"; +import type { ITurnRepo } from "./repo.js"; + +// Test fake mirroring FSTurnRepo semantics (create-if-absent, validate on +// write and read boundaries, per-turn locking) without touching disk. +export class InMemoryTurnRepo implements ITurnRepo { + private files = new Map>>(); + private mutex = new KeyedMutex(); + + async create(event: z.infer): Promise { + const parsed = TurnCreated.parse(event); + if (this.files.has(parsed.turnId)) { + throw new Error(`turn already exists: ${parsed.turnId}`); + } + this.files.set(parsed.turnId, [structuredClone(parsed)]); + } + + async read(turnId: string): Promise>> { + const events = this.files.get(turnId); + if (!events) { + throw new Error(`turn not found: ${turnId}`); + } + return structuredClone(events); + } + + async append( + turnId: string, + events: Array>, + ): Promise { + const file = this.files.get(turnId); + if (!file) { + throw new Error(`turn not found: ${turnId}`); + } + for (const event of events) { + const parsed = TurnEvent.parse(event); + if (parsed.turnId !== turnId) { + throw new Error( + `event turnId ${parsed.turnId} does not match ${turnId}`, + ); + } + file.push(structuredClone(parsed)); + } + } + + async withLock(turnId: string, fn: () => Promise): Promise { + return this.mutex.run(turnId, fn); + } + + // Test helper: seed a raw event log (validated) for recovery scenarios. + seed(events: Array>): void { + if (events.length === 0) { + throw new Error("cannot seed an empty log"); + } + const turnId = events[0].turnId; + this.files.set( + turnId, + events.map((e) => structuredClone(TurnEvent.parse(e))), + ); + } +} diff --git a/apps/x/packages/core/src/turns/index.ts b/apps/x/packages/core/src/turns/index.ts new file mode 100644 index 00000000..e7cbccf3 --- /dev/null +++ b/apps/x/packages/core/src/turns/index.ts @@ -0,0 +1,14 @@ +export * from "./api.js"; +export * from "./agent-resolver.js"; +export * from "./bus.js"; +export * from "./clock.js"; +export * from "./context-resolver.js"; +export * from "./fs-repo.js"; +export * from "./in-memory-turn-repo.js"; +export * from "./keyed-mutex.js"; +export * from "./model-registry.js"; +export * from "./permission.js"; +export * from "./repo.js"; +export * from "./runtime.js"; +export * from "./stream.js"; +export * from "./tool-registry.js"; diff --git a/apps/x/packages/core/src/turns/keyed-mutex.ts b/apps/x/packages/core/src/turns/keyed-mutex.ts new file mode 100644 index 00000000..e39c2404 --- /dev/null +++ b/apps/x/packages/core/src/turns/keyed-mutex.ts @@ -0,0 +1,21 @@ +// In-process per-key exclusion. Cross-process coordination is explicitly out +// of scope for the turn runtime (single Electron main process). +export class KeyedMutex { + private tails = new Map>(); + + async run(key: string, fn: () => Promise): Promise { + const prev = this.tails.get(key) ?? Promise.resolve(); + const next = prev.then( + () => fn(), + () => fn(), + ); + this.tails.set(key, next); + try { + return await next; + } finally { + if (this.tails.get(key) === next) { + this.tails.delete(key); + } + } + } +} diff --git a/apps/x/packages/core/src/turns/model-registry.ts b/apps/x/packages/core/src/turns/model-registry.ts new file mode 100644 index 00000000..d87aa097 --- /dev/null +++ b/apps/x/packages/core/src/turns/model-registry.ts @@ -0,0 +1,46 @@ +import type { z } from "zod"; +import type { AssistantMessage } from "@x/shared/dist/message.js"; +import type { + ConversationMessage, + DurableLlmStepStreamEvent, + JsonValue, + ModelDescriptor, + ToolDescriptor, + TurnUsage, +} from "@x/shared/dist/turns.js"; + +// One stream() call performs exactly one model step; the turn loop drives +// multi-step behavior. The stream yields normalized events and must end with +// exactly one "completed" event, or throw (a throw is a model failure; an +// abort-triggered throw is cancellation). +export type LlmStreamEvent = + | { type: "text_delta"; delta: string } + | { type: "reasoning_delta"; delta: string } + | { type: "step_event"; event: z.infer } + | { + type: "completed"; + message: z.infer; + finishReason: string; + usage: z.infer; + providerMetadata?: JsonValue; + }; + +export interface ModelStreamRequest { + systemPrompt: string; + messages: Array>; + tools: Array>; + parameters: Record; + signal: AbortSignal; +} + +export interface ResolvedModel { + descriptor: z.infer; + stream(request: ModelStreamRequest): AsyncIterable; +} + +// Resolves the persisted model descriptor to a live model during advanceTurn. +// A rejection is an infrastructure error: the execution is rejected and the +// turn is left unchanged. +export interface IModelRegistry { + resolve(descriptor: z.infer): Promise; +} diff --git a/apps/x/packages/core/src/turns/permission.ts b/apps/x/packages/core/src/turns/permission.ts new file mode 100644 index 00000000..7f386623 --- /dev/null +++ b/apps/x/packages/core/src/turns/permission.ts @@ -0,0 +1,52 @@ +import type { JsonValue } from "@x/shared/dist/turns.js"; + +export interface PermissionCheckInput { + turnId: string; + toolCallId: string; + toolId: string; + toolName: string; + input: unknown; +} + +export interface PermissionCheckAllowed { + required: false; +} + +export interface PermissionCheckRequired { + required: true; + // Presentation payload persisted on tool_permission_required and shown to + // the human/classifier. + request: JsonValue; +} + +// Tool-specific policy (command analysis, filesystem boundaries, allowlists) +// lives behind this seam, outside the loop. A thrown error fails closed: the +// call is recorded as permission-required and never executes automatically. +export interface IPermissionChecker { + check( + input: PermissionCheckInput, + ): Promise; +} + +export interface PermissionClassificationInput { + toolCallId: string; + toolName: string; + input: unknown; + request: JsonValue; +} + +export interface PermissionClassification { + toolCallId: string; + decision: "allow" | "deny" | "defer"; + reason: string; +} + +// Handles all permission-required calls from one model response in one batch +// when automatic permission is enabled. Internal model calls are opaque to +// the turn loop. Failures and omitted decisions normalize to defer. +export interface IPermissionClassifier { + classify( + requests: PermissionClassificationInput[], + signal: AbortSignal, + ): Promise; +} diff --git a/apps/x/packages/core/src/turns/repo.ts b/apps/x/packages/core/src/turns/repo.ts new file mode 100644 index 00000000..6824f529 --- /dev/null +++ b/apps/x/packages/core/src/turns/repo.ts @@ -0,0 +1,15 @@ +import type { z } from "zod"; +import type { TurnCreated, TurnEvent } from "@x/shared/dist/turns.js"; + +// Loop-facing repository contract. Listing, deletion, session lookup, and +// presentation metadata are deliberately not part of it. +export interface ITurnRepo { + // Fails if the turn already exists. + create(event: z.infer): Promise; + // Validates every line strictly; corrupt files are rejected whole. + read(turnId: string): Promise>>; + // Validates events before writing. + append(turnId: string, events: Array>): Promise; + // In-process per-turn exclusion. + withLock(turnId: string, fn: () => Promise): Promise; +} diff --git a/apps/x/packages/core/src/turns/runtime.test.ts b/apps/x/packages/core/src/turns/runtime.test.ts new file mode 100644 index 00000000..b9019b60 --- /dev/null +++ b/apps/x/packages/core/src/turns/runtime.test.ts @@ -0,0 +1,1495 @@ +import { describe, expect, it } from "vitest"; +import type { z } from "zod"; +import { + MODEL_CALL_LIMIT_ERROR_CODE, + type JsonValue, + type ResolvedAgent, + type ToolDescriptor, + type TurnEvent, + type TurnStreamEvent, +} from "@x/shared/dist/turns.js"; +import type { IAgentResolver } from "./agent-resolver.js"; +import type { TurnExecution, TurnOutcome } from "./api.js"; +import { TurnDependencyError, TurnInputError } from "./api.js"; +import type { TurnLifecycleEvent } from "./bus.js"; +import { TurnRepoContextResolver } from "./context-resolver.js"; +import { InMemoryTurnRepo } from "./in-memory-turn-repo.js"; +import type { + IModelRegistry, + LlmStreamEvent, + ModelStreamRequest, + ResolvedModel, +} from "./model-registry.js"; +import type { + IPermissionChecker, + IPermissionClassifier, + PermissionCheckInput, + PermissionClassification, + PermissionClassificationInput, +} from "./permission.js"; +import { TurnRuntime } from "./runtime.js"; +import type { + IToolRegistry, + RuntimeTool, + SyncRuntimeTool, + ToolExecutionContext, +} from "./tool-registry.js"; + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +const TS = "2026-07-02T10:00:00Z"; + +const echoDescriptor: z.infer = { + toolId: "tool.echo", + name: "echo", + description: "Echo tool", + inputSchema: {}, + execution: "sync", + requiresHuman: false, +}; + +const fetchDescriptor: z.infer = { + toolId: "tool.fetch", + name: "fetch", + description: "Async fetch tool", + inputSchema: {}, + execution: "async", + requiresHuman: false, +}; + +const askHumanDescriptor: z.infer = { + toolId: "tool.ask-human", + name: "ask-human", + description: "Ask the human", + inputSchema: {}, + execution: "async", + requiresHuman: true, +}; + +const defaultAgent: z.infer = { + agentId: "copilot", + systemPrompt: "SYS", + model: { provider: "fake", model: "m" }, + tools: [echoDescriptor, fetchDescriptor, askHumanDescriptor], +}; + +function user(text: string) { + return { role: "user" as const, content: text }; +} + +function assistantText(text: string) { + return { role: "assistant" as const, content: text }; +} + +function toolCallPart(id: string, name: string, args: JsonValue = {}) { + return { + type: "tool-call" as const, + toolCallId: id, + toolName: name, + arguments: args, + }; +} + +function assistantCalls(...parts: Array>) { + return { role: "assistant" as const, content: parts }; +} + +function completedResp( + message: Extract["message"], +): LlmStreamEvent { + return { + type: "completed", + message, + finishReason: "stop", + usage: { inputTokens: 10, outputTokens: 5, totalTokens: 15 }, + }; +} + +type ScriptedCall = ( + request: ModelStreamRequest, +) => AsyncGenerator; + +function respond(...events: LlmStreamEvent[]): ScriptedCall { + return async function* () { + yield* events; + }; +} + +function failCall(error: string): ScriptedCall { + // eslint-disable-next-line require-yield + return async function* () { + throw new Error(error); + }; +} + +function hangUntilAbort(onStarted?: () => void): ScriptedCall { + // eslint-disable-next-line require-yield + return async function* (request) { + onStarted?.(); + await new Promise((resolve) => { + if (request.signal.aborted) { + resolve(); + } else { + request.signal.addEventListener("abort", () => resolve(), { + once: true, + }); + } + }); + throw new Error("aborted"); + }; +} + +class FakeModelRegistry implements IModelRegistry { + requests: ModelStreamRequest[] = []; + private next = 0; + + constructor(private readonly calls: ScriptedCall[]) {} + + async resolve( + descriptor: ResolvedModel["descriptor"], + ): Promise { + return { + descriptor, + stream: (request) => { + this.requests.push(request); + const call = this.calls[this.next++]; + if (!call) { + throw new Error("no scripted model call remaining"); + } + return call(request); + }, + }; + } +} + +function syncTool( + descriptor: z.infer, + execute: SyncRuntimeTool["execute"], +): SyncRuntimeTool { + return { + descriptor: descriptor as SyncRuntimeTool["descriptor"], + execute, + }; +} + +const defaultTools: RuntimeTool[] = [ + syncTool(echoDescriptor, async (input) => ({ + output: { echoed: (input ?? null) as JsonValue }, + isError: false, + })), + { descriptor: fetchDescriptor as { execution: "async" } & typeof fetchDescriptor }, + { descriptor: askHumanDescriptor as { execution: "async" } & typeof askHumanDescriptor }, +]; + +class FakeToolRegistry implements IToolRegistry { + constructor(private readonly tools: RuntimeTool[]) {} + + async resolve( + descriptor: z.infer, + ): Promise { + const tool = this.tools.find( + (t) => t.descriptor.toolId === descriptor.toolId, + ); + if (!tool) { + throw new TurnDependencyError(`no live tool for ${descriptor.toolId}`); + } + return tool; + } +} + +type CheckerRule = "allow" | { request: JsonValue } | "throw"; + +class FakePermissionChecker implements IPermissionChecker { + calls: PermissionCheckInput[] = []; + + constructor(private readonly rules: Record = {}) {} + + async check(input: PermissionCheckInput) { + this.calls.push(input); + const rule = this.rules[input.toolName] ?? "allow"; + if (rule === "allow") { + return { required: false as const }; + } + if (rule === "throw") { + throw new Error("checker exploded"); + } + return { required: true as const, request: rule.request }; + } +} + +class FakePermissionClassifier implements IPermissionClassifier { + batches: PermissionClassificationInput[][] = []; + + constructor( + private readonly impl?: ( + requests: PermissionClassificationInput[], + ) => PermissionClassification[], + private readonly throws?: string, + ) {} + + async classify( + requests: PermissionClassificationInput[], + ): Promise { + this.batches.push(requests); + if (this.throws) { + throw new Error(this.throws); + } + if (!this.impl) { + throw new Error("classifier must not be called in this test"); + } + return this.impl(requests); + } +} + +class FakeAgentResolver implements IAgentResolver { + constructor( + private readonly agent: z.infer, + private readonly error?: string, + ) {} + + async resolve() { + if (this.error) { + throw new Error(this.error); + } + return this.agent; + } +} + +class FakeBus { + events: TurnLifecycleEvent[] = []; + publish(event: TurnLifecycleEvent): void { + this.events.push(event); + } +} + +class FakeIdGen { + private n = 0; + async next(): Promise { + this.n += 1; + return `2026-07-02T10-00-00Z-${String(this.n).padStart(7, "0")}-000`; + } +} + +class FakeClock { + now(): string { + return TS; + } +} + +function makeRuntime(opts: { + models?: ScriptedCall[]; + tools?: RuntimeTool[]; + checker?: FakePermissionChecker; + classifier?: FakePermissionClassifier; + agent?: z.infer; + agentError?: string; + repo?: InMemoryTurnRepo; +} = {}) { + const repo = opts.repo ?? new InMemoryTurnRepo(); + const models = new FakeModelRegistry(opts.models ?? []); + const checker = opts.checker ?? new FakePermissionChecker(); + const classifier = opts.classifier ?? new FakePermissionClassifier(); + const bus = new FakeBus(); + const runtime = new TurnRuntime({ + turnRepo: repo, + idGenerator: new FakeIdGen(), + clock: new FakeClock(), + agentResolver: new FakeAgentResolver(opts.agent ?? defaultAgent, opts.agentError), + modelRegistry: models, + toolRegistry: new FakeToolRegistry(opts.tools ?? defaultTools), + contextResolver: new TurnRepoContextResolver({ turnRepo: repo }), + permissionChecker: checker, + permissionClassifier: classifier, + bus, + }); + return { runtime, repo, models, checker, classifier, bus }; +} + +async function newTurn( + runtime: TurnRuntime, + overrides: Partial[0]> = {}, +): Promise { + return runtime.createTurn({ + agent: { agentId: "copilot" }, + context: [], + input: user("hello"), + config: { humanAvailable: true }, + ...overrides, + }); +} + +async function settle(execution: TurnExecution): Promise<{ + outcome?: TurnOutcome; + error?: unknown; + events: TurnStreamEvent[]; +}> { + const events: TurnStreamEvent[] = []; + const consumer = (async () => { + try { + for await (const event of execution.events) { + events.push(event); + } + } catch { + // Infrastructure failures also reject the outcome; tests assert there. + } + })(); + let outcome: TurnOutcome | undefined; + let error: unknown; + try { + outcome = await execution.outcome; + } catch (err) { + error = err; + } + await consumer; + return { outcome, error, events }; +} + +async function advanceAndSettle( + runtime: TurnRuntime, + turnId: string, + input?: Parameters[1], + options?: Parameters[2], +) { + return settle(runtime.advanceTurn(turnId, input, options)); +} + +function typesOf(events: Array<{ type: string }>): string[] { + return events.map((e) => e.type); +} + +async function persisted( + repo: InMemoryTurnRepo, + turnId: string, +): Promise>> { + return repo.read(turnId); +} + +// --------------------------------------------------------------------------- +// §26.1 Plain model response +// --------------------------------------------------------------------------- + +describe("plain model response (26.1)", () => { + it("runs one model step to completion with exact persisted request", async () => { + const { runtime, repo, bus } = makeRuntime({ + models: [ + respond( + { type: "text_delta", delta: "do" }, + { type: "step_event", event: { type: "text_end", text: "done" } }, + { type: "text_delta", delta: "ne" }, + completedResp(assistantText("done")), + ), + ], + }); + const turnId = await newTurn(runtime); + const { outcome, events } = await advanceAndSettle(runtime, turnId); + + expect(outcome).toMatchObject({ + status: "completed", + output: assistantText("done"), + finishReason: "stop", + }); + + const log = await persisted(repo, turnId); + expect(typesOf(log)).toEqual([ + "turn_created", + "model_call_requested", + "model_step_event", + "model_call_completed", + "turn_completed", + ]); + const request = log[1]; + expect(request).toMatchObject({ + modelCallIndex: 0, + request: { + systemPrompt: "SYS", + messages: [user("hello")], + }, + }); + // Deltas are streamed but never persisted. + expect(events.filter((e) => e.type === "text_delta")).toHaveLength(2); + expect(typesOf(log)).not.toContain("text_delta"); + // Terminal event duplicates output and aggregate usage. + const terminal = log[log.length - 1]; + expect(terminal).toMatchObject({ + type: "turn_completed", + output: assistantText("done"), + usage: { inputTokens: 10, outputTokens: 5, totalTokens: 15 }, + }); + // Lifecycle bus events are emitted but not persisted. + expect(bus.events.map((e) => e.type)).toEqual([ + "turn-processing-start", + "turn-processing-end", + ]); + }); + + it("streams durable events only after persistence, matching the file", async () => { + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(assistantText("done")))], + }); + const turnId = await newTurn(runtime); + const { events } = await advanceAndSettle(runtime, turnId); + const durable = events.filter( + (e) => e.type !== "text_delta" && e.type !== "reasoning_delta", + ); + const log = await persisted(repo, turnId); + expect(durable).toEqual(log.slice(1)); + }); + + it("outcome resolves without consuming the event stream", async () => { + const { runtime } = makeRuntime({ + models: [respond(completedResp(assistantText("done")))], + }); + const turnId = await newTurn(runtime); + const execution = runtime.advanceTurn(turnId); + const outcome = await execution.outcome; + expect(outcome.status).toBe("completed"); + }); +}); + +// --------------------------------------------------------------------------- +// §26.2 Mixed sync and async tools +// --------------------------------------------------------------------------- + +describe("mixed sync and async tools (26.2)", () => { + it("executes sync tools, suspends on async, and preserves source order in the next request", async () => { + const batch = assistantCalls( + toolCallPart("A", "echo", { i: 1 }), + toolCallPart("B", "fetch", { i: 2 }), + toolCallPart("C", "echo", { i: 3 }), + toolCallPart("D", "fetch", { i: 4 }), + ); + const { runtime, repo, models } = makeRuntime({ + models: [ + respond(completedResp(batch)), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime); + + const first = await advanceAndSettle(runtime, turnId); + expect(first.outcome).toMatchObject({ + status: "suspended", + pendingAsyncTools: [ + expect.objectContaining({ toolCallId: "B" }), + expect.objectContaining({ toolCallId: "D" }), + ], + }); + + // Async results arrive in reverse order. + const second = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "D", + result: { output: "d-result", isError: false }, + }); + expect(second.outcome?.status).toBe("suspended"); + + const third = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "B", + result: { output: "b-result", isError: false }, + }); + expect(third.outcome?.status).toBe("completed"); + + // The next model request carries results in original source order + // even though physical completion order was A, C, D, B. + const log = await persisted(repo, turnId); + const secondRequest = log.find( + (e) => e.type === "model_call_requested" && e.modelCallIndex === 1, + ); + expect(secondRequest).toBeDefined(); + const toolMessages = + secondRequest?.type === "model_call_requested" + ? secondRequest.request.messages.filter((m) => m.role === "tool") + : []; + expect(toolMessages.map((m) => (m.role === "tool" ? m.toolCallId : ""))).toEqual( + ["A", "B", "C", "D"], + ); + // The live model call saw the same ordering. + expect( + models.requests[1].messages + .filter((m) => m.role === "tool") + .map((m) => (m.role === "tool" ? m.toolCallId : "")), + ).toEqual(["A", "B", "C", "D"]); + }); + + it("rejects async results for calls that are not pending", async () => { + const { runtime } = makeRuntime({ + models: [respond(completedResp(assistantText("done")))], + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + const { error } = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "ghost", + result: { output: "x", isError: false }, + }); + expect(error).toBeInstanceOf(TurnInputError); + }); +}); + +// --------------------------------------------------------------------------- +// §26.3 Partial human permission decisions +// --------------------------------------------------------------------------- + +describe("partial human permission decisions (26.3)", () => { + async function setup() { + const batch = assistantCalls( + toolCallPart("P1", "echo"), + toolCallPart("P2", "echo"), + toolCallPart("F1", "fetch"), + ); + const fixture = makeRuntime({ + models: [ + respond(completedResp(batch)), + respond(completedResp(assistantText("done"))), + ], + checker: new FakePermissionChecker({ + echo: { request: { kind: "command" } }, + }), + }); + const turnId = await newTurn(fixture.runtime); + const first = await advanceAndSettle(fixture.runtime, turnId); + return { ...fixture, turnId, first }; + } + + it("records all required permissions and exposes allowed async tools in one suspension", async () => { + const { first } = await setup(); + expect(first.outcome).toMatchObject({ + status: "suspended", + pendingPermissions: [ + expect.objectContaining({ toolCallId: "P1" }), + expect.objectContaining({ toolCallId: "P2" }), + ], + pendingAsyncTools: [expect.objectContaining({ toolCallId: "F1" })], + }); + }); + + it("one approval advances only its tool; denial yields an error result; no model call until all settle", async () => { + const { runtime, repo, turnId } = await setup(); + + const afterAllow = await advanceAndSettle(runtime, turnId, { + type: "permission_decision", + toolCallId: "P1", + decision: "allow", + }); + expect(afterAllow.outcome).toMatchObject({ + status: "suspended", + pendingPermissions: [expect.objectContaining({ toolCallId: "P2" })], + }); + let log = await persisted(repo, turnId); + expect( + log.some( + (e) => e.type === "tool_result" && e.toolCallId === "P1" && e.source === "sync", + ), + ).toBe(true); + + // Async result may arrive while a permission is still unresolved. + const afterAsync = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "F1", + result: { output: "fetched", isError: false }, + }); + expect(afterAsync.outcome?.status).toBe("suspended"); + + const afterDeny = await advanceAndSettle(runtime, turnId, { + type: "permission_decision", + toolCallId: "P2", + decision: "deny", + }); + expect(afterDeny.outcome?.status).toBe("completed"); + + log = await persisted(repo, turnId); + const denial = log.find( + (e) => e.type === "tool_result" && e.toolCallId === "P2", + ); + expect(denial).toMatchObject({ + source: "runtime", + result: { isError: true }, + }); + // Exactly two model calls: the batch, then the follow-up. + expect(log.filter((e) => e.type === "model_call_requested")).toHaveLength(2); + }); + + it("rejects decisions for permissions that are not pending", async () => { + const { runtime, turnId } = await setup(); + await advanceAndSettle(runtime, turnId, { + type: "permission_decision", + toolCallId: "P1", + decision: "allow", + }); + const { error } = await advanceAndSettle(runtime, turnId, { + type: "permission_decision", + toolCallId: "P1", + decision: "deny", + }); + expect(error).toBeInstanceOf(TurnInputError); + }); +}); + +// --------------------------------------------------------------------------- +// §26.4 Automatic permission classification +// --------------------------------------------------------------------------- + +describe("automatic permission classification (26.4)", () => { + const batch = assistantCalls( + toolCallPart("CA", "echo"), + toolCallPart("CD", "echo"), + toolCallPart("CF", "echo"), + ); + const checkerRules = { echo: { request: { kind: "command" as const } } }; + const classifierImpl = (requests: PermissionClassificationInput[]) => + requests.map((r): PermissionClassification => { + const decision = + r.toolCallId === "CA" ? "allow" : r.toolCallId === "CD" ? "deny" : "defer"; + return { toolCallId: r.toolCallId, decision, reason: `because ${decision}` }; + }); + + it("handles allow, deny, and defer in one batch with a human available", async () => { + const { runtime, repo, classifier } = makeRuntime({ + models: [respond(completedResp(batch))], + checker: new FakePermissionChecker(checkerRules), + classifier: new FakePermissionClassifier(classifierImpl), + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, autoPermission: true }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + + // Deferred call asks the human. + expect(outcome).toMatchObject({ + status: "suspended", + pendingPermissions: [expect.objectContaining({ toolCallId: "CF" })], + }); + expect(classifier.batches).toHaveLength(1); + expect(classifier.batches[0].map((r) => r.toolCallId)).toEqual([ + "CA", + "CD", + "CF", + ]); + + const log = await persisted(repo, turnId); + // Classifier provenance and effective decisions are distinct records. + expect( + log.filter((e) => e.type === "tool_permission_classified"), + ).toHaveLength(3); + const resolved = log.filter((e) => e.type === "tool_permission_resolved"); + expect(resolved).toEqual([ + expect.objectContaining({ toolCallId: "CA", decision: "allow", source: "classifier" }), + expect.objectContaining({ toolCallId: "CD", decision: "deny", source: "classifier" }), + ]); + // Allow executed; deny got an error result without invocation. + expect( + log.some((e) => e.type === "tool_result" && e.toolCallId === "CA" && e.source === "sync"), + ).toBe(true); + expect( + log.some( + (e) => e.type === "tool_invocation_requested" && e.toolCallId === "CD", + ), + ).toBe(false); + }); + + it("denies deferred calls when no human is available", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond(completedResp(batch)), + respond(completedResp(assistantText("done"))), + ], + checker: new FakePermissionChecker(checkerRules), + classifier: new FakePermissionClassifier(classifierImpl), + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: false, autoPermission: true }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, turnId); + expect( + log.find( + (e) => e.type === "tool_permission_resolved" && e.toolCallId === "CF", + ), + ).toMatchObject({ decision: "deny", source: "human_unavailable" }); + }); + + it("classifier failure records the failure and defers to the human", async () => { + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(batch))], + checker: new FakePermissionChecker(checkerRules), + classifier: new FakePermissionClassifier(undefined, "classifier exploded"), + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, autoPermission: true }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome).toMatchObject({ status: "suspended" }); + expect( + outcome?.status === "suspended" + ? outcome.pendingPermissions.map((p) => p.toolCallId) + : [], + ).toEqual(["CA", "CD", "CF"]); + const log = await persisted(repo, turnId); + expect( + log.find((e) => e.type === "tool_permission_classification_failed"), + ).toMatchObject({ toolCallIds: ["CA", "CD", "CF"] }); + // A later advance does not re-classify failed calls. + const again = await advanceAndSettle(runtime, turnId); + expect(again.outcome?.status).toBe("suspended"); + }); + + it("missing decisions are recorded as per-call classification failures", async () => { + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(batch))], + checker: new FakePermissionChecker(checkerRules), + classifier: new FakePermissionClassifier((requests) => + requests + .filter((r) => r.toolCallId !== "CF") + .map((r) => ({ + toolCallId: r.toolCallId, + decision: "allow", + reason: "ok", + })), + ), + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, autoPermission: true }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome).toMatchObject({ + status: "suspended", + pendingPermissions: [expect.objectContaining({ toolCallId: "CF" })], + }); + const log = await persisted(repo, turnId); + expect( + log.find((e) => e.type === "tool_permission_classification_failed"), + ).toMatchObject({ toolCallIds: ["CF"] }); + }); + + it("manual mode never calls the classifier", async () => { + const classifier = new FakePermissionClassifier(); // throws if called + const { runtime } = makeRuntime({ + models: [respond(completedResp(batch))], + checker: new FakePermissionChecker(checkerRules), + classifier, + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, autoPermission: false }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("suspended"); + expect(classifier.batches).toHaveLength(0); + }); + + it("checker failure fails closed: recorded, routed to human, never auto-executed", async () => { + const { runtime, repo, classifier } = makeRuntime({ + models: [respond(completedResp(assistantCalls(toolCallPart("X", "echo"))))], + checker: new FakePermissionChecker({ echo: "throw" }), + classifier: new FakePermissionClassifier(() => [ + { toolCallId: "X", decision: "allow", reason: "should not matter" }, + ]), + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, autoPermission: true }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome).toMatchObject({ + status: "suspended", + pendingPermissions: [expect.objectContaining({ toolCallId: "X" })], + }); + // The classifier is bypassed for checker-error calls. + expect(classifier.batches).toHaveLength(0); + const log = await persisted(repo, turnId); + expect( + log.find((e) => e.type === "tool_permission_required"), + ).toMatchObject({ checkerError: "checker exploded" }); + expect(log.some((e) => e.type === "tool_invocation_requested")).toBe(false); + }); + + it("checker failure with no human denies without executing", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("X", "echo")))), + respond(completedResp(assistantText("done"))), + ], + checker: new FakePermissionChecker({ echo: "throw" }), + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: false, autoPermission: false }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, turnId); + expect( + log.find((e) => e.type === "tool_permission_resolved"), + ).toMatchObject({ decision: "deny", source: "human_unavailable" }); + expect(log.some((e) => e.type === "tool_invocation_requested")).toBe(false); + }); +}); + +// --------------------------------------------------------------------------- +// Human-dependent tools +// --------------------------------------------------------------------------- + +describe("human-dependent tools", () => { + it("ask-human suspends as a pending async tool when a human is available", async () => { + const { runtime } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("H", "ask-human")))), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime); + const first = await advanceAndSettle(runtime, turnId); + expect(first.outcome).toMatchObject({ + status: "suspended", + pendingAsyncTools: [expect.objectContaining({ toolCallId: "H" })], + }); + const second = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "H", + result: { output: "42", isError: false }, + }); + expect(second.outcome?.status).toBe("completed"); + }); + + it("requiresHuman tools get an immediate error result when no human is available", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("H", "ask-human")))), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: false }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, turnId); + expect( + log.some((e) => e.type === "tool_invocation_requested" && e.toolCallId === "H"), + ).toBe(true); + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "H")).toMatchObject({ + source: "runtime", + result: { isError: true }, + }); + }); +}); + +// --------------------------------------------------------------------------- +// §26.5 Cancellation +// --------------------------------------------------------------------------- + +describe("cancellation (26.5)", () => { + it("cancels a suspended turn via the cancel input", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("B", "fetch")))), + ], + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + + const { outcome } = await advanceAndSettle(runtime, turnId, { + type: "cancel", + reason: "user stop", + }); + expect(outcome).toMatchObject({ status: "cancelled", reason: "user stop" }); + + const log = await persisted(repo, turnId); + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "B")).toMatchObject({ + source: "runtime", + result: { isError: true }, + }); + expect(log[log.length - 1]).toMatchObject({ + type: "turn_cancelled", + reason: "user stop", + }); + + // Late external inputs are rejected. + const late = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "B", + result: { output: "late", isError: false }, + }); + expect(late.error).toBeInstanceOf(TurnInputError); + }); + + it("cancels during a model call via the abort signal", async () => { + const controller = new AbortController(); + let abortNow: (() => void) | undefined; + const started = new Promise((resolve) => { + abortNow = () => resolve(); + }); + const { runtime, repo } = makeRuntime({ + models: [hangUntilAbort(() => abortNow?.())], + }); + const turnId = await newTurn(runtime); + const execution = runtime.advanceTurn(turnId, undefined, { + signal: controller.signal, + }); + await started; + controller.abort(); + const { outcome } = await settle(execution); + expect(outcome?.status).toBe("cancelled"); + const log = await persisted(repo, turnId); + expect(typesOf(log)).toEqual([ + "turn_created", + "model_call_requested", + "model_call_failed", + "turn_cancelled", + ]); + }); + + it("cancels during a sync tool with a synthetic result", async () => { + const controller = new AbortController(); + const tools: RuntimeTool[] = [ + syncTool(echoDescriptor, async (_input, ctx: ToolExecutionContext) => { + controller.abort(); + if (!ctx.signal.aborted) { + await new Promise((resolve) => { + ctx.signal.addEventListener("abort", () => resolve(), { + once: true, + }); + }); + } + throw new Error("aborted mid-tool"); + }), + { descriptor: fetchDescriptor as { execution: "async" } & typeof fetchDescriptor }, + { descriptor: askHumanDescriptor as { execution: "async" } & typeof askHumanDescriptor }, + ]; + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(assistantCalls(toolCallPart("S", "echo"))))], + tools, + }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId, undefined, { + signal: controller.signal, + }); + expect(outcome?.status).toBe("cancelled"); + const log = await persisted(repo, turnId); + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "S")).toMatchObject({ + source: "runtime", + result: { isError: true }, + }); + expect(log[log.length - 1].type).toBe("turn_cancelled"); + }); +}); + +// --------------------------------------------------------------------------- +// §26.6 Failures +// --------------------------------------------------------------------------- + +describe("failures (26.6)", () => { + it("provider failure appends model and turn failures", async () => { + const { runtime, repo } = makeRuntime({ models: [failCall("boom")] }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome).toMatchObject({ status: "failed", error: "boom" }); + const log = await persisted(repo, turnId); + expect(typesOf(log)).toEqual([ + "turn_created", + "model_call_requested", + "model_call_failed", + "turn_failed", + ]); + }); + + it("a sync tool throw becomes an error result and the turn continues", async () => { + const tools: RuntimeTool[] = [ + syncTool(echoDescriptor, async () => { + throw new Error("tool exploded"); + }), + { descriptor: fetchDescriptor as { execution: "async" } & typeof fetchDescriptor }, + { descriptor: askHumanDescriptor as { execution: "async" } & typeof askHumanDescriptor }, + ]; + const { runtime, repo } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("T", "echo")))), + respond(completedResp(assistantText("recovered"))), + ], + tools, + }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome).toMatchObject({ + status: "completed", + output: assistantText("recovered"), + }); + const log = await persisted(repo, turnId); + expect(log.find((e) => e.type === "tool_result")).toMatchObject({ + source: "sync", + result: { output: "tool exploded", isError: true }, + }); + }); + + it("an async failure result becomes an error tool result and the turn continues", async () => { + const { runtime } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("B", "fetch")))), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + const { outcome } = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "B", + result: { output: "network unreachable", isError: true }, + }); + expect(outcome?.status).toBe("completed"); + }); + + it("unknown tools become runtime error results, not turn failures", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("U", "no-such-tool")))), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, turnId); + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "U")).toMatchObject({ + source: "runtime", + result: { isError: true }, + }); + }); + + it("repository errors reject stream and outcome without a false turn_failed", async () => { + const { runtime } = makeRuntime(); + const execution = runtime.advanceTurn("2026-07-02T10-00-00Z-9999999-000"); + await expect(execution.outcome).rejects.toThrowError(/turn not found/); + await expect( + (async () => { + for await (const event of execution.events) { + void event; // drain + } + })(), + ).rejects.toThrowError(/turn not found/); + }); + + it("missing live tools reject the execution and leave the turn unchanged", async () => { + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(assistantText("done")))], + tools: [], // registry knows nothing + }); + const turnId = await newTurn(runtime); + const { error } = await advanceAndSettle(runtime, turnId); + expect(error).toBeInstanceOf(TurnDependencyError); + expect((await persisted(repo, turnId)).length).toBe(1); // only turn_created + }); + + it("agent resolution failure rejects createTurn without creating a file", async () => { + const { runtime, repo } = makeRuntime({ agentError: "no such agent" }); + await expect(newTurn(runtime)).rejects.toThrowError("no such agent"); + await expect( + repo.read("2026-07-02T10-00-00Z-0000001-000"), + ).rejects.toThrowError(/turn not found/); + }); +}); + +// --------------------------------------------------------------------------- +// §26.7 Crash recovery +// --------------------------------------------------------------------------- + +describe("crash recovery (26.7)", () => { + const SEED_ID = "2026-07-02T10-00-00Z-0000001-000"; + + function seedCreated(config?: { + autoPermission: boolean; + humanAvailable: boolean; + maxModelCalls: number; + }): z.infer { + return { + type: "turn_created", + schemaVersion: 1, + turnId: SEED_ID, + ts: TS, + sessionId: null, + agent: { requested: { agentId: "copilot" }, resolved: defaultAgent }, + context: [], + input: user("hello"), + config: config ?? { + autoPermission: false, + humanAvailable: true, + maxModelCalls: 20, + }, + }; + } + + function seedRequested(index: number, messages: Array | ReturnType | { role: "assistant"; content: string } | { role: "tool"; content: string; toolCallId: string; toolName: string }>): z.infer { + return { + type: "model_call_requested", + turnId: SEED_ID, + ts: TS, + modelCallIndex: index, + request: { + systemPrompt: "SYS", + messages, + tools: defaultAgent.tools, + parameters: {}, + }, + }; + } + + function seedCompleted(index: number, message: Parameters[0]): z.infer { + return { + type: "model_call_completed", + turnId: SEED_ID, + ts: TS, + modelCallIndex: index, + message, + finishReason: "stop", + usage: {}, + }; + } + + it("created-only log initiates the first model call", async () => { + const repo = new InMemoryTurnRepo(); + repo.seed([seedCreated()]); + const { runtime } = makeRuntime({ + repo, + models: [respond(completedResp(assistantText("done")))], + }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome?.status).toBe("completed"); + }); + + it("an unmatched model request is closed as interrupted and re-issued", async () => { + const repo = new InMemoryTurnRepo(); + repo.seed([seedCreated(), seedRequested(0, [user("hello")])]); + const { runtime, models } = makeRuntime({ + repo, + models: [respond(completedResp(assistantText("done")))], + }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, SEED_ID); + expect(typesOf(log)).toEqual([ + "turn_created", + "model_call_requested", + "model_call_failed", + "model_call_requested", + "model_call_completed", + "turn_completed", + ]); + expect(log[2]).toMatchObject({ error: expect.stringMatching(/interrupted/) }); + expect(models.requests).toHaveLength(1); // only the re-issued call hit the provider + }); + + it("the re-issued call counts against the model-call budget", async () => { + const repo = new InMemoryTurnRepo(); + repo.seed([ + seedCreated({ + autoPermission: false, + humanAvailable: true, + maxModelCalls: 1, + }), + seedRequested(0, [user("hello")]), + ]); + const { runtime } = makeRuntime({ repo }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome).toMatchObject({ + status: "failed", + code: MODEL_CALL_LIMIT_ERROR_CODE, + }); + }); + + it("an unmatched sync invocation gets an indeterminate result and the turn continues", async () => { + const repo = new InMemoryTurnRepo(); + const batch = assistantCalls(toolCallPart("S", "echo")); + repo.seed([ + seedCreated(), + seedRequested(0, [user("hello")]), + seedCompleted(0, batch), + { + type: "tool_invocation_requested", + turnId: SEED_ID, + ts: TS, + toolCallId: "S", + toolId: "tool.echo", + toolName: "echo", + execution: "sync", + input: {}, + }, + ]); + const { runtime } = makeRuntime({ + repo, + models: [respond(completedResp(assistantText("done")))], + }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome?.status).toBe("completed"); + const log = await persisted(repo, SEED_ID); + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "S")).toMatchObject({ + source: "runtime", + result: { + output: expect.stringMatching(/interrupted; its outcome is unknown/), + isError: true, + }, + }); + // No turn_failed anywhere: the turn completed. + expect(typesOf(log)).not.toContain("turn_failed"); + }); + + it("an unmatched async invocation remains suspended, appending the missing snapshot", async () => { + const repo = new InMemoryTurnRepo(); + const batch = assistantCalls(toolCallPart("B", "fetch")); + repo.seed([ + seedCreated(), + seedRequested(0, [user("hello")]), + seedCompleted(0, batch), + { + type: "tool_invocation_requested", + turnId: SEED_ID, + ts: TS, + toolCallId: "B", + toolId: "tool.fetch", + toolName: "fetch", + execution: "async", + input: {}, + }, + ]); + const { runtime } = makeRuntime({ repo }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome).toMatchObject({ + status: "suspended", + pendingAsyncTools: [expect.objectContaining({ toolCallId: "B" })], + }); + const log = await persisted(repo, SEED_ID); + expect(log[log.length - 1].type).toBe("turn_suspended"); + }); + + it("re-advancing an already-suspended turn appends no duplicate snapshot", async () => { + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(assistantCalls(toolCallPart("B", "fetch"))))], + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + const before = (await persisted(repo, turnId)).length; + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("suspended"); + expect((await persisted(repo, turnId)).length).toBe(before); + }); + + it("a completed tool batch proceeds to the next model call", async () => { + const repo = new InMemoryTurnRepo(); + const batch = assistantCalls(toolCallPart("S", "echo")); + repo.seed([ + seedCreated(), + seedRequested(0, [user("hello")]), + seedCompleted(0, batch), + { + type: "tool_invocation_requested", + turnId: SEED_ID, + ts: TS, + toolCallId: "S", + toolId: "tool.echo", + toolName: "echo", + execution: "sync", + input: {}, + }, + { + type: "tool_result", + turnId: SEED_ID, + ts: TS, + toolCallId: "S", + toolName: "echo", + source: "sync", + result: { output: "ok", isError: false }, + }, + ]); + const { runtime, models } = makeRuntime({ + repo, + models: [respond(completedResp(assistantText("done")))], + }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome?.status).toBe("completed"); + expect( + models.requests[0].messages.filter((m) => m.role === "tool"), + ).toHaveLength(1); + }); + + it("a permission resolved allow before invocation safely invokes the tool", async () => { + const repo = new InMemoryTurnRepo(); + const batch = assistantCalls(toolCallPart("P", "echo")); + repo.seed([ + seedCreated(), + seedRequested(0, [user("hello")]), + seedCompleted(0, batch), + { + type: "tool_permission_required", + turnId: SEED_ID, + ts: TS, + toolCallId: "P", + toolName: "echo", + request: {}, + }, + { + type: "tool_permission_resolved", + turnId: SEED_ID, + ts: TS, + toolCallId: "P", + decision: "allow", + source: "human", + }, + ]); + const { runtime, repo: r } = makeRuntime({ + repo, + models: [respond(completedResp(assistantText("done")))], + }); + const { outcome } = await advanceAndSettle(runtime, SEED_ID); + expect(outcome?.status).toBe("completed"); + const log = await persisted(r, SEED_ID); + expect(log.some((e) => e.type === "tool_invocation_requested" && e.toolCallId === "P")).toBe(true); + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "P")).toMatchObject({ + source: "sync", + result: { isError: false }, + }); + }); + + it("a terminal turn returns its outcome and performs no writes", async () => { + const { runtime, repo } = makeRuntime({ + models: [respond(completedResp(assistantText("done")))], + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + const before = (await persisted(repo, turnId)).length; + + const again = await advanceAndSettle(runtime, turnId); + expect(again.outcome).toMatchObject({ + status: "completed", + output: assistantText("done"), + }); + expect((await persisted(repo, turnId)).length).toBe(before); + + const withInput = await advanceAndSettle(runtime, turnId, { + type: "cancel", + }); + expect(withInput.error).toBeInstanceOf(TurnInputError); + }); +}); + +// --------------------------------------------------------------------------- +// §26.8 Historical and live reconstruction +// --------------------------------------------------------------------------- + +describe("historical and live reconstruction (26.8)", () => { + it("getTurn is read-only and matches live durable events", async () => { + const { runtime, repo } = makeRuntime({ + models: [ + respond( + { type: "text_delta", delta: "d" }, + completedResp(assistantText("done")), + ), + ], + }); + const turnId = await newTurn(runtime); + const { events } = await advanceAndSettle(runtime, turnId); + + const before = (await persisted(repo, turnId)).length; + const turn = await runtime.getTurn(turnId); + expect((await persisted(repo, turnId)).length).toBe(before); + + const durable = events.filter( + (e) => e.type !== "text_delta" && e.type !== "reasoning_delta", + ); + expect(turn.events.slice(1)).toEqual(durable); + // Ephemeral deltas are absent after reload. + expect(turn.events.some((e) => (e.type as string) === "text_delta")).toBe(false); + }); +}); + +// --------------------------------------------------------------------------- +// §26.9 Model-call limit +// --------------------------------------------------------------------------- + +describe("model-call limit (26.9)", () => { + it("a tool-free response on the final allowed call completes normally", async () => { + const { runtime } = makeRuntime({ + models: [ + respond(completedResp(assistantCalls(toolCallPart("S", "echo")))), + respond(completedResp(assistantText("done"))), + ], + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, maxModelCalls: 2 }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome?.status).toBe("completed"); + }); + + it("tool calls on the final call are fully processed, then the turn fails with the limit code", async () => { + const { runtime, repo, models } = makeRuntime({ + models: [respond(completedResp(assistantCalls(toolCallPart("S", "echo"))))], + }); + const turnId = await newTurn(runtime, { + config: { humanAvailable: true, maxModelCalls: 1 }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId); + expect(outcome).toMatchObject({ + status: "failed", + code: MODEL_CALL_LIMIT_ERROR_CODE, + }); + const log = await persisted(repo, turnId); + // The batch was fully processed before failing… + expect(log.find((e) => e.type === "tool_result" && e.toolCallId === "S")).toMatchObject({ + result: { isError: false }, + }); + // …and no further model call was made. + expect(models.requests).toHaveLength(1); + expect(log[log.length - 1]).toMatchObject({ + type: "turn_failed", + code: MODEL_CALL_LIMIT_ERROR_CODE, + }); + }); +}); + +// --------------------------------------------------------------------------- +// Tool progress +// --------------------------------------------------------------------------- + +describe("tool progress", () => { + it("sync progress persists before the callback resolves; async progress arrives as input", async () => { + const tools: RuntimeTool[] = [ + syncTool(echoDescriptor, async (_input, ctx) => { + await ctx.reportProgress({ pct: 50 }); + return { output: "done", isError: false }; + }), + { descriptor: fetchDescriptor as { execution: "async" } & typeof fetchDescriptor }, + { descriptor: askHumanDescriptor as { execution: "async" } & typeof askHumanDescriptor }, + ]; + const { runtime, repo } = makeRuntime({ + models: [ + respond( + completedResp( + assistantCalls(toolCallPart("S", "echo"), toolCallPart("B", "fetch")), + ), + ), + respond(completedResp(assistantText("done"))), + ], + tools, + }); + const turnId = await newTurn(runtime); + await advanceAndSettle(runtime, turnId); + + await advanceAndSettle(runtime, turnId, { + type: "async_tool_progress", + toolCallId: "B", + progress: { pct: 10 }, + }); + const { outcome } = await advanceAndSettle(runtime, turnId, { + type: "async_tool_result", + toolCallId: "B", + result: { output: "fetched", isError: false }, + }); + expect(outcome?.status).toBe("completed"); + + const log = await persisted(repo, turnId); + const progress = log.filter((e) => e.type === "tool_progress"); + expect(progress).toEqual([ + expect.objectContaining({ toolCallId: "S", source: "sync" }), + expect.objectContaining({ toolCallId: "B", source: "async" }), + ]); + }); +}); diff --git a/apps/x/packages/core/src/turns/runtime.ts b/apps/x/packages/core/src/turns/runtime.ts new file mode 100644 index 00000000..4a532a6e --- /dev/null +++ b/apps/x/packages/core/src/turns/runtime.ts @@ -0,0 +1,928 @@ +import type { z } from "zod"; +import { + DEFAULT_MAX_MODEL_CALLS, + MODEL_CALL_LIMIT_ERROR_CODE, + type JsonValue, + type ModelCallFailed, + type ModelRequest, + type ToolCallState, + type ToolInvocationRequested, + ToolResultData, + type ToolDescriptor, + type ToolPermissionRequired, + type ToolPermissionResolved, + type ToolResult, + TurnCreated, + type TurnEvent, + type TurnState, + type TurnStreamEvent, + type TurnSuspended, + outstandingAsyncTools, + outstandingPermissions, + reduceTurn, + turnTranscript, +} from "@x/shared/dist/turns.js"; +import type { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js"; +import type { IAgentResolver } from "./agent-resolver.js"; +import { + type CreateTurnInput, + type ITurnRuntime, + type Turn, + TurnDependencyError, + type TurnExecution, + type TurnExternalInput, + TurnInputError, + type TurnOutcome, +} from "./api.js"; +import type { ITurnLifecycleBus } from "./bus.js"; +import type { IClock } from "./clock.js"; +import type { IContextResolver } from "./context-resolver.js"; +import type { IModelRegistry, LlmStreamEvent } from "./model-registry.js"; +import type { IPermissionChecker, IPermissionClassifier } from "./permission.js"; +import type { ITurnRepo } from "./repo.js"; +import { HotStream } from "./stream.js"; +import type { IToolRegistry, RuntimeTool, SyncRuntimeTool } from "./tool-registry.js"; + +type TEvent = z.infer; + +const INTERRUPTED_TOOL_MESSAGE = + "Tool execution was interrupted; its outcome is unknown and it was not retried."; + +export interface TurnRuntimeDependencies { + turnRepo: ITurnRepo; + idGenerator: IMonotonicallyIncreasingIdGenerator; + clock: IClock; + agentResolver: IAgentResolver; + modelRegistry: IModelRegistry; + toolRegistry: IToolRegistry; + contextResolver: IContextResolver; + permissionChecker: IPermissionChecker; + permissionClassifier: IPermissionClassifier; + bus: ITurnLifecycleBus; +} + +// Immutable dependency container: holds no mutable per-turn state. All active +// turn state is reconstructed from the repository inside each invocation. +export class TurnRuntime implements ITurnRuntime { + private readonly turnRepo: ITurnRepo; + private readonly idGenerator: IMonotonicallyIncreasingIdGenerator; + private readonly clock: IClock; + private readonly agentResolver: IAgentResolver; + private readonly modelRegistry: IModelRegistry; + private readonly toolRegistry: IToolRegistry; + private readonly contextResolver: IContextResolver; + private readonly permissionChecker: IPermissionChecker; + private readonly permissionClassifier: IPermissionClassifier; + private readonly bus: ITurnLifecycleBus; + + constructor({ + turnRepo, + idGenerator, + clock, + agentResolver, + modelRegistry, + toolRegistry, + contextResolver, + permissionChecker, + permissionClassifier, + bus, + }: TurnRuntimeDependencies) { + this.turnRepo = turnRepo; + this.idGenerator = idGenerator; + this.clock = clock; + this.agentResolver = agentResolver; + this.modelRegistry = modelRegistry; + this.toolRegistry = toolRegistry; + this.contextResolver = contextResolver; + this.permissionChecker = permissionChecker; + this.permissionClassifier = permissionClassifier; + this.bus = bus; + } + + async createTurn(input: CreateTurnInput): Promise { + const resolved = await this.agentResolver.resolve(input.agent); + const turnId = await this.idGenerator.next(); + const event = TurnCreated.parse({ + type: "turn_created", + schemaVersion: 1, + turnId, + ts: this.clock.now(), + sessionId: input.sessionId ?? null, + agent: { requested: input.agent, resolved }, + context: input.context, + input: input.input, + config: { + autoPermission: input.config.autoPermission ?? false, + humanAvailable: input.config.humanAvailable, + maxModelCalls: input.config.maxModelCalls ?? DEFAULT_MAX_MODEL_CALLS, + }, + }); + await this.turnRepo.create(event); + return turnId; + } + + async getTurn(turnId: string): Promise { + const events = await this.turnRepo.read(turnId); + return { turnId, events }; + } + + advanceTurn( + turnId: string, + input?: TurnExternalInput, + options?: { signal?: AbortSignal }, + ): TurnExecution { + const stream = new HotStream(); + void this.turnRepo + .withLock(turnId, () => + this.advanceLocked(turnId, input, options?.signal, stream), + ) + .then( + (outcome) => stream.end(outcome), + (error: unknown) => stream.fail(error), + ); + return { events: stream.events, outcome: stream.outcome }; + } + + private async advanceLocked( + turnId: string, + input: TurnExternalInput | undefined, + externalSignal: AbortSignal | undefined, + stream: HotStream, + ): Promise { + this.bus.publish({ type: "turn-processing-start", turnId }); + try { + return await this.advance(turnId, input, externalSignal, stream); + } finally { + this.bus.publish({ type: "turn-processing-end", turnId }); + } + } + + private async advance( + turnId: string, + input: TurnExternalInput | undefined, + externalSignal: AbortSignal | undefined, + stream: HotStream, + ): Promise { + const events = await this.turnRepo.read(turnId); + let state = reduceTurn(events); + + if (state.terminal) { + if (input) { + throw new TurnInputError( + `turn ${turnId} is terminal; input rejected`, + ); + } + return outcomeFromTerminal(state); + } + + const definition = state.definition; + + // Materialize context and live dependencies. Failures here are + // infrastructure errors: the execution rejects, the turn is unchanged. + const resolvedContext = await this.contextResolver.resolve( + definition.context, + ); + const model = await this.modelRegistry.resolve( + definition.agent.resolved.model, + ); + const toolsByName = new Map(); + for (const descriptor of definition.agent.resolved.tools) { + const tool = await this.toolRegistry.resolve(descriptor); + if ( + tool.descriptor.toolId !== descriptor.toolId || + tool.descriptor.execution !== descriptor.execution + ) { + throw new TurnDependencyError( + `resolved tool ${descriptor.toolId} does not match its persisted descriptor`, + ); + } + toolsByName.set(descriptor.name, tool); + } + + const controller = new AbortController(); + const forwardAbort = () => controller.abort(); + if (externalSignal) { + if (externalSignal.aborted) { + controller.abort(); + } else { + externalSignal.addEventListener("abort", forwardAbort, { + once: true, + }); + } + } + + let appended = false; + const append = async (...batch: TEvent[]): Promise => { + await this.turnRepo.append(turnId, batch); + events.push(...batch); + state = reduceTurn(events); + appended = true; + for (const event of batch) { + stream.push(event); + } + }; + const now = () => this.clock.now(); + // Checker "allowed" outcomes are deliberately not durable: after a + // crash the checker is simply re-consulted. + const checkerAllowed = new Set(); + let cancelReason: string | undefined; + + const cancelTurn = async (): Promise => { + const open = state.modelCalls.find( + (c) => c.response === undefined && c.error === undefined, + ); + if (open) { + await append( + modelCallFailedEvent(turnId, now(), open.index, "model call was cancelled"), + ); + } + for (const tc of state.toolCalls.filter((t) => !t.result)) { + await append( + runtimeResultEvent(turnId, now(), tc, { + output: "Tool call was cancelled before completion.", + isError: true, + }), + ); + } + await append({ + type: "turn_cancelled", + turnId, + ts: now(), + ...(cancelReason === undefined ? {} : { reason: cancelReason }), + usage: state.usage, + }); + return { + status: "cancelled", + ...(cancelReason === undefined ? {} : { reason: cancelReason }), + usage: state.usage, + }; + }; + + try { + // Apply the optional single external input against durable + // pending state. + if (input) { + switch (input.type) { + case "cancel": + cancelReason = input.reason; + return await cancelTurn(); + case "permission_decision": { + const tc = state.toolCalls.find( + (t) => t.toolCallId === input.toolCallId, + ); + if (!tc?.permission || tc.permission.resolved || tc.result) { + throw new TurnInputError( + `no pending permission for tool call ${input.toolCallId}`, + ); + } + await append({ + type: "tool_permission_resolved", + turnId, + ts: now(), + toolCallId: input.toolCallId, + decision: input.decision, + source: "human", + ...(input.metadata === undefined + ? {} + : { metadata: input.metadata }), + }); + if (input.decision === "deny") { + await append( + runtimeResultEvent(turnId, now(), tc, { + output: "Permission denied by user.", + isError: true, + }), + ); + } + break; + } + case "async_tool_progress": { + const tc = requirePendingAsync(state, input.toolCallId); + await append({ + type: "tool_progress", + turnId, + ts: now(), + toolCallId: tc.toolCallId, + source: "async", + progress: input.progress, + }); + break; + } + case "async_tool_result": { + const tc = requirePendingAsync(state, input.toolCallId); + await append({ + type: "tool_result", + turnId, + ts: now(), + toolCallId: tc.toolCallId, + toolName: tc.toolName, + source: "async", + result: input.result, + }); + break; + } + } + } + + for (;;) { + if (controller.signal.aborted) { + return await cancelTurn(); + } + + // Recovery: close a model call interrupted by a crash, then + // re-issue it as a new call (counts against the budget). + const open = state.modelCalls.find( + (c) => c.response === undefined && c.error === undefined, + ); + if (open) { + await append( + modelCallFailedEvent( + turnId, + now(), + open.index, + "model call was interrupted before a response was recorded", + ), + ); + continue; + } + + // Recovery: close sync invocations interrupted by a crash + // with an indeterminate result; the turn continues. + const interruptedSync = state.toolCalls.filter( + (tc) => tc.invocation && tc.execution === "sync" && !tc.result, + ); + if (interruptedSync.length > 0) { + for (const tc of interruptedSync) { + await append( + runtimeResultEvent(turnId, now(), tc, { + output: INTERRUPTED_TOOL_MESSAGE, + isError: true, + }), + ); + } + continue; + } + + // Permission requirements for freshly extracted tool calls. + const fresh = state.toolCalls.filter( + (tc) => + !tc.result && + !tc.invocation && + !tc.permission && + !checkerAllowed.has(tc.toolCallId), + ); + for (const tc of fresh) { + if (controller.signal.aborted) { + break; + } + const tool = toolsByName.get(tc.toolName); + if (!tool) { + await append( + runtimeResultEvent(turnId, now(), tc, { + output: `Unknown tool: ${tc.toolName}`, + isError: true, + }), + ); + continue; + } + if ( + tool.descriptor.requiresHuman && + !definition.config.humanAvailable + ) { + await append( + invocationEvent(turnId, now(), tc, tool.descriptor), + ); + await append( + runtimeResultEvent(turnId, now(), tc, { + output: "Human input is unavailable for this turn.", + isError: true, + }), + ); + continue; + } + try { + const check = await this.permissionChecker.check({ + turnId, + toolCallId: tc.toolCallId, + toolId: tool.descriptor.toolId, + toolName: tc.toolName, + input: tc.input, + }); + if (!check.required) { + checkerAllowed.add(tc.toolCallId); + } else { + await append( + permissionRequiredEvent( + turnId, + now(), + tc, + check.request, + ), + ); + } + } catch (error) { + // Checker failure fails closed: record it and route + // to a human (or denial below); never execute. + await append( + permissionRequiredEvent(turnId, now(), tc, {}, errorMessage(error)), + ); + } + } + if (controller.signal.aborted) { + return await cancelTurn(); + } + + // Automatic classification, one batch. Checker-error calls + // and previously failed classifications go straight to the + // human/deny fallback. + if (definition.config.autoPermission) { + const candidates = state.toolCalls.filter( + (tc) => + tc.permission && + !tc.permission.resolved && + !tc.permission.classification && + !tc.permission.classificationFailed && + tc.permission.required.checkerError === undefined && + !tc.result, + ); + if (candidates.length > 0) { + try { + const decisions = await this.permissionClassifier.classify( + candidates.map((tc) => ({ + toolCallId: tc.toolCallId, + toolName: tc.toolName, + input: tc.input, + request: tc.permission!.required.request, + })), + controller.signal, + ); + for (const tc of candidates) { + const decision = decisions.find( + (d) => d.toolCallId === tc.toolCallId, + ); + if (!decision) { + await append({ + type: "tool_permission_classification_failed", + turnId, + ts: now(), + toolCallIds: [tc.toolCallId], + error: "classifier returned no decision", + }); + continue; + } + await append({ + type: "tool_permission_classified", + turnId, + ts: now(), + toolCallId: tc.toolCallId, + decision: decision.decision, + reason: decision.reason, + }); + if (decision.decision === "allow") { + await append( + resolvedEvent(turnId, now(), tc.toolCallId, "allow", "classifier", decision.reason), + ); + } else if (decision.decision === "deny") { + await append( + resolvedEvent(turnId, now(), tc.toolCallId, "deny", "classifier", decision.reason), + ); + await append( + runtimeResultEvent(turnId, now(), tc, { + output: `Permission denied: ${decision.reason}`, + isError: true, + }), + ); + } + // "defer" falls through to human/deny fallback. + } + } catch (error) { + if (controller.signal.aborted) { + return await cancelTurn(); + } + await append({ + type: "tool_permission_classification_failed", + turnId, + ts: now(), + toolCallIds: candidates.map((c) => c.toolCallId), + error: errorMessage(error), + }); + } + } + } + + // No human available: deny whatever remains unresolved. + if (!definition.config.humanAvailable) { + const unresolved = state.toolCalls.filter( + (tc) => tc.permission && !tc.permission.resolved && !tc.result, + ); + for (const tc of unresolved) { + await append( + resolvedEvent(turnId, now(), tc.toolCallId, "deny", "human_unavailable"), + ); + await append( + runtimeResultEvent(turnId, now(), tc, { + output: "Permission denied: no human is available for this turn.", + isError: true, + }), + ); + } + } + + // Execute allowed sync tools sequentially; expose allowed + // async tools. Source order. + const executable = state.toolCalls.filter( + (tc) => + !tc.result && + !tc.invocation && + (checkerAllowed.has(tc.toolCallId) || + tc.permission?.resolved?.decision === "allow"), + ); + for (const tc of executable) { + if (controller.signal.aborted) { + break; + } + const tool = toolsByName.get(tc.toolName); + if (!tool) { + await append( + runtimeResultEvent(turnId, now(), tc, { + output: `Unknown tool: ${tc.toolName}`, + isError: true, + }), + ); + continue; + } + await append(invocationEvent(turnId, now(), tc, tool.descriptor)); + if (tool.descriptor.execution === "async") { + continue; + } + const syncTool = tool as SyncRuntimeTool; + try { + const result = await syncTool.execute(tc.input, { + signal: controller.signal, + reportProgress: async (progress) => { + await append({ + type: "tool_progress", + turnId, + ts: now(), + toolCallId: tc.toolCallId, + source: "sync", + progress, + }); + }, + }); + await append({ + type: "tool_result", + turnId, + ts: now(), + toolCallId: tc.toolCallId, + toolName: tc.toolName, + source: "sync", + result: ToolResultData.parse(result), + }); + } catch (error) { + if (controller.signal.aborted) { + await append( + runtimeResultEvent(turnId, now(), tc, { + output: "Tool execution was cancelled.", + isError: true, + }), + ); + break; + } + // A tool failure is conversational, not terminal. + await append({ + type: "tool_result", + turnId, + ts: now(), + toolCallId: tc.toolCallId, + toolName: tc.toolName, + source: "sync", + result: { output: errorMessage(error), isError: true }, + }); + } + } + if (controller.signal.aborted) { + return await cancelTurn(); + } + + // Suspend while external work remains outstanding. + const pendingPerms = outstandingPermissions(state); + const pendingAsync = outstandingAsyncTools(state); + if (pendingPerms.length + pendingAsync.length > 0) { + const last = events[events.length - 1]; + if (appended || last.type !== "turn_suspended") { + await append({ + type: "turn_suspended", + turnId, + ts: now(), + pendingPermissions: permissionsSnapshot(pendingPerms), + pendingAsyncTools: asyncSnapshot(pendingAsync), + usage: state.usage, + }); + } + return { + status: "suspended", + pendingPermissions: permissionsSnapshot(pendingPerms), + pendingAsyncTools: asyncSnapshot(pendingAsync), + usage: state.usage, + }; + } + + // Tool batch complete. Completion, budget, or the next call. + const lastCall = state.modelCalls[state.modelCalls.length - 1]; + if ( + lastCall?.response !== undefined && + state.toolCalls.every((tc) => tc.modelCallIndex !== lastCall.index) + ) { + const output = lastCall.response; + const finishReason = lastCall.finishReason ?? "unknown"; + await append({ + type: "turn_completed", + turnId, + ts: now(), + output, + finishReason, + usage: state.usage, + }); + return { + status: "completed", + output, + finishReason, + usage: state.usage, + }; + } + + if (state.modelCalls.length >= definition.config.maxModelCalls) { + const error = `Model call limit of ${definition.config.maxModelCalls} reached before the turn completed.`; + await append({ + type: "turn_failed", + turnId, + ts: now(), + error, + code: MODEL_CALL_LIMIT_ERROR_CODE, + usage: state.usage, + }); + return { + status: "failed", + error, + code: MODEL_CALL_LIMIT_ERROR_CODE, + usage: state.usage, + }; + } + + // One model step. The durable request barrier precedes the + // provider call; step events persist before the next provider + // stream read; deltas bypass storage. + const index = state.modelCalls.length; + const transcript = turnTranscript(state); + const isRef = !Array.isArray(definition.context); + const request: z.infer = { + systemPrompt: definition.agent.resolved.systemPrompt, + ...(isRef ? { contextRef: definition.context as { previousTurnId: string } } : {}), + messages: isRef + ? transcript + : [...(definition.context as typeof transcript), ...transcript], + tools: definition.agent.resolved.tools, + parameters: {}, + }; + await append({ + type: "model_call_requested", + turnId, + ts: now(), + modelCallIndex: index, + request, + }); + + let completion: Extract< + LlmStreamEvent, + { type: "completed" } + > | null = null; + try { + for await (const event of model.stream({ + systemPrompt: request.systemPrompt, + messages: [...resolvedContext, ...transcript], + tools: request.tools, + parameters: request.parameters, + signal: controller.signal, + })) { + switch (event.type) { + case "text_delta": + stream.push({ + type: "text_delta", + turnId, + modelCallIndex: index, + delta: event.delta, + }); + break; + case "reasoning_delta": + stream.push({ + type: "reasoning_delta", + turnId, + modelCallIndex: index, + delta: event.delta, + }); + break; + case "step_event": + await append({ + type: "model_step_event", + turnId, + ts: now(), + modelCallIndex: index, + event: event.event, + }); + break; + case "completed": + completion = event; + break; + } + } + if (!completion) { + throw new Error( + "model stream ended without a completed response", + ); + } + } catch (error) { + if (controller.signal.aborted) { + await append( + modelCallFailedEvent(turnId, now(), index, "model call was cancelled"), + ); + return await cancelTurn(); + } + const message = errorMessage(error); + await append(modelCallFailedEvent(turnId, now(), index, message)); + await append({ + type: "turn_failed", + turnId, + ts: now(), + error: message, + usage: state.usage, + }); + return { status: "failed", error: message, usage: state.usage }; + } + + await append({ + type: "model_call_completed", + turnId, + ts: now(), + modelCallIndex: index, + message: completion.message, + finishReason: completion.finishReason, + usage: completion.usage, + ...(completion.providerMetadata === undefined + ? {} + : { providerMetadata: completion.providerMetadata }), + }); + } + } finally { + if (externalSignal) { + externalSignal.removeEventListener("abort", forwardAbort); + } + } + } +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function outcomeFromTerminal(state: TurnState): TurnOutcome { + const terminal = state.terminal; + if (!terminal) { + throw new Error("turn is not terminal"); + } + switch (terminal.type) { + case "turn_completed": + return { + status: "completed", + output: terminal.output, + finishReason: terminal.finishReason, + usage: terminal.usage, + }; + case "turn_failed": + return { + status: "failed", + error: terminal.error, + ...(terminal.code === undefined ? {} : { code: terminal.code }), + usage: terminal.usage, + }; + case "turn_cancelled": + return { + status: "cancelled", + ...(terminal.reason === undefined ? {} : { reason: terminal.reason }), + usage: terminal.usage, + }; + } +} + +function requirePendingAsync(state: TurnState, toolCallId: string): ToolCallState { + const tc = state.toolCalls.find((t) => t.toolCallId === toolCallId); + if (!tc?.invocation || tc.execution !== "async" || tc.result) { + throw new TurnInputError( + `no pending async tool call ${toolCallId}`, + ); + } + return tc; +} + +function permissionsSnapshot( + calls: ToolCallState[], +): z.infer["pendingPermissions"] { + return calls.map((tc) => ({ + toolCallId: tc.toolCallId, + toolName: tc.toolName, + request: (tc.permission as NonNullable) + .required.request, + })); +} + +function asyncSnapshot( + calls: ToolCallState[], +): z.infer["pendingAsyncTools"] { + return calls.map((tc) => ({ + toolCallId: tc.toolCallId, + toolId: (tc.invocation as z.infer).toolId, + toolName: tc.toolName, + input: (tc.invocation as z.infer).input, + })); +} + +function modelCallFailedEvent( + turnId: string, + ts: string, + modelCallIndex: number, + error: string, +): z.infer { + return { type: "model_call_failed", turnId, ts, modelCallIndex, error }; +} + +function runtimeResultEvent( + turnId: string, + ts: string, + tc: ToolCallState, + result: { output: JsonValue; isError: boolean }, +): z.infer { + return { + type: "tool_result", + turnId, + ts, + toolCallId: tc.toolCallId, + toolName: tc.toolName, + source: "runtime", + result, + }; +} + +function permissionRequiredEvent( + turnId: string, + ts: string, + tc: ToolCallState, + request: JsonValue, + checkerError?: string, +): z.infer { + return { + type: "tool_permission_required", + turnId, + ts, + toolCallId: tc.toolCallId, + toolName: tc.toolName, + request, + ...(checkerError === undefined ? {} : { checkerError }), + }; +} + +function resolvedEvent( + turnId: string, + ts: string, + toolCallId: string, + decision: "allow" | "deny", + source: z.infer["source"], + reason?: string, +): z.infer { + return { + type: "tool_permission_resolved", + turnId, + ts, + toolCallId, + decision, + source, + ...(reason === undefined ? {} : { reason }), + }; +} + +function invocationEvent( + turnId: string, + ts: string, + tc: ToolCallState, + descriptor: z.infer, +): z.infer { + return { + type: "tool_invocation_requested", + turnId, + ts, + toolCallId: tc.toolCallId, + toolId: descriptor.toolId, + toolName: tc.toolName, + execution: descriptor.execution, + input: (tc.input ?? null) as JsonValue, + }; +} diff --git a/apps/x/packages/core/src/turns/stream.test.ts b/apps/x/packages/core/src/turns/stream.test.ts new file mode 100644 index 00000000..f2b35614 --- /dev/null +++ b/apps/x/packages/core/src/turns/stream.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, it } from "vitest"; +import { HotStream } from "./stream.js"; + +async function collect(iterable: AsyncIterable): Promise { + const out: T[] = []; + for await (const item of iterable) { + out.push(item); + } + return out; +} + +describe("HotStream", () => { + it("buffers events pushed before a consumer attaches", async () => { + const stream = new HotStream(); + stream.push(1); + stream.push(2); + stream.end("done"); + expect(await collect(stream.events)).toEqual([1, 2]); + expect(await stream.outcome).toBe("done"); + }); + + it("outcome resolves without draining events", async () => { + const stream = new HotStream(); + stream.push(1); + stream.push(2); + stream.end("done"); + expect(await stream.outcome).toBe("done"); + }); + + it("delivers events in order across await boundaries", async () => { + const stream = new HotStream(); + const consumer = collect(stream.events); + stream.push(1); + await Promise.resolve(); + stream.push(2); + stream.push(3); + stream.end("done"); + expect(await consumer).toEqual([1, 2, 3]); + }); + + it("closing the consumer drops future events without affecting outcome", async () => { + const stream = new HotStream(); + stream.push(1); + stream.push(2); + for await (const event of stream.events) { + expect(event).toBe(1); + break; // closes the iterator + } + stream.push(3); // dropped + stream.end("done"); + expect(await collect(stream.events)).toEqual([]); + expect(await stream.outcome).toBe("done"); + }); + + it("failure drains queued events, then throws, and rejects outcome with the same error", async () => { + const stream = new HotStream(); + const boom = new Error("boom"); + stream.push(1); + stream.fail(boom); + const seen: number[] = []; + await expect( + (async () => { + for await (const event of stream.events) { + seen.push(event); + } + })(), + ).rejects.toBe(boom); + expect(seen).toEqual([1]); + await expect(stream.outcome).rejects.toBe(boom); + }); + + it("ignores pushes and settlement after completion", async () => { + const stream = new HotStream(); + stream.end("first"); + stream.push(9); + stream.end("second"); + stream.fail(new Error("late")); + expect(await stream.outcome).toBe("first"); + expect(await collect(stream.events)).toEqual([]); + }); + + it("a waiting consumer wakes on end", async () => { + const stream = new HotStream(); + const consumer = collect(stream.events); + await Promise.resolve(); + stream.end("done"); + expect(await consumer).toEqual([]); + }); +}); diff --git a/apps/x/packages/core/src/turns/stream.ts b/apps/x/packages/core/src/turns/stream.ts new file mode 100644 index 00000000..9092b454 --- /dev/null +++ b/apps/x/packages/core/src/turns/stream.ts @@ -0,0 +1,94 @@ +// Hot execution stream (turn-runtime-design.md §16). Execution starts +// independently of event consumption; events buffer in an unbounded in-memory +// queue until the single assumed consumer attaches. If the consumer closes, +// subsequent events are dropped; closing never cancels execution. On +// infrastructure failure, iteration drains already-queued events and then +// throws, and the outcome rejects with the same error. +export class HotStream { + private queue: TEvent[] = []; + private waiters: Array<() => void> = []; + private done = false; + private failure: { error: unknown } | null = null; + private consumerClosed = false; + + readonly outcome: Promise; + private resolveOutcome!: (outcome: TOutcome) => void; + private rejectOutcome!: (error: unknown) => void; + + constructor() { + this.outcome = new Promise((resolve, reject) => { + this.resolveOutcome = resolve; + this.rejectOutcome = reject; + }); + // The outcome may legitimately never be awaited (fire-and-forget + // callers); don't surface unhandled rejections for it. + this.outcome.catch(() => undefined); + } + + push(event: TEvent): void { + if (this.done || this.consumerClosed) { + return; + } + this.queue.push(event); + this.wake(); + } + + end(outcome: TOutcome): void { + if (this.done) { + return; + } + this.done = true; + this.resolveOutcome(outcome); + this.wake(); + } + + fail(error: unknown): void { + if (this.done) { + return; + } + this.done = true; + this.failure = { error }; + this.rejectOutcome(error); + this.wake(); + } + + private wake(): void { + const waiters = this.waiters; + this.waiters = []; + for (const waiter of waiters) { + waiter(); + } + } + + get events(): AsyncIterable { + return { + [Symbol.asyncIterator]: (): AsyncIterator => ({ + next: async (): Promise> => { + for (;;) { + if (this.consumerClosed) { + return { value: undefined, done: true }; + } + const event = this.queue.shift(); + if (event !== undefined) { + return { value: event, done: false }; + } + if (this.done) { + if (this.failure) { + throw this.failure.error; + } + return { value: undefined, done: true }; + } + await new Promise((resolve) => + this.waiters.push(resolve), + ); + } + }, + return: async (): Promise> => { + this.consumerClosed = true; + this.queue.length = 0; + return { value: undefined, done: true }; + }, + }), + }; + } +} diff --git a/apps/x/packages/core/src/turns/tool-registry.ts b/apps/x/packages/core/src/turns/tool-registry.ts new file mode 100644 index 00000000..82670e72 --- /dev/null +++ b/apps/x/packages/core/src/turns/tool-registry.ts @@ -0,0 +1,35 @@ +import type { z } from "zod"; +import type { + JsonValue, + ToolDescriptor, + ToolResultData, +} from "@x/shared/dist/turns.js"; + +export interface ToolExecutionContext { + signal: AbortSignal; + // The loop appends a durable tool_progress event before resolving. + reportProgress(progress: JsonValue): Promise; +} + +export interface SyncRuntimeTool { + descriptor: z.infer & { execution: "sync" }; + execute( + input: unknown, + context: ToolExecutionContext, + ): Promise>; +} + +// An async tool has no in-process executor. Its invocation is exposed +// externally and its progress/result arrives later through advanceTurn. +export interface AsyncRuntimeTool { + descriptor: z.infer & { execution: "async" }; +} + +export type RuntimeTool = SyncRuntimeTool | AsyncRuntimeTool; + +// Resolves persisted tool descriptors to live implementations during +// advanceTurn. A rejection or a descriptor mismatch is an infrastructure +// error: the execution is rejected and the turn is left unchanged. +export interface IToolRegistry { + resolve(descriptor: z.infer): Promise; +} diff --git a/apps/x/packages/shared/src/turns.test.ts b/apps/x/packages/shared/src/turns.test.ts index bcdae400..3c77faf5 100644 --- a/apps/x/packages/shared/src/turns.test.ts +++ b/apps/x/packages/shared/src/turns.test.ts @@ -506,6 +506,7 @@ describe("tool execution", () => { permResolved("tc1", "deny", "human"), result("tc1", "echo", "runtime", "Permission denied", true), ]); + expect(state.toolCalls[0].permission?.classificationFailed).toBe(true); expect(state.toolCalls[0].result?.result.isError).toBe(true); }); diff --git a/apps/x/packages/shared/src/turns.ts b/apps/x/packages/shared/src/turns.ts index 52e39c23..f8b87c5b 100644 --- a/apps/x/packages/shared/src/turns.ts +++ b/apps/x/packages/shared/src/turns.ts @@ -382,6 +382,9 @@ export interface ToolCallState { permission?: { required: z.infer; classification?: z.infer; + // Set when a tool_permission_classification_failed event named this + // call; such calls are treated as defer and never re-classified. + classificationFailed?: boolean; resolved?: z.infer; }; invocation?: z.infer; @@ -635,8 +638,8 @@ function applyToolPermissionClassificationFailed( if (toolCall.permission.resolved) { fail(`classification failure after permission resolution: ${toolCallId}`); } + toolCall.permission.classificationFailed = true; } - // Recorded for the audit trail; does not currently affect derived state. } function applyToolPermissionResolved(