refactor(agent): derive recovery state from record queries

packages/agent/docs/harness-v2.md

@@ -647,13 +647,13 @@ Opening a session restores every lane independently. Restore reads; it never app
 
 Recovery starts with indexed discovery, not a full log scan:
 
-1. `findOpenOperations(lane, { limit: 2 })` returns unfinished `operation_started` records newest first. Zero means idle, one means suspended, and two means corruption. Backends must answer this from replayed/indexed operation state; callers cannot infer it from only the newest start.
-2. For an idle lane, one indexed query finds the newest run-kind `operation_started`, then filtered `queue_enqueued` / `queue_cancelled` queries above it reconstruct pending `nextRun` items. With no prior run, the same type-filtered queries read only pre-run queue state; unrelated usage adjustments are never scanned.
+1. `getLanes()` returns each lane's `openOperationId`. `null` means idle; otherwise `findRecords({ lane, type: "operation_started", runId: openOperationId, limit: 1 })` loads and validates the suspended operation's start. Storage updates the projection atomically with the start or finish record, rejects a second start while one is open, and validates the same invariant during replay.
+2. For an idle lane, one indexed `findRecords({ lane, type: "operation_started", operationKind: "run", limit: 1 })` query finds the newest run start, then filtered `queue_enqueued` / `queue_cancelled` queries above it reconstruct pending `nextRun` items. With no prior run, the same type-filtered queries read only pre-run queue state; unrelated usage adjustments are never scanned.
 3. For a suspended lane, the open operation selects two bounded payload reads:
    - **The lane's records** since that `operation_started`. Everything after the finish of the previous operation is irrelevant history.
    - **The lane's own entries**: the path from its leaf back to the operation's anchor (`sourceLeafId`). These are exactly the entries this operation appended.
 
-Reduction may additionally perform point lookups for provisioned entry ids and bounded branch lookups for effective model, thinking, and active-tool configuration at the operation anchor. These are indexed lookups, not extra history scans. Every scan is bounded by the open operation or the still-relevant idle queue, not by total session history or another lane's activity.
+Reduction may additionally perform point lookups for provisioned entry ids. Every scan is bounded by the open operation or the still-relevant idle queue, not by total session history or another lane's activity.
 
 An idle lane's remaining state is pending next-run queue items. Next-run messages can be enqueued at any time; only the acceptance of a run consumes them — compaction and navigation pass over the queue. Pending items are the `queue_enqueued` records after the lane's most recent run-kind `operation_started` whose provisioned entries do not exist and that no `queue_cancelled` retracts. Items a run captured are listed in its intent's `initialMessages`, so a captured-but-unappended item is completed by that run's recovery and is never offered to the next run.
 
@@ -1457,14 +1457,25 @@ Read-only opens keep the physical v3 file unchanged; the first v4 write persists
 The tree-facing contract. Each lane exposes one view (`lane.session`); `Session` itself implements it for `main`. Reads pass through always. A write through a lane view enters that lane's mutation line: while a run is open — including suspension and cancellation — it becomes a durable deferred write; during compaction or navigation it waits for the operation to end; on an idle lane it appends directly. Writes on a standalone `Session` (no harness attached) apply immediately.
 
 ```ts
-interface EntryQuery {
-  type?: Entry["type"];
-  customType?: string;                     // for type "custom"
+interface EntryQueryBase {
   order?: "newestFirst" | "oldestFirst";   // default newestFirst
   limit?: number;
   cursor?: EntryCursor;
 }
 
+type EntryQuery = EntryQueryBase & (
+  | {
+      /** Omit to query every entry type unless customType is present. */
+      type?: "custom";
+      /** When present, implicitly restricts results to custom entries. */
+      customType?: string;
+    }
+  | {
+      type: Exclude<Entry["type"], "custom">;
+      customType?: never;
+    }
+);
+
 /** Bounds of a branch scan. Default: the whole path, leaf to root. */
 interface BranchBounds {
   start?: string;              // default: the view's lane leaf
@@ -1502,7 +1513,7 @@ interface SessionTree {
 Query semantics: a branch scan takes the path from `start` to root, walks it in `order` direction, stops after a `stopAt` match (inclusive), filters, then applies `limit` and `cursor`.
 
 - `newestFirst` with `stopAtType: "compaction"` ends at the newest compaction: the context window.
-- `type` and `customType` filter results; a `stopAt` entry is returned only if it passes the filter.
+- `type` and `customType` filter results; `customType` alone selects custom entries, and cannot be combined with a non-custom `type`. A `stopAt` entry is returned only if it passes the filter.
 - Extension patterns: effective state = `findEntryOnBranch({ type: "custom", customType })`; collections = `findEntriesOnBranch(...)`; global inventory = `findEntries(...)`.
 - Context build is a branch scan with `stopAtType: "compaction"`, projected through `entryProjectors` and `toProviderMessages`. Its projection is the compaction summary, the materialized `retainedTail`, then the entries after the compaction; nothing before the compaction is read.
 - `SessionTree` has no navigation; moving a lane is `navigateTree()` on the lane.
@@ -1526,7 +1537,7 @@ class Session implements SessionTree {          // bound to "main"
   view(lane: string): SessionTree;
 
   // Lanes — permanent named pointers. Durable via storage (section 13).
-  getLanes(): Promise<{ lane: string; leafId: string | null }[]>;
+  getLanes(): Promise<LanePointer[]>;
   createLane(lane: string, at: string | null): Promise<void>;   // rejects existing names
   moveLane(lane: string, to: string | null): Promise<void>;
 
@@ -1542,14 +1553,17 @@ class Session implements SessionTree {          // bound to "main"
     query: RecordQuery & { type: K },
   ): Promise<Extract<LaneRecord, { type: K }>[]>;
   findRecords(query?: RecordQuery): Promise<LaneRecord[]>;
-  /** Unfinished operation starts, newest first. limit: 2 distinguishes the
-      valid zero/one states from multiple-open-operation corruption. */
-  findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]>;
   /** Full chronological view: entries, records, facts, lane moves,
       merged by seq. Debugging and tests. */
   getLog(options?: { afterSeq?: number; limit?: number }): Promise<LogItem[]>;
 }
 
+interface LanePointer {
+  lane: string;
+  leafId: string | null;
+  openOperationId: string | null;
+}
+
 interface IdGenerator { next(): string; }
 
 interface RecordQuery {
@@ -1582,14 +1596,14 @@ interface RecordQuery {
 
 ### Contract
 
-One session per storage instance. Storage persists and answers queries; `Session` owns validation and view binding. Storage never executes operations, queues, or recovery. Record payloads are opaque except for indexed columns and the required open-operation recovery projection.
+One session per storage instance. Storage persists and answers queries; `Session` owns validation and view binding. Storage never executes operations, queues, or recovery. Record payloads are opaque except for indexed columns and the lane open-operation projection.
 
 ```ts
 interface SessionStorage {
   getMetadata(): Promise<SessionMetadata>;
 
   // Lanes
-  getLanes(): Promise<{ lane: string; leafId: string | null }[]>;
+  getLanes(): Promise<LanePointer[]>;
   createLane(lane: string, at: string | null): Promise<void>;
   moveLane(lane: string, to: string | null): Promise<void>;
 
@@ -1609,7 +1623,6 @@ interface SessionStorage {
     query: RecordQuery & { type: K },
   ): Promise<Extract<LaneRecord, { type: K }>[]>;
   findRecords(query?: RecordQuery): Promise<LaneRecord[]>;
-  findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]>;
   getLog(options?): Promise<LogItem[]>;
 
   // Global facts
@@ -1627,7 +1640,6 @@ 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 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.
@@ -1697,7 +1709,9 @@ branch_tips    (session_id, branch_id, tip_id)      -- PRIMARY KEY (session_id,
 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)
+records:        (session_id, lane, seq), (session_id, lane, type, seq)
+                (session_id, type, op_kind, seq), (session_id, lane, type, op_kind, seq)
+                (session_id, run_id, seq)
 branch_entries: (session_id, branch_id, entry_type, entry_seq)
                 (session_id, entry_id)              -- reverse lookup: entry → branches
 ```
@@ -1762,7 +1776,7 @@ Case 4 — a branch still ends at an entry that has children.
 
 Stale branches (no lane resolves through them) are kept.
 
-Every restore query is an index seek plus a bounded scan: a lane's open operation via `(lane, type, seq)`, its last run-kind start via `(lane, type, op_kind, seq)`, its records above the operation via the same index, its own entries via the read plan from its leaf. No query touches another lane's traffic.
+SQLite returns open-operation ids with lane pointers from the `lanes` primary key and loads each projected start from `(session_id, run_id, seq)`. Its latest run-kind start uses `(session_id, lane, type, op_kind, seq)`. Once discovery identifies a boundary, records above it use the lane indexes and the lane's own entries use the read plan from its leaf. No query touches another lane's traffic.
 
 SQLite implementation follow-ups:
 
@@ -3235,9 +3249,9 @@ 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 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.
+  - Return `openOperationId` with each lane pointer, load the projected start through `findRecords`, and retain `RecordQuery.operationKind` for direct indexed latest-run lookup.
+  - Prove that idle/open operations are distinguishable, normal writes cannot start a second operation on a busy lane, and replay rejects the same invalid transition. Keep the lane open-operation projection consistent across backends.
+  - Acceptance: memory and SQLite have identical query behavior, projected starts are validated, and recovery discovery requires no backend-specific API or historical open-operation scan.
 
 - [x] **R1 — pure record-log validity.** Dependencies: R0.
   - Primary files: `packages/agent/src/harness/reducer.ts`, `packages/agent/test/harness/reducer.test.ts`.

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

@@ -9,7 +9,6 @@ import {
 	type LogItem,
 	type LogOptions,
 	type NewRecord,
-	type OperationStartedRecord,
 	type ProvisionedEntry,
 	type RecordQuery,
 	SessionError,
@@ -172,7 +171,7 @@ 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;
+			const currentOpenOperationId = this.state.getOpenOperationId(newRecord.lane);
 			if (newRecord.type === "operation_started" && currentOpenOperationId !== undefined) {
 				throw new SessionError(
 					"storage",
@@ -212,10 +211,6 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
 		return structuredClone(this.state.findRecords(query));
 	}
 
-	async findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]> {
-		return structuredClone(this.state.findOpenOperations(lane, options));
-	}
-
 	async getLog(options: LogOptions = {}): Promise<LogItem[]> {
 		return structuredClone(this.state.getLog(options));
 	}

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

@@ -11,7 +11,6 @@ import {
 	type LogItem,
 	type LogOptions,
 	type NewRecord,
-	type OperationStartedRecord,
 	type ProvisionedEntry,
 	type RecordQuery,
 	type SessionCreateOptions,
@@ -72,7 +71,7 @@ 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;
+		const currentOpenOperationId = this.state.getOpenOperationId(newRecord.lane);
 		if (newRecord.type === "operation_started" && currentOpenOperationId !== undefined) {
 			throw new SessionError(
 				"storage",
@@ -109,10 +108,6 @@ export class InMemorySessionStorage implements SessionStorage {
 		return structuredClone(this.state.findRecords(query));
 	}
 
-	async findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]> {
-		return structuredClone(this.state.findOpenOperations(lane, options));
-	}
-
 	async getLog(options: LogOptions = {}): Promise<LogItem[]> {
 		return structuredClone(this.state.getLog(options));
 	}

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

@@ -10,7 +10,6 @@ import type {
 	LogItem,
 	LogOptions,
 	NewRecord,
-	OperationStartedRecord,
 	ProvisionedEntry,
 	RecordBase,
 	RecordQuery,
@@ -214,11 +213,6 @@ export class Session<TMetadata extends SessionMetadata = SessionMetadata> implem
 		return this.queryRecords(query);
 	}
 
-	async findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]> {
-		assertValidLimit(options?.limit);
-		return this.storage.findOpenOperations(lane, options);
-	}
-
 	async getLog(options?: LogOptions): Promise<LogItem[]> {
 		return this.queryLog(options);
 	}

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

@@ -53,7 +53,7 @@ export class SessionState {
 	private readonly entries: Entry[] = [];
 	private readonly entriesById = new Map<string, Entry>();
 	private readonly records: LaneRecord[] = [];
-	private readonly openOperationsByLane = new Map<string, Map<string, OperationStartedRecord>>();
+	private readonly openOperationByLane = new Map<string, OperationStartedRecord>();
 	private readonly lanes = new Map<string, string | null>([["main", null]]);
 	private readonly log: LogItem[] = [];
 	private readonly stats: SessionStats = {
@@ -71,7 +71,11 @@ export class SessionState {
 	}
 
 	getLanes(): LanePointer[] {
-		return [...this.lanes].map(([lane, leafId]) => ({ lane, leafId }));
+		return [...this.lanes].map(([lane, leafId]) => ({
+			lane,
+			leafId,
+			openOperationId: this.getOpenOperationId(lane) ?? null,
+		}));
 	}
 
 	requireLane(lane: string): string | null {
@@ -126,18 +130,22 @@ export class SessionState {
 			case "record": {
 				if (!this.lanes.has(mutation.record.lane)) invalid(`references missing lane ${mutation.record.lane}`);
 				if (this.usedIds.has(mutation.record.id)) invalid(`contains duplicate id ${mutation.record.id}`);
+				if (mutation.record.type === "operation_started") {
+					const currentOpenOperationId = this.getOpenOperationId(mutation.record.lane);
+					if (currentOpenOperationId !== undefined) {
+						invalid(`starts operation ${mutation.record.id} while ${currentOpenOperationId} remains open`);
+					}
+				}
 				this.sequence = seq;
 				this.usedIds.add(mutation.record.id);
 				this.records.push(mutation.record);
 				if (mutation.record.type === "operation_started") {
-					let openOperations = this.openOperationsByLane.get(mutation.record.lane);
-					if (!openOperations) {
-						openOperations = new Map();
-						this.openOperationsByLane.set(mutation.record.lane, openOperations);
-					}
-					openOperations.set(mutation.record.id, mutation.record);
-				} else if (mutation.record.type === "operation_finished") {
-					this.openOperationsByLane.get(mutation.record.lane)?.delete(mutation.record.runId);
+					this.openOperationByLane.set(mutation.record.lane, mutation.record);
+				} else if (
+					mutation.record.type === "operation_finished" &&
+					this.openOperationByLane.get(mutation.record.lane)?.id === mutation.record.runId
+				) {
+					this.openOperationByLane.delete(mutation.record.lane);
 				}
 				this.log.push({ kind: "record", seq, record: mutation.record });
 				if (mutation.record.type === "usage") {
@@ -226,11 +234,8 @@ export class SessionState {
 		return results;
 	}
 
-	findOpenOperations(lane: string, options?: { limit?: number }): OperationStartedRecord[] {
-		assertValidLimit(options?.limit);
-		const openOperationsById = this.openOperationsByLane.get(lane);
-		const openOperations = openOperationsById ? [...openOperationsById.values()].reverse() : [];
-		return options?.limit === undefined ? openOperations : openOperations.slice(0, options.limit);
+	getOpenOperationId(lane: string): string | undefined {
+		return this.openOperationByLane.get(lane)?.id;
 	}
 
 	getLog(options: LogOptions = {}): LogItem[] {
@@ -275,7 +280,7 @@ export class SessionState {
 				targetId = position === "at" ? entry.id : entry.parentId;
 			}
 			copiedEntries = targetId === null ? [] : this.findEntriesOnBranch({ start: targetId, order: "oldestFirst" });
-			forkLanes = [{ lane: "main", leafId: targetId }];
+			forkLanes = [{ lane: "main", leafId: targetId, openOperationId: null }];
 		}
 
 		const mutations: SessionMutation[] = [];