diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 22f7a42f1..371f8b629 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -9,7 +9,6 @@ - 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 (a run's 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. - A task cut off by a crash 5 times in a row without committing a new state is parked instead of run again (see the crash-loop guard under Added); set `settings.restartLimit` to `Infinity` to keep the old behavior. `TaskInspection`'s blocked reasons include `restart_limit`. -- `onReport` gets a second argument, `info` (see below), and `TaskRuntime.report()` takes an optional `from`; an `onReport` written for one argument still works. ### Added @@ -26,6 +25,7 @@ ### Changed - A task phase that faults is reported through `onReport` (with `faulted: true`), not only recorded in its outcome. +- `onReport` gets a second argument, `info`, and `TaskRuntime.report()` takes an optional `from` (see Added); an `onReport` written for one argument still works. - Generation retry backoffs, deferred-response polls, and compaction retry backoffs sleep instead of waiting in a running phase, so a process can stop or hibernate during them. A record an earlier version stored mid-backoff sleeps again when it resumes before its time. A deferred poll whose model is gone after a reopen now fails `no_model` once the poll is due, not at once. - Code mode measures script deadlines and wall time on the Harness clock (`HarnessOptions.now`) instead of `Date.now()`. - SQLite drivers, portable interfaces, and the generic migration runner now come from `@earendil-works/chord/sqlite`. Existing Durable SQLite entry points keep re-exporting their public names; storage helpers and Durable schema migrations stay here. diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 84201b5d5..8a3a41d9c 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -5323,20 +5323,26 @@ dropped, since nothing is left to report it to. A throwing `now()` is reported once, and `Date.now` serves instead. **Every report says where it came from.** `onReport(error, info)` gets a -`ReportInfo`: `source` is `session` for the error that failed the Session (a -Storage error, often a platform reset, or the Session's own bookkeeping), -`task` for a task's phase that faulted (`faulted: true`), a task's -`runtime.report()`, a failed migration, or a task parked by the restart limit -(`blocked: "restart_limit"`), with `taskId`, `taskKind`, and `conversationId`; `extension` for hooks, -wrappers, system prompt sections, and a tool's invalid structured output; -`listener`; `host` for the clock, settings, registry, environment builder, and -model provider calls such as cancelling a deferred response; an `extension` -report names the extension (`extension`) when the failing code is known to -belong to one: the hook's, the wrapper's, or the last selected extension that -provides or wraps the section or tool. A `task` report names the extension that -installed the task's definition; built-in tasks have none; `timer` for `setTimer`; and `wake` for `onWake`. A host acts on them -outside the scheduler, for example counting faults of one task kind toward a -breaker. +`ReportInfo` whose `source` is one of: + +- `session`: the error that failed the Session (a Storage error, often a + platform reset, or the Session's own bookkeeping); +- `task`: a task's phase that faulted (`faulted: true`), a task's + `runtime.report()`, a failed migration, or a task parked or faulted by the + restart limit (`blocked: "restart_limit"`), with `taskId`, `taskKind`, and + `conversationId`; +- `extension`: hooks, wrappers, system prompt sections, and a tool's invalid + structured output; +- `listener`; `host` for the clock, settings, registry, environment builder, and + model provider calls such as cancelling a deferred response; `timer` for + `setTimer`; and `wake` for `onWake`. + +`extension` names the extension whose code failed, when known. A `task` report +names the extension that installed the definition the task ran, as of its +reservation; built-in tasks have none. An `extension` report names the hook's or +the wrapper's extension, or the last selected extension that provides or wraps +the section or tool. A host acts on reports outside the scheduler, for example +counting faults of one task kind toward a breaker. ## 11. Backends diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index ac52ebf22..816ca996b 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -96,6 +96,8 @@ type Fit = type Invocation = { readonly taskId: TaskId; readonly kind: string; + /** The extension that installed the definition this invocation runs, as of its reservation; none for a built-in. */ + readonly extension: string | undefined; readonly conversationId: ConversationId; readonly mode: "run" | "abort"; readonly controller: AbortController; @@ -194,8 +196,8 @@ export type TaskSchedulerOptions = { readonly onWake: ((wake: Wake) => void) | undefined; /** The host's timers (`HarnessOptions.setTimer`). */ readonly timers: Timers; - /** Write records straight to Storage while closing, after admission is sealed. */ - readonly writeAtClose: (writes: readonly StorageWrite[]) => Promise; + /** While closing, on the line after admitted commits settle, write what the callback returns straight to Storage. */ + readonly writeAtClose: (writes: () => readonly StorageWrite[]) => Promise; readonly report: (error: unknown, info: ReportInfo) => void; /** Harness cleanup staged in the commit that makes an outcome the scheduler wrote itself terminal. */ readonly settleOutcome: (tx: Transaction, record: AnyTaskRecord, outcome: SchedulerOutcome) => Promise; @@ -244,7 +246,7 @@ export class TaskScheduler { readonly #now: () => number; readonly #onWake: ((wake: Wake) => void) | undefined; readonly #timers: Timers; - readonly #writeAtClose: (writes: readonly StorageWrite[]) => Promise; + readonly #writeAtClose: (writes: () => readonly StorageWrite[]) => Promise; readonly #report: (error: unknown, info: ReportInfo) => void; readonly #settleOutcome: TaskSchedulerOptions["settleOutcome"]; readonly #withdrawInputs: TaskSchedulerOptions["withdrawInputs"]; @@ -313,8 +315,8 @@ export class TaskScheduler { /** Tasks a pass left out as parked; each pass checks them again, as `settings.restartLimit` may have risen. */ readonly #parkedTasks = new Set(); #reconcileScheduled = false; - /** Reconcile passes underway (they may overlap): their commits finalize held outcomes and mark cascades, recovery work. */ - #reconciling = 0; + /** Reconcile passes underway with work (they may overlap): their commits finalize held outcomes and mark cascades. */ + #reconcilingWork = 0; #sweepScheduled = false; #cascadePending = false; #unsubscribeRegistry: () => void = () => {}; @@ -410,7 +412,8 @@ export class TaskScheduler { await this.#session.commitWith(async (tx) => { const current = await tx.task(id); if (current === undefined) throw new Error(`Task ${id} does not exist`); - if (!this.#parked(current)) return; + const status = current.state.status; + if (status === "terminal" || status === "completing" || !this.#parked(current)) return; tx.setTask(progressed(current)); }, context); this.resume(); @@ -451,9 +454,9 @@ export class TaskScheduler { */ wake(): Wake { const now = this.#now(); - // A reconcile pass scheduled or underway is work too: it may finalize held outcomes and mark cascades, at open as - // recovery. A process that dies before it lands must be woken again. - let active = this.#invocations.size > 0 || this.#reconcileScheduled || this.#reconciling > 0; + // A reconcile pass with work to do is work too: it finalizes held outcomes and marks cascades, at open as + // recovery. A process that dies before it lands must be woken again. A pass with nothing to do is not. + let active = this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork(); let at: number | undefined; let snapshot: RegistrySnapshot | undefined; let snapshotFailed = false; @@ -496,15 +499,18 @@ export class TaskScheduler { const stopped = [...this.#invocations.values()]; for (const invocation of stopped) invocation.controller.abort(); await Promise.allSettled(stopped.map((invocation) => invocation.done)); - // A close is not a crash: a task it stopped is `pending` again, so the next open counts no restart for it. - const released = stopped.flatMap((invocation) => { - const record = this.#live.get(invocation.taskId); - if (record?.state.status !== "running") return []; - return [withState(record, { status: "pending", checkpoint: record.state.checkpoint })]; - }); - if (released.length === 0) return; - // A failure leaves them `running`, counted as if cut off; the Storage guard fails the Session and reports it. - await this.#writeAtClose(released.map((value) => ({ type: "task", value }) as const)).catch(() => {}); + if (stopped.length === 0) return; + // A close is not a crash: a task it stopped is `pending` again, so the next open counts no restart for it. Read + // on the line after commits admitted before the seal, such as an abort mark, so the write keeps them. A failure + // leaves them `running`, counted as if cut off; the Storage guard fails the Session and reports it. + await this.#writeAtClose(() => + stopped.flatMap((invocation) => { + const record = this.#live.get(invocation.taskId); + if (record?.state.status !== "running") return []; + const value = withState(record, { status: "pending", checkpoint: record.state.checkpoint }); + return [{ type: "task", value } as const]; + }), + ).catch(() => {}); } /** @@ -747,7 +753,8 @@ export class TaskScheduler { */ async #reconcile(): Promise { this.#reconcileScheduled = false; - this.#reconciling++; + const work = this.#reconcileWork(); + if (work) this.#reconcilingWork++; const cascade = this.#cascadePending; this.#cascadePending = false; const checks = [...this.#failFastChecks]; @@ -794,11 +801,16 @@ export class TaskScheduler { // running the pass again, so it fails the Session, which reports it. this.#failSession(error); } - this.#reconciling--; + if (work) this.#reconcilingWork--; this.#scheduleWake(); this.#settleIdle(); } + /** Whether a reconcile pass would have work: a cascade to derive, held outcomes to check, or `failFast` waits. */ + #reconcileWork(): boolean { + return this.#cascadePending || this.#finalizeChecks.size > 0 || this.#failFastChecks.size > 0; + } + /** Whether any of `ids` holds or ended with an outcome other than `completed`. */ async #anyFailed(ids: readonly TaskId[]): Promise { for (const id of ids) { @@ -1331,19 +1343,19 @@ export class TaskScheduler { readonly taskId: TaskId; readonly kind: string; readonly conversationId: ConversationId; + /** The installer of the definition whose code ran (an invocation's); absent: the current one's. */ + readonly extension?: string | undefined; }): ReportInfo { const info = taskInfo(task); let extension: string | undefined; - try { - const snapshot = this.#registry.snapshot(); - const definition = snapshot.task(task.kind); - if (definition !== undefined) { - extension = snapshot - .installed() - .findLast((installed) => installed.tasks?.includes(definition) === true)?.name; + if ("extension" in task) extension = task.extension; + else { + try { + const snapshot = this.#registry.snapshot(); + extension = installerOf(snapshot, snapshot.task(task.kind)); + } catch { + // A registry that throws is reported where it is read for scheduling; a report goes out without the name. } - } catch { - // A registry that throws is reported where it is read for scheduling; a report goes out without the name. } return extension === undefined ? info : { ...info, extension }; } @@ -1504,7 +1516,7 @@ export class TaskScheduler { // Registered on the line, so marks and later reservations see it and close joins it; `#end` makes it a // candidate again. this.#runnable.delete(id); - const invocation = this.#createInvocation(record, mode); + const invocation = this.#createInvocation(record, mode, installerOf(snapshot, resolution.task)); reservations.push({ invocation, task: resolution.task, snapshot }); } }, this.#context); @@ -1559,9 +1571,10 @@ export class TaskScheduler { // Blocked now without a commit: the graph and the host's wake must hear it. queueMicrotask(() => this.#notifyBlocked()); this.#scheduleWake(); + const extension = installerOf(snapshot, fit.task); this.#report( error, - this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }), + this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId, extension }), ); return { kind: "blocked", reason: "migration_failed" }; } @@ -1624,12 +1637,13 @@ export class TaskScheduler { return { kind: "ready", migrates: fit.migrates }; } - #createInvocation(record: AnyTaskRecord, mode: "run" | "abort"): Invocation { + #createInvocation(record: AnyTaskRecord, mode: "run" | "abort", extension: string | undefined): Invocation { const controller = new AbortController(); const { promise: done, resolve: finish } = Promise.withResolvers(); const invocation: Invocation = { taskId: record.id, kind: record.kind, + extension, conversationId: record.conversationId, mode, controller, @@ -1668,7 +1682,9 @@ export class TaskScheduler { const runtime = this.#runtime(invocation, phase); let previous: PhaseResult | undefined; for (;;) { - const current = await this.#step(invocation, (tx, current) => this.#decide(tx, current, previous, state)); + const current = await this.#step(invocation, (tx, current) => + this.#decide(tx, current, previous, state, invocation.extension), + ); // Close may seal between the decision and dispatch. if (current === undefined || this.#closing) return; const checkpoint = current.state.checkpoint; @@ -1692,6 +1708,7 @@ export class TaskScheduler { current: ErasedRunningTask, previous: PhaseResult | undefined, state: { task: AnyTask; snapshot: RegistrySnapshot; reported: ReportedTask }, + extension: string | undefined, ): Decision { // 3. abort mark: end; a fresh abort invocation starts once the task's ordinary owned work is gone. if (current.abortRequested) return false; @@ -1722,7 +1739,12 @@ export class TaskScheduler { const cause = next === undefined ? "missing_task" : "incompatible_task"; this.#report( new Error(`Task ${current.id} keeps running under its old ${current.kind} definition`, { cause }), - this.#taskInfo({ taskId: current.id, kind: current.kind, conversationId: current.conversationId }), + this.#taskInfo({ + taskId: current.id, + kind: current.kind, + conversationId: current.conversationId, + extension, + }), ); } } @@ -2207,6 +2229,12 @@ export class TaskScheduler { const HOST: ReportInfo = { source: "host" }; +/** The extension that installed `task` in `snapshot`, the last one that lists it; none for a built-in task. */ +function installerOf(snapshot: RegistrySnapshot, task: AnyTask | undefined): string | undefined { + if (task === undefined) return undefined; + return snapshot.installed().findLast((installed) => installed.tasks?.includes(task) === true)?.name; +} + /** The report info of a failure of a task. */ function taskInfo(task: { readonly taskId: TaskId; @@ -2300,10 +2328,10 @@ function progressed(record: AnyTaskRecord): AnyTaskRecord { return rest as AnyTaskRecord; } -/** Replace a live record's state; memos disappear once an outcome is decided. */ +/** Replace a live record's state; memos and the restart count disappear once an outcome is decided. */ function withState(record: AnyTaskRecord, state: TaskState): AnyTaskRecord { if (state.status !== "terminal" && state.status !== "completing") return { ...record, state } as AnyTaskRecord; - const { memos: _memos, ...rest } = record; + const { memos: _memos, restarts: _restarts, ...rest } = record; return { ...rest, state }; } diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index e5c670aae..b9ad627d4 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -561,7 +561,11 @@ async function finalResult( } else { // The model never sees structured output, so its call stands; the host learns of the broken tool. final = rest; - const extension = toolExtension(await runtime.agent(context), tool.name); + // Only names the extension: a failed resolution leaves the report without the name. + const extension = await runtime.agent(context).then( + (agent) => toolExtension(agent, tool.name), + () => undefined, + ); runtime.report(new Error(broken), { source: "extension", ...(extension === undefined ? {} : { extension }) }); } } diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index eb3d3b3d7..10593c818 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -552,14 +552,15 @@ export class SessionImpl implements Session { } /** - * Internal, for `beforeClose()`: write records straight to Storage, on the line, after admission is sealed. Nothing - * is published, since the Session is closing; a failed Session writes nothing. + * Internal, for `beforeClose()`: on the line, after every commit admitted before the seal has settled, write what + * `writes()` returns then straight to Storage. Nothing is published, since the Session is closing; a failed Session + * writes nothing. */ - protected writeAtClose(writes: readonly StorageWrite[]): Promise { - if (writes.length === 0) return Promise.resolve(); + protected writeAtClose(writes: () => readonly StorageWrite[]): Promise { return this.#enqueue(async () => { if (this.#failure !== undefined) return; - await this.#storage.commit(writes, BACKGROUND_CONTEXT); + const batch = writes(); + if (batch.length > 0) await this.#storage.commit(batch, BACKGROUND_CONTEXT); }); } diff --git a/packages/durable/test/harness-blocked.test.ts b/packages/durable/test/harness-blocked.test.ts index 537b65677..7131cf8d2 100644 --- a/packages/durable/test/harness-blocked.test.ts +++ b/packages/durable/test/harness-blocked.test.ts @@ -6,17 +6,23 @@ import { defineExtension, defineTask, type Registry, + 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"; -import { completed, eventually, openTasks } from "./task-support.ts"; +import { completed, deferred, eventually, openTasks } from "./task-support.ts"; const directories = new Set(); +/** Storages of killed Harnesses, never closed by them: closed after each test. */ +const killed = new Set(); + afterEach(async () => { + for (const storage of killed) await storage.close(context).catch(() => {}); + killed.clear(); for (const directory of directories) await rm(directory, { recursive: true, force: true }); directories.clear(); }); @@ -67,6 +73,7 @@ async function openWith(path: string, registry: Registry, settings: { restartLim }); const kill = (): void => { dead = true; + killed.add(inner); }; return { harness, wakes, reports, kill }; } @@ -134,7 +141,7 @@ describe("tasks that cannot run", () => { let opened = await openWith(path, registry, { restartLimit: 3 }); const root = await opened.harness.root(context); const id = await root.commit((tx) => tx.createTask(Loop, null, { ownership: { kind: "conversation" } }), context); - // Each open runs the phase, and each close cuts it off: three restarts. + // Each open runs the phase, and each crash cuts it off: three restarts. for (let run = 1; run <= 3; run++) { opened.harness.resume(); await eventually(() => state.started === run); @@ -189,6 +196,33 @@ describe("tasks that cannot run", () => { await opened.harness.close(context); }); + it("keeps an abort mark admitted before close in the record close writes back", async () => { + const path = await sqlitePath(); + const state = { crashing: true, started: 0 }; + const registry = createRegistry(); + const Loop = crashLoop(state); + registry.install(defineExtension({ name: "loop", tasks: [Loop] })); + const { harness } = await openWith(path, registry); + const root = await harness.root(context); + const id = await root.commit((tx) => tx.createTask(Loop, null, { ownership: { kind: "conversation" } }), context); + harness.resume(); + await eventually(() => state.started === 1); + // Hold the line; the abort mark is admitted behind it, then close begins. + const gate = deferred(); + const holding = root.commit(() => gate.promise, context); + const aborting = harness.abortTask(id, context); + const closing = harness.close(context); + // The invocation ends while the line is held; its write-back waits behind the abort mark. + await flush(); + gate.resolve(); + await holding; + expect(await aborting).toBe("marked"); + await closing; + const raw = await openNodeSqliteStorage(path); + expect(await raw.task(id, context)).toMatchObject({ abortRequested: true, state: { status: "pending" } }); + await raw.close(context); + }); + it("faults a task whose abort handler is cut off restartLimit times", async () => { const path = await sqlitePath(); let aborts = 0; @@ -222,6 +256,8 @@ describe("tasks that cannot run", () => { status: "faulted", error: { detail: { reason: "restart_limit" } }, }); + // The receipt keeps no restart count. + expect(settled.restarts).toBeUndefined(); expect(aborts).toBe(1); await eventually(() => opened.reports.length > 0); expect(opened.reports.map((error) => (error as Error).cause)).toEqual(["restart_limit"]); diff --git a/packages/durable/test/harness-reports.test.ts b/packages/durable/test/harness-reports.test.ts index f8a37d558..a575229a0 100644 --- a/packages/durable/test/harness-reports.test.ts +++ b/packages/durable/test/harness-reports.test.ts @@ -17,7 +17,7 @@ import { import { describe, expect, it } from "vitest"; import { chatSetup } from "./chat-support.ts"; import { ControlledStorage, context } from "./session-support.ts"; -import { completed, eventually } from "./task-support.ts"; +import { completed, deferred, eventually } from "./task-support.ts"; type Report = { readonly message: string; readonly info: ReportInfo }; @@ -75,6 +75,43 @@ describe("reports", () => { await harness.close(context); }); + it("name the extension whose definition ran, also after another one replaced it", async () => { + const gate = deferred(); + const started = deferred(); + const definition = (name: string) => + defineTask({ + name: "test.replaced", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async () => { + started.resolve(); + await gate.promise; + throw new Error(`broken in ${name}`); + }, + }, + abort: async () => {}, + }); + const Old = definition("old"); + const { harness, registry, reports } = await open([]); + const old = defineExtension({ name: "old-extension", tasks: [Old] }); + registry.install(old); + const root = await harness.root(context); + const id = await root.commit((tx) => tx.createTask(Old, null, { ownership: { kind: "conversation" } }), context); + harness.resume(); + await started.promise; + registry.uninstall(old); + registry.install(defineExtension({ name: "new-extension", tasks: [definition("new")] })); + gate.resolve(); + await harness.waitForTask(id, context); + await eventually(() => reports.some((report) => report.info.faulted === true)); + expect(reports.find((report) => report.info.faulted === true)).toMatchObject({ + message: "broken in old", + info: { extension: "old-extension" }, + }); + await harness.close(context); + }); + it("name the extension of a failing section or hook, and tell them from a failing environment builder", async () => { const setup = chatSetup(); setup.faux.setResponses([fauxAssistantMessage("answer")]);