diff --git a/src/session/incremental-jsonl.test.ts b/src/session/incremental-jsonl.test.ts index 53e86d3ac..233cb2f11 100644 --- a/src/session/incremental-jsonl.test.ts +++ b/src/session/incremental-jsonl.test.ts @@ -237,6 +237,34 @@ describe("createSegmentedJSONLWriter", () => { }); }); +describe("createSegmentedJSONLWriter stale keepBytes", () => { + test("does not pad null bytes when on-disk file shrank below keepBytes", async () => { + const dir = tempDir(); + const write = createSegmentedJSONLWriter(dir, BASE); + + const a = { id: 1, text: "first-record" }; + const b = { id: 2, text: "second-record" }; + const c = { id: 3, text: "third-record" }; + await write([a, b, c]); + + const full = path.join(dir, BASE); + // Simulate external shrink/compaction that left the in-memory offsets stale: + // file is shorter than the writer's remembered keepBytes for a shared prefix. + const keptOnDisk = fullSnapshot([a]); + fs.writeFileSync(full, keptOnDisk); + expect(fs.statSync(full).size).toBeLessThan(Buffer.byteLength(fullSnapshot([a, b, c]))); + + // Shared prefix [a, b] would compute keepBytes past the shrunken file size. + // Writer must rebuild rather than truncate-extend with null padding. + const d = { id: 4, text: "after-shrink" }; + await write([a, b, d]); + + const onDisk = fs.readFileSync(full); + expect(onDisk.includes(0)).toBe(false); + expect(await combined(dir)).toBe(fullSnapshot([a, b, d])); + }); +}); + describe("segment readers", () => { test("readExtraSegmentTexts returns tail segments in order", async () => { const dir = tempDir(); diff --git a/src/session/incremental-jsonl.ts b/src/session/incremental-jsonl.ts index edce3810b..857d2c8df 100644 --- a/src/session/incremental-jsonl.ts +++ b/src/session/incremental-jsonl.ts @@ -211,12 +211,31 @@ export function createSegmentedJSONLWriter( const full = path.join(dir, name); const truncateInPlace = isFirst && state !== null && entry.keepBytes > 0; if (truncateInPlace) { - const handle = await fs.promises.open(full, "r+"); + // Stale keepBytes (e.g. after external shrink/compaction) can exceed the + // on-disk size. POSIX truncate-past-EOF pads with null bytes, which + // poisons the JSONL and breaks resume with `\u0000` parse errors. + // Never extend via truncate — rewrite the full segment instead. + let existingSize = 0; try { - await handle.truncate(entry.keepBytes); - if (entry.text.length > 0) await handle.write(entry.text, entry.keepBytes); - } finally { - await handle.close(); + existingSize = (await fs.promises.stat(full)).size; + } catch { + existingSize = 0; + } + if (entry.keepBytes > existingSize) { + // Offsets are wrong relative to disk. Rebuild the kept prefix from + // the in-memory records that belong in this segment, then append + // the planned text (the post-prefix lines for this segment). + const keptRecords = records.slice(firstSegStartRecord, prefix); + const fullText = keptRecords.map((r) => lineFor(r)).join("") + entry.text; + await fs.promises.writeFile(full, fullText); + } else { + const handle = await fs.promises.open(full, "r+"); + try { + await handle.truncate(entry.keepBytes); + if (entry.text.length > 0) await handle.write(entry.text, entry.keepBytes); + } finally { + await handle.close(); + } } } else { await fs.promises.writeFile(full, entry.text); diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 195ef965c..4c56a9f1b 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -61,6 +61,104 @@ describe("createOptimizedContextStore load", () => { expect(loaded.turns.map((t) => (t.content[0] as { text: string }).text)).toEqual(["a", "b"]); }); + test("recovers usable turns when turns.jsonl has a mid-file null-byte hole", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + + const head = jsonl([turn("a"), turn("b")]); + const tail = jsonl([turn("c")]); + // Simulate truncate-past-EOF null padding between valid JSONL records. + const poisoned = Buffer.concat([ + Buffer.from(head, "utf8"), + Buffer.alloc(64, 0), + Buffer.from(tail, "utf8"), + ]); + fs.writeFileSync(path.join(dir, TURNS_FILE), poisoned); + + const loaded = await store.load(); + expect(loaded.turns.map((t) => (t.content[0] as { text: string }).text)).toEqual([ + "a", + "b", + "c", + ]); + }); + + test("preserves pendingOperations when turns are poisoned but metadata is valid", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + + const head = jsonl([turn("a"), turn("b")]); + const tail = jsonl([turn("c")]); + const poisoned = Buffer.concat([ + Buffer.from(head, "utf8"), + Buffer.alloc(64, 0), + Buffer.from(tail, "utf8"), + ]); + fs.writeFileSync(path.join(dir, TURNS_FILE), poisoned); + + // Valid non-empty metadata must survive recovery so rehydrateGates can re-arm. + const pendingOp = { + correlationId: "corr-1", + kind: "approval" as const, + registeredAt: 1_700_000_000_000, + gateId: "gate-1", + }; + fs.writeFileSync( + path.join(dir, "metadata.json"), + JSON.stringify({ + pendingOperations: [pendingOp], + tokenUsage: { input: 10, output: 20, cacheRead: 1, cacheWrite: 2, thinking: 3 }, + connectorState: null, + }), + ); + + const loaded = await store.load(); + expect(loaded.turns.map((t) => (t.content[0] as { text: string }).text)).toEqual([ + "a", + "b", + "c", + ]); + expect(loaded.pendingOperations).toEqual([pendingOp]); + expect(loaded.tokenUsage).toEqual({ + input: 10, + output: 20, + cacheRead: 1, + cacheWrite: 2, + thinking: 3, + }); + expect(loaded.connectorState).toBeNull(); + }); + + test("soft-defaults metadata when metadata.json is corrupt but turns load", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + + fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl([turn("kept")])); + // Corrupt metadata alone must not abort resume when turns are fine. + // Base load parses turns first then metadata — if metadata throws, recovery + // path soft-defaults and still returns turns. + fs.writeFileSync(path.join(dir, "metadata.json"), "{not-json\x00"); + + const loaded = await store.load(); + expect(loaded.turns).toHaveLength(1); + expect((loaded.turns[0]!.content[0] as { text: string }).text).toBe("kept"); + expect(loaded.pendingOperations).toEqual([]); + expect(loaded.connectorState).toBeNull(); + }); + + test("unrecoverable turns.jsonl names the file in the error", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + + // Mid-file garbage that is not null padding and not a torn tail — unrecoverable. + fs.writeFileSync( + path.join(dir, TURNS_FILE), + jsonl([turn("a")]) + "THIS IS NOT JSON\n" + jsonl([turn("b")]), + ); + + await expect(store.load()).rejects.toThrow(/turns\.jsonl/); + }); + // Compacted head rewrites segment 0 while a prior multi-segment history's // tails stay on disk. Concatenating them reintroduces tool_call ids that the // compact head already kept — drop the orphan tails so the session can resume. diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index e9c3eae7f..e9db1d7fa 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -2,7 +2,13 @@ import fs from "node:fs"; import path from "node:path"; import { type } from "arktype"; import { createIsogitStore } from "@intx/storage-isogit"; -import { ContentBlock, type ConversationTurn } from "@intx/types/runtime"; +import { + ContentBlock, + type ConnectorThreadState, + type ConversationTurn, + type PendingOperation, + type TokenUsage, +} from "@intx/types/runtime"; import { getLogger } from "@intx/log"; import { createSegmentedJSONLWriter, @@ -68,31 +74,86 @@ function sanitizeCallId(callId: string): string { * Parse conversation turns out of one JSONL segment. A crash can tear the final * line of the active (last) segment mid-write; when `tolerateTornTail` is set a * final line that fails to parse is dropped rather than aborting the resume. + * + * Null bytes (truncate-past-EOF padding from a stale keepBytes write) are stripped + * so a poisoned segment can still yield its usable turns on resume. Errors name + * `fileName` when provided so diagnostics point at the on-disk file, not a bare + * Bun JSON token. */ -function parseSegmentTurns(text: string, tolerateTornTail: boolean): ConversationTurn[] { +function parseSegmentTurns( + text: string, + tolerateTornTail: boolean, + fileName = "turns segment", +): ConversationTurn[] { if (text.length === 0) return []; - const lines = text.split("\n"); + // POSIX truncate past EOF pads with `\0`. Strip them so the rest of the JSONL + // remains parseable instead of dying on Unrecognized token '\u0000'. + const cleaned = text.includes("\0") ? text.replaceAll("\0", "") : text; + if (cleaned.length === 0) return []; + const lines = cleaned.split("\n"); if (lines[lines.length - 1] === "") lines.pop(); const turns: ConversationTurn[] = []; for (let i = 0; i < lines.length; i++) { + const line = lines[i]!; + if (line.length === 0) continue; const isLast = i === lines.length - 1; let raw: unknown; try { - raw = JSON.parse(lines[i]!); + raw = JSON.parse(line); } catch (cause) { if (tolerateTornTail && isLast) break; - throw new Error("turns segment has malformed JSON", { cause }); + throw new Error(`${fileName} has malformed JSON at line ${i + 1}`, { cause }); } const result = ConversationTurnSchema(raw); if (result instanceof type.errors) { - throw new Error(`turns segment has unexpected structure: ${result.summary}`); + throw new Error(`${fileName} has unexpected structure at line ${i + 1}: ${result.summary}`); } turns.push(result); } return turns; } +const EMPTY_TOKEN_USAGE = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, +} as const; + +type SessionMetadata = { + pendingOperations: PendingOperation[]; + tokenUsage: TokenUsage; + connectorState: ConnectorThreadState | null; +}; + +function emptyMetadata(): SessionMetadata { + return { + pendingOperations: [], + tokenUsage: { ...EMPTY_TOKEN_USAGE }, + connectorState: null, + }; +} + +/** + * Prefer real metadata via the base store schema on recovery. Soft-default only + * when metadata.json is missing, corrupt, or otherwise unreadable so poisoned + * turns still resume without wiping pendingOperations / tokenUsage / connectorState. + */ +async function loadMetadataSoft( + loadMetadata: () => Promise, +): Promise { + try { + return await loadMetadata(); + } catch (cause) { + log.warn("metadata.json unreadable during resilient load; using empty defaults", { + cause: cause instanceof Error ? cause.message : String(cause), + }); + return emptyMetadata(); + } +} + // Mirrors assertWellFormedToolSequence without throwing. Used to choose the // longest segment prefix the reactor will accept after a load. Unpaired // trailing tool_calls are allowed; dups and orphan results fail. @@ -333,12 +394,50 @@ export async function createOptimizedContextStore(dir: string): Promise base.loadMetadata()); + return { turns, ...metadata }; + } }, setConnectorState: (state) => base.setConnectorState(state), branch: (name, signal) => base.branch(name, signal),