diff --git a/packages/chord/src/output.ts b/packages/chord/src/output.ts index 80804c33e..a31cdfd2a 100644 --- a/packages/chord/src/output.ts +++ b/packages/chord/src/output.ts @@ -376,7 +376,8 @@ export class Progress { else this.#timer = this.#clock.setTimer(() => { this.#timer = undefined; - this.#flush(); + // Checks the clock again: a timer that fires early waits for the rest. + this.#schedule(); }, wait); } diff --git a/packages/chord/test/output.test.ts b/packages/chord/test/output.test.ts index ae598c026..00e7d647e 100644 --- a/packages/chord/test/output.test.ts +++ b/packages/chord/test/output.test.ts @@ -299,15 +299,20 @@ describe("Progress", () => { now = 1_040; const second = progress.markAndWait(); expect(timers.map(({ delayMs }) => delayMs)).toEqual([60]); - now = 1_100; + // A timer that fires early commits nothing and waits for the rest. + now = 1_080; timers[0]!.callback(); + expect(commits).toEqual([1_000]); + expect(timers.map(({ delayMs }) => delayMs)).toEqual([60, 20]); + now = 1_100; + timers[1]!.callback(); await second; expect(commits).toEqual([1_000, 1_100]); await new Promise((resolve) => setTimeout(resolve, 0)); progress.mark(); - expect(timers.map(({ delayMs }) => delayMs)).toEqual([60, 100]); + expect(timers.map(({ delayMs }) => delayMs)).toEqual([60, 20, 100]); await progress.stop(); - expect(timers[1]!.disarmed).toBe(true); + expect(timers[2]!.disarmed).toBe(true); }); it("rejects the waiters of a failed commit and reports its error", async () => { diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 29550359f..816836c7a 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -6,14 +6,14 @@ - `TaskRuntime.sleep()` is removed: it waited inside a phase, which holds an invocation and, after a restart, reruns the phase and restarts the wait. A task now ends its phase asleep with `runtime.sleepUntil(until, checkpoint, context)` or by returning `{ status: "sleeping", until, checkpoint }` from `runtime.commit()`; move the work after the wait into the next phase and carry what it needs in the checkpoint. `TaskRuntime.setTimer()` paces work inside a phase without waiting for it. - Tasks have a `sleeping` status, so code that handles every `TaskState`, `TaskGraphState`, or `TaskInspection` state needs a case for it. -- Opening a SQLite storage migrates it to schema version 2 automatically, rebuilding its task table. A migrated database cannot be opened by 1.3.0 or earlier, so a deployment cannot roll back past this release. +- Opening a SQLite storage migrates it to schema version 2 automatically, rebuilding its task table. A migrated database cannot be opened by 1.3.0 or earlier, so a deployment cannot roll back past this release. JSONL storage needs no migration, but 1.3.0 or earlier never resumes a sleeping task it finds, so its conversation stays busy. - `TaskRuntime` has `sleepUntil()` and `setTimer()`, and `ToolExecutionApi` has `now()`: code that builds either object by hand, such as a test fake or a tool wrapper that does not spread the original, must add them. ### Added - `settings.compaction.modelOverrides` sets compaction thresholds per model, keyed by exact `"provider/modelId"`, so one harness can serve small- and large-window models; the threshold check, overflow recovery, and the compaction task use the entry of the model they act for ([#10942](https://github.com/earendil-works/pi/issues/10942)). - Durable sleeps: a sleeping task holds no invocation, survives restarts, and resumes at its checkpoint once the Harness clock reaches `until`. -- `HarnessOptions.setTimer` arms every timer of the Session and Harness (progress throttles, idle-memory eviction, sleeps), also for `createSession()`; with `now`, a Harness runs entirely on a host or fake clock. `ToolExecutionApi.now()` reads the Harness clock. +- `HarnessOptions.setTimer` arms the timers of the Session, the scheduler, and its tasks (progress throttles, idle-memory eviction, sleeps), also for `createSession()`; with `now`, they run on a host or fake clock. A foreground timer the host throws on or declines falls back to `setTimeout`. `ToolExecutionApi.now()` reads the Harness clock. - `Harness.wake()` and `HarnessOptions.onWake` tell a host when the Session next needs a process: `active` while work can run, `at` for the earliest sleep. With `onWake`, the Harness keeps no timer for sleeps, so a Cloudflare Durable Object can hibernate and wake on its alarm, calling `resume()`. - `HarnessOptions.models` and `HarnessOptions.settings` also accept a function returning the current value, like `env`. Generation, compaction, hooks and tools call it at each use (`api.models` reads the current value at each access), so a host can replace its model configuration or settings without reopening the Harness. A plain object still works. diff --git a/packages/durable/README.md b/packages/durable/README.md index 257a3b439..4c68184ed 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -149,7 +149,7 @@ const Reminder = defineTask<{ text: string; at: number }, { phase: "wait" } | { There is no sleep in the middle of a phase: a phase reruns from its top after a restart, so work after the wait goes in the next phase, and what it needs goes in the checkpoint. `{ status: "sleeping", until, checkpoint }` returned from `runtime.commit()` is the same sleep, committed together with other writes. Retry backoffs and deferred-response polls of the built-in tasks sleep the same way. -Every timer the Harness uses goes through `setTimer` (default `setTimeout`), and every time through `now` (default `Date.now`); pass both to run a Harness on a fake clock. Sleep deadlines are stored on `now`, so every process that opens one storage must use the same clock. While scheduling runs, a timer resumes sleeping tasks when they are due and keeps a Node.js process alive meanwhile. +The timers of the Session, the scheduler, and its tasks go through `setTimer` (default `setTimeout`), and their times through `now` (default `Date.now`); pass both to run a Harness on a fake clock. The Node.js environment, file watcher, and image worker keep their own timers. Sleep deadlines are stored on `now`, so every process that opens one storage must use the same clock. While scheduling runs, a timer resumes sleeping tasks when they are due and keeps a Node.js process alive meanwhile. A host whose process can be evicted, such as a Cloudflare Durable Object, wakes it itself. With `onWake`, the Harness arms no timer for sleeps; it tells the host what `harness.wake()` says whenever that changes: `active` while some task can run now or is running, and `at`, the earliest sleep. The host keeps its alarm armed and calls `resume()` when it fires: @@ -161,7 +161,10 @@ export class Session extends DurableObject { super(ctx, env); void ctx.blockConcurrencyWhile(async () => { const storage = await openDurableObjectSqliteStorage(ctx.storage); - this.#harness = await Harness.open(storage, { models, registry, onWake: (wake) => void this.#arm(wake) }, context); + const onWake = (wake: Wake) => + // A failed alarm write resets the object; it arms again when it next starts. + void this.#arm(wake).catch(() => ctx.abort("Alarm write failed")); + this.#harness = await Harness.open(storage, { models, registry, onWake }, context); this.#harness.resume(); }); } @@ -171,8 +174,9 @@ export class Session extends DurableObject { await this.#arm(this.#harness.wake()); } + // Arm before any call that can start work (submit, compact, abortTask, abort, a registry change): `onWake` runs + // after the commit. An early alarm is harmless, a missing one strands the work. async submit(conversation: Conversation, input: string, requestId: string): Promise { - // Arm before admitting work: an early alarm is harmless, a missing one strands the work. await this.#arm({ ...this.#harness.wake(), active: true }); await conversation.submit({ type: "input", content: input, requestId }, context); } diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 65d0350d4..e8ebd993f 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -685,7 +685,7 @@ and open rejects a registry whose snapshot lacks any of them. Open changes surviving `running` tasks to `pending`, keeps `waiting` tasks waiting and `sleeping` tasks asleep, and re-evaluates finalization and `failFast` for `completing` and `waiting` tasks (section 5.5). It never migrates or terminalizes a task because -its definition is missing or unmigratable. Such a task stays `pending` or `waiting` and is **blocked**: the +its definition is missing or unmigratable. Such a task stays `pending`, `waiting`, or `sleeping` and is **blocked**: the scheduler skips it and reconsiders it whenever the registry changes (section 5.4). Any built-in document touched by recovery migrates through its ordinary typed access path. Open does not scan or migrate other documents. No handler @@ -1748,7 +1748,7 @@ type TaskRecord = { readonly endedAt?: number; } & ( | { - readonly state: Extract, { status: "pending" | "running" | "waiting" }>; + readonly state: Extract, { status: "pending" | "running" | "waiting" | "sleeping" }>; readonly memos?: Readonly>; } | { @@ -1765,7 +1765,7 @@ type RunningTask = TaskRecord & { * State a task commits for itself: a replacement checkpoint, a wait, or its outcome. A returned `terminal` state * becomes `completing` while ordinary owned work below is live (section 5.5). */ -type NextTaskState = Extract, { status: "running" | "waiting" | "terminal" }>; +type NextTaskState = Extract, { status: "running" | "waiting" | "sleeping" | "terminal" }>; interface HookRunner { each(name: K, invoke: (handler: NonNullable) => void | Promise): Promise; @@ -1882,10 +1882,11 @@ 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, 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` state ends the invocation even -when its checkpoint is unchanged; it rejects when `on` names a missing task or -closes a wait cycle (section 5.5), which no task on it could finish first, and -an empty `on` resumes at the next scheduling pass. `pending` is never returned; +completing state drops the memos. A `waiting` or `sleeping` state ends the +invocation even when its checkpoint is unchanged. A wait rejects when `on` names +a missing task or closes a wait cycle (section 5.5), which no task on it could +finish first, and an empty `on` resumes at the next scheduling pass; a sleep +rejects an `until` that is not a time (section 5.6). `pending` is never returned; only reconciliation and handover write it. `Tx` has no task replacement operation. A task changes only its own state, @@ -1944,7 +1945,7 @@ invocation queued earlier therefore either lands before the step and counts, or reaches the line after it and rejects. Before the first phase only rules 1–3 apply: -1. Terminal, `completing`, or `waiting`: stop. +1. Terminal, `completing`, `waiting`, or `sleeping`: stop. 2. Session closing: stop; preserve the checkpoint and any abort mark for reopen. 3. Run mode with a durable abort mark: end and join the run invocation; a fresh abort invocation starts once the task's ordinary owned work is no longer @@ -1964,10 +1965,11 @@ again. On open, running-task reconciliation changes surviving `running` tasks back to `pending`, preserving their checkpoint and abort mark; `waiting` tasks stay -waiting, and `completing` and `waiting` tasks are re-evaluated (section 5.5). Task migration runs at -reservation, atomically with `pending` or `waiting` -> `running`. One callback handles every +waiting and `sleeping` tasks asleep, and `completing` and `waiting` tasks are +re-evaluated (section 5.5). Task migration runs at reservation, atomically with +`pending`, `waiting`, or `sleeping` -> `running`. One callback handles every supported older version. A task whose definition is missing, older than the -stored version, or fails migration stays `pending` or `waiting` and blocked until a fitting +stored version, or fails migration stays `pending`, `waiting`, or `sleeping` and blocked until a fitting definition is installed or the task is aborted (section 5.4). `close()` marks the runtime closing, seals admission and reservation, signals invocations, and ends states and watches. Outside the Session line it settles admitted @@ -2361,8 +2363,11 @@ 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 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 may fire early -or late; every caller checks its clock again. +process alive, since a sleeping task is work left to do. A timer must not fire +before its delay and may fire late; the default splits a delay beyond what +`setTimeout` supports itself. A host that throws on or declines a foreground +timer has it armed with `setTimeout` instead, and the first failure is reported. +Callers still check their clock when a timer fires. **Wakeups.** `Harness.wake()` says, from memory and also while scheduling is paused, when the Session next needs a running process: @@ -2380,9 +2385,12 @@ whenever the value changes, off the Session line; a throw is reported. The host arms its own wakeup and calls `resume()` when it fires, which makes due sleeps runnable. A host whose process can be evicted, such as a Durable Object with its one alarm, arms the alarm at `at`, or at `now + heartbeat` while `active` -if that is earlier, and deletes it when neither is set. It arms the recovery wakeup -before admitting work, since an early wakeup is harmless and a missing one -leaves work stranded. +if that is earlier, and deletes it when neither is set. `onWake` runs after the +commit that changes the value, so a host call that can make an idle Session +active arms the recovery wakeup first: `submit()`, `compact()`, `abortTask()`, +`Conversation.abort()`, a registry change, and `resume()` after open. An early +wakeup is harmless; a missing one leaves work stranded. A failed alarm write +resets the object, which arms again when it next starts. ## 6. Submissions and inbox diff --git a/packages/durable/src/codemode/codemode.ts b/packages/durable/src/codemode/codemode.ts index a8bcde21a..a58b2b345 100644 --- a/packages/durable/src/codemode/codemode.ts +++ b/packages/durable/src/codemode/codemode.ts @@ -799,6 +799,8 @@ class CodemodeCall { })); running.then( (done) => { + // A result after the outcome is decided has nowhere to go, and the call's API has ended. + if (this.stopped) return; this.queue.push({ id: call.id, at: this.api.now(), ...done }); void this.pump(); }, diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index ffd708c7a..9176625ae 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -237,9 +237,9 @@ export const GenerationTask = defineTask { const { attempt, compacted, model: ref, cutoff, handle, pollAt } = task.state.checkpoint; + if (pollAt > runtime.now()) return runtime.sleepUntil(pollAt, task.state.checkpoint, context); const model = runtime.models.getModel(ref.provider, ref.modelId); if (model === undefined) return failNoModel(runtime, ref, context); - if (pollAt > runtime.now()) return runtime.sleepUntil(pollAt, task.state.checkpoint, context); const message = await runtime.models.fetchDeferred(model, handle, { signal: runtime.signal }); const request = { attempt, compacted, model: ref, cutoff, pollAt }; await classify(runtime, request, message, context); diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 9381b407d..0c5b84ae0 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -1164,9 +1164,10 @@ export class TaskScheduler { } const last = this.#lastWake; if (last !== undefined && last.active === wake.active && last.at === wake.at) return; - this.#lastWake = wake; try { this.#onWake(wake); + // Only a delivered wake counts: after a throw, the next change or kick offers it again. + this.#lastWake = wake; } catch (error) { this.#report(error); } @@ -1200,7 +1201,9 @@ export class TaskScheduler { } /** A sleep whose `until` the clock has not reached; an abort mark ends a sleep. */ - #notDue(record: AnyTaskRecord): boolean { + #notDue( + record: AnyTaskRecord, + ): record is AnyTaskRecord & { readonly state: { readonly status: "sleeping"; readonly until: number } } { const state = record.state; return state.status === "sleeping" && !record.abortRequested && state.until > this.#now(); } @@ -1249,6 +1252,9 @@ export class TaskScheduler { return; } await this.#loadScopes(false); + // Every pass picks up sleeps whose time has come: a pass started by any change may be the first to run after + // a deadline, and the wake it schedules no longer keeps a timer for a due sleep. + this.#wakeDue(); // Taken once per pass, and only when some task is a candidate. let snapshot: RegistrySnapshot | undefined; for (const id of [...this.#runnable]) { @@ -1262,7 +1268,7 @@ export class TaskScheduler { this.#runnable.delete(id); continue; } - // It returns when its time comes: `#wakeDue()` from the timer or `resume()`. + // It returns at the first pass after its time comes, which the timer, `resume()`, or any change starts. if (this.#notDue(record)) { this.#runnable.delete(id); continue; @@ -1392,8 +1398,7 @@ export class TaskScheduler { if (record.state.status === "completing") return { kind: "completing" }; const on = this.#waitingOn(record); if (on.length > 0) return { kind: "waiting", on }; - if (this.#notDue(record) && record.state.status === "sleeping") - return { kind: "sleeping", until: record.state.until }; + if (this.#notDue(record)) return { kind: "sleeping", until: record.state.until }; const fit = this.#fit(record, snapshot.task(record.kind)); if ("reason" in fit) return { kind: "blocked", ...fit }; if (fit.migrates && fit.task.definition.migrate === undefined) { diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index eae03bc8c..cc894d876 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -7,6 +7,7 @@ import { type OutputLimits, PROGRESS_BYTES_PER_SECOND, Progress, + type ProgressClock, } from "@earendil-works/chord/output"; import type { ImageContent, TextContent, ToolCall, ToolResultMessage } from "@earendil-works/pi-ai"; import { validateToolArguments } from "@earendil-works/pi-ai/utils/validation"; @@ -442,6 +443,25 @@ async function run( for (const waiter of pending) waiter.resolve(); } +/** + * The Harness clock and timers for progress pacing. After the invocation ends nothing more commits, and the clock + * answers with the last time it read. + */ +function progressClock(runtime: Runtime): ProgressClock { + let last = 0; + return { + now: () => { + try { + last = runtime.now(); + } catch { + // The invocation ended. + } + return last; + }, + setTimer: (callback, delayMs) => runtime.setTimer(callback, delayMs), + }; +} + /** * 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. @@ -493,17 +513,7 @@ function publishProgress(runtime: Runtime, reported: Reported, enabled: boolean, if (!runtime.signal.aborted) runtime.report(error); }, runtime.settings.progress.outputIntervalMs, - { - // After the invocation ends nothing more commits; the clock only has to answer. - now: () => { - try { - return runtime.now(); - } catch { - return Date.now(); - } - }, - setTimer: (callback, delayMs) => runtime.setTimer(callback, delayMs), - }, + progressClock(runtime), ); } diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index b17091ded..81d620f44 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -609,8 +609,9 @@ export type HarnessOptions = { */ readonly now?: () => number; /** - * Arms every timer the Harness uses: pacing progress commits, freeing idle memory (`background`), and, without - * `onWake`, resuming sleeping tasks; and the timers of `TaskRuntime.setTimer()`. Default `setTimeout`, with background + * Arms the timers of the Session, the scheduler, and its tasks: pacing progress commits, freeing idle memory + * (`background`), and, without `onWake`, resuming sleeping tasks; and the timers of `TaskRuntime.setTimer()`. The + * Node.js environment, file watcher, and image worker keep their own timers. Default `setTimeout`, with background * timers unreferenced, or declined in Cloudflare Workers and where they cannot be unreferenced. Tests can pass fake * timers; a host can keep timers off a process that should hibernate. */ @@ -619,8 +620,9 @@ export type HarnessOptions = { * Wakes the process for sleeping tasks instead of a `setTimer` timer. Without it, while scheduling runs, a timer that * keeps the process alive resumes sleeping tasks when they are due. With it, the Harness arms no timer for them: it * calls `onWake` after open and whenever `Harness.wake()` changes, and the host arms its own wakeup, such as a - * Durable Object alarm, and calls `resume()` when it fires. Called off the Session line; it may call `wake()` and - * `resume()`, and must not block. A throw is reported and ignored. + * Durable Object alarm, and calls `resume()` when it fires. Called off the Session line, after the commit that + * changed the value, so a host arms a recovery wakeup before calls that can start work. It may call `wake()` and + * `resume()`, and must not block. A throw is reported, and the value is offered again at the next change. */ readonly onWake?: (wake: Wake) => void; /** diff --git a/packages/durable/src/timers.ts b/packages/durable/src/timers.ts index e4a2e2c2c..0e270c9b4 100644 --- a/packages/durable/src/timers.ts +++ b/packages/durable/src/timers.ts @@ -1,6 +1,6 @@ import type { SetTimer } from "./types.ts"; -/** Longest delay `setTimeout` supports; a longer one fires early, and its caller arms again. */ +/** Longest delay `setTimeout` supports; a longer one is waited for in several steps. */ const MAX_TIMER_DELAY = 2_147_483_647; /** @@ -10,44 +10,62 @@ const MAX_TIMER_DELAY = 2_147_483_647; const inWorkers = typeof navigator !== "undefined" && navigator.userAgent === "Cloudflare-Workers"; /** - * The default `HarnessOptions.setTimer`: `setTimeout`. A background timer is unreferenced, so it never keeps the - * process alive. It is declined in Cloudflare Workers and wherever timers cannot be unreferenced. + * The default `HarnessOptions.setTimer`: `setTimeout`, never firing early, also for delays beyond its limit. A + * background timer is unreferenced, so it never keeps the process alive. It is declined in Cloudflare Workers and + * wherever timers cannot be unreferenced. */ export const nativeTimer: SetTimer = (callback, delayMs, options) => { if (options.background && inWorkers) return undefined; - const handle = setTimeout(callback, Math.min(Math.max(0, delayMs), MAX_TIMER_DELAY)); - if (options.background) { + let handle: ReturnType; + const arm = (remaining: number): boolean => { + handle = setTimeout( + () => (remaining > MAX_TIMER_DELAY ? arm(remaining - MAX_TIMER_DELAY) : callback()), + Math.min(remaining, MAX_TIMER_DELAY), + ); + if (!options.background) return true; const unref = (handle as { unref?: unknown }).unref; - if (typeof unref !== "function") { - clearTimeout(handle); - return undefined; - } + if (typeof unref !== "function") return false; unref.call(handle); + return true; + }; + if (!arm(Math.max(0, delayMs))) { + clearTimeout(handle!); + return undefined; } return () => clearTimeout(handle); }; /** Timers the Session and Harness arm through the host's `setTimer`. */ export type Timers = { - /** A timer that paces or resumes work. A host that declines it is reported, and the timer never fires. */ + /** A timer that paces or resumes work; when the host throws or declines it, `setTimeout` arms it instead. */ foreground(callback: () => void, delayMs: number): () => void; /** A timer that only frees memory; `undefined` when the host declines it. */ background(callback: () => void, delayMs: number): (() => void) | undefined; }; -/** Wrap a host `setTimer`: a throw or a declined foreground timer is reported, and arms nothing. */ +/** + * Wrap a host `setTimer`. A foreground timer must fire, or work stalls: one the host throws on or declines is armed + * with `setTimeout`, and the first such failure is reported. A background one the host throws on is reported and + * declined. + */ export function hostTimers(setTimer: SetTimer | undefined, report: (error: unknown) => void): Timers { const arm = setTimer ?? nativeTimer; + let reported = false; + const fallBack = (error: unknown): void => { + if (reported) return; + reported = true; + report(error); + }; return { foreground: (callback, delayMs) => { try { const disarm = arm(callback, delayMs, { background: false }); if (disarm !== undefined) return disarm; - report(new Error("HarnessOptions.setTimer declined a timer that paces work")); + fallBack(new Error("HarnessOptions.setTimer declined a timer that paces work; using setTimeout")); } catch (error) { - report(error); + fallBack(error); } - return () => {}; + return nativeTimer(callback, delayMs, { background: false })!; }, background: (callback, delayMs) => { try { diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 11819eb59..a07657a2e 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -134,11 +134,10 @@ export type TaskDocFamilyToken = DocF >; /** - * A host timer, given as `HarnessOptions.setTimer`: call `callback` once, about `delayMs` from now, and return a - * function that disarms it. `background` timers only free memory, and a host may decline one by returning `undefined`, - * for example where a pending timer keeps the process from hibernating; the Harness then frees that memory at later - * commits. Other timers pace or resume work and must be armed. A timer may fire early or late: callers check their - * clock again. + * A host timer, given as `HarnessOptions.setTimer`: call `callback` once, no sooner than `delayMs` from now, and return + * a function that disarms it. `background` timers only free memory, and a host may decline one by returning + * `undefined`, for example where a pending timer keeps the process from hibernating; the Harness then frees that memory + * at later commits. Other timers pace or resume work: one the host throws on or declines is armed with `setTimeout`. */ export type SetTimer = ( callback: () => void, diff --git a/packages/durable/test-workers/cloudflare.d.ts b/packages/durable/test-workers/cloudflare.d.ts index c3b6eb11e..6842c0485 100644 --- a/packages/durable/test-workers/cloudflare.d.ts +++ b/packages/durable/test-workers/cloudflare.d.ts @@ -16,6 +16,8 @@ declare module "cloudflare:workers" { export type DurableObjectState = { readonly storage: DurableObjectStorage; blockConcurrencyWhile(callback: () => Promise): Promise; + /** Reset the object: its instance is dropped, and the next event starts a new one. */ + abort(reason?: string): void; }; export type DurableObjectId = { toString(): string }; diff --git a/packages/durable/test-workers/wake.test.ts b/packages/durable/test-workers/wake.test.ts index 41c746bb3..a28a739c9 100644 --- a/packages/durable/test-workers/wake.test.ts +++ b/packages/durable/test-workers/wake.test.ts @@ -87,10 +87,20 @@ describe("a Harness woken by its Durable Object alarm", () => { await eventually(stub, async (_instance, state) => (await state.storage.getAlarm()) === null); }); - it("declines background timers by default, so no pending timer keeps an object from hibernating", () => { + it("can evict an object whose Harness keeps idle documents and contexts in memory", async () => { + const stub = fresh(); + await runInDurableObject(stub, async (instance: TestSession) => { + instance.script(["answer"]); + await instance.submit("hi", "turn-1"); + const root = await instance.root(); + await (await root.submit({ type: "input", content: "hi", requestId: "turn-1" }, context)).wait(context); + }); + // Retention timers would keep the object running; the default declines them in Workers. expect(nativeTimer(() => {}, 10, { background: true })).toBeUndefined(); - const disarm = nativeTimer(() => {}, 10, { background: false }); - expect(typeof disarm).toBe("function"); - disarm?.(); + const evicted = await Promise.race([ + evictDurableObject(stub).then(() => "evicted"), + new Promise((resolve) => setTimeout(() => resolve("still running"), 2_000)), + ]); + expect(evicted).toBe("evicted"); }); }); diff --git a/packages/durable/test-workers/worker.ts b/packages/durable/test-workers/worker.ts index 7a94c0656..888f298af 100644 --- a/packages/durable/test-workers/worker.ts +++ b/packages/durable/test-workers/worker.ts @@ -65,7 +65,8 @@ export class TestSession extends DurableObject { now: () => this.now(), onWake: (wake) => { this.wakes.push(wake); - void this.arm(wake); + // A failed alarm write resets the object; it arms again when it next starts. + void this.arm(wake).catch(() => ctx.abort("Alarm write failed")); }, }, context, diff --git a/packages/durable/test/codemode.test.ts b/packages/durable/test/codemode.test.ts index 84d7f2a30..9df95eb1d 100644 --- a/packages/durable/test/codemode.test.ts +++ b/packages/durable/test/codemode.test.ts @@ -663,6 +663,30 @@ describe("durable code mode", () => { await harness.close(context); }); + it("drops a global's result that arrives after the script finished", async () => { + const late = deferred(); + const chat = codemodeChat([ + { name: "slowly", execute: () => late.promise }, + { name: "quickly", execute: async () => "quick" }, + ]); + chat.setup.faux.setResponses([script(`const pending = slowly(); await quickly(); return 1;`), DONE]); + const rejections: unknown[] = []; + const onRejection = (reason: unknown) => rejections.push(reason); + process.on("unhandledRejection", onRejection); + try { + const { harness, root } = await openChat(await openNodeSqliteStorage(await sqlitePath()), chat.setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(await resultText(chat.setup, root)).toEqual({ text: "1", isError: false }); + // The script's tool call has ended; its pending global resolves now. + late.resolve("late"); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(rejections).toEqual([]); + await harness.close(context); + } finally { + process.off("unhandledRejection", onRejection); + } + }); + it("rejects a failed tool call with its error text and structured output", async () => { const chat = codemodeChat(); addTool( diff --git a/packages/durable/test/harness-compaction.test.ts b/packages/durable/test/harness-compaction.test.ts index ebea5e8a8..cc4a85797 100644 --- a/packages/durable/test/harness-compaction.test.ts +++ b/packages/durable/test/harness-compaction.test.ts @@ -1,6 +1,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { type AssistantMessage, fauxAssistantMessage, @@ -1733,11 +1734,26 @@ describe("compaction recovery", () => { chat.setup.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 60_000 }; chat.faux.summaries.push(failure("overloaded")); const id = await chat.root.compact(undefined, context); - await waitFor(async () => (await live(chat)).compactions?.[0]?.retry !== undefined); + await waitFor(async () => (await chat.harness.getTask(id, context))?.state.status === "sleeping"); + const until = now + 60_000; expect((await chat.harness.getTask(id, context))?.state).toMatchObject({ - checkpoint: { phase: "retry", attempt: 1 }, + status: "sleeping", + checkpoint: { phase: "retry", attempt: 1, until }, + until, }); await chat.harness.close(context); + // Earlier versions slept inside a running `retry` phase: reopened before its time, it sleeps again. + const database = new DatabaseSync(path); + database + .prepare( + "UPDATE tasks SET status = 'running', record = json_set(json_remove(record, '$.state.until'), '$.state.status', 'running') WHERE id = ?", + ) + .run(id); + database.close(); + chat = await reopen(path, chat.setup, chat.faux); + await waitFor(async () => (await chat.harness.getTask(id, context))?.state.status === "sleeping"); + expect(chat.faux.summaryRequests).toHaveLength(1); + await chat.harness.close(context); now += 120_000; chat = await reopen(path, chat.setup, chat.faux); chat.faux.summaries.push(summary()); diff --git a/packages/durable/test/harness-sleep.test.ts b/packages/durable/test/harness-sleep.test.ts index bd98ec2fe..790fa880f 100644 --- a/packages/durable/test/harness-sleep.test.ts +++ b/packages/durable/test/harness-sleep.test.ts @@ -1,7 +1,15 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { defineTask, MemoryStorage, type Storage, type TaskId, type Wake } from "@earendil-works/pi-durable"; +import { + defineExtension, + defineTask, + MemoryStorage, + type SetTimer, + type Storage, + type TaskId, + type Wake, +} from "@earendil-works/pi-durable"; import { afterEach, describe, expect, it } from "vitest"; import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; import { context, flush } from "./session-support.ts"; @@ -53,6 +61,22 @@ async function stateOf(harness: Awaited>["harness"] return (await harness.getTask(id, context))?.state; } +type FakeTimer = { readonly delayMs: number; readonly callback: () => void; disarmed: boolean }; + +/** A host `setTimer` that records the timers that pace work and declines background ones. */ +function fakeTimers() { + const armed: FakeTimer[] = []; + const setTimer: SetTimer = (callback, delayMs, options) => { + if (options.background) return undefined; + const timer: FakeTimer = { delayMs, callback, disarmed: false }; + armed.push(timer); + return () => { + timer.disarmed = true; + }; + }; + return { armed, setTimer, live: () => armed.filter((timer) => !timer.disarmed) }; +} + /** Let microtasks and a few timer turns run. */ async function settle(): Promise { await flush(); @@ -139,43 +163,68 @@ describe("sleeping tasks", () => { await harness.close(context); }); - it("resumes a due sleep with a timer when there is no onWake", async () => { - const Nap = napTask(30); - const { harness } = await openTasks(new MemoryStorage(), [Nap]); + it("resumes a due sleep with its timer without onWake, with background timers declined", async () => { + let clock = 1_000; + const timers = fakeTimers(); + const Nap = napTask(500); + const { harness } = await openTasks(new MemoryStorage(), [Nap], { now: () => clock, setTimer: timers.setTimer }); const id = await startNap(harness, Nap); - const started = Date.now(); - // Only this call: the timer wakes the task. harness.resume(); - const receipt = await harness.waitForTask(id, context); - expect(receipt.state.outcome.status).toBe("completed"); - expect(Date.now() - started).toBeGreaterThanOrEqual(25); + await eventually(() => timers.live().some((timer) => timer.delayMs === 500)); + expect((await stateOf(harness, id))?.status).toBe("sleeping"); + clock = 1_500; + timers + .live() + .find((timer) => timer.delayMs === 500)! + .callback(); + // Polled: `waitForTask()` would call `resume()`, which wakes due sleeps itself. + await eventually(async () => (await stateOf(harness, id))?.status === "terminal"); + expect(await stateOf(harness, id)).toMatchObject({ outcome: { status: "completed", result: 1_500 } }); await harness.close(context); }); - it("arms its timers through setTimer: a sleep's in the foreground, a declined background one is not needed", async () => { + it("runs a due sleep at any scheduling pass, before its timer fires", async () => { let clock = 1_000; - const armed: { readonly delayMs: number; readonly background: boolean; readonly callback: () => void }[] = []; + const timers = fakeTimers(); const Nap = napTask(500); - const { harness } = await openTasks(new MemoryStorage(), [Nap], { + const { harness, registry } = await openTasks(new MemoryStorage(), [Nap], { now: () => clock, - setTimer: (callback, delayMs, options) => { - armed.push({ delayMs, background: options.background, callback }); - // Declines background timers, as a host that must hibernate would. - return options.background ? undefined : () => {}; - }, + setTimer: timers.setTimer, }); const id = await startNap(harness, Nap); harness.resume(); - await eventually(() => armed.some((timer) => !timer.background && timer.delayMs === 500)); - expect((await stateOf(harness, id))?.status).toBe("sleeping"); + await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); + // The deadline passes, and an unrelated change starts a pass before the timer fires. clock = 1_500; - armed.find((timer) => !timer.background && timer.delayMs === 500)!.callback(); - expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "completed", result: 1_500 }); + registry.install(defineExtension({ name: "unrelated" })); + // Polled: `waitForTask()` would call `resume()`, which wakes due sleeps itself. + await eventually(async () => (await stateOf(harness, id))?.status === "terminal"); + expect(await stateOf(harness, id)).toMatchObject({ outcome: { status: "completed", result: 1_500 } }); await harness.close(context); }); - it("disarms an invocation's timers when it ends", async () => { - const disarmed: number[] = []; + it("disarms the sleep timer when the sleep is aborted and when the Harness closes", async () => { + const clock = 1_000; + const timers = fakeTimers(); + const Short = napTask(500); + const { harness } = await openTasks(new MemoryStorage(), [Short], { + now: () => clock, + setTimer: timers.setTimer, + }); + const aborted = await startNap(harness, Short); + harness.resume(); + await eventually(() => timers.live().some((timer) => timer.delayMs === 500)); + await harness.abortTask(aborted, context); + await harness.waitForTask(aborted, context); + await eventually(() => timers.live().length === 0); + const kept = await startNap(harness, Short); + await eventually(async () => (await stateOf(harness, kept))?.status === "sleeping"); + expect(timers.live().map((timer) => timer.delayMs)).toEqual([500]); + await harness.close(context); + expect(timers.live()).toEqual([]); + }); + + it("disarms an invocation's timers when it ends, and a late fire runs nothing", async () => { const fired: string[] = []; const Timed = defineTask({ name: "test.timed", @@ -189,13 +238,8 @@ describe("sleeping tasks", () => { }, abort: async () => {}, }); - let next = 0; - const { harness } = await openTasks(new MemoryStorage(), [Timed], { - setTimer: (_callback, delayMs) => { - const id = next++; - return delayMs === 60_000 ? () => disarmed.push(id) : () => {}; - }, - }); + const timers = fakeTimers(); + const { harness } = await openTasks(new MemoryStorage(), [Timed], { setTimer: timers.setTimer }); const root = await harness.root(context); const id = await root.commit( (tx) => tx.createTask(Timed, null, { ownership: { kind: "conversation" } }), @@ -203,24 +247,92 @@ describe("sleeping tasks", () => { ); harness.resume(); await harness.waitForTask(id, context); - await eventually(() => disarmed.length === 1); + const timer = timers.armed.find((armed) => armed.delayMs === 60_000)!; + await eventually(() => timer.disarmed); + // A host timer that fires anyway: the ended invocation's callback does not run. + timer.callback(); expect(fired).toEqual([]); await harness.close(context); }); - it("keeps no timer with onWake", async () => { - const Nap = napTask(10); - const { harness } = await openTasks(new MemoryStorage(), [Nap], { onWake: () => {} }); + it("arms no timer for sleeps with onWake", async () => { + let clock = 1_000; + const timers = fakeTimers(); + const Nap = napTask(500); + const { harness } = await openTasks(new MemoryStorage(), [Nap], { + now: () => clock, + setTimer: timers.setTimer, + onWake: () => {}, + }); const id = await startNap(harness, Nap); harness.resume(); - await new Promise((resolve) => setTimeout(resolve, 60)); - expect((await stateOf(harness, id))?.status).toBe("sleeping"); + await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); + await settle(); + expect(timers.armed).toEqual([]); + clock = 1_500; expect(harness.wake()).toEqual({ active: true }); harness.resume(); expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("completed"); await harness.close(context); }); + it("arms a foreground timer the host declines with setTimeout, reporting it once", async () => { + const Nap = napTask(10); + const { harness, reports } = await openTasks(new MemoryStorage(), [Nap], { setTimer: () => undefined }); + const first = await startNap(harness, Nap); + const second = await startNap(harness, Nap); + harness.resume(); + expect((await harness.waitForTask(first, context)).state.outcome.status).toBe("completed"); + expect((await harness.waitForTask(second, context)).state.outcome.status).toBe("completed"); + expect(reports.map((error) => (error as Error).message)).toEqual([ + "HarnessOptions.setTimer declined a timer that paces work; using setTimeout", + ]); + await harness.close(context); + }); + + it("counts neither a wait on a sleeping task nor an outcome held over one as active", async () => { + const clock = 1_000; + const Child = napTask(500); + type ParentState = { phase: "start" } | { phase: "joined"; child: TaskId }; + const Parent = defineTask<{ hold: boolean }, ParentState, null>({ + name: "test.parent", + version: 1, + initial: () => ({ phase: "start" }), + phases: { + start: async (task, runtime, ctx) => { + await runtime.commit(async (tx) => { + const child = await tx.createTask(Child, null, { ownership: { kind: "task", taskId: task.id } }); + // Held: completes next, holding its outcome over the sleeping child. Otherwise waits for it. + return task.input.hold + ? { status: "running", checkpoint: { phase: "joined", child } } + : { status: "waiting", checkpoint: { phase: "joined", child }, on: [child], policy: "allSettled" }; + }, ctx); + }, + joined: async (_task, runtime, ctx) => { + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + for (const hold of [false, true]) { + const { harness } = await openTasks(new MemoryStorage(), [Parent, Child], { + now: () => clock, + onWake: () => {}, + }); + const root = await harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Parent, { hold }, { ownership: { kind: "conversation" } }), + context, + ); + harness.resume(); + const parked = hold ? "completing" : "waiting"; + await eventually(async () => (await stateOf(harness, id))?.status === parked); + await eventually(() => harness.wake().at === 1_500); + expect(harness.wake()).toEqual({ active: false, at: 1_500 }); + await harness.close(context); + } + }); + it("runs the abort handler of a sleeping task without waiting for its time", async () => { const clock = 1_000; const Nap = napTask(60_000); @@ -288,7 +400,6 @@ describe("sleeping tasks", () => { }); it("does not count a task without a definition as active", async () => { - const Nap = napTask(10); const Missing = napTask(10); const { harness } = await openTasks(new MemoryStorage(), []); const root = await harness.root(context); @@ -298,33 +409,36 @@ describe("sleeping tasks", () => { harness.resume(); await settle(); expect(harness.wake()).toEqual({ active: false }); - void Nap; await harness.close(context); }); - it("reports a throwing onWake and keeps calling it for later changes", async () => { + it("reports a throwing onWake and offers the same wake again at the next pass", async () => { let clock = 1_000; const wakes: Wake[] = []; const failure = new Error("alarm failed"); + let failing = true; const Nap = napTask(1_000); - const { harness, reports } = await openTasks(new MemoryStorage(), [Nap], { + const { harness, registry, reports } = await openTasks(new MemoryStorage(), [Nap], { now: () => clock, onWake: (wake) => { wakes.push(wake); - if (wake.at !== undefined) throw failure; + if (failing && wake.at !== undefined) throw failure; }, }); const id = await startNap(harness, Nap); harness.resume(); - await eventually(async () => (await stateOf(harness, id))?.status === "sleeping"); - await settle(); - expect(reports.length).toBeGreaterThan(0); - expect(reports.every((error) => error === failure)).toBe(true); + await eventually(() => wakes.some((wake) => wake.at === 2_000 && !wake.active)); + expect(reports).toContain(failure); + // The host recovers; an unrelated change offers the same, undelivered wake again. + failing = false; + const offered = wakes.length; + registry.install(defineExtension({ name: "unrelated" })); + await eventually(() => wakes.length > offered); + expect(wakes.at(-1)).toEqual({ active: false, at: 2_000 }); clock = 2_000; harness.resume(); await harness.waitForTask(id, context); - await settle(); - expect(wakes.at(-1)).toEqual({ active: false }); + await eventually(() => wakes.at(-1)?.active === false && wakes.at(-1)?.at === undefined); await harness.close(context); }); });