From 6c2fd73b0c545a4034672cb60e34d53e035dd751 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sat, 24 Oct 2026 17:49:30 +0000 Subject: [PATCH] feat(durable): deferred storage of throttled commits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A commit may be marked deferrable ({ persistence: { kind: "deferred", withinMs? } }) on Session, Conversation, TaskRuntime and a tool's api.commit. When every write changes an existing conversation document with history "latest" and fork "initial", it is adopted in memory and shown to watches at once, and stored with the next commit that reads or writes the document, before the document leaves memory, at close, or within withinMs (default settings.progress.persistWithinMs, 1000 ms). The generation partial throttle and the tool output throttle use it. subscribeCommits() gets these commits only once stored, with exactly what was stored; internal observers get memory-only publications. Spec §1/§13 changed, new §4.1. 32 streams over SQLite: storage transactions 562/s -> 99/s, disk 14.3 -> 4.9 MB/s, CPU 29% -> 22% of one core. A crash loses up to about 1 s of streamed content. --- packages/durable/CHANGELOG.md | 2 + packages/durable/README.md | 18 +- packages/durable/docs/spec.md | 172 ++++- packages/durable/src/harness/agent.ts | 4 +- packages/durable/src/harness/events.ts | 4 +- packages/durable/src/harness/generation.ts | 32 +- packages/durable/src/harness/harness.ts | 16 +- packages/durable/src/harness/json.ts | 12 +- packages/durable/src/harness/live.ts | 18 +- .../durable/src/harness/partial-changes.ts | 52 +- packages/durable/src/harness/scheduler.ts | 13 +- packages/durable/src/harness/submissions.ts | 16 +- packages/durable/src/harness/task-graph.ts | 17 +- packages/durable/src/harness/tool.ts | 57 +- packages/durable/src/harness/types.ts | 19 +- packages/durable/src/harness/view.ts | 10 +- packages/durable/src/index.ts | 2 + packages/durable/src/session/session.ts | 295 +++++++- packages/durable/src/session/transaction.ts | 192 ++++- packages/durable/src/types.ts | 35 +- .../test/harness-deferred-progress.test.ts | 444 ++++++++++++ packages/durable/test/harness-events.test.ts | 10 +- .../durable/test/harness-generation.test.ts | 11 +- .../durable/test/harness-live-deltas.test.ts | 6 +- .../test/harness-partial-changes.test.ts | 12 +- .../durable/test/harness-registry.test.ts | 1 + .../durable/test/harness-structured.test.ts | 3 + .../test/harness-tools-recovery.test.ts | 2 +- .../test/session-deferred-commits.test.ts | 653 ++++++++++++++++++ packages/durable/test/session-support.ts | 22 +- 30 files changed, 1944 insertions(+), 206 deletions(-) create mode 100644 packages/durable/test/harness-deferred-progress.test.ts create mode 100644 packages/durable/test/session-deferred-commits.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 41abda519..eff3b87a9 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -9,9 +9,11 @@ - 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. - Document, conversation view, and task graph watches deliver the commits made while their listener runs (or before `start()`) in one callback: the newest value, the commits' operations folded (appends to one string join, sets of one field keep the last) and the newest commit's Context. Past 10,000 pending operations the callback gets one root replacement `[["r", newest]]`. Before, each commit was its own callback and more than 100 pending commits became a root replacement. `watchEvents()` still delivers every event batch in order, with one `snapshot` past 100 pending batches. - 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()`. +- Partial answers and running tool output are stored later than they are shown: their throttled commits are deferred (`CommitPersistence`, spec §4.1), so watches, views, and agent events see each at once, while Storage gets them together with the next commit that reads `pi.live` (the end of the answer, a tool settling, a run ending) and at the latest after `settings.progress.persistWithinMs` (default 1,000 ms). With 32 streams over SQLite this cuts storage transactions about 5.7x and disk writes about 2.9x for turns with parallel tools, 27x and 9x for long answers, and CPU by about a quarter. A crash now loses up to about a second of streamed content instead of 100 ms; a stream that crashes before its first store leaves no aborted entry, and that attempt's usage is not recorded. `subscribeCommits()` gets these commits only once they are stored, with exactly what was stored (held operations first), and gets a publication for a deadline's store too; it no longer sees every partial. Migration: a `subscribeCommits()` listener that showed live progress should use `watchDoc()`, `Conversation.watch()`, or `watchEvents()`; one that mirrors Storage keeps working; set `settings.progress.persistWithinMs: 0` to store progress right after each commit, about as before (one storage transaction per partial). ### Added +- Deferred commits: `Session.commit()`, `Conversation.commit()`, `TaskRuntime.commit()`, and a tool's `api.commit()` take `{ persistence: { kind: "deferred", withinMs? } }`. A commit that changes only existing conversation documents with `history: "latest"` and `fork: "initial"` is then adopted in memory and shown to watches at once, and stored with the next commit that reads or writes the document, before the document leaves memory, at close, or within `withinMs` (default `settings.progress.persistWithinMs`, or `createSession(storage, { persistWithinMs })`); any other write makes it immediate. A crash loses what is held. See spec §4.1. - `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. ### Changed diff --git a/packages/durable/README.md b/packages/durable/README.md index 257faa8d1..452b7c740 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -413,7 +413,7 @@ const harness = await Harness.open(storage, { stream: { timeoutMs: 120_000 }, retry: { maxRetries: 3 }, compaction: { reserveTokens: 16384 }, - progress: { partialIntervalMs: 100, outputIntervalMs: 100 }, + progress: { partialIntervalMs: 100, outputIntervalMs: 100, persistWithinMs: 1_000 }, toolExecution: "parallel", contextRetentionMs: 600_000, // idle conversations keep their context in memory this long (checked lazily where background timers are declined, as in Workers) documentRetentionMs: 600_000, // loaded documents nothing observes stay in memory this long after their last use; observed ones always @@ -492,7 +492,7 @@ Watches are Chord's `WatchHandle` (exported by `@earendil-works/chord`; Durable Commits that land while a callback runs are delivered together in the next callback: the newest view and the operations of all of them, folded (appends to one string join, sets of one field keep the last). Past 10,000 pending operations the callback gets one root replacement `[["r", view]]` holding the whole newest view instead. A client that joins late or reconnects starts from the current view; nothing is replayed. -Partial answers and tool output are committed at most every 100 ms by default, so a crash loses at most that window. `settings.progress` changes the intervals; a host whose storage is remote can commit less often, for example `{ partialIntervalMs: 500, outputIntervalMs: 500 }`. +Partial answers and tool output are committed at most every 100 ms by default, and watches see each commit at once. Storage gets them later and together, as deferred commits (see [Your Own State](#your-own-state)): with the next commit that reads `pi.live`, such as the end of the answer or of a tool, and at the latest after `settings.progress.persistWithinMs` (default 1,000 ms). A crash loses at most about that window of an answer or of tool output; a stream that crashes in its first second may leave no aborted entry, and that attempt's usage is not recorded. `settings.progress` changes the intervals; a host whose storage is remote can commit less often, for example `{ partialIntervalMs: 500, outputIntervalMs: 500 }`. ## Busy Conversations @@ -739,6 +739,20 @@ await root.commit(async (tx) => { console.log(await harness.snapshot(Todos, root.id, context)); ``` +A commit whose changes are cheap to lose, such as your own streamed progress, can be shown at once and stored later: + +```typescript +await root.commit( + async (tx) => { + (await tx.doc(Todos, root.id)).items.push("draft"); // a document like Todos: latest, fork "initial" + }, + context, + { persistence: { kind: "deferred", withinMs: 1_000 } }, // without withinMs: settings.progress.persistWithinMs +); +``` + +Watches and the view see it at once. Storage gets it with the next commit that reads or writes the document, before the document leaves memory, at close, and at the latest about `withinMs` later; `subscribeCommits()` sees it then, with exactly what was stored. A crash loses what was not stored yet, so derive nothing durable from such a value outside a commit that reads it. Only a commit that changes nothing but existing documents with `history: "latest"` and `fork: "initial"` is deferred; anything else in it, such as an entry, a new document, or a task state, stores it at once. + `harness.watchDoc()` watches one document like the view above. `HarnessOptions.conversationCreated(tx, conversation)` runs in every commit that creates or forks a conversation, including a tool's raw `tx.createConversation()`, so every conversation gets your documents; `init` in `createConversation()`, `fork()`, and `root()` writes per-call data in the same commit. An extension's tools, sections, and hooks read their own documents through `api` or `input.read`, and treat an absent one as its default ([`10-extension-state.ts`](test/examples/10-extension-state.ts)). ## Usage and Cost diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index d63ba63e5..1b95481b1 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -41,7 +41,8 @@ type once the PR lands. The core rule is: > A Session atomically commits immutable entries, full task records, and -> Chord-tracked documents. Only committed state is observable. +> Chord-tracked documents. Only committed state is observable. A commit marked +> deferred may be observed before it is stored (section 4.1). ## 1. Terms and invariants @@ -63,8 +64,16 @@ The core rule is: Required invariants: 1. One Session commit is atomic across all record and document writes. -2. A document update is published only after its storage commit succeeds. -3. All visible progress is durable. There is no volatile publication path. +2. A document update is published only after its storage commit succeeds, + except a deferred commit's (section 4.1): it is published to watches and the + conversation view when memory adopts it, and to `subscribeCommits()` once it + is stored. +3. All visible progress is durable, except what a deferred commit holds: what + watches see can be ahead of Storage by up to that commit's window + (`withinMs`), and a crash loses it. A commit is deferred only when its caller + marks it so, and only when it changes nothing but existing latest, + initial-fork conversation documents (section 4.1). Every stored document is + a state memory had. 4. External effects do not run inside the Session mutation transaction. 5. Entries and IDs are immutable and never reused after a committed write. 6. Document drafts are fully revoked when their transaction callback settles: @@ -649,6 +658,8 @@ interface Conversation { commit( change: (tx: Tx) => T | Promise, context: Context, + /** As for `Session.commit()` (section 4.1). */ + options?: CommitOptions, ): Promise; /** With `at`, the context as of that visible entry, as `fork(at)` would start with. */ context(context: Context, options?: { readonly at?: EntryId }): Promise; @@ -1286,8 +1297,14 @@ There is no mutable `session.document()` API. ```ts type SessionEnd = { readonly reason: "closed" } | { readonly reason: "failed"; readonly error: unknown }; +type CommitPersistence = + | { readonly kind: "immediate" } + | { readonly kind: "deferred"; readonly withinMs?: number }; +type CommitOptions = { readonly persistence?: CommitPersistence }; + interface Session extends DocumentObserver { - commit(change: (tx: Tx) => T | Promise, context: Context): Promise; + /** `options.persistence` may defer storing the commit (section 4.1). */ + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise; close(context: Context): Promise; /** Settles once closed, Storage included: `{ reason: "closed" }`, or `{ reason: "failed", error }` (section 10.1). */ readonly closed: Promise; @@ -1482,6 +1499,12 @@ callback succeeds with no pending acquisition prepare every open change -> immutable next revision + self-contained Chord Op[] Session evaluates each required/ordinary document write exactly once Storage.commit persists the atomic batch while the Session line remains held +commit deferred and holdable (section 4.1) + adopt every prepared change by pointer swap; queue its ops per document, unstored + watches and view mounts capture it synchronously; subscribeCommits() does not + release the line; invoke listeners later +otherwise, right before Storage admission + take the held ops of every document the commit acquired and put them in front of its own storage succeeds adopt every prepared change by pointer swap and enqueue immutable revision/ops publication commit listeners, including conversation view mounts (section 9.3), capture it synchronously @@ -1552,8 +1575,11 @@ The Session evaluates this predicate exactly once after tracker preparation. base, excluding the change being evaluated. Storage reports it when it materializes the current value; the Session advances it after each adopted write and resets it after every base, so evaluation performs no Storage read. -Creation and version transitions require bases and do not call it. The Session -then gives Storage only the selected representation: +Creation and version transitions require bases and do not call it. A deferred +commit's predicate also runs once, when it is held, against the stored count; a +base it selects is stored with the held changes, as a store whose held +operations overflowed is (section 4.1). The Session then gives Storage only the +selected representation: ```text required or predicate true -> base with complete value @@ -1698,14 +1724,14 @@ line through callback execution, preparation, storage settlement, committed baseline adoption, and publication enqueue. Adoption runs inside `Storage.commit()`: the Session passes an `adopt(seq)` callback, which the storage calls synchronously once the batch is durable and before its next -commit (section 10), so memory always shows a complete prefix of the log. The -callback keeps its own error: a failed adoption fails the Session once +commit (section 10), so memory always shows a complete prefix of the log, plus +what deferred commits hold (section 4.1). The callback keeps its own error: a failed adoption fails the Session once `Storage.commit()` returns. A commit that rejects after its adoption fails the Session too, and its adopted changes stay adopted. Commit observers run on the line; watch listeners run later. External model, process, tool, network, and human effects run outside it. -`subscribeCommits()` observes complete immutable publications synchronously on -the line after adoption. `subscribeClose()` observes close synchronously when it +`subscribeCommits()` observes stored commits synchronously on the line after +adoption, each with exactly what that storage transaction stored (section 4.1). `subscribeClose()` observes close synchronously when it begins, after admission is sealed, also when a failure closes the Session; the Harness ends waits, watches, and streams there, and signals task invocations after every close listener has run, so what a task holds ends with the Session's @@ -1747,6 +1773,84 @@ Storage ownership: - Every storage read returns a detached JSON value. - Immutable strings may be shared; mutable arrays and objects may not. +### 4.1 Deferred commits + +A commit whose writes are only cheap to lose, such as a streamed answer or +running tool output, may be stored later than it is shown: + +```ts +await runtime.commit( + async (tx) => { (await tx.doc(LiveDoc, conversationId)).generation!.message = partial; }, + context, + { persistence: { kind: "deferred", withinMs: 1_000 } }, +); +``` + +`Session.commit()`, `Conversation.commit()`, `TaskRuntime.commit()`, and a +tool's `api.commit()` take the option; without it, or +with `{ kind: "immediate" }`, a commit is stored before it resolves. `withinMs` +is a number of milliseconds, `0` or more (`Infinity` waits for one of the other +triggers below); anything else rejects with a `TypeError`. Without it, a +Harness uses `settings.progress.persistWithinMs` (default 1,000) and a plain +Session its `persistWithinMs` option (default 1,000). + +The mark is a permission. A deferred commit is **held** only when, after its +callback, settlement, and the read-only check, every write it makes changes an +existing conversation document with `history: "latest"` and `fork: "initial"` +at its stored version. Anything else, such as an entry, a task state, a +submission, another document, a document's creation, migration, or retirement, +makes it immediate, a normal commit. A held commit is adopted in memory by +pointer swap, resolves, and is published to the Session's own observers: document +watches, the conversation view and agent events (section 9). Its stored +version and delta count stay; its operations wait, folded per incarnation +(Chord's `PendingOps`), and the incarnation is pinned in the cache, so idle +eviction skips it at any retention. + +What is held is stored by the first of: + +1. a normal commit that acquires the document (`tx.doc()`), whether it changes + it or only reads it, and stores anything: the held operations go in front of + its own in the one content write of that incarnation, in its storage + transaction. A stored delta of a held document therefore always begins with + its held operations; a base the commit writes covers them; +2. the document's deadline, the earliest `withinMs` deadline of its held + commits on the Session clock, kept by one foreground timer (section 5.6); + when it fires, every held document is stored in one storage transaction: + a deadline is the latest time, and the others cost little more there; +3. `close()`, after the writes at close and before the trackers are dropped; + a failure of that store fails the Session, and the backend still closes; +4. any drop of a tracker that holds them: `unloadDocuments()`, and a token of + another definition version, which reloads the document from Storage; + retirement is case 1. + +Held operations leave their queue only at the last step before +`Storage.commit()`, after every fallible step: a callback, settlement, or +checkpoint failure of a normal commit keeps them. A store fails like any +storage write: it fails the Session (section 10.1). Past Chord's +`MAX_PENDING_OPS` (10,000) held operations, or when a held commit's +`checkpointWhen` selected a base, the store writes a base of the current value. +A failed Session stores nothing; after a reopen nothing is held. + +Publication: the Session's own observers see each held commit once, when memory +adopts it, and of a normal commit that stores held changes only its own +operations. `subscribeCommits()` sees every storage transaction with exactly +what it stored: nothing for a held commit, and the stored operations, held +first, for a commit or a deadline that stores them. A consumer that applies +`subscribeCommits()` publications to stored state, such as a replica, stays +equal to Storage. Its `seq` values have gaps where nothing was published. + +Consistency: every stored document is a state it had in memory, and a normal +commit that reads a held document through `tx` stores it with whatever it +derives. What a crash loses is a held commit's changes: up to `withinMs` plus the +line and I/O delay. A value read outside a commit (`snapshot()`, a watch) may be +ahead of Storage; nothing durable may be derived from it without reading it in +a commit. Other readers of Storage, such as another process, a replica, or an +export, see held changes once they are stored. + +The Harness defers exactly two kinds of commits: generation partials (section +8.3) and running tool progress (section 7.3). Every commit that consumes or +clears them reads `pi.live` and so stores them first (section 8.2). + ## 5. Tasks ### 5.1 Definition @@ -1853,6 +1957,8 @@ interface TaskRuntime extends DocumentObserver, Docum current: RunningTask, ) => NextTaskState | undefined | Promise | undefined>, context: Context, + /** May defer storing a commit that changes only documents (section 4.1); one that returns a state is stored. */ + options?: CommitOptions, ): Promise; memo(name: string, context: Context): Promise; @@ -1937,7 +2043,8 @@ Harness is closing, the task is terminal, or a run invocation's task carries an abort mark. Its `tx.createTask()` defaults to the task's conversation, and every entry it appends records the task as `byTaskId`. When the callback returns a state, the runtime replaces the task's state in the same -commit, so the checkpoint or outcome is atomic with the callback's entries, +commit (stored at once even when `options` defers it, section 4.1), so the +checkpoint or outcome is atomic with the callback's entries, documents, and child tasks and is type-checked against the task's checkpoint and result types. Returning nothing leaves the state unchanged. A terminal or completing state drops the memos. A `waiting` or `sleeping` state ends the @@ -2448,7 +2555,8 @@ commits and task changes. The default declines them in Cloudflare Workers, since a pending timer keeps a Durable Object from hibernating or being evicted even when `nodejs_compat` gives it an `unref()`, and wherever timers cannot be unreferenced. Foreground timers pace or resume work: throttled -progress commits through `TaskRuntime.setTimer()`, and, without `onWake`, the +progress commits through `TaskRuntime.setTimer()`, the deadline of deferred +commits (section 4.1), and, without `onWake`, the timer for the earliest sleep while scheduling runs. That one keeps a Node.js process alive, since a sleeping task is work left to do. A timer should not fire early and may fire late; every caller checks its clock again when it fires. The @@ -3087,7 +3195,8 @@ interface ToolExecutionApi extends Docum retainedOutput(): { readonly text: string; readonly truncated: boolean }; diagnostic(diagnostic: ToolDiagnostic): void; details(value: TDetails, context: Context): Promise; - commit(change: (tx: Tx) => T | Promise, context: Context): Promise; + /** A commit of the call's task; `options` may defer storing it (section 4.1). */ + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise; memo(name: string, context: Context): Promise; memo(name: string, candidate: T, context: Context): Promise; createTask( @@ -3424,7 +3533,8 @@ canonical path before opening it does not prevent a concurrent rename or symlink swap, so callers that restrict paths do so for hygiene, not security. The `details()` promise resolves after the corresponding or coalesced document -commit. During normal settlement the tool task stops its throttle and awaits the +commit, which memory adopts at once and Storage gets with the next commit that +reads `pi.live` or within `settings.progress.persistWithinMs` (section 4.1). During normal settlement the tool task stops its throttle and awaits the commit in flight; the terminal commit, which appends the result entry, is the final flush and settles any `details()` promise still pending. Abort and close obey invocation and Session admission gates: uncommitted buffered updates may be @@ -3964,8 +4074,12 @@ continuation calls `models.fetchDeferred()` and `models.cancelDeferred()` with that same model and signal. Missing models and synchronous/streamed pi-ai errors are classified into the durable generation outcomes below. -Generation and tool progress are throttled durable document commits. A crash may -lose only the uncommitted throttle window. Recovery converts committed partials +Generation and tool progress are throttled document commits, deferred +(section 4.1): watches see each at once, and Storage gets them with the next +commit that reads `pi.live` or within `settings.progress.persistWithinMs` +(default 1,000 ms). A crash may lose the throttle window plus what was not +stored yet: a stream that crashes before its first store leaves no aborted entry, +and that attempt's usage is not recorded. Recovery converts stored partials to normal interrupted/aborted transcript entries, clears presentation state, and then retries or terminates according to the task phase. Retry deadlines, attempts, compaction, and tool progress are current document state for late @@ -4108,7 +4222,8 @@ each commit writes at most one window, and the throttle bounds the rate. Generation creates `tools` with one `pending` slot per call in the commit that appends the tool-calling answer; a call its request did not offer starts `done` with its result entry. A tool task sets its slot `running` in its intent commit, -publishes throttled output and details into it, and in its terminal commit sets +publishes throttled output and details into it with deferred commits (section +4.1), and in its terminal commit sets `done` and `entry` and removes `output`, `droppedBytes`, `droppedLines`, `details`, and `diagnostics`, which the result entry now carries (section 7.3). The generation's `tools` phase removes `tools`. Slot updates apply only while a slot with the task's @@ -4137,6 +4252,17 @@ and resolves the transaction's latest candidate submission record, falling back to committed state, during assembly, like task-document validation (section 3.3), so it is not a caller table read and works after the commit's first table write. +The streamed fields, `generation.message` and a running slot's `output`, +`droppedBytes`, `droppedLines`, `details`, and `diagnostics`, are written by +deferred commits (section 4.1). Every commit that derives anything from them or +clears them acquires `pi.live`: converting a partial into an aborted entry, +settling a tool from its slot, handing a generation over to its tool round, +ending a run, and clearing a safe rerun's progress. It therefore stores them +first, with its own changes, and after a crash Storage holds a `pi.live` its +memory had: control fields as of the last normal commit, streamed fields as of +the last store. Recovery's own conversions and clears are normal commits, so a +second crash never converts the same partial twice. + ### 8.3 Generation ```ts @@ -4233,7 +4359,9 @@ The run's inputs live in `pi.live.run`, not in the task input. model context through `cutoff` with the invocation signal, the thinking level as `reasoning` (omitted for `off`), the conversation's persisted provider `sessionId`, and the pinned `streamOptions`, committing throttled partials at - most every `settings.progress.partialIntervalMs` (default 100 ms). + most every `settings.progress.partialIntervalMs` (default 100 ms), deferred + (section 4.1), so Storage gets them at the next commit that reads `pi.live` + or within `settings.progress.persistWithinMs`. Before streaming, the `beforeRequest` chain may replace the messages for this request only. Recovery resends the same committed messages with the same pinned model, thinking level, and stream @@ -4766,8 +4894,8 @@ The Harness keeps at most one mount per conversation. The first `watch()` or event attachment builds its revision on the Session line from the committed active transcript and documents; the mount is dropped when its last observer detaches. A mount derives each next revision from the Session's -`subscribeCommits()` publications, like `watchDoc()`: every publication is -already durable, so the view shows only committed state. One publication that +commits as memory adopts them, like `watchDoc()`: it shows committed state, +including deferred commits before they are stored (section 4.1). One publication that touches the conversation yields one Chord batch: ```text @@ -5587,7 +5715,9 @@ These are contracts, not invitations to add defensive machinery: Durable has no: - whole-Session DOM; -- visible-undurable publication; +- visible-undurable publication beyond deferred commits (section 4.1): an + unmarked commit, and every commit that writes anything but existing latest, + initial-fork conversation documents, is stored before anything observes it; - Session-kernel semantic event journal or independently maintained event state; - session-scoped rewindable documents; - automatic checkpoint heuristic; diff --git a/packages/durable/src/harness/agent.ts b/packages/durable/src/harness/agent.ts index 64a15bb6a..7706d945b 100644 --- a/packages/durable/src/harness/agent.ts +++ b/packages/durable/src/harness/agent.ts @@ -1,6 +1,6 @@ import { copyJson, type Draft } from "@earendil-works/chord"; import { defineDoc } from "../documents.ts"; -import { documentRetention } from "../session/session.ts"; +import { documentRetention, persistWithin } from "../session/session.ts"; import type { ConversationId, ConversationRecord, Tx } from "../types.ts"; import type { Agent, @@ -34,6 +34,7 @@ export const DEFAULT_COMPACTION_POLICY: CompactionPolicy = { export const DEFAULT_PROGRESS_POLICY: ProgressPolicy = { partialIntervalMs: 100, outputIntervalMs: 100, + persistWithinMs: 1_000, }; /** The reserved section key of the agent's `instructions`. */ @@ -73,6 +74,7 @@ export function resolveSettings(settings: HarnessSettings | undefined): Settings progress: { partialIntervalMs: settings?.progress?.partialIntervalMs ?? DEFAULT_PROGRESS_POLICY.partialIntervalMs, outputIntervalMs: settings?.progress?.outputIntervalMs ?? DEFAULT_PROGRESS_POLICY.outputIntervalMs, + persistWithinMs: persistWithin(settings?.progress?.persistWithinMs), }, toolExecution: settings?.toolExecution ?? "parallel", steeringMode: settings?.steeringMode ?? "one-at-a-time", diff --git a/packages/durable/src/harness/events.ts b/packages/durable/src/harness/events.ts index 5d5d638ce..ae3e4de9d 100644 --- a/packages/durable/src/harness/events.ts +++ b/packages/durable/src/harness/events.ts @@ -1,9 +1,9 @@ import { type Context, type JsonValue, NativeWatch, type WatchEnd } from "@earendil-works/chord"; import type { Op, Path } from "@earendil-works/chord/delta"; import type { AssistantMessage, Message, Usage } from "@earendil-works/pi-ai"; +import type { ObservedCommit } from "../session/session.ts"; import type { CommitChange, - CommitPublication, ConversationId, EntryRecord, JsonObject, @@ -228,7 +228,7 @@ function translate( before: ConversationView, after: ConversationView, viewOps: readonly Op[], - publication: CommitPublication, + publication: ObservedCommit, held: Set, ): AgentEvent[] { const entries: EntryRecord[] = []; diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index d617713bd..087ec9a97 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -1,4 +1,4 @@ -import { type Context, copyJson, type Draft, type JsonValue } from "@earendil-works/chord"; +import { type Context, copyJson, type Draft } from "@earendil-works/chord"; import type { Api, AssistantMessage, @@ -31,8 +31,8 @@ import type { import { addTools, sectionExtension } from "./agent.ts"; import { compactionPolicy, createCompaction, estimateContext, selectCut } from "./compaction.ts"; import { applyBoundary, type Boundary, prepareBoundary, withdrawQueuedInputs } from "./inbox.ts"; -import { endRun, LiveDoc, type LiveState, type ToolSlot } from "./live.ts"; -import { PartialChanges } from "./partial-changes.ts"; +import { endRun, LiveDoc, type LiveState, PROGRESS_COMMIT, type ToolSlot } from "./live.ts"; +import { PartialChanges, type PartialMessage } from "./partial-changes.ts"; import { planSystemEntries, renderSections, replaySections } from "./prompt.ts"; import { ensureProviderSessionId } from "./provider.ts"; import { appendToolResult, harnessError, ToolTask, type ToolTaskResult } from "./tool.ts"; @@ -440,9 +440,10 @@ export async function convertPartial(tx: Tx, live: Draft, conversatio /** * Stream one request and return the terminal message. Partials commit as trailing writes at most every - * `progress.partialIntervalMs` (default 100 ms) with one commit in flight; `finally` stops the throttle and awaits that - * commit, so no stale partial lands after the outcome. A commit appends what text and thinking grew by and applies - * the argument changes since the last one (`PartialChanges`), so its cost follows the new text, not the strings' length. + * `progress.partialIntervalMs` (default 100 ms) with one commit in flight, deferred (`PROGRESS_COMMIT`); `finally` stops + * the throttle and awaits that commit, so no stale partial lands after the outcome. A commit appends what text and + * thinking grew by and applies the argument changes since the last one (`PartialChanges`), so its cost follows the new + * text, not the strings' length. */ async function streamResponse( runtime: Runtime, @@ -465,15 +466,20 @@ async function streamResponse( if (partial === undefined || stopped) return; inFlight = (async () => { // Copy synchronously: the provider keeps mutating its partial. The changes since the last commit belong to it. - const message = copyJson(partial, { omitUndefinedProperties: true }) as { content: JsonValue[] }; + // `copyJson()` returns untyped JSON; a copy of a message is that message's JSON representation. + const message = copyJson(partial, { omitUndefinedProperties: true }) as PartialMessage; const taken = changes.take(message); try { - await runtime.commit(async (tx) => { - const live = await tx.doc(LiveDoc, runtime.conversationId); - live.generation ??= { attempt }; - changes.commit(live.generation as Draft>, message, taken); - return undefined; - }, context); + await runtime.commit( + async (tx) => { + const live = await tx.doc(LiveDoc, runtime.conversationId); + live.generation ??= { attempt }; + changes.commit(live.generation, message, taken); + return undefined; + }, + context, + PROGRESS_COMMIT, + ); } catch (error) { changes.restore(taken); throw error; diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index d22f368cc..7ce8de9c8 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -20,6 +20,7 @@ import { contained, SessionImpl } from "../session/session.ts"; import type { Transaction } from "../session/transaction.ts"; import { hostTimers } from "../timers.ts"; import type { + CommitOptions, ConversationId, ConversationOwnership, ConversationRecord, @@ -138,8 +139,8 @@ class ConversationImpl implements Conversation { await this.#host.submissions.submit(this.id, { type: "write", entry }, context); } - commit(change: (tx: Tx) => T | Promise, context: Context): Promise { - return this.#host.harness.commitWith(change, context, { conversationId: this.id }); + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise { + return this.#host.harness.commitWith(change, context, { conversationId: this.id }, options); } context(context: Context, options?: { readonly at?: EntryId }): Promise { @@ -213,8 +214,15 @@ class HarnessImpl extends SessionImpl implements const report = safeReport(options.onReport, options.telemetry ?? NOOP_TELEMETRY_CONTEXT); 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, options.readOnly === true); + // Read at each use, as all settings are, but only these fields: they are read on every commit. + super( + rawStorage, + now, + () => hostSettings()?.documentRetentionMs, + options.setTimer, + options.readOnly === true, + () => hostSettings()?.progress?.persistWithinMs, + ); // Every component reads and writes through the Session's failure guard: `rawStorage` serves only `super()`. const storage = this.storage; this.#options = options; diff --git a/packages/durable/src/harness/json.ts b/packages/durable/src/harness/json.ts index c96119fe8..0922fe45c 100644 --- a/packages/durable/src/harness/json.ts +++ b/packages/durable/src/harness/json.ts @@ -1,8 +1,6 @@ import type { JsonValue } from "@earendil-works/chord"; import { appendText, type Draft } from "@earendil-works/chord/delta"; -type JsonContainer = Record | JsonValue[]; - /** How `assignJson()` writes one value instead of assigning it leaf by leaf. */ export type AssignPlan = /** The string grew by `text` since it was last assigned: append it, when the lengths agree. */ @@ -19,10 +17,16 @@ export type AssignPlan = * string member grew by, and the committed string plus that text has the new string's length, the text is appended * with `appendText()` instead. */ -export function assignJson(target: JsonContainer, key: string | number, value: JsonValue, plan?: AssignPlan): void { +export function assignJson( + target: T, + key: K, + value: T[K] & JsonValue, + plan?: AssignPlan, +): void { if (plan?.kind === "keep") return; + // Every target is a document container: JSON members or items, whatever its declared type names them. const slots = target as Record; - const current = slots[key]; + const current: JsonValue | undefined = slots[key]; const fields = plan?.kind === "fields" ? plan.fields : undefined; if (isRecord(current) && isRecord(value)) { for (const name of Object.keys(current)) if (!Object.hasOwn(value, name)) delete current[name]; diff --git a/packages/durable/src/harness/live.ts b/packages/durable/src/harness/live.ts index 4d5397f71..55759479f 100644 --- a/packages/durable/src/harness/live.ts +++ b/packages/durable/src/harness/live.ts @@ -2,7 +2,16 @@ import type { Draft, JsonRepresentation, JsonValue } from "@earendil-works/chord import type { AssistantMessage, Usage } from "@earendil-works/pi-ai"; import { defineDoc } from "../documents.ts"; import type { Transaction } from "../session/transaction.ts"; -import type { EntryId, JsonObject, SubmissionId, SubmissionSettlement, TaskId, TaskRecord, Tx } from "../types.ts"; +import type { + CommitOptions, + EntryId, + JsonObject, + SubmissionId, + SubmissionSettlement, + TaskId, + TaskRecord, + Tx, +} from "../types.ts"; import { convertPartial } from "./generation.ts"; import type { SchedulerOutcome } from "./scheduler.ts"; import type { CompactionReason, CompactionResult, ToolDiagnostic } from "./types.ts"; @@ -126,6 +135,13 @@ export const LiveDoc = defineDoc({ !(value.nestedTools ?? []).some((slot) => slot.status === "running"), }); +/** + * How the progress throttles commit a partial answer and running tool output into `pi.live`: deferred, so storage gets + * them together, at the next commit that reads or writes `pi.live` or after `settings.progress.persistWithinMs`. Every + * commit that consumes or clears streamed fields reads `pi.live`, so it stores them with its own changes (spec §8.2). + */ +export const PROGRESS_COMMIT: CommitOptions = { persistence: { kind: "deferred" } }; + /** Built-in task kinds that can own `pi.live.run`. */ const RUN_TASK_KINDS: ReadonlySet = new Set(["pi.generation"]); const TOOL_TASK_KIND = "pi.tool"; diff --git a/packages/durable/src/harness/partial-changes.ts b/packages/durable/src/harness/partial-changes.ts index 355a9c93c..bf1a8ab04 100644 --- a/packages/durable/src/harness/partial-changes.ts +++ b/packages/durable/src/harness/partial-changes.ts @@ -1,18 +1,25 @@ // What a streamed answer changed in its partial message between two partial commits, so a commit costs the new text. -import type { JsonValue } from "@earendil-works/chord"; +import type { JsonRepresentation, JsonValue } from "@earendil-works/chord"; import { appendText, type Draft } from "@earendil-works/chord/delta"; -import type { AssistantMessageEvent } from "@earendil-works/pi-ai"; +import type { AssistantMessage, AssistantMessageEvent, ToolCall } from "@earendil-works/pi-ai"; import { type JsonChange, StreamingJsonParser } from "@earendil-works/pi-ai/utils/streaming-json"; import { type AssignPlan, assignJson } from "./json.ts"; +import type { LiveState } from "./live.ts"; type JsonContainer = Record | JsonValue[]; +/** A partial message as `pi.live.generation.message` holds it. */ +export type PartialMessage = JsonRepresentation; + +/** Committed arguments of a streamed tool call. */ +type Arguments = JsonRepresentation["arguments"]; + /** What one partial commit writes besides the partial message itself; from `PartialChanges.take()`. */ export type PartialCommit = { /** The plan `assignJson()` writes the message with: text appends, and the arguments it leaves to `changes`. */ readonly plan: AssignPlan; /** Each tracked tool call by content index: its argument changes since the last commit, in order, and its snapshot. */ - readonly toolCalls: ReadonlyMap; + readonly toolCalls: ReadonlyMap; }; /** What a tracked tool call changed since the last commit, and whether any commit took its changes yet. */ @@ -80,16 +87,18 @@ export class PartialChanges { * Take what changed since the last call, for a commit of `message`, a copy of the partial. Clears the arguments of * tracked tool calls in `message`: the commit writes them from the changes. Call `restore()` if the commit fails. */ - take(message: { readonly content: readonly JsonValue[] }): PartialCommit { + take(message: PartialMessage): PartialCommit { const blocks = new Map(); for (const [index, { field, text }] of this.#text) { blocks.set(index, { kind: "fields", fields: new Map([[field, { kind: "append", text }]]) }); } this.#text = new Map(); - const toolCalls = new Map(); + const toolCalls = new Map(); for (const [index, call] of this.#toolCalls) { - const block = message.content[index] as Record; - toolCalls.set(index, { changes: call.changes, snapshot: block.arguments! }); + const block = message.content[index]; + // A tracked call is the tool call its `toolcall_start` named; the narrowing only tells TypeScript. + if (block?.type !== "toolCall") continue; + toolCalls.set(index, { changes: call.changes, snapshot: block.arguments }); block.arguments = {}; blocks.set(index, { kind: "fields", fields: new Map([["arguments", { kind: "keep" }]]) }); call.changes = []; @@ -106,25 +115,32 @@ export class PartialChanges { * its first piece; when the committed message lost a call whose changes were already committed (another writer reset * it), the call falls back to its snapshot from now on. */ - commit(generation: Draft>, message: JsonValue, commit: PartialCommit): void { - const committedBlocks = (generation.message as { content?: unknown[] } | undefined)?.content?.length ?? 0; + commit( + generation: Draft>, + message: PartialMessage, + commit: PartialCommit, + ): void { + const committedBlocks = generation.message?.content.length ?? 0; const apply: [number, readonly JsonChange[]][] = []; for (const [index, { changes, snapshot }] of commit.toolCalls) { const call = this.#toolCalls.get(index); if (call === undefined) continue; - if (index >= committedBlocks && call.committed) { + const block = message.content[index]; + if (index >= committedBlocks && call.committed && block?.type === "toolCall") { // The earlier changes are not in the committed message: commit the snapshot and stop tracking the call. this.#toolCalls.delete(index); - (message as { content: Record[] }).content[index]!.arguments = snapshot; + block.arguments = snapshot; continue; } call.committed = true; apply.push([index, changes]); } assignJson(generation, "message", message, commit.plan); - const content = (generation.message as unknown as { content: Draft>[] }).content; + const content = generation.message?.content ?? []; for (const [index, changes] of apply) { - for (const change of changes) applyChange(content[index]!, change); + const block = content[index]; + if (block?.type !== "toolCall") continue; + for (const change of changes) applyChange(block, change); } } @@ -138,13 +154,15 @@ export class PartialChanges { } /** Apply one argument change to the committed arguments of `block`. */ -function applyChange(block: Draft>, change: JsonChange): void { +function applyChange(block: Draft>, change: JsonChange): void { const key = change.path.at(-1); if (key === undefined) { - // Only an object root is tracked, and only `set` starts it. - if (change.kind === "set") block.arguments = value(change); + // Only an object root is tracked (`observe()` drops any other), and only `set` starts it. + const root = change.kind === "set" ? value(change) : undefined; + if (isRecord(root)) block.arguments = root; return; } + // The path follows the parsed argument text, not a type: each segment names a container `observe()` saw set. let parent = block.arguments as Draft; for (const segment of change.path.slice(0, -1)) { parent = (parent as Record>)[segment]!; @@ -161,6 +179,6 @@ function value(change: Extract): JsonValue { return change.value as JsonValue; } -function isRecord(value: JsonValue | undefined): boolean { +function isRecord(value: JsonValue | undefined): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 8fde30b63..9a7fcf2e2 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -11,11 +11,11 @@ import { } from "@earendil-works/pi-telemetry"; import type { ExecutionEnv } from "../env/index.ts"; import { errorMessage, ReadOnlySession } from "../errors.ts"; -import type { SessionImpl } from "../session/session.ts"; +import type { ObservedCommit, SessionImpl } from "../session/session.ts"; import type { Transaction } from "../session/transaction.ts"; import type { Timers } from "../timers.ts"; import type { - CommitPublication, + CommitOptions, ConversationId, DocumentWatch, EntryId, @@ -635,7 +635,9 @@ export class TaskScheduler { // ─── Scheduling ──────────────────────────────────────────────────────── - #observe(publication: CommitPublication): void { + #observe(publication: ObservedCommit): void { + // A commit held in memory changes documents only. + if (publication.kind === "memory") return; const updated: AnyTaskRecord[] = []; const failed: TaskId[] = []; let changed = false; @@ -2145,7 +2147,7 @@ export class TaskScheduler { get registry() { return phase.snapshot(); }, - commit: async (change, context) => { + commit: async (change, context, options) => { let checkpoint: string | undefined; await this.#gated( invocation, @@ -2157,6 +2159,7 @@ export class TaskScheduler { } }, context, + options, ); if (checkpoint !== undefined) { invocation.context.telemetry.addEvent("pi.durable.checkpoint", { @@ -2343,6 +2346,7 @@ export class TaskScheduler { invocation: Invocation, change: (tx: Transaction, current: ErasedRunningTask) => T | Promise, context: Context, + options?: CommitOptions, ): Promise { if (invocation.ended) return Promise.reject(endedError(invocation)); return this.#session.commitWith( @@ -2360,6 +2364,7 @@ export class TaskScheduler { }, context, { conversationId: invocation.conversationId, taskId: invocation.taskId }, + options, ); } diff --git a/packages/durable/src/harness/submissions.ts b/packages/durable/src/harness/submissions.ts index 0cf838a93..8f44e0793 100644 --- a/packages/durable/src/harness/submissions.ts +++ b/packages/durable/src/harness/submissions.ts @@ -7,16 +7,8 @@ import { type TelemetrySpan, } from "@earendil-works/pi-telemetry"; import { ConversationBusy } from "../errors.ts"; -import type { SessionImpl } from "../session/session.ts"; -import type { - CommitPublication, - ConversationId, - JsonObject, - Storage, - SubmissionId, - SubmissionRecord, - Tx, -} from "../types.ts"; +import type { ObservedCommit, SessionImpl } from "../session/session.ts"; +import type { ConversationId, JsonObject, Storage, SubmissionId, SubmissionRecord, Tx } from "../types.ts"; import { startRun } from "./generation.ts"; import { appendInput, @@ -142,7 +134,9 @@ export class Submissions { }, context); } - #observe(publication: CommitPublication): void { + #observe(publication: ObservedCommit): void { + // A commit held in memory changes documents only. + if (publication.kind === "memory") return; for (const change of publication.changes) { if (change.type !== "submission" || !isSettled(change.value)) continue; this.#waiters.resolve(change.value.id, change.value); diff --git a/packages/durable/src/harness/task-graph.ts b/packages/durable/src/harness/task-graph.ts index ec4e55a23..55ee659ed 100644 --- a/packages/durable/src/harness/task-graph.ts +++ b/packages/durable/src/harness/task-graph.ts @@ -1,16 +1,8 @@ import { type Context, type JsonValue, Watch, type WatchHandle } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT, withoutAbortSignal } from "@earendil-works/chord/context"; import { applyImmutable, type Op } from "@earendil-works/chord/delta"; -import { contained, type SessionImpl } from "../session/session.ts"; -import type { - CommitPublication, - ConversationId, - JoinPolicy, - Storage, - TaskId, - TaskOutcome, - TaskRecord, -} from "../types.ts"; +import { contained, type ObservedCommit, type SessionImpl } from "../session/session.ts"; +import type { ConversationId, JoinPolicy, Storage, TaskId, TaskOutcome, TaskRecord } from "../types.ts"; import { type BlockedReason, type BlockedSubject, blockedSubject } from "./scheduler.ts"; import { closedError, scanAll } from "./util.ts"; @@ -88,7 +80,8 @@ export class TaskGraphView { this.#storage = storage; this.#blocking = blocking; session.observeCommits((publication, context) => { - if (this.#mount === undefined) return; + // A commit held in memory changes documents only, which the graph does not show. + if (this.#mount === undefined || publication.kind === "memory") return; advance(this.#mount, publication, context, blocking.blocked, (error) => session.report(error)); }); blocking.subscribeBlocked(() => this.#refreshBlocked()); @@ -190,7 +183,7 @@ export class TaskGraphView { */ function advance( mount: Mount, - publication: CommitPublication, + publication: ObservedCommit, context: Context, blocked: TaskGraphBlocking["blocked"], report: (error: unknown) => void, diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index 215e8f063..483d3b4fe 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -38,6 +38,7 @@ import { LiveDoc, type NestedToolSlot, type NestedToolSummary, + PROGRESS_COMMIT, removeNestedSlots, type ToolSlot, toolSlot, @@ -358,12 +359,16 @@ async function run( committed.catch(() => {}); return awaitWithContext(committed, detailsContext); }, - commit: async (change, commitContext) => { + commit: async (change, commitContext, options) => { let result: Awaited> | undefined; - await runtime.commit(async (tx) => { - result = await change(tx); - return undefined; - }, commitContext); + await runtime.commit( + async (tx) => { + result = await change(tx); + return undefined; + }, + commitContext, + options, + ); return result as Awaited>; }, memo: runtime.memo, @@ -478,7 +483,7 @@ function progressClock(runtime: Runtime): ProgressClock { /** * Throttled commits of what the tool reported into its slot, in `pi.live.tools` or `pi.live.nestedTools`, each writing - * only what changed since the last one; none when `enabled` is false. + * only what changed since the last one, deferred (`PROGRESS_COMMIT`); none when `enabled` is false. */ function publishProgress(runtime: Runtime, reported: Reported, enabled: boolean, context: Context): Progress { let written = { text: "", details: undefined as JsonValue | undefined, diagnostics: 0 }; @@ -501,24 +506,28 @@ function publishProgress(runtime: Runtime, reported: Reported, enabled: boolean, } if (detailsChanged) bytes += utf8ByteLength(JSON.stringify(current.details ?? null)); if (added.length > 0) bytes += utf8ByteLength(JSON.stringify(added)); - await runtime.commit(async (tx) => { - const slot = toolSlot(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId); - if (slot === undefined) return undefined; - // REMINDER: assign `output` as one string field. Chord then diffs it into an append, or a trim plus an - // append for a sliding tail; replacing the slot object would record the whole window on every commit. - if ((slot.output ?? "") !== snapshot.text) slot.output = snapshot.text; - if (snapshot.droppedBytes > 0) slot.droppedBytes = snapshot.droppedBytes; - if (snapshot.droppedLines > 0) slot.droppedLines = snapshot.droppedLines; - // Diff details leaf by leaf and append new diagnostics, so each commit writes only what changed. - if (detailsChanged && current.details !== undefined) { - assignJson(slot as unknown as Record, "details", current.details); - } - if (added.length > 0) { - if (slot.diagnostics === undefined) slot.diagnostics = []; - for (const diagnostic of added) slot.diagnostics.push(diagnostic); - } - return undefined; - }, context); + await runtime.commit( + async (tx) => { + const slot = toolSlot(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId); + if (slot === undefined) return undefined; + // REMINDER: assign `output` as one string field. Chord then diffs it into an append, or a trim plus an + // append for a sliding tail; replacing the slot object would record the whole window on every commit. + if ((slot.output ?? "") !== snapshot.text) slot.output = snapshot.text; + if (snapshot.droppedBytes > 0) slot.droppedBytes = snapshot.droppedBytes; + if (snapshot.droppedLines > 0) slot.droppedLines = snapshot.droppedLines; + // Diff details leaf by leaf and append new diagnostics, so each commit writes only what changed. + if (detailsChanged && current.details !== undefined) { + assignJson(slot, "details", current.details); + } + if (added.length > 0) { + if (slot.diagnostics === undefined) slot.diagnostics = []; + for (const diagnostic of added) slot.diagnostics.push(diagnostic); + } + return undefined; + }, + context, + PROGRESS_COMMIT, + ); written = current; return bytes; }, diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 66d8cb75f..fbe636dd7 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -18,6 +18,7 @@ import type { import type { TelemetryContext } from "@earendil-works/pi-telemetry"; import type { ExecutionEnv, ShellOutputSkip, ShellOutputWindow } from "../env/index.ts"; import type { + CommitOptions, ConversationId, ConversationOwnership, ConversationRecord, @@ -240,7 +241,8 @@ export interface ToolExecutionApi extend diagnostic(diagnostic: ToolDiagnostic): void; /** Replace running details; the last value becomes the result details when the result omits `details`. */ details(value: TDetails, context: Context): Promise; - commit(change: (tx: Tx) => T | Promise, context: Context): Promise; + /** A commit of this call's task; `options` may defer storing it (`CommitPersistence`). */ + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise; memo(name: string, context: Context): Promise; memo(name: string, candidate: T, context: Context): Promise; /** @@ -521,14 +523,21 @@ export type CompactionSettings = Partial & { }; /** - * How often running progress is committed. Each progress commit is a storage write; a host whose storage is remote can - * commit less often, so live answers and tool output appear in larger steps. + * How often running progress is committed, and how long it may wait in memory before it is stored. Progress commits are + * deferred (`CommitPersistence`): watches see each at once, and Storage gets them together, at the run's next phase + * boundary or after `persistWithinMs`. A host can commit less often, so live answers and tool output appear in larger steps. */ export type ProgressPolicy = { /** Minimum pause between commits of the answer being generated. */ partialIntervalMs: number; /** Minimum pause between commits of running tool output; large commits also pause in proportion to their size. */ outputIntervalMs: number; + /** + * How long these commits, and other deferred commits without their own `withinMs`, may stay in memory before they are + * stored (`CommitPersistence`); a crash loses at most about this much of an answer or of tool output. Default 1,000; + * `Infinity` stores them only at the next commit that reads or writes their document and at close. + */ + persistWithinMs: number; }; /** Why a compaction runs: `compact()`, a threshold in generation preparation, or a context overflow. */ @@ -753,8 +762,8 @@ export interface Conversation { */ compact(instructions: string | undefined, context: Context): Promise>; - /** Session commit whose `tx.createTask()` defaults to this conversation. */ - commit(change: (tx: Tx) => T | Promise, context: Context): Promise; + /** Session commit whose `tx.createTask()` defaults to this conversation; `options` as for `Session.commit()`. */ + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise; /** * Committed raw active transcript and model context. With `at`, the context as of that visible entry: the same view * `fork(at)` would start with, without creating a conversation. diff --git a/packages/durable/src/harness/view.ts b/packages/durable/src/harness/view.ts index d47d2e913..7f6f1a408 100644 --- a/packages/durable/src/harness/view.ts +++ b/packages/durable/src/harness/view.ts @@ -1,9 +1,8 @@ import { type Context, type JsonValue, Watch, type WatchHandle } from "@earendil-works/chord"; import { withoutAbortSignal } from "@earendil-works/chord/context"; import { applyImmutable, type NonEmptyPath, type Op, type Path } from "@earendil-works/chord/delta"; -import { contained, type SessionImpl } from "../session/session.ts"; +import { contained, type ObservedCommit, type SessionImpl } from "../session/session.ts"; import type { - CommitPublication, ConversationDocToken, ConversationId, ConversationRecord, @@ -37,7 +36,7 @@ export type ViewObserver = { before: ConversationView, after: ConversationView, ops: readonly Op[], - publication: CommitPublication, + publication: ObservedCommit, context: Context, ): void; /** The Session closed, or failed with `failure`. */ @@ -72,7 +71,8 @@ export function conversationViews(harness: object): ConversationViews { /** * The Harness's conversation view mounts: at most one per conversation, built on the Session line by its first - * observer and dropped with its last. Each mount advances from the Session's commit publications, which are durable. + * observer and dropped with its last. Each mount advances from the Session's commits as memory adopts them, deferred ones + * (`CommitPersistence`) before they are stored. */ export class ConversationViews { readonly #session: SessionImpl; @@ -172,7 +172,7 @@ export class ConversationViews { function advance( id: ConversationId, mount: Mount, - publication: CommitPublication, + publication: ObservedCommit, context: Context, report: (error: unknown) => void, ): void { diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 60dad5337..b26a50316 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -130,6 +130,8 @@ export { defineTask } from "./tasks.ts"; export type { CheckpointInfo, CommitChange, + CommitOptions, + CommitPersistence, CommitPublication, CommonDocDefinition, ContextEdit, diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index 4a6563b70..d66045b19 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -1,6 +1,6 @@ import { type Context, type JsonValue, Watch } from "@earendil-works/chord"; import { awaitWithContext, BACKGROUND_CONTEXT, withoutAbortSignal } from "@earendil-works/chord/context"; -import { type Op, track } from "@earendil-works/chord/delta"; +import { type Op, PendingOps, track } from "@earendil-works/chord/delta"; import { type AnyDocToken, addressId, @@ -14,6 +14,8 @@ import { idFromNumber } from "../ids.ts"; import { hostTimers, type Timers } from "../timers.ts"; import type { CommitChange, + CommitOptions, + CommitPersistence, CommitPublication, ConversationDocFamilyToken, ConversationDocToken, @@ -56,11 +58,42 @@ import type { TaskRecord, Tx, } from "../types.ts"; -import { type LoadedDocument, Transaction, type TransactionHost, type TransactionScope } from "./transaction.ts"; +import { + type AdoptedDocuments, + type HeldDocument, + type LoadedDocument, + storeHeld, + Transaction, + type TransactionHost, + type TransactionScope, +} from "./transaction.ts"; /** What a document watch observes: the committed value of its incarnation, `null` once it retired. */ type ObservedDocumentValue = Readonly | null; +/** + * What the Session's own components observe of one commit (`observeCommits()`): a stored commit with its own changes, or + * a deferred one adopted in memory only (`CommitPersistence`), which changes documents only. Memory is never behind: + * a later store of held changes is not observed again. + */ +export type ObservedCommit = + | { readonly kind: "stored"; readonly seq: Seq; readonly changes: readonly CommitChange[] } + | { readonly kind: "memory"; readonly changes: readonly DocumentCommitChange[] }; + +/** A held document (`HeldDocument`) as the Session keeps it: pinned in the cache until stored. */ +type Hold = HeldDocument & { + /** When the earliest of its held commits must be stored, on the Session clock. */ + deadline: number; + readonly unpin: () => void; +}; + +/** The armed store of held documents: a foreground timer for the earliest deadline, `fired` once its job is queued. */ +type HoldTimer = { readonly at: number; disarm: () => void; fired: boolean }; + +const NO_HOLDS: readonly Hold[] = []; + +const NO_OPERATIONS: readonly Op[] = []; + /** The default source of a report: a listener, which never fails what ran it. */ const LISTENER: ReportInfo = { type: "listener_error" }; @@ -71,14 +104,23 @@ const LISTENER: ReportInfo = { type: "listener_error" }; */ export function createSession( storage: Storage, - options?: { readonly now?: () => number; readonly documentRetentionMs?: number; readonly setTimer?: SetTimer }, + options?: { + readonly now?: () => number; + readonly documentRetentionMs?: number; + readonly setTimer?: SetTimer; + /** Default `withinMs` of a deferred commit (`CommitPersistence`); default 1,000. */ + readonly persistWithinMs?: number; + }, ): Session { const retention = options?.documentRetentionMs; + const within = options?.persistWithinMs; return new SessionImpl( storage, options?.now, retention === undefined ? undefined : () => retention, options?.setTimer, + false, + () => within, ); } @@ -98,6 +140,17 @@ export function documentRetention(ms: number | undefined): number { return typeof ms === "number" && ms >= 0 ? ms : defaultRetention; } +/** Default `withinMs` of a deferred commit, `settings.progress.persistWithinMs`. */ +const DEFAULT_PERSIST_WITHIN_MS = 1_000; + +/** + * A deferred commit's default window as given, when it is a number of milliseconds, `0` or more (`Infinity` waits for a + * reading or writing commit or close); otherwise the default. + */ +export function persistWithin(ms: number | undefined): number { + return typeof ms === "number" && ms >= 0 ? ms : DEFAULT_PERSIST_WITHIN_MS; +} + /** * Session kernel: one mutation line, the loaded document tracker cache, and committed publication. * @@ -105,8 +158,11 @@ export function documentRetention(ms: number | undefined): number { * written for the retention period, and the next access loads it from Storage. Eviction runs as a job on the mutation * line, so no transaction ever holds a tracker that leaves. * - * Only committed state is observable. Every commit callback, preparation, Storage settlement, adoption, and - * publication enqueue runs while the line is held; listeners run later. + * Only committed state is observable. A deferred commit (`CommitPersistence`) is committed in memory first: it is + * adopted and shown to the Session's own observers, and its changes wait, folded per document and pinned in the cache, for + * the first commit that reads or writes the document, the document's deadline, a drop of its tracker, or close, which + * store them. Every commit callback, preparation, Storage settlement, adoption, and publication enqueue runs while the + * line is held; listeners run later. * * Storage failures are final. Every Storage call goes through one guard: the first error a Storage method throws, other * than for a cancelled caller, fails the Session, as does a commit it cannot adopt. That call gets the error; every later @@ -123,6 +179,10 @@ export class SessionImpl implements Session { readonly #attachedDocuments = new Map(); readonly #clock: () => number; readonly #documentRetentionMs: () => number; + readonly #persistWithinMs: () => number; + /** Documents with held changes, by incarnation (record ID); each is pinned in the cache until they are stored. */ + readonly #holds = new Map(); + #holdTimer: HoldTimer | undefined; /** Refuse every commit with writes (`HarnessOptions.readOnly`). */ readonly #readOnly: boolean; readonly #timers: Timers; @@ -133,7 +193,7 @@ export class SessionImpl implements Session { #documentTimersUnavailable = false; readonly #commitListeners = new Set<(publication: CommitPublication, context: Context) => void>(); /** The Session's own listeners, such as its scheduler's: they keep memory in step with storage, so a throw fails it. */ - readonly #internalListeners = new Set<(publication: CommitPublication, context: Context) => void>(); + readonly #internalListeners = new Set<(publication: ObservedCommit, context: Context) => void>(); readonly #closeListeners = new Set<() => void>(); readonly #host: TransactionHost; #tail: Promise = Promise.resolve(); @@ -163,6 +223,7 @@ export class SessionImpl implements Session { documentRetentionMs: () => number | undefined = () => undefined, setTimer?: SetTimer, readOnly = false, + persistWithinMs: () => number | undefined = () => undefined, ) { // Waits race it; none has to. this.#failed.promise.catch(() => {}); @@ -185,6 +246,13 @@ export class SessionImpl implements Session { return documentRetention(undefined); } }; + this.#persistWithinMs = () => { + try { + return persistWithin(persistWithinMs()); + } catch { + return persistWithin(undefined); + } + }; this.#host = { storage: this.#storage, now, @@ -210,8 +278,8 @@ export class SessionImpl implements Session { return this.#storage; } - commit(change: (tx: Tx) => T | Promise, context: Context): Promise { - return this.commitWith(change, context); + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise { + return this.commitWith(change, context, undefined, options); } /** @@ -234,13 +302,32 @@ export class SessionImpl implements Session { * bootstrap and task replacement. `scope` sets the default `tx.createTask()` conversation and the task attributed to * appended entries. */ - commitWith(change: (tx: Transaction) => T | Promise, context: Context, scope?: TransactionScope): Promise { + commitWith( + change: (tx: Transaction) => T | Promise, + context: Context, + scope?: TransactionScope, + options?: CommitOptions, + ): Promise { + let withinMs: number | undefined; try { this.assertUsable(); + withinMs = this.#deferral(options?.persistence); } catch (error) { return Promise.reject(error); } - return this.#enqueue(() => this.#runCommit(change, context, scope)); + return this.#enqueue(() => this.#runCommit(change, context, scope, withinMs)); + } + + /** How long a commit may stay in memory before it is stored; `undefined` when it is immediate. */ + #deferral(persistence: CommitPersistence | undefined): number | undefined { + if (persistence?.kind !== "deferred") return undefined; + const ms = persistence.withinMs; + if (ms === undefined) return this.#persistWithinMs(); + // Public options: checked here, as the types cannot. + if (typeof ms !== "number" || !(ms >= 0)) { + throw new TypeError("CommitPersistence.withinMs must be a number of milliseconds, 0 or more"); + } + return ms; } /** Internal: run a read-only job on the mutation line so multi-read derivations observe one committed state. */ @@ -437,11 +524,21 @@ export class SessionImpl implements Session { .then(() => this.beforeClose()) .then(() => this.#enqueue(async () => { + // Last, after the writes at close (`writeAtClose()`), while the trackers are still here. A failed Session + // writes nothing; a failing store fails it, and the backend still closes. + let unstored: { readonly error: unknown } | undefined; + // Only with something to store: otherwise the backend's close begins in this very job. + if (this.#failure === undefined && this.#holds.size > 0) { + await this.#storeHolds([...this.#holds.values()], cleanup).catch((error: unknown) => { + unstored = { error }; + }); + } this.#commitListeners.clear(); this.#internalListeners.clear(); this.#documents.clear(); this.#idleDocuments.clear(); this.#attachedDocuments.clear(); + this.#holds.clear(); try { await this.#storage.close(cleanup); } catch (error) { @@ -452,6 +549,7 @@ export class SessionImpl implements Session { } throw error; } + if (unstored !== undefined) throw unstored.error; }), ) .finally(() => { @@ -469,6 +567,8 @@ export class SessionImpl implements Session { #notifyClose(): void { this.#documentSweep?.disarm?.(); this.#documentSweep = undefined; + this.#holdTimer?.disarm(); + this.#holdTimer = undefined; const listeners = [...this.#closeListeners]; this.#closeListeners.clear(); for (const listener of listeners) contained(listener, (error) => this.report(error)); @@ -506,7 +606,10 @@ export class SessionImpl implements Session { return Promise.resolve(); } - /** Register a synchronous post-adoption listener. It must not block or call Session operations; a throw is reported. */ + /** + * Register a synchronous listener for every stored commit, with exactly what it stored (`CommitPublication`). It must + * not block or call Session operations; a throw is reported. + */ subscribeCommits(listener: (publication: CommitPublication, context: Context) => void): () => void { this.assertUsable(); this.#commitListeners.add(listener); @@ -514,10 +617,11 @@ export class SessionImpl implements Session { } /** - * Internal: `subscribeCommits()` for the Session's own components, whose state follows each commit. A throw leaves - * that state behind storage, so it fails the Session instead of being reported. They run before host listeners. + * Internal: observe every commit as memory adopts it (`ObservedCommit`), for the Session's own components, whose state + * follows memory. A throw leaves that state behind, so it fails the Session instead of being reported. They run + * before host listeners. */ - observeCommits(listener: (publication: CommitPublication, context: Context) => void): () => void { + observeCommits(listener: (publication: ObservedCommit, context: Context) => void): () => void { this.assertUsable(); this.#internalListeners.add(listener); return () => this.#internalListeners.delete(listener); @@ -533,29 +637,32 @@ export class SessionImpl implements Session { return () => this.#closeListeners.delete(listener); } - /** Drop every loaded tracker on the mutation line; later access cold-loads from Storage. */ + /** Drop every loaded tracker on the mutation line, held changes stored first; later access cold-loads from Storage. */ unloadDocuments(): Promise { return this.#enqueue(async () => { + if (this.#failure === undefined) await this.#storeHolds([...this.#holds.values()], BACKGROUND_CONTEXT); this.#documents.clear(); this.#idleDocuments.clear(); }); } /** - * Internal, for tests: how many documents the cache holds, how many are idle, how many incarnations are pinned, and how - * many sweep jobs were queued on the line so far. + * Internal, for tests: how many documents the cache holds, how many are idle, how many incarnations are pinned, how + * many sweep jobs were queued on the line so far, and how many documents have held changes. */ get documentCacheSizes(): { readonly cached: number; readonly idle: number; readonly attached: number; readonly sweeps: number; + readonly held: number; } { return { cached: this.#documents.size, idle: this.#idleDocuments.size, attached: this.#attachedDocuments.size, sweeps: this.#sweepJobs, + held: this.#holds.size, }; } @@ -653,7 +760,8 @@ export class SessionImpl implements Session { async #runCommit( change: (tx: Transaction) => T | Promise, context: Context, - scope?: TransactionScope, + scope: TransactionScope | undefined, + withinMs: number | undefined, ): Promise { this.#assertHealthy(); context.abortSignal?.throwIfAborted(); @@ -687,9 +795,19 @@ export class SessionImpl implements Session { this.#sweepDocuments(); throw new ReadOnlySession("a commit with writes"); } + if (withinMs !== undefined) { + const held = tx.holdable(); + if (held !== undefined) { + this.#hold(tx, held, withinMs, context); + this.#sweepDocuments(); + return result; + } + } + // The last step before Storage: held changes leave their queue only once nothing but Storage can fail. + const taken = this.#holds.size === 0 ? NO_HOLDS : tx.takeHeld((id) => this.#holds.get(id)); // 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; + let adoption: { readonly documents: AdoptedDocuments } | { readonly error: unknown } | undefined; const adopt = (committed: Seq): void => { try { adoption = { documents: tx.adopt(committed) }; @@ -717,18 +835,126 @@ export class SessionImpl implements Session { throw adoption.error; } const documents = adoption.documents; - for (const document of documents) this.#touchDocument(addressId(document.record)); + if (taken.length > 0) this.#release(taken); + for (const document of documents.stored) this.#touchDocument(addressId(document.record)); this.#publish(seq, writes, documents, context); this.#sweepDocuments(); return result; } - #publish( - seq: Seq, - writes: readonly StorageWrite[], - documents: readonly DocumentCommitChange[], + /** + * Hold a deferred commit (`Transaction.holdable()`): adopt it in memory, queue its changes per document, pinned, by + * the earliest deadline of the document's held commits, and show it to the Session's own observers. + */ + #hold( + tx: Transaction, + held: NonNullable>, + withinMs: number, context: Context, ): void { + const documents = tx.adoptHeld(); + const deadline = this.#clock() + withinMs; + for (const { loaded, ops, base } of held) { + const id = loaded.record.id; + let hold = this.#holds.get(id); + if (hold === undefined) { + const unpin = this.#pinDocument(loaded.addressId, id); + hold = { loaded, pending: new PendingOps(), base: false, deadline, unpin }; + this.#holds.set(id, hold); + } + hold.pending.push(ops); + if (base) hold.base = true; + if (deadline < hold.deadline) hold.deadline = deadline; + } + this.#armHoldTimer(); + const publication: ObservedCommit = { kind: "memory", changes: documents }; + for (const listener of [...this.#internalListeners]) { + contained( + () => listener(publication, context), + (error) => this.fail(error, context), + ); + } + } + + /** + * Store the held changes of `holds` in one storage transaction, on the line. Memory already has them, so only + * `subscribeCommits()` listeners see it. A failure fails the Session, as every Storage failure does. + */ + async #storeHolds(holds: readonly Hold[], context: Context): Promise { + if (holds.length === 0) return; + const writes: Extract[] = []; + const changes: DocumentCommitChange[] = []; + for (const hold of holds) { + const { record, tracker, valueVersion: version } = hold.loaded; + const value = tracker.value; + const stored = storeHeld(hold, NO_OPERATIONS, value, version); + writes.push({ type: "document.change", id: record.id, content: stored.content }); + // Held documents are conversation documents. + const conversationId = record.scope.kind === "conversation" ? record.scope.conversationId : undefined; + changes.push({ type: "document", record, conversationId, version, value, ops: stored.ops }); + } + let adopted = false; + const adopt = (): void => { + adopted = true; + for (const [index, hold] of holds.entries()) { + if (writes[index]!.content.kind === "base") hold.loaded.deltasSinceBase = 0; + else hold.loaded.deltasSinceBase++; + } + }; + const seq = await this.#storage.commit(writes, adopt, withoutAbortSignal(context)); + if (!adopted) { + const error = new Error("Storage committed without adopting the batch"); + this.fail(error); + throw error; + } + this.#release(holds); + const publication: CommitPublication = { seq, changes }; + for (const listener of [...this.#commitListeners]) { + contained( + () => listener(publication, context), + (error) => this.report(error, LISTENER, context), + ); + } + } + + /** After their store: forget `holds`, unpin their documents, and move the timer to the next deadline. */ + #release(holds: readonly Hold[]): void { + for (const hold of holds) { + this.#holds.delete(hold.loaded.record.id); + hold.unpin(); + } + this.#armHoldTimer(); + } + + /** + * Keep a foreground timer (Workers refuse background ones) for the earliest deadline of the held documents. When it + * fires, every held document is stored, in one storage transaction: a deadline is the latest time, and the others + * cost little more in the same transaction. Once fired, it stays until its job ran. + */ + #armHoldTimer(): void { + if (this.#closing !== undefined) return; + let at = Infinity; + for (const hold of this.#holds.values()) if (hold.deadline < at) at = hold.deadline; + const armed = this.#holdTimer; + if (armed !== undefined && (armed.fired || armed.at === at)) return; + armed?.disarm(); + this.#holdTimer = undefined; + if (at === Infinity) return; + const timer: HoldTimer = { at, disarm: () => {}, fired: false }; + const run = (): void => { + timer.fired = true; + this.#enqueue(async () => { + if (this.#holdTimer === timer) this.#holdTimer = undefined; + // Close stores what is held itself. + if (this.#closing !== undefined || this.#failure !== undefined) return; + await this.#storeHolds([...this.#holds.values()], BACKGROUND_CONTEXT); + }).catch(() => {}); + }; + this.#holdTimer = timer; + timer.disarm = this.#timers.foreground(run, Math.max(0, at - this.#clock())); + } + + #publish(seq: Seq, writes: readonly StorageWrite[], documents: AdoptedDocuments, context: Context): void { const changes: CommitChange[] | undefined = this.#commitListeners.size > 0 || this.#internalListeners.size > 0 ? [] : undefined; for (const write of writes) { @@ -756,18 +982,24 @@ export class SessionImpl implements Session { } } if (changes === undefined) return; - for (const document of documents) changes.push(document); - const publication: CommitPublication = { seq, changes }; + const tables = changes.length; + for (const document of documents.observed) changes.push(document); + const observed: ObservedCommit = { kind: "stored", seq, changes }; + // One publication for both, unless the commit also stored held changes, which only durable listeners get. + const stored: CommitPublication = + documents.stored === documents.observed + ? observed + : { seq, changes: [...changes.slice(0, tables), ...documents.stored] }; // The commit is durable: a failing listener neither fails the commit nor skips the others. for (const listener of [...this.#internalListeners]) { contained( - () => listener(publication, context), + () => listener(observed, context), (error) => this.fail(error, context), ); } for (const listener of [...this.#commitListeners]) { contained( - () => listener(publication, context), + () => listener(stored, context), (error) => this.report(error, LISTENER, context), ); } @@ -822,7 +1054,12 @@ export class SessionImpl implements Session { this.#touchDocument(addressId); return cached; } - if (cached !== undefined) this.#dropDocument(addressId); + if (cached !== undefined) { + // The reload must find what memory holds: held changes are stored before their tracker goes. + const hold = this.#holds.get(cached.record.id); + if (hold !== undefined) await this.#storeHolds([hold], context); + this.#dropDocument(addressId); + } const record = await this.#storage.findDocument(address, "current", context); if (record === undefined) return undefined; const stored = await this.#storage.document(record.id, "current", context); diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts index e7b783408..7a62c440b 100644 --- a/packages/durable/src/session/transaction.ts +++ b/packages/durable/src/session/transaction.ts @@ -1,6 +1,6 @@ import { type Context, copyJson, type Draft, type JsonValue } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import { type Change, type Op, type Prepared, type Tracker, track } from "@earendil-works/chord/delta"; +import { type Change, type Op, type PendingOps, type Prepared, type Tracker, track } from "@earendil-works/chord/delta"; import type { TelemetryCarrier } from "@earendil-works/pi-telemetry"; import { type AnyDocDefinition, @@ -24,6 +24,7 @@ import type { Cursor, DocumentAddress, DocumentCommitChange, + DocumentContent, DocumentCopySource, DocumentCreate, DocumentId, @@ -97,6 +98,25 @@ export type LoadedDocument = { readonly tracker: Tracker; }; +/** + * The changes of a cached incarnation that deferred commits adopted in memory and the Session has not stored yet + * (`CommitPersistence`), folded. The Session owns it; a commit that stores them takes them. + */ +export type HeldDocument = { + readonly loaded: LoadedDocument; + readonly pending: PendingOps; + /** Set when a held commit selected a base (`checkpointWhen`): their store writes the whole value. */ + base: boolean; +}; + +/** What adoption publishes: to the Session's own observers, and the stored changes to `subscribeCommits()`. */ +export type AdoptedDocuments = { + /** This commit's own changes, as memory adopted them. */ + readonly observed: readonly DocumentCommitChange[]; + /** What Storage got: `observed`, unless the commit also stored held changes, which precede its own. */ + readonly stored: readonly DocumentCommitChange[]; +}; + /** Session services used by a transaction while it holds the mutation line. */ export interface TransactionHost { readonly storage: Storage; @@ -188,6 +208,11 @@ type DocumentPlan = { }; /** Resolved before Storage admission so adoption performs no reads. */ conversationId?: ConversationId; + /** + * Set when this commit also stores held changes of the incarnation (`takeHeld()`): the operations its stored change + * publishes, held first, and whether it has a change of its own to show the Session's observers. + */ + held?: { readonly ops: readonly Op[]; readonly observed: boolean }; }; /** @@ -724,9 +749,102 @@ export class Transaction implements Tx { this.#abortChanges(); } - /** Adopt every prepared change by pointer swap after Storage success and describe the publication. */ - adopt(seq: Seq): DocumentCommitChange[] { + /** + * Internal: what a deferred commit holds in memory instead of storing (`CommitPersistence`), each changed incarnation + * with its operations and whether they selected a base; `undefined` when the commit must be stored. It may be held + * only when every write changes an existing `latest`, `initial`-fork conversation document at its stored version. + */ + holdable(): { readonly loaded: LoadedDocument; readonly ops: readonly Op[]; readonly base: boolean }[] | undefined { + if (this.#writes.some((write) => write.type !== "document.change")) return undefined; + const held = []; + for (const plan of this.#plans) { + const { content, change } = plan; + if (content === undefined && !plan.retire) continue; + if (plan.retire || change?.loaded === undefined || content?.type !== "document.change") return undefined; + const { loaded, prepared, version } = change; + const record = loaded.record; + if (record.scope.kind !== "conversation" || record.history !== "latest" || record.fork !== "initial") { + return undefined; + } + // A migration stores a base at once. + if (loaded.storedVersion !== version) return undefined; + held.push({ loaded, ops: prepared.ops, base: content.content.kind === "base" }); + } + return held; + } + + /** + * Internal: adopt a held commit (`holdable()`) in memory only. Trackers advance by pointer swap; stored versions and + * delta counts stay, as nothing was stored. Returns its publication for the Session's own observers. + */ + adoptHeld(): DocumentCommitChange[] { const publications: DocumentCommitChange[] = []; + for (const plan of this.#plans) { + // Every plan of a held commit changes or reads a loaded incarnation (`holdable()`). + const { tracker, prepared, version, loaded } = plan.change!; + if (prepared.ops.length === 0) { + prepared.abort(); + continue; + } + tracker.adopt(prepared); + const record = loaded!.record; + const conversationId = plan.conversationId; + publications.push({ + type: "document", + record, + conversationId, + version, + value: prepared.value, + ops: prepared.ops, + }); + } + return publications; + } + + /** + * Internal: store the held changes of every incarnation this commit reads or writes in front of its own, so a stored + * document never misses an operation its later ones build on. Called right before Storage admission, after every + * fallible step, because it empties the holds: a failure earlier keeps them. Returns the holds it took. + */ + takeHeld(held: (id: DocumentId) => H | undefined): H[] { + const taken: H[] = []; + for (const plan of this.#plans) { + // A new incarnation holds nothing yet. + if (!("createdAt" in plan.record)) continue; + const hold = held(plan.record.id); + if (hold === undefined) continue; + taken.push(hold); + const own = plan.content?.type === "document.change" ? plan.content : undefined; + const change = plan.change; + const value = change?.prepared.value ?? hold.loaded.tracker.value; + const stored = storeHeld(hold, change?.prepared.ops ?? EMPTY_OPERATIONS, value, hold.loaded.valueVersion); + // A base this commit selected or must write covers what was held. + const content = own?.content.kind === "base" ? own.content : stored.content; + const write = { type: "document.change", id: plan.record.id, content } as const; + if (own !== undefined) this.#writes[this.#writes.indexOf(own)] = write; + else this.#insertBeforeRetirement(write); + if (!plan.retire) plan.held = { ops: stored.ops, observed: own !== undefined }; + plan.content = write; + // Held documents are conversation documents; one this commit only read has no owner resolved yet. + if (plan.record.scope.kind === "conversation") plan.conversationId = plan.record.scope.conversationId; + } + return taken; + } + + /** Stage a content write for an incarnation, before its retirement when it retires. */ + #insertBeforeRetirement(write: Extract): void { + const at = this.#writes.findIndex( + (candidate) => candidate.type === "document.retire" && candidate.id === write.id, + ); + if (at < 0) this.#writes.push(write); + else this.#writes.splice(at, 0, write); + } + + /** Adopt every prepared change by pointer swap after Storage success and describe the publication. */ + adopt(seq: Seq): AdoptedDocuments { + const publications: DocumentCommitChange[] = []; + // Only when the commit stored held changes: then what Storage got differs from what memory adopted. + let stored: DocumentCommitChange[] | undefined; for (const plan of this.#plans) { const committed = "createdAt" in plan.record; let record: DocumentRecord = committed ? (plan.record as DocumentRecord) : { ...plan.record, createdAt: seq }; @@ -754,31 +872,21 @@ export class Transaction implements Tx { }); } } - const conversationId = plan.conversationId; - if (plan.retire) { - if (committed) this.#host.evict(plan.addressId, record.id); - const retired = { version: undefined, value: null, ops: EMPTY_OPERATIONS }; - publications.push({ type: "document", record, conversationId, ...retired }); - } else if (plan.content?.type === "document.copy") { - publications.push({ - type: "document.copy", - record, - conversationId: conversationId!, - source: plan.content.source, - }); - } else if (change !== undefined && publishes(plan)) { - const ops = change.loaded === undefined ? EMPTY_OPERATIONS : change.prepared.ops; - publications.push({ - type: "document", - record, - conversationId, - version: change.version, - value: change.prepared.value, - ops, - }); + if (plan.retire && committed) this.#host.evict(plan.addressId, record.id); + const publication = describe(plan, record); + if (publication === undefined) continue; + const held = plan.held; + if (held === undefined) { + publications.push(publication); + stored?.push(publication); + continue; } + stored ??= [...publications]; + if (held.observed) publications.push(publication); + // Only a document change of a loaded incarnation carries held operations. + stored.push(publication.type === "document" ? { ...publication, ops: held.ops } : publication); } - return publications; + return { observed: publications, stored: stored ?? publications }; } async #assemble(): Promise { @@ -1062,6 +1170,38 @@ function planDocument(document: DocumentEntry): DocumentPlan | undefined { } } +/** The publication of one adopted plan, if it publishes (`publishes()`), with `record` as committed. */ +function describe(plan: DocumentPlan, record: DocumentRecord): DocumentCommitChange | undefined { + const conversationId = plan.conversationId; + if (plan.retire) + return { type: "document", record, conversationId, version: undefined, value: null, ops: EMPTY_OPERATIONS }; + if (plan.content?.type === "document.copy") { + return { type: "document.copy", record, conversationId: conversationId!, source: plan.content.source }; + } + const change = plan.change; + if (change === undefined || !publishes(plan)) return undefined; + const ops = change.loaded === undefined ? EMPTY_OPERATIONS : change.prepared.ops; + return { type: "document", record, conversationId, version: change.version, value: change.prepared.value, ops }; +} + +/** + * Take a hold's changes after `ops`, a commit's own on top of them, as one stored content for an incarnation now at + * `value`, and the operations a reader of the stored state applies to reach it: the folded ones, or a root replacement + * after the queue overflowed, which stores a base, as does a hold whose commits selected one. + */ +export function storeHeld( + hold: HeldDocument, + ops: readonly Op[], + value: JsonObject, + version: number, +): { readonly content: DocumentContent; readonly ops: readonly Op[] } { + hold.pending.push(ops); + const taken = hold.pending.take(); + if (taken.kind === "replace") return { content: { version, kind: "base", value }, ops: [["r", value]] }; + if (hold.base) return { content: { version, kind: "base", value }, ops: taken.ops }; + return { content: { version, kind: "delta", ops: taken.ops }, ops: taken.ops }; +} + /** * Whether adoption publishes the plan: every creation, copy, and retirement, and a loaded incarnation that writes * content, which includes a migration-only base so observers of the older shape receive the new value. diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 24c44f9ec..5844fccb1 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -268,6 +268,8 @@ export interface TaskRuntime extends DocumentObserver * Harness is closing, or, in a run invocation, the task carries an abort mark. A returned state replaces the task's * state in the same commit; returning nothing leaves it unchanged. `tx.createTask()` defaults to the task's * conversation. Inside the callback read only through `tx`: `snapshot()`, `watchDoc()`, `getTask()`, and the other Session and runtime reads may queue on the line the callback holds, and then never resolve. + * `options.persistence` may defer storing a commit that changes only documents (`CommitPersistence`); one that returns + * a state is immediate. */ commit( change: ( @@ -275,6 +277,7 @@ export interface TaskRuntime extends DocumentObserver current: RunningTask, ) => NextTaskState | undefined | Promise | undefined>, context: Context, + options?: CommitOptions, ): Promise; /** Read a durable memo of this task. */ memo(name: string, context: Context): Promise; @@ -897,7 +900,12 @@ export type TableCommitChange = Extract< export type CommitChange = TableCommitChange | DocumentCommitChange; -/** Every immutable change from one successful Session commit. Change order is unspecified. */ +/** + * Every immutable change one Session storage transaction stored, with its `seq`. Change order is unspecified. A commit + * held in memory (`CommitPersistence`) is published once it is stored: by the next commit that reads or writes its + * document, alone (`seq` of its own), or with that commit, whose document change then carries the held operations + * followed by its own. + */ export type CommitPublication = { readonly seq: Seq; readonly changes: readonly CommitChange[]; @@ -1048,12 +1056,30 @@ export interface DocumentObserver { /** Why a Session ended: `close()`, or a failed Storage call, whose error it carries. */ export type SessionEnd = { readonly reason: "closed" } | { readonly reason: "failed"; readonly error: unknown }; +/** + * When a commit reaches Storage. `immediate`, the default: before `commit()` resolves. `deferred`: when every write of + * the commit changes an existing conversation document with `history: "latest"` and `fork: "initial"` (no creation, + * migration, or retirement), the Session adopts it in memory, shows it to watches and views, and resolves, but stores it + * later: with the next commit that reads (`tx.doc()`) or writes that document, before the document leaves memory or + * retires, at `close()`, and at the latest about `withinMs` after the commit (default `settings.progress.persistWithinMs`, + * 1,000 ms; `Infinity` waits for one of the others). A commit that writes anything else is immediate. A crash loses what + * is held, so only state whose loss is acceptable, such as streamed progress, should be deferred, and nothing durable may + * depend on it without reading it in a commit. + */ +export type CommitPersistence = + | { readonly kind: "immediate" } + | { readonly kind: "deferred"; readonly withinMs?: number }; + +/** Options of one Session commit. */ +export type CommitOptions = { readonly persistence?: CommitPersistence }; + /** Owner of one mutation line, its records, and its tracked documents. */ export interface Session extends DocumentObserver { /** * Run one atomic transaction on the Session mutation line. Inside the callback read only through `tx`: `snapshot()`, `watchDoc()`, `getTask()`, and the other Session and runtime reads may queue on the line the callback holds, and then never resolve. + * `options.persistence` may defer storing it (`CommitPersistence`). */ - commit(change: (tx: Tx) => T | Promise, context: Context): Promise; + commit(change: (tx: Tx) => T | Promise, context: Context, options?: CommitOptions): Promise; /** Seal admission, settle admitted commits, then close storage. */ close(context: Context): Promise; /** @@ -1061,7 +1087,10 @@ export interface Session extends DocumentObserver { * which closes the Session itself. A failed Session is reopened from Storage; nothing else is left to clean up. */ readonly closed: Promise; - /** Observe complete commits synchronously after adoption. The listener must not block or call Session APIs. */ + /** + * Observe stored commits synchronously after adoption, each with exactly what it stored; a commit held in memory is + * published when it is stored (`CommitPublication`). The listener must not block or call Session APIs. + */ subscribeCommits(listener: (publication: CommitPublication, context: Context) => void): () => void; /** * Observe close synchronously when it begins, also when a failed Storage call closes the Session. The listener must diff --git a/packages/durable/test/harness-deferred-progress.test.ts b/packages/durable/test/harness-deferred-progress.test.ts new file mode 100644 index 000000000..8f6221af7 --- /dev/null +++ b/packages/durable/test/harness-deferred-progress.test.ts @@ -0,0 +1,444 @@ +// Deferred progress commits (`PROGRESS_COMMIT`): the partial answer and tool output reach watches at once and Storage at +// the next commit that reads or writes `pi.live` or within `settings.progress.persistWithinMs`. Crashes are simulated +// by failing the Harness: from that moment nothing more reaches Storage, as when the process dies; the reopened Harness +// resumes from what was stored. +import { applyImmutable } from "@earendil-works/chord/delta"; +import { fauxAssistantMessage, fauxText, fauxToolCall, Type } from "@earendil-works/pi-ai"; +import { + AssistantEntry, + type ConversationId, + defineTask, + defineTool, + type EntryRecord, + type Harness, + LiveDoc, + type LiveState, + type TaskId, + ToolResultEntry, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import type { PartialMessage } from "../src/harness/partial-changes.ts"; +import type { SessionImpl } from "../src/session/session.ts"; +import { allEntries, type ChatSetup, chatSetup, openChat, textOf, waitFor } from "./chat-support.ts"; +import { addTask, addTool } from "./harness-support.ts"; +import { ControlledStorage, context, observeCommits } from "./session-support.ts"; +import { aborted, deferred } from "./task-support.ts"; + +/** A slow stream: 4 characters 80 times a second. */ +function slowSetup(persistWithinMs: number): ChatSetup { + const setup = chatSetup({ tokensPerSecond: 80, tokenSize: { min: 4, max: 4 } }); + setup.settings.progress = { persistWithinMs }; + return setup; +} + +const LONG = "streamed ".repeat(400); + +/** Open (or reopen) a Harness over `storage` and let it run tasks. */ +async function open(storage: ControlledStorage, setup: ChatSetup) { + const opened = await openChat(storage.reopen(), setup); + opened.harness.resume(); + return { ...opened, memory: memoryPartials(opened.harness) }; +} + +/** The partial text of every `pi.live` revision memory adopted, the last one last. */ +function memoryPartials(harness: Harness): string[] { + const texts: string[] = []; + observeCommits(harness, (publication) => { + for (const change of publication.changes) { + if (change.type !== "document" || change.record.kind !== LiveDoc.definition.kind) continue; + // Publications carry untyped document values; this one is `pi.live`. + const message = (change.value as LiveState | null)?.generation?.message; + if (message !== undefined) texts.push(partialText(message) ?? ""); + } + }); + return texts; +} + +/** Fail the Harness now: nothing more is stored, held changes are lost with it. */ +async function crash(harness: Harness): Promise { + // `fail()` is internal, so the public type lacks it; every Harness is a `SessionImpl`. + (harness as unknown as SessionImpl).fail(new Error("crash")); + await harness.closed; +} + +/** `pi.live` as Storage has it, read past the Session, also after a crash closed it. */ +async function storedLive(storage: ControlledStorage, conversationId: ConversationId): Promise { + storage.reopen(); + const address = { kind: LiveDoc.definition.kind, scope: { kind: "conversation", conversationId } } as const; + const record = await storage.findDocument(address, "current", context); + if (record === undefined) return undefined; + // Storage values are untyped; this one is `pi.live`. + return (await storage.document(record.id, "current", context))?.value as LiveState | undefined; +} + +/** The text of a partial message's first text block. */ +function partialText(message: PartialMessage | undefined): string | undefined { + const block = message?.content.find((candidate) => candidate.type === "text"); + return block?.type === "text" ? block.text : undefined; +} + +function storedText(live: LiveState | undefined): string | undefined { + const message = live?.generation?.message; + return message === undefined ? undefined : (partialText(message) ?? ""); +} + +/** Assistant entries: their text and stop reason. */ +function answers(entries: readonly EntryRecord[]): { text: string | undefined; stopReason: string | undefined }[] { + return entries + .filter((entry) => AssistantEntry.is(entry)) + .map((entry) => { + const message = entry.model![0]!; + return { text: textOf(message), stopReason: message.role === "assistant" ? message.stopReason : undefined }; + }); +} + +async function generationId(harness: Harness, conversationId: ConversationId): Promise { + return (await harness.snapshot(LiveDoc, conversationId, context))!.run!.taskId; +} + +describe("generation partials", () => { + it("hold partials in memory and store them together, at most every persistWithinMs", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(400); + setup.faux.setResponses([fauxAssistantMessage(LONG.slice(0, 640))]); + const { harness, root, memory } = await open(storage, setup); + const from = storage.commits.length; + await (await root.submit({ type: "input", content: "hi" }, context)).wait(context); + const writesLive = storage.commits + .slice(from) + .filter((batch) => batch.some((write) => write.type === "document.change")); + // About 2 s of streaming at 100 ms per partial: some 20 partials in memory, a store about every 400 ms. + expect(memory.length).toBeGreaterThan(12); + expect(writesLive.length).toBeLessThan(memory.length / 1.5); + await harness.close(context); + }); + + it("converts what was stored into the aborted entry after a crash, and resends", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(150); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + let opened = await open(storage, setup); + // A replica fed by subscribeCommits() alone, as a replication feed would be. + let replica = await storedLive(storage, opened.root.id); + opened.harness.subscribeCommits((publication) => { + for (const change of publication.changes) { + if (change.type === "document" && change.record.kind === LiveDoc.definition.kind) { + replica = applyImmutable(replica, change.ops); + } + } + }); + const submission = await opened.root.submit({ type: "input", content: "hi" }, context); + await waitFor(async () => (storedText(await storedLive(storage, opened.root.id))?.length ?? 0) > 40); + const inMemory = opened.memory.at(-1)!; + await crash(opened.harness); + const stored = storedText(await storedLive(storage, opened.root.id))!; + // What Storage has is a state memory had, behind it by at most the window, and what the replica has. + expect(opened.memory).toContain(stored); + expect(inMemory.startsWith(stored)).toBe(true); + expect(replica).toEqual(await storedLive(storage, opened.root.id)); + + setup.faux.setResponses([fauxAssistantMessage("answer")]); + opened = await open(storage, setup); + expect((await (await opened.harness.submission(submission.id, context))!.wait(context)).status).toBe("done"); + expect(answers(await allEntries(opened.root))).toEqual([ + { text: stored, stopReason: "aborted" }, + { text: "answer", stopReason: "stop" }, + ]); + expect(await opened.harness.snapshot(LiveDoc, opened.root.id, context)).toEqual({}); + await opened.harness.close(context); + }); + + it("leaves no aborted entry for a stream that crashed before its first store", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(Number.POSITIVE_INFINITY); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + let opened = await open(storage, setup); + const submission = await opened.root.submit({ type: "input", content: "hi" }, context); + await waitFor(() => (opened.memory.at(-1)?.length ?? 0) > 40); + await crash(opened.harness); + expect(await storedLive(storage, opened.root.id)).toMatchObject({ generation: { attempt: 1 } }); + expect(storedText(await storedLive(storage, opened.root.id))).toBeUndefined(); + + setup.faux.setResponses([fauxAssistantMessage("answer")]); + opened = await open(storage, setup); + expect((await (await opened.harness.submission(submission.id, context))!.wait(context)).status).toBe("done"); + expect(answers(await allEntries(opened.root))).toEqual([{ text: "answer", stopReason: "stop" }]); + await opened.harness.close(context); + }); + + it("converts a stored partial once through two crashes, before and after the conversion", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(150); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + let opened = await open(storage, setup); + const submission = await opened.root.submit({ type: "input", content: "hi" }, context); + await waitFor(async () => (storedText(await storedLive(storage, opened.root.id))?.length ?? 0) > 40); + await crash(opened.harness); + const stored = storedText(await storedLive(storage, opened.root.id))!; + + // Crash again before anything ran: nothing was converted, so nothing is lost or doubled. + opened = { ...(await openChat(storage.reopen(), setup)), memory: [] }; + await crash(opened.harness); + expect(storedText(await storedLive(storage, opened.root.id))).toBe(stored); + + // Crash again while the resend streams: its conversion is stored, its own partials only held. + setup.settings.progress = { persistWithinMs: Number.POSITIVE_INFINITY }; + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + opened = await open(storage, setup); + await waitFor(() => (opened.memory.at(-1)?.length ?? 0) > 40); + await crash(opened.harness); + expect(storedText(await storedLive(storage, opened.root.id))).toBeUndefined(); + + setup.faux.setResponses([fauxAssistantMessage("answer")]); + opened = await open(storage, setup); + expect((await (await opened.harness.submission(submission.id, context))!.wait(context)).status).toBe("done"); + expect(answers(await allEntries(opened.root))).toEqual([ + { text: stored, stopReason: "aborted" }, + { text: "answer", stopReason: "stop" }, + ]); + await opened.harness.close(context); + }); + + it("stores everything held at a clean close", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(Number.POSITIVE_INFINITY); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + let opened = await open(storage, setup); + const submission = await opened.root.submit({ type: "input", content: "hi" }, context); + await waitFor(() => (opened.memory.at(-1)?.length ?? 0) > 40); + await opened.harness.close(context); + const atClose = opened.memory.at(-1)!; + expect(storedText(await storedLive(storage, opened.root.id))).toBe(atClose); + + setup.faux.setResponses([fauxAssistantMessage("answer")]); + opened = await open(storage, setup); + await (await opened.harness.submission(submission.id, context))!.wait(context); + expect(answers(await allEntries(opened.root))[0]).toEqual({ text: atClose, stopReason: "aborted" }); + await opened.harness.close(context); + }); + + it("never counts partial commits as progress of a task cut off again and again", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(50); + let opened = await open(storage, setup); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + await opened.root.submit({ type: "input", content: "hi" }, context); + const id = await generationId(opened.harness, opened.root.id); + for (let crashes = 1; crashes <= 2; crashes++) { + await waitFor(async () => (storedText(await storedLive(storage, opened.root.id))?.length ?? 0) > 40); + await crash(opened.harness); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + opened = await open(storage, setup); + expect((await opened.harness.getTask(id, context))?.restarts).toBe(crashes); + } + await opened.harness.close(context); + }); + + it("resumes polling a deferred response after a crash", async () => { + const storage = new ControlledStorage(); + const setup = chatSetup({ deferred: { pollAfterMs: 60_000 } }); + let now = 1_000; + setup.now = () => now; + setup.faux.setResponses([fauxAssistantMessage("deferred answer")]); + let opened = await open(storage, setup); + setup.settings.stream = { deferred: true }; + const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id; + await waitFor(async () => (await storedLive(storage, opened.root.id))?.generation?.deferred !== undefined); + await crash(opened.harness); + opened = await open(storage, setup); + now = 61_000; + opened.harness.resume(); + const settled = await (await opened.harness.submission(id, context))!.wait(context); + if (settled.status !== "done" || settled.type !== "input") throw new Error(`Unexpected ${settled.status}`); + const answer = await opened.root.commit((tx) => tx.entry(AssistantEntry, settled.answer), context); + expect(textOf(answer?.model?.[0])).toBe("deferred answer"); + expect(setup.faux.state.deferredFetchCount).toBe(1); + await opened.harness.close(context); + }); +}); + +describe("tool output", () => { + /** + * A tool that writes a line every 20 ms until its invocation is cancelled on its first run. A later run waits for + * cancellation while `stall` is set (`stalled` resolves then), and otherwise writes one line and returns. + */ + function streamingTool(replay: "safe" | "unsafe") { + const lines: string[] = []; + const state = { runs: 0, lines, stall: false, stalled: deferred() }; + const registration = defineTool({ + name: "work", + description: "work", + parameters: Type.Object({}), + replay, + execute: async (_args, api, callContext) => { + const run = ++state.runs; + const signal = callContext.abortSignal!; + if (run > 1 && state.stall) { + state.stalled.resolve(); + return aborted(signal); + } + if (run > 1) { + api.output(`rerun ${run}\n`); + return {}; + } + for (let line = 0; !signal.aborted; line++) { + api.output(`line ${line}\n`); + state.lines.push(`line ${line}\n`); + await new Promise((resolve) => setTimeout(resolve, 20)); + } + return aborted(signal); + }, + }); + return { registration, state }; + } + + async function storedOutput(storage: ControlledStorage, conversationId: ConversationId) { + return (await storedLive(storage, conversationId))?.tools?.[0]?.output; + } + + function results(entries: readonly EntryRecord[]): string[] { + return entries + .filter((entry) => ToolResultEntry.is(entry)) + .map((entry) => { + const message = entry.model![0]!; + if (message.role !== "toolResult") return ""; + return message.content.map((item) => (item.type === "text" ? item.text : "")).join("|"); + }); + } + + it("answers a tool without replay that crashed with the output stored by then", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(150); + const { registration, state } = streamingTool("unsafe"); + addTool(setup.registry, registration); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("work", {}, { id: "c1" })], { stopReason: "toolUse" }), + ]); + let opened = await open(storage, setup); + const submission = await opened.root.submit({ type: "input", content: "go" }, context); + await waitFor(async () => ((await storedOutput(storage, opened.root.id))?.length ?? 0) > 20); + await crash(opened.harness); + const stored = (await storedOutput(storage, opened.root.id))!; + expect(state.lines.join("").startsWith(stored)).toBe(true); + + setup.faux.setResponses([fauxAssistantMessage([fauxText("done")])]); + opened = await open(storage, setup); + expect((await (await opened.harness.submission(submission.id, context))!.wait(context)).status).toBe("done"); + expect(state.runs).toBe(1); + expect(results(await allEntries(opened.root))).toEqual([ + `${stored}|\n[error] Tool work was interrupted and may have partially run\n`, + ]); + await opened.harness.close(context); + }); + + it("reruns a replay-safe tool after a crash with its stored progress cleared, also through a second crash", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(150); + const { registration, state } = streamingTool("safe"); + addTool(setup.registry, registration); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("work", {}, { id: "c1" })], { stopReason: "toolUse" }), + ]); + let opened = await open(storage, setup); + const submission = await opened.root.submit({ type: "input", content: "go" }, context); + await waitFor(async () => ((await storedOutput(storage, opened.root.id))?.length ?? 0) > 20); + await crash(opened.harness); + + // The rerun's first commit stores the cleared progress; it crashes again before writing any of its own. + state.stall = true; + opened = await open(storage, setup); + await state.stalled.promise; + expect(await storedOutput(storage, opened.root.id)).toBeUndefined(); + await crash(opened.harness); + expect(await storedOutput(storage, opened.root.id)).toBeUndefined(); + + state.stall = false; + setup.faux.setResponses([fauxAssistantMessage([fauxText("done")])]); + opened = await open(storage, setup); + expect((await (await opened.harness.submission(submission.id, context))!.wait(context)).status).toBe("done"); + expect(state.runs).toBe(3); + expect(results(await allEntries(opened.root))).toEqual(["rerun 3\n"]); + await opened.harness.close(context); + }); +}); + +describe("compaction", () => { + it("reruns a summarization cut off by a crash from its checkpoint", async () => { + const storage = new ControlledStorage(); + const setup = chatSetup(); + setup.settings.compaction = { enabled: false, reserveTokens: 1000, keepRecentTokens: 150, backgroundTokens: 0 }; + const summarizing = deferred(); + setup.faux.setResponses([ + ...["a1", "a2", "a3"].map((label) => fauxAssistantMessage(`${label} ${"word ".repeat(100)}`)), + async (_request, options) => { + summarizing.resolve(); + return aborted(options!.signal!); + }, + ]); + let opened = await open(storage, setup); + for (const label of ["u1", "u2", "u3"]) { + await (await opened.root.submit({ type: "input", content: `${label} ${"word ".repeat(100)}` }, context)).wait( + context, + ); + } + const id = await opened.root.compact(undefined, context); + await summarizing.promise; + await crash(opened.harness); + + setup.faux.setResponses([fauxAssistantMessage("SUMMARY")]); + opened = await open(storage, setup); + expect((await opened.harness.waitForTask(id, context)).state.outcome.status).toBe("completed"); + await waitFor(async () => (await allEntries(opened.root)).some((entry) => entry.kind === "pi.compaction")); + expect(await opened.harness.snapshot(LiveDoc, opened.root.id, context)).toEqual({}); + await opened.harness.close(context); + }); +}); + +describe("reading pi.live in a commit", () => { + it("stores the held partial a third-party task's commit read, with what it derived", async () => { + const storage = new ControlledStorage(); + const setup = slowSetup(Number.POSITIVE_INFINITY); + const Audit = defineTask<{ conversationId: ConversationId }, { phase: "run" }, null>({ + name: "test.audit", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, ctx) => { + await runtime.commit(async (tx) => { + const live = await tx.doc(LiveDoc, task.input.conversationId); + const text = partialText(live.generation?.message) ?? ""; + await tx.appendEntry(task.input.conversationId, { kind: "test.audit", data: text }); + return { status: "terminal", outcome: { status: "completed", result: null } }; + }, ctx); + }, + }, + abort: async () => {}, + }); + addTask(setup.registry, Audit); + setup.faux.setResponses([fauxAssistantMessage(LONG)]); + let opened = await open(storage, setup); + const submission = await opened.root.submit({ type: "input", content: "hi" }, context); + await waitFor(() => (opened.memory.at(-1)?.length ?? 0) > 40); + const conversationId = opened.root.id; + const audit = await opened.harness.commit( + (tx) => tx.createTask(Audit, { conversationId }, { ownership: { kind: "conversation" }, conversationId }), + context, + ); + await opened.harness.waitForTask(audit, context); + const derived = String((await allEntries(opened.root)).find((entry) => entry.kind === "test.audit")!.data); + expect(derived.length).toBeGreaterThan(40); + // The read stored the partial it saw, in the same storage transaction as the entry. + const batch = storage.commits.find((writes) => + writes.some((write) => write.type === "entry" && write.value.kind === "test.audit"), + )!; + expect(batch.some((write) => write.type === "document.change")).toBe(true); + await crash(opened.harness); + expect(storedText(await storedLive(storage, conversationId))).toBe(derived); + + setup.faux.setResponses([fauxAssistantMessage("answer")]); + opened = await open(storage, setup); + await (await opened.harness.submission(submission.id, context))!.wait(context); + expect(answers(await allEntries(opened.root))[0]).toEqual({ text: derived, stopReason: "aborted" }); + await opened.harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-events.test.ts b/packages/durable/test/harness-events.test.ts index 8f561a205..029a6a53d 100644 --- a/packages/durable/test/harness-events.test.ts +++ b/packages/durable/test/harness-events.test.ts @@ -28,7 +28,7 @@ import { import { describe, expect, it } from "vitest"; import { type ChatSetup, chatSetup, openChat, textOf, waitFor } from "./chat-support.ts"; import { addTool } from "./harness-support.ts"; -import { context, documentChanges } from "./session-support.ts"; +import { context, documentChanges, observeCommits } from "./session-support.ts"; import { aborted, deferred } from "./task-support.ts"; /** Attach and start an event stream that records every delivered batch. */ @@ -95,7 +95,7 @@ function applyChanges(message: AssistantMessage, changes: readonly MessageChange /** Committed partials of `conversation`'s generation, one per commit that has one. */ function partialsOf(harness: Harness): AssistantMessage[] { const partials: AssistantMessage[] = []; - harness.subscribeCommits((publication) => { + observeCommits(harness, (publication) => { for (const change of documentChanges(publication)) { const message = (change.value as LiveState | null)?.generation?.message as AssistantMessage | undefined; if (change.record.kind === "pi.live" && message !== undefined) partials.push(message); @@ -113,7 +113,7 @@ describe("agent events", () => { const { stream, events } = await listen(harness, root); expect(stream.snapshot).toMatchObject({ type: "snapshot", entries: [], tools: [], inbox: [] }); const partials: string[] = []; - harness.subscribeCommits((publication) => { + observeCommits(harness, (publication) => { for (const change of documentChanges(publication)) { const message = (change.value as LiveState | null)?.generation?.message as AssistantMessage | undefined; if (change.record.kind === "pi.live" && message !== undefined) partials.push(textOf(message) ?? ""); @@ -354,7 +354,7 @@ describe("agent events", () => { ]); const { harness, root } = await openChat(new MemoryStorage(), setup); const outputs: string[] = []; - harness.subscribeCommits((publication) => { + observeCommits(harness, (publication) => { for (const change of documentChanges(publication)) { const output = (change.value as LiveState | null)?.tools?.[0]?.output; if (change.record.kind === "pi.live" && output !== undefined && output !== outputs.at(-1)) { @@ -724,7 +724,7 @@ describe("streamed partials", () => { ]); const { harness, root } = await openChat(new MemoryStorage(), setup); let commits = 0; - harness.subscribeCommits(() => commits++); + observeCommits(harness, () => commits++); const bytes = await allocatedBytes(async () => { await (await root.submit({ type: "input", content: "go" }, context)).wait(context); }); diff --git a/packages/durable/test/harness-generation.test.ts b/packages/durable/test/harness-generation.test.ts index 50bae0a58..3cb077264 100644 --- a/packages/durable/test/harness-generation.test.ts +++ b/packages/durable/test/harness-generation.test.ts @@ -9,7 +9,6 @@ import { } from "@earendil-works/pi-ai"; import { AssistantEntry, - type CommitPublication, type Conversation, createRegistry, defineDoc, @@ -31,7 +30,7 @@ import { resolveSettings } from "../src/harness/agent.ts"; import type { SessionImpl } from "../src/session/session.ts"; import { allEntries, type ChatSetup, chatSetup, openChat, textOf, unanswered, waitFor } from "./chat-support.ts"; import { addSection } from "./harness-support.ts"; -import { ControlledStorage, context } from "./session-support.ts"; +import { ControlledStorage, context, observeCommits } from "./session-support.ts"; const ERROR_503 = fauxAssistantMessage([], { stopReason: "error", errorMessage: "503 Service Unavailable" }); @@ -76,7 +75,7 @@ function invalidFinalStream(): ReturnType { function livePublications(harness: Harness): LiveState[] { const values: LiveState[] = []; - harness.subscribeCommits((publication: CommitPublication) => { + observeCommits(harness, (publication) => { for (const change of publication.changes) { if (change.type === "document" && change.record.kind === "pi.live" && change.value !== null) { values.push(change.value as LiveState); @@ -509,7 +508,7 @@ describe("generation", () => { stream: {}, retry: { enabled: true, maxRetries: 3, baseDelayMs: 2000, maxAgentDelayMs: 60000 }, compaction: { enabled: true, reserveTokens: 16384, keepRecentTokens: 20000, backgroundTokens: 32768 }, - progress: { partialIntervalMs: 100, outputIntervalMs: 100 }, + progress: { partialIntervalMs: 100, outputIntervalMs: 100, persistWithinMs: 1000 }, toolExecution: "parallel", steeringMode: "one-at-a-time", followUpMode: "one-at-a-time", @@ -527,7 +526,11 @@ describe("generation", () => { expect(resolveSettings({ progress: { outputIntervalMs: 500 } }).progress).toEqual({ partialIntervalMs: 100, outputIntervalMs: 500, + persistWithinMs: 1000, }); + // Not a number of milliseconds, 0 or more: the default. + expect(resolveSettings({ progress: { persistWithinMs: -1 } }).progress.persistWithinMs).toBe(1000); + expect(resolveSettings({ progress: { persistWithinMs: 0 } }).progress.persistWithinMs).toBe(0); }); it("commits partials no more often than progress.partialIntervalMs", async () => { diff --git a/packages/durable/test/harness-live-deltas.test.ts b/packages/durable/test/harness-live-deltas.test.ts index df5593384..26db00d26 100644 --- a/packages/durable/test/harness-live-deltas.test.ts +++ b/packages/durable/test/harness-live-deltas.test.ts @@ -13,7 +13,7 @@ import { import { describe, expect, it } from "vitest"; import { chatSetup, openChat, waitFor } from "./chat-support.ts"; import { addTool } from "./harness-support.ts"; -import { context, documentChanges } from "./session-support.ts"; +import { context, documentChanges, observeCommits } from "./session-support.ts"; type Action = (api: ToolExecutionApi) => void | Promise; @@ -52,7 +52,7 @@ async function drive(outputLimits: ToolRegistration["outputLimits"] = {}) { ]); const { harness, root } = await openChat(new MemoryStorage(), setup); const commits: Op[][] = []; - harness.subscribeCommits((publication) => { + observeCommits(harness, (publication) => { for (const change of documentChanges(publication)) { if (change.record.kind === "pi.live" && change.ops.length > 0) commits.push([...change.ops]); } @@ -261,7 +261,7 @@ describe("pi.live deltas", () => { setup.faux.setResponses([fauxAssistantMessage([fauxText("word ".repeat(250))])]); const { harness, root } = await openChat(new MemoryStorage(), setup); const commits: Op[][] = []; - harness.subscribeCommits((publication) => { + observeCommits(harness, (publication) => { for (const change of documentChanges(publication)) { if (change.record.kind === "pi.live" && change.ops.length > 0) commits.push([...change.ops]); } diff --git a/packages/durable/test/harness-partial-changes.test.ts b/packages/durable/test/harness-partial-changes.test.ts index 3b61133b7..4e2ebf78d 100644 --- a/packages/durable/test/harness-partial-changes.test.ts +++ b/packages/durable/test/harness-partial-changes.test.ts @@ -1,4 +1,4 @@ -import type { JsonValue } from "@earendil-works/chord"; +import { copyJson } from "@earendil-works/chord"; import { type Op, track } from "@earendil-works/chord/delta"; import { type AssistantMessage, @@ -7,11 +7,12 @@ import { ToolCallArgumentsStream, } from "@earendil-works/pi-ai"; import { describe, expect, it } from "vitest"; -import { PartialChanges } from "../src/harness/partial-changes.ts"; +import type { LiveState } from "../src/harness/live.ts"; +import { PartialChanges, type PartialMessage } from "../src/harness/partial-changes.ts"; // #9265: partial commits reassigned the growing strings of a streamed answer, and Chord compared each one whole. -type Live = { generation: Record }; +type Live = { generation: NonNullable }; /** * A response streaming one text block (started empty) and then one tool call, as a provider produces it: the partial @@ -63,10 +64,11 @@ function setup() { }, /** One partial commit, as the generation's throttle makes it; `fail` aborts it like a rejected commit. */ commit(fail = false) { - const message = structuredClone(partial) as unknown as { content: JsonValue[] }; + // As the generation copies it: `copyJson()` returns untyped JSON, here a message's JSON representation. + const message = copyJson(partial) as PartialMessage; const taken = changes.take(message); const change = tracker.beginChange(); - changes.commit(change.state.generation, message as unknown as JsonValue, taken); + changes.commit(change.state.generation, message, taken); if (fail) { change.abort(); changes.restore(taken); diff --git a/packages/durable/test/harness-registry.test.ts b/packages/durable/test/harness-registry.test.ts index bf2edf002..af852b0c1 100644 --- a/packages/durable/test/harness-registry.test.ts +++ b/packages/durable/test/harness-registry.test.ts @@ -142,6 +142,7 @@ describe("settings", () => { expect(resolveSettings({ progress: { partialIntervalMs: undefined, outputIntervalMs: 250 } }).progress).toEqual({ partialIntervalMs: 100, outputIntervalMs: 250, + persistWithinMs: 1000, }); }); }); diff --git a/packages/durable/test/harness-structured.test.ts b/packages/durable/test/harness-structured.test.ts index 7bc47916d..77fa781cc 100644 --- a/packages/durable/test/harness-structured.test.ts +++ b/packages/durable/test/harness-structured.test.ts @@ -1336,6 +1336,9 @@ describe("tool rounds", () => { }, }); const setup = { ...base, models }; + // The partial is stored at once, so the failed final commit leaves it for the reopened run to convert; held, it would + // be lost with the failed Session. + setup.settings.progress = { persistWithinMs: 0 }; addTask(setup.registry, Node); let opened: Harness | undefined; addHooks(setup.registry, GenerationTask, { diff --git a/packages/durable/test/harness-tools-recovery.test.ts b/packages/durable/test/harness-tools-recovery.test.ts index 226bade81..31485ab25 100644 --- a/packages/durable/test/harness-tools-recovery.test.ts +++ b/packages/durable/test/harness-tools-recovery.test.ts @@ -83,7 +83,7 @@ async function toolTaskId(harness: Harness): Promise { /** * A tool that writes output, then blocks until its invocation is cancelled the first `blocking` times it runs. - * `started` resolves once the output is durable. + * `started` resolves once the output is committed; a close stores it. */ function blockingTool(name: string, extra: Partial = {}) { const started = deferred(); diff --git a/packages/durable/test/session-deferred-commits.test.ts b/packages/durable/test/session-deferred-commits.test.ts new file mode 100644 index 000000000..e4281c5e8 --- /dev/null +++ b/packages/durable/test/session-deferred-commits.test.ts @@ -0,0 +1,653 @@ +// Deferred commits (`CommitPersistence`): what a deferred commit holds in memory, what stores it, and what Storage and +// each kind of listener then sees. +import { setFlagsFromString } from "node:v8"; +import { runInNewContext } from "node:vm"; +import type { JsonValue } from "@earendil-works/chord"; +import { applyImmutable } from "@earendil-works/chord/delta"; +import { + type CommitChange, + type CommitOptions, + type CommitPublication, + type ConversationId, + type DocumentId, + defineDoc, + type JsonObject, + ReadOnlySession, + type SetTimer, + type StorageWrite, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { type ObservedCommit, SessionImpl } from "../src/session/session.ts"; +import { ControlledStorage, context, createConversation, flush } from "./session-support.ts"; + +type Text = { text: string; base?: boolean; explode?: boolean; keys?: Record }; + +/** A latest, initial-fork conversation document, the kind a deferred commit may hold. */ +const TextDoc = defineDoc({ + kind: "test.deferred.text", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ text: "" }), + checkpointWhen: (value) => { + if (value.explode === true) throw new Error("checkpoint failed"); + return value.base === true; + }, +}); + +/** `TextDoc` at version 2, which a reload migrates to. */ +const TextDocV2 = defineDoc({ + kind: "test.deferred.text", + version: 2, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ text: "", migrated: true }), + // `migrate()` gets the older version's untyped value: version 1's `Text`. + migrate: (value) => ({ ...(value as Text), migrated: true }), + checkpointWhen: () => false, +}); + +const RewindableDoc = defineDoc({ + kind: "test.deferred.rewindable", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ text: "" }), + checkpointWhen: () => false, +}); + +const CurrentForkDoc = defineDoc({ + kind: "test.deferred.current-fork", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ text: "" }), + checkpointWhen: () => false, +}); + +const SessionTextDoc = defineDoc({ + kind: "test.deferred.session", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + checkpointWhen: () => false, +}); + +/** Chord's `MAX_PENDING_OPS`: held operations past which a store writes the whole value. */ +const MAX_PENDING_OPS = 10_000; + +const DEFERRED: CommitOptions = { persistence: { kind: "deferred" } }; + +const IMMEDIATE: CommitOptions = { persistence: { kind: "immediate" } }; + +const within = (withinMs: number): CommitOptions => ({ persistence: { kind: "deferred", withinMs } }); + +/** One foreground timer the Session armed. */ +type Armed = { readonly callback: () => void; readonly delayMs: number; disarmed: boolean; fired: boolean }; + +/** + * A Session over controlled storage with a manual clock and manual foreground timers (background ones are declined, so + * commits evict), recording what `subscribeCommits()` and the Session's own observers get. + */ +async function open(options: { readonly retentionMs?: number; readonly persistWithinMs?: number } = {}) { + const clock = { now: 1_000 }; + const timers: Armed[] = []; + const setTimer: SetTimer = (callback, delayMs, { background }) => { + if (background) return undefined; + const timer: Armed = { callback, delayMs, disarmed: false, fired: false }; + timers.push(timer); + return () => { + timer.disarmed = true; + }; + }; + const storage = new ControlledStorage(); + const session = new SessionImpl( + storage, + () => clock.now, + () => options.retentionMs ?? 600_000, + setTimer, + false, + () => options.persistWithinMs, + ); + const stored: CommitPublication[] = []; + session.subscribeCommits((publication) => { + stored.push(publication); + }); + const observed: ObservedCommit[] = []; + session.observeCommits((publication) => { + observed.push(publication); + }); + const conversationId = await createConversation(session); + await session.commit(async (tx) => { + await tx.doc(TextDoc, conversationId); + }, context); + const recordId = (await storage.findDocument(address(conversationId), "current", context))!.id; + return { + clock, + storage, + session, + stored, + observed, + conversationId, + recordId, + /** Timers armed and neither disarmed nor fired. */ + armed: () => timers.filter((timer) => !timer.disarmed && !timer.fired), + /** Fire every armed timer and let the line run what they queued. */ + async fire(): Promise { + for (const timer of timers) { + if (timer.disarmed || timer.fired) continue; + timer.fired = true; + timer.callback(); + } + await session.readOnLine(async () => {}); + }, + /** Batches admitted to Storage from here on. */ + mark(): () => (readonly StorageWrite[])[] { + const from = storage.commits.length; + return () => storage.commits.slice(from); + }, + async storedText(id: DocumentId = recordId): Promise { + return (await storage.document(id, "current", context))?.value; + }, + sizes: () => session.documentCacheSizes, + }; +} + +function address(conversationId: ConversationId) { + return { kind: TextDoc.definition.kind, scope: { kind: "conversation", conversationId } } as const; +} + +function append( + session: SessionImpl, + conversationId: ConversationId, + text: string, + options: CommitOptions = DEFERRED, +): Promise { + return session.commit( + async (tx) => { + (await tx.doc(TextDoc, conversationId)).text += text; + }, + context, + options, + ); +} + +function changesOf( + batches: readonly (readonly StorageWrite[])[], +): Extract[] { + return batches.flat().filter((write) => write.type === "document.change"); +} + +describe("holding a deferred commit", () => { + it("adopts a commit that only changes an existing latest, initial-fork conversation document in memory, unstored", async () => { + const { session, storage, stored, observed, conversationId, sizes, storedText } = await open(); + const admitted = storage.commits.length; + const publications = stored.length; + await append(session, conversationId, "a"); + await append(session, conversationId, "b"); + expect(storage.commits.length).toBe(admitted); + expect(stored.length).toBe(publications); + expect(await session.snapshot(TextDoc, conversationId, context)).toEqual({ text: "ab" }); + expect(await storedText()).toEqual({ text: "" }); + expect(sizes().held).toBe(1); + expect(observed.slice(-2).map((publication) => publication.kind)).toEqual(["memory", "memory"]); + }); + + it("stores a deferred commit at once when it writes anything else", async () => { + const { session, storage, conversationId, sizes } = await open(); + const stores = async (change: Parameters[0]): Promise => { + const admitted = storage.commits.length; + await session.commit(change, context, DEFERRED); + return storage.commits.length > admitted; + }; + // An entry next to the held document's change. + expect( + await stores(async (tx) => { + (await tx.doc(TextDoc, conversationId)).text += "x"; + await tx.appendEntry(conversationId, { kind: "test.note" }); + }), + ).toBe(true); + // Creation, then other document semantics. + expect(await stores(async (tx) => void (await tx.doc(RewindableDoc, conversationId)))).toBe(true); + expect( + await stores(async (tx) => { + (await tx.doc(RewindableDoc, conversationId)).text = "r"; + }), + ).toBe(true); + expect(await stores(async (tx) => void (await tx.doc(CurrentForkDoc, conversationId)))).toBe(true); + expect( + await stores(async (tx) => { + (await tx.doc(CurrentForkDoc, conversationId)).text = "c"; + }), + ).toBe(true); + expect(await stores(async (tx) => void (await tx.doc(SessionTextDoc)))).toBe(true); + expect( + await stores(async (tx) => { + (await tx.doc(SessionTextDoc)).text = "s"; + }), + ).toBe(true); + // A held document next to an immediate one: the whole commit is stored. + expect( + await stores(async (tx) => { + (await tx.doc(TextDoc, conversationId)).text += "y"; + (await tx.doc(SessionTextDoc)).text += "s"; + }), + ).toBe(true); + // Retirement. + expect(await stores((tx) => tx.retireDoc(CurrentForkDoc, conversationId))).toBe(true); + expect(sizes().held).toBe(0); + // The control: the same change alone is held. + expect( + await stores(async (tx) => { + (await tx.doc(TextDoc, conversationId)).text += "z"; + }), + ).toBe(false); + }); + + it("stores a deferred commit that migrates its document at once, as a base", async () => { + const { session, conversationId, mark, sizes } = await open(); + const batches = mark(); + await session.commit( + async (tx) => { + (await tx.doc(TextDocV2, conversationId)).text = "v2"; + }, + context, + DEFERRED, + ); + expect(changesOf(batches()).map((write) => write.content.kind)).toEqual(["base"]); + expect(sizes().held).toBe(0); + }); + + it("rejects a deferred commit on a read-only Session like any commit with writes", async () => { + const { storage, conversationId } = await open(); + const viewer = new SessionImpl(storage, Date.now, () => 600_000, undefined, true); + await expect(append(viewer, conversationId, "a")).rejects.toBeInstanceOf(ReadOnlySession); + }); + + it("rejects a withinMs that is not a number of milliseconds, 0 or more", async () => { + const { session, conversationId } = await open(); + await expect(append(session, conversationId, "a", within(-1))).rejects.toBeInstanceOf(TypeError); + await expect(append(session, conversationId, "a", within(Number.NaN))).rejects.toBeInstanceOf(TypeError); + await expect(append(session, conversationId, "a", { persistence: { kind: "immediate" } })).resolves.toBe( + undefined, + ); + }); +}); + +describe("storing held changes", () => { + it("stores them first, folded with its own, in a commit that writes the document", async () => { + const { session, conversationId, mark, sizes, storedText } = await open(); + await append(session, conversationId, "a"); + await append(session, conversationId, "b"); + const batches = mark(); + await append(session, conversationId, "c", IMMEDIATE); + expect(batches()).toHaveLength(1); + expect(changesOf(batches()).map((write) => write.content)).toEqual([ + { version: 1, kind: "delta", ops: [["a", ["text"], "abc"]] }, + ]); + expect(await storedText()).toEqual({ text: "abc" }); + expect(sizes().held).toBe(0); + }); + + it("stores them in a commit that only reads the document, with its other writes", async () => { + const { session, conversationId, mark, storedText } = await open(); + await append(session, conversationId, "held"); + const batches = mark(); + await session.commit(async (tx) => { + const text = (await tx.doc(TextDoc, conversationId)).text; + await tx.appendEntry(conversationId, { kind: "test.derived", data: text }); + }, context); + expect(batches()).toHaveLength(1); + expect( + batches()[0]! + .map((write) => write.type) + .sort(), + ).toEqual(["document.change", "entry"]); + expect(await storedText()).toEqual({ text: "held" }); + }); + + it("leaves them held through commits that neither read nor write the document, or write nothing", async () => { + const { session, conversationId, mark, sizes } = await open(); + await append(session, conversationId, "held"); + const batches = mark(); + await session.commit((tx) => tx.appendEntry(conversationId, { kind: "test.unrelated" }).then(() => {}), context); + await session.commit(async (tx) => void (await tx.doc(TextDoc, conversationId)), context); + expect(changesOf(batches())).toEqual([]); + expect(sizes().held).toBe(1); + }); + + it("keeps them when a commit that reads the document fails before Storage", async () => { + const { session, conversationId, sizes, storedText, fire } = await open(); + await append(session, conversationId, "held"); + await expect( + session.commit(async (tx) => { + (await tx.doc(TextDoc, conversationId)).text += "!"; + await tx.appendEntry(conversationId, { kind: "test.note" }); + throw new Error("callback failed"); + }, context), + ).rejects.toThrow("callback failed"); + // Its checkpoint predicate throws: settlement fails after every other step. + await expect( + session.commit(async (tx) => { + (await tx.doc(TextDoc, conversationId)).explode = true; + }, context), + ).rejects.toThrow("checkpoint failed"); + expect(sizes().held).toBe(1); + await fire(); + expect(await storedText()).toEqual({ text: "held" }); + }); + + it("stores every held document in one transaction when the earliest deadline comes", async () => { + const { session, conversationId, mark, armed, fire, stored, sizes, storage } = await open(); + const other = await createConversation(session); + await session.commit(async (tx) => void (await tx.doc(TextDoc, other)), context); + await append(session, conversationId, "one"); + await append(session, other, "two"); + expect(armed().map((timer) => timer.delayMs)).toEqual([1_000]); + const publications = stored.length; + const batches = mark(); + await fire(); + expect(batches()).toHaveLength(1); + expect(changesOf(batches())).toHaveLength(2); + expect(stored.length).toBe(publications + 1); + expect(sizes().held).toBe(0); + expect(armed()).toEqual([]); + const otherId = (await storage.findDocument(address(other), "current", context))!.id; + expect((await storage.document(otherId, "current", context))?.value).toEqual({ text: "two" }); + }); + + it("arms the earliest deadline of the held commits, each with its withinMs or the default", async () => { + const { session, conversationId, armed, clock } = await open({ persistWithinMs: 2_000 }); + await append(session, conversationId, "a"); + expect(armed().map((timer) => timer.delayMs)).toEqual([2_000]); + clock.now += 500; + // Earlier: the timer moves up. + await append(session, conversationId, "b", within(100)); + expect(armed().map((timer) => timer.delayMs)).toEqual([100]); + // Later: the deadline stays. + await append(session, conversationId, "c", within(60_000)); + expect(armed().map((timer) => timer.delayMs)).toEqual([100]); + }); + + it("arms no timer for an infinite window, and moves the timer on when a commit stores the held document", async () => { + const { session, conversationId, armed, sizes } = await open(); + await append(session, conversationId, "a", within(Number.POSITIVE_INFINITY)); + expect(armed()).toEqual([]); + expect(sizes().held).toBe(1); + await append(session, conversationId, "b"); + expect(armed().map((timer) => timer.delayMs)).toEqual([1_000]); + await append(session, conversationId, "c", IMMEDIATE); + expect(armed()).toEqual([]); + expect(sizes().held).toBe(0); + }); + + it("stores them at close, after the commits admitted before it, and closes Storage when that store fails", async () => { + const { session, storage, conversationId, recordId } = await open(); + await append(session, conversationId, "held"); + await session.close(context); + expect((await storage.reopen().document(recordId, "current", context))?.value).toEqual({ text: "held" }); + + const failing = await open(); + await append(failing.session, failing.conversationId, "lost"); + const failure = new Error("disk gone"); + failing.storage.failNextCommit(failure); + await expect(failing.session.close(context)).rejects.toBe(failure); + expect(await failing.session.closed).toEqual({ reason: "failed", error: failure }); + // Closed: the backend refuses calls until reopened. + await expect(failing.storage.document(failing.recordId, "current", context)).rejects.toThrow("closed"); + expect((await failing.storage.reopen().document(failing.recordId, "current", context))?.value).toEqual({ + text: "", + }); + }); + + it("stores nothing once the Session failed", async () => { + const { session, storage, conversationId, recordId, mark } = await open(); + await append(session, conversationId, "lost"); + const batches = mark(); + session.fail(new Error("crash")); + await session.closed; + expect(batches()).toEqual([]); + expect((await storage.reopen().document(recordId, "current", context))?.value).toEqual({ text: "" }); + }); + + it("stores them before unloadDocuments() drops the trackers", async () => { + const { session, conversationId, storedText, sizes } = await open(); + await append(session, conversationId, "held"); + await session.unloadDocuments(); + expect(await storedText()).toEqual({ text: "held" }); + expect(sizes()).toMatchObject({ cached: 0, held: 0 }); + expect(await session.snapshot(TextDoc, conversationId, context)).toEqual({ text: "held" }); + }); + + it("stores them before a token of another definition version reloads the document", async () => { + const { session, conversationId, storedText, mark } = await open(); + await append(session, conversationId, "held"); + const batches = mark(); + expect(await session.snapshot(TextDocV2, conversationId, context)).toEqual({ text: "held", migrated: true }); + expect(changesOf(batches()).map((write) => write.content.kind)).toEqual(["delta"]); + expect(await storedText()).toEqual({ text: "held" }); + // The same inside a commit, which then migrates with a base. + await session.commit(async (tx) => void (await tx.doc(TextDoc, conversationId)), context); + await append(session, conversationId, "+"); + await session.commit(async (tx) => { + (await tx.doc(TextDocV2, conversationId)).text += "v2"; + }, context); + expect(await storedText()).toEqual({ text: "held+v2", migrated: true }); + }); + + it("stores them before the document's retirement, in the retiring commit", async () => { + const { session, conversationId, mark, sizes } = await open(); + await append(session, conversationId, "held"); + const batches = mark(); + await session.commit((tx) => tx.retireDoc(TextDoc, conversationId), context); + expect(batches()[0]!.map((write) => write.type)).toEqual(["document.change", "document.retire"]); + expect(changesOf(batches())[0]!.content).toEqual({ version: 1, kind: "delta", ops: [["a", ["text"], "held"]] }); + expect(sizes().held).toBe(0); + expect(await session.snapshot(TextDoc, conversationId, context)).toBeUndefined(); + }); +}); + +describe("bases and overflow", () => { + it("stores a base when a held commit selected one, and counts the store as one delta otherwise", async () => { + const { session, conversationId, mark, fire } = await open(); + const batches = mark(); + await session.commit( + async (tx) => { + const doc = await tx.doc(TextDoc, conversationId); + doc.text = "x"; + doc.base = true; + }, + context, + DEFERRED, + ); + await append(session, conversationId, "y"); + await fire(); + expect(changesOf(batches()).map((write) => write.content)).toEqual([ + { version: 1, kind: "base", value: { text: "xy", base: true } }, + ]); + }); + + it("stores the whole value once more than MAX_PENDING_OPS operations are held", async () => { + const { session, conversationId, mark, fire, stored } = await open(); + await overflow(session, conversationId); + const batches = mark(); + await fire(); + const value = (await session.snapshot(TextDoc, conversationId, context))!; + expect(changesOf(batches()).map((write) => write.content)).toEqual([{ version: 1, kind: "base", value }]); + // A durable listener replaces its copy with the stored value. + expect(stored.at(-1)!.changes).toMatchObject([{ type: "document", ops: [["r", value]] }]); + }); + + it("stores a base of the commit's own value when it stores an overflowed hold", async () => { + const { session, conversationId, mark } = await open(); + await overflow(session, conversationId); + await append(session, conversationId, "a"); + const batches = mark(); + await append(session, conversationId, "b", IMMEDIATE); + const value = (await session.snapshot(TextDoc, conversationId, context))!; + expect(value.text).toBe("ab"); + expect(changesOf(batches()).map((write) => write.content)).toEqual([{ version: 1, kind: "base", value }]); + }); +}); + +/** + * Hold more than `MAX_PENDING_OPS` operations: deferred commits of one set per key, each below the tracker's own limit, + * past which a commit is one root replacement. + */ +async function overflow(session: SessionImpl, conversationId: ConversationId): Promise { + await session.commit(async (tx) => { + (await tx.doc(TextDoc, conversationId)).keys = {}; + }, context); + for (let from = 0; from <= MAX_PENDING_OPS; from += 4_000) { + await session.commit( + async (tx) => { + const keys = (await tx.doc(TextDoc, conversationId)).keys!; + for (let index = from; index < from + 4_000; index++) keys[`k${index}`] = index; + }, + context, + DEFERRED, + ); + } +} + +describe("publications", () => { + it("give subscribeCommits() exactly what was stored, and nothing for a held commit", async () => { + const { session, storage, conversationId, recordId, stored, fire } = await open(); + const from = storage.commits.length; + const publications = stored.length; + await append(session, conversationId, "a"); + await append(session, conversationId, "b"); + await append(session, conversationId, "c", IMMEDIATE); + await append(session, conversationId, "d"); + await fire(); + await append(session, conversationId, "e"); + await session.commit(async (tx) => { + await tx.doc(TextDoc, conversationId); + await tx.appendEntry(conversationId, { kind: "test.note" }); + }, context); + await append(session, conversationId, "f"); + await session.close(context); + const batches = storage.commits.slice(from); + const published = stored.slice(publications); + expect(published).toHaveLength(batches.length); + // A replica built from the publications alone always equals what Storage holds. + let fromStorage: JsonValue = { text: "" }; + let fromPublications: JsonValue = { text: "" }; + for (const [index, batch] of batches.entries()) { + const write = changesOf([batch]).find((candidate) => candidate.id === recordId)!; + const change = published[index]!.changes.find( + (candidate): candidate is Extract => + candidate.type === "document" && candidate.record.id === recordId, + )!; + fromStorage = + write.content.kind === "base" ? write.content.value : applyImmutable(fromStorage, write.content.ops); + fromPublications = applyImmutable(fromPublications, change.ops); + expect(fromPublications).toEqual(fromStorage); + expect(change.value).toEqual(fromStorage); + } + expect(fromStorage).toEqual({ text: "abcdef" }); + expect(published.map((publication) => publication.changes.map((change) => change.type).sort())).toEqual([ + ["document"], + ["document"], + ["document", "entry"], + ["document"], + ]); + }); + + it("show every held commit to watches and the Session's own observers, once", async () => { + const { session, conversationId, observed, fire } = await open(); + const watch = (await session.watchDoc(TextDoc, conversationId, context))!; + const values: string[] = []; + watch.start(async (value) => { + values.push(value?.text ?? ""); + }); + const from = observed.length; + for (const text of ["a", "b", "c"]) { + await append(session, conversationId, text); + await flush(); + } + await append(session, conversationId, "d", IMMEDIATE); + await flush(); + await fire(); + expect(values).toEqual(["a", "ab", "abc", "abcd"]); + const seen = observed.slice(from); + expect(seen.map((publication) => publication.kind)).toEqual(["memory", "memory", "memory", "stored"]); + // The storing commit shows its own operations; the held ones were shown already. The timer showed nothing. + expect( + seen.map((publication) => publication.changes.map((change) => change.type === "document" && change.ops)), + ).toEqual([ + [[["a", ["text"], "a"]]], + [[["a", ["text"], "b"]]], + [[["a", ["text"], "c"]]], + [[["a", ["text"], "d"]]], + ]); + await watch.stop(); + }); +}); + +describe("pinning", () => { + it("keeps a held document cached with retention 0 until it is stored, then lets it go", async () => { + const { session, conversationId, sizes, fire } = await open({ retentionMs: 0 }); + await session.readOnLine(async () => {}); + expect(sizes()).toMatchObject({ cached: 0, held: 0, attached: 0 }); + await append(session, conversationId, "held"); + await session.readOnLine(async () => {}); + expect(sizes()).toMatchObject({ cached: 1, held: 1, attached: 1 }); + await fire(); + await session.commit(async () => {}, context); + expect(sizes()).toMatchObject({ cached: 0, held: 0, attached: 0 }); + expect(await session.snapshot(TextDoc, conversationId, context)).toEqual({ text: "held" }); + }); + + it("retains a held revision only until it is stored", async () => { + setFlagsFromString("--expose-gc"); + const gc = runInNewContext("gc") as () => void; + const collect = async (): Promise => { + for (let round = 0; round < 5; round++) { + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); + } + }; + const { session, conversationId, fire, stored, observed } = await open({ retentionMs: 0 }); + await append(session, conversationId, "x".repeat(1_000)); + const revision = new WeakRef((await session.snapshot(TextDoc, conversationId, context))!); + await collect(); + expect(revision.deref()).toBeDefined(); + await fire(); + await session.commit(async () => {}, context); + // The recorded publications would keep it. + stored.length = 0; + observed.length = 0; + await collect(); + expect(revision.deref()).toBeUndefined(); + }); +}); + +describe("crash", () => { + it("leaves Storage at the last stored state, which memory had, and nothing held after reopen", async () => { + const { session, storage, conversationId, recordId, fire } = await open(); + const states: string[] = []; + session.observeCommits((publication) => { + for (const change of publication.changes) { + // Publications carry untyped document values; this one is `TextDoc`'s. + if (change.type === "document" && change.record.id === recordId) states.push((change.value as Text).text); + } + }); + await append(session, conversationId, "a"); + await fire(); + await append(session, conversationId, "b"); + await append(session, conversationId, "c"); + session.fail(new Error("crash")); + await session.closed; + const reopened = new SessionImpl(storage.reopen(), Date.now, () => 600_000); + const value = (await reopened.snapshot(TextDoc, conversationId, context))!; + expect(value).toEqual({ text: "a" }); + expect(states).toContain(value.text); + expect(reopened.documentCacheSizes.held).toBe(0); + await reopened.close(context); + }); +}); diff --git a/packages/durable/test/session-support.ts b/packages/durable/test/session-support.ts index b5c4df8b6..e3bd17d44 100644 --- a/packages/durable/test/session-support.ts +++ b/packages/durable/test/session-support.ts @@ -1,6 +1,7 @@ import type { Context } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { + type CommitChange, type CommitPublication, type ConversationId, type DocumentAddress, @@ -11,9 +12,10 @@ import { type Id, MemoryStorage, type Seq, + type Session, type StorageWrite, } from "@earendil-works/pi-durable"; -import { SessionImpl } from "../src/session/session.ts"; +import { type ObservedCommit, SessionImpl } from "../src/session/session.ts"; export const context: Context = BACKGROUND_CONTEXT; @@ -138,14 +140,26 @@ export function openTestSession(): { return { storage, session, publications }; } -export function documentChanges( - publication: CommitPublication, -): readonly Extract[] { +export function documentChanges(publication: { + readonly changes: readonly CommitChange[]; +}): readonly Extract[] { return publication.changes.filter( (change): change is Extract => change.type === "document", ); } +/** + * Observe every commit as memory adopts it, deferred ones before they are stored, as views and document watches do; + * `subscribeCommits()` sees only what was stored. + */ +export function observeCommits( + session: Session, + listener: (publication: ObservedCommit, context: Context) => void, +): () => void { + // `observeCommits()` is internal, so the public type lacks it; every Session and Harness is a `SessionImpl`. + return (session as unknown as SessionImpl).observeCommits(listener); +} + export function documentCopyChanges( publication: CommitPublication, ): readonly Extract[] {