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

fix(durable): review fixes for the crash-loop guard, reports, and blocked tasks

Browse source 
- A close writes the tasks it stopped back as pending, so only a crash counts a restart.
- An abort mark starts the count over; an abort handler cut off restartLimit times ends the task faulted.
- Only a changed state resets the count; raised and lowered limits reach the scheduler and the graph.
- wake() stays active while a reconcile pass is scheduled or underway; failed migrations recompute it.
- Reports name the extension whose code failed (hooks, wrappers, sections, tools, a task's installer),
  runtime.report() takes a source, and the parked report carries blocked: restart_limit.
- The task graph keeps only what blocked reasons read, and shows abandoned tasks waiting for a definition.
This commit is contained in:
Mario Zechner 2026-10-18 00:53:14 +00:00
parent a30c2a81df
commit 440a5fcdb8
21 changed files with 730 additions and 113 deletions
Download patch file Download diff file Expand all files Collapse all files

8
packages/durable/CHANGELOG.md
View file

@ -8,7 +8,8 @@
- Tasks have a `sleeping` status, so code that handles every `TaskState`, `TaskGraphState`, or `TaskInspection` state needs a case for it.
- Opening a SQLite storage migrates it to schema version 2 automatically, rebuilding its task table. A migrated database cannot be opened by 1.3.0 or earlier, so a deployment cannot roll back past this release. JSONL storage needs no migration, but 1.3.0 or earlier never resumes a sleeping task it finds (a run's conversation stays busy).
- `TaskRuntime` has `sleepUntil()` and `setTimer()`, and `ToolExecutionApi` has `now()`: code that builds either object by hand, such as a test fake or a tool wrapper that does not spread the original, must add them.
- A task cut off by a crash or close 5 times in a row without committing a new state is parked as blocked (`restart_limit`) instead of run again; set `settings.restartLimit` to `Infinity` to keep the old behavior. `TaskInspection`'s blocked reasons include `restart_limit`.
- A task cut off by a crash 5 times in a row without committing a new state is parked instead of run again (see the crash-loop guard under Added); set `settings.restartLimit` to `Infinity` to keep the old behavior. `TaskInspection`'s blocked reasons include `restart_limit`.
- `onReport` gets a second argument, `info` (see below), and `TaskRuntime.report()` takes an optional `from`; an `onReport` written for one argument still works.
### Added
@ -16,14 +17,15 @@
- Durable sleeps: a sleeping task holds no invocation, survives restarts, and resumes at its checkpoint once the Harness clock reaches `until`.
- `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 or close cut off (`TaskRecord.restarts`), and a 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.
- `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, and conversation (and `faulted` for a phase that faulted the task, which is now reported too), or `extension`, `listener`, `host`, `timer`, or `wake`.
- 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`.
- `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.
- Generation retry backoffs, deferred-response polls, and compaction retry backoffs sleep instead of waiting in a running phase, so a process can stop or hibernate during them. A record an earlier version stored mid-backoff sleeps again when it resumes before its time. A deferred poll whose model is gone after a reopen now fails `no_model` once the poll is due, not at once.
- Code mode measures script deadlines and wall time on the Harness clock (`HarnessOptions.now`) instead of `Date.now()`.
- SQLite drivers, portable interfaces, and the generic migration runner now come from `@earendil-works/chord/sqlite`. Existing Durable SQLite entry points keep re-exporting their public names; storage helpers and Durable schema migrations stay here.

10
packages/durable/README.md
View file

@ -199,15 +199,15 @@ export class Session extends DurableObject<Env> {
The default `setTimer` declines the timers that only free memory in Workers, so no pending timer keeps the object from hibernating or being evicted; memory is then freed at later commits.
### When Work Keeps Failing
### Failures and Recovery
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 or close cut off; a state the task commits itself 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. `harness.retryTask(id)` runs it again; `abortTask()` still runs its abort handler.
**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`.
Every report says where it came from: `onReport(error, info)` gets `info.source`, which is `session` for the error that failed the Harness (a Storage error, often a platform reset), `task` for a task that faulted (`faulted: true`) or reported a failure, with `taskId`, `taskKind`, and `conversationId`, and `extension`, `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 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.
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`.
**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`.
Tasks blocked on a missing or unmigratable definition show their reason in the task graph too, and a registry change that unblocks them makes `onWake` report the Session active, so a host need not poll them.
**Blocked tasks.** Tasks blocked on a missing or unmigratable definition show their reason in the task graph too, and a registry change that unblocks them makes `onWake` report the Session active, so a host need not poll them.
## Extensions

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

@ -391,8 +391,10 @@ type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
* changes, off the Session line; the host arms its own wakeup and calls `resume()` when it fires.
*/
readonly onWake?: (wake: Wake) => void;
/** Failures that do not fail the calling operation, and the Session's failure, once (section 10.1). */
/** `info` says where it came from: a failed Session, a task (with `faulted` when it ended one), extension code, and so on (section 10.1). */
/**
* Failures that do not fail the calling operation, and the Session's failure, once; `info` says where each came
* from (section 10.1).
*/
readonly onReport?: (error: unknown, info: ReportInfo) => void;
};
@ -600,7 +602,9 @@ type ReportInfo = {
readonly taskId?: TaskId;
readonly taskKind?: string;
readonly conversationId?: ConversationId;
readonly extension?: string;
readonly faulted?: true;
readonly blocked?: "restart_limit";
};
type HarnessInspection = {
@ -1027,7 +1031,9 @@ joins task, tool, and hook invocations, including code that ignores its signal,
then closes Storage. Once `close()` resolves, no invocation code of this Harness
runs and no handler holds Storage, so a new Harness may open the same Storage.
Already-running watch callbacks remain caller-owned and may finish
independently. Close writes no task outcome. Cancelling a `close()` call cancels
independently. Close writes no task outcome; it writes the tasks it stopped
back as `pending`, so a reopen counts no restart for them (section 5.4).
Cancelling a `close()` call cancels
only that wait; shutdown continues, and every later `close()` awaits the same
shutdown.
@ -1764,7 +1770,7 @@ type TaskRecord<I, S, R> = {
readonly startedAt?: number;
/** Wall clock at the change to `terminal`. */
readonly endedAt?: number;
/** Times in a row a crash or close cut it off without a state change since (section 5.4). */
/** Times in a row a crash cut it off without a state change since; a close does not count (section 5.4). */
readonly restarts?: number;
} & (
| {
@ -1842,7 +1848,11 @@ interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver, Docum
/** The Harness clock. */
now(): number;
/** Forward a non-fatal failure to `HarnessOptions.onReport`. */
report(error: unknown): void;
/** `from`: whose code failed, the task's own (default), an extension's (with its name), or a `host` callback's. */
report(
error: unknown,
from?: { source: "task" } | { source: "extension"; extension?: string } | { source: "host" },
): void;
/** Commit `{ status: "sleeping", until, checkpoint }`; resolves once stored (section 5.6). */
sleepUntil(until: number, checkpoint: S, context: Context): Promise<void>;
/** A `HarnessOptions.setTimer` timer for pacing work in a phase; the invocation's end disarms it. */
@ -1994,7 +2004,9 @@ definition is installed or the task is aborted (section 5.4).
`close()` marks the runtime closing, seals admission and reservation, signals
invocations, and ends states and watches. Outside the Session line it settles admitted
commits and joins invocations before closing storage. Already-running watch callbacks remain
caller-owned. Later runtime commits reject, and close writes no task outcome. Closing
caller-owned. Later runtime commits reject, and close writes no task outcome; after
joining, it writes the tasks it stopped back as `pending`, straight to Storage, with
nothing published. Closing
starts no fresh phase or abort invocation. It does not set abort marks,
terminalize tasks, retire task documents, or publish document retirement. The
hosting layer stops exposing Harness state to clients before it closes the
@ -2152,17 +2164,26 @@ equal version -> commit running
```
**Restarts.** Open counts every surviving `running` task as cut off: it writes
the task back as `pending` with `restarts` one higher. Any state the task commits
itself (a new checkpoint, a wait, a sleep, an outcome) is progress and removes
the count; memos and document writes are not. Once `restarts` reaches
the task back as `pending` with `restarts` one higher. A close is not a crash:
after joining the invocations it stopped, close writes those tasks back as
`pending` without a count, so only a process that died, was evicted, or was
reset leaves `running` records. A state the task commits
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
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.
`Harness.retryTask(id)` removes the count and runs the task again; it does
nothing for a task that is not parked. An abort mark still runs a parked task's
abort handler.
checkpoint thus stops after a few attempts instead of running forever. Every
task running when a process dies is counted, not only the one that killed it,
so a long stream next to the culprit can be parked too; a host shows parked
tasks and offers `Harness.retryTask(id)`, which removes the count and runs the
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"`.
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
@ -2411,7 +2432,9 @@ paused, when the Session next needs a running process:
- `active`: some task can run now or is running: a pending task that is not
blocked, a running invocation, a waiting task whose `on` is settled, a due
sleep, or an abort-marked task whose owned work drained. A host keeps a
sleep, or an abort-marked task whose owned work drained; also while a
reconcile pass (section 5.5) is scheduled or underway, such as open's
recovery, since it may finalize held outcomes and mark cascades. A host keeps a
recovery wakeup armed while it is set, so work cut off by a crash resumes.
- `at`: the earliest sleep not yet due.
@ -2516,6 +2539,8 @@ own inputs: `keep` (the default) leaves them queued; `continue` applies the
final boundary, so the next run starts with them, as after an answer;
`withdraw` settles them `unanswered` with reason `run_failed`, keeping queued
writes. A run whose task is aborted, faults, or is orphaned keeps its inbox.
With `continue`, a cause that persists, such as a missing model, fails the next
run too, so the queue drains one failed run per input.
An idle conversation with a non-empty inbox, for example after a failed run,
queues every new input and every write that is not display-only behind the
@ -5301,10 +5326,15 @@ once, and `Date.now` serves instead.
`ReportInfo`: `source` is `session` for the error that failed the Session (a
Storage error, often a platform reset, or the Session's own bookkeeping),
`task` for a task's phase that faulted (`faulted: true`), a task's
`runtime.report()`, a failed migration, or a task parked by the restart limit,
with `taskId`, `taskKind`, and `conversationId`; `extension` for hooks and
wrappers; `listener`; `host` for the clock, settings, registry, and environment
builder; `timer` for `setTimer`; and `wake` for `onWake`. A host acts on them
`runtime.report()`, a failed migration, or a task parked by the restart limit
(`blocked: "restart_limit"`), with `taskId`, `taskKind`, and `conversationId`; `extension` for hooks,
wrappers, system prompt sections, and a tool's invalid structured output;
`listener`; `host` for the clock, settings, registry, environment builder, and
model provider calls such as cancelling a deferred response; an `extension`
report names the extension (`extension`) when the failing code is known to
belong to one: the hook's, the wrapper's, or the last selected extension that
provides or wraps the section or tool. A `task` report names the extension that
installed the task's definition; built-in tasks have none; `timer` for `setTimer`; and `wake` for `onWake`. A host acts on them
outside the scheduler, for example counting faults of one task kind toward a
breaker.

45
packages/durable/src/harness/agent.ts
View file

@ -56,9 +56,9 @@ const DEFAULT_CONTEXT_RETENTION_MS = 600_000;
/** Default `settings.restartLimit`. */
export const DEFAULT_RESTART_LIMIT = 5;
/** A restart limit as given when it is a number of restarts, `1` or more (`Infinity` never parks); else the default. */
/** A restart limit as given when it is a number of restarts, `1` or more, rounded down (`Infinity` never parks); else the default. */
function restartLimit(limit: number | undefined): number {
return typeof limit === "number" && limit >= 1 ? limit : DEFAULT_RESTART_LIMIT;
return typeof limit === "number" && limit >= 1 ? Math.floor(limit) : DEFAULT_RESTART_LIMIT;
}
/** Resolve the host settings: every field over its built-in default, object fields merged. */
@ -180,14 +180,41 @@ export async function createAgent(tx: Tx, conversation: ConversationRecord): Pro
}
/** Handlers of the selected extensions' hooks for a task name, in extension order. */
export function agentHooks(agent: Agent, taskName: string): object[] {
const handlers: object[] = [];
export function agentHooks(
agent: Agent,
taskName: string,
): { readonly extension: string; readonly handlers: object }[] {
const handlers: { readonly extension: string; readonly handlers: object }[] = [];
for (const extension of agent.extensions) {
for (const hook of extension.hooks ?? []) if (hook.task === taskName) handlers.push(hook.handlers);
for (const hook of extension.hooks ?? []) {
if (hook.task === taskName) handlers.push({ extension: extension.name, handlers: hook.handlers });
}
}
return handlers;
}
/** The extension whose code a section of `agent` runs: the last selected one that provides or wraps it. */
export function sectionExtension(agent: Agent, key: string): string | undefined {
let found: string | undefined;
for (const extension of agent.extensions) {
const provides = extension.sections?.some((section) => section.key === key) === true;
const wraps = extension.wraps?.some((wrap) => "section" in wrap && wrap.section === key) === true;
if (provides || wraps) found = extension.name;
}
return found;
}
/** The extension whose code a tool of `agent` runs: the last selected one that provides or wraps it. */
export function toolExtension(agent: Agent, name: string): string | undefined {
let found: string | undefined;
for (const extension of agent.extensions) {
const provides = extension.tools?.some((tool) => tool.name === name) === true;
const wraps = extension.wraps?.some((wrap) => "tool" in wrap && wrap.tool === name) === true;
if (provides || wraps) found = extension.name;
}
return found;
}
/**
* Resolve an agent from its stored state (absent: every field unset), a registry snapshot, and resolved settings. A
* wrapper that throws or renames drops its target and is reported; a wrapper without a target does nothing.
@ -196,7 +223,8 @@ export function resolveAgent<Tool extends ToolRegistration>(
state: Readonly<AgentState> | undefined,
snapshot: RegistrySnapshot<Tool>,
settings: Settings,
report: (error: unknown) => void,
/** Gets a wrapper's failure with the name of the wrapper's extension. */
report: (error: unknown, extension: string) => void,
): Agent<Tool> {
const extensions = selectExtensions(state?.extensions, snapshot, settings);
@ -208,13 +236,14 @@ export function resolveAgent<Tool extends ToolRegistration>(
}
for (const extension of extensions) {
for (const wrap of extension.wraps ?? []) {
const wrapReport = (error: unknown) => report(error, extension.name);
if ("tool" in wrap)
applyWrap(
composed,
wrap.tool,
(tool) => wrap.wrap(tool),
(tool) => tool.name,
report,
wrapReport,
);
else
applyWrap(
@ -222,7 +251,7 @@ export function resolveAgent<Tool extends ToolRegistration>(
wrap.section,
(section) => wrap.wrap(section),
(section) => section.key,
report,
wrapReport,
);
}
}

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

@ -27,7 +27,7 @@ import type {
Tx,
TypedEntry,
} from "../types.ts";
import { addTools } from "./agent.ts";
import { addTools, sectionExtension } from "./agent.ts";
import { compactionPolicy, createCompaction, estimateContext, selectCut } from "./compaction.ts";
import { applyBoundary, type Boundary, prepareBoundary, withdrawQueuedInputs } from "./inbox.ts";
import { assignJson } from "./json.ts";
@ -152,16 +152,25 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
}
const view = await runtime.context(conversationId, context);
const shown = replaySections(view.messages);
const report = (error: unknown) => runtime.report(error);
let env: ExecutionEnv | undefined;
try {
env = await runtime.env(context);
} catch (error) {
if (context.abortSignal?.aborted) throw error;
report(error);
runtime.report(error, { source: "host" });
}
const input: PromptInput = { conversationId, agent, env, shown: Object.fromEntries(shown), read: runtime };
const desired = await renderSections(agent.sections, input, shown, report, context);
const desired = await renderSections(
agent.sections,
input,
shown,
(error, section) =>
runtime.report(error, {
source: "extension",
...optionalExtension(sectionExtension(agent, section.key)),
}),
context,
);
const entries = planSystemEntries(view, desired, agent.tools, runtime.now());
const threshold =
compacted === undefined
@ -274,7 +283,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
try {
await runtime.models.cancelDeferred(model, checkpoint.handle, { signal: runtime.signal });
} catch (error) {
runtime.report(error);
runtime.report(error, { source: "host" });
}
}
}
@ -750,3 +759,8 @@ 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 };
}

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

@ -213,6 +213,7 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
now,
onWake: options.onWake,
timers: hostTimers(options.setTimer, (error) => report(error, { source: "timer" })),
writeAtClose: (writes) => this.writeAtClose(writes),
report: this.#report,
settleOutcome: settleSchedulerOutcome,
withdrawInputs: withdrawQueuedInputs,
@ -246,8 +247,8 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
): Promise<Agent<Tool>> {
const registry = snapshot ?? this.#options.registry.snapshot();
const state = await this.snapshot(AgentDoc, id, context);
return resolveAgent(state, registry, resolveSettings(this.#hostSettings()), (error) =>
this.#report(error, { source: "extension", conversationId: id }),
return resolveAgent(state, registry, resolveSettings(this.#hostSettings()), (error, extension) =>
this.#report(error, { source: "extension", conversationId: id, extension }),
);
}

4
packages/durable/src/harness/prompt.ts
View file

@ -26,7 +26,7 @@ export async function renderSections<Tool extends ToolRegistration>(
sections: readonly PromptSection<Tool>[],
input: PromptInput<Tool>,
shown: ReadonlyMap<string, string>,
report: (error: unknown) => void,
report: (error: unknown, section: PromptSection<Tool>) => void,
context: Context,
): Promise<Map<string, string>> {
const desired = new Map<string, string>();
@ -36,7 +36,7 @@ export async function renderSections<Tool extends ToolRegistration>(
text = await section.render(input, context);
} catch (error) {
if (context.abortSignal?.aborted) throw error;
report(error);
report(error, section);
const kept = shown.get(section.key);
if (kept !== undefined) desired.set(section.key, kept);
continue;

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

@ -18,6 +18,7 @@ import type {
ReportInfo,
RunningTask,
Storage,
StorageWrite,
TaskDefinition,
TaskId,
TaskOutcome,
@ -55,6 +56,27 @@ type ErasedRunningTask = RunningTask<JsonValue, Checkpoint, JsonValue>;
const SCAN_PAGE_SIZE = 256;
const LIVE_STATUSES = ["pending", "running", "waiting", "sleeping", "completing"] as const;
/** What `blockedReason()` reads of a task: the task graph keeps only this of each record. */
export type BlockedSubject = Pick<
AnyTaskRecord,
"id" | "kind" | "version" | "abortRequested" | "abortReason" | "restarts"
> & {
readonly status: TaskState<JsonValue, JsonValue>["status"];
};
/** The part of a record `blockedReason()` reads. */
export function blockedSubject(record: AnyTaskRecord): BlockedSubject {
return {
id: record.id,
kind: record.kind,
version: record.version,
abortRequested: record.abortRequested,
...(record.abortReason === undefined ? {} : { abortReason: record.abortReason }),
...(record.restarts === undefined ? {} : { restarts: record.restarts }),
status: record.state.status,
};
}
/**
* Why a task cannot be reserved: no registered definition can take it, or it was cut off `restartLimit` times in a row.
* Derived from the registry, the settings, and the record's `restarts`.
@ -172,6 +194,8 @@ export type TaskSchedulerOptions = {
readonly onWake: ((wake: Wake) => void) | undefined;
/** The host's timers (`HarnessOptions.setTimer`). */
readonly timers: Timers;
/** Write records straight to Storage while closing, after admission is sealed. */
readonly writeAtClose: (writes: readonly StorageWrite[]) => Promise<void>;
readonly report: (error: unknown, info: ReportInfo) => void;
/** Harness cleanup staged in the commit that makes an outcome the scheduler wrote itself terminal. */
readonly settleOutcome: (tx: Transaction, record: AnyTaskRecord, outcome: SchedulerOutcome) => Promise<void>;
@ -220,6 +244,7 @@ export class TaskScheduler {
readonly #now: () => number;
readonly #onWake: ((wake: Wake) => void) | undefined;
readonly #timers: Timers;
readonly #writeAtClose: (writes: readonly StorageWrite[]) => Promise<void>;
readonly #report: (error: unknown, info: ReportInfo) => void;
readonly #settleOutcome: TaskSchedulerOptions["settleOutcome"];
readonly #withdrawInputs: TaskSchedulerOptions["withdrawInputs"];
@ -283,7 +308,13 @@ export class TaskScheduler {
#lastWake: Wake | undefined;
#wakeScheduled = false;
readonly #blockedListeners = new Set<() => void>();
/** A registry snapshot that threw for the task graph was reported; it is read per record, so only once. */
#snapshotReported = false;
/** Tasks a pass left out as parked; each pass checks them again, as `settings.restartLimit` may have risen. */
readonly #parkedTasks = new Set<TaskId>();
#reconcileScheduled = false;
/** Reconcile passes underway (they may overlap): their commits finalize held outcomes and mark cascades, recovery work. */
#reconciling = 0;
#sweepScheduled = false;
#cascadePending = false;
#unsubscribeRegistry: () => void = () => {};
@ -303,6 +334,7 @@ export class TaskScheduler {
this.#now = options.now;
this.#onWake = options.onWake;
this.#timers = options.timers;
this.#writeAtClose = options.writeAtClose;
this.#report = options.report;
this.#settleOutcome = options.settleOutcome;
this.#withdrawInputs = options.withdrawInputs;
@ -351,7 +383,8 @@ export class TaskScheduler {
`Task ${record.id} (${record.kind}) was cut off ${restarts} times without progress; it is parked until retryTask()`,
{ cause: "restart_limit" },
);
this.#report(error, taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }));
const info = this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId });
this.#report(error, { ...info, blocked: "restart_limit" });
}
// Derive abort marks a crash left unapplied below cancelled owners, and finalize held outcomes.
this.#cascadePending = true;
@ -377,7 +410,7 @@ export class TaskScheduler {
await this.#session.commitWith(async (tx) => {
const current = await tx.task(id);
if (current === undefined) throw new Error(`Task ${id} does not exist`);
if (current.state.status === "terminal" || current.restarts === undefined) return;
if (!this.#parked(current)) return;
tx.setTask(progressed(current));
}, context);
this.resume();
@ -387,17 +420,18 @@ export class TaskScheduler {
* Why a live task cannot run here, if it cannot: its definition (missing, too old, failed migration) or the restart
* limit. A running task, an abort-marked one, and one holding its outcome are never blocked.
*/
blockedReason(record: AnyTaskRecord): BlockedReason | undefined {
const status = record.state.status;
if (status === "running" || status === "completing" || status === "terminal" || record.abortRequested) {
return undefined;
}
blockedReason(record: BlockedSubject): BlockedReason | undefined {
const status = record.status;
if (status === "running" || status === "completing" || status === "terminal") return undefined;
// An abort request runs its handler or orphans the task; a `restart` mark waits for the definition, blocked.
if (record.abortRequested && record.abortReason === undefined) return undefined;
if (this.#parked(record)) return "restart_limit";
let snapshot: RegistrySnapshot;
try {
snapshot = this.#registry.snapshot();
} catch (error) {
this.#report(error, HOST);
if (!this.#snapshotReported) this.#report(error, HOST);
this.#snapshotReported = true;
return undefined;
}
const fit = this.#fit(record, snapshot.task(record.kind));
@ -417,7 +451,9 @@ export class TaskScheduler {
*/
wake(): Wake {
const now = this.#now();
let active = this.#invocations.size > 0;
// A reconcile pass scheduled or underway is work too: it may finalize held outcomes and mark cascades, at open as
// recovery. A process that dies before it lands must be woken again.
let active = this.#invocations.size > 0 || this.#reconcileScheduled || this.#reconciling > 0;
let at: number | undefined;
let snapshot: RegistrySnapshot | undefined;
let snapshotFailed = false;
@ -452,12 +488,23 @@ export class TaskScheduler {
}
/**
* Signal every invocation and wait for them; writes nothing. Close calls this after every close listener has run, so
* waits, watches, and streams a task holds end with the Session's reason, not as cancelled by its signal.
* Signal every invocation and wait for them, then write the tasks they ran back as `pending` (`writeAtClose`), as a
* close is not a crash. Close calls this after every close listener has run, so waits, watches, and streams a task
* holds end with the Session's reason, not as cancelled by its signal.
*/
async join(): Promise<void> {
for (const invocation of this.#invocations.values()) invocation.controller.abort();
await Promise.allSettled([...this.#invocations.values()].map((invocation) => invocation.done));
const stopped = [...this.#invocations.values()];
for (const invocation of stopped) invocation.controller.abort();
await Promise.allSettled(stopped.map((invocation) => invocation.done));
// A close is not a crash: a task it stopped is `pending` again, so the next open counts no restart for it.
const released = stopped.flatMap((invocation) => {
const record = this.#live.get(invocation.taskId);
if (record?.state.status !== "running") return [];
return [withState(record, { status: "pending", checkpoint: record.state.checkpoint })];
});
if (released.length === 0) return;
// A failure leaves them `running`, counted as if cut off; the Storage guard fails the Session and reports it.
await this.#writeAtClose(released.map((value) => ({ type: "task", value }) as const)).catch(() => {});
}
/**
@ -700,6 +747,7 @@ export class TaskScheduler {
*/
async #reconcile(): Promise<void> {
this.#reconcileScheduled = false;
this.#reconciling++;
const cascade = this.#cascadePending;
this.#cascadePending = false;
const checks = [...this.#failFastChecks];
@ -746,6 +794,8 @@ export class TaskScheduler {
// running the pass again, so it fails the Session, which reports it.
this.#failSession(error);
}
this.#reconciling--;
this.#scheduleWake();
this.#settleIdle();
}
@ -1269,10 +1319,49 @@ export class TaskScheduler {
}
/** A task cut off `restartLimit` times in a row; an abort mark still runs its abort handler. */
#parked(record: AnyTaskRecord): boolean {
#parked(record: Pick<AnyTaskRecord, "abortRequested" | "restarts">): boolean {
return !record.abortRequested && record.restarts !== undefined && record.restarts >= this.#restartLimit();
}
/**
* 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: {
readonly taskId: TaskId;
readonly kind: string;
readonly conversationId: ConversationId;
}): ReportInfo {
const info = taskInfo(task);
let extension: string | undefined;
try {
const snapshot = this.#registry.snapshot();
const definition = snapshot.task(task.kind);
if (definition !== undefined) {
extension = snapshot
.installed()
.findLast((installed) => installed.tasks?.includes(definition) === true)?.name;
}
} catch {
// A registry that throws is reported where it is read for scheduling; a report goes out without the name.
}
return extension === undefined ? info : { ...info, extension };
}
/** Make parked tasks that a raised `settings.restartLimit` no longer parks candidates again, and tell the graph. */
#unparkRaised(): void {
let unparked = false;
for (const id of this.#parkedTasks) {
const record = this.#live.get(id);
if (record !== undefined && this.#parked(record)) continue;
this.#parkedTasks.delete(id);
if (record === undefined) continue;
this.#runnable.add(id);
unparked = true;
}
if (unparked) queueMicrotask(() => this.#notifyBlocked());
}
/** `settings.restartLimit`, or the default when the host's settings throw. */
#restartLimit(): number {
try {
@ -1332,6 +1421,7 @@ export class TaskScheduler {
*/
async #reserve(): Promise<Reservation[]> {
const reservations: Reservation[] = [];
const faulted: { readonly record: AnyTaskRecord; readonly message: string }[] = [];
let abandoned: TaskId[] = [];
try {
await this.#session.commitWith(async (tx) => {
@ -1348,6 +1438,7 @@ export class TaskScheduler {
// Every pass picks up sleeps whose time has come: a pass started by any change may be the first to run after
// a deadline, and the wake it schedules no longer keeps a timer for a due sleep.
this.#wakeDue();
this.#unparkRaised();
// Taken once per pass, and only when some task is a candidate.
let snapshot: RegistrySnapshot | undefined;
for (const id of [...this.#runnable]) {
@ -1366,9 +1457,12 @@ export class TaskScheduler {
this.#runnable.delete(id);
continue;
}
// It returns when `retry()` clears its restart count.
// It returns when `retry()` clears its restart count, or at a pass after `settings.restartLimit` rose.
if (this.#parked(record)) {
this.#runnable.delete(id);
// Newly parked without a commit, as after a lowered limit: the graph must hear it.
if (!this.#parkedTasks.has(id)) queueMicrotask(() => this.#notifyBlocked());
this.#parkedTasks.add(id);
continue;
}
const runnable = record as RunnableTaskRecord;
@ -1380,6 +1474,16 @@ export class TaskScheduler {
this.#scheduleReconcile();
continue;
}
// An abort handler cut off `restartLimit` times in a row would run forever; nothing is left to park it for.
if (mode === "abort" && (record.restarts ?? 0) >= this.#restartLimit()) {
const message = `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times`;
await this.#terminate(tx, record, {
status: "faulted",
error: { message, detail: { reason: "restart_limit" } },
});
faulted.push({ record, message });
continue;
}
snapshot ??= this.#registry.snapshot();
const resolution = this.#resolve(runnable, snapshot);
if (resolution.kind === "blocked") {
@ -1412,6 +1516,14 @@ 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",
});
}
// Marked, or already marked or gone: done with them. Reserve again, now seeing the marks.
for (const id of abandoned) this.#abandoned.delete(id);
if (abandoned.length > 0) this.#dirty = true;
@ -1444,13 +1556,18 @@ export class TaskScheduler {
return { kind: "ready", task: fit.task, record: migratedRecord as RunnableTaskRecord };
} catch (error) {
this.#failedMigrations.set(record.id, { task: fit.task, error });
// Blocked now without a commit: the graph and the host's wake must hear it.
queueMicrotask(() => this.#notifyBlocked());
this.#report(error, taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }));
this.#scheduleWake();
this.#report(
error,
this.#taskInfo({ taskId: record.id, kind: record.kind, conversationId: record.conversationId }),
);
return { kind: "blocked", reason: "migration_failed" };
}
}
#fit(record: AnyTaskRecord, task: AnyTask | undefined): Fit {
#fit(record: Pick<AnyTaskRecord, "id" | "version">, task: AnyTask | undefined): Fit {
if (task === undefined) return { reason: "missing_task" };
const version = task.definition.version;
if (version === record.version) return { task, migrates: false };
@ -1534,7 +1651,7 @@ export class TaskScheduler {
if (invocation.mode === "run") await this.#run(reservation);
else await this.#runAbort(reservation);
} catch (error) {
this.#report(error, taskInfo(invocation));
this.#report(error, this.#taskInfo(invocation));
} finally {
this.#end(invocation);
invocation.finish();
@ -1605,7 +1722,7 @@ export class TaskScheduler {
const cause = next === undefined ? "missing_task" : "incompatible_task";
this.#report(
new Error(`Task ${current.id} keeps running under its old ${current.kind} definition`, { cause }),
taskInfo({ taskId: current.id, kind: current.kind, conversationId: current.conversationId }),
this.#taskInfo({ taskId: current.id, kind: current.kind, conversationId: current.conversationId }),
);
}
}
@ -1655,7 +1772,7 @@ 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, { ...taskInfo(invocation), faulted: true });
if (fault !== undefined) this.#report(fault.error, { ...this.#taskInfo(invocation), faulted: true });
return continued;
} catch (error) {
// The step writes the scheduler's own decision, a fault or a terminal record; a throw there would otherwise
@ -1702,8 +1819,10 @@ export class TaskScheduler {
return;
}
}
// A state the task committed is progress: its restart count starts over.
tx.setTask(withState(progressed(current), next));
// A state the task committed that differs from its current one is progress: its restart count starts over. The
// same checkpoint committed again, for example as intent before an effect, is not.
const same = jsonEqual(current.state as unknown as JsonValue, next as unknown as JsonValue);
tx.setTask(withState(same ? current : progressed(current), next));
}
/**
@ -1839,14 +1958,14 @@ export class TaskScheduler {
};
const hooks: HookRunner<Record<string, unknown>> = {
each: async (name, invoke) => {
for (const handlers of agentHooks(await agent(), phase.task().definition.name)) {
for (const { extension, handlers } of agentHooks(await agent(), phase.task().definition.name)) {
const handler = (handlers as Record<string, unknown>)[name];
if (typeof handler !== "function") continue;
try {
await invoke(handler.bind(handlers));
} catch (error) {
if (invocation.controller.signal.aborted) throw error;
this.#report(error, { ...taskInfo(invocation), source: "extension" });
this.#report(error, { ...taskInfo(invocation), source: "extension", extension });
}
}
},
@ -2030,9 +2149,13 @@ export class TaskScheduler {
disarm();
};
},
report: (error) => {
report: (error, from = { source: "task" }) => {
if (invocation.ended) throw endedError(invocation);
this.#report(error, taskInfo(invocation));
// The task's own failure names the task's extension; any other names whose code failed, if anyone's.
this.#report(
error,
from.source === "task" ? this.#taskInfo(invocation) : { ...taskInfo(invocation), ...from },
);
},
};
}
@ -2093,9 +2216,12 @@ function taskInfo(task: {
return { source: "task", taskId: task.taskId, taskKind: task.kind, conversationId: task.conversationId };
}
/** The record with an abort mark: `restart` for an abandonment; without a reason, a request, which replaces `restart`. */
/**
* The record with an abort mark: `restart` for an abandonment; without a reason, a request, which replaces `restart`.
* The restart count starts over: it now counts cut-offs of the abort handler.
*/
function withAbortMark(record: AnyTaskRecord, reason?: "restart"): AnyTaskRecord {
const { abortReason: _reason, ...rest } = record;
const { abortReason: _reason, restarts: _restarts, ...rest } = record;
return { ...rest, abortRequested: true, ...(reason === undefined ? {} : { abortReason: reason }) } as AnyTaskRecord;
}

22
packages/durable/src/harness/task-graph.ts
View file

@ -13,7 +13,7 @@ import type {
TaskRecord,
WatchHandle,
} from "../types.ts";
import type { BlockedReason } from "./scheduler.ts";
import { type BlockedReason, type BlockedSubject, blockedSubject } from "./scheduler.ts";
import { closedError, scanAll } from "./util.ts";
/** A live task's durable status without its checkpoint and outcome payloads (spec §9.5). */
@ -61,13 +61,13 @@ type Observer = CommittedStateSource<TaskGraph> | CommittedWatch<TaskGraph>;
type Mount = {
value: TaskGraph;
readonly observers: Set<Observer>;
/** The live records behind the nodes, so a registry change can derive their blocked reasons again. */
readonly records: Map<string, AnyTaskRecord>;
/** What the blocked reason reads of each live task, so a registry change can derive it again without payloads. */
readonly records: Map<string, BlockedSubject>;
};
/** What the graph asks the scheduler: a record's blocked reason, and when reasons may change without a commit. */
/** What the graph asks the scheduler: a task's blocked reason, and when reasons may change without a commit. */
export type TaskGraphBlocking = {
readonly blocked: (record: AnyTaskRecord) => BlockedReason | undefined;
readonly blocked: (record: BlockedSubject) => BlockedReason | undefined;
readonly subscribeBlocked: (listener: () => void) => () => void;
};
@ -179,7 +179,7 @@ export class TaskGraphView {
async #build(context: Context): Promise<Omit<Mount, "observers">> {
const tasks: Record<string, TaskGraphNode> = {};
const live = new Map<string, AnyTaskRecord>();
const live = new Map<string, BlockedSubject>();
const records: AnyTaskRecord[] = [];
for (const status of LIVE_STATUSES) {
records.push(
@ -192,8 +192,9 @@ export class TaskGraphView {
this.#storage.scanConversations({ ownerTaskId: record.id }, SCAN_PAGE_SIZE, cursor, context),
);
const conversations = owned.map((conversation) => conversation.id).sort((a, b) => a - b);
tasks[String(record.id)] = nodeOf(record, conversations, this.#blocking.blocked(record));
live.set(String(record.id), record);
const subject = blockedSubject(record);
tasks[String(record.id)] = nodeOf(record, conversations, this.#blocking.blocked(subject));
live.set(String(record.id), subject);
}
return { value: { tasks }, records: live };
}
@ -227,8 +228,9 @@ function advance(
changed.set(key, undefined);
continue;
}
mount.records.set(key, record);
const next = nodeOf(record, previous?.conversations ?? [], blocked(record));
const subject = blockedSubject(record);
mount.records.set(key, subject);
const next = nodeOf(record, previous?.conversations ?? [], blocked(subject));
if (previous !== undefined && JSON.stringify(next) === JSON.stringify(previous)) continue;
ops.push(["s", ["tasks", key], next as unknown as JsonValue]);
changed.set(key, next);

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

@ -30,6 +30,7 @@ import type {
Tx,
TypedEntry,
} from "../types.ts";
import { toolExtension } from "./agent.ts";
import { assignJson } from "./json.ts";
import {
clearProgress,
@ -560,7 +561,8 @@ async function finalResult(
} else {
// The model never sees structured output, so its call stands; the host learns of the broken tool.
final = rest;
runtime.report(new Error(broken));
const extension = toolExtension(await runtime.agent(context), tool.name);
runtime.report(new Error(broken), { source: "extension", ...(extension === undefined ? {} : { extension }) });
}
}
// Programs get the tool's own value: without a schema, its output. The model reads only `output`, so model-issued

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

@ -568,9 +568,9 @@ export type HarnessSettings = {
*/
readonly afterFailedRun?: AfterFailedRun;
/**
* How many times in a row a task may be cut off by a crash or close without changing its state before it is parked
* as blocked (`restart_limit`) instead of run again; `Harness.retryTask()` runs it again. Default 5;
* `Infinity` never parks.
* How many times in a row a task may be cut off by a crash, eviction, or reset (a close does not count) without
* changing its state before it is parked as blocked (`restart_limit`) instead of run again; `Harness.retryTask()`
* runs it again. Default 5; `Infinity` never parks.
*/
readonly restartLimit?: number;
};
@ -653,8 +653,9 @@ export type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
};
/**
* When a Session next needs a running process (`Harness.wake()`). `active`: some task can run now or is running, so a
* process that dies must be woken again soon, for example by a recovery alarm. `at`: the earliest sleep not yet due,
* When a Session next needs a running process (`Harness.wake()`). `active`: some task can run now or is running, or a
* reconcile pass such as open's recovery has not landed yet, so a process that dies must be woken again soon, for
* example by a recovery alarm. `at`: the earliest sleep not yet due,
* when there is one. Neither: nothing needs a process until new work is submitted. Tasks blocked on a missing or
* unmigratable definition, `completing` tasks, and waits on live tasks do not count; the work they wait for does.
*/

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

@ -169,6 +169,7 @@ export type {
NextTaskState,
Page,
PhaseHandler,
ReportFrom,
ReportInfo,
ReportSource,
RewindableConversationDocFamilyToken,

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

@ -551,6 +551,18 @@ export class SessionImpl implements Session {
return Promise.resolve();
}
/**
* Internal, for `beforeClose()`: write records straight to Storage, on the line, after admission is sealed. Nothing
* is published, since the Session is closing; a failed Session writes nothing.
*/
protected writeAtClose(writes: readonly StorageWrite[]): Promise<void> {
if (writes.length === 0) return Promise.resolve();
return this.#enqueue(async () => {
if (this.#failure !== undefined) return;
await this.#storage.commit(writes, BACKGROUND_CONTEXT);
});
}
/** Runs after close seals admission and before the line closes Storage; must not reject. */
protected beforeClose(): Promise<void> {
return Promise.resolve();

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

@ -145,6 +145,12 @@ export type TaskDocFamilyToken<T extends JsonObject, I extends JsonValue> = DocF
*/
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;
@ -152,8 +158,15 @@ export type ReportInfo = {
/** 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";
};
/**
@ -279,8 +292,12 @@ export interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver
* nothing is armed. Returns a function that disarms it. Wait for time between phases with `sleepUntil()`.
*/
setTimer(callback: () => void, delayMs: number): () => void;
/** Forward a non-fatal failure to `HarnessOptions.onReport`. */
report(error: unknown): 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.
*/
report(error: unknown, from?: ReportFrom): void;
/**
* Commit a sleep: the task holds no invocation until the Harness clock reaches `until`, then resumes at `checkpoint`.
* Resolves once the sleep is stored, not when the time comes; the phase returns right after. Returning
@ -629,8 +646,9 @@ type TaskRecordBase<I, R> = {
/** Wall-clock milliseconds of the change to `terminal`, stamped by the Session; absent while live. */
readonly endedAt?: number;
/**
* How many times in a row a crash or close cut the task off while it ran, without the task changing its state since
* (open counts them). At `HarnessSettings.restartLimit` the task is parked as blocked until `Harness.retryTask()`.
* How many times in a row a crash, eviction, or reset cut the task off while it ran, without the task changing its
* state since; open counts them, and a close does not. At `HarnessSettings.restartLimit` the task is parked as blocked
* until `Harness.retryTask()`.
*/
readonly restarts?: number;
};

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

@ -11,7 +11,7 @@ import {
} from "@earendil-works/pi-durable";
import { afterEach, describe, expect, it } from "vitest";
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
import { context } from "./session-support.ts";
import { context, flush } from "./session-support.ts";
import { completed, eventually, openTasks } from "./task-support.ts";
const directories = new Set<string>();
@ -45,16 +45,85 @@ function crashLoop(state: { crashing: boolean; started: number }) {
});
}
/**
* Open a Harness over `path`. `kill()` simulates a crash: every later call of its storage hangs, so it writes nothing
* more, not even the `pending` records a close writes for the tasks it stops.
*/
async function openWith(path: string, registry: Registry, settings: { restartLimit?: number } = {}) {
const wakes: Wake[] = [];
const { harness, reports } = await openTasks(await openNodeSqliteStorage(path), [], {
const inner = await openNodeSqliteStorage(path);
let dead = false;
const storage = new Proxy(inner, {
get(target, property) {
const value = Reflect.get(target, property, target);
if (typeof value !== "function") return value;
return (...args: unknown[]) => (dead ? new Promise(() => {}) : value.apply(target, args));
},
});
const { harness, reports } = await openTasks(storage, [], {
registry,
settings,
onWake: (wake) => wakes.push(wake),
});
return { harness, wakes, reports };
const kill = (): void => {
dead = true;
};
return { harness, wakes, reports, kill };
}
describe("recovery at open", () => {
it("keeps the host's wake active until open's recovery pass has landed", async () => {
const path = await sqlitePath();
const Held = defineTask<null, { phase: "run" }, null>({
name: "test.held",
version: 1,
initial: () => ({ phase: "run" }),
phases: { run: async () => {} },
abort: async () => {},
});
const registry = createRegistry();
registry.install(defineExtension({ name: "held", tasks: [Held] }));
const first = await openWith(path, registry);
const id = await (await first.harness.root(context)).commit(
(tx) => tx.createTask(Held, null, { ownership: { kind: "conversation" } }),
context,
);
await first.harness.close(context);
// A crash left the task holding its outcome: the next open's recovery pass makes it terminal.
const raw = await openNodeSqliteStorage(path);
const record = (await raw.task(id, context))!;
const { memos: _memos, ...rest } = record;
const completing = {
...rest,
state: { status: "completing", outcome: { status: "completed", result: null } },
} as const;
await raw.commit([{ type: "task", value: completing }], context);
await raw.close(context);
// That pass's commit fails, as on a platform reset.
const inner = await openNodeSqliteStorage(path);
let failNext = true;
const failing = new Proxy(inner, {
get(target, property) {
if (property === "commit") {
return (...args: Parameters<typeof inner.commit>) => {
if (!failNext) return target.commit(...args);
failNext = false;
return Promise.reject(new Error("reset"));
};
}
const value = Reflect.get(target, property, target);
return typeof value === "function" ? value.bind(target) : value;
},
});
const wakes: Wake[] = [];
const { harness } = await openTasks(failing, [], { registry, onWake: (wake) => wakes.push(wake) });
expect((await harness.closed).reason).toBe("failed");
// The host was told work is underway, and never that it ended: its recovery alarm stays.
expect(wakes).toEqual([{ active: true }]);
});
});
describe("tasks that cannot run", () => {
it("parks a task cut off restartLimit times in a row, shows why, and runs it again on retryTask()", async () => {
const path = await sqlitePath();
@ -69,7 +138,7 @@ describe("tasks that cannot run", () => {
for (let run = 1; run <= 3; run++) {
opened.harness.resume();
await eventually(() => state.started === run);
await opened.harness.close(context);
opened.kill(); // a crash: a close would not count
opened = await openWith(path, registry, { restartLimit: 3 });
}
const { harness, reports } = opened;
@ -83,9 +152,11 @@ describe("tasks that cannot run", () => {
const graph = await harness.taskGraph(context);
expect(graph.value.tasks[String(id)]?.blocked).toBe("restart_limit");
// Parked work needs no process: nothing is active.
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false });
harness.resume();
await new Promise((resolve) => setTimeout(resolve, 20));
// A scheduling pass runs within a few turns; it leaves the parked task out.
for (let turn = 0; turn < 5; turn++) await flush();
expect(state.started).toBe(3);
state.crashing = false;
@ -96,6 +167,98 @@ describe("tasks that cannot run", () => {
await harness.close(context);
});
it("counts no restart for a task a close stopped", async () => {
const path = await sqlitePath();
const state = { crashing: true, started: 0 };
const registry = createRegistry();
const Loop = crashLoop(state);
registry.install(defineExtension({ name: "loop", tasks: [Loop] }));
let opened = await openWith(path, registry, { restartLimit: 1 });
const id = await (await opened.harness.root(context)).commit(
(tx) => tx.createTask(Loop, null, { ownership: { kind: "conversation" } }),
context,
);
opened.harness.resume();
await eventually(() => state.started === 1);
await opened.harness.close(context);
opened = await openWith(path, registry, { restartLimit: 1 });
expect(await opened.harness.getTask(id, context)).toMatchObject({ state: { status: "pending" } });
expect((await opened.harness.getTask(id, context))?.restarts).toBeUndefined();
state.crashing = false;
expect((await opened.harness.waitForTask(id, context)).state.outcome.status).toBe("completed");
await opened.harness.close(context);
});
it("faults a task whose abort handler is cut off restartLimit times", async () => {
const path = await sqlitePath();
let aborts = 0;
const Stuck = defineTask<null, { phase: "run" }, null>({
name: "test.stuck-abort",
version: 1,
initial: () => ({ phase: "run" }),
phases: {
run: (_task, runtime) =>
new Promise((resolve) => runtime.signal.addEventListener("abort", () => resolve())),
},
abort: (_task, runtime) => {
aborts++;
return new Promise((resolve) => runtime.signal.addEventListener("abort", () => resolve()));
},
});
const registry = createRegistry();
registry.install(defineExtension({ name: "stuck", tasks: [Stuck] }));
let opened = await openWith(path, registry, { restartLimit: 1 });
const id = await (await opened.harness.root(context)).commit(
(tx) => tx.createTask(Stuck, null, { ownership: { kind: "conversation" } }),
context,
);
opened.harness.resume();
void opened.harness.abortTask(id, context);
await eventually(() => aborts === 1);
opened.kill(); // a crash in the abort handler
opened = await openWith(path, registry, { restartLimit: 1 });
const settled = await opened.harness.waitForTask(id, context);
expect(settled.state.outcome).toMatchObject({
status: "faulted",
error: { detail: { reason: "restart_limit" } },
});
expect(aborts).toBe(1);
await eventually(() => opened.reports.length > 0);
expect(opened.reports.map((error) => (error as Error).cause)).toEqual(["restart_limit"]);
await opened.harness.close(context);
});
it("runs a parked task once a raised restartLimit no longer parks it", async () => {
const path = await sqlitePath();
const state = { crashing: true, started: 0 };
const registry = createRegistry();
const Loop = crashLoop(state);
registry.install(defineExtension({ name: "loop", tasks: [Loop] }));
const settings: { restartLimit?: number } = { restartLimit: 1 };
let opened = await openWith(path, registry, settings);
const id = await (await opened.harness.root(context)).commit(
(tx) => tx.createTask(Loop, null, { ownership: { kind: "conversation" } }),
context,
);
opened.harness.resume();
await eventually(() => state.started === 1);
opened.kill(); // a crash: a close would not count
opened = await openWith(path, registry, settings);
const graph = await opened.harness.taskGraph(context);
opened.harness.resume();
await eventually(() => graph.value.tasks[String(id)]?.blocked === "restart_limit");
// Let a scheduling pass leave the parked task out.
for (let turn = 0; turn < 5; turn++) await flush();
state.crashing = false;
// The host's live settings change; the next pass runs the task.
settings.restartLimit = Number.POSITIVE_INFINITY;
opened.harness.resume();
expect((await opened.harness.waitForTask(id, context)).state.outcome.status).toBe("completed");
expect(state.started).toBe(2);
graph.dispose();
await opened.harness.close(context);
});
it("still runs the abort handler of a parked task", async () => {
const path = await sqlitePath();
const state = { crashing: true, started: 0 };
@ -116,7 +279,7 @@ describe("tasks that cannot run", () => {
);
opened.harness.resume();
await eventually(() => state.started === 1);
await opened.harness.close(context);
opened.kill(); // a crash: a close would not count
opened = await openWith(path, registry, { restartLimit: 1 });
await opened.harness.abortTask(id, context);
expect((await opened.harness.waitForTask(id, context)).state.outcome.status).toBe("aborted");
@ -125,6 +288,45 @@ describe("tasks that cannot run", () => {
await opened.harness.close(context);
});
it("does not count the same checkpoint committed again as progress", async () => {
const path = await sqlitePath();
let started = 0;
const Intent = defineTask<null, { phase: "run" }, null>({
name: "test.intent",
version: 1,
initial: () => ({ phase: "run" }),
phases: {
run: async (task, runtime, ctx) => {
started++;
// Intent before an effect, then the effect never returns: the checkpoint is unchanged.
await runtime.commit(() => ({ status: "running", checkpoint: task.state.checkpoint }), ctx);
await new Promise<void>((resolve) => runtime.signal.addEventListener("abort", () => resolve()));
},
},
abort: async () => {},
});
const registry = createRegistry();
registry.install(defineExtension({ name: "intent", tasks: [Intent] }));
let opened = await openWith(path, registry, { restartLimit: 2 });
const id = await (await opened.harness.root(context)).commit(
(tx) => tx.createTask(Intent, null, { ownership: { kind: "conversation" } }),
context,
);
for (let run = 1; run <= 2; run++) {
opened.harness.resume();
await eventually(() => started === run);
await eventually(async () => (await opened.harness.getTask(id, context))?.state.status === "running");
opened.kill(); // a crash: a close would not count
opened = await openWith(path, registry, { restartLimit: 2 });
}
expect((await opened.harness.getTask(id, context))?.restarts).toBe(2);
expect((await opened.harness.inspect(context)).tasks[0]?.state).toEqual({
kind: "blocked",
reason: "restart_limit",
});
await opened.harness.close(context);
});
it("starts the count over when the task makes progress", async () => {
const path = await sqlitePath();
let crashing = true;
@ -156,7 +358,7 @@ describe("tasks that cannot run", () => {
);
opened.harness.resume();
await eventually(() => started === 1);
await opened.harness.close(context);
opened.kill(); // a crash: a close would not count
opened = await openWith(path, registry, { restartLimit: 2 });
expect((await opened.harness.getTask(id, context))?.restarts).toBe(1);
// Progress: phase one commits phase two, which resets the count.
@ -167,6 +369,73 @@ describe("tasks that cannot run", () => {
await opened.harness.close(context);
});
it("tells the graph and the host when a migration fails", async () => {
const path = await sqlitePath();
const run = { phases: { run: async () => {} }, abort: async () => {} } as const;
const V1 = defineTask<null, { phase: "run" }, null>({
name: "test.migrating",
version: 1,
initial: () => ({ phase: "run" }),
...run,
});
const V2 = defineTask<null, { phase: "run" }, null>({
name: "test.migrating",
version: 2,
initial: () => ({ phase: "run" }),
...run,
migrate: () => {
throw new Error("cannot migrate");
},
});
const first = createRegistry();
first.install(defineExtension({ name: "v1", tasks: [V1] }));
let opened = await openWith(path, first);
// Created, never run: scheduling stays off.
const id = await (await opened.harness.root(context)).commit(
(tx) => tx.createTask(V1, null, { ownership: { kind: "conversation" } }),
context,
);
await opened.harness.close(context);
const second = createRegistry();
second.install(defineExtension({ name: "v2", tasks: [V2] }));
opened = await openWith(path, second);
const graph = await opened.harness.taskGraph(context);
opened.harness.resume();
await eventually(() => graph.value.tasks[String(id)]?.blocked === "migration_failed");
await eventually(() => !opened.harness.wake().active);
expect(opened.harness.wake()).toEqual({ active: false });
await eventually(() => opened.wakes.at(-1)?.active === false);
graph.dispose();
await opened.harness.close(context);
});
it("shows an abandoned task waiting for its definition as blocked", async () => {
const path = await sqlitePath();
const Abandoned = defineTask<null, { phase: "run" }, null>({
name: "test.abandoned",
version: 1,
initial: () => ({ phase: "run" }),
phases: { run: async () => {} },
abort: async () => {},
});
const registry = createRegistry();
registry.install(defineExtension({ name: "abandoned", tasks: [Abandoned] }));
let opened = await openWith(path, registry);
const id = await (await opened.harness.root(context)).commit(
(tx) => tx.createTask(Abandoned, null, { ownership: { kind: "conversation" }, abandonOnRestart: true }),
context,
);
await opened.harness.close(context);
// The next open marks it `restart`; without its definition, it waits for one to clean up.
opened = await openWith(path, createRegistry());
const graph = await opened.harness.taskGraph(context);
opened.harness.resume();
await eventually(async () => (await opened.harness.getTask(id, context))?.abortReason === "restart");
await eventually(() => graph.value.tasks[String(id)]?.blocked === "missing_task");
graph.dispose();
await opened.harness.close(context);
});
it("shows a task without a definition as blocked, and wakes the host once a registry change unblocks it", async () => {
const Missing = defineTask<null, { phase: "run" }, null>({
name: "test.missing",
@ -186,7 +455,7 @@ describe("tasks that cannot run", () => {
);
harness.resume();
await eventually(() => graph.value.tasks[String(id)]?.blocked === "missing_task");
expect(harness.wake()).toEqual({ active: false });
await eventually(() => !harness.wake().active);
registry.install(defineExtension({ name: "late", tasks: [Missing] }));
// The graph follows the registry; the host hears it can run work now.
await eventually(() => graph.value.tasks[String(id)]?.blocked === undefined);

21
packages/durable/test/harness-failed-run-inbox.test.ts
View file

@ -53,6 +53,27 @@ describe("inputs queued behind a failed run", () => {
await harness.close(context);
});
it("place queued writes, and keep them with withdraw", async () => {
for (const policy of ["continue", "withdraw"] as const) {
const setup = chatSetup();
setup.settings.afterFailedRun = policy;
const failing = failingOnce();
setup.faux.setResponses([failing.step, fauxAssistantMessage("second answer")]);
const { harness, root } = await openChat(new MemoryStorage(), setup);
const first = await root.submit({ type: "input", content: "first" }, context);
await failing.reached.promise;
const note = { role: "user", content: "a note the model sees", timestamp: 1 } as const;
const write = await root.submit({ type: "write", entry: { kind: "test.note", model: [note] } }, context);
expect(await write.status(context)).toMatchObject({ status: "queued" });
failing.gate.resolve();
await first.wait(context);
await harness.waitForIdle(context);
// `continue` places the write at the failed run's boundary; `withdraw` withdraws inputs only.
expect(await write.status(context)).toMatchObject({ status: policy === "continue" ? "done" : "queued" });
await harness.close(context);
}
});
it("settle unanswered with run_failed with withdraw", async () => {
const { harness, second } = await failWithQueuedInput("withdraw");
expect(await second.wait(context)).toMatchObject({ status: "unanswered", reason: "run_failed" });

4
packages/durable/test/harness-generation-recovery.test.ts
View file

@ -9,7 +9,7 @@ import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
import { allEntries, type ChatSetup, chatSetup, openChat, textOf, unanswered, waitFor } from "./chat-support.ts";
import { addSection } from "./harness-support.ts";
import { context } from "./session-support.ts";
import { aborted, deferred } from "./task-support.ts";
import { aborted, deferred, eventually } from "./task-support.ts";
const directories = new Set<string>();
@ -188,6 +188,7 @@ describe("generation recovery", () => {
checkpoint: { phase: "retry", attempt: 1, until: 61_000 },
until: 61_000,
});
await eventually(() => !opened.harness.wake().active);
expect(opened.harness.wake()).toEqual({ active: false, at: 61_000 });
// An early wake retries nothing.
opened.harness.resume();
@ -274,6 +275,7 @@ describe("generation recovery", () => {
status: "sleeping",
until: 61_000,
});
await eventually(() => !opened.harness.wake().active);
expect(opened.harness.wake()).toEqual({ active: false, at: 61_000 });
now = 61_000;
opened.harness.resume();

6
packages/durable/test/harness-registry.test.ts
View file

@ -285,8 +285,10 @@ describe("agent resolution", () => {
defineExtension<AppTool>({ name: "a", hooks: [hook(ToolTask, first), hook(GenerationTask, onYield)] }),
);
local.install(defineExtension<AppTool>({ name: "b", hooks: [hook(ToolTask, second)] }));
expect(agentHooks(resolve(undefined, local.snapshot()), "pi.tool")).toEqual([first, second]);
expect(agentHooks(resolve({ extensions: ["b", "a"] }, local.snapshot()), "pi.tool")).toEqual([second, first]);
const a = { extension: "a", handlers: first };
const b = { extension: "b", handlers: second };
expect(agentHooks(resolve(undefined, local.snapshot()), "pi.tool")).toEqual([a, b]);
expect(agentHooks(resolve({ extensions: ["b", "a"] }, local.snapshot()), "pi.tool")).toEqual([b, a]);
expect(agentHooks(resolve({ extensions: ["b"] }, local.snapshot()), "pi.generation")).toEqual([]);
const defaults = resolve(undefined, local.snapshot());

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

@ -1,17 +1,21 @@
import { createModels, Type } from "@earendil-works/pi-ai";
import { createModels, fauxAssistantMessage, Type } from "@earendil-works/pi-ai";
import {
type AnyTask,
createRegistry,
defineExtension,
defineTask,
defineTool,
GenerationTask,
Harness,
hook,
MemoryStorage,
type ReportInfo,
type SetTimer,
section,
wrapTool,
} from "@earendil-works/pi-durable";
import { describe, expect, it } from "vitest";
import { chatSetup } from "./chat-support.ts";
import { ControlledStorage, context } from "./session-support.ts";
import { completed, eventually } from "./task-support.ts";
@ -62,7 +66,8 @@ describe("reports", () => {
harness.resume();
expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("faulted");
await eventually(() => reports.length === 2);
const task = { source: "task", taskId: id, taskKind: "test.faulty", conversationId: root.id };
// 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" };
expect(reports).toEqual([
{ message: "noted", info: task },
{ message: "broken", info: { ...task, faulted: true } },
@ -70,6 +75,57 @@ describe("reports", () => {
await harness.close(context);
});
it("name the extension of a failing section or hook, and tell them from a failing environment builder", async () => {
const setup = chatSetup();
setup.faux.setResponses([fauxAssistantMessage("answer")]);
setup.registry.install(
defineExtension({
name: "sections",
sections: [
section("broken", () => {
throw new Error("section");
}),
],
}),
);
setup.registry.install(
defineExtension({
name: "hooks",
hooks: [
hook(GenerationTask, {
afterResponse: () => {
throw new Error("hook");
},
}),
],
}),
);
const reports: Report[] = [];
const harness = await Harness.open(
new MemoryStorage(),
{
models: setup.models,
registry: setup.registry,
env: () => {
throw new Error("env");
},
onReport: (error, info) => reports.push({ message: (error as Error).message, info }),
},
context,
);
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" });
await harness.close(context);
});
it("tell extension, listener, wake, and timer failures from a failed Session", async () => {
const Sleeper = defineTask<null, { phase: "run" } | { phase: "done" }, null>({
name: "test.sleeper",
@ -116,7 +172,11 @@ describe("reports", () => {
["alarm", "no timers"].every((message) => reports.some((report) => report.message === message)),
);
const infoOf = (message: string) => reports.find((report) => report.message === message)?.info;
expect(infoOf("Wrapper renamed echo to renamed")).toEqual({ source: "extension", conversationId: root.id });
expect(infoOf("Wrapper renamed echo to renamed")).toEqual({
source: "extension",
conversationId: root.id,
extension: "wraps",
});
expect(infoOf("listener")).toEqual({ source: "listener" });
expect(infoOf("alarm")).toEqual({ source: "wake" });
expect(infoOf("no timers")).toEqual({ source: "timer" });

9
packages/durable/test/harness-sleep.test.ts
View file

@ -106,6 +106,7 @@ describe("sleeping tasks", () => {
const inspection = await harness.inspect(context);
expect(inspection.tasks.find((task) => task.record.id === id)?.state).toEqual({ kind: "sleeping", until: 2_000 });
expect(graph.value.tasks[String(id)]?.state).toEqual({ status: "sleeping", phase: "after", until: 2_000 });
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false, at: 2_000 });
await settle();
expect(wakes.at(-1)).toEqual({ active: false, at: 2_000 });
@ -125,11 +126,11 @@ describe("sleeping tasks", () => {
expect(receipt.state.outcome).toEqual({ status: "completed", result: 2_000 });
expect(runs.after).toBe(1);
await settle();
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false });
expect(wakes.at(-1)).toEqual({ active: false });
// Only changes: idle at open, running, asleep until 2000, running again, idle.
for (let index = 1; index < wakes.length; index++) expect(wakes[index]).not.toEqual(wakes[index - 1]);
expect(wakes[0]).toEqual({ active: false });
expect(wakes).toContainEqual({ active: true });
expect(wakes).toContainEqual({ active: false, at: 2_000 });
graph.dispose();
@ -148,7 +149,8 @@ describe("sleeping tasks", () => {
await harness.close(context);
({ harness } = await open(await openNodeSqliteStorage(path)));
// Known from storage before scheduling starts.
// Known from storage before scheduling starts, once open's recovery pass is done.
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false, at: 61_000 });
harness.resume();
await settle();
@ -328,6 +330,7 @@ describe("sleeping tasks", () => {
const parked = hold ? "completing" : "waiting";
await eventually(async () => (await stateOf(harness, id))?.status === parked);
await eventually(() => harness.wake().at === 1_500);
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false, at: 1_500 });
await harness.close(context);
}
@ -405,9 +408,11 @@ describe("sleeping tasks", () => {
const root = await harness.root(context);
// Created through a token the registry does not hold: nothing can run it.
await root.commit((tx) => tx.createTask(Missing, null, { ownership: { kind: "conversation" } }), context);
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false });
harness.resume();
await settle();
await eventually(() => !harness.wake().active);
expect(harness.wake()).toEqual({ active: false });
await harness.close(context);
});

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

@ -1145,7 +1145,14 @@ describe("task close", () => {
proceed.resolve();
await closing;
expect(await aborting).toBe("marked");
expect(storage.commits.length).toBe(commits);
// No outcome: close writes only the stopped task back as pending, so a reopen counts no restart.
expect(storage.commits.length).toBe(commits + 1);
expect(storage.commits.at(-1)).toMatchObject([
{
type: "task",
value: { id, abortRequested: true, state: { status: "pending", checkpoint: { phase: "two" } } },
},
]);
expect(phases).toEqual(["one"]);
expect(abortRan).toBe(false);
expect(record).toMatchObject({
@ -1212,9 +1219,11 @@ describe("task close", () => {
const taskWrites = storage.commits.flatMap((writes) =>
writes.filter((write) => write.type === "task" && write.value.id === id),
);
// Created, reserved, and written back as pending by close: no fault.
expect(taskWrites.map((write) => (write.type === "task" ? write.value.state.status : ""))).toEqual([
"pending",
"running",
"pending",
]);
});
@ -1238,8 +1247,12 @@ describe("task close", () => {
held.release();
await closing;
expect(ran).toBe(false);
const writes = storage.commits.at(-1)!.filter((write) => write.type === "task");
expect(writes).toMatchObject([{ value: { id, abortRequested: true, state: { status: "running" } } }]);
// The reservation landed; close wrote the task back as pending.
const reserved = storage.commits.at(-2)!.filter((write) => write.type === "task");
expect(reserved).toMatchObject([{ value: { id, abortRequested: true, state: { status: "running" } } }]);
expect(storage.commits.at(-1)).toMatchObject([
{ type: "task", value: { id, abortRequested: true, state: { status: "pending" } } },
]);
});
it("rejects a runtime commit that was queued on the line when close sealed it", async () => {
@ -1272,7 +1285,12 @@ describe("task close", () => {
const taskWrites = storage.commits.flatMap((writes) =>
writes.filter((write) => write.type === "task" && write.value.id === id),
);
expect(taskWrites).toHaveLength(2);
// Created, reserved, and written back as pending by close; the queued outcome never landed.
expect(taskWrites.map((write) => (write.type === "task" ? write.value.state.status : ""))).toEqual([
"pending",
"running",
"pending",
]);
});
it("starts no next phase when close seals during a step that decided to continue", async () => {
@ -1334,7 +1352,9 @@ describe("task close", () => {
held.release();
await closing;
expect(ran).toBe(false);
const writes = storage.commits.at(-1)!.filter((write) => write.type === "task");
expect(writes).toMatchObject([{ type: "task", value: { id, state: { status: "running" } } }]);
// The reservation landed; close wrote the task back as pending.
const reserved = storage.commits.at(-2)!.filter((write) => write.type === "task");
expect(reserved).toMatchObject([{ type: "task", value: { id, state: { status: "running" } } }]);
expect(storage.commits.at(-1)).toMatchObject([{ type: "task", value: { id, state: { status: "pending" } } }]);
});
});