fix: project open operation id to lanes

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.

packages/agent/src/harness/session/jsonl/storage.ts

@@ -132,6 +132,13 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
 		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,

packages/agent/src/harness/session/memory.ts

@@ -119,6 +119,13 @@ export class InMemorySessionStorage implements SessionStorage {
 	async appendRecord<TRecord extends LaneRecord>(newRecord: NewRecord<TRecord>): Promise<TRecord> {
 		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,

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]);
 		}),
 

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 (

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<SqliteSessionMetadata> {
 			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<SqliteSessionMetadata> {
 				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<SqliteSessionMetadata> {
 
 	async findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]> {
 		const rows = readOpenOperationRows(this.db, this.metadata.id, lane, options);
+
 		return rows.map((row) => {
 			const record = decodeRecord(row);
 			if (record.type !== "operation_started") {

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<LaneRow>(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];