From 91b33cab3b5cae4d3f612664db53e155639db662 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sun, 18 Oct 2026 01:58:07 +0000 Subject: [PATCH] feat(durable): ReportInfo says what happened, with typed reasons onReport info is tagged by type (session_failed, task_faulted, task_parked, migration_failed, task_outdated, task_error, extension_error, host_error, listener_error). Task reports carry a ReportTask; restart-limit and outdated definition reasons are typed fields instead of error.cause strings; host_error names the failing HarnessOptions callback. --- packages/durable/CHANGELOG.md | 4 +- packages/durable/README.md | 4 +- packages/durable/docs/spec.md | 92 +++++++++++------- packages/durable/src/harness/generation.ts | 12 +-- packages/durable/src/harness/harness.ts | 8 +- packages/durable/src/harness/scheduler.ts | 89 +++++++++--------- packages/durable/src/harness/tool.ts | 3 +- packages/durable/src/harness/types.ts | 6 +- packages/durable/src/harness/util.ts | 5 + packages/durable/src/index.ts | 3 +- packages/durable/src/session/session.ts | 8 +- packages/durable/src/types.ts | 94 ++++++++++++------- packages/durable/test/harness-blocked.test.ts | 14 +-- packages/durable/test/harness-reports.test.ts | 32 +++---- .../test/harness-tasks-recovery.test.ts | 12 ++- packages/durable/test/task-support.ts | 18 +++- 16 files changed, 236 insertions(+), 168 deletions(-) diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 371f8b629..6645eb0dd 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -17,14 +17,14 @@ - `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()`. - Crash-loop guard: open counts the restarts of each task that a crash, eviction, or reset cut off (`TaskRecord.restarts`); a close writes the tasks it stopped back as `pending` uncounted, and a new state the task commits starts the count over. At `settings.restartLimit` the task is parked: blocked with reason `restart_limit` in `inspect()` and the task graph, not active in `wake()`, and reported once. `Harness.retryTask(id)` runs it again. An abort handler cut off that many times ends its task `faulted`. -- `onReport(error, info)`: `info` (`ReportInfo`) says where a failure came from, `session` for the error that failed the Session, `task` with the task's ID, kind, conversation, and installing extension (`faulted` for a fault, `blocked` for the restart limit), `extension` with the extension's name, or `listener`, `host`, `timer`, or `wake`. +- `onReport(error, info)`: `info` (`ReportInfo`) says what happened in `type`: `session_failed`; `task_faulted`, `task_parked`, `migration_failed`, `task_outdated`, or `task_error` with the task and the extension that installed it; `extension_error` with the extension's name; `host_error` with the failing `HostCallback`; or `listener_error`. - `settings.afterFailedRun`: a run that fails with `model_error` or `no_model` keeps the inputs queued behind it (`keep`, the default), starts the next run with them (`continue`), or settles them `unanswered` with reason `run_failed` (`withdraw`). - The task graph shows why a task cannot run (`TaskGraphNode.blocked`: `missing_task`, `task_too_old`, `migration_failed`, or `restart_limit`) and follows registry changes and failed migrations. - `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. ### Changed -- A task phase that faults is reported through `onReport` (with `faulted: true`), not only recorded in its outcome. +- A task phase that faults is reported through `onReport` (as `task_faulted`), 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()`. diff --git a/packages/durable/README.md b/packages/durable/README.md index 61f8bb388..fea9629de 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -201,9 +201,9 @@ The default `setTimer` declines the timers that only free memory in Workers, so ### Failures and Recovery -**Crash loops.** Work that deterministically crashes its process, by exhausting memory or CPU before its next checkpoint, would otherwise be resumed and crash again forever. Open counts each task a crash, eviction, or reset cut off; a close is not counted, and a new state the task commits starts the count over. Once the count reaches `settings.restartLimit` (default 5; `Infinity` never parks), the task is parked: `inspect()` and the task graph show it as blocked with reason `restart_limit`, `wake()` does not count it as active, and `onReport` gets it once. Every task running when a process dies is counted, so show parked tasks and offer `harness.retryTask(id)`, which runs one again. `abortTask()` still runs its abort handler; an abort handler cut off that many times ends the task `faulted`. +**Crash loops.** Work that deterministically crashes its process, by exhausting memory or CPU before its next checkpoint, would otherwise be resumed and crash again forever. Open counts each task a crash, eviction, or reset cut off; a close is not counted, and a new state the task commits starts the count over. Once the count reaches `settings.restartLimit` (default 5; `Infinity` never parks), the task is parked: `inspect()` and the task graph show it as blocked with reason `restart_limit`, `wake()` does not count it as active, and `onReport` gets it once as `task_parked`. Every task running when a process dies is counted, so show parked tasks and offer `harness.retryTask(id)`, which runs one again. `abortTask()` still runs its abort handler; an abort handler cut off that many times ends the task `faulted`. -**Reports.** `onReport(error, info)` says where a failure came from: `info.source` is `session` for the error that failed the Harness (a Storage error, often a platform reset), `task` for a task that faulted (`faulted: true`), reported a failure, or was parked (`blocked: "restart_limit"`), with `taskId`, `taskKind`, `conversationId`, and the name of the extension that installed the task in `info.extension`, `extension` for hooks, sections, wrappers, and tools, again with the extension's name, and `listener`, `host`, `timer`, or `wake` for the rest. A host can feed task faults of one kind to its own breaker. +**Reports.** `onReport(error, info)` says what happened in `info.type`: `session_failed` for the error that failed the Harness (a Storage error, often a platform reset); `task_faulted`, `task_parked`, `migration_failed`, `task_outdated`, or `task_error` for a task, with `info.task` (its `id`, `kind`, and `conversationId`) and the extension that installed it in `info.extension`; `extension_error` for hooks, sections, wrappers, and tools, with the extension's name; `host_error` for a host callback, named in `info.callback`; and `listener_error`. A host can feed `task_faulted` reports of one kind to its own breaker. **Queued inputs after a failed run.** A run that fails with a model error leaves the inputs queued behind it waiting until the next input arrives. `settings.afterFailedRun` changes that: `continue` starts the next run with them, as after an answer, and `withdraw` settles them `unanswered` with reason `run_failed`. diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 8a3a41d9c..eb957dbea 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -596,16 +596,34 @@ type TaskInspection = { /** When the Session next needs a running process (section 5.6). */ type Wake = { readonly active: boolean; readonly at?: number }; -/** Where a report came from (section 10.1). */ -type ReportInfo = { - readonly source: "session" | "task" | "extension" | "listener" | "host" | "timer" | "wake"; - readonly taskId?: TaskId; - readonly taskKind?: string; - readonly conversationId?: ConversationId; - readonly extension?: string; - readonly faulted?: true; - readonly blocked?: "restart_limit"; -}; +/** What a report is about (section 10.1). */ +type ReportTask = { readonly id: TaskId; readonly kind: string; readonly conversationId: ConversationId }; +type HostCallback = "now" | "settings" | "registry" | "env" | "models" | "setTimer" | "onWake"; +type ReportInfo = + | { readonly type: "session_failed" } + | { + readonly type: "task_faulted"; + readonly task: ReportTask; + readonly extension?: string; + readonly reason: "threw" | "restart_limit"; + } + | { readonly type: "task_parked"; readonly task: ReportTask; readonly extension?: string } + | { readonly type: "migration_failed"; readonly task: ReportTask; readonly extension?: string } + | { + readonly type: "task_outdated"; + readonly task: ReportTask; + readonly extension?: string; + readonly reason: "missing_task" | "incompatible_task"; + } + | { readonly type: "task_error"; readonly task: ReportTask; readonly extension?: string } + | { + readonly type: "extension_error"; + readonly extension?: string; + readonly task?: ReportTask; + readonly conversationId?: ConversationId; + } + | { readonly type: "host_error"; readonly callback: HostCallback; readonly task?: ReportTask } + | { readonly type: "listener_error" }; type HarnessInspection = { readonly scheduling: "paused" | "running" | "closing"; @@ -1848,10 +1866,10 @@ interface TaskRuntime extends DocumentObserver, Docum /** The Harness clock. */ now(): number; /** Forward a non-fatal failure to `HarnessOptions.onReport`. */ - /** `from`: whose code failed, the task's own (default), an extension's (with its name), or a `host` callback's. */ + /** A `task_error`, or with `from` an `extension_error` or `host_error`, with this task (section 10.1). */ report( error: unknown, - from?: { source: "task" } | { source: "extension"; extension?: string } | { source: "host" }, + from?: { type: "extension"; extension?: string } | { type: "host"; callback: HostCallback }, ): void; /** Commit `{ status: "sleeping", until, checkpoint }`; resolves once stored (section 5.6). */ sleepUntil(until: number, checkpoint: S, context: Context): Promise; @@ -2172,7 +2190,7 @@ itself that differs from its current one (a new checkpoint, a wait, a sleep, an outcome) is progress and removes the count; the same checkpoint committed again, memos, and document writes are not. Once `restarts` reaches `settings.restartLimit` (default 5; `Infinity` never parks), the task is parked: -blocked with reason `restart_limit`, reported once through `onReport` at the open +blocked with reason `restart_limit`, reported once as `task_parked` at the open that parks it, shown in `inspect()` and the task graph, and not active in `wake()`. Work that deterministically exhausts memory or CPU before its next checkpoint thus stops after a few attempts instead of running forever. Every @@ -2183,13 +2201,13 @@ task again, and does nothing for a task that is not parked. A run task that is parked keeps its conversation busy. An abort mark starts the count over and still runs a parked task's abort handler; an abort handler cut off `restartLimit` times is not run again, and the task ends `faulted` with detail -`{ reason: "restart_limit" }`, reported with `blocked: "restart_limit"`. +`{ reason: "restart_limit" }`, reported as `task_faulted` with reason `restart_limit`. An abort invocation resolves the definition the same way. When the definition can take the task, its abort handler runs; otherwise the task is orphaned as described below. -A failed migration is reported once through `onReport` and retried only after +A failed migration is reported once as `migration_failed` and retried only after the registry resolves a different definition object for the task's kind. The Harness never terminalizes a task merely because registry code is missing or @@ -2276,7 +2294,7 @@ over: the step commits the task back to `pending` with its checkpoint, memos, and abort mark and ends the invocation, and the next reservation starts a fresh invocation under the new definition, applying the reservation rules above. When the definition is missing or cannot reserve the task, the invocation keeps running under its old -definition, reports once through `onReport` per resolved definition, and +definition, reports once as `task_outdated` per resolved definition, and reconsiders at its next boundary. A handler that never settles never hands over. @@ -5322,26 +5340,34 @@ listener throws ends with `listener_error`. A throwing or rejecting `onReport` i 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` whose `source` is one of: +**Every report says what happened.** `onReport(error, info)` gets a +`ReportInfo` whose `type` is one of: -- `session`: the error that failed the Session (a Storage error, often a +- `session_failed`: 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`. +- `task_faulted`: a task ended `faulted`, because its phase threw (reason + `threw`) or its abort handler was cut off `restartLimit` times (reason + `restart_limit`); +- `task_parked`: the restart limit parked a task (section 5.4); +- `migration_failed`: a task's migration threw, and the task is blocked; +- `task_outdated`: a running task keeps its old definition, which is missing + (`missing_task`) or cannot take the task (`incompatible_task`); +- `task_error`: a task's own `runtime.report()`, or a failure running it; +- `extension_error`: hooks, wrappers, system prompt sections, and a tool's + invalid structured output, with the task that ran them, or for a wrapper + outside a task the conversation whose agent was resolved; +- `host_error`: a host callback, by its `HarnessOptions` name: `now`, + `settings`, `registry`, `env`, `models` for provider calls such as cancelling + a deferred response, `setTimer`, or `onWake`; with the task that called it, + if one did; +- `listener_error`: a commit, close, watch, view, or document listener. -`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 +Every task report carries `task` (its ID, kind, and conversation). `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_error` 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/generation.ts b/packages/durable/src/harness/generation.ts index b7bc86933..832c84721 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -48,6 +48,7 @@ import type { UserInput, } from "./types.ts"; import { recordUsage } from "./usage.ts"; +import { optionalExtension } from "./util.ts"; export type GenerationInput = Record; @@ -157,7 +158,7 @@ export const GenerationTask = defineTask runtime.report(error, { - source: "extension", + type: "extension", ...optionalExtension(sectionExtension(agent, section.key)), }), context, @@ -283,7 +284,7 @@ export const GenerationTask = defineTask, from: TaskId, to: TaskId): void { if (live.run?.taskId === from) live.run.taskId = to; } - -/** `{ extension }` when the extension is known. */ -function optionalExtension(extension: string | undefined): { readonly extension?: string } { - return extension === undefined ? {} : { extension }; -} diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 4d0345c19..ea4762572 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -212,7 +212,7 @@ class HarnessImpl extends SessionImpl implements env: (id, callContext) => this.buildEnv(id, callContext), now, onWake: options.onWake, - timers: hostTimers(options.setTimer, (error) => report(error, { source: "timer" })), + timers: hostTimers(options.setTimer, (error) => report(error, { type: "host_error", callback: "setTimer" })), writeAtClose: (writes) => this.writeAtClose(writes), report: this.#report, settleOutcome: settleSchedulerOutcome, @@ -248,7 +248,7 @@ class HarnessImpl extends SessionImpl implements const registry = snapshot ?? this.#options.registry.snapshot(); const state = await this.snapshot(AgentDoc, id, context); return resolveAgent(state, registry, resolveSettings(this.#hostSettings()), (error, extension) => - this.#report(error, { source: "extension", conversationId: id, extension }), + this.#report(error, { type: "extension_error", conversationId: id, extension }), ); } @@ -365,7 +365,7 @@ class HarnessImpl extends SessionImpl implements return this.#create({ kind: "independent", ownership: options.ownership }, options, context); } - override report(error: unknown, info: ReportInfo = { source: "listener" }): void { + override report(error: unknown, info: ReportInfo = { type: "listener_error" }): void { this.#report(error, info); } @@ -436,7 +436,7 @@ export function safeNow( // Set first: a report handler that reads the clock again must not report again. if (!reported) { reported = true; - report(error, { source: "host" }); + report(error, { type: "host_error", callback: "now" }); } return Date.now(); } diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 816ca996b..f41557584 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -16,6 +16,7 @@ import type { JsonObject, NextTaskState, ReportInfo, + ReportTask, RunningTask, Storage, StorageWrite, @@ -40,7 +41,7 @@ import type { TaskInspection, Wake, } from "./types.ts"; -import { closedError, scanAll, Waiters } from "./util.ts"; +import { closedError, optionalExtension, scanAll, Waiters } from "./util.ts"; type AnyTaskRecord = TaskRecord; /** A record that can still run code: pending, running, waiting, or sleeping. */ @@ -383,10 +384,8 @@ export class TaskScheduler { const restarts = (record.restarts ?? 0) + 1; const error = new Error( `Task ${record.id} (${record.kind}) was cut off ${restarts} times without progress; it is parked until retryTask()`, - { cause: "restart_limit" }, ); - const info = this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }); - this.#report(error, { ...info, blocked: "restart_limit" }); + this.#report(error, { type: "task_parked", ...this.#aboutTask(recordTask(record)) }); } // Derive abort marks a crash left unapplied below cancelled owners, and finalize held outcomes. this.#cascadePending = true; @@ -433,7 +432,7 @@ export class TaskScheduler { try { snapshot = this.#registry.snapshot(); } catch (error) { - if (!this.#snapshotReported) this.#report(error, HOST); + if (!this.#snapshotReported) this.#report(error, REGISTRY); this.#snapshotReported = true; return undefined; } @@ -480,7 +479,7 @@ export class TaskScheduler { snapshot = this.#registry.snapshot(); } catch (error) { snapshotFailed = true; - this.#report(error, HOST); + this.#report(error, REGISTRY); } } if (snapshot !== undefined && "reason" in this.#fit(record, snapshot.task(record.kind))) continue; @@ -687,7 +686,7 @@ export class TaskScheduler { try { return this.#settings().contextRetentionMs; } catch (error) { - this.#report(error, HOST); + this.#report(error, SETTINGS); return 0; } } @@ -1268,7 +1267,7 @@ export class TaskScheduler { try { this.#unsubscribeRegistry(); } catch (error) { - this.#report(error, HOST); + this.#report(error, REGISTRY); } const error = closedError(this.#session); this.#taskWaiters.rejectAll(error); @@ -1299,7 +1298,7 @@ export class TaskScheduler { // Only a delivered wake counts: after a throw, the next change or kick offers it again. this.#lastWake = wake; } catch (error) { - this.#report(error, { source: "wake" }); + this.#report(error, ON_WAKE); } }); } @@ -1339,14 +1338,14 @@ export class TaskScheduler { * The report info of a task's failure, with the extension that installed its definition: the code that failed is * that extension's. Built-in tasks have none. */ - #taskInfo(task: { + /** What a report about `task`'s own code says of it: the task, and the extension that installed its definition. */ + #aboutTask(task: { 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); + }): { readonly task: ReportTask; readonly extension?: string } { let extension: string | undefined; if ("extension" in task) extension = task.extension; else { @@ -1357,7 +1356,7 @@ export class TaskScheduler { // 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 }; + return { task: reportTask(task), ...optionalExtension(extension) }; } /** Make parked tasks that a raised `settings.restartLimit` no longer parks candidates again, and tell the graph. */ @@ -1379,7 +1378,7 @@ export class TaskScheduler { try { return this.#settings().restartLimit; } catch (error) { - this.#report(error, HOST); + this.#report(error, SETTINGS); return DEFAULT_RESTART_LIMIT; } } @@ -1389,7 +1388,7 @@ export class TaskScheduler { try { listener(); } catch (error) { - this.#report(error, { source: "listener" }); + this.#report(error, { type: "listener_error" }); } } } @@ -1529,11 +1528,10 @@ export class TaskScheduler { throw error; } for (const { record, message } of faulted) { - const info = this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }); - this.#report(new Error(message, { cause: "restart_limit" }), { - ...info, - faulted: true, - blocked: "restart_limit", + this.#report(new Error(message), { + type: "task_faulted", + ...this.#aboutTask(recordTask(record)), + reason: "restart_limit", }); } // Marked, or already marked or gone: done with them. Reserve again, now seeing the marks. @@ -1572,10 +1570,7 @@ export class TaskScheduler { 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, extension }), - ); + this.#report(error, { type: "migration_failed", ...this.#aboutTask({ ...recordTask(record), extension }) }); return { kind: "blocked", reason: "migration_failed" }; } } @@ -1665,7 +1660,7 @@ export class TaskScheduler { if (invocation.mode === "run") await this.#run(reservation); else await this.#runAbort(reservation); } catch (error) { - this.#report(error, this.#taskInfo(invocation)); + this.#report(error, { type: "task_error", ...this.#aboutTask(invocation) }); } finally { this.#end(invocation); invocation.finish(); @@ -1736,16 +1731,11 @@ export class TaskScheduler { } if (state.reported === undefined || state.reported.task !== next) { state.reported = { task: next }; - 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, - extension, - }), - ); + this.#report(new Error(`Task ${current.id} keeps running under its old ${current.kind} definition`), { + type: "task_outdated", + ...this.#aboutTask({ ...recordTask(current), extension }), + reason: next === undefined ? "missing_task" : "incompatible_task", + }); } } return true; @@ -1794,7 +1784,8 @@ export class TaskScheduler { return undefined; }, this.#context); // Reported once the fault is durable: a host may count it, for example toward a breaker. - if (fault !== undefined) this.#report(fault.error, { ...this.#taskInfo(invocation), faulted: true }); + if (fault !== undefined) + this.#report(fault.error, { type: "task_faulted", ...this.#aboutTask(invocation), reason: "threw" }); return continued; } catch (error) { // The step writes the scheduler's own decision, a fault or a terminal record; a throw there would otherwise @@ -1987,7 +1978,7 @@ export class TaskScheduler { await invoke(handler.bind(handlers)); } catch (error) { if (invocation.controller.signal.aborted) throw error; - this.#report(error, { ...taskInfo(invocation), source: "extension", extension }); + this.#report(error, { type: "extension_error", extension, task: reportTask(invocation) }); } } }, @@ -2171,12 +2162,17 @@ export class TaskScheduler { disarm(); }; }, - report: (error, from = { source: "task" }) => { + report: (error, from) => { if (invocation.ended) throw endedError(invocation); // The task's own failure names the task's extension; any other names whose code failed, if anyone's. + const task = reportTask(invocation); this.#report( error, - from.source === "task" ? this.#taskInfo(invocation) : { ...taskInfo(invocation), ...from }, + from === undefined + ? { type: "task_error", ...this.#aboutTask(invocation) } + : from.type === "extension" + ? { type: "extension_error", task, ...optionalExtension(from.extension) } + : { type: "host_error", callback: from.callback, task }, ); }, }; @@ -2227,7 +2223,9 @@ export class TaskScheduler { } } -const HOST: ReportInfo = { source: "host" }; +const REGISTRY: ReportInfo = { type: "host_error", callback: "registry" }; +const SETTINGS: ReportInfo = { type: "host_error", callback: "settings" }; +const ON_WAKE: ReportInfo = { type: "host_error", callback: "onWake" }; /** 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 { @@ -2236,12 +2234,17 @@ function installerOf(snapshot: RegistrySnapshot, task: AnyTask | undefined): str } /** The report info of a failure of a task. */ -function taskInfo(task: { +function reportTask(task: { readonly taskId: TaskId; readonly kind: string; readonly conversationId: ConversationId; -}): ReportInfo { - return { source: "task", taskId: task.taskId, taskKind: task.kind, conversationId: task.conversationId }; +}): ReportTask { + return { id: task.taskId, kind: task.kind, conversationId: task.conversationId }; +} + +/** A stored task's identity, as an invocation names it. */ +function recordTask(record: Pick) { + return { taskId: record.id, kind: record.kind, conversationId: record.conversationId }; } /** diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index b9ad627d4..709a8c180 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -53,6 +53,7 @@ import type { ToolRegistration, } from "./types.ts"; import { recordUsage } from "./usage.ts"; +import { optionalExtension } from "./util.ts"; /** * A model-issued call, read from its assistant entry, or a nested call a running tool made through `executeTool()`, @@ -566,7 +567,7 @@ async function finalResult( (agent) => toolExtension(agent, tool.name), () => undefined, ); - runtime.report(new Error(broken), { source: "extension", ...(extension === undefined ? {} : { extension }) }); + runtime.report(new Error(broken), { type: "extension", ...optionalExtension(extension) }); } } // Programs get the tool's own value: without a schema, its output. The model reads only `output`, so model-issued diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index a15c34474..cca3ef601 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -645,9 +645,9 @@ export type HarnessOptions = { readonly onWake?: (wake: Wake) => void; /** * Receives failures that do not fail the calling operation, such as a throwing hook or listener, a task that faulted, - * and the error that fails the Session, once. `info` says where it came from: a host can tell a failed Session - * (`source: "session"`, often a platform reset) from a faulted task (`source: "task"`, `faulted`). Its own throw or - * rejection is dropped. + * and the error that fails the Session, once. `info` says what happened: a host can tell a failed Session + * (`session_failed`, often a platform reset) from a faulted task (`task_faulted`) or a parked one (`task_parked`). + * Its own throw or rejection is dropped. */ readonly onReport?: (error: unknown, info: ReportInfo) => void; }; diff --git a/packages/durable/src/harness/util.ts b/packages/durable/src/harness/util.ts index 0b41372f2..cf2222a5a 100644 --- a/packages/durable/src/harness/util.ts +++ b/packages/durable/src/harness/util.ts @@ -59,3 +59,8 @@ export function closedError(session: { readonly failure: { readonly error: unkno const failure = session.failure; return failure === undefined ? new Error("Harness is closed") : new SessionFailed(failure.error); } + +/** `{ extension }` when the extension is known. */ +export function optionalExtension(extension: string | undefined): { readonly extension?: string } { + return extension === undefined ? {} : { extension }; +} diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index b61e8c280..74408d514 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -162,6 +162,7 @@ export type { EntryQuery, EntryRecord, HookRunner, + HostCallback, Id, JoinPolicy, JsonObject, @@ -171,7 +172,7 @@ export type { PhaseHandler, ReportFrom, ReportInfo, - ReportSource, + ReportTask, RewindableConversationDocFamilyToken, RewindableConversationDocToken, RewindableConversationSemantics, diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index 10593c818..a94b32719 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -66,7 +66,7 @@ import { import { type LoadedDocument, Transaction, type TransactionHost, type TransactionScope } from "./transaction.ts"; /** The default source of a report: a listener, which never fails what ran it. */ -const LISTENER: ReportInfo = { source: "listener" }; +const LISTENER: ReportInfo = { type: "listener_error" }; /** * Open a Session kernel over one storage backend. `now` is the wall clock for task times; default `Date.now`. @@ -168,7 +168,7 @@ export class SessionImpl implements Session { // Waits race it; none has to. this.#failed.promise.catch(() => {}); this.#storage = new GuardedStorage(storage, this); - this.#timers = hostTimers(setTimer, (error) => this.report(error, { source: "timer" })); + this.#timers = hostTimers(setTimer, (error) => this.report(error, { type: "host_error", callback: "setTimer" })); // Cache bookkeeping must never change the outcome of the operation it follows: a clock or a retention setting // that throws counts as now, and as the default. this.#clock = () => { @@ -226,7 +226,7 @@ export class SessionImpl implements Session { // Seal first, so a report handler that calls back finds the Session failed. Closing runs the close listeners // synchronously; a failing backend close is the caller's to see, and here nobody waits. this.close(BACKGROUND_CONTEXT).catch(() => {}); - this.report(error, { source: "session" }); + this.report(error, { type: "session_failed" }); } /** @@ -511,7 +511,7 @@ export class SessionImpl implements Session { // A Storage that cannot close is a failed one; an earlier failure stays the cause. if (this.#failure === undefined) { this.#failure = { error }; - this.report(error, { source: "session" }); + this.report(error, { type: "session_failed" }); } throw error; } diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index c0843286f..2042b62cc 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -133,42 +133,64 @@ export type TaskDocFamilyToken = DocF DocFamilyDefinition & { readonly scope: "task" } >; -/** - * Where a reported failure came from (`HarnessOptions.onReport`): - * - `session`: the error failed the Session, which closes (a Storage error, or the Session's own bookkeeping). - * - `task`: a task's phase faulted, a task's runtime reported it, or a task cannot run; `taskId` names it. - * - `extension`: extension code other than a task phase: a hook, a system prompt section, or a wrapper. - * - `listener`: a commit, close, watch, view, or document listener. - * - `host`: a host callback: the clock, the settings, the registry, or the environment builder. - * - `timer`: `HarnessOptions.setTimer`. - * - `wake`: `HarnessOptions.onWake`. - */ -export type ReportSource = "session" | "task" | "extension" | "listener" | "host" | "timer" | "wake"; - -/** Whose code a task's `runtime.report()` is about: its own (default), an extension's, or the host's. */ -export type ReportFrom = - | { readonly source: "task" } - | { readonly source: "extension"; readonly extension?: string } - | { readonly source: "host" }; - -/** What a report is about: its source and, when a task is involved, the task. */ -export type ReportInfo = { - readonly source: ReportSource; - readonly taskId?: TaskId; +/** The task a report is about. */ +export type ReportTask = { + readonly id: TaskId; /** The task's registered name. */ - readonly taskKind?: string; - readonly conversationId?: ConversationId; - /** - * The extension whose code failed, when known: the one that installed a failing task's definition, or a hook's, a - * section's, a wrapper's, or a tool's. Built-in tasks have none. - */ - readonly extension?: string; - /** Set when the error ended its task `faulted`. */ - readonly faulted?: true; - /** Set when the report is about the task being parked by `settings.restartLimit`, or faulted at it. */ - readonly blocked?: "restart_limit"; + readonly kind: string; + readonly conversationId: ConversationId; }; +/** A host callback, by its `HarnessOptions` name; `env` is the environment builder, `models` a provider call. */ +export type HostCallback = "now" | "settings" | "registry" | "env" | "models" | "setTimer" | "onWake"; + +/** + * What a report to `HarnessOptions.onReport` is about. `extension`, where present, names the extension whose code + * failed, when known; for a task's own code, the one that installed the definition the task ran. + * - `session_failed`: the error failed the Session, which closes: a Storage error, often a platform reset. + * - `task_faulted`: the task ended `faulted`: its phase threw (`reason: "threw"`), or its abort handler was cut off + * `settings.restartLimit` times (`reason: "restart_limit"`). + * - `task_parked`: the restart limit parked the task until `Harness.retryTask()`. + * - `migration_failed`: the task's stored state could not be migrated; it is blocked until the registry changes. + * - `task_outdated`: a running task keeps running under its old definition, which was uninstalled (`missing_task`) + * or replaced by one that cannot take over its state (`incompatible_task`). + * - `task_error`: the task's own `runtime.report()`, or a failure in running it. + * - `extension_error`: a hook, a section, a wrapper, or a tool's invalid structured output; with the task that ran it, + * or, outside a task, the conversation whose agent was resolved. + * - `host_error`: a host callback threw; with the task that called it, if one did. + * - `listener_error`: a commit, close, watch, view, or document listener threw. + */ +export type ReportInfo = + | { readonly type: "session_failed" } + | { + readonly type: "task_faulted"; + readonly task: ReportTask; + readonly extension?: string; + readonly reason: "threw" | "restart_limit"; + } + | { readonly type: "task_parked"; readonly task: ReportTask; readonly extension?: string } + | { readonly type: "migration_failed"; readonly task: ReportTask; readonly extension?: string } + | { + readonly type: "task_outdated"; + readonly task: ReportTask; + readonly extension?: string; + readonly reason: "missing_task" | "incompatible_task"; + } + | { readonly type: "task_error"; readonly task: ReportTask; readonly extension?: string } + | { + readonly type: "extension_error"; + readonly extension?: string; + readonly task?: ReportTask; + readonly conversationId?: ConversationId; + } + | { readonly type: "host_error"; readonly callback: HostCallback; readonly task?: ReportTask } + | { readonly type: "listener_error" }; + +/** Whose code a task's `runtime.report()` is about when not the task's own: an extension's, or a host callback. */ +export type ReportFrom = + | { readonly type: "extension"; readonly extension?: string } + | { readonly type: "host"; readonly callback: HostCallback }; + /** * 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 @@ -293,9 +315,9 @@ export interface TaskRuntime extends DocumentObserver */ setTimer(callback: () => void, delayMs: number): () => void; /** - * Forward a non-fatal failure to `HarnessOptions.onReport`, with this task in its info. `from` says whose code - * failed: the task's own (default), `extension` code it called, such as a section, with the extension's name when - * known, or a `host` callback, such as the environment builder. + * Forward a non-fatal failure to `HarnessOptions.onReport`, with this task in its info: a `task_error` about the + * task's own code, or, with `from`, an `extension_error` about extension code it called, such as a section, or a + * `host_error` about a host callback, such as the environment builder. */ report(error: unknown, from?: ReportFrom): void; /** diff --git a/packages/durable/test/harness-blocked.test.ts b/packages/durable/test/harness-blocked.test.ts index 7131cf8d2..0cd7aa484 100644 --- a/packages/durable/test/harness-blocked.test.ts +++ b/packages/durable/test/harness-blocked.test.ts @@ -66,7 +66,7 @@ async function openWith(path: string, registry: Registry, settings: { restartLim return (...args: unknown[]) => (dead ? new Promise(() => {}) : value.apply(target, args)); }, }); - const { harness, reports } = await openTasks(storage, [], { + const { harness, infos } = await openTasks(storage, [], { registry, settings, onWake: (wake) => wakes.push(wake), @@ -75,7 +75,7 @@ async function openWith(path: string, registry: Registry, settings: { restartLim dead = true; killed.add(inner); }; - return { harness, wakes, reports, kill }; + return { harness, wakes, infos, kill }; } describe("recovery at open", () => { @@ -148,9 +148,11 @@ describe("tasks that cannot run", () => { opened.kill(); // a crash: a close would not count opened = await openWith(path, registry, { restartLimit: 3 }); } - const { harness, reports } = opened; + const { harness, infos } = opened; expect((await harness.getTask(id, context))?.restarts).toBe(3); - expect(reports.map((error) => (error as Error).cause)).toEqual(["restart_limit"]); + expect(infos).toEqual([ + { type: "task_parked", task: { id, kind: "test.crash-loop", conversationId: root.id }, extension: "loop" }, + ]); const inspection = await harness.inspect(context); expect(inspection.tasks.find((task) => task.record.id === id)?.state).toEqual({ kind: "blocked", @@ -259,8 +261,8 @@ describe("tasks that cannot run", () => { // 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"]); + await eventually(() => opened.infos.length > 0); + expect(opened.infos).toMatchObject([{ type: "task_faulted", task: { id }, reason: "restart_limit" }]); await opened.harness.close(context); }); diff --git a/packages/durable/test/harness-reports.test.ts b/packages/durable/test/harness-reports.test.ts index a575229a0..0bc2304cd 100644 --- a/packages/durable/test/harness-reports.test.ts +++ b/packages/durable/test/harness-reports.test.ts @@ -67,10 +67,10 @@ describe("reports", () => { expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("faulted"); await eventually(() => reports.length === 2); // The task's own failures name the extension that installed it. - const task = { source: "task", taskId: id, taskKind: "test.faulty", conversationId: root.id, extension: "test" }; + const task = { task: { id, kind: "test.faulty", conversationId: root.id }, extension: "test" }; expect(reports).toEqual([ - { message: "noted", info: task }, - { message: "broken", info: { ...task, faulted: true } }, + { message: "noted", info: { type: "task_error", ...task } }, + { message: "broken", info: { type: "task_faulted", ...task, reason: "threw" } }, ]); await harness.close(context); }); @@ -104,8 +104,9 @@ describe("reports", () => { 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({ + const faulted = (report: Report) => report.info.type === "task_faulted"; + await eventually(() => reports.some(faulted)); + expect(reports.find(faulted)).toMatchObject({ message: "broken in old", info: { extension: "old-extension" }, }); @@ -153,13 +154,10 @@ describe("reports", () => { const root = await harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } }); await (await root.submit({ type: "input", content: "hi" }, context)).wait(context); const infoOf = (message: string) => reports.find((report) => report.message === message)?.info; - expect(infoOf("env")).toMatchObject({ source: "host", taskKind: "pi.generation", conversationId: root.id }); - expect(infoOf("section")).toMatchObject({ - source: "extension", - extension: "sections", - taskKind: "pi.generation", - }); - expect(infoOf("hook")).toMatchObject({ source: "extension", extension: "hooks", taskKind: "pi.generation" }); + const generation = { kind: "pi.generation", conversationId: root.id }; + expect(infoOf("env")).toMatchObject({ type: "host_error", callback: "env", task: generation }); + expect(infoOf("section")).toMatchObject({ type: "extension_error", extension: "sections", task: generation }); + expect(infoOf("hook")).toMatchObject({ type: "extension_error", extension: "hooks", task: generation }); await harness.close(context); }); @@ -210,18 +208,18 @@ describe("reports", () => { ); const infoOf = (message: string) => reports.find((report) => report.message === message)?.info; expect(infoOf("Wrapper renamed echo to renamed")).toEqual({ - source: "extension", + type: "extension_error", conversationId: root.id, extension: "wraps", }); - expect(infoOf("listener")).toEqual({ source: "listener" }); - expect(infoOf("alarm")).toEqual({ source: "wake" }); - expect(infoOf("no timers")).toEqual({ source: "timer" }); + expect(infoOf("listener")).toEqual({ type: "listener_error" }); + expect(infoOf("alarm")).toEqual({ type: "host_error", callback: "onWake" }); + expect(infoOf("no timers")).toEqual({ type: "host_error", callback: "setTimer" }); const failure = new Error("disk gone"); storage.failNextCommit(failure); await root.commit((tx) => tx.createConversation({ ownership: { kind: "ownerless" } }), context).catch(() => {}); await harness.closed; - expect(infoOf("disk gone")).toEqual({ source: "session" }); + expect(infoOf("disk gone")).toEqual({ type: "session_failed" }); }); }); diff --git a/packages/durable/test/harness-tasks-recovery.test.ts b/packages/durable/test/harness-tasks-recovery.test.ts index 1c2458773..ebaea6e1b 100644 --- a/packages/durable/test/harness-tasks-recovery.test.ts +++ b/packages/durable/test/harness-tasks-recovery.test.ts @@ -721,7 +721,7 @@ describe("definition handover", () => { it("hands over to a newer version whose migration fails and leaves the task blocked", async () => { const log: string[] = []; const gate = deferred(); - const { harness, registry, id, reports } = await startHandover(log, { a: gate.promise }); + const { harness, registry, id, infos } = await startHandover(log, { a: gate.promise }); // The same extension name replaces the old one in place. addTask( registry, @@ -732,7 +732,8 @@ describe("definition handover", () => { }), ); gate.resolve(); - await eventually(() => reports.length === 1); + await eventually(() => infos.length === 1); + expect(infos).toMatchObject([{ type: "migration_failed", task: { id } }]); await flush(); expect(await harness.getTask(id, context)).toMatchObject({ version: 1, @@ -746,7 +747,7 @@ describe("definition handover", () => { const log: string[] = []; const gateA = deferred(); const gateB = deferred(); - const { harness, registry, old, id, reports } = await startHandover(log, { a: gateA.promise, b: gateB.promise }); + const { harness, registry, old, id, infos } = await startHandover(log, { a: gateA.promise, b: gateB.promise }); old.dispose(); gateA.resolve(); await eventually(() => log.includes("old:b start")); @@ -755,7 +756,10 @@ describe("definition handover", () => { gateB.resolve(); await harness.waitForTask(id, context); expect(log).toEqual(["old:a start", "old:a end", "old:b start", "old:b end", "old:c"]); - expect(reports.map((report) => (report as Error).cause)).toEqual(["missing_task", "incompatible_task"]); + expect(infos).toMatchObject([ + { type: "task_outdated", task: { id }, reason: "missing_task" }, + { type: "task_outdated", task: { id }, reason: "incompatible_task" }, + ]); await harness.close(context); }); diff --git a/packages/durable/test/task-support.ts b/packages/durable/test/task-support.ts index fc2006b2b..a94373050 100644 --- a/packages/durable/test/task-support.ts +++ b/packages/durable/test/task-support.ts @@ -8,6 +8,7 @@ import { type HarnessSettings, type Registry, type RegistryReader, + type ReportInfo, type Storage, } from "@earendil-works/pi-durable"; import { context, flush } from "./session-support.ts"; @@ -60,7 +61,7 @@ export async function settled(promise: Promise): Promise { return done; } -/** Open a Harness whose registry holds `tasks`; failures passed to `onReport` are collected. */ +/** Open a Harness whose registry holds `tasks`; failures passed to `onReport` are collected, their infos in `infos`. */ export async function openTasks( storage: Storage, tasks: readonly AnyTask[], @@ -71,16 +72,25 @@ export async function openTasks( readonly onWake?: HarnessOptions["onWake"]; readonly setTimer?: HarnessOptions["setTimer"]; } = {}, -): Promise<{ readonly harness: Harness; readonly registry: Registry; readonly reports: unknown[] }> { +): Promise<{ + readonly harness: Harness; + readonly registry: Registry; + readonly reports: unknown[]; + readonly infos: ReportInfo[]; +}> { const registry = options.registry ?? createRegistry(); if (tasks.length > 0) registry.install(defineExtension({ name: "tasks", tasks })); const reports: unknown[] = []; + const infos: ReportInfo[] = []; const harness = await Harness.open( storage, { models: createModels(), registry, - onReport: (error) => reports.push(error), + onReport: (error, info) => { + reports.push(error); + infos.push(info); + }, ...(options.now === undefined ? {} : { now: options.now }), ...(options.settings === undefined ? {} : { settings: options.settings }), ...(options.onWake === undefined ? {} : { onWake: options.onWake }), @@ -88,7 +98,7 @@ export async function openTasks( }, context, ); - return { harness, registry, reports }; + return { harness, registry, reports, infos }; } /** Next state that completes a task with `result`. */