feat(agent): add indexed harness recovery queries (#7646)

Implements R0 of the harness v2 plan, which adds RecordQuery.operationKind filtering and adds Session.findOpenOperations() for recovery purposes. The in-memory implementation keeps the open-operation projections in memory. SQLite implementation adds an index idx_records_session_lane_run_id_type.

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 (

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<SqliteSessionMetadata> {
 		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<Entry[]> {
 		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<SqliteSessionMetadata> {
 		return structuredClone(rows.map(decodeRecord));
 	}
 
+	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") {
+				throw new SessionError("storage", "Expected operation_started record");
+			}
+			return record;
+		});
+	}
+
 	async getLog(options: LogOptions = {}): Promise<LogItem[]> {
 		const afterSeq = options.afterSeq ?? 0;
 		const entryRows = readEntryRows(this.db, this.metadata.id, { afterSeq, order: "oldestFirst" });

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<RecordRow>(...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<RecordRow>(...params);
+}