mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(durable): run-end tasks with the afterRun hook
GenerationHooks.afterRun asks, after a run answered, for durable tasks (runEndTask(task, input)). The answer commits and the run's inputs settle done first; the generation then creates the tasks as its children in a finish phase and waits for them in drain, keeping pi.live.run (run.afterRun) and the conversation busy, so idle waits include them, Stop aborts them and a crash resumes them. Queued input is placed when they are terminal. pi.live.run.first and runFailure.first name where a run's history starts, also for a continue after a failure. closes #11141
This commit is contained in:
Mario Zechner
parent
65fba65b2e
commit
7e15f96019
21 changed files with 1954 additions and 86 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -628,6 +628,10 @@ type SettledSubmissionRecord = SubmissionRecord & {
|
|||
interface Submission {
|
||||
readonly id: SubmissionId;
|
||||
status(context: Context): Promise<SubmissionRecord>;
|
||||
/**
|
||||
* 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<SettledSubmissionRecord>;
|
||||
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<void>;
|
||||
/** 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<ToolTaskResult>[];
|
||||
/** 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
|
||||
|
|
|
|||
|
|
@ -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<Tool extends ToolRegistration = ToolRegistration>(
|
||||
|
|
@ -30,6 +40,14 @@ export function hook<K extends AnyTask>(task: K, handlers: Partial<HooksOf<K>>):
|
|||
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<I extends JsonValue, S extends { phase: string }, R, H extends object>(
|
||||
task: Task<I, S, R, H>,
|
||||
input: I,
|
||||
): RunEndTask {
|
||||
return { task, input };
|
||||
}
|
||||
|
||||
/** Wrap the tool named like `tool` wherever the wrapping extension is selected. */
|
||||
export function wrapTool<Tool extends ToolRegistration>(tool: Tool, wrapper: (tool: Tool) => Tool): Wrap<Tool> {
|
||||
return { tool: tool.name, wrap: wrapper };
|
||||
|
|
|
|||
|
|
@ -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<ToolTaskResult>[];
|
||||
/** 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<GenerationInput, GenerationCheckpoint,
|
|||
return { status: "waiting", checkpoint, on: [taskId], policy: "allSettled" };
|
||||
}, context);
|
||||
},
|
||||
/**
|
||||
* Ask the `afterRun` hooks for tasks and create them, as the generation's children, in the commit that moves to
|
||||
* `drain`; without any, end the run now.
|
||||
*/
|
||||
finish: async (task, runtime, context) => {
|
||||
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<Next> => {
|
||||
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<GenerationInput, GenerationCheckpoint,
|
|||
: harnessError("aborted", `Tool ${call.name} was aborted`);
|
||||
await appendToolResult(tx, conversationId, call, result, { timestamp: runtime.now() });
|
||||
}
|
||||
// In `finish` and `drain` the run answered already: it ends without a failure, its inputs stay `done`.
|
||||
endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "aborted" });
|
||||
return { status: "terminal", outcome: { status: "aborted" } };
|
||||
}, context);
|
||||
|
|
@ -594,8 +643,8 @@ async function endFailedRun(
|
|||
endRun(tx, live, runtime.taskId, end);
|
||||
if (failure.policy === "withdraw") await withdrawQueuedInputs(tx, conversationId, "run_failed");
|
||||
if (failure.boundary === undefined) return;
|
||||
const { users } = await applyBoundary(tx, failure.boundary, "final", runtime.now());
|
||||
if (users.length > 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<void> {
|
||||
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<Next> => {
|
||||
// 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<boolean> {
|
||||
const hooks = agentHooks(await runtime.agent(context), GenerationTask.definition.name);
|
||||
return hooks.some(({ handlers }) => typeof (handlers as Partial<GenerationHooks>).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<LiveState>, runtime: Runtime, end: Omit<RunSummary, "first">): 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<JsonValue, { phase: string }, JsonValue, object>; 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<void> {
|
||||
const conversationId = runtime.conversationId;
|
||||
await runtime.commit(async (tx): Promise<Next> => {
|
||||
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<void> {
|
||||
const text = `The generation was interrupted ${cutOffCount(task)} in its ${task.state.checkpoint.phase} phase`;
|
||||
await runtime.commit(async (tx): Promise<Next> => {
|
||||
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<Next> => {
|
||||
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<LiveState>,
|
||||
inputs: SubmissionId[],
|
||||
run: { readonly inputs: SubmissionId[]; readonly first: EntryId | undefined },
|
||||
options?: { readonly keepFailure?: boolean },
|
||||
): Promise<void> {
|
||||
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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<Extract<InboxItem, { mode: "write" }>>;
|
||||
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`. */
|
||||
|
|
|
|||
|
|
@ -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<LiveState>, 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<LiveState>, 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<AnyTaskRecord, "id" | "kind" | "conversationId"
|
|||
* The restart count starts over: it now counts cut-offs of the abort handler.
|
||||
*/
|
||||
/** Why a failed outcome's error says crashes ended the task (`detail.reason`), if it does. */
|
||||
/**
|
||||
* Terminal states a built-in task commits because it was cut off `settings.restartLimit` times, when their outcome does
|
||||
* not say so (a failure with detail `{ reason: "restart_limit" }` does), with the message the scheduler reports them
|
||||
* `task_interrupted` with. Keyed by the state object the task returns, so nothing is added to the stored state.
|
||||
*/
|
||||
const endedByCrashesStates = new WeakMap<object, string>();
|
||||
|
||||
/** `state`, marked as ending its task for crashes (`task_interrupted` with `message`), whatever its outcome. */
|
||||
export function endedByCrashes<T extends object>(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;
|
||||
|
|
|
|||
|
|
@ -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" };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
/**
|
||||
* 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 };
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
140
packages/durable/test/examples/32-tests-after-edits.ts
Normal file
140
packages/durable/test/examples/32-tests-after-edits.ts
Normal file
|
|
@ -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);
|
||||
1204
packages/durable/test/harness-after-run.test.ts
Normal file
1204
packages/durable/test/harness-after-run.test.ts
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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" } },
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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"]);
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue