From 2444dabfbf3c1e1fa1fd8135bb7c4f465b6c6243 Mon Sep 17 00:00:00 2001 From: Noobear Date: Thu, 8 Oct 2026 08:17:00 +0900 Subject: [PATCH 1/2] fix(chat): read long Pi histories past 64 MiB; hold every agent's reader to page budgets The active-branch byte cap refused Pi sessions over 64 MiB (branch_unreadable), so the chat fell back to scrollback. Bound index/record/cache work instead and read records one at a time. Claude and Codex already page from the tail; a new test holds all three readers past 64 MiB. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 3 + server/conversation.ts | 74 ++++++++++++++----------- server/large-history-agents.test.ts | 57 +++++++++++++++++++ server/pi-large-history.test.ts | 86 +++++++++++++++++++++++++++++ server/pi-tree.ts | 81 ++++++++++++++++++--------- server/pi.test.ts | 23 ++++++-- server/retention-budget.ts | 18 ++++++ 7 files changed, 279 insertions(+), 63 deletions(-) create mode 100644 server/large-history-agents.test.ts create mode 100644 server/pi-large-history.test.ts create mode 100644 server/retention-budget.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index c368356f5e..8d5225cb97 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,9 @@ between releases do not reach them. Remote-PC runtime bundles are versioned sepa ([#860](https://github.com/devswha/herdr-web-ui/pull/860)) ### Fixed +- 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. - Web UI SSH connections no longer open extra tunnels configured in the account’s SSH config. ([#606](https://github.com/devswha/herdr-web-ui/pull/606) by @sobir-git) - **Send now** on a working Codex pane sends the message at once. Codex kept it until its next diff --git a/server/conversation.ts b/server/conversation.ts index 1abd24fb9d..e8ee8f6c96 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; @@ -435,7 +437,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) { @@ -760,6 +762,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; @@ -824,6 +827,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); @@ -959,14 +971,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 }; @@ -977,6 +990,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(); @@ -1541,8 +1561,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 }; } @@ -1650,16 +1669,14 @@ 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; - } - } finally { closeSync(fd); } - return null; + let output: string | null = null; + for (const line of piBranchRecords(path, size)) { + let entry: unknown; + try { entry = JSON.parse(line); } catch { continue; } + if (isContextClear(entry, "pi-transcript")) output = null; + else if (output === null) output = outputInText("pi-transcript", line, ref); + } + return output; } /** @@ -1675,22 +1692,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..8324cad051 --- /dev/null +++ b/server/large-history-agents.test.ts @@ -0,0 +1,57 @@ +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 { 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-")); +afterAll(() => 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..42f01c1333 --- /dev/null +++ b/server/pi-large-history.test.ts @@ -0,0 +1,86 @@ +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 { 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-")); +afterAll(() => 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, but too large to retain in the index cache + 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("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..cc3f56744e 100644 --- a/server/pi-tree.ts +++ b/server/pi-tree.ts @@ -9,15 +9,12 @@ */ import { closeSync, openSync, readSync, statSync } from "node:fs"; -/** - * 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. - */ -export const MAX_BRANCH_BYTES = 64 * 1024 * 1024; +/** 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; +export const PI_INDEX_CACHE_BYTES = 8 * 1024 * 1024; +const MAX_PI_INDEX_METADATA_BYTES = 32 * 1024 * 1024; +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 +23,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 +44,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 +54,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 +87,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 +97,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 +122,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 +142,16 @@ 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); + if (fresh.retainedBytes <= PI_INDEX_CACHE_BYTES) piIndexes.set(path, fresh); + while (piIndexes.size > PI_INDEX_CACHE_FILES || [...piIndexes.values()].reduce((sum, item) => sum + item.retainedBytes, 0) > PI_INDEX_CACHE_BYTES) + piIndexes.delete(piIndexes.keys().next().value!); + 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 +168,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 +182,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 +199,33 @@ 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; + const fd = openSync(path, "r"); + 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); +} From 8946f0c8e60d4dbf1b8962ed42c9b3a40345cb3c Mon Sep 17 00:00:00 2001 From: Noobear Date: Mon, 12 Oct 2026 18:03:12 +0900 Subject: [PATCH 2/2] fix(chat): keep a valid Pi index cached whatever its size; skip unrelated records before parsing Review follow-up. An index above 8 MiB of estimated metadata was returned but never cached, so an unchanged history was indexed again on every read, even a few MB of small records. The cache now keeps any index the builder accepts (it refuses more than 32 MiB) and holds at most 32 MiB across four files, least recently read out first, never the one just read. The Pi tool-output lookup parses only a record that can be a clear or holds the ref again, as the segment reader did. A branch record read whose file has gone ends the iteration instead of throwing. The two large-history test files drop the module-wide caches they leave. Co-Authored-By: Claude Opus 5.5 --- server/conversation.ts | 13 +++++++++---- server/large-history-agents.test.ts | 5 +++-- server/pi-large-history.test.ts | 24 +++++++++++++++++++++--- server/pi-tree.ts | 22 +++++++++++++++++----- 4 files changed, 50 insertions(+), 14 deletions(-) diff --git a/server/conversation.ts b/server/conversation.ts index e8ee8f6c96..02afd1cd73 100644 --- a/server/conversation.ts +++ b/server/conversation.ts @@ -1669,12 +1669,17 @@ 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; + // 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)) { - let entry: unknown; - try { entry = JSON.parse(line); } catch { continue; } - if (isContextClear(entry, "pi-transcript")) output = null; - else if (output === null) output = outputInText("pi-transcript", line, ref); + if (line.includes("context_clear")) { + let entry: unknown; + try { entry = JSON.parse(line); } catch { continue; } + if (isContextClear(entry, "pi-transcript")) { output = null; continue; } + } + if (output === null && line.includes(needle)) output = outputInText("pi-transcript", line, ref); } return output; } diff --git a/server/large-history-agents.test.ts b/server/large-history-agents.test.ts index 8324cad051..4a42a5ba3b 100644 --- a/server/large-history-agents.test.ts +++ b/server/large-history-agents.test.ts @@ -2,12 +2,13 @@ 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 { transcriptPage } from "./conversation.ts"; +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-")); -afterAll(() => rmSync(dir, { recursive: true, force: true })); +// 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] : []); diff --git a/server/pi-large-history.test.ts b/server/pi-large-history.test.ts index 42f01c1333..b600616a5e 100644 --- a/server/pi-large-history.test.ts +++ b/server/pi-large-history.test.ts @@ -2,11 +2,12 @@ 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 { piBranchSegments, piEntryIndex, piIndexCacheUsage, PI_INDEX_CACHE_BYTES, MAX_PI_ENTRY_BYTES, MAX_PI_INDEX_ENTRIES } from "./pi-tree.ts"; +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-")); -afterAll(() => rmSync(dir, { recursive: true, force: true })); +// 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", () => { @@ -60,10 +61,27 @@ test("oversized individual records and excessive entry counts fail closed withou 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, but too large to retain in the index cache + 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++) { diff --git a/server/pi-tree.ts b/server/pi-tree.ts index cc3f56744e..e0257d6e3f 100644 --- a/server/pi-tree.ts +++ b/server/pi-tree.ts @@ -12,8 +12,13 @@ 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; -export const PI_INDEX_CACHE_BYTES = 8 * 1024 * 1024; const MAX_PI_INDEX_METADATA_BYTES = 32 * 1024 * 1024; +/** + * 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 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. */ @@ -147,9 +152,15 @@ export function piEntryIndex(path: string): PiIndex | null { 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); - if (fresh.retainedBytes <= PI_INDEX_CACHE_BYTES) piIndexes.set(path, fresh); - while (piIndexes.size > PI_INDEX_CACHE_FILES || [...piIndexes.values()].reduce((sum, item) => sum + item.retainedBytes, 0) > PI_INDEX_CACHE_BYTES) - piIndexes.delete(piIndexes.keys().next().value!); + 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; } } @@ -203,7 +214,8 @@ export function piBranchSegments(path: string, size: number): TranscriptSegment[ 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; - const fd = openSync(path, "r"); + 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;