apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-07-12 21:02:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(x/core): turn runtime with event-sourced execution loop (stage 2)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh 2026-07-02 09:32:27 +05:30
parent d821c419fe
commit cda17c2d40
21 changed files with 3656 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

14
apps/x/packages/core/src/turns/agent-resolver.ts Normal file
View file

@ -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<typeof RequestedAgent>,
): Promise<z.infer<typeof ResolvedAgent>>;
}

112
apps/x/packages/core/src/turns/api.ts Normal file
View file

@ -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<typeof RequestedAgent>;
sessionId?: string | null;
context: z.infer<typeof TurnContext>;
input: z.infer<typeof UserMessage>;
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<typeof ToolResultData>;
}
| {
type: "cancel";
reason?: string;
};
export type TurnOutcome =
| {
status: "completed";
output: z.infer<typeof AssistantMessage>;
finishReason: string;
usage: z.infer<typeof TurnUsage>;
}
| {
status: "suspended";
pendingPermissions: z.infer<typeof TurnSuspended>["pendingPermissions"];
pendingAsyncTools: z.infer<typeof TurnSuspended>["pendingAsyncTools"];
usage: z.infer<typeof TurnUsage>;
}
| {
status: "failed";
error: string;
// Mirrors turn_failed.code (e.g. MODEL_CALL_LIMIT_ERROR_CODE).
code?: string;
usage: z.infer<typeof TurnUsage>;
}
| {
status: "cancelled";
reason?: string;
usage: z.infer<typeof TurnUsage>;
};
export interface TurnExecution {
events: AsyncIterable<TurnStreamEvent>;
outcome: Promise<TurnOutcome>;
}
export interface Turn {
turnId: string;
events: Array<z.infer<typeof TurnEvent>>;
}
export interface ITurnRuntime {
createTurn(input: CreateTurnInput): Promise<string>;
advanceTurn(
turnId: string,
input?: TurnExternalInput,
options?: { signal?: AbortSignal },
): TurnExecution;
getTurn(turnId: string): Promise<Turn>;
}
// 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";
}
}

18
apps/x/packages/core/src/turns/bus.ts Normal file
View file

@ -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;
}

11
apps/x/packages/core/src/turns/clock.ts Normal file
View file

@ -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();
}
}

257
apps/x/packages/core/src/turns/context-resolver.test.ts Normal file
View file

@ -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<typeof TurnEvent>;
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<typeof TurnContext>,
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);
});
});

55
apps/x/packages/core/src/turns/context-resolver.ts Normal file
View file

@ -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<typeof TurnContext>,
): Promise<Array<z.infer<typeof ConversationMessage>>>;
}
export class TurnRepoContextResolver implements IContextResolver {
private readonly turnRepo: ITurnRepo;
constructor({ turnRepo }: { turnRepo: ITurnRepo }) {
this.turnRepo = turnRepo;
}
async resolve(
context: z.infer<typeof TurnContext>,
): Promise<Array<z.infer<typeof ConversationMessage>>> {
// 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<Array<z.infer<typeof ConversationMessage>>> = [];
const visited = new Set<string>();
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();
}
}

205
apps/x/packages/core/src/turns/fs-repo.test.ts Normal file
View file

@ -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<typeof TurnCreated> {
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<typeof TurnEvent> {
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<typeof TurnEvent> {
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<typeof TurnEvent>]),
).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<void> {
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");
});
});

128
apps/x/packages/core/src/turns/fs-repo.ts Normal file
View file

@ -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<z.infer<typeof TurnEvent>>,
): 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<typeof TurnCreated>): Promise<void> {
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<Array<z.infer<typeof TurnEvent>>> {
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<z.infer<typeof TurnEvent>> = [];
for (const [index, line] of lines.entries()) {
let parsed: z.infer<typeof TurnEvent>;
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<z.infer<typeof TurnEvent>>,
): Promise<void> {
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<T>(turnId: string, fn: () => Promise<T>): Promise<T> {
return this.mutex.run(turnId, fn);
}
}

62
apps/x/packages/core/src/turns/in-memory-turn-repo.ts Normal file
View file

@ -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<string, Array<z.infer<typeof TurnEvent>>>();
private mutex = new KeyedMutex();
async create(event: z.infer<typeof TurnCreated>): Promise<void> {
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<Array<z.infer<typeof TurnEvent>>> {
const events = this.files.get(turnId);
if (!events) {
throw new Error(`turn not found: ${turnId}`);
}
return structuredClone(events);
}
async append(
turnId: string,
events: Array<z.infer<typeof TurnEvent>>,
): Promise<void> {
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<T>(turnId: string, fn: () => Promise<T>): Promise<T> {
return this.mutex.run(turnId, fn);
}
// Test helper: seed a raw event log (validated) for recovery scenarios.
seed(events: Array<z.infer<typeof TurnEvent>>): 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))),
);
}
}

14
apps/x/packages/core/src/turns/index.ts Normal file
View file

@ -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";

21
apps/x/packages/core/src/turns/keyed-mutex.ts Normal file
View file

@ -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<string, Promise<unknown>>();
async run<T>(key: string, fn: () => Promise<T>): Promise<T> {
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);
}
}
}
}

46
apps/x/packages/core/src/turns/model-registry.ts Normal file
View file

@ -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<typeof DurableLlmStepStreamEvent> }
| {
type: "completed";
message: z.infer<typeof AssistantMessage>;
finishReason: string;
usage: z.infer<typeof TurnUsage>;
providerMetadata?: JsonValue;
};
export interface ModelStreamRequest {
systemPrompt: string;
messages: Array<z.infer<typeof ConversationMessage>>;
tools: Array<z.infer<typeof ToolDescriptor>>;
parameters: Record<string, JsonValue>;
signal: AbortSignal;
}
export interface ResolvedModel {
descriptor: z.infer<typeof ModelDescriptor>;
stream(request: ModelStreamRequest): AsyncIterable<LlmStreamEvent>;
}
// 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<typeof ModelDescriptor>): Promise<ResolvedModel>;
}

52
apps/x/packages/core/src/turns/permission.ts Normal file
View file

@ -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<PermissionCheckAllowed | PermissionCheckRequired>;
}
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<PermissionClassification[]>;
}

15
apps/x/packages/core/src/turns/repo.ts Normal file
View file

@ -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<typeof TurnCreated>): Promise<void>;
// Validates every line strictly; corrupt files are rejected whole.
read(turnId: string): Promise<Array<z.infer<typeof TurnEvent>>>;
// Validates events before writing.
append(turnId: string, events: Array<z.infer<typeof TurnEvent>>): Promise<void>;
// In-process per-turn exclusion.
withLock<T>(turnId: string, fn: () => Promise<T>): Promise<T>;
}

1495
apps/x/packages/core/src/turns/runtime.test.ts Normal file
View file

File diff suppressed because it is too large Load diff

928
apps/x/packages/core/src/turns/runtime.ts Normal file
View file

@ -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<typeof TurnEvent>;
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<string> {
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<Turn> {
const events = await this.turnRepo.read(turnId);
return { turnId, events };
}
advanceTurn(
turnId: string,
input?: TurnExternalInput,
options?: { signal?: AbortSignal },
): TurnExecution {
const stream = new HotStream<TurnStreamEvent, TurnOutcome>();
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<TurnStreamEvent, TurnOutcome>,
): Promise<TurnOutcome> {
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<TurnStreamEvent, TurnOutcome>,
): Promise<TurnOutcome> {
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<string, RuntimeTool>();
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<void> => {
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<string>();
let cancelReason: string | undefined;
const cancelTurn = async (): Promise<TurnOutcome> => {
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<typeof ModelRequest> = {
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<typeof TurnSuspended>["pendingPermissions"] {
return calls.map((tc) => ({
toolCallId: tc.toolCallId,
toolName: tc.toolName,
request: (tc.permission as NonNullable<ToolCallState["permission"]>)
.required.request,
}));
}
function asyncSnapshot(
calls: ToolCallState[],
): z.infer<typeof TurnSuspended>["pendingAsyncTools"] {
return calls.map((tc) => ({
toolCallId: tc.toolCallId,
toolId: (tc.invocation as z.infer<typeof ToolInvocationRequested>).toolId,
toolName: tc.toolName,
input: (tc.invocation as z.infer<typeof ToolInvocationRequested>).input,
}));
}
function modelCallFailedEvent(
turnId: string,
ts: string,
modelCallIndex: number,
error: string,
): z.infer<typeof ModelCallFailed> {
return { type: "model_call_failed", turnId, ts, modelCallIndex, error };
}
function runtimeResultEvent(
turnId: string,
ts: string,
tc: ToolCallState,
result: { output: JsonValue; isError: boolean },
): z.infer<typeof ToolResult> {
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<typeof ToolPermissionRequired> {
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<typeof ToolPermissionResolved>["source"],
reason?: string,
): z.infer<typeof ToolPermissionResolved> {
return {
type: "tool_permission_resolved",
turnId,
ts,
toolCallId,
decision,
source,
...(reason === undefined ? {} : { reason }),
};
}
function invocationEvent(
turnId: string,
ts: string,
tc: ToolCallState,
descriptor: z.infer<typeof ToolDescriptor>,
): z.infer<typeof ToolInvocationRequested> {
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,
};
}

89
apps/x/packages/core/src/turns/stream.test.ts Normal file
View file

@ -0,0 +1,89 @@
import { describe, expect, it } from "vitest";
import { HotStream } from "./stream.js";
async function collect<T>(iterable: AsyncIterable<T>): Promise<T[]> {
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<number, string>();
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<number, string>();
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<number, string>();
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<number, string>();
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<number, string>();
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<number, string>();
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<number, string>();
const consumer = collect(stream.events);
await Promise.resolve();
stream.end("done");
expect(await consumer).toEqual([]);
});
});

94
apps/x/packages/core/src/turns/stream.ts Normal file
View file

@ -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<TEvent, TOutcome> {
private queue: TEvent[] = [];
private waiters: Array<() => void> = [];
private done = false;
private failure: { error: unknown } | null = null;
private consumerClosed = false;
readonly outcome: Promise<TOutcome>;
private resolveOutcome!: (outcome: TOutcome) => void;
private rejectOutcome!: (error: unknown) => void;
constructor() {
this.outcome = new Promise<TOutcome>((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<TEvent> {
return {
[Symbol.asyncIterator]: (): AsyncIterator<TEvent> => ({
next: async (): Promise<IteratorResult<TEvent>> => {
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<void>((resolve) =>
this.waiters.push(resolve),
);
}
},
return: async (): Promise<IteratorResult<TEvent>> => {
this.consumerClosed = true;
this.queue.length = 0;
return { value: undefined, done: true };
},
}),
};
}
}

35
apps/x/packages/core/src/turns/tool-registry.ts Normal file
View file

@ -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<void>;
}
export interface SyncRuntimeTool {
descriptor: z.infer<typeof ToolDescriptor> & { execution: "sync" };
execute(
input: unknown,
context: ToolExecutionContext,
): Promise<z.infer<typeof ToolResultData>>;
}
// 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<typeof ToolDescriptor> & { 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<typeof ToolDescriptor>): Promise<RuntimeTool>;
}

1
apps/x/packages/shared/src/turns.test.ts
View file

@ -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);
});

5
apps/x/packages/shared/src/turns.ts
View file

@ -382,6 +382,9 @@ export interface ToolCallState {
permission?: {
required: z.infer<typeof ToolPermissionRequired>;
classification?: z.infer<typeof ToolPermissionClassified>;
// 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<typeof ToolPermissionResolved>;
};
invocation?: z.infer<typeof ToolInvocationRequested>;
@ -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(