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, + ); +}); diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index b31afd8c8..368b24ec0 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, @@ -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; } @@ -926,6 +945,125 @@ describe("concurrent sessions write disjoint transcripts", () => { ).toEqual(turnTexts(turnsB)); }); + test("a diverted write is not the log load() re-arms", 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); + // 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 () => { + 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 283ddca46..265dca56a 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,12 @@ 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, + pendingApprovalRecord, + type PendingApprovalRecord, + StoredPendingOperationSchema, +} from "./pending-approval-record.js"; const TURNS_FILE = "turns.jsonl"; const PROMPT_FILE = "prompt.jsonl"; @@ -180,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 @@ -218,6 +223,12 @@ function parseSegmentTurns( cause, }); } + // Approval records share the log but are never history. + const approval = parsePendingApprovalRecord(raw); + if (approval !== null) { + approvals?.push(approval); + continue; + } const result = ConversationTurnSchema(raw); if (result instanceof type.errors) { if (skipMalformed) { @@ -264,17 +275,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, @@ -509,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( @@ -1119,6 +1126,53 @@ async function managedPathsDiffer( return matrix.some(([, head, workdir]) => head !== workdir); } +/** + * 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 Map(); + const answered = new Set(); + for (const turn of turns) { + for (const block of turn.content) { + 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; + 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; +} + export interface SessionStores { storage: ContextStore; audit: AuditStore; @@ -1211,28 +1265,216 @@ 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 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(); + // 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) { + if (!approvalRecords.has(op.correlationId)) { + approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); + } + loggedApprovalIds.add(op.correlationId); + } + } + + function hasUnloggedApproval(): boolean { + for (const id of approvalRecords.keys()) { + if (!loggedApprovalIds.has(id)) 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[]): { + 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); + // 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); + canonicalLines = lines; + } + + // 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); + } + } + 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); + } + 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); + } + for (const record of pending) { + loggedApprovalIds.add(record.pendingApproval.correlationId); + } + invalidateTurnsCache(); + } + + 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; }): 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). Parked approvals never go into metadata.json. + for (const op of scrubPersistedPendingOperations( + metadata.pendingOperations.filter( + (pending) => + pending.suspendedCall !== undefined && + !approvalRecords.has(pending.correlationId), + ), + )) { + approvalRecords.set(op.correlationId, pendingApprovalRecord(op)); + } + // 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; + } + // 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 }); + pendingSegmentPaths.add(METADATA_FILE); + writtenMetadataJson = null; + } + 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( - 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; @@ -1279,9 +1521,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); } @@ -1304,9 +1546,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; @@ -1357,13 +1599,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 = @@ -1376,11 +1619,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 @@ -1396,22 +1639,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( @@ -1466,7 +1711,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeSegmented(writeTurnsSegmented, turns); + await writeTurnLog(turns); liveTurnRefs = [...turns]; invalidateTurnsCache(); return; @@ -1481,7 +1726,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeSegmented(writeTurnsSegmented, turns); + await writeTurnLog(turns); liveTurnRefs = [...turns]; invalidateTurnsCache(); return; @@ -1496,7 +1741,7 @@ export async function createSessionStores( unpublishedRewrite = [...turns]; return; } - await writeSegmented(writeTurnsSegmented, turns); + await writeTurnLog(turns); liveTurnRefs = [...turns]; invalidateTurnsCache(); }); @@ -1504,13 +1749,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. @@ -1526,8 +1774,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 = { @@ -1538,9 +1789,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( @@ -1562,7 +1813,29 @@ export async function createSessionStores( ); metadata = emptyMetadata(); } - return { turns, ...metadata }; + // 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)); + } + 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 @@ -1572,8 +1845,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; @@ -1671,11 +1953,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); @@ -1826,6 +2109,10 @@ export async function createSessionStores( } catch { // A partial restore must not mask the real commit error. } + // 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); } // Drop the writer's stale in-memory state so a retry rewrites from // scratch. @@ -1834,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 new file mode 100644 index 000000000..c9cd07481 --- /dev/null +++ b/src/session/pending-approval-log.test.ts @@ -0,0 +1,641 @@ +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, +} 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 }); +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(""); +} + +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); + }); +}); + +describe("a parked approval in the turn log", () => { + 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]); + 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); + + // 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(afterRecord); + 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]); + + 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" }); + 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([ + 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 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)).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"); + }); +}); + +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([]); + }); + + 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([]); + }); +}); diff --git a/src/session/pending-approval-record.ts b/src/session/pending-approval-record.ts new file mode 100644 index 000000000..021b6ea8b --- /dev/null +++ b/src/session/pending-approval-record.ts @@ -0,0 +1,50 @@ +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", +}); + +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, +): PendingOperation | null { + const result = PendingApprovalRecordSchema(raw); + if (result instanceof type.errors) return null; + return result.pendingApproval; +} 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 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; }