diff --git a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md index c349e8098a..4b734a8e3e 100644 --- a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md +++ b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md @@ -8,9 +8,9 @@ Status: implemented ## Decision -Extract a backend-agnostic `PersistenceCoordinator` into `dsh-session-persistence`. The coordinator owns the orchestration once; each first-party backend composes one (`new PersistenceCoordinator(ctx, this)`), implements a small `PersistenceBackend` hook interface, and delegates its four public service methods (`create`/`append`/`load`/`list`) to it. +Extract a backend-agnostic `PersistenceCoordinator` into `dsh-session-persistence`. The coordinator owns the orchestration once; each first-party backend composes one (`new PersistenceCoordinator(ctx, this)`), implements a small `PersistenceBackend` hook interface, and delegates its stateful public methods (`create`/`append`/`load`/`inspect`) to it. Backend-owned metadata and revision listing bypass the coordinator. -Composition, not inheritance. The coordinator is a concrete class the backend holds, not a base class the backend extends. The Agent Note's risk — "a coordinator must not make unusual backends fight an inheritance hierarchy" — is avoided: a backend exposes only the hooks; it cannot reach the coordinator's private orchestration state, and the public `SessionPersistence` service shape is unchanged, so a third-party backend MAY still implement the abstract service directly without the coordinator at all. +Composition, not inheritance. The coordinator is a concrete class the backend holds, not a base class the backend extends. The Agent Note's risk — "a coordinator must not make unusual backends fight an inheritance hierarchy" — is avoided: a backend exposes only the hooks and cannot reach the coordinator's private orchestration state. A third-party backend MAY still implement the abstract service directly without the coordinator, including the non-mutating `inspect` contract used by read models. The coordinator retires each live session from its `session/disposed` notification: it waits for that exact Session object's initialization, serializes a final drain, and then removes the owned state, buffer, and init entries. Failed drains retain their buffers for backend teardown to retry. Settled per-id chain tails remove themselves only when they are still the current tail, so a completion cannot erase a newer operation for the same id. Backend teardown unregisters the write-path listeners before awaiting all admitted retirements, remaining buffers, and chains, then closes the backend. @@ -19,7 +19,7 @@ The coordinator retires each live session from its `session/disposed` notificati Five required members plus an optional lifecycle hook form the only boundary between the coordinator and storage: - `name` — backend label for the dispose-failure `AggregateError`. -- `loadStored(id)` — read one stored prefix by id across every storage scope (every JSONL cwd bucket; SQLite's id is globally unique). Resume/load, live adoption, and the create-collision probe share this lookup. The coordinator asserts the returned id and rejects a stored/live cwd mismatch before repair or state publication. +- `loadStored(id)` — read one stored prefix by id across every storage scope (every JSONL cwd bucket; SQLite's id is globally unique). Resume/load, non-mutating inspection, live adoption, and the create-collision probe share this lookup. The coordinator asserts the returned id and rejects a stored/live cwd mismatch before repair or state publication. - `appendBatch(meta, events, isMaterialized)` — durably append a contiguous batch, lazily materializing the session ATOMICALLY when not yet materialized (the materialize-write and the first event batch must commit together — a crash between them must not leave a materialized-but-empty session; this is why there is no separate `materialize` hook). - `commitRepair(meta, tornMarker, closers)` — make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined`) and append `closers`. **NOT required to be atomic** — JSONL legitimately truncates-then-appends in two fsync'd steps, SQLite does DELETE+INSERT in one transaction. Used by `load` (truncate + synthetic closers) and live-adoption (truncate only, `closers = []`). - `list()` — list all stored metadata. @@ -31,7 +31,7 @@ The single design choice that keeps the seam clean: the crash-repair "where is t ## Testing -The shared `runPersistenceContract` (public-API contract) keeps running for every backend. `runCoordinatorContract` (`tests/coordinator-contract.ts`) holds the write-path orchestration — adoption, HMR, collision, session and backend disposal drains, and crash-tail repair — and runs once per backend through a `CoordinatorFixture` (an in-memory reference + jsonl + sqlite). Coordinator-specific tests pin retirement map cleanup, same-id chain-tail races, failed-drain retry, and close ordering. The per-backend specs retain storage mechanics only (JSONL: path safety, fsync rollback, bucket listing; SQLite: schema version, `scanRows`, transaction rollback). A through-coordinator torn-tail→load→`commitRepair` test per real backend (via a `corruptTail` fixture hook) keeps the coordinator's torn-marker repair branch covered under the 100% per-file gate — the contract crash test only produces synthetic closers, never a torn marker, so it could not reach that branch. +The shared `runPersistenceContract` (public-API contract) keeps running for every backend and proves that `inspect` leaves interrupted logs and revisions unchanged before `load` performs recovery. `runCoordinatorContract` (`tests/coordinator-contract.ts`) holds the write-path orchestration — adoption, HMR, collision, session and backend disposal drains, and crash-tail repair — and runs once per backend through a `CoordinatorFixture` (an in-memory reference + jsonl + sqlite). Coordinator-specific tests pin retirement map cleanup, same-id chain-tail races, failed-drain retry, and close ordering. The per-backend specs retain storage mechanics only (JSONL: path safety, fsync rollback, bucket listing; SQLite: schema version, `scanRows`, transaction rollback). A through-coordinator torn-tail→load→`commitRepair` test per real backend (via a `corruptTail` fixture hook) keeps the coordinator's torn-marker repair branch covered under the 100% per-file gate — the contract crash test only produces synthetic closers, never a torn marker, so it could not reach that branch. ## Alternatives considered @@ -40,4 +40,4 @@ The shared `runPersistenceContract` (public-API contract) keeps running for ever ## Consequences -The coordinator adds one indirection, an opaque torn marker, and detached session-retirement tasks, but centralizes correctness-heavy orchestration previously duplicated by every backend. Session disposal remains an observe-only event, so the session owner does not await persistence retirement; the coordinator contains failures, preserves uncommitted buffers, and makes backend teardown the quiescence boundary. Its hook surface stays narrow: identity, adoption, and collision checks reuse `loadStored`; materialization stays atomic inside `appendBatch`; and listing bypasses the coordinator. New backends implement storage primitives rather than copy the event-buffer-flush lifecycle. +The coordinator adds one indirection, an opaque torn marker, and detached session-retirement tasks, but centralizes correctness-heavy orchestration previously duplicated by every backend. Session disposal remains an observe-only event, so the session owner does not await persistence retirement; the coordinator contains failures, preserves uncommitted buffers, and makes backend teardown the quiescence boundary. Its hook surface stays narrow: identity, adoption, collision checks, and non-mutating inspection reuse `loadStored`; materialization stays atomic inside `appendBatch`; and listing bypasses the coordinator. Read models use `inspect` rather than `load`, so observing a persisted open turn cannot race a new live owner by committing interruption closers. New backends implement storage primitives rather than copy the event-buffer-flush lifecycle. diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 0150c984bd..35bb0c8201 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -924,6 +924,16 @@ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise */ abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> +/** + * Inspect a header and its valid contiguous stored prefix without repairing + * a torn tail, closing an interrupted turn, or publishing coordinator state. + * This read is serialized with writes for the same id and returns detached + * values, so observers cannot mutate backend-owned state. + * @param id - the persisted session to inspect. + * @returns the header and valid stored event prefix exactly as observed. + */ +abstract inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> + /** * Lightweight listing from metadata, without a full-log parse. * @returns one header per materialized session. diff --git a/docs/core-data-structures/persistence.md b/docs/core-data-structures/persistence.md index e598e908d3..9bfdca56bd 100644 --- a/docs/core-data-structures/persistence.md +++ b/docs/core-data-structures/persistence.md @@ -2,7 +2,7 @@ The **durability seam** for the event log. [session.md](session.md) describes the in-memory `Session` — the append-only `SessionEvent` log that is the source of truth. This page describes how that log is made durable: the abstract `SessionPersistence` service, its backends, the flush checkpoint, crash recovery, and the metadata header that travels alongside the log. The event vocabulary the log carries is enumerated, member by member, in the generated [persistence log event catalog](../persistence-catalog.md). -The seam is a textbook [capability seam](../../.agents/notes/implemented/architecture/2026-06-13-capability-seams.md): one abstract service ([dsh-session-persistence](../../packages/session-persistence/session-persistence), `ctx.sessionPersistence`) defining locate/create/append/load plus lightweight list/snapshot observation over the existing `SessionEvent` — **no parallel persisted type** — and two interchangeable backends that pass the same `runPersistenceContract` suite. See the [session-persistence Agent Note](../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). +The seam is a textbook [capability seam](../../.agents/notes/implemented/architecture/2026-06-13-capability-seams.md): one abstract service ([dsh-session-persistence](../../packages/session-persistence/session-persistence), `ctx.sessionPersistence`) defining locate/create/append, crash-repairing load, non-mutating inspect, and lightweight list/snapshot observation over the existing `SessionEvent` — **no parallel persisted type** — and two interchangeable backends that pass the same `runPersistenceContract` suite. See the [session-persistence Agent Note](../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). ## The flush checkpoint @@ -12,6 +12,8 @@ The seam is a textbook [capability seam](../../.agents/notes/implemented/archite A backend that reloads a log crashed mid-turn finds an open `turn/start` with no `turn/end`. It does **not** truncate — a single turn can be huge in a long-horizon task (many steps, large tool output), and those events were durably appended before the crash. Instead it closes the orphaned turn with a synthetic `turn/end { reason: { kind: 'interrupted' } }`, keeping the log balanced and the turn-enclosure invariant intact. `interrupted` is the one `TurnEndReason` no loop emits (see [session.md](session.md#why-a-turn-ended-turnendreasonmap)). +`SessionPersistence.inspect(id)` is the observer counterpart to recovery: it returns a detached valid stored prefix without truncating a torn record, adding interruption closers, or publishing write state. Same-id serialization keeps it coherent with backend writes. Derived read models use `inspect`, never `load`, so observing a checkpointed open turn cannot mutate the log if live ownership begins concurrently. + ## `SessionLocation` — optional per-session artifact target `SessionPersistence.locate(meta)` synchronously resolves a backend-owned independent artifact without reading, creating, or flushing it. JSONL returns its absolute target path; SQLite returns `undefined` because sessions share one database. A returned path can therefore name a file that does not yet exist or lacks the current unflushed turn; it is a location hint, not authorization or a freshness guarantee. @@ -122,7 +124,7 @@ interface SessionPersistenceSnapshot { ## The backends -Both implement the same abstract `SessionPersistence` (locate/create/append/load/list/listSnapshots over `SessionEvent`) and pass `runPersistenceContract`, proving the seam is genuinely backend-agnostic: +Both implement the same abstract `SessionPersistence` (locate/create/append/load/inspect/list/listSnapshots over `SessionEvent`) and pass `runPersistenceContract`, proving the seam is genuinely backend-agnostic: - **[dsh-session-persistence-jsonl](../../packages/session-persistence/session-persistence-jsonl)** — an append-only logical JSONL log per session, stored as checksummed concatenated Zstandard frames by default or raw lines by configuration, with crash-safe atomic writes, interrupted-turn recovery, and a read/replay path. - **[dsh-session-persistence-sqlite](../../packages/session-persistence/session-persistence-sqlite)** — `node:sqlite`, one row per `SessionEvent`. The row shape `(session_id, seq, type, time, data, source_event_seqs, surface_op)` maps 1:1 onto the event, including optional surface metadata, so there is no parallel persisted schema to keep in sync. diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index d80d2cbf2e..ae8c55924a 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -464,6 +464,10 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ signature: 'abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', }, + { + signature: 'abstract inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', + jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values, so observers cannot mutate backend-owned state.\n * @param id - the persisted session to inspect.\n * @returns the header and valid stored event prefix exactly as observed.\n */', + }, { signature: 'abstract list(): Promise', jsDoc: '/**\n * Lightweight listing from metadata, without a full-log parse.\n * @returns one header per materialized session.\n */', diff --git a/packages/session-persistence/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts b/packages/session-persistence/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts index 3310b191b8..3cbb3b9832 100644 --- a/packages/session-persistence/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts +++ b/packages/session-persistence/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts @@ -19,6 +19,9 @@ class TestPersistence extends SessionPersistence { load(_id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { return Promise.reject(new Error('not used')) } + inspect(_id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return Promise.reject(new Error('not used')) + } list(): Promise { return Promise.resolve([]) } listSnapshots(): Promise { return Promise.resolve([]) } } diff --git a/packages/session-persistence/session-persistence-jsonl/README.md b/packages/session-persistence/session-persistence-jsonl/README.md index d21c47a39b..dfd90b3ce2 100644 --- a/packages/session-persistence/session-persistence-jsonl/README.md +++ b/packages/session-persistence/session-persistence-jsonl/README.md @@ -37,6 +37,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the - **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s the encoded header and first batch in a temporary file. POSIX publishes it without overwrite via a hard link and `fsync`s the parent directory. Windows publishes it without overwrite via `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` and creates missing directories through the same write-through pattern. A created-but-never-appended session leaves nothing on disk and is absent from `list`. - **Append-only.** Flushed events are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length. - **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and rejects. +- **Non-mutating inspection.** `inspect()` returns the detached valid prefix without truncating an incomplete tail or closing an interrupted turn, and leaves the lightweight revision unchanged. - **Contiguous-seq.** `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type. - **Lightweight revisions.** `listSnapshots()` identifies a log by its device, inode, size, and nanosecond timestamps, avoiding a full-log parse while changing after append, repair, replacement, or store changes. diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index b02b147841..629c0e3ff1 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -131,6 +131,10 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi return this.coordinator.load(id) } + inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.coordinator.inspect(id) + } + // One method serves both public `list` and the backend hook; delegating it to // the coordinator would call this hook recursively. diff --git a/packages/session-persistence/session-persistence-sqlite/README.md b/packages/session-persistence/session-persistence-sqlite/README.md index 3e7fc42fbe..6dcfa2d125 100644 --- a/packages/session-persistence/session-persistence-sqlite/README.md +++ b/packages/session-persistence/session-persistence-sqlite/README.md @@ -19,6 +19,7 @@ On filesystems with POSIX modes, the backend requests mode `0700` for missing di - **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.) - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `list()` (which reports exactly the sessions that have a row). - **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md): preserve the valid interrupted turn, append its synthetic closing events in one transaction, and remove only a torn tail row. Committed parse errors or sequence gaps make the session unloadable. Because recovery mutates stored rows, the next append starts from a balanced log and accurate cursor. +- **Non-mutating inspection.** `inspect()` returns the detached valid row prefix without deleting a torn tail row or appending recovery closers, and leaves the lightweight revision unchanged. - **Lightweight revisions.** `listSnapshots()` combines the immutable store and database-file identity, a per-materialization incarnation id, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and distinguishes independent stores and recreated same-id logs. ## Configuration (schemastery) diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 693d3119e7..5804c18282 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -157,6 +157,10 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers return this.coordinator.load(id) } + inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.coordinator.inspect(id) + } + // One method serves both public `list` and the backend hook; delegating it to // the coordinator would call this hook recursively. diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 8efe22c20c..ea9265cb7a 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -12,6 +12,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l | `create(meta): Promise` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). | | `append(id, events): Promise` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. | | `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. | +| `inspect(id): Promise<{ meta; events }>` | Return a detached valid stored prefix without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; intended for read models and other observers that must never recover a log. | | `list(): Promise` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. | | `listSnapshots(): Promise` | Lightweight metadata plus an opaque branded per-log revision, without loading event logs. A revision stays equal while that log and its backing store are unchanged, changes after append or mutating load repair, and cannot collide solely because two stores use the same local counter. | @@ -37,13 +38,13 @@ The `PersistenceBackend` hooks (the only seam between the coordinato | Hook | Role | |---|---| | `name` | Backend label for the dispose-failure `AggregateError`. | -| `loadStored(id)` | Read a stored prefix by id across every storage scope. Used by resume/load, live adoption, and the create-collision probe. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. | +| `loadStored(id)` | Read a stored prefix by id across every storage scope. Used by resume/load, non-mutating inspect, live adoption, and the create-collision probe. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. | | `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. | | `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). | | `list()` | List all stored metadata. | | `close?()` | Optional lifecycle teardown (e.g. close a db handle), awaited after the dispose drain. | -The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). A third-party backend MAY implement the abstract service directly without the coordinator, but it must also provide trustworthy lightweight snapshot revisions. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md). +The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. Its `inspect()` path validates and clones the prefix without calling `commitRepair` or publishing write state. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). A third-party backend MAY implement the abstract service directly without the coordinator, but it must provide the same non-mutating inspection and trustworthy lightweight snapshot revisions. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md). ## Testing backends diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index f558caa789..ce536c571c 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -252,6 +252,28 @@ export class PersistenceCoordinator { return this.serialize(id, () => this.loadCore(id)) } + /** + * Read a detached valid stored prefix without recovery mutations or + * coordinator-state publication. + * @param id - persisted session to inspect. + * @returns stored header and events before any synthetic recovery closers. + */ + inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.serialize(id, () => this.inspectCore(id)) + } + + private async inspectCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + const stored = await this.backend.loadStored(id) + if (stored === undefined) throw new Error(`session "${id}" not found`) + this.assertStoredId(id, stored.meta) + this.assertVersion(stored.meta) + assertSupportedEvents(stored.events, id) + return { + meta: structuredClone(stored.meta), + events: structuredClone(stored.events), + } + } + private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { const stored = await this.backend.loadStored(id) if (stored === undefined) throw new Error(`session "${id}" not found`) diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index 7c98e37734..276b4e5bcf 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -93,6 +93,16 @@ export abstract class SessionPersistence extends Service { */ abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> + /** + * Inspect a header and its valid contiguous stored prefix without repairing + * a torn tail, closing an interrupted turn, or publishing coordinator state. + * This read is serialized with writes for the same id and returns detached + * values, so observers cannot mutate backend-owned state. + * @param id - the persisted session to inspect. + * @returns the header and valid stored event prefix exactly as observed. + */ + abstract inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> + /** * Lightweight listing from metadata, without a full-log parse. * @returns one header per materialized session. diff --git a/packages/session-persistence/session-persistence/tests/contract.ts b/packages/session-persistence/session-persistence/tests/contract.ts index 1e50347f55..ae07bf77aa 100644 --- a/packages/session-persistence/session-persistence/tests/contract.ts +++ b/packages/session-persistence/session-persistence/tests/contract.ts @@ -99,6 +99,15 @@ export function runPersistenceContract(name: string, make: () => Promise snapshot.header.id === m.id)?.revision + const inspected = await persistence.inspect(m.id) + const afterInspect = (await persistence.listSnapshots()) + .find(snapshot => snapshot.header.id === m.id)?.revision + expect(afterInspect).toBe(beforeRepair) + expect(inspected.events.map(e => e.type)).toEqual([ + 'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', + 'turn/start', 'step/start', + ]) + // load PRESERVES the interrupted turn's events (a turn can be huge — they // must not be truncated) and closes the orphaned turn with synthetic // boundary events: step/end (the step was open) then turn/end {interrupted}. diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index 620d069d32..c42bf935e8 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -643,11 +643,12 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< } }) - it('load rejects a missing session', async () => { + it('load and inspect reject a missing session', async () => { const fix = await makeFixture() const { ctx, fiber } = await freshCtx(fix) try { await expect(ctx.sessionPersistence.load(SessionId('nope'))).rejects.toThrow(/not found/) + await expect(ctx.sessionPersistence.inspect(SessionId('nope'))).rejects.toThrow(/not found/) } finally { await fiber.dispose() await fix.cleanup() diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 3ea7bc6adb..dc0ee1b7df 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -96,6 +96,10 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend return this.coordinator.load(id) } + inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.coordinator.inspect(id) + } + // --- PersistenceBackend hooks (the Map storage primitives) --- // A Map-backed store has no torn tails, so `tornMarker` is never set. diff --git a/packages/session-query/session-query-sqlite/README.md b/packages/session-query/session-query-sqlite/README.md index af37ff74c8..a2c48a669f 100644 --- a/packages/session-query/session-query-sqlite/README.md +++ b/packages/session-query/session-query-sqlite/README.md @@ -12,11 +12,11 @@ All three surfaces (`current`, `shadowed`, and `log-only`) are searchable by def ## Source and index lifecycle -The service requires `ctx.sessions` and observes optional `ctx.sessionPersistence` dynamically. One serialized state machine compares source-qualified lightweight durable snapshot revisions, loads only new or changed logs, extracts shared semantic documents, reconciles changes transactionally, and runs the query. It never invokes the persistence backend's mutating `load()` for an id currently owned by `ctx.sessions`; the TEMP live row still records persisted availability, and the durable base refreshes after that live owner detaches. Repeated queries and unchanged same-store reopen load no full durable logs; switching stores, or observing new, changed, deleted, or load-repaired sources, reconciles on the next stable observation. Source or transaction failure commits nothing, and the next search retries. +The service requires `ctx.sessions` and observes optional `ctx.sessionPersistence` dynamically. One serialized state machine compares source-qualified lightweight durable snapshot revisions, non-mutatingly inspects only new or changed logs, extracts shared semantic documents, reconciles changes transactionally, and runs the query. Session queries never invoke the persistence backend's crash-repairing `load()`; an owner attaching during inspection cannot mutate its log, and the stable-observation retry makes the result live-preferred. The TEMP live row still records persisted availability, and the durable base refreshes after that live owner detaches. Repeated queries and an unchanged same-store reopen perform no full durable-log inspection; switching stores, or observing new, changed, deleted, or externally load-repaired sources, reconciles on the next stable observation. Source or transaction failure commits nothing, and the next search retries. Persisted FTS rows live in a dedicated derived database. Connection-local TEMP tables hold live rows, which shadow the durable base for the same session and reveal it when the live owner disappears. Unmounting persistence hides durable rows without discarding the cache; remounting reconciles it. Closing or reopening the database drops every live overlay while retaining persisted rows. -The database is disposable but reset is guarded: a recognized incompatible search schema rebuilds in place, while an unrelated or canonical database is refused before mutating journal mode. Never point `path` at the session-persistence database. On filesystems with POSIX modes, missing directories and databases are created owner-only (`0700` and `0600` before the process umask), and SQLite sidecars inherit the database mode; existing modes are preserved. Exactly one service in one process owns a derived-index path; external writers or a second process are unsupported because generations and TEMP shadow state are connection-owned. +The database is disposable but reset is guarded: every recognized schema version rejects unknown user tables before mutating journal mode, and only a recognized incompatible schema containing derived tables rebuilds in place. An unrelated or canonical database is refused. Never point `path` at the session-persistence database. On filesystems with POSIX modes, missing directories and databases are created owner-only (`0700` and `0600` before the process umask), and SQLite sidecars inherit the database mode; existing modes are preserved. Exactly one service in one process owns a derived-index path; external writers or a second process are unsupported because generations and TEMP shadow state are connection-owned. ## Configuration diff --git a/packages/session-query/session-query-sqlite/src/index.ts b/packages/session-query/session-query-sqlite/src/index.ts index 049999d2e7..a5fa0761e4 100644 --- a/packages/session-query/session-query-sqlite/src/index.ts +++ b/packages/session-query/session-query-sqlite/src/index.ts @@ -438,11 +438,12 @@ export class SessionQuerySqlite extends SessionQueryService { persisted = materializePersistenceSnapshots(before) for (const entry of persisted.values()) { if (canReuseIndexed && indexed.get(entry.header.id)?.revision === entry.revision) continue - // `load()` may durably repair an interrupted tail. Never invoke it - // for a session currently owned by the live store: a checkpointed - // open turn is active, not crash-interrupted. + // Skip work already shadowed by a live owner. `inspect()` is + // non-mutating, so an owner attaching after this check cannot cause + // crash-repair side effects; the live-membership retry below makes + // the returned observation live-preferred. if (initiallyLive.has(entry.header.id) || this.ctx.sessions.get(entry.header.id) !== undefined) continue - const loaded = await waitWithAbort(persistence.load(entry.header.id), signal) + const loaded = await waitWithAbort(persistence.inspect(entry.header.id), signal) assertSessionHeadersCompatible(entry.header, loaded.meta) entry.loaded = observeSession(loaded.meta, loaded.events) } diff --git a/packages/session-query/session-query-sqlite/src/schema.ts b/packages/session-query/session-query-sqlite/src/schema.ts index 045c84d960..b88e04b536 100644 --- a/packages/session-query/session-query-sqlite/src/schema.ts +++ b/packages/session-query/session-query-sqlite/src/schema.ts @@ -60,8 +60,9 @@ export async function openSearchDatabase(path: string, journalMode: JournalMode) if (applicationId === 0 && userTables.length > 0) { throw new Error(`session-search database at "${actual}" is not an empty or recognized derived index`) } - if (applicationId === SESSION_QUERY_SQLITE_APPLICATION_ID && version !== SESSION_QUERY_SQLITE_SCHEMA_VERSION) { - resetDerivedSchema(db, actual, userTables) + if (applicationId === SESSION_QUERY_SQLITE_APPLICATION_ID) { + assertDerivedUserTables(actual, userTables) + if (version !== SESSION_QUERY_SQLITE_SCHEMA_VERSION) resetDerivedSchema(db, userTables) } // Apply mutating pragmas only after refusing foreign or canonical files. // journalMode is a validated closed union, not caller-controlled SQL. @@ -82,13 +83,16 @@ function listUserTables(db: DatabaseSync): string[] { return rows.map(row => row.name) } -function resetDerivedSchema(db: DatabaseSync, path: string, userTables: readonly string[]): void { +function assertDerivedUserTables(path: string, userTables: readonly string[]): void { const unknownTables = userTables.filter(name => !DERIVED_USER_TABLES.has(name)) if (unknownTables.length > 0) { throw new Error( `session-search database at "${path}" has unrecognized user tables: ${unknownTables.join(', ')}`, ) } +} + +function resetDerivedSchema(db: DatabaseSync, userTables: readonly string[]): void { for (const name of userTables) { db.exec(`DROP TABLE IF EXISTS ${quoteIdentifier(name)}`) } diff --git a/packages/session-query/session-query-sqlite/tests/sqlite.spec.ts b/packages/session-query/session-query-sqlite/tests/sqlite.spec.ts index 2e5aebfeeb..8a1a3454f1 100644 --- a/packages/session-query/session-query-sqlite/tests/sqlite.spec.ts +++ b/packages/session-query/session-query-sqlite/tests/sqlite.spec.ts @@ -67,7 +67,9 @@ class TestPersistence extends SessionPersistence { static revisions = new Map() static nextRevision = 0 static loads = new Map() + static inspections = new Map() static loadEffect: ((entry: { meta: SessionHeader; events: SessionEvent[] }) => void) | undefined + static inspectEffect: ((entry: { meta: SessionHeader; events: SessionEvent[] }) => void | Promise) | undefined static listGate: Promise | undefined static listStarted: (() => void) | undefined static snapshotEffect: (() => void | Promise) | undefined @@ -82,7 +84,9 @@ class TestPersistence extends SessionPersistence { this.entries = new Map() this.revisions = new Map() this.loads = new Map() + this.inspections = new Map() this.loadEffect = undefined + this.inspectEffect = undefined for (const entry of entries) this.set(entry) this.listGate = undefined this.listStarted = undefined @@ -123,6 +127,16 @@ class TestPersistence extends SessionPersistence { return structuredClone(entry) } + async inspect(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + TestPersistence.inspections.set(id, (TestPersistence.inspections.get(id) ?? 0) + 1) + if (TestPersistence.failure !== undefined) throw TestPersistence.failure + const entry = TestPersistence.entries.get(id) + if (entry === undefined) throw new Error('missing test session') + await TestPersistence.inspectEffect?.(entry) + TestPersistence.inspectEffect = undefined + return structuredClone(entry) + } + async list(): Promise { TestPersistence.listStarted?.() await TestPersistence.listGate @@ -602,11 +616,13 @@ describe('SQLite reconciliation and source lifecycle', () => { items: [{ header: shared, live: true, persisted: true }], }) expect(TestPersistence.loads.get(shared.id)).toBeUndefined() + expect(TestPersistence.inspections.get(shared.id)).toBeUndefined() detach() await expect(ctx.sessionQuery.searchSessions({ query: 'persisted' })) .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] }) - expect(TestPersistence.loads.get(shared.id)).toBe(1) + expect(TestPersistence.loads.get(shared.id)).toBeUndefined() + expect(TestPersistence.inspections.get(shared.id)).toBe(1) await persistence.dispose() }) @@ -623,6 +639,28 @@ describe('SQLite reconciliation and source lifecycle', () => { .resolves.toMatchObject({ items: [{ header: { id: SessionId('attached') } }] }) }) + it('cannot crash-repair a log when live ownership begins during persisted inspection', async () => { + const shared = header('attach-during-inspect', 10) + const persistedEvents = messageEvents('persisted needle') + TestPersistence.reset([{ meta: shared, events: persistedEvents }]) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + TestPersistence.loadEffect = (entry) => { + entry.events = messageEvents('incorrect repair') + } + TestPersistence.inspectEffect = () => { + ctx.sessions.create(shared.id, { + seed: messageEvents('live needle'), + meta: { createdAt: shared.createdAt }, + }) + } + + await expect(ctx.sessionQuery.searchSessions({ query: 'live' })) + .resolves.toMatchObject({ items: [{ header: shared, live: true, persisted: true }] }) + expect(TestPersistence.loads.get(shared.id)).toBeUndefined() + expect(TestPersistence.entries.get(shared.id)?.events).toEqual(persistedEvents) + }) + it('retries when one live owner replaces another during persistence observation', async () => { TestPersistence.reset() const ctx = await liveContext() @@ -733,10 +771,10 @@ describe('SQLite reconciliation and source lifecycle', () => { TestPersistence.revisions.set(durable.id, revision) const replacement = await ctx.plugin(TestPersistence) const page = await ctx.sessionQuery.searchSessions({ query: 'new needle' }) - expect(TestPersistence.loads.get(durable.id)).toBe(2) + expect(TestPersistence.inspections.get(durable.id)).toBe(2) expect(page).toMatchObject({ items: [{ header: durable }] }) await expect(ctx.sessionQuery.searchSessions({ query: 'old' })).resolves.toEqual({ items: [] }) - expect(TestPersistence.loads.get(durable.id)).toBe(2) + expect(TestPersistence.inspections.get(durable.id)).toBe(2) await replacement.dispose() }) @@ -768,8 +806,8 @@ describe('SQLite reconciliation and source lifecycle', () => { const page = await ctx.sessionQuery.searchSessions({ query: 'needle' }) expect(page.items.map(item => item.header.id).sort()).toEqual([added.id, first.id].sort()) - expect(TestPersistence.loads.get(first.id)).toBe(2) - expect(TestPersistence.loads.get(added.id)).toBe(1) + expect(TestPersistence.inspections.get(first.id)).toBe(2) + expect(TestPersistence.inspections.get(added.id)).toBe(1) }) it('fails after one retry when persistence snapshots keep changing', async () => { @@ -811,7 +849,7 @@ describe('SQLite reconciliation and source lifecycle', () => { await expect(ctx.sessionQuery.searchSessions({ query: 'needle' })) .resolves.toMatchObject({ items: [{ header: durable }] }) - expect(TestPersistence.loads.get(durable.id)).toBe(2) + expect(TestPersistence.inspections.get(durable.id)).toBe(2) list.mockRestore() }) @@ -869,9 +907,9 @@ describe('SQLite reconciliation and source lifecycle', () => { const firstPersistence = await first.plugin(TestPersistence) const firstSearch = await first.plugin(SessionQuerySqlite, { path }) await first.sessionQuery.searchSessions({ query: 'needle' }) - expect(Object.fromEntries(TestPersistence.loads)).toEqual({ unchanged: 1, changed: 1, deleted: 1 }) + expect(Object.fromEntries(TestPersistence.inspections)).toEqual({ unchanged: 1, changed: 1, deleted: 1 }) await first.sessionQuery.searchSessions({ query: 'needle' }) - expect(Object.fromEntries(TestPersistence.loads)).toEqual({ unchanged: 1, changed: 1, deleted: 1 }) + expect(Object.fromEntries(TestPersistence.inspections)).toEqual({ unchanged: 1, changed: 1, deleted: 1 }) await firstSearch.dispose() await firstPersistence.dispose() @@ -890,7 +928,7 @@ describe('SQLite reconciliation and source lifecycle', () => { const secondSearch = await second.plugin(SessionQuerySqlite, { path }) const result = await second.sessionQuery.searchSessions({ query: 'needle' }) expect(result.items.map(item => item.header.id).sort()).toEqual([added.id, changed.id, unchanged.id].sort()) - expect(Object.fromEntries(TestPersistence.loads)).toEqual({ + expect(Object.fromEntries(TestPersistence.inspections)).toEqual({ unchanged: 1, changed: 2, deleted: 1, @@ -929,25 +967,30 @@ describe('SQLite reconciliation and source lifecycle', () => { await expect(second.sessionQuery.searchSessions({ query: 'live' })).resolves.toEqual({ items: [] }) await expect(second.sessionQuery.searchSessions({ query: 'persisted' })) .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] }) - expect(TestPersistence.loads.get(shared.id)).toBe(1) + expect(TestPersistence.inspections.get(shared.id)).toBe(1) await searchAgain.dispose() await persistenceAgain.dispose() }) - it('refreshes the stored revision after a mutating load repair', async () => { + it('refreshes after an external mutating load repair without loading from the query path', async () => { const durable = header('repair') TestPersistence.reset([{ meta: durable, events: messageEvents('before repair') }]) + const ctx = await liveContext() + const persistence = await ctx.plugin(TestPersistence) + await expect(ctx.sessionQuery.searchSessions({ query: 'before' })) + .resolves.toMatchObject({ items: [{ header: durable }] }) TestPersistence.loadEffect = (entry) => { entry.events = messageEvents('repaired needle') } - const ctx = await liveContext() - await ctx.plugin(TestPersistence) + await ctx.sessionPersistence.load(durable.id) await expect(ctx.sessionQuery.searchSessions({ query: 'repaired' })) .resolves.toMatchObject({ items: [{ header: durable }] }) - expect(TestPersistence.loads.get(durable.id)).toBe(2) + expect(TestPersistence.inspections.get(durable.id)).toBe(2) await ctx.sessionQuery.searchSessions({ query: 'repaired' }) - expect(TestPersistence.loads.get(durable.id)).toBe(2) + expect(TestPersistence.inspections.get(durable.id)).toBe(2) + expect(TestPersistence.loads.get(durable.id)).toBe(1) + await persistence.dispose() }) it('recovers on the next search after source and SQLite transaction failures', async () => { @@ -1066,6 +1109,27 @@ describe('SQLite schema, cancellation, and real persistence integration', () => expect(stillAugmented.prepare('PRAGMA user_version').get()).toEqual({ user_version: 999 }) stillAugmented.close() + const currentAugmentedPath = await temporaryPath('current-augmented.db') + const currentAugmentedOwner = await liveContext({ path: currentAugmentedPath }) + await (currentAugmentedOwner.sessionQuery as SessionQuerySqlite).close() + const currentAugmented = new DatabaseSync(currentAugmentedPath) + currentAugmented.exec('CREATE TABLE unrelated(value TEXT)') + currentAugmented.exec("INSERT INTO unrelated VALUES ('safe')") + currentAugmented.close() + const currentAugmentedCtx = new Context() + await currentAugmentedCtx.plugin(SessionStore) + await expect(currentAugmentedCtx.plugin(SessionQuerySqlite, { + path: currentAugmentedPath, + journalMode: 'delete', + })).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED')) + expect(currentAugmentedCtx.sessionQuery).toBeUndefined() + const stillCurrentAugmented = new DatabaseSync(currentAugmentedPath) + expect(stillCurrentAugmented.prepare('SELECT value FROM unrelated').get()).toEqual({ value: 'safe' }) + expect(stillCurrentAugmented.prepare('PRAGMA user_version').get()) + .toEqual({ user_version: SESSION_QUERY_SQLITE_SCHEMA_VERSION }) + expect(stillCurrentAugmented.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' }) + stillCurrentAugmented.close() + const foreignPath = await temporaryPath('foreign.db') const foreign = new DatabaseSync(foreignPath) foreign.exec('PRAGMA journal_mode = WAL') @@ -1260,22 +1324,22 @@ describe('SQLite schema, cancellation, and real persistence integration', () => const persistenceA = await first.plugin(SessionPersistenceSqlite, { path: persistencePathA }) await first.sessionPersistence.create(shared) await first.sessionPersistence.append(shared.id, messageEvents('alpha source')) - const loadA = vi.spyOn(first.sessionPersistence, 'load') + const inspectA = vi.spyOn(first.sessionPersistence, 'inspect') const searchA = await first.plugin(SessionQuerySqlite, { path: searchPath }) await expect(first.sessionQuery.searchSessions({ query: 'alpha' })) .resolves.toMatchObject({ items: [{ header: shared }] }) - expect(loadA).toHaveBeenCalledTimes(1) + expect(inspectA).toHaveBeenCalledTimes(1) await searchA.dispose() await persistenceA.dispose() const reopened = new Context() await reopened.plugin(SessionStore) const persistenceAAgain = await reopened.plugin(SessionPersistenceSqlite, { path: persistencePathA }) - const reopenedLoad = vi.spyOn(reopened.sessionPersistence, 'load') + const reopenedInspect = vi.spyOn(reopened.sessionPersistence, 'inspect') const searchAAgain = await reopened.plugin(SessionQuerySqlite, { path: searchPath }) await expect(reopened.sessionQuery.searchSessions({ query: 'alpha' })) .resolves.toMatchObject({ items: [{ header: shared }] }) - expect(reopenedLoad).not.toHaveBeenCalled() + expect(reopenedInspect).not.toHaveBeenCalled() await searchAAgain.dispose() await persistenceAAgain.dispose() @@ -1284,12 +1348,12 @@ describe('SQLite schema, cancellation, and real persistence integration', () => const persistenceB = await second.plugin(SessionPersistenceSqlite, { path: persistencePathB }) await second.sessionPersistence.create(shared) await second.sessionPersistence.append(shared.id, messageEvents('bravo source')) - const loadB = vi.spyOn(second.sessionPersistence, 'load') + const inspectB = vi.spyOn(second.sessionPersistence, 'inspect') const searchB = await second.plugin(SessionQuerySqlite, { path: searchPath }) await expect(second.sessionQuery.searchSessions({ query: 'bravo' })) .resolves.toMatchObject({ items: [{ header: shared }] }) await expect(second.sessionQuery.searchSessions({ query: 'alpha' })).resolves.toEqual({ items: [] }) - expect(loadB).toHaveBeenCalledTimes(1) + expect(inspectB).toHaveBeenCalledTimes(1) await searchB.dispose() await persistenceB.dispose() }) diff --git a/packages/session-query/session-query/src/corpus.ts b/packages/session-query/session-query/src/corpus.ts index 4a27c34e58..0e1753d5ce 100644 --- a/packages/session-query/session-query/src/corpus.ts +++ b/packages/session-query/session-query/src/corpus.ts @@ -72,16 +72,18 @@ export class SessionCorpus { if (persistence === undefined) throw notFound(sessionId) const listed = (await listPersisted(persistence)).find(header => header.id === sessionId) if (listed === undefined) throw notFound(sessionId) - let loaded: Awaited> + let loaded: Awaited> try { - loaded = await persistence.load(sessionId) + loaded = await persistence.inspect(sessionId) } catch (error: unknown) { throw new SessionQueryError( - `failed to load session "${sessionId}": ${errorMessage(error)}`, + `failed to inspect session "${sessionId}": ${errorMessage(error)}`, 'SESSION_QUERY_PERSISTENCE_FAILED', { cause: error }, ) } + const attached = this._ctx.sessions.get(sessionId) + if (attached !== undefined) return snapshotLive(attached) assertSessionHeadersCompatible(loaded.meta, listed) return { header: structuredClone(loaded.meta), diff --git a/packages/session-query/session-query/tests/session-query.spec.ts b/packages/session-query/session-query/tests/session-query.spec.ts index 98ca9a7863..a2ea051dd0 100644 --- a/packages/session-query/session-query/tests/session-query.spec.ts +++ b/packages/session-query/session-query/tests/session-query.spec.ts @@ -27,13 +27,15 @@ function eventLog(text = 'hello'): SessionEvent[] { class TestPersistence extends SessionPersistence { static entries = new Map() static listFailure: unknown - static loadFailure: unknown + static inspectFailure: unknown + static inspectEffect: (() => void) | undefined static afterList: (() => void) | undefined static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void { this.entries = new Map(entries.map(entry => [entry.meta.id, structuredClone(entry)])) this.listFailure = undefined - this.loadFailure = undefined + this.inspectFailure = undefined + this.inspectEffect = undefined this.afterList = undefined } @@ -54,10 +56,17 @@ class TestPersistence extends SessionPersistence { } load(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - if (TestPersistence.loadFailure !== undefined) return rejectUnknown(TestPersistence.loadFailure) + return this.inspect(id) + } + + inspect(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + if (TestPersistence.inspectFailure !== undefined) return rejectUnknown(TestPersistence.inspectFailure) const entry = TestPersistence.entries.get(id) if (entry === undefined) return Promise.reject(new Error('missing test session')) - return Promise.resolve(structuredClone(entry)) + const result = structuredClone(entry) + TestPersistence.inspectEffect?.() + TestPersistence.inspectEffect = undefined + return Promise.resolve(result) } list(): Promise { @@ -96,6 +105,22 @@ function rejectUnknown(reason: unknown): Promise { } describe('session-query exact reads', () => { + it('prefers a live owner that attaches while its persisted prefix is inspected', async () => { + const shared = header('attach-during-inspect', 2) + TestPersistence.reset([{ meta: shared, events: eventLog('persisted') }]) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + TestPersistence.inspectEffect = () => { + ctx.sessions.create(shared.id, { + seed: eventLog('live'), + meta: { createdAt: shared.createdAt }, + }) + } + + await expect(ctx.sessionQuery.filterEvents(shared.id, [])) + .resolves.toMatchObject([{ sessionId: shared.id, text: 'live' }]) + }) + it('reads the latest title from one live-preferred or persisted log without widening listSessions', async () => { const persistedHeader = header('persisted-title', 2) const sharedHeader = header('shared-title', 3) @@ -363,7 +388,7 @@ describe('session-query exact reads', () => { ) await ctx.plugin(TestPersistence) TestPersistence.listFailure = new Error('list unavailable') - TestPersistence.loadFailure = new Error('load unavailable') + TestPersistence.inspectFailure = new Error('inspect unavailable') await expect(ctx.sessionQuery.listEvents(live.id)).resolves.toHaveLength(2) await expect(ctx.sessionQuery.readEvent({ sessionId: live.id, seq: 1 })).resolves.toMatchObject({ target: { seq: 1 } }) @@ -381,10 +406,10 @@ describe('session-query exact reads', () => { await expect(ctx.sessionQuery.listEvents(SessionId('absent'))) .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND')) - TestPersistence.loadFailure = 'raw failure' + TestPersistence.inspectFailure = 'raw failure' await expect(ctx.sessionQuery.listEvents(durable.id)) .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED')) - TestPersistence.loadFailure = undefined + TestPersistence.inspectFailure = undefined const durableEntry = TestPersistence.entries.get(durable.id)! durableEntry.meta = { ...durableEntry.meta, cwd: '/changed-after-list' } TestPersistence.afterList = () => { diff --git a/packages/session-query/session-query/tests/tracing.spec.ts b/packages/session-query/session-query/tests/tracing.spec.ts index dc7aa41641..2f115d2dc9 100644 --- a/packages/session-query/session-query/tests/tracing.spec.ts +++ b/packages/session-query/session-query/tests/tracing.spec.ts @@ -31,17 +31,17 @@ function appendEvent(seq: number, sources?: number[]): SessionEvent { class TracePersistence extends SessionPersistence { static entries = new Map() static listCalls = 0 - static loadCalls = 0 + static inspectCalls = 0 static listFailure: Error | undefined - static loadFailure: Error | undefined + static inspectFailure: Error | undefined static afterList: (() => void) | undefined static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void { this.entries = new Map(entries.map(entry => [entry.meta.id, structuredClone(entry)])) this.listCalls = 0 - this.loadCalls = 0 + this.inspectCalls = 0 this.listFailure = undefined - this.loadFailure = undefined + this.inspectFailure = undefined this.afterList = undefined } @@ -62,8 +62,12 @@ class TracePersistence extends SessionPersistence { } load(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - TracePersistence.loadCalls += 1 - if (TracePersistence.loadFailure !== undefined) return Promise.reject(TracePersistence.loadFailure) + return this.inspect(id) + } + + inspect(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + TracePersistence.inspectCalls += 1 + if (TracePersistence.inspectFailure !== undefined) return Promise.reject(TracePersistence.inspectFailure) const entry = TracePersistence.entries.get(id) if (entry === undefined) return Promise.reject(new Error('missing test session')) return Promise.resolve(structuredClone(entry)) @@ -209,7 +213,7 @@ describe('session lineage tracing', () => { complete: true, }) expect(TracePersistence.listCalls).toBe(1) - expect(TracePersistence.loadCalls).toBe(0) + expect(TracePersistence.inspectCalls).toBe(0) TracePersistence.listFailure = new Error('unavailable') await expect(ctx.sessionQuery.traceSession(durable.id)) @@ -301,7 +305,7 @@ describe('session event tracing', () => { expect(repeated.derivedEventSeqs).toEqual([8]) }) - it('loads persisted logs once, prefers live logs, and preserves failures and conflicts', async () => { + it('inspects persisted logs once, prefers live logs, and preserves failures and conflicts', async () => { const durable = header('shared', 1, { cwd: '/same' }) TracePersistence.reset([{ meta: durable, events: [appendEvent(0)] }]) const ctx = await queryContext() @@ -309,7 +313,7 @@ describe('session event tracing', () => { await expect(ctx.sessionQuery.traceEvent({ sessionId: durable.id, seq: 0 })) .resolves.toMatchObject({ target: { type: 'user/message', surface: 'current' } }) - expect([TracePersistence.listCalls, TracePersistence.loadCalls]).toEqual([1, 1]) + expect([TracePersistence.listCalls, TracePersistence.inspectCalls]).toEqual([1, 1]) const live = ctx.sessions.create(durable.id, { meta: { createdAt: 1, cwd: '/same' } }) live.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) @@ -319,10 +323,10 @@ describe('session event tracing', () => { { surfaceOp: 'append' }, ) TracePersistence.listFailure = new Error('list unavailable') - TracePersistence.loadFailure = new Error('load unavailable') + TracePersistence.inspectFailure = new Error('inspect unavailable') await expect(ctx.sessionQuery.traceEvent({ sessionId: durable.id, seq: 1 })) .resolves.toMatchObject({ target: { type: 'context/message' } }) - expect([TracePersistence.listCalls, TracePersistence.loadCalls]).toEqual([1, 1]) + expect([TracePersistence.listCalls, TracePersistence.inspectCalls]).toEqual([1, 1]) TracePersistence.reset([{ meta: durable, events: [appendEvent(0)] }]) const failedCtx = await queryContext() @@ -331,10 +335,10 @@ describe('session event tracing', () => { await expect(failedCtx.sessionQuery.traceEvent({ sessionId: durable.id, seq: 0 })) .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED')) TracePersistence.listFailure = undefined - TracePersistence.loadFailure = new Error('load unavailable') + TracePersistence.inspectFailure = new Error('inspect unavailable') await expect(failedCtx.sessionQuery.traceEvent({ sessionId: durable.id, seq: 0 })) .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED')) - TracePersistence.loadFailure = undefined + TracePersistence.inspectFailure = undefined TracePersistence.afterList = () => { mutableHeader(TracePersistence.entries.get(durable.id)!.meta).cwd = '/changed' }