Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
77 changes: 46 additions & 31 deletions server/conversation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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;
Expand Down Expand Up @@ -446,7 +448,7 @@ export function parseClaudeTranscript(text: string, maxTurns = MAX_TURNS, option
const writtenSessions = new Set<string>();

/** Re-parse on file changes, including replacement and same-size rewrites. */
const cache = new Map<string, { signature: string; turns: ConversationTurn[]; metadata: ConversationMetadata; cursor: string | null; abandoned?: { count: number; branches: number; summary: string | null } }>();
const cache = new Map<string, { source: RecognizedConversation["source"]; signature: string; turns: ConversationTurn[]; metadata: ConversationMetadata; cursor: string | null; abandoned?: { count: number; branches: number; summary: string | null } }>();

export class ConversationUnavailable extends Error {
constructor(reason: string) {
Expand Down Expand Up @@ -783,6 +785,7 @@ interface LiveScan {
const liveScans = new Map<string, LiveScan>();

interface SettledTurns {
source: RecognizedConversation["source"];
id: string;
/** the page start these turns begin at, and the turn start they end at */
start: number;
Expand Down Expand Up @@ -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<T extends { source: RecognizedConversation["source"] }>(map: Map<string, T>, 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);
Expand Down Expand Up @@ -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 };
Expand All @@ -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();
Expand Down Expand Up @@ -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 };
}

Expand Down Expand Up @@ -1673,16 +1692,19 @@ function piToolOutput(path: string, ref: string): string | null {
let branch: ReturnType<typeof piBranchSegments>;
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;
}

/**
Expand All @@ -1698,22 +1720,15 @@ export function piTranscriptImage(path: string, ref: string): { mediaType: strin
let branch: ReturnType<typeof piBranchSegments>;
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;
}

Expand Down
58 changes: 58 additions & 0 deletions server/large-history-agents.test.ts
Original file line number Diff line number Diff line change
@@ -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<typeof transcriptPage>) => 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<string> {
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<string>();
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);
});
Loading
Loading