diff --git a/CHANGELOG.md b/CHANGELOG.md index dd6b870593..99e874aa0e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,9 @@ between releases do not reach them. Remote-PC runtime bundles are versioned sepa to the start of the message in the order the files were picked, once a `/command` being typed or an IME composition is finished. It used to go where the caret was, so a path that arrived while you typed landed in the middle of your words and moved the caret. +- A Pi conversation whose session file has grown past 64 MiB reads as a conversation again, where + the chat fell back to the terminal's text (`branch_unreadable`): the reader no longer refuses a + long active branch as a whole, and pages it the way it pages a short one, within the same budgets. - One paste, pick or drop attaches up to 20 files instead of silently keeping the first four, and the message box says how many it left out. **Settings → Chat → Files at once** sets the number for the device, from 1 to 50. diff --git a/server/conversation.ts b/server/conversation.ts index d3fa4477a0..1a409fd705 100644 --- a/server/conversation.ts +++ b/server/conversation.ts @@ -48,12 +48,13 @@ import { processStartedAt } from "./process-start.ts"; import { ompHeldTranscript } from "./omp.ts"; import { forgetOpencodeRead, forgetOpencodeState, OPENCODE_IMAGE_REF, opencodeConversation, opencodeDatabasePath, opencodeImage, opencodeReadKey, opencodeSessionId, opencodeToolOutput } from "./opencode.ts"; import { piTranscriptInStore, piTranscriptPath, unwrittenSession } from "./pi.ts"; -import { forgetAllPiIndexes, forgetPiIndex, piAbandonedTurns, piBranchSegments } from "./pi-tree.ts"; import { defaultDevinDbPath, DevinHistoryUnavailable, devinConversation, forgetDevinState } from "./devin.ts"; import { forgetGrokState, grokProjection, grokToolOutput, parseGrokTranscript, GrokUnavailable } from "./grok.ts"; import { resolveGrokStore } from "./grok-store.ts"; import { AgyUnavailable, agyCatalog, agyToolOutput, forgetAgyState, parseAgyTranscript, withAgyHistory, type AgyProjection } from "./agy.ts"; import { resolveAgyStore, sameAgyStore, type AgyStore } from "./agy-store.ts"; +import { withinRetentionBudget } from "./retention-budget.ts"; +import { forgetAllPiIndexes, forgetPiIndex, piAbandonedTurns, piBranchRecords, piBranchSegments } from "./pi-tree.ts"; import { trimOutput } from "./tool-output.ts"; import { parseConversationMetadata } from "./conversation-metadata.ts"; import { forgetHermesTranscriptState, hermesConversationPage, hermesToolOutput, hermesTranscriptForPane } from "./hermes.ts"; @@ -71,6 +72,7 @@ export { isOmoProcess } from "./omo.ts"; * tool-heavy session; the chat asks for the pages before it as the reader scrolls up. */ export const TRANSCRIPT_WINDOW_BYTES = 16 * 1024 * 1024; +const PI_PAGE_CACHE_BYTES = 4 * 1024 * 1024; /** A page never holds more prompts than this (each opens a user + assistant pair). */ const MAX_PAGE_PROMPTS = MAX_TURNS / 2; @@ -446,7 +448,7 @@ export function parseClaudeTranscript(text: string, maxTurns = MAX_TURNS, option const writtenSessions = new Set(); /** Re-parse on file changes, including replacement and same-size rewrites. */ -const cache = new Map(); +const cache = new Map(); export class ConversationUnavailable extends Error { constructor(reason: string) { @@ -783,6 +785,7 @@ interface LiveScan { const liveScans = new Map(); interface SettledTurns { + source: RecognizedConversation["source"]; id: string; /** the page start these turns begin at, and the turn start they end at */ start: number; @@ -847,6 +850,15 @@ function rememberPaneDevin(paneId: string, sessionId: string, cwd: string, dbPat remember(paneReads, paneId, reads, PANE_READS_MAX); } +/** Pi retains at most four small parsed pages per cache; oversized pages remain readable uncached. */ +function rememberPage(map: Map, key: string, value: T, limit: number): void { + map.delete(key); + if (value.source === "pi-transcript" && !withinRetentionBudget(value, PI_PAGE_CACHE_BYTES)) return; + remember(map, key, value, limit); + const piKeys = [...map].filter(([, item]) => item.source === "pi-transcript").map(([key]) => key); + while (piKeys.length > 4) map.delete(piKeys.shift()!); +} + /** The newest page's start and every turn start in it (pageBefore without widening), or null when it starts mid-turn. */ function newestPage(path: string, stream: TranscriptStream, source: StreamSource): { start: number; starts: number[] } | null { const from = Math.max(stream.floor, stream.length - TRANSCRIPT_WINDOW_BYTES); @@ -982,14 +994,15 @@ function liveTurns(path: string, stream: TranscriptStream, source: RecognizedCon const to = Math.min(start, kept.observedEnd); if (to > from) parseTurns(source, path, readStream(stream, from, to).toString("utf8"), inherited); } - settled = { id: stream.id, start, end: start, turns: [], metadata: parseConversationMetadata(`${head}\n`, source), tail: bytesBefore(stream, start), taskTitles: new Map(inherited), noticed: new Set(), inherited, observedEnd: stream.length }; + settled = { source, id: stream.id, start, end: start, turns: [], metadata: parseConversationMetadata(`${head} +`, source), tail: bytesBefore(stream, start), taskTitles: new Map(inherited), noticed: new Set(), inherited, observedEnd: stream.length }; } if (settled.end < last) { const text = readStream(stream, settled.end, last).toString("utf8"); settled = { ...settled, end: last, turns: [...settled.turns, ...parseTurns(source, path, text, settled.taskTitles, settled.noticed)], metadata: parseConversationMetadata(text, source, settled.metadata), tail: bytesBefore(stream, last) }; } settled.observedEnd = stream.length; - remember(settledTurns, key, settled, 8); + rememberPage(settledTurns, key, settled, 8); if (source === "codex-transcript") { const live = codexLiveTurn(path, stream, last, settled.metadata); return { turns: [...settled.turns, ...live.turns], metadata: live.metadata }; @@ -1000,6 +1013,13 @@ function liveTurns(path: string, stream: TranscriptStream, source: RecognizedCon return { turns: [...settled.turns, ...parseTurns(source, path, text, new Map(settled.taskTitles), new Set(settled.noticed))], metadata: parseConversationMetadata(text, source, settled.metadata) }; } +/** Retention diagnostics only: counts/budget adherence, never transcript contents. */ +export function piPageCacheUsage(): { pages: number; settled: number; withinBudget: boolean } { + const pages = [...cache.values()].filter(item => item.source === "pi-transcript"); + const settled = [...settledTurns.values()].filter(item => item.source === "pi-transcript"); + return { pages: pages.length, settled: settled.length, withinBudget: [...pages, ...settled].every(item => withinRetentionBudget(item, PI_PAGE_CACHE_BYTES)) }; +} + /** Forget every scan and parse kept between polls (tests compare against a cold read). */ export function forgetTranscriptState(): void { cache.clear(); @@ -1564,8 +1584,7 @@ function transcriptPageRead(source: StreamSource, path: string, page: Conversati // the abandoned turns live outside the branch's byte ranges, so the page cannot see them: a // straight line answers { count: 0 }, which is most sessions and costs the client nothing const abandoned = source === "pi-transcript" ? piAbandonedTurns(path, stat.size) ?? undefined : undefined; - cache.set(key, { signature, turns, metadata, cursor, abandoned }); - if (cache.size > 32) cache.delete(cache.keys().next().value!); + rememberPage(cache, key, { source, signature, turns, metadata, cursor, abandoned }, 32); return { source, turns, metadata, cursor, abandoned, history_id: stream.id, version }; } @@ -1673,16 +1692,19 @@ function piToolOutput(path: string, ref: string): string | null { let branch: ReturnType; try { size = statSync(path).size; branch = piBranchSegments(path, size); } catch { return null; } if (branch === null) return null; - const fd = openSync(path, "r"); - try { - for (const segment of branch) { - const buffer = Buffer.alloc(segment.end - segment.start); - if (readSync(fd, buffer, 0, buffer.length, segment.start) !== buffer.length) continue; - const output = outputInText("pi-transcript", buffer.toString("utf8"), ref); - if (output !== null) return output; + // only a record that can be a clear or can hold the ref is parsed: an unrelated record (a picture + // the agent read, a long tool result) is skipped on a substring check, as the segment reader did + const needle = JSON.stringify(ref); + let output: string | null = null; + for (const line of piBranchRecords(path, size)) { + if (line.includes("context_clear")) { + let entry: unknown; + try { entry = JSON.parse(line); } catch { continue; } + if (isContextClear(entry, "pi-transcript")) { output = null; continue; } } - } finally { closeSync(fd); } - return null; + if (output === null && line.includes(needle)) output = outputInText("pi-transcript", line, ref); + } + return output; } /** @@ -1698,22 +1720,15 @@ export function piTranscriptImage(path: string, ref: string): { mediaType: strin let branch: ReturnType; try { size = statSync(path).size; branch = piBranchSegments(path, size); } catch { return null; } if (branch === null) return null; - const fd = openSync(path, "r"); - try { - for (const segment of branch) { - const buffer = Buffer.alloc(segment.end - segment.start); - if (readSync(fd, buffer, 0, buffer.length, segment.start) !== buffer.length) continue; - for (const line of buffer.toString("utf8").split("\n")) { - if (!line.includes(callId)) continue; - let entry: unknown; - try { entry = JSON.parse(line); } catch { continue; } - const message = piMessage(entry); - if (message === null) continue; - const image = piImageBlock(message, callId, Number(nth)); - if (image !== null) return { mediaType: image.media_type, bytes: new Uint8Array(Buffer.from(image.data, "base64")) }; - } - } - } finally { closeSync(fd); } + for (const line of piBranchRecords(path, size)) { + if (!line.includes(callId)) continue; + let entry: unknown; + try { entry = JSON.parse(line); } catch { continue; } + const message = piMessage(entry); + if (message === null) continue; + const image = piImageBlock(message, callId, Number(nth)); + if (image !== null) return { mediaType: image.media_type, bytes: new Uint8Array(Buffer.from(image.data, "base64")) }; + } return null; } diff --git a/server/large-history-agents.test.ts b/server/large-history-agents.test.ts new file mode 100644 index 0000000000..4a42a5ba3b --- /dev/null +++ b/server/large-history-agents.test.ts @@ -0,0 +1,58 @@ +import { afterAll, expect, test } from "bun:test"; +import { appendFileSync, mkdirSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { forgetTranscriptState, transcriptPage } from "./conversation.ts"; + +// Every agent's chat keeps reading past 64 MiB of history: pages bound each read, the file's size +// does not. Pi has its own test (pi-large-history.test.ts); this holds Claude and Codex to the same. +const dir = mkdtempSync(join(tmpdir(), "large-history-agents-")); +// the readers keep module-wide caches keyed by path: leave none behind for the files removed here +afterAll(() => { forgetTranscriptState(); rmSync(dir, { recursive: true, force: true }); }); +const pad = "x".repeat(512 * 1024); +const TURNS = 140; +const texts = (page: ReturnType) => page.turns.flatMap((turn) => turn.parts).flatMap((part) => part.kind === "text" ? [part.text] : []); + +function walk(source: "claude-transcript" | "codex-transcript", path: string, home?: string): Set { + expect(statSync(path).size).toBeGreaterThan(64 * 1024 * 1024); + let page = transcriptPage(source, path, {}, home); + expect(page.source).toBe(source); + expect(texts(page).some((text) => text.startsWith(`answer ${TURNS - 1}`))).toBe(true); + const prompts = new Set(); + for (let pages = 0; ; pages++) { + expect(pages).toBeLessThan(200); + for (const text of texts(page)) if (text.startsWith("prompt ")) prompts.add(text); + if (!page.cursor) break; + page = transcriptPage(source, path, { before: page.cursor }, home); + } + return prompts; +} + +test("a Claude transcript past 64 MiB pages back to its first prompt", () => { + const path = join(dir, "claude.jsonl"); + writeFileSync(path, ""); + for (let n = 0; n < TURNS; n++) { + appendFileSync(path, `${JSON.stringify({ type: "user", message: { role: "user", content: `prompt ${n}` } })}\n`); + appendFileSync(path, `${JSON.stringify({ type: "assistant", message: { role: "assistant", content: [{ type: "text", text: `answer ${n} ${pad}` }] } })}\n`); + } + const prompts = walk("claude-transcript", path); + expect(prompts.size).toBe(TURNS); + expect(prompts.has("prompt 0")).toBe(true); +}); + +test("a Codex rollout past 64 MiB pages back to its first prompt", () => { + const home = join(dir, "codex"); + mkdirSync(join(home, "sessions", "2026", "10", "08"), { recursive: true }); + const path = join(home, "sessions", "2026", "10", "08", "rollout-2026-10-08T00-00-00-01a0a337-19e8-7712-92f5-aa0883392afd.jsonl"); + writeFileSync(path, `${JSON.stringify({ type: "session_meta", payload: { source: "cli", thread_source: "user" } })}\n`); + for (let n = 0; n < TURNS; n++) { + appendFileSync(path, [ + { type: "event_msg", timestamp: "2026-10-08T00:00:00.000Z", payload: { type: "task_started" } }, + { type: "response_item", timestamp: "2026-10-08T00:00:00.000Z", payload: { type: "message", role: "user", content: [{ type: "input_text", text: `prompt ${n}` }] } }, + { type: "response_item", timestamp: "2026-10-08T00:00:01.000Z", payload: { type: "message", role: "assistant", content: [{ type: "output_text", text: `answer ${n} ${pad}` }] } }, + ].map((entry) => JSON.stringify(entry)).join("\n") + "\n"); + } + const prompts = walk("codex-transcript", path, home); + expect(prompts.size).toBe(TURNS); + expect(prompts.has("prompt 0")).toBe(true); +}); diff --git a/server/pi-large-history.test.ts b/server/pi-large-history.test.ts new file mode 100644 index 0000000000..b600616a5e --- /dev/null +++ b/server/pi-large-history.test.ts @@ -0,0 +1,104 @@ +import { afterAll, expect, test } from "bun:test"; +import { appendFileSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { forgetAllPiIndexes, piBranchSegments, piEntryIndex, piIndexCacheUsage, PI_INDEX_CACHE_BYTES, MAX_PI_ENTRY_BYTES, MAX_PI_INDEX_ENTRIES } from "./pi-tree.ts"; +import { transcriptPage, transcriptToolOutput, piTranscriptImage, piPageCacheUsage, forgetTranscriptState } from "./conversation.ts"; +import { withinRetentionBudget } from "./retention-budget.ts"; +const dir = mkdtempSync(join(tmpdir(), "pi-bounded-history-")); +// the readers keep module-wide caches keyed by path: leave none behind for the files removed here +afterAll(() => { forgetTranscriptState(); forgetAllPiIndexes(); rmSync(dir, { recursive: true, force: true }); }); +const textOf = (page: ReturnType) => page.turns.flatMap(t => t.parts).flatMap(p => p.kind === "text" ? [p.text] : []); +const message = (id: string, parentId: string | null, role: string, text: string) => ({ id, parentId, type: "message", message: { role, content: [{ type: "text", text }] } }); +test("history beyond 64MiB remains paged and complete, stable on append, with bounded index retention", () => { + const path = join(dir, "large.jsonl"); writeFileSync(path, ""); let parent: string | null = null; + const add = (record: Record) => { appendFileSync(path, JSON.stringify({ parentId: parent, ...record }) + "\n"); parent = record.id as string; }; + for (let i = 0; i < 140; i++) { + add(message("u" + i, parent, "user", "question " + i)); + add({ id: "pad" + i, type: "custom", data: "x".repeat(512 * 1024) }); + add(message("a" + i, parent, "assistant", "answer " + i)); + } + expect(statSync(path).size).toBeGreaterThan(64 * 1024 * 1024); + expect(piBranchSegments(path, statSync(path).size)).not.toBeNull(); + let page = transcriptPage("pi-transcript", path); const history = page.history_id; + expect(textOf(page)).toContain("answer 139"); expect(page.cursor).not.toBeNull(); expect(page.turns.length).toBeLessThan(80); + const all = new Set(); let pages = 0; + for (;;) { + textOf(page).forEach(t => { expect(all.has(t)).toBe(false); all.add(t); }); + if (!page.cursor) break; + expect(++pages).toBeLessThan(100); page = transcriptPage("pi-transcript", path, { before: page.cursor }); + } + expect(all.size).toBe(280); expect(all.has("question 0")).toBe(true); + add(message("u-next", parent, "user", "appended")); + const newest = transcriptPage("pi-transcript", path); expect(newest.history_id).toBe(history); expect(textOf(newest)).toContain("appended"); + forgetTranscriptState(); expect(textOf(transcriptPage("pi-transcript", path))).toEqual(textOf(newest)); + add({ id: "tool", type: "message", message: { role: "toolResult", toolCallId: "fixture-call", toolName: "read", content: [{ type: "text", text: "kept output" }, { type: "image", mimeType: "image/png", data: Buffer.from("fixture-image").toString("base64") }] } }); + expect(transcriptToolOutput("pi-transcript", path, "fixture-call")).toContain("kept output"); + expect(Buffer.from(piTranscriptImage(path, "pi:fixture-call:0")!.bytes).toString()).toBe("fixture-image"); + // Abandoned records remain inaccessible, even with large preceding history. + add(message("other-branch", "u0", "assistant", "new branch")); + expect(transcriptToolOutput("pi-transcript", path, "fixture-call")).toBeNull(); + expect(piTranscriptImage(path, "pi:fixture-call:0")).toBeNull(); + expect(textOf(transcriptPage("pi-transcript", path))).toEqual(["question 0", "new branch"]); + expect(piIndexCacheUsage().estimatedBytes).toBeLessThanOrEqual(PI_INDEX_CACHE_BYTES); +}, 30000); +test("index discards unused compaction summaries and evicts cached files", () => { + for (let i = 0; i < 7; i++) { + const path = join(dir, `cache-${i}.jsonl`); + writeFileSync(path, JSON.stringify({ id: "c", parentId: null, type: "compaction", summary: "x".repeat(1024 * 1024) }) + "\n"); + const index = piEntryIndex(path)!; expect(index.entries[0]!.summary).toBeNull(); expect(index.retainedBytes).toBeLessThan(1000); + } + expect(piIndexCacheUsage().files).toBeLessThanOrEqual(4); + expect(piIndexCacheUsage().estimatedBytes).toBeLessThanOrEqual(PI_INDEX_CACHE_BYTES); +}); +test("oversized individual records and excessive entry counts fail closed without a partial cached graph", () => { + const path = join(dir, "oversized.jsonl"); + writeFileSync(path, JSON.stringify({ id: "large", type: "custom", data: "x".repeat(MAX_PI_ENTRY_BYTES) }) + "\n"); + expect(piEntryIndex(path)).toBeNull(); expect(piBranchSegments(path, statSync(path).size)).toBeNull(); + writeFileSync(path, JSON.stringify(message("good", null, "user", "restored")) + "\n"); expect(piEntryIndex(path)!.entries).toHaveLength(1); + const many = join(dir, "many.jsonl"); writeFileSync(many, Array.from({ length: MAX_PI_INDEX_ENTRIES + 1 }, (_, i) => JSON.stringify({ id: String(i), parentId: i ? String(i - 1) : null })).join("\n") + "\n"); + expect(piEntryIndex(many)).toBeNull(); + writeFileSync(path, JSON.stringify({ id: "bad", type: "x".repeat(65) }) + "\n"); expect(piEntryIndex(path)).toBeNull(); + const summary = "s".repeat(9 * 1024 * 1024); + writeFileSync(path, JSON.stringify({ id: "one", type: "branch_summary", parentId: null, summary }) + "\n"); + expect(piEntryIndex(path)).not.toBeNull(); // valid: large, and still within the metadata budget + appendFileSync(path, JSON.stringify({ id: "two", type: "branch_summary", parentId: "one", summary }) + "\n"); + expect(piEntryIndex(path)).toBeNull(); // retained metadata budget, independent of history byte size +}, 15000); +test("a valid index above 8 MiB of metadata is cached: an unchanged history is not indexed again, an append extends it", () => { + forgetAllPiIndexes(); + const path = join(dir, "many-small.jsonl"); + writeFileSync(path, Array.from({ length: 40_000 }, (_, i) => JSON.stringify(message("m" + i, i ? "m" + (i - 1) : null, "user", "q" + i))).join("\n") + "\n"); + const first = piEntryIndex(path)!; + expect(first.retainedBytes).toBeGreaterThan(8 * 1024 * 1024); + expect(piEntryIndex(path)).toBe(first); // the same index object: nothing was rebuilt + appendFileSync(path, JSON.stringify(message("next", "m39999", "user", "appended")) + "\n"); + const extended = piEntryIndex(path)!; + expect(extended).toBe(first); // extended in place from the scanned offset + expect(extended.entries).toHaveLength(40_001); + expect(piIndexCacheUsage().estimatedBytes).toBeLessThanOrEqual(PI_INDEX_CACHE_BYTES); + // other files read after it push it out once the budget is spent, never the one just read + for (let i = 0; i < 4; i++) { const other = join(dir, `after-${i}.jsonl`); writeFileSync(other, JSON.stringify(message("u", null, "user", "x")) + "\n"); piEntryIndex(other); } + expect(piIndexCacheUsage().files).toBe(4); + expect(piEntryIndex(path)).not.toBe(first); // evicted as least recently read, then rebuilt +}, 30000); +test("parsed page caches admit at most four small pages each and still serve oversized pages uncached", () => { + forgetTranscriptState(); + for (let i = 0; i < 7; i++) { + const path = join(dir, `page-cache-${i}.jsonl`); + writeFileSync(path, JSON.stringify(message("u", null, "user", "hello")) + "\n"); + transcriptPage("pi-transcript", path); + } + expect(piPageCacheUsage()).toEqual({ pages: 4, settled: 4, withinBudget: true }); + forgetTranscriptState(); + const path = join(dir, "large-page.jsonl"), body = "y".repeat(3 * 1024 * 1024); + writeFileSync(path, [message("u", null, "user", body), message("a", "u", "assistant", "ok"), message("u2", "a", "user", "next")].map(r => JSON.stringify(r)).join("\n") + "\n"); + expect(textOf(transcriptPage("pi-transcript", path))).toContain(body); + expect(piPageCacheUsage()).toEqual({ pages: 0, settled: 0, withinBudget: true }); +}); +test("retention estimator bounds large strings, nested structures and Maps without following cycles forever", () => { + expect(withinRetentionBudget({ text: "x".repeat(300) }, 100)).toBe(false); + expect(withinRetentionBudget({ text: "small" }, 1000)).toBe(true); + expect(withinRetentionBudget(new Map([["title", "x".repeat(300)]]), 100)).toBe(false); + const cycle: any = {}; cycle.self = cycle; expect(withinRetentionBudget(cycle, 1000)).toBe(true); +}); diff --git a/server/pi-tree.ts b/server/pi-tree.ts index 8ba425e9e2..e0257d6e3f 100644 --- a/server/pi-tree.ts +++ b/server/pi-tree.ts @@ -9,15 +9,17 @@ */ import { closeSync, openSync, readSync, statSync } from "node:fs"; +/** Bound work/retention, not cumulative history bytes: pages already bound transcript reads. */ +export const MAX_PI_INDEX_ENTRIES = 100_000; +export const MAX_PI_ENTRY_BYTES = 16 * 1024 * 1024; +const MAX_PI_INDEX_METADATA_BYTES = 32 * 1024 * 1024; /** - * A branch longer than this is not returned at all. The caller treats that as a transcript it - * cannot read (`branch_unreadable`) and the chat falls back to the pane's terminal output, so an - * oversized session shows scrollback rather than a projection built by holding that many bytes of - * paths in memory. The largest branch across 147 real session files here is 23.3 MiB (24,451,068 - * bytes), so the cap sits 2.7x above what pi has been seen to write — close enough that a long-lived - * pane could reach it, which is why it degrades to the fallback instead of throwing. + * The index cache keeps any index the builder accepts, so an unchanged history is never indexed + * twice whatever its size, and holds at most this much estimated metadata across at most + * PI_INDEX_CACHE_FILES files: the least recently read goes first, never the one just read. */ -export const MAX_BRANCH_BYTES = 64 * 1024 * 1024; +export const PI_INDEX_CACHE_BYTES = MAX_PI_INDEX_METADATA_BYTES; +const PI_INDEX_CACHE_FILES = 4; /** One range of the transcript file, in the order it should be read. */ export interface TranscriptSegment { start: number; end: number } @@ -26,6 +28,7 @@ interface PiEntry { id: string; parent: string | null; start: number; end: numbe interface PiIndex { /** the file the index was built from (device and inode): another file at the same path starts over */ file: string; + retainedBytes: number; /** the tree's children in append order; entries without an id (the session header) are not nodes */ entries: PiEntry[]; children: Map; @@ -46,7 +49,9 @@ function indexEntry(line: string, start: number, end: number): PiEntry | null { // type and role are taken here because the line is parsed for this anyway: what counts as a // turn is decided later by a reader that must not pay for a second pass over the file const message = entry.message; + if (entry.id.length > 256 || (typeof entry.parentId === "string" && entry.parentId.length > 256)) throw new Error("Pi entry identity exceeds limit"); const role = message !== null && typeof message === "object" && typeof message.role === "string" ? message.role : ""; + if ((typeof entry.type === "string" && entry.type.length > 64) || role.length > 64) throw new Error("Pi entry metadata exceeds limit"); return { id: entry.id, parent: typeof entry.parentId === "string" && entry.parentId.length > 0 ? entry.parentId : null, @@ -54,12 +59,12 @@ function indexEntry(line: string, start: number, end: number): PiEntry | null { end, type: typeof entry.type === "string" ? entry.type : "", role, - summary: typeof entry.summary === "string" ? entry.summary : null, + summary: entry.type === "branch_summary" && typeof entry.summary === "string" ? entry.summary : null, }; } function buildIndex(path: string, size: number, file: string): PiIndex { - const index: PiIndex = { file, entries: [], children: new Map(), scanned: 0, tail: "" }; + const index: PiIndex = { file, retainedBytes: 0, entries: [], children: new Map(), scanned: 0, tail: "" }; extendIndex(index, path, size); return index; } @@ -87,6 +92,7 @@ function extendIndex(index: PiIndex, path: string, size: number): void { const read = chunk.subarray(0, got); if (!read.includes(0x0a)) { open.push(read); + if (carry.length + open.reduce((sum, part) => sum + part.length, 0) > MAX_PI_ENTRY_BYTES) throw new Error("Pi record exceeds read limit"); end += got; continue; } @@ -96,11 +102,13 @@ function extendIndex(index: PiIndex, path: string, size: number): void { const base = end - held - carry.length; // absolute offset of bytes[0] let offset = 0; for (let newline = bytes.indexOf(0x0a); newline !== -1; newline = bytes.indexOf(0x0a, offset)) { + if (newline - offset > MAX_PI_ENTRY_BYTES) throw new Error("Pi record exceeds read limit"); const entry = indexEntry(bytes.subarray(offset, newline).toString("utf8"), base + offset, base + newline + 1); if (entry) addEntry(index, entry); offset = newline + 1; } carry = bytes.subarray(offset); + if (carry.length > MAX_PI_ENTRY_BYTES) throw new Error("Pi record exceeds read limit"); end += got; } index.scanned = end - carry.length - open.reduce((sum, part) => sum + part.length, 0); @@ -119,6 +127,9 @@ function readBefore(path: string, offset: number, bytes: number): string { } function addEntry(index: PiIndex, entry: PiEntry): void { + if (index.entries.length >= MAX_PI_INDEX_ENTRIES) throw new Error("Pi index entry limit reached"); + index.retainedBytes += 192 + 2 * (entry.id.length + (entry.parent?.length ?? 0) + entry.type.length + entry.role.length + (entry.summary?.length ?? 0)); + if (index.retainedBytes > MAX_PI_INDEX_METADATA_BYTES) throw new Error("Pi index metadata budget reached"); index.entries.push(entry); if (entry.parent === null) return; const siblings = index.children.get(entry.parent); @@ -136,14 +147,22 @@ export function piEntryIndex(path: string): PiIndex | null { // a truncation or an in-place replacement can, and only the tail check tells them apart; // a file put in the old one's place (an import, a restore) is another file even where its // tail reads the same - if (cached && cached.file === file && cached.scanned <= size && readBefore(path, cached.scanned, 64) === cached.tail) { - extendIndex(cached, path, size); - return cached; - } - const fresh = buildIndex(path, size, file); - piIndexes.set(path, fresh); - if (piIndexes.size > 32) piIndexes.delete(piIndexes.keys().next().value!); - return fresh; + piIndexes.delete(path); // a failed extension must not leave a partially indexed graph cached + try { + const reusable = cached && cached.file === file && cached.scanned <= size && readBefore(path, cached.scanned, 64) === cached.tail; + const fresh = reusable ? cached : buildIndex(path, size, file); + if (reusable) extendIndex(fresh, path, size); + piIndexes.set(path, fresh); // last in the map's order: the most recently read + let total = 0; + for (const item of piIndexes.values()) total += item.retainedBytes; + for (const [other, item] of piIndexes) { + if (piIndexes.size <= PI_INDEX_CACHE_FILES && total <= PI_INDEX_CACHE_BYTES) break; + if (other === path) break; // the builder never accepts more than the budget, so this one fits alone + piIndexes.delete(other); + total -= item.retainedBytes; + } + return fresh; + } catch { return null; } } /** Drops one file's index: the next read rebuilds it from the file, which is all it ever was. */ @@ -160,9 +179,8 @@ export function forgetAllPiIndexes(): void { * The active branch: the last entry written back to its root, oldest first, as byte * ranges of the file. Adjacent entries merge into one range, so a session no /tree * touched reads as the single prefix it is. An id repeated (never pi, but never - * trusted) keeps its last append; a branch over MAX_BRANCH_BYTES is refused, and the chat falls - * back to terminal output - * of projected. Null when the tree cannot be read, or holds no entry at all. + * trusted) keeps its last append. Only offsets are projected, never a branch-sized buffer. + * Index entry/record budgets still fail closed rather than invent a partial branch. Null when the tree cannot be read, or holds no entry at all. */ export function piBranchSegments(path: string, size: number): TranscriptSegment[] | null { const index = piEntryIndex(path); @@ -175,12 +193,9 @@ export function piBranchSegments(path: string, size: number): TranscriptSegment[ const chain: PiEntry[] = []; const seen = new Set(); let next: PiEntry | undefined = leaf; - let total = 0; while (next !== undefined && !seen.has(next.id)) { seen.add(next.id); chain.push(next); - total += next.end - next.start; - if (total > MAX_BRANCH_BYTES) return null; const parent: PiEntry | undefined = next.parent === null ? undefined : byId.get(next.parent); // a parent that never appears (a foreign id, a torn index) ends the walk short: // what it still resolves to is shown, the unreadable head is not invented @@ -195,6 +210,34 @@ export function piBranchSegments(path: string, size: number): TranscriptSegment[ return segments; } +/** Read auxiliary tool/image records one at a time, never allocating a merged branch range. */ +export function* piBranchRecords(path: string, size: number): Generator { + const branch = piBranchSegments(path, size), index = piEntryIndex(path); + if (!branch || !index || (index.entries.at(-1)?.end ?? 0) > size) return; + let fd: number; + try { fd = openSync(path, "r"); } catch { return; } // moved or removed since it was indexed: nothing to read + try { + for (const segment of branch) { + let low = 0, high = index.entries.length; + while (low < high) { const mid = (low + high) >>> 1; if (index.entries[mid]!.start < segment.start) low = mid + 1; else high = mid; } + for (let at = low; at < index.entries.length; at++) { + const entry = index.entries[at]!; + if (entry.end > segment.end) break; + const length = entry.end - entry.start; + if (length > MAX_PI_ENTRY_BYTES + 1) return; + const bytes = Buffer.alloc(length); + if (readSync(fd, bytes, 0, length, entry.start) !== length) return; + yield bytes.toString("utf8"); + } + } + } finally { closeSync(fd); } +} + +/** Test/diagnostic summary: metadata estimates only; never session contents. */ +export function piIndexCacheUsage(): { files: number; estimatedBytes: number } { + return { files: piIndexes.size, estimatedBytes: [...piIndexes.values()].reduce((sum, item) => sum + item.retainedBytes, 0) }; +} + /** * What the file holds that the chat cannot show: the turns on paths pi walked away from. * diff --git a/server/pi.test.ts b/server/pi.test.ts index f6b550e2fd..9db98d6a7a 100644 --- a/server/pi.test.ts +++ b/server/pi.test.ts @@ -8,9 +8,14 @@ import type { ConversationTurn } from "../shared/protocol.ts"; const root = mkdtempSync(join(tmpdir(), "herdr-pi-store-")); const store = join(root, "sessions"); -const slug = join(store, `--${root.replaceAll("/", "-")}--`); +const slug = join(store, `--${root.replace(/[\\/:]/g, "-")}--`); mkdirSync(slug, { recursive: true }); afterAll(() => rmSync(root, { recursive: true, force: true })); +const canSymlink = (() => { + const probe = join(root, "symlink-capability"); + try { symlinkSync(store, probe, "dir"); rmSync(probe); return true; } + catch (error) { if (process.platform === "win32" && (error as NodeJS.ErrnoException).code === "EPERM") return false; throw error; } +})(); const session = (name: string, body: unknown[] = []) => { const path = join(slug, `${name}.jsonl`); @@ -36,12 +41,14 @@ describe("pi's session store holds the transcript a pane reads", () => { expect(piTranscriptInStore(join(root, "missing.jsonl"), store)).toBeNull(); }); - it("refuses a link out of the store, a directory and a file that is not jsonl", () => { + it.skipIf(!canSymlink)("refuses a link out of the store", () => { const outside = join(root, "target.jsonl"); writeFileSync(outside, "{}\n"); const link = join(slug, "link.jsonl"); symlinkSync(outside, link); expect(piTranscriptInStore(link, store)).toBeNull(); + }); + it("refuses a directory and a file that is not jsonl", () => { mkdirSync(join(slug, "dir.jsonl"), { recursive: true }); expect(piTranscriptInStore(join(slug, "dir.jsonl"), store)).toBeNull(); const other = join(slug, "notes.txt"); @@ -58,21 +65,20 @@ describe("pi's session store holds the transcript a pane reads", () => { expect(unwrittenSession(join(slug, "fresh.txt"), store)).toBeNull(); expect(unwrittenSession("fresh.jsonl", store)).toBeNull(); expect(unwrittenSession(join(root, "no-store", "fresh.jsonl"), join(root, "no-store"))).toBeNull(); + }); + it.skipIf(!canSymlink)("refuses a dangling session link", () => { const dangling = join(slug, "dangling.jsonl"); symlinkSync(join(root, "nowhere.jsonl"), dangling); expect(unwrittenSession(dangling, store)).toBeNull(); }); - it("resolves the directories around an unwritten session before trusting its place", () => { + it.skipIf(!canSymlink)("resolves the directories around an unwritten session before trusting its place", () => { // a link in the store to a directory outside it does not make its children the store's mkdirSync(join(root, "outside-dir"), { recursive: true }); symlinkSync(join(root, "outside-dir"), join(store, "escape")); expect(unwrittenSession(join(store, "escape", "fresh.jsonl"), store)).toBeNull(); symlinkSync(join(root, "nowhere-dir"), join(store, "broken")); expect(unwrittenSession(join(store, "broken", "fresh.jsonl"), store)).toBeNull(); - // a file where a directory should be holds no session - writeFileSync(join(slug, "plain"), ""); - expect(unwrittenSession(join(slug, "plain", "fresh.jsonl"), store)).toBeNull(); // the store reached through a link is still the store const alias = join(root, "store-alias"); symlinkSync(store, alias); @@ -80,6 +86,11 @@ describe("pi's session store holds the transcript a pane reads", () => { expect(unwrittenSession(join(alias, "fresh.jsonl"), store)).toEqual({ id: "fresh", path: join(realpathSync(store), "fresh.jsonl") }); }); + it("a file where a directory should be holds no unwritten session", () => { + writeFileSync(join(slug, "plain"), ""); + expect(unwrittenSession(join(slug, "plain", "fresh.jsonl"), store)).toBeNull(); + }); + it("follows PI_CODING_AGENT_SESSION_DIR the way Codex follows CODEX_HOME", () => { const previous = process.env["PI_CODING_AGENT_SESSION_DIR"]; process.env["PI_CODING_AGENT_SESSION_DIR"] = store; diff --git a/server/retention-budget.ts b/server/retention-budget.ts new file mode 100644 index 0000000000..ba364538cf --- /dev/null +++ b/server/retention-budget.ts @@ -0,0 +1,18 @@ +/** Conservative retained-content estimate, not a claim about engine RSS. Stops at the budget. */ +export function withinRetentionBudget(value: unknown, limit: number): boolean { + let bytes = 0; + const seen = new WeakSet(); + function visit(v: unknown, depth: number): boolean { + if (depth > 100) return false; + if (typeof v === "string") bytes += 2 * v.length; + else if (v && typeof v === "object") { + if (seen.has(v)) return true; + seen.add(v); bytes += 64; + if (bytes > limit) return false; + if (v instanceof Map) { for (const [key, item] of v) if (!visit(key, depth + 1) || !visit(item, depth + 1)) return false; } + else { for (const [key, item] of Object.entries(v)) if (!visit(key, depth + 1) || !visit(item, depth + 1)) return false; } + } else bytes += 8; + return bytes <= limit; + } + return visit(value, 0); +}