From 607083474f7b5b609fc950558238775358201198 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 13 Aug 2026 10:53:13 -0700 Subject: [PATCH 1/2] Stop null-padding turns.jsonl and recover poisoned resume loads When keepBytes exceeds on-disk size, rebuild the segment instead of truncate-past-EOF (which pads null bytes). On load, if the base store fails, recover usable turns with null-strip parse and soft-default metadata; unrecoverable errors name the file. --- src/session/incremental-jsonl.test.ts | 28 +++++ src/session/incremental-jsonl.ts | 29 ++++- src/session/optimized-context-store.test.ts | 52 +++++++++ src/session/optimized-context-store.ts | 112 ++++++++++++++++++-- 4 files changed, 206 insertions(+), 15 deletions(-) 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..51860cbbc 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -61,6 +61,58 @@ 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("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..d503c90da 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -68,31 +68,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; + +function emptyMetadata(): { + pendingOperations: never[]; + tokenUsage: typeof EMPTY_TOKEN_USAGE; + connectorState: null; +} { + return { + pendingOperations: [], + tokenUsage: { ...EMPTY_TOKEN_USAGE }, + connectorState: null, + }; +} + +/** + * Soft-default metadata when the recovery path cannot use the base store. + * Corrupt or missing metadata.json must not abort resume of usable turns. + */ +async function loadMetadataSoft(dir: string): Promise> { + const metadataPath = path.join(dir, METADATA_FILE); + try { + if (!(await pathExists(metadataPath))) return emptyMetadata(); + const text = await fs.promises.readFile(metadataPath, "utf-8"); + JSON.parse(text); + // Schema lives in the base store; recovery only needs a safe shell. + return emptyMetadata(); + } 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 +388,49 @@ export async function createOptimizedContextStore(dir: string): Promise base.setConnectorState(state), branch: (name, signal) => base.branch(name, signal), From 7f880254a09a6357d703be28ff1d80d253dda13e Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 17 Aug 2026 10:11:03 -0700 Subject: [PATCH 2/2] Preserve valid metadata when recovering poisoned turns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turn recovery after a hard base.load failure was always soft-emptying metadata even when metadata.json parsed cleanly under the real schema. That dropped pendingOperations, tokenUsage, and connectorState for sessions whose turns.jsonl had null-byte holes but whose gates were still parked — rehydrateGates then found nothing to re-arm and left suspended agents wedged. Prefer base.loadMetadata() on the recovery path so a good metadata file survives; soft-empty only when that load itself fails. Regression covers null-hole turns plus non-empty pendingOperations. --- src/session/optimized-context-store.test.ts | 46 +++++++++++++++++++++ src/session/optimized-context-store.ts | 41 ++++++++++-------- 2 files changed, 70 insertions(+), 17 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 51860cbbc..4c56a9f1b 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -83,6 +83,52 @@ describe("createOptimizedContextStore load", () => { ]); }); + 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); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index d503c90da..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, @@ -116,11 +122,13 @@ const EMPTY_TOKEN_USAGE = { thinking: 0, } as const; -function emptyMetadata(): { - pendingOperations: never[]; - tokenUsage: typeof EMPTY_TOKEN_USAGE; - connectorState: null; -} { +type SessionMetadata = { + pendingOperations: PendingOperation[]; + tokenUsage: TokenUsage; + connectorState: ConnectorThreadState | null; +}; + +function emptyMetadata(): SessionMetadata { return { pendingOperations: [], tokenUsage: { ...EMPTY_TOKEN_USAGE }, @@ -129,17 +137,15 @@ function emptyMetadata(): { } /** - * Soft-default metadata when the recovery path cannot use the base store. - * Corrupt or missing metadata.json must not abort resume of usable turns. + * 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(dir: string): Promise> { - const metadataPath = path.join(dir, METADATA_FILE); +async function loadMetadataSoft( + loadMetadata: () => Promise, +): Promise { try { - if (!(await pathExists(metadataPath))) return emptyMetadata(); - const text = await fs.promises.readFile(metadataPath, "utf-8"); - JSON.parse(text); - // Schema lives in the base store; recovery only needs a safe shell. - return emptyMetadata(); + return await loadMetadata(); } catch (cause) { log.warn("metadata.json unreadable during resilient load; using empty defaults", { cause: cause instanceof Error ? cause.message : String(cause), @@ -391,7 +397,8 @@ export async function createOptimizedContextStore(dir: string): Promise base.loadMetadata()); return { turns, ...metadata }; } },