diff --git a/CLAUDE.md b/CLAUDE.md index 176b4f4f..20602021 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -150,6 +150,7 @@ Sessions live in the daemon. Clients are stateless. - **Scrollback buffer**: circular ring (5000 entries / 20MB) replayed on attach - **Transcript persistence**: JSONL per session, user prompts written before API call - **Session resume**: on daemon restart, meta files scanned, sessions rebuilt, scrollback restored +- **Lazy loading** (`session.lazyLoad`, default on): a resumed session holds its metadata only (`Session.prepareResume`: the log's next seq and a mark of each of its files, its newest canonical turn, the turn index) and loads its scrollback and canonical history the first time it is used, exactly as a resume loaded them: the load reads the files as they stood at resume (`LoadTranscriptOptions.asOf`: by identity and size, so whatever was appended since is left out, whatever its seqs; an old log's seqs can go back down), and `#seq` then continues past the highest seq loaded. A transcript that can't be read at start fails the session's resume (logged, left on disk), as before. A transcript, canonical log or turn index that can't be read at a later load fails that use (the session left unloaded, read again on the next). A canonical log that can't be read when the history loads at resume (`lazyLoad: false`, or a session loaded then) leaves the session running without it, as before, but never written over (`Session#historyUnread`): going back or forking from it is refused (`RewindError` `history_unavailable`), a whole rewrite of it is never written (a rotation, which empties it anyway, excepted), turns are still appended, and each use reads it again (`#rereadHistory`, not mid-turn). Entry points await `Session.ensureHydrated()` (one shared load): `SessionManager.handle` for any message naming a session (less `NO_HISTORY_MESSAGES`, less a caller without the scopes its handler needs, `HISTORY_SCOPES`, and less what a loop worker's guard refuses, `#refusedBeforeHistory`; while one waits, every later message naming that session waits behind it, `#historyGate`, so they run in arrival order; a prompt waiting is a preparing send, `Session.holdSend`, which the drain stops and waits out; a client gone meanwhile is detached after), `send`, and the async history readers; `#scrollback` / `#accumulator` are getters that load synchronously, logged, if anything reaches them first (`Session.syncHydrations`). A notice while unloaded is appended at the next seq and kept for the load (`#buffer`); a re-seed owed by a teardown waits for it (`#reseedAfterTeardown`). Sessions from before the canonical log or turn index, or with a fresh backing, load at resume. Detach from terminal -> attach from Telegram -> same session, same context, scrollback replayed. diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index f9293e0b..6841f4b9 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -87,6 +87,9 @@ CODEOID_RESUME_MAX_SESSIONS=200 # sessions restored from disk at startu # Resume is also time-boxed, so raising this costs # startup time; the remainder stays on disk and loads # on a later restart. +CODEOID_SESSION_LAZY_LOAD=true # a resumed session loads its scrollback and history the + # first time it is used (opened, sent to, forked, …), + # not at start; false loads every session's at start CODEOID_IDLE_RELEASE_MINUTES=30 # minutes idle before a session's backend process and # everything it started (MCP servers, commands still # running) are stopped; the next message resumes it diff --git a/e2e/specs/lazy-load.spec.ts b/e2e/specs/lazy-load.spec.ts new file mode 100644 index 00000000..e925c261 --- /dev/null +++ b/e2e/specs/lazy-load.spec.ts @@ -0,0 +1,147 @@ +import { readFileSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import type { Page } from "@playwright/test"; +import { type Codeoid, createSessionInUi, expect, focusSession, messages, openUi, send, test, waitIdle } from "../harness"; + +// A resumed session holds its metadata only, and loads its scrollback and +// history the first time it is opened (`session.lazyLoad`, on by default). + +/** Long enough that a few turns overflow the attach tail set below. */ +const FILLER = "lorem ipsum dolor sit amet ".repeat(5).trim(); +const TURNS = 8; + +/** The sidebar row's text, less its relative "created …" line. */ +async function rowText(page: Page, name: string): Promise { + const row = page.getByRole("listitem").filter({ has: page.getByText(name, { exact: true }) }); + return (await row.innerText()).replace(/created .*$/m, "").trim(); +} + +/** Sessions whose history the daemon has loaded since `from` in its log. */ +function loadsSince(codeoid: Codeoid, from: number): string[] { + return codeoid.log + .slice(from) + .split("\n") + .filter((l) => l.includes("history loaded:")); +} + +/** `codeoid ls`, one block per session, sorted (a resume lists newest first), less what attaching changes. */ +async function ls(codeoid: Codeoid): Promise { + return (await codeoid.cliOk("ls")) + .replace(/clients: \d+/g, "clients: _") + .split(/\n\s*\n/) + .map((b) => b.trim()) + .filter((b) => b && !b.startsWith("Sessions:")) + .sort(); +} + +/** Scroll the transcript (virtualized: only what is in view is rendered). */ +async function scrollTranscript(page: Page, to: "top" | "bottom"): Promise { + await page.getByRole("main").evaluate((el, end) => { + type Box = { scrollHeight: number; clientHeight: number; scrollTop: number }; + for (const n of [el, ...el.querySelectorAll("*")] as unknown as Box[]) { + if (n.scrollHeight > n.clientHeight) n.scrollTop = end === "top" ? 0 : n.scrollHeight; + } + }, to); +} + +/** Page back to the session's first message, then show it. */ +async function pageToTop(page: Page): Promise { + const sentinel = page.getByTestId("older-history-sentinel"); + for (let i = 0; i < 40 && (await sentinel.count()) > 0; i++) { + const more = sentinel.getByRole("button"); + // Scrolling it into view may load the page first: the button goes away. + if ((await more.count()) > 0) await more.click({ timeout: 1_000 }).catch(() => {}); + await page.waitForTimeout(100); + } + await expect(sentinel).toHaveCount(0); + await scrollTranscript(page, "top"); +} + +test("resumed sessions list before they load, and load their whole history when opened, across restarts", async ({ page, browser, codeoid }) => { + test.setTimeout(180_000); + // Memory on (embeddings off: nothing to download), so each turn's usage is + // recorded and restored with the session, as on a default install; and a + // small attach tail, so opening the session pages its history. + const configPath = join(codeoid.configDir, "codeoid", "config.json"); + const config = JSON.parse(readFileSync(configPath, "utf8")) as { session: Record }; + writeFileSync( + configPath, + JSON.stringify({ ...config, memory: { enabled: true, embeddings: { enabled: false } }, session: { ...config.session, attachTailBytes: 2048 } }), + ); + await codeoid.restart(); + await openUi(page, codeoid); + await createSessionInUi(page, "long", codeoid.workdir); + for (let i = 1; i <= TURNS; i++) { + await send(page, `turn ${i} ${FILLER}`); + await expect(messages(page).filter({ hasText: `scripted: turn ${i} ` })).toHaveCount(1); + await waitIdle(page); + } + await codeoid.cliOk("new", "other", codeoid.workdir); + // The sidebar lists another client's new session on its next list. + await page.reload(); + await expect(page.getByText("live", { exact: true })).toBeVisible(); + await focusSession(page, "other"); + await send(page, "hello other"); + await expect(messages(page).filter({ hasText: "scripted: hello other" })).toHaveCount(1); + await waitIdle(page); + const rows = { long: await rowText(page, "long"), other: await rowText(page, "other") }; + expect(rows.long).toContain(`${TURNS}t`); + const listed = await ls(codeoid); + const idOf = (name: string) => /id: ([0-9a-f-]{36})/.exec(listed.find((b) => b.startsWith(`${name} `))!)![1]!; + const ids: Record = { long: idOf("long"), other: idOf("other") }; + const longId = ids.long!; + const loadsOf = (name: string, from: number) => loadsSince(codeoid, from).filter((l) => l.includes(ids[name]!.slice(0, 8))); + + + // This page re-attaches its focused session when it reconnects: park it. + await page.goto("about:blank"); + + for (const round of [1, 2]) { + const from = codeoid.log.length; + await codeoid.restart(); + + // A fresh browser: nothing focused, so nothing is opened. + const ctx = await browser.newContext(); + const fresh = await ctx.newPage(); + const sent: string[] = []; + fresh.on("websocket", (ws) => ws.on("framesent", (f) => sent.push(String(f.payload)))); + await openUi(fresh, codeoid); + await expect(fresh.getByRole("listitem").filter({ has: fresh.getByText("long", { exact: true }) })).toBeVisible(); + // The list says what it said before the restart. The UI opens the newest + // session by itself; the other one, listed but not opened, isn't loaded. + expect(await rowText(fresh, "long")).toBe(rows.long); + expect(await rowText(fresh, "other")).toBe(rows.other); + expect(await ls(codeoid)).toEqual(listed); + const opened = (await fresh.getByRole("heading", { level: 2 }).innerText()).trim(); + const notOpened = opened === "long" ? "other" : "long"; + expect(loadsOf(notOpened, from)).toEqual([]); + + // Opening one loads it (and only it): the newest turns at once, the rest paged. + await focusSession(fresh, "long"); + await expect(messages(fresh).filter({ hasText: `scripted: turn ${TURNS} ` })).toHaveCount(1); + await pageToTop(fresh); + await expect(messages(fresh).filter({ hasText: "scripted: turn 1 " })).toBeVisible(); + // The first turns came by paging: the attach sent only the newest. + expect(sent.some((f) => f.includes('"scrollback.page"') && f.includes(longId))).toBe(true); + // Loaded once, when opened. + expect(loadsOf("long", from)).toHaveLength(1); + + // It takes a new prompt and answers it. + await scrollTranscript(fresh, "bottom"); + await send(fresh, `after restart ${round}`); + await scrollTranscript(fresh, "bottom"); + await expect(messages(fresh).filter({ hasText: `scripted: after restart ${round}` })).toHaveCount(1); + await waitIdle(fresh); + rows.long = await rowText(fresh, "long"); + expect(rows.long).toContain(`${TURNS + round}t`); + await ctx.close(); + } + + // The first browser signs in again and finds the whole history. + await openUi(page, codeoid); + await focusSession(page, "long"); + await expect(messages(page).filter({ hasText: "scripted: after restart 2" })).toHaveCount(1); + await expect(messages(page).filter({ hasText: "scripted: after restart 1" })).toHaveCount(1); + await pageToTop(page); + await expect(messages(page).filter({ hasText: "scripted: turn 1 " })).toBeVisible(); +}); diff --git a/src/config.ts b/src/config.ts index 856a1497..89a22088 100644 --- a/src/config.ts +++ b/src/config.ts @@ -501,6 +501,15 @@ const SessionSchema = z * restart. */ resumeMaxSessions: z.number().int().min(1).default(200), + /** + * Resume sessions without loading their history: a resumed session holds + * its metadata only, and loads its scrollback and canonical history the + * first time something uses them (attach, a prompt, a fork, …). Most + * resumed sessions are never opened before the next restart, and their + * history was most of the daemon's memory. Off = load every session's at + * start, as before. + */ + lazyLoad: z.boolean().default(true), /** * Per-turn workspace snapshots (#354): at the start of every turn the * session's working directory (minus .gitignore'd files, secrets and @@ -556,6 +565,7 @@ const SessionSchema = z mcpToolTimeoutMs: 120_000, attachTailBytes: 512 * 1024, resumeMaxSessions: 200, + lazyLoad: true, seedBudgetChars: 0, seedCharsPerToken: 2, seedWindowFraction: 0.7, @@ -1506,6 +1516,8 @@ export interface CodeoidConfig { idleReleaseMinutes?: number; /** Sessions restored from disk at daemon start, newest-first by last activity. Resume is also time-boxed, so this is the coarse guard; the remainder stays on disk. Defaults to 200 when omitted. */ resumeMaxSessions?: number; + /** Load a resumed session's history the first time it is used, not at start. Defaults to true when omitted. */ + lazyLoad?: boolean; /** Per-turn workspace snapshots (#354). Enabled unless `enabled: false`. */ checkpoints?: { enabled?: boolean; @@ -1817,6 +1829,7 @@ const ENV_OVERRIDES: readonly EnvOverride[] = [ { env: "CODEOID_TURN_STALL_TIMEOUT_MS", path: "session.turnStallTimeoutMs", kind: "int" }, { env: "CODEOID_MCP_TOOL_TIMEOUT_MS", path: "session.mcpToolTimeoutMs", kind: "int" }, { env: "CODEOID_RESUME_MAX_SESSIONS", path: "session.resumeMaxSessions", kind: "int" }, + { env: "CODEOID_SESSION_LAZY_LOAD", path: "session.lazyLoad", kind: "boolean" }, { env: "CODEOID_SEED_BUDGET_CHARS", path: "session.seedBudgetChars", kind: "int" }, { env: "CODEOID_SEED_CHARS_PER_TOKEN", path: "session.seedCharsPerToken", kind: "float" }, { env: "CODEOID_SEED_WINDOW_FRACTION", path: "session.seedWindowFraction", kind: "float" }, diff --git a/src/daemon/dispatch.ts b/src/daemon/dispatch.ts index c19d6e49..ecb6f535 100644 --- a/src/daemon/dispatch.ts +++ b/src/daemon/dispatch.ts @@ -146,6 +146,8 @@ export interface DispatcherHost { workerStatus(sessionId: string): SessionStatus | null; /** Compressed completion digest for a finished worker — never raw transcript. */ buildWorkerDigest(task: DispatchTaskRow): string; + /** Load what `buildWorkerDigest` reads (a resumed worker's history), before it runs. */ + prepareWorkerDigest?(task: DispatchTaskRow): Promise; /** Tear down a finished worker session (best-effort). */ destroyWorker(sessionId: string, reason: string): Promise; /** @@ -1075,6 +1077,7 @@ export class Dispatcher { try { // Digest BEFORE teardown — it reads the live session + memory. + await this.#host.prepareWorkerDigest?.(task); const digest = this.#host.buildWorkerDigest(task); if (status === "idle") { // The read above and this write are not one transaction; the store's diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index 78b3dd6a..899a9547 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -11,7 +11,8 @@ import { existsSync, mkdirSync, realpathSync, statSync } from "node:fs"; import { homedir } from "node:os"; import { dirname, join, resolve, sep } from "node:path"; -import { type ForkPoint, RewindError, SendStoppedError, SHUTDOWN_AUTH, Session, type AttachedClient, type WindowScope } from "./session.js"; +import { type ForkPoint, RewindError, type SendHold, SendStoppedError, SHUTDOWN_AUTH, Session, type AttachedClient, type WindowScope } from "./session.js"; +import type { CanonicalTurn } from "./providers/canonical.js"; import { sweepCheckpoints } from "./checkpoints.js"; import type { CompareRunRow } from "./store.js"; import { type CompareState, type CompareTargetState, type ErrorCode, LIMITS, backendHasAccounts, DEFAULT_ACCOUNT, type AccountKind, type ClaudeAccountKind } from "../protocol/types.js"; @@ -31,7 +32,6 @@ import { applyPatches, getManifest, getSnapshot, previewPatches, readMcpServerEn import { adoptability, adoptableName, adoptedEntry, adoptionRefusal, claudeMcpServers, fromRepoFile, fullUrlRequired, isSameAdoption, shellArg } from "./mcp/adopt.js"; import { readWorkdirServers } from "./mcp/workdir-servers.js"; import { cleanMcpError, commandSummary, MAX_SHOWN_KEYS, quotedVisible, redactMcpUrl, serverProgram, shownLabel, visibleCompact, visibleVerbatim } from "./mcp/redact.js"; -import { canonicalFromTranscript, turnIndexFromHistory } from "./canonical-restore.js"; import { BackendLoginBroker, BackendLoginError, type LoginOutcome, redact } from "./auth/backend-login.js"; import { ClaudeAccountError, ClaudeAccountStore } from "./claude-accounts.js"; import { AccountError, type AccountStore, LOGIN_KINDS, validateAccountName } from "./accounts.js"; @@ -411,6 +411,71 @@ export function cutRowsAfterTurn( */ const RESUME_CANONICAL_MAX_BYTES = 4 * 1024 * 1024; +/** + * Messages that name a session but never read its scrollback or canonical + * history: they don't load a resumed session's (lazy session loading). A + * notice one of them adds is kept for the load (`Session#buffer`); anything + * that does read the history loads it there (`Session#hydrateNow`). + */ +const NO_HISTORY_MESSAGES: ReadonlySet = new Set([ + "session.detach", + "session.destroy", + "session.interrupt", + "session.ui_response", + "session.commands", + "session.set_mode", + "session.set_sandbox", + "session.pin", + "session.unpin", + "session.rename", + "session.turns", + "session.approvals", + "session.set_model", + "session.set_effort", + "session.set_account", + "session.set_claude_account", + "session.mcp_reconnect", + "session.mcp_refresh", + "session.export", + "claude.config", + "mcp.approve", + "mcp.approve.preview", + "fs.list", + "fs.read", + "fs.browse_dir", + "blackboard.index", + "blackboard.read", + "collaboration.panels", + "compare.get", + "compare.list", + "compare.keep", + "mcp.adopt", +]); + +/** + * The scopes each history-reading message's handler requires (every group, + * any scope within one): a caller without them is refused there, so its + * message loads nothing. Types not listed load for any caller of the tenant. + */ +const HISTORY_SCOPES: Readonly>> = { + "session.attach": [[SCOPES.SESSION_ATTACH, SCOPES.SESSION_WATCH]], + "scrollback.page": [[SCOPES.SESSION_ATTACH, SCOPES.SESSION_WATCH]], + "session.send": [[SCOPES.SESSION_SEND]], + "session.part_action": [[SCOPES.SESSION_SEND]], + "session.rotate": [[SCOPES.SESSION_SEND]], + "session.rewind": [[SCOPES.SESSION_SEND]], + "session.approve": [[SCOPES.SESSION_APPROVE]], + "session.set_provider": [[SCOPES.SESSION_APPROVE]], + "session.fork": [[SCOPES.SESSION_CREATE]], + "session.compare": [[SCOPES.SESSION_CREATE], [SCOPES.SESSION_SEND]], +}; + +/** Whether `auth` holds the scopes `type`'s handler requires before it reads a session's history. */ +function mayReadHistory(type: string, auth: AuthContext): boolean { + const groups = HISTORY_SCOPES[type]; + return !groups || groups.every((any) => any.some((scope) => hasScope(auth.scopes as string[], scope))); +} + /** Sort key for resume ordering: most-recently-active first. Falls back to * createdAt, then 0, so a malformed timestamp never throws. */ function resumeSortKey(m: { lastActivityAt?: string; createdAt?: string }): number { @@ -427,6 +492,12 @@ const LOOP_WORKER_PROVIDERS: ReadonlySet = new Set(["claude"]); export class SessionManager { #sessions = new Map(); + /** Per session, the newest message waiting on its history's load (`#historyGate`): the next one waits behind it. */ + #historyGates = new Map>(); + /** Clients with messages waiting on a load, and how many. */ + #clientsWaiting = new Map(); + /** Of those, the ones that disconnected meanwhile (`disconnectClient`). */ + #clientsGone = new Set(); #store: Store; #transcriptStore: TranscriptStore; #identityManager?: AgentIdentityManager; @@ -959,6 +1030,7 @@ export class SessionManager { if (m.collaboration) goalConfigs.set(m.sessionId, m.collaboration); } const deadline = Date.now() + RESUME_DEADLINE_MS; + const lazyLoad = this.#config?.session?.lazyLoad ?? true; let resumed = 0; let skippedDeadline = 0; let resumedChildren = 0; @@ -1083,54 +1155,22 @@ mcpHub: this.#mcpHub, session.adoptReportedBackgroundTasks(meta.reportedBackgroundTasks); session.adoptKnownBackgroundTasks(meta.knownBackgroundTasks); - // Restore scrollback from transcript, seeding the seq counter past - // the persisted tail so new appends continue the monotonic sequence. - // Byte-budgeted + deadline-aware: one huge transcript can neither - // OOM the daemon nor eat the whole resume window by itself. - const loadStats: { truncated?: boolean } = {}; - const entries = await this.#transcriptStore.loadTranscript(meta.sessionId, { - maxBytes: RESUME_TRANSCRIPT_MAX_BYTES, - deadlineAt: deadline, - stats: loadStats, - }); - const messages = entries.map((e) => e.message); - const maxSeq = entries.reduce((max, e) => Math.max(max, e.seq), -1); - session.restoreScrollback(messages, maxSeq + 1, entries.map((e) => e.bytes), { - partialHistory: loadStats.truncated === true, - }); - // The backend-neutral conversation (#354). Without it a restored - // session forked / switched backend with no history, and stateless - // backends answered the next prompt with no memory of the session. - // Sessions from before the log existed are rebuilt from the - // transcript once, then read from the log. - try { - const canonical = await this.#transcriptStore.loadCanonical(meta.sessionId, { - maxBytes: RESUME_CANONICAL_MAX_BYTES, - }); - const index = await this.#transcriptStore.loadTurnIndex(meta.sessionId); - // No log: a session from before it existed. Rebuild once from the - // WHOLE transcript (the scrollback window above may be a tail) and - // persist it — a long session's history must not shrink to the - // window, or to just the next turn on the following restart. - let rebuildFrom = messages; - if (!canonical && loadStats.truncated === true) { - const all = await this.#transcriptStore.loadTranscript(meta.sessionId, { deadlineAt: deadline }); - rebuildFrom = all.map((e) => e.message); - } - const history = - canonical?.turns ?? (rebuildFrom.length > 0 ? canonicalFromTranscript(rebuildFrom, providerId) : []); - session.restoreTurns(history, index ?? turnIndexFromHistory(history), { - persistHistory: !canonical && history.length > 0, - persistIndex: !index && history.some((t) => t.turnId), - partial: canonical?.partial ?? false, - }); - if (meta.redactedBackings) session.adoptRedactedBackings(meta.redactedBackings); - if (meta.backingFresh) await session.reseedFreshBacking(); - } catch (err) { - console.error( - `[codeoid/resume] canonical history restore failed for ${meta.sessionId} (session resumes without it): ${err instanceof Error ? err.message : String(err)}`, - ); - } + // The history (scrollback + canonical conversation) loads the first + // time something uses the session (`session.lazyLoad`, default on): + // only what it reports and appends with before then is read here. + // Sessions from before the canonical log or the turn index load now, + // since the load rebuilds and persists them. + if (meta.redactedBackings) session.adoptRedactedBackings(meta.redactedBackings); + const loadNow = !(await session.prepareResume({ + transcriptMaxBytes: RESUME_TRANSCRIPT_MAX_BYTES, + canonicalMaxBytes: RESUME_CANONICAL_MAX_BYTES, + deadlineMs: RESUME_DEADLINE_MS, + providerId, + backingFresh: meta.backingFresh === true, + })); + // A fresh backing is re-seeded at resume, as before: it is rare (a + // rewind not yet followed by a turn) and its backend is owed it now. + if (loadNow || !lazyLoad || meta.backingFresh) await session.ensureHydrated({ deadlineAt: deadline, atResume: true }); this.#sessions.set(session.id, session); // Track a resumed child's mount so teardown revokes it. Without this a @@ -1151,8 +1191,12 @@ mcpHub: this.#mcpHub, // "Concurrent session limit reached" until the user // /destroys some. resumed++; - } catch { - // Skip sessions that fail to resume + } catch (err) { + // Skip sessions that fail to resume (a transcript that can't be + // read): left on disk as they are, tried again at the next start. + console.error( + `[codeoid/resume] could not resume ${meta.sessionId}: ${err instanceof Error ? err.message : String(err)}`, + ); } } @@ -1332,6 +1376,49 @@ mcpHub: this.#mcpHub, */ rawToken?: string; }, + ): Promise { + // A resumed session loads its history the first time it is used. Only + // then is there an await here: handlers that claim state synchronously + // (a rewind's stop point) keep doing so before the caller's next call. + const gate = this.#historyGate(msg, auth); + if (!gate) return this.#route(msg, auth, client, opts); + return this.#routeAfterLoad(gate, msg, auth, client, opts); + } + + /** `handle`, once the session `msg` names has loaded its history (`#historyGate`). */ + async #routeAfterLoad( + gate: { session: Session; wait: Promise }, + msg: ClientMessage, + auth: AuthContext, + client: AttachedClient, + opts?: { rawToken?: string }, + ): Promise { + // A prompt waiting on the load is a send preparing its turn: the drain + // sees it, and a Stop meanwhile stops it, as a queued send. + const hold = msg.type === "session.send" ? gate.session.holdSend() : undefined; + this.#clientsWaiting.set(client.id, (this.#clientsWaiting.get(client.id) ?? 0) + 1); + try { + const unloadable = await gate.wait; + if (unloadable) return unloadable; + return await this.#route(msg, auth, client, opts, hold); + } finally { + hold?.release(); + const waiting = (this.#clientsWaiting.get(client.id) ?? 1) - 1; + if (waiting > 0) this.#clientsWaiting.set(client.id, waiting); + else { + this.#clientsWaiting.delete(client.id); + // Gone while its messages waited: what they attached it to lets go of it. + if (this.#clientsGone.delete(client.id)) this.disconnectClient(client.id); + } + } + } + + async #route( + msg: ClientMessage, + auth: AuthContext, + client: AttachedClient, + opts?: { rawToken?: string }, + sendHold?: SendHold, ): Promise { switch (msg.type) { case "ping": @@ -1365,7 +1452,7 @@ mcpHub: this.#mcpHub, case "session.detach": return this.#detach(msg, client); case "session.send": - return this.#send(msg, auth); + return this.#send(msg, auth, sendHold); case "session.interrupt": return this.#interrupt(msg, auth); case "session.approve": @@ -1582,6 +1669,51 @@ mcpHub: this.#mcpHub, } } + /** + * Load the history of the session `msg` names (lazy session loading), so + * every handler finds it in memory: the one place a client's use of a + * session loads it. Only for a session the caller's tenant owns, with the + * scope its handler needs (`HISTORY_SCOPES`), and not for messages that + * never read it (`NO_HISTORY_MESSAGES`). While a load is awaited here, + * every later message naming that session waits behind it, whether it + * reads the history or not, so they are handled in the order they came + * (an interrupt never overtakes the prompt it stops); other sessions' + * don't. Null when nothing waits; `wait` resolves to an error response + * when the transcript can't be read. + */ + #historyGate(msg: ClientMessage, auth: AuthContext): { session: Session; wait: Promise } | null { + const id = (msg as { sessionId?: unknown }).sessionId; + if (typeof id !== "string") return null; + const session = this.#sessions.get(id); + if (!session || session.accountId !== auth.accountId || session.projectId !== auth.projectId) return null; + const load = (): Promise | null => { + if (NO_HISTORY_MESSAGES.has(msg.type) || !session.historyPending || !mayReadHistory(msg.type, auth)) return null; + if (this.#refusedBeforeHistory(msg, auth)) return null; + return session.ensureHydrated().then( + () => null, + (err): DaemonMessage => { + console.error( + `[codeoid/resume] could not load the history of ${id}: ${err instanceof Error ? err.message : String(err)}`, + ); + return { + type: "response.error", + requestId: (msg as { id?: string }).id ?? "", + error: "The session's history could not be loaded from disk — try again, or see the daemon's log.", + code: "internal", + }; + }, + ); + }; + const behind = this.#historyGates.get(id); + const wait = behind ? behind.then(load) : load(); + if (!wait) return null; + this.#historyGates.set(id, wait); + void wait.finally(() => { + if (this.#historyGates.get(id) === wait) this.#historyGates.delete(id); + }); + return { session, wait }; + } + async #sessionExport( msg: Extract, auth: AuthContext, @@ -1808,6 +1940,10 @@ mcpHub: this.#mcpHub, }, ); + // The imported log's lines are seqs 1..N: the session's own appends + // continue past them, so its seqs only grow along the log. + this.#sessions.get(result.newSessionId)?.restoreScrollback([], result.importedMessages + 1); + return { type: "session.import.result", requestId: msg.id, @@ -3906,6 +4042,9 @@ mcpHub: this.#mcpHub, /** Remove a client from all sessions (e.g. on disconnect). */ disconnectClient(clientId: string): void { + // A message of its own still waiting on a session's load runs after + // this: whatever it attaches the client to is let go of then. + if (this.#clientsWaiting.has(clientId)) this.#clientsGone.add(clientId); for (const session of this.#sessions.values()) { session.detach(clientId); } @@ -4019,7 +4158,9 @@ mcpHub: this.#mcpHub, } waiting++; } - if (waiting === 0) break; + // A message waiting on a session's load is handled after it: wait for + // it too (a prompt among them is preparing, and is stopped above). + if (waiting === 0 && this.#historyGates.size === 0) break; await Bun.sleep(Math.min(pollMs, remaining())); pollMs = Math.min(pollMs * 2, 50); } @@ -5054,26 +5195,26 @@ mcpHub: this.#mcpHub, code: "invalid_request", }; } - if (msg.afterTurnId) { - try { - point = await parent.forkPoint(msg.afterTurnId); - } catch (err) { - return { - type: "response.error", - requestId: msg.id, - error: err instanceof Error ? err.message : String(err), - code: - err instanceof RewindError - ? err.code === "not_found" - ? "not_found" - : err.code === "restore_failed" - ? "internal" - : "invalid_request" // before_rotation, busy: pick another point / wait - : "internal", - }; - } + let parentHistory: readonly CanonicalTurn[]; + try { + if (msg.afterTurnId) point = await parent.forkPoint(msg.afterTurnId); + parentHistory = point ? point.history : await parent.fullCanonicalHistory(); + } catch (err) { + return { + type: "response.error", + requestId: msg.id, + error: err instanceof Error ? err.message : String(err), + code: + err instanceof RewindError + ? err.code === "not_found" + ? "not_found" + : err.code === "restore_failed" || err.code === "history_unavailable" + ? "internal" + : "invalid_request" // before_rotation, busy: pick another point / wait + : "internal", + }; } - const history = (point ? point.history : await parent.fullCanonicalHistory()).map((t) => ({ ...t })); + const history = parentHistory.map((t) => ({ ...t })); // From an earlier turn: the parent's background launches after that turn // are its own business, not news for the fork's model (they stay foreign). // Only the parent's unfinished work matters, and it is looked for in the @@ -5678,7 +5819,9 @@ mcpHub: this.#mcpHub, // COMPLETE (or ask for input) on GENUINELY NEW final text — never on the // PRIOR phase's ⟦PHASE-COMPLETE⟧ that still sits in lastAssistantText (e.g. // behind a tools-only turn whose own text is empty). Without this, phase N - // "completes" instantly by reading phase N-1's marker. + // "completes" instantly by reading phase N-1's marker. A pipeline resumed + // after a restart drives a session whose history may not be loaded yet. + if (session.historyPending) await session.ensureHydrated(); let textAtSend = session.lastAssistantText ?? ""; // Every NEW assistant text this phase produced, in order. The phase's // summary is the whole of what the model said across its turns, not only @@ -6712,6 +6855,27 @@ mcpHub: this.#mcpHub, }; } + /** + * A history-reading message its handler refuses before it reads the + * history, for a reason other than scope (`#loopWorkerGuard`, a delegated + * approval of a loop worker's): it loads nothing. A handler that goes on + * after all loads the history itself. + */ + #refusedBeforeHistory(msg: ClientMessage, auth: AuthContext): boolean { + const id = (msg as { sessionId?: unknown }).sessionId; + if (typeof id !== "string") return false; + switch (msg.type) { + case "session.rotate": + case "session.rewind": + case "session.set_provider": + return this.#isLiveLoopWorker(id); + case "session.approve": + return auth.delegationDepth > 0 && this.#isLiveLoopWorker(id); + default: + return false; + } + } + /** The session works in an active loop run's worktree: its current worker, * or an earlier one (replaced after a restart) still holding the same tree. */ #isLiveLoopWorker(sessionId: string): boolean { @@ -7269,6 +7433,11 @@ mcpHub: this.#mcpHub, workerStatus: (sessionId: string) => this.#sessions.get(sessionId)?.status ?? null, + prepareWorkerDigest: async (task: DispatchTaskRow): Promise => { + const worker = this.#sessionForTask(task.workerSessionId, task); + if (worker?.historyPending) await worker.ensureHydrated().catch(() => {}); + }, + buildWorkerDigest: (task: DispatchTaskRow): string => { const worker = this.#sessionForTask(task.workerSessionId, task); const header = `task ${task.id.slice(0, 8)} (${task.kind}/${task.shape}) in ${task.workdir ?? task.targetSession ?? "?"}`; @@ -8203,6 +8372,8 @@ mcpHub: this.#mcpHub, async #freezeBranch(t: CompareRunRow["targets"][number], s: Session): Promise { const turnId = t.baseTurnId!; + // A branch settled after a restart may not have loaded its history yet. + if (s.historyPending) await s.ensureHydrated().catch(() => {}); const status: "idle" | "error" = s.status === "error" ? "error" : "idle"; // Nothing after the compared turn yet: the branch's totals are that turn's // (a fork starts with none). Read before awaiting — the user may go on. @@ -8449,7 +8620,7 @@ mcpHub: this.#mcpHub, type: "response.error", requestId: msg.id, error: err.message, - code: err.code === "not_found" ? "not_found" : "invalid_request", + code: err.code === "not_found" ? "not_found" : err.code === "history_unavailable" ? "internal" : "invalid_request", }; } throw err; @@ -8495,6 +8666,7 @@ mcpHub: this.#mcpHub, #send( msg: Extract, auth: AuthContext, + hold?: SendHold, ): DaemonMessage | Promise { if (!hasScope(auth.scopes as string[], SCOPES.SESSION_SEND)) { return { type: "response.error", requestId: msg.id, error: "Missing scope: session:send", code: "forbidden" }; @@ -8543,7 +8715,7 @@ mcpHub: this.#mcpHub, // returned a false "ok" and dropped the error — silent data loss). // priority controls mid-turn insertion semantics (default "later" = FIFO). session - .send(msg.text, auth, msg.attachments, msg.priority) + .send(msg.text, auth, msg.attachments, msg.priority, hold) .catch((err) => { // Stopped before its turn started: the user already got a notice. if (err instanceof SendStoppedError) return; diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 656ddaca..2a8d87a8 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -130,7 +130,8 @@ import type { Store } from "./store.js"; import type { AgentIdentityManager } from "./agent-identity.js"; import { ScrollbackBuffer } from "./scrollback.js"; import { reconcileResumedMessage } from "./resume-reconcile.js"; -import type { TranscriptStore } from "./transcript.js"; +import type { TranscriptEntry, TranscriptFileMark, TranscriptStore } from "./transcript.js"; +import { canonicalFromTranscript, turnIndexFromHistory } from "./canonical-restore.js"; import { backgroundLaunchIds, FOREIGN_AGENT_DENY, @@ -755,7 +756,7 @@ export interface ForkPoint { /** A rewind that can't proceed (unknown turn, a turn that wouldn't stop). */ export class RewindError extends Error { constructor( - readonly code: "not_found" | "busy" | "restore_failed" | "before_rotation", + readonly code: "not_found" | "busy" | "restore_failed" | "before_rotation" | "history_unavailable", message: string, ) { super(message); @@ -873,6 +874,90 @@ export function writesClaudeSettingsFile(target: string, workdir: string): boole * counts as spending. */ /** What one turn's event consumer counted (#430): model work, requests sent, usage reports received. */ +/** + * How a resumed session loads its history (`Session.prepareResume`): the + * resume read budgets, which SessionManager owns, and what its meta says. + */ +export interface ResumeHistoryPlan { + /** Newest transcript bytes read into the scrollback. */ + transcriptMaxBytes: number; + /** Newest canonical-log bytes read into the history. */ + canonicalMaxBytes: number; + /** Time a load may spend parsing the transcript before it keeps what it has. */ + deadlineMs: number; + /** The backend a history rebuilt from the transcript is for (a session from before the log). */ + providerId: string; + /** The meta's `backingFresh`: the backend is re-seeded from the history once it is loaded. */ + backingFresh: boolean; +} + +/** + * A resumed session whose history isn't loaded yet (lazy session loading): + * what it needs meanwhile, and what the load must add. + */ +interface Dehydrated { + plan: ResumeHistoryPlan; + /** + * The log's files as they were at resume: the load reads them up to there, + * exactly what an eager resume read, whatever this process appended since + * and whatever their seqs. + */ + files: TranscriptFileMark[]; + /** Messages appended since (on disk too), pushed onto the scrollback after the load, as if live. */ + appended: SessionMessage[]; + /** The session had a turn index on disk (else the load rebuilds it from the history). */ + hadIndex: boolean; + /** The turn index or canonical log couldn't be read at resume: the load reads the index again, and fails as a resume did. */ + indexUnread?: boolean; + /** The newest canonical turn, for `lastTurnId` / `lastTurnRole`. */ + lastTurn: Pick | null; + /** A teardown owed the backend a re-seed from the history (`#reseedAfterTeardown`): paid on load. */ + reseed: boolean; + /** The load in flight. */ + loading?: Promise; +} + +/** + * The canonical log (or turn index) couldn't be read when a resume loaded + * the history (`Session#historyUnread`): what the session holds in memory is + * only what came since. + */ +interface HistoryUnread { + plan: ResumeHistoryPlan; + /** The turn index wasn't read either: it is read again with the log. */ + indexUnread: boolean; + /** The last read's error, logged once per distinct one. */ + error: string; + /** The read in flight. */ + retry?: Promise; +} + +/** Why going back or forking is refused while the log couldn't be read. */ +const HISTORY_UNREAD_REFUSAL = + "This session's earlier conversation couldn't be read from disk, so going back or forking now would lose it. Nothing was changed — try again, or see the daemon's log."; + +/** A send counted as preparing before it reaches `Session.send` (`Session.holdSend`). */ +export interface SendHold { + /** The Stop generation when it arrived: a Stop since stops it. */ + readonly stopGen: number; + /** Take the count over (true once; false after `take` or `release`). */ + take(): boolean; + /** Give the count back, unless taken. */ + release(): void; +} + +/** What a history load read, for `Session#installHistory`. */ +interface LoadedHistory { + entries: TranscriptEntry[]; + truncated: boolean; + canonical: { turns: CanonicalTurn[]; partial: boolean } | null; + /** The whole transcript, when there's no log to read and the scrollback read was a tail. */ + rebuildFrom?: DaemonMessage[]; + /** The turn index, read by the load (`Dehydrated.indexUnread`). */ + index?: TurnIndexEntry[] | null; + canonicalError?: unknown; +} + interface TurnCounts { activity: boolean; requests: number; @@ -1034,7 +1119,8 @@ export class Session { #transcriptStore: TranscriptStore; #identityManager?: AgentIdentityManager; #agentIdentity: MessageIdentity; - #scrollback = new ScrollbackBuffer(); + /** The replay buffer. Read it as `#scrollback`, which loads a resumed session's history first. */ + #scrollbackBuffer = new ScrollbackBuffer(); /** * Identity of this Session instance's replay buffer (`replay.resume`). * A client cursor (`sinceSeq`) is only valid against the buffer that @@ -1090,7 +1176,8 @@ export class Session { // this to detect a turn whose event stream has gone silent (hung tool / dead // subprocess) so the session can self-recover instead of wedging forever. #lastEventAt = 0; - #accumulator = new CanonicalHistoryAccumulator(); + /** The canonical history. Read it as `#accumulator`, which loads a resumed session's history first. */ + #historyAccumulator = new CanonicalHistoryAccumulator(); /** * The turn messages are currently attributed to (#354): minted when a * prompt (or a background continuation) starts a turn, stamped on every @@ -1108,7 +1195,7 @@ export class Session { * bounded resume read). Features that need the whole conversation (going * back to, or forking from, an old turn) load the full log on demand. */ - #canonicalPartial = false; + #canonicalPartialFlag = false; /** Pluggable seed policy for switch/fork. Default `transcript` (no change); * `CODEOID_CONTEXT_STRATEGY=vws` opts into the compact session map. */ #contextStrategy: ContextStrategy = selectContextStrategy(); @@ -1222,8 +1309,8 @@ export class Session { } | null = null; /** Set by a turn_done whose seed overflowed; the consumer's exit retries it. */ #seedOverflowRetry = false; - /** The last turn failed with "prompt too long" — /rotate then skips its min-turns guard. */ - #promptTooLong = false; + /** The last turn failed with "prompt too long" — /rotate then skips its min-turns guard. Derived from the scrollback on resume. */ + #promptTooLongFlag = false; // Execution mode + turn budget (autonomous mode only). // Default `guarded` (≈ Claude Code's default): read-only tools (Read/Grep/Glob // + memory) auto-approve, while Write/Edit/Bash and other mutations prompt. @@ -1626,8 +1713,14 @@ export class Session { // Durable canonical history (#354): every change is appended to the // session's log so forks, backend switches and rewinds after a restart // still carry the conversation. - this.#accumulator.onChange = (change) => { + this.#historyAccumulator.onChange = (change) => { if (this.#destroyed) return; + // Never write a history that couldn't be read over the log it lost; + // what is appended still is (`#historyUnread`). + if (change.op === "replace" && this.#historyUnread) { + console.error(`[codeoid/session ${this.id}] canonical history not rewritten: the log couldn't be read, and would lose its turns`); + return; + } void this.#transcriptStore.recordCanonical(this.id, change); }; this.#identityManager = opts.identityManager; @@ -2075,9 +2168,7 @@ export class Session { // A backend that can't resume its own conversation (the Gemini CLI // starts a new ACP session) is handed the history, as after an MCP // rebuild — an account switch must never start the conversation over. - if (RESEED_AFTER_TEARDOWN.has(this.#provider.id) && this.#accumulator.history.length > 0) { - await this.#seedProviderFromHistory(); - } + await this.#reseedAfterTeardown(); }) .catch((e) => { console.error(`[codeoid/session ${this.id}] releasing the idle ${backend} backend failed: ${e instanceof Error ? e.message : String(e)}`); @@ -2385,6 +2476,7 @@ export class Session { // Rejecting here means we never tear down an actively-running provider. const busy = this.#switchBusyReason(); if (busy) return busy; + if (this.#loadPending) await this.ensureHydrated(); const previous = this.#provider.id; @@ -2528,6 +2620,8 @@ export class Session { } /** turnId of the newest turn in the canonical history (null before any, or when unstamped). */ get lastTurnId(): string | null { + // Before the history loads: the newest turn of its log. + if (this.#dehydrated) return this.#dehydrated.lastTurn?.turnId ?? null; return this.#accumulator.history.at(-1)?.turnId ?? null; } /** Number of turns (user + assistant) in the canonical history. */ @@ -2563,6 +2657,7 @@ export class Session { * length watermark was not). */ get lastTurnRole(): "user" | "assistant" | undefined { + if (this.#dehydrated) return this.#dehydrated.lastTurn?.role; const h = this.#accumulator.history; return h[h.length - 1]?.role; } @@ -2890,9 +2985,7 @@ export class Session { } if (view.fingerprint === this.#mcpBuiltWith) return false; await this.#teardownProvider(); // records what's applied, and lets go of what was removed - if (RESEED_AFTER_TEARDOWN.has(this.#provider.id) && this.#accumulator.history.length > 0) { - await this.#seedProviderFromHistory(); - } + await this.#reseedAfterTeardown(); return true; } @@ -3575,6 +3668,7 @@ export class Session { maxBytes?: number, ): Promise<{ messages: SessionMessage[]; hasMore: boolean; source: "buffer" | "transcript" }> { const budget = Math.min(Math.max(1, maxBytes ?? PAGE_DEFAULT_BYTES), PAGE_MAX_BYTES); + if (this.#loadPending) await this.ensureHydrated(); if (beforeMessageId === undefined) { const tail = this.#scrollback.readTailChunked(budget, budget); @@ -3828,6 +3922,7 @@ export class Session { data: Record | undefined, sender: AuthContext, ): Promise<{ ok: true } | { ok: false; code: "not_found" | "invalid_request"; error: string }> { + if (this.#loadPending) await this.ensureHydrated(); const msg = this.#scrollback.getMessage(messageId) as SessionMessage | undefined; if (!msg) { return { ok: false, code: "not_found", error: "Message not found" }; @@ -3901,6 +3996,8 @@ export class Session { sender: AuthContext, attachments?: readonly Attachment[], priority?: "now" | "next" | "later", + /** Counted as preparing since it arrived (`holdSend`): this send takes it over. */ + hold?: SendHold, ): Promise { // Shutting down: nothing is recorded, nothing starts. if (this.#draining) throw new DaemonStoppingError(); @@ -3914,11 +4011,17 @@ export class Session { // A send is "preparing" from the moment it is queued until its turn has // started (or it failed): drain() must treat that as work, and a Stop in // that window must cancel it — see #stoppedBeforeStart. - const stopGen = this.#stopGen; - this.#preparingSends++; + const held = hold?.take() === true; + const stopGen = held ? hold!.stopGen : this.#stopGen; + if (!held) this.#preparingSends++; const next = this.#sendChain .catch(() => undefined) - .then(() => this.#sendInner(text, sender, attachments, priority, stopGen)) + // A resumed session loads its history before its first turn. + .then(() => + this.#dehydrated || this.#historyUnread + ? this.ensureHydrated().then(() => this.#sendInner(text, sender, attachments, priority, stopGen)) + : this.#sendInner(text, sender, attachments, priority, stopGen), + ) .finally(() => { this.#preparingSends--; }); @@ -3929,6 +4032,30 @@ export class Session { return next; } + /** + * Count a send as preparing before it reaches send(): one the daemon holds + * while the session's history loads (`SessionManager.handle`). The drain + * sees it, and a Stop meanwhile stops it as it stops a queued send. Hand + * it to send(), which takes it over, or release() it. + */ + holdSend(): SendHold { + this.#preparingSends++; + let held = true; + return { + stopGen: this.#stopGen, + take: () => { + const was = held; + held = false; + return was; + }, + release: () => { + if (!held) return; + held = false; + this.#preparingSends--; + }, + }; + } + /** Sends queued or preparing their turn (see send()). */ #preparingSends = 0; /** Bumped by every Stop, so a send queued before it knows it was stopped. */ @@ -6459,11 +6586,383 @@ export class Session { }; } + // ── Lazy session loading ────────────────────────────────────────────── + // + // A resumed session starts with its metadata only (`prepareResume`): its + // scrollback and canonical history stay on disk until something uses them, + // and load exactly as a resume used to load them. Entry points await + // `ensureHydrated()`; the synchronous internals read the two through + // `#scrollback` / `#accumulator`, which load them first if nothing has + // (`#hydrateNow`, logged: a path that should have awaited). + + /** Set while the history isn't loaded; null once it is (and for every session not resumed). */ + #dehydrated: Dehydrated | null = null; + /** Work the load leaves the backend (a re-seed), awaited by `ensureHydrated()`. */ + #afterHydrate: Promise | null = null; + /** + * The canonical log couldn't be read when a resume loaded the history: the + * session runs without it, as a resume always did, but the history in + * memory is only what came since. It is never written over the log (going + * back or forking is refused, `#assertHistoryRead`), turns are still + * appended to it, and every use reads it again (`#rereadHistory`) until it + * can. Null otherwise. + */ + #historyUnread: HistoryUnread | null = null; + /** Loads `#hydrateNow` had to do synchronously, daemon-wide (tests read it). */ + static syncHydrations = 0; + + get #scrollback(): ScrollbackBuffer { + if (this.#dehydrated) this.#hydrateNow("scrollback"); + return this.#scrollbackBuffer; + } + + get #accumulator(): CanonicalHistoryAccumulator { + if (this.#dehydrated) this.#hydrateNow("history"); + return this.#historyAccumulator; + } + + get #canonicalPartial(): boolean { + if (this.#dehydrated) this.#hydrateNow("history"); + return this.#canonicalPartialFlag; + } + + set #canonicalPartial(v: boolean) { + if (this.#dehydrated) this.#hydrateNow("history"); + this.#canonicalPartialFlag = v; + } + + get #promptTooLong(): boolean { + if (this.#dehydrated) this.#hydrateNow("scrollback"); + return this.#promptTooLongFlag; + } + + set #promptTooLong(v: boolean) { + if (this.#dehydrated) this.#hydrateNow("scrollback"); + this.#promptTooLongFlag = v; + } + + /** The history isn't in place yet (or couldn't be read), or a load's re-seed is still running: await `ensureHydrated()`. */ + get #loadPending(): boolean { + return this.#dehydrated !== null || this.#afterHydrate !== null || this.#historyUnread !== null; + } + + /** `ensureHydrated()` has work to do (a load, or a load's re-seed still running). */ + get historyPending(): boolean { + return this.#loadPending; + } + + /** Whether the scrollback and canonical history are in memory. */ + get hydrated(): boolean { + return this.#dehydrated === null; + } + + /** + * Resume this session from disk without loading its history: only what it + * reports and appends with before the load — the seq its log continues + * from, the newest canonical turn, the turn index (small) — is read, from + * the files' ends. `#seq` continues past the log, the backend is told it + * has a conversation to resume, and a fresh backing's flags are set, as a + * full resume sets them. Returns false when the history must be loaded now: + * a session from before the canonical log or the turn index (the load + * rebuilds and persists them), or one whose log or index couldn't be read + * (the load fails as a resume did: the history is resumed without). + * Rejects when the transcript is there but can't be read: the session + * can't resume, as when a resume couldn't load it — never as one with no + * history, whose seq starts over and whose backend starts a new conversation. + */ + async prepareResume(plan: ResumeHistoryPlan): Promise { + const tail = await this.#transcriptStore.transcriptTail(this.id); + let canonical: Awaited> = null; + let index: TurnIndexEntry[] | null = null; + let indexUnread = false; + try { + [canonical, index] = await Promise.all([ + this.#transcriptStore.canonicalTail(this.id), + this.#transcriptStore.loadTurnIndex(this.id), + ]); + } catch { + indexUnread = true; + } + if (tail.nextSeq > this.#seq) this.#seq = tail.nextSeq; + // A session with prior scrollback already exists in its backend's own + // store — the next send() must resume it, not re-create it. + if (tail.hasEntries) this.#provider.setHasQueried(true); + this.#dehydrated = { + plan, + files: tail.files, + appended: [], + hadIndex: index !== null, + ...(indexUnread ? { indexUnread } : {}), + lastTurn: canonical?.last ?? null, + reseed: false, + }; + if (index) { + this.#turnIndex = index; + this.#currentTurnId = index.at(-1)?.turnId ?? null; + this.#endSnapshotTurn = this.#currentTurnId; + } + if (plan.backingFresh) { + // reseedFreshBacking's flags; its seed waits for the history. + this.#backingFresh = true; + this.#provider.setHasQueried(false); + } + // An empty index over a non-empty history names its current turn from the history. + return !indexUnread && canonical !== null && index !== null && (index.length > 0 || canonical.last === null); + } + + /** + * Load the history of a session resumed by `prepareResume`, if it isn't + * loaded: idempotent, and concurrent callers share one load. Rejects when + * the transcript, canonical log or turn index can't be read (the session + * stays unloaded; the next call tries again). At resume (`atResume`) a + * canonical log that can't be read is logged and the session goes on + * without it, as a resume always did (`#historyUnread`); each later call + * reads it again. Resolves once a fresh backing has been re-seeded too. + */ + ensureHydrated(opts: { deadlineAt?: number; atResume?: boolean } = {}): Promise { + const d = this.#dehydrated; + if (!d && this.#historyUnread) return this.#rereadHistory().then(() => this.#afterHydrate ?? undefined); + if (!d) return this.#afterHydrate ?? Promise.resolve(); + if (!d.loading) { + const loading = this.#hydrate(d, opts.deadlineAt, opts.atResume === true).then(() => this.#afterHydrate ?? undefined); + d.loading = loading; + loading.catch(() => { + if (d.loading === loading) d.loading = undefined; + }); + } + return d.loading; + } + + async #hydrate(d: Dehydrated, deadlineAt = Date.now() + d.plan.deadlineMs, atResume = false): Promise { + const started = performance.now(); + // The window an eager resume read: the log as it was at the restart. + // What was appended since is left out (it is in memory). + const stats: { truncated?: boolean } = {}; + const window = { maxBytes: d.plan.transcriptMaxBytes, deadlineAt, stats, asOf: d.files }; + const entries = await this.#transcriptStore.loadTranscript(this.id, window); + const loaded: LoadedHistory = { entries, truncated: stats.truncated === true, canonical: null }; + try { + loaded.canonical = await this.#transcriptStore.loadCanonical(this.id, { maxBytes: d.plan.canonicalMaxBytes }); + if (d.indexUnread) loaded.index = await this.#transcriptStore.loadTurnIndex(this.id); + // No log: a session from before it existed. Rebuild once from the + // WHOLE transcript (the scrollback window may be a tail) and persist it + // — a long session's history must not shrink to the window. + if (!loaded.canonical && loaded.truncated) { + const all = await this.#transcriptStore.loadTranscript(this.id, { deadlineAt, asOf: d.files }); + loaded.rebuildFrom = all.map((e) => e.message); + } + } catch (err) { + // After the resume, the session is used: it must not go on without its + // conversation. The request fails, and the next use loads it again. + if (!atResume) throw err; + loaded.canonicalError = err; + } + // Loaded synchronously meanwhile (#hydrateNow): that load stands. + if (this.#dehydrated !== d) return; + this.#installHistory(d, loaded, atResume ? null : started); + } + + /** + * Load the history now, synchronously: a synchronous path needs it and + * nothing awaited `ensureHydrated()` first. Correct, but it blocks the + * daemon for the read, so it is logged with where it came from. + */ + #hydrateNow(what: string): void { + const d = this.#dehydrated; + if (!d) return; + const started = performance.now(); + Session.syncHydrations++; + const where = new Error().stack?.split("\n").slice(3, 6).map((l) => l.trim()).join(" < ") ?? ""; + console.warn(`[codeoid/session ${this.id}] ${what} needed before it was loaded — loading it synchronously (${where})`); + const stats: { truncated?: boolean } = {}; + const deadlineAt = Date.now() + d.plan.deadlineMs; + const window = { maxBytes: d.plan.transcriptMaxBytes, deadlineAt, stats, asOf: d.files }; + // Any read error throws, the session left unloaded, as in `#hydrate`. + const entries = this.#transcriptStore.loadTranscriptSync(this.id, window); + const loaded: LoadedHistory = { entries, truncated: stats.truncated === true, canonical: null }; + loaded.canonical = this.#transcriptStore.loadCanonicalSync(this.id, { maxBytes: d.plan.canonicalMaxBytes }); + if (d.indexUnread) loaded.index = this.#transcriptStore.loadTurnIndexSync(this.id); + if (!loaded.canonical && loaded.truncated) { + loaded.rebuildFrom = this.#transcriptStore + .loadTranscriptSync(this.id, { deadlineAt, asOf: d.files }) + .map((e) => e.message); + } + this.#installHistory(d, loaded, started); + } + + /** Put a load's history in place, as a resume did: one synchronous step. */ + #installHistory(d: Dehydrated, loaded: LoadedHistory, started: number | null): void { + this.#dehydrated = null; + const { entries } = loaded; + // Restore scrollback from transcript. Byte-budgeted + deadline-aware: one + // huge transcript can neither OOM the daemon nor stall the load. The seq + // counter continues past the log's newest line (prepareResume), and now + // past its highest seq too: a log whose seqs went back down (an older + // daemon restarted them, an import) holds it further up. + const messages = entries.map((e) => e.message); + const maxSeq = entries.reduce((max, e) => Math.max(max, e.seq), -1); + this.restoreScrollback(messages, maxSeq + 1, entries.map((e) => e.bytes), { + partialHistory: loaded.truncated, + markQueried: false, + }); + // The backend-neutral conversation (#354). Without it a restored session + // forked / switched backend with no history, and stateless backends + // answered the next prompt with no memory of the session. + let canonicalError = loaded.canonicalError; + if (canonicalError === undefined) { + try { + const rebuildFrom = loaded.rebuildFrom ?? messages; + const history = + loaded.canonical?.turns ?? + (rebuildFrom.length > 0 ? canonicalFromTranscript(rebuildFrom, d.plan.providerId) : []); + // The index read at resume, or by the load when it couldn't be then. + const index = loaded.index !== undefined ? loaded.index : d.hadIndex ? this.#turnIndex : null; + this.restoreTurns(history, index ?? turnIndexFromHistory(history), { + persistHistory: !loaded.canonical && history.length > 0, + persistIndex: !index && history.some((t) => t.turnId), + partial: loaded.canonical?.partial ?? false, + }); + } catch (err) { + canonicalError = err; + } + } + if (canonicalError !== undefined) { + const error = canonicalError instanceof Error ? canonicalError.message : String(canonicalError); + this.#historyUnread = { plan: d.plan, indexUnread: d.indexUnread === true, error }; + console.error( + `[codeoid/resume] canonical history restore failed for ${this.id} (session resumes without it, and reads it again on its next use): ${error}`, + ); + } + // What this process appended before the load, in order, as it was pushed live. + for (const msg of d.appended) this.#scrollbackBuffer.push(msg); + const reseedFresh = d.plan.backingFresh && canonicalError === undefined; + if (reseedFresh || d.reseed) { + const after: Promise = (async () => { + if (reseedFresh) await this.reseedFreshBacking(); + if (d.reseed) await this.#reseedAfterTeardown(); + })() + .catch((err) => { + console.error( + `[codeoid/resume] canonical history restore failed for ${this.id} (session resumes without it): ${err instanceof Error ? err.message : String(err)}`, + ); + }) + .finally(() => { + if (this.#afterHydrate === after) this.#afterHydrate = null; + }); + this.#afterHydrate = after; + } + if (started === null) return; // part of the resume: not worth a line per session + const ms = Math.round(performance.now() - started); + console.log( + `[codeoid/session ${this.id.slice(0, 8)}] history loaded: ${this.#scrollbackBuffer.length} message(s), ${this.#historyAccumulator.history.length} turn(s) in ${ms}ms`, + ); + } + + /** + * Read the canonical log again, for a session that couldn't at resume + * (`#historyUnread`): never rejects, the session goes on without it until + * a read succeeds. Not while a turn runs (its reply is being built in the + * history), and a read that something changed the history or the turn + * list under is dropped: the next use reads again. + */ + #rereadHistory(): Promise { + const u = this.#historyUnread; + if (!u || this.#turnActive() || this.#destroyed) return Promise.resolve(); + u.retry ??= this.#reread(u).finally(() => { + u.retry = undefined; + }); + return u.retry; + } + + async #reread(u: HistoryUnread): Promise { + const history = this.#historyAccumulator.history; + const indexed = this.#turnIndex; + let canonical: Awaited>; + let index: TurnIndexEntry[] | null = null; + try { + canonical = await this.#transcriptStore.loadCanonical(this.id, { maxBytes: u.plan.canonicalMaxBytes }); + if (u.indexUnread) index = await this.#transcriptStore.loadTurnIndex(this.id); + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + if (error !== u.error) { + u.error = error; + console.error(`[codeoid/session ${this.id}] canonical history still can't be read (the session goes on without it): ${error}`); + } + return; + } + if ( + this.#historyUnread !== u || + this.#destroyed || + this.#turnActive() || + this.#historyAccumulator.history !== history || + this.#historyAccumulator.history.length !== history.length || + this.#turnIndex !== indexed || + this.#turnIndex.length !== indexed.length + ) { + return; + } + // What this process appended since is in the log too. No log: nothing was + // written since, and nothing on disk to keep. + const turns = canonical?.turns ?? history; + try { + const turnIndex = u.indexUnread ? (index ?? turnIndexFromHistory(turns)) : indexed; + this.restoreTurns(turns, turnIndex, { + persistHistory: false, + persistIndex: u.indexUnread && !index && turns.some((t) => t.turnId), + partial: canonical?.partial ?? false, + }); + } catch (err) { + console.error(`[codeoid/session ${this.id}] canonical history read again but not restored: ${err instanceof Error ? err.message : String(err)}`); + return; + } + this.#historyUnread = null; + console.log(`[codeoid/session ${this.id.slice(0, 8)}] canonical history read again: ${turns.length} turn(s)`); + // A fresh backing no turn has used yet is owed the history it missed. + if (u.plan.backingFresh && this.#backingFresh && turns.length > 0) { + await this.reseedFreshBacking().catch((err) => { + console.error( + `[codeoid/session ${this.id}] re-seeding the backend from the history read again failed: ${err instanceof Error ? err.message : String(err)}`, + ); + }); + } + } + + /** Refuse what would rewrite the canonical log from a history that couldn't be read (`#historyUnread`). */ + #assertHistoryRead(): void { + if (this.#historyUnread) throw new RewindError("history_unavailable", HISTORY_UNREAD_REFUSAL); + } + + /** + * A backend that can't resume its own conversation (the Gemini CLI starts + * a new ACP session) is handed the history after a teardown. A session + * whose history isn't loaded owes it until the load: nothing reaches the + * backend before then. + */ + async #reseedAfterTeardown(): Promise { + if (!RESEED_AFTER_TEARDOWN.has(this.#provider.id)) return; + if (this.#dehydrated) { + this.#dehydrated.reseed = true; + return; + } + if (this.#accumulator.history.length > 0) await this.#seedProviderFromHistory(); + } + + /** + * Push a message onto the scrollback — or, history not loaded yet, keep it + * for the load, which pushes it after what it restores, as if it had been + * pushed live (the load reads the log as it was before it was appended). + */ + #buffer(msg: SessionMessage): void { + if (this.#dehydrated) this.#dehydrated.appended.push(msg); + else this.#scrollbackBuffer.push(msg); + } + restoreScrollback( messages: DaemonMessage[], nextSeq?: number, sizeHints?: ReadonlyArray, - opts?: { partialHistory?: boolean }, + /** `markQueried: false`: the caller has told the backend already (`prepareResume`). */ + opts?: { partialHistory?: boolean; markQueried?: boolean }, ): void { // Restart-restore with a byte-budgeted transcript window: the buffer // holds only the tail, so history paging must know disk has more. @@ -6498,7 +6997,7 @@ export class Session { } // A session with prior scrollback already exists in Claude Code's own // persistent session store — next send() must use `resume`, not re-create. - if (messages.length > 0) { + if (messages.length > 0 && opts?.markQueried !== false) { this.#provider.setHasQueried(true); } } @@ -7433,9 +7932,7 @@ export class Session { // Switched since to a backend that takes it per turn: nothing to rebuild. if (this.#provider.effortChange !== "restart") return; await this.#teardownProvider(); - if (RESEED_AFTER_TEARDOWN.has(this.#provider.id) && this.#accumulator.history.length > 0) { - await this.#seedProviderFromHistory(); - } + await this.#reseedAfterTeardown(); } /** @@ -7479,6 +7976,7 @@ export class Session { * then rotating is the only way forward, however few turns there are. */ async manualRotate(sender: AuthContext): Promise { + if (this.#loadPending) await this.ensureHydrated(); const stuck = this.#promptTooLong || this.#status === "error"; if (!stuck && this.#usage.numTurns < (this.#config?.autoRotate.minTurnsBeforeRotate ?? 3)) { const infoMsg = this.#makeMessage( @@ -7559,6 +8057,9 @@ export class Session { // CRITICAL: reset the provider so it creates a fresh session rather // than trying to resume an id that has no persisted history. this.#provider.resetToNewSession(newBackingId); + // A rotation empties the log whatever it held: one that couldn't be read + // (`#historyUnread`) loses nothing by it, and is now what memory holds. + this.#historyUnread = null; this.#accumulator.reset(); // Re-snapshot the frozen workspace index for the fresh context (the rotated // session gets an up-to-date orientation; within the context it stays stable @@ -9944,6 +10445,8 @@ export class Session { /** The whole canonical history: from the log when memory holds only its tail. */ async fullCanonicalHistory(): Promise { + if (this.#loadPending) await this.ensureHydrated(); + this.#assertHistoryRead(); if (!this.#canonicalPartial) return this.#accumulator.history; const full = await this.#transcriptStore.loadCanonical(this.id); return full && full.turns.length >= this.#accumulator.history.length ? full.turns : this.#accumulator.history; @@ -9994,6 +10497,8 @@ export class Session { * all of it comes from codeoid's own records. */ async forkPoint(turnId: string): Promise { + if (this.#loadPending) await this.ensureHydrated(); + this.#assertHistoryRead(); // One copy of the turn list: a rewind during the awaits below must not // make the turns and the conversation disagree. const index = this.#turnIndex.map((e) => ({ ...e })); @@ -10201,6 +10706,8 @@ export class Session { /** What going back to before `turnId` would remove, restore and not undo. */ async #planRewind(turnId: string, restoreFiles: boolean): Promise { + if (this.#loadPending) await this.ensureHydrated(); + this.#assertHistoryRead(); const pos = this.#turnIndex.findIndex((e) => e.turnId === turnId); if (pos === -1) throw new RewindError("not_found", "That turn isn't in this session (or is older than its turn list)."); const removedIds = this.#turnIndex.slice(pos).map((e) => e.turnId); @@ -10333,6 +10840,9 @@ export class Session { if (this.rewindNeedsStop && opts.canStop !== true) { throw new RewindError("busy", "The session is busy; going back would stop it, which needs session:interrupt."); } + if (this.#loadPending) await this.ensureHydrated(); + // Before anything is stopped: the log couldn't be read, so nothing changes. + this.#assertHistoryRead(); this.#rewinding = true; try { return await this.#rewindStopped(turnId, opts, sender); @@ -10604,7 +11114,9 @@ export class Session { /** Persist to transcript + scrollback buffer + memory chunker */ #persistAndBuffer(msg: SessionMessage): void { - this.#scrollback.push(msg); + // A notice for a session whose history isn't loaded (a mode change, a + // rename) doesn't load it: it is kept for the load. + this.#buffer(msg); // Log instead of swallowing — a silently-dropped transcript write means a // turn that's missing on resume with no trace. (Scrollback above is the // in-memory handoff source and is synchronous, so device-handoff replay diff --git a/src/daemon/settings/manifest.ts b/src/daemon/settings/manifest.ts index b393cf76..6e4c3cc6 100644 --- a/src/daemon/settings/manifest.ts +++ b/src/daemon/settings/manifest.ts @@ -93,6 +93,12 @@ const general: SettingsTab = { default: 200, advanced: true, }), + cfg("session.lazyLoad", "Load history on first use", "A resumed session loads its scrollback and history the first time it is opened or used, not at daemon start. Off loads every session's at start (more memory).", { + kind: "boolean", + envVar: "CODEOID_SESSION_LAZY_LOAD", + default: true, + advanced: true, + }), ], }, { diff --git a/src/daemon/transcript.ts b/src/daemon/transcript.ts index 1daea68b..35e02ed8 100644 --- a/src/daemon/transcript.ts +++ b/src/daemon/transcript.ts @@ -13,7 +13,7 @@ * - Use append-only JSONL — no reads on the hot path */ -import { existsSync, mkdirSync } from "node:fs"; +import { closeSync, existsSync, fstatSync, mkdirSync, openSync, read, readFileSync, readSync, statSync } from "node:fs"; import { appendFile, rename, rm, writeFile } from "node:fs/promises"; import { join } from "node:path"; import type { CanonicalHistoryChange, CanonicalToolCall, CanonicalTurn } from "./providers/canonical.js"; @@ -130,6 +130,20 @@ export interface LoadTranscriptOptions { * paging (`scrollback.page`) knows older messages exist on disk. */ stats?: { truncated?: boolean }; + /** + * Read the log as it was when these marks were taken (`transcriptTail`): + * each file up to its size then, wherever rotation has moved it since. What + * a resumed session appended before its scrollback was loaded, which it + * holds itself (`Session#hydrate`), is left out whatever its seqs. + */ + asOf?: readonly TranscriptFileMark[]; +} + +/** One transcript file as it was at a moment: which file (it may be renamed by rotation) and its size then. */ +export interface TranscriptFileMark { + dev: number; + ino: number; + size: number; } /** Tuning knobs, injectable for tests. */ @@ -390,27 +404,56 @@ export class TranscriptStore { async #readCanonicalOnce(sessionId: string, maxBytes?: number): Promise<{ turns: CanonicalTurn[]; partial: boolean }> { const path = this.canonicalPath(sessionId); - const size = Bun.file(path).size; - const offset = maxBytes !== undefined && size > maxBytes ? size - maxBytes : 0; + const offset = canonicalOffset(Bun.file(path).size, maxBytes); const turns: CanonicalTurn[] = []; for await (const line of readLines(path, offset)) { - if (!line.trim()) continue; - try { - const entry = JSON.parse(line) as { op?: string; turn?: CanonicalTurn }; - if (entry.op === "append" && entry.turn && (entry.turn.role === "user" || entry.turn.role === "assistant")) { - turns.push(entry.turn); - } - } catch { - // torn line - } + const turn = canonicalTurnOf(line); + if (turn) turns.push(turn); } - const partial = offset > 0; - if (partial) { - // A tail read may start mid-turn: drop leading assistant turns. - const firstUser = turns.findIndex((t) => t.role === "user"); - turns.splice(0, firstUser === -1 ? turns.length : firstUser); + return canonicalWindow(turns, offset); + } + + /** + * `loadCanonical` read synchronously, for a session whose history a + * synchronous path needs before anything awaited loading it + * (`Session#hydrateNow`). Doesn't wait for pending writes: the caller only + * reads a log nothing has appended to in this process. + */ + loadCanonicalSync(sessionId: string, opts: { maxBytes?: number } = {}): { turns: CanonicalTurn[]; partial: boolean } | null { + const path = this.canonicalPath(sessionId); + const size = fileSize(path); + if (size === null) return null; + let window = opts.maxBytes; + for (;;) { + const offset = canonicalOffset(size, window); + const turns: CanonicalTurn[] = []; + for (const line of readLinesSync(path, offset)) { + const turn = canonicalTurnOf(line); + if (turn) turns.push(turn); + } + const r = canonicalWindow(turns, offset); + if (!r.partial || r.turns.length > 0 || window === undefined) return r; + window *= 2; } - return { turns, partial }; + } + + /** + * The newest turn of a session's canonical log, read from the end of the + * file only: what a resumed session reports (`lastTurnId`, `lastTurnRole`) + * before its history is loaded. Null when the session has no log; + * `last: null` for an empty one. + */ + async canonicalTail(sessionId: string): Promise<{ last: Pick | null } | null> { + await this.#canonicalChain.get(sessionId); + const path = this.canonicalPath(sessionId); + if (fileSize(path) === null) return null; + let last: CanonicalTurn | null = null; + tailLines(path, (line) => { + last = canonicalTurnOf(line); + return last !== null; + }); + const turn = last as CanonicalTurn | null; + return { last: turn ? { role: turn.role, ...(turn.turnId !== undefined ? { turnId: turn.turnId } : {}) } : null }; } /** Append one turn-index entry (#354). Serialized per session; never throws. */ @@ -437,17 +480,23 @@ export class TranscriptStore { await this.#turnIndexChain.get(sessionId); const file = Bun.file(this.turnIndexPath(sessionId)); if (!(await file.exists())) return null; - const out: TurnIndexEntry[] = []; - for (const line of (await file.text()).split("\n")) { - if (!line.trim()) continue; - try { - const e = JSON.parse(line) as TurnIndexEntry; - if (typeof e.turnId === "string" && (e.kind === "prompt" || e.kind === "background")) out.push(e); - } catch { - // torn line - } + return turnIndexOf(await file.text()); + } + + /** + * `loadTurnIndex` read synchronously (`Session#hydrateNow`). Doesn't wait + * for pending writes: the caller only reads an index nothing has written + * to in this process. + */ + loadTurnIndexSync(sessionId: string): TurnIndexEntry[] | null { + let text: string; + try { + text = readFileSync(this.turnIndexPath(sessionId), "utf-8"); + } catch (err) { + if (isMissing(err)) return null; + throw err; } - return out; + return turnIndexOf(text); } async #writeAtomic(path: string, body: string): Promise { @@ -505,13 +554,7 @@ export class TranscriptStore { async append(sessionId: string, msg: DaemonMessage, seq: number): Promise { if (!PERSISTED_TYPES.has(msg.type)) return; - const entry: TranscriptEntry = { - seq, - timestamp: "timestamp" in msg ? (msg as { timestamp: string }).timestamp : new Date().toISOString(), - message: capPersistedToolOutput(msg), - }; - - const line = `${JSON.stringify(entry)}\n`; + const line = transcriptLine(msg, seq); // True append. The previous implementation read the entire file // and rewrote it with the new line concatenated — O(n) per append, @@ -704,32 +747,77 @@ export class TranscriptStore { sessionId: string, opts: LoadTranscriptOptions = {}, ): Promise { - // Oldest → newest: [.N, …, .1, live]. - const candidates: string[] = []; - for (let i = this.#maxRotatedSegments; i >= 1; i--) { - candidates.push(this.#segmentPath(sessionId, i)); + const merged = new TranscriptMerge(); + let sinceDeadlineCheck = 0; + outer: for (const part of this.#transcriptWindow(sessionId, opts)) { + const lines = part.mark ? this.#readMarked(sessionId, part.mark, part.offset, opts) : readLines(part.path, part.offset); + for await (const line of lines) { + if (++sinceDeadlineCheck >= DEADLINE_CHECK_EVERY) { + sinceDeadlineCheck = 0; + if (deadlineHit(sessionId, opts, merged.size)) break outer; + } + merged.add(line); + } + } + return merged.entries(); + } + + /** + * `loadTranscript` read synchronously (same window, deadline and merge), + * for a session whose scrollback a synchronous path needs before anything + * awaited loading it (`Session#hydrateNow`). + */ + loadTranscriptSync(sessionId: string, opts: LoadTranscriptOptions = {}): TranscriptEntry[] { + const merged = new TranscriptMerge(); + let sinceDeadlineCheck = 0; + outer: for (const part of this.#transcriptWindow(sessionId, opts)) { + const lines = part.mark ? this.#readMarkedSync(sessionId, part.mark, part.offset, opts) : readLinesSync(part.path, part.offset); + for (const line of lines) { + if (++sinceDeadlineCheck >= DEADLINE_CHECK_EVERY) { + sinceDeadlineCheck = 0; + if (deadlineHit(sessionId, opts, merged.size)) break outer; + } + merged.add(line); + } } - candidates.push(this.transcriptPath(sessionId)); + return merged.entries(); + } + + /** + * The files (and start offsets) holding the newest `opts.maxBytes` of a + * session's log, oldest first: the files as they are, or as `opts.asOf` + * marked them. Sets `opts.stats.truncated` when older history is left out. + */ + #transcriptWindow( + sessionId: string, + opts: LoadTranscriptOptions, + ): Array<{ path: string; offset: number; mark?: TranscriptFileMark }> { + // Oldest → newest: [.N, …, .1, live]. + const candidates: Array<{ path: string; size: number; mark?: TranscriptFileMark }> = opts.asOf + ? opts.asOf.map((mark) => ({ path: "", size: mark.size, mark })) + : this.#transcriptFiles(sessionId).flatMap((path) => { + const size = fileSize(path); + return size === null ? [] : [{ path, size }]; + }); // Walk newest-first, keeping files while budget remains; the oldest // surviving file may enter mid-way (offset > 0, first partial line // dropped by the reader). - const chosen: Array<{ path: string; offset: number }> = []; + const chosen: Array<{ path: string; offset: number; mark?: TranscriptFileMark }> = []; let budget = opts.maxBytes ?? Number.POSITIVE_INFINITY; let skippedBytes = 0; for (let i = candidates.length - 1; i >= 0; i--) { - const f = Bun.file(candidates[i]!); - if (!(await f.exists())) continue; + const { path, size, mark } = candidates[i]!; if (budget <= 0) { - skippedBytes += f.size; + skippedBytes += size; continue; } - if (f.size <= budget) { - chosen.unshift({ path: candidates[i]!, offset: 0 }); - budget -= f.size; + if (size <= budget) { + chosen.unshift({ path, offset: 0, ...(mark ? { mark } : {}) }); + budget -= size; } else { - skippedBytes += f.size - budget; - chosen.unshift({ path: candidates[i]!, offset: f.size - budget }); + skippedBytes += size - budget; + chosen.unshift({ path, offset: size - budget, ...(mark ? { mark } : {}) }); budget = 0; } } @@ -739,84 +827,110 @@ export class TranscriptStore { `[codeoid] transcript ${sessionId}: replay window capped at ${opts.maxBytes} bytes — ${skippedBytes} bytes of older history left on disk (still exported by share.pack).`, ); } + return chosen; + } - // Merge entries by messageId — later entries update earlier ones. - const byMessageId = new Map(); - const order: string[] = []; - let sinceDeadlineCheck = 0; - - outer: for (const { path, offset } of chosen) { - for await (const line of readLines(path, offset)) { - if (++sinceDeadlineCheck >= DEADLINE_CHECK_EVERY) { - sinceDeadlineCheck = 0; - if (opts.deadlineAt !== undefined && Date.now() > opts.deadlineAt) { - if (opts.stats) opts.stats.truncated = true; - console.warn( - `[codeoid] transcript ${sessionId}: resume deadline hit mid-parse — replaying the ${order.length} message(s) merged so far.`, - ); - break outer; - } - } - if (!line.trim()) continue; - let entry: TranscriptEntry; - try { - entry = JSON.parse(line); - } catch { - continue; // Skip corrupted lines - } - // Valid JSON isn't necessarily a valid entry — `null`, arrays, or - // primitives would throw on the field accesses below. Skip them the - // same way as corrupted lines. - if ( - entry === null || - typeof entry !== "object" || - Array.isArray(entry) || - typeof entry.message !== "object" || - entry.message === null - ) { - continue; - } - const bytes = Buffer.byteLength(line, "utf-8"); - - const msg = entry.message; - const messageId = (msg as { messageId?: string }).messageId; + /** A session's log files, oldest first: [.N, …, .1, live]. */ + #transcriptFiles(sessionId: string): string[] { + const files: string[] = []; + for (let i = this.#maxRotatedSegments; i >= 1; i--) files.push(this.#segmentPath(sessionId, i)); + files.push(this.transcriptPath(sessionId)); + return files; + } - if (!messageId) { - // No messageId (e.g. status_change) — keep as-is with synthetic key - const key = `_seq_${entry.seq}`; - byMessageId.set(key, { ...entry, bytes }); - order.push(key); - continue; - } + /** + * Open the file `mark` names, wherever rotation has moved it since: null + * when it is gone (rotated out of retention; logged, and the read flagged + * as missing history). + */ + #openMarked(sessionId: string, mark: TranscriptFileMark, opts: LoadTranscriptOptions): number | null { + for (const path of this.#transcriptFiles(sessionId).reverse()) { + let fd: number; + try { + fd = openSync(path, "r"); + } catch (err) { + if (isMissing(err)) continue; + throw err; + } + const st = fstatSync(fd); + if (st.dev === mark.dev && st.ino === mark.ino) return fd; + closeSync(fd); + } + if (opts.stats) opts.stats.truncated = true; + console.warn(`[codeoid] transcript ${sessionId}: a file of the log was rotated out before it was read — ${mark.size} bytes of older history left out.`); + return null; + } - if (byMessageId.has(messageId)) { - // Update: merge the newer entry over the older one - const existing = byMessageId.get(messageId)!; - const existingMsg = existing.message as unknown as Record; - const newMsg = msg as unknown as Record; - - // Shallow merge — newer fields overwrite older, preserving what's not in the update - for (const [k, v] of Object.entries(newMsg)) { - if (v !== undefined) existingMsg[k] = v; - } - // Deep merge tool state specifically - if (newMsg.tool && existingMsg.tool) { - Object.assign(existingMsg.tool as Record, newMsg.tool as Record); - } - existing.seq = entry.seq; // update seq to latest - // The merged message is at least as large as its latest full line — - // good enough for scrollback's byte accounting. - existing.bytes = Math.max(existing.bytes ?? 0, bytes); - } else { - // First occurrence — insert - byMessageId.set(messageId, { ...entry, message: { ...msg as object } as typeof msg, bytes }); - order.push(messageId); - } + /** The lines of the file `mark` names, from `offset` up to its size then. */ + async *#readMarked( + sessionId: string, + mark: TranscriptFileMark, + offset: number, + opts: LoadTranscriptOptions, + ): AsyncGenerator { + const fd = this.#openMarked(sessionId, mark, opts); + if (fd === null) return; + try { + const lines = new LineSplitter(offset > 0); + const buf = Buffer.allocUnsafe(READ_CHUNK_BYTES); + for (let pos = offset; pos < mark.size; ) { + const n = await readAt(fd, buf, Math.min(buf.length, mark.size - pos), pos); + if (n <= 0) break; + pos += n; + yield* lines.push(buf.subarray(0, n)); } + yield* lines.end(); + } finally { + closeSync(fd); } + } - // "Went back a turn" (#355): hide what was taken back. - return applyRewinds(order.map((key) => byMessageId.get(key)!)); + /** `#readMarked`, synchronously. */ + *#readMarkedSync(sessionId: string, mark: TranscriptFileMark, offset: number, opts: LoadTranscriptOptions): Generator { + const fd = this.#openMarked(sessionId, mark, opts); + if (fd === null) return; + try { + yield* readFdLinesSync(fd, offset, mark.size); + } finally { + closeSync(fd); + } + } + + /** + * The newest end of a session's log, read from the tail of its newest + * file holding an entry, and a mark of each of its files as they are now + * (`LoadTranscriptOptions.asOf`): what a resumed session needs to keep + * appending before its scrollback is loaded, and to load it later exactly + * as it is now. `nextSeq` is one past the newest entry's seq: the highest + * in a log whose seqs only grow; one whose seqs went back down (an older + * daemon restarted them, an import) continues past the highest once its + * scrollback is loaded (`Session#installHistory`). `hasEntries`: the log + * holds any entry at all, with a seq or not. Throws when a file is there + * but can't be read. + */ + async transcriptTail(sessionId: string): Promise<{ nextSeq: number; hasEntries: boolean; files: TranscriptFileMark[] }> { + await this.#appendChain.get(sessionId); + const paths = this.#transcriptFiles(sessionId); + const files: TranscriptFileMark[] = []; + for (const path of paths) { + const st = fileStat(path); + if (st) files.push({ dev: st.dev, ino: st.ino, size: st.size }); + } + let hasEntries = false; + for (let i = paths.length - 1; i >= 0; i--) { + if (!fileSize(paths[i]!)) continue; + let newest: number | null = null; + tailLines(paths[i]!, (line) => { + const entry = transcriptEntryOf(line); + if (!entry) return false; + hasEntries = true; + if (typeof entry.seq !== "number" || !Number.isFinite(entry.seq)) return false; + newest = entry.seq; + return true; + }); + if (newest !== null) return { nextSeq: (newest as number) + 1, hasEntries, files }; + } + return { nextSeq: 0, hasEntries, files }; } /** @@ -886,6 +1000,302 @@ export function applyRewinds(rows: T[]): T return out; } +/** A turn index file's entries (torn lines left out). */ +function turnIndexOf(text: string): TurnIndexEntry[] { + const out: TurnIndexEntry[] = []; + for (const line of text.split("\n")) { + if (!line.trim()) continue; + try { + const e = JSON.parse(line) as TurnIndexEntry; + if (typeof e.turnId === "string" && (e.kind === "prompt" || e.kind === "background")) out.push(e); + } catch { + // torn line + } + } + return out; +} + +/** One transcript line for `msg` at `seq`, as `append` writes it. */ +function transcriptLine(msg: DaemonMessage, seq: number): string { + const entry: TranscriptEntry = { + seq, + timestamp: "timestamp" in msg ? (msg as { timestamp: string }).timestamp : new Date().toISOString(), + message: withoutBufferSeq(capPersistedToolOutput(msg)), + }; + return `${JSON.stringify(entry)}\n`; +} + +/** + * A message without the replay buffer's `seq` stamp: a cursor into one daemon + * run's scrollback, meaningless once it is read back after a restart (the + * replay frame's `maxSeq` is the client's cursor), and present or not + * depending on whether the session's scrollback was in memory when it was + * appended (lazy session loading). + */ +function withoutBufferSeq(msg: DaemonMessage): DaemonMessage { + if (msg.type !== "session.message" || (msg as { seq?: number }).seq === undefined) return msg; + const { seq: _seq, ...rest } = msg as SessionMessage; + return rest as DaemonMessage; +} + +/** A transcript line's entry, or null for a blank, torn or malformed line. */ +function transcriptEntryOf(line: string): TranscriptEntry | null { + if (!line.trim()) return null; + let entry: TranscriptEntry; + try { + entry = JSON.parse(line); + } catch { + return null; // Skip corrupted lines + } + // Valid JSON isn't necessarily a valid entry — `null`, arrays, or + // primitives would throw on the field accesses below. Skip them the + // same way as corrupted lines. + if ( + entry === null || + typeof entry !== "object" || + Array.isArray(entry) || + typeof entry.message !== "object" || + entry.message === null + ) { + return null; + } + return entry; +} + +/** + * A transcript's lines merged by messageId, in file order — later entries + * update earlier ones — with rewinds applied (`entries()`). + */ +class TranscriptMerge { + #byMessageId = new Map(); + #order: string[] = []; + + get size(): number { + return this.#order.length; + } + + add(line: string): void { + const entry = transcriptEntryOf(line); + if (!entry) return; + const bytes = Buffer.byteLength(line, "utf-8"); + + const msg = entry.message; + const messageId = (msg as { messageId?: string }).messageId; + + if (!messageId) { + // No messageId (e.g. status_change) — keep as-is with synthetic key + const key = `_seq_${entry.seq}`; + this.#byMessageId.set(key, { ...entry, bytes }); + this.#order.push(key); + return; + } + + const existing = this.#byMessageId.get(messageId); + if (existing) { + // Update: merge the newer entry over the older one + const existingMsg = existing.message as unknown as Record; + const newMsg = msg as unknown as Record; + + // Shallow merge — newer fields overwrite older, preserving what's not in the update + for (const [k, v] of Object.entries(newMsg)) { + if (v !== undefined) existingMsg[k] = v; + } + // Deep merge tool state specifically + if (newMsg.tool && existingMsg.tool) { + Object.assign(existingMsg.tool as Record, newMsg.tool as Record); + } + existing.seq = entry.seq; // update seq to latest + // The merged message is at least as large as its latest full line — + // good enough for scrollback's byte accounting. + existing.bytes = Math.max(existing.bytes ?? 0, bytes); + } else { + // First occurrence — insert + this.#byMessageId.set(messageId, { ...entry, message: { ...msg as object } as typeof msg, bytes }); + this.#order.push(messageId); + } + } + + entries(): TranscriptEntry[] { + // "Went back a turn" (#355): hide what was taken back. + return applyRewinds(this.#order.map((key) => this.#byMessageId.get(key)!)); + } +} + +/** Whether a transcript read ran past its deadline (logged, and flagged in `opts.stats`). */ +function deadlineHit(sessionId: string, opts: LoadTranscriptOptions, merged: number): boolean { + if (opts.deadlineAt === undefined || Date.now() <= opts.deadlineAt) return false; + if (opts.stats) opts.stats.truncated = true; + console.warn( + `[codeoid] transcript ${sessionId}: resume deadline hit mid-parse — replaying the ${merged} message(s) merged so far.`, + ); + return true; +} + +/** A canonical log line's turn, or null for a blank, torn or foreign line. */ +function canonicalTurnOf(line: string): CanonicalTurn | null { + if (!line.trim()) return null; + try { + const entry = JSON.parse(line) as { op?: string; turn?: CanonicalTurn }; + if (entry.op === "append" && entry.turn && (entry.turn.role === "user" || entry.turn.role === "assistant")) { + return entry.turn; + } + } catch { + // torn line + } + return null; +} + +/** Where a canonical read of the newest `maxBytes` starts. */ +function canonicalOffset(size: number, maxBytes: number | undefined): number { + return maxBytes !== undefined && size > maxBytes ? size - maxBytes : 0; +} + +/** The turns read from `offset` on, as a history: a tail starts at a prompt. */ +function canonicalWindow(turns: CanonicalTurn[], offset: number): { turns: CanonicalTurn[]; partial: boolean } { + const partial = offset > 0; + if (partial) { + // A tail read may start mid-turn: drop leading assistant turns. + const firstUser = turns.findIndex((t) => t.role === "user"); + turns.splice(0, firstUser === -1 ? turns.length : firstUser); + } + return { turns, partial }; +} + +/** Bytes read per step by the synchronous and marked readers. */ +const READ_CHUNK_BYTES = 1024 * 1024; + +/** A missing file (or a missing dir on its path): what the readers treat as no file. */ +function isMissing(err: unknown): boolean { + const code = (err as NodeJS.ErrnoException).code; + return code === "ENOENT" || code === "ENOTDIR"; +} + +/** A regular file's stat, or null when there is none at `path`. Throws when it can't be looked at. */ +function fileStat(path: string): import("node:fs").Stats | null { + let st: import("node:fs").Stats; + try { + st = statSync(path); + } catch (err) { + if (isMissing(err)) return null; + throw err; + } + return st.isFile() ? st : null; +} + +/** A regular file's size, or null when there is none at `path`. Throws when it can't be looked at. */ +function fileSize(path: string): number | null { + return fileStat(path)?.size ?? null; +} + +/** `pread`, awaited. */ +function readAt(fd: number, buf: Buffer, length: number, position: number): Promise { + return new Promise((resolve, reject) => { + read(fd, buf, 0, length, position, (err, n) => (err ? reject(err) : resolve(n))); + }); +} + +/** + * A byte stream cut into lines, as the readers yield them: with `dropFirst` + * (a read starting mid-file) the first, partial line is left out. + */ +class LineSplitter { + #decoder = new TextDecoder(); + #remainder = ""; + #dropFirst: boolean; + + constructor(dropFirst: boolean) { + this.#dropFirst = dropFirst; + } + + *push(bytes: Uint8Array): Generator { + this.#remainder += this.#decoder.decode(bytes, { stream: true }); + const lines = this.#remainder.split("\n"); + this.#remainder = lines.pop()!; + for (const line of lines) { + if (this.#dropFirst) { + this.#dropFirst = false; + continue; + } + yield line; + } + } + + *end(): Generator { + this.#remainder += this.#decoder.decode(); + if (this.#remainder && !this.#dropFirst) yield this.#remainder; + } +} + +/** `readLines`, synchronously. */ +function* readLinesSync(path: string, offset: number): Generator { + let fd: number; + try { + fd = openSync(path, "r"); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "ENOENT") return; + throw err; + } + try { + yield* readFdLinesSync(fd, offset, Number.POSITIVE_INFINITY); + } finally { + closeSync(fd); + } +} + +/** An open file's lines from `offset` up to `end`. */ +function* readFdLinesSync(fd: number, offset: number, end: number): Generator { + const lines = new LineSplitter(offset > 0); + const buf = Buffer.allocUnsafe(READ_CHUNK_BYTES); + for (let pos = offset; pos < end; ) { + const n = readSync(fd, buf, 0, Math.min(buf.length, end - pos), pos); + if (n <= 0) break; + pos += n; + yield* lines.push(buf.subarray(0, n)); + } + yield* lines.end(); +} + +/** + * Hand a file's whole lines to `take`, newest first, reading backwards from + * its end in growing windows, until `take` returns true or the file's start + * is reached. For the few newest lines of a log too big to read whole. + */ +function tailLines(path: string, take: (line: string) => boolean): void { + let fd: number; + try { + fd = openSync(path, "r"); + } catch (err) { + // Gone since it was looked at: nothing to read. Anything else (a file + // that can't be read) is the caller's to fail on, never an empty log. + if (isMissing(err)) return; + throw err; + } + try { + const size = fstatSync(fd).size; + let window = 64 * 1024; + for (;;) { + const start = Math.max(0, size - window); + const buf = Buffer.allocUnsafe(size - start); + let read = 0; + while (read < buf.length) { + const n = readSync(fd, buf, read, buf.length - read, start + read); + if (n <= 0) break; + read += n; + } + const lines = buf.subarray(0, read).toString("utf-8").split("\n"); + // Mid-file, the first line is cut: the next, wider window reads it whole. + const from = start > 0 ? 1 : 0; + for (let i = lines.length - 1; i >= from; i--) { + if (take(lines[i]!)) return; + } + if (start === 0) return; + window *= 4; + } + } finally { + closeSync(fd); + } +} + /** * Stream a file's lines without materialising the whole file. With * `offset > 0` the read starts mid-file and the first (partial) line is diff --git a/src/tests/lazy-session-loading.test.ts b/src/tests/lazy-session-loading.test.ts new file mode 100644 index 00000000..0de02565 --- /dev/null +++ b/src/tests/lazy-session-loading.test.ts @@ -0,0 +1,1262 @@ +/** + * Lazy session loading: a resumed session holds its metadata only, and loads + * its scrollback and canonical history the first time something uses it. + * + * Every client-visible behaviour of a resumed session must be the one an + * eagerly loaded session has. So each test here runs the same scenario on two + * copies of one daemon's data — one resumed with `session.lazyLoad: false` + * (the history loaded at start, as before) and one lazily — and compares what + * a client sees: ids, timestamps and the data dirs normalized, nothing else. + */ + +import { afterAll, afterEach, beforeAll, describe, expect, it, setDefaultTimeout } from "bun:test"; +import { execFileSync } from "node:child_process"; +import { cpSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { CodeoidConfig } from "../config.js"; +import type { CanonicalTurn } from "../daemon/providers/canonical.js"; +import type { ProviderEvent, SessionScopedEvent } from "../daemon/providers/interface.js"; +import { MockSessionProvider } from "../daemon/providers/mock/session-provider.js"; +import { ProviderRegistry } from "../daemon/providers/registry.js"; +import { type AttachedClient, Session } from "../daemon/session.js"; +import { SessionManager } from "../daemon/session-manager.js"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import { ALL_SCOPES } from "../protocol/scopes.js"; +import { type AuthContext, CAPABILITIES, type DaemonMessage, type SessionInfo, type SessionMessage } from "../protocol/types.js"; + +// Two daemons per test, each resuming and shutting down: slower than the default under a loaded suite. +setDefaultTimeout(60_000); + +const AUTH: AuthContext = { + sub: "user:lazy", + scopes: [...ALL_SCOPES] as AuthContext["scopes"], + delegationDepth: 0, + accountId: "acc-lazy", + projectId: "proj-lazy", +}; + +const RESULT = { + providerId: "mock", + model: "mock", + inputTokens: 10, + outputTokens: 5, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalCostUsd: 0.01, + durationMs: 1, +}; + +function say(text: string): ProviderEvent[] { + return [{ type: "text_done", content: text } as ProviderEvent, { type: "turn_done", result: RESULT } as ProviderEvent]; +} + +/** A turn that reads a file (a safe tool: it runs unasked), then answers. */ +function readThenSay(n: number, text: string): ProviderEvent[] { + return [ + { type: "tool_start", toolId: `t${n}`, sdkToolUseId: `sdk-t${n}`, name: "Read", input: { file_path: "a.txt" }, approvalId: `ap-${n}` } as ProviderEvent, + { type: "tool_complete", sdkToolUseId: `sdk-t${n}`, output: `v0 (read ${n})`, success: true } as ProviderEvent, + ...say(text), + ]; +} + +/** Replies long enough that a 2 KiB attach tail and paging have work to do. */ +const long = (n: number): string => `answer ${n}: ${"lorem ipsum dolor sit amet ".repeat(12)}`; + +function mkConfig(root: string, lazyLoad?: boolean): CodeoidConfig { + return { + daemonUrl: "ws://127.0.0.1:7400", + dbPath: join(root, "codeoid.db"), + transcriptDir: join(root, "transcripts"), + auth: { baseUrl: "http://localhost:8899" }, + zeroidUrl: "http://localhost:8899", + workspaceIndex: { enabled: false, episodeThreshold: 5, timeThresholdMs: 60_000, debounceMs: 15_000 }, + compress: { enabled: false, excludeCommands: [], excludePatterns: [], compressPipes: false, minBytes: 1024 }, + labeling: {}, + telemetry: { osc8: "auto" }, + autoRotate: { enabled: false, warnPct: 0.6, rotatePct: 0.8, hardRotatePct: 0.9, minTurnsBeforeRotate: 3, strategy: "task-anchor" }, + session: { attachTailBytes: 2048, ...(lazyLoad !== undefined ? { lazyLoad } : {}) }, + conductor: { enabled: false, name: "conductor", provider: "claude" }, + } as CodeoidConfig; +} + +/** One daemon over one data dir. */ +interface Env { + label: string; + root: string; + repo: string; + store: Store; + transcript: TranscriptStore; + m: SessionManager; + /** Every provider the manager built, oldest first. */ + made: MockSessionProvider[]; +} + +/** Each provider replies to the next prompts with these, in order. */ +const REPLIES = (): ProviderEvent[][] => [say("r1"), say("r2"), say("r3"), say("r4")]; + +function openEnv(label: string, root: string, lazyLoad?: boolean, script: () => ProviderEvent[][] = REPLIES): Env { + const made: MockSessionProvider[] = []; + const registry = new ProviderRegistry("claude"); + for (const id of ["claude", "codex", "gemini-cli"]) { + registry.register({ + id, + displayName: id, + create: () => { + const p = new MockSessionProvider(id, script()); + made.push(p); + return p; + }, + }); + } + const store = new Store(join(root, "codeoid.db")); + const transcript = new TranscriptStore(join(root, "transcripts")); + const m = new SessionManager(store, transcript, undefined, undefined, undefined, { + config: mkConfig(root, lazyLoad), + providers: registry, + }); + return { label, root, repo: join(root, "repo"), store, transcript, m, made }; +} + +async function closeEnv(env: Env): Promise { + env.m.stopRoutines(); + env.m.stopLoops(); + await env.m.drain(3_000).catch(() => {}); + await env.transcript.flush(); + try { + env.store.close(); + } catch {} +} + +const client = (id: string, sink: DaemonMessage[], capabilities?: string[]): AttachedClient => ({ + id, + auth: AUTH, + send: (msg: DaemonMessage) => sink.push(msg), + ...(capabilities ? { capabilities } : {}), +}); + +async function call(env: Env, msg: Record, sink: DaemonMessage[] = []): Promise { + return env.m.handle({ id: "req", ...msg } as never, AUTH, client("cli", sink)); +} + +async function settle(env: Env, sessionId: string): Promise { + const s = env.m._sessionForTest(sessionId)!; + const busy = () => s.status !== "idle" || s.turnInFlight || s.preparingTurn; + for (let i = 0; i < 400 && busy(); i++) await Bun.sleep(5); + expect(s.status).toBe("idle"); + await Bun.sleep(20); +} + +async function sendAndSettle(env: Env, sessionId: string, text: string): Promise { + const r = await call(env, { type: "session.send", sessionId, text }); + await settle(env, sessionId); + return r; +} + +// ── The data both daemons resume ───────────────────────────────────────── + +let base: string; +/** Session ids in the fixture, by name. */ +const ids: Record = {}; +const roots: string[] = []; + +/** + * One daemon's data: a long session with tool calls and notices, a short + * one, an empty one, one on a backend that is re-seeded after a teardown, + * and one from before the canonical log (loaded at start either way). + */ +beforeAll(async () => { + base = mkdtempSync(join(tmpdir(), "codeoid-lazy-base-")); + const repo = join(base, "repo"); + mkdirSync(repo); + const git = (...args: string[]) => execFileSync("git", args, { cwd: repo, encoding: "utf8" }); + git("init", "-q"); + git("config", "user.email", "t@example.com"); + git("config", "user.name", "t"); + writeFileSync(join(repo, "a.txt"), "v0\n"); + git("add", "."); + git("commit", "-qm", "init"); + + const script = () => [ + say(long(1)), + readThenSay(2, long(2)), + say(long(3)), + readThenSay(4, long(4)), + say(long(5)), + say(long(6)), + ]; + const env = openEnv("base", base, undefined, script); + const create = async (name: string, providerId?: string): Promise => { + const r = await call(env, { type: "session.create", name, workdir: repo, ...(providerId ? { providerId } : {}) }); + if (r.type !== "response.ok") throw new Error(JSON.stringify(r)); + return (r.data as { id: string }).id; + }; + ids.rich = await create("rich"); + for (let i = 1; i <= 6; i++) await sendAndSettle(env, ids.rich, `prompt ${i}: ${"please ".repeat(20)}`); + await call(env, { type: "session.pin", sessionId: ids.rich, path: "a.txt" }); + await call(env, { type: "session.set_mode", sessionId: ids.rich, mode: "autonomous", maxTurns: 7 }); + + ids.short = await create("short"); + await sendAndSettle(env, ids.short, "hello"); + await call(env, { type: "session.rename", sessionId: ids.short, name: "short-renamed" }); + + ids.empty = await create("empty"); + + ids.gemini = await create("gemini", "gemini-cli"); + await sendAndSettle(env, ids.gemini, "hi gemini"); + + ids.legacy = await create("legacy"); + await sendAndSettle(env, ids.legacy, "old times"); + + // A routine that keeps its session across runs: its session is resumed. + const routine = await call(env, { + type: "routine.create", + spec: { name: "daily", prompt: "report", workdir: repo, approvals: "guarded", session: "persistent" }, + }); + if (routine.type !== "routine.snapshot") throw new Error(JSON.stringify(routine)); + ids.routine = routine.routine.id; + const run = await call(env, { type: "routine.run", routine: "daily" }); + if (run.type !== "routine.run.result") throw new Error(JSON.stringify(run)); + ids.routineSession = (await runEnds(env, ids.routine, run.run.id)).sessionId!; + + await closeEnv(env); + // From before the canonical log and the turn index (#354). + rmSync(join(base, "transcripts", `${ids.legacy}.canonical.jsonl`)); + rmSync(join(base, "transcripts", `${ids.legacy}.turns.jsonl`)); +}, 60_000); + +afterAll(() => { + rmSync(base, { recursive: true, force: true }); +}); + +const envs: Env[] = []; +afterEach(async () => { + for (const e of envs.splice(0)) await closeEnv(e); + for (const r of roots.splice(0)) rmSync(r, { recursive: true, force: true }); +}); + +/** A copy of the fixture, its sessions pointed at the copy's repo. */ +function cloneBase(label: string): string { + // Same-length names: a path inside an export counts toward its size. + const root = mkdtempSync(join(tmpdir(), `codeoid-lazy-${label.slice(0, 4)}-`)); + roots.push(root); + cpSync(base, root, { recursive: true }); + const dir = join(root, "transcripts"); + for (const f of readdirSync(dir)) { + if (!f.endsWith(".meta.json")) continue; + const p = join(dir, f); + writeFileSync(p, readFileSync(p, "utf8").split(base).join(root)); + } + // A copied repository's index has stale stat data, which makes a fork's + // `git stash create` fail: refresh it, as any git command would. + execFileSync("git", ["status", "--porcelain"], { cwd: join(root, "repo") }); + return root; +} + +/** The fixture resumed eagerly and lazily. */ +async function both(script?: () => ProviderEvent[][]): Promise<{ eager: Env; lazy: Env }> { + const eager = openEnv("eager", cloneBase("eager"), false, script); + const lazy = openEnv("lazy", cloneBase("lazy"), true, script); + envs.push(eager, lazy); + await eager.m.resumeSessions(); + await lazy.m.resumeSessions(); + return { eager, lazy }; +} + +const UUID = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/g; +const ISO = /\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z/g; +/** A git object id (a checkpoint taken during the test embeds the clock). */ +const SHA = /\b[0-9a-f]{40}\b/g; +/** Epoch milliseconds (a turn's `createdAt`). */ +const EPOCH_MS = /\b1[7-9]\d{11}\b/g; + +/** A stable placeholder per distinct value, numbered by first appearance. */ +function token(seen: Map, value: string, kind: string): string { + let t = seen.get(value); + if (t === undefined) { + t = `<${kind}${seen.size}>`; + seen.set(value, t); + } + return t; +} + +/** What a client sees, with each daemon's own ids, clock and paths taken out. */ +function norm(env: Env, v: unknown): unknown { + const fixed = new Map(Object.values(ids).map((id) => [id, id])); + const fresh = new Map(); + const shas = new Map(); + const text = JSON.stringify(v ?? null) + .split(env.root) + .join("") + .split(base) + .join("") + .replace(ISO, "") + .replace(EPOCH_MS, "\"\"") + .replace(SHA, (sha) => token(shas, sha, "sha")) + .replace(UUID, (id) => fixed.get(id) ?? token(fresh, id, "id")); + return JSON.parse(text); +} + +/** Run `scenario` on both and expect the same result. */ +async function same(pair: { eager: Env; lazy: Env }, scenario: (env: Env) => Promise): Promise { + const before = Session.syncHydrations; + const a = await scenario(pair.eager); + const b = await scenario(pair.lazy); + expect(norm(pair.lazy, b)).toEqual(norm(pair.eager, a)); + // Every path above awaited the load: none had to block on it. + expect(Session.syncHydrations).toBe(before); + return b; +} + +const sessionOf = (env: Env, name: keyof typeof ids) => env.m._sessionForTest(ids[name]!)!; +const loaded = (env: Env, name: keyof typeof ids) => sessionOf(env, name).hydrated; + +/** Infos as a client lists them. */ +async function list(env: Env): Promise { + const r = await call(env, { type: "session.list" }); + if (r.type !== "session.list.result") throw new Error(JSON.stringify(r)); + return [...r.sessions].sort((x, y) => x.id.localeCompare(y.id)); +} + +/** What attaching shows. */ +async function attach(env: Env, sessionId: string, capabilities?: string[], resume?: { key: string; sinceSeq: number }): Promise { + const frames: DaemonMessage[] = []; + const r = await env.m.handle( + { type: "session.attach", id: "att", sessionId, ...(resume ? { resume } : {}) } as never, + AUTH, + client(`att-${Math.random()}`, frames, capabilities), + ); + return [r, ...frames.filter((f) => f.type === "scrollback.replay")]; +} + +async function runEnds(env: Env, routine: string, runId: string): Promise<{ status: string; sessionId?: string; summary?: string }> { + for (let i = 0; i < 400; i++) { + const r = await call(env, { type: "routine.runs", routine, runId }); + const run = r.type === "routine.runs.result" ? r.runs[0] : undefined; + if (run && !["queued", "checking", "running"].includes(run.status)) return run; + await Bun.sleep(25); + } + throw new Error("routine run never ended"); +} + +/** The transcript's seqs, in file order. */ +function seqsOnDisk(env: Env, sessionId: string): number[] { + const text = readFileSync(env.transcript.transcriptPath(sessionId), "utf8"); + return text + .split("\n") + .filter((l) => l.trim()) + .map((l) => (JSON.parse(l) as { seq: number }).seq); +} + +const PAGING = [CAPABILITIES.SCROLLBACK_PAGING]; + +// ── Tests ──────────────────────────────────────────────────────────────── + +describe("lazy session loading: a resumed session before it is used", () => { + it("lists with the same info, and loads nothing (a session from before the canonical log, or with none yet, loads at start)", async () => { + const pair = await both(); + await same(pair, list); + await same(pair, async (env) => Object.keys(ids).filter((k) => k !== "routine").map((k) => env.m._sessionForTest(ids[k]!)?.toInfo())); + for (const name of ["rich", "short", "gemini", "routineSession"] as const) { + expect([name, loaded(pair.eager, name)]).toEqual([name, true]); + expect([name, loaded(pair.lazy, name)]).toEqual([name, false]); + } + expect(loaded(pair.lazy, "legacy")).toBe(true); + // Never had a turn: no log to read lazily, and nothing to hold. + expect(loaded(pair.lazy, "empty")).toBe(true); + }); + + it("reports the same newest turn (what a routine reads before its run) without loading", async () => { + const pair = await both(); + await same(pair, async (env) => + (["rich", "short", "empty", "gemini"] as const).map((n) => [sessionOf(env, n).lastTurnId, sessionOf(env, n).lastTurnRole, sessionOf(env, n).currentTurnId]), + ); + expect(loaded(pair.lazy, "rich")).toBe(false); + }); + + it("the fleet board shows the same, without loading", async () => { + const pair = await both(); + await same(pair, async (env) => { + const r = await call(env, { type: "fleet.subscribe", scope: "tenant" }); + return r; + }); + expect(loaded(pair.lazy, "rich")).toBe(false); + }); + + it("the turn list is the same, without loading", async () => { + const pair = await both(); + await same(pair, async (env) => call(env, { type: "session.turns", sessionId: ids.rich })); + expect(loaded(pair.lazy, "rich")).toBe(false); + }); + + it("notices while unloaded (model, effort) are in the scrollback and on disk in order, their seqs never reused", async () => { + const pair = await both(); + const before = seqsOnDisk(pair.lazy, ids.short!).length; + await same(pair, async (env) => { + const results = [ + await call(env, { type: "session.set_mode", sessionId: ids.short, mode: "guarded" }), + await call(env, { type: "session.rename", sessionId: ids.short, name: "short-again" }), + await call(env, { type: "session.set_model", sessionId: ids.short, model: "sonnet" }), + await call(env, { type: "session.set_effort", sessionId: ids.short, effort: "high" }), + ]; + expect(results.map((r) => r.type)).toEqual(["response.ok", "response.ok", "response.ok", "response.ok"]); + await env.transcript.flush(); + return { results, info: sessionOf(env, "short").toInfo() }; + }); + // Appended while the history stayed on disk. + expect(loaded(pair.lazy, "short")).toBe(false); + expect(seqsOnDisk(pair.lazy, ids.short!).length).toBeGreaterThan(before); + await same(pair, async (env) => attach(env, ids.short!)); + await same(pair, async (env) => { + await sendAndSettle(env, ids.short!, "after the notices"); + await env.transcript.flush(); + return attach(env, ids.short!); + }); + for (const env of [pair.eager, pair.lazy]) { + const seqs = seqsOnDisk(env, ids.short!); + expect(new Set(seqs).size).toBe(seqs.length); + expect([...seqs].sort((x, y) => x - y)).toEqual(seqs); + } + expect(seqsOnDisk(pair.lazy, ids.short!)).toEqual(seqsOnDisk(pair.eager, ids.short!)); + }); +}); + +describe("lazy session loading: using a resumed session", () => { + it("attach replays the same: tail-first with paging, whole for a legacy client, and a stale cursor gets a snapshot", async () => { + const pair = await both(); + await same(pair, (env) => attach(env, ids.rich!, PAGING)); + await same(pair, (env) => attach(env, ids.rich!)); + await same(pair, (env) => attach(env, ids.rich!, PAGING, { key: "from-before-the-restart", sinceSeq: 3 })); + expect(loaded(pair.lazy, "rich")).toBe(true); + }); + + it("incremental resume by seq continues the same after a new turn", async () => { + const pair = await both(); + await same(pair, async (env) => { + const first = await attach(env, ids.rich!, PAGING); + const replay = first.find((f) => f.type === "scrollback.replay") as { resumeKey: string; maxSeq: number }; + await sendAndSettle(env, ids.rich!, "one more"); + const again = await attach(env, ids.rich!, PAGING, { key: replay.resumeKey, sinceSeq: replay.maxSeq }); + return { maxSeq: replay.maxSeq, again }; + }); + }); + + it("paging back through the whole history gives the same pages; MCP control's read_transcript page too", async () => { + const pair = await both(); + await same(pair, async (env) => { + const pages: DaemonMessage[] = []; + // read_transcript: the newest page, no anchor. + pages.push(await call(env, { type: "scrollback.page", sessionId: ids.rich, maxBytes: 4096 })); + const replay = (await attach(env, ids.rich!, PAGING)).find((f) => f.type === "scrollback.replay") as { messages: SessionMessage[] }; + let anchor = replay.messages[0]?.messageId; + for (let i = 0; anchor && i < 50; i++) { + const page = await call(env, { type: "scrollback.page", sessionId: ids.rich, beforeMessageId: anchor, maxBytes: 1500 }); + pages.push(page); + const p = page as { messages: SessionMessage[]; hasMore: boolean }; + if (!p.hasMore || p.messages.length === 0) break; + anchor = p.messages[0]!.messageId; + } + return pages; + }); + }); + + it("forking at a turn carries the same conversation, turns and scrollback", async () => { + const pair = await both(); + const forkId = await same(pair, async (env) => { + const turns = (await call(env, { type: "session.turns", sessionId: ids.rich })) as { turns: Array<{ turnId: string }> }; + const r = await call(env, { type: "session.fork", sessionId: ids.rich, name: "branch", afterTurnId: turns.turns[2]!.turnId, isolate: false }); + expect(r.type).toBe("response.ok"); + return r; + }); + const id = (forkId as { data?: { id: string } }).data?.id ?? (forkId as { session?: { id: string } }).session?.id; + expect(id).toBeString(); + await same(pair, async (env) => { + const r = (await call(env, { type: "session.list" })) as { sessions: SessionInfo[] }; + const fork = r.sessions.find((s) => s.name === "branch")!; + const p = env.made.at(-1)!; + return { + fork, + replay: await attach(env, fork.id, PAGING), + turns: await call(env, { type: "session.turns", sessionId: fork.id }), + history: env.m._sessionForTest(fork.id)!.canonicalHistory.map((t: CanonicalTurn) => [t.role, t.content, t.turnId]), + seeded: p.seededHistory?.map((t) => [t.role, t.content]), + }; + }); + const fork = (await list(pair.lazy)).find((s) => s.name === "branch")!; + expect(pair.lazy.m._sessionForTest(fork.id)!.canonicalHistory.length).toBe(6); + }); + + it("going back a turn removes the same, and the next prompt carries the same history", async () => { + const pair = await both(); + await same(pair, async (env) => { + const turns = (await call(env, { type: "session.turns", sessionId: ids.rich })) as { turns: Array<{ turnId: string }> }; + const r = await call(env, { type: "session.rewind", sessionId: ids.rich, turnId: turns.turns[4]!.turnId }); + expect(r.type).toBe("session.rewind.result"); + await settle(env, ids.rich!); + const after = await attach(env, ids.rich!, PAGING); + await sendAndSettle(env, ids.rich!, "instead"); + const opts = env.made.at(-1)!.capturedOpts.at(-1)!; + return { r, after, history: opts.history.map((t) => [t.role, t.content]), user: opts.userMessage }; + }); + }); + + it("a comparison runs from the same conversation and settles the same", async () => { + const pair = await both(); + await same(pair, async (env) => { + const r = (await call(env, { + type: "session.compare", + sessionId: ids.rich, + prompt: "do it two ways", + targets: [{ providerId: "claude" }, { providerId: "codex" }], + })) as { type: string; compare?: { compareId: string } }; + expect([r.type, (r as { error?: string }).error]).toEqual(["compare.state", undefined]); + if (r.type !== "compare.state") return r; + for (let i = 0; i < 400; i++) { + const g = (await call(env, { type: "compare.get", compareId: r.compare!.compareId })) as { compare: { targets: Array<{ done?: boolean }> } }; + if (g.compare.targets.every((t) => t.done)) { + const seeded = env.made.slice(-2).map((p) => p.seededHistory?.map((t) => [t.role, t.content])); + return { g, seeded }; + } + await Bun.sleep(10); + } + throw new Error("comparison never settled"); + }); + }); + + it("a backend switch hands the new backend the same history", async () => { + const pair = await both(); + await same(pair, async (env) => { + const r = await call(env, { type: "session.set_provider", sessionId: ids.rich, providerId: "codex" }); + expect(r.type).toBe("response.ok"); + const p = env.made.at(-1)!; + expect(p.id).toBe("codex"); + return { r, id: p.id, seeded: p.seededHistory?.map((t) => [t.role, t.content, t.turnId]), replay: await attach(env, ids.rich!, PAGING) }; + }); + }); + + it("export bundles the same", async () => { + const pair = await both(); + await same(pair, async (env) => { + const r = (await call(env, { type: "session.export", sessionId: ids.rich, includeMemory: false })) as unknown as Record; + expect(r.type).toBe("session.export.result"); + return r; + }); + }); + + it("a new prompt hands the backend the same history, and the transcript's seq continues past the log", async () => { + const pair = await both(); + await same(pair, async (env) => { + const before = seqsOnDisk(env, ids.rich!); + const r = await sendAndSettle(env, ids.rich!, "next"); + await env.transcript.flush(); + const after = seqsOnDisk(env, ids.rich!); + expect(after.slice(0, before.length)).toEqual(before); + expect(Math.min(...after.slice(before.length))).toBe(Math.max(...before) + 1); + expect(new Set(after).size).toBe(after.length); + const opts = env.made.at(-1)!.capturedOpts.at(-1)!; + return { r, history: opts.history.map((t) => [t.role, t.content, t.turnId]), hasQueried: env.made.at(-1)!.hasQueried, info: sessionOf(env, "rich").toInfo() }; + }); + }); + + it("a background task settling on a resumed session wakes it on the same history", async () => { + const pair = await both(); + await same(pair, async (env) => { + const s = sessionOf(env, "short"); + const p = env.made.find((x) => x.onSessionEvent && env.m._sessionForTest(ids.short!)?.providerId === x.id && s.toInfo().rotation?.claudeCodeSessionId === x.backingSessionId)!; + p.onSessionEvent!({ type: "background_tasks", tasks: [{ id: "bg1", kind: "local_bash", description: "build", status: "running" }] } as SessionScopedEvent); + p.onSessionEvent!({ type: "background_task_settled", taskId: "bg1", status: "completed", summary: "built" } as SessionScopedEvent); + for (let i = 0; i < 200 && p.capturedOpts.length === 0; i++) await Bun.sleep(5); + await settle(env, ids.short!); + expect(p.capturedOpts.length).toBe(1); + return { turns: p.capturedOpts.map((o) => ({ user: o.userMessage, history: o.history.map((t) => [t.role, t.content]) })), replay: await attach(env, ids.short!, PAGING) }; + }); + }); + + it("a persistent routine's run on its resumed session ends the same", async () => { + const pair = await both(); + await same(pair, async (env) => { + const r = await call(env, { type: "routine.run", routine: "daily" }); + if (r.type !== "routine.run.result") throw new Error(JSON.stringify(r)); + const run = await runEnds(env, ids.routine!, r.run.id); + expect(run.status).toBe("succeeded"); + expect(run.sessionId).toBe(ids.routineSession); + const p = env.made.find((x) => x.capturedOpts.length > 0)!; + expect(p.capturedOpts.at(-1)!.history.length).toBeGreaterThan(2); + return { run, history: p.capturedOpts.at(-1)?.history.map((t) => [t.role, t.content]) }; + }); + }); + + it("an idle release before first use re-seeds a backend that needs it once the history loads", async () => { + const pair = await both(); + await same(pair, async (env) => { + const s = sessionOf(env, "gemini"); + s.releaseIdleProvider({ idle: true }); + await Bun.sleep(50); + const p = env.made.find((x) => x.id === "gemini-cli")!; + // Before anything uses it: the lazy one owes the seed until it loads. + if (env.label === "lazy") expect(s.hydrated).toBe(false); + await sendAndSettle(env, ids.gemini!, "back again"); + expect(p.seededHistory?.length).toBe(2); + return { seeded: p.seededHistory?.map((t) => [t.role, t.content]), teardowns: p.teardownCount, history: p.capturedOpts.at(-1)?.history.map((t) => [t.role, t.content]) }; + }); + }); +}); + +describe("lazy session loading: restarts", () => { + it("restarting twice, with notices (some while unloaded) and a turn in between, keeps everything the same", async () => { + const pair = await both(); + const restart = async (env: Env): Promise => { + await closeEnv(env); + envs.splice(envs.indexOf(env), 1); + const next = openEnv(env.label, env.root, env.label === "lazy"); + envs.push(next); + await next.m.resumeSessions(); + return next; + }; + let { eager, lazy } = pair; + const models: Record = {}; + const step = async (env: Env) => { + await call(env, { type: "session.set_mode", sessionId: ids.short, mode: "autonomous", maxTurns: 2 }); + // A notice appended while "short" is unloaded (lazy), persisted across the restarts. + const n = (models[env.label] ?? 0) + 1; + models[env.label] = n; + const r = await call(env, { type: "session.set_model", sessionId: ids.short, model: n === 1 ? "sonnet" : "haiku" }); + expect([r.type, (r as { error?: string }).error]).toEqual(["response.ok", undefined]); + if (env.label === "lazy") expect(sessionOf(env, "short").hydrated).toBe(false); + await sendAndSettle(env, ids.rich!, "between restarts"); + return list(env); + }; + await same({ eager, lazy }, step); + eager = await restart(eager); + lazy = await restart(lazy); + await same({ eager, lazy }, list); + await same({ eager, lazy }, step); + eager = await restart(eager); + lazy = await restart(lazy); + await same({ eager, lazy }, list); + await same({ eager, lazy }, async (env) => ({ + short: await attach(env, ids.short!), + rich: await attach(env, ids.rich!, PAGING), + turns: await call(env, { type: "session.turns", sessionId: ids.rich }), + })); + for (const env of [eager, lazy]) { + const seqs = seqsOnDisk(env, ids.short!); + expect(new Set(seqs).size).toBe(seqs.length); + const replay = (await attach(env, ids.short!)).find((f) => f.type === "scrollback.replay") as { messages: SessionMessage[] }; + expect(replay.messages.filter((m) => m.metadata?.event === "session.model_changed").length).toBe(2); + } + }); +}); + +/** + * A session from a long-lived daemon: a transcript past the 24 MiB the + * scrollback is read from, and a canonical log past the 4 MiB the history + * is, written into both copies. `seqOf` gives line i of n its seq (none: undefined). + */ +function writeBig(root: string, seqOf: (i: number, n: number) => number | undefined = (i) => i): string { + const id = "0b16b16b-0000-4000-8000-00000000b16b"; + const dir = join(root, "transcripts"); + const at = (i: number) => new Date(Date.UTC(2026, 0, 1, 0, 0, i)).toISOString(); + const lines: string[] = []; + const canon: string[] = []; + const index: string[] = []; + const entries: Array<{ timestamp: string; message: object }> = []; + for (let t = 0; t < 1100; t++) { + const turnId = `turn-${String(t).padStart(5, "0")}`; + const prompt = `prompt ${t} ${"x".repeat(200)}`; + const reply = `reply ${t} ${"y".repeat(24_000)}`; + const user = { type: "session.message", sessionId: id, messageId: `u-${t}`, role: "user", content: prompt, identity: { sub: AUTH.sub, type: "human" }, timestamp: at(2 * t), turnId }; + const asst = { type: "session.message", sessionId: id, messageId: `a-${t}`, role: "assistant", content: reply, identity: { sub: `agent:${id}`, type: "agent", name: "big (Claude)" }, timestamp: at(2 * t + 1), turnId }; + entries.push({ timestamp: user.timestamp, message: user }, { timestamp: asst.timestamp, message: asst }); + canon.push(JSON.stringify({ op: "append", turn: { role: "user", content: prompt, turnId, prompt, at: user.timestamp } })); + canon.push(JSON.stringify({ op: "append", turn: { role: "assistant", content: reply.slice(0, 4_000), turnId } })); + index.push(JSON.stringify({ turnId, kind: "prompt", preview: `prompt ${t}`, startedAt: user.timestamp })); + } + for (let i = 0; i < entries.length; i++) lines.push(JSON.stringify({ seq: seqOf(i, entries.length), ...entries[i]! })); + writeFileSync(join(dir, `${id}.jsonl`), `${lines.join("\n")}\n`); + writeFileSync(join(dir, `${id}.canonical.jsonl`), `${canon.join("\n")}\n`); + writeFileSync(join(dir, `${id}.turns.jsonl`), `${index.join("\n")}\n`); + writeFileSync( + join(dir, `${id}.meta.json`), + JSON.stringify({ + sessionId: id, + sessionName: "big", + workdir: join(root, "repo"), + createdBy: AUTH.sub, + createdAt: at(0), + lastStatus: "idle", + lastActivityAt: at(2200), + accountId: AUTH.accountId, + projectId: AUTH.projectId, + providerId: "claude", + }), + ); + return id; +} + +describe("lazy session loading: a history past the resume budgets", () => { + it("loads the same windows: a partial scrollback paged from disk, a partial history read whole on demand", async () => { + const eager = openEnv("eager", cloneBase("eager"), false); + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(eager, lazy); + const id = writeBig(eager.root); + writeBig(lazy.root); + ids.big = id; + try { + await eager.m.resumeSessions(); + await lazy.m.resumeSessions(); + const pair = { eager, lazy }; + expect(lazy.m._sessionForTest(id)!.hydrated).toBe(false); + // A notice while unloaded: the load's window still starts where a resume's did. + await same(pair, async (env) => call(env, { type: "session.set_mode", sessionId: id, mode: "guarded" })); + await same(pair, async (env) => { + const replay = await attach(env, id, PAGING); + const s = env.m._sessionForTest(id)!; + const full = await s.fullCanonicalHistory(); + const pages: DaemonMessage[] = []; + let anchor = (replay.find((f) => f.type === "scrollback.replay") as { messages: SessionMessage[] }).messages[0]?.messageId; + // Back past the buffer's floor, into the transcript on disk. + for (let i = 0; anchor && i < 400; i++) { + const page = (await call(env, { type: "scrollback.page", sessionId: id, beforeMessageId: anchor, maxBytes: 2_000_000 })) as { messages: SessionMessage[]; hasMore: boolean; source?: string }; + pages.push({ ...page, messages: page.messages.map((m) => m.messageId) } as never); + if (!page.hasMore || page.messages.length === 0) break; + anchor = page.messages[0]!.messageId; + } + return { + replay, + partial: s.historyIsPartial, + inMemory: s.historyLength, + full: full.length, + last: [s.lastTurnId, s.lastTurnRole, s.lastAssistantText?.slice(0, 20)], + pages, + }; + }); + expect(lazy.m._sessionForTest(id)!.historyIsPartial).toBe(true); + } finally { + delete ids.big; + } + }); +}); + +describe("lazy session loading: the load itself", () => { + it("concurrent first uses share one load", async () => { + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(lazy); + await lazy.m.resumeSessions(); + const reads: string[] = []; + const load = lazy.transcript.loadTranscript.bind(lazy.transcript); + lazy.transcript.loadTranscript = (sessionId, opts) => { + reads.push(sessionId); + return load(sessionId, opts); + }; + const results = await Promise.all([ + attach(lazy, ids.rich!, PAGING), + call(lazy, { type: "scrollback.page", sessionId: ids.rich, maxBytes: 4096 }), + attach(lazy, ids.rich!), + call(lazy, { type: "session.fork", sessionId: ids.rich, name: "f", isolate: false }), + ]); + expect(results[1]!.type).toBe("scrollback.page.result"); + expect(results[3]!.type).toBe("response.ok"); + // One load; the fork reads the parent's transcript itself, as it always has. + expect(reads.filter((r) => r === ids.rich).length).toBe(2); + expect(sessionOf(lazy, "rich").hydrated).toBe(true); + }); + + it("an unreadable transcript is an error for the request that needed it, and the next one loads it", async () => { + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(lazy); + await lazy.m.resumeSessions(); + const path = lazy.transcript.transcriptPath(ids.short!); + const { chmodSync } = await import("node:fs"); + chmodSync(path, 0o000); + try { + const r = await call(lazy, { type: "session.attach", sessionId: ids.short }); + expect(r.type).toBe("response.error"); + expect(sessionOf(lazy, "short").hydrated).toBe(false); + const send = await call(lazy, { type: "session.send", sessionId: ids.short, text: "hi" }); + expect(send.type).toBe("response.error"); + } finally { + chmodSync(path, 0o600); + } + const ok = await attach(lazy, ids.short!, PAGING); + expect(ok[0]!.type).toBe("response.ok"); + expect(sessionOf(lazy, "short").hydrated).toBe(true); + expect((ok[1] as { messages: unknown[] }).messages.length).toBeGreaterThan(0); + }); +}); + +describe("lazy session loading: a log that can't be read at start", () => { + it("a transcript that can't be read fails the session's resume, as an eager resume did, and leaves it as it was for the next start", async () => { + const { chmodSync } = await import("node:fs"); + const roots = { eager: cloneBase("eager"), lazy: cloneBase("lazy") }; + const paths = Object.values(roots).map((r) => join(r, "transcripts", `${ids.short}.jsonl`)); + const before = readFileSync(paths[0]!, "utf8"); + for (const p of paths) chmodSync(p, 0o000); + try { + const pair = { eager: openEnv("eager", roots.eager, false), lazy: openEnv("lazy", roots.lazy, true) }; + envs.push(pair.eager, pair.lazy); + await pair.eager.m.resumeSessions(); + await pair.lazy.m.resumeSessions(); + // Not listed as a session with no history (its seq at 0, its backend starting over): not listed. + await same(pair, async (env) => (await list(env)).map((i) => i.name)); + expect(pair.lazy.m._sessionForTest(ids.short!)).toBeUndefined(); + await same(pair, (env) => call(env, { type: "session.set_model", sessionId: ids.short, model: "sonnet" })); + for (const env of [pair.eager, pair.lazy]) await closeEnv(env); + envs.splice(0); + } finally { + for (const p of paths) chmodSync(p, 0o600); + } + for (const p of paths) expect(readFileSync(p, "utf8")).toBe(before); + // Readable again: the next start resumes it whole. + const pair = { eager: openEnv("eager", roots.eager, false), lazy: openEnv("lazy", roots.lazy, true) }; + envs.push(pair.eager, pair.lazy); + await pair.eager.m.resumeSessions(); + await pair.lazy.m.resumeSessions(); + const queried = await same(pair, async (env) => env.made.map((p) => p.hasQueried)); + expect(queried.filter((q) => !q).length).toBe(1); // the empty session's + await same(pair, async (env) => attach(env, ids.short!, PAGING)); + await same(pair, async (env) => { + await sendAndSettle(env, ids.short!, "again"); + return attach(env, ids.short!, PAGING); + }); + }); + + it("a canonical log or turn index that can't be read: the session resumes without its history, as an eager resume did, and reads it on its next use once it can", async () => { + const { chmodSync } = await import("node:fs"); + const roots = { eager: cloneBase("eager"), lazy: cloneBase("lazy") }; + const paths = Object.values(roots).map((r) => join(r, "transcripts", `${ids.short}.canonical.jsonl`)); + for (const p of paths) chmodSync(p, 0o000); + const pair = { eager: openEnv("eager", roots.eager, false), lazy: openEnv("lazy", roots.lazy, true) }; + envs.push(pair.eager, pair.lazy); + try { + await pair.eager.m.resumeSessions(); + await pair.lazy.m.resumeSessions(); + await same(pair, async (env) => { + const s = sessionOf(env, "short"); + return { names: (await list(env)).map((i) => i.name), history: s.historyLength, last: s.lastTurnId, replay: await attach(env, ids.short!, PAGING) }; + }); + } finally { + for (const p of paths) chmodSync(p, 0o600); + } + // Readable again: the next use reads it. + await same(pair, async (env) => { + await attach(env, ids.short!, PAGING); + const s = sessionOf(env, "short"); + return { history: s.historyLength, last: s.lastTurnId }; + }); + expect(sessionOf(pair.lazy, "short").historyLength).toBe(2); + }); +}); + +describe("lazy session loading: logs whose seqs don't grow along the file", () => { + /** Line i of n's seq in each shape the audits found on disk (undefined: the line has none). */ + const SHAPES: Record number | undefined> = { + // A daemon from before 2026-07-03 started every run's seqs at 0. + "seqs restarted at 0 by an older daemon": (i, n) => (i < n - 2 ? i : i - (n - 2)), + "the newest lines' seqs back down": (i, n) => (i < n - 4 ? i : i - (n - 4)), + // An import writes 1..N; the imported session's own prompt then started at 0. + "an import, then a prompt": (i, n) => (i < n - 2 ? i + 1 : i - (n - 2)), + "no seqs at all": () => undefined, + "the newest lines without a seq": (i, n) => (i < n - 2 ? i : undefined), + }; + + /** Rewrite a session's transcript with `shape`'s seqs, its lines unchanged otherwise. */ + function reshape(root: string, sessionId: string, shape: (i: number, n: number) => number | undefined): void { + const path = join(root, "transcripts", `${sessionId}.jsonl`); + const lines = readFileSync(path, "utf8").split("\n").filter((l) => l.trim()); + const out = lines.map((l, i) => { + const { seq: _seq, ...rest } = JSON.parse(l) as { seq?: number }; + const seq = shape(i, lines.length); + return JSON.stringify(seq === undefined ? rest : { seq, ...rest }); + }); + writeFileSync(path, `${out.join("\n")}\n`); + } + + /** What a resume read before lazy loading (origin/main's loader): the log's newest 24 MiB, merged, as it is now. */ + async function eagerRead(root: string, sessionId: string): Promise<{ messages: unknown[]; truncated: boolean }> { + const stats: { truncated?: boolean } = {}; + const entries = await new TranscriptStore(join(root, "transcripts")).loadTranscript(sessionId, { maxBytes: 24 * 1024 * 1024, stats }); + return { messages: entries.map((e) => e.message), truncated: stats.truncated === true }; + } + + /** A replayed message as the log holds it (the buffer stamps its own seq; a resumed tool call is reconciled). */ + const shown = (m: unknown) => { + const { seq: _seq, ...rest } = m as SessionMessage; + return { messageId: rest.messageId, role: rest.role, content: rest.content }; + }; + + for (const [label, shape] of Object.entries(SHAPES)) { + it(`${label}: every line loads, as an eager resume read it, with and without lazy loading`, async () => { + const roots = { eager: cloneBase("eager"), lazy: cloneBase("lazy") }; + for (const root of Object.values(roots)) { + reshape(root, ids.rich!, shape); + reshape(root, ids.legacy!, shape); + } + const want = await eagerRead(roots.eager, ids.rich!); + const wantLegacy = await eagerRead(roots.eager, ids.legacy!); + expect(want.messages.length).toBeGreaterThan(10); + const pair = { eager: openEnv("eager", roots.eager, false), lazy: openEnv("lazy", roots.lazy, true) }; + envs.push(pair.eager, pair.lazy); + await pair.eager.m.resumeSessions(); + await pair.lazy.m.resumeSessions(); + expect(loaded(pair.lazy, "rich")).toBe(false); + // Every backend with a conversation on disk is told to resume it, seqs or none. + expect(pair.eager.made.length).toBeGreaterThan(0); + const queried = await same(pair, async (env) => env.made.map((p) => p.hasQueried)); + // All but the empty session's. + expect(queried.filter((q) => !q).length).toBe(1); + // A notice while unloaded is appended past the log, whatever its seqs. + await same(pair, (env) => call(env, { type: "session.set_model", sessionId: ids.rich, model: "sonnet" })); + const replays = await same(pair, async (env) => { + const replay = (await attach(env, ids.rich!))[1] as { messages: SessionMessage[] }; + const s = sessionOf(env, "rich"); + return { messages: replay.messages, partial: s.historyIsPartial }; + }); + // What the log held, in its order, then the notice. + expect(replays.messages.slice(0, -1).map(shown)).toEqual(want.messages.map(shown)); + expect(replays.messages.at(-1)!.content).toContain("Model switched"); + expect(replays.partial).toBe(want.truncated); + // A session from before the canonical log rebuilds its history from every line, and persists it. + await same(pair, async (env) => { + const s = sessionOf(env, "legacy"); + const replay = (await attach(env, ids.legacy!))[1] as { messages: SessionMessage[] }; + expect(replay.messages.map(shown)).toEqual(wantLegacy.messages.map(shown)); + return { history: s.historyLength, turns: (await call(env, { type: "session.turns", sessionId: ids.legacy })) }; + }); + // A new prompt's lines continue past every seq the log holds. + await same(pair, (env) => sendAndSettle(env, ids.rich!, "after the restart")); + for (const env of [pair.eager, pair.lazy]) { + const seqs = seqsOnDisk(env, ids.rich!); + const before = seqs.slice(0, want.messages.length).filter((q) => q !== undefined); + const newest = seqs.slice(-2); + if (before.length > 0) for (const q of newest) expect(q).toBeGreaterThan(Math.max(...before)); + } + // And the next restart reads the same log from both. + const after = await Promise.all([eagerRead(roots.eager, ids.rich!), eagerRead(roots.lazy, ids.rich!)]); + expect(norm(pair.lazy, after[1].messages.map(shown))).toEqual(norm(pair.eager, after[0].messages.map(shown))); + }); + } + + it("a log past the read budget whose seqs restarted: the same window, the same partial flag", async () => { + const pair = { eager: openEnv("eager", cloneBase("eager"), false), lazy: openEnv("lazy", cloneBase("lazy"), true) }; + envs.push(pair.eager, pair.lazy); + // Seqs restarted at 0 for the newest 20 turns. + const restart = (i: number, n: number) => (i < n - 40 ? i : i - (n - 40)); + const id = writeBig(pair.eager.root, restart); + writeBig(pair.lazy.root, restart); + const want = await eagerRead(pair.eager.root, id); + expect(want.truncated).toBe(true); + ids.big = id; + try { + await pair.eager.m.resumeSessions(); + await pair.lazy.m.resumeSessions(); + await same(pair, (env) => call(env, { type: "session.set_model", sessionId: id, model: "sonnet" })); + const got = await same(pair, async (env) => { + const s = env.m._sessionForTest(id)!; + const page = (await call(env, { type: "scrollback.page", sessionId: id, maxBytes: 2_000_000 })) as { messages: SessionMessage[] }; + return { ids: page.messages.map((m) => m.messageId), partial: s.historyIsPartial }; + }); + // The newest page is what was read, in its order, up to its newest line, then the notice. + const read = want.messages.map((m) => (m as SessionMessage).messageId); + const kept = got.ids.slice(0, -1); + expect(kept.length).toBeGreaterThan(50); + const at = kept.map((k) => read.indexOf(k!)); + expect(at.every((i, n) => i >= 0 && (n === 0 || i > at[n - 1]!))).toBe(true); + expect(kept.at(-1)).toBe(read.at(-1)); + // It reaches back past where the seqs restarted (turn 1080). + expect(kept).toContain("a-1070"); + expect(got.partial).toBe(true); + } finally { + delete ids.big; + } + }); +}); + +describe("lazy session loading: messages while a load is in flight", () => { + /** The fixture resumed eagerly and lazily, every transcript read held `ms` (a large log). */ + async function slowPair(ms: number): Promise<{ eager: Env; lazy: Env }> { + const pair = await both(); + for (const env of [pair.eager, pair.lazy]) { + const load = env.transcript.loadTranscript.bind(env.transcript); + env.transcript.loadTranscript = async (sessionId, opts) => { + await Bun.sleep(ms); + return load(sessionId, opts); + }; + } + return pair; + } + + /** A turn's worth of provider sends the session's backends were handed. */ + const prompts = (env: Env) => env.made.reduce((n, p) => n + p.capturedOpts.length, 0); + + it("are handled in the order they came: an interrupt stops the prompt sent before it, a detach follows its attach", async () => { + const pair = await slowPair(50); + await same(pair, async (env) => { + const sent = env.m.handle({ type: "session.send", id: "s", sessionId: ids.rich, text: "go" } as never, AUTH, client("c", [])); + const stop = env.m.handle({ type: "session.interrupt", id: "i", sessionId: ids.rich } as never, AUTH, client("c", [])); + const responses = await Promise.all([sent, stop]); + await settle(env, ids.rich!); + return { responses, prompts: prompts(env), turns: sessionOf(env, "rich").toInfo().usage?.numTurns }; + }); + expect(prompts(pair.lazy)).toBe(0); + await same(pair, async (env) => { + const frames: DaemonMessage[] = []; + const c = client("ws", frames); + const attached = env.m.handle({ type: "session.attach", id: "a", sessionId: ids.short } as never, AUTH, c); + const detached = env.m.handle({ type: "session.detach", id: "d", sessionId: ids.short } as never, AUTH, c); + const responses = await Promise.all([attached, detached]); + return { responses: responses.map((r) => r.type), clients: sessionOf(env, "short").attachedClientCount }; + }); + expect(sessionOf(pair.lazy, "short").attachedClientCount).toBe(0); + }); + + it("an attach whose socket closed while it waited leaves no client attached", async () => { + const pair = await slowPair(50); + await same(pair, async (env) => { + const attached = env.m.handle({ type: "session.attach", id: "a", sessionId: ids.rich } as never, AUTH, client("ws1", [])); + env.m.disconnectClient("ws1"); // the socket's close handler + await attached; + return sessionOf(env, "rich").attachedClientCount; + }); + expect(sessionOf(pair.lazy, "rich").attachedClientCount).toBe(0); + // Another client of the same session is unaffected. + const frames: DaemonMessage[] = []; + await pair.lazy.m.handle({ type: "session.attach", id: "b", sessionId: ids.rich } as never, AUTH, client("ws2", frames)); + expect(sessionOf(pair.lazy, "rich").attachedClientCount).toBe(1); + }); + + it("a prompt waiting on the load is work the shutdown drain sees: no turn starts after it", async () => { + const pair = await slowPair(300); + for (const env of [pair.eager, pair.lazy]) { + const sent = env.m.handle({ type: "session.send", id: "s", sessionId: ids.rich, text: "go" } as never, AUTH, client("c", [])); + expect(sessionOf(env, "rich").preparingTurn).toBe(true); + await env.m.drain(5_000); + const drained = prompts(env); + const reply = (await sent) as { type: string; error?: string }; + if (env === pair.eager) { + // It was already preparing when the drain began: accepted, then stopped before it went out. + expect(reply.type).toBe("response.ok"); + } else { + // Still waiting on the load when the drain began: refused, saying so (no turn starts once a shutdown has). + expect(reply.type).toBe("response.error"); + expect(reply.error).toContain("shutting down"); + } + await Bun.sleep(100); + expect(prompts(env)).toBe(drained); + expect(sessionOf(env, "rich").preparingTurn).toBe(false); + expect(sessionOf(env, "rich").status).toBe("idle"); + } + expect(prompts(pair.lazy)).toBe(0); + }); + + it("the drain waits for a message waiting on a load", async () => { + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(lazy); + await lazy.m.resumeSessions(); + const load = lazy.transcript.loadTranscript.bind(lazy.transcript); + lazy.transcript.loadTranscript = async (sessionId, opts) => { + await Bun.sleep(300); + return load(sessionId, opts); + }; + let answered = false; + const attached = lazy.m.handle({ type: "session.attach", id: "a", sessionId: ids.rich } as never, AUTH, client("c", [])).then((r) => { + answered = true; + return r; + }); + await lazy.m.drain(5_000); + expect(answered).toBe(true); + expect((await attached).type).toBe("response.ok"); + }); + + it("a request its handler refuses on scope loads nothing", async () => { + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(lazy); + await lazy.m.resumeSessions(); + const asked: Array<[string, Record]> = [ + ["session.attach", {}], + ["scrollback.page", { maxBytes: 4096 }], + ["session.send", { text: "hi" }], + ["session.part_action", { messageId: "m", action: "a" }], + ["session.rotate", {}], + ["session.rewind", { turnId: "t" }], + ["session.approve", { approvalId: "x", decision: "allow" }], + ["session.fork", { name: "f" }], + ["session.set_provider", { providerId: "codex" }], + ]; + for (const scopes of [[], ["session:watch"]] as const) { + const auth: AuthContext = { ...AUTH, scopes: [...scopes] as AuthContext["scopes"] }; + for (const [type, rest] of asked) { + if (type === "session.attach" || type === "scrollback.page") { + if (scopes.length > 0) continue; // a watcher may read the history + } + const r = await lazy.m.handle({ type, id: "r", sessionId: ids.rich, ...rest } as never, auth, client("c", [])); + expect([type, r.type, (r as { error?: string }).error?.startsWith("Missing scope")]).toEqual([type, "response.error", true]); + expect(sessionOf(lazy, "rich").hydrated).toBe(false); + } + } + // A watcher's attach does load it. + const watcher: AuthContext = { ...AUTH, scopes: ["session:watch"] as AuthContext["scopes"] }; + const r = await lazy.m.handle({ type: "session.attach", id: "w", sessionId: ids.rich } as never, watcher, client("w", [])); + expect(r.type).toBe("response.ok"); + expect(sessionOf(lazy, "rich").hydrated).toBe(true); + }); + + it("a request its handler refuses for another reason, or that never reads the history, loads nothing", async () => { + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(lazy); + await lazy.m.resumeSessions(); + const delegated: AuthContext = { ...AUTH, delegationDepth: 1 }; + const ask = async (msg: Record, auth: AuthContext = AUTH) => { + const r = await lazy.m.handle({ id: "r", sessionId: ids.rich, ...msg } as never, auth, client("c", [])); + expect(sessionOf(lazy, "rich").hydrated).toBe(false); + return r; + }; + // mcp.adopt only reads the session's workdir; compare.keep its status. + expect(((await ask({ type: "mcp.adopt", server: "srv" }, delegated)) as { error?: string }).error).toContain("delegated token"); + expect(((await ask({ type: "compare.keep", compareId: "nope" })) as { error?: string }).error).toBe("Comparison not found"); + // A running loop's worker: what its guard refuses never reaches the history. + const loops = lazy.m.loops!; + const runOfWorker = loops.runOfWorker.bind(loops); + loops.runOfWorker = (id) => (id === ids.rich ? "loop_test" : runOfWorker(id)); + try { + for (const msg of [ + { type: "session.rotate" }, + { type: "session.rewind", turnId: "t" }, + { type: "session.set_provider", providerId: "codex" }, + ]) { + expect(((await ask(msg)) as { error?: string }).error).toContain("running loop's worker"); + } + const approve = await ask({ type: "session.approve", approvalId: "x", decision: "allow" }, delegated); + expect((approve as { error?: string }).error).toContain("loop worker"); + } finally { + loops.runOfWorker = runOfWorker; + } + }); +}); + +describe("lazy session loading: a canonical log that can't be read", () => { + /** The canonical log's lines on disk. */ + const canonicalLines = (env: Env, sessionId: string) => + readFileSync(env.transcript.canonicalPath(sessionId), "utf8").split("\n").filter((l) => l.trim()).length; + const newestTurn = async (env: Env, sessionId: string) => + ((await call(env, { type: "session.turns", sessionId })) as { turns: Array<{ turnId: string }> }).turns.at(-1)!.turnId; + + it("at the first use: that request fails, nothing is lost, and the next use loads the whole conversation", async () => { + const pair = await both(); + const { chmodSync } = await import("node:fs"); + const path = pair.lazy.transcript.canonicalPath(ids.rich!); + chmodSync(path, 0o000); + try { + const r = await call(pair.lazy, { type: "session.attach", sessionId: ids.rich }); + expect(r.type).toBe("response.error"); + expect(sessionOf(pair.lazy, "rich").hydrated).toBe(false); + const send = await call(pair.lazy, { type: "session.send", sessionId: ids.rich, text: "hi" }); + expect(send.type).toBe("response.error"); + } finally { + chmodSync(path, 0o600); + } + expect(pair.lazy.made.reduce((n, p) => n + p.capturedOpts.length, 0)).toBe(0); + await same(pair, async (env) => { + const replay = await attach(env, ids.rich!, PAGING); + const s = sessionOf(env, "rich"); + return { replay, history: s.historyLength, last: s.lastTurnId, text: s.lastAssistantText }; + }); + expect(sessionOf(pair.lazy, "rich").historyLength).toBe(12); + // Going back the newest turn keeps the rest on disk. + await same(pair, async (env) => { + const r = await call(env, { type: "session.rewind", sessionId: ids.rich, turnId: await newestTurn(env, ids.rich!) }); + expect(r.type).toBe("session.rewind.result"); + await settle(env, ids.rich!); + return canonicalLines(env, ids.rich!); + }); + expect(canonicalLines(pair.lazy, ids.rich!)).toBe(10); + }); + + it("a synchronous load that can't read it throws, and leaves the session unloaded for the next use", async () => { + const lazy = openEnv("lazy", cloneBase("lazy"), true); + envs.push(lazy); + await lazy.m.resumeSessions(); + const { chmodSync } = await import("node:fs"); + const path = lazy.transcript.canonicalPath(ids.rich!); + const s = sessionOf(lazy, "rich"); + chmodSync(path, 0o000); + try { + expect(() => s.canonicalHistory).toThrow(); + expect(s.hydrated).toBe(false); + } finally { + chmodSync(path, 0o600); + } + expect(s.canonicalHistory.length).toBe(12); + expect(s.hydrated).toBe(true); + }); + + it("at resume: the session runs without it, as before, never writes over it, and reads it again on its next use", async () => { + const pair = { eager: openEnv("eager", cloneBase("eager"), false), lazy: openEnv("lazy", cloneBase("lazy"), true) }; + envs.push(pair.eager, pair.lazy); + let failing = true; + for (const env of [pair.eager, pair.lazy]) { + const load = env.transcript.loadCanonical.bind(env.transcript); + env.transcript.loadCanonical = (sessionId, opts) => { + if (failing && sessionId === ids.rich) return Promise.reject(new Error("EIO: i/o error, read")); + return load(sessionId, opts); + }; + } + await pair.eager.m.resumeSessions(); + await pair.lazy.m.resumeSessions(); + const env = pair.eager; + expect(sessionOf(env, "rich").historyLength).toBe(0); + const turnId = await newestTurn(env, ids.rich!); + // Going back or forking would rewrite the log from what's in memory: refused, the log untouched. + const rewind = await call(env, { type: "session.rewind", sessionId: ids.rich, turnId }); + expect([rewind.type, (rewind as { error?: string }).error]).toEqual(["response.error", expect.stringContaining("couldn't be read")]); + const preview = await call(env, { type: "session.rewind", sessionId: ids.rich, turnId, dryRun: true }); + expect(preview.type).toBe("response.error"); + for (const fork of [{}, { afterTurnId: turnId }]) { + const r = await call(env, { type: "session.fork", sessionId: ids.rich, name: "f", isolate: false, ...fork }); + expect([r.type, (r as { error?: string }).error]).toEqual(["response.error", expect.stringContaining("couldn't be read")]); + } + expect(canonicalLines(env, ids.rich!)).toBe(12); + // A turn still runs, and is appended to the log. + expect((await sendAndSettle(env, ids.rich!, "more")).type).toBe("response.ok"); + expect(canonicalLines(env, ids.rich!)).toBe(14); + expect(sessionOf(env, "rich").historyLength).toBe(2); + // Readable again: the next use reads it, the turn since included. + failing = false; + await attach(env, ids.rich!, PAGING); + expect(sessionOf(env, "rich").historyLength).toBe(14); + const r = await call(env, { type: "session.rewind", sessionId: ids.rich, turnId: await newestTurn(env, ids.rich!) }); + expect(r.type).toBe("session.rewind.result"); + await settle(env, ids.rich!); + expect(canonicalLines(env, ids.rich!)).toBe(12); + // The lazy daemon read it at the first use, failing or not. + expect(sessionOf(pair.lazy, "rich").hydrated).toBe(false); + await attach(pair.lazy, ids.rich!, PAGING); + expect(sessionOf(pair.lazy, "rich").historyLength).toBe(12); + }); + + it("at resume: a rotation empties it, as it always does, and nothing is read back after it", async () => { + const env = openEnv("eager", cloneBase("eager"), false); + envs.push(env); + let failing = true; + const load = env.transcript.loadCanonical.bind(env.transcript); + env.transcript.loadCanonical = (sessionId, opts) => + failing && sessionId === ids.rich ? Promise.reject(new Error("EIO: i/o error, read")) : load(sessionId, opts); + await env.m.resumeSessions(); + // Past the turns a rotation needs, each appended to the log. + for (const text of ["one", "two", "three"]) await sendAndSettle(env, ids.rich!, text); + await env.transcript.flush(); + expect(canonicalLines(env, ids.rich!)).toBe(18); + expect((await call(env, { type: "session.rotate", sessionId: ids.rich })).type).toBe("response.ok"); + await settle(env, ids.rich!); + expect(sessionOf(env, "rich").toInfo().rotation?.count).toBe(1); + await env.transcript.flush(); + expect(canonicalLines(env, ids.rich!)).toBe(0); + failing = false; + await attach(env, ids.rich!, PAGING); + expect(sessionOf(env, "rich").historyLength).toBe(0); + expect((await sendAndSettle(env, ids.rich!, "after")).type).toBe("response.ok"); + expect(sessionOf(env, "rich").historyLength).toBe(2); + expect(canonicalLines(env, ids.rich!)).toBe(2); + }); +}); diff --git a/src/tests/transcript.test.ts b/src/tests/transcript.test.ts index 8a032b37..a3671b75 100644 --- a/src/tests/transcript.test.ts +++ b/src/tests/transcript.test.ts @@ -352,3 +352,43 @@ describe("TranscriptStore — rotation, bounded load, output cap (#85)", () => { expect(await Bun.file(`${small.transcriptPath("gone")}.2`).exists()).toBe(false); }); }); + +describe("TranscriptStore — the log as it was (lazy session loading)", () => { + test("a read as of a tail's marks leaves out what was appended since, whatever its seqs, rotated or not", async () => { + const small = new TranscriptStore(tmpDir, { segmentMaxBytes: 1500, maxRotatedSegments: 3 }); + const ids: string[] = []; + // Seqs that go back down, as an older daemon's restarts left them. + for (let i = 0; i < 6; i++) { + const msg = makeMsg(`before ${i} ${"x".repeat(120)}`); + ids.push(msg.messageId); + await small.append("asof", msg, i < 4 ? i : i - 4); + } + await small.flush(); + const want = (await small.loadTranscript("asof")).map((e) => (e.message as SessionMessage).messageId); + expect(want).toEqual(ids); + const tail = await small.transcriptTail("asof"); + expect(tail).toMatchObject({ nextSeq: 2, hasEntries: true }); + // Appended since, at seqs the log already holds, and enough to rotate the files. + for (let i = 0; i < 4; i++) await small.append("asof", makeMsg(`after ${i} ${"y".repeat(200)}`), 2 + i); + await small.flush(); + expect(await Bun.file(`${small.transcriptPath("asof")}.2`).exists()).toBe(true); + const got = await small.loadTranscript("asof", { asOf: tail.files }); + expect(got.map((e) => (e.message as SessionMessage).messageId)).toEqual(want); + const gotSync = small.loadTranscriptSync("asof", { asOf: tail.files }); + expect(gotSync.map((e) => (e.message as SessionMessage).messageId)).toEqual(want); + }); + + test("a log with entries but no seqs has entries; an unreadable one is an error, never an empty log", async () => { + const { appendFileSync, chmodSync } = await import("node:fs"); + const path = store.transcriptPath("noseq"); + appendFileSync(path, `${JSON.stringify({ timestamp: "t", message: makeMsg("no seq") })}\n`); + expect(await store.transcriptTail("noseq")).toMatchObject({ nextSeq: 0, hasEntries: true }); + chmodSync(path, 0o000); + try { + await expect(store.transcriptTail("noseq")).rejects.toThrow(); + } finally { + chmodSync(path, 0o600); + } + expect(await store.transcriptTail("missing")).toEqual({ nextSeq: 0, hasEntries: false, files: [] }); + }); +});