From ccf38c6a0fd5e53e41bb74af228274fc89cf8443 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 2 Jul 2026 09:57:20 +0530 Subject: [PATCH] feat(x/core): session layer with write-through index (stage 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionsImpl per session-design.md, constructor-injected and tested entirely against a mocked ITurnRuntime: - sendMessage: per-session lock, typed TurnNotSettledError while the latest turn is non-terminal (terminal statuses including failed/ cancelled allow continuation), context as { previousTurnId } ref, turn-file-first write ordering (a failed session append leaves a benign orphan turn and no advance), denormalized agent/model on turn_appended, default title from the first message. - Dedicated respondToAskHuman endpoint (async_tool_result wrapper); respondToPermission / deliverAsyncToolResult pass turn-runtime rejections through. stopTurn aborts a live advance or cancels an at-rest turn; resumeTurn re-enters idle turns (never at startup). - In-memory index: startup scan (session files + each latest turn for status; corrupt files yield errored entries without aborting), write-through updates, deletion guard against late-settle resurrection. Bus events (turn-event / index-changed) defined in @x/shared as the renderer IPC contract. - FSSessionRepo: date-partitioned append-only JSONL + list/delete; deleteSession removes the session file only. - runHeadlessTurn: standalone turns (sessionId null, auto permission, no human) for background/knowledge/scheduled callers. 29 new tests covering the session-design §13 matrix. Co-Authored-By: Claude Fable 5 --- apps/x/packages/core/src/sessions/api.ts | 79 ++ apps/x/packages/core/src/sessions/bus.ts | 7 + .../core/src/sessions/fs-repo.test.ts | 108 +++ apps/x/packages/core/src/sessions/fs-repo.ts | 149 ++++ apps/x/packages/core/src/sessions/headless.ts | 41 + .../src/sessions/in-memory-session-repo.ts | 83 ++ apps/x/packages/core/src/sessions/index.ts | 8 + apps/x/packages/core/src/sessions/repo.ts | 20 + .../core/src/sessions/session-index.ts | 25 + .../core/src/sessions/sessions.test.ts | 801 ++++++++++++++++++ apps/x/packages/core/src/sessions/sessions.ts | 418 +++++++++ apps/x/packages/shared/src/sessions.ts | 22 +- 12 files changed, 1760 insertions(+), 1 deletion(-) create mode 100644 apps/x/packages/core/src/sessions/api.ts create mode 100644 apps/x/packages/core/src/sessions/bus.ts create mode 100644 apps/x/packages/core/src/sessions/fs-repo.test.ts create mode 100644 apps/x/packages/core/src/sessions/fs-repo.ts create mode 100644 apps/x/packages/core/src/sessions/headless.ts create mode 100644 apps/x/packages/core/src/sessions/in-memory-session-repo.ts create mode 100644 apps/x/packages/core/src/sessions/index.ts create mode 100644 apps/x/packages/core/src/sessions/repo.ts create mode 100644 apps/x/packages/core/src/sessions/session-index.ts create mode 100644 apps/x/packages/core/src/sessions/sessions.test.ts create mode 100644 apps/x/packages/core/src/sessions/sessions.ts diff --git a/apps/x/packages/core/src/sessions/api.ts b/apps/x/packages/core/src/sessions/api.ts new file mode 100644 index 00000000..7f787b49 --- /dev/null +++ b/apps/x/packages/core/src/sessions/api.ts @@ -0,0 +1,79 @@ +import type { z } from "zod"; +import type { UserMessage } from "@x/shared/dist/message.js"; +import type { + SessionIndexEntry, + SessionState, +} from "@x/shared/dist/sessions.js"; +import type { + JsonValue, + RequestedAgent, + ToolResultData, +} from "@x/shared/dist/turns.js"; +import type { Turn } from "../turns/api.js"; + +// Per-message configuration; it lands on the turn (sessions store none). +export interface SendMessageConfig { + agent: z.infer; + autoPermission?: boolean; + maxModelCalls?: number; +} + +export interface ISessions { + // Startup scan: builds the in-memory index from session files (reading + // each session's latest turn for status). Must run before listSessions. + initialize(): Promise; + + createSession(input?: { title?: string }): Promise; + listSessions(): SessionIndexEntry[]; + getSession(sessionId: string): Promise; + getTurn(turnId: string): Promise; + + // Rejects with TurnNotSettledError while the latest turn is non-terminal. + // Returns as soon as the turn is created, referenced, and advancing; + // progress flows through the bus. + sendMessage( + sessionId: string, + input: z.infer, + config: SendMessageConfig, + ): Promise<{ turnId: string }>; + + // External inputs, one advanceTurn each. These settle with that + // invocation's outcome; turn-runtime input rejections pass through. + respondToPermission( + turnId: string, + toolCallId: string, + decision: "allow" | "deny", + metadata?: JsonValue, + ): Promise; + // The dedicated ask-human endpoint; sendMessage never routes here. + respondToAskHuman( + turnId: string, + toolCallId: string, + answer: string, + ): Promise; + deliverAsyncToolResult( + turnId: string, + toolCallId: string, + result: z.infer, + ): Promise; + + stopTurn(turnId: string, reason?: string): Promise; + // Recovery entry for turns left idle by a crash; runs in the background. + resumeTurn(sessionId: string): Promise; + + setTitle(sessionId: string, title: string): Promise; + deleteSession(sessionId: string): Promise; +} + +export class TurnNotSettledError extends Error { + constructor( + readonly sessionId: string, + readonly turnId: string, + readonly turnStatus: string, + ) { + super( + `session ${sessionId} has a non-terminal turn ${turnId} (${turnStatus})`, + ); + this.name = "TurnNotSettledError"; + } +} diff --git a/apps/x/packages/core/src/sessions/bus.ts b/apps/x/packages/core/src/sessions/bus.ts new file mode 100644 index 00000000..e82f6f47 --- /dev/null +++ b/apps/x/packages/core/src/sessions/bus.ts @@ -0,0 +1,7 @@ +import type { SessionBusEvent } from "@x/shared/dist/sessions.js"; + +// Ephemeral fan-out toward the renderer (bridged over IPC in the app layer). +// Publishing is fire-and-forget; nothing durable depends on delivery. +export interface ISessionBus { + publish(event: SessionBusEvent): void; +} diff --git a/apps/x/packages/core/src/sessions/fs-repo.test.ts b/apps/x/packages/core/src/sessions/fs-repo.test.ts new file mode 100644 index 00000000..dc85993d --- /dev/null +++ b/apps/x/packages/core/src/sessions/fs-repo.test.ts @@ -0,0 +1,108 @@ +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 { + SessionCorruptionError, + type SessionCreated, + type SessionEvent, +} from "@x/shared/dist/sessions.js"; +import { FSSessionRepo } from "./fs-repo.js"; + +const S1 = "2026-07-02T09-00-00Z-0000001-000"; +const S2 = "2026-07-03T09-00-00Z-0000002-000"; + +function created(sessionId = S1): z.infer { + return { + type: "session_created", + schemaVersion: 1, + sessionId, + ts: "2026-07-02T09:00:00Z", + title: "T", + }; +} + +function appended(sessionId = S1): z.infer { + return { + type: "turn_appended", + sessionId, + ts: "2026-07-02T09:01:00Z", + turnId: "turn-1", + sessionSeq: 1, + agentId: "copilot", + model: { provider: "fake", model: "m" }, + }; +} + +describe("FSSessionRepo", () => { + let root: string; + let repo: FSSessionRepo; + + beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "session-repo-")); + repo = new FSSessionRepo({ sessionsRootDir: root }); + }); + + afterEach(async () => { + await fs.rm(root, { recursive: true, force: true }); + }); + + it("creates date-partitioned files and round-trips events", async () => { + await repo.create(created()); + await repo.append(S1, [appended()]); + const file = path.join(root, "2026", "07", "02", `${S1}.jsonl`); + await fs.access(file); + const events = await repo.read(S1); + expect(events.map((e) => e.type)).toEqual([ + "session_created", + "turn_appended", + ]); + }); + + it("rejects malformed and path-like session ids", async () => { + for (const bad of ["../evil", "a/b", "nope", ""]) { + await expect(repo.read(bad)).rejects.toThrowError(/invalid session id/); + } + }); + + it("create fails if the session exists; append fails if it doesn't", async () => { + await repo.create(created()); + await expect(repo.create(created())).rejects.toThrowError(); + await expect(repo.append(S2, [appended(S2)])).rejects.toThrowError( + /session not found/, + ); + }); + + it("read rejects corrupt files whole", async () => { + const file = path.join(root, "2026", "07", "02", `${S1}.jsonl`); + await fs.mkdir(path.dirname(file), { recursive: true }); + await fs.writeFile(file, `${JSON.stringify(created())}\nnot json\n`); + await expect(repo.read(S1)).rejects.toThrowError(SessionCorruptionError); + await fs.writeFile(file, `${JSON.stringify(created())}\n{"torn`); + await expect(repo.read(S1)).rejects.toThrowError( + /does not end with a complete line/, + ); + }); + + it("lists session ids across date partitions", async () => { + await repo.create(created(S1)); + await repo.create(created(S2)); + expect(await repo.listSessionIds()).toEqual([S1, S2]); + }); + + it("listing an empty root returns no ids", async () => { + const empty = new FSSessionRepo({ + sessionsRootDir: path.join(root, "missing"), + }); + expect(await empty.listSessionIds()).toEqual([]); + }); + + it("delete removes the file only; deleting a missing session throws", async () => { + await repo.create(created(S1)); + await repo.create(created(S2)); + await repo.delete(S1); + expect(await repo.listSessionIds()).toEqual([S2]); + await expect(repo.delete(S1)).rejects.toThrowError(/session not found/); + }); +}); diff --git a/apps/x/packages/core/src/sessions/fs-repo.ts b/apps/x/packages/core/src/sessions/fs-repo.ts new file mode 100644 index 00000000..63eccecd --- /dev/null +++ b/apps/x/packages/core/src/sessions/fs-repo.ts @@ -0,0 +1,149 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { z } from "zod"; +import { + SessionCorruptionError, + SessionCreated, + SessionEvent, +} from "@x/shared/dist/sessions.js"; +import { KeyedMutex } from "../turns/keyed-mutex.js"; +import type { ISessionRepo } from "./repo.js"; + +// Session IDs come from the same generator as turn IDs. +const SESSION_ID_PATTERN = /^(\d{4})-(\d{2})-(\d{2})T[A-Za-z0-9-]+$/; + +export class FSSessionRepo implements ISessionRepo { + private readonly rootDir: string; + private readonly mutex = new KeyedMutex(); + + constructor({ sessionsRootDir }: { sessionsRootDir: string }) { + this.rootDir = sessionsRootDir; + } + + private filePath(sessionId: string): string { + const match = SESSION_ID_PATTERN.exec(sessionId); + if (!match) { + throw new Error(`invalid session id: ${sessionId}`); + } + const [, year, month, day] = match; + return path.join(this.rootDir, year, month, day, `${sessionId}.jsonl`); + } + + private serialize( + sessionId: string, + events: Array>, + ): string { + let payload = ""; + for (const event of events) { + const parsed = SessionEvent.parse(event); + if (parsed.sessionId !== sessionId) { + throw new Error( + `event sessionId ${parsed.sessionId} does not match ${sessionId}`, + ); + } + payload += `${JSON.stringify(parsed)}\n`; + } + return payload; + } + + async create(event: z.infer): Promise { + const parsed = SessionCreated.parse(event); + const file = this.filePath(parsed.sessionId); + await fs.mkdir(path.dirname(file), { recursive: true }); + await fs.writeFile(file, this.serialize(parsed.sessionId, [parsed]), { + flag: "wx", + }); + } + + async read(sessionId: string): Promise>> { + const file = this.filePath(sessionId); + let raw: string; + try { + raw = await fs.readFile(file, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + throw new Error(`session not found: ${sessionId}`); + } + throw error; + } + if (raw.length === 0) { + throw new SessionCorruptionError(`session file is empty: ${sessionId}`); + } + const lines = raw.split("\n"); + const trailing = lines.pop(); + if (trailing !== "") { + throw new SessionCorruptionError( + `session file does not end with a complete line: ${sessionId}`, + ); + } + const events: Array> = []; + for (const [index, line] of lines.entries()) { + let parsed: z.infer; + try { + parsed = SessionEvent.parse(JSON.parse(line)); + } catch (error) { + throw new SessionCorruptionError( + `malformed session event at ${sessionId}:${index + 1}: ${String( + error instanceof Error ? error.message : error, + )}`, + ); + } + if (parsed.sessionId !== sessionId) { + throw new SessionCorruptionError( + `event sessionId ${parsed.sessionId} does not match file ${sessionId}`, + ); + } + events.push(parsed); + } + return events; + } + + async append( + sessionId: string, + events: Array>, + ): Promise { + if (events.length === 0) { + return; + } + const payload = this.serialize(sessionId, events); + const file = this.filePath(sessionId); + try { + await fs.access(file); + } catch { + throw new Error(`session not found: ${sessionId}`); + } + await fs.appendFile(file, payload); + } + + async withLock(sessionId: string, fn: () => Promise): Promise { + return this.mutex.run(sessionId, fn); + } + + async listSessionIds(): Promise { + let names: string[]; + try { + names = (await fs.readdir(this.rootDir, { recursive: true })) as string[]; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return []; + } + throw error; + } + return names + .filter((name) => name.endsWith(".jsonl")) + .map((name) => path.basename(name, ".jsonl")) + .sort(); + } + + async delete(sessionId: string): Promise { + const file = this.filePath(sessionId); + try { + await fs.unlink(file); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + throw new Error(`session not found: ${sessionId}`); + } + throw error; + } + } +} diff --git a/apps/x/packages/core/src/sessions/headless.ts b/apps/x/packages/core/src/sessions/headless.ts new file mode 100644 index 00000000..071d7fd4 --- /dev/null +++ b/apps/x/packages/core/src/sessions/headless.ts @@ -0,0 +1,41 @@ +import type { z } from "zod"; +import type { UserMessage } from "@x/shared/dist/message.js"; +import type { + ConversationMessage, + RequestedAgent, +} from "@x/shared/dist/turns.js"; +import type { ITurnRuntime, TurnOutcome } from "../turns/api.js"; + +// Standalone turns for non-session callers (background tasks, live notes, +// knowledge pipelines, scheduled agents): sessionId null, automatic +// permissions, no human. Never appears in the session index; callers keep +// the turnId if they need history. +export async function runHeadlessTurn( + turnRuntime: ITurnRuntime, + input: { + agent: z.infer; + context?: Array>; + input: z.infer; + maxModelCalls?: number; + signal?: AbortSignal; + }, +): Promise<{ turnId: string; outcome: TurnOutcome }> { + const turnId = await turnRuntime.createTurn({ + agent: input.agent, + sessionId: null, + context: input.context ?? [], + input: input.input, + config: { + autoPermission: true, + humanAvailable: false, + ...(input.maxModelCalls === undefined + ? {} + : { maxModelCalls: input.maxModelCalls }), + }, + }); + const execution = turnRuntime.advanceTurn(turnId, undefined, { + signal: input.signal, + }); + const outcome = await execution.outcome; + return { turnId, outcome }; +} diff --git a/apps/x/packages/core/src/sessions/in-memory-session-repo.ts b/apps/x/packages/core/src/sessions/in-memory-session-repo.ts new file mode 100644 index 00000000..35c54765 --- /dev/null +++ b/apps/x/packages/core/src/sessions/in-memory-session-repo.ts @@ -0,0 +1,83 @@ +import { z } from "zod"; +import { + SessionCorruptionError, + SessionCreated, + SessionEvent, +} from "@x/shared/dist/sessions.js"; +import { KeyedMutex } from "../turns/keyed-mutex.js"; +import type { ISessionRepo } from "./repo.js"; + +// Test fake mirroring FSSessionRepo semantics without touching disk. +export class InMemorySessionRepo implements ISessionRepo { + private files = new Map>>(); + private corrupt = new Set(); + private mutex = new KeyedMutex(); + + async create(event: z.infer): Promise { + const parsed = SessionCreated.parse(event); + if (this.files.has(parsed.sessionId) || this.corrupt.has(parsed.sessionId)) { + throw new Error(`session already exists: ${parsed.sessionId}`); + } + this.files.set(parsed.sessionId, [structuredClone(parsed)]); + } + + async read(sessionId: string): Promise>> { + if (this.corrupt.has(sessionId)) { + throw new SessionCorruptionError(`session file is corrupt: ${sessionId}`); + } + const events = this.files.get(sessionId); + if (!events) { + throw new Error(`session not found: ${sessionId}`); + } + return structuredClone(events); + } + + async append( + sessionId: string, + events: Array>, + ): Promise { + const file = this.files.get(sessionId); + if (!file) { + throw new Error(`session not found: ${sessionId}`); + } + for (const event of events) { + const parsed = SessionEvent.parse(event); + if (parsed.sessionId !== sessionId) { + throw new Error( + `event sessionId ${parsed.sessionId} does not match ${sessionId}`, + ); + } + file.push(structuredClone(parsed)); + } + } + + async withLock(sessionId: string, fn: () => Promise): Promise { + return this.mutex.run(sessionId, fn); + } + + async listSessionIds(): Promise { + return [...this.files.keys(), ...this.corrupt].sort(); + } + + async delete(sessionId: string): Promise { + const existed = this.files.delete(sessionId) || this.corrupt.delete(sessionId); + if (!existed) { + throw new Error(`session not found: ${sessionId}`); + } + } + + // Test helpers + seed(events: Array>): void { + if (events.length === 0) { + throw new Error("cannot seed an empty log"); + } + this.files.set( + events[0].sessionId, + events.map((e) => structuredClone(SessionEvent.parse(e))), + ); + } + + seedCorrupt(sessionId: string): void { + this.corrupt.add(sessionId); + } +} diff --git a/apps/x/packages/core/src/sessions/index.ts b/apps/x/packages/core/src/sessions/index.ts new file mode 100644 index 00000000..df906d01 --- /dev/null +++ b/apps/x/packages/core/src/sessions/index.ts @@ -0,0 +1,8 @@ +export * from "./api.js"; +export * from "./bus.js"; +export * from "./fs-repo.js"; +export * from "./headless.js"; +export * from "./in-memory-session-repo.js"; +export * from "./repo.js"; +export * from "./session-index.js"; +export * from "./sessions.js"; diff --git a/apps/x/packages/core/src/sessions/repo.ts b/apps/x/packages/core/src/sessions/repo.ts new file mode 100644 index 00000000..eb10f4e1 --- /dev/null +++ b/apps/x/packages/core/src/sessions/repo.ts @@ -0,0 +1,20 @@ +import type { z } from "zod"; +import type { SessionCreated, SessionEvent } from "@x/shared/dist/sessions.js"; + +export interface ISessionRepo { + // Fails if the session already exists. + create(event: z.infer): Promise; + // Validates every line strictly; corrupt files are rejected whole. + read(sessionId: string): Promise>>; + // Validates events before writing. + append( + sessionId: string, + events: Array>, + ): Promise; + // In-process per-session exclusion. + withLock(sessionId: string, fn: () => Promise): Promise; + // Enumerates all session ids for the startup scan. + listSessionIds(): Promise; + // Removes the session file only; referenced turn files stay as orphans. + delete(sessionId: string): Promise; +} diff --git a/apps/x/packages/core/src/sessions/session-index.ts b/apps/x/packages/core/src/sessions/session-index.ts new file mode 100644 index 00000000..f09dcbf6 --- /dev/null +++ b/apps/x/packages/core/src/sessions/session-index.ts @@ -0,0 +1,25 @@ +import type { SessionIndexEntry } from "@x/shared/dist/sessions.js"; + +// In-memory projection over session files. Never a source of truth: rebuilt +// by the startup scan, maintained write-through by the sessions service. +export class SessionIndex { + private entries = new Map(); + + list(): SessionIndexEntry[] { + return [...this.entries.values()].sort((a, b) => + b.updatedAt.localeCompare(a.updatedAt), + ); + } + + get(sessionId: string): SessionIndexEntry | undefined { + return this.entries.get(sessionId); + } + + upsert(entry: SessionIndexEntry): void { + this.entries.set(entry.sessionId, entry); + } + + remove(sessionId: string): boolean { + return this.entries.delete(sessionId); + } +} diff --git a/apps/x/packages/core/src/sessions/sessions.test.ts b/apps/x/packages/core/src/sessions/sessions.test.ts new file mode 100644 index 00000000..f0c53fcf --- /dev/null +++ b/apps/x/packages/core/src/sessions/sessions.test.ts @@ -0,0 +1,801 @@ +import { describe, expect, it } from "vitest"; +import type { z } from "zod"; +import type { SessionBusEvent, SessionEvent } from "@x/shared/dist/sessions.js"; +import type { + ResolvedAgent, + TurnEvent, + TurnStreamEvent, +} from "@x/shared/dist/turns.js"; +import type { + CreateTurnInput, + ITurnRuntime, + Turn, + TurnExecution, + TurnExternalInput, + TurnOutcome, +} from "../turns/api.js"; +import { TurnInputError } from "../turns/api.js"; +import { HotStream } from "../turns/stream.js"; +import { TurnNotSettledError } from "./api.js"; +import { runHeadlessTurn } from "./headless.js"; +import { InMemorySessionRepo } from "./in-memory-session-repo.js"; +import type { ISessionRepo } from "./repo.js"; +import { SessionsImpl } from "./sessions.js"; + +type TEvent = z.infer; + +const TS = "2026-07-02T10:00:00Z"; +const FIXTURE_MODEL = { provider: "openai", model: "gpt-fixture" }; +const FIXTURE_AGENT: z.infer = { + agentId: "copilot", + systemPrompt: "SYS", + model: FIXTURE_MODEL, + tools: [], +}; + +function user(text: string) { + return { role: "user" as const, content: text }; +} + +function assistantText(text: string) { + return { role: "assistant" as const, content: text }; +} + +function completedOutcome(): TurnOutcome { + return { + status: "completed", + output: assistantText("ok"), + finishReason: "stop", + usage: {}, + }; +} + +function createdEvent(turnId: string, input: CreateTurnInput): TEvent { + return { + type: "turn_created", + schemaVersion: 1, + turnId, + ts: TS, + sessionId: input.sessionId ?? null, + agent: { requested: input.agent, resolved: FIXTURE_AGENT }, + context: input.context, + input: input.input, + config: { + autoPermission: input.config.autoPermission ?? false, + humanAvailable: input.config.humanAvailable, + maxModelCalls: input.config.maxModelCalls ?? 20, + }, + }; +} + +// Minimal valid event logs reducing to each turn status. +function turnLog( + turnId: string, + sessionId: string, + status: "idle" | "completed" | "failed" | "cancelled" | "suspended", +): TEvent[] { + const created: TEvent = createdEvent(turnId, { + agent: { agentId: "copilot" }, + sessionId, + context: [], + input: user("hi"), + config: { humanAvailable: true }, + }); + const requested: TEvent = { + type: "model_call_requested", + turnId, + ts: TS, + modelCallIndex: 0, + request: { systemPrompt: "SYS", messages: [user("hi")], tools: [], parameters: {} }, + }; + switch (status) { + case "idle": + return [created]; + case "completed": + return [ + created, + requested, + { + type: "model_call_completed", + turnId, + ts: TS, + modelCallIndex: 0, + message: assistantText("ok"), + finishReason: "stop", + usage: {}, + }, + { + type: "turn_completed", + turnId, + ts: TS, + output: assistantText("ok"), + finishReason: "stop", + usage: {}, + }, + ]; + case "failed": + return [ + created, + requested, + { type: "model_call_failed", turnId, ts: TS, modelCallIndex: 0, error: "boom" }, + { type: "turn_failed", turnId, ts: TS, error: "boom", usage: {} }, + ]; + case "cancelled": + return [ + created, + { type: "turn_cancelled", turnId, ts: TS, usage: {} }, + ]; + case "suspended": + return [ + created, + requested, + { + type: "model_call_completed", + turnId, + ts: TS, + modelCallIndex: 0, + message: { + role: "assistant", + content: [ + { + type: "tool-call", + toolCallId: "B", + toolName: "fetch", + arguments: {}, + }, + ], + }, + finishReason: "tool-calls", + usage: {}, + }, + { + type: "tool_invocation_requested", + turnId, + ts: TS, + toolCallId: "B", + toolId: "tool.fetch", + toolName: "fetch", + execution: "async", + input: {}, + }, + { + type: "turn_suspended", + turnId, + ts: TS, + pendingPermissions: [], + pendingAsyncTools: [ + { toolCallId: "B", toolId: "tool.fetch", toolName: "fetch", input: {} }, + ], + usage: {}, + }, + ]; + } +} + +type AdvanceScript = (call: { + turnId: string; + input?: TurnExternalInput; + signal?: AbortSignal; +}) => + | { events?: TurnStreamEvent[]; outcome: TurnOutcome } + | { error: unknown } + | { untilAbort: true }; + +class FakeTurnRuntime implements ITurnRuntime { + createTurnInputs: CreateTurnInput[] = []; + advanceCalls: Array<{ turnId: string; input?: TurnExternalInput }> = []; + logs = new Map(); + createError?: Error; + script?: AdvanceScript; + private n = 0; + + async createTurn(input: CreateTurnInput): Promise { + if (this.createError) { + throw this.createError; + } + this.createTurnInputs.push(input); + this.n += 1; + const turnId = `2026-07-02T10-00-00Z-${String(this.n).padStart(7, "0")}-000`; + this.logs.set(turnId, [createdEvent(turnId, input)]); + return turnId; + } + + advanceTurn( + turnId: string, + input?: TurnExternalInput, + options?: { signal?: AbortSignal }, + ): TurnExecution { + this.advanceCalls.push({ turnId, input }); + const stream = new HotStream(); + const result = this.script?.({ turnId, input, signal: options?.signal }) ?? { + outcome: completedOutcome(), + }; + if ("error" in result) { + stream.fail(result.error); + } else if ("untilAbort" in result) { + const signal = options?.signal; + if (!signal) { + throw new Error("untilAbort script requires a signal"); + } + const finish = () => stream.end({ status: "cancelled", usage: {} }); + if (signal.aborted) { + finish(); + } else { + signal.addEventListener("abort", finish, { once: true }); + } + } else { + for (const event of result.events ?? []) { + stream.push(event); + } + stream.end(result.outcome); + } + return { events: stream.events, outcome: stream.outcome }; + } + + async getTurn(turnId: string): Promise { + const log = this.logs.get(turnId); + if (!log) { + throw new Error(`turn not found: ${turnId}`); + } + return { turnId, events: structuredClone(log) }; + } + + setLog(turnId: string, events: TEvent[]): void { + this.logs.set(turnId, events); + } +} + +class RecordingBus { + events: SessionBusEvent[] = []; + publish(event: SessionBusEvent): void { + this.events.push(event); + } +} + +class FakeIdGen { + private n = 0; + async next(): Promise { + this.n += 1; + return `2026-07-02T09-00-00Z-${String(this.n).padStart(7, "0")}-000`; + } +} + +class FakeClock { + now(): string { + return TS; + } +} + +// Wrapper to simulate a crash between turn-file creation and session append. +class FlakySessionRepo implements ISessionRepo { + failNextAppend = false; + constructor(private readonly inner: InMemorySessionRepo) {} + create(event: Parameters[0]) { + return this.inner.create(event); + } + read(sessionId: string) { + return this.inner.read(sessionId); + } + async append( + sessionId: string, + events: Array>, + ): Promise { + if (this.failNextAppend) { + this.failNextAppend = false; + throw new Error("disk full"); + } + return this.inner.append(sessionId, events); + } + withLock(sessionId: string, fn: () => Promise): Promise { + return this.inner.withLock(sessionId, fn); + } + listSessionIds() { + return this.inner.listSessionIds(); + } + delete(sessionId: string) { + return this.inner.delete(sessionId); + } +} + +function makeSessions(opts: { repo?: ISessionRepo; fake?: FakeTurnRuntime } = {}) { + const repo = opts.repo ?? new InMemorySessionRepo(); + const fake = opts.fake ?? new FakeTurnRuntime(); + const bus = new RecordingBus(); + const sessions = new SessionsImpl({ + sessionRepo: repo, + turnRuntime: fake, + idGenerator: new FakeIdGen(), + clock: new FakeClock(), + bus, + }); + return { sessions, repo, fake, bus }; +} + +const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + +describe("createSession and listing", () => { + it("persists session_created and publishes an index entry with status none", async () => { + const { sessions, repo, bus } = makeSessions(); + const sessionId = await sessions.createSession({ title: "My chat" }); + const events = await (repo as InMemorySessionRepo).read(sessionId); + expect(events).toEqual([ + expect.objectContaining({ + type: "session_created", + schemaVersion: 1, + sessionId, + title: "My chat", + }), + ]); + expect(sessions.listSessions()).toEqual([ + expect.objectContaining({ + sessionId, + title: "My chat", + turnCount: 0, + latestTurnStatus: "none", + }), + ]); + expect(bus.events).toEqual([ + expect.objectContaining({ kind: "index-changed", sessionId }), + ]); + }); +}); + +describe("sendMessage (13.3)", () => { + it("first message: inline empty context, config on the turn, denormalized ref, default title", async () => { + const { sessions, repo, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("Fix the bug in parser"), { + agent: { agentId: "copilot", overrides: { model: { provider: "x", model: "y" } } }, + autoPermission: true, + maxModelCalls: 5, + }); + + expect(fake.createTurnInputs[0]).toEqual({ + agent: { agentId: "copilot", overrides: { model: { provider: "x", model: "y" } } }, + sessionId, + context: [], + input: user("Fix the bug in parser"), + config: { humanAvailable: true, autoPermission: true, maxModelCalls: 5 }, + }); + + const events = await (repo as InMemorySessionRepo).read(sessionId); + expect(events[1]).toEqual( + expect.objectContaining({ + type: "turn_appended", + turnId, + sessionSeq: 1, + agentId: "copilot", + model: FIXTURE_MODEL, // resolved, not requested override input + }), + ); + expect(events[2]).toEqual( + expect.objectContaining({ + type: "title_changed", + title: "Fix the bug in parser", + }), + ); + expect(fake.advanceCalls).toEqual([{ turnId, input: undefined }]); + }); + + it("subsequent messages reference the latest turn and skip the title", async () => { + const { sessions, repo, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const first = await sessions.sendMessage(sessionId, user("one"), { + agent: { agentId: "copilot" }, + }); + fake.setLog(first.turnId, turnLog(first.turnId, sessionId, "completed")); + + const second = await sessions.sendMessage(sessionId, user("two"), { + agent: { agentId: "copilot" }, + }); + expect(fake.createTurnInputs[1].context).toEqual({ + previousTurnId: first.turnId, + }); + const events = await (repo as InMemorySessionRepo).read(sessionId); + const appends = events.filter((e) => e.type === "turn_appended"); + expect(appends.map((e) => (e.type === "turn_appended" ? e.sessionSeq : 0))).toEqual([1, 2]); + expect(appends[1]).toEqual( + expect.objectContaining({ turnId: second.turnId }), + ); + expect(events.filter((e) => e.type === "title_changed")).toHaveLength(1); + }); + + it("rejects while the latest turn is idle or suspended; allows all terminal statuses", async () => { + const { sessions, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("one"), { + agent: { agentId: "copilot" }, + }); + + for (const status of ["idle", "suspended"] as const) { + fake.setLog(turnId, turnLog(turnId, sessionId, status)); + const attempt = sessions.sendMessage(sessionId, user("nope"), { + agent: { agentId: "copilot" }, + }); + await expect(attempt).rejects.toThrowError(TurnNotSettledError); + await expect(attempt).rejects.toMatchObject({ + sessionId, + turnId, + turnStatus: status, + }); + } + + let latest = turnId; + for (const status of ["completed", "failed", "cancelled"] as const) { + fake.setLog(latest, turnLog(latest, sessionId, status)); + const result = await sessions.sendMessage(sessionId, user(`after ${status}`), { + agent: { agentId: "copilot" }, + }); + latest = result.turnId; + } + }); + + it("serializes concurrent sends: exactly one wins", async () => { + const { sessions, repo } = makeSessions(); + const sessionId = await sessions.createSession(); + const results = await Promise.allSettled([ + sessions.sendMessage(sessionId, user("a"), { agent: { agentId: "copilot" } }), + sessions.sendMessage(sessionId, user("b"), { agent: { agentId: "copilot" } }), + ]); + expect(results.filter((r) => r.status === "fulfilled")).toHaveLength(1); + const rejected = results.find((r) => r.status === "rejected"); + expect((rejected as PromiseRejectedResult).reason).toBeInstanceOf( + TurnNotSettledError, + ); + const events = await (repo as InMemorySessionRepo).read(sessionId); + expect(events.filter((e) => e.type === "turn_appended")).toHaveLength(1); + }); +}); + +describe("write ordering and crash simulation (13.4)", () => { + it("a failed session append leaves a benign orphan turn and no advance; retry works", async () => { + const inner = new InMemorySessionRepo(); + const flaky = new FlakySessionRepo(inner); + const { sessions, fake } = makeSessions({ repo: flaky }); + const sessionId = await sessions.createSession(); + + flaky.failNextAppend = true; + await expect( + sessions.sendMessage(sessionId, user("one"), { agent: { agentId: "copilot" } }), + ).rejects.toThrowError("disk full"); + + // Turn file exists (orphan), session unchanged, nothing advanced. + expect(fake.logs.size).toBe(1); + expect(fake.advanceCalls).toHaveLength(0); + expect(await inner.read(sessionId)).toHaveLength(1); + + // Retry produces a fresh turn at sessionSeq 1. + const { turnId } = await sessions.sendMessage(sessionId, user("one again"), { + agent: { agentId: "copilot" }, + }); + const events = await inner.read(sessionId); + expect(events.filter((e) => e.type === "turn_appended")).toEqual([ + expect.objectContaining({ turnId, sessionSeq: 1 }), + ]); + }); + + it("a failed createTurn leaves the session untouched", async () => { + const { sessions, repo, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + fake.createError = new Error("agent resolution failed"); + await expect( + sessions.sendMessage(sessionId, user("one"), { agent: { agentId: "ghost" } }), + ).rejects.toThrowError("agent resolution failed"); + expect(await (repo as InMemorySessionRepo).read(sessionId)).toHaveLength(1); + expect(fake.advanceCalls).toHaveLength(0); + }); +}); + +describe("external input routing (13.5)", () => { + async function setupSuspended() { + const fixture = makeSessions(); + const sessionId = await fixture.sessions.createSession(); + const { turnId } = await fixture.sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + fixture.fake.setLog(turnId, turnLog(turnId, sessionId, "suspended")); + fixture.fake.advanceCalls.length = 0; + return { ...fixture, sessionId, turnId }; + } + + it("respondToPermission advances with a permission_decision input", async () => { + const { sessions, fake, turnId } = await setupSuspended(); + await sessions.respondToPermission(turnId, "tc1", "allow", { scope: "once" }); + expect(fake.advanceCalls).toEqual([ + { + turnId, + input: { + type: "permission_decision", + toolCallId: "tc1", + decision: "allow", + metadata: { scope: "once" }, + }, + }, + ]); + }); + + it("respondToAskHuman is a dedicated async_tool_result wrapper", async () => { + const { sessions, fake, turnId } = await setupSuspended(); + await sessions.respondToAskHuman(turnId, "B", "the answer is 42"); + expect(fake.advanceCalls).toEqual([ + { + turnId, + input: { + type: "async_tool_result", + toolCallId: "B", + result: { output: "the answer is 42", isError: false }, + }, + }, + ]); + }); + + it("deliverAsyncToolResult passes the result through verbatim", async () => { + const { sessions, fake, turnId } = await setupSuspended(); + await sessions.deliverAsyncToolResult(turnId, "B", { + output: { rows: 3 }, + isError: false, + }); + expect(fake.advanceCalls[0].input).toEqual({ + type: "async_tool_result", + toolCallId: "B", + result: { output: { rows: 3 }, isError: false }, + }); + }); + + it("turn-runtime input rejections pass through", async () => { + const { sessions, fake, turnId } = await setupSuspended(); + fake.script = () => ({ error: new TurnInputError("no pending async tool call X") }); + await expect( + sessions.deliverAsyncToolResult(turnId, "X", { output: "x", isError: false }), + ).rejects.toThrowError(TurnInputError); + }); +}); + +describe("stopTurn and resumeTurn", () => { + it("aborts an actively running advance instead of issuing a cancel input", async () => { + const { sessions, fake } = makeSessions(); + fake.script = () => ({ untilAbort: true }); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + expect(fake.advanceCalls).toHaveLength(1); + await sessions.stopTurn(turnId); + // No second advance: the running invocation's signal was aborted. + expect(fake.advanceCalls).toHaveLength(1); + }); + + it("cancels an at-rest turn through a cancel input", async () => { + const { sessions, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + fake.setLog(turnId, turnLog(turnId, sessionId, "suspended")); + fake.advanceCalls.length = 0; + await sessions.stopTurn(turnId, "user stop"); + expect(fake.advanceCalls).toEqual([ + { turnId, input: { type: "cancel", reason: "user stop" } }, + ]); + }); + + it("resumeTurn re-enters the latest turn with no input", async () => { + const { sessions, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + fake.advanceCalls.length = 0; + await sessions.resumeTurn(sessionId); + expect(fake.advanceCalls).toEqual([{ turnId, input: undefined }]); + }); + + it("resumeTurn on a session with no turns throws", async () => { + const { sessions } = makeSessions(); + const sessionId = await sessions.createSession(); + await expect(sessions.resumeTurn(sessionId)).rejects.toThrowError( + /no turns to resume/, + ); + }); +}); + +describe("event forwarding and index maintenance (13.6, 13.7)", () => { + it("forwards stream events to the bus tagged with sessionId in order", async () => { + const { sessions, fake, bus } = makeSessions(); + const streamed: TurnStreamEvent[] = [ + { type: "text_delta", turnId: "x", modelCallIndex: 0, delta: "he" }, + { type: "text_delta", turnId: "x", modelCallIndex: 0, delta: "y" }, + ]; + fake.script = () => ({ events: streamed, outcome: completedOutcome() }); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + const turnEvents = bus.events.filter((e) => e.kind === "turn-event"); + expect(turnEvents).toEqual( + streamed.map((event) => ({ + kind: "turn-event", + sessionId, + turnId, + event, + })), + ); + }); + + it("outcome settlement updates the index entry's latest turn status", async () => { + const { sessions, fake, bus } = makeSessions(); + fake.script = () => ({ + outcome: { + status: "suspended", + pendingPermissions: [], + pendingAsyncTools: [ + { toolCallId: "B", toolId: "tool.fetch", toolName: "fetch", input: {} }, + ], + usage: {}, + }, + }); + const sessionId = await sessions.createSession(); + await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + expect(sessions.listSessions()[0].latestTurnStatus).toBe("suspended"); + const last = bus.events[bus.events.length - 1]; + expect(last).toMatchObject({ + kind: "index-changed", + entry: { latestTurnStatus: "suspended" }, + }); + }); + + it("setTitle appends and updates the index preserving status", async () => { + const { sessions, repo } = makeSessions(); + const sessionId = await sessions.createSession(); + await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + await sessions.setTitle(sessionId, "Renamed"); + const events = await (repo as InMemorySessionRepo).read(sessionId); + expect(events[events.length - 1]).toMatchObject({ + type: "title_changed", + title: "Renamed", + }); + expect(sessions.listSessions()[0]).toMatchObject({ + title: "Renamed", + latestTurnStatus: "completed", + }); + }); + + it("deleteSession removes the file and entry; turn files stay; late settles don't resurrect", async () => { + const { sessions, repo, fake, bus } = makeSessions(); + fake.script = () => ({ untilAbort: true }); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + + await sessions.deleteSession(sessionId); + await expect( + (repo as InMemorySessionRepo).read(sessionId), + ).rejects.toThrowError(/session not found/); + expect(sessions.listSessions()).toEqual([]); + expect(bus.events[bus.events.length - 1]).toEqual({ + kind: "index-changed", + sessionId, + entry: null, + }); + // Turn file untouched (orphaned, inert). + expect(fake.logs.has(turnId)).toBe(true); + + // The still-running advance settles after deletion: no resurrection. + await sessions.stopTurn(turnId); + await flush(); + expect(sessions.listSessions()).toEqual([]); + }); +}); + +describe("startup scan (13.6)", () => { + it("builds the index from session files and each latest turn; corrupt files yield errored entries", async () => { + const repo = new InMemorySessionRepo(); + const fake = new FakeTurnRuntime(); + const s1 = "2026-07-02T09-00-00Z-0000001-000"; + const s2 = "2026-07-02T09-00-00Z-0000002-000"; + const s3 = "2026-07-02T09-00-00Z-0000003-000"; + repo.seed([ + { type: "session_created", schemaVersion: 1, sessionId: s1, ts: TS, title: "Done one" }, + { + type: "turn_appended", + sessionId: s1, + ts: TS, + turnId: "t1", + sessionSeq: 1, + agentId: "copilot", + model: FIXTURE_MODEL, + }, + ]); + fake.setLog("t1", turnLog("t1", s1, "completed")); + repo.seed([ + { type: "session_created", schemaVersion: 1, sessionId: s2, ts: TS }, + { + type: "turn_appended", + sessionId: s2, + ts: TS, + turnId: "t2", + sessionSeq: 1, + agentId: "researcher", + model: { provider: "anthropic", model: "claude-x" }, + }, + ]); + fake.setLog("t2", turnLog("t2", s2, "suspended")); + repo.seedCorrupt(s3); + + const { sessions } = makeSessions({ repo, fake }); + await sessions.initialize(); + + const entries = sessions.listSessions(); + expect(entries).toHaveLength(3); + const byId = new Map(entries.map((e) => [e.sessionId, e])); + expect(byId.get(s1)).toMatchObject({ + title: "Done one", + turnCount: 1, + lastAgentId: "copilot", + lastModel: FIXTURE_MODEL, + latestTurnStatus: "completed", + }); + expect(byId.get(s2)).toMatchObject({ + lastAgentId: "researcher", + latestTurnStatus: "suspended", + }); + expect(byId.get(s3)).toMatchObject({ + latestTurnStatus: "none", + error: expect.stringMatching(/corrupt/), + }); + }); + + it("a rebuilt index matches write-through state for the same history", async () => { + const repo = new InMemorySessionRepo(); + const fake = new FakeTurnRuntime(); + const live = makeSessions({ repo, fake }); + const sessionId = await live.sessions.createSession(); + const { turnId } = await live.sessions.sendMessage(sessionId, user("hello world"), { + agent: { agentId: "copilot" }, + }); + await flush(); + fake.setLog(turnId, turnLog(turnId, sessionId, "completed")); + + const rebuilt = makeSessions({ repo, fake }); + await rebuilt.sessions.initialize(); + expect(rebuilt.sessions.listSessions()).toEqual([ + expect.objectContaining({ + sessionId, + title: "hello world", + turnCount: 1, + latestTurnId: turnId, + latestTurnStatus: "completed", + }), + ]); + }); +}); + +describe("headless standalone turns (13.8)", () => { + it("runs a standalone turn with sessionId null, auto permission, no human", async () => { + const fake = new FakeTurnRuntime(); + const { turnId, outcome } = await runHeadlessTurn(fake, { + agent: { agentId: "background-task-agent" }, + input: user("summarize"), + maxModelCalls: 3, + }); + expect(fake.createTurnInputs[0]).toEqual({ + agent: { agentId: "background-task-agent" }, + sessionId: null, + context: [], + input: user("summarize"), + config: { autoPermission: true, humanAvailable: false, maxModelCalls: 3 }, + }); + expect(outcome.status).toBe("completed"); + expect(fake.advanceCalls).toEqual([{ turnId, input: undefined }]); + }); +}); diff --git a/apps/x/packages/core/src/sessions/sessions.ts b/apps/x/packages/core/src/sessions/sessions.ts new file mode 100644 index 00000000..c258c2eb --- /dev/null +++ b/apps/x/packages/core/src/sessions/sessions.ts @@ -0,0 +1,418 @@ +import type { z } from "zod"; +import type { UserMessage } from "@x/shared/dist/message.js"; +import { + SessionCreated, + type SessionEvent, + type SessionIndexEntry, + type SessionLatestTurnStatus, + type SessionState, + reduceSession, + sessionIndexEntry, +} from "@x/shared/dist/sessions.js"; +import { + type JsonValue, + type ModelDescriptor, + type ToolResultData, + deriveTurnStatus, + reduceTurn, +} from "@x/shared/dist/turns.js"; +import type { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js"; +import type { + ITurnRuntime, + Turn, + TurnExecution, + TurnExternalInput, + TurnOutcome, +} from "../turns/api.js"; +import type { IClock } from "../turns/clock.js"; +import { + type ISessions, + type SendMessageConfig, + TurnNotSettledError, +} from "./api.js"; +import type { ISessionBus } from "./bus.js"; +import type { ISessionRepo } from "./repo.js"; +import { SessionIndex } from "./session-index.js"; + +export interface SessionsDependencies { + sessionRepo: ISessionRepo; + turnRuntime: ITurnRuntime; + idGenerator: IMonotonicallyIncreasingIdGenerator; + clock: IClock; + bus: ISessionBus; +} + +// The session layer per session-design.md: owns conversations as ordered +// chains of turn references, enforces one active turn per session, assembles +// context as a reference to the previous turn, and maintains the in-memory +// index write-through. It never reads or writes turn-file contents beyond +// what ITurnRuntime exposes. +export class SessionsImpl implements ISessions { + private readonly sessionRepo: ISessionRepo; + private readonly turnRuntime: ITurnRuntime; + private readonly idGenerator: IMonotonicallyIncreasingIdGenerator; + private readonly clock: IClock; + private readonly bus: ISessionBus; + + private readonly index = new SessionIndex(); + // Ephemeral: executions this process started, for stopTurn's abort path. + private readonly active = new Map< + string, + { sessionId: string | null; controller: AbortController; execution: TurnExecution } + >(); + + constructor({ + sessionRepo, + turnRuntime, + idGenerator, + clock, + bus, + }: SessionsDependencies) { + this.sessionRepo = sessionRepo; + this.turnRuntime = turnRuntime; + this.idGenerator = idGenerator; + this.clock = clock; + this.bus = bus; + } + + // §8.2: scan session files, read each session's latest turn for status. + // Corrupt files yield errored entries; the scan never aborts. + async initialize(): Promise { + for (const sessionId of await this.sessionRepo.listSessionIds()) { + this.index.upsert(await this.scanSession(sessionId)); + } + } + + private async scanSession(sessionId: string): Promise { + try { + const state = reduceSession(await this.sessionRepo.read(sessionId)); + const status = await this.latestTurnStatus(state); + return sessionIndexEntry(state, status); + } catch (error) { + return { + sessionId, + createdAt: "", + updatedAt: "", + turnCount: 0, + latestTurnStatus: "none", + error: errorMessage(error), + }; + } + } + + private async latestTurnStatus( + state: SessionState, + ): Promise { + if (!state.latestTurnId) { + return "none"; + } + const turn = await this.turnRuntime.getTurn(state.latestTurnId); + return deriveTurnStatus(reduceTurn(turn.events)); + } + + async createSession(input?: { title?: string }): Promise { + const sessionId = await this.idGenerator.next(); + const event = SessionCreated.parse({ + type: "session_created", + schemaVersion: 1, + sessionId, + ts: this.clock.now(), + ...(input?.title === undefined ? {} : { title: input.title }), + }); + await this.sessionRepo.create(event); + this.publishEntry(sessionIndexEntry(reduceSession([event]), "none")); + return sessionId; + } + + listSessions(): SessionIndexEntry[] { + return this.index.list(); + } + + async getSession(sessionId: string): Promise { + return reduceSession(await this.sessionRepo.read(sessionId)); + } + + async getTurn(turnId: string): Promise { + return this.turnRuntime.getTurn(turnId); + } + + // §9.1. Write order per §7: turn file first, then turn_appended, then the + // first advance — so an orphan turn (crash between the writes) is benign + // and an executing turn is always referenced. + async sendMessage( + sessionId: string, + input: z.infer, + config: SendMessageConfig, + ): Promise<{ turnId: string }> { + return this.sessionRepo.withLock(sessionId, async () => { + const events = await this.sessionRepo.read(sessionId); + const state = reduceSession(events); + + if (state.latestTurnId) { + const status = await this.latestTurnStatus(state); + if ( + status !== "completed" && + status !== "failed" && + status !== "cancelled" + ) { + throw new TurnNotSettledError( + sessionId, + state.latestTurnId, + status, + ); + } + } + + const turnId = await this.turnRuntime.createTurn({ + agent: config.agent, + sessionId, + context: state.latestTurnId + ? { previousTurnId: state.latestTurnId } + : [], + input, + config: { + humanAvailable: true, + ...(config.autoPermission === undefined + ? {} + : { autoPermission: config.autoPermission }), + ...(config.maxModelCalls === undefined + ? {} + : { maxModelCalls: config.maxModelCalls }), + }, + }); + + const batch: Array> = [ + { + type: "turn_appended", + sessionId, + ts: this.clock.now(), + turnId, + sessionSeq: state.turns.length + 1, + agentId: config.agent.agentId, + model: await this.resolvedModelOf(turnId), + }, + ]; + if (!state.title) { + batch.push({ + type: "title_changed", + sessionId, + ts: this.clock.now(), + title: defaultTitle(input), + }); + } + await this.sessionRepo.append(sessionId, batch); + + this.publishEntry( + sessionIndexEntry(reduceSession([...events, ...batch]), "idle"), + ); + this.startTrackedAdvance(sessionId, turnId); + return { turnId }; + }); + } + + async respondToPermission( + turnId: string, + toolCallId: string, + decision: "allow" | "deny", + metadata?: JsonValue, + ): Promise { + await this.advanceWithInput(turnId, { + type: "permission_decision", + toolCallId, + decision, + ...(metadata === undefined ? {} : { metadata }), + }); + } + + async respondToAskHuman( + turnId: string, + toolCallId: string, + answer: string, + ): Promise { + await this.advanceWithInput(turnId, { + type: "async_tool_result", + toolCallId, + result: { output: answer, isError: false }, + }); + } + + async deliverAsyncToolResult( + turnId: string, + toolCallId: string, + result: z.infer, + ): Promise { + await this.advanceWithInput(turnId, { + type: "async_tool_result", + toolCallId, + result, + }); + } + + async stopTurn(turnId: string, reason?: string): Promise { + const running = this.active.get(turnId); + if (running) { + running.controller.abort(); + await running.execution.outcome.catch(() => undefined); + return; + } + await this.advanceWithInput(turnId, { + type: "cancel", + ...(reason === undefined ? {} : { reason }), + }); + } + + // Recovery entry for idle (crash-interrupted) turns. Deliberately not run + // at startup: recovery re-issues interrupted model calls, so resumption + // must be an explicit action. Runs in the background. + async resumeTurn(sessionId: string): Promise { + const state = reduceSession(await this.sessionRepo.read(sessionId)); + if (!state.latestTurnId) { + throw new Error(`session ${sessionId} has no turns to resume`); + } + this.startTrackedAdvance(sessionId, state.latestTurnId); + } + + async setTitle(sessionId: string, title: string): Promise { + await this.sessionRepo.withLock(sessionId, async () => { + const events = await this.sessionRepo.read(sessionId); + const batch: Array> = [ + { type: "title_changed", sessionId, ts: this.clock.now(), title }, + ]; + await this.sessionRepo.append(sessionId, batch); + const state = reduceSession([...events, ...batch]); + const existing = this.index.get(sessionId); + this.publishEntry( + sessionIndexEntry(state, existing?.latestTurnStatus ?? "none"), + ); + }); + } + + // §9.4: removes the session file and index entry only. Referenced turn + // files stay behind as inert orphans. + async deleteSession(sessionId: string): Promise { + await this.sessionRepo.withLock(sessionId, async () => { + await this.sessionRepo.delete(sessionId); + this.index.remove(sessionId); + this.bus.publish({ kind: "index-changed", sessionId, entry: null }); + }); + } + + // ------------------------------------------------------------------ + // Internals + // ------------------------------------------------------------------ + + private async resolvedModelOf( + turnId: string, + ): Promise> { + const turn = await this.turnRuntime.getTurn(turnId); + const created = turn.events[0]; + if (created.type !== "turn_created") { + throw new Error(`turn ${turnId} has no turn_created event`); + } + return created.agent.resolved.model; + } + + private async sessionIdOf(turnId: string): Promise { + const turn = await this.turnRuntime.getTurn(turnId); + const created = turn.events[0]; + if (created.type !== "turn_created") { + throw new Error(`turn ${turnId} has no turn_created event`); + } + return created.sessionId; + } + + private async advanceWithInput( + turnId: string, + input: TurnExternalInput, + ): Promise { + const sessionId = await this.sessionIdOf(turnId); + const execution = this.startTrackedAdvance(sessionId, turnId, input); + await execution.outcome; + } + + // Every advance this layer initiates: forward its stream to the bus + // tagged with sessionId, keep the abort controller for stopTurn, and + // update the index entry when the outcome settles. + private startTrackedAdvance( + sessionId: string | null, + turnId: string, + input?: TurnExternalInput, + ): TurnExecution { + const controller = new AbortController(); + const execution = this.turnRuntime.advanceTurn(turnId, input, { + signal: controller.signal, + }); + this.active.set(turnId, { sessionId, controller, execution }); + + void (async () => { + try { + for await (const event of execution.events) { + if (sessionId !== null) { + this.bus.publish({ + kind: "turn-event", + sessionId, + turnId, + event, + }); + } + } + } catch { + // Infrastructure failures surface through the outcome. + } + })(); + + void execution.outcome + .then((outcome) => this.onSettled(sessionId, turnId, outcome)) + .catch(() => undefined) + .finally(() => this.active.delete(turnId)); + + return execution; + } + + private onSettled( + sessionId: string | null, + turnId: string, + outcome: TurnOutcome, + ): void { + if (sessionId === null) { + return; + } + const entry = this.index.get(sessionId); + // The session may have been deleted, or a newer turn appended. + if (!entry || entry.latestTurnId !== turnId) { + return; + } + this.publishEntry({ + ...entry, + latestTurnStatus: outcome.status, + updatedAt: this.clock.now(), + }); + } + + private publishEntry(entry: SessionIndexEntry): void { + this.index.upsert(entry); + this.bus.publish({ + kind: "index-changed", + sessionId: entry.sessionId, + entry, + }); + } +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function defaultTitle(input: z.infer): string { + const text = + typeof input.content === "string" + ? input.content + : input.content + .map((part) => (part.type === "text" ? part.text : "")) + .join(" "); + const collapsed = text.trim().replace(/\s+/g, " "); + if (collapsed.length === 0) { + return "New session"; + } + return collapsed.length > 80 ? `${collapsed.slice(0, 79)}…` : collapsed; +} diff --git a/apps/x/packages/shared/src/sessions.ts b/apps/x/packages/shared/src/sessions.ts index 52cfd11b..1a52d38f 100644 --- a/apps/x/packages/shared/src/sessions.ts +++ b/apps/x/packages/shared/src/sessions.ts @@ -1,5 +1,9 @@ import { z } from "zod"; -import { ModelDescriptor, type TurnStatus } from "./turns.js"; +import { + ModelDescriptor, + type TurnStatus, + type TurnStreamEvent, +} from "./turns.js"; // Durable session contract for the session layer (see // packages/core/docs/session-design.md). A session is an append-only chain @@ -167,6 +171,22 @@ export interface SessionIndexEntry { error?: string; } +// What the renderer's single feed consumer receives over IPC: live turn +// stream events (durable events + ephemeral deltas) and index updates. +// entry: null signals deletion. +export type SessionBusEvent = + | { + kind: "turn-event"; + sessionId: string; + turnId: string; + event: TurnStreamEvent; + } + | { + kind: "index-changed"; + sessionId: string; + entry: SessionIndexEntry | null; + }; + export function sessionIndexEntry( state: SessionState, latestTurnStatus: SessionLatestTurnStatus,