From be14d9732f98fcc33929cf103a92b96afaa6523e Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Thu, 22 Oct 2026 14:24:58 +0200 Subject: [PATCH] feat(durable): open a Harness read-only HarnessOptions.readOnly opens a viewer of a session another process drives, such as a synced copy on a device that only shows it. Open leaves running tasks as committed, commits with writes reject with ReadOnlySession before they reach Storage, and resume() and every call that would run tasks reject with it. --- packages/durable/CHANGELOG.md | 4 + packages/durable/README.md | 2 +- packages/durable/docs/spec.md | 9 ++ packages/durable/src/errors.ts | 11 ++ packages/durable/src/harness/harness.ts | 13 +- packages/durable/src/harness/scheduler.ts | 22 ++- packages/durable/src/harness/types.ts | 15 +- packages/durable/src/index.ts | 2 +- packages/durable/src/session/session.ts | 12 +- .../durable/test/harness-read-only.test.ts | 144 ++++++++++++++++++ 10 files changed, 218 insertions(+), 16 deletions(-) create mode 100644 packages/durable/test/harness-read-only.test.ts 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); + }); +});