Stop null-padding turns.jsonl and recover poisoned resume loads (#480)

* 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.

* Preserve valid metadata when recovering poisoned turns

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.

Author: TheGreatAxios

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();

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);

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.

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<SessionMetadata>,
+): Promise<SessionMetadata> {
+  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<ContextS
     // the complete turn history is the actual live conversation state, not an
     // optional convenience — callers that only need a recent tail (e.g. TUI
     // resume hydration) should use `loadRecentTurns` instead.
+    //
+    // When the base isogit store hard-fails (e.g. null-padded turns.jsonl from
+    // a stale truncate), recover usable turns via resilient segment parse and
+    // re-read metadata via the base schema (soft-empty only if that fails too)
+    // so resume does not die on a bare Bun JSON token or wipe pending ops.
     async load(signal) {
-      const baseResult = await base.load(signal);
-      const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE);
-      if (extraTexts.length === 0) return baseResult;
-      const turns = await loadTurnsWithoutMalformedToolSequence(baseResult.turns, extraTexts);
-      return { ...baseResult, turns };
+      try {
+        const baseResult = await base.load(signal);
+        const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE);
+        if (extraTexts.length === 0) return baseResult;
+        const turns = await loadTurnsWithoutMalformedToolSequence(baseResult.turns, extraTexts);
+        return { ...baseResult, turns };
+      } catch (cause) {
+        log.warn(
+          "base context store load failed; recovering turns from disk segments",
+          { cause: cause instanceof Error ? cause.message : String(cause) },
+        );
+        let baseTurns: ConversationTurn[];
+        try {
+          // Prefer resilient parse of segment 0 alone so orphan-tail heal still runs.
+          const basePath = path.join(dir, TURNS_FILE);
+          if (await pathExists(basePath)) {
+            const text = await fs.promises.readFile(basePath, "utf-8");
+            baseTurns = parseSegmentTurns(text, false, TURNS_FILE);
+          } else {
+            baseTurns = [];
+          }
+        } catch (parseCause) {
+          // Unrecoverable: rethrow with the file name in the message.
+          throw new Error(
+            `failed to load ${TURNS_FILE}: ${
+              parseCause instanceof Error ? parseCause.message : String(parseCause)
+            }`,
+            { cause: parseCause },
+          );
+        }
+        const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE);
+        const turns =
+          extraTexts.length === 0
+            ? baseTurns
+            : await loadTurnsWithoutMalformedToolSequence(baseTurns, extraTexts);
+        const metadata = await loadMetadataSoft(() => base.loadMetadata());
+        return { turns, ...metadata };
+      }
     },
     setConnectorState: (state) => base.setConnectorState(state),
     branch: (name, signal) => base.branch(name, signal),