mirror of
https://github.com/earendil-works/pi.git
synced 2026-10-28 09:44:37 +08:00
fix(durable): end restart loops visibly; a host death charges a generation's attempt
A generation whose request or poll fetch was in flight at a host death counts one failed attempt of the retry policy (backoff, the partial kept, the same deferred handle polled again); without retries left the run ends with runFailure interrupted. restartLimit ends every other task visibly instead of parking it (tool error result, compactionFailure, runFailure restart_limit, faulted extension tasks) and reports task_interrupted; retryTask, task_parked and the restart_limit blocked reason are removed. closes #11139
This commit is contained in:
Mario Zechner committed
1 parent
f3255de765
commit
b3a29f1111
20 files changed
+1295
-246
No files matched your search
@@ -80,6 +80,8 @@
|
||||
| a run that ended without an answer (`run_end` with an `unanswered` `submission`) | `pi.live.runFailure`: `reason` (`model_error`, `no_model`, `aborted`, `reset`, `faulted`, `orphaned`), `detail`, `taskId`, `inputs` |
|
||||
|
||||
Some event data has no state in the view. A tool's arguments after repair and `beforeTool` at its start: in the nested slot's `arguments` for a nested call, only in the tool task's checkpoint for a model-issued one. A submission's settlement: `Submission.status()`/`wait()` and `Harness.submission(id)`; an input settled without a run (withdrawn, `stale`, aborted while queued, `run_failed`) just leaves `pi.inbox`. A failed task's error: `Harness.getTask()` and `onReport`; the task graph keeps only live tasks, not terminal outcomes. A nested call's full result: its caller's `NestedCallDoc` (`watchDoc()`), which retires when the caller settles. Exact turn boundaries: the run's `taskId` and the task graph, with no explicit event. Whether a compaction did anything: its `CompactionResult` (`waitForTask()`), and the summary entry when it is placed.
|
||||
- Restart-limited tasks no longer park: a task cut off `settings.restartLimit` times in a row ends instead, visibly. A tool call ends with an error result (diagnostic code `restart_limit`) and its task fails with detail `{ reason: "restart_limit" }`, a compaction fails into `pi.live.compactionFailure`, a generation ends its run with `pi.live.runFailure` reason `restart_limit` (its inputs settle `unanswered` with that reason), and a task of an extension's kind ends `faulted` with detail `{ reason: "restart_limit" }`, reported as `task_faulted` with reason `restart_limit`. The built-in tasks are reported as `task_interrupted` (below). An abort handler cut off at the limit still faults its task, but a run someone aborted now ends with `runFailure` reason `aborted`, the cut-off as `detail`, and an aborted compaction leaves no `compactionFailure`. `Harness.retryTask()`, the `restart_limit` blocked reason of `TaskInspection` and `TaskGraphNode`, and the `task_parked` report (and its telemetry value) are removed. Migration: a parked task kept its conversation busy until `retryTask()`; it now ends, so show `runFailure`, `compactionFailure`, the tool's error result, or the `task_interrupted` and `task_faulted` reports instead, and to run the work again, submit it again (a new input, `compact()`, or a new task); raise `restartLimit` before the task is next reserved to let it run longer ([#11139](https://github.com/earendil-works/pi/issues/11139)).
|
||||
- A generation whose model request was in flight when the host died no longer resends it on open: the attempt counts as failed under `settings.retry`. Its partial becomes an aborted entry, as before; the next attempt waits for the backoff, shown in `pi.live.generation.retry`, and with no retry left the run ends with `pi.live.runFailure` reason `interrupted` and the inputs settle `unanswered` with that reason. A deferred poll whose fetch was in flight is charged the same way, even when the replacement host's clock is behind, and its next attempt polls the same deferred response again instead of submitting a new request; a run that ends instead cancels it. A close still resends. `restartLimit` still bounds these phases: a charge that itself keeps being cut off ends the run `restart_limit`, past the limit without converting the partial ([#11139](https://github.com/earendil-works/pi/issues/11139)).
|
||||
|
||||
### Added
|
||||
|
||||
@@ -87,6 +89,8 @@
|
||||
- Synchronous observers for adapters, such as a Chord source over a Durable view: `Session.observeDoc(token, ...scope, observer, context)`, `Harness.observeConversation(id, observer, context)`, and `Harness.observeTaskGraph(observer, context)` attach an `Observer<T>` (`advance(change, context)`, `end(failure)`) atomically on the Session line, like a watch, and return an `Observation<T>` (`value`, `detach()`; `observeDoc` gives `undefined` for an absent document). `advance` gets every change unfolded as memory adopts it, for a stored commit inside Storage's `adopt` and for a deferred one before it is stored, so it must only record it: no awaits, no Session calls, no throws. A document observer gets one call per change at its address (two for a commit that retires and recreates it), the task graph also a call outside any commit when a registry change or a failed migration changes a `blocked` reason, and a change can arrive before the attaching promise resolves. The attaching context governs only the attachment; a closed Session or an invalid argument rejects the promise. One that throws is reported as `listener_error`, detached, and ended with the error. `end` comes once, at Session close or failure, or after a task document's retirement with its task. Document, view, and task graph watches are built on them (spec §9.1).
|
||||
- `ConversationTranscript` keeps what a client has of one conversation over any `ConversationView`: `merge(view)` returns the view with `items`, the client's whole transcript (entries, a `gap` where a later window did not continue it, and an `earlier` marker until the start is loaded), instead of the window; `loadEarlier(context)` and `fillGap(gap, context)` page entries through a loader such as `Conversation.entries()`, and `resume` is the window that continues it after a drop: `conversation.watch(context, transcript.resume)`. A merge costs O(items) and returns the same `items` array when it adds nothing; concurrent loads of the same page are shared ([#11327](https://github.com/earendil-works/pi/issues/11327)).
|
||||
- `pi.live.runFailure` (`RunFailure`, `RunFailureReason`) records why the newest run ended without an answer, `model_error`, `no_model`, `aborted`, `reset`, `faulted`, or `orphaned`, with the run's task and inputs and a `detail` where one exists (the error text, the missing model, the fault, the orphan reason). The commit that ends the run sets it; a later commit that starts a run removes it ([#11327](https://github.com/earendil-works/pi/issues/11327)).
|
||||
- `RunFailureReason` gains `interrupted` (the host stopped while the model was answering, with no retry left) and `restart_limit` (the run's task was cut off `settings.restartLimit` times in a row) ([#11139](https://github.com/earendil-works/pi/issues/11139)).
|
||||
- `onReport` gets `task_interrupted` (`reason` `restart_limit` or `interrupted`, `restarts`, and the task) when crashes end a built-in task: a tool call, a compaction, or a generation at the restart limit, or a run that ends `interrupted`. The telemetry report attributes gain the type `task_interrupted` and the reason `interrupted` ([#11139](https://github.com/earendil-works/pi/issues/11139)).
|
||||
- `HarnessOptions.readOnly` opens a Harness as a viewer of a session another process drives: open leaves running tasks as committed instead of recovering them, every commit with writes rejects with `ReadOnlySession` before it reaches Storage, and `resume()` and every call that would run tasks reject with it.
|
||||
|
||||
### Changed
|
||||
@@ -108,6 +112,7 @@
|
||||
- Fixed partial commits polluting `Object.prototype` when a streamed tool call's arguments gained a `__proto__` member after an earlier commit: the merge read the missing member through to `Object.prototype` and wrote into it
|
||||
- Fixed a commit that Storage rejects after adopting it (a failed JSONL reclaim) publishing nothing: it is published and its waits resolve before the Session fails, so a replica built from `subscribeCommits()` stays equal to Storage and `pi.durable.tasks.active` stays balanced.
|
||||
- Fixed a throwing telemetry adapter (a commit's `pi.durable.entry` or `pi.durable.submission` event, `pi.durable.tasks.active`) rejecting a durable commit and leaving the views and the scheduler without it: it is reported as `host_error` with callback `"telemetry"`, and the commit resolves. A throwing report metric or report event is dropped.
|
||||
- Fixed a host that crashes during a model request resending it forever, adding an aborted entry each time: the interrupted request is a failed attempt under the retry policy, and the run ends with `runFailure` reason `interrupted` once no retry is left; other work cut off `restartLimit` times ends instead of keeping its conversation busy ([#11139](https://github.com/earendil-works/pi/issues/11139))
|
||||
|
||||
## [1.4.0] - 2026-10-21
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
> **Experimental.** The API changes without notice between releases.
|
||||
|
||||
A durable agent harness. Conversations, model turns, tool calls, and your own state are committed to storage before anything is shown. If the process dies mid-turn, reopening the storage picks the work up where it stopped.
|
||||
A durable agent harness. Conversations, model turns, tool calls, and your own state are committed to storage before anything is shown. If the process dies mid-turn, reopening the storage picks the work up where it stopped; a model request it died during counts as a failed attempt and is retried, so a request that keeps killing the process ends (see [Persist and Resume](#persist-and-resume)).
|
||||
|
||||
Built on [`@earendil-works/pi-ai`](../ai/README.md) for model access and `@earendil-works/chord` for document state.
|
||||
|
||||
@@ -213,11 +213,11 @@ 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 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`.
|
||||
**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. A model request the host died during is a failed attempt under `settings.retry`: its partial answer is kept as an aborted entry, the next attempt waits for the backoff (shown in `docs["pi.live"].generation.retry`), and with no retry left the run ends with `docs["pi.live"].runFailure` reason `interrupted`. A deferred response that was being fetched is polled again as the next attempt, not requested again, and cancelled when the run ends. A close is not a crash: a request it stopped is sent again. Other work is counted: open counts each task a crash, eviction, or reset cut off, and a new state the task commits starts the count over. Once the count reaches `settings.restartLimit` (default 5; `Infinity` never ends a task for it), the task ends instead of running again: a tool call with an error result (diagnostic code `restart_limit`), a compaction failed into `compactionFailure`, a run with `runFailure` reason `restart_limit` (also when charging an interrupted request keeps crashing), and a task of your own kinds `faulted`, reported as `task_faulted` with reason `restart_limit`. The built-in ones are reported as `task_interrupted`, as is a run that ends `interrupted`, so a crash loop always reaches `onReport`. Every task running when a process dies is counted, not only the one that killed it. An abort handler cut off that many times ends its task `faulted` too; a run someone aborted still ends `aborted`, with the cut-off in `detail`.
|
||||
|
||||
**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.
|
||||
**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_interrupted` (crashes ended it, with `info.reason` and `info.restarts`), `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`.
|
||||
**Queued inputs after a failed run.** A run that fails with a model error, a missing model, or crashes (`interrupted`, `restart_limit`) 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`.
|
||||
|
||||
**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.
|
||||
|
||||
@@ -274,7 +274,7 @@ const count = defineTool({
|
||||
registry.install(defineExtension({ name: "count", tools: [count] }));
|
||||
```
|
||||
|
||||
Each call runs as its own durable task. Its intent is committed before `execute()` runs. If the process dies mid-call, the tool reruns on reopen only when it is declared `replay: "safe"`; otherwise the model gets an `interrupted` error result with the output committed so far. Throwing from `execute()` gives the model an error result. A result can also return `usage`, which is added to the conversation's [usage](#usage-and-cost). It can also return `control: { terminate: true }`: when every result of the round asks for it, the run ends without another model request.
|
||||
Each call runs as its own durable task. Its intent is committed before `execute()` runs. If the process dies mid-call, the tool reruns on reopen only when it is declared `replay: "safe"`; otherwise the model gets an `interrupted` error result with the output committed so far. A replay-safe call cut off `settings.restartLimit` times in a row ends with a `restart_limit` error result instead of running again. Throwing from `execute()` gives the model an error result. A result can also return `usage`, which is added to the conversation's [usage](#usage-and-cost). It can also return `control: { terminate: true }`: when every result of the round asks for it, the run ends without another model request.
|
||||
|
||||
A result has three channels: `output` for the model, `structuredOutput` for programs that call the tool (see [Calling tools from tools](#calling-tools-from-tools)), and `details` for UIs. A tool that declares `structuredOutputSchema` returns a matching `structuredOutput`, which the Harness validates. Programs calling a tool without one get its bounded output: one text item as a string, one image as its `ImageContent`, nothing as `""`, and anything else as the content list, for errors too. `structuredOutput` is always the tool's own value: a schema tool's error result keeps the value it returned, and the text of one without is in a `tool_error` diagnostic, as a thrown error's message is. The transcript stores only `output` and `details`:
|
||||
|
||||
@@ -538,7 +538,7 @@ await root.submit({ type: "write", entry: { kind: "app.note", data: "user opened
|
||||
|
||||
If a run fails, queued items stay in the inbox until the next input or queued write places them, oldest first.
|
||||
|
||||
When a run ends without an answer, `docs["pi.live"].runFailure` says why until the next run starts: `model_error` and `no_model` (with the error text or the missing model in `detail`), `aborted`, `reset`, or `faulted` and `orphaned` (the run's task could not go on). The run's inputs are in it too, settled `unanswered`; the settlement's reason is the same, except that an orphaned run's inputs carry why it was orphaned (`missing_task`, `task_too_old`, `migration_failed`) and `no_model` has no `detail` there (spec §8.2).
|
||||
When a run ends without an answer, `docs["pi.live"].runFailure` says why until the next run starts: `model_error` and `no_model` (with the error text or the missing model in `detail`), `aborted`, `reset`, `interrupted` (the host stopped while the model answered, with no retry left), or `faulted`, `orphaned`, and `restart_limit` (the run's task could not go on). The run's inputs are in it too, settled `unanswered`; the settlement's reason is the same, except that an orphaned run's inputs carry why it was orphaned (`missing_task`, `task_too_old`, `migration_failed`) and `no_model` has no `detail` there (spec §8.2).
|
||||
|
||||
An input is placed as a `pi.user` entry. To show it as something else, give it an `entry` with your kind and data; the model still sees `content` as the user message:
|
||||
|
||||
|
||||
@@ -477,7 +477,7 @@ type HarnessSettings = {
|
||||
readonly documentRetentionMs?: number;
|
||||
/** What a failed run does with the inputs queued behind it (section 6). */
|
||||
readonly afterFailedRun?: "keep" | "continue" | "withdraw";
|
||||
/** Restarts in a row without progress after which a task is parked (section 5.4); `Infinity` never parks. */
|
||||
/** Restarts in a row without progress after which a task ends instead of running again (section 5.4). */
|
||||
readonly restartLimit?: number;
|
||||
};
|
||||
|
||||
@@ -605,7 +605,7 @@ type TaskInspection = {
|
||||
| { readonly kind: "completing" }
|
||||
| {
|
||||
readonly kind: "blocked";
|
||||
readonly reason: "missing_task" | "task_too_old" | "migration_failed" | "restart_limit";
|
||||
readonly reason: "missing_task" | "task_too_old" | "migration_failed";
|
||||
readonly error?: unknown;
|
||||
};
|
||||
};
|
||||
@@ -624,7 +624,13 @@ type ReportInfo =
|
||||
readonly extension?: string;
|
||||
readonly reason: "threw" | "restart_limit";
|
||||
}
|
||||
| { readonly type: "task_parked"; readonly task: ReportTask; readonly extension?: string }
|
||||
| {
|
||||
readonly type: "task_interrupted";
|
||||
readonly task: ReportTask;
|
||||
readonly extension?: string;
|
||||
readonly reason: "restart_limit" | "interrupted";
|
||||
readonly restarts: number;
|
||||
}
|
||||
| { readonly type: "migration_failed"; readonly task: ReportTask; readonly extension?: string }
|
||||
| {
|
||||
readonly type: "task_outdated";
|
||||
@@ -712,8 +718,6 @@ interface Harness extends Session {
|
||||
conversationId?: ConversationId,
|
||||
): Promise<"aborted" | "already_placed" | "settled" | "not_found">;
|
||||
abortTask(id: TaskId, context: Context): Promise<"marked" | "terminal">;
|
||||
/** Run a task parked by the restart limit again (section 5.4). */
|
||||
retryTask(id: TaskId, context: Context): Promise<void>;
|
||||
waitForTask<R>(id: TaskId<R>, context: Context): Promise<SettledTask<R>>;
|
||||
waitForIdle(context: Context): Promise<void>;
|
||||
/** Session total of every conversation's `pi.usage` (section 8.6). */
|
||||
@@ -2360,7 +2364,7 @@ Every run invocation start resolves the task's definition by `TaskRecord.kind`
|
||||
from the scheduler's current registry snapshot:
|
||||
|
||||
```text
|
||||
restarts at restartLimit -> stay pending, blocked: restart_limit
|
||||
restarts at restartLimit -> a kind of an extension: terminal faulted (built-in tasks end themselves)
|
||||
definition missing -> stay pending, blocked: missing_task
|
||||
stored version newer than it -> stay pending, blocked: task_too_old
|
||||
stored version older than it -> migrate(input, checkpoint, storedVersion)
|
||||
@@ -2376,20 +2380,43 @@ after joining the invocations it stopped, close writes those tasks back as
|
||||
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 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
|
||||
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 as `task_faulted` with reason `restart_limit`.
|
||||
memos, and document writes are not. Work that deterministically exhausts memory
|
||||
or CPU before its next checkpoint thus shows as a growing count, and once
|
||||
`restarts` reaches `settings.restartLimit` (default 5; `Infinity` never ends a
|
||||
task for it), the task ends visibly instead of running again:
|
||||
|
||||
| task | at the limit |
|
||||
|---|---|
|
||||
| tool call (`pi.tool`) | ends with an error result, diagnostic code `restart_limit` ("Tool X was interrupted 3 times in a row; it is not run again"), built from its slot like an interrupted call's; its task fails with detail `{ reason: "restart_limit" }` |
|
||||
| compaction | fails with that error and detail, recorded in `pi.live.compactionFailure` |
|
||||
| generation | ends its run `unanswered` with `pi.live.runFailure` reason `restart_limit`, failing with that detail |
|
||||
| a kind of an extension | the scheduler ends it `faulted` with detail `{ reason: "restart_limit" }`, reported as `task_faulted` with reason `restart_limit` |
|
||||
|
||||
A built-in task that ends itself this way, and a generation whose run ends
|
||||
`interrupted` (section 8.3), is reported as `task_interrupted` with that reason,
|
||||
the task, and its `restarts`, so a host keeps a crash-loop signal. The built-in
|
||||
tasks check the count when a phase starts, a tool call after reading its call;
|
||||
reading the setting only when the task has a count, so the normal path does no
|
||||
more work. The scheduler checks an extension's task when it reserves it,
|
||||
reading the current setting, so a raised limit lets the task run; a task that
|
||||
already ended stays ended. A generation's request or deferred poll in flight
|
||||
when the host died is a failed attempt under the retry policy (section 8.3)
|
||||
first; the limit still bounds it, so a charge that itself keeps cutting the
|
||||
task off (converting a huge partial, say) ends the run. At the limit a task
|
||||
still does its usual cleanup, such as converting a partial or cancelling a
|
||||
deferred response; past it, after that cleanup was cut off too, it ends
|
||||
without. 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 end too. A nested
|
||||
call runs only while its replay-safe caller runs, so a crash counts both and
|
||||
the caller reaches the limit no later than its nested calls; the caller's own
|
||||
`restart_limit` result is what the model sees. An abort mark starts the count
|
||||
over and still runs the abort handler; an abort handler cut off `restartLimit`
|
||||
times is not run again, and the task ends `faulted` with detail
|
||||
`{ reason: "restart_limit" }`, reported as `task_faulted` with reason
|
||||
`restart_limit`. Someone aborted that task, so its conversation shows the
|
||||
abort: a run ends with `runFailure` reason `aborted`, the cut-off as `detail`,
|
||||
without converting its partial, and a compaction's status is removed without a
|
||||
failure.
|
||||
|
||||
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
|
||||
@@ -2403,9 +2430,8 @@ incompatible, neither at open nor later. A blocked task without an abort mark
|
||||
keeps its durable record unchanged, remains live, still blocks ordinary idle
|
||||
waits, and is reconsidered whenever the registry changes. The blocked reason is derived
|
||||
runtime state, not a persisted task status. A blocked task is not active in
|
||||
`wake()` (section 5.6): only a registry change, or `retryTask()` for a parked
|
||||
one, can let it run, and either makes `onWake` report it active, so a host need
|
||||
not poll it. The task graph shows the reason and follows registry changes
|
||||
`wake()` (section 5.6): only a registry change can let it run, and it makes
|
||||
`onWake` report it active, so a host need not poll it. The task graph shows the reason and follows registry changes
|
||||
(section 9.4). Document migration remains
|
||||
access-driven.
|
||||
|
||||
@@ -2740,8 +2766,8 @@ for the other submission type rejects. A busy input with `whenBusy: "reject"`
|
||||
writes no record and reports `ConversationBusy`. An idle input queues with mode
|
||||
`steer` when `whenBusy` is `steer` and `followUp` otherwise.
|
||||
|
||||
`settings.afterFailedRun` decides what a run that fails with `model_error` or
|
||||
`no_model` does with the inputs queued behind it, in the commit that settles its
|
||||
`settings.afterFailedRun` decides what a run that fails with `model_error`,
|
||||
`no_model`, `interrupted`, or `restart_limit` does with the inputs queued behind it, in the commit that settles its
|
||||
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
|
||||
@@ -3431,7 +3457,8 @@ error diagnostic already says what went wrong, so programs get it, for example
|
||||
diagnostic too. So a hook that redacts a schema-less tool's output redacts
|
||||
what programs get. An
|
||||
error result the Harness writes itself (unavailable, invalid arguments,
|
||||
blocked, interrupted, aborted, abandoned, or a scheduler fault) has no
|
||||
blocked, interrupted, aborted, abandoned, ended at the restart limit, or a
|
||||
scheduler fault) has no
|
||||
`structuredOutput`; its diagnostics say what went wrong. A thrown error
|
||||
becomes a `tool_error` diagnostic with its message. A nested call whose result breaks this gets an error result
|
||||
with an `invalid_structured_output` diagnostic, without the broken value: a
|
||||
@@ -3799,7 +3826,8 @@ retain their tool-result entry ID for the generation's `tools` phase.
|
||||
|
||||
Error results the Harness writes itself set `isError` and carry an `error`
|
||||
diagnostic with one of the codes `tool_unavailable`, `invalid_arguments`,
|
||||
`blocked`, `interrupted`, `aborted`, or `tool_error` (a throw from `execute()`)
|
||||
`blocked`, `interrupted`, `abandoned`, `restart_limit` (section 5.4),
|
||||
`aborted`, or `tool_error` (a throw from `execute()`)
|
||||
and the error text as its message. Their content is the durable partial output,
|
||||
if any, and `details` is the tool's last reported value, if any.
|
||||
|
||||
@@ -4267,7 +4295,7 @@ type RunFailure = {
|
||||
taskId: TaskId;
|
||||
/** The run's inputs, settled `unanswered`. */
|
||||
inputs: SubmissionId[];
|
||||
reason: "model_error" | "no_model" | "aborted" | "reset" | "faulted" | "orphaned";
|
||||
reason: "model_error" | "no_model" | "aborted" | "reset" | "faulted" | "orphaned" | "interrupted" | "restart_limit";
|
||||
/**
|
||||
* `model_error`: the error text; `no_model`: which model is missing; `faulted`: the error message; `orphaned`: why
|
||||
* the task could not run (`missing_task`, `task_too_old`, `migration_failed`).
|
||||
@@ -4351,10 +4379,12 @@ that settles its inputs `unanswered`:
|
||||
|---|---|---|
|
||||
| `model_error` | the response failed after any retries, or a blocking or overflow compaction it needed failed; `detail` is the error text | `model_error` with the text |
|
||||
| `no_model` | no model is configured, or the configured one is not available; `detail` names it | `no_model` |
|
||||
| `aborted` | its task was aborted (`Conversation.abort()`, `abortTask()`) | `aborted` |
|
||||
| `aborted` | its task was aborted (`Conversation.abort()`, `abortTask()`); `detail` says when its abort handler was cut off `restartLimit` times (section 5.4) | `aborted`, with that detail |
|
||||
| `reset` | a queued reset started a new context at the post-tools boundary (a tool's handoff continues the run) | `reset` |
|
||||
| `faulted` | the scheduler faulted its task (section 5.4); `detail` is the error message | `faulted` with the message |
|
||||
| `orphaned` | its task was aborted while no definition could take it; `detail` is the reason | the reason |
|
||||
| `interrupted` | the host stopped while the model was answering, with no retry left (section 8.3); `detail` names the attempt | `interrupted` with the text |
|
||||
| `restart_limit` | its generation was cut off `restartLimit` times in a row (section 5.4), also while an interrupted request was charged; `detail` says in which phase | `restart_limit` with the text |
|
||||
|
||||
The next commit that starts a run removes it. A run started in the same commit,
|
||||
the queue's next run under `afterFailedRun: "continue"` (section 6) or after a
|
||||
@@ -4469,7 +4499,22 @@ The run's inputs live in `pi.live.run`, not in the task input.
|
||||
`prepare`. `request` converts a leftover partial (below) before it resolves
|
||||
the model.
|
||||
- `request` first converts a committed partial left in `pi.live` by an
|
||||
interrupted attempt into an aborted `pi.assistant` entry. It then streams the
|
||||
interrupted attempt into an aborted `pi.assistant` entry. When the host died
|
||||
while this request was in flight (the record's `restarts`, section 5.4), the
|
||||
attempt failed: that commit also either sleeps at `retry` until
|
||||
`now + retryDelayMs(policy, attempt)`, with `pi.live.generation.retry` saying
|
||||
"The host stopped while the model was answering, on attempt N", when the
|
||||
retry policy allows another attempt, or settles the inputs `unanswered` and
|
||||
ends the run with `runFailure` reason `interrupted` and that text, failing the
|
||||
task. A request a close stopped is not counted and is sent again, below. A
|
||||
`poll` whose fetch was in flight is charged the same way ("The host stopped
|
||||
while the deferred answer was fetched"), checked before its `pollAt`, which a
|
||||
replacement host's clock may not have reached: its `retry` keeps the deferred
|
||||
response and polls it again as the next attempt instead of sending another
|
||||
request, and a run that ends instead cancels it, as the abort handler does.
|
||||
A generation asleep in `retry` or `poll`, or waiting for its tools, is not
|
||||
running and never charged. Past `restartLimit` the run ends `restart_limit`
|
||||
instead (section 5.4). Otherwise it streams the
|
||||
model context through `cutoff` with the invocation signal, the thinking level
|
||||
as `reasoning` (omitted for `off`), the conversation's persisted provider
|
||||
`sessionId`, and the pinned `streamOptions`, committing throttled partials at
|
||||
@@ -4477,9 +4522,10 @@ The run's inputs live in `pi.live.run`, not in the task input.
|
||||
(section 4.1), so Storage gets them at the next commit that reads `pi.live`
|
||||
or within `settings.progress.persistWithinMs`.
|
||||
Before streaming, the `beforeRequest` chain may
|
||||
replace the messages for this request only. Recovery resends the same
|
||||
committed messages with the same pinned model, thinking level, and stream
|
||||
options, and reruns `beforeRequest`.
|
||||
replace the messages for this request only. After a close, recovery resends
|
||||
the same committed messages with the same pinned model, thinking level, and
|
||||
stream options, and reruns `beforeRequest`; after a crash the attempt is
|
||||
charged instead, as above.
|
||||
- `afterResponse` observes every terminal message, from `request` or `poll`,
|
||||
before classification; a still deferred result is not terminal.
|
||||
- Before classifying, the handler stops the partial throttle and awaits any
|
||||
@@ -4790,7 +4836,7 @@ Phases:
|
||||
`sessionId`, and the pinned stream options without `deferred`, with
|
||||
`cacheRetention: "none"` and the pinned `maxTokens`. Providers such as Codex
|
||||
may suppress that identity when caching is disabled. Recovery resends the same
|
||||
request. The response is classified in one commit
|
||||
request, until the task was cut off `restartLimit` times in a row (section 5.4). The response is classified in one commit
|
||||
that adds its usage to `pi.usage` (section 8.6):
|
||||
- `stop` with non-empty text and no tool call: the text is the summary, placed
|
||||
as below.
|
||||
@@ -5288,7 +5334,7 @@ type TaskGraphNode = {
|
||||
/** Conversations this task owns, in ID order. */
|
||||
readonly conversations: readonly ConversationId[];
|
||||
/** Why it cannot run here, when it cannot (section 5.4); follows registry changes. */
|
||||
readonly blocked?: "missing_task" | "task_too_old" | "migration_failed" | "restart_limit";
|
||||
readonly blocked?: "missing_task" | "task_too_old" | "migration_failed";
|
||||
};
|
||||
|
||||
type TaskGraph = {
|
||||
@@ -5643,9 +5689,12 @@ views keep it.
|
||||
- `session_failed`: the error that failed the Session (a Storage error, often a
|
||||
platform reset, or the Session's own bookkeeping);
|
||||
- `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);
|
||||
`threw`), or it or its abort handler was cut off `restartLimit` times (reason
|
||||
`restart_limit`, section 5.4);
|
||||
- `task_interrupted`: crashes ended a task: a built-in task ended itself at the
|
||||
restart limit (reason `restart_limit`), or a generation's run ended because
|
||||
the host kept stopping while the model answered (reason `interrupted`), with
|
||||
the task's `restarts` (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`);
|
||||
|
||||
@@ -57,7 +57,7 @@ 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, rounded down (`Infinity` never parks); else the default. */
|
||||
/** A restart limit as given when it is a number of restarts, `1` or more, rounded down (`Infinity`: no task ends for restarts); else the default. */
|
||||
function restartLimit(limit: number | undefined): number {
|
||||
return typeof limit === "number" && limit >= 1 ? Math.floor(limit) : DEFAULT_RESTART_LIMIT;
|
||||
}
|
||||
|
||||
@@ -18,6 +18,8 @@ import { orderToolResults } from "./context.ts";
|
||||
import {
|
||||
addCompactionStatus,
|
||||
compactionStatus,
|
||||
cutOffCount,
|
||||
cutOffTooOften,
|
||||
failCompactionStatus,
|
||||
LiveDoc,
|
||||
type LiveState,
|
||||
@@ -115,6 +117,7 @@ export const CompactionTask = defineTask<CompactionInput, CompactionCheckpoint,
|
||||
initial: () => ({ phase: "select" }),
|
||||
phases: {
|
||||
select: async (task, runtime, context) => {
|
||||
if (cutOffTooOften(task, runtime)) return failCutOff(runtime, task, context);
|
||||
const { conversationId } = runtime;
|
||||
const agent = await runtime.agent(context);
|
||||
const settings = runtime.settings;
|
||||
@@ -155,6 +158,7 @@ export const CompactionTask = defineTask<CompactionInput, CompactionCheckpoint,
|
||||
await runtime.commit(() => ({ status: "running", checkpoint: { phase: "summarize", ...request } }), context);
|
||||
},
|
||||
summarize: async (task, runtime, context) => {
|
||||
if (cutOffTooOften(task, runtime)) return failCutOff(runtime, task, context);
|
||||
const { phase: _, ...request } = task.state.checkpoint;
|
||||
const { model: ref, thinkingLevel, streamOptions, maxTokens, tail, firstKept, attempt } = request;
|
||||
const model = runtime.models.getModel(ref.provider, ref.modelId);
|
||||
@@ -219,6 +223,7 @@ export const CompactionTask = defineTask<CompactionInput, CompactionCheckpoint,
|
||||
/** After the backoff's sleep; a run before `until` (an early wake, or a record of an earlier version) sleeps again. */
|
||||
retry: async (task, runtime, context) => {
|
||||
const { phase: _, until, ...request } = task.state.checkpoint;
|
||||
if (cutOffTooOften(task, runtime)) return failCutOff(runtime, task, context);
|
||||
if (until > runtime.now()) return runtime.sleepUntil(until, task.state.checkpoint, context);
|
||||
const attempt = request.attempt + 1;
|
||||
await runtime.commit(async (tx) => {
|
||||
@@ -473,6 +478,18 @@ async function complete(runtime: Runtime, context: Context): Promise<void> {
|
||||
}, context);
|
||||
}
|
||||
|
||||
/** Fail a compaction cut off `settings.restartLimit` times in a row (spec §5.4) instead of running it again. */
|
||||
async function failCutOff(runtime: Runtime, task: { readonly restarts?: number }, context: Context): Promise<void> {
|
||||
const message = `Compaction was interrupted ${cutOffCount(task)}`;
|
||||
await runtime.commit(async (tx) => {
|
||||
failCompactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId, message);
|
||||
return {
|
||||
status: "terminal",
|
||||
outcome: { status: "failed", error: { message, detail: { reason: "restart_limit" } } },
|
||||
};
|
||||
}, context);
|
||||
}
|
||||
|
||||
async function failNoModel(runtime: Runtime, ref: ModelRef | undefined, context: Context): Promise<void> {
|
||||
const message =
|
||||
ref === undefined ? "No model is configured" : `Model ${ref.provider}/${ref.modelId} is not available`;
|
||||
|
||||
@@ -30,7 +30,17 @@ import type {
|
||||
import { addTools, sectionExtension } from "./agent.ts";
|
||||
import { compactionPolicy, createCompaction, estimateContext, selectCut } from "./compaction.ts";
|
||||
import { applyBoundary, type Boundary, prepareBoundary, withdrawQueuedInputs } from "./inbox.ts";
|
||||
import { endRun, LiveDoc, type LiveState, PROGRESS_COMMIT, type RunEnd, type ToolSlot } from "./live.ts";
|
||||
import {
|
||||
cutOffCount,
|
||||
cutOffTooOften,
|
||||
endRun,
|
||||
LiveDoc,
|
||||
type LiveState,
|
||||
PROGRESS_COMMIT,
|
||||
type RunEnd,
|
||||
salvageable,
|
||||
type ToolSlot,
|
||||
} from "./live.ts";
|
||||
import { PartialChanges, type PartialMessage } from "./partial-changes.ts";
|
||||
import { planSystemEntries, renderSections, replaySections } from "./prompt.ts";
|
||||
import { ensureProviderSessionId } from "./provider.ts";
|
||||
@@ -74,7 +84,17 @@ export type GenerationCheckpoint =
|
||||
/** Newest entry included in the request. */
|
||||
cutoff: EntryId;
|
||||
}
|
||||
| { phase: "retry"; attempt: number; compacted?: TaskId<CompactionResult>; until: number }
|
||||
| {
|
||||
phase: "retry";
|
||||
attempt: number;
|
||||
compacted?: TaskId<CompactionResult>;
|
||||
until: number;
|
||||
/**
|
||||
* The deferred response a poll was fetching when the host died: the next attempt polls it again instead of
|
||||
* sending another request, so the provider's job is neither lost nor paid twice. Absent otherwise.
|
||||
*/
|
||||
poll?: { model: ModelRef; thinkingLevel?: ModelThinkingLevel; cutoff: EntryId; handle: DeferredHandle };
|
||||
}
|
||||
| {
|
||||
phase: "poll";
|
||||
attempt: number;
|
||||
@@ -133,6 +153,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
||||
* may land meanwhile.
|
||||
*/
|
||||
prepare: async (task, runtime, context) => {
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, context);
|
||||
const { conversationId } = runtime;
|
||||
const agent = await runtime.agent(context);
|
||||
const settings = runtime.settings;
|
||||
@@ -209,6 +230,10 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
||||
}, context);
|
||||
},
|
||||
request: async (task, runtime, context) => {
|
||||
// Charging the attempt may itself be what keeps failing, such as converting a huge partial: the limit ends it.
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, context);
|
||||
// The host stopped while this request was in flight: a failed attempt, not a resend (spec §8.3).
|
||||
if (task.restarts !== undefined) return interrupted(runtime, task.state.checkpoint, context);
|
||||
const { attempt, compacted, model: ref, thinkingLevel, streamOptions, cutoff } = task.state.checkpoint;
|
||||
const conversationId = runtime.conversationId;
|
||||
await runtime.commit(async (tx) => {
|
||||
@@ -238,21 +263,32 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
||||
},
|
||||
/** After the backoff's sleep; a run before `until` (an early wake, or a record of an earlier version) sleeps again. */
|
||||
retry: async (task, runtime, context) => {
|
||||
const { attempt, compacted, until } = task.state.checkpoint;
|
||||
const { attempt, compacted, until, poll } = task.state.checkpoint;
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, context);
|
||||
if (until > runtime.now()) return runtime.sleepUntil(until, task.state.checkpoint, context);
|
||||
await runtime.commit(async (tx) => {
|
||||
(await tx.doc(LiveDoc, runtime.conversationId)).generation = { attempt: attempt + 1 };
|
||||
const checkpoint: GenerationCheckpoint = {
|
||||
phase: "prepare",
|
||||
attempt: attempt + 1,
|
||||
...(compacted === undefined ? {} : { compacted }),
|
||||
};
|
||||
return { status: "running", checkpoint };
|
||||
const live = await tx.doc(LiveDoc, runtime.conversationId);
|
||||
const kept = compacted === undefined ? {} : { compacted };
|
||||
// A poll the host died during polls its response again, as the next attempt.
|
||||
if (poll !== undefined) {
|
||||
const pollAt = runtime.now();
|
||||
live.generation = { attempt: attempt + 1, deferred: { pollAt } };
|
||||
return {
|
||||
status: "running",
|
||||
checkpoint: { phase: "poll", attempt: attempt + 1, ...kept, ...poll, pollAt },
|
||||
};
|
||||
}
|
||||
live.generation = { attempt: attempt + 1 };
|
||||
return { status: "running", checkpoint: { phase: "prepare", attempt: attempt + 1, ...kept } };
|
||||
}, context);
|
||||
},
|
||||
/** After the sleep until `pollAt`; a run before it (an early wake, or a record of an earlier version) sleeps again. */
|
||||
poll: async (task, runtime, context) => {
|
||||
const { attempt, compacted, model: ref, thinkingLevel, cutoff, handle, pollAt } = task.state.checkpoint;
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, context);
|
||||
// The host stopped while the poll's fetch was in flight: a failed attempt, as for a request. Checked before the
|
||||
// time, which may have moved backwards on a replacement host; a sleeping poll is never cut off.
|
||||
if (task.restarts !== undefined) return interrupted(runtime, task.state.checkpoint, context);
|
||||
if (pollAt > runtime.now()) return runtime.sleepUntil(pollAt, task.state.checkpoint, context);
|
||||
const model = runtime.models.getModel(ref.provider, ref.modelId);
|
||||
if (model === undefined) return failNoModel(runtime, ref, context);
|
||||
@@ -264,6 +300,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
||||
await classify(runtime, request, message, context);
|
||||
},
|
||||
tools: async (task, runtime, context) => {
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, context);
|
||||
const { assistant, tools, pending } = task.state.checkpoint;
|
||||
const [next, ...rest] = pending;
|
||||
if (next === undefined) return finishToolRound(runtime, assistant, tools, context);
|
||||
@@ -285,19 +322,8 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
||||
},
|
||||
abort: async (task, runtime, context) => {
|
||||
const checkpoint = task.state.checkpoint;
|
||||
if (checkpoint.phase === "poll") {
|
||||
const model = runtime.models.getModel(checkpoint.model.provider, checkpoint.model.modelId);
|
||||
if (model !== undefined) {
|
||||
try {
|
||||
await runtime.models.cancelDeferred(model, checkpoint.handle, {
|
||||
signal: runtime.signal,
|
||||
telemetry: context.telemetry,
|
||||
});
|
||||
} catch (error) {
|
||||
runtime.report(error, { type: "host", callback: "models" });
|
||||
}
|
||||
}
|
||||
}
|
||||
const polled = deferredOf(checkpoint);
|
||||
if (polled !== undefined) await cancelDeferred(runtime, polled, context);
|
||||
const conversationId = runtime.conversationId;
|
||||
// Runs after the round's tool tasks are terminal; calls never started get `aborted` results (spec §8.5).
|
||||
const unstarted =
|
||||
@@ -388,6 +414,109 @@ async function failModelError(runtime: Runtime, text: string, context: Context):
|
||||
}, context);
|
||||
}
|
||||
|
||||
/**
|
||||
* The host stopped while this attempt's request or poll was in flight (`TaskRecord.restarts`): a failed attempt under
|
||||
* `settings.retry` (spec §8.3). The committed partial becomes an aborted entry; the next attempt follows the backoff, a
|
||||
* poll's by polling its deferred response again, or, with no retry left, the run ends `interrupted`.
|
||||
*/
|
||||
async function interrupted(
|
||||
runtime: Runtime,
|
||||
checkpoint: Extract<GenerationCheckpoint, { phase: "request" | "poll" }>,
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
const { attempt, compacted } = checkpoint;
|
||||
const policy = runtime.settings.retry;
|
||||
const retry = policy.enabled && attempt <= policy.maxRetries;
|
||||
const until = retry ? runtime.now() + retryDelayMs(policy, attempt) : 0;
|
||||
const polled = checkpoint.phase === "poll" ? checkpoint : undefined;
|
||||
const text =
|
||||
polled === undefined
|
||||
? `The host stopped while the model was answering, on attempt ${attempt}`
|
||||
: `The host stopped while the deferred answer was fetched, on attempt ${attempt}`;
|
||||
// A run that ends does not leave the provider's job running.
|
||||
if (!retry && polled !== undefined) await cancelDeferred(runtime, polled, context);
|
||||
await runtime.commit(async (tx): Promise<Next> => {
|
||||
// Read before the first table write; only a failure uses it.
|
||||
const failure = retry ? undefined : await prepareFailure(tx, runtime);
|
||||
const live = await tx.doc(LiveDoc, runtime.conversationId);
|
||||
await convertPartial(tx, live, runtime.conversationId);
|
||||
if (failure === undefined) {
|
||||
live.generation = { attempt, retry: { at: until, error: text } };
|
||||
const poll =
|
||||
polled === undefined
|
||||
? {}
|
||||
: {
|
||||
poll: {
|
||||
model: polled.model,
|
||||
...(polled.thinkingLevel === undefined ? {} : { thinkingLevel: polled.thinkingLevel }),
|
||||
cutoff: polled.cutoff,
|
||||
handle: polled.handle,
|
||||
},
|
||||
};
|
||||
const next = { phase: "retry", attempt, ...(compacted === undefined ? {} : { compacted }), until, ...poll };
|
||||
return { status: "sleeping", checkpoint: next as GenerationCheckpoint, until };
|
||||
}
|
||||
await endFailedRun(tx, runtime, live, failure, { status: "unanswered", reason: "interrupted", detail: text });
|
||||
return {
|
||||
status: "terminal",
|
||||
outcome: { status: "failed", error: { message: text, detail: { reason: "interrupted" } } },
|
||||
};
|
||||
}, context);
|
||||
}
|
||||
|
||||
/**
|
||||
* End a run whose generation was cut off `settings.restartLimit` times in a row (spec §5.4). At the limit it still
|
||||
* keeps a committed partial as an aborted entry and cancels a deferred response it polled; past the limit that cleanup
|
||||
* may be what cut it off, so it ends without it.
|
||||
*/
|
||||
async function endCutOff(
|
||||
runtime: Runtime,
|
||||
task: { readonly restarts?: number; readonly state: { readonly checkpoint: GenerationCheckpoint } },
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
const checkpoint = task.state.checkpoint;
|
||||
const text = `The generation was interrupted ${cutOffCount(task)} in its ${checkpoint.phase} phase`;
|
||||
const salvage = salvageable(task, runtime);
|
||||
const polled = deferredOf(checkpoint);
|
||||
if (salvage && polled !== undefined) await cancelDeferred(runtime, polled, context);
|
||||
await runtime.commit(async (tx): Promise<Next> => {
|
||||
const failure = await prepareFailure(tx, runtime);
|
||||
const live = await tx.doc(LiveDoc, runtime.conversationId);
|
||||
if (salvage) await convertPartial(tx, live, runtime.conversationId);
|
||||
await endFailedRun(tx, runtime, live, failure, { status: "unanswered", reason: "restart_limit", detail: text });
|
||||
return {
|
||||
status: "terminal",
|
||||
outcome: { status: "failed", error: { message: text, detail: { reason: "restart_limit" } } },
|
||||
};
|
||||
}, context);
|
||||
}
|
||||
|
||||
/** The deferred response a checkpoint polls, or will poll again after a retry's backoff. */
|
||||
function deferredOf(
|
||||
checkpoint: GenerationCheckpoint,
|
||||
): { readonly model: ModelRef; readonly handle: DeferredHandle } | undefined {
|
||||
if (checkpoint.phase === "poll") return checkpoint;
|
||||
return checkpoint.phase === "retry" ? checkpoint.poll : undefined;
|
||||
}
|
||||
|
||||
/** Cancel a deferred response no attempt will poll again; a failure is reported, never thrown. */
|
||||
async function cancelDeferred(
|
||||
runtime: Runtime,
|
||||
checkpoint: { readonly model: ModelRef; readonly handle: DeferredHandle },
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
const model = runtime.models.getModel(checkpoint.model.provider, checkpoint.model.modelId);
|
||||
if (model === undefined) return;
|
||||
try {
|
||||
await runtime.models.cancelDeferred(model, checkpoint.handle, {
|
||||
signal: runtime.signal,
|
||||
telemetry: context.telemetry,
|
||||
});
|
||||
} catch (error) {
|
||||
runtime.report(error, { type: "host", callback: "models" });
|
||||
}
|
||||
}
|
||||
|
||||
/** Settle the run's inputs `unanswered` with `no_model` and fail. */
|
||||
async function failNoModel(runtime: Runtime, ref: ModelRef | undefined, context: Context): Promise<void> {
|
||||
const message =
|
||||
|
||||
@@ -43,7 +43,7 @@ import { AgentDoc, configure, createAgent, resolveAgent, resolveSettings } from
|
||||
import { createCompaction } from "./compaction.ts";
|
||||
import { readContext } from "./context.ts";
|
||||
import { InboxDoc, withdrawQueuedInputs } from "./inbox.ts";
|
||||
import { LiveDoc, settleSchedulerOutcome } from "./live.ts";
|
||||
import { LIMITS_RESTARTS, LiveDoc, settleSchedulerOutcome } from "./live.ts";
|
||||
import { ProviderDoc } from "./provider.ts";
|
||||
import { BUILTIN_TASKS } from "./registry.ts";
|
||||
import { type InvocationBinding, TaskScheduler } from "./scheduler.ts";
|
||||
@@ -245,6 +245,7 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
|
||||
timers: hostTimers(options.setTimer, (error) => report(error, { type: "host_error", callback: "setTimer" })),
|
||||
writeAtClose: (writes) => this.writeAtClose(writes),
|
||||
report: this.#report,
|
||||
limitsRestarts: (kind) => LIMITS_RESTARTS.has(kind),
|
||||
settleOutcome: settleSchedulerOutcome,
|
||||
withdrawInputs: withdrawQueuedInputs,
|
||||
conversation: async (id, binding, callContext) => {
|
||||
@@ -343,11 +344,6 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
|
||||
return this.#tasks.abort(id, context);
|
||||
}
|
||||
|
||||
retryTask(id: TaskId, context: Context): Promise<void> {
|
||||
this.assertUsable();
|
||||
return this.#tasks.retry(id, context);
|
||||
}
|
||||
|
||||
async waitForTask<R>(id: TaskId<R>, context: Context): Promise<SettledTask<R>> {
|
||||
this.#tasks.resume();
|
||||
return this.#tasks.waitForTask(id, context) as Promise<SettledTask<R>>;
|
||||
@@ -471,12 +467,12 @@ function safeReport(
|
||||
};
|
||||
switch (info.type) {
|
||||
case "task_faulted":
|
||||
case "task_interrupted":
|
||||
case "task_outdated":
|
||||
attributes.reason = info.reason;
|
||||
attributes.extension = info.extension;
|
||||
attributes["task.id"] = info.task.id;
|
||||
break;
|
||||
case "task_parked":
|
||||
case "migration_failed":
|
||||
case "task_error":
|
||||
case "extension_error":
|
||||
|
||||
@@ -91,13 +91,25 @@ export type CompactionFailure = {
|
||||
* Why a run ended without an answer:
|
||||
* - `model_error`: the model's response failed, after any retries, or the compaction a request needed failed;
|
||||
* - `no_model`: no model is configured, or the configured one is not available;
|
||||
* - `aborted`: the run was aborted (`Conversation.abort()`, `abortTask()`);
|
||||
* - `aborted`: the run was aborted (`Conversation.abort()`, `abortTask()`); `detail` says when its abort handler was
|
||||
* cut off `settings.restartLimit` times and did not finish;
|
||||
* - `reset`: a queued reset started a new context at the post-tools boundary, before an answer (a tool's handoff
|
||||
* continues the run);
|
||||
* - `faulted`: the run's task threw where it cannot recover, or storage rejected its commit;
|
||||
* - `orphaned`: the run's task was aborted while no registered definition could take it.
|
||||
* - `orphaned`: the run's task was aborted while no registered definition could take it;
|
||||
* - `interrupted`: the host stopped while the model was answering, more often than `settings.retry` retries;
|
||||
* - `restart_limit`: the run's generation was cut off `settings.restartLimit` times in a row; a request or poll it was
|
||||
* cut off during counts as a failed attempt first, so this ends one whose charge itself keeps failing.
|
||||
*/
|
||||
export type RunFailureReason = "model_error" | "no_model" | "aborted" | "reset" | "faulted" | "orphaned";
|
||||
export type RunFailureReason =
|
||||
| "model_error"
|
||||
| "no_model"
|
||||
| "aborted"
|
||||
| "reset"
|
||||
| "faulted"
|
||||
| "orphaned"
|
||||
| "interrupted"
|
||||
| "restart_limit";
|
||||
|
||||
/** The newest run that ended without an answer (spec §8.2). */
|
||||
export type RunFailure = {
|
||||
@@ -108,7 +120,8 @@ export type RunFailure = {
|
||||
reason: RunFailureReason;
|
||||
/**
|
||||
* `model_error`: the error text; `no_model`: which model is missing; `faulted`: the error message; `orphaned`: why
|
||||
* the task could not run (`missing_task`, `task_too_old`, `migration_failed`).
|
||||
* the task could not run (`missing_task`, `task_too_old`, `migration_failed`); `interrupted` and `restart_limit`: what
|
||||
* was cut off, and how often.
|
||||
*/
|
||||
detail?: string;
|
||||
};
|
||||
@@ -179,6 +192,38 @@ const RUN_TASK_KINDS: ReadonlySet<string> = new Set(["pi.generation"]);
|
||||
const TOOL_TASK_KIND = "pi.tool";
|
||||
const COMPACTION_TASK_KIND = "pi.compaction";
|
||||
|
||||
/**
|
||||
* Built-in task kinds that end themselves once cut off `settings.restartLimit` times (`TaskRecord.restarts`); the
|
||||
* scheduler faults a task of any other kind instead (spec §5.4).
|
||||
*/
|
||||
export const LIMITS_RESTARTS: ReadonlySet<string> = new Set([...RUN_TASK_KINDS, TOOL_TASK_KIND, COMPACTION_TASK_KIND]);
|
||||
|
||||
/** Whether `task` was cut off `settings.restartLimit` times in a row, so its phase ends it instead of running. */
|
||||
export function cutOffTooOften(
|
||||
task: { readonly restarts?: number },
|
||||
runtime: { readonly settings: { readonly restartLimit: number } },
|
||||
): boolean {
|
||||
// The settings are read only for a task that was cut off at all, so the normal path resolves nothing more.
|
||||
return task.restarts !== undefined && task.restarts >= runtime.settings.restartLimit;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a task at the limit may still do the cleanup it does when it ends, such as converting a partial: only at the
|
||||
* limit. Past it, that cleanup may be what cut the task off, so the task ends without it.
|
||||
*/
|
||||
export function salvageable(
|
||||
task: { readonly restarts?: number },
|
||||
runtime: { readonly settings: { readonly restartLimit: number } },
|
||||
): boolean {
|
||||
return (task.restarts ?? 0) <= runtime.settings.restartLimit;
|
||||
}
|
||||
|
||||
/** How often a task was cut off, for a message: "once", or "3 times in a row". */
|
||||
export function cutOffCount(task: { readonly restarts?: number }): string {
|
||||
const restarts = task.restarts ?? 0;
|
||||
return restarts === 1 ? "once" : `${restarts} times in a row`;
|
||||
}
|
||||
|
||||
/** How a run ends: with an answer, or without one, settling its inputs `unanswered` and recording why. */
|
||||
export type RunEnd =
|
||||
| { readonly status: "done"; readonly answer: EntryId }
|
||||
@@ -319,21 +364,28 @@ export async function settleSchedulerOutcome(
|
||||
removeNestedSlots(live, record.id);
|
||||
return;
|
||||
}
|
||||
// Only an abort handler is faulted for the limit here (the built-in phases end themselves): the task was aborted,
|
||||
// and keeps that meaning. Its run ends `aborted`, with the limit as detail, and an aborted compaction is no failure.
|
||||
const detail = outcome.status === "faulted" ? outcome.error.detail : undefined;
|
||||
const cutOff =
|
||||
typeof detail === "object" && detail !== null && "reason" in detail && detail.reason === "restart_limit";
|
||||
if (record.kind === COMPACTION_TASK_KIND) {
|
||||
const error = outcome.status === "faulted" ? outcome.error.message : outcome.reason;
|
||||
failCompactionStatus(await tx.doc(LiveDoc, record.conversationId), record.id, error);
|
||||
const live = await tx.doc(LiveDoc, record.conversationId);
|
||||
if (cutOff) removeCompactionStatus(live, record.id);
|
||||
else failCompactionStatus(live, record.id, outcome.status === "faulted" ? outcome.error.message : outcome.reason);
|
||||
return;
|
||||
}
|
||||
if (!RUN_TASK_KINDS.has(record.kind)) return;
|
||||
const live = await tx.doc(LiveDoc, record.conversationId);
|
||||
if (live.run?.taskId !== record.id) return;
|
||||
await convertPartial(tx, live, record.conversationId);
|
||||
// Cut off, the abort handler may have died converting the partial: the run ends without it.
|
||||
if (!cutOff) await convertPartial(tx, live, record.conversationId);
|
||||
endRun(
|
||||
tx,
|
||||
live,
|
||||
record.id,
|
||||
outcome.status === "faulted"
|
||||
? { status: "unanswered", reason: "faulted", detail: outcome.error.message }
|
||||
? { status: "unanswered", reason: cutOff ? "aborted" : "faulted", detail: outcome.error.message }
|
||||
: {
|
||||
status: "unanswered",
|
||||
reason: "orphaned",
|
||||
|
||||
@@ -31,6 +31,7 @@ import type {
|
||||
TaskDefinition,
|
||||
TaskId,
|
||||
TaskOutcome,
|
||||
TaskOutcomeError,
|
||||
TaskRecord,
|
||||
TaskRuntime,
|
||||
TaskState,
|
||||
@@ -68,10 +69,7 @@ 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"
|
||||
> & {
|
||||
export type BlockedSubject = Pick<AnyTaskRecord, "id" | "kind" | "version" | "abortRequested" | "abortReason"> & {
|
||||
readonly status: TaskState<JsonValue, JsonValue>["status"];
|
||||
};
|
||||
|
||||
@@ -83,16 +81,12 @@ export function blockedSubject(record: AnyTaskRecord): BlockedSubject {
|
||||
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`.
|
||||
*/
|
||||
export type BlockedReason = "missing_task" | "task_too_old" | "migration_failed" | "restart_limit";
|
||||
/** Why a task cannot be reserved: no registered definition can take it. Derived from the registry. */
|
||||
export type BlockedReason = "missing_task" | "task_too_old" | "migration_failed";
|
||||
|
||||
type Resolution =
|
||||
| { readonly kind: "ready"; readonly task: AnyTask; readonly record: RunnableTaskRecord }
|
||||
@@ -215,6 +209,11 @@ export type TaskSchedulerOptions = {
|
||||
/** While closing, on the line after admitted commits settle, write what the callback returns straight to Storage. */
|
||||
readonly writeAtClose: (writes: () => readonly StorageWrite[]) => Promise<void>;
|
||||
readonly report: (error: unknown, info: ReportInfo, context?: Context) => void;
|
||||
/**
|
||||
* Task kinds whose run phases end the task themselves once it was cut off `settings.restartLimit` times (the
|
||||
* built-in tasks, from `TaskRecord.restarts`); the scheduler ends a task of any other kind `faulted` instead.
|
||||
*/
|
||||
readonly limitsRestarts: (kind: string) => boolean;
|
||||
/** 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>;
|
||||
/** Withdraw a conversation's queued inputs, for conversation abort and abort cascades. */
|
||||
@@ -274,6 +273,7 @@ export class TaskScheduler {
|
||||
readonly #writeAtClose: (writes: () => readonly StorageWrite[]) => Promise<void>;
|
||||
readonly #report: (error: unknown, info: ReportInfo, context?: Context) => void;
|
||||
readonly #settleOutcome: TaskSchedulerOptions["settleOutcome"];
|
||||
readonly #limitsRestarts: TaskSchedulerOptions["limitsRestarts"];
|
||||
readonly #withdrawInputs: TaskSchedulerOptions["withdrawInputs"];
|
||||
readonly #conversation: TaskSchedulerOptions["conversation"];
|
||||
readonly #context: Context;
|
||||
@@ -337,8 +337,6 @@ export class TaskScheduler {
|
||||
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 with work (they may overlap): their commits finalize held outcomes and mark cascades. */
|
||||
#reconcilingWork = 0;
|
||||
@@ -369,6 +367,7 @@ export class TaskScheduler {
|
||||
this.#writeAtClose = options.writeAtClose;
|
||||
this.#report = options.report;
|
||||
this.#settleOutcome = options.settleOutcome;
|
||||
this.#limitsRestarts = options.limitsRestarts;
|
||||
this.#withdrawInputs = options.withdrawInputs;
|
||||
this.#conversation = options.conversation;
|
||||
this.#context = options.context;
|
||||
@@ -376,9 +375,9 @@ export class TaskScheduler {
|
||||
}
|
||||
|
||||
/**
|
||||
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing. A
|
||||
* task that reaches the restart limit this way is parked and reported. Read-only, it only loads them: a `running`
|
||||
* task stays as committed, running in another process.
|
||||
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
|
||||
* a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
|
||||
* `running` task stays as committed, running in another process.
|
||||
*/
|
||||
async open(context: Context): Promise<void> {
|
||||
this.#session.observeCommits((publication) => this.#observe(publication));
|
||||
@@ -387,8 +386,6 @@ export class TaskScheduler {
|
||||
this.#notifyBlocked();
|
||||
this.#kick();
|
||||
});
|
||||
const limit = this.#restartLimit();
|
||||
const parked: AnyTaskRecord[] = [];
|
||||
await this.#session.commitWith(async (tx) => {
|
||||
// Every table read before the first write.
|
||||
const scans = [];
|
||||
@@ -405,7 +402,6 @@ export class TaskScheduler {
|
||||
const restarts = (record.restarts ?? 0) + 1;
|
||||
const pending = withState(record, { status: "pending", checkpoint: record.state.checkpoint });
|
||||
tx.setTask({ ...pending, restarts } as AnyTaskRecord);
|
||||
if (restarts >= limit && !record.abortRequested) parked.push(record);
|
||||
}
|
||||
if (record.state.status === "waiting" && record.state.policy === "failFast") {
|
||||
this.#failFastChecks.add(record.id);
|
||||
@@ -413,13 +409,6 @@ export class TaskScheduler {
|
||||
}
|
||||
}
|
||||
}, context);
|
||||
for (const record of parked) {
|
||||
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()`,
|
||||
);
|
||||
this.#report(error, { type: "task_parked", ...this.#aboutTask(recordTask(record)) });
|
||||
}
|
||||
if (this.#readOnly) return;
|
||||
// Derive abort marks a crash left unapplied below cancelled owners, and finalize held outcomes.
|
||||
this.#cascadePending = true;
|
||||
@@ -439,30 +428,14 @@ export class TaskScheduler {
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a task parked by the restart limit again: clears its restart count and enables scheduling. Does nothing for a
|
||||
* task that is not parked; rejects for an unknown one.
|
||||
*/
|
||||
async retry(id: TaskId, context: Context): Promise<void> {
|
||||
await this.#session.commitWith(async (tx) => {
|
||||
const current = await tx.task(id);
|
||||
if (current === undefined) throw new Error(`Task ${id} does not exist`);
|
||||
const status = current.state.status;
|
||||
if (status === "terminal" || status === "completing" || !this.#parked(current)) return;
|
||||
tx.setTask(progressed(current));
|
||||
}, context);
|
||||
this.resume();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Why a live task cannot run here, if it cannot: its definition is missing, too old, or failed to migrate. A running
|
||||
* task, an abort-marked one, and one holding its outcome are never blocked.
|
||||
*/
|
||||
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();
|
||||
@@ -506,7 +479,6 @@ export class TaskScheduler {
|
||||
continue;
|
||||
}
|
||||
if (active) continue;
|
||||
if (this.#parked(record)) continue;
|
||||
// An abort invocation waits for the live work below it, which counts itself.
|
||||
if (record.abortRequested && this.#waitingOn(record, true).length > 0) continue;
|
||||
// A task no definition can take waits for a registry change, except an abort-marked one, which is orphaned.
|
||||
@@ -1395,11 +1367,6 @@ export class TaskScheduler {
|
||||
}
|
||||
}
|
||||
|
||||
/** A task cut off `restartLimit` times in a row; an abort mark still runs its abort handler. */
|
||||
#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.
|
||||
@@ -1425,20 +1392,6 @@ export class TaskScheduler {
|
||||
return { task: reportTask(task), ...optionalExtension(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 {
|
||||
@@ -1515,7 +1468,6 @@ 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]) {
|
||||
@@ -1534,14 +1486,6 @@ export class TaskScheduler {
|
||||
this.#runnable.delete(id);
|
||||
continue;
|
||||
}
|
||||
// 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;
|
||||
const mode = record.abortRequested ? "abort" : "run";
|
||||
// Work below an owner with cancellation intent waits for its cascade mark instead of running a phase; schedule
|
||||
@@ -1551,9 +1495,15 @@ 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`;
|
||||
// Work cut off `restartLimit` times in a row would run forever: an abort handler, or a phase of a kind that
|
||||
// does not end itself at the limit, ends the task `faulted` instead.
|
||||
// The settings are read only for a task that was cut off at all.
|
||||
const cutOff = record.restarts !== undefined && record.restarts >= this.#restartLimit();
|
||||
if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
|
||||
const message =
|
||||
mode === "abort"
|
||||
? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times`
|
||||
: `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
|
||||
await this.#terminate(tx, record, {
|
||||
status: "faulted",
|
||||
error: { message, detail: { reason: "restart_limit" } },
|
||||
@@ -1689,7 +1639,6 @@ export class TaskScheduler {
|
||||
const on = this.#waitingOn(record);
|
||||
if (on.length > 0) return { kind: "waiting", on };
|
||||
if (this.#notDue(record)) return { kind: "sleeping", until: record.state.until };
|
||||
if (this.#parked(record)) return { kind: "blocked", reason: "restart_limit" };
|
||||
const fit = this.#fit(record, snapshot.task(record.kind));
|
||||
if ("reason" in fit) return { kind: "blocked", ...fit };
|
||||
if (fit.migrates && fit.task.definition.migrate === undefined) {
|
||||
@@ -2165,6 +2114,7 @@ export class TaskScheduler {
|
||||
},
|
||||
commit: async (change, context, options) => {
|
||||
let checkpoint: string | undefined;
|
||||
let interrupted: { readonly error: TaskOutcomeError; readonly restarts: number } | undefined;
|
||||
await this.#gated(
|
||||
invocation,
|
||||
async (tx, current) => {
|
||||
@@ -2173,10 +2123,28 @@ export class TaskScheduler {
|
||||
if (next?.status === "running" && current.state.checkpoint.phase !== next.checkpoint.phase) {
|
||||
checkpoint = next.checkpoint.phase;
|
||||
}
|
||||
// A task that ends for crashes says so (`task_interrupted`), so a host keeps a crash-loop signal.
|
||||
if (next?.status === "terminal" && next.outcome.status === "failed") {
|
||||
if (interruptionOf(next.outcome.error) !== undefined) {
|
||||
interrupted = { error: next.outcome.error, restarts: current.restarts ?? 0 };
|
||||
}
|
||||
}
|
||||
},
|
||||
context,
|
||||
options,
|
||||
);
|
||||
if (interrupted !== undefined) {
|
||||
this.#report(
|
||||
new Error(interrupted.error.message),
|
||||
{
|
||||
type: "task_interrupted",
|
||||
...this.#aboutTask(invocation),
|
||||
reason: interruptionOf(interrupted.error)!,
|
||||
restarts: interrupted.restarts,
|
||||
},
|
||||
invocation.context,
|
||||
);
|
||||
}
|
||||
if (checkpoint !== undefined) {
|
||||
invocation.context.telemetry.addEvent("pi.durable.checkpoint", {
|
||||
phase: checkpoint,
|
||||
@@ -2427,6 +2395,13 @@ function recordTask(record: Pick<AnyTaskRecord, "id" | "kind" | "conversationId"
|
||||
* 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.
|
||||
*/
|
||||
/** Why a failed outcome's error says crashes ended the task (`detail.reason`), if it does. */
|
||||
function interruptionOf(error: TaskOutcomeError): "restart_limit" | "interrupted" | undefined {
|
||||
const detail = error.detail;
|
||||
if (typeof detail !== "object" || detail === null || !("reason" in detail)) return undefined;
|
||||
return detail.reason === "restart_limit" || detail.reason === "interrupted" ? detail.reason : undefined;
|
||||
}
|
||||
|
||||
function withAbortMark(record: AnyTaskRecord, reason?: "restart"): AnyTaskRecord {
|
||||
const { abortReason: _reason, restarts: _restarts, ...rest } = record;
|
||||
return { ...rest, abortRequested: true, ...(reason === undefined ? {} : { abortReason: reason }) } as AnyTaskRecord;
|
||||
|
||||
@@ -40,7 +40,7 @@ export type TaskGraphNode = {
|
||||
readonly state: TaskGraphState;
|
||||
/**
|
||||
* Why it cannot run here, when it cannot: no registered definition takes it (`missing_task`, `task_too_old`,
|
||||
* `migration_failed`), or the restart limit parked it (`restart_limit`). Follows registry changes.
|
||||
* `migration_failed`). Follows registry changes.
|
||||
*/
|
||||
readonly blocked?: BlockedReason;
|
||||
/** Conversations this task owns, in ID order. */
|
||||
|
||||
@@ -15,7 +15,7 @@ const reportAttributes = {
|
||||
values: [
|
||||
"session_failed",
|
||||
"task_faulted",
|
||||
"task_parked",
|
||||
"task_interrupted",
|
||||
"migration_failed",
|
||||
"task_outdated",
|
||||
"task_error",
|
||||
@@ -28,7 +28,7 @@ const reportAttributes = {
|
||||
type: "string",
|
||||
required: false,
|
||||
description: "Fixed report reason, when present.",
|
||||
values: ["threw", "restart_limit", "missing_task", "incompatible_task"],
|
||||
values: ["threw", "restart_limit", "interrupted", "missing_task", "incompatible_task"],
|
||||
},
|
||||
extension: { type: "string", required: false, description: "Registered extension name, when present." },
|
||||
"task.id": { type: "number", required: false, description: "Task ID, when present.", cardinality: "high" },
|
||||
|
||||
@@ -34,6 +34,8 @@ import { toolExtension } from "./agent.ts";
|
||||
import { assignJson } from "./json.ts";
|
||||
import {
|
||||
clearProgress,
|
||||
cutOffCount,
|
||||
cutOffTooOften,
|
||||
finishSlot,
|
||||
LiveDoc,
|
||||
type NestedToolSlot,
|
||||
@@ -135,6 +137,7 @@ export const ToolTask = defineTask<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
|
||||
phases: {
|
||||
call: async (task, runtime, context) => {
|
||||
const call = await readCall(runtime, task.input, context);
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, call, context);
|
||||
const tool = await resolveTool(runtime, task.input, call.name, context);
|
||||
if (tool === undefined) {
|
||||
const error = harnessError("tool_unavailable", `Tool ${call.name} is not available`);
|
||||
@@ -184,6 +187,7 @@ export const ToolTask = defineTask<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
|
||||
const call = await readCall(runtime, task.input, context);
|
||||
const tool = await resolveTool(runtime, task.input, call.name, context);
|
||||
if (replay === "safe" && tool?.replay === "safe") {
|
||||
if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, call, context);
|
||||
// The rerun reports from scratch; clear what the interrupted attempt published.
|
||||
await runtime.commit(async (tx) => {
|
||||
const slot = toolSlot(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);
|
||||
@@ -212,6 +216,21 @@ export const ToolTask = defineTask<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* End a call cut off `settings.restartLimit` times in a row (spec §5.4) with an error result, built from the slot like
|
||||
* an interrupted call's, instead of running it again. `failed`, like an interruption, so its owned work is aborted.
|
||||
*/
|
||||
async function endCutOff(
|
||||
runtime: Runtime,
|
||||
task: { readonly restarts?: number; readonly input: ToolTaskInput },
|
||||
call: ToolHookCall,
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
const message = `Tool ${call.name} was interrupted ${cutOffCount(task)}; it is not run again`;
|
||||
const ending = { status: "failed", message, reason: "restart_limit" } as const;
|
||||
await settle(runtime, task.input, call, ending, (slot) => fromSlot(slot, "restart_limit", message), context);
|
||||
}
|
||||
|
||||
/** The called tool: among the tools the model is offered for a model-issued call, else among those tools may call. */
|
||||
async function resolveTool(
|
||||
runtime: Runtime,
|
||||
@@ -699,7 +718,10 @@ async function settle(
|
||||
if (madeNested) removeNestedSlots(live, runtime.taskId);
|
||||
if (ending.status === "aborted") return { status: "terminal", outcome: { status: "aborted", result: settled } };
|
||||
if (ending.status === "failed") {
|
||||
const error = { message: ending.message };
|
||||
const error = {
|
||||
message: ending.message,
|
||||
...(ending.reason === undefined ? {} : { detail: { reason: ending.reason } }),
|
||||
};
|
||||
return { status: "terminal", outcome: { status: "failed", error, result: settled } };
|
||||
}
|
||||
return { status: "terminal", outcome: { status: "completed", result: settled } };
|
||||
@@ -826,7 +848,9 @@ function fallbackResult(taskId: TaskId, name: string, outcome: TaskOutcome<ToolT
|
||||
* How a tool task ends; the result entry is appended either way. `failed` (execution threw or was interrupted)
|
||||
* records cancellation intent for the conversations the call owns; a result with `isError` still completes.
|
||||
*/
|
||||
type Ending = { readonly status: "completed" | "aborted" } | { readonly status: "failed"; readonly message: string };
|
||||
type Ending =
|
||||
| { readonly status: "completed" | "aborted" }
|
||||
| { readonly status: "failed"; readonly message: string; readonly reason?: string };
|
||||
|
||||
const COMPLETED: Ending = { status: "completed" };
|
||||
|
||||
|
||||
@@ -578,15 +578,18 @@ export type HarnessSettings = {
|
||||
*/
|
||||
readonly documentRetentionMs?: number;
|
||||
/**
|
||||
* What a run that fails (`model_error`, `no_model`) does with the inputs queued behind it: `keep` leaves them
|
||||
* What a run that fails (`model_error`, `no_model`, `interrupted`, `restart_limit`) does with the inputs queued behind it: `keep` leaves them
|
||||
* queued until the next input or queued write, `continue` starts the next run with them, as an answer would, and
|
||||
* `withdraw` settles them `unanswered` with reason `run_failed`. Default `keep`.
|
||||
*/
|
||||
readonly afterFailedRun?: AfterFailedRun;
|
||||
/**
|
||||
* 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.
|
||||
* changing its state before it ends instead of running again: a tool call with an error result, a compaction failed
|
||||
* into `pi.live.compactionFailure`, a run with `pi.live.runFailure` reason `restart_limit`, any other task `faulted`
|
||||
* with detail `{ reason: "restart_limit" }`. A generation's request or poll the host died during is a failed attempt
|
||||
* under `retry` first; the limit bounds it only when charging that attempt keeps being cut off. Default 5;
|
||||
* `Infinity` never ends a task for it.
|
||||
*/
|
||||
readonly restartLimit?: number;
|
||||
};
|
||||
@@ -672,7 +675,7 @@ export type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
|
||||
/**
|
||||
* 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 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`).
|
||||
* (`session_failed`, often a platform reset) from a faulted task (`task_faulted`).
|
||||
* Its own throw or rejection is dropped.
|
||||
*/
|
||||
readonly onReport?: (error: unknown, info: ReportInfo) => void;
|
||||
@@ -711,13 +714,10 @@ export type TaskInspection = {
|
||||
| { readonly kind: "sleeping"; readonly until: number }
|
||||
/** Outcome held until its ordinary owned work drains. */
|
||||
| { readonly kind: "completing" }
|
||||
/**
|
||||
* Cannot run here: no registered definition can take it (aborting it settles it as `orphaned`), or it is parked by
|
||||
* the restart limit (`retryTask()` runs it again).
|
||||
*/
|
||||
/** Cannot run here: no registered definition can take it (aborting it settles it as `orphaned`). */
|
||||
| {
|
||||
readonly kind: "blocked";
|
||||
readonly reason: "missing_task" | "task_too_old" | "migration_failed" | "restart_limit";
|
||||
readonly reason: "missing_task" | "task_too_old" | "migration_failed";
|
||||
readonly error?: unknown;
|
||||
};
|
||||
};
|
||||
@@ -845,11 +845,6 @@ export interface Harness extends Session {
|
||||
* definition cannot take it settles as `orphaned` instead.
|
||||
*/
|
||||
abortTask(id: TaskId, context: Context): Promise<"marked" | "terminal">;
|
||||
/**
|
||||
* Run a task parked by `settings.restartLimit` again: clears its restart count and enables scheduling. Does nothing
|
||||
* for a task that is not parked; rejects for an unknown one.
|
||||
*/
|
||||
retryTask(id: TaskId, context: Context): Promise<void>;
|
||||
/** Resolve with the terminal receipt; cancelling `context` cancels only this wait. */
|
||||
waitForTask<R>(id: TaskId<R>, context: Context): Promise<SettledTask<R>>;
|
||||
/** Resolve when the ordinary ownership scope of every ownerless conversation has no live non-background task. */
|
||||
|
||||
@@ -153,8 +153,10 @@ export type HostCallback = "now" | "settings" | "registry" | "env" | "models" |
|
||||
* 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()`.
|
||||
* `settings.restartLimit` times (`reason: "restart_limit"`), as was a phase of a task of an extension's kind.
|
||||
* - `task_interrupted`: crashes ended the task: it ended itself after being cut off `settings.restartLimit` times in a
|
||||
* row (`reason: "restart_limit"`; a tool call, a compaction, a generation), or a generation's run ended because the
|
||||
* host kept stopping while the model answered (`reason: "interrupted"`). `restarts` is the task's count when it ended.
|
||||
* - `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`).
|
||||
@@ -172,7 +174,13 @@ export type ReportInfo =
|
||||
readonly extension?: string;
|
||||
readonly reason: "threw" | "restart_limit";
|
||||
}
|
||||
| { readonly type: "task_parked"; readonly task: ReportTask; readonly extension?: string }
|
||||
| {
|
||||
readonly type: "task_interrupted";
|
||||
readonly task: ReportTask;
|
||||
readonly extension?: string;
|
||||
readonly reason: "restart_limit" | "interrupted";
|
||||
readonly restarts: number;
|
||||
}
|
||||
| { readonly type: "migration_failed"; readonly task: ReportTask; readonly extension?: string }
|
||||
| {
|
||||
readonly type: "task_outdated";
|
||||
@@ -686,8 +694,9 @@ type TaskRecordBase<I, R> = {
|
||||
readonly endedAt?: number;
|
||||
/**
|
||||
* 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()`.
|
||||
* state since; open counts them, and a close does not. At `HarnessSettings.restartLimit` the task ends (the built-in
|
||||
* tasks end themselves, visibly; the scheduler faults any other kind), and a generation reads it to charge an
|
||||
* interrupted model request as a failed attempt.
|
||||
*/
|
||||
readonly restarts?: number;
|
||||
};
|
||||
|
||||
@@ -59,7 +59,7 @@ describe("a Harness woken by its Durable Object alarm", () => {
|
||||
await eventually(stub, async (_instance, state) => (await state.storage.getAlarm()) === null);
|
||||
});
|
||||
|
||||
it("recovers a turn an eviction cut off mid-stream through the heartbeat alarm", async () => {
|
||||
it("recovers a turn an eviction cut off mid-stream through the alarms, retrying the charged attempt", async () => {
|
||||
const stub = fresh();
|
||||
await runInDurableObject(stub, async (instance: TestSession, state) => {
|
||||
instance.script(["hang", "recovered"]);
|
||||
|
||||
@@ -66,6 +66,8 @@ export class TestSession extends DurableObject {
|
||||
{
|
||||
models,
|
||||
registry,
|
||||
// An eviction during a request is a failed attempt (#11139); its retry need not wait long here.
|
||||
settings: { retry: { enabled: true, maxRetries: 3, baseDelayMs: 1 } },
|
||||
now: () => this.now(),
|
||||
onWake: (wake) => {
|
||||
this.wakes.push(wake);
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
type HarnessOptions,
|
||||
type HarnessSettings,
|
||||
type Registry,
|
||||
type ReportInfo,
|
||||
type Storage,
|
||||
type ToolRegistration,
|
||||
} from "@earendil-works/pi-durable";
|
||||
@@ -28,6 +29,8 @@ export type ChatSetup = {
|
||||
readonly models: Models;
|
||||
readonly registry: Registry;
|
||||
readonly reports: unknown[];
|
||||
/** What each report was about, in the order of `reports`. */
|
||||
readonly infos: ReportInfo[];
|
||||
/** Live Harness settings; tests assign fields to change them between decisions. */
|
||||
readonly settings: { -readonly [K in keyof HarnessSettings]: HarnessSettings[K] };
|
||||
now: () => number;
|
||||
@@ -39,7 +42,7 @@ export function chatSetup(options: RegisterFauxProviderOptions = {}): ChatSetup
|
||||
const faux = fauxProvider(options);
|
||||
const models = createModels();
|
||||
models.setProvider(faux.provider);
|
||||
return { faux, models, registry: createRegistry(), reports: [], settings: {}, now: () => Date.now() };
|
||||
return { faux, models, registry: createRegistry(), reports: [], infos: [], settings: {}, now: () => Date.now() };
|
||||
}
|
||||
|
||||
/** Open a Harness over `storage` and return its root, configured with the faux model on first creation. */
|
||||
@@ -60,7 +63,10 @@ export async function openChat(
|
||||
? {}
|
||||
: { env: typeof options.env === "function" ? options.env : () => options.env as ExecutionEnv }),
|
||||
now: () => setup.now(),
|
||||
onReport: (error) => setup.reports.push(error),
|
||||
onReport: (error, info) => {
|
||||
setup.reports.push(error);
|
||||
setup.infos.push(info);
|
||||
},
|
||||
},
|
||||
context,
|
||||
);
|
||||
|
||||
@@ -132,7 +132,8 @@ describe("recovery at open", () => {
|
||||
});
|
||||
|
||||
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 () => {
|
||||
// #11139
|
||||
it("faults an extension's task cut off restartLimit times in a row instead of running it again", async () => {
|
||||
const path = await sqlitePath();
|
||||
const state = { crashing: true, started: 0 };
|
||||
const registry = createRegistry();
|
||||
@@ -150,31 +151,29 @@ describe("tasks that cannot run", () => {
|
||||
}
|
||||
const { harness, infos } = opened;
|
||||
expect((await harness.getTask(id, context))?.restarts).toBe(3);
|
||||
expect(infos).toEqual([
|
||||
{ type: "task_parked", task: { id, kind: "test.crash-loop", conversationId: root.id }, extension: "loop" },
|
||||
]);
|
||||
// Nothing is parked: the task is ready, and the host is asked to run, which ends it.
|
||||
const inspection = await harness.inspect(context);
|
||||
expect(inspection.tasks.find((task) => task.record.id === id)?.state).toEqual({
|
||||
kind: "blocked",
|
||||
reason: "restart_limit",
|
||||
});
|
||||
const graph = await harness.watchTaskGraph(context);
|
||||
// A started watch keeps its value current.
|
||||
graph.start(async () => {});
|
||||
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 });
|
||||
expect(inspection.tasks.find((task) => task.record.id === id)?.state).toEqual({ kind: "ready", migrates: false });
|
||||
expect(harness.wake().active).toBe(true);
|
||||
harness.resume();
|
||||
// A scheduling pass runs within a few turns; it leaves the parked task out.
|
||||
for (let turn = 0; turn < 5; turn++) await flush();
|
||||
const settled = await harness.waitForTask(id, context);
|
||||
expect(settled.state.outcome).toEqual({
|
||||
status: "faulted",
|
||||
error: {
|
||||
message: `Task ${id} (test.crash-loop) was cut off 3 times without progress`,
|
||||
detail: { reason: "restart_limit" },
|
||||
},
|
||||
});
|
||||
expect(state.started).toBe(3);
|
||||
|
||||
state.crashing = false;
|
||||
await harness.retryTask(id, context);
|
||||
expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("completed");
|
||||
expect(state.started).toBe(4);
|
||||
await graph.stop();
|
||||
await eventually(() => infos.length > 0);
|
||||
expect(infos).toEqual([
|
||||
{
|
||||
type: "task_faulted",
|
||||
task: { id, kind: "test.crash-loop", conversationId: root.id },
|
||||
extension: "loop",
|
||||
reason: "restart_limit",
|
||||
},
|
||||
]);
|
||||
await harness.close(context);
|
||||
});
|
||||
|
||||
@@ -268,7 +267,7 @@ describe("tasks that cannot run", () => {
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("runs a parked task once a raised restartLimit no longer parks it", async () => {
|
||||
it("reads restartLimit when it reserves the task: a raised one runs it again", async () => {
|
||||
const path = await sqlitePath();
|
||||
const state = { crashing: true, started: 0 };
|
||||
const registry = createRegistry();
|
||||
@@ -284,24 +283,17 @@ describe("tasks that cannot run", () => {
|
||||
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.watchTaskGraph(context);
|
||||
// A started watch keeps its value current.
|
||||
graph.start(async () => {});
|
||||
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();
|
||||
expect((await opened.harness.getTask(id, context))?.restarts).toBe(1);
|
||||
state.crashing = false;
|
||||
// The host's live settings change; the next pass runs the task.
|
||||
// The host's live settings change before the task is reserved; it runs instead of ending.
|
||||
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);
|
||||
await graph.stop();
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("still runs the abort handler of a parked task", async () => {
|
||||
it("runs the abort handler of a task at the limit", async () => {
|
||||
const path = await sqlitePath();
|
||||
const state = { crashing: true, started: 0 };
|
||||
const aborted: TaskId[] = [];
|
||||
@@ -362,10 +354,12 @@ describe("tasks that cannot run", () => {
|
||||
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",
|
||||
opened.harness.resume();
|
||||
expect((await opened.harness.waitForTask(id, context)).state.outcome).toMatchObject({
|
||||
status: "faulted",
|
||||
error: { detail: { reason: "restart_limit" } },
|
||||
});
|
||||
expect(started).toBe(2);
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
|
||||
@@ -13,7 +13,6 @@ import {
|
||||
type Harness,
|
||||
LiveDoc,
|
||||
type LiveState,
|
||||
type TaskId,
|
||||
ToolResultEntry,
|
||||
} from "@earendil-works/pi-durable";
|
||||
import { describe, expect, it } from "vitest";
|
||||
@@ -28,6 +27,8 @@ import { aborted, deferred } from "./task-support.ts";
|
||||
function slowSetup(persistWithinMs: number): ChatSetup {
|
||||
const setup = chatSetup({ tokensPerSecond: 80, tokenSize: { min: 4, max: 4 } });
|
||||
setup.settings.progress = { persistWithinMs };
|
||||
// A request a crash interrupted is a failed attempt (#11139): retry at once.
|
||||
setup.settings.retry = { enabled: true, maxRetries: 5, baseDelayMs: 1 };
|
||||
return setup;
|
||||
}
|
||||
|
||||
@@ -92,10 +93,6 @@ function answers(entries: readonly EntryRecord[]): { text: string | undefined; s
|
||||
});
|
||||
}
|
||||
|
||||
async function generationId(harness: Harness, conversationId: ConversationId): Promise<TaskId> {
|
||||
return (await harness.snapshot(LiveDoc, conversationId, context))!.run!.taskId;
|
||||
}
|
||||
|
||||
describe("generation partials", () => {
|
||||
it("hold partials in memory and store them together, at most every persistWithinMs", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
@@ -113,7 +110,7 @@ describe("generation partials", () => {
|
||||
await harness.close(context);
|
||||
});
|
||||
|
||||
it("converts what was stored into the aborted entry after a crash, and resends", async () => {
|
||||
it("converts what was stored into the aborted entry after a crash, and retries the charged attempt", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
const setup = slowSetup(150);
|
||||
setup.faux.setResponses([fauxAssistantMessage(LONG)]);
|
||||
@@ -217,23 +214,6 @@ describe("generation partials", () => {
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("never counts partial commits as progress of a task cut off again and again", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
const setup = slowSetup(50);
|
||||
let opened = await open(storage, setup);
|
||||
setup.faux.setResponses([fauxAssistantMessage(LONG)]);
|
||||
await opened.root.submit({ type: "input", content: "hi" }, context);
|
||||
const id = await generationId(opened.harness, opened.root.id);
|
||||
for (let crashes = 1; crashes <= 2; crashes++) {
|
||||
await waitFor(async () => (storedText(await storedLive(storage, opened.root.id))?.length ?? 0) > 40);
|
||||
await crash(opened.harness);
|
||||
setup.faux.setResponses([fauxAssistantMessage(LONG)]);
|
||||
opened = await open(storage, setup);
|
||||
expect((await opened.harness.getTask(id, context))?.restarts).toBe(crashes);
|
||||
}
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("resumes polling a deferred response after a crash", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
const setup = chatSetup({ deferred: { pollAfterMs: 60_000 } });
|
||||
@@ -306,6 +286,47 @@ describe("tool output", () => {
|
||||
});
|
||||
}
|
||||
|
||||
// #11139: a generation's request is charged at once after a crash, so its partials never stream with a count; a
|
||||
// replay-safe tool reruns and streams its output with one.
|
||||
it("never counts progress commits as progress of a task cut off again and again", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
const setup = slowSetup(50);
|
||||
let runs = 0;
|
||||
addTool(
|
||||
setup.registry,
|
||||
defineTool({
|
||||
name: "work",
|
||||
description: "work",
|
||||
parameters: Type.Object({}),
|
||||
replay: "safe",
|
||||
execute: async (_args, api, callContext) => {
|
||||
runs++;
|
||||
for (let line = 0; !callContext.abortSignal!.aborted; line++) {
|
||||
api.output(`run ${runs} line ${line}\n`);
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
return aborted(callContext.abortSignal!);
|
||||
},
|
||||
}),
|
||||
);
|
||||
setup.faux.setResponses([
|
||||
fauxAssistantMessage([fauxToolCall("work", {}, { id: "c1" })], { stopReason: "toolUse" }),
|
||||
]);
|
||||
let opened = await open(storage, setup);
|
||||
await opened.root.submit({ type: "input", content: "go" }, context);
|
||||
for (let crashes = 1; crashes <= 2; crashes++) {
|
||||
// This run stored output of its own, a progress commit, before it was cut off.
|
||||
await waitFor(
|
||||
async () => (await storedOutput(storage, opened.root.id))?.startsWith(`run ${crashes} `) === true,
|
||||
);
|
||||
await crash(opened.harness);
|
||||
opened = await open(storage, setup);
|
||||
const tool = (await opened.harness.commit((tx) => tx.scanTasks({ kind: "pi.tool" }, 1), context)).items[0]!;
|
||||
expect(tool.restarts).toBe(crashes);
|
||||
}
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("answers a tool without replay that crashed with the output stored by then", async () => {
|
||||
const storage = new ControlledStorage();
|
||||
const setup = slowSetup(150);
|
||||
|
||||
@@ -0,0 +1,775 @@
|
||||
// What a crash (a host that dies, is evicted, or is reset) does to work that was running: a generation's interrupted
|
||||
// request is a failed attempt under the retry policy, and every other task ends visibly once it was cut off
|
||||
// `settings.restartLimit` times in a row (#11139).
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import {
|
||||
type AssistantMessage,
|
||||
type FauxResponseStep,
|
||||
fauxAssistantMessage,
|
||||
fauxText,
|
||||
fauxToolCall,
|
||||
type Message,
|
||||
type Models,
|
||||
type ToolResultMessage,
|
||||
Type,
|
||||
} from "@earendil-works/pi-ai";
|
||||
import {
|
||||
defineTool,
|
||||
type GenerationCheckpoint,
|
||||
type Harness,
|
||||
LiveDoc,
|
||||
type LiveState,
|
||||
MemoryStorage,
|
||||
ROOT_CONVERSATION_ID,
|
||||
type Storage,
|
||||
type SubmissionId,
|
||||
type TaskId,
|
||||
} from "@earendil-works/pi-durable";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
|
||||
import { allEntries, type ChatSetup, chatSetup, openChat, textOf, unanswered, waitFor } from "./chat-support.ts";
|
||||
import { addSection, addTool } from "./harness-support.ts";
|
||||
import { context, observeCommits } from "./session-support.ts";
|
||||
import { aborted, deferred } from "./task-support.ts";
|
||||
|
||||
const directories = new Set<string>();
|
||||
/** Storages of killed Harnesses, never closed by them: closed after each test. */
|
||||
const killed = new Set<Storage>();
|
||||
|
||||
afterEach(async () => {
|
||||
for (const storage of killed) await storage.close(context).catch(() => {});
|
||||
killed.clear();
|
||||
for (const directory of directories) await rm(directory, { recursive: true, force: true });
|
||||
directories.clear();
|
||||
});
|
||||
|
||||
async function sqlitePath(): Promise<string> {
|
||||
const directory = await mkdtemp(join(tmpdir(), "pi-durable-restart-"));
|
||||
directories.add(directory);
|
||||
return join(directory, "session.sqlite");
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a chat over the SQLite file at `path`. `kill()` simulates a crash: every later storage call hangs, so the Harness
|
||||
* writes nothing more, not even the `pending` records a close writes for the tasks it stops.
|
||||
*/
|
||||
async function open(path: string, setup: ChatSetup) {
|
||||
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 opened = await openChat(storage, setup);
|
||||
const kill = (): void => {
|
||||
dead = true;
|
||||
killed.add(inner);
|
||||
};
|
||||
return { ...opened, kill };
|
||||
}
|
||||
|
||||
async function live(harness: Harness): Promise<Readonly<LiveState>> {
|
||||
return (await harness.snapshot(LiveDoc, ROOT_CONVERSATION_ID, context)) ?? {};
|
||||
}
|
||||
|
||||
/** Every `pi.live` value from now on, as memory adopts it. */
|
||||
function liveValues(harness: Harness): LiveState[] {
|
||||
const values: LiveState[] = [];
|
||||
observeCommits(harness, (publication) => {
|
||||
for (const change of publication.changes) {
|
||||
if (change.type === "document" && change.record.kind === "pi.live" && change.value !== null) {
|
||||
values.push(change.value as LiveState);
|
||||
}
|
||||
}
|
||||
});
|
||||
return values;
|
||||
}
|
||||
|
||||
/** Let deferred progress commits (`persistWithinMs: 0`) reach Storage before a crash. */
|
||||
function stored(): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, 50));
|
||||
}
|
||||
|
||||
/** A setup whose model streams a long answer slowly; partials are stored as soon as they are committed. */
|
||||
function streaming(): ChatSetup {
|
||||
const setup = chatSetup({ tokensPerSecond: 20, tokenSize: { min: 1, max: 1 } });
|
||||
setup.faux.setResponses([fauxAssistantMessage("z".repeat(400))]);
|
||||
setup.settings.progress = { persistWithinMs: 0 };
|
||||
return setup;
|
||||
}
|
||||
|
||||
/** Submit an input to a fresh chat at `path` and crash once its streamed partial is stored. */
|
||||
async function crashWhileStreaming(path: string, setup: ChatSetup) {
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
await waitFor(async () => (await live(opened.harness)).generation?.message !== undefined);
|
||||
const taskId = (await live(opened.harness)).run!.taskId;
|
||||
await stored();
|
||||
opened.kill();
|
||||
return { id, taskId };
|
||||
}
|
||||
|
||||
/** Reopen after a crash in the middle of a request, with another streaming answer, and crash again. */
|
||||
async function crashAgain(path: string, setup: ChatSetup, taskId: TaskId) {
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
// The first attempt was charged; the second one streams.
|
||||
await waitFor(async () => {
|
||||
const state = (await opened.harness.getTask(taskId, context))?.state;
|
||||
if (state?.status !== "running") return false;
|
||||
const checkpoint = state.checkpoint as GenerationCheckpoint;
|
||||
return checkpoint.phase === "request" && checkpoint.attempt === 2;
|
||||
});
|
||||
await waitFor(async () => (await live(opened.harness)).generation?.message !== undefined);
|
||||
await stored();
|
||||
opened.kill();
|
||||
}
|
||||
|
||||
async function settled(harness: Harness, id: SubmissionId) {
|
||||
return (await harness.submission(id, context))!.wait(context);
|
||||
}
|
||||
|
||||
describe("a generation interrupted while the model answers (#11139)", () => {
|
||||
it("charges the attempt: keeps the partial, waits for the retry backoff, and answers on the next attempt", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 1 };
|
||||
const { id, taskId } = await crashWhileStreaming(path, first);
|
||||
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 1 };
|
||||
const sent: string[][] = [];
|
||||
setup.faux.setResponses([
|
||||
(request) => {
|
||||
sent.push(request.messages.map((message) => message.role));
|
||||
return fauxAssistantMessage("answer");
|
||||
},
|
||||
]);
|
||||
const opened = await open(path, setup);
|
||||
const values = liveValues(opened.harness);
|
||||
const partial = textOf((await live(opened.harness)).generation?.message as Message)!;
|
||||
expect((await opened.harness.getTask(taskId, context))?.restarts).toBe(1);
|
||||
opened.harness.resume();
|
||||
expect((await settled(opened.harness, id)).status).toBe("done");
|
||||
// The backoff showed in `pi.live` with why.
|
||||
const retry = values.find((value) => value.generation?.retry !== undefined)?.generation;
|
||||
expect(retry).toEqual({
|
||||
attempt: 1,
|
||||
retry: { at: expect.any(Number), error: "The host stopped while the model was answering, on attempt 1" },
|
||||
});
|
||||
expect(values.some((value) => value.generation?.attempt === 2)).toBe(true);
|
||||
const entries = await allEntries(opened.root);
|
||||
expect(entries.map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant", "pi.assistant"]);
|
||||
const converted = entries[1]!.model![0] as AssistantMessage;
|
||||
expect(converted.stopReason).toBe("aborted");
|
||||
expect(textOf(converted)).toBe(partial);
|
||||
// The aborted partial is excluded from the next request.
|
||||
expect(sent).toEqual([["user"]]);
|
||||
expect(await live(opened.harness)).toEqual({});
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("sleeps the backoff of `settings.retry` before the next attempt", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 60_000 };
|
||||
const { taskId } = await crashWhileStreaming(path, first);
|
||||
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 60_000 };
|
||||
const now = 1_000_000;
|
||||
setup.now = () => now;
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
await waitFor(async () => (await opened.harness.getTask(taskId, context))?.state.status === "sleeping");
|
||||
const record = await opened.harness.getTask(taskId, context);
|
||||
expect(record?.state).toMatchObject({ status: "sleeping", until: now + 60_000, checkpoint: { phase: "retry" } });
|
||||
expect(record?.restarts).toBeUndefined();
|
||||
expect((await live(opened.harness)).generation).toEqual({
|
||||
attempt: 1,
|
||||
retry: { at: now + 60_000, error: "The host stopped while the model was answering, on attempt 1" },
|
||||
});
|
||||
expect(opened.harness.wake()).toEqual({ active: false, at: now + 60_000 });
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("ends the run interrupted once no retry is left, keeping every partial", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.retry = { enabled: true, maxRetries: 1, baseDelayMs: 1 };
|
||||
const { id, taskId } = await crashWhileStreaming(path, first);
|
||||
// Attempt 2 is interrupted too.
|
||||
const second = streaming();
|
||||
second.settings.retry = { enabled: true, maxRetries: 1, baseDelayMs: 1 };
|
||||
await crashAgain(path, second, taskId);
|
||||
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: true, maxRetries: 1, baseDelayMs: 1 };
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const text = "The host stopped while the model was answering, on attempt 2";
|
||||
expect(await settled(opened.harness, id)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "interrupted",
|
||||
detail: text,
|
||||
});
|
||||
expect(await live(opened.harness)).toEqual({
|
||||
runFailure: { taskId, inputs: [id], reason: "interrupted", detail: text },
|
||||
});
|
||||
expect((await opened.harness.waitForTask(taskId, context)).state.outcome).toEqual({
|
||||
status: "failed",
|
||||
error: { message: text, detail: { reason: "interrupted" } },
|
||||
});
|
||||
const entries = await allEntries(opened.root);
|
||||
expect(entries.map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant", "pi.assistant"]);
|
||||
expect(entries.slice(1).map((entry) => (entry.model![0] as AssistantMessage).stopReason)).toEqual([
|
||||
"aborted",
|
||||
"aborted",
|
||||
]);
|
||||
// No request was sent after the second crash.
|
||||
expect(setup.faux.state.callCount).toBe(0);
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("ends the run at once when retries are disabled", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
const { id } = await crashWhileStreaming(path, first);
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: false };
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
expect(await settled(opened.harness, id)).toMatchObject({ status: "unanswered", reason: "interrupted" });
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("resends after a clean close: a close is not a crash", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.retry = { enabled: false };
|
||||
const opened = await open(path, first);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
await waitFor(async () => (await live(opened.harness)).generation?.message !== undefined);
|
||||
const taskId = (await live(opened.harness)).run!.taskId;
|
||||
await opened.harness.close(context);
|
||||
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: false };
|
||||
setup.faux.setResponses([fauxAssistantMessage("answer")]);
|
||||
const reopened = await open(path, setup);
|
||||
expect((await reopened.harness.getTask(taskId, context))?.restarts).toBeUndefined();
|
||||
reopened.harness.resume();
|
||||
expect((await settled(reopened.harness, id)).status).toBe("done");
|
||||
await reopened.harness.close(context);
|
||||
});
|
||||
|
||||
it("does not charge a generation that sleeps in a retry or waits for its tools", async () => {
|
||||
const path = await sqlitePath();
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: true, maxRetries: 1, baseDelayMs: 60_000 };
|
||||
const gate = deferred();
|
||||
const started = deferred();
|
||||
addTool(
|
||||
setup.registry,
|
||||
defineTool({
|
||||
name: "hold",
|
||||
description: "Waits",
|
||||
parameters: Type.Object({}),
|
||||
replay: "safe",
|
||||
execute: async () => {
|
||||
started.resolve();
|
||||
await gate.promise;
|
||||
return {};
|
||||
},
|
||||
}),
|
||||
);
|
||||
const error = fauxAssistantMessage([], { stopReason: "error", errorMessage: "503 Service Unavailable" });
|
||||
setup.faux.setResponses([error]);
|
||||
let opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
const taskId = (await live(opened.harness)).run!.taskId;
|
||||
await waitFor(async () => (await opened.harness.getTask(taskId, context))?.state.status === "sleeping");
|
||||
opened.kill();
|
||||
|
||||
// Asleep in its retry: not running, so not cut off.
|
||||
const later = chatSetup();
|
||||
later.settings.retry = { enabled: true, maxRetries: 1, baseDelayMs: 60_000 };
|
||||
later.now = () => Date.now() + 120_000;
|
||||
addTool(later.registry, setup.registry.snapshot().tools()[0]!.tool);
|
||||
later.faux.setResponses([
|
||||
fauxAssistantMessage([fauxToolCall("hold", {}, { id: "c1" })], { stopReason: "toolUse" }),
|
||||
fauxAssistantMessage([fauxText("done")]),
|
||||
]);
|
||||
opened = await open(path, later);
|
||||
expect((await opened.harness.getTask(taskId, context))?.restarts).toBeUndefined();
|
||||
opened.harness.resume();
|
||||
// Waiting for its tool round: not running either.
|
||||
await started.promise;
|
||||
const waiting = await opened.harness.getTask(taskId, context);
|
||||
expect(waiting?.state).toMatchObject({ status: "waiting", checkpoint: { phase: "tools" } });
|
||||
opened.kill();
|
||||
|
||||
opened = await open(path, later);
|
||||
expect((await opened.harness.getTask(taskId, context))?.restarts).toBeUndefined();
|
||||
gate.resolve();
|
||||
opened.harness.resume();
|
||||
expect((await settled(opened.harness, id)).status).toBe("done");
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("ends a run whose generation is cut off restartLimit times outside a request", async () => {
|
||||
const path = await sqlitePath();
|
||||
const setup = chatSetup();
|
||||
setup.settings.restartLimit = 1;
|
||||
const reached = deferred();
|
||||
// Preparation never finishes, as one that exhausts the host's memory would not.
|
||||
addSection(setup.registry, "preamble", async (_input, ctx) => {
|
||||
reached.resolve();
|
||||
return aborted(ctx.abortSignal!);
|
||||
});
|
||||
let opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
await reached.promise;
|
||||
const taskId = (await live(opened.harness)).run!.taskId;
|
||||
opened.kill();
|
||||
|
||||
opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const text = "The generation was interrupted once in its prepare phase";
|
||||
expect(await settled(opened.harness, id)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "restart_limit",
|
||||
detail: text,
|
||||
});
|
||||
expect(await live(opened.harness)).toEqual({
|
||||
runFailure: { taskId, inputs: [id], reason: "restart_limit", detail: text },
|
||||
});
|
||||
await waitFor(() => setup.infos.length > 0);
|
||||
expect(setup.infos).toContainEqual(
|
||||
expect.objectContaining({ type: "task_interrupted", reason: "restart_limit", restarts: 1 }),
|
||||
);
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
});
|
||||
|
||||
describe("other tasks cut off restartLimit times in a row (#11139)", () => {
|
||||
it("ends a tool call with an error result instead of running it again", async () => {
|
||||
const path = await sqlitePath();
|
||||
const setup = chatSetup();
|
||||
setup.settings.restartLimit = 2;
|
||||
setup.settings.progress = { outputIntervalMs: 0, persistWithinMs: 0 };
|
||||
let runs = 0;
|
||||
const running = { current: deferred() };
|
||||
addTool(
|
||||
setup.registry,
|
||||
defineTool({
|
||||
name: "heavy",
|
||||
description: "Never finishes",
|
||||
parameters: Type.Object({}),
|
||||
replay: "safe",
|
||||
execute: async (_args, api, ctx) => {
|
||||
runs++;
|
||||
api.output("partial\n");
|
||||
running.current.resolve();
|
||||
return aborted(ctx.abortSignal!);
|
||||
},
|
||||
}),
|
||||
);
|
||||
const answers: FauxResponseStep[] = [
|
||||
() => fauxAssistantMessage([fauxToolCall("heavy", {}, { id: "c1" })], { stopReason: "toolUse" }),
|
||||
(request) => {
|
||||
const result = request.messages.at(-1) as ToolResultMessage;
|
||||
return fauxAssistantMessage(`saw: ${textOf(result)}`);
|
||||
},
|
||||
];
|
||||
setup.faux.setResponses(answers);
|
||||
let opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "go" }, context)).id;
|
||||
for (let crash = 1; crash <= 2; crash++) {
|
||||
await running.current.promise;
|
||||
running.current = deferred();
|
||||
await stored();
|
||||
opened.kill();
|
||||
opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
}
|
||||
expect((await settled(opened.harness, id)).status).toBe("done");
|
||||
expect(runs).toBe(2);
|
||||
const result = (await allEntries(opened.root)).find((entry) => entry.kind === "pi.tool-result")!;
|
||||
const message = result.model![0] as ToolResultMessage;
|
||||
expect(message.isError).toBe(true);
|
||||
expect(result.data).toMatchObject({
|
||||
diagnostics: [
|
||||
{
|
||||
severity: "error",
|
||||
code: "restart_limit",
|
||||
message: "Tool heavy was interrupted 2 times in a row; it is not run again",
|
||||
},
|
||||
],
|
||||
});
|
||||
// The slot's partial output is kept, as for an interrupted call.
|
||||
expect(textOf(message)).toContain("partial");
|
||||
const toolTask = (await opened.harness.commit((tx) => tx.scanTasks({ kind: "pi.tool" }, 10), context)).items[0]!;
|
||||
expect(toolTask.state).toMatchObject({
|
||||
status: "terminal",
|
||||
outcome: { status: "failed", error: { detail: { reason: "restart_limit" } } },
|
||||
});
|
||||
expect(setup.infos).toContainEqual({
|
||||
type: "task_interrupted",
|
||||
task: { id: toolTask.id, kind: "pi.tool", conversationId: ROOT_CONVERSATION_ID },
|
||||
reason: "restart_limit",
|
||||
restarts: 2,
|
||||
});
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("fails a compaction into compactionFailure", async () => {
|
||||
const path = await sqlitePath();
|
||||
const setup = chatSetup();
|
||||
setup.settings.restartLimit = 1;
|
||||
// Little recent context is kept, so the first turn is summarized.
|
||||
setup.settings.compaction = { enabled: false, keepRecentTokens: 10 };
|
||||
setup.faux.setResponses([fauxAssistantMessage("a1"), fauxAssistantMessage("a2")]);
|
||||
let opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
await (await opened.root.submit({ type: "input", content: "u1 ".repeat(200) }, context)).wait(context);
|
||||
await (await opened.root.submit({ type: "input", content: "u2 ".repeat(200) }, context)).wait(context);
|
||||
// The summary request never returns.
|
||||
const hanging = unanswered();
|
||||
setup.faux.setResponses([hanging.step]);
|
||||
const compaction = await opened.root.compact(undefined, context);
|
||||
await hanging.reached;
|
||||
opened.kill();
|
||||
|
||||
opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const error = "Compaction was interrupted once";
|
||||
expect((await opened.harness.waitForTask(compaction, context)).state.outcome).toEqual({
|
||||
status: "failed",
|
||||
error: { message: error, detail: { reason: "restart_limit" } },
|
||||
});
|
||||
expect((await live(opened.harness)).compactionFailure).toEqual({
|
||||
taskId: compaction,
|
||||
reason: "manual",
|
||||
error,
|
||||
});
|
||||
expect((await live(opened.harness)).compactions).toBeUndefined();
|
||||
await waitFor(() => setup.infos.length > 0);
|
||||
expect(setup.infos).toContainEqual({
|
||||
type: "task_interrupted",
|
||||
task: { id: compaction, kind: "pi.compaction", conversationId: ROOT_CONVERSATION_ID },
|
||||
reason: "restart_limit",
|
||||
restarts: 1,
|
||||
});
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* A setup whose model answers deferred, and whose first fetch of each deferred response is in flight until the Harness
|
||||
* stops; later fetches answer. `fetching` resolves when the first fetch began.
|
||||
*/
|
||||
function deferredSetup(pollAfterMs = 10) {
|
||||
const base = chatSetup({ deferred: { pollAfterMs } });
|
||||
base.settings.stream = { deferred: true };
|
||||
base.faux.setResponses([fauxAssistantMessage("deferred answer")]);
|
||||
const fetching = deferred();
|
||||
let fetches = 0;
|
||||
const fetchDeferred: Models["fetchDeferred"] = (model, handle, options) => {
|
||||
fetches++;
|
||||
if (fetches > 1) return base.models.fetchDeferred(model, handle, options);
|
||||
fetching.resolve();
|
||||
return aborted(options!.signal!);
|
||||
};
|
||||
const models = new Proxy(base.models, {
|
||||
get(target, property) {
|
||||
if (property === "fetchDeferred") return fetchDeferred;
|
||||
const value = Reflect.get(target, property, target);
|
||||
return typeof value === "function" ? value.bind(target) : value;
|
||||
},
|
||||
});
|
||||
const setup: ChatSetup = { ...base, models };
|
||||
return { setup, fetching, fetches: () => fetches };
|
||||
}
|
||||
|
||||
/** Submit to a fresh chat at `path`, let its deferred response be fetched, and crash during that fetch. */
|
||||
async function crashWhilePolling(path: string, setup: ChatSetup, fetching: Promise<void>) {
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
await fetching;
|
||||
const taskId = (await live(opened.harness)).run!.taskId;
|
||||
expect((await opened.harness.getTask(taskId, context))?.state).toMatchObject({
|
||||
status: "running",
|
||||
checkpoint: { phase: "poll" },
|
||||
});
|
||||
opened.kill();
|
||||
return { id, taskId };
|
||||
}
|
||||
|
||||
describe("crash edges (#11139)", () => {
|
||||
it("charges a crash during a poll's fetch, then polls the same deferred response again", async () => {
|
||||
const path = await sqlitePath();
|
||||
const { setup, fetching, fetches } = deferredSetup();
|
||||
setup.settings.retry = { enabled: true, maxRetries: 2, baseDelayMs: 1 };
|
||||
const { id, taskId } = await crashWhilePolling(path, setup, fetching.promise);
|
||||
|
||||
const opened = await open(path, setup);
|
||||
const values = liveValues(opened.harness);
|
||||
expect((await opened.harness.getTask(taskId, context))?.restarts).toBe(1);
|
||||
opened.harness.resume();
|
||||
const answer = await settled(opened.harness, id);
|
||||
expect(answer.status).toBe("done");
|
||||
expect(values.find((value) => value.generation?.retry !== undefined)?.generation).toEqual({
|
||||
attempt: 1,
|
||||
retry: {
|
||||
at: expect.any(Number),
|
||||
error: "The host stopped while the deferred answer was fetched, on attempt 1",
|
||||
},
|
||||
});
|
||||
expect(values.some((value) => value.generation?.attempt === 2 && value.generation.deferred !== undefined)).toBe(
|
||||
true,
|
||||
);
|
||||
// One request, polled twice: the provider's job is neither dropped nor submitted again.
|
||||
expect(setup.faux.state.callCount).toBe(1);
|
||||
expect(fetches()).toBe(2);
|
||||
expect(setup.faux.state.cancelledDeferred).toEqual([]);
|
||||
const entries = await allEntries(opened.root);
|
||||
expect(textOf(entries.at(-1)!.model![0])).toBe("deferred answer");
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("cancels the deferred response when a crash during its poll ends the run", async () => {
|
||||
const path = await sqlitePath();
|
||||
const { setup, fetching } = deferredSetup();
|
||||
setup.settings.retry = { enabled: false };
|
||||
const { id, taskId } = await crashWhilePolling(path, setup, fetching.promise);
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const text = "The host stopped while the deferred answer was fetched, on attempt 1";
|
||||
expect(await settled(opened.harness, id)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "interrupted",
|
||||
detail: text,
|
||||
});
|
||||
expect(await live(opened.harness)).toEqual({
|
||||
runFailure: { taskId, inputs: [id], reason: "interrupted", detail: text },
|
||||
});
|
||||
expect(setup.faux.state.cancelledDeferred).toHaveLength(1);
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("charges an interrupted poll even when the clock moved backwards", async () => {
|
||||
const path = await sqlitePath();
|
||||
const { setup, fetching } = deferredSetup(10);
|
||||
let now = 1000;
|
||||
setup.now = () => now;
|
||||
setup.settings.retry = { enabled: false };
|
||||
const first = await open(path, setup);
|
||||
first.harness.resume();
|
||||
const id = (await first.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
await waitFor(async () => (await live(first.harness)).generation?.deferred !== undefined);
|
||||
now = 1010;
|
||||
first.harness.resume();
|
||||
await fetching.promise;
|
||||
first.kill();
|
||||
// A replacement host whose clock is behind: the poll looks not due yet, but its fetch was cut off.
|
||||
now = 1000;
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
expect(await settled(opened.harness, id)).toMatchObject({ status: "unanswered", reason: "interrupted" });
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("charges a request once when a crash is followed by a clean close before the charge", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.retry = { enabled: true, maxRetries: 3, baseDelayMs: 1 };
|
||||
const { id, taskId } = await crashWhileStreaming(path, first);
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: true, maxRetries: 3, baseDelayMs: 1 };
|
||||
setup.faux.setResponses([fauxAssistantMessage("answer")]);
|
||||
let opened = await open(path, setup);
|
||||
await opened.harness.close(context);
|
||||
opened = await open(path, setup);
|
||||
expect((await opened.harness.getTask(taskId, context))?.restarts).toBe(1);
|
||||
const values = liveValues(opened.harness);
|
||||
opened.harness.resume();
|
||||
expect((await settled(opened.harness, id)).status).toBe("done");
|
||||
const charged = values.filter((value) => value.generation?.retry !== undefined);
|
||||
expect(charged.map((value) => value.generation?.attempt)).toEqual([1]);
|
||||
expect((await allEntries(opened.root)).map((entry) => entry.kind)).toEqual([
|
||||
"pi.user",
|
||||
"pi.assistant",
|
||||
"pi.assistant",
|
||||
]);
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("ends a run restart_limit when charging the interrupted request keeps being cut off", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.restartLimit = 1;
|
||||
const { id, taskId } = await crashWhileStreaming(path, first);
|
||||
// At the limit: the partial is still kept.
|
||||
const setup = chatSetup();
|
||||
setup.settings.restartLimit = 1;
|
||||
const opened = await open(path, setup);
|
||||
const partial = textOf((await live(opened.harness)).generation?.message as Message)!;
|
||||
opened.harness.resume();
|
||||
const text = "The generation was interrupted once in its request phase";
|
||||
expect(await settled(opened.harness, id)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "restart_limit",
|
||||
detail: text,
|
||||
});
|
||||
expect(await live(opened.harness)).toEqual({
|
||||
runFailure: { taskId, inputs: [id], reason: "restart_limit", detail: text },
|
||||
});
|
||||
const entries = await allEntries(opened.root);
|
||||
expect(entries.map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant"]);
|
||||
expect(textOf(entries[1]!.model![0])).toBe(partial);
|
||||
expect(setup.faux.state.callCount).toBe(0);
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("drops the partial past the limit, where converting it may be what keeps failing", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
const { id, taskId } = await crashWhileStreaming(path, first);
|
||||
// As if the charge had been cut off twice before: the open after this crash counts 3, past the limit of 2.
|
||||
const raw = await openNodeSqliteStorage(path);
|
||||
const record = await raw.task(taskId, context);
|
||||
await raw.commit([{ type: "task", value: { ...record!, restarts: 2 } }], () => {}, context);
|
||||
await raw.close(context);
|
||||
const setup = chatSetup();
|
||||
setup.settings.restartLimit = 2;
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
expect(await settled(opened.harness, id)).toMatchObject({
|
||||
status: "unanswered",
|
||||
reason: "restart_limit",
|
||||
detail: "The generation was interrupted 3 times in a row in its request phase",
|
||||
});
|
||||
expect((await allEntries(opened.root)).map((entry) => entry.kind)).toEqual(["pi.user"]);
|
||||
expect((await live(opened.harness)).generation).toBeUndefined();
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
|
||||
it("ends a run someone aborted as aborted when its abort handler is cut off restartLimit times", async () => {
|
||||
const path = await sqlitePath();
|
||||
const setup = chatSetup({ deferred: { pollAfterMs: 3_600_000 } });
|
||||
setup.settings.restartLimit = 2;
|
||||
setup.settings.stream = { deferred: true };
|
||||
setup.faux.setResponses([fauxAssistantMessage("never")]);
|
||||
let cancels = 0;
|
||||
// The abort handler of a polling generation cancels its response: make that hang.
|
||||
setup.models.cancelDeferred = () => {
|
||||
cancels++;
|
||||
return new Promise(() => {});
|
||||
};
|
||||
let opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id;
|
||||
const taskId = (await live(opened.harness)).run!.taskId;
|
||||
await waitFor(async () => (await opened.harness.getTask(taskId, context))?.state.status === "sleeping");
|
||||
void opened.root.abort(context).catch(() => {});
|
||||
for (let crash = 1; crash <= 2; crash++) {
|
||||
await waitFor(() => cancels === crash);
|
||||
opened.kill();
|
||||
opened = await open(path, setup);
|
||||
expect((await opened.harness.getTask(taskId, context))?.restarts).toBe(crash);
|
||||
opened.harness.resume();
|
||||
}
|
||||
const detail = `Abort handler of task ${taskId} (pi.generation) was cut off 2 times`;
|
||||
expect(await settled(opened.harness, id)).toMatchObject({ status: "unanswered", reason: "aborted", detail });
|
||||
expect(await live(opened.harness)).toEqual({ runFailure: { taskId, inputs: [id], reason: "aborted", detail } });
|
||||
expect((await opened.harness.waitForTask(taskId, context)).state.outcome).toMatchObject({
|
||||
status: "faulted",
|
||||
error: { detail: { reason: "restart_limit" } },
|
||||
});
|
||||
await waitFor(() => setup.infos.some((info) => info.type === "task_faulted"));
|
||||
expect(setup.infos).toContainEqual(expect.objectContaining({ type: "task_faulted", reason: "restart_limit" }));
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
});
|
||||
|
||||
describe("reports of work crashes ended (#11139)", () => {
|
||||
it("reports task_interrupted for a run that ends interrupted and for a task that ends at the limit", async () => {
|
||||
const path = await sqlitePath();
|
||||
const first = streaming();
|
||||
first.settings.retry = { enabled: false };
|
||||
const { taskId } = await crashWhileStreaming(path, first);
|
||||
const setup = chatSetup();
|
||||
setup.settings.retry = { enabled: false };
|
||||
const opened = await open(path, setup);
|
||||
opened.harness.resume();
|
||||
await opened.harness.waitForTask(taskId, context);
|
||||
await waitFor(() => setup.infos.length > 0);
|
||||
expect(setup.infos).toEqual([
|
||||
{
|
||||
type: "task_interrupted",
|
||||
task: { id: taskId, kind: "pi.generation", conversationId: ROOT_CONVERSATION_ID },
|
||||
reason: "interrupted",
|
||||
restarts: 1,
|
||||
},
|
||||
]);
|
||||
expect(String(setup.reports[0])).toContain("The host stopped while the model was answering, on attempt 1");
|
||||
await opened.harness.close(context);
|
||||
});
|
||||
});
|
||||
|
||||
describe("the restart limit on the normal path (#11139)", () => {
|
||||
it("resolves the settings no more often than before: only a task that was cut off reads the limit", async () => {
|
||||
const setup = chatSetup();
|
||||
let resolutions = 0;
|
||||
// Every resolution of the host settings reads `extensions` once.
|
||||
(setup as { settings: object }).settings = new Proxy(
|
||||
{},
|
||||
{
|
||||
get(target, key) {
|
||||
if (key === "extensions") resolutions++;
|
||||
return Reflect.get(target, key);
|
||||
},
|
||||
},
|
||||
);
|
||||
addTool(
|
||||
setup.registry,
|
||||
defineTool({
|
||||
name: "t",
|
||||
description: "t",
|
||||
parameters: Type.Object({}),
|
||||
replay: "safe",
|
||||
execute: async () => ({}),
|
||||
}),
|
||||
);
|
||||
const rounds = 5;
|
||||
const responses: FauxResponseStep[] = [];
|
||||
for (let round = 0; round <= rounds; round++) {
|
||||
responses.push(() =>
|
||||
fauxAssistantMessage([fauxToolCall("t", {}, { id: `c${round}` })], { stopReason: "toolUse" }),
|
||||
);
|
||||
responses.push(() => fauxAssistantMessage([fauxText("done")]));
|
||||
}
|
||||
setup.faux.setResponses(responses);
|
||||
const { harness, root } = await openChat(new MemoryStorage(), setup);
|
||||
await (await root.submit({ type: "input", content: "warm up" }, context)).wait(context);
|
||||
resolutions = 0;
|
||||
for (let round = 0; round < rounds; round++) {
|
||||
await (await root.submit({ type: "input", content: `round ${round}` }, context)).wait(context);
|
||||
}
|
||||
// One generation round with one tool call, as before the restart limit applied to every kind (#11327: 49).
|
||||
expect(resolutions / rounds).toBe(49);
|
||||
await harness.close(context);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user