diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index f76b9543f..98f05d090 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -112,6 +112,7 @@ - `whenIdle: "reject"` on input submissions: an idle conversation rejects the input with the new `ConversationIdle` and writes nothing, so it never starts a run. With `whenBusy: "steer"` it steers the current run or is refused, in one admission instead of a host's check and a separate submit that can race the run's end. A known request ID still returns its submission ([#11149](https://github.com/earendil-works/pi/issues/11149)) - `Conversation.submit(tx, draft)` admits a submission inside the caller's commit (`Harness.commit()`, `Conversation.commit()`) with the same rules as `submit(draft, context)`, and returns its ID: a host checks its own documents and `pi.live`, admits, and writes its own state in one commit, which stores all of it or none. `ConversationHandle.submit(tx, draft)` does the same inside a task's or tool's own commit. The Harness supplies the clock, queue modes, the `pi.durable.submit` span (covering the admission only; its `result` is provisional until the commit is stored), and the request for progress. It must come before the commit's first table write, once per commit, and throws `ReadAfterWrite` otherwise on every path ([#11149](https://github.com/earendil-works/pi/issues/11149)) - `HookApi.entry()` and `entries()`, and `TaskRuntime.entries()`: a hook or task reads the conversation's raw history, which compaction never changes, so a range bounded by entry IDs reads the same before and after a compaction and on a rerun ([#11141](https://github.com/earendil-works/pi/issues/11141)) +- Run-end tasks: a `GenerationHooks.afterRun(run, api)` hook returns `{ tasks }`, built with the typed `runEndTask(task, input)`, after a run answered (a final answer or a `terminate` round; not when an `onYield` continuation is taken, not after a failure or an abort). The answer is committed and the run's inputs settle `done` first, so `Submission.wait()` resolves at the answer; the generation then creates the tasks as its children and waits for them, keeping `pi.live.run` (with `run.afterRun`) and the conversation busy, so idle waits include them, Stop aborts them, and a crash resumes them. Input queued meanwhile, steers included, is placed when they are terminal and starts the next run; a task admits a model-visible write or a fix-run follow-up in its own commit with `submit(tx, draft)`. A crash before the tasks exist asks the hooks again; a broken request (an uninstalled or built-in task, a non-JSON input, a throwing `initial()`) is reported and skipped. `run` names the run's entries: `first` (`pi.live.run.first`, carried into `runFailure.first` so a continue covers a failed run's history), `answer`, and `last` ([#11141](https://github.com/earendil-works/pi/issues/11141)) ### Changed diff --git a/packages/durable/README.md b/packages/durable/README.md index 3403a4997..703731bf9 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -540,7 +540,7 @@ await root.submit({ type: "write", entry: { kind: "app.note", data: "user opened - **Steers** are placed after the current tool round and join the running work. - **Follow-ups** are placed when the run answers, and start the next run. -- **`whenIdle: "reject"`** refuses an input when no run is working, so it never starts a run of its own. With `whenBusy: "steer"`, it steers the current run or throws `ConversationIdle`. +- **`whenIdle: "reject"`** refuses an input when no run is working, so it never starts a run of its own. With `whenBusy: "steer"`, it steers the current run or throws `ConversationIdle`. A run that answered and waits for its [after-run tasks](#work-at-the-end-of-a-run) is still working, so such an input queues and starts the next run. - **`submit(tx, draft)`** admits inside your own commit, with the same rules, and returns the submission's ID. Your checks, the admission, and your writes are stored together or not at all: ```typescript @@ -659,11 +659,47 @@ const Guard = defineExtension({ }); ``` -- **Generation:** `beforeRequest` (replace the messages of one request), `afterResponse`, `onYield` (continue the run with another user message), and `afterTools` (runs once a round's tools are done). +- **Generation:** `beforeRequest` (replace the messages of one request), `afterResponse`, `onYield` (continue the run with another user message), `afterTools` (runs once a round's tools are done), and `afterRun` (tasks to run before the run ends, below). - **Tools:** `beforeTool` (block or rewrite arguments) and `afterTool` (replace the result). `afterTool` sees every result a tool task writes, also the error results the Harness writes itself (unavailable, invalid, blocked, interrupted, restart limit, aborted), told apart by their diagnostic code; past `settings.restartLimit` it no longer runs. On an abort it runs inside the abort handler, so a slow hook delays `abort()`. To limit a hook to some conversations, select its extension only there, for example with `configure({ extensions: { add: [Guard] } })`. +Hooks read committed documents (`api.snapshot()`) and the conversation's raw history (`api.entry()`, `api.entries()`), which compaction never changes. + +### Work at the end of a run + +An `afterRun` hook starts durable work after a run answered, without a tool call: for example the project's tests after a run that edited files. It returns tasks, built with `runEndTask()`, and the generation creates them as its children and waits for them before the run ends: + +```typescript +import { defineExtension, GenerationTask, hook, runEndTask, ToolResultEntry } from "@earendil-works/pi-durable"; + +const TestsAfterEdits = defineExtension({ + name: "tests-after-edits", + tasks: [RunTests], // a task of yours, defineTask(): runs the tests, then commits its result + hooks: [ + hook(GenerationTask, { + afterRun: async (run, api, context) => { + // The run's own entries, also those a compaction summarized. + const range = { minEntryId: run.first, maxEntryId: run.last, order: "ascending" } as const; + const { items } = await api.entries(range, 1000, undefined, context); // page with `next` for longer runs + const edited = items.some((entry) => { + const message = entry.model?.[0]; + return ToolResultEntry.is(entry) && message?.role === "toolResult" && message.toolName === "edit"; + }); + return edited ? { tasks: [runEndTask(RunTests, { command: "npm test" })] } : undefined; + }, + }), + ], +}); +``` + +- **When:** after a final answer or a `terminate` round, not while an `onYield` continuation keeps the run going, never after a failure or an abort. +- **The answer first:** the answer is committed and the run's inputs settle `done` before the hooks are asked, so `Submission.wait()` resolves at the answer. +- **Busy until done:** `pi.live.run` stays, with `run.afterRun` listing the tasks, so the conversation is busy and `waitForIdle()` includes them; Stop aborts them; a crash resumes them. A crash before they exist asks the hooks again, never twice once they do. +- **Input meanwhile:** steers and follow-ups queue and start the next run once the tasks are done; `whenBusy: "reject"` and a continue throw `ConversationBusy`. +- **Results for the model:** a task gets its conversation's handle before its commit, `const handle = (await runtime.conversation(runtime.conversationId, context))!`, and admits a write or a follow-up (say "the tests failed: ...") inside it with `handle.submit(tx, draft)`, before its own table writes. Both are placed when the run ends: a write before any queued user input, a follow-up in queue order with the rest. A follow-up starts a fix run, which asks `afterRun` again with a fresh turn budget, so cap such loops in your extension. +- **Broken requests** (an uninstalled or built-in task, a non-JSON input, a throwing `initial()`) are reported through `onReport` against the hook's extension and skipped; the run still ends. + ## More Conversations and Forks ```typescript @@ -917,6 +953,7 @@ node --conditions=source --experimental-strip-types test/examples/13-chat.ts | [29-tool-override](test/examples/29-tool-override.ts) | A same-name bash for some conversations, and a wrapper that times whichever bash won | | [30-reload-and-restart](test/examples/30-reload-and-restart.ts) | Reloading an extension mid-call, and stored choices surviving a restart | | [31-transcript](test/examples/31-transcript.ts) | A client's transcript over the windowed view: live appends, a stall with a gap, loading it and earlier pages, and resuming with `from` | +| [32-tests-after-edits](test/examples/32-tests-after-edits.ts) | An `afterRun` hook that runs the tests after every run that edited files, a fix run when they fail, and input waiting for them | | [00](test/examples/00-conversation.ts)–[12](test/examples/12-recovery.ts) | The layers underneath: sessions, documents, forks, watches, the Harness, agent configuration, reload, extension state, tasks, recovery | Examples that call OpenAI need `OPENAI_API_KEY`; most use the faux provider otherwise. diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index e795d5556..72d630c1b 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -628,6 +628,10 @@ type SettledSubmissionRecord = SubmissionRecord & { interface Submission { readonly id: SubmissionId; status(context: Context): Promise; + /** + * Resolves once the input is answered or unanswered: at the answer, also while the run still waits for its after-run + * tasks (section 8.3), so the conversation may still be busy. + */ wait(context: Context): Promise; abort(context: Context): Promise<"aborted" | "already_placed" | "settled">; } @@ -2488,7 +2492,7 @@ task for it), the task ends visibly instead of running again: |---|---| | 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 | +| generation | ends its run `unanswered` with `pi.live.runFailure` reason `restart_limit`, failing with that detail; a run that answered already, in `finish` or `drain`, ends without a failure and the generation completes (section 8.3) | | 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 @@ -2625,7 +2629,9 @@ The **ordinary owned work** of a task `T` is every task that ordinary traversal reaches below `T`: the tasks `T` owns, the tasks in the conversations `T` owns, and, transitively, the same for each of those that is not background. A background task and everything below it are excluded. A conversation-owned task -may be background; a child task may not. +may be background; a child task may not. The tasks a generation creates for its +`afterRun` hooks (section 8.3) are such children, so Stop and idle waits include +them, and a cascade from the generation aborts them first. Two rules follow from the tree. @@ -2843,7 +2849,11 @@ inputs, including while its tool round runs, then the next generation. Tool tasks never own the run; the current round's tool tasks are listed in `pi.live.tools`. The ID list is mutable state because a boundary adds placed steering inputs to an active run; every terminal path -settles exactly the listed inputs. +settles exactly the listed inputs. A run that answered while `afterRun` hooks +are selected (section 8.3) settles its inputs `done` at the answer and keeps +`run`, with an empty ID list and `run.afterRun` set, until the tasks the hooks +asked for are terminal: the conversation stays busy, and `Submission.wait()` +has already resolved. Admission and terminal transitions: @@ -2858,7 +2868,7 @@ Admission and terminal transitions: | input or other write, busy or non-empty inbox | `queued` | append inbox item | | boundary places user item | `placed`, with entry | add ID to current or successor run | | boundary places write | `done`, with entry | append entry | -| run answers | input or continue `done`, with required answer entry | remove `run`, or hand it to a successor | +| run answers | input or continue `done`, with required answer entry | remove `run`, or hand it to a successor; with `afterRun` hooks selected, keep `run` until the tasks they ask for are terminal | | run fails or its task aborts | input `unanswered`, with reason | remove `run`; inbox per `afterFailedRun` below | | tool round reaches `maxTurns` | input `unanswered`, reason `turn_limit` | as a failed run | | withdraw queued item | `unanswered`, reason `aborted` | remove inbox item | @@ -2868,9 +2878,11 @@ Admission and terminal transitions: for the other submission type rejects. A busy input with `whenBusy: "reject"` writes no record and reports `ConversationBusy`. An idle input with `whenIdle: "reject"` writes no record and reports `ConversationIdle`, also when -the inbox holds queued items, so the input never starts a run: with -`whenBusy: "steer"` it steers the current run or is refused. Idle means -`run === undefined`, the inverse of `busy` above. An idle input queues with +the inbox holds queued items, so the input never starts a run of its own: with +`whenBusy: "steer"` it steers the current run or is refused, except as below. Idle means +`run === undefined`, the inverse of `busy` above. A run that answered and waits +for its after-run tasks is busy, so such an input queues there; it starts the +next run when that run ends, as every item queued behind a final answer does. An idle input queues with mode `steer` when `whenBusy` is `steer` and `followUp` otherwise. `Conversation.submit(tx, draft)` runs the same admission inside the caller's @@ -2985,8 +2997,15 @@ except for an `onYield` continuation (section 8.3), which applies only when the boundary selected no user item and no reset. Only successful run ends apply the final boundary: an answer or `terminate`. +With `afterRun` hooks selected, it applies when the run ends after its +after-run tasks, not with the answer, so whether items queued before the answer +land right after it depends on the conversation's extension selection. Items +queued meanwhile wait as well, steers included: there is no next model request +in the run for them to join. A run whose after-run phase was cut off +`settings.restartLimit` times ends without this boundary, as a failed run does. Failure, a run task's abort handler, fault, and orphan settle the run's inputs -`unanswered` and leave the inbox alone (`Conversation.abort()` separately withdraws queued user items); the queued items stay visible in +`unanswered`, unless the run answered already and waits for its after-run tasks +(its inputs are `done`), and leave the inbox alone (`Conversation.abort()` separately withdraws queued user items); the queued items stay visible in `pi.inbox` until the next input or queued write runs a boundary, or their withdrawal. Selected and stale items are removed positionally while retained item order is @@ -3274,6 +3293,7 @@ consuming commit may rerun them. Abort errors always propagate. | `beforeTool` | argument replacement chain; first block wins | block tool with error text | | `afterTool` | result replacement chain | report, continue | | `afterTools` | all observers | report, continue | +| `afterRun` | all observers; requested tasks concatenated in extension order | report, continue | | `beforeCompact` | first decision wins | report, continue | Hooks use task memos for durable first-writer-wins decisions. There is no public @@ -3315,8 +3335,25 @@ interface GenerationHooks { onYield(answer: AssistantMessage, api: HookApi, context: Context): HookResult<{ readonly continue: UserInput }>; /** After every tool of the round is terminal; `results` are the round's result entries in call order. */ afterTools(assistant: EntryId, results: readonly EntryId[], api: HookApi, context: Context): void | Promise; + /** After a run answered: the tasks to run before it ends (section 8.3). */ + afterRun(run: RunSummary, api: HookApi, context: Context): HookResult<{ readonly tasks: readonly RunEndTask[] }>; } +/** How a run ended, for `afterRun`. */ +type RunSummary = { + /** A final answer, or a tool round whose every call asked to `terminate`. */ + readonly reason: "answer" | "terminate"; + /** The run's first entry (`pi.live.run.first`, section 8.2), or the answer for a run without one. */ + readonly first: EntryId; + /** The answer, or the terminating round's tool-calling answer. */ + readonly answer: EntryId; + /** The run's last entry: the answer; for a `terminate` round its newest result, or the reset of its handoff. */ + readonly last: EntryId; +}; + +/** Built by `runEndTask(task, input)`, which types `input` by the task. */ +type RunEndTask = { readonly task: AnyTask; readonly input: JsonValue }; + /** `parent` is set for a nested call (section 7.3). */ type ToolHookCall = ToolCall & { readonly parent?: { readonly taskId: TaskId; readonly callId: string } }; @@ -3373,10 +3410,11 @@ namespace, so hook authors prefix their memo names. Hooks read committed state as it is when they run, off the line, so a hook a crash reruns may see more. Entries never change and compaction never removes -them (section 8.7): a range bounded by entry IDs reads the same before and -after a compaction and in a rerun. `HookApi` has no `context()`: the active -transcript changes in the middle of a run, when a compaction summary or a -handoff's reset lands, so a hook asking what a run did reads its history +them (section 8.7): a range bounded by entry IDs, such as an `afterRun` hook's +`entries({ minEntryId: run.first, maxEntryId: run.last })`, reads the same +before and after a compaction and in a rerun. `HookApi` has no `context()`: +the active transcript changes in the middle of a run, when a compaction summary +or a handoff's reset lands, so a hook asking what a run did reads its history instead. A range also holds entries of others, such as display-only writes, compaction summaries, and entries of other tasks; hooks filter by `kind`. @@ -3405,6 +3443,53 @@ await root.commit(async (tx) => { }, context); ``` +An `afterRun` hook starts durable work at the end of a run, without a tool +call, and the run waits for it (section 8.3). Here the project's tests run after +every run that edited files, and a failure starts a fix run: + +```ts +const RunTests = defineTask<{ command: string }, { phase: "run" }, string>({ + name: "app.run-tests", version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, context) => { + // The app's helper over `ExecutionEnv.exec()`: { passed: boolean; output: string }. A rerun after a crash runs it again. + const result = await runTests(await runtime.env(context), task.input.command, context); + const handle = (await runtime.conversation(runtime.conversationId, context))!; + await runtime.commit(async (tx) => { + // Admitted in this commit, before its table writes; placed when the run ends, in queue order with other input. + if (!result.passed) { + await handle.submit(tx, { type: "input", content: `Tests failed:\n${result.output}`, whenBusy: "followUp" }); + } + await tx.appendEntry(runtime.conversationId, { kind: "app.tests", data: { passed: result.passed } }); + return { status: "terminal", outcome: { status: "completed", result: result.output } }; + }, context); + }, + }, + abort: async (_task, runtime, context) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), context), +}); +export const TestsAfterEdits = defineExtension({ + name: "tests-after-edits", + tasks: [RunTests], + hooks: [hook(GenerationTask, { + afterRun: async (run, api, context) => { + const range = { minEntryId: run.first, maxEntryId: run.last, order: "ascending" } as const; + const { items } = await api.entries(range, 1000, undefined, context); + const edited = items.some((entry) => { + const message = entry.model?.[0]; + return ToolResultEntry.is(entry) && message?.role === "toolResult" && ["edit", "write"].includes(message.toolName); + }); + return edited ? { tasks: [runEndTask(RunTests, { command: "npm test" })] } : undefined; + }, + })], +}); +``` + +A fix run starts another run, which asks `afterRun` again, with a fresh turn +budget (`AgentState.maxTurns`); the extension caps such a loop itself, for +example with a counter in its own document. + ### 7.3 Tools ```ts @@ -4401,7 +4486,20 @@ type LiveState = { * Run control (section 6); present exactly while the conversation is busy. `turns` counts the run's tool rounds, * which `AgentState.maxTurns` bounds (section 8.5); an `onYield` continuation starts it over (section 8.3). */ - run?: { taskId: TaskId; inputs: SubmissionId[]; turns: number }; + run?: { + taskId: TaskId; + /** Placed inputs, settled when the run ends; `[]` once it answered and waits for its after-run tasks. */ + inputs: SubmissionId[]; + turns: number; + /** + * The run's first entry, where `afterRun` hooks start reading (section 8.3): the first one the boundary that + * started it placed, a write's or an input's; a continue's `runFailure.first`; otherwise the run's first + * tool-calling answer. A run started before it was recorded may lack it and gets its answer instead. + */ + first?: EntryId; + /** The run answered and waits for the tasks its `afterRun` hooks asked for, listed once created (section 8.3). */ + afterRun?: { tasks?: TaskId[] }; + }; /** Presentation of the current generation attempt. */ generation?: { attempt: number; @@ -4492,6 +4590,8 @@ type RunFailure = { * what was cut off, and how often; `turn_limit`: the tool rounds the run took, `"1 turn"` or `"N turns"`. */ detail?: string; + /** The failed run's `run.first`, which a continue takes over. */ + first?: EntryId; }; type CompactionFailure = { @@ -4520,7 +4620,7 @@ type CompactionStatus = { | version | `2` | | scope/history/fork | conversation, `latest`, `initial` | | `initial()` | `{}` | -| `migrate()` | from version 1: `reason` of each `compactions` status and of `compactionFailure` renamed `trigger`, and `run.turns` set to `0` (section 3.6) | +| `migrate()` | from version 1: `reason` of each `compactions` status and of `compactionFailure` renamed `trigger`, and `run.turns` set to `0` (section 3.6); `run.first`, `run.afterRun`, and `runFailure.first` are optional and need no step | | checkpoint | complete base whenever nothing runs: `generation` absent and no `running` tool or nested slot | | view mount | `docs["pi.live"]` | | created | with every Harness conversation (section 2.2) | @@ -4564,7 +4664,10 @@ that outcome holds as `completing` (section 5.5), or the scheduler's cleanup It stays small: the summary text lives in the task's placement, never here. Partials are normalized to strict JSON before assignment. Every terminal path of a run task removes `run`, `generation`, and `tools` in the commit that settles the run's -inputs. +inputs. A run that answered while `afterRun` hooks are selected is the one +exception: the answer's commit settles its inputs and removes `generation` and +`tools`, and `run` stays, with `run.afterRun` set, until the run ends after its +after-run tasks (section 8.3). A run that ends without an answer records why in `runFailure`, in the commit that settles its inputs `unanswered`: @@ -4586,7 +4689,10 @@ 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 reset, keeps it, so the failure is shown until a later run starts. A run that ends with an answer, including a round whose tools all ask to terminate, leaves -it alone. Inputs settled without a run (withdrawn, stale, aborted while +it alone; so does one that answered and then ends without its after-run tasks +(aborted, orphaned, or cut off): it answered. A run that fails copies its +`first` into `runFailure.first`, so a continue reads the failed run's history +too. Inputs settled without a run (withdrawn, stale, aborted while queued, `run_failed`) are not run failures; their submission records say why. `tx.settleSubmission()` stages each input's new status and resolves the transaction's latest candidate submission record, falling back to committed state, during assembly, like task-document validation (section 3.3), @@ -4645,6 +4751,17 @@ type GenerationCheckpoint = tools: TaskId[]; /** Calls of a sequential round not started yet, in call order. */ pending: string[]; + } + | { + /** The run answered; its `afterRun` hooks are asked for tasks, again after a crash. */ + phase: "finish"; + run: RunSummary; + } + | { + /** Waiting on the tasks the hooks asked for, which the generation owns; `waiting` holds this checkpoint. */ + phase: "drain"; + answer: EntryId; + tasks: TaskId[]; }; type GenerationResult = { entryId: EntryId }; ``` @@ -4653,7 +4770,8 @@ type GenerationResult = { entryId: EntryId }; `migrate` keeps a version 1 record. Version 2 keeps calls to tools not offered in a sequential round's `pending`, flagged `unoffered` in their slots (section 8.5), which a version 1 Harness would start as tool tasks without the flag, so -it leaves version 2 blocked (`task_too_old`). +it leaves version 2 blocked (`task_too_old`); the `finish` and `drain` phases +below are version 2 as well. The run's inputs live in `pi.live.run`, not in the task input. - `prepare` runs section 7.4 with the phase's agent resolution and resolves @@ -4749,7 +4867,15 @@ The run's inputs live in `pi.live.run`, not in the task input. Otherwise it settles the run's inputs `done`, removes `run` and `generation`, and starts a successor run for the selected user items, if any; a dropped continuation is not retried. Both complete with - `{ entryId }`. + `{ entryId }`. Whether a continuation applies is decided from the inbox + before the boundary places anything: it applies when the inbox holds no + steer, no follow-up, and no reset. + When the phase's agent selects an `afterRun` hook and no continuation + applies, the commit instead appends the answer, settles the run's inputs + `done`, keeps `run` with `inputs: []` and `afterRun: {}`, removes + `generation`, applies no boundary, and moves to `finish` with + `{ reason: "answer", first, answer, last }`: `first` is `run.first`, or the + answer when the run has none, and `answer` and `last` are the answer. - `toolUse` with at least one tool call: the commit appends the assistant entry and starts the tool round described below, moving to `waiting` in the `tools` phase. A `toolUse` message without calls is classified like `stop`. @@ -4779,9 +4905,34 @@ The run's inputs live in `pi.live.run`, not in the task input. it sleeps again. A still deferred result sleeps until a strictly later `pollAt`; any other result is classified as above. A model gone after reopen fails `no_model` once the poll is due. +- `finish` asks the `afterRun` hooks with its `run` (section 7.2) and checks + each requested task off the line: its kind must be an installed task that is + not built-in (`pi.*`), its input JSON, and the installed definition's + `initial()` must return a JSON checkpoint of one of its phases. A request + that fails a check is skipped and reported as an `extension_error` of the + hook that asked for it; the run never fails for one. The task is + created from the installed definition, not the hook's token, with the + checkpoint `initial()` returned, so `initial()` runs once. One commit creates + the tasks as children of the generation, lists them in `run.afterRun.tasks`, + and commits `waiting` on them with `allSettled` at `{ phase: "drain", answer, + tasks }`. Without tasks it ends the run at once, as `drain` does. A crash in + `finish` runs it again and asks the hooks again; nothing was created. Once the + tasks exist, the hooks are not asked again for this run, and a crash leaves + the wait and the tasks to resume. +- `drain` runs once every after-run task is terminal, whatever its outcome. One + commit applies the final boundary (section 6), removes `run`, starts a + successor run for the selected user items, and completes with + `{ entryId: answer }`. +- `finish` and `drain` check the restart count like every phase (section 5.4). + At the limit the run ends `completed` with the answer, without its tasks, + without a `runFailure`, and without the final boundary, so queued items stay + queued as after a failed run; the scheduler reports it `task_interrupted`. - The abort handler calls `models.cancelDeferred()` in `poll` when the model is known (a failure is reported), converts a committed partial, settles the inputs `unanswered` with - `aborted`, removes `run` and `generation`, and ends `aborted`. + `aborted`, removes `run` and `generation`, and ends `aborted`. In `finish` and + `drain` the run answered already: the handler runs after the after-run tasks + are terminal (section 5.5), removes `run`, and records no `runFailure`; the + inputs stay `done`. A tool round starts in the commit that appends the tool-calling answer: @@ -4802,7 +4953,8 @@ A tool round starts in the commit that appends the tool-calling answer: 3. The generation commits `waiting` on its tool tasks with `allSettled` and the `tools` checkpoint (section 8.5); `pi.live.run` stays with it. 4. `pi.live.tools` receives the round's slots (section 8.2), `generation` is - removed, and `pi.live.run.turns` grows by one. + removed, and `pi.live.run.turns` grows by one; a run without `first` gets the + tool-calling answer as its first entry. Input and continue submissions settle `unanswered` with one of these reasons: `no_model`, `model_error` (detail: provider error text), `nothing_to_answer`, @@ -4917,7 +5069,11 @@ and then commits once: text, and a queued head write targeting an entry before it is stale. - When every result of the round requests `terminate`, it settles the run's inputs `done` with the tool-calling answer, removes `run` and `tools`, and - applies the final boundary. + applies the final boundary. When the phase's agent selects an `afterRun` + hook, it keeps `run` instead and moves to `finish` (section 8.3) with + `{ reason: "terminate", first, answer, last }`: `answer` is the tool-calling + answer and `last` the newest of the round's result entries and the handoff's + reset; the final boundary waits for the run's end. - Otherwise, when the run's `turns`, counted in the commit that started the round, reached the phase agent's `maxTurns`, the run ends as a failed run (section 6): its inputs settle `unanswered` with `turn_limit` (detail: the @@ -5016,9 +5172,10 @@ ways to start one; the task is the same, only ownership and placement differ: Only a blocking compaction runs while the conversation's run waits for it. The others never take run control: the conversation keeps working while they -summarize, and their summary is placed like any passive write. Compactions do -not coordinate with each other; several may run at once, and section 6 decides -between their summaries. +summarize, and their summary is placed like any passive write: while a run +that answered waits for its after-run tasks (section 8.3), at the boundary that +ends it. Compactions do not coordinate with each other; several may run at +once, and section 6 decides between their summaries. **Range selection** is a pure function of a `ContextView`, whose `contributions` give each active entry's model messages after every edit in the @@ -5412,8 +5569,10 @@ exposed through their own Chord objects, not mounted. Everything a renderer draws is state in this value: the transcript in `entries`, the run and its presentation in `pi.live` (`run`, `generation`, `tools`, `nestedTools`, `compactions`, `compactionFailure`, `runFailure`, section 8.2), the queue in -`pi.inbox`, the agent in `pi.agent`, spend in `pi.usage`. A renderer renders -the newest value; it never assembles state from a stream of events. +`pi.inbox`, the agent in `pi.agent`, spend in `pi.usage`. A run that answered +and waits for its after-run tasks shows as `run` with `run.afterRun` and no +`generation`. A renderer renders the newest value; it never assembles state +from a stream of events. **The window.** Each watch and observer has its own window. It starts with the newest `min` entries and the ID of the entry before them in `before`. With @@ -6123,6 +6282,23 @@ These are contracts, not invitations to add defensive machinery: `waitForIdle()`; a hanging one holds them until the host restarts, and the handler's last attempt before the restart limit runs without hooks (section 8.4). +- **Busy after the answer:** with `afterRun` hooks selected, `Submission.wait()` + resolves at the answer while the conversation stays busy until the after-run + tasks end: `whenBusy: "reject"` and a continue throw `ConversationBusy`, and a + steer, also with `whenIdle: "reject"`, queues and starts the next run. +- **After-run tasks waiting on their own run:** an after-run task that awaits + `waitForIdle()` of its own conversation, or the settlement of a follow-up it + submitted, waits for the run that waits for it, and never resolves. Submit and + let the run's end place it. +- **Missing after-run tasks:** an after-run task whose definition is uninstalled + stays blocked, and so does its run: the conversation is busy until Stop + orphans the task. +- **Fix loops:** a follow-up an after-run task submits starts a run with a fresh + turn budget, which asks `afterRun` again; `maxTurns` does not bound the loop, + the extension does, for example with a counter in its own document. +- **Submissions from after-run tasks:** admit them in the task's own commit with + `submit(tx, draft)`, before its table writes; one submitted outside a commit + should carry a stable `requestId`, as a rerun after a crash submits it again. - **Hook memo names:** hooks share the asking task's memo namespace with the task and other hooks. Prefix memo names. - **Long transactions:** an async commit callback holds the Session mutation diff --git a/packages/durable/src/harness/define.ts b/packages/durable/src/harness/define.ts index b9bbd913f..dd4bd2e8d 100644 --- a/packages/durable/src/harness/define.ts +++ b/packages/durable/src/harness/define.ts @@ -1,6 +1,16 @@ import type { JsonValue } from "@earendil-works/chord"; import type { TSchema } from "@earendil-works/pi-ai"; -import type { AnyTask, Extension, HookRegistration, HooksOf, PromptSection, ToolRegistration, Wrap } from "./types.ts"; +import type { Task } from "../types.ts"; +import type { + AnyTask, + Extension, + HookRegistration, + HooksOf, + PromptSection, + RunEndTask, + ToolRegistration, + Wrap, +} from "./types.ts"; /** Identity function that types an extension. */ export function defineExtension( @@ -30,6 +40,14 @@ export function hook(task: K, handlers: Partial>): return { task: task.definition.name, handlers }; } +/** A task an `afterRun` hook asks for, its input typed by the task (spec §8.3). */ +export function runEndTask( + task: Task, + input: I, +): RunEndTask { + return { task, input }; +} + /** Wrap the tool named like `tool` wherever the wrapping extension is selected. */ export function wrapTool(tool: Tool, wrapper: (tool: Tool) => Tool): Wrap { return { tool: tool.name, wrap: wrapper }; diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 2b756bd37..b86af2ed0 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -1,4 +1,4 @@ -import { type Context, copyJson, type Draft } from "@earendil-works/chord"; +import { type Context, copyJson, type Draft, type JsonValue } from "@earendil-works/chord"; import type { Api, AssistantMessage, @@ -21,13 +21,14 @@ import type { EntryId, NextTaskState, SubmissionId, + Task, TaskId, TaskOutcome, TaskRuntime, Tx, TypedEntry, } from "../types.ts"; -import { addTools, sectionExtension } from "./agent.ts"; +import { addTools, agentHooks, sectionExtension } from "./agent.ts"; import { compactionPolicy, createCompaction, estimateContext, selectCut } from "./compaction.ts"; import { applyBoundary, type Boundary, hasQueuedReset, prepareBoundary, withdrawQueuedInputs } from "./inbox.ts"; import { @@ -44,6 +45,7 @@ import { import { PartialChanges, type PartialMessage } from "./partial-changes.ts"; import { planSystemEntries, renderSections, replaySections } from "./prompt.ts"; import { ensureProviderSessionId } from "./provider.ts"; +import { endedByCrashes } from "./scheduler.ts"; import { appendToolResult, harnessError, ToolTask, type ToolTaskResult } from "./tool.ts"; import type { AfterFailedRun, @@ -54,6 +56,8 @@ import type { GenerationHooks, ModelRef, PromptInput, + RunEndTask, + RunSummary, ToolControl, UserInput, } from "./types.ts"; @@ -115,6 +119,20 @@ export type GenerationCheckpoint = tools: TaskId[]; /** Calls of a sequential round not started yet, in call order. */ pending: string[]; + } + | { + /** + * The run answered: its answer is committed and its inputs settled, and it keeps `pi.live.run` while `afterRun` + * hooks are asked for tasks (spec §8.3). A crash here asks them again: nothing was created yet. + */ + phase: "finish"; + run: RunSummary; + } + | { + /** Waiting on the tasks `afterRun` hooks asked for, which the generation owns; then the run ends. */ + phase: "drain"; + answer: EntryId; + tasks: TaskId[]; }; export type GenerationResult = { entryId: EntryId }; @@ -331,6 +349,36 @@ export const GenerationTask = defineTask { + const { run } = task.state.checkpoint; + if (cutOffTooOften(task, runtime)) return endAnsweredCutOff(runtime, task, run.answer, context); + const prepared: PreparedTask[] = []; + await runtime.hooks.each("afterRun", async (hook) => { + const asked = await hook(run, runtime, context); + if (asked !== undefined) prepareRunEndTasks(runtime, asked.tasks, prepared); + }); + if (prepared.length === 0) return endAnsweredRun(runtime, run.answer, context); + await runtime.commit(async (tx): Promise => { + const tasks: TaskId[] = []; + for (const { task, input } of prepared) { + tasks.push(await tx.createTask(task, input, { ownership: { kind: "task", taskId: runtime.taskId } })); + } + const live = await tx.doc(LiveDoc, runtime.conversationId); + if (live.run?.taskId === runtime.taskId && live.run.afterRun !== undefined) live.run.afterRun.tasks = tasks; + const checkpoint = { phase: "drain", answer: run.answer, tasks } as const; + return { status: "waiting", checkpoint, on: tasks, policy: "allSettled" }; + }, context); + }, + /** The after-run tasks are terminal, whatever their outcomes: end the run. */ + drain: async (task, runtime, context) => { + const { answer } = task.state.checkpoint; + if (cutOffTooOften(task, runtime)) return endAnsweredCutOff(runtime, task, answer, context); + return endAnsweredRun(runtime, answer, context); + }, }, abort: async (task, runtime, context) => { const checkpoint = task.state.checkpoint; @@ -354,6 +402,7 @@ export const GenerationTask = defineTask 0) await startRun(tx, conversationId, live, users, { keepFailure: true }); + const { users, first } = await applyBoundary(tx, failure.boundary, "final", runtime.now()); + if (users.length > 0) await startRun(tx, conversationId, live, { inputs: users, first }, { keepFailure: true }); } /** @@ -780,8 +829,9 @@ async function classify( /** * A final answer; the final boundary places queued items (spec §6). The first `onYield` continuation appends a user - * message and hands the run to a successor generation, but only when the boundary selected no user item and no reset. - * Otherwise the run's inputs settle `done`, and selected user items start the next run. + * message and hands the run to a successor generation, but only when the inbox holds no user item and no reset. + * Otherwise the run's inputs settle `done`; with `afterRun` hooks selected the run waits for the tasks they ask for, + * in `finish` and `drain`, before the final boundary; without, selected user items start the next run now. */ async function answer(runtime: Runtime, message: AssistantMessage, context: Context): Promise { let continuation: UserInput | undefined; @@ -789,6 +839,7 @@ async function answer(runtime: Runtime, message: AssistantMessage, context: Cont if (continuation !== undefined) return; continuation = (await hook(message, runtime, context))?.continue; }); + const afterRun = await asksAfterRun(runtime, context); const conversationId = runtime.conversationId; await runtime.commit(async (tx): Promise => { // Queue modes are read on the Session line, when the boundary is decided. @@ -797,7 +848,11 @@ async function answer(runtime: Runtime, message: AssistantMessage, context: Cont const entry = await appendAssistant(tx, conversationId, message); const result: Next = { status: "terminal", outcome: { status: "completed", result: { entryId: entry.id } } }; const continues = takesContinuation(continuation, boundary) ? continuation : undefined; - const { users } = await applyBoundary(tx, boundary, "final", runtime.now()); + if (continues === undefined && afterRun) { + const held = holdRun(tx, live, runtime, { reason: "answer", answer: entry.id, last: entry.id }); + if (held !== undefined) return held; + } + const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now()); if (continues !== undefined) { const user = { role: "user", content: continues, timestamp: runtime.now() } as const; await tx.appendEntry(UserEntry, conversationId, { model: [user] }); @@ -808,11 +863,107 @@ async function answer(runtime: Runtime, message: AssistantMessage, context: Cont return result; } endRun(tx, live, runtime.taskId, { status: "done", answer: entry.id }); - if (users.length > 0) await startRun(tx, conversationId, live, users); + if (users.length > 0) await startRun(tx, conversationId, live, { inputs: users, first }); return result; }, context); } +/** Whether the phase's agent selects an `afterRun` hook, so a run that answers waits for its tasks (spec §8.3). */ +async function asksAfterRun(runtime: Runtime, context: Context): Promise { + const hooks = agentHooks(await runtime.agent(context), GenerationTask.definition.name); + return hooks.some(({ handlers }) => typeof (handlers as Partial).afterRun === "function"); +} + +/** + * Commit A of a run that answered with `afterRun` hooks selected (spec §8.3): settle the run's inputs `done` with the + * answer and keep `pi.live.run` for the after-run tasks, then ask the hooks in `finish`. The final boundary waits for + * the run's end, so queued items stay queued. + */ +function holdRun(tx: Tx, live: Draft, runtime: Runtime, end: Omit): Next | undefined { + const run = live.run; + if (run?.taskId !== runtime.taskId) return undefined; + for (const id of run.inputs) tx.settleSubmission(id, { status: "done", answer: end.answer }); + run.inputs = []; + run.afterRun = {}; + delete live.generation; + delete live.tools; + delete live.nestedTools; + // A run started before `first` was recorded has its answer as its first entry. + const summary: RunSummary = { ...end, first: run.first ?? end.answer }; + return { status: "running", checkpoint: { phase: "finish", run: summary } }; +} + +/** A task an `afterRun` hook asked for, checked and ready to create. */ +type PreparedTask = { readonly task: Task; readonly input: JsonValue }; + +/** + * Check the tasks one `afterRun` hook asked for, off the line, and add those that pass to `prepared`, so a broken request + * is skipped instead of failing the creating commit: its kind must be an installed task that is not built-in, its input + * JSON, and its definition's `initial()` must return a JSON checkpoint of one of its phases. The task is created from + * the installed definition, as every task runs under it, with that checkpoint, so `initial()` runs once. Then throws + * the rejections, which `hooks.each` reports against the hook's extension. + */ +function prepareRunEndTasks(runtime: Runtime, requests: readonly RunEndTask[], prepared: PreparedTask[]): void { + const rejected: unknown[] = []; + for (const request of requests) { + try { + const name = request.task.definition.name; + if (name.startsWith("pi.")) throw new Error(`An afterRun hook cannot start the built-in task ${name}`); + const installed = runtime.registry.task(name); + if (installed === undefined) throw new Error(`Task ${name} is not installed`); + const input = copyJson(request.input, { omitUndefinedProperties: true }); + // The registry erases definitions; an installed one is a task definition of its kind. + const definition = installed.definition as PreparedTask["task"]["definition"]; + const checkpoint = copyJson(definition.initial(input), { omitUndefinedProperties: true }) as { phase: string }; + if (!Object.hasOwn(definition.phases, checkpoint.phase)) { + throw new Error(`Task ${name} starts at the unknown phase ${String(checkpoint.phase)}`); + } + prepared.push({ task: { definition: { ...definition, initial: () => checkpoint } }, input }); + } catch (error) { + rejected.push(error); + } + } + if (rejected.length === 1) throw rejected[0]; + if (rejected.length > 1) + throw new AggregateError(rejected, `${rejected.length} afterRun task requests were rejected`); +} + +/** + * Commit B of a run that answered (spec §8.3): its after-run tasks are terminal, or there were none. Place what the + * final boundary selects, end the run, and start the next one for selected user items. + */ +async function endAnsweredRun(runtime: Runtime, answer: EntryId, context: Context): Promise { + const conversationId = runtime.conversationId; + await runtime.commit(async (tx): Promise => { + const boundary = await prepareBoundary(tx, conversationId, runtime.settings); + const live = await tx.doc(LiveDoc, conversationId); + const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now()); + endRun(tx, live, runtime.taskId, { status: "done", answer }); + if (users.length > 0) await startRun(tx, conversationId, live, { inputs: users, first }); + return { status: "terminal", outcome: { status: "completed", result: { entryId: answer } } }; + }, context); +} + +/** + * End a run that answered once its `finish` or `drain` phase was cut off `settings.restartLimit` times (spec §5.4): + * the run answered, so it ends `completed`, without a failure and without the tasks it would still have waited for, + * marked so the scheduler reports it `task_interrupted`. Queued items stay queued, as after a failed run. + */ +async function endAnsweredCutOff( + runtime: Runtime, + task: { readonly restarts?: number; readonly state: { readonly checkpoint: GenerationCheckpoint } }, + answer: EntryId, + context: Context, +): Promise { + const text = `The generation was interrupted ${cutOffCount(task)} in its ${task.state.checkpoint.phase} phase`; + await runtime.commit(async (tx): Promise => { + const live = await tx.doc(LiveDoc, runtime.conversationId); + endRun(tx, live, runtime.taskId, { status: "done", answer }); + const completed: Next = { status: "terminal", outcome: { status: "completed", result: { entryId: answer } } }; + return endedByCrashes(completed, text); + }, context); +} + /** * Whether an answer's `onYield` continuation is taken (spec §8.3), decided from the inbox before the final boundary * places anything: only when it holds no user item, of which the boundary would select one, and no reset, which would @@ -871,8 +1022,12 @@ async function startToolRound( } delete live.generation; live.tools = slots; - // A tool round is one turn of the run (`AgentState.maxTurns`). - if (live.run?.taskId === runtime.taskId) live.run.turns += 1; + if (live.run?.taskId === runtime.taskId) { + // A tool round is one turn of the run (`AgentState.maxTurns`). + live.run.turns += 1; + // A run its boundary started without placing anything, as a continue after an answer, begins here. + live.run.first ??= entry.id; + } const checkpoint = { phase: "tools", assistant: entry.id, tools, pending } as const; return { status: "waiting", checkpoint, on: tools, policy: "allSettled" }; }, context); @@ -913,22 +1068,28 @@ async function finishToolRound( const handoff = ordered.findLast((control) => control?.handoff !== undefined)?.handoff; // Read from the phase's agent, as every agent field; a `configure()` during the round applies at its end. const maxTurns = (await runtime.agent(context)).maxTurns; + const afterRun = terminate && (await asksAfterRun(runtime, context)); await runtime.commit(async (tx): Promise => { const boundary = await prepareBoundary(tx, conversationId, runtime.settings); if (added.length > 0) await addTools(tx, conversationId, added); const live = await tx.doc(LiveDoc, conversationId); const now = runtime.now(); + // The run's last entry so far: the newest result, or the assistant when no result has an entry. + let last = results.reduce((newest, id) => (id > newest ? id : newest), assistant); if (handoff !== undefined) { // The handoff's reset precedes every queued item, so queued writes and selected steers land in its context. const message = { role: "user", content: handoff, timestamp: now } as const; const entry = await tx.appendEntry(ResetEntry, conversationId, { head: "self", model: [message] }); boundary.head = entry.id; + last = entry.id; } const turns = live.run?.taskId === runtime.taskId ? live.run.turns : 0; + const held = afterRun ? holdRun(tx, live, runtime, { reason: "terminate", answer: assistant, last }) : undefined; + if (held !== undefined) return held; if (terminate) { - const { users } = await applyBoundary(tx, boundary, "final", now); + const { users, first } = await applyBoundary(tx, boundary, "final", now); endRun(tx, live, runtime.taskId, { status: "done", answer: assistant }); - if (users.length > 0) await startRun(tx, conversationId, live, users); + if (users.length > 0) await startRun(tx, conversationId, live, { inputs: users, first }); } else if (maxTurns !== undefined && turns >= maxTurns) { // The turn limit ends the run like a failure: `afterFailedRun` decides the queued items, and a continue resumes // it on the round's results. The table was read for the boundary already, before any write. @@ -940,11 +1101,13 @@ async function finishToolRound( detail: turns === 1 ? "1 turn" : `${turns} turns`, }); } else { - const { users, reset } = await applyBoundary(tx, boundary, "postTools", now); + const { users, first, reset } = await applyBoundary(tx, boundary, "postTools", now); if (reset) { // The queued reset cut the run's context before an answer. endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "reset" }); - if (users.length > 0) await startRun(tx, conversationId, live, users, { keepFailure: true }); + if (users.length > 0) { + await startRun(tx, conversationId, live, { inputs: users, first }, { keepFailure: true }); + } } else { delete live.tools; delete live.nestedTools; @@ -970,17 +1133,20 @@ async function appendAssistant( } /** - * Start a run for `inputs`, placed input submissions: a new generation takes `pi.live.run`. Removes `runFailure`, unless - * `keepFailure`: a run started in the commit that ended the failed one keeps its failure. + * Start a run for `inputs`, placed input submissions, whose first entry is `first` (`LiveRun.first`): a new generation + * takes `pi.live.run`. Removes `runFailure`, unless `keepFailure`: a run started in the commit that ended the failed one + * keeps its failure. */ export async function startRun( tx: Tx, conversationId: ConversationId, live: Draft, - inputs: SubmissionId[], + run: { readonly inputs: SubmissionId[]; readonly first: EntryId | undefined }, options?: { readonly keepFailure?: boolean }, ): Promise { - live.run = { taskId: await createGeneration(tx, conversationId, { submissionIds: inputs }), inputs, turns: 0 }; + const { inputs, first } = run; + const taskId = await createGeneration(tx, conversationId, { submissionIds: inputs }); + live.run = { taskId, inputs, turns: 0, ...(first === undefined ? {} : { first }) }; if (options?.keepFailure !== true) delete live.runFailure; } diff --git a/packages/durable/src/harness/inbox.ts b/packages/durable/src/harness/inbox.ts index ced697767..3469e8e79 100644 --- a/packages/durable/src/harness/inbox.ts +++ b/packages/durable/src/harness/inbox.ts @@ -41,8 +41,15 @@ export type Boundary = { head: EntryId | undefined; }; -/** Selected user items, in ID order, and whether a `head: "self"` write (a reset) was placed. */ -export type BoundaryResult = { readonly users: SubmissionId[]; readonly reset: boolean }; +/** + * Selected user items, in ID order, the first entry the boundary placed, a write's or a user item's (a run it starts + * begins there, `LiveRun.first`), and whether a `head: "self"` write (a reset) was placed. + */ +export type BoundaryResult = { + readonly users: SubmissionId[]; + readonly first: EntryId | undefined; + readonly reset: boolean; +}; /** * Read what a boundary needs. Table reads must precede the commit's first table write, so callers prepare the @@ -86,6 +93,7 @@ export async function applyBoundary( users.sort((a, b) => a - b); // `appendEntry()` copies the drafts' values; the items are removed only afterwards. + let first: EntryId | undefined; for (const index of writes) { const item = items[index] as Draft>; const draft = item.entry as unknown as EntryDraft; @@ -96,6 +104,7 @@ export async function applyBoundary( const entry = await tx.appendEntry(conversationId, draft); if (draft.head !== undefined) boundary.head = draft.head === "self" ? entry.id : draft.head; tx.placeSubmission(item.id, entry.id); + first ??= entry.id; } const placed: SubmissionId[] = []; for (const index of users) { @@ -103,10 +112,11 @@ export async function applyBoundary( const entry = await appendInput(tx, conversationId, item.content as UserInput, item.entry as InputEntry, now); tx.placeSubmission(item.id, entry.id); placed.push(item.id); + first ??= entry.id; } const removed = [...writes, ...users].sort((a, b) => b - a); for (const index of removed) items.splice(index, 1); - return { users: placed, reset }; + return { users: placed, first, reset }; } /** Append the entry an input is placed as: `entry`'s kind and data, or `pi.user`, with the user message of `content`. */ diff --git a/packages/durable/src/harness/live.ts b/packages/durable/src/harness/live.ts index a0fcccd3b..d9696f46d 100644 --- a/packages/durable/src/harness/live.ts +++ b/packages/durable/src/harness/live.ts @@ -130,15 +130,35 @@ export type RunFailure = { * was cut off, and how often; `turn_limit`: the tool rounds the run took, `"1 turn"` or `"N turns"`. */ detail?: string; + /** The failed run's `run.first`, so a continue's `afterRun` hooks see its history too. */ + first?: EntryId; +}; + +/** Run control (spec §8.2). */ +export type LiveRun = { + /** The task that settles the run's inputs: its generation, handed from one to the next. */ + taskId: TaskId; + /** The run's placed inputs, settled when it ends; `[]` once it answered and runs only its `afterRun` tasks. */ + inputs: SubmissionId[]; + /** Tool rounds so far, which `AgentState.maxTurns` bounds; an `onYield` continuation starts the count over. */ + turns: number; + /** + * The run's first entry: the first one the boundary that started it placed, a continue's the failed run's it + * continues; otherwise its first tool-calling answer. A run started before it was recorded may lack it; `afterRun` + * hooks then get the answer as the first entry. + */ + first?: EntryId; + /** + * Present once the run answered and its generation waits for the tasks `afterRun` hooks asked for, listed once + * created: queued items stay queued until they are terminal, and the run then ends without a failure. + */ + afterRun?: { tasks?: TaskId[] }; }; /** Built-in live conversation state: run control and presentation of the current generation and tool round. */ export type LiveState = { - /** - * Run control: the task that settles the run's inputs, those inputs, and the run's tool rounds so far, which - * `AgentState.maxTurns` bounds (an `onYield` continuation starts the count over); present exactly while busy. - */ - run?: { taskId: TaskId; inputs: SubmissionId[]; turns: number }; + /** Run control; present exactly while busy, also while a run that answered runs its `afterRun` tasks. */ + run?: LiveRun; /** Presentation of the current generation attempt. */ generation?: { attempt: number; @@ -263,14 +283,16 @@ export type RunEnd = /** * End the run owned by `taskId`: settle each of its inputs, remove `run`, and, without an answer, record - * `runFailure`. Always removes `generation`, `tools`, and `nestedTools`, whose presentation belongs to the ending run. + * `runFailure`. A run that answered already, and runs only its `afterRun` tasks, records no failure, however it ends. + * Always removes `generation`, `tools`, and `nestedTools`, whose presentation belongs to the ending run. */ export function endRun(tx: Tx, live: Draft, taskId: TaskId, end: RunEnd): void { if (live.run?.taskId === taskId) { const inputs = [...live.run.inputs]; + const { first } = live.run; if (end.status === "done") { for (const id of inputs) tx.settleSubmission(id, { status: "done", answer: end.answer }); - } else { + } else if (live.run.afterRun === undefined) { const { reason, detail } = end; const settlement = end.settlement ?? { status: "unanswered", @@ -278,7 +300,13 @@ export function endRun(tx: Tx, live: Draft, taskId: TaskId, end: RunE ...(detail === undefined ? {} : { detail }), }; for (const id of inputs) tx.settleSubmission(id, settlement); - live.runFailure = { taskId, inputs, reason, ...(detail === undefined ? {} : { detail }) }; + live.runFailure = { + taskId, + inputs, + reason, + ...(detail === undefined ? {} : { detail }), + ...(first === undefined ? {} : { first }), + }; } delete live.run; } diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index de26aceb3..fc3c2107c 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -2114,7 +2114,13 @@ export class TaskScheduler { }, commit: async (change, context, options) => { let checkpoint: string | undefined; - let interrupted: { readonly error: TaskOutcomeError; readonly restarts: number } | undefined; + let interrupted: + | { + readonly message: string; + readonly reason: "restart_limit" | "interrupted"; + readonly restarts: number; + } + | undefined; await this.#gated( invocation, async (tx, current) => { @@ -2124,9 +2130,16 @@ export class TaskScheduler { 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 }; + if (next?.status === "terminal") { + const restarts = current.restarts ?? 0; + const reason = next.outcome.status === "failed" ? interruptionOf(next.outcome.error) : undefined; + if (reason !== undefined && next.outcome.status === "failed") { + interrupted = { message: next.outcome.error.message, reason, restarts }; + } else { + // A task that ended for crashes with another outcome says so itself, as a generation whose run + // answered ends `completed` without its after-run tasks. + const message = endedByCrashesStates.get(next); + if (message !== undefined) interrupted = { message, reason: "restart_limit", restarts }; } } }, @@ -2135,11 +2148,11 @@ export class TaskScheduler { ); if (interrupted !== undefined) { this.#report( - new Error(interrupted.error.message), + new Error(interrupted.message), { kind: "task_interrupted", ...this.#aboutTask(invocation), - reason: interruptionOf(interrupted.error)!, + reason: interrupted.reason, restarts: interrupted.restarts, }, invocation.context, @@ -2407,6 +2420,19 @@ function recordTask(record: Pick(); + +/** `state`, marked as ending its task for crashes (`task_interrupted` with `message`), whatever its outcome. */ +export function endedByCrashes(state: T, message: string): T { + endedByCrashesStates.set(state, message); + return state; +} + function interruptionOf(error: TaskOutcomeError): "restart_limit" | "interrupted" | undefined { const detail = error.detail; if (typeof detail !== "object" || detail === null || !("reason" in detail)) return undefined; diff --git a/packages/durable/src/harness/submissions.ts b/packages/durable/src/harness/submissions.ts index df0363e67..66f116af3 100644 --- a/packages/durable/src/harness/submissions.ts +++ b/packages/durable/src/harness/submissions.ts @@ -269,9 +269,10 @@ export async function admitSubmission( if (busy) throw new ConversationBusy(conversationId); const boundary = await prepareBoundary(tx, conversationId, queueModes); const { id } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" }); - // Items an earlier failed run left queued (`afterFailedRun: "keep"`) are placed first and join the run. - const { users } = await applyBoundary(tx, boundary, "final", now); - await startRun(tx, conversationId, live, [...users, id]); + // Items an earlier failed run left queued (`afterFailedRun: "keep"`) are placed first and join the run. Its history + // starts with the failed run's, so `afterRun` hooks see what that run did. + const { users, first } = await applyBoundary(tx, boundary, "final", now); + await startRun(tx, conversationId, live, { inputs: [...users, id], first: live.runFailure?.first ?? first }); return { id, status: "placed" }; } // A boundary reads the table, so it is prepared before the first table write; a busy one needs none. @@ -295,8 +296,8 @@ export async function admitSubmission( items.push({ id, mode, content, ...entry }); } if (boundary === undefined) return { id, status: "queued" }; - const { users } = await applyBoundary(tx, boundary, "final", now); - if (users.length > 0) await startRun(tx, conversationId, live, users); + const { users, first } = await applyBoundary(tx, boundary, "final", now); + if (users.length > 0) await startRun(tx, conversationId, live, { inputs: users, first }); return { id, status: draft.type === "write" ? "done" : users.at(-1) === id ? "placed" : "queued" }; } if (draft.type === "write") { @@ -314,6 +315,6 @@ export async function admitSubmission( const entry = await appendInput(tx, conversationId, draft.content, draft.entry, now); const input = { ...fields, type: "input", status: "placed", entry: entry.id } as const; const { id } = await tx.createSubmission(input); - await startRun(tx, conversationId, live, [id]); + await startRun(tx, conversationId, live, { inputs: [id], first: entry.id }); return { id, status: "placed" }; } diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 99a024c6a..d7f4a05ec 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -79,7 +79,8 @@ export type SubmissionDraft = { readonly whenBusy?: "steer" | "followUp" | "reject"; /** * `reject`: an idle conversation (no run) rejects the input with `ConversationIdle` and writes nothing, so the input - * never starts a run. With `whenBusy: "steer"` it steers the current run or is refused. + * never starts a run of its own. With `whenBusy: "steer"` it steers the current run or is refused; a run that + * answered and waits for its `afterRun` tasks is still busy, and such an input queues and starts the next run. */ readonly whenIdle?: "reject"; /** Place the input as this entry; default `pi.user` without data. The kind must not start with `pi.`. */ @@ -975,8 +976,34 @@ export interface GenerationHooks { onYield(answer: AssistantMessage, api: HookApi, context: Context): HookResult<{ readonly continue: UserInput }>; /** After every tool of the round is terminal; `results` are the round's result entries in call order. */ afterTools(assistant: EntryId, results: readonly EntryId[], api: HookApi, context: Context): void | Promise; + /** + * After a run answered, its answer committed and its inputs settled `done`: a final answer, or a `terminate` round; + * not when an `onYield` continuation is taken, and never after a failure or an abort. The requested tasks of every + * hook are created together as children of the generation, which keeps the run, and the conversation busy, until + * they are terminal; then queued items are placed and the run ends (spec §8.3). A crash before they are created asks + * again; once created, never again for this run. + */ + afterRun(run: RunSummary, api: HookApi, context: Context): HookResult<{ readonly tasks: readonly RunEndTask[] }>; } +/** How a run ended, for `afterRun` hooks. Read the run's history with `api.entries({ minEntryId: first, maxEntryId: last })`. */ +export type RunSummary = { + /** A final answer, or a tool round whose every call asked to `terminate`. */ + readonly reason: "answer" | "terminate"; + /** + * The run's first entry (`LiveRun.first`): the first one the boundary that started it placed, for a continue the failed + * run's it continues, otherwise its first tool-calling answer; the answer itself for a run that has none of these. + */ + readonly first: EntryId; + /** The answer, or the terminating round's tool-calling answer, whose results follow it. */ + readonly answer: EntryId; + /** The run's last entry: the answer; for a `terminate` round its newest result, or the reset of its handoff. */ + readonly last: EntryId; +}; + +/** A task an `afterRun` hook asks for, built by `runEndTask()`. */ +export type RunEndTask = { readonly task: AnyTask; readonly input: JsonValue }; + /** A call as tool hooks see it; `parent` is set for a nested call, one a running tool made through `executeTool()`. */ export type ToolHookCall = ToolCall & { readonly parent?: { readonly taskId: TaskId; readonly callId: string }; diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index a08c8ceb5..4b4671129 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -31,7 +31,7 @@ export { CompactionTask, type SummaryRequest, } from "./harness/compaction.ts"; -export { defineExtension, defineTool, hook, section, wrapSection, wrapTool } from "./harness/define.ts"; +export { defineExtension, defineTool, hook, runEndTask, section, wrapSection, wrapTool } from "./harness/define.ts"; export { type GenerationCheckpoint, type GenerationInput, @@ -44,6 +44,7 @@ export { type CompactionFailure, type CompactionStatus, LiveDoc, + type LiveRun, type LiveState, type NestedToolSlot, type NestedToolSummary, @@ -112,6 +113,8 @@ export type { Registry, RegistryReader, RegistrySnapshot, + RunEndTask, + RunSummary, Settings, SettledSubmissionRecord, SettledTask, diff --git a/packages/durable/test/examples/32-tests-after-edits.ts b/packages/durable/test/examples/32-tests-after-edits.ts new file mode 100644 index 000000000..32292c521 --- /dev/null +++ b/packages/durable/test/examples/32-tests-after-edits.ts @@ -0,0 +1,140 @@ +// Run the project's tests at the end of every run that edited files, with an `afterRun` hook: the run waits for the +// tests, a failure starts a fix run, and input sent meanwhile waits for them too. +// Run from packages/durable: +// node --conditions=source --experimental-strip-types test/examples/32-tests-after-edits.ts +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { Type } from "@earendil-works/pi-ai"; +import { createModels } from "@earendil-works/pi-ai/models"; +import { fauxAssistantMessage, fauxProvider, fauxToolCall } from "@earendil-works/pi-ai/providers/faux"; +import { + createRegistry, + defineExtension, + defineTask, + defineTool, + GenerationTask, + Harness, + hook, + LiveDoc, + MemoryStorage, + runEndTask, + ToolResultEntry, +} from "../../src/index.ts"; + +const context = BACKGROUND_CONTEXT; + +// ─── Product code: the project, an edit tool, and the test runner ───────────── + +// The project under test, in memory: the tests pass once `add` adds. +const files = new Map([["math.ts", "export const add = (a, b) => a - b;"]]); +const runTests = () => { + const passed = files.get("math.ts")!.includes("a + b"); + return { passed, output: passed ? "1 passed" : "add(1, 2): expected 3, got -1" }; +}; + +const edit = defineTool({ + name: "edit", + description: "Replace a file's content", + parameters: Type.Object({ path: Type.String(), content: Type.String() }), + execute: async (args) => { + files.set(args.path, args.content); + return { output: [{ type: "text", text: `wrote ${args.path}` }] }; + }, +}); + +// A durable task: it runs once the run answered, and the run ends after it. A crash resumes it. +const RunTests = defineTask<{ attempt: number }, { phase: "run" }, boolean>({ + name: "app.run-tests", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, ctx) => { + const result = runTests(); + console.log(` [tests, attempt ${task.input.attempt}] ${result.output}`); + // The handle is fetched before the commit: inside it, runtime reads would wait for the commit itself. + const conversation = (await runtime.conversation(runtime.conversationId, ctx))!; + await runtime.commit(async (tx) => { + // Admitted with this commit, before its own table writes: placed when the run ends. + if (!result.passed && task.input.attempt < 3) { + await conversation.submit(tx, { + type: "input", + content: `The tests failed: ${result.output}. Fix it.`, + whenBusy: "followUp", + requestId: `fix-${task.input.attempt}`, + }); + } + await tx.appendEntry(runtime.conversationId, { kind: "app.tests", data: result }); + return { status: "terminal", outcome: { status: "completed", result: result.passed } }; + }, ctx); + }, + }, + abort: async (_task, runtime, ctx) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }) as const, ctx), +}); + +let attempts = 0; +const TestsAfterEdits = defineExtension({ + name: "tests-after-edits", + tools: [edit], + tasks: [RunTests], + hooks: [ + hook(GenerationTask, { + afterRun: async (run, api, ctx) => { + // The run's own entries, from its first input to its answer; a compaction never removes them. + const range = { minEntryId: run.first, maxEntryId: run.last, order: "ascending" } as const; + const { items } = await api.entries(range, 1000, undefined, ctx); + const edited = items.some((entry) => { + const message = entry.model?.[0]; + return ToolResultEntry.is(entry) && message?.role === "toolResult" && message.toolName === "edit"; + }); + // Each fix run asks again with a fresh turn budget: the extension caps the loop itself. + return edited ? { tasks: [runEndTask(RunTests, { attempt: ++attempts })] } : undefined; + }, + }), + ], +}); + +// ─── Host setup ───────────────────────────────────────────────────────────── + +const faux = fauxProvider(); +const models = createModels(); +models.setProvider(faux.provider); +faux.setResponses([ + // The first run edits, wrongly. + fauxAssistantMessage(fauxToolCall("edit", { path: "math.ts", content: "export const add = (a, b) => a - b + 0;" }), { + stopReason: "toolUse", + }), + fauxAssistantMessage("I cleaned up add."), + // The user's question, queued first, runs first; it edits nothing, so no tests run after it. + fauxAssistantMessage("Not yet: add subtracts. The fix is queued next."), + // The fix run the failed tests started. + fauxAssistantMessage(fauxToolCall("edit", { path: "math.ts", content: "export const add = (a, b) => a + b;" }), { + stopReason: "toolUse", + }), + fauxAssistantMessage("Fixed: add adds now."), +]); + +const registry = createRegistry(); +registry.install(TestsAfterEdits); +const harness = await Harness.open(new MemoryStorage(), { models, registry }, context); +const root = await harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } }); + +const submission = await root.submit({ type: "input", content: "Clean up math.ts." }, context); +// Resolves with the answer; the conversation stays busy while the tests run. +console.log("answered:", (await submission.wait(context)).status); +const live = await harness.snapshot(LiveDoc, root.id, context); +console.log("busy with after-run tasks:", live?.run?.afterRun !== undefined); +// Sent while the tests run: it waits for them. Follow-ups run in queue order, so it runs before the fix run. +await root.submit({ type: "input", content: "Are the tests green now?" }, context); + +await root.waitForIdle(context); +const page = await root.entries({ order: "ascending" }, 100, undefined, context); +for (const entry of page.items) { + const message = entry.model?.[0]; + if (entry.kind === "app.tests") console.log("tests:", (entry.data as { output: string }).output); + else if (message?.role === "user") console.log(">", message.content); + else if (message?.role === "assistant") { + const text = message.content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join(""); + if (text !== "") console.log("agent:", text); + } +} +await harness.close(context); diff --git a/packages/durable/test/harness-after-run.test.ts b/packages/durable/test/harness-after-run.test.ts new file mode 100644 index 000000000..373922122 --- /dev/null +++ b/packages/durable/test/harness-after-run.test.ts @@ -0,0 +1,1204 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { Draft, JsonValue } from "@earendil-works/chord"; +import { fauxAssistantMessage, fauxToolCall, Type } from "@earendil-works/pi-ai"; +import { + CompactionTask, + ConversationBusy, + type ConversationHandle, + defineExtension, + defineTask, + defineTool, + type EntryRecord, + type GenerationHooks, + GenerationTask, + hook, + InboxDoc, + LiveDoc, + type LiveState, + MemoryStorage, + type RunEndTask, + type RunSummary, + runEndTask, + type Storage, + type TaskId, + type Tx, +} from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { endRun } from "../src/harness/live.ts"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { allEntries, type ChatSetup, chatSetup, openChat, textOf } from "./chat-support.ts"; +import { context } from "./session-support.ts"; +import { aborted, completed, type Deferred, deferred, eventually, settled } from "./task-support.ts"; + +const directories = new Set(); +/** Storages of killed Harnesses, never closed by them: closed after each test. */ +const killed = new Set(); + +afterEach(async () => { + for (const storage of killed) await storage.close(context).catch(() => {}); + killed.clear(); + for (const directory of directories) await rm(directory, { recursive: true, force: true }); + directories.clear(); +}); + +async function sqlitePath(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-after-run-")); + directories.add(directory); + return join(directory, "session.sqlite"); +} + +/** Open a chat over SQLite at `path`; `kill()` simulates a crash: every later storage call hangs, so nothing more lands. */ +async function openKillable(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 }; +} + +/** What the after-run task and hook of a test did, and how they behave. */ +type Probe = { + /** Labels of the after-run task's runs, in order, reruns included. */ + runs: string[]; + /** Summaries the `afterRun` hook was asked with. */ + asked: RunSummary[]; + aborted: number; + /** Holds every run of the after-run task until resolved. */ + gate?: Deferred; + /** Holds the hook until resolved. */ + hookGate?: Deferred; + /** Admitted in the after-run task's commit, before its own entry. */ + admit?: (handle: ConversationHandle, tx: Tx) => Promise; +}; + +function probe(): Probe { + return { runs: [], asked: [], aborted: 0 }; +} + +/** The after-run task: records its run, waits for the gate, then appends a `test.tests` entry and completes. */ +function testsTask(state: Probe) { + return defineTask<{ label: string }, { phase: "run" }, string>({ + name: "test.run-tests", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, ctx) => { + state.runs.push(task.input.label); + if (state.gate !== undefined) await Promise.race([state.gate.promise, aborted(runtime.signal)]); + const handle = + state.admit === undefined ? undefined : await runtime.conversation(runtime.conversationId, ctx); + await runtime.commit(async (tx) => { + if (state.admit !== undefined) await state.admit(handle!, tx); + await tx.appendEntry(runtime.conversationId, { kind: "test.tests", data: task.input.label }); + return completed(task.input.label); + }, ctx); + }, + }, + abort: async (_task, runtime, ctx) => { + state.aborted++; + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }) as const, ctx); + }, + }); +} + +/** + * Install the after-run extension: its task and an `afterRun` hook, by default asking for one task per run. Returns + * the task. + */ +function install(setup: ChatSetup, state: Probe, afterRun?: GenerationHooks["afterRun"]) { + const Tests = testsTask(state); + const handler: GenerationHooks["afterRun"] = + afterRun ?? + (async (run) => { + state.asked.push(run); + if (state.hookGate !== undefined) await state.hookGate.promise; + return { tasks: [runEndTask(Tests, { label: `run ${state.asked.length}` })] }; + }); + setup.registry.install( + defineExtension({ name: "after-run", tasks: [Tests], hooks: [hook(GenerationTask, { afterRun: handler })] }), + ); + return Tests; +} + +const Parameters = Type.Object({}); +const tools = defineExtension({ + name: "tools", + tools: [ + defineTool({ + name: "edit", + description: "Edit a file", + parameters: Parameters, + execute: async () => ({ output: [{ type: "text", text: "edited" }] }), + }), + defineTool({ + name: "finish", + description: "End the run", + parameters: Parameters, + execute: async () => ({ output: [{ type: "text", text: "finished" }], control: { terminate: true } }), + }), + defineTool({ + name: "handoff", + description: "End the run in a new context", + parameters: Parameters, + execute: async () => ({ + output: [{ type: "text", text: "handed off" }], + control: { terminate: true, handoff: "fresh start" }, + }), + }), + ], +}); + +function call(name: string, id: string) { + return fauxAssistantMessage([fauxToolCall(name, {}, { id })], { stopReason: "toolUse" }); +} + +function kindOf(entries: readonly EntryRecord[], kind: string): EntryRecord[] { + return entries.filter((entry) => entry.kind === kind); +} + +/** Texts of the user messages of the conversation's entries, in order. */ +function userTexts(entries: readonly EntryRecord[]): string[] { + return entries.flatMap((entry) => + (entry.model ?? []).flatMap((message) => (message.role === "user" ? [textOf(message) ?? ""] : [])), + ); +} + +describe("afterRun", () => { + // #11141 + it("waits for the tasks an answer's hooks ask for: inputs done at the answer, the run busy until they end", async () => { + const setup = chatSetup(); + const state = probe(); + state.gate = deferred(); + install(setup, state); + setup.registry.install(tools); + setup.faux.setResponses([call("edit", "a"), fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const submission = await root.submit({ type: "input", content: "go" }, context); + // The input settles with the answer, before the task finishes. + expect((await submission.wait(context)).status).toBe("done"); + await eventually(() => state.runs.length === 1); + const live = (await harness.snapshot(LiveDoc, root.id, context))!; + expect(live.run?.inputs).toEqual([]); + expect(live.run?.afterRun?.tasks).toHaveLength(1); + const child = await harness.getTask(live.run!.afterRun!.tasks![0]!, context); + expect(child?.owner).toBe(live.run!.taskId); + expect(child?.background).toBe(false); + const idle = root.waitForIdle(context); + expect(await settled(idle)).toBe(false); + + const entries = await allEntries(root); + const answer = kindOf(entries, "pi.assistant").at(-1)!; + expect(state.asked).toEqual([ + { reason: "answer", first: kindOf(entries, "pi.user")[0]!.id, answer: answer.id, last: answer.id }, + ]); + state.gate.resolve(); + await idle; + expect((await harness.snapshot(LiveDoc, root.id, context))?.run).toBeUndefined(); + expect((await harness.snapshot(LiveDoc, root.id, context))?.runFailure).toBeUndefined(); + expect((await allEntries(root)).at(-1)?.kind).toBe("test.tests"); + const generation = await harness.getTask(live.run!.taskId, context); + expect(generation?.state).toMatchObject({ + status: "terminal", + outcome: { status: "completed", result: { entryId: answer.id } }, + }); + await harness.close(context); + }); + + // #11141 + it("asks after a terminate round, the run's last entry its newest result or its handoff's reset", async () => { + for (const [tool, lastKind] of [ + ["finish", "pi.tool-result"], + ["handoff", "pi.reset"], + ] as const) { + const setup = chatSetup(); + const state = probe(); + install(setup, state); + setup.registry.install(tools); + setup.faux.setResponses([call(tool, "a")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + const entries = await allEntries(root); + const assistant = kindOf(entries, "pi.assistant").at(-1)!; + expect(state.asked).toEqual([ + { + reason: "terminate", + first: kindOf(entries, "pi.user")[0]!.id, + answer: assistant.id, + last: kindOf(entries, lastKind).at(-1)!.id, + }, + ]); + expect(state.runs).toEqual(["run 1"]); + expect((await harness.snapshot(LiveDoc, root.id, context))?.run).toBeUndefined(); + await harness.close(context); + } + }); + + // #11141 + it("ends the run in one commit without an afterRun hook, and after the hook when it asks for nothing", async () => { + const plain = chatSetup(); + const state = probe(); + plain.registry.install(defineExtension({ name: "tasks-only", tasks: [testsTask(state)] })); + plain.faux.setResponses([fauxAssistantMessage("done")]); + const opened = await openChat(new MemoryStorage(), plain); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + // No hook: the answer's commit ended the run. + expect(await opened.harness.snapshot(LiveDoc, opened.root.id, context)).toEqual({}); + await opened.harness.close(context); + + const setup = chatSetup(); + install(setup, state, async (run) => { + state.asked.push(run); + return undefined; + }); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + expect(state.asked).toHaveLength(1); + expect(state.runs).toEqual([]); + expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({}); + await harness.close(context); + }); + + // #11141 + it("queues input during the tasks and places it after them: writes first, then steers and follow-ups", async () => { + const setup = chatSetup(); + const state = probe(); + state.gate = deferred(); + const Tests = testsTask(state); + setup.registry.install( + defineExtension({ + name: "after-run", + tasks: [Tests], + hooks: [ + hook(GenerationTask, { + afterRun: async (run) => { + state.asked.push(run); + // Only the first run gets a task. + return state.asked.length === 1 ? { tasks: [runEndTask(Tests, { label: "run 1" })] } : undefined; + }, + }), + ], + }), + ); + // The task admits its result for the model in its own commit. + state.admit = (handle, tx) => + handle.submit(tx, { + type: "write", + entry: { kind: "test.result", model: [{ role: "user", content: "tests passed", timestamp: 1 }] }, + }); + setup.faux.setResponses([ + fauxAssistantMessage("first"), + fauxAssistantMessage("second"), + fauxAssistantMessage("third"), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.runs.length === 1); + + const steer = await root.submit({ type: "input", content: "steer", whenBusy: "steer" }, context); + const followUp = await root.submit({ type: "input", content: "later" }, context); + await expect(root.submit({ type: "input", content: "x", whenBusy: "reject" }, context)).rejects.toBeInstanceOf( + ConversationBusy, + ); + await expect(root.submit({ type: "continue" }, context)).rejects.toBeInstanceOf(ConversationBusy); + // Busy, so not idle: it queues and starts the next run after the tasks. + const idleSteer = await root.submit( + { type: "input", content: "steer 2", whenBusy: "steer", whenIdle: "reject" }, + context, + ); + const note = await root.submit({ type: "write", entry: { kind: "test.note" } }, context); + expect((await note.status(context)).status).toBe("done"); + expect((await steer.status(context)).status).toBe("queued"); + expect(await harness.snapshot(InboxDoc, root.id, context)).toMatchObject({ items: { length: 3 } }); + + state.gate.resolve(); + await root.waitForIdle(context); + for (const submission of [steer, followUp, idleSteer]) { + expect((await submission.wait(context)).status).toBe("done"); + } + const entries = await allEntries(root); + const kinds = entries.filter((entry) => entry.kind !== "pi.system").map((entry) => entry.kind); + expect(kinds).toEqual([ + "pi.user", + "pi.assistant", + "test.note", + "test.tests", + "test.result", + "pi.user", + "pi.user", + "pi.assistant", + "pi.user", + "pi.assistant", + ]); + expect(userTexts(entries)).toEqual(["go", "tests passed", "steer", "later", "steer 2"]); + await harness.close(context); + }); + + // #11141 + it("places a reset queued during the tasks after them", async () => { + const setup = chatSetup(); + const state = probe(); + state.gate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.runs.length === 1); + await root.reset(undefined, context); + expect(kindOf(await allEntries(root), "pi.reset")).toEqual([]); + state.gate.resolve(); + await root.waitForIdle(context); + const kinds = (await allEntries(root)).map((entry) => entry.kind); + expect(kinds.slice(-2)).toEqual(["test.tests", "pi.reset"]); + await harness.close(context); + }); + + // #11141 + it("asks once a run really ends: not while onYield continues it, and when a queued steer drops the continuation", async () => { + const setup = chatSetup(); + const state = probe(); + install(setup, state); + let yields = 0; + setup.registry.install( + defineExtension({ + name: "again", + hooks: [hook(GenerationTask, { onYield: async () => (++yields <= 2 ? { continue: "again" } : undefined) })], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("1"), fauxAssistantMessage("2"), fauxAssistantMessage("3")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + const entries = await allEntries(root); + expect(state.asked).toEqual([ + { + reason: "answer", + first: kindOf(entries, "pi.user")[0]!.id, + answer: kindOf(entries, "pi.assistant").at(-1)!.id, + last: kindOf(entries, "pi.assistant").at(-1)!.id, + }, + ]); + await harness.close(context); + + // A steer queued while the answer's hooks run: the continuation is dropped, and the run ends after its task. + const dropped = chatSetup(); + const second = probe(); + install(dropped, second); + let root2: Awaited>["root"] | undefined; + dropped.registry.install( + defineExtension({ + name: "steering", + hooks: [ + hook(GenerationTask, { + onYield: async () => { + if (second.asked.length > 0) return undefined; + await root2!.submit({ type: "input", content: "steer", whenBusy: "steer" }, context); + return { continue: "again" }; + }, + }), + ], + }), + ); + dropped.faux.setResponses([fauxAssistantMessage("1"), fauxAssistantMessage("2")]); + const opened = await openChat(new MemoryStorage(), dropped); + root2 = opened.root; + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await opened.root.waitForIdle(context); + expect(second.asked).toHaveLength(2); + expect(second.runs).toEqual(["run 1", "run 2"]); + expect(userTexts(await allEntries(opened.root))).toEqual(["go", "steer"]); + await opened.harness.close(context); + }); + + // #11141 + it("asks the hooks again after a crash before the tasks exist, and never once they do", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const state = probe(); + state.hookGate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.asked.length === 1); + opened.kill(); // a crash while the hook runs, in `finish` + state.hookGate = undefined; + state.gate = deferred(); + opened = await openKillable(path, setup); + opened.harness.resume(); + await eventually(() => state.runs.length === 1); + expect(state.asked).toHaveLength(2); + opened.kill(); // a crash while the task runs, in `drain` + state.gate = undefined; + opened = await openKillable(path, setup); + await opened.root.waitForIdle(context); + // The task ran again; the hook was not asked again, and only one task exists. + expect(state.asked).toHaveLength(2); + expect(state.runs).toEqual(["run 2", "run 2"]); + expect(kindOf(await allEntries(opened.root), "test.tests")).toHaveLength(1); + const tasks = (await opened.harness.inspect(context)).tasks; + expect(tasks).toEqual([]); + await opened.harness.close(context); + + // An orderly close in `drain` resumes the same way. + const closing = await sqlitePath(); + const other = probe(); + other.gate = deferred(); + const setup2 = chatSetup(); + install(setup2, other); + setup2.faux.setResponses([fauxAssistantMessage("done")]); + let reopened = await openKillable(closing, setup2); + await (await reopened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => other.runs.length === 1); + await reopened.harness.close(context); + other.gate = undefined; + reopened = await openKillable(closing, setup2); + await reopened.root.waitForIdle(context); + expect(other.asked).toHaveLength(1); + expect(other.runs).toEqual(["run 1", "run 1"]); + await reopened.harness.close(context); + }); + + // #11141 + it("ends an answered run without a failure when Stop comes while the hooks run or the tasks run", async () => { + for (const stage of ["finish", "drain"] as const) { + const setup = chatSetup(); + const state = probe(); + const Tests = testsTask(state); + state.gate = deferred(); + setup.registry.install( + defineExtension({ + name: "after-run", + tasks: [Tests], + hooks: [ + hook(GenerationTask, { + afterRun: async (run, _api, ctx) => { + state.asked.push(run); + if (stage === "finish") await aborted(ctx.abortSignal!); + return { tasks: [runEndTask(Tests, { label: "run 1" })] }; + }, + }), + ], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const submission = await root.submit({ type: "input", content: "go" }, context); + await submission.wait(context); + await eventually(() => (stage === "finish" ? state.asked.length === 1 : state.runs.length === 1)); + await root.abort(context); + const live = await harness.snapshot(LiveDoc, root.id, context); + expect(live?.run).toBeUndefined(); + expect(live?.runFailure).toBeUndefined(); + expect((await submission.status(context)).status).toBe("done"); + expect(state.aborted).toBe(stage === "drain" ? 1 : 0); + expect(state.runs).toEqual(stage === "drain" ? ["run 1"] : []); + await harness.close(context); + } + }); + + // #11141 + it("ends the run normally when one of its tasks is aborted on its own", async () => { + const setup = chatSetup(); + const state = probe(); + state.gate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.runs.length === 1); + const live = (await harness.snapshot(LiveDoc, root.id, context))!; + await harness.abortTask(live.run!.afterRun!.tasks![0]!, context); + await root.waitForIdle(context); + expect(state.aborted).toBe(1); + expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({}); + expect((await harness.getTask(live.run!.taskId, context))?.state).toMatchObject({ + status: "terminal", + outcome: { status: "completed" }, + }); + await harness.close(context); + }); + + // #11141 + it("reports broken requests and hooks and creates the rest", async () => { + const setup = chatSetup(); + const state = probe(); + const Tests = testsTask(state); + const Throwing = defineTask({ + name: "test.throwing", + version: 1, + initial: () => { + throw new Error("initial failed"); + }, + phases: { run: async () => {} }, + abort: async () => {}, + }); + // Its `initial()` starts at a phase the definition does not have. + const BadStart = defineTask({ + name: "test.bad-start", + version: 1, + initial: () => ({ phase: "nowhere" }) as unknown as { phase: "run" }, + phases: { run: async () => {} }, + abort: async () => {}, + }); + const requests: RunEndTask[] = [ + { task: { definition: { ...Tests.definition, name: "test.unknown" } }, input: null }, + { task: GenerationTask, input: {} }, + { task: Tests, input: { label: () => "not JSON" } as unknown as JsonValue }, + runEndTask(Throwing, null), + runEndTask(BadStart, null), + runEndTask(Tests, { label: "ok" }), + ]; + setup.registry.install( + defineExtension({ + name: "after-run", + tasks: [Tests, Throwing, BadStart], + hooks: [hook(GenerationTask, { afterRun: async () => ({ tasks: requests }) })], + }), + ); + setup.registry.install( + defineExtension({ + name: "broken", + hooks: [ + hook(GenerationTask, { + afterRun: async () => { + throw new Error("hook failed"); + }, + }), + ], + }), + ); + setup.registry.install( + defineExtension({ + name: "not-a-list", + hooks: [hook(GenerationTask, { afterRun: async () => ({ tasks: "x" as unknown as RunEndTask[] }) })], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + expect(state.runs).toEqual(["ok"]); + // One report per hook, against the hook's extension; the requests of one hook come together. + expect(setup.infos.filter((info) => info.kind === "extension_error")).toEqual([ + expect.objectContaining({ extension: "after-run" }), + expect.objectContaining({ extension: "broken" }), + expect.objectContaining({ extension: "not-a-list" }), + ]); + const rejected = setup.reports[0] as AggregateError; + expect(rejected.errors.map((error: Error) => error.message)).toEqual([ + "Task test.unknown is not installed", + "An afterRun hook cannot start the built-in task pi.generation", + "Value contains a non-JSON function; expected strict JSON", + "initial failed", + "Task test.bad-start starts at the unknown phase nowhere", + ]); + expect((setup.reports[1] as Error).message).toBe("hook failed"); + expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({}); + await harness.close(context); + }); + + // #11141 + it("creates a requested task from the installed definition, calling its initial() once, off the line", async () => { + const setup = chatSetup(); + const calls = { installed: 0, token: 0 }; + const define = (version: number, count: () => void) => + defineTask({ + name: "test.versioned", + version, + initial: () => { + count(); + return { phase: "run" }; + }, + phases: { run: async (_task, runtime, ctx) => runtime.commit(() => completed(null), ctx) }, + abort: async () => {}, + }); + const Installed = define(2, () => calls.installed++); + // The hook's extension holds an older copy of the task. + const Stale = define(1, () => calls.token++); + setup.registry.install(defineExtension({ name: "tasks", tasks: [Installed] })); + setup.registry.install( + defineExtension({ + name: "asker", + hooks: [hook(GenerationTask, { afterRun: async () => ({ tasks: [runEndTask(Stale, null)] }) })], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + const created = await harness.commit( + (tx) => tx.scanTasks({ conversationId: root.id, kind: "test.versioned" }, 10), + context, + ); + expect(created.items).toHaveLength(1); + expect(created.items[0]).toMatchObject({ version: 2, state: { status: "terminal" } }); + expect(calls).toEqual({ installed: 1, token: 0 }); + await harness.close(context); + }); + + // #11141 + it("lets a hook read the run's history, tool results included", async () => { + const setup = chatSetup(); + const kinds: string[] = []; + setup.registry.install(tools); + setup.registry.install( + defineExtension({ + name: "reader", + hooks: [ + hook(GenerationTask, { + afterRun: async (run, api, ctx) => { + const page = await api.entries( + { minEntryId: run.first, maxEntryId: run.last, order: "ascending" }, + 100, + undefined, + ctx, + ); + kinds.push(...page.items.map((entry) => entry.kind)); + expect((await api.entry(run.answer, ctx))?.kind).toBe("pi.assistant"); + return undefined; + }, + }), + ], + }), + ); + setup.faux.setResponses([call("edit", "a"), fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + expect(kinds.filter((kind) => kind !== "pi.system")).toEqual([ + "pi.user", + "pi.assistant", + "pi.tool-result", + "pi.assistant", + ]); + await harness.close(context); + }); + + // #11141 + it("lets a hook read the run's history after a compaction summarized the start of the run", async () => { + const setup = chatSetup(); + setup.settings.compaction = { enabled: false, reserveTokens: 1000, keepRecentTokens: 1, backgroundTokens: 0 }; + const gate = deferred(); + const started = deferred(); + const runs: string[][] = []; + setup.registry.install( + defineExtension({ + name: "slow", + tools: [ + defineTool({ + name: "edit", + description: "Edit a file", + parameters: Parameters, + execute: async () => { + started.resolve(); + await gate.promise; + return { output: [{ type: "text", text: "edited" }] }; + }, + }), + ], + hooks: [ + hook(CompactionTask, { beforeCompact: async () => ({ summary: "earlier work" }) }), + hook(GenerationTask, { + afterRun: async (run, api, ctx) => { + const page = await api.entries( + { minEntryId: run.first, maxEntryId: run.last, order: "ascending" }, + 100, + undefined, + ctx, + ); + runs.push(page.items.map((entry) => entry.kind)); + return undefined; + }, + }), + ], + }), + ); + setup.faux.setResponses([call("edit", "a"), fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const submission = await root.submit({ type: "input", content: "go" }, context); + await started.promise; + // Its summary is placed at the round's post-tools boundary, in the middle of the run. + const compaction = await root.compact(undefined, context); + await harness.waitForTask(compaction, context); + gate.resolve(); + await submission.wait(context); + await root.waitForIdle(context); + const active = (await root.context(context)).entries.map((entry) => entry.kind); + // The active context lost the run's input to the summary; the run's history did not. + expect(active).not.toContain("pi.user"); + expect(runs).toHaveLength(1); + expect(runs[0]!.filter((kind) => kind !== "pi.system")).toEqual([ + "pi.user", + "pi.assistant", + "pi.tool-result", + "pi.compaction", + "pi.assistant", + ]); + await harness.close(context); + }); + + // #11141 + it("ends an answered run completed, without a failure, once finish was cut off restartLimit times", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + setup.settings.restartLimit = 1; + const state = probe(); + state.hookGate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.asked.length === 1); + await opened.root.submit({ type: "input", content: "later" }, context); + opened.kill(); + opened = await openKillable(path, setup); + await opened.root.waitForIdle(context); + expect(state.asked).toHaveLength(1); + const live = await opened.harness.snapshot(LiveDoc, opened.root.id, context); + expect(live?.run).toBeUndefined(); + expect(live?.runFailure).toBeUndefined(); + // Queued items stay, as after a failed run. + expect((await opened.harness.snapshot(InboxDoc, opened.root.id, context))?.items).toHaveLength(1); + expect(setup.infos).toContainEqual( + expect.objectContaining({ kind: "task_interrupted", reason: "restart_limit" }), + ); + await opened.harness.close(context); + }); + + // #11141 + it("keeps the run busy while an after-run task cannot run, until Stop orphans it", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const state = probe(); + state.gate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.runs.length === 1); + opened.kill(); + // Reopened without the extension: its task is blocked. + const bare = chatSetup(); + opened = await openKillable(path, bare); + opened.harness.resume(); + const live = (await opened.harness.snapshot(LiveDoc, opened.root.id, context))!; + expect(live.run?.afterRun?.tasks).toHaveLength(1); + await eventually(async () => + (await opened.harness.inspect(context)).tasks.some((task) => task.state.kind === "blocked"), + ); + expect(await settled(opened.root.waitForIdle(context))).toBe(false); + await opened.root.abort(context); + const after = await opened.harness.snapshot(LiveDoc, opened.root.id, context); + expect(after?.run).toBeUndefined(); + expect(after?.runFailure).toBeUndefined(); + expect((await opened.harness.getTask(live.run!.afterRun!.tasks![0]!, context))?.state).toMatchObject({ + status: "terminal", + outcome: { status: "orphaned" }, + }); + await opened.harness.close(context); + }); + + // #11141 + it("starts a fix run with a follow-up a task admits, after the tasks", async () => { + const setup = chatSetup(); + const state = probe(); + const Tests = testsTask(state); + setup.registry.install( + defineExtension({ + name: "after-run", + tasks: [Tests], + hooks: [ + hook(GenerationTask, { + afterRun: async (run) => { + state.asked.push(run); + return state.asked.length === 1 ? { tasks: [runEndTask(Tests, { label: "run 1" })] } : undefined; + }, + }), + ], + }), + ); + state.admit = (handle, tx) => + handle.submit(tx, { type: "input", content: "fix the tests", requestId: "fix-1", whenBusy: "followUp" }); + setup.faux.setResponses([fauxAssistantMessage("done"), fauxAssistantMessage("fixed")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + const entries = await allEntries(root); + expect(entries.filter((entry) => entry.kind !== "pi.system").map((entry) => entry.kind)).toEqual([ + "pi.user", + "pi.assistant", + "test.tests", + "pi.user", + "pi.assistant", + ]); + expect(userTexts(entries)).toEqual(["go", "fix the tests"]); + expect(state.asked).toHaveLength(2); + // The fix run's history starts with its own input. + expect(state.asked[1]!.first).toBe(kindOf(entries, "pi.user")[1]!.id); + await harness.close(context); + }); + + // #11141 + it("gives a continue after a failed run the failed run's history", async () => { + const setup = chatSetup(); + const state = probe(); + install(setup, state); + setup.registry.install(tools); + setup.faux.setResponses([call("edit", "a"), fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await root.configure({ maxTurns: 1 }, context); + const failed = await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(failed.status).toBe("unanswered"); + const user = kindOf(await allEntries(root), "pi.user")[0]!; + expect((await harness.snapshot(LiveDoc, root.id, context))?.runFailure).toMatchObject({ + reason: "turn_limit", + first: user.id, + }); + expect(state.asked).toEqual([]); + await harness.commit((tx) => root.submit(tx, { type: "continue" }), context); + await root.waitForIdle(context); + const answer = kindOf(await allEntries(root), "pi.assistant").at(-1)!; + expect(state.asked).toEqual([{ reason: "answer", first: user.id, answer: answer.id, last: answer.id }]); + await harness.close(context); + }); + + // #11141 + it("ends the run without asking when the hooks are deselected before finish runs", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const state = probe(); + state.hookGate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.asked.length === 1); + opened.kill(); + setup.registry.uninstall(defineExtension({ name: "after-run" })); + opened = await openKillable(path, setup); + await opened.root.waitForIdle(context); + expect(state.asked).toHaveLength(1); + expect(state.runs).toEqual([]); + expect(await opened.harness.snapshot(LiveDoc, opened.root.id, context)).toEqual({}); + await opened.harness.close(context); + }); + + // #11141 + it("starts a fork of a conversation running its after-run tasks idle", async () => { + const setup = chatSetup(); + const state = probe(); + state.gate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.runs.length === 1); + await root.submit({ type: "input", content: "later" }, context); + const answer = kindOf(await allEntries(root), "pi.assistant").at(-1)!; + const fork = await root.fork(answer.id, { ownership: { kind: "ownerless" } }, context); + expect((await harness.snapshot(LiveDoc, fork.id, context))?.run).toBeUndefined(); + expect((await harness.snapshot(InboxDoc, fork.id, context))?.items ?? []).toEqual([]); + state.gate.resolve(); + await root.waitForIdle(context); + await harness.close(context); + }); + + // #11141 + it("gives a run without a recorded first entry its answer as first", async () => { + const setup = chatSetup(); + const state = probe(); + install(setup, state); + let root: Awaited>["root"] | undefined; + setup.registry.install( + defineExtension({ + name: "legacy", + hooks: [ + hook(GenerationTask, { + // As a run started before `first` existed. + onYield: async () => { + await root!.commit(async (tx) => { + delete (await tx.doc(LiveDoc, root!.id)).run!.first; + }, context); + return undefined; + }, + }), + ], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const opened = await openChat(new MemoryStorage(), setup); + root = opened.root; + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + await root.waitForIdle(context); + const answer = kindOf(await allEntries(root), "pi.assistant").at(-1)!; + expect(state.asked).toEqual([{ reason: "answer", first: answer.id, answer: answer.id, last: answer.id }]); + await opened.harness.close(context); + }); + + // #11141 + it("ends the run normally when its after-run task is cut off restartLimit times", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + setup.settings.restartLimit = 1; + const state = probe(); + state.gate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.runs.length === 1); + const generation = (await opened.harness.snapshot(LiveDoc, opened.root.id, context))!.run!.taskId; + // The task is cut off, and the waiting generation is not: the scheduler faults the task, and the run ends. + opened.kill(); + state.gate = undefined; + opened = await openKillable(path, setup); + await opened.root.waitForIdle(context); + expect((await opened.harness.getTask(generation, context))?.state).toMatchObject({ + status: "terminal", + outcome: { status: "completed" }, + }); + expect((await opened.harness.snapshot(LiveDoc, opened.root.id, context))?.runFailure).toBeUndefined(); + expect(state.runs).toEqual(["run 1"]); + expect(setup.infos).toContainEqual(expect.objectContaining({ kind: "task_faulted", reason: "restart_limit" })); + await opened.harness.close(context); + }); + + // #11141 + it("reports no task_interrupted for a task that ends normally after the limit was lowered", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const state = { started: 0 }; + const Worker = defineTask({ + name: "test.worker", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime, ctx) => { + state.started++; + if (state.started === 1) return new Promise(() => {}); // the run a crash cuts off + // The host lowers the limit while the rerun runs. + setup.settings.restartLimit = 1; + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + setup.registry.install(defineExtension({ name: "worker", tasks: [Worker] })); + let opened = await openKillable(path, setup); + const id = await opened.root.commit( + (tx) => tx.createTask(Worker, null, { ownership: { kind: "conversation" } }), + context, + ); + opened.harness.resume(); + await eventually(() => state.started === 1); + opened.kill(); + opened = await openKillable(path, setup); + expect((await opened.harness.waitForTask(id, context)).state.outcome.status).toBe("completed"); + expect(setup.infos.filter((info) => info.kind === "task_interrupted")).toEqual([]); + setup.settings.restartLimit = undefined; + await opened.harness.close(context); + }); + + // #11141 + it("reports no task_interrupted for a generation that ends normally after the limit was lowered in finish", async () => { + const path = await sqlitePath(); + const setup = chatSetup(); + const state = probe(); + state.hookGate = deferred(); + install(setup, state); + setup.faux.setResponses([fauxAssistantMessage("done")]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.asked.length === 1); + opened.kill(); // `finish` cut off once + const rerun = deferred(); + state.hookGate = rerun; + opened = await openKillable(path, setup); + opened.harness.resume(); + await eventually(() => state.asked.length === 2); + // The host lowers the limit while the hook runs; the phase had started under the old one and ends normally. + setup.settings.restartLimit = 1; + rerun.resolve(); + await opened.root.waitForIdle(context); + expect(state.runs).toEqual(["run 2"]); + expect(setup.infos.filter((info) => info.kind === "task_interrupted")).toEqual([]); + await opened.harness.close(context); + }); + + // #11141 + it("asks after a terminate round at the turn limit: terminate wins", async () => { + const setup = chatSetup(); + const state = probe(); + install(setup, state); + setup.registry.install(tools); + setup.faux.setResponses([call("finish", "a")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await root.configure({ maxTurns: 1 }, context); + expect((await (await root.submit({ type: "input", content: "go" }, context)).wait(context)).status).toBe("done"); + await root.waitForIdle(context); + expect(state.asked).toMatchObject([{ reason: "terminate" }]); + expect(state.runs).toEqual(["run 1"]); + expect((await harness.snapshot(LiveDoc, root.id, context))?.runFailure).toBeUndefined(); + await harness.close(context); + }); + + // #11141 + it("starts a continue's history at its first tool round when nothing failed and nothing was placed", async () => { + const setup = chatSetup(); + const kinds: string[][] = []; + setup.registry.install(tools); + setup.registry.install( + defineExtension({ + name: "reader", + hooks: [ + hook(GenerationTask, { + afterRun: async (run, api, ctx) => { + const range = { minEntryId: run.first, maxEntryId: run.last, order: "ascending" } as const; + kinds.push((await api.entries(range, 100, undefined, ctx)).items.map((entry) => entry.kind)); + return undefined; + }, + }), + ], + }), + ); + setup.faux.setResponses([call("edit", "a"), fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + // A host writes the user message itself, then continues. + await root.submit( + { type: "write", entry: { kind: "app.prompt", model: [{ role: "user", content: "go", timestamp: 1 }] } }, + context, + ); + await (await root.submit({ type: "continue" }, context)).wait(context); + await root.waitForIdle(context); + expect(kinds).toEqual([["pi.assistant", "pi.tool-result", "pi.assistant"]]); + await harness.close(context); + }); + + // #11141 + it("leaves a summary queued during the tasks stale behind a terminate round's handoff", async () => { + const setup = chatSetup(); + setup.settings.compaction = { enabled: false, reserveTokens: 1000, keepRecentTokens: 1, backgroundTokens: 0 }; + const state = probe(); + install(setup, state); + const gate = deferred(); + const started = deferred(); + setup.registry.install( + defineExtension({ + name: "slow-handoff", + tools: [ + defineTool({ + name: "handoff", + description: "End the run in a new context", + parameters: Parameters, + execute: async () => { + started.resolve(); + await gate.promise; + return { + output: [{ type: "text", text: "handed off" }], + control: { terminate: true, handoff: "fresh start" }, + }; + }, + }), + ], + hooks: [hook(CompactionTask, { beforeCompact: async () => ({ summary: "earlier work" }) })], + }), + ); + setup.faux.setResponses([call("handoff", "a")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const submission = await root.submit({ type: "input", content: "go" }, context); + await started.promise; + const compaction = await root.compact(undefined, context); + await harness.waitForTask(compaction, context); + gate.resolve(); + await submission.wait(context); + await root.waitForIdle(context); + const entries = await allEntries(root); + // The handoff's reset landed with the answer; the summary, cutting before it, settled stale at the run's end. + expect(kindOf(entries, "pi.compaction")).toEqual([]); + expect(entries.map((entry) => entry.kind).slice(-2)).toEqual(["pi.reset", "test.tests"]); + expect(state.asked[0]).toMatchObject({ reason: "terminate", last: kindOf(entries, "pi.reset")[0]!.id }); + await harness.close(context); + }); + + // #11141 + it("stores the answer's held progress with commit A, and asks again after an orderly close in finish", async () => { + const path = await sqlitePath(); + const setup = chatSetup({ tokensPerSecond: 200, tokenSize: { min: 1, max: 1 } }); + setup.settings.progress = { persistWithinMs: Number.POSITIVE_INFINITY }; + const state = probe(); + const Tests = testsTask(state); + let hookGate: Deferred | undefined = deferred(); + setup.registry.install( + defineExtension({ + name: "after-run", + tasks: [Tests], + hooks: [ + hook(GenerationTask, { + afterRun: async (run, _api, ctx) => { + state.asked.push(run); + if (hookGate !== undefined) await Promise.race([hookGate.promise, aborted(ctx.abortSignal!)]); + return { tasks: [runEndTask(Tests, { label: "run" })] }; + }, + }), + ], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("x".repeat(40))]); + let opened = await openKillable(path, setup); + await (await opened.root.submit({ type: "input", content: "go" }, context)).wait(context); + await eventually(() => state.asked.length === 1); + opened.kill(); + opened = await openKillable(path, setup); + const stored = (await opened.harness.snapshot(LiveDoc, opened.root.id, context))!; + expect(stored.generation).toBeUndefined(); + expect(stored.run?.afterRun).toEqual({}); + expect(textOf(kindOf(await allEntries(opened.root), "pi.assistant")[0]!.model![0])).toBe("x".repeat(40)); + opened.harness.resume(); + await eventually(() => state.asked.length === 2); + await opened.harness.close(context); + hookGate = undefined; + opened = await openKillable(path, setup); + await opened.root.waitForIdle(context); + expect(state.asked).toHaveLength(3); + expect(state.runs).toEqual(["run"]); + await opened.harness.close(context); + }); +}); + +describe("endRun", () => { + const tx = { settleSubmission: () => {} } as unknown as Tx; + + // #11141 + it("leaves a run it does not own alone, and ends an answered run without a failure", () => { + const live = { + run: { taskId: 1 as TaskId, inputs: [], turns: 0, afterRun: { tasks: [5 as TaskId] } }, + } as unknown as Draft; + endRun(tx, live, 2 as TaskId, { status: "unanswered", reason: "faulted" }); + expect(live.run?.afterRun).toEqual({ tasks: [5] }); + endRun(tx, live, 1 as TaskId, { status: "unanswered", reason: "aborted" }); + expect(live.run).toBeUndefined(); + expect(live.runFailure).toBeUndefined(); + }); + + // #11141 + it("copies a failed run's first entry into runFailure", () => { + const live = { run: { taskId: 1 as TaskId, inputs: [], turns: 0, first: 7 } } as unknown as Draft; + endRun(tx, live, 1 as TaskId, { status: "unanswered", reason: "model_error" }); + expect(live.runFailure).toMatchObject({ reason: "model_error", first: 7 }); + }); +}); diff --git a/packages/durable/test/harness-generation-recovery.test.ts b/packages/durable/test/harness-generation-recovery.test.ts index 52c0f62be..d88c34ea7 100644 --- a/packages/durable/test/harness-generation-recovery.test.ts +++ b/packages/durable/test/harness-generation-recovery.test.ts @@ -197,7 +197,7 @@ describe("generation recovery", () => { await new Promise((resolve) => setTimeout(resolve, 20)); expect(setup.faux.state.callCount).toBe(1); expect(await opened.harness.snapshot(LiveDoc, opened.root.id, context)).toEqual({ - run: { taskId, inputs: [id], turns: 0 }, + run: { taskId, inputs: [id], turns: 0, first: expect.any(Number) }, generation: { attempt: 1, retry: { at: 61_000, error: "503 Service Unavailable" } }, }); now = 61_000; diff --git a/packages/durable/test/harness-generation.test.ts b/packages/durable/test/harness-generation.test.ts index 46a3fe58f..a835aea34 100644 --- a/packages/durable/test/harness-generation.test.ts +++ b/packages/durable/test/harness-generation.test.ts @@ -156,7 +156,9 @@ describe("generation", () => { // The earlier settlement stays; the run's own settlement leaves it unchanged. expect(await submission.status(context)).toMatchObject({ status: "unanswered", reason: "withdrawn" }); // A run ends without an answer even when its inputs were settled already. - expect(await live(harness, root)).toEqual({ runFailure: { taskId, inputs: [submission.id], reason: "aborted" } }); + expect(await live(harness, root)).toEqual({ + runFailure: { taskId, inputs: [submission.id], reason: "aborted", first: expect.any(Number) }, + }); await harness.close(context); }); @@ -186,6 +188,7 @@ describe("generation", () => { inputs: [expect.any(Number)], reason: "no_model", detail: "Model faux/missing is not available", + first: expect.any(Number), }, }); expect((await live(harness, plain))?.runFailure).toMatchObject({ @@ -315,7 +318,9 @@ describe("generation", () => { const converted = entries[1]!.model![0] as AssistantMessage; expect(converted.stopReason).toBe("aborted"); expect(textOf(converted)!.startsWith(textOf(partial as Message)!)).toBe(true); - expect(await live(harness, root)).toEqual({ runFailure: { taskId, inputs: [submission.id], reason: "aborted" } }); + expect(await live(harness, root)).toEqual({ + runFailure: { taskId, inputs: [submission.id], reason: "aborted", first: expect.any(Number) }, + }); expect((await harness.waitForTask(taskId, context)).state.outcome).toEqual({ status: "aborted" }); await harness.close(context); }); @@ -332,7 +337,9 @@ describe("generation", () => { await harness.abortTask(taskId, context); expect(await submission.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); expect(setup.faux.state.cancelledDeferred).toHaveLength(1); - expect(await live(harness, root)).toEqual({ runFailure: { taskId, inputs: [submission.id], reason: "aborted" } }); + expect(await live(harness, root)).toEqual({ + runFailure: { taskId, inputs: [submission.id], reason: "aborted", first: expect.any(Number) }, + }); await harness.close(context); }); @@ -356,7 +363,9 @@ describe("generation", () => { await harness.abortTask(taskId, context); expect(await submission.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); expect(setup.reports).toContainEqual(new Error("cancel failed")); - expect(await live(harness, root)).toEqual({ runFailure: { taskId, inputs: [submission.id], reason: "aborted" } }); + expect(await live(harness, root)).toEqual({ + runFailure: { taskId, inputs: [submission.id], reason: "aborted", first: expect.any(Number) }, + }); await harness.close(context); }); @@ -691,6 +700,7 @@ describe("generation", () => { inputs: [submission.id], reason: "faulted", detail: expect.stringContaining("non-JSON function"), + first: expect.any(Number), }, }); const tasks = await harness.commit((tx) => tx.scanTasks({ conversationId: root.id }, 10), context); diff --git a/packages/durable/test/harness-input-entries.test.ts b/packages/durable/test/harness-input-entries.test.ts index 44d68a9ac..f0c44b9fb 100644 --- a/packages/durable/test/harness-input-entries.test.ts +++ b/packages/durable/test/harness-input-entries.test.ts @@ -377,6 +377,7 @@ describe("data-only writes", () => { inputs: [input.id], reason: "model_error", detail: "invalid request", + first: expect.any(Number), }, }); expect(setup.faux.state.callCount).toBe(1); diff --git a/packages/durable/test/harness-live-deltas.test.ts b/packages/durable/test/harness-live-deltas.test.ts index 696820e7c..518399392 100644 --- a/packages/durable/test/harness-live-deltas.test.ts +++ b/packages/durable/test/harness-live-deltas.test.ts @@ -131,7 +131,13 @@ describe("pi.live deltas", () => { const result = entries.items.find((entry) => entry.kind === "pi.tool-result")!.id; expect(commits).toEqual([ // submission - [["s", ["run"], { taskId: firstGeneration, inputs: [expect.any(Number)], turns: 0 }]], + [ + [ + "s", + ["run"], + { taskId: firstGeneration, inputs: [expect.any(Number)], turns: 0, first: expect.any(Number) }, + ], + ], // request [["s", ["generation"], { attempt: 1 }]], // the generation starts its tool round and keeps the run, counting the round as one turn (#11043) diff --git a/packages/durable/test/harness-restart-failure.test.ts b/packages/durable/test/harness-restart-failure.test.ts index f0552a953..356f2318e 100644 --- a/packages/durable/test/harness-restart-failure.test.ts +++ b/packages/durable/test/harness-restart-failure.test.ts @@ -220,7 +220,7 @@ describe("a generation interrupted while the model answers (#11139)", () => { detail: text, }); expect(await live(opened.harness)).toEqual({ - runFailure: { taskId, inputs: [id], reason: "interrupted", detail: text }, + runFailure: { taskId, inputs: [id], reason: "interrupted", detail: text, first: expect.any(Number) }, }); expect((await opened.harness.waitForTask(taskId, context)).state.outcome).toEqual({ status: "failed", @@ -351,7 +351,7 @@ describe("a generation interrupted while the model answers (#11139)", () => { detail: text, }); expect(await live(opened.harness)).toEqual({ - runFailure: { taskId, inputs: [id], reason: "restart_limit", detail: text }, + runFailure: { taskId, inputs: [id], reason: "restart_limit", detail: text, first: expect.any(Number) }, }); await waitFor(() => setup.infos.length > 0); expect(setup.infos).toContainEqual( @@ -563,7 +563,7 @@ describe("crash edges (#11139)", () => { detail: text, }); expect(await live(opened.harness)).toEqual({ - runFailure: { taskId, inputs: [id], reason: "interrupted", detail: text }, + runFailure: { taskId, inputs: [id], reason: "interrupted", detail: text, first: expect.any(Number) }, }); expect(setup.faux.state.cancelledDeferred).toHaveLength(1); await opened.harness.close(context); @@ -634,7 +634,7 @@ describe("crash edges (#11139)", () => { detail: text, }); expect(await live(opened.harness)).toEqual({ - runFailure: { taskId, inputs: [id], reason: "restart_limit", detail: text }, + runFailure: { taskId, inputs: [id], reason: "restart_limit", detail: text, first: expect.any(Number) }, }); const entries = await allEntries(opened.root); expect(entries.map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant"]); @@ -693,7 +693,9 @@ describe("crash edges (#11139)", () => { } 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 live(opened.harness)).toEqual({ + runFailure: { taskId, inputs: [id], reason: "aborted", detail, first: expect.any(Number) }, + }); expect((await opened.harness.waitForTask(taskId, context)).state.outcome).toMatchObject({ status: "faulted", error: { detail: { reason: "restart_limit" } }, diff --git a/packages/durable/test/harness-run-failure.test.ts b/packages/durable/test/harness-run-failure.test.ts index 4c21cc970..ded8c8ebc 100644 --- a/packages/durable/test/harness-run-failure.test.ts +++ b/packages/durable/test/harness-run-failure.test.ts @@ -49,6 +49,7 @@ describe("pi.live.runFailure", () => { inputs: [submission.id], reason: "model_error", detail: "Invalid request", + first: expect.any(Number), } satisfies RunFailure); const next = await root.submit({ type: "input", content: "again" }, context); // The commit that starts the next run removes it. @@ -85,7 +86,12 @@ describe("pi.live.runFailure", () => { const taskId = (await live(harness, root)).run!.taskId; await root.abort(context); expect(await submission.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); - expect((await live(harness, root)).runFailure).toEqual({ taskId, inputs: [submission.id], reason: "aborted" }); + expect((await live(harness, root)).runFailure).toEqual({ + taskId, + inputs: [submission.id], + reason: "aborted", + first: expect.any(Number), + }); await harness.close(context); }); diff --git a/packages/durable/test/harness-structured.test.ts b/packages/durable/test/harness-structured.test.ts index 87fa7c72d..ac7cc9b55 100644 --- a/packages/durable/test/harness-structured.test.ts +++ b/packages/durable/test/harness-structured.test.ts @@ -1253,7 +1253,7 @@ describe("tool rounds", () => { const results = (await allEntries(root)).filter((entry) => ToolResultEntry.is(entry)); expect(results.map((entry) => (entry.model![0] as ToolResultMessage).toolCallId)).toEqual(["c1", "c2"]); expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({ - runFailure: { taskId: generation, inputs: [submission.id], reason: "aborted" }, + runFailure: { taskId: generation, inputs: [submission.id], reason: "aborted", first: expect.any(Number) }, }); expect((await harness.waitForTask(generation, context)).state.outcome.status).toBe("aborted"); await harness.close(context); @@ -1407,6 +1407,7 @@ describe("tool rounds", () => { inputs: [submission.id], reason: "faulted", detail: "Value contains a non-JSON function; expected strict JSON", + first: expect.any(Number), }, }); expect((await allEntries(reopened.root)).map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant"]); diff --git a/packages/durable/test/harness-view-run.test.ts b/packages/durable/test/harness-view-run.test.ts index a87fc009c..61431cd13 100644 --- a/packages/durable/test/harness-view-run.test.ts +++ b/packages/durable/test/harness-view-run.test.ts @@ -103,7 +103,12 @@ describe("a run as the conversation view shows it", () => { expect(partialsIn(states).map(textOf)).toEqual(partials.map(textOf)); // The run shows from the input's commit until the commit that appends the answer and clears the partial. const started = states.findIndex((state) => state.run !== undefined); - expect(states[started]!.run).toEqual({ taskId: expect.any(Number), inputs: [submission.id], turns: 0 }); + expect(states[started]!.run).toEqual({ + taskId: expect.any(Number), + inputs: [submission.id], + turns: 0, + first: expect.any(Number), + }); expect(history.views[started]!.entries.at(-1)!.kind).toBe("pi.user"); const ended = states.findIndex((state, index) => index > started && state.run === undefined); expect(states[ended]!.generation).toBeUndefined();