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
28 changes: 28 additions & 0 deletions src/session/incremental-jsonl.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
29 changes: 24 additions & 5 deletions src/session/incremental-jsonl.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
98 changes: 98 additions & 0 deletions src/session/optimized-context-store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
121 changes: 110 additions & 11 deletions src/session/optimized-context-store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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),
Expand Down
Loading