diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 0f8d53b5d..96be36d6c 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -9,6 +9,10 @@ - Watches use Chord's `WatchHandle` and `WatchEnd` (Durable re-exports both). End reasons are `stopped`, `cancelled`, `retired`, `ended` (was `session_closed`), `listener_error`, and `failed` (was `session_failed`). Retirement is an explicit event of the watched document incarnation (last frame `[["r", null]]`, then `retired`), never inferred from a `null` value. An ended watch drops its listener and every reference into its Session, so a stopped watch that is still referenced keeps only its last value. - Removed the Chord state getters `Session.documentState()`, `Conversation.viewState()`, and `Harness.taskGraph()`, and the `DocumentState` type; use `watchDoc()`, `Conversation.watch()`, and `Harness.watchTaskGraph()`. Durable no longer uses Chord's `replicatedState()`. +### Added + +- `HarnessOptions.readOnly` opens a Harness as a viewer of a session another process drives: open leaves running tasks as committed instead of recovering them, every commit with writes rejects with `ReadOnlySession` before it reaches Storage, and `resume()` and every call that would run tasks reject with it. + ## [1.4.0] - 2026-10-21 ### Breaking Changes diff --git a/packages/durable/README.md b/packages/durable/README.md index ca9826575..8882f9a3a 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -124,7 +124,7 @@ const root = await harness.root(context); // the same root as last time harness.resume(); // continue any run the last process left unfinished ``` -Work interrupted by a crash or close stays pending. `resume()` starts the task scheduler; submitting or waiting starts it too. Each conversation has its own persisted UUIDv7 in `pi.provider`, forwarded to pi-ai as `sessionId` for provider prompt-cache and session affinity. It survives reopen, retries, reset, compaction, and model changes; a child or fork receives a fresh identity. A legacy conversation receives and persists one before its first generation or compaction request. +Work interrupted by a crash or close stays pending. `resume()` starts the task scheduler; submitting or waiting starts it too. A process that only shows a session another one drives opens it with `readOnly: true`: open recovers nothing, and every write, `resume()`, and every call that would run tasks rejects with `ReadOnlySession`. Each conversation has its own persisted UUIDv7 in `pi.provider`, forwarded to pi-ai as `sessionId` for provider prompt-cache and session affinity. It survives reopen, retries, reset, compaction, and model changes; a child or fork receives a fresh identity. A legacy conversation receives and persists one before its first generation or compaction request. A retried submission with the same `requestId` returns the existing submission instead of submitting twice: diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 99ebf614e..ae40f01b1 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -753,6 +753,15 @@ with them. A viewer that only reads never enables scheduling: `inspect()`, `getTask()`, `submission()`, `Submission.status()`, `usage()`, document reads, every conversation read, state, and watch, and the task graph. +`HarnessOptions.readOnly` opens a viewer of a session another process drives, +such as a synced copy on a device that only shows it. Open loads live tasks as +committed, `running` ones included, and reconciles nothing. Every commit with +writes rejects with `ReadOnlySession` before it reaches Storage, so the Session +stays healthy; a commit that writes nothing succeeds. `resume()` throws +`ReadOnlySession`, and every call that enables scheduling rejects with it. +`wake()` is always inactive and `onWake` is never called. Reopening without +`readOnly` recovers as usual: the viewer counted no restart. + `createConversation({ ownership, agent, init })` and `fork(at, { ownership, agent, init })` commit atomically: the conversation, its explicitly selected ownership, the creation hook's documents, the `agent` diff --git a/packages/durable/src/errors.ts b/packages/durable/src/errors.ts index c66a1cee2..e2a36dfc1 100644 --- a/packages/durable/src/errors.ts +++ b/packages/durable/src/errors.ts @@ -45,6 +45,17 @@ export class SessionFailed extends Error { } } +/** + * The Harness was opened read-only (`HarnessOptions.readOnly`): a commit with writes, or a call that would run tasks, was + * refused. Nothing was written, and the Harness stays usable for reads. + */ +export class ReadOnlySession extends Error { + constructor(what: string) { + super(`Read-only Harness: ${what} refused`); + this.name = "ReadOnlySession"; + } +} + /** A submission reached a busy conversation and was not admitted. */ export class ConversationBusy extends Error { readonly conversationId: ConversationId; diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 7b315be5f..d22f368cc 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -123,7 +123,7 @@ class ConversationImpl implements Conversation { return this.#host.submissions.submit(this.id, submission, context); } - compact(instructions: string | undefined, context: Context): Promise> { + async compact(instructions: string | undefined, context: Context): Promise> { this.#host.tasks.resume(); const input = { reason: "manual", ...(instructions === undefined ? {} : { instructions }) } as const; return this.#host.harness.commitWith((tx) => createCompaction(tx, this.id, input), context); @@ -165,12 +165,12 @@ class ConversationImpl implements Conversation { return this.#host.create({ kind: "fork", parentId: this.id, at, ownership: options.ownership }, options, context); } - abort(context: Context, options?: ConversationAbortOptions): Promise { + async abort(context: Context, options?: ConversationAbortOptions): Promise { this.#host.tasks.resume(); return this.#host.tasks.abortConversation(this.id, options?.background === true, context); } - waitForIdle(context: Context): Promise { + async waitForIdle(context: Context): Promise { this.#host.tasks.resume(); return this.#host.tasks.waitForIdle(this.id, context); } @@ -214,7 +214,7 @@ class HarnessImpl extends SessionImpl implements const now = safeNow(options.now, report); const hostSettings = currentSettings(options.settings); // Read at each use, as all settings are, but only this field: it is read on every commit. - super(rawStorage, now, () => hostSettings()?.documentRetentionMs, options.setTimer); + super(rawStorage, now, () => hostSettings()?.documentRetentionMs, options.setTimer, options.readOnly === true); // Every component reads and writes through the Session's failure guard: `rawStorage` serves only `super()`. const storage = this.storage; this.#options = options; @@ -242,6 +242,7 @@ class HarnessImpl extends SessionImpl implements return record === undefined ? undefined : boundConversation(id, binding, this.#submissions, this.#tasks); }, context: withoutAbortSignal(context), + readOnly: options.readOnly === true, }); this.#submissions = new Submissions(this, storage, now, settings, () => this.#tasks.resume(), options.telemetry); this.#taskGraph = new TaskGraphView(this, storage, { @@ -337,12 +338,12 @@ class HarnessImpl extends SessionImpl implements return this.#tasks.retry(id, context); } - waitForTask(id: TaskId, context: Context): Promise> { + async waitForTask(id: TaskId, context: Context): Promise> { this.#tasks.resume(); return this.#tasks.waitForTask(id, context) as Promise>; } - waitForIdle(context: Context): Promise { + async waitForIdle(context: Context): Promise { this.#tasks.resume(); return this.#tasks.waitForIdle(undefined, context); } diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 23951e899..0ce1b3de3 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -9,7 +9,7 @@ import { type TelemetryCarrier, } from "@earendil-works/pi-telemetry"; import type { ExecutionEnv } from "../env/index.ts"; -import { errorMessage } from "../errors.ts"; +import { errorMessage, ReadOnlySession } from "../errors.ts"; import type { SessionImpl } from "../session/session.ts"; import type { Transaction } from "../session/transaction.ts"; import type { Timers } from "../timers.ts"; @@ -226,6 +226,11 @@ export type TaskSchedulerOptions = { ) => Promise; /** Context for scheduler commits and invocations; carries no caller cancellation. */ readonly context: Context; + /** + * `HarnessOptions.readOnly`: open loads live tasks as committed, `running` ones included, and the scheduler never + * commits, runs anything, or asks for a wakeup. + */ + readonly readOnly?: boolean; }; /** @@ -340,6 +345,7 @@ export class TaskScheduler { #cascadePending = false; #unsubscribeRegistry: () => void = () => {}; #enabled = false; + readonly #readOnly: boolean; #closing = false; #dirty = false; #draining = false; @@ -365,11 +371,13 @@ export class TaskScheduler { this.#withdrawInputs = options.withdrawInputs; this.#conversation = options.conversation; this.#context = options.context; + this.#readOnly = options.readOnly === true; } /** * Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing. A - * task that reaches the restart limit this way is parked and reported. + * task that reaches the restart limit this way is parked and reported. Read-only, it only loads them: a `running` + * task stays as committed, running in another process. */ async open(context: Context): Promise { this.#session.observeCommits((publication) => this.#observe(publication)); @@ -390,6 +398,7 @@ export class TaskScheduler { for (const record of records) { this.#track(undefined, record); this.#resumed.add(record.id); + if (this.#readOnly) continue; if (record.abandonOnRestart === true) this.#abandoned.add(record.id); if (record.state.status === "running") { const restarts = (record.restarts ?? 0) + 1; @@ -410,6 +419,7 @@ export class TaskScheduler { ); this.#report(error, { type: "task_parked", ...this.#aboutTask(recordTask(record)) }); } + if (this.#readOnly) return; // Derive abort marks a crash left unapplied below cancelled owners, and finalize held outcomes. this.#cascadePending = true; this.#scheduleReconcile(); @@ -421,6 +431,7 @@ export class TaskScheduler { * Idempotent; the kick does nothing once closing. */ resume(): void { + if (this.#readOnly) throw new ReadOnlySession("running tasks"); this.#enabled = true; this.#wakeDue(); this.#kick(); @@ -475,6 +486,8 @@ export class TaskScheduler { * sleep not yet due. Synchronous, from memory; a task whose chain is not loaded yet counts as it would alone. */ wake(): Wake { + // A read-only Harness runs nothing, so it never needs a process. + if (this.#readOnly) return { active: false }; const now = this.#now(); // A reconcile pass with work to do is work too: it finalizes held outcomes and marks cascades, at open as // recovery. A process that dies before it lands must be woken again. A pass with nothing to do is not. @@ -762,7 +775,8 @@ export class TaskScheduler { // ─── Ownership ─────────────────────────────────────────────────────────── #scheduleReconcile(): void { - if (this.#reconcileScheduled || this.#closing) return; + // A read-only reconcile could only fail: what it writes is refused, and its own commits fail the Session. + if (this.#reconcileScheduled || this.#closing || this.#readOnly) return; this.#reconcileScheduled = true; queueMicrotask(() => void this.#reconcile()); } @@ -1314,7 +1328,7 @@ export class TaskScheduler { /** Recompute the wake in a microtask, off the Session line: arm the built-in timer, or tell `onWake` of a change. */ #scheduleWake(): void { - if (this.#wakeScheduled || this.#closing) return; + if (this.#wakeScheduled || this.#closing || this.#readOnly) return; this.#wakeScheduled = true; queueMicrotask(() => { this.#wakeScheduled = false; diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 6935fb797..545f12e6b 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -660,6 +660,14 @@ export type HarnessOptions = { * Its own throw or rejection is dropped. */ readonly onReport?: (error: unknown, info: ReportInfo) => void; + /** + * Open for reading only, as a viewer of a session that another process drives, for example a synced copy on a device + * that only shows it. Open recovers nothing: a `running` task stays as committed. Every commit with writes rejects + * with `ReadOnlySession` before it reaches Storage, and so does `resume()` and every call that enables scheduling; + * the Harness stays usable for reads and watches. `wake()` is always inactive, and `onWake` is never called. Reopen + * without it to write. + */ + readonly readOnly?: boolean; }; /** @@ -779,9 +787,10 @@ export type ConversationWatch = WatchHandle; /** Durable agent harness over one Session. */ export interface Harness extends Session { /** - * Enable task scheduling. Idempotent; throws after close. Calls that ask for progress enable it too: - * `Conversation.submit()`, `Conversation.compact()`, `Conversation.abort()`, `Submission.wait()`, `waitForTask()`, - * `Harness.waitForIdle()`, and `Conversation.waitForIdle()`. Read-only viewers never do. + * Enable task scheduling. Idempotent; throws after close, and `ReadOnlySession` on a read-only Harness. Calls that + * ask for progress enable it too: `Conversation.submit()`, `Conversation.compact()`, `Conversation.abort()`, + * `Submission.wait()`, `waitForTask()`, `Harness.waitForIdle()`, and `Conversation.waitForIdle()`. Read-only viewers + * never do. */ resume(): void; /** diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 5012b44d1..60dad5337 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -9,7 +9,7 @@ export { ToolResultEntry, UserEntry, } from "./entries.ts"; -export { ConversationBusy, ReadAfterWrite, SessionFailed, StorageRequestError } from "./errors.ts"; +export { ConversationBusy, ReadAfterWrite, ReadOnlySession, SessionFailed, StorageRequestError } from "./errors.ts"; export { AgentDoc, configure, diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index d87507e54..99513ce91 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -9,7 +9,7 @@ import { materializeDocument, resolveAddress, } from "../documents.ts"; -import { SessionFailed, StorageRequestError } from "../errors.ts"; +import { ReadOnlySession, SessionFailed, StorageRequestError } from "../errors.ts"; import { idFromNumber } from "../ids.ts"; import { hostTimers, type Timers } from "../timers.ts"; import type { @@ -121,6 +121,8 @@ export class SessionImpl implements Session { readonly #attachedDocuments = new Map(); readonly #clock: () => number; readonly #documentRetentionMs: () => number; + /** Refuse every commit with writes (`HarnessOptions.readOnly`). */ + readonly #readOnly: boolean; readonly #timers: Timers; /** The pending eviction: a timer for the oldest idle document, or a line job when the retention is 0. */ #documentSweep: { readonly at: number; disarm: (() => void) | undefined } | undefined; @@ -158,9 +160,11 @@ export class SessionImpl implements Session { now: () => number = Date.now, documentRetentionMs: () => number | undefined = () => undefined, setTimer?: SetTimer, + readOnly = false, ) { // Waits race it; none has to. this.#failed.promise.catch(() => {}); + this.#readOnly = readOnly; this.#storage = new GuardedStorage(storage, this); this.#timers = hostTimers(setTimer, (error) => this.report(error, { type: "host_error", callback: "setTimer" })); // Cache bookkeeping must never change the outcome of the operation it follows: a clock or a retention setting @@ -675,6 +679,12 @@ export class SessionImpl implements Session { this.#sweepDocuments(); return result; } + if (this.#readOnly) { + // Refused before Storage: the Session stays healthy, and the caller sees why. + tx.discard(); + this.#sweepDocuments(); + throw new ReadOnlySession("a commit with writes"); + } // Storage adopts the batch synchronously once it is durable, before its next commit; a throw here would reach // Storage, so the callback keeps its own error. let adoption: { readonly documents: DocumentCommitChange[] } | { readonly error: unknown } | undefined; diff --git a/packages/durable/test/harness-read-only.test.ts b/packages/durable/test/harness-read-only.test.ts new file mode 100644 index 000000000..da6539789 --- /dev/null +++ b/packages/durable/test/harness-read-only.test.ts @@ -0,0 +1,144 @@ +// A read-only Harness: a viewer of a session another process drives, such as a synced copy on another device. +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { Harness, LiveDoc, ReadOnlySession, UserEntry, type Wake } from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { allEntries, type ChatSetup, chatSetup, openChat, unanswered } from "./chat-support.ts"; +import { context } from "./session-support.ts"; + +const directories = new Set(); + +afterEach(async () => { + for (const directory of directories) await rm(directory, { recursive: true, force: true }); + directories.clear(); +}); + +async function sqlitePath(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-read-only-")); + directories.add(directory); + return join(directory, "session.sqlite"); +} + +async function openReadOnly(path: string, setup: ChatSetup, onWake?: (wake: Wake) => void): Promise { + return Harness.open( + await openNodeSqliteStorage(path), + { + models: setup.models, + registry: setup.registry, + readOnly: true, + onReport: (error) => setup.reports.push(error), + ...(onWake === undefined ? {} : { onWake }), + }, + context, + ); +} + +/** The next sequence the file would commit, read beside every open handle; `undefined` before the schema exists. */ +function head(path: string): number | undefined { + const db = new DatabaseSync(path); + try { + const exists = db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'durable_metadata'").get(); + if (exists === undefined) return undefined; + return Number(db.prepare("SELECT next_seq FROM durable_metadata WHERE singleton = 1").get()?.next_seq); + } finally { + db.close(); + } +} + +/** A session whose root generation is running in another Harness, still open, waiting for the model. */ +async function runningSession(path: string, setup: ChatSetup) { + const busy = unanswered(); + setup.faux.setResponses([busy.step]); + const driver = await openChat(await openNodeSqliteStorage(path), setup); + await driver.root.submit({ type: "input", content: "slow" }, context); + await busy.reached; + return driver; +} + +describe("read-only Harness", () => { + it("opens a session with work running elsewhere without writing", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const driver = await runningSession(path, setup); + const before = head(path); + const wakes: Wake[] = []; + + const viewer = await openReadOnly(path, chatSetup(), (wake) => wakes.push(wake)); + const root = await viewer.root(context); + expect((await allEntries(root)).map((entry) => entry.kind)).toEqual([UserEntry.kind]); + const run = (await viewer.snapshot(LiveDoc, root.id, context))?.run; + expect(run).toBeDefined(); + expect((await viewer.getTask(run!.taskId, context))?.state.status).toBe("running"); + const inspection = await viewer.inspect(context); + expect(inspection.scheduling).toBe("paused"); + expect(viewer.wake()).toEqual({ active: false }); + await viewer.close(context); + + expect(head(path)).toBe(before); + expect(wakes).toEqual([]); + await driver.harness.close(context); + }); + + it("refuses writes and scheduling, and stays usable", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + setup.faux.setResponses([fauxAssistantMessage("one")]); + const driver = await openChat(await openNodeSqliteStorage(path), setup); + await (await driver.root.submit({ type: "input", content: "hi" }, context)).wait(context); + await driver.harness.close(context); + const before = head(path); + + const viewerSetup = chatSetup(); + const viewer = await openReadOnly(path, viewerSetup); + const root = await viewer.root(context); + await expect(root.commit((tx) => tx.appendEntry(root.id, { kind: "app.note" }), context)).rejects.toBeInstanceOf( + ReadOnlySession, + ); + await expect(root.submit({ type: "input", content: "again" }, context)).rejects.toBeInstanceOf(ReadOnlySession); + expect(() => viewer.resume()).toThrow(ReadOnlySession); + await expect(viewer.waitForIdle(context)).rejects.toBeInstanceOf(ReadOnlySession); + await expect(root.abort(context)).rejects.toBeInstanceOf(ReadOnlySession); + await expect(viewer.createConversation({ ownership: { kind: "ownerless" } }, context)).rejects.toBeInstanceOf( + ReadOnlySession, + ); + + // Still healthy: reads work, and a commit that writes nothing succeeds. + expect(await root.commit(() => "nothing", context)).toBe("nothing"); + expect((await allEntries(root)).length).toBe(2); + await viewer.close(context); + expect(head(path)).toBe(before); + expect(viewerSetup.reports).toEqual([]); + }); + + it("refuses to create the root of an empty storage", async () => { + const path = await sqlitePath(); + const viewer = await openReadOnly(path, chatSetup()); + await expect( + viewer.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } }), + ).rejects.toBeInstanceOf(ReadOnlySession); + await viewer.close(context); + expect(head(path)).toBe(1); + }); + + it("leaves recovery to the next writable open", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const driver = await runningSession(path, setup); + const taskId = (await driver.harness.snapshot(LiveDoc, driver.root.id, context))!.run!.taskId; + + const viewer = await openReadOnly(path, chatSetup()); + await viewer.close(context); + + // The writable open recovers the task once, as if the viewer had never opened it. + const writer = await openChat(await openNodeSqliteStorage(path), chatSetup()); + const record = await writer.harness.getTask(taskId, context); + expect(record?.state.status).toBe("pending"); + expect(record?.restarts).toBe(1); + await writer.harness.close(context); + await driver.harness.close(context); + }); +});