sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(durable): ReportInfo says what happened, with typed reasons

Browse source 
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.
This commit is contained in:
Mario Zechner 2026-10-18 01:58:07 +00:00
parent 5ba1cbaf3d
commit 91b33cab3b
16 changed files with 236 additions and 168 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/durable/CHANGELOG.md
View file

@ -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()`.

4
packages/durable/README.md
View file

@ -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`.

92
packages/durable/docs/spec.md
View file

@ -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<I, S, R, H extends object> 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<void>;
@ -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

12
packages/durable/src/harness/generation.ts
View file

@ -48,6 +48,7 @@ import type {
UserInput,
} from "./types.ts";
import { recordUsage } from "./usage.ts";
import { optionalExtension } from "./util.ts";
export type GenerationInput = Record<string, never>;
@ -157,7 +158,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
env = await runtime.env(context);
} catch (error) {
if (context.abortSignal?.aborted) throw error;
runtime.report(error, { source: "host" });
runtime.report(error, { type: "host", callback: "env" });
}
const input: PromptInput = { conversationId, agent, env, shown: Object.fromEntries(shown), read: runtime };
const desired = await renderSections(
@ -166,7 +167,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
shown,
(error, section) =>
runtime.report(error, {
source: "extension",
type: "extension",
...optionalExtension(sectionExtension(agent, section.key)),
}),
context,
@ -283,7 +284,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
try {
await runtime.models.cancelDeferred(model, checkpoint.handle, { signal: runtime.signal });
} catch (error) {
runtime.report(error, { source: "host" });
runtime.report(error, { type: "host", callback: "models" });
}
}
}
@ -759,8 +760,3 @@ function createGeneration(tx: Tx, conversationId: ConversationId): Promise<TaskI
export function handOver(live: Draft<LiveState>, 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 };
}

8
packages/durable/src/harness/harness.ts
View file

@ -212,7 +212,7 @@ class HarnessImpl<Tool extends ToolRegistration> 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<Tool extends ToolRegistration> 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<Tool extends ToolRegistration> 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();
}

89
packages/durable/src/harness/scheduler.ts
View file

@ -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<JsonValue, JsonValue, JsonValue>;
/** 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<AnyTaskRecord, "id" | "kind" | "conversationId">) {
return { taskId: record.id, kind: record.kind, conversationId: record.conversationId };
}
/**

3
packages/durable/src/harness/tool.ts
View file

@ -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

6
packages/durable/src/harness/types.ts
View file

@ -645,9 +645,9 @@ export type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
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;
};

5
packages/durable/src/harness/util.ts
View file

@ -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 };
}

3
packages/durable/src/index.ts
View file

@ -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,

8
packages/durable/src/session/session.ts
View file

@ -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;
}

94
packages/durable/src/types.ts
View file

@ -133,42 +133,64 @@ export type TaskDocFamilyToken<T extends JsonObject, I extends JsonValue> = DocF
DocFamilyDefinition<T, I> & { 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<I, S, R, H extends object> 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;
/**

14
packages/durable/test/harness-blocked.test.ts
View file

@ -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);
});

32
packages/durable/test/harness-reports.test.ts
View file

@ -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" });
});
});

12
packages/durable/test/harness-tasks-recovery.test.ts
View file

@ -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);
});

18
packages/durable/test/task-support.ts
View file

@ -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<unknown>): Promise<boolean> {
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`. */