diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index b5458d327..b31afd8c8 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -1,4 +1,5 @@ import { defined } from "../../testkit/defined.js"; +import { withFileLogSink } from "../../testkit/file-log-sink.js"; import { describe, test, expect } from "bun:test"; import fs from "node:fs"; import os from "node:os"; @@ -13,6 +14,7 @@ import { } from "./optimized-context-store.js"; import { segmentFileName, listSegmentFiles } from "./incremental-jsonl.js"; import { createPruningCompactor } from "./compactor.js"; +import { LOG_NAMESPACE_ROOT } from "../branding.js"; const TURNS_FILE = "turns.jsonl"; @@ -421,6 +423,81 @@ describe("createOptimizedContextStore load", () => { }); }); +describe("createOptimizedContextStore load warnings", () => { + function contextStoreWarnings(logged: string): string[] { + return logged + .split("\n") + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as Record) + .filter( + (record) => + record.level === "warning" && + record.category === `${LOG_NAMESPACE_ROOT}.session.context-store`, + ) + .map((record) => JSON.stringify(record)); + } + + const pendingOp = { + correlationId: "corr-1", + kind: "approval" as const, + registeredAt: 1_700_000_000_000, + gateId: "gate-1", + }; + const usage = { + input: 7, + output: 3, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }; + + test("a cold load of a store written the current way logs no warning", async () => { + const dir = tempDir(); + const writer = await createOptimizedContextStore(dir); + await writer.writeTurns([turn("a"), turn("b")]); + await writer.writeMetadata({ + pendingOperations: [pendingOp], + tokenUsage: usage, + }); + await writer.commit({ message: "checkpoint: tool-execution" }); + + let loaded: Awaited> | undefined; + const logged = await withFileLogSink(async () => { + const resumed = await createOptimizedContextStore(dir); + loaded = await resumed.load(); + }); + + expect(contextStoreWarnings(logged)).toEqual([]); + expect(turnTexts(defined(loaded).turns)).toEqual(["a", "b"]); + expect(defined(loaded).pendingOperations).toEqual([pendingOp]); + expect(defined(loaded).tokenUsage).toEqual(usage); + }); + + test("a malformed metadata.json still warns on load", async () => { + const dir = tempDir(); + const writer = await createOptimizedContextStore(dir); + await writer.writeTurns([turn("a")]); + await writer.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await writer.commit({ message: "checkpoint: tool-execution" }); + fs.writeFileSync( + path.join(dir, "metadata.json"), + JSON.stringify({ pendingOperations: 42 }), + ); + + let loaded: Awaited> | undefined; + const logged = await withFileLogSink(async () => { + const resumed = await createOptimizedContextStore(dir); + loaded = await resumed.load(); + }); + + const warnings = contextStoreWarnings(logged); + expect(warnings).toHaveLength(1); + expect(warnings[0]).toContain("metadata.json has unexpected structure"); + expect(turnTexts(defined(loaded).turns)).toEqual(["a"]); + expect(defined(loaded).pendingOperations).toEqual([]); + }); +}); + describe("loadRecentTurns", () => { test("reads only the tail segments needed to satisfy the window, in order", async () => { const dir = tempDir(); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index d051cacd5..283ddca46 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -510,9 +510,8 @@ export async function loadRecentTurns( } /** - * Resilient parse of `turns.jsonl` alone (extras merged by the caller). - * Mirrors `load()`'s recovery when the isogit base hard-fails, so a torn or - * poisoned base cannot block the healing write. + * Resilient parse of `turns.jsonl` alone (extras merged by the caller), so a + * torn or poisoned base cannot block the healing write. */ async function readBaseTurnsFromDisk(dir: string): Promise { const basePath = path.join(dir, TURNS_FILE); @@ -1358,9 +1357,7 @@ export async function createSessionStores( return count; } - async function readLiveTurnsCached( - signal?: AbortSignal, - ): Promise { + async function readLiveTurnsCached(): Promise { const segments = await turnSegmentStats(); if ( turnsCache !== null && @@ -1375,23 +1372,14 @@ export async function createSessionStores( : sharedSegmentPrefix(turnsCache.segments, segments); // Base segment stat unchanged: its parse is still exactly what disk // holds, so only the newer segments get re-read and re-parsed. - let baseTurns: ConversationTurn[]; - if (reused >= 1 && turnsCache !== null) { - baseTurns = turnsCache.baseTurns; - } else { - try { - baseTurns = (await base.load(signal)).turns; - } catch (cause) { - // A torn or poisoned base must not block the healing write; recover - // the usable turns the same way load() always has. Corrupt - // metadata.json also slides — callers here only need the turns. - log.warn( - "base context store load failed; recovering base segment from disk", - { cause: cause instanceof Error ? cause.message : String(cause) }, - ); - baseTurns = await readBaseTurnsFromDisk(dir); - } - } + // Otherwise parse turns.jsonl here rather than through the isogit base's + // load(): that also validates metadata.json against the vendor shape, + // which this store no longer writes (tokenUsage lives on the sidecar). + // Metadata is read separately by readSessionMetadata. + const baseTurns = + reused >= 1 && turnsCache !== null + ? turnsCache.baseTurns + : await readBaseTurnsFromDisk(dir); const parsedExtras: ConversationTurn[][] = []; for (let index = 1; index < segments.length; index++) { const cached = @@ -1422,10 +1410,8 @@ export async function createSessionStores( return merged; } - async function readLiveTurns( - signal?: AbortSignal, - ): Promise { - return readLiveTurnsCached(signal); + async function readLiveTurns(): Promise { + return readLiveTurnsCached(); } async function resolveFirstWrite( @@ -1546,15 +1532,15 @@ export async function createSessionStores( const store: ContextStore & AuditStore = { // Full-history read: the reactor's initialization contract. Turns come - // through the shared generation cache (strict base parse with resilient - // recovery, resilient extras, orphan-tail heal), so a repeated load within - // one store generation does not re-parse segments whose stat is unchanged. + // through the shared generation cache (resilient base and extras, + // orphan-tail heal), so a repeated load within one store generation does + // not re-parse segments whose stat is unchanged. // Metadata stays a fresh read: corrupt metadata.json soft-defaults so // resume does not die or wipe pending ops. - async load(signal) { + async load() { let turns: ConversationTurn[]; try { - turns = await readLiveTurnsCached(signal); + turns = await readLiveTurnsCached(); } catch (cause) { // Unrecoverable history: rethrow with the file name in the message. throw new Error(