diff --git a/packages/agent/docs/harness-v2.md b/packages/agent/docs/harness-v2.md index 9c0cd4fdd1e..71d65eda0e4 100644 --- a/packages/agent/docs/harness-v2.md +++ b/packages/agent/docs/harness-v2.md @@ -1619,8 +1619,8 @@ Contract rules, all backends: - `Session` and the harness provision ids with `session.idGenerator`; storage enforces per-session uniqueness at append. - Every durable payload must be JSON-serializable. `Session` validates before dispatch so Memory, JSONL, and SQLite accept the same values; Memory does not retain values JSONL would reject. - Reads return immutable data. -- `findOpenOperations` is a required recovery projection: Memory maintains it with its record state, JSONL derives it while replaying the file, and SQLite answers it with indexed records. It returns unfinished starts newest first and must expose a second result so recovery can reject multiple open operations. -- No conditional writes exist. Single-writer plus the lane mutation line make compare-and-set unnecessary; storage stays plain appends and pointer/fact updates. +- `findOpenOperations` is a required recovery projection: Memory maintains it with its record state, JSONL derives it while replaying the file, and SQLite answers it from the lane's current open-operation projection. It returns unfinished starts newest first and must expose a second result when a replayed/imported backend observes multiple open operations so recovery can reject corruption. Backends with conditional current-state projections may reject a second `operation_started` append instead of creating that corruption through their normal write API. +- No general conditional writes exist. Single-writer plus the lane mutation line make compare-and-set unnecessary for normal appends and pointer/fact updates. The lane open-operation projection is the narrow exception: starting an operation conditionally sets the lane's open operation from `null` to the run id, and a failed update means the lane is already busy. - One writer per session, enforced by the serving layer; SQLite additionally rejects a second writer itself. Per session, not per backend: one SQLite database hosts many sessions, each with its own single writer. - Any write failure faults the harness (section 4). The store is left a valid prefix. - Global-fact and lane-move history is kept, never rewritten: latest by `seq` wins. History is the cheaper implementation (insert, never update), and lane-move history is a reflog if anyone ever wants one. @@ -1680,7 +1680,7 @@ Greenfield schema; existing WIP databases are discarded. The engine design is th session_sequences (session_id, next_seq) -- atomic seq allocator entries (session_id, seq, id, parent_id, type, timestamp, payload) records (session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload) -lanes (session_id, lane, leaf_id) -- current pointer per lane +lanes (session_id, lane, leaf_id, open_operation_id) -- current pointer + open op projection lane_moves (session_id, seq, lane, leaf_id) -- history; getLog parity facts (session_id, seq, kind, key, value) -- name, labels; latest by seq branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) @@ -1689,7 +1689,6 @@ writer_leases (session_id, owner_id, fence, expires_at_ms) -- writer claim -- indexes records: (session_id, lane, type, seq), (session_id, lane, type, op_kind, seq) - (session_id, lane, run_id, type) branch_entries: (session_id, branch_id, entry_type, entry_seq) (session_id, entry_id) -- reverse lookup: entry → branches ``` @@ -3015,8 +3014,8 @@ These packages merge R0 → R1 → R2 → R3. R1 and R2 add a reducer module ins - [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. + - 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 from the lane open-operation projection. + - Prove that zero/one open operations are distinguishable, that normal writes cannot start a second operation on a busy lane, and that the latest run-kind start is an indexed query. Add the lane open-operation projection. - Acceptance: memory and SQLite have identical query behavior, invalid query combinations reject, and no restore algorithm needs a full historical scan. - [x] **R1 — pure record-log validity.** Dependencies: R0. diff --git a/packages/agent/src/harness/session/jsonl/storage.ts b/packages/agent/src/harness/session/jsonl/storage.ts index a74bf81358e..236dfddecec 100644 --- a/packages/agent/src/harness/session/jsonl/storage.ts +++ b/packages/agent/src/harness/session/jsonl/storage.ts @@ -132,6 +132,13 @@ export class JsonlSessionStorage implements SessionStorage return this.enqueue(async () => { this.state.requireLane(newRecord.lane); this.state.validateUnusedId(newRecord.id); + const currentOpenOperationId = this.state.findOpenOperations(newRecord.lane, { limit: 1 })[0]?.id; + if (newRecord.type === "operation_started" && currentOpenOperationId !== undefined) { + throw new SessionError( + "storage", + `Lane ${newRecord.lane} already has an open operation ${currentOpenOperationId}`, + ); + } const record = { ...structuredClone(newRecord), seq: this.state.nextSequence, diff --git a/packages/agent/src/harness/session/memory.ts b/packages/agent/src/harness/session/memory.ts index b270500863d..5ec394737ac 100644 --- a/packages/agent/src/harness/session/memory.ts +++ b/packages/agent/src/harness/session/memory.ts @@ -119,6 +119,13 @@ export class InMemorySessionStorage implements SessionStorage { async appendRecord(newRecord: NewRecord): Promise { this.state.requireLane(newRecord.lane); this.state.validateUnusedId(newRecord.id); + const currentOpenOperationId = this.state.findOpenOperations(newRecord.lane, { limit: 1 })[0]?.id; + if (newRecord.type === "operation_started" && currentOpenOperationId !== undefined) { + throw new SessionError( + "storage", + `Lane ${newRecord.lane} already has an open operation ${currentOpenOperationId}`, + ); + } const record = { ...structuredClone(newRecord), seq: this.state.nextSequence, diff --git a/packages/agent/src/harness/session/testing/conformance.ts b/packages/agent/src/harness/session/testing/conformance.ts index 5e1798f1024..be33b221308 100644 --- a/packages/agent/src/harness/session/testing/conformance.ts +++ b/packages/agent/src/harness/session/testing/conformance.ts @@ -416,18 +416,27 @@ export function createSessionBackendConformance( 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", + type: "operation_finished", + id: "run-old-finished", lane: "main", - sourceLeafId: null, - intent: { kind: "compaction", resultEntryId: "compaction-result" }, + runId: "run-old", + outcome: "completed", }); + await session.appendRecord(operationStarted("compaction", { lane: "main", kind: "compaction" })); await session.appendRecord({ - type: "operation_started", - id: "navigation", + type: "operation_finished", + id: "compaction-finished", lane: "main", - sourceLeafId: null, - intent: { kind: "navigation", targetId: null, summarize: false }, + runId: "compaction", + outcome: "completed", + }); + await session.appendRecord(operationStarted("navigation", { lane: "main", kind: "navigation" })); + await session.appendRecord({ + type: "operation_finished", + id: "navigation-finished", + lane: "main", + runId: "navigation", + outcome: "completed", }); await session.appendRecord(operationStarted("run-new", { lane: "main", kind: "run" })); @@ -471,72 +480,56 @@ export function createSessionBackendConformance( ); }), - 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]); + createCase(factory, "records and log", "tracks and enforces one open operation per lane", async (repository) => { + const session = await repository.create({ id: "session" }); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), []); - await session.appendRecord({ - type: "operation_finished", - id: "finish-first", - lane: "main", - runId: first.id, - outcome: "completed", - }); - deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [second]); + const first = await session.appendRecord(operationStarted("first", { lane: "main", kind: "run" })); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [first]); + await rejectsWithCode( + session.appendRecord(operationStarted("second", { lane: "main", kind: "run" })), + "storage", + ); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [first]); - await session.appendRecord({ - type: "operation_finished", - id: "finish-second", - lane: "main", - runId: second.id, - outcome: "failed", - }); - deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), []); - }, - ), + await session.appendRecord({ + type: "operation_finished", + id: "finish-first", + lane: "main", + runId: first.id, + outcome: "completed", + }); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), []); + }), createCase( factory, "records and log", - "does not let an earlier finish close a later operation start", + "does not let an earlier finish close a later start", async (repository) => { const session = await repository.create({ id: "session" }); await session.appendRecord({ type: "operation_finished", - id: "early-finish", + id: "finish-before-start", lane: "main", - runId: "late-start", + runId: "run", outcome: "completed", }); - const started = await session.appendRecord(operationStarted("late-start", { lane: "main", kind: "run" })); - - deepStrictEqual(await session.findOpenOperations("main"), [started]); + const started = await session.appendRecord(operationStarted("run", { lane: "main", kind: "run" })); + deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [started]); }, ), - createCase(factory, "records and log", "scopes open operations by lane kind and limit", async (repository) => { + createCase(factory, "records and log", "scopes open operations by lane 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("main"), [mainRun]); + deepStrictEqual(await session.findOpenOperations("main", { limit: 1 }), [mainRun]); deepStrictEqual(await session.findOpenOperations("thread", { limit: 2 }), [threadNavigation]); }), 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 9fbc42434d2..e400a4367c7 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 @@ -61,11 +61,10 @@ CREATE TABLE IF NOT EXISTS lanes ( session_id TEXT NOT NULL, lane TEXT NOT NULL, leaf_id TEXT NULL, + open_operation_id TEXT NULL, PRIMARY KEY (session_id, lane) ) WITHOUT ROWID; -CREATE INDEX IF NOT EXISTS idx_lanes_session_leaf ON lanes(session_id, leaf_id); - CREATE TABLE IF NOT EXISTS records ( session_id TEXT NOT NULL, seq INTEGER NOT NULL, @@ -83,7 +82,6 @@ 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 e261c0c4175..d75c7a5abe1 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/repo.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/repo.ts @@ -35,12 +35,14 @@ import { appendFact, deleteFactRows, readFactRows, readLatestFact, readLatestLab import { createInitialLane, deleteLaneRows, + finishLaneOperation, createLane as insertLane, readLane, readLaneHead, readLaneMoveRows, readLanes, setLaneLeaf, + startLaneOperation, moveLane as updateLane, } from "./storage/lanes.ts"; import { @@ -496,6 +498,9 @@ class SqliteSessionStorage implements SessionStorage { assertUnusedId(this.db, this.metadata.id, record.id); const seq = getNextSequence(this.db, this.metadata.id); const committed: LaneRecord = { ...record, seq, timestamp: Date.now() }; + if (record.type === "operation_started") { + startLaneOperation(this.db, this.metadata.id, record.lane, record.id); + } appendRecordRow(this.db, this.metadata.id, { seq, id: record.id, @@ -506,6 +511,9 @@ class SqliteSessionStorage implements SessionStorage { timestamp: timestampToText(committed.timestamp), payload: JSON.stringify(record), }); + if (record.type === "operation_finished") { + finishLaneOperation(this.db, this.metadata.id, record.lane, record.runId); + } if (record.type === "usage") addUsageToStats(this.db, this.metadata.id, record.usage); advanceSequence(this.db, this.metadata.id, seq); return structuredClone(committed); @@ -546,6 +554,7 @@ class SqliteSessionStorage implements SessionStorage { 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") { diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts index 645d9179079..d6f54a3a585 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts @@ -5,6 +5,7 @@ export interface LaneRow { session_id: string; lane: string; leaf_id: string | null; + open_operation_id: string | null; } export interface LaneMoveRow { @@ -15,7 +16,11 @@ export interface LaneMoveRow { } export function createInitialLane(db: SqliteDatabase, sessionId: string, lane = "main", leafId: string | null = null) { - db.prepare("INSERT INTO lanes (session_id, lane, leaf_id) VALUES (?, ?, ?)").run(sessionId, lane, leafId); + db.prepare("INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) VALUES (?, ?, ?, NULL)").run( + sessionId, + lane, + leafId, + ); } export function readLanes(db: SqliteDatabase, sessionId: string) { @@ -25,6 +30,7 @@ export function readLanes(db: SqliteDatabase, sessionId: string) { l.session_id, l.lane, l.leaf_id, + l.open_operation_id, (l.leaf_id IS NULL OR EXISTS ( SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id )) AS leaf_exists @@ -38,12 +44,17 @@ export function readLanes(db: SqliteDatabase, sessionId: string) { throw new SessionError("storage", `Lane ${row.lane} points at missing entry ${row.leaf_id}`); } } - return rows.map(({ session_id, lane, leaf_id }) => ({ session_id, lane, leaf_id })); + return rows.map(({ session_id, lane, leaf_id, open_operation_id }) => ({ + session_id, + lane, + leaf_id, + open_operation_id, + })); } export function readLane(db: SqliteDatabase, sessionId: string, lane: string) { return db - .prepare("SELECT session_id, lane, leaf_id FROM lanes WHERE session_id = ? AND lane = ?") + .prepare("SELECT session_id, lane, leaf_id, open_operation_id FROM lanes WHERE session_id = ? AND lane = ?") .get(sessionId, lane); } @@ -65,7 +76,11 @@ export function readLaneHead(db: SqliteDatabase, sessionId: string, lane: string } export function createLane(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) { - db.prepare("INSERT INTO lanes (session_id, lane, leaf_id) VALUES (?, ?, ?)").run(sessionId, lane, leafId); + db.prepare("INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) VALUES (?, ?, ?, NULL)").run( + sessionId, + lane, + leafId, + ); appendLaneMove(db, sessionId, seq, lane, leafId); } @@ -84,6 +99,22 @@ export function setLaneLeaf(db: SqliteDatabase, sessionId: string, lane: string, if (result.changes !== 1) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); } +export function startLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) { + const result = db + .prepare("UPDATE lanes SET open_operation_id = ? WHERE session_id = ? AND lane = ? AND open_operation_id IS NULL") + .run(runId, sessionId, lane); + if (result.changes === 1) return; + const current = readLane(db, sessionId, lane); + if (!current) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); + throw new SessionError("storage", `Lane ${lane} already has an open operation ${current.open_operation_id}`); +} + +export function finishLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) { + db.prepare( + "UPDATE lanes SET open_operation_id = NULL WHERE session_id = ? AND lane = ? AND open_operation_id = ?", + ).run(sessionId, lane, runId); +} + export function readLaneMoveRows(db: SqliteDatabase, sessionId: string, options: { afterSeq?: number } = {}) { const predicates = ["session_id = ?"]; const params: unknown[] = [sessionId]; 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 a14f90d25cc..c68276877ce 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts @@ -1,3 +1,4 @@ +import { SessionError } from "@earendil-works/pi-agent-core"; import type { SqliteDatabase } from "../types.ts"; export interface RecordRow { @@ -103,29 +104,26 @@ export function readOpenOperationRows( db: SqliteDatabase, sessionId: string, lane: string, - options: { limit?: number } = {}, + _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 + const laneRow = db + .prepare("SELECT open_operation_id FROM lanes WHERE session_id = ? AND lane = ?") + .get<{ open_operation_id: string | null }>(sessionId, lane); + if (!laneRow?.open_operation_id) return []; + + const record = 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}`, + `SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload + FROM records + WHERE session_id = ? + AND id = ?`, ) - .all(...params); + .get(sessionId, laneRow.open_operation_id); + if (!record) { + throw new SessionError("storage", `Lane ${lane} points at missing open operation ${laneRow.open_operation_id}`); + } + if (record.lane !== lane || record.type !== "operation_started") { + throw new SessionError("storage", `Lane ${lane} points at invalid open operation ${laneRow.open_operation_id}`); + } + return [record]; } diff --git a/packages/session-backends/sqlite-node/test/migrations.test.ts b/packages/session-backends/sqlite-node/test/migrations.test.ts index 07c215d4d52..5d59f55d613 100644 --- a/packages/session-backends/sqlite-node/test/migrations.test.ts +++ b/packages/session-backends/sqlite-node/test/migrations.test.ts @@ -34,8 +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"); + const laneColumns = db.prepare("PRAGMA table_info(lanes)").all<{ name: string }>(); + expect(laneColumns.map((column) => column.name)).toContain("open_operation_id"); } finally { db.close(); }