diff --git a/packages/agent/docs/harness-v2.md b/packages/agent/docs/harness-v2.md index e48bba502..911689ac8 100644 --- a/packages/agent/docs/harness-v2.md +++ b/packages/agent/docs/harness-v2.md @@ -2974,9 +2974,7 @@ These packages merge QA1 → QA2 → QA3. QA packages own the audit document and These packages merge R0 → R1 → R2 → R3. R1 and R2 add a reducer module instead of growing `agent-harness.ts`. R3 is the first package in this track that owns `agent-harness.ts` and therefore runs after F0. -**In progress and reserved: R0 by @vegarsti.** Other agents must not pick R0 while this ownership marker remains. - -- [ ] **R0 — recovery-query contract.** Dependencies: none. +- [x] **R0 — recovery-query contract.** Dependencies: none. - Primary files: `packages/agent/src/harness/session/types.ts`, `session.ts`, `memory.ts`, SQLite record storage/repository files, backend conformance, and focused recovery-query tests. - Add `RecordQuery.operationKind` and `findOpenOperations(lane, { limit })` exactly as specified in sections 7, 12, and 13. Memory maintains the projection, JSONL will derive it during replay, and SQLite answers it with indexed records. - Prove that zero/one open operations are distinguishable from multiple-open-operation corruption, and that the latest run-kind start is an indexed query. Add the SQLite `(session_id, lane, run_id, type)` index. diff --git a/packages/agent/src/harness/session/memory.ts b/packages/agent/src/harness/session/memory.ts index 6e2525140..1ff1cd96f 100644 --- a/packages/agent/src/harness/session/memory.ts +++ b/packages/agent/src/harness/session/memory.ts @@ -11,6 +11,7 @@ import { type LogItem, type LogOptions, type NewRecord, + type OperationStartedRecord, type ProvisionedEntry, type RecordQuery, type SessionCreateOptions, @@ -50,6 +51,7 @@ export class InMemorySessionStorage implements SessionStorage { private readonly entries: Entry[] = []; private readonly entriesById = new Map(); private readonly records: LaneRecord[] = []; + private readonly openOperationsByLane = new Map>(); private readonly lanes = new Map([["main", null]]); private readonly log: LogItem[] = []; private readonly stats: SessionStats = { @@ -156,6 +158,17 @@ export class InMemorySessionStorage implements SessionStorage { const record = provisionRecord(clonedRecord, this.nextSequence()); this.usedIds.add(record.id); this.records.push(record); + // Maintain the recovery projection incrementally so restore never scans the full record log. + if (record.type === "operation_started") { + let openOperations = this.openOperationsByLane.get(record.lane); + if (!openOperations) { + openOperations = new Map(); + this.openOperationsByLane.set(record.lane, openOperations); + } + openOperations.set(record.id, record); + } else if (record.type === "operation_finished") { + this.openOperationsByLane.get(record.lane)?.delete(record.runId); + } this.log.push({ kind: "record", seq: record.seq, record }); if (record.type === "usage") { this.stats.cachedTokens += record.usage.cacheRead; @@ -212,6 +225,12 @@ export class InMemorySessionStorage implements SessionStorage { return structuredClone(results); } + async findOpenOperations(lane: string, options?: { limit?: number }): Promise { + const openOperationsById = this.openOperationsByLane.get(lane); + const openOperations = openOperationsById ? [...openOperationsById.values()].reverse() : []; + return structuredClone(options?.limit === undefined ? openOperations : openOperations.slice(0, options.limit)); + } + async getLog(options: LogOptions = {}): Promise { const results: LogItem[] = []; for (const item of this.log) { @@ -305,6 +324,8 @@ export class InMemorySessionStorage implements SessionStorage { (record.type === "operation_started" ? record.id === query.runId : "runId" in record && record.runId === query.runId)) && + (query.operationKind === undefined || + (record.type === "operation_started" && record.intent.kind === query.operationKind)) && (query.afterSeq === undefined || record.seq > query.afterSeq) ); } diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index d57d5a917..262cdcd8c 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -10,6 +10,7 @@ import type { LogItem, LogOptions, NewRecord, + OperationStartedRecord, ProvisionedEntry, RecordBase, RecordQuery, @@ -213,12 +214,17 @@ export class Session implem return this.queryRecords(query); } + async findOpenOperations(lane: string, options?: { limit?: number }): Promise { + assertValidLimit(options?.limit); + return this.storage.findOpenOperations(lane, options); + } + async getLog(options?: LogOptions): Promise { return this.queryLog(options); } private async getLeafIdForLane(lane: string): Promise { - const pointer = (await this.storage.getLanes()).find((candidate) => candidate.lane === lane); + const pointer = (await this.getLanes()).find((candidate) => candidate.lane === lane); if (!pointer) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); return pointer.leafId; } @@ -245,6 +251,9 @@ export class Session implem private async queryRecords(query: RecordQuery = {}): Promise { assertValidLimit(query.limit); assertValidCursor(query.afterSeq); + if (query.operationKind !== undefined && query.type !== "operation_started") { + throw new SessionError("invalid_query", 'operationKind requires type "operation_started"'); + } return this.storage.findRecords(query); } diff --git a/packages/agent/src/harness/session/testing/conformance.ts b/packages/agent/src/harness/session/testing/conformance.ts index f18ed25a3..c158f6036 100644 --- a/packages/agent/src/harness/session/testing/conformance.ts +++ b/packages/agent/src/harness/session/testing/conformance.ts @@ -39,14 +39,23 @@ function createAssistantMessage(text: string): AgentMessage { }; } -function operationStarted(id: string, lane = "main"): NewRecord { - return { - type: "operation_started", - id, - lane, - sourceLeafId: null, - intent: { kind: "run", originalPrompt: [], initialMessages: [] }, - }; +function operationStarted( + id: string, + { lane, kind }: { lane: string; kind: OperationStartedRecord["intent"]["kind"] }, +): NewRecord { + let intent: OperationStartedRecord["intent"]; + switch (kind) { + case "run": + intent = { kind, originalPrompt: [], initialMessages: [] }; + break; + case "compaction": + intent = { kind, resultEntryId: `${id}-result` }; + break; + case "navigation": + intent = { kind, targetId: null, summarize: false }; + break; + } + return { type: "operation_started", id, lane, sourceLeafId: null, intent }; } async function entryIds(entries: Promise): Promise { @@ -99,7 +108,7 @@ export function createSessionBackendConformance( { type: "custom", id: "child", customType: "note", data: { value: 1 } }, "thread", ); - const record = await session.appendRecord(operationStarted("run", "thread")); + const record = await session.appendRecord(operationStarted("run", { lane: "thread", kind: "run" })); await session.setName("Example"); await session.setLabel(root.id, "checkpoint"); await session.moveLane("main", child.id); @@ -175,8 +184,11 @@ export function createSessionBackendConformance( { type: "message", id: "shared", message: createUserMessage("root") }, "main", ); - await rejectsWithCode(session.appendRecord(operationStarted("shared")), "already_exists"); - await session.appendRecord(operationStarted("run")); + await rejectsWithCode( + session.appendRecord(operationStarted("shared", { lane: "main", kind: "run" })), + "already_exists", + ); + await session.appendRecord(operationStarted("run", { lane: "main", kind: "run" })); await rejectsWithCode( session.appendEntry({ type: "custom", id: "run", customType: "note" }, "main"), "already_exists", @@ -228,6 +240,10 @@ export function createSessionBackendConformance( await rejectsWithCode(thread.findEntriesOnBranch({ cursor: { afterSeq: -1 } }), "invalid_query"); await rejectsWithCode(thread.findEntryOnBranch({ limit: 0 }), "invalid_query"); await rejectsWithCode(session.findRecords({ limit: 0 }), "invalid_query"); + await rejectsWithCode(session.findRecords({ operationKind: "run" }), "invalid_query"); + await rejectsWithCode(session.findRecords({ type: "step_attempt", operationKind: "run" }), "invalid_query"); + await rejectsWithCode(session.findOpenOperations("main", { limit: 0 }), "invalid_query"); + await rejectsWithCode(session.findOpenOperations("main", { limit: -1 }), "invalid_query"); await rejectsWithCode(session.getLog({ afterSeq: -1 }), "invalid_query"); }), @@ -296,7 +312,7 @@ export function createSessionBackendConformance( async (repository) => { const session = await repository.create({ id: "session" }); await session.createLane("thread", null); - await session.appendRecord(operationStarted("old-run", "thread")); + await session.appendRecord(operationStarted("old-run", { lane: "thread", kind: "run" })); await session.appendRecord({ type: "queue_enqueued", id: "old-next-run", @@ -355,7 +371,7 @@ export function createSessionBackendConformance( "filters records by lane type run sequence and order", async (repository) => { const session = await repository.create({ id: "session" }); - await session.appendRecord(operationStarted("run-1")); + await session.appendRecord(operationStarted("run-1", { lane: "main", kind: "run" })); await session.appendRecord({ type: "step_attempt", id: "attempt-1", @@ -366,7 +382,7 @@ export function createSessionBackendConformance( resultEntryId: "assistant-1", }); await session.createLane("thread", null); - await session.appendRecord(operationStarted("run-2", "thread")); + await session.appendRecord(operationStarted("run-2", { lane: "thread", kind: "run" })); await session.appendRecord({ type: "step_attempt", id: "attempt-2", @@ -396,6 +412,149 @@ export function createSessionBackendConformance( }, ), + createCase(factory, "records and log", "filters operation starts by operation kind", async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendRecord(operationStarted("run-old", { lane: "main", kind: "run" })); + await session.appendRecord({ + type: "operation_started", + id: "compaction", + lane: "main", + sourceLeafId: null, + intent: { kind: "compaction", resultEntryId: "compaction-result" }, + }); + await session.appendRecord({ + type: "operation_started", + id: "navigation", + lane: "main", + sourceLeafId: null, + intent: { kind: "navigation", targetId: null, summarize: false }, + }); + await session.appendRecord(operationStarted("run-new", { lane: "main", kind: "run" })); + + deepStrictEqual( + ( + await session.findRecords({ + type: "operation_started", + operationKind: "run", + order: "oldestFirst", + }) + ).map((record) => record.id), + ["run-old", "run-new"], + ); + deepStrictEqual( + ( + await session.findRecords({ + type: "operation_started", + operationKind: "compaction", + }) + ).map((record) => record.id), + ["compaction"], + ); + deepStrictEqual( + ( + await session.findRecords({ + type: "operation_started", + operationKind: "navigation", + }) + ).map((record) => record.id), + ["navigation"], + ); + deepStrictEqual( + ( + await session.findRecords({ + type: "operation_started", + operationKind: "run", + limit: 1, + }) + ).map((record) => record.id), + ["run-new"], + ); + }), + + createCase( + factory, + "records and log", + "distinguishes zero one and multiple open operations", + async (repository) => { + const session = await repository.create({ id: "session" }); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), []); + + const first = await session.appendRecord(operationStarted("first", { lane: "main", kind: "run" })); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [first]); + + const second = await session.appendRecord(operationStarted("second", { lane: "main", kind: "run" })); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [second, first]); + + await session.appendRecord({ + type: "operation_finished", + id: "finish-first", + lane: "main", + runId: first.id, + outcome: "completed", + }); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [second]); + + await session.appendRecord({ + type: "operation_finished", + id: "finish-second", + lane: "main", + runId: second.id, + outcome: "failed", + }); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), []); + }, + ), + + createCase( + factory, + "records and log", + "does not let an earlier finish close a later operation start", + async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendRecord({ + type: "operation_finished", + id: "early-finish", + lane: "main", + runId: "late-start", + outcome: "completed", + }); + const started = await session.appendRecord(operationStarted("late-start", { lane: "main", kind: "run" })); + + deepStrictEqual(await session.findOpenOperations("main"), [started]); + }, + ), + + createCase(factory, "records and log", "scopes open operations by lane kind and limit", async (repository) => { + const session = await repository.create({ id: "session" }); + await session.createLane("thread", null); + const mainRun = await session.appendRecord(operationStarted("main-run", { lane: "main", kind: "run" })); + const mainCompaction = await session.appendRecord( + operationStarted("main-compaction", { lane: "main", kind: "compaction" }), + ); + const threadNavigation = await session.appendRecord( + operationStarted("thread-navigation", { lane: "thread", kind: "navigation" }), + ); + + deepStrictEqual(await session.findOpenOperations("main"), [mainCompaction, mainRun]); + deepStrictEqual(await session.findOpenOperations("main", { limit: 1 }), [mainCompaction]); + deepStrictEqual(await session.findOpenOperations("thread", { limit: 2 }), [threadNavigation]); + }), + + createCase( + factory, + "validation and immutability", + "returns immutable open-operation records", + async (repository) => { + const session = await repository.create({ id: "session" }); + const committed = await session.appendRecord(operationStarted("run", { lane: "main", kind: "run" })); + const [read] = await session.findOpenOperations("main"); + if (read?.intent.kind !== "run") throw new Error("Expected an open run operation"); + read.intent.originalPrompt.push(createUserMessage("mutated")); + + deepStrictEqual(await session.findOpenOperations("main"), [committed]); + }, + ), + createCase( factory, "queries and facts", @@ -645,7 +804,10 @@ export function createSessionBackendConformance( deepStrictEqual(await session.findRecords(), []); deepStrictEqual(await session.getLog(), []); - strictEqual((await session.appendRecord(operationStarted("valid-record"))).seq, 1); + strictEqual( + (await session.appendRecord(operationStarted("valid-record", { lane: "main", kind: "run" }))).seq, + 1, + ); }, ), @@ -720,7 +882,7 @@ export function createSessionBackendConformance( await source.setName("Source"); await source.setLabel(shared, "copied"); await source.setLabel(threadChild, "excluded"); - await source.appendRecord(operationStarted("run")); + await source.appendRecord(operationStarted("run", { lane: "main", kind: "run" })); await source.appendRecord({ type: "usage", id: "source-usage", diff --git a/packages/agent/src/harness/session/types.ts b/packages/agent/src/harness/session/types.ts index 2e466b950..281c5b506 100644 --- a/packages/agent/src/harness/session/types.ts +++ b/packages/agent/src/harness/session/types.ts @@ -238,6 +238,8 @@ export interface RecordQuery { lane?: string; type?: LaneRecord["type"]; runId?: string; + /** Valid only with type "operation_started". */ + operationKind?: OperationStartedRecord["intent"]["kind"]; afterSeq?: number; order?: EntryOrder; limit?: number; @@ -295,6 +297,7 @@ export interface SessionStorage[]>; findRecords(query?: RecordQuery): Promise; + findOpenOperations(lane: string, options?: { limit?: number }): Promise; getLog(options?: { afterSeq?: number; limit?: number }): Promise; // Global facts diff --git a/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql b/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql index 3937389a4..d68cbdd06 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql +++ b/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql @@ -83,6 +83,7 @@ CREATE TABLE IF NOT EXISTS records ( CREATE INDEX IF NOT EXISTS idx_records_session_seq ON records(session_id, seq); CREATE INDEX IF NOT EXISTS idx_records_session_lane_type_seq ON records(session_id, lane, type, seq); CREATE INDEX IF NOT EXISTS idx_records_session_lane_type_op_kind_seq ON records(session_id, lane, type, op_kind, seq); +CREATE INDEX IF NOT EXISTS idx_records_session_lane_run_id_type ON records(session_id, lane, run_id, type); CREATE INDEX IF NOT EXISTS idx_records_session_run_id_seq ON records(session_id, run_id, seq); CREATE TABLE IF NOT EXISTS lane_moves ( diff --git a/packages/session-backends/sqlite-node/src/sqlite/repo.ts b/packages/session-backends/sqlite-node/src/sqlite/repo.ts index 8afabf003..3ea21426f 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/repo.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/repo.ts @@ -8,6 +8,7 @@ import { type LogItem, type LogOptions, type NewRecord, + type OperationStartedRecord, type ProvisionedEntry, type RecordQuery, Session, @@ -49,7 +50,13 @@ import { renewSessionLease, type SessionLease, } from "./storage/leases.ts"; -import { appendRecordRow, deleteRecordRows, idExistsInRecords, readRecordRows } from "./storage/records.ts"; +import { + appendRecordRow, + deleteRecordRows, + idExistsInRecords, + readOpenOperationRows, + readRecordRows, +} from "./storage/records.ts"; import { advanceSequence, createSequence, @@ -510,6 +517,8 @@ class SqliteSessionStorage implements SessionStorage { return row ? decodeEntry(row) : undefined; } + // TODO: Remove redundant structuredClone calls from findEntries, findEntriesOnBranch, + // findRecords, and getLog; SQLite row decoding already returns fresh object graphs. async findEntries(query: EntryQuery = {}): Promise { const rows = readEntryRows(this.db, this.metadata.id, { order: query.order }); const entries = rows.map(decodeEntry).filter((entry) => matchesEntryQuery(entry, query)); @@ -537,6 +546,17 @@ class SqliteSessionStorage implements SessionStorage { return structuredClone(rows.map(decodeRecord)); } + async findOpenOperations(lane: string, options?: { limit?: number }): Promise { + const rows = readOpenOperationRows(this.db, this.metadata.id, lane, options); + return rows.map((row) => { + const record = decodeRecord(row); + if (record.type !== "operation_started") { + throw new SessionError("storage", "Expected operation_started record"); + } + return record; + }); + } + async getLog(options: LogOptions = {}): Promise { const afterSeq = options.afterSeq ?? 0; const entryRows = readEntryRows(this.db, this.metadata.id, { afterSeq, order: "oldestFirst" }); diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts index c615a5096..a14f90d25 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts @@ -58,6 +58,7 @@ export function readRecordRows( lane?: string; type?: string; runId?: string; + operationKind?: string; afterSeq?: number; order?: "newestFirst" | "oldestFirst"; limit?: number; @@ -77,6 +78,10 @@ export function readRecordRows( predicates.push("run_id = ?"); params.push(query.runId); } + if (query.operationKind !== undefined) { + predicates.push("op_kind = ?"); + params.push(query.operationKind); + } if (query.afterSeq !== undefined) { predicates.push("seq > ?"); params.push(query.afterSeq); @@ -93,3 +98,34 @@ export function readRecordRows( ) .all(...params); } + +export function readOpenOperationRows( + db: SqliteDatabase, + sessionId: string, + lane: string, + options: { limit?: number } = {}, +): RecordRow[] { + const params: unknown[] = [sessionId, lane]; + const limit = options.limit === undefined ? "" : " LIMIT ?"; + if (options.limit !== undefined) params.push(options.limit); + return db + .prepare( + `SELECT started.session_id, started.seq, started.id, started.lane, started.run_id, + started.type, started.op_kind, started.timestamp, started.payload + FROM records AS started + WHERE started.session_id = ? + AND started.lane = ? + AND started.type = 'operation_started' + AND NOT EXISTS ( + SELECT 1 + FROM records AS finished + WHERE finished.session_id = started.session_id + AND finished.lane = started.lane + AND finished.run_id = started.id + AND finished.type = 'operation_finished' + AND finished.seq > started.seq + ) + ORDER BY started.seq DESC${limit}`, + ) + .all(...params); +} diff --git a/packages/session-backends/sqlite-node/test/migrations.test.ts b/packages/session-backends/sqlite-node/test/migrations.test.ts index 051b985d8..79cbd5418 100644 --- a/packages/session-backends/sqlite-node/test/migrations.test.ts +++ b/packages/session-backends/sqlite-node/test/migrations.test.ts @@ -34,6 +34,8 @@ describe("SQLite migrations", () => { ); const sessionColumns = db.prepare("PRAGMA table_info(sessions)").all<{ name: string }>(); expect(sessionColumns.map((column) => column.name)).not.toContain("leaf_id"); + const recordIndexes = db.prepare("PRAGMA index_list(records)").all<{ name: string }>(); + expect(recordIndexes.map((index) => index.name)).toContain("idx_records_session_lane_run_id_type"); } finally { db.close(); }