feat(x/core): session layer with write-through index (stage 3)

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 <noreply@anthropic.com>

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<typeof RequestedAgent>;
+    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<void>;
+
+    createSession(input?: { title?: string }): Promise<string>;
+    listSessions(): SessionIndexEntry[];
+    getSession(sessionId: string): Promise<SessionState>;
+    getTurn(turnId: string): Promise<Turn>;
+
+    // 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<typeof UserMessage>,
+        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<void>;
+    // The dedicated ask-human endpoint; sendMessage never routes here.
+    respondToAskHuman(
+        turnId: string,
+        toolCallId: string,
+        answer: string,
+    ): Promise<void>;
+    deliverAsyncToolResult(
+        turnId: string,
+        toolCallId: string,
+        result: z.infer<typeof ToolResultData>,
+    ): Promise<void>;
+
+    stopTurn(turnId: string, reason?: string): Promise<void>;
+    // Recovery entry for turns left idle by a crash; runs in the background.
+    resumeTurn(sessionId: string): Promise<void>;
+
+    setTitle(sessionId: string, title: string): Promise<void>;
+    deleteSession(sessionId: string): Promise<void>;
+}
+
+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";
+    }
+}

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

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<typeof SessionCreated> {
+    return {
+        type: "session_created",
+        schemaVersion: 1,
+        sessionId,
+        ts: "2026-07-02T09:00:00Z",
+        title: "T",
+    };
+}
+
+function appended(sessionId = S1): z.infer<typeof SessionEvent> {
+    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/);
+    });
+});

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

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<typeof RequestedAgent>;
+        context?: Array<z.infer<typeof ConversationMessage>>;
+        input: z.infer<typeof UserMessage>;
+        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 };
+}

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<string, Array<z.infer<typeof SessionEvent>>>();
+    private corrupt = new Set<string>();
+    private mutex = new KeyedMutex();
+
+    async create(event: z.infer<typeof SessionCreated>): Promise<void> {
+        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<Array<z.infer<typeof SessionEvent>>> {
+        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<z.infer<typeof SessionEvent>>,
+    ): Promise<void> {
+        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<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
+        return this.mutex.run(sessionId, fn);
+    }
+
+    async listSessionIds(): Promise<string[]> {
+        return [...this.files.keys(), ...this.corrupt].sort();
+    }
+
+    async delete(sessionId: string): Promise<void> {
+        const existed = this.files.delete(sessionId) || this.corrupt.delete(sessionId);
+        if (!existed) {
+            throw new Error(`session not found: ${sessionId}`);
+        }
+    }
+
+    // Test helpers
+    seed(events: Array<z.infer<typeof SessionEvent>>): 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);
+    }
+}

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

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<typeof SessionCreated>): Promise<void>;
+    // Validates every line strictly; corrupt files are rejected whole.
+    read(sessionId: string): Promise<Array<z.infer<typeof SessionEvent>>>;
+    // Validates events before writing.
+    append(
+        sessionId: string,
+        events: Array<z.infer<typeof SessionEvent>>,
+    ): Promise<void>;
+    // In-process per-session exclusion.
+    withLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T>;
+    // Enumerates all session ids for the startup scan.
+    listSessionIds(): Promise<string[]>;
+    // Removes the session file only; referenced turn files stay as orphans.
+    delete(sessionId: string): Promise<void>;
+}

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<string, SessionIndexEntry>();
+
+    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);
+    }
+}

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<typeof TurnEvent>;
+
+const TS = "2026-07-02T10:00:00Z";
+const FIXTURE_MODEL = { provider: "openai", model: "gpt-fixture" };
+const FIXTURE_AGENT: z.infer<typeof ResolvedAgent> = {
+    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<string, TEvent[]>();
+    createError?: Error;
+    script?: AdvanceScript;
+    private n = 0;
+
+    async createTurn(input: CreateTurnInput): Promise<string> {
+        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<TurnStreamEvent, TurnOutcome>();
+        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<Turn> {
+        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<string> {
+        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<InMemorySessionRepo["create"]>[0]) {
+        return this.inner.create(event);
+    }
+    read(sessionId: string) {
+        return this.inner.read(sessionId);
+    }
+    async append(
+        sessionId: string,
+        events: Array<z.infer<typeof SessionEvent>>,
+    ): Promise<void> {
+        if (this.failNextAppend) {
+            this.failNextAppend = false;
+            throw new Error("disk full");
+        }
+        return this.inner.append(sessionId, events);
+    }
+    withLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
+        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<void>((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 }]);
+    });
+});

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<void> {
+        for (const sessionId of await this.sessionRepo.listSessionIds()) {
+            this.index.upsert(await this.scanSession(sessionId));
+        }
+    }
+
+    private async scanSession(sessionId: string): Promise<SessionIndexEntry> {
+        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<SessionLatestTurnStatus> {
+        if (!state.latestTurnId) {
+            return "none";
+        }
+        const turn = await this.turnRuntime.getTurn(state.latestTurnId);
+        return deriveTurnStatus(reduceTurn(turn.events));
+    }
+
+    async createSession(input?: { title?: string }): Promise<string> {
+        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<SessionState> {
+        return reduceSession(await this.sessionRepo.read(sessionId));
+    }
+
+    async getTurn(turnId: string): Promise<Turn> {
+        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<typeof UserMessage>,
+        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<z.infer<typeof SessionEvent>> = [
+                {
+                    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<void> {
+        await this.advanceWithInput(turnId, {
+            type: "permission_decision",
+            toolCallId,
+            decision,
+            ...(metadata === undefined ? {} : { metadata }),
+        });
+    }
+
+    async respondToAskHuman(
+        turnId: string,
+        toolCallId: string,
+        answer: string,
+    ): Promise<void> {
+        await this.advanceWithInput(turnId, {
+            type: "async_tool_result",
+            toolCallId,
+            result: { output: answer, isError: false },
+        });
+    }
+
+    async deliverAsyncToolResult(
+        turnId: string,
+        toolCallId: string,
+        result: z.infer<typeof ToolResultData>,
+    ): Promise<void> {
+        await this.advanceWithInput(turnId, {
+            type: "async_tool_result",
+            toolCallId,
+            result,
+        });
+    }
+
+    async stopTurn(turnId: string, reason?: string): Promise<void> {
+        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<void> {
+        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<void> {
+        await this.sessionRepo.withLock(sessionId, async () => {
+            const events = await this.sessionRepo.read(sessionId);
+            const batch: Array<z.infer<typeof SessionEvent>> = [
+                { 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<void> {
+        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<z.infer<typeof ModelDescriptor>> {
+        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<string | null> {
+        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<void> {
+        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<typeof UserMessage>): 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;
+}

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,