From 1c87cd2ac782a1ad150f884f91a6df969b8e5751 Mon Sep 17 00:00:00 2001 From: Randalix Date: Fri, 23 Oct 2026 15:35:12 +0200 Subject: [PATCH] =?UTF-8?q?fix(edit-history):=20review=20round=203=20?= =?UTF-8?q?=E2=80=94=20restored=20sessions=20read=20again,=20/clear=20read?= =?UTF-8?q?s=20the=20old=20conversation=20to=20its=20end,=20bounded=20memo?= =?UTF-8?q?ry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - remove() drops the session's slot once the file is gone: a session that comes back with the same id (reboot restore) starts afresh instead of answering `removed` until the next restart. A run in flight stays on its own slot's store, and the routes start a sync in the same tick as their session check, so a deletion right after it still stops that sync. A store that is not loaded is deleted without being parsed (deleteStore). - A new conversation (/clear, resume) first reads the previous transcript, next to the new one, to its end with its subagents, so edits it got after the last read are not lost with its cursors. "Load older" reads the stored conversation while the switch is pending. - At most MAX_LOADED_STORES (3) stores stay in memory; the least recently used one without a run is written if it changed (settle) and loads again on demand. - Tmp files are named per process start, so a reused pid (a container's pid 1) no longer keeps a stale one through the startup sweep. - load() keeps only a plain transcript id (the previous transcript's path is built from it) and no cursors without one; a pending Bash call with an id, cwd or agent id over 4,096 characters is not kept at parse time either. - nextRange() is gone: production reads forward through the reader. The store tests mirror the reader's range choice, and a new end-to-end test runs the real reader, sync and store (/clear cases, 40 seeded random transcripts). - The wiring guard scans all of src for holders of the sync. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/api-reference.md | 4 +- docs/architecture-invariants.md | 6 +- docs/wiki/Working-With-Files.md | 3 + src/edit-history-parse.ts | 3 + src/edit-history-store.ts | 76 +++++---- src/web/edit-history-sync.ts | 143 ++++++++++++---- src/web/routes/edit-history-routes.ts | 13 +- src/web/server.ts | 1 + test/edit-history-parse.test.ts | 12 ++ test/edit-history-pipeline.test.ts | 214 ++++++++++++++++++++++++ test/edit-history-store.test.ts | 70 +++++++- test/edit-history-sync.test.ts | 113 +++++++++++-- test/routes/edit-history-routes.test.ts | 13 ++ 13 files changed, 581 insertions(+), 90 deletions(-) create mode 100644 test/edit-history-pipeline.test.ts diff --git a/docs/api-reference.md b/docs/api-reference.md index 458450ca..47192502 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -1087,11 +1087,11 @@ Lists are capped (300 files and 50 commits per repository) while the counts stay What the Edit history panel reads (Settings → Header & Panels → Header, per-device, default off): the edits a **local claude session** made, main agent and subagents, read from its transcripts. Read-only. Every route resolves the session through `findSessionOrFail` (another user's session is `404`). Remote, Docker and non-claude sessions answer `state: 'unsupported'` and are never read. -`GET /api/sessions/:id/edits[?sync=1]` lists the stored edits, newest first, without diffs: `{ state, reason?, backfillTruncated, agents, entries }`. With `sync=1` it first reads what the transcripts gained since the last read; without it the server reads nothing. `state` is `ok`, `unavailable` (stored entries only; `reason` is `no-transcript` before the session has a transcript Codeman can find, `removed` once the session's history was deleted, or the read error; the transcript is the one the session's hooks report, else the newest `.jsonl` under the session's `CLAUDE_CONFIG_DIR` override or the server's config dir) or `unsupported` (`reason`: `remote`, `docker`, `mode`). Each entry is `{ id, ts, path, relPath, kind: 'code'|'image', tool, created, added, removed, truncated, agentId? }`, where `tool` is the edit tool, `bash` for an image a Bash call produced or `sent` for one the agent sent; `relPath` is relative to the workspace (a subagent's worktree for its edits). `agents` maps a subagent id to `{ description, agentType, worktreeBranch? }`. `backfillTruncated: true` means older edits can still be loaded. +`GET /api/sessions/:id/edits[?sync=1]` lists the stored edits, newest first, without diffs: `{ state, reason?, backfillTruncated, agents, entries }`. With `sync=1` it first reads what the transcripts gained since the last read (after `/clear` or a resume, the previous conversation's transcripts to their end first); without it the server reads nothing. `state` is `ok`, `unavailable` (stored entries only; `reason` is `no-transcript` before the session has a transcript Codeman can find, `removed` while the session's history is being deleted, or the read error; the transcript is the one the session's hooks report, else the newest `.jsonl` under the session's `CLAUDE_CONFIG_DIR` override or the server's config dir) or `unsupported` (`reason`: `remote`, `docker`, `mode`). Each entry is `{ id, ts, path, relPath, kind: 'code'|'image', tool, created, added, removed, truncated, agentId? }`, where `tool` is the edit tool, `bash` for an image a Bash call produced or `sent` for one the agent sent; `relPath` is relative to the workspace (a subagent's worktree for its edits). `agents` maps a subagent id to `{ description, agentType, worktreeBranch? }`. `backfillTruncated: true` means older edits can still be loaded. `GET /api/sessions/:id/edits/:entryId` returns one edit's diff as unified text: `{ diff, truncated }` (each edit's diff is capped at 64 KiB). `404` for an edit the session does not have. -`POST /api/sessions/:id/edits/older` reads the next older block of up to three transcripts, a full read budget each (with more transcripts, main and subagents take turns), and answers like the list. +`POST /api/sessions/:id/edits/older` reads the next older block, within the current conversation, of up to three transcripts, a full read budget each (with more transcripts, main and subagents take turns), and answers like the list. The edits are stored per session in `~/.codeman/edit-history/.json` (mode 0600, at most 500 edits and 4 MiB of diff text, the oldest dropped), deleted when the session's record is removed (a killed pinned session keeps it; at startup, a file whose session no longer exists is deleted too) and never fed into `/api/v1/search`. No route here serves a file: the panel shows thumbnails through `file-raw` inside the workspace only; an image outside it opens in the file preview, which registers it through the attachment route. diff --git a/docs/architecture-invariants.md b/docs/architecture-invariants.md index dad3a0f0..ac8fafd6 100644 --- a/docs/architecture-invariants.md +++ b/docs/architecture-invariants.md @@ -728,11 +728,11 @@ Matching semantics: a query containing `/` matches the relative path, otherwise ⚠️ **Reads only on request, opt-in all the way down.** `server.ts` builds the `EditHistorySync` and hands it to the routes; its only own calls are `remove` (through `forgetEditHistory`, when the session's persisted record goes), `sweep` (startup) and `flushAll` (shutdown). No hook, no timer, no second watcher. The panel drives every read: `?sync=1` when it opens, and while it is open on that session a debounced (600 ms) re-read after a `transcript:tool_end` of an edit-capable tool (Edit, Write, MultiEdit, NotebookEdit, Bash, Agent, Task, SendUserFile), which the existing `TranscriptWatcher` already broadcasts, or the session's `session:idle` (the end of a turn, via `_onSessionIdle`; it also catches a background subagent that writes after the tool end that started it). Not `transcript:complete`: it fires on a `result` line, which claude's interactive transcript never has. A closed panel reads nothing and keeps no badge. Guarded by the "reads only on request" wiring test in `test/edit-history-sync.test.ts`. -⚠️ **Which transcript** (`resolveClaudeTranscript`). The main transcript is the path the session's `TranscriptWatcher` follows (the hook-reported `transcript_path`), so a session is read from where its claude really writes, whatever its `CLAUDE_CONFIG_DIR`. Without a watcher (server restarted, no hook since), `findClaudeTranscript()` looks for `.jsonl` under `claudeProjectsDir(session.claudeConfigDir())`: the session's own override (#255, the same dir the relaunch pin checks), else the server's. A config dir set outside Codeman (a shell profile, a wrapper) is never guessed: until the next hook such a session answers `unavailable` / `no-transcript`, and the panel says so. Subagents are `
/subagents/agent-.jsonl`. A different main file (`/clear`, resume) is read from its end with fresh cursors; the stored entries stay. +⚠️ **Which transcript** (`resolveClaudeTranscript`). The main transcript is the path the session's `TranscriptWatcher` follows (the hook-reported `transcript_path`), so a session is read from where its claude really writes, whatever its `CLAUDE_CONFIG_DIR`. Without a watcher (server restarted, no hook since), `findClaudeTranscript()` looks for `.jsonl` under `claudeProjectsDir(session.claudeConfigDir())`: the session's own override (#255, the same dir the relaunch pin checks), else the server's. A config dir set outside Codeman (a shell profile, a wrapper) is never guessed: until the next hook such a session answers `unavailable` / `no-transcript`, and the panel says so. Subagents are `
/subagents/agent-.jsonl`. A different main file (`/clear`, resume): the previous one, next to it in the same directory, is first read to its end with its subagents (over as many syncs as that takes), then the new one from its end with fresh cursors; the stored entries stay. "Load older" reaches back within the current conversation only, and reads the stored conversation's transcript while the switch is still pending. -⚠️ **Bounded reads.** A sync is at most `MAX_FORWARD_PASSES` (3) reads, each at most `READ_BUDGET_BYTES` (8 MiB) shared by the main transcript and all subagents (main first); a first read starts that far before the end. A file whose chunk the shared budget cut short is not applied but read in a later pass (the store would take a chunk with no line end for one line longer than the budget). "Load older" reads a full-budget block of up to `MAX_OLDER_SOURCES` (3) transcripts per click; with more, they take turns. A line longer than the budget is skipped, going forward and going back, and the line after (before) it is read whole. A subagent transcript that vanished loses its cursor, not its entries. Parsing happens in the request, line by line; nothing runs on the PTY data path. One sync per session at a time, and requests during a run fold into one more run. +⚠️ **Bounded reads.** A sync is at most `MAX_FORWARD_PASSES` (3) reads, each at most `READ_BUDGET_BYTES` (8 MiB) shared by the main transcript and all subagents (main first), plus up to 8 KiB of `meta.json` per subagent seen for the first time; a first read starts that far before the end. A file whose chunk the shared budget cut short is not applied but read in a later pass (the store would take a chunk with no line end for one line longer than the budget). "Load older" reads a full-budget block of up to `MAX_OLDER_SOURCES` (3) transcripts per click; with more, they take turns. A line longer than the budget is skipped, going forward and going back, and the line after (before) it is read whole. A subagent transcript that vanished loses its cursor, not its entries. Parsing happens in the request, line by line; nothing runs on the PTY data path. One sync per session at a time, and requests during a run fold into one more run. At most `MAX_LOADED_STORES` (3) stores stay in memory: past that the least recently used one without a run in flight is written (only if it changed) and dropped, and a later request loads it again after that write. -⚠️ **The store holds secrets.** Every diff an agent wrote ends up in `dataPath('edit-history')/.json`, API keys and `.env` contents included. It is written 0600 via tmp + rename (a fresh file each time, so an existing store never keeps wider bits; a new directory is created 0700) and bounded per session: `MAX_ENTRIES` (500) and `MAX_STORE_BYTES` (4 MiB of diff text, paths and ids), oldest dropped, also on load. Each edit's diff is capped at `MAX_HUNK_BYTES` (64 KiB); the file also holds at most `MAX_PENDING_BASH` (200) unanswered Bash calls of at most 8,192 characters of command text each and `MAX_AGENTS` (501) subagent labels of at most 1,024 characters per string. `load()` takes the file as untrusted: one over `MAX_FILE_BYTES` (32 MiB) is not parsed, and the rest is rebuilt from its known fields under the same bounds (unknown fields and non-string diff lines dropped, the per-edit cap applied again). Writes are serialized. The store lives as long as the session's persisted record: `forgetEditHistory` runs when `cleanupSession` with `killMux` removes the record (a pinned session's record is only demoted and keeps its history), on the persisted-only `DELETE /api/sessions/:id`, for every record `cleanupStaleSessions` drops at startup and every one `POST /api/cleanup-state` drops. After that startup cleanup, `sweep` deletes the `edit-history/*.json` files whose session is neither live nor recorded (left behind by a crash between removing the record and deleting the file, or by an older build) and tmp files of other processes. After `remove()` nothing writes the file again, neither a sync in flight nor a later request (the session answers `unavailable` / `removed`). The store is blocked in `isSensitivePath` (`/.codeman*/edit-history/`, same reasoning as `intents.json`) and never fed to `/api/search`. +⚠️ **The store holds secrets.** Every diff an agent wrote ends up in `dataPath('edit-history')/.json`, API keys and `.env` contents included. It is written 0600 via tmp + rename (a fresh file each time, so an existing store never keeps wider bits; a new directory is created 0700) and bounded per session: `MAX_ENTRIES` (500) and `MAX_STORE_BYTES` (4 MiB of diff text, paths and ids), oldest dropped, also on load. Each edit's diff is capped at `MAX_HUNK_BYTES` (64 KiB); the file also holds one cursor per transcript file of the current conversation, at most `MAX_PENDING_BASH` (200) unanswered Bash calls of at most 8,192 characters of command text each (a call with an id, cwd or agent id over 4,096 characters is not kept, and dropped on load) and `MAX_AGENTS` (501) subagent labels of at most 1,024 characters per string. `load()` takes the file as untrusted: one over `MAX_FILE_BYTES` (32 MiB) is not parsed, and the rest is rebuilt from its known fields under the same bounds (unknown fields and non-string diff lines dropped, the per-edit cap applied again, a transcript id with anything but `[A-Za-z0-9._-]` dropped, since the previous transcript's path is built from it). Writes are serialized. The store lives as long as the session's persisted record: `forgetEditHistory` runs when `cleanupSession` with `killMux` removes the record (a pinned session's record is only demoted and keeps its history), on the persisted-only `DELETE /api/sessions/:id`, for every record `cleanupStaleSessions` drops at startup and every one `POST /api/cleanup-state` drops. After that startup cleanup, `sweep` deletes the `edit-history/*.json` files whose session is neither live nor recorded (left behind by a crash between removing the record and deleting the file, or by an older build) and tmp files of earlier runs (named per process start, so a pid used again, a container's pid 1, does not keep one). While `remove()` runs, nothing writes the file again: a sync in flight stops after its current read, and a request meanwhile reads nothing (`unavailable` / `removed`). The routes start a sync in the same tick as their session check, so a request that passed the check before the deletion cannot start one afterwards. Then the session's slot goes, and a session that comes back with the same id (reboot restore re-creates it) starts afresh. A store that is not loaded is deleted without being parsed. The store is blocked in `isSensitivePath` (`/.codeman*/edit-history/`, same reasoning as `intents.json`) and never fed to `/api/search`. ⚠️ **Images only through existing file surfaces.** A Bash image candidate counts only when the file's mtime lies in the call's window (start − 2 s … end + 2 s) and no entry already covers that file version; `SendUserFile` images need no window. The panel shows a thumbnail only inside the workspace, through the confined `file-raw` route. An image outside it gets no thumbnail (each would mint an attachment id per render); a click opens the shared file preview, which registers it once through `POST /api/sessions/:id/attachments` with `notify: false` (its extension allowlist and `isSensitivePath` apply). SVGs get no thumbnail. No route of this feature serves a file, least of all a transcript path. diff --git a/docs/wiki/Working-With-Files.md b/docs/wiki/Working-With-Files.md index cd3e430c..470b4b8b 100644 --- a/docs/wiki/Working-With-Files.md +++ b/docs/wiki/Working-With-Files.md @@ -275,6 +275,9 @@ Click it for a draggable window listing the active session's edits, newest first to open it in the preview. - **Load older** reads further back in the conversation, in the main agent's transcript and the subagents' alike (up to three transcripts per click; with more, they take turns). +- After `/clear` or a resume, the edits of the earlier conversation stay listed, and the window + first catches up on the ones it had not read yet. **Load older** then goes back within the new + conversation. It works for local Claude sessions only. Remote (SSH), Docker and other CLIs' sessions say so instead. diff --git a/src/edit-history-parse.ts b/src/edit-history-parse.ts index e89a2ce6..b86d24b9 100644 --- a/src/edit-history-parse.ts +++ b/src/edit-history-parse.ts @@ -34,6 +34,8 @@ export const MAX_PENDING_BASH = 200; * this bounds it; an image path past this point of a command is still found in the output. */ export const MAX_PENDING_COMMAND_CHARS = 8192; +/** Longest id, cwd or agent id of an unanswered Bash call; a call with a longer one is not kept. */ +export const MAX_PENDING_FIELD_CHARS = 4096; export type EditKind = 'code' | 'image'; /** `edit` = a result with a path and a patch from a tool other than Edit/Write. */ @@ -291,6 +293,7 @@ export function parseClaudeEditLines(text: string, pending: readonly PendingBash if (typeof block.id !== 'string' || !isRecord(block.input)) continue; const command = block.input.command; if (typeof command !== 'string') continue; + if ([block.id, cwd, agentId ?? ''].some((f) => f.length > MAX_PENDING_FIELD_CHARS)) continue; const entry: PendingBash = { id: block.id, ts, command: command.slice(0, MAX_PENDING_COMMAND_CHARS), cwd }; if (agentId) entry.agentId = agentId; open.delete(block.id); diff --git a/src/edit-history-store.ts b/src/edit-history-store.ts index 60a262f5..a8df090b 100644 --- a/src/edit-history-store.ts +++ b/src/edit-history-store.ts @@ -6,11 +6,11 @@ * (`subagents/agent-.jsonl`). Each has a cursor: * - `offset` — the next byte to read forward; always a line boundary. * - `floor` — the lowest byte already parsed; > 0 while older lines are unread - * ("ältere laden" reads `[floor − budget, floor)`). - * The store decides WHICH byte range to read (`nextRange`/`olderRange`); the caller reads - * it — locally or over ssh — and hands the bytes to `apply`. Cuts happen only at `\n`, - * which never occurs inside a UTF-8 sequence, so multibyte characters split by a chunk - * boundary are never decoded half. + * ("ältere laden" reads `[floor − budget, floor)`, see `olderRange`). + * The reader (`edit-history-local-source.ts`) reads forward from each `offset` of the + * current conversation, a file without one from `budget` bytes before its end, and hands the + * bytes to `apply`. Cuts happen only at `\n`, which never occurs inside a UTF-8 sequence, so + * multibyte characters split by a chunk boundary are never decoded half. * * A first read of a file starts `budget` bytes before its end (transcripts reach 100 MB+), * drops the cut first line and leaves `floor` there. A line longer than the budget is @@ -23,8 +23,8 @@ * A new conversation (`/clear`, resume) arrives as a new `transcriptId`: cursors and * pending Bash calls start over, entries stay. Entries are capped at {@link MAX_ENTRIES} * and {@link MAX_STORE_BYTES} of diff text; the oldest by time are dropped. Besides them - * the file holds the cursors, at most {@link MAX_PENDING_BASH} unanswered Bash calls and at - * most {@link MAX_AGENTS} subagent labels. + * the file holds one cursor per transcript file of the current conversation, at most + * {@link MAX_PENDING_BASH} unanswered Bash calls and at most {@link MAX_AGENTS} subagent labels. * * ⚠️ The file holds secrets: a diff carries whatever the agent wrote, keys and `.env` * contents included. So, like `intents.json`, it is written 0600 via tmp+rename, bounded, @@ -40,6 +40,7 @@ * @module edit-history-store */ +import { randomBytes } from 'node:crypto'; import { mkdir, readdir, readFile, rename, stat, unlink, writeFile } from 'node:fs/promises'; import { join } from 'node:path'; import { dataPath } from './config/instance.js'; @@ -48,6 +49,7 @@ import { MAX_HUNK_BYTES, MAX_PENDING_BASH, MAX_PENDING_COMMAND_CHARS, + MAX_PENDING_FIELD_CHARS, parseClaudeEditLines, type EditRecord, type EditTool, @@ -72,12 +74,17 @@ export const READ_BUDGET_BYTES = 8 * 1024 * 1024; * above stays far below it; JSON escaping at most doubles ordinary text. */ export const MAX_FILE_BYTES = 32 * 1024 * 1024; -/** Longest id, cwd or agent id of a loaded pending Bash call; a longer one is no real one. */ -const MAX_LOADED_FIELD_CHARS = 4096; export const MAIN_SOURCE = 'main'; const SAVE_DEBOUNCE_MS = 1000; const FILE_VERSION = 1; const NL = 0x0a; +/** A transcript id as the reader takes it (`.jsonl`): no path separators. */ +const TRANSCRIPT_ID = /^[A-Za-z0-9._-]+$/; +/** + * Names this process's tmp files (`.json..tmp`); any other tmp file is left over from + * an earlier run (`sweep`), even one whose pid this process has again (a container's pid 1). + */ +export const TMP_TAG = `${process.pid}-${randomBytes(4).toString('hex')}`; /** `bash`: an image a Bash call wrote; `sent`: an image claude delivered to the user. */ export type EditHistoryTool = EditTool | 'bash' | 'sent'; @@ -198,7 +205,7 @@ function cursorFrom(raw: unknown): SourceCursor | null { return cursor; } -const loadedField = (v: unknown): v is string => typeof v === 'string' && v.length <= MAX_LOADED_FIELD_CHARS; +const loadedField = (v: unknown): v is string => typeof v === 'string' && v.length <= MAX_PENDING_FIELD_CHARS; function pendingBashFrom(raw: unknown): PendingBash | null { if (!isObj(raw)) return null; @@ -279,8 +286,12 @@ export class EditHistoryStore { if (!isObj(data)) return store; if (data.version !== FILE_VERSION) return store; const f = data as unknown as StoreFile; - store.transcriptId = typeof f.transcriptId === 'string' ? f.transcriptId : null; - for (const [k, raw] of Object.entries(isObj(f.cursors) ? f.cursors : {})) { + // The sync builds the previous transcript's path from it (`/clear`), so only a plain id counts. + store.transcriptId = + typeof f.transcriptId === 'string' && TRANSCRIPT_ID.test(f.transcriptId) ? f.transcriptId : null; + // Cursors are offsets into that conversation's files; without it they mean nothing. + const cursors = store.transcriptId !== null && isObj(f.cursors) ? f.cursors : {}; + for (const [k, raw] of Object.entries(cursors)) { const cursor = cursorFrom(raw); if (cursor) store.cursors.set(k, cursor); } @@ -319,9 +330,9 @@ export class EditHistoryStore { } let removed = 0; for (const name of names) { - const tmp = /^(.+)\.json\.(\d+)\.tmp$/.exec(name); + const tmp = /\.json\.[\w-]+\.tmp$/.test(name); const stale = tmp - ? Number(tmp[2]) !== process.pid + ? !name.endsWith(`.json.${TMP_TAG}.tmp`) : name.endsWith('.json') && !keep(name.slice(0, -'.json'.length)); if (!stale) continue; await unlink(join(dir, name)).catch(() => undefined); @@ -330,6 +341,11 @@ export class EditHistoryStore { return removed; } + /** Delete the stored history of `sessionId` without loading it. */ + static async delete(sessionId: string, opts: Pick = {}): Promise { + await new EditHistoryStore(sessionId, opts).remove(); + } + getTranscriptId(): string | null { return this.transcriptId; } @@ -375,21 +391,6 @@ export class EditHistoryStore { this.scheduleSave(); } - /** - * The next forward range to read from `source`, or null when nothing is new. - * A conversation other than the stored one starts from scratch (see `apply`). - */ - nextRange(source: string, size: number, transcriptId: string, budget = READ_BUDGET_BYTES): ByteRange | null { - const cursor = transcriptId === this.transcriptId ? this.cursors.get(source) : undefined; - if (cursor && size >= cursor.offset) { - if (size === cursor.offset) return null; - return { offset: cursor.offset, len: Math.min(budget, size - cursor.offset) }; - } - // Unknown file, new conversation, or a file that shrank (replaced): read its end. - const offset = Math.max(0, size - budget); - return size > offset ? { offset, len: size - offset } : null; - } - /** The next older range of `source` ("ältere laden"), or null when it is read to its start. */ olderRange(source: string, budget = READ_BUDGET_BYTES): ByteRange | null { const cursor = this.cursors.get(source); @@ -418,8 +419,10 @@ export class EditHistoryStore { } /** - * Apply a forward chunk read at `nextRange`: a full budget unless it reaches the file's end. - * Returns new entries and Bash image candidates. + * Apply a forward chunk: read from the source's cursor, or, without one, its last `budget` + * bytes; a full budget unless it reaches the file's end. A file that shrank below its cursor + * (replaced) loses the cursor and is read from its end next time. Returns new entries and + * Bash image candidates. */ apply(input: ApplyInput): ApplyResult { if (this.removed) return { added: 0, candidates: [] }; @@ -546,6 +549,15 @@ export class EditHistoryStore { return added; } + /** + * Write pending changes, if any, and wait for a write under way (the store is dropped from + * memory and loaded again later). Unlike `flush()`, a store without changes writes nothing. + */ + async settle(): Promise { + if (this.saveTimer) return this.flush(); + await this.writing; + } + /** Write now (cancels a pending debounced write). Writes run one after another. */ async flush(): Promise { if (this.saveTimer) { @@ -593,7 +605,7 @@ export class EditHistoryStore { await mkdir(dir, { recursive: true, mode: 0o700 }); // `mode` applies only to a file being created: the fresh tmp file is 0600, and the rename // makes it the store, so an existing store never keeps wider bits. - const tmp = `${this.file}.${process.pid}.tmp`; + const tmp = `${this.file}.${TMP_TAG}.tmp`; await writeFile(tmp, JSON.stringify(data), { mode: 0o600 }); if (this.removed) { await unlink(tmp).catch(() => undefined); diff --git a/src/web/edit-history-sync.ts b/src/web/edit-history-sync.ts index eceb2c25..c886811d 100644 --- a/src/web/edit-history-sync.ts +++ b/src/web/edit-history-sync.ts @@ -13,9 +13,18 @@ * for a line longer than the budget) but read in the next pass; up to * {@link MAX_FORWARD_PASSES} passes per sync, the rest with the next one. * - * GONE. `remove()` (the session's record is deleted) marks the session gone before it deletes - * the file: a run in flight stops after its current read, its store no longer writes, and a - * later request reads nothing (`unavailable`, reason `removed`) instead of loading a new store. + * NEW CONVERSATION. When the transcript path names another conversation (`/clear`, resume), + * the previous transcript (same directory) is first read to its end, so edits it got after + * the last read are not lost with its cursors; then the new one is read from its end. + * "Load older" reaches back within the current conversation only. + * + * MEMORY. At most {@link MAX_LOADED_STORES} stores stay loaded; past that the least recently + * used one without a run is written (if it changed) and dropped, and loaded again on demand. + * + * GONE. `remove()` (the session's record is deleted) marks the session gone, deletes the file + * and then drops the session's slot: a run in flight stops after its current read and its + * store no longer writes; a request during the removal reads nothing (`unavailable`, reason + * `removed`). A session that comes back with the same id (reboot restore) starts afresh. * `sweep()` deletes, at startup, the stored history of sessions that no longer exist. A * subagent transcript that vanished loses its cursor (`dropSource`), not its entries. * @@ -33,6 +42,7 @@ * @module web/edit-history-sync */ +import { basename, dirname, join } from 'node:path'; import type { SessionRemote } from '../types/session.js'; import type { EditSourceReadOptions, EditSourceRequest, EditSources, PathProbe } from '../edit-history-local-source.js'; import type { ImageCandidate } from '../edit-history-parse.js'; @@ -45,6 +55,8 @@ export const IMAGE_MTIME_SLACK_MS = 2000; export const MAX_FORWARD_PASSES = 3; /** Sources read back per "Load older" click, a full budget each. */ export const MAX_OLDER_SOURCES = 3; +/** Stores kept in memory; the least recently used one past this is dropped. */ +export const MAX_LOADED_STORES = 3; /** `reason` when the session has no transcript yet (no prompt yet, or not found). */ export const NO_TRANSCRIPT = 'no-transcript'; /** `reason` once the session's history was deleted. */ @@ -79,6 +91,8 @@ export interface EditHistorySyncDeps { ): Promise; probe(paths: readonly string[]): Promise>; loadStore(sessionId: string): Promise; + /** Delete a session's stored history without loading it. */ + deleteStore(sessionId: string): Promise; /** Delete the stored history of every session `keep` rejects; returns how many. */ sweepStores(keep: (sessionId: string) => boolean): Promise; budget?: number; @@ -151,7 +165,10 @@ interface SessionSlot { } export class EditHistorySync { + /** By session, least recently used first. */ private readonly slots = new Map(); + /** Writes of dropped stores; loading that session again waits for its write. */ + private readonly unloading = new Map>(); private readonly budget: number; constructor(private readonly deps: EditHistorySyncDeps) { @@ -160,15 +177,7 @@ export class EditHistorySync { /** The session's store, loaded from disk on first use (one load for concurrent callers). */ getStore(sessionId: string): Promise { - const slot = this.slot(sessionId); - if (!slot.store) { - const loading = this.deps.loadStore(sessionId); - slot.store = loading; - loading.catch(() => { - if (slot.store === loading) slot.store = undefined; - }); - } - return slot.store; + return this.storeOf(sessionId, this.slot(sessionId)); } getStatus(sessionId: string): EditHistoryStatus { @@ -182,10 +191,10 @@ export class EditHistorySync { async sync(session: EditHistorySession, opts: SyncOptions = {}): Promise { const why = unsupportedReason(session); const slot = this.slot(session.id); - if (slot.gone) return this.getStatus(session.id); + if (slot.gone) return { ...slot.status }; if (why) { slot.status = { state: 'unsupported', reason: why }; - return this.getStatus(session.id); + return { ...slot.status }; } const want: SyncWant = { forward: !opts.older, older: !!opts.older }; if (slot.running) { @@ -196,33 +205,48 @@ export class EditHistorySync { await slot.running; // The queued run starts when the current one ends; wait for it too. if (slot.running) await slot.running; - return this.getStatus(session.id); + return { ...slot.status }; } slot.running = this.runLoop(session, slot, want); try { await slot.running; } finally { slot.running = undefined; + this.unloadIdle(slot); } - return this.getStatus(session.id); + return { ...slot.status }; } /** Write every loaded store's pending changes (server shutdown). */ async flushAll(): Promise { - await Promise.all([...this.slots.values()].map(async (slot) => (await slot.store)?.flush())); + await Promise.all([ + ...[...this.slots.values()].map(async (slot) => (await slot.store?.catch(() => undefined))?.settle()), + ...this.unloading.values(), + ]); } /** - * Delete a session's stored history (its record is gone). The session stays marked, so a - * run in flight or a late request cannot load and write a new store for it. + * Delete a session's stored history (its record is gone). While it runs the session is + * marked, so a run in flight or a request meanwhile cannot load and write a new store for + * it; then its slot goes, and a session that comes back with this id starts afresh. */ async remove(sessionId: string): Promise { const slot = this.slot(sessionId); slot.gone = true; slot.again = undefined; slot.status = { state: 'unavailable', reason: REMOVED }; - const store = await this.getStore(sessionId); - await store.remove(); + try { + const store = await slot.store?.catch(() => undefined); + if (store) { + await store.remove(); + } else { + await this.unloading.get(sessionId); + await this.deps.deleteStore(sessionId); + } + await slot.running; + } finally { + if (this.slots.get(sessionId) === slot) this.slots.delete(sessionId); + } } /** @@ -233,15 +257,53 @@ export class EditHistorySync { return this.deps.sweepStores((id) => this.slots.has(id) || exists(id)); } + /** The session's slot, now the most recently used. */ private slot(sessionId: string): SessionSlot { - let slot = this.slots.get(sessionId); - if (!slot) { - slot = { status: { state: 'ok' } }; - this.slots.set(sessionId, slot); - } + const slot = this.slots.get(sessionId) ?? { status: { state: 'ok' } }; + this.slots.delete(sessionId); + this.slots.set(sessionId, slot); return slot; } + private storeOf(sessionId: string, slot: SessionSlot): Promise { + if (!slot.store) { + const loading = (async () => { + await this.unloading.get(sessionId); + return this.deps.loadStore(sessionId); + })(); + slot.store = loading; + loading.catch(() => { + if (slot.store === loading) slot.store = undefined; + }); + this.unloadIdle(slot); + } + return slot.store; + } + + /** + * Past {@link MAX_LOADED_STORES}, drop the least recently used stores no run uses: each is + * written if it changed, and a later request loads it again (after that write). + */ + private unloadIdle(keep?: SessionSlot): void { + let loaded = 0; + for (const slot of this.slots.values()) if (slot.store) loaded++; + for (const [sessionId, slot] of this.slots) { + if (loaded <= MAX_LOADED_STORES) break; + if (!slot.store || slot.running || slot.gone || slot === keep) continue; + this.slots.delete(sessionId); + loaded--; + const writing: Promise = slot.store + .then((store) => store.settle()) + .catch((err) => { + console.warn(`[EditHistory] Saving the history of ${sessionId} failed:`, err); + }) + .finally(() => { + if (this.unloading.get(sessionId) === writing) this.unloading.delete(sessionId); + }); + this.unloading.set(sessionId, writing); + } + } + private async runLoop(session: EditHistorySession, slot: SessionSlot, want: SyncWant): Promise { let next: SyncWant | undefined = want; while (next && !slot.gone) { @@ -253,9 +315,10 @@ export class EditHistorySync { } private async runOnce(session: EditHistorySession, slot: SessionSlot, older: boolean): Promise { - const store = await this.getStore(session.id); let status: EditHistoryStatus; try { + // This slot's store: a run still in flight after `remove()` cannot reach a later one. + const store = await this.storeOf(session.id, slot); const path = slot.gone ? null : await this.deps.transcriptPath(session); let found: boolean; if (!path) { @@ -289,11 +352,31 @@ export class EditHistorySync { /** * One forward read of all sources. `more` = some file has unread bytes after it, or was cut * short by the shared budget and left for the next pass. A transcript other than the stored - * one (`/clear`, resume) is read like a first read. + * one (`/clear`, resume) is read like a first read, once the stored one is read to its end. */ private async readForward(path: string, store: EditHistoryStore): Promise<{ found: boolean; more: boolean }> { + const own = this.storedTranscript(path, store); + if (own !== path && store.knownSources().length > 0) { + const previous = await this.readFrom(own, store, this.forwardRequests(store)); + if (previous.more) return { found: true, more: true }; + } + return this.readFrom(path, store, own === path ? this.forwardRequests(store) : []); + } + + /** + * The path of the conversation the store holds cursors for: `path` itself, or the previous + * conversation's transcript next to it (`/clear` and resume keep the project directory). + */ + private storedTranscript(path: string, store: EditHistoryStore): string { const known = store.getTranscriptId(); - const requests = known !== null && path.endsWith(`/${known}.jsonl`) ? this.forwardRequests(store) : []; + return known === null || basename(path) === `${known}.jsonl` ? path : join(dirname(path), `${known}.jsonl`); + } + + private async readFrom( + path: string, + store: EditHistoryStore, + requests: EditSourceRequest[] + ): Promise<{ found: boolean; more: boolean }> { const res = await this.deps.readSources(path, requests, this.budget); if (!res) return { found: false, more: false }; this.dropMissing(store, requests, res); @@ -326,6 +409,8 @@ export class EditHistorySync { private async readOlder(path: string, store: EditHistoryStore, slot: SessionSlot): Promise { if (!store.backfillTruncated) return true; + // Older lines belong to the stored conversation, even before a forward read moved on. + path = this.storedTranscript(path, store); const sources = store.sourcesWithOlder().sort(); if (sources.length === 0) return true; // Take turns: start after the last source the previous click read. diff --git a/src/web/routes/edit-history-routes.ts b/src/web/routes/edit-history-routes.ts index 5741a055..9e24e313 100644 --- a/src/web/routes/edit-history-routes.ts +++ b/src/web/routes/edit-history-routes.ts @@ -23,6 +23,7 @@ import { unsupportedReason, type EditHistorySession, type EditHistoryState, + type EditHistoryStatus, type EditHistorySync, } from '../edit-history-sync.js'; @@ -104,10 +105,14 @@ export function registerEditHistoryRoutes(app: FastifyInstance, ctx: SessionPort const { sync: doSync } = req.query as { sync?: string }; const { session, view } = sessionOf(id, req); const why = unsupportedReason(view); - const store = await sync.getStore(session.id); - if (why) return { success: true, data: listOf(store, session.workingDir, { state: 'unsupported', reason: why }) }; - const status = doSync === '1' ? await sync.sync(view) : sync.getStatus(session.id); - return { success: true, data: listOf(store, session.workingDir, status) }; + // The sync starts in the same tick as the session check: a deletion right after it marks + // the running sync gone (EditHistorySync.remove), so nothing writes a deleted session. + const status: EditHistoryStatus = why + ? { state: 'unsupported', reason: why } + : doSync === '1' + ? await sync.sync(view) + : sync.getStatus(session.id); + return { success: true, data: listOf(await sync.getStore(session.id), session.workingDir, status) }; }); app.post('/api/sessions/:id/edits/older', async (req): Promise> => { diff --git a/src/web/server.ts b/src/web/server.ts index 7313c60a..6e964d0b 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -348,6 +348,7 @@ export class WebServer extends EventEmitter { readSources: readClaudeEditSources, probe: probeLocalPaths, loadStore: (sessionId) => EditHistoryStore.load(sessionId), + deleteStore: (sessionId) => EditHistoryStore.delete(sessionId), sweepStores: (keep) => EditHistoryStore.sweep(keep), }); private respawnControllers: Map = new Map(); diff --git a/test/edit-history-parse.test.ts b/test/edit-history-parse.test.ts index f1dede92..49c64cef 100644 --- a/test/edit-history-parse.test.ts +++ b/test/edit-history-parse.test.ts @@ -11,6 +11,7 @@ import { MAX_IMAGE_CANDIDATES, MAX_PENDING_BASH, MAX_PENDING_COMMAND_CHARS, + MAX_PENDING_FIELD_CHARS, parseClaudeEditLines, } from '../src/edit-history-parse.js'; @@ -255,6 +256,17 @@ describe('parseClaudeEditLines: Bash image candidates', () => { expect(pending[0].id).toBe('toolu_5'); }); + it(`keeps no call with an id, cwd or agent id over ${MAX_PENDING_FIELD_CHARS} characters (the store would refuse it on load)`, () => { + const long = 'x'.repeat(MAX_PENDING_FIELD_CHARS + 1); + const lines = [ + bashUseLine(long, start, 'render > a.png'), + bashUseLine('toolu_cwd', start, 'render > a.png', { cwd: '/' + long }), + bashUseLine('toolu_agent', start, 'render > a.png', { agentId: long }), + bashUseLine('toolu_ok', start, 'render > a.png', { cwd: '/' + 'x'.repeat(MAX_PENDING_FIELD_CHARS - 1) }), + ]; + expect(parseClaudeEditLines(lines.join('\n')).pending.map((p) => p.id)).toEqual(['toolu_ok']); + }); + it('keeps only the head of a long command (the pending list is persisted)', () => { const command = 'render > out.png && ' + 'x'.repeat(MAX_PENDING_COMMAND_CHARS * 2); const { pending } = parseClaudeEditLines(bashUseLine('toolu_long', start, command)); diff --git a/test/edit-history-pipeline.test.ts b/test/edit-history-pipeline.test.ts new file mode 100644 index 00000000..5ed0a959 --- /dev/null +++ b/test/edit-history-pipeline.test.ts @@ -0,0 +1,214 @@ +/** + * Edit history end to end: the real reader (edit-history-local-source.ts), sync and store on + * temp transcripts — what the fakes in edit-history-sync.test.ts cannot show (file order, + * a new conversation next to the old one, every line read exactly once). + */ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { appendFileSync, existsSync, mkdirSync, mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { readClaudeEditSources } from '../src/edit-history-local-source.js'; +import { EditHistoryStore } from '../src/edit-history-store.js'; +import { EditHistorySync, MAX_FORWARD_PASSES, type EditHistorySession } from '../src/web/edit-history-sync.js'; + +let root: string; +const opened: EditHistoryStore[] = []; +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'edit-pipeline-')); +}); +afterEach(async () => { + await Promise.all(opened.splice(0).map((s) => s.remove())); + rmSync(root, { recursive: true, force: true }); +}); + +const T0 = Date.parse('2026-10-15T20:00:00.000Z'); + +/** One edit result line of `pad` bytes or more, newline included. */ +function editLine(id: string, i: number, pad = 0, agentId?: string): string { + const rec: Record = { + type: 'user', + timestamp: new Date(T0 + i * 1000).toISOString(), + cwd: '/w', + message: { role: 'user', content: [{ tool_use_id: id, type: 'tool_result', content: 'ok' }] }, + toolUseResult: { + filePath: `/w/${id}.ts`, + oldString: 'a', + newString: 'b', + structuredPatch: [{ oldStart: 1, oldLines: 1, newStart: 1, newLines: 1, lines: ['-a', '+b'] }], + }, + pad: '', + }; + if (agentId) rec.agentId = agentId; + rec.pad = 'p'.repeat(Math.max(0, pad - JSON.stringify(rec).length - 1)); + return JSON.stringify(rec) + '\n'; +} + +function setup(budget?: number, base = root) { + const project = join(base, 'projects', 'p'); + mkdirSync(project, { recursive: true }); + const at = (transcriptId: string) => join(project, `${transcriptId}.jsonl`); + /** The transcript watcher's path (the hooks report it; `/clear` moves it). */ + let current = at('convA'); + const sync = new EditHistorySync({ + transcriptPath: async () => (existsSync(current) ? current : null), + readSources: readClaudeEditSources, + probe: async (paths) => paths.map(() => null), + loadStore: async (id) => { + const store = await EditHistoryStore.load(id, { dir: join(base, 'store'), debounceMs: 10 }); + opened.push(store); + return store; + }, + deleteStore: (id) => EditHistoryStore.delete(id, { dir: join(base, 'store') }), + sweepStores: async () => 0, + budget, + }); + const session: EditHistorySession = { id: 's1', mode: 'claude', remote: undefined, claudeSessionId: 'convA' }; + const ids = async () => + (await sync.getStore('s1')) + .list() + .map((e) => e.id) + .sort(); + return { + project, + at, + sync, + session, + ids, + switchTo: (transcriptId: string) => (current = at(transcriptId)), + }; +} + +describe('a new conversation (/clear, resume)', () => { + it('reads the previous transcript to its end before the new one', async () => { + const { at, sync, session, ids, switchTo } = setup(); + appendFileSync(at('convA'), editLine('a1', 1)); + await sync.sync(session); + // The turn's last edits land, then /clear before the panel's debounced re-read. + appendFileSync(at('convA'), editLine('a2', 2) + editLine('a3', 3)); + switchTo('convB'); + appendFileSync(at('convB'), editLine('b1', 4)); + await sync.sync(session); + expect(await ids()).toEqual(['a1', 'a2', 'a3', 'b1']); + const store = await sync.getStore('s1'); + expect(store.getTranscriptId()).toBe('convB'); + expect(store.knownSources()).toEqual(['main']); + }); + + it('a previous backlog larger than one sync reads on with the next ones; the new transcript follows', async () => { + const { at, sync, session, ids, switchTo } = setup(600); + appendFileSync(at('convA'), editLine('a0', 0)); + await sync.sync(session); + const backlog = Array.from({ length: MAX_FORWARD_PASSES * 3 }, (_, i) => `a${i + 1}`); + appendFileSync(at('convA'), backlog.map((id, i) => editLine(id, i + 1, 500)).join('')); + switchTo('convB'); + appendFileSync(at('convB'), editLine('b1', 99)); + await sync.sync(session); + expect(await ids()).not.toContain('b1'); + for (let i = 0; i < 5; i++) await sync.sync(session); + expect(await ids()).toEqual(['a0', ...backlog, 'b1'].sort()); + }); + + it('subagents of the previous conversation are read to their end too', async () => { + const { project, at, sync, session, ids, switchTo } = setup(); + const sub = join(project, 'convA', 'subagents'); + mkdirSync(sub, { recursive: true }); + appendFileSync(at('convA'), editLine('a1', 1)); + appendFileSync(join(sub, 'agent-x.jsonl'), editLine('x1', 2, 0, 'x')); + await sync.sync(session); + appendFileSync(join(sub, 'agent-x.jsonl'), editLine('x2', 3, 0, 'x')); + switchTo('convB'); + appendFileSync(at('convB'), editLine('b1', 4)); + await sync.sync(session); + expect(await ids()).toEqual(['a1', 'b1', 'x1', 'x2']); + }); + + it('a previous transcript that is gone does not hold up the new one', async () => { + const { at, sync, session, ids, switchTo } = setup(); + appendFileSync(at('convA'), editLine('a1', 1)); + await sync.sync(session); + rmSync(at('convA')); + switchTo('convB'); + appendFileSync(at('convB'), editLine('b1', 2)); + await sync.sync(session); + expect(await ids()).toEqual(['a1', 'b1']); + }); + + it('"Load older" before the forward read moved on reads the stored conversation, not the new file', async () => { + const { at, sync, session, ids, switchTo } = setup(600); + const older = Array.from({ length: 4 }, (_, i) => `a${i}`); + appendFileSync(at('convA'), older.map((id, i) => editLine(id, i, 500)).join('')); + await sync.sync(session); + expect(await ids()).toEqual(['a3']); + switchTo('convB'); + appendFileSync(at('convB'), editLine('b1', 99)); + await sync.sync(session, { older: true }); + expect(await ids()).toEqual(['a2', 'a3']); + }); +}); + +describe('every line exactly once (seeded random transcripts)', () => { + /** Deterministic: a seed fixes budget, line lengths, subagents, partial writes and sync points. */ + async function run(seed: number): Promise { + let state = seed; + const rnd = () => { + state = (state * 1103515245 + 12345) & 0x7fffffff; + return state / 0x7fffffff; + }; + const ri = (a: number, b: number) => a + Math.floor(rnd() * (b - a + 1)); + const budget = ri(400, 1500); + const { project, at, sync, session } = setup(budget, mkdtempSync(join(root, 'run-'))); + const main = at('convA'); + const subDir = join(project, 'convA', 'subagents'); + const expected = new Set(); + const tooLong = new Set(); + const agents: string[] = []; + const tails = new Map(); + let n = 0; + const write = () => { + if (rnd() < 0.15 && agents.length < 6) { + mkdirSync(subDir, { recursive: true }); + agents.push(`a${agents.length}`); + } + const k = ri(0, agents.length); + const agent = k === 0 ? undefined : agents[k - 1]; + const file = agent ? join(subDir, `agent-${agent}.jsonl`) : main; + const id = `e${n++}`; + // Some lines are longer than the budget: skipped, never shown, never blocking. + const line = editLine(id, n, rnd() < 0.12 ? ri(budget + 2, budget * 3) : ri(250, budget - 2), agent); + if (Buffer.byteLength(line) < budget) expected.add(id); + else if (Buffer.byteLength(line) > budget) tooLong.add(id); + // Sometimes half a line: claude is still writing it. + const cut = rnd() < 0.15 ? ri(1, line.length - 1) : line.length; + appendFileSync(file, (tails.get(file) ?? '') + line.slice(0, cut)); + if (cut < line.length) tails.set(file, line.slice(cut)); + else tails.delete(file); + }; + for (let i = ri(0, 20); i > 0; i--) write(); // backlog before the panel opens + if (!existsSync(main)) appendFileSync(main, ''); + for (let r = ri(5, 25); r > 0; r--) { + for (let i = ri(0, 6); i > 0; i--) write(); + await sync.sync(session); + } + for (const [file, rest] of tails) appendFileSync(file, rest); + for (let i = 0; i < 40; i++) await sync.sync(session); + const store = await sync.getStore('s1'); + for (let clicks = 0; store.backfillTruncated && clicks < 400; clicks++) await sync.sync(session, { older: true }); + const got = new Set(store.list().map((e) => e.id)); + const missing = [...expected].filter((id) => !got.has(id)); + const extra = [...got].filter((id) => tooLong.has(id)); + if (missing.length || extra.length || store.backfillTruncated) { + return `seed ${seed} (budget ${budget}): missing ${missing.join(',')} extra ${extra.join(',')}`; + } + return null; + } + + it('reads every line shorter than the budget once, forward and back, and none longer', async () => { + const failures: string[] = []; + for (let seed = 1; seed <= 40; seed++) { + await Promise.all(opened.splice(0).map((s) => s.remove())); + const r = await run(seed); + if (r) failures.push(r); + } + expect(failures).toEqual([]); + }); +}); diff --git a/test/edit-history-store.test.ts b/test/edit-history-store.test.ts index 26b54045..dbe7f1ca 100644 --- a/test/edit-history-store.test.ts +++ b/test/edit-history-store.test.ts @@ -16,6 +16,8 @@ import { MAX_FILE_BYTES, MAX_STORE_BYTES, parseAgentMeta, + READ_BUDGET_BYTES, + TMP_TAG, } from '../src/edit-history-store.js'; import { MAX_HUNK_BYTES } from '../src/edit-history-parse.js'; @@ -77,12 +79,24 @@ function bashResult(id: string, sec: number): string { const jsonl = (lines: string[]) => Buffer.from(lines.map((l) => l + '\n').join('')); -/** Read like the sync will: ask the store for a range, slice it out of `file`, apply. */ -function syncForward(store: EditHistoryStore, file: Buffer, budget?: number, source = MAIN_SOURCE, tid = 'conv-1') { - const range = store.nextRange(source, file.length, tid, budget); - if (!range) return null; - const chunk = file.subarray(range.offset, range.offset + range.len); - return store.apply({ transcriptId: tid, source, startOffset: range.offset, chunk, size: file.length }); +/** + * Read like the reader (edit-history-local-source.ts): on from the cursor of the current + * conversation (clamped to the size), else the file's last `budget` bytes; then apply. + * Null when nothing is new. + */ +function syncForward( + store: EditHistoryStore, + file: Buffer, + budget = READ_BUDGET_BYTES, + source = MAIN_SOURCE, + tid = 'conv-1' +) { + const cursor = tid === store.getTranscriptId() ? store.getCursor(source) : undefined; + const offset = cursor ? Math.min(cursor.offset, file.length) : Math.max(0, file.length - budget); + const len = Math.min(budget, file.length - offset); + if (cursor && offset === cursor.offset && len === 0) return null; + const chunk = file.subarray(offset, offset + len); + return store.apply({ transcriptId: tid, source, startOffset: offset, chunk, size: file.length }); } function syncOlder(store: EditHistoryStore, file: Buffer, budget?: number, source = MAIN_SOURCE, tid = 'conv-1') { @@ -192,10 +206,13 @@ describe('EditHistoryStore: forward reads', () => { expect(store.get('e2')?.path).toBe(path); }); - it('restarts a file that shrank (replaced) from its end', () => { + it('restarts a file that shrank (replaced) from its end, with the next read', () => { const store = new EditHistoryStore('s1', { dir }); syncForward(store, jsonl([editLine('e1', 1), editLine('e2', 2)])); const replaced = jsonl([editLine('n1', 3)]); + // The reader clamps the cursor to the size: an empty chunk at the end, and the cursor goes. + expect(syncForward(store, replaced)?.added).toBe(0); + expect(store.getCursor(MAIN_SOURCE)).toBeUndefined(); expect(syncForward(store, replaced)?.added).toBe(1); expect(store.getCursor(MAIN_SOURCE)?.offset).toBe(replaced.length); }); @@ -214,7 +231,6 @@ describe('EditHistoryStore: conversations and sources', () => { const store = new EditHistoryStore('s1', { dir }); syncForward(store, jsonl([editLine('e1', 1)])); const second = jsonl([editLine('f1', 5)]); - expect(store.nextRange(MAIN_SOURCE, second.length, 'conv-2')).toEqual({ offset: 0, len: second.length }); syncForward(store, second, undefined, MAIN_SOURCE, 'conv-2'); expect(store.getTranscriptId()).toBe('conv-2'); expect(ids(store)).toEqual(['f1', 'e1']); @@ -590,16 +606,52 @@ describe('EditHistoryStore: persistence', () => { 'live.json', 'gone.json', `gone.json.${process.pid + 1}.tmp`, + // An earlier run that had this process's pid (a container's pid 1): stale as well. `live.json.${process.pid}.tmp`, + `live.json.${process.pid}-0a1b2c3d.tmp`, + `live.json.${TMP_TAG}.tmp`, 'notes.txt', ]) { writeFileSync(join(dir, name), '{}'); } expect(await EditHistoryStore.sweep((id) => id === 'live', { dir })).toBe(1); - expect(readdirSync(dir).sort()).toEqual([`live.json.${process.pid}.tmp`, 'live.json', 'notes.txt'].sort()); + expect(readdirSync(dir).sort()).toEqual([`live.json.${TMP_TAG}.tmp`, 'live.json', 'notes.txt'].sort()); expect(await EditHistoryStore.sweep(() => false, { dir: join(dir, 'missing') })).toBe(0); }); + it('takes only a plain transcript id from the file (the sync builds a path from it)', async () => { + const cursors = { main: { offset: 10, floor: 0 } }; + for (const transcriptId of ['../../etc/passwd', 'a/b', '', 7]) { + writeFileSync(join(dir, 's1.json'), JSON.stringify({ version: 1, transcriptId, cursors, entries: [] })); + const store = await EditHistoryStore.load('s1', { dir }); + expect(store.getTranscriptId()).toBeNull(); + expect(store.knownSources()).toEqual([]); // offsets into no known conversation + } + writeFileSync(join(dir, 's1.json'), JSON.stringify({ version: 1, transcriptId: 'conv-1.a_b', cursors })); + const store = await EditHistoryStore.load('s1', { dir }); + expect(store.getTranscriptId()).toBe('conv-1.a_b'); + expect(store.knownSources()).toEqual(['main']); + }); + + it('settle() writes pending changes and nothing for a store without any', async () => { + const store = new EditHistoryStore('s1', { dir, debounceMs: 60_000 }); + syncForward(store, jsonl([editLine('e1', 1)])); + await store.settle(); + expect((await EditHistoryStore.load('s1', { dir })).get('e1')).toBeDefined(); + const loaded = await EditHistoryStore.load('s1', { dir }); + rmSync(join(dir, 's1.json')); + await loaded.settle(); + await new EditHistoryStore('empty', { dir }).settle(); + expect(readdirSync(dir)).toEqual([]); + }); + + it('delete() removes a stored history without parsing it', async () => { + writeFileSync(join(dir, 's1.json'), 'not json'); + await EditHistoryStore.delete('s1', { dir }); + await EditHistoryStore.delete('missing', { dir }); + expect(readdirSync(dir)).toEqual([]); + }); + it('ignores a file of an unknown version', async () => { writeFileSync(join(dir, 'v9.json'), JSON.stringify({ version: 9, entries: [{ id: 'x' }] })); expect((await EditHistoryStore.load('v9', { dir })).list()).toEqual([]); diff --git a/test/edit-history-sync.test.ts b/test/edit-history-sync.test.ts index 5ec69b42..aab8f8c4 100644 --- a/test/edit-history-sync.test.ts +++ b/test/edit-history-sync.test.ts @@ -3,7 +3,7 @@ * "reads only on request" wiring. Reader and probe are fakes; the store is real (temp dir). */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs'; +import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { EditHistoryStore } from '../src/edit-history-store.js'; @@ -16,6 +16,7 @@ import type { import { confirmImages, EditHistorySync, + MAX_LOADED_STORES, MAX_OLDER_SOURCES, NO_TRANSCRIPT, REMOVED, @@ -126,6 +127,8 @@ function setup(over: Partial = {}) { const host = new FakeHost(); const probes = new Map(); const lookups: string[] = []; + /** Session ids in the order their stores were loaded. */ + const loads: string[] = []; const deps: EditHistorySyncDeps = { transcriptPath: async (session) => { lookups.push(session.id); @@ -134,16 +137,18 @@ function setup(over: Partial = {}) { readSources: (path, req, budget, opts) => host.read(path, req, budget, opts), probe: async (paths) => paths.map((p) => probes.get(p) ?? null), loadStore: (id) => EditHistoryStore.load(id, { dir, debounceMs: 0 }), + deleteStore: (id) => EditHistoryStore.delete(id, { dir }), sweepStores: async () => 0, ...over, }; const load = deps.loadStore; deps.loadStore = async (id) => { + loads.push(id); const store = await load(id); opened.push(store); return store; }; - return { host, probes, lookups, deps, sync: new EditHistorySync(deps) }; + return { host, probes, lookups, loads, deps, sync: new EditHistorySync(deps) }; } const local: EditHistorySession = { id: 's1', mode: 'claude', remote: undefined, claudeSessionId: 'conv1' }; @@ -317,28 +322,107 @@ describe('EditHistorySync', () => { expect(() => readFileSync(join(dir, 's1.json'))).toThrow(); }); - it('a deleted session stays deleted: neither a run in flight nor a later request writes it again', async () => { + it('neither a run in flight nor a request during the removal writes the deleted history again', async () => { const { host, sync } = setup(); host.append('main', editLine('t1', T0, '/w/a.ts')); let release!: () => void; const gate = new Promise((r) => (release = r)); + let entered!: () => void; + const reading = new Promise((r) => (entered = r)); const orig = host.read; host.read = async (...a) => { + entered(); await gate; return orig(...a); }; const runs = [sync.sync(local), sync.sync(local)]; // one running, one queued + await reading; const removing = sync.remove('s1'); + // The removal waits for the read under way; a request meanwhile reads nothing. + expect(await sync.sync(local)).toEqual({ state: 'unavailable', reason: REMOVED }); release(); await Promise.all([...runs, removing]); await new Promise((r) => setTimeout(r, 20)); // past the store's debounce expect(readdirSync(dir)).toEqual([]); - const reads = host.calls.length; - expect(await sync.sync(local)).toEqual({ state: 'unavailable', reason: REMOVED }); - await new Promise((r) => setTimeout(r, 20)); - expect(host.calls).toHaveLength(reads); + expect(host.calls).toHaveLength(1); // the read under way; the queued run never read + }); + + it('the same conversation under an unnormalized path reads on from its cursors, nothing else', async () => { + const path = '/p/./conv1.jsonl'; + const { host, sync } = setup({ transcriptPath: async () => path }); + host.append('main', editLine('t1', T0, '/w/a.ts')); + await sync.sync(local); + host.append('main', editLine('t2', T0 + 1, '/w/a.ts')); + await sync.sync(local); + expect(host.paths).toEqual([path, path]); + expect(host.calls[1]).toEqual([ + { relPath: 'main', offset: host.files.get('main')!.length - editLine('t2', T0 + 1, '/w/a.ts').length }, + ]); + }); + + it('a session that comes back with the same id (reboot restore) reads its history afresh', async () => { + const { host, sync } = setup(); + host.append('main', editLine('t1', T0, '/w/a.ts')); + await sync.sync(local); + // Boot: the unpinned record is pruned as stale; then "restore" brings the session back. + await sync.remove('s1'); + expect(await sync.sync(local)).toEqual({ state: 'ok' }); + expect((await sync.getStore('s1')).list().map((e) => e.id)).toEqual(['t1']); + }); + + it('remove deletes a history it has not loaded without loading it', async () => { + const { sync, loads } = setup(); + const other = new EditHistoryStore('s1', { dir }); + other.addImageEntries([{ toolUseId: 'b1', path: '/w/a.png', ts: T0 }]); + await other.flush(); + await sync.remove('s1'); + expect(loads).toEqual([]); + expect(existsSync(join(dir, 's1.json'))).toBe(false); + }); + + it(`keeps at most ${MAX_LOADED_STORES} stores loaded: a dropped one is written and loads again whole`, async () => { + const { host, sync, loads } = setup({ loadStore: (id) => EditHistoryStore.load(id, { dir, debounceMs: 60_000 }) }); + host.append('main', editLine('t1', T0, '/w/a.ts')); + const ids = Array.from({ length: MAX_LOADED_STORES + 2 }, (_, i) => `s${i}`); + for (const id of ids) await sync.sync({ ...local, id }); + expect(loads).toEqual(ids); + // The two least recently used went: written (no debounced save ran), not kept in memory. + await sync.flushAll(); + expect(existsSync(join(dir, 's0.json'))).toBe(true); + expect((await sync.getStore('s0')).list().map((e) => e.id)).toEqual(['t1']); + expect(loads.slice(ids.length)).toEqual(['s0']); + // The most recently used ones stay loaded. + await sync.getStore(ids[ids.length - 1]); + expect(loads.slice(ids.length)).toEqual(['s0']); + }); + + it('a dropped store that never changed writes nothing (no file for a session without edits)', async () => { + const { sync } = setup(); // no transcript: nothing to read + for (let i = 0; i <= MAX_LOADED_STORES; i++) await sync.sync({ ...local, id: `s${i}` }); + await sync.flushAll(); expect(readdirSync(dir)).toEqual([]); - expect((await sync.getStore('s1')).list()).toEqual([]); + }); + + it('a store whose run is in flight is not dropped, and loads once', async () => { + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const { host, sync, loads } = setup(); + host.append('main', editLine('t1', T0, '/w/a.ts')); + const orig = host.read; + let first = true; + host.read = async (...a) => { + if (first) { + first = false; + await gate; + } + return orig(...a); + }; + const slow = sync.sync(local); + for (let i = 1; i <= MAX_LOADED_STORES + 1; i++) await sync.sync({ ...local, id: `s1-${i}` }); + release(); + expect(await slow).toEqual({ state: 'ok' }); + expect((await sync.getStore('s1')).list().map((e) => e.id)).toEqual(['t1']); + expect(loads.filter((id) => id === 's1')).toEqual(['s1']); }); it('concurrent first callers share one store', async () => { @@ -563,9 +647,16 @@ describe('reads only on request (wiring guard)', () => { }); it('no other module holds the reader, and the edit history modules run no timer or watcher', () => { - for (const file of ['src/web/session-listener-wiring.ts', 'src/web/routes/hook-event-routes.ts']) { - expect(read(file)).not.toMatch(/editHistory/i); - } + const root = join(__dirname, '..'); + const holders = readdirSync(join(root, 'src'), { recursive: true, encoding: 'utf-8' }) + .filter((f) => /\.(ts|js)$/.test(f)) + .map((f) => join('src', f).split('\\').join('/')) + .filter((f) => /EditHistorySync|editHistorySync/.test(read(f))); + expect(holders.sort()).toEqual([ + 'src/web/edit-history-sync.ts', + 'src/web/routes/edit-history-routes.ts', + 'src/web/server.ts', + ]); for (const file of [ 'src/web/edit-history-sync.ts', 'src/edit-history-store.ts', diff --git a/test/routes/edit-history-routes.test.ts b/test/routes/edit-history-routes.test.ts index 6823ade0..1106f77c 100644 --- a/test/routes/edit-history-routes.test.ts +++ b/test/routes/edit-history-routes.test.ts @@ -68,6 +68,7 @@ async function setup( opened.push(store); return store; }, + deleteStore: (id) => EditHistoryStore.delete(id, { dir }), sweepStores: async () => 0, ...deps, }); @@ -114,6 +115,18 @@ describe('GET /api/sessions/:id/edits', () => { expect(data.entries[0]).not.toHaveProperty('hunks'); }); + it('starts the sync before it loads the store (a deletion right after the session check stops the sync)', async () => { + const { app, sync } = await setup(); + const calls: string[] = []; + const syncFn = sync.sync.bind(sync); + const getStore = sync.getStore.bind(sync); + vi.spyOn(sync, 'sync').mockImplementation((...a) => (calls.push('sync'), syncFn(...a))); + vi.spyOn(sync, 'getStore').mockImplementation((...a) => (calls.push('getStore'), getStore(...a))); + await app.inject({ method: 'GET', url: `${URL}?sync=1` }); + await app.inject({ method: 'POST', url: `${URL}/older` }); + expect(calls).toEqual(['sync', 'getStore', 'sync', 'getStore']); + }); + it('without sync=1 answers from the store and does not read', async () => { const { app, read } = await setup(); const res = await app.inject({ method: 'GET', url: URL });