diff --git a/packages/agent/docs/harness-v2.md b/packages/agent/docs/harness-v2.md index b27a5a83a81..5febbbea5ef 100644 --- a/packages/agent/docs/harness-v2.md +++ b/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; + 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; createLane(lane: string, at: string | null): Promise; // rejects existing names moveLane(lane: string, to: string | null): Promise; @@ -1542,14 +1553,17 @@ class Session implements SessionTree { // bound to "main" query: RecordQuery & { type: K }, ): Promise[]>; findRecords(query?: RecordQuery): Promise; - /** 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; /** Full chronological view: entries, records, facts, lane moves, merged by seq. Debugging and tests. */ getLog(options?: { afterSeq?: number; limit?: number }): Promise; } +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; // Lanes - getLanes(): Promise<{ lane: string; leafId: string | null }[]>; + getLanes(): Promise; createLane(lane: string, at: string | null): Promise; moveLane(lane: string, to: string | null): Promise; @@ -1609,7 +1623,6 @@ interface SessionStorage { query: RecordQuery & { type: K }, ): Promise[]>; findRecords(query?: RecordQuery): Promise; - findOpenOperations(lane: string, options?: { limit?: number }): Promise; getLog(options?): Promise; // 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`. diff --git a/packages/agent/src/harness/session/jsonl/storage.ts b/packages/agent/src/harness/session/jsonl/storage.ts index 92f758dd2ae..34d54fd313b 100644 --- a/packages/agent/src/harness/session/jsonl/storage.ts +++ b/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 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 return structuredClone(this.state.findRecords(query)); } - async findOpenOperations(lane: string, options?: { limit?: number }): Promise { - return structuredClone(this.state.findOpenOperations(lane, options)); - } - async getLog(options: LogOptions = {}): Promise { return structuredClone(this.state.getLog(options)); } diff --git a/packages/agent/src/harness/session/memory.ts b/packages/agent/src/harness/session/memory.ts index 598d8dcf4fc..8e23877bfaa 100644 --- a/packages/agent/src/harness/session/memory.ts +++ b/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(newRecord: NewRecord): Promise { 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 { - return structuredClone(this.state.findOpenOperations(lane, options)); - } - async getLog(options: LogOptions = {}): Promise { return structuredClone(this.state.getLog(options)); } diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index f4be8adfacd..6457590f357 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/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 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); } diff --git a/packages/agent/src/harness/session/state.ts b/packages/agent/src/harness/session/state.ts index edc16f85b63..7a7441dc987 100644 --- a/packages/agent/src/harness/session/state.ts +++ b/packages/agent/src/harness/session/state.ts @@ -53,7 +53,7 @@ export class SessionState { private readonly entries: Entry[] = []; private readonly entriesById = new Map(); private readonly records: LaneRecord[] = []; - private readonly openOperationsByLane = new Map>(); + private readonly openOperationByLane = new Map(); private readonly lanes = new Map([["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[] = []; diff --git a/packages/agent/src/harness/session/testing/conformance.ts b/packages/agent/src/harness/session/testing/conformance.ts index be33b221308..867ac2f8427 100644 --- a/packages/agent/src/harness/session/testing/conformance.ts +++ b/packages/agent/src/harness/session/testing/conformance.ts @@ -135,8 +135,8 @@ export function createSessionBackendConformance( ], ); deepStrictEqual(await session.getLanes(), [ - { lane: "main", leafId: "child" }, - { lane: "thread", leafId: "child" }, + { lane: "main", leafId: "child", openOperationId: null }, + { lane: "thread", leafId: "child", openOperationId: "run" }, ]); }, ), @@ -160,9 +160,9 @@ export function createSessionBackendConformance( }); strictEqual(finished.seq, 2); - deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: "root" }]); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: "root", openOperationId: null }]); await session.moveLane("main", null); - deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: null }]); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: null, openOperationId: null }]); deepStrictEqual(await session.getLog(), [ { kind: "entry", seq: 1, entry: root }, { kind: "record", seq: 2, record: finished }, @@ -216,8 +216,8 @@ export function createSessionBackendConformance( ); deepStrictEqual(await session.getLanes(), [ - { lane: "main", leafId: "main-child" }, - { lane: "thread", leafId: "thread-child" }, + { lane: "main", leafId: "main-child", openOperationId: null }, + { lane: "thread", leafId: "thread-child", openOperationId: null }, ]); deepStrictEqual(await entryIds(session.findEntriesOnBranch({ start: "main-child", order: "oldestFirst" })), [ "root", @@ -242,8 +242,6 @@ export function createSessionBackendConformance( 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"); }), @@ -279,9 +277,14 @@ export function createSessionBackendConformance( await entryIds(session.findEntries({ order: "oldestFirst", cursor: { afterSeq: 2 }, limit: 2 })), ["compact", "new-note"], ); - deepStrictEqual(await entryIds(session.findEntries({ customType: "note" })), ["new-note", "old-note"]); + deepStrictEqual(await entryIds(session.findEntries({ type: "custom", customType: "note" })), [ + "new-note", + "old-note", + ]); deepStrictEqual( - await entryIds(session.findEntriesOnBranch({ start: "tail", customType: "note", limit: 1 })), + await entryIds( + session.findEntriesOnBranch({ start: "tail", type: "custom", customType: "note", limit: 1 }), + ), ["new-note"], ); deepStrictEqual( @@ -480,17 +483,23 @@ export function createSessionBackendConformance( ); }), - createCase(factory, "records and log", "tracks and enforces one open operation per lane", async (repository) => { + createCase(factory, "records and log", "enforces one open operation per lane", 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]); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: null, openOperationId: first.id }]); + deepStrictEqual( + await session.findRecords({ + lane: "main", + type: "operation_started", + runId: first.id, + limit: 1, + }), + [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", @@ -499,7 +508,9 @@ export function createSessionBackendConformance( runId: first.id, outcome: "completed", }); - deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), []); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: null, openOperationId: null }]); + const second = await session.appendRecord(operationStarted("second", { lane: "main", kind: "run" })); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: null, openOperationId: second.id }]); }), createCase( @@ -515,39 +526,26 @@ export function createSessionBackendConformance( runId: "run", outcome: "completed", }); - const started = await session.appendRecord(operationStarted("run", { lane: "main", kind: "run" })); - deepStrictEqual(await session.findOpenOperations("main", { limit: 2 }), [started]); + await session.appendRecord(operationStarted("run", { lane: "main", kind: "run" })); + await rejectsWithCode( + session.appendRecord(operationStarted("second", { lane: "main", kind: "run" })), + "storage", + ); }, ), - createCase(factory, "records and log", "scopes open operations by lane and limit", async (repository) => { + createCase(factory, "records and log", "scopes open-operation enforcement by lane", 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 threadNavigation = await session.appendRecord( - operationStarted("thread-navigation", { lane: "thread", kind: "navigation" }), - ); + await session.appendRecord(operationStarted("main-run", { lane: "main", kind: "run" })); + await session.appendRecord(operationStarted("thread-navigation", { lane: "thread", kind: "navigation" })); - deepStrictEqual(await session.findOpenOperations("main"), [mainRun]); - deepStrictEqual(await session.findOpenOperations("main", { limit: 1 }), [mainRun]); - deepStrictEqual(await session.findOpenOperations("thread", { limit: 2 }), [threadNavigation]); + await rejectsWithCode( + session.appendRecord(operationStarted("main-second", { lane: "main", kind: "run" })), + "storage", + ); }), - 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,15 +643,38 @@ export function createSessionBackendConformance( const metadata = await session.getMetadata(); const data = { nested: { value: 1 } }; await session.appendEntry({ type: "custom", id: "custom", customType: "note", data }, "main"); + const committedRecord = await session.appendRecord({ + type: "operation_started", + id: "run", + lane: "main", + sourceLeafId: "custom", + intent: { kind: "run", originalPrompt: [createUserMessage("original")], initialMessages: [] }, + }); + const expectedRecord = structuredClone(committedRecord); + data.nested.value = 50; const read = await session.getEntry("custom"); if (read?.type !== "custom") throw new Error("Expected custom entry"); (read.data as { nested: { value: number } }).nested.value = 99; + const [recordRead] = await session.findRecords({ type: "operation_started" }); + if (recordRead?.intent.kind !== "run") throw new Error("Expected operation-start record"); + deepStrictEqual(recordRead, expectedRecord); + recordRead.intent.originalPrompt.push(createUserMessage("mutated record read")); const readMetadata = await session.getMetadata(); readMetadata.id = "changed"; const log = await session.getLog(); if (log[0]?.kind !== "entry" || log[0].entry.type !== "custom") throw new Error("Expected entry log"); (log[0].entry.data as { nested: { value: number } }).nested.value = 100; + const recordLog = log.find((item) => item.kind === "record"); + if ( + recordLog?.kind !== "record" || + recordLog.record.type !== "operation_started" || + recordLog.record.intent.kind !== "run" + ) { + throw new Error("Expected operation-start record in log"); + } + deepStrictEqual(recordLog.record, expectedRecord); + recordLog.record.intent.originalPrompt.push(createUserMessage("mutated record log")); deepStrictEqual(await session.getMetadata(), metadata); deepStrictEqual(await session.getEntry("custom"), { @@ -665,6 +686,7 @@ export function createSessionBackendConformance( seq: 1, timestamp: read.timestamp, }); + deepStrictEqual(await session.findRecords({ type: "operation_started" }), [expectedRecord]); }), createCase(factory, "entries and lanes", "validates lane lifecycle and targets", async (repository) => { @@ -903,7 +925,7 @@ export function createSessionBackendConformance( }); deepStrictEqual(await entryIds(fork.findEntries({ order: "oldestFirst" })), [root, shared, mainChild]); - deepStrictEqual(await fork.getLanes(), [{ lane: "main", leafId: mainChild }]); + deepStrictEqual(await fork.getLanes(), [{ lane: "main", leafId: mainChild, openOperationId: null }]); strictEqual(await fork.getName(), "Source"); strictEqual(await fork.getLabel(shared), "copied"); strictEqual(await fork.getLabel(threadChild), undefined); @@ -936,8 +958,8 @@ export function createSessionBackendConformance( const fork = await repository.fork(await source.getMetadata(), { scope: "tree", id: "tree-fork" }); deepStrictEqual(await entryIds(fork.findEntries({ order: "oldestFirst" })), [root, mainChild, threadChild]); deepStrictEqual(await fork.getLanes(), [ - { lane: "main", leafId: mainChild }, - { lane: "thread", leafId: threadChild }, + { lane: "main", leafId: mainChild, openOperationId: null }, + { lane: "thread", leafId: threadChild, openOperationId: null }, ]); strictEqual(await fork.getLabel(threadChild), "thread-tip"); strictEqual((await fork.getStats()).messageCount, 3); diff --git a/packages/agent/src/harness/session/types.ts b/packages/agent/src/harness/session/types.ts index 5df425f1574..b6ccf7d5c0e 100644 --- a/packages/agent/src/harness/session/types.ts +++ b/packages/agent/src/harness/session/types.ts @@ -220,14 +220,28 @@ export interface EntryCursor { afterSeq: number; } -export interface EntryQuery { - type?: Entry["type"]; - customType?: string; // for type "custom" +interface EntryQueryBase { order?: EntryOrder; // default newestFirst limit?: number; cursor?: EntryCursor; } +export type EntryQuery = EntryQueryBase & + ( + | { + type?: never; + customType?: never; + } + | { + type: Exclude; + customType?: never; + } + | { + type: "custom"; + customType?: string; + } + ); + /** Bounds of a branch scan. Default: the whole path, leaf to root. */ export interface BranchBounds { start?: string; // default: the view's lane leaf @@ -273,6 +287,7 @@ export interface SessionStats { export interface LanePointer { lane: string; leafId: string | null; + openOperationId: string | null; } export type LogItem = @@ -291,7 +306,7 @@ export interface SessionStorage; // Lanes - getLanes(): Promise<{ lane: string; leafId: string | null }[]>; + getLanes(): Promise; createLane(lane: string, at: string | null): Promise; moveLane(lane: string, to: string | null): Promise; @@ -308,13 +323,6 @@ export interface SessionStorage[]>; findRecords(query?: RecordQuery): Promise; - /** - * Returns unfinished operation starts newest first. Recovery uses `limit: 2`: - * zero results mean the lane is idle, one means it is suspended, and two - * mean at least two operations are open, which is corruption. Further - * results provide no additional recovery state. - */ - findOpenOperations(lane: string, options?: { limit?: number }): Promise; getLog(options?: { afterSeq?: number; limit?: number }): Promise; // Global facts diff --git a/packages/agent/test/harness/session/jsonl-storage.test.ts b/packages/agent/test/harness/session/jsonl-storage.test.ts index 6b971dd6e29..150f5711d6d 100644 --- a/packages/agent/test/harness/session/jsonl-storage.test.ts +++ b/packages/agent/test/harness/session/jsonl-storage.test.ts @@ -178,7 +178,9 @@ describe("JSONL v4 per-session storage", () => { }) ).map((entry) => entry.id), ).toEqual(["compaction", "branch-summary"]); - expect((await restored.findEntries({ customType: "note" })).map((entry) => entry.id)).toEqual(["custom"]); + expect((await restored.findEntries({ type: "custom", customType: "note" })).map((entry) => entry.id)).toEqual([ + "custom", + ]); expect(await restored.getStats()).toEqual({ messageCount: 3, cachedTokens: 0, @@ -200,7 +202,7 @@ describe("JSONL v4 per-session storage", () => { expect(await restored.findEntries({ order: "oldestFirst" })).toEqual(committed); }); - it("round trips every record type, recovery projection, and ledger statistics", async () => { + it("round trips every record type and ledger statistics", async () => { const root = createTempDir(); const session = await createRepository(root).create({ id: "records", cwd: root }); await session.appendCustomEntry("anchor"); @@ -382,15 +384,6 @@ describe("JSONL v4 per-session storage", () => { const restored = await reopen(root, session); expect(await restored.findRecords({ order: "oldestFirst" })).toEqual(records); - expect( - ( - await restored.findRecords({ - type: "operation_started", - operationKind: "run", - limit: 1, - }) - ).map((record) => record.id), - ).toEqual(["run"]); expect( (await restored.findRecords({ runId: "compaction", order: "oldestFirst" })).map((record) => record.id), ).toEqual(["compaction", "compaction-attempt", "compaction-finished"]); @@ -399,9 +392,6 @@ describe("JSONL v4 per-session storage", () => { (record) => record.id, ), ).toEqual(["adjustment", "hook-usage"]); - expect((await restored.findOpenOperations("main", { limit: 2 })).map((record) => record.id)).toEqual([ - "navigation", - ]); expect(await restored.getStats()).toEqual({ messageCount: 0, cachedTokens: 45, @@ -409,11 +399,6 @@ describe("JSONL v4 per-session storage", () => { totalTokens: 150, costTotal: 15, }); - - const [started] = await restored.findRecords({ type: "operation_started", operationKind: "run" }); - if (started?.intent.kind !== "run") throw new Error("Expected restored run record"); - started.intent.originalPrompt.push(userMessage("mutated")); - expect(await restored.findRecords({ order: "oldestFirst" })).toEqual(records); }); it("persists concurrent cross-lane writes in shared sequence order", async () => { diff --git a/packages/agent/test/harness/session/jsonl.test.ts b/packages/agent/test/harness/session/jsonl.test.ts index 6b3a7671a94..9ec2f053770 100644 --- a/packages/agent/test/harness/session/jsonl.test.ts +++ b/packages/agent/test/harness/session/jsonl.test.ts @@ -292,21 +292,12 @@ describe("JSONL v4 persistence", () => { const reopenedRepository = createRepository(root); const reopened = await reopenedRepository.open(metadata); expect(await reopened.getLanes()).toEqual([ - { lane: "main", leafId: null }, - { lane: "thread", leafId: entryId }, + { lane: "main", leafId: null, openOperationId: null }, + { lane: "thread", leafId: entryId, openOperationId: "run" }, ]); expect(await reopened.getName()).toBe("Example"); expect(await reopened.getLabel(entryId)).toBe("checkpoint"); expect((await reopened.findRecords()).map((record) => record.id)).toEqual(["run"]); - expect( - ( - await reopened.findRecords({ - type: "operation_started", - operationKind: "run", - }) - ).map((record) => record.id), - ).toEqual(["run"]); - expect((await reopened.findOpenOperations("thread", { limit: 2 })).map((record) => record.id)).toEqual(["run"]); expect((await reopened.getLog()).map((item) => item.seq)).toEqual([1, 2, 3, 4, 5, 6]); expect( ( @@ -319,7 +310,6 @@ describe("JSONL v4 persistence", () => { }) ).seq, ).toBe(7); - expect(await reopened.findOpenOperations("thread", { limit: 2 })).toEqual([]); }); it("recomputes fork message counts when reopening", async () => { @@ -371,8 +361,8 @@ describe("JSONL v4 persistence", () => { threadId, ]); expect(await reopened.getLanes()).toEqual([ - { lane: "main", leafId: mainId }, - { lane: "thread", leafId: threadId }, + { lane: "main", leafId: mainId, openOperationId: null }, + { lane: "thread", leafId: threadId, openOperationId: null }, ]); expect(await reopened.getName()).toBe("Source"); expect(await reopened.getLabel(threadId)).toBe("tip"); @@ -677,6 +667,33 @@ describe("JSONL v4 persistence", () => { }, ], }, + { + name: "multiple open operations on one lane", + message: "while first remains open", + mutations: [ + { + kind: "record", + type: "operation_started", + id: "first", + lane: "main", + seq: 1, + timestamp: 1, + sourceLeafId: null, + intent: { kind: "run", originalPrompt: [], initialMessages: [] }, + }, + { + kind: "record", + type: "operation_started", + id: "second", + lane: "main", + seq: 2, + timestamp: 2, + sourceLeafId: null, + intent: { kind: "run", originalPrompt: [], initialMessages: [] }, + }, + ], + }, + { name: "a lane move referencing a missing entry", message: "missing lane target", diff --git a/packages/agent/test/harness/session/types.test.ts b/packages/agent/test/harness/session/types.test.ts new file mode 100644 index 00000000000..b161ca92d4c --- /dev/null +++ b/packages/agent/test/harness/session/types.test.ts @@ -0,0 +1,13 @@ +import { describe, expectTypeOf, it } from "vitest"; +import type { EntryQuery } from "../../../src/harness/session/index.ts"; + +describe("EntryQuery", () => { + it("constrains customType to custom entries", () => { + expectTypeOf<{ limit: 1 }>().toExtend(); + expectTypeOf<{ type: "custom" }>().toExtend(); + expectTypeOf<{ type: "custom"; customType: "note" }>().toExtend(); + expectTypeOf<{ type: "message" }>().toExtend(); + expectTypeOf<{ customType: "note" }>().not.toExtend(); + expectTypeOf<{ type: "message"; customType: "note" }>().not.toExtend(); + }); +}); diff --git a/packages/session-backends/sqlite-node/src/sqlite/repo.ts b/packages/session-backends/sqlite-node/src/sqlite/repo.ts index da4b13c0125..49c5e03ccd3 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/repo.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/repo.ts @@ -4,11 +4,11 @@ import { type Entry, type EntryQuery, type ForkOptions, + type LanePointer, type LaneRecord, type LogItem, type LogOptions, type NewRecord, - type OperationStartedRecord, type ProvisionedEntry, type RecordQuery, Session, @@ -46,13 +46,7 @@ import { startLaneOperation, moveLane as updateLane, } from "./storage/lanes.ts"; -import { - appendRecordRow, - deleteRecordRows, - idExistsInRecords, - readOpenOperationRows, - readRecordRows, -} from "./storage/records.ts"; +import { appendRecordRow, deleteRecordRows, idExistsInRecords, readRecordRows } from "./storage/records.ts"; import { advanceSequence, createSequence, @@ -434,8 +428,12 @@ class SqliteSessionStorage implements SessionStorage { return this.metadata.id === sessionId; } - async getLanes(): Promise<{ lane: string; leafId: string | null }[]> { - return readLanes(this.db, this.metadata.id).map((row) => ({ lane: row.lane, leafId: row.leaf_id })); + async getLanes(): Promise { + return readLanes(this.db, this.metadata.id).map((row) => ({ + lane: row.lane, + leafId: row.leaf_id, + openOperationId: row.open_operation_id, + })); } async createLane(lane: string, at: string | null): Promise { @@ -565,18 +563,6 @@ class SqliteSessionStorage implements SessionStorage { return 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 limit = options.limit; 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 1dc35722754..3f094afc74d 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts @@ -1,4 +1,3 @@ -import { SessionError } from "@earendil-works/pi-agent-core"; import { joinSqlFragments, sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; @@ -69,27 +68,3 @@ export function readRecordRows( WHERE ${joinSqlFragments(predicates, " AND ")} ORDER BY seq ${direction}${limit}`.all(db); } - -export function readOpenOperationRows( - db: SqliteDatabase, - sessionId: string, - lane: string, - _options: { limit?: number } = {}, -): RecordRow[] { - const laneRow = sql`SELECT open_operation_id FROM lanes WHERE session_id = ${sessionId} AND lane = ${lane}`.get<{ - open_operation_id: string | null; - }>(db); - if (!laneRow?.open_operation_id) return []; - - const record = sql`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload - FROM records - WHERE session_id = ${sessionId} - AND id = ${laneRow.open_operation_id}`.get(db); - 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/branch-query.test.ts b/packages/session-backends/sqlite-node/test/branch-query.test.ts index 54f017b68a4..40bdb146494 100644 --- a/packages/session-backends/sqlite-node/test/branch-query.test.ts +++ b/packages/session-backends/sqlite-node/test/branch-query.test.ts @@ -82,7 +82,7 @@ describe("SQLite branch queries", () => { } finally { await invalidJsonDb.close(); } - expect(await session.findEntriesOnBranch({ start: leafId, customType: "other" })).toEqual([]); + expect(await session.findEntriesOnBranch({ start: leafId, type: "custom", customType: "other" })).toEqual([]); }); it("does not validate ancestors beyond newest-first stop bounds", async () => { diff --git a/packages/session-backends/sqlite-node/test/migrations.test.ts b/packages/session-backends/sqlite-node/test/migrations.test.ts index fb16311009f..8d0cb68d874 100644 --- a/packages/session-backends/sqlite-node/test/migrations.test.ts +++ b/packages/session-backends/sqlite-node/test/migrations.test.ts @@ -44,14 +44,17 @@ describe("SQLite migrations", () => { const branchEntryIndexes = db.prepare("PRAGMA index_list(branch_entries)").all<{ name: string }>(); expect(branchEntryIndexes.map((index) => index.name)).toContain("idx_branch_entries_session_entry"); const recordIndexes = db.prepare("PRAGMA index_list(records)").all<{ name: string }>(); - expect(recordIndexes.map((index) => index.name)).toEqual( + const recordIndexNames = recordIndexes.map((index) => index.name); + expect(recordIndexNames).toEqual( expect.arrayContaining([ "idx_records_session_lane_seq", - "idx_records_session_type_seq", + "idx_records_session_lane_type_seq", "idx_records_session_type_op_kind_seq", + "idx_records_session_lane_type_op_kind_seq", + "idx_records_session_run_id_seq", ]), ); - expect(recordIndexes.map((index) => index.name)).not.toContain("idx_records_session_seq"); + expect(recordIndexNames).not.toContain("idx_records_session_seq"); const laneMoveIndexes = db.prepare("PRAGMA index_list(lane_moves)").all<{ name: string }>(); expect(laneMoveIndexes.map((index) => index.name)).not.toContain("idx_lane_moves_session_lane_seq"); } finally {