From ae259b652c2e1148956b9393cf1d1299b498f20a Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 16 Oct 2026 22:54:56 -0400 Subject: [PATCH 1/6] test(e2e): cover reloading the agent after an approved ask --- e2e/integration-harness.ts | 28 ++- e2e/reactor-approval-reload.test.ts | 280 ++++++++++++++++++++++++++++ 2 files changed, 306 insertions(+), 2 deletions(-) create mode 100644 e2e/reactor-approval-reload.test.ts diff --git a/e2e/integration-harness.ts b/e2e/integration-harness.ts index e87cc9245..6a7bb29fb 100644 --- a/e2e/integration-harness.ts +++ b/e2e/integration-harness.ts @@ -149,6 +149,8 @@ export interface OpenIntegrationSessionOpts { retryPolicy?: RetryPolicy; /** Fields merged over harness.deps, e.g. a real scheduler when retry delays must elapse. */ depsOverrides?: Partial; + /** Reuse an existing session cwd, so the agent loads that session's store. */ + cwd?: string; } export async function openIntegrationSession( @@ -170,7 +172,8 @@ export async function openIntegrationSession( throw new Error("openIntegrationSession requires at least one source"); } const defaultSource = opts.defaultSourceId ?? primarySource.id; - const cwd = mkdtempSync(join(tmpdir(), "corbits-integration-cwd-")); + const cwd = + opts.cwd ?? mkdtempSync(join(tmpdir(), "corbits-integration-cwd-")); const workdir = join(cwd, ".agent-state", "integration-session"); const evidenceArchiveHolder: { current: CompactionArchive | undefined } = { current: undefined, @@ -380,7 +383,7 @@ export async function openIntegrationSession( }; } -export async function closeIntegrationSession( +async function releaseIntegrationAgent( session: IntegrationSession, ): Promise { try { @@ -388,10 +391,31 @@ export async function closeIntegrationSession( } finally { await session.toolset.dispose(); session.harness.dispose(); + } +} + +export async function closeIntegrationSession( + session: IntegrationSession, +): Promise { + try { + await releaseIntegrationAgent(session); + } finally { rmSync(session.cwd, { recursive: true, force: true }); } } +/** + * The TUI reload path: close the agent without a final commit, then build a + * fresh one on the same session store. The new session owns the cwd. + */ +export async function rebuildIntegrationSession( + session: IntegrationSession, + opts: OpenIntegrationSessionOpts, +): Promise { + await releaseIntegrationAgent(session); + return openIntegrationSession({ ...opts, cwd: session.cwd }); +} + export interface TurnResult { events: ReactorEmittedEvent[]; reply: string; diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts new file mode 100644 index 000000000..0ccf547f7 --- /dev/null +++ b/e2e/reactor-approval-reload.test.ts @@ -0,0 +1,280 @@ +/** + * An approved ask whose agent is rebuilt from the same store before the next + * cycle commit (the TUI's deferred reload) must not leave the parked call's + * pending operation behind: the rebuilt reactor would re-arm its gate and, at + * the original deadline, answer an already-answered call with "approval timed + * out". + */ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; + +import type { ReactorEmittedEvent } from "@intx/inference"; +import type { ConversationTurn } from "@intx/types/runtime"; + +import { createPermissionGate } from "../src/permission/gate.js"; +import { createReactorAuthorize } from "../src/permission/reactor-authorize.js"; +import { + createApprovalResume, + resolveParkedCallIdFromStore, +} from "../src/session/approval-resume.js"; +import { + closeIntegrationSession, + openIntegrationSession, + rebuildIntegrationSession, + runUntilDone, + runUntilSuspended, + type IntegrationSession, +} from "./integration-harness.js"; + +// Ask-tier, but answered locally: nothing listens on the discard port. +const CLOSED_PORT_PROBE = "curl -sS --max-time 1 http://127.0.0.1:9/"; + +// Reactor gates run on the real clock (setTimeout + Date.now), not the +// inference harness clock, so the approval deadline is driven here. Only +// timers at least this long are held; everything shorter keeps real timing. +const HELD_TIMER_MIN_MS = 30 * 60_000; + +interface HeldTimer { + dueAt: number; + fire: () => void; + cleared: boolean; +} + +function holdLongTimers() { + const realNow = Date.now.bind(Date); + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + let offset = 0; + const held = new Map(); + const now = spyOn(Date, "now").mockImplementation(() => realNow() + offset); + const set = spyOn(globalThis, "setTimeout").mockImplementation((( + fire: () => void, + delay?: number, + ...args: unknown[] + ) => { + if (delay === undefined || delay < HELD_TIMER_MIN_MS) { + return realSetTimeout(fire, delay, ...args); + } + const handle = {}; + held.set(handle, { dueAt: Date.now() + delay, fire, cleared: false }); + return handle; + }) as typeof setTimeout); + const clear = spyOn(globalThis, "clearTimeout").mockImplementation((( + handle: Parameters[0], + ) => { + const timer = + typeof handle === "object" && handle !== null + ? held.get(handle) + : undefined; + if (timer !== undefined) { + timer.cleared = true; + return; + } + realClearTimeout(handle); + }) as typeof clearTimeout); + return { + /** Move the clock to `at` and fire every held timer due by then. */ + advanceTo(at: number): void { + offset = at - realNow(); + for (const timer of held.values()) { + if (timer.cleared || timer.dueAt > at) continue; + timer.cleared = true; + timer.fire(); + } + }, + restore(): void { + clear.mockRestore(); + set.mockRestore(); + now.mockRestore(); + }, + }; +} + +function deferredApprovalGate() { + let release: ((outcome: { allow: boolean }) => void) | undefined; + return { + asked: () => release !== undefined, + gate: createPermissionGate({ + approvals: [], + interactive: true, + skipPermissions: false, + auto: false, + reactorGated: true, + requestApproval: () => + new Promise((resolve) => { + release = resolve; + }), + }), + approve: () => release?.({ allow: true }), + }; +} + +async function waitFor( + predicate: () => boolean, + what: string, + ms = 5_000, +): Promise { + const deadline = performance.now() + ms; + while (!predicate()) { + if (performance.now() > deadline) throw new Error(`timed out: ${what}`); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +function toolResultsFor( + turns: readonly ConversationTurn[], + callId: string, +): string[] { + return turns + .flatMap((turn) => turn.content) + .flatMap((block) => + block.type === "tool_result" && block.callId === callId + ? [JSON.stringify(block.content)] + : [], + ); +} + +function collect(session: IntegrationSession): ReactorEmittedEvent[] { + const events: ReactorEmittedEvent[] = []; + void (async () => { + for await (const event of session.agent.stream()) events.push(event); + })().catch(() => undefined); + return events; +} + +const cleanups: (() => Promise | void)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +interface Reloaded { + session: IntegrationSession; + /** Events from the rebuilt agent, collected from its start. */ + events: ReactorEmittedEvent[]; + callId: string; + deadline: number; + clock: ReturnType; +} + +/** + * Parks `command` on an ask, approves it, and once `reloadWhen` holds for + * the approved run, rebuilds the agent on the same store. The model turn + * after the approved result stays open, so the reload always lands before + * the reply terminal commits the cycle. + */ +async function approveThenReload( + command: string, + reloadWhen: ( + session: IntegrationSession, + events: ReactorEmittedEvent[], + ) => boolean, +): Promise { + const ctx = deferredApprovalGate(); + const sessionOpts = { + permissionGate: ctx.gate, + authorize: createReactorAuthorize(ctx.gate), + }; + let session = await openIntegrationSession(sessionOpts); + cleanups.push(() => closeIntegrationSession(session)); + + session.harness.scenario.replyOnce("anthropic", { + toolCalls: [{ name: "run_shell", args: { command } }], + }); + session.harness.scenario.stall(); + + const turn = await runUntilSuspended(session, "Probe the local port."); + if (turn.result.type !== "suspended") throw new Error("did not park"); + const { correlationId } = turn.result; + const parked = (await session.storage.load()).pendingOperations.find( + (op) => op.correlationId === correlationId, + ); + const callId = parked?.suspendedCall?.id; + const deadline = parked?.timeoutAt; + if (callId === undefined || deadline === undefined) { + throw new Error("parked call has no persisted call id or deadline"); + } + + const resume = createApprovalResume({ + getAgent: () => session.agent, + gate: ctx.gate, + resolveParkedCallId: (id) => + resolveParkedCallIdFromStore(session.storage, id), + }); + const handling = resume.handle(turn.result); + await waitFor(ctx.asked, "approval prompt"); + ctx.approve(); + expect(await handling).toBe(true); + void session.harness + .run({ wallClockBudgetMs: 30_000 }) + .catch(() => undefined); + const before = session; + await waitFor(() => reloadWhen(before, turn.events), "reload point"); + + const clock = holdLongTimers(); + cleanups.push(() => clock.restore()); + session = await rebuildIntegrationSession(session, sessionOpts); + return { session, events: collect(session), callId, deadline, clock }; +} + +/** Moves past the original approval deadline and lets the reactor react. */ +async function passDeadline(reloaded: Reloaded): Promise { + reloaded.clock.advanceTo(reloaded.deadline + 1_000); + await reloaded.session.harness + .run({ wallClockBudgetMs: 2_000 }) + .catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 50)); +} + +describe("integration: approving an ask, then reloading the agent", () => { + test.serial( + "after the approved result: the rebuilt agent continues, and the old deadline adds no result", + async () => { + const reloaded = await approveThenReload( + CLOSED_PORT_PROBE, + (session) => session.harness.scenario.matchedRequests().length === 2, + ); + const { session, callId } = reloaded; + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + + session.harness.scenario.replyOnce("anthropic", { text: "Port closed." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Port closed."); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + const results = toolResultsFor(await session.agent.history(), callId); + expect(results).toHaveLength(1); + expect(results.join("")).not.toContain("approval timed out"); + }, + 20_000, + ); + + // The reload interrupts the approved call, which the closing reactor + // answers itself. What an interrupted call should report is not pinned + // here; only that the old deadline cannot add a second result. + test.serial( + "during the approved call: the old deadline adds no second result", + async () => { + const reloaded = await approveThenReload( + `sleep 20; ${CLOSED_PORT_PROBE}`, + (_session, events) => events.some((e) => e.type === "tool.start"), + ); + const { session, callId } = reloaded; + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + }, + 20_000, + ); +}); From c3bcb82bb09f0cf32eeb1afb6815565267015426 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 17 Oct 2026 01:21:21 -0400 Subject: [PATCH 2/6] feat(session): skip approval record lines when reading turns A parked call's approval is moving into the turn log as a role-less line next to the call. Teach every reader of the log to pass over that line first, so history, readAt, the recent window, and the subagent trace never see it as a turn or count it as malformed. readAt stops going through the base store's strict parser, which knows turns only. --- src/session/optimized-context-store.ts | 32 +++--- src/session/pending-approval-log.test.ts | 136 +++++++++++++++++++++++ src/session/pending-approval-record.ts | 42 +++++++ src/subagent/trace-reader.test.ts | 22 ++++ src/subagent/trace-reader.ts | 3 +- 5 files changed, 219 insertions(+), 16 deletions(-) create mode 100644 src/session/pending-approval-log.test.ts create mode 100644 src/session/pending-approval-record.ts diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 283ddca46..a378acae2 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -7,14 +7,12 @@ import { type CommitSigner, } from "@intx/storage-isogit/node"; import { - ApprovalSnapshot, ContentBlock, ConnectorThreadState, type AuditStore, TokenUsage, type ConversationTurn, type PendingOperation, - ToolCall, } from "@intx/types/runtime"; import { getLogger } from "@intx/log"; import { @@ -28,6 +26,10 @@ import { LOG_NAMESPACE_ROOT } from "../branding.js"; import { loadOrCreateCommitSigner } from "./commit-signer.js"; import { withResolvedDirLock } from "./session-dir-lock.js"; import { scrubPersistedPendingOperations } from "./pending-snapshot-scrub.js"; +import { + parsePendingApprovalRecord, + StoredPendingOperationSchema, +} from "./pending-approval-record.js"; const TURNS_FILE = "turns.jsonl"; const PROMPT_FILE = "prompt.jsonl"; @@ -218,6 +220,8 @@ function parseSegmentTurns( cause, }); } + // Approval records share the log but are never history. + if (parsePendingApprovalRecord(raw) !== null) continue; const result = ConversationTurnSchema(raw); if (result instanceof type.errors) { if (skipMalformed) { @@ -264,17 +268,6 @@ function emptyMetadata(): SessionMetadata { * Legacy files still carry tokenUsage (the old vendor shape) and read as the * fallback when the sidecar holds nothing. */ -const StoredPendingOperationSchema = type({ - correlationId: "string", - kind: "'approval'", - "expectedFrom?": "string", - registeredAt: "number", - gateId: "string", - "timeoutAt?": "number", - "suspendedCall?": ToolCall, - "approvalSnapshot?": ApprovalSnapshot, -}); - const StoredMetadataSchema = type({ pendingOperations: StoredPendingOperationSchema.array(), "tokenUsage?": TokenUsage, @@ -1572,8 +1565,17 @@ export async function createSessionStores( }, branch: (name, signal) => base.branch(name, signal), log: (limit, signal) => base.log(limit, signal), - async readAt(hash, signal) { - const baseTurns = await base.readAt(hash, signal); + async readAt(hash) { + // The base store's strict parser knows turns only; the log also holds + // approval records. + let baseText: string | null = null; + try { + baseText = await blobTextAtCommit(dir, hash, TURNS_FILE); + } catch (cause) { + if (!isNotFoundError(cause)) throw cause; + } + const baseTurns = + baseText === null ? [] : parseSegmentTurns(baseText, false, TURNS_FILE); const extraNames = await turnSegmentNamesAtCommit(dir, hash); if (extraNames.length === 0) return baseTurns; diff --git a/src/session/pending-approval-log.test.ts b/src/session/pending-approval-log.test.ts new file mode 100644 index 000000000..9cc276346 --- /dev/null +++ b/src/session/pending-approval-log.test.ts @@ -0,0 +1,136 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import type { ConversationTurn, PendingOperation } from "@intx/types/runtime"; + +import { withFileLogSink } from "../../testkit/file-log-sink.js"; +import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import { + createOptimizedContextStore, + loadRecentTurns, +} from "./optimized-context-store.js"; + +const dirs: string[] = []; +afterEach(() => { + for (const dir of dirs.splice(0)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +function tempDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pending-approval-log-")); + dirs.push(dir); + return dir; +} + +const SECRET = "sk-test-0123456789abcdefghij"; + +const user: ConversationTurn = { + role: "user", + content: [{ type: "text", text: "probe" }], + timestamp: 1, +}; + +const assistant: ConversationTurn = { + role: "assistant", + content: [ + { + type: "tool_call", + id: "call-1", + name: "run_shell", + arguments: { api_key: SECRET }, + }, + { + type: "tool_call", + id: "call-2", + name: "read_file", + arguments: { path: "a" }, + }, + ], + timestamp: 2, +}; + +function resultTurn(callId: string, timestamp: number): ConversationTurn { + return { + role: "user", + content: [ + { type: "tool_result", callId, content: [{ type: "text", text: "ok" }] }, + ], + timestamp, + }; +} + +const siblingResult = resultTurn("call-2", 3); + +function approvalFor( + callId: string, + args: Record, +): PendingOperation { + return { + correlationId: `corr-${callId}`, + kind: "approval", + registeredAt: 1_700_000_000_000, + gateId: `pending-corr-${callId}`, + timeoutAt: 1_700_003_600_000, + suspendedCall: { id: callId, name: "run_shell", arguments: args }, + approvalSnapshot: { + name: "run_shell", + description: "runs a shell command", + inputSchema: {}, + arguments: args, + }, + }; +} + +const op = approvalFor("call-1", { api_key: SECRET }); + +function jsonl(records: readonly unknown[]): string { + return records.map((record) => `${JSON.stringify(record)}\n`).join(""); +} + +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)); +} + +describe("approval records in the turn log", () => { + test("load, readAt, and the recent window read them as no turn, without warnings", async () => { + const dir = tempDir(); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + jsonl([user, assistant, { pendingApproval: op }]), + ); + fs.writeFileSync( + path.join(dir, "turns-0001.jsonl"), + jsonl([siblingResult]), + ); + const writer = await createOptimizedContextStore(dir); + const commit = await writer.commit({ message: "hand-written log" }); + + let loaded: ConversationTurn[] = []; + let atCommit: ConversationTurn[] = []; + let recent: ConversationTurn[] = []; + const logged = await withFileLogSink(async () => { + const store = await createOptimizedContextStore(dir); + loaded = (await store.load()).turns; + atCommit = await store.readAt(commit.hash); + recent = (await loadRecentTurns(dir, 10)).turns; + }); + + const history = [user, assistant, siblingResult]; + expect(contextStoreWarnings(logged)).toEqual([]); + expect(loaded).toEqual(history); + expect(atCommit).toEqual(history); + expect(recent).toEqual(history); + }); +}); diff --git a/src/session/pending-approval-record.ts b/src/session/pending-approval-record.ts new file mode 100644 index 000000000..300e8e808 --- /dev/null +++ b/src/session/pending-approval-record.ts @@ -0,0 +1,42 @@ +import { type } from "arktype"; +import { + ApprovalSnapshot, + type PendingOperation, + ToolCall, +} from "@intx/types/runtime"; + +/** + * A pending operation as it rests in session state (metadata.json, or a turn + * log record). Validated on read because both files are external input. + */ +export const StoredPendingOperationSchema = type({ + correlationId: "string", + kind: "'approval'", + "expectedFrom?": "string", + registeredAt: "number", + gateId: "string", + "timeoutAt?": "number", + "suspendedCall?": ToolCall, + "approvalSnapshot?": ApprovalSnapshot, +}); + +/** + * A parked call's approval rests in the turn log as its own line, right after + * the turn that holds the call, so the call's tool_result line is what closes + * it. The line carries no `role`, so no turn parser can mistake it for a turn. + */ +const PendingApprovalRecordSchema = type({ + pendingApproval: StoredPendingOperationSchema.and({ + suspendedCall: ToolCall, + }), + "+": "reject", +}); + +/** The approval a turn log line records, or null when the line is not one. */ +export function parsePendingApprovalRecord( + raw: unknown, +): PendingOperation | null { + const result = PendingApprovalRecordSchema(raw); + if (result instanceof type.errors) return null; + return result.pendingApproval; +} diff --git a/src/subagent/trace-reader.test.ts b/src/subagent/trace-reader.test.ts index 475e514d1..534475a55 100644 --- a/src/subagent/trace-reader.test.ts +++ b/src/subagent/trace-reader.test.ts @@ -169,6 +169,28 @@ describe("readAgentTrace", () => { expect(result.entries).toHaveLength(1); }); + test("a pending approval record line is not a turn and not a warning", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + const call = { id: "call-1", name: "run_shell", arguments: {} }; + writeTurns(childDir, [ + { role: "assistant", content: [{ type: "tool_call", ...call }] }, + { + pendingApproval: { + correlationId: "corr-1", + kind: "approval", + registeredAt: 1, + gateId: "gate-1", + suspendedCall: call, + }, + }, + ]); + + const result = await readAgentTrace(root, "worker-1"); + expect(result.totalTurns).toBe(1); + expect(result.parseWarnings).toBe(0); + }); + test("bounds the entry count to the requested limit and reports omission", async () => { const root = tempDir(); const childDir = path.join(root, "subagents", "worker-1"); diff --git a/src/subagent/trace-reader.ts b/src/subagent/trace-reader.ts index ded9a16bb..30d7697b5 100644 --- a/src/subagent/trace-reader.ts +++ b/src/subagent/trace-reader.ts @@ -17,6 +17,7 @@ import { readdir, realpath } from "node:fs/promises"; import path, { join } from "node:path"; import { listSegmentFiles } from "../session/incremental-jsonl.js"; +import { parsePendingApprovalRecord } from "../session/pending-approval-record.js"; const TURNS_FILE = "turns.jsonl"; @@ -187,7 +188,7 @@ function parseTurnsTolerant(text: string): { try { const raw: unknown = JSON.parse(line); if (isRawTurn(raw)) turns.push(raw); - else warnings += 1; + else if (parsePendingApprovalRecord(raw) === null) warnings += 1; } catch { warnings += 1; } From c285fa69f190c26218a1d48f9d080de491fd2d69 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 17 Oct 2026 01:27:05 -0400 Subject: [PATCH 3/6] fix(session): keep a parked approval in the turn log beside its call An approved call's pending operation was removed only in memory: the tool cycle writes turns, not metadata, so an agent rebuilt from the store before the next commit re-armed the gate and later answered the call a second time with "approval timed out". Store the approval as a record line right after the turn holding the call, written in the same commit, so the call's tool_result closes it with no second write. load() returns only approvals whose call is in history and unanswered, from the log and from a legacy metadata.json, and metadata.json now holds only what the log does not, removed from the commit once empty. --- src/session/optimized-context-store.test.ts | 35 +- src/session/optimized-context-store.ts | 264 ++++++++++--- src/session/pending-approval-log.test.ts | 393 ++++++++++++++++++++ src/session/pending-approval-record.ts | 8 + src/session/pending-snapshot-scrub.test.ts | 28 +- src/session/pending-snapshot-scrub.ts | 5 +- 6 files changed, 667 insertions(+), 66 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index b31afd8c8..9919ace26 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -34,8 +34,11 @@ const EMPTY_CHECKPOINT_METADATA = { tokenUsage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, }; -async function commitEmptyCheckpoint(dir: string): Promise { +// A turn gives the checkpoint content: with no pending operations there is +// no metadata.json, and a checkpoint with nothing changed commits nothing. +async function commitTurnCheckpoint(dir: string): Promise { const store = await createOptimizedContextStore(dir); + await store.writeTurns([turn("checkpoint")]); await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); await store.commit({ message: "checkpoint: tool-execution" }); } @@ -75,12 +78,17 @@ describe("createOptimizedContextStore writeMetadata approval snapshot scrubbing" const dir = tempDir(); const store = await createOptimizedContextStore(dir); const secretArg = { api_key: "sk-test-0123456789abcdefghij" }; + const suspendedCall = { + id: "call-1", + name: "run_shell", + arguments: secretArg, + }; const op = { correlationId: "corr-1", kind: "approval" as const, registeredAt: 1_700_000_000_000, gateId: "gate-1", - suspendedCall: { id: "call-1", name: "run_shell", arguments: secretArg }, + suspendedCall, approvalSnapshot: { name: "run_shell", description: "runs a shell command", @@ -88,6 +96,14 @@ describe("createOptimizedContextStore writeMetadata approval snapshot scrubbing" arguments: secretArg, }, }; + // The parked call is in history before its approval is written. + await store.writeTurns([ + { + role: "assistant", + content: [{ type: "tool_call", ...suspendedCall }], + timestamp: 1, + }, + ]); await store.writeMetadata({ pendingOperations: [op], tokenUsage: { @@ -100,11 +116,14 @@ describe("createOptimizedContextStore writeMetadata approval snapshot scrubbing" }); await store.commit({ message: "checkpoint: pending" }); - const persisted = JSON.parse( - fs.readFileSync(path.join(dir, "metadata.json"), "utf8"), - ) as { pendingOperations: { approvalSnapshot: { arguments: object } }[] }; + const [, recordLine] = fs + .readFileSync(path.join(dir, TURNS_FILE), "utf8") + .split("\n"); + const persisted = JSON.parse(defined(recordLine)) as { + pendingApproval: { approvalSnapshot: { arguments: object } }; + }; const persistedSubset = JSON.stringify( - persisted.pendingOperations[0]?.approvalSnapshot?.arguments, + persisted.pendingApproval.approvalSnapshot.arguments, ); // The approver-facing snapshot is redacted at rest ... expect(persistedSubset).not.toContain("sk-test-0123456789abcdefghij"); @@ -616,7 +635,7 @@ describe("createOptimizedContextStore checkpoint", () => { test("signs checkpoints as the interchange harness author", async () => { const dir = tempDir(); - await commitEmptyCheckpoint(dir); + await commitTurnCheckpoint(dir); const commit = await headCommit(dir); expect(commit.gpgsig).toContain("BEGIN SSH SIGNATURE"); }); @@ -641,7 +660,7 @@ describe("createOptimizedContextStore checkpoint", () => { ); }) as typeof Bun.spawn; try { - await commitEmptyCheckpoint(dir); + await commitTurnCheckpoint(dir); } finally { Bun.spawn = original; } diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index a378acae2..91bf1e9e8 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -28,6 +28,8 @@ import { withResolvedDirLock } from "./session-dir-lock.js"; import { scrubPersistedPendingOperations } from "./pending-snapshot-scrub.js"; import { parsePendingApprovalRecord, + pendingApprovalRecord, + type PendingApprovalRecord, StoredPendingOperationSchema, } from "./pending-approval-record.js"; @@ -182,6 +184,7 @@ function parseSegmentTurns( tolerateTornTail: boolean, fileName = "turns segment", skipMalformed = false, + approvals?: PendingOperation[], ): ConversationTurn[] { if (text.length === 0) return []; // POSIX truncate past EOF pads with `\0`. Strip them so the rest of the JSONL @@ -221,7 +224,11 @@ function parseSegmentTurns( }); } // Approval records share the log but are never history. - if (parsePendingApprovalRecord(raw) !== null) continue; + const approval = parsePendingApprovalRecord(raw); + if (approval !== null) { + approvals?.push(approval); + continue; + } const result = ConversationTurnSchema(raw); if (result instanceof type.errors) { if (skipMalformed) { @@ -502,15 +509,22 @@ export async function loadRecentTurns( return { turns, truncated }; } +interface ParsedLogSegment { + turns: ConversationTurn[]; + approvals: PendingOperation[]; +} + /** * 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 { +async function readBaseLogFromDisk(dir: string): Promise { const basePath = path.join(dir, TURNS_FILE); - if (!(await pathExists(basePath))) return []; + if (!(await pathExists(basePath))) return { turns: [], approvals: [] }; const text = await fs.promises.readFile(basePath, "utf-8"); - return parseSegmentTurns(text, true, TURNS_FILE, true); + const approvals: PendingOperation[] = []; + const turns = parseSegmentTurns(text, true, TURNS_FILE, true, approvals); + return { turns, approvals }; } async function readOwnerMarker( @@ -1112,6 +1126,38 @@ async function managedPathsDiffer( return matrix.some(([, head, workdir]) => head !== workdir); } +/** + * A parked call's approval is open while its call is in history and no + * tool_result answers it; an op with no parked call is an async marker and + * always passes. First occurrence of a correlationId wins, so callers list + * the log before legacy metadata.json. + */ +function openPendingOperations( + turns: readonly ConversationTurn[], + candidates: readonly PendingOperation[], +): PendingOperation[] { + const called = new Set(); + const answered = new Set(); + for (const turn of turns) { + for (const block of turn.content) { + if (block.type === "tool_call") called.add(block.id); + else if (block.type === "tool_result") answered.add(block.callId); + } + } + const seen = new Set(); + const open: PendingOperation[] = []; + for (const op of candidates) { + if (seen.has(op.correlationId)) continue; + seen.add(op.correlationId); + const callId = op.suspendedCall?.id; + if (callId !== undefined && (!called.has(callId) || answered.has(callId))) { + continue; + } + open.push(op); + } + return open; +} + export interface SessionStores { storage: ContextStore; audit: AuditStore; @@ -1204,28 +1250,122 @@ export async function createSessionStores( sidecarUsage = { ...tokenUsage }; } + // A parked call's approval lives in the turn log, on a record line right + // after the turn holding the call; the call's tool_result closes it, so an + // approved call needs no second write to stop it re-arming. Keyed by + // correlationId in log order. Record objects stay stable so the segmented + // writer sees a new one as a prefix extension, and a closed one keeps its + // line: dropping it would turn every later append into a rewrite. + const approvalRecords = new Map(); + // Approvals the log on disk holds; any other parked op rides metadata.json. + const loggedApprovalIds = new Set(); + + function adoptLoggedApprovals(ops: readonly PendingOperation[]): void { + for (const op of ops) { + if (!approvalRecords.has(op.correlationId)) { + approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); + } + loggedApprovalIds.add(op.correlationId); + } + } + + // Located by call id, not turn identity, so a record follows an excerpted + // copy of its turn through a compaction and is dropped with its call. + function withApprovalRecords(turns: readonly ConversationTurn[]): { + lines: readonly unknown[]; + included: string[]; + } { + if (approvalRecords.size === 0) return { lines: turns, included: [] }; + const byCallId = new Map(); + for (const [correlationId, record] of approvalRecords) { + const callId = record.pendingApproval.suspendedCall?.id; + if (callId === undefined) continue; + const ids = byCallId.get(callId); + if (ids === undefined) byCallId.set(callId, [correlationId]); + else ids.push(correlationId); + } + const lines: unknown[] = []; + const included: string[] = []; + for (const turn of turns) { + lines.push(turn); + for (const block of turn.content) { + if (block.type !== "tool_call") continue; + const ids = byCallId.get(block.id); + if (ids === undefined) continue; + // A call id seen twice is malformed history; record it once. + byCallId.delete(block.id); + for (const id of ids) { + const record = approvalRecords.get(id); + if (record === undefined) continue; + lines.push(record); + included.push(id); + } + } + } + return { lines, included }; + } + + async function writeTurnLog( + turns: readonly ConversationTurn[], + ): Promise { + const { lines, included } = withApprovalRecords(turns); + await writeSegmented(writeTurnsSegmented, lines); + for (const id of included) loggedApprovalIds.add(id); + } + async function writeMetadataLive(metadata: { pendingOperations: PendingOperation[]; tokenUsage: TokenUsage; }): Promise { await appendUsageSidecar(metadata.tokenUsage); - // Approval snapshot args persist into metadata.json in the clear; scrub - // them at this app-owned write (suspendedCall stays raw for the approved - // call's re-dispatch). + // Approval snapshot args would persist in the clear; scrub them at this + // app-owned write (suspendedCall stays raw for the approved call's + // re-dispatch). + let added = false; + for (const op of scrubPersistedPendingOperations( + metadata.pendingOperations.filter( + (pending) => + pending.suspendedCall !== undefined && + !approvalRecords.has(pending.correlationId), + ), + )) { + approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); + added = true; + } + // The reactor writes the parked call's turn just before this in the same + // commit cycle, so the record lands in the same commit as its call. + if (added && unpublishedRewrite === null && liveTurnRefs !== null) { + await writeTurnLog(liveTurnRefs); + invalidateTurnsCache(); + } + // Only what the log does not hold yet stays here: async markers, and an + // approval waiting on a staged rewrite or this process's first turn + // write. Legacy files are still read, so dropping a logged op is safe. + const sideOps = metadata.pendingOperations.filter( + (pending) => + pending.suspendedCall === undefined || + !loggedApprovalIds.has(pending.correlationId), + ); + if (!metadataJsonSeeded) { + writtenMetadataJson = await readTextOrNull(path.join(dir, METADATA_FILE)); + metadataJsonSeeded = true; + } + if (sideOps.length === 0 && connectorStateBuffer === null) { + if (writtenMetadataJson !== null) { + await fs.promises.rm(path.join(dir, METADATA_FILE), { force: true }); + pendingSegmentPaths.add(METADATA_FILE); + writtenMetadataJson = null; + } + return; + } const desired = JSON.stringify( { - pendingOperations: scrubPersistedPendingOperations( - metadata.pendingOperations, - ), + pendingOperations: scrubPersistedPendingOperations(sideOps), connectorState: connectorStateBuffer, }, null, 2, ); - if (!metadataJsonSeeded) { - writtenMetadataJson = await readTextOrNull(path.join(dir, METADATA_FILE)); - metadataJsonSeeded = true; - } if (writtenMetadataJson !== desired) { await publishMetadataDocument(dir, desired); writtenMetadataJson = desired; @@ -1272,9 +1412,9 @@ export async function createSessionStores( async function writeSegmented( writer: ReturnType, - turns: readonly ConversationTurn[], + lines: readonly unknown[], ): Promise { - const { modifiedPaths } = await writer(turns); + const { modifiedPaths } = await writer(lines); for (const filepath of modifiedPaths) pendingSegmentPaths.add(filepath); } @@ -1297,9 +1437,9 @@ export async function createSessionStores( type TurnsGenerationCache = { segments: SegmentFingerprint[]; - baseTurns: ConversationTurn[]; - parsedExtras: ConversationTurn[][]; - merged: ConversationTurn[]; + base: ParsedLogSegment; + parsedExtras: ParsedLogSegment[]; + merged: ParsedLogSegment; }; let turnsCache: TurnsGenerationCache | null = null; @@ -1350,13 +1490,14 @@ export async function createSessionStores( return count; } - async function readLiveTurnsCached(): Promise { + async function readLiveLogCached(): Promise { const segments = await turnSegmentStats(); if ( turnsCache !== null && turnsCache.segments.length === segments.length && sharedSegmentPrefix(turnsCache.segments, segments) === segments.length ) { + adoptLoggedApprovals(turnsCache.merged.approvals); return turnsCache.merged; } const reused = @@ -1369,11 +1510,11 @@ export async function createSessionStores( // 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 = + const baseLog = reused >= 1 && turnsCache !== null - ? turnsCache.baseTurns - : await readBaseTurnsFromDisk(dir); - const parsedExtras: ConversationTurn[][] = []; + ? turnsCache.base + : await readBaseLogFromDisk(dir); + const parsedExtras: ParsedLogSegment[] = []; for (let index = 1; index < segments.length; index++) { const cached = index < reused && turnsCache !== null @@ -1389,22 +1530,24 @@ export async function createSessionStores( path.join(dir, segment.name), "utf-8", ); - parsedExtras.push( - parseSegmentTurns( - text, - index === segments.length - 1, - segment.name, - true, - ), + const approvals: PendingOperation[] = []; + const turns = parseSegmentTurns( + text, + index === segments.length - 1, + segment.name, + true, + approvals, ); + parsedExtras.push({ turns, approvals }); } - const merged = await mergedLiveTurns(baseTurns, parsedExtras); - turnsCache = { segments, baseTurns, parsedExtras, merged }; + const merged = await mergedLiveLog(baseLog, parsedExtras); + turnsCache = { segments, base: baseLog, parsedExtras, merged }; + adoptLoggedApprovals(merged.approvals); return merged; } async function readLiveTurns(): Promise { - return readLiveTurnsCached(); + return (await readLiveLogCached()).turns; } async function resolveFirstWrite( @@ -1459,7 +1602,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeSegmented(writeTurnsSegmented, turns); + await writeTurnLog(turns); liveTurnRefs = [...turns]; invalidateTurnsCache(); return; @@ -1474,7 +1617,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeSegmented(writeTurnsSegmented, turns); + await writeTurnLog(turns); liveTurnRefs = [...turns]; invalidateTurnsCache(); return; @@ -1489,7 +1632,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeSegmented(writeTurnsSegmented, turns); + await writeTurnLog(turns); liveTurnRefs = [...turns]; invalidateTurnsCache(); }); @@ -1497,13 +1640,16 @@ export async function createSessionStores( // Prefer the longest loadable prefix of base + extras; orphan tails from a // fresh-writer compaction rewrite are dropped and unlinked so the next load - // does not re-poison the session. - async function mergedLiveTurns( - baseTurns: ConversationTurn[], - parsedExtras: ConversationTurn[][], - ): Promise { - if (parsedExtras.length === 0) return baseTurns; - const keepExtras = longestWellFormedExtraCount(baseTurns, parsedExtras); + // does not re-poison the session. A dropped segment's records go with it. + async function mergedLiveLog( + baseLog: ParsedLogSegment, + parsedExtras: ParsedLogSegment[], + ): Promise { + if (parsedExtras.length === 0) return baseLog; + const keepExtras = longestWellFormedExtraCount( + baseLog.turns, + parsedExtras.map((extra) => extra.turns), + ); if (keepExtras < parsedExtras.length) { // Segment file index for the first extra is 1. @@ -1519,8 +1665,11 @@ export async function createSessionStores( invalidateTurnsCache(); } - if (keepExtras === 0) return baseTurns; - return [...baseTurns, ...parsedExtras.slice(0, keepExtras).flat()]; + const kept = [baseLog, ...parsedExtras.slice(0, keepExtras)]; + return { + turns: kept.flatMap((segment) => segment.turns), + approvals: kept.flatMap((segment) => segment.approvals), + }; } const store: ContextStore & AuditStore = { @@ -1531,9 +1680,9 @@ export async function createSessionStores( // Metadata stays a fresh read: corrupt metadata.json soft-defaults so // resume does not die or wipe pending ops. async load() { - let turns: ConversationTurn[]; + let liveLog: ParsedLogSegment; try { - turns = await readLiveTurnsCached(); + liveLog = await readLiveLogCached(); } catch (cause) { // Unrecoverable history: rethrow with the file name in the message. throw new Error( @@ -1555,7 +1704,17 @@ export async function createSessionStores( ); metadata = emptyMetadata(); } - return { turns, ...metadata }; + const { turns } = liveLog; + return { + turns, + ...metadata, + // Legacy sessions park approvals on metadata.json only; the next + // commit moves a still-open one into the log next to its call. + pendingOperations: openPendingOperations(turns, [ + ...liveLog.approvals, + ...metadata.pendingOperations, + ]), + }; }, // Buffer is carried only by this store: base.writeMetadata is never // called through the wrapper, so base's pendingConnectorState would @@ -1673,11 +1832,12 @@ export async function createSessionStores( return headWithinLock; }; const headBefore = stagedRewrite === null ? null : await lockedHead(); + const loggedBefore = new Set(loggedApprovalIds); let extraPaths: string[] = []; try { if (stagedRewrite !== null) { - await writeSegmented(writeTurnsSegmented, stagedRewrite); + await writeTurnLog(stagedRewrite); invalidateTurnsCache(); } const listing = await surveyWorktree(dir); @@ -1828,6 +1988,10 @@ export async function createSessionStores( } catch { // A partial restore must not mask the real commit error. } + // Records only the rewrite carried are gone again; those + // approvals stay on metadata.json until a rewrite lands. + loggedApprovalIds.clear(); + for (const id of loggedBefore) loggedApprovalIds.add(id); } // Drop the writer's stale in-memory state so a retry rewrites from // scratch. diff --git a/src/session/pending-approval-log.test.ts b/src/session/pending-approval-log.test.ts index 9cc276346..c2826db72 100644 --- a/src/session/pending-approval-log.test.ts +++ b/src/session/pending-approval-log.test.ts @@ -2,11 +2,13 @@ import { afterEach, describe, expect, test } from "bun:test"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import git from "isomorphic-git"; import type { ConversationTurn, PendingOperation } from "@intx/types/runtime"; import { withFileLogSink } from "../../testkit/file-log-sink.js"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import { createPruningCompactor } from "./compactor.js"; import { createOptimizedContextStore, loadRecentTurns, @@ -85,6 +87,145 @@ function approvalFor( } const op = approvalFor("call-1", { api_key: SECRET }); +const answered = resultTurn("call-1", 4); + +const USAGE = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }; + +type LogLine = Record; + +function logLines(dir: string): LogLine[] { + return fs + .readFileSync(path.join(dir, "turns.jsonl"), "utf8") + .split("\n") + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as LogLine); +} + +function lineKinds(dir: string): string[] { + return logLines(dir).map((line) => ("role" in line ? "turn" : "record")); +} + +function recordedApprovals(dir: string): PendingOperation[] { + return logLines(dir).flatMap((line) => + "pendingApproval" in line ? [line.pendingApproval as PendingOperation] : [], + ); +} + +function metadataPath(dir: string): string { + return path.join(dir, "metadata.json"); +} + +async function headPaths(dir: string): Promise { + return git.listFiles({ fs, dir, ref: "HEAD" }); +} + +async function openOps(dir: string): Promise { + const loaded = await (await createOptimizedContextStore(dir)).load(); + return loaded.pendingOperations.map((pending) => pending.correlationId); +} + +async function parkCall( + dir: string, + turns: ConversationTurn[], + ops: PendingOperation[], +) { + const store = await createOptimizedContextStore(dir); + await store.writeTurns(turns); + await store.writeMetadata({ pendingOperations: ops, tokenUsage: USAGE }); + const commit = await store.commit({ message: "suspend" }); + return { store, commit }; +} + +// Forces the next commit to fail: every object write hits ENOTDIR. +async function failingCommit(dir: string, run: () => Promise) { + const objectsDir = path.join(dir, ".git", "objects"); + const held = `${objectsDir}.held`; + fs.renameSync(objectsDir, held); + fs.writeFileSync(objectsDir, "held"); + try { + await expect(run()).rejects.toThrow(); + } finally { + fs.unlinkSync(objectsDir); + fs.renameSync(held, objectsDir); + } +} + +function textTurn(text: string): ConversationTurn { + return { role: "user", content: [{ type: "text", text }], timestamp: 1 }; +} + +function readCall(id: string): ConversationTurn { + return { + role: "assistant", + content: [ + { + type: "tool_call", + id, + name: "read_file", + arguments: { path: "src/hot.ts" }, + }, + ], + timestamp: 1, + }; +} + +function readResult(id: string, text: string): ConversationTurn { + return { + role: "user", + content: [ + { type: "tool_result", callId: id, content: [{ type: "text", text }] }, + ], + timestamp: 1, + }; +} + +// Enough tool history for the pruning compactor to fold, ending on a parked +// call it must keep. +function longHistoryEndingParked(): ConversationTurn[] { + return [ + textTurn("start the migration and keep the auth notes"), + textTurn("constraint: do not rewrite the session log"), + readCall("old-a"), + readResult("old-a", `FOLD_${"c".repeat(8000)}`), + readCall("old-b"), + readResult("old-b", `FOLD_${"d".repeat(8000)}`), + readCall("old-c"), + readResult("old-c", `FOLD_${"e".repeat(8000)}`), + readCall("r1"), + readResult("r1", `OLD_${"a".repeat(3000)}`), + readCall("r2"), + readResult("r2", `NEW_${"b".repeat(3000)}`), + assistant, + siblingResult, + ]; +} + +function callIds(turns: readonly ConversationTurn[]): string[] { + return turns.flatMap((candidate) => + candidate.content.flatMap((block) => + block.type === "tool_call" ? [block.id] : [], + ), + ); +} + +/** Index of the turn line holding `callId`, and of the record for it. */ +function recordFollowsCall(dir: string, callId: string): boolean { + const lines = logLines(dir); + const callAt = lines.findIndex( + (line) => + Array.isArray(line.content) && + (line.content as { type: string; id?: string }[]).some( + (block) => block.type === "tool_call" && block.id === callId, + ), + ); + const next = lines[callAt + 1]; + return ( + callAt >= 0 && + next !== undefined && + "pendingApproval" in next && + (next.pendingApproval as PendingOperation).suspendedCall?.id === callId + ); +} function jsonl(records: readonly unknown[]): string { return records.map((record) => `${JSON.stringify(record)}\n`).join(""); @@ -134,3 +275,255 @@ describe("approval records in the turn log", () => { expect(recent).toEqual(history); }); }); + +describe("a parked approval in the turn log", () => { + test("the suspend commit writes it right after its call, snapshot scrubbed, and no metadata.json", async () => { + const dir = tempDir(); + await parkCall(dir, [user, assistant], [op]); + + expect(lineKinds(dir)).toEqual(["turn", "turn", "record"]); + const [record] = recordedApprovals(dir); + expect(record?.correlationId).toBe(op.correlationId); + expect(record?.timeoutAt).toBe(op.timeoutAt); + // The approver-facing snapshot is redacted at rest; the re-dispatch + // artifact stays raw, same as the call block it parks. + expect(JSON.stringify(record?.approvalSnapshot)).not.toContain(SECRET); + expect(record?.suspendedCall?.arguments).toEqual({ api_key: SECRET }); + expect(fs.existsSync(metadataPath(dir))).toBe(false); + + const loaded = await (await createOptimizedContextStore(dir)).load(); + expect(loaded.turns).toEqual([user, assistant]); + expect(loaded.pendingOperations).toHaveLength(1); + expect(loaded.pendingOperations[0]?.suspendedCall?.arguments).toEqual({ + api_key: SECRET, + }); + expect(loaded.pendingOperations[0]?.timeoutAt).toBe(op.timeoutAt); + }); + + test("the call's tool_result closes it with no metadata write or commit", async () => { + const dir = tempDir(); + const { store } = await parkCall(dir, [user, assistant], [op]); + // Sibling results land after the batch: the parked call is still open. + await store.writeTurns([user, assistant, siblingResult]); + expect(await openOps(dir)).toEqual([op.correlationId]); + + // Approved: the reactor drops the op in memory and the tool cycle writes + // turns only, then the process goes away before any commit. + await store.writeTurns([user, assistant, siblingResult, answered]); + expect(await openOps(dir)).toEqual([]); + // Append-only: the record kept its line, later turns follow it. + expect(lineKinds(dir)).toEqual(["turn", "turn", "record", "turn", "turn"]); + }); + + test("a fresh writer after a restart keeps it on its first rewrite", async () => { + const dir = tempDir(); + await parkCall(dir, [user, assistant], [op]); + + const second = await createOptimizedContextStore(dir); + const loaded = await second.load(); + await second.writeTurns([...loaded.turns, siblingResult]); + await second.writeMetadata({ + pendingOperations: loaded.pendingOperations, + tokenUsage: USAGE, + }); + await second.commit({ message: "siblings" }); + + expect(lineKinds(dir)).toEqual(["turn", "turn", "record", "turn"]); + expect(await openOps(dir)).toEqual([op.correlationId]); + }); + + test("a torn record line is dropped like any torn tail", async () => { + const dir = tempDir(); + const whole = JSON.stringify({ pendingApproval: op }); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + jsonl([user, assistant]) + whole.slice(0, whole.length / 2), + ); + + const store = await createOptimizedContextStore(dir); + const loaded = await store.load(); + expect(loaded.turns).toEqual([user, assistant]); + expect(loaded.pendingOperations).toEqual([]); + await store.writeTurns([...loaded.turns, siblingResult]); + expect(lineKinds(dir)).toEqual(["turn", "turn", "turn"]); + }); + + test("in a parallel batch, answering one parked call leaves only the other open", async () => { + const dir = tempDir(); + const second = approvalFor("call-2", { path: "a" }); + const { store } = await parkCall(dir, [user, assistant], [op, second]); + expect(lineKinds(dir)).toEqual(["turn", "turn", "record", "record"]); + expect(await openOps(dir)).toEqual([ + op.correlationId, + second.correlationId, + ]); + + await store.writeTurns([user, assistant, answered]); + await store.writeMetadata({ + pendingOperations: [second], + tokenUsage: USAGE, + }); + await store.commit({ message: "first answered" }); + + expect(await openOps(dir)).toEqual([second.correlationId]); + expect(fs.existsSync(metadataPath(dir))).toBe(false); + }); + + test("an async marker with no parked call stays on metadata.json", async () => { + const dir = tempDir(); + const marker: PendingOperation = { + correlationId: "corr-async", + kind: "approval", + registeredAt: 1, + gateId: "gate-async", + }; + await parkCall(dir, [user, assistant], [op, marker]); + + expect(recordedApprovals(dir).map((r) => r.correlationId)).toEqual([ + op.correlationId, + ]); + const metadata = JSON.parse(fs.readFileSync(metadataPath(dir), "utf8")) as { + pendingOperations: PendingOperation[]; + }; + expect(metadata.pendingOperations).toEqual([marker]); + expect(await openOps(dir)).toEqual([ + op.correlationId, + marker.correlationId, + ]); + }); +}); + +describe("a parked approval across a compaction", () => { + for (const writer of ["same writer", "fresh writer"] as const) { + test(`a ${writer} rewrite keeps the parked call and its record together`, async () => { + const dir = tempDir(); + const original = longHistoryEndingParked(); + const parked = await parkCall(dir, original, [op]); + const folded = ( + await createPruningCompactor().apply(original, {} as never) + ).output; + // The compactor must keep a parked call: dropping it would orphan the + // approval regardless of where the approval is stored. + expect(folded.length).toBeLessThan(original.length); + expect(callIds(folded)).toContain("call-1"); + + const store = + writer === "same writer" + ? parked.store + : await createOptimizedContextStore(dir); + if (writer === "fresh writer") await store.load(); + await store.writeTurns(folded); + await store.writeMetadata({ + pendingOperations: [op], + tokenUsage: USAGE, + }); + await store.commit({ message: "compact" }); + + expect(recordFollowsCall(dir, "call-1")).toBe(true); + expect(recordedApprovals(dir)).toHaveLength(1); + const loaded = await (await createOptimizedContextStore(dir)).load(); + expect(loaded.turns).toEqual(folded); + expect(loaded.pendingOperations.map((p) => p.correlationId)).toEqual([ + op.correlationId, + ]); + }); + } + + test("a failed compaction commit neither loses nor duplicates the parked approval", async () => { + const dir = tempDir(); + const original = longHistoryEndingParked(); + const { store } = await parkCall(dir, original, [op]); + const folded = (await createPruningCompactor().apply(original, {} as never)) + .output; + + await store.writeTurns(folded); + await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); + await failingCommit(dir, () => store.commit({ message: "compact" })); + + expect(recordedApprovals(dir)).toHaveLength(1); + expect(await openOps(dir)).toEqual([op.correlationId]); + expect((await store.load()).pendingOperations).toHaveLength(1); + // The next cycle's metadata write still finds the approval logged. + await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); + expect(fs.existsSync(metadataPath(dir))).toBe(false); + + await store.commit({ message: "retry compact" }); + expect(recordFollowsCall(dir, "call-1")).toBe(true); + expect(recordedApprovals(dir)).toHaveLength(1); + expect(await openOps(dir)).toEqual([op.correlationId]); + }); + + test("an approval parked while a rewrite is staged survives that rewrite's failed commit", async () => { + const dir = tempDir(); + const original = longHistoryEndingParked(); + const { store } = await parkCall(dir, original, []); + const folded = (await createPruningCompactor().apply(original, {} as never)) + .output; + + await store.writeTurns(folded); + await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); + // The live log is still the published one, so the approval waits on + // metadata.json until the rewrite lands. + expect(recordedApprovals(dir)).toEqual([]); + expect(fs.existsSync(metadataPath(dir))).toBe(true); + await failingCommit(dir, () => store.commit({ message: "compact" })); + + expect(recordedApprovals(dir)).toEqual([]); + expect(await openOps(dir)).toEqual([op.correlationId]); + + await store.commit({ message: "retry compact" }); + expect(recordFollowsCall(dir, "call-1")).toBe(true); + expect(await openOps(dir)).toEqual([op.correlationId]); + await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); + await store.commit({ message: "settle metadata" }); + expect(fs.existsSync(metadataPath(dir))).toBe(false); + expect(await headPaths(dir)).not.toContain("metadata.json"); + expect(await openOps(dir)).toEqual([op.correlationId]); + }); +}); + +describe("a pending approval left on a legacy metadata.json", () => { + async function legacySession(turns: ConversationTurn[]): Promise { + const dir = tempDir(); + const seed = await createOptimizedContextStore(dir); + await seed.writeTurns(turns); + fs.writeFileSync( + metadataPath(dir), + JSON.stringify({ pendingOperations: [op], connectorState: null }), + ); + await seed.commit({ message: "legacy" }); + expect(await headPaths(dir)).toContain("metadata.json"); + return dir; + } + + test("an open one loads, then moves into the log and the file leaves the commit", async () => { + const dir = await legacySession([user, assistant]); + + const store = await createOptimizedContextStore(dir); + const loaded = await store.load(); + expect(loaded.pendingOperations.map((p) => p.correlationId)).toEqual([ + op.correlationId, + ]); + await store.writeTurns([...loaded.turns, siblingResult]); + await store.writeMetadata({ + pendingOperations: loaded.pendingOperations, + tokenUsage: USAGE, + }); + await store.commit({ message: "migrate" }); + + expect(recordFollowsCall(dir, "call-1")).toBe(true); + expect(fs.existsSync(metadataPath(dir))).toBe(false); + expect(await headPaths(dir)).not.toContain("metadata.json"); + expect(await openOps(dir)).toEqual([op.correlationId]); + }); + + test("one whose call is answered does not re-arm", async () => { + const dir = await legacySession([user, assistant, siblingResult, answered]); + expect(await openOps(dir)).toEqual([]); + }); + + test("one whose call is not in history does not re-arm", async () => { + const dir = await legacySession([user]); + expect(await openOps(dir)).toEqual([]); + }); +}); diff --git a/src/session/pending-approval-record.ts b/src/session/pending-approval-record.ts index 300e8e808..021b6ea8b 100644 --- a/src/session/pending-approval-record.ts +++ b/src/session/pending-approval-record.ts @@ -32,6 +32,14 @@ const PendingApprovalRecordSchema = type({ "+": "reject", }); +export type PendingApprovalRecord = { pendingApproval: PendingOperation }; + +export function pendingApprovalRecord( + op: PendingOperation, +): PendingApprovalRecord { + return { pendingApproval: op }; +} + /** The approval a turn log line records, or null when the line is not one. */ export function parsePendingApprovalRecord( raw: unknown, diff --git a/src/session/pending-snapshot-scrub.test.ts b/src/session/pending-snapshot-scrub.test.ts index 474376d8c..998969bf6 100644 --- a/src/session/pending-snapshot-scrub.test.ts +++ b/src/session/pending-snapshot-scrub.test.ts @@ -91,6 +91,21 @@ describe("scrubPersistedPendingOperations", () => { arguments: args, }, }; + // The parked call is in history before its approval is written. + await store.writeTurns([ + { + role: "assistant", + content: [ + { + type: "tool_call", + id: "call-9", + name: "run_shell", + arguments: args, + }, + ], + timestamp: 1, + }, + ]); await store.writeMetadata({ pendingOperations: [op], tokenUsage: { @@ -103,15 +118,16 @@ describe("scrubPersistedPendingOperations", () => { }); await store.commit({ message: "checkpoint: pending" }); - // Redacted at rest ... + // Redacted at rest in the turn log record next to the call ... + const [, recordLine] = fs + .readFileSync(path.join(dir, "turns.jsonl"), "utf8") + .split("\n"); const persistedArgs = JSON.stringify( ( - JSON.parse( - fs.readFileSync(path.join(dir, "metadata.json"), "utf8"), - ) as { - pendingOperations: { approvalSnapshot: { arguments: object } }[]; + JSON.parse(recordLine ?? "null") as { + pendingApproval: { approvalSnapshot: { arguments: object } }; } - ).pendingOperations[0]?.approvalSnapshot?.arguments, + ).pendingApproval.approvalSnapshot.arguments, ); expect(persistedArgs).not.toContain(SECRET); diff --git a/src/session/pending-snapshot-scrub.ts b/src/session/pending-snapshot-scrub.ts index f95b41bab..62d609f02 100644 --- a/src/session/pending-snapshot-scrub.ts +++ b/src/session/pending-snapshot-scrub.ts @@ -1,5 +1,6 @@ -// Approval snapshot args at rest. metadata.json (git-backed session state -// under .agent-state) persists each PendingOperation verbatim, +// Approval snapshot args at rest. Session state (git-backed, under +// .agent-state: the turn log record next to a parked call, or metadata.json) +// persists each PendingOperation verbatim, // `approvalSnapshot.arguments` included — the raw model arguments of an ask // the operator has not answered yet. The result-side secret scrubber never // sees these, so a parked call carrying a bearer-shaped argument persisted it From 9556d30fd7a0955636ff1099d4821a0e8e6121ad Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 17 Oct 2026 09:18:44 -0700 Subject: [PATCH 4/6] fix(session): stop dropping approvals load() can still see A diverted turn write marked the approval logged even though load() only reads turns.jsonl, so the next metadata write deleted it. A closed log record also tombstoned a later open op with the same correlation id, and a record was accepted when only the call id matched. Record an id as logged only on the canonical turn log. Accept a correlation id only once the op is open, and only when the suspended call matches the unanswered tool call's name and arguments. --- src/session/optimized-context-store.test.ts | 110 +++++++++++++++++++- src/session/optimized-context-store.ts | 41 ++++++-- src/session/pending-approval-log.test.ts | 88 ++++++++++++++++ 3 files changed, 227 insertions(+), 12 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 9919ace26..3801bf2f7 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -5,7 +5,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import git, { type CommitObject } from "isomorphic-git"; -import type { ConversationTurn } from "@intx/types/runtime"; +import type { ConversationTurn, PendingOperation } from "@intx/types/runtime"; import { createOptimizedContextStore, createSessionStores, @@ -945,6 +945,114 @@ describe("concurrent sessions write disjoint transcripts", () => { ).toEqual(turnTexts(turnsB)); }); + test("a diverted write does not drop an approval load() still has to return", async () => { + const dir = tempDir(); + const parked: ConversationTurn = { + role: "assistant", + content: [ + { + type: "tool_call", + id: "call-1", + name: "run_shell", + arguments: { cmd: "true" }, + }, + ], + timestamp: 2, + }; + const turns = [turn("probe"), parked]; + const approval: PendingOperation = { + correlationId: "corr-call-1", + kind: "approval", + registeredAt: 1, + gateId: "pending-corr-call-1", + suspendedCall: { + id: "call-1", + name: "run_shell", + arguments: { cmd: "true" }, + }, + }; + const sessionA = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await sessionA.writeTurns(turns); + await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionA.commit({ message: "session A" }); + + const sessionB = await createOptimizedContextStore(dir, { + ownerId: "session-B", + }); + await sessionB.writeTurns(turns); + await sessionB.writeMetadata({ + pendingOperations: [approval], + tokenUsage: EMPTY_CHECKPOINT_METADATA.tokenUsage, + }); + await sessionB.commit({ message: "divert" }); + await sessionB.writeMetadata({ + pendingOperations: [approval], + tokenUsage: EMPTY_CHECKPOINT_METADATA.tokenUsage, + }); + await sessionB.commit({ message: "divert again" }); + + const loaded = await (await createOptimizedContextStore(dir)).load(); + expect( + loaded.turns.some((candidate) => + candidate.content.some( + (block) => block.type === "tool_call" && block.id === "call-1", + ), + ), + ).toBe(true); + expect( + loaded.pendingOperations.map((pending) => pending.correlationId), + ).toEqual(["corr-call-1"]); + }); + + test("a same-owner approval is logged in turns.jsonl and dropped from metadata.json", async () => { + const dir = tempDir(); + const parked: ConversationTurn = { + role: "assistant", + content: [ + { + type: "tool_call", + id: "call-1", + name: "run_shell", + arguments: { cmd: "true" }, + }, + ], + timestamp: 2, + }; + const approval: PendingOperation = { + correlationId: "corr-call-1", + kind: "approval", + registeredAt: 1, + gateId: "pending-corr-call-1", + suspendedCall: { + id: "call-1", + name: "run_shell", + arguments: { cmd: "true" }, + }, + }; + const owner = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await owner.writeTurns([turn("probe"), parked]); + await owner.writeMetadata({ + pendingOperations: [approval], + tokenUsage: EMPTY_CHECKPOINT_METADATA.tokenUsage, + }); + await owner.commit({ message: "park" }); + + expect(fs.existsSync(path.join(dir, "metadata.json"))).toBe(false); + const lines = fs + .readFileSync(path.join(dir, TURNS_FILE), "utf8") + .split("\n") + .filter((line) => line.length > 0); + expect(lines.some((line) => line.includes("corr-call-1"))).toBe(true); + const loaded = await (await createOptimizedContextStore(dir)).load(); + expect( + loaded.pendingOperations.map((pending) => pending.correlationId), + ).toEqual(["corr-call-1"]); + }); + test("markerless resume can still extend the shared log", async () => { const dir = tempDir(); const sessionA = await createOptimizedContextStore(dir); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 91bf1e9e8..e448a8c7c 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -1127,32 +1127,47 @@ async function managedPathsDiffer( } /** - * A parked call's approval is open while its call is in history and no - * tool_result answers it; an op with no parked call is an async marker and - * always passes. First occurrence of a correlationId wins, so callers list - * the log before legacy metadata.json. + * A parked call's approval is open while its call is in history, unanswered, + * and the same call (id, name, and arguments). An op with no parked call is + * an async marker and always passes. The first op actually accepted as open + * wins for its correlationId: a closed record must not tombstone a later open + * one, because load() lists the log before legacy metadata.json. */ function openPendingOperations( turns: readonly ConversationTurn[], candidates: readonly PendingOperation[], ): PendingOperation[] { - const called = new Set(); + const called = new Map(); const answered = new Set(); for (const turn of turns) { for (const block of turn.content) { - if (block.type === "tool_call") called.add(block.id); - else if (block.type === "tool_result") answered.add(block.callId); + if (block.type === "tool_call") { + if (!called.has(block.id)) { + called.set(block.id, { + name: block.name, + arguments: block.arguments, + }); + } + } else if (block.type === "tool_result") answered.add(block.callId); } } const seen = new Set(); const open: PendingOperation[] = []; for (const op of candidates) { if (seen.has(op.correlationId)) continue; - seen.add(op.correlationId); - const callId = op.suspendedCall?.id; - if (callId !== undefined && (!called.has(callId) || answered.has(callId))) { - continue; + const call = op.suspendedCall; + if (call !== undefined) { + const historical = called.get(call.id); + if ( + historical === undefined || + answered.has(call.id) || + historical.name !== call.name || + JSON.stringify(historical.arguments) !== JSON.stringify(call.arguments) + ) { + continue; + } } + seen.add(op.correlationId); open.push(op); } return open; @@ -1310,6 +1325,10 @@ export async function createSessionStores( ): Promise { const { lines, included } = withApprovalRecords(turns); await writeSegmented(writeTurnsSegmented, lines); + // Diversion writes a versioned base load() never reads. Counting that as + // logged would drop the approval from metadata.json while the unanswered + // call is still only in the canonical log. + if (turnsBaseName !== TURNS_FILE) return; for (const id of included) loggedApprovalIds.add(id); } diff --git a/src/session/pending-approval-log.test.ts b/src/session/pending-approval-log.test.ts index c2826db72..2a893ad51 100644 --- a/src/session/pending-approval-log.test.ts +++ b/src/session/pending-approval-log.test.ts @@ -351,6 +351,11 @@ describe("a parked approval in the turn log", () => { test("in a parallel batch, answering one parked call leaves only the other open", async () => { const dir = tempDir(); const second = approvalFor("call-2", { path: "a" }); + second.suspendedCall = { + id: "call-2", + name: "read_file", + arguments: { path: "a" }, + }; const { store } = await parkCall(dir, [user, assistant], [op, second]); expect(lineKinds(dir)).toEqual(["turn", "turn", "record", "record"]); expect(await openOps(dir)).toEqual([ @@ -526,4 +531,87 @@ describe("a pending approval left on a legacy metadata.json", () => { const dir = await legacySession([user]); expect(await openOps(dir)).toEqual([]); }); + + test("a closed log record does not tombstone an open op with the same correlationId", async () => { + const dir = tempDir(); + const closed = { + pendingApproval: { + ...op, + suspendedCall: { + id: "gone", + name: "run_shell", + arguments: { api_key: SECRET }, + }, + }, + }; + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + jsonl([user, assistant, closed]), + ); + fs.writeFileSync( + metadataPath(dir), + JSON.stringify({ pendingOperations: [op], connectorState: null }), + ); + + const loaded = await (await createOptimizedContextStore(dir)).load(); + expect(callIds(loaded.turns)).toContain("call-1"); + expect( + loaded.pendingOperations.map((pending) => pending.correlationId), + ).toEqual([op.correlationId]); + }); + + test("an answered call stays closed when metadata.json still lists the logged approval", async () => { + const dir = tempDir(); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + jsonl([ + user, + assistant, + siblingResult, + { pendingApproval: op }, + answered, + ]), + ); + fs.writeFileSync( + metadataPath(dir), + JSON.stringify({ pendingOperations: [op], connectorState: null }), + ); + expect(await openOps(dir)).toEqual([]); + }); +}); + +describe("a parked approval must be the historical tool call", () => { + test("a suspendedCall whose body differs from the unanswered tool_call is not open", async () => { + const dir = tempDir(); + const parked: ConversationTurn = { + role: "assistant", + content: [ + { + type: "tool_call", + id: "call-1", + name: "run_shell", + arguments: { cmd: "true" }, + }, + ], + timestamp: 2, + }; + const mismatched = { + pendingApproval: { + ...op, + suspendedCall: { + id: "call-1", + name: "run_shell", + arguments: { cmd: "false" }, + }, + }, + }; + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + jsonl([user, parked, mismatched]), + ); + + const loaded = await (await createOptimizedContextStore(dir)).load(); + expect(callIds(loaded.turns)).toContain("call-1"); + expect(loaded.pendingOperations).toEqual([]); + }); }); From 0584e1a9e8d27fa2d69dfc2163a9bcfdb44c248a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 17 Oct 2026 09:51:22 -0700 Subject: [PATCH 5/6] fix(session): flush a parked approval once into the turn log writeTurns stages an append so writeMetadata can attach the approval and flush the segment a single time. commit only git-adds that flush. A parked approval is never written to metadata.json. A versioned divert is not the log load() reads, and is not copied into metadata to compensate. Legacy open approvals move into the log on the next commit and are not written back. Async markers and connectorState stay on metadata.json. --- src/session/optimized-context-store.test.ts | 25 ++- src/session/optimized-context-store.ts | 225 +++++++++++++++++--- src/session/pending-approval-log.test.ts | 27 ++- 3 files changed, 229 insertions(+), 48 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 3801bf2f7..e6df14686 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -724,9 +724,10 @@ describe("createOptimizedContextStore checkpoint", () => { await store.writeTurns([...turns]); } const extraTurns = segmentFileName(TURNS_FILE, 1); - expect(fs.existsSync(path.join(dir, extraTurns))).toBe(true); - + // The append stays staged until commit flushes it. That flush is what + // the empty-checkpoint skip must not miss. const second = await store.commit({ message: "pending extra" }); + expect(fs.existsSync(path.join(dir, extraTurns))).toBe(true); expect(second.hash).not.toBe(first.hash); expect(await gitLsTree(dir)).toContain(extraTurns); }); @@ -945,7 +946,7 @@ describe("concurrent sessions write disjoint transcripts", () => { ).toEqual(turnTexts(turnsB)); }); - test("a diverted write does not drop an approval load() still has to return", async () => { + test("a diverted write is not the log load() re-arms", async () => { const dir = tempDir(); const parked: ConversationTurn = { role: "assistant", @@ -1001,9 +1002,20 @@ describe("concurrent sessions write disjoint transcripts", () => { ), ), ).toBe(true); - expect( - loaded.pendingOperations.map((pending) => pending.correlationId), - ).toEqual(["corr-call-1"]); + // The approval landed on the versioned base. load() reads turns.jsonl + // only, and the approval is not copied into metadata.json. + expect(loaded.pendingOperations).toEqual([]); + expect(fs.existsSync(path.join(dir, "metadata.json"))).toBe(false); + const versioned = versionedTurnsFiles(dir); + expect(versioned).toHaveLength(1); + const diverted = fs.readFileSync( + path.join(dir, defined(versioned[0])), + "utf8", + ); + expect(diverted).toContain("corr-call-1"); + expect(fs.readFileSync(path.join(dir, TURNS_FILE), "utf8")).not.toContain( + "corr-call-1", + ); }); test("a same-owner approval is logged in turns.jsonl and dropped from metadata.json", async () => { @@ -1390,6 +1402,7 @@ describe("concurrent sessions write disjoint transcripts", () => { ownerId: "session-A", }); await sessionA.writeTurns([turn("seed")]); + await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); fs.unlinkSync(path.join(dir, TURNS_FILE)); const sessionB = await createOptimizedContextStore(dir, { diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index e448a8c7c..0eb9f86e3 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -1272,8 +1272,16 @@ export async function createSessionStores( // writer sees a new one as a prefix extension, and a closed one keeps its // line: dropping it would turn every later append into a rewrite. const approvalRecords = new Map(); - // Approvals the log on disk holds; any other parked op rides metadata.json. + // Approvals a canonical turns.jsonl flush has already written. A versioned + // base is not this set: load() never reads it, and the approval is not + // copied onto metadata.json to compensate. const loggedApprovalIds = new Set(); + // Append writeTurns holds turns here until one flush. The commit cycle + // flushes from writeMetadata, once the parked approval exists to attach. + // commit only git-adds that flush. A tool-result append flushes here, + // because that cycle never calls writeMetadata and the result is what + // closes the record. + let stagedAppend: readonly ConversationTurn[] | null = null; function adoptLoggedApprovals(ops: readonly PendingOperation[]): void { for (const op of ops) { @@ -1284,6 +1292,28 @@ export async function createSessionStores( } } + function hasUnloggedApproval(): boolean { + for (const id of approvalRecords.keys()) { + if (!loggedApprovalIds.has(id)) return true; + } + return false; + } + + function newTurnsHaveToolResult( + prev: readonly ConversationTurn[], + next: readonly ConversationTurn[], + ): boolean { + const start = refPrefixLength(prev, next); + for (let index = start; index < next.length; index++) { + const turn = next[index]; + if (turn === undefined) continue; + for (const block of turn.content) { + if (block.type === "tool_result") return true; + } + } + return false; + } + // Located by call id, not turn identity, so a record follows an excerpted // copy of its turn through a compaction and is dropped with its call. function withApprovalRecords(turns: readonly ConversationTurn[]): { @@ -1325,13 +1355,83 @@ export async function createSessionStores( ): Promise { const { lines, included } = withApprovalRecords(turns); await writeSegmented(writeTurnsSegmented, lines); - // Diversion writes a versioned base load() never reads. Counting that as - // logged would drop the approval from metadata.json while the unanswered - // call is still only in the canonical log. + // Diversion writes a versioned base load() never reads. That flush is not + // the canonical log, and the approval is not written to metadata.json. if (turnsBaseName !== TURNS_FILE) return; for (const id of included) loggedApprovalIds.add(id); } + async function flushStagedAppend(): Promise { + if (stagedAppend === null) return; + const turns = stagedAppend; + stagedAppend = null; + await writeTurnLog(turns); + liveTurnRefs = [...turns]; + invalidateTurnsCache(); + } + + // One canonical flush of the staged append, or of the already-flushed turns + // when a legacy approval still has to be inserted. A staged rewrite stays + // unpublished until commit, which is the flush for that path. + async function flushCanonicalApprovals(): Promise { + if (turnsBaseName !== TURNS_FILE || unpublishedRewrite !== null) return; + if (stagedAppend !== null) { + await flushStagedAppend(); + return; + } + if (!hasUnloggedApproval()) return; + const turns = liveTurnRefs ?? (await readLiveTurns()); + if (turns.length === 0) return; + await writeTurnLog(turns); + liveTurnRefs = [...turns]; + invalidateTurnsCache(); + } + + async function noteAppend(turns: readonly ConversationTurn[]): Promise { + const previous = liveTurnRefs; + stagedAppend = [...turns]; + if (previous !== null && newTurnsHaveToolResult(previous, turns)) { + await flushStagedAppend(); + return; + } + // A fresh store has no in-memory baseline. A tool-result suffix on top of + // the log already on disk still has to flush here: that cycle does not + // call writeMetadata. The first publish of a history (empty log) stays + // staged so writeMetadata can attach a parked approval and flush once. + if (previous === null && (await diskSuffixHasToolResult(turns))) { + await flushStagedAppend(); + } + } + + async function diskSuffixHasToolResult( + turns: readonly ConversationTurn[], + ): Promise { + const live = await readLiveTurns(); + if (live.length === 0 || turns.length <= live.length) return false; + if (contentPrefixLength(live, turns) < live.length) return false; + for (let index = live.length; index < turns.length; index++) { + const turn = turns[index]; + if (turn === undefined) continue; + for (const block of turn.content) { + if (block.type === "tool_result") return true; + } + } + return false; + } + + function retainedLegacyApprovals(raw: string | null): PendingOperation[] { + if (raw === null || raw.length === 0) return []; + try { + return readMetadataDocument(raw).pendingOperations.filter( + (op) => + op.suspendedCall !== undefined && + !loggedApprovalIds.has(op.correlationId), + ); + } catch { + return []; + } + } + async function writeMetadataLive(metadata: { pendingOperations: PendingOperation[]; tokenUsage: TokenUsage; @@ -1339,8 +1439,7 @@ export async function createSessionStores( await appendUsageSidecar(metadata.tokenUsage); // Approval snapshot args would persist in the clear; scrub them at this // app-owned write (suspendedCall stays raw for the approved call's - // re-dispatch). - let added = false; + // re-dispatch). Parked approvals never go into metadata.json. for (const op of scrubPersistedPendingOperations( metadata.pendingOperations.filter( (pending) => @@ -1349,26 +1448,24 @@ export async function createSessionStores( ), )) { approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); - added = true; } // The reactor writes the parked call's turn just before this in the same - // commit cycle, so the record lands in the same commit as its call. - if (added && unpublishedRewrite === null && liveTurnRefs !== null) { - await writeTurnLog(liveTurnRefs); - invalidateTurnsCache(); - } - // Only what the log does not hold yet stays here: async markers, and an - // approval waiting on a staged rewrite or this process's first turn - // write. Legacy files are still read, so dropping a logged op is safe. - const sideOps = metadata.pendingOperations.filter( - (pending) => - pending.suspendedCall === undefined || - !loggedApprovalIds.has(pending.correlationId), - ); + // commit cycle. Attaching the record here is the cycle's one segment flush. + await flushCanonicalApprovals(); if (!metadataJsonSeeded) { writtenMetadataJson = await readTextOrNull(path.join(dir, METADATA_FILE)); metadataJsonSeeded = true; } + // Async markers (no suspendedCall) and connectorState stay in this file. + // A legacy parked approval already on disk stays until the canonical log + // has it; this write does not put it back, and does not add a new one. + const retained = retainedLegacyApprovals(writtenMetadataJson); + const sideOps = [ + ...metadata.pendingOperations.filter( + (pending) => pending.suspendedCall === undefined, + ), + ...retained, + ]; if (sideOps.length === 0 && connectorStateBuffer === null) { if (writtenMetadataJson !== null) { await fs.promises.rm(path.join(dir, METADATA_FILE), { force: true }); @@ -1377,6 +1474,28 @@ export async function createSessionStores( } return; } + if (retained.length > 0 && writtenMetadataJson !== null) { + try { + const onDisk = readMetadataDocument(writtenMetadataJson); + const asyncOnDisk = onDisk.pendingOperations.filter( + (op) => op.suspendedCall === undefined, + ); + const asyncFromCall = metadata.pendingOperations.filter( + (pending) => pending.suspendedCall === undefined, + ); + const connectorSame = + JSON.stringify(onDisk.connectorState ?? null) === + JSON.stringify(connectorStateBuffer); + if ( + connectorSame && + JSON.stringify(asyncOnDisk) === JSON.stringify(asyncFromCall) + ) { + return; + } + } catch { + // Unreadable legacy document: fall through and publish the side ops. + } + } const desired = JSON.stringify( { pendingOperations: scrubPersistedPendingOperations(sideOps), @@ -1616,14 +1735,17 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - if (liveTurnRefs !== null) { - if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { + if (liveTurnRefs !== null || stagedAppend !== null) { + const baseline = stagedAppend ?? liveTurnRefs; + if ( + baseline !== null && + refPrefixLength(baseline, turns) < baseline.length + ) { + stagedAppend = null; unpublishedRewrite = [...turns]; return; } - await writeTurnLog(turns); - liveTurnRefs = [...turns]; - invalidateTurnsCache(); + await noteAppend(turns); return; } await withResolvedDirLock(dir, async () => { @@ -1631,14 +1753,17 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - if (liveTurnRefs !== null) { - if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { + if (liveTurnRefs !== null || stagedAppend !== null) { + const baseline = stagedAppend ?? liveTurnRefs; + if ( + baseline !== null && + refPrefixLength(baseline, turns) < baseline.length + ) { + stagedAppend = null; unpublishedRewrite = [...turns]; return; } - await writeTurnLog(turns); - liveTurnRefs = [...turns]; - invalidateTurnsCache(); + await noteAppend(turns); return; } const decision = await resolveFirstWrite(turns); @@ -1651,9 +1776,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeTurnLog(turns); - liveTurnRefs = [...turns]; - invalidateTurnsCache(); + await noteAppend(turns); }); } @@ -1723,6 +1846,31 @@ export async function createSessionStores( ); metadata = emptyMetadata(); } + // A still-open legacy approval is remembered so the next canonical + // flush moves it into the log. It is not marked logged, and it is not + // written back. + for (const op of scrubPersistedPendingOperations( + openPendingOperations(liveLog.turns, metadata.pendingOperations).filter( + (pending) => + pending.suspendedCall !== undefined && + !approvalRecords.has(pending.correlationId), + ), + )) { + approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); + } + if (stagedAppend !== null && unpublishedRewrite === null) { + await flushStagedAppend(); + try { + liveLog = await readLiveLogCached(); + } catch (cause) { + throw new Error( + `failed to load ${TURNS_FILE}: ${ + cause instanceof Error ? cause.message : String(cause) + }`, + { cause }, + ); + } + } const { turns } = liveLog; return { turns, @@ -1835,6 +1983,13 @@ export async function createSessionStores( async commit(options, signal) { return withResolvedDirLock(dir, async () => { const stagedRewrite = unpublishedRewrite; + // An append the commit cycle did not flush (no writeMetadata) still + // has to land before git add. A rewrite stays staged until the + // snapshot below, then one flush — commit does not flush again after + // writeMetadata already did. + if (stagedAppend !== null && stagedRewrite === null) { + await flushStagedAppend(); + } // The staged rewrite lands on the working-tree segments before the git // operations below; snapshot them so a failed commit can restore the // published generation. The rewrite stays staged so a retry can still @@ -2007,8 +2162,8 @@ export async function createSessionStores( } catch { // A partial restore must not mask the real commit error. } - // Records only the rewrite carried are gone again; those - // approvals stay on metadata.json until a rewrite lands. + // Records only the rewrite carried are not canonical until a + // rewrite lands. Approvals already on the published log stay. loggedApprovalIds.clear(); for (const id of loggedBefore) loggedApprovalIds.add(id); } diff --git a/src/session/pending-approval-log.test.ts b/src/session/pending-approval-log.test.ts index 2a893ad51..cba8f1702 100644 --- a/src/session/pending-approval-log.test.ts +++ b/src/session/pending-approval-log.test.ts @@ -277,6 +277,22 @@ describe("approval records in the turn log", () => { }); describe("a parked approval in the turn log", () => { + test("writeTurns stages the parked call and writeMetadata flushes it once", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + await store.writeTurns([user, assistant]); + expect(fs.existsSync(path.join(dir, "turns.jsonl"))).toBe(false); + + await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); + expect(lineKinds(dir)).toEqual(["turn", "turn", "record"]); + expect(fs.existsSync(metadataPath(dir))).toBe(false); + const flushed = fs.readFileSync(path.join(dir, "turns.jsonl")); + + await store.commit({ message: "suspend" }); + expect(fs.readFileSync(path.join(dir, "turns.jsonl"))).toEqual(flushed); + expect(await openOps(dir)).toEqual([op.correlationId]); + }); + test("the suspend commit writes it right after its call, snapshot scrubbed, and no metadata.json", async () => { const dir = tempDir(); await parkCall(dir, [user, assistant], [op]); @@ -467,23 +483,20 @@ describe("a parked approval across a compaction", () => { await store.writeTurns(folded); await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); - // The live log is still the published one, so the approval waits on - // metadata.json until the rewrite lands. + // The rewrite is still unpublished. The approval is not on metadata.json; + // it rides the staged rewrite and lands when that commit flushes. expect(recordedApprovals(dir)).toEqual([]); - expect(fs.existsSync(metadataPath(dir))).toBe(true); + expect(fs.existsSync(metadataPath(dir))).toBe(false); await failingCommit(dir, () => store.commit({ message: "compact" })); expect(recordedApprovals(dir)).toEqual([]); - expect(await openOps(dir)).toEqual([op.correlationId]); + expect(await openOps(dir)).toEqual([]); await store.commit({ message: "retry compact" }); expect(recordFollowsCall(dir, "call-1")).toBe(true); expect(await openOps(dir)).toEqual([op.correlationId]); - await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); - await store.commit({ message: "settle metadata" }); expect(fs.existsSync(metadataPath(dir))).toBe(false); expect(await headPaths(dir)).not.toContain("metadata.json"); - expect(await openOps(dir)).toEqual([op.correlationId]); }); }); From 0c3a31407f9f6b2751c6a9eea21e56bd0915382b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 17 Oct 2026 10:11:23 -0700 Subject: [PATCH 6/6] fix(session): persist turn appends before the approval record writeTurns of an append hits turns.jsonl before it returns, so a new store can load those turns with no metadata write and no commit. writeMetadata then appends only the parked approval's record line. That append also lands on the published log while a rewrite is still staged, and a failed commit restores it with the published generation. --- src/session/optimized-context-store.test.ts | 6 +- src/session/optimized-context-store.ts | 174 +++++++------------- src/session/pending-approval-log.test.ts | 29 +++- 3 files changed, 83 insertions(+), 126 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index e6df14686..368b24ec0 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -724,10 +724,9 @@ describe("createOptimizedContextStore checkpoint", () => { await store.writeTurns([...turns]); } const extraTurns = segmentFileName(TURNS_FILE, 1); - // The append stays staged until commit flushes it. That flush is what - // the empty-checkpoint skip must not miss. - const second = await store.commit({ message: "pending extra" }); expect(fs.existsSync(path.join(dir, extraTurns))).toBe(true); + + const second = await store.commit({ message: "pending extra" }); expect(second.hash).not.toBe(first.hash); expect(await gitLsTree(dir)).toContain(extraTurns); }); @@ -1402,7 +1401,6 @@ describe("concurrent sessions write disjoint transcripts", () => { ownerId: "session-A", }); await sessionA.writeTurns([turn("seed")]); - await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); fs.unlinkSync(path.join(dir, TURNS_FILE)); const sessionB = await createOptimizedContextStore(dir, { diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 0eb9f86e3..265dca56a 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -1276,12 +1276,10 @@ export async function createSessionStores( // base is not this set: load() never reads it, and the approval is not // copied onto metadata.json to compensate. const loggedApprovalIds = new Set(); - // Append writeTurns holds turns here until one flush. The commit cycle - // flushes from writeMetadata, once the parked approval exists to attach. - // commit only git-adds that flush. A tool-result append flushes here, - // because that cycle never calls writeMetadata and the result is what - // closes the record. - let stagedAppend: readonly ConversationTurn[] | null = null; + // Line objects last passed to the canonical segmented writer, same refs, so + // a later record can extend that prefix by one append instead of rewriting + // the turns. + let canonicalLines: readonly unknown[] | null = null; function adoptLoggedApprovals(ops: readonly PendingOperation[]): void { for (const op of ops) { @@ -1299,21 +1297,6 @@ export async function createSessionStores( return false; } - function newTurnsHaveToolResult( - prev: readonly ConversationTurn[], - next: readonly ConversationTurn[], - ): boolean { - const start = refPrefixLength(prev, next); - for (let index = start; index < next.length; index++) { - const turn = next[index]; - if (turn === undefined) continue; - for (const block of turn.content) { - if (block.type === "tool_result") return true; - } - } - return false; - } - // Located by call id, not turn identity, so a record follows an excerpted // copy of its turn through a compaction and is dropped with its call. function withApprovalRecords(turns: readonly ConversationTurn[]): { @@ -1359,64 +1342,48 @@ export async function createSessionStores( // the canonical log, and the approval is not written to metadata.json. if (turnsBaseName !== TURNS_FILE) return; for (const id of included) loggedApprovalIds.add(id); + canonicalLines = lines; } - async function flushStagedAppend(): Promise { - if (stagedAppend === null) return; - const turns = stagedAppend; - stagedAppend = null; - await writeTurnLog(turns); - liveTurnRefs = [...turns]; - invalidateTurnsCache(); - } - - // One canonical flush of the staged append, or of the already-flushed turns - // when a legacy approval still has to be inserted. A staged rewrite stays - // unpublished until commit, which is the flush for that path. - async function flushCanonicalApprovals(): Promise { - if (turnsBaseName !== TURNS_FILE || unpublishedRewrite !== null) return; - if (stagedAppend !== null) { - await flushStagedAppend(); - return; + // The turns are already on disk. Extend the canonical log by the new record + // lines only. A prefix extension appends; it does not rewrite the turns. A + // writer with no memory of those bytes appends the line to the active + // segment instead of rebuilding it. + async function appendParkedRecords( + published: readonly ConversationTurn[], + ): Promise { + if (turnsBaseName !== TURNS_FILE) return; + const callIds = new Set(); + for (const turn of published) { + for (const block of turn.content) { + if (block.type === "tool_call") callIds.add(block.id); + } } - if (!hasUnloggedApproval()) return; - const turns = liveTurnRefs ?? (await readLiveTurns()); - if (turns.length === 0) return; - await writeTurnLog(turns); - liveTurnRefs = [...turns]; - invalidateTurnsCache(); - } - - async function noteAppend(turns: readonly ConversationTurn[]): Promise { - const previous = liveTurnRefs; - stagedAppend = [...turns]; - if (previous !== null && newTurnsHaveToolResult(previous, turns)) { - await flushStagedAppend(); - return; + const pending: PendingApprovalRecord[] = []; + for (const [id, record] of approvalRecords) { + if (loggedApprovalIds.has(id)) continue; + const callId = record.pendingApproval.suspendedCall?.id; + if (callId === undefined || !callIds.has(callId)) continue; + pending.push(record); } - // A fresh store has no in-memory baseline. A tool-result suffix on top of - // the log already on disk still has to flush here: that cycle does not - // call writeMetadata. The first publish of a history (empty log) stays - // staged so writeMetadata can attach a parked approval and flush once. - if (previous === null && (await diskSuffixHasToolResult(turns))) { - await flushStagedAppend(); + if (pending.length === 0) return; + if (canonicalLines !== null) { + const lines = [...canonicalLines, ...pending]; + await writeSegmented(writeTurnsSegmented, lines); + canonicalLines = lines; + } else { + const names = await listSegmentFiles(dir, TURNS_FILE); + const target = names.at(-1) ?? TURNS_FILE; + const text = pending + .map((record) => `${JSON.stringify(record)}\n`) + .join(""); + await fs.promises.appendFile(path.join(dir, target), text); + pendingSegmentPaths.add(target); } - } - - async function diskSuffixHasToolResult( - turns: readonly ConversationTurn[], - ): Promise { - const live = await readLiveTurns(); - if (live.length === 0 || turns.length <= live.length) return false; - if (contentPrefixLength(live, turns) < live.length) return false; - for (let index = live.length; index < turns.length; index++) { - const turn = turns[index]; - if (turn === undefined) continue; - for (const block of turn.content) { - if (block.type === "tool_result") return true; - } + for (const record of pending) { + loggedApprovalIds.add(record.pendingApproval.correlationId); } - return false; + invalidateTurnsCache(); } function retainedLegacyApprovals(raw: string | null): PendingOperation[] { @@ -1449,9 +1416,13 @@ export async function createSessionStores( )) { approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); } - // The reactor writes the parked call's turn just before this in the same - // commit cycle. Attaching the record here is the cycle's one segment flush. - await flushCanonicalApprovals(); + // Turns are already on disk. A parked approval appends its record line + // once, including while a rewrite is still unpublished, so load() of the + // published log sees it before commit. No parked approval adds no write. + if (hasUnloggedApproval()) { + const published = liveTurnRefs ?? (await readLiveTurns()); + await appendParkedRecords(published); + } if (!metadataJsonSeeded) { writtenMetadataJson = await readTextOrNull(path.join(dir, METADATA_FILE)); metadataJsonSeeded = true; @@ -1735,17 +1706,14 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - if (liveTurnRefs !== null || stagedAppend !== null) { - const baseline = stagedAppend ?? liveTurnRefs; - if ( - baseline !== null && - refPrefixLength(baseline, turns) < baseline.length - ) { - stagedAppend = null; + if (liveTurnRefs !== null) { + if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { unpublishedRewrite = [...turns]; return; } - await noteAppend(turns); + await writeTurnLog(turns); + liveTurnRefs = [...turns]; + invalidateTurnsCache(); return; } await withResolvedDirLock(dir, async () => { @@ -1753,17 +1721,14 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - if (liveTurnRefs !== null || stagedAppend !== null) { - const baseline = stagedAppend ?? liveTurnRefs; - if ( - baseline !== null && - refPrefixLength(baseline, turns) < baseline.length - ) { - stagedAppend = null; + if (liveTurnRefs !== null) { + if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { unpublishedRewrite = [...turns]; return; } - await noteAppend(turns); + await writeTurnLog(turns); + liveTurnRefs = [...turns]; + invalidateTurnsCache(); return; } const decision = await resolveFirstWrite(turns); @@ -1776,7 +1741,9 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await noteAppend(turns); + await writeTurnLog(turns); + liveTurnRefs = [...turns]; + invalidateTurnsCache(); }); } @@ -1858,19 +1825,6 @@ export async function createSessionStores( )) { approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); } - if (stagedAppend !== null && unpublishedRewrite === null) { - await flushStagedAppend(); - try { - liveLog = await readLiveLogCached(); - } catch (cause) { - throw new Error( - `failed to load ${TURNS_FILE}: ${ - cause instanceof Error ? cause.message : String(cause) - }`, - { cause }, - ); - } - } const { turns } = liveLog; return { turns, @@ -1983,13 +1937,6 @@ export async function createSessionStores( async commit(options, signal) { return withResolvedDirLock(dir, async () => { const stagedRewrite = unpublishedRewrite; - // An append the commit cycle did not flush (no writeMetadata) still - // has to land before git add. A rewrite stays staged until the - // snapshot below, then one flush — commit does not flush again after - // writeMetadata already did. - if (stagedAppend !== null && stagedRewrite === null) { - await flushStagedAppend(); - } // The staged rewrite lands on the working-tree segments before the git // operations below; snapshot them so a failed commit can restore the // published generation. The rewrite stays staged so a retry can still @@ -2174,6 +2121,7 @@ export async function createSessionStores( turnsBaseName, ); liveTurnRefs = null; + canonicalLines = null; // The restore rewrote the published generation; the cached // live-turns parse predates it in general, so rebuild from disk. invalidateTurnsCache(); diff --git a/src/session/pending-approval-log.test.ts b/src/session/pending-approval-log.test.ts index cba8f1702..c9cd07481 100644 --- a/src/session/pending-approval-log.test.ts +++ b/src/session/pending-approval-log.test.ts @@ -277,19 +277,28 @@ describe("approval records in the turn log", () => { }); describe("a parked approval in the turn log", () => { - test("writeTurns stages the parked call and writeMetadata flushes it once", async () => { + test("writeTurns is on disk before writeMetadata, which only appends the record", async () => { const dir = tempDir(); const store = await createOptimizedContextStore(dir); await store.writeTurns([user, assistant]); - expect(fs.existsSync(path.join(dir, "turns.jsonl"))).toBe(false); + const afterTurns = fs.readFileSync(path.join(dir, "turns.jsonl")); + expect(lineKinds(dir)).toEqual(["turn", "turn"]); + const early = await (await createOptimizedContextStore(dir)).load(); + expect(early.turns).toEqual([user, assistant]); + expect(early.pendingOperations).toEqual([]); await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); + const afterRecord = fs.readFileSync(path.join(dir, "turns.jsonl")); + expect(afterRecord.subarray(0, afterTurns.length)).toEqual(afterTurns); expect(lineKinds(dir)).toEqual(["turn", "turn", "record"]); expect(fs.existsSync(metadataPath(dir))).toBe(false); - const flushed = fs.readFileSync(path.join(dir, "turns.jsonl")); + + // No parked approval: the segment is left alone. + await store.writeMetadata({ pendingOperations: [], tokenUsage: USAGE }); + expect(fs.readFileSync(path.join(dir, "turns.jsonl"))).toEqual(afterRecord); await store.commit({ message: "suspend" }); - expect(fs.readFileSync(path.join(dir, "turns.jsonl"))).toEqual(flushed); + expect(fs.readFileSync(path.join(dir, "turns.jsonl"))).toEqual(afterRecord); expect(await openOps(dir)).toEqual([op.correlationId]); }); @@ -483,17 +492,19 @@ describe("a parked approval across a compaction", () => { await store.writeTurns(folded); await store.writeMetadata({ pendingOperations: [op], tokenUsage: USAGE }); - // The rewrite is still unpublished. The approval is not on metadata.json; - // it rides the staged rewrite and lands when that commit flushes. - expect(recordedApprovals(dir)).toEqual([]); + // The published log still has the call, so the record is on it before + // commit. A failed rewrite must not drop it, and must not use metadata.json. + expect(recordedApprovals(dir)).toHaveLength(1); expect(fs.existsSync(metadataPath(dir))).toBe(false); + expect(await openOps(dir)).toEqual([op.correlationId]); await failingCommit(dir, () => store.commit({ message: "compact" })); - expect(recordedApprovals(dir)).toEqual([]); - expect(await openOps(dir)).toEqual([]); + expect(recordedApprovals(dir)).toHaveLength(1); + expect(await openOps(dir)).toEqual([op.correlationId]); await store.commit({ message: "retry compact" }); expect(recordFollowsCall(dir, "call-1")).toBe(true); + expect(recordedApprovals(dir)).toHaveLength(1); expect(await openOps(dir)).toEqual([op.correlationId]); expect(fs.existsSync(metadataPath(dir))).toBe(false); expect(await headPaths(dir)).not.toContain("metadata.json");