mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(agent): add durable generation procedures
This commit is contained in:
Mario Zechner
parent
9f666bfbbb
commit
9b034fc45e
8 changed files with 2131 additions and 31 deletions
|
|
@ -173,7 +173,7 @@ TX[ upsert pi.op.state/O = effect_pending (reserves response n2, usage u1) ]
|
|||
… provider streams … ← the uncertain window
|
||||
TX[ append pi.pending.assistant_frame/O:n2 += frame ] ← one per streamed event,
|
||||
enqueued without awaiting
|
||||
TX[ insert entry n2, upsert pi.lane.leaf = n2, insert usage u1,
|
||||
TX[ insert entry n2, insert usage u1, upsert pi.lane.leaf = n2,
|
||||
delete list pi.pending.assistant_frame/O:n2,
|
||||
upsert pi.op.state/O = tools (result id n3 reserved) ]
|
||||
TX[ upsert pi.op.tool_args/O:s1:0, upsert pi.op.state/O = call 0 effect_pending ]
|
||||
|
|
@ -245,7 +245,7 @@ Source type provenance:
|
|||
|
||||
- `AgentMessage`, `AgentTool`, `AgentToolResult`, `QueueMode`, and `ThinkingLevel`: `packages/agent/src/types.ts`.
|
||||
- `Skill`, `PromptTemplate`, `AgentHarnessResources` (`Resources` below), `AgentHarnessTool`, `AgentHarnessToolContextSource`, `AgentHarnessToolInvocation`, `AgentHarnessToolUpdateCallback`, `AgentHarnessToolUpdateOptions`, `AgentHarnessStreamOptions`, and `AgentHarnessStreamOptionsPatch`: `packages/agent/src/harness/types.ts`.
|
||||
- `Model`, `Models`, `Usage`, `RetryPolicy`, `StopReason`, `AssistantMessage`, `ImageContent`, provider messages, stream options, and deferred handles: `packages/ai`. `AiContext` below aliases pi-ai's provider request `Context` to distinguish it from the harness invocation `Context`.
|
||||
- `Model`, `Models`, `Tool`, `Usage`, `RetryPolicy`, `StopReason`, `AssistantMessage`, `ImageContent`, provider messages, stream options, and deferred handles: `packages/ai`. `AiContext` below aliases pi-ai's provider request `Context` to distinguish it from the harness invocation `Context`.
|
||||
- `AssistantMessageFrame`, `assistantMessageEventToFrame`, and `reduceAssistantMessageFrames`: `packages/ai` (`@earendil-works/pi-ai`), `src/utils/assistant-message-frame.ts`. The harness defines no second frame codec or reducer.
|
||||
- `CompactionSettings`, `CompactionPreparation`, `CompactResult`, `BranchPreparation`, and `BranchSummaryResult`: `packages/agent/src/harness/compaction/`. Existing preparation and split-turn algorithms remain the implementation starting point unless this document explicitly changes them.
|
||||
- `TelemetryContext` and typed schema helpers: `packages/telemetry`; the agent-owned schemas remain in `packages/agent/src/harness/telemetry.ts`.
|
||||
|
|
@ -1397,14 +1397,14 @@ The complete transition table — every row is one `commit()`; classification or
|
|||
| assistant `effect_pending` | settlement classifies `deferred` with a valid handle | §3.7's deferred row | suspended, `poll: 0`, `sourceEntryId: R` |
|
||||
| suspended, poll *k* | captured model is unavailable when a drive supplies `pollDeferred: true` | `TX[ S(failure_drain{error:{code:"model_unavailable", ...}, provenance:{kind:"configuration"}}) ]` — the handle is abandoned without fabricating a fetch response or usage | failure_drain |
|
||||
| suspended, poll *k* | captured model resolves and a drive supplies `pollDeferred: true`: the poll's `before_request` settlement commits its intent, consuming the pass's single poll permit | mint fresh R′ and U′, then `TX[ S(deferred{effect_pending, poll k+1, responseEntryId R′, usageId U′}) ]` | effect_pending, poll *k*+1 |
|
||||
| effect_pending, poll *k*+1 | fetch returns **pending** with a completely equal handle | `TX[ insert response entry R′, upsert pi.lane.leaf = R′, insert usage U′, S(latestAssistantEntryId=R′, deferred{suspended, sourceEntryId R′, poll k+1}) ]` — the pending response becomes the next source and the operation re-suspends; no second poll this invocation | suspended, poll *k*+1 |
|
||||
| effect_pending | fetch returns **pending** with a mismatched handle | normalize to a durable `error` response explaining the mismatch: `TX[ insert normalized response R′, upsert pi.lane.leaf = R′, insert usage U′, S(latestAssistantEntryId=R′, failure_drain{error, provenance:response R′}) ]` | failure_drain |
|
||||
| effect_pending | fetch returns **ready** with tool calls | `TX[ insert response R′, upsert pi.lane.leaf = R′, insert usage U′, S(latestAssistantEntryId=R′, tools{plan with reserved result ids}) ]` — result ids minted as followers of R′ (§1.2) | tools |
|
||||
| effect_pending | fetch returns **ready** without tool calls | `TX[ insert response R′, upsert pi.lane.leaf = R′, insert usage U′, S(latestAssistantEntryId=R′, checkpoint{may_finish, includeFinalAssistant:true}) ]` | checkpoint |
|
||||
| effect_pending | fetch settles as a provider `error` | `TX[ insert response R′, upsert pi.lane.leaf = R′, insert usage U′, S(latestAssistantEntryId=R′, failure_drain{error, provenance:response R′}) ]` — polls have no retry path | failure_drain |
|
||||
| effect_pending, poll *k*+1 | fetch returns **pending** with a completely equal handle | `TX[ insert response entry R′, insert usage U′, upsert pi.lane.leaf = R′, S(latestAssistantEntryId=R′, deferred{suspended, sourceEntryId R′, poll k+1}) ]` — the pending response becomes the next source and the operation re-suspends; no second poll this invocation | suspended, poll *k*+1 |
|
||||
| effect_pending | fetch returns **pending** with a mismatched handle | normalize to a durable `error` response explaining the mismatch: `TX[ insert normalized response R′, insert usage U′, upsert pi.lane.leaf = R′, S(latestAssistantEntryId=R′, failure_drain{error, provenance:response R′}) ]` | failure_drain |
|
||||
| effect_pending | fetch returns **ready** with tool calls | `TX[ insert response R′, insert usage U′, upsert pi.lane.leaf = R′, S(latestAssistantEntryId=R′, tools{plan with reserved result ids}) ]` — result ids minted as followers of R′ (§1.2) | tools |
|
||||
| effect_pending | fetch returns **ready** without tool calls | `TX[ insert response R′, insert usage U′, upsert pi.lane.leaf = R′, S(latestAssistantEntryId=R′, checkpoint{may_finish, includeFinalAssistant:true}) ]` | checkpoint |
|
||||
| effect_pending | fetch settles as a provider `error` | `TX[ insert response R′, insert usage U′, upsert pi.lane.leaf = R′, S(latestAssistantEntryId=R′, failure_drain{error, provenance:response R′}) ]` — polls have no retry path | failure_drain |
|
||||
| effect_pending, restored, running control | crash left the poll's outcome unknown; a drive with a poll permit finds the captured model unavailable | `TX[ delete list pi.pending.assistant_frame/O:R′, S(failure_drain{error:{code:"model_unavailable", ...}, provenance:{kind:"configuration"}}) ]` — abandon the old reserved response/usage strings without fabricating settlement | failure_drain |
|
||||
| effect_pending, restored, running control | crash left the poll's outcome unknown; the next drive with a poll permit resolves the captured model and replaces it | mint fresh R″/U″ and commit a fresh intent at the **same** poll number — an unknown-outcome poll never completed, so `poll` does not increment; the old reserved id strings are abandoned, never materialized, and the intent transaction deletes the abandoned old frame list; the replacement poll starts a fresh list under its fresh response id | effect_pending, poll *k*+1 |
|
||||
| effect_pending, cancelled control | reconciliation, live or restored (§4.5, §4.6) | synthetic settlement under the **existing** reserved ids, preserving partial content reduced from committed frames: `TX[ insert synthetic aborted response R′, upsert pi.lane.leaf = R′, insert zero usage U′, delete frame list O:R′, S(latestAssistantEntryId=R′, cancelled checkpoint{may_finish}) ]` | cancelled checkpoint → aborted finish |
|
||||
| effect_pending, cancelled control | reconciliation, live or restored (§4.5, §4.6) | synthetic settlement under the **existing** reserved ids, preserving partial content reduced from committed frames: `TX[ insert synthetic aborted response R′, insert zero usage U′, upsert pi.lane.leaf = R′, delete frame list O:R′, S(latestAssistantEntryId=R′, cancelled checkpoint{may_finish}) ]` | cancelled checkpoint → aborted finish |
|
||||
| suspended, cancelled control | reconciliation | no fetch starts; best-effort `cancel_deferred` targets the newest source (§4.6), and the operation finishes through the aborted terminal transaction | terminal |
|
||||
|
||||
### Structural work
|
||||
|
|
@ -1559,13 +1559,13 @@ Pre-acceptance rejections write **nothing**: `LaneBusy`, `NothingToCompact`, `In
|
|||
| checkpoint `need_assistant` | ordinary procedure | snapshot current lane config, stream options, and normalized retry policy inline into the context in `TX[ S(assistant{ready, nextAttempt:1}) ]` | ready |
|
||||
| assistant `ready` | captured model or configured active-tool definition is unavailable before request admission | `TX[ S(failure_drain{error:{code:"model_unavailable" | "configured_tools_unavailable", ...}, provenance:{kind:"configuration"}}) ]` — no response/usage ids are reserved and no assistant entry or usage row is fabricated | failure_drain |
|
||||
| assistant `ready` | identities resolve and `before_request` completes | mint R and U, then `TX[ S(assistant{effect_pending, attempt=nextAttempt, responseEntryId R, usageId U, intendedOutputLimit, contextWindow}) ]` | effect_pending |
|
||||
| effect_pending | settles with tool calls | `TX[ insert response entry R, upsert pi.lane.leaf = R, insert usage U, delete frame list O:R, S(latestAssistantEntryId=R, tools{plan with reserved result ids}) ]` | tools |
|
||||
| effect_pending | retryable error, attempts remain | `TX[ insert response entry R, upsert pi.lane.leaf = R, insert usage U, S(latestAssistantEntryId=R, assistant{retry_wait, nextAttempt k+1, notBefore}) ]` | retry_wait |
|
||||
| effect_pending | first overflow, preparation non-empty | `TX[ insert response entry R **normalized to error**, upsert pi.lane.leaf = R, insert usage U, upsert pi.op.preparation/O:{taskId} = P, S(latestAssistantEntryId=R, compaction{reason:overflow, structural:{deciding, taskId}, resumeAfter:{checkpoint, prior trigger, need_assistant(true)}}) ]` | compaction |
|
||||
| effect_pending | first overflow, preparation empty | `TX[ insert normalized response entry R, upsert pi.lane.leaf = R, insert usage U, S(latestAssistantEntryId=R, failure_drain{error, provenance:response R}) ]` | failure_drain |
|
||||
| effect_pending | `stopReason: "deferred"` | `TX[ insert response entry R, upsert pi.lane.leaf = R, insert usage U, S(latestAssistantEntryId=R, deferred{suspended, sourceEntryId R, poll 0, configuration/options copied}) ]` | deferred |
|
||||
| effect_pending | `stop` or genuine `length` | `TX[ insert response entry R, upsert pi.lane.leaf = R, insert usage U, S(latestAssistantEntryId=R, checkpoint{may_finish, includeFinalAssistant:true}) ]` | checkpoint |
|
||||
| effect_pending | terminal error, retries exhausted, or 2nd overflow | `TX[ insert response entry R, upsert pi.lane.leaf = R, insert usage U, S(latestAssistantEntryId=R, failure_drain{error, provenance:response R}) ]` | failure_drain |
|
||||
| effect_pending | settles with tool calls | `TX[ insert response entry R, insert usage U, upsert pi.lane.leaf = R, delete frame list O:R, S(latestAssistantEntryId=R, tools{plan with reserved result ids}) ]` | tools |
|
||||
| effect_pending | retryable error, attempts remain | `TX[ insert response entry R, insert usage U, upsert pi.lane.leaf = R, S(latestAssistantEntryId=R, assistant{retry_wait, nextAttempt k+1, notBefore}) ]` | retry_wait |
|
||||
| effect_pending | first overflow, preparation non-empty | `TX[ insert response entry R **normalized to error**, insert usage U, upsert pi.lane.leaf = R, upsert pi.op.preparation/O:{taskId} = P, S(latestAssistantEntryId=R, compaction{reason:overflow, structural:{deciding, taskId}, resumeAfter:{checkpoint, prior trigger, need_assistant(true)}}) ]` | compaction |
|
||||
| effect_pending | first overflow, preparation empty | `TX[ insert normalized response entry R, insert usage U, upsert pi.lane.leaf = R, S(latestAssistantEntryId=R, failure_drain{error, provenance:response R}) ]` | failure_drain |
|
||||
| effect_pending | `stopReason: "deferred"` | `TX[ insert response entry R, insert usage U, upsert pi.lane.leaf = R, S(latestAssistantEntryId=R, deferred{suspended, sourceEntryId R, poll 0, configuration/options copied}) ]` | deferred |
|
||||
| effect_pending | `stop` or genuine `length` | `TX[ insert response entry R, insert usage U, upsert pi.lane.leaf = R, S(latestAssistantEntryId=R, checkpoint{may_finish, includeFinalAssistant:true}) ]` | checkpoint |
|
||||
| effect_pending | terminal error, retries exhausted, or 2nd overflow | `TX[ insert response entry R, insert usage U, upsert pi.lane.leaf = R, S(latestAssistantEntryId=R, failure_drain{error, provenance:response R}) ]` | failure_drain |
|
||||
| retry_wait | gated timer reaches `notBefore` | `TX[ S(assistant{ready, nextAttempt:k+1}) ]` | ready |
|
||||
|
||||
Every settlement row above — tool plan, retry, overflow, deferred, checkpoint, and failure alike — additionally executes `deleteList(pendingAssistantFrames(O, R))`; the first row shows it explicitly and the rest abbreviate. Settlement and frame cleanup are one atomic commit.
|
||||
|
|
@ -1702,7 +1702,7 @@ Structural provider streams are internal: they emit **no** public assistant-mess
|
|||
```
|
||||
TX[ insert e_41 = { …assistant response, stopReason: "error",
|
||||
errorMessage: "context window exceeded: …" },
|
||||
upsert pi.lane.leaf/main = "e_41", insert usage u_41,
|
||||
insert usage u_41, upsert pi.lane.leaf/main = "e_41",
|
||||
upsert pi.op.preparation/op_9:t_1 = <structural preparation>,
|
||||
S(compaction{ reason: overflow,
|
||||
structural: { deciding, taskId: "t_1" },
|
||||
|
|
@ -3152,7 +3152,9 @@ interface AssistantResponseMetadata {
|
|||
}
|
||||
|
||||
interface AssistantStreamObserver {
|
||||
start(message: AssistantMessage, context: Context): void | Promise<void>;
|
||||
start(message: AssistantMessage,
|
||||
event: Extract<AssistantMessageEvent, { type: "start" }> | undefined,
|
||||
context: Context): void | Promise<void>;
|
||||
update(message: AssistantMessage, event: AssistantMessageEvent,
|
||||
context: Context): void | Promise<void>;
|
||||
end(message: SettledAssistantMessage, context: Context): void | Promise<void>;
|
||||
|
|
@ -3161,7 +3163,7 @@ interface AssistantStreamObserver {
|
|||
interface HarnessAssistantStreamConfig {
|
||||
model: Model;
|
||||
systemPrompt: string;
|
||||
tools?: AgentTool[];
|
||||
tools?: Tool[];
|
||||
thinkingLevel: ThinkingLevel;
|
||||
streamOptions: AgentHarnessStreamOptions;
|
||||
transformContext?: (
|
||||
|
|
@ -3201,7 +3203,7 @@ transformContext(requestContext, context)
|
|||
→ map curated stream options + thinking level to SimpleStreamOptions
|
||||
→ install context.abortSignal, context.telemetryContext, beforePayload, and metadata capture
|
||||
→ request(aiContext, options, context)
|
||||
→ observer.start(message, context)
|
||||
→ observer.start(message, startEvent, context)
|
||||
→ observer.update(message, event, context)*
|
||||
→ settle the stream completely
|
||||
→ afterResponse(settled message, captured metadata, context)
|
||||
|
|
@ -3209,7 +3211,7 @@ transformContext(requestContext, context)
|
|||
→ return the final settled message
|
||||
```
|
||||
|
||||
It never mutates `messages`. Every callback receives the same explicit invocation Context unless its adapter deliberately derives a child span Context. The harness-supplied observer converts each streamed event with `assistantMessageEventToFrame` and synchronously enqueues its invocation-fenced frame append without awaiting storage (§3.7); `afterResponse` is always installed, even with no hook listeners, because it first stops frame admission and awaits the latest frame-write promise before invoking the optional `after_response` pipeline. If the stream terminates without a start event, it emits `observer.start` for the final message before `observer.end`, matching the harness lifecycle contract; this synthesized lifecycle start occurs after stream consumption and is not converted into or appended as a durable frame. If abort interrupts the parked `afterResponse` adapter, the block awaits the carried abort-mutation promise, skips that hook, emits `observer.end` with the raw settled message, and returns it so the caller can commit it under the now-current cancellation control. `beforePayload` maps to pi-ai's payload callback. Response metadata capture maps to pi-ai's `onResponse`; it is distinct from `afterResponse`, because `onResponse` runs before the response body is consumed while the harness hook transforms the settled assistant message afterward. The harness exposes neither callback through `AgentHarnessStreamOptions`.
|
||||
It never mutates `messages`. Every callback receives the same explicit invocation Context unless its adapter deliberately derives a child span Context. The harness-supplied observer converts each streamed event with `assistantMessageEventToFrame` and synchronously enqueues its invocation-fenced frame append without awaiting storage (§3.7); `afterResponse` is always installed, even with no hook listeners, because it first stops frame admission and awaits the latest frame-write promise before invoking the optional `after_response` pipeline. If the stream terminates without a start event, it emits `observer.start` with `event: undefined` for the final message before `observer.end`, matching the harness lifecycle contract; this synthesized lifecycle start occurs after stream consumption and is not converted into or appended as a durable frame. If abort interrupts the parked `afterResponse` adapter, the block awaits the carried abort-mutation promise, skips that hook, emits `observer.end` with the raw settled message, and returns it so the caller can commit it under the now-current cancellation control. `beforePayload` maps to pi-ai's payload callback. Response metadata capture maps to pi-ai's `onResponse`; it is distinct from `afterResponse`, because `onResponse` runs before the response body is consumed while the harness hook transforms the settled assistant message afterward. The harness exposes neither callback through `AgentHarnessStreamOptions`.
|
||||
|
||||
The request function, not this block, owns registry dispatch, auth, and operation admission:
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
# WP05 — Direct durable drive
|
||||
|
||||
**Status: M2 complete; M3 next.** Independent Opus and GPT-5.6-Sol reviews approved the authoritative-`Lane.state`, straight-line procedure correction and M2 implementation with no blockers.
|
||||
**Status: M3 complete; M4 next.** Independent Opus, GPT-5.6-Sol, and Fable reviews approved the authoritative-`Lane.state` design and M3 generation implementation with no blockers.
|
||||
|
||||
Removes breakpoints/manual drive, then implements the complete durable execution graph. Public drive stays disabled until every reachable phase and reconciliation path exists.
|
||||
|
||||
|
|
@ -1316,7 +1316,7 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig
|
|||
**Modify**
|
||||
| Path | Change |
|
||||
|---|---|
|
||||
| `src/harness/execution/assistant.ts` | `request` admitted through `Gate`; observer receives Context |
|
||||
| `src/harness/execution/assistant.ts` | use provider-facing `Tool` declarations; observer callbacks receive trailing Context, and `start` also receives the raw start event (or `undefined` for a synthesized lifecycle start) so frame persistence uses pi-ai's codec; the runtime request adapter admits `Models.streamSimple` through `Gate` |
|
||||
|
||||
**Behavior/invariants landed.** Preparation precedes intent; ids reserved at intent and honoured at settlement; frames enqueued synchronously without per-frame awaits; settlement deletes the frame list; every transition plans from the current owned projection and publishes its exact next projection; checkpoint separates tree parent/leaf from `triggerEntryId` for unprojected custom writes.
|
||||
|
||||
|
|
@ -1672,7 +1672,7 @@ Both methods run on `Lane`; procedures do not remove ownership. The external fin
|
|||
|
||||
`compact()` and `navigateTree()` are *accept + drive*, so they cannot be written before `accept` admits their operation kinds. Doing the composition first would force a second admission path that only the convenience method can reach.
|
||||
|
||||
**Behavior/invariants landed.** Owner-before-idle claim order; join observes delivered events; usage row `seq` from `CommitResult`, and usage-event totals read `commit.stats.usage` inside `events(commit)` — no process accumulator (§2.12); every hook/event/observation carries the emitting Context. `AgentHarness.watchSession` is **explicitly deferred** and keeps its rejection.
|
||||
**Behavior/invariants landed.** Owner-before-idle claim order; join observes delivered events; usage row `seq` from `CommitResult`, and usage-event totals read `commit.stats.usage` inside `events(commit)` — no process accumulator (§2.12); every hook/event/observation carries the emitting Context. Before public exposure, pi-ai provider streams must guarantee that a queued start event's partial cannot be mutated before `assistantMessageEventToFrame` consumes it; pin this with a back-to-back shared-partial provider test that does not yield between start and block-start publication. `AgentHarness.watchSession` is **explicitly deferred** and keeps its rejection.
|
||||
|
||||
**Focused tests**
|
||||
| Path | Theme |
|
||||
|
|
@ -1681,7 +1681,7 @@ Both methods run on `Lane`; procedures do not remove ownership. The external fin
|
|||
| `test/harness/runtime2/drive-ownership.test.ts` (independent completion/removal) | run each case with a **non-cooperative provider** and again with a **non-cooperative tool** that ignores its signal and never settles. **External finalization**: after the terminal event is delivered, public completion resolves with the finalizer's outcome while the effect is still stuck; `accept` returns `LaneBusy` throughout; exactly **one** `run_end` is delivered; after the effect finally returns the owner clears and `accept` succeeds. **Invocation abandonment**: the public promise rejects `DriveAbandoned`, a new `drive` claims the lane **before** the old effect returns, durable state is still `effect_pending`, and the new pass runs orphan recovery. **Close/fault**: resolves without waiting; public promises reject `HarnessClosed`/`HarnessFault`. In every case the detached pass produces no unhandled rejection, and `Drive.settle`/`Drive.fail` and exact `removeDrive` remain idempotent |
|
||||
| `test/harness/runtime2/drive-ownership.test.ts` (ABA fencing) | abandon an invocation whose effect is still running, let a **replacement owner claim the same `operationId`**, then let the old effect return: its transition planner and its progress writes both decline via `isDriveActive` even though the operation id is unchanged; zero commits attributed to the old task (asserted through `InstrumentedStorage`); the replacement owner is **not** cleared by the old task's cleanup; the replacement's own commits succeed |
|
||||
| `test/harness/runtime2/drive-surfaces.test.ts` | `accept({kind:"compaction"})` and `accept({kind:"navigation"})` admit and drive before any convenience method is used; convenience = accept + drive; `resume`; `requestAbort` through the public method reaches the M7 primitive; `Closed` returned not thrown; listeners settle before the next procedure hook; usage totals per settlement equal `getStats()`; **reopen a session with historical usage** → the first new usage event's totals include the history; **two lanes committing usage concurrently** → each event's totals match its own commit position and the final totals equal `getStats()`; Context lineage |
|
||||
| `test/harness/runtime2/drive-crash-matrix.test.ts` | release N commits → discard → reopen → drive, for every N in every phase path; every gated boundary exposes its exact committed writes; gated vs ungated byte-identical writes |
|
||||
| `test/harness/runtime2/drive-crash-matrix.test.ts` | release N commits → discard → reopen → drive, for every N in every phase path; every gated boundary exposes its exact committed writes; gated vs ungated byte-identical writes; a provider that publishes start and then mutates one shared partial for block start without yielding still persists a reducible immutable start frame |
|
||||
|
||||
**Commands.** `npm run check` · `./test.sh`
|
||||
|
||||
|
|
|
|||
|
|
@ -7,8 +7,9 @@ import type {
|
|||
Message,
|
||||
Model,
|
||||
SimpleStreamOptions,
|
||||
Tool,
|
||||
} from "@earendil-works/pi-ai";
|
||||
import type { AgentMessage, AgentTool, ThinkingLevel } from "../../types.ts";
|
||||
import type { AgentMessage, ThinkingLevel } from "../../types.ts";
|
||||
import type { Context } from "../context.ts";
|
||||
import type { SettledAssistantMessage } from "../session/types.ts";
|
||||
import type { AgentHarnessStreamOptions } from "../types.ts";
|
||||
|
|
@ -22,7 +23,11 @@ export interface AssistantResponseMetadata {
|
|||
|
||||
/** Process-local lifecycle observer for one assistant stream. */
|
||||
export interface AssistantStreamObserver {
|
||||
start(message: AssistantMessage, context: Context): void | Promise<void>;
|
||||
start(
|
||||
message: AssistantMessage,
|
||||
event: Extract<AssistantMessageEvent, { type: "start" }> | undefined,
|
||||
context: Context,
|
||||
): void | Promise<void>;
|
||||
update(message: AssistantMessage, event: AssistantMessageEvent, context: Context): void | Promise<void>;
|
||||
end(message: SettledAssistantMessage, context: Context): void | Promise<void>;
|
||||
}
|
||||
|
|
@ -31,7 +36,7 @@ export interface AssistantStreamObserver {
|
|||
export interface HarnessAssistantStreamConfig {
|
||||
model: Model<Api>;
|
||||
systemPrompt: string;
|
||||
tools?: AgentTool[];
|
||||
tools?: Tool[];
|
||||
thinkingLevel: ThinkingLevel;
|
||||
streamOptions: AgentHarnessStreamOptions;
|
||||
transformContext?: (
|
||||
|
|
@ -126,7 +131,7 @@ export async function streamHarnessAssistant(
|
|||
for await (const event of stream) {
|
||||
if (event.type === "start") {
|
||||
started = true;
|
||||
await config.observer.start({ ...event.partial }, context);
|
||||
await config.observer.start({ ...event.partial }, event, context);
|
||||
} else if (isUpdateEvent(event)) {
|
||||
await config.observer.update({ ...event.partial }, event, context);
|
||||
}
|
||||
|
|
@ -134,7 +139,7 @@ export async function streamHarnessAssistant(
|
|||
|
||||
const settled = (await stream.result()) as SettledAssistantMessage;
|
||||
if (!started) {
|
||||
await config.observer.start({ ...settled }, context);
|
||||
await config.observer.start({ ...settled }, undefined, context);
|
||||
}
|
||||
let finalMessage = settled;
|
||||
if (config.afterResponse) {
|
||||
|
|
|
|||
608
packages/agent/src/harness/runtime2/drive/checkpoint.ts
Normal file
608
packages/agent/src/harness/runtime2/drive/checkpoint.ts
Normal file
|
|
@ -0,0 +1,608 @@
|
|||
import type { AgentMessage } from "../../../types.ts";
|
||||
import type { HarnessEvent, TerminalOperationOutcome } from "../../agent-harness.ts";
|
||||
import { insertEntry } from "../../session/commit.ts";
|
||||
import { buildSessionContext } from "../../session/context.ts";
|
||||
import { SessionInvariantError } from "../../session/session.ts";
|
||||
import type {
|
||||
CheckpointPhase,
|
||||
Entry,
|
||||
LaneLastResult,
|
||||
MessageEntry,
|
||||
NewEntry,
|
||||
RunState,
|
||||
SettledAssistantMessage,
|
||||
} from "../../session/types.ts";
|
||||
import {
|
||||
deleteValue,
|
||||
laneLastResult as laneLastResultValue,
|
||||
laneLeaf,
|
||||
laneState as laneStateValue,
|
||||
operationState as operationStateValue,
|
||||
pendingEntry,
|
||||
setValue,
|
||||
} from "../../session/values.ts";
|
||||
import type { Lane } from "../lane.ts";
|
||||
import type { Drive, LaneCommand, LostOwnership, ProcedureResult } from "../types.ts";
|
||||
import { operationCleanupWrites } from "./terminal.ts";
|
||||
|
||||
type ReadResult<T> = { kind: "value"; value: T } | { kind: "state_changed" } | LostOwnership;
|
||||
|
||||
function isSettledAssistant(message: AgentMessage): message is SettledAssistantMessage {
|
||||
return message.role === "assistant" && message.stopReason !== "pending";
|
||||
}
|
||||
|
||||
function sameCheckpoint(current: CheckpointPhase, expected: CheckpointPhase): boolean {
|
||||
return (
|
||||
current.triggerEntryId === expected.triggerEntryId &&
|
||||
current.thresholdCheckedTriggerEntryId === expected.thresholdCheckedTriggerEntryId &&
|
||||
current.skipInboxOnce === expected.skipInboxOnce &&
|
||||
current.continuation.kind === expected.continuation.kind &&
|
||||
(current.continuation.kind !== "need_assistant" ||
|
||||
(expected.continuation.kind === "need_assistant" &&
|
||||
current.continuation.overflowRecoveryUsed === expected.continuation.overflowRecoveryUsed)) &&
|
||||
(current.continuation.kind !== "may_finish" ||
|
||||
(expected.continuation.kind === "may_finish" &&
|
||||
current.continuation.includeFinalAssistant === expected.continuation.includeFinalAssistant))
|
||||
);
|
||||
}
|
||||
|
||||
function lifecycleEvents(entry: Entry, lane: string, runId: string): HarnessEvent[] {
|
||||
return entry.type === "message"
|
||||
? [
|
||||
{ type: "message_start", lane, runId, message: entry.message },
|
||||
{ type: "message_end", lane, runId, message: entry.message, entryId: entry.id },
|
||||
{ type: "entry_added", lane, entry },
|
||||
]
|
||||
: [{ type: "entry_added", lane, entry }];
|
||||
}
|
||||
|
||||
async function readRunContext<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
phase: CheckpointPhase,
|
||||
): Promise<ReadResult<AgentMessage[]>> {
|
||||
const entries = await lane.commandDriveOwned<ReadResult<Entry[]>>(
|
||||
drive,
|
||||
async (state, reader): Promise<LaneCommand<ReadResult<Entry[]>>> => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Checkpoint context reached a non-run operation");
|
||||
}
|
||||
if (operation.state.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "state_changed" } as const };
|
||||
}
|
||||
if (operation.state.phase.kind !== "checkpoint" || !sameCheckpoint(operation.state.phase, phase)) {
|
||||
throw new SessionInvariantError("Checkpoint context lost its durable boundary");
|
||||
}
|
||||
if (state.leafId === null) throw new SessionInvariantError("Run checkpoint has no branch leaf");
|
||||
const path = (
|
||||
await reader.scanBranch(
|
||||
{ start: state.leafId, stopAtType: "compaction", order: "newestFirst" },
|
||||
drive.context,
|
||||
)
|
||||
).reverse();
|
||||
return { kind: "return", result: { kind: "value", value: path } as const };
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
if (entries.kind !== "value") return entries;
|
||||
return {
|
||||
kind: "value",
|
||||
value: await buildSessionContext(
|
||||
entries.value,
|
||||
{ entryProjectors: lane.readConfig().entryProjectors },
|
||||
drive.context,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
/** Consume before_run and commit the initial checkpoint. */
|
||||
export async function startRun<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
run: RunState,
|
||||
): Promise<ProcedureResult> {
|
||||
if (run.phase.kind !== "starting") throw new SessionInvariantError("startRun requires the starting phase");
|
||||
const prompt = await lane.commandDriveOwned<ReadResult<AgentMessage[]>>(
|
||||
drive,
|
||||
async (state, reader): Promise<LaneCommand<ReadResult<AgentMessage[]>>> => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run" || operation.meta.intent.kind !== "run") {
|
||||
throw new SessionInvariantError("before_run reached a non-run operation");
|
||||
}
|
||||
if (operation.state.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "state_changed" } as const };
|
||||
}
|
||||
if (operation.state.phase.kind !== "starting") {
|
||||
throw new SessionInvariantError("before_run lost the starting boundary");
|
||||
}
|
||||
const entries = await reader.getEntries(operation.meta.intent.promptEntryIds, drive.context);
|
||||
const messages = operation.meta.intent.promptEntryIds.map((id) => {
|
||||
const entry = entries.get(id);
|
||||
if (entry?.type !== "message") {
|
||||
throw new SessionInvariantError(`Run prompt entry ${id} is missing its message`);
|
||||
}
|
||||
return entry.message;
|
||||
});
|
||||
return { kind: "return", result: { kind: "value", value: messages } as const };
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
if (prompt.kind === "lost_ownership") return prompt;
|
||||
if (prompt.kind === "state_changed") return { kind: "continue" };
|
||||
|
||||
const hook = await lane.hooks.runWithGate(
|
||||
"before_run",
|
||||
{ lane: lane.name, runId: drive.operationId, prompt: prompt.value, resources: lane.readConfig().resources },
|
||||
drive.gate,
|
||||
drive.context,
|
||||
);
|
||||
const injected = hook?.messages ?? [];
|
||||
for (const message of injected) {
|
||||
if (message.role === "assistant" && message.stopReason === "pending") {
|
||||
throw new SessionInvariantError("before_run returned a pending assistant message");
|
||||
}
|
||||
}
|
||||
const reserved = injected.map((message) => ({ id: lane.session.idGenerator.next(), message }));
|
||||
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
(state) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("before_run settlement reached a non-run operation");
|
||||
}
|
||||
if (operation.state.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (operation.state.phase.kind !== "starting") {
|
||||
throw new SessionInvariantError("before_run settlement lost the starting boundary");
|
||||
}
|
||||
let parentId = state.leafId;
|
||||
const entries = reserved.map(({ id, message }) => {
|
||||
const entry: NewEntry<MessageEntry> = { id, parentId, type: "message", message };
|
||||
parentId = id;
|
||||
return entry;
|
||||
});
|
||||
const triggerEntryId = parentId;
|
||||
if (triggerEntryId === null) throw new SessionInvariantError("Run start has no trigger entry");
|
||||
const nextState: RunState = {
|
||||
...operation.state,
|
||||
phase: {
|
||||
kind: "checkpoint",
|
||||
continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
|
||||
triggerEntryId,
|
||||
},
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
...entries.map((entry) => insertEntry(entry)),
|
||||
...(entries.length === 0 ? [] : [setValue(laneLeaf(lane.name), triggerEntryId)]),
|
||||
setValue(operationStateValue(drive.operationId), nextState),
|
||||
],
|
||||
next: {
|
||||
...state,
|
||||
leafId: triggerEntryId,
|
||||
operation: { meta: operation.meta, state: nextState },
|
||||
},
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
events: (commit) =>
|
||||
entries.flatMap((entry, index) =>
|
||||
lifecycleEvents(
|
||||
{ ...entry, seq: commit.seqs[index]!, timestamp: commit.timestamp },
|
||||
lane.name,
|
||||
drive.operationId,
|
||||
),
|
||||
),
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
async function applyPendingWrites<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
phase: CheckpointPhase,
|
||||
): Promise<ProcedureResult> {
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
async (state, reader) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Pending-write drain reached a non-run operation");
|
||||
}
|
||||
const current = operation.state;
|
||||
if (current.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (current.phase.kind !== "checkpoint" || !sameCheckpoint(current.phase, phase)) {
|
||||
throw new SessionInvariantError("Pending-write drain lost its checkpoint boundary");
|
||||
}
|
||||
const ids = current.inbox.writes;
|
||||
if (ids.length === 0) return { kind: "return", result: { kind: "continue" } as const };
|
||||
const pending = await Promise.all(
|
||||
ids.map(async (id) => {
|
||||
const value = await reader.getValue(pendingEntry(id), drive.context);
|
||||
if (value === undefined) throw new SessionInvariantError(`Pending write ${id} is missing its payload`);
|
||||
return { id, pending: value.value };
|
||||
}),
|
||||
);
|
||||
let parentId = state.leafId;
|
||||
let triggerEntryId: string | undefined;
|
||||
const entries: NewEntry[] = pending.map(({ id, pending: item }) => {
|
||||
const entry: NewEntry =
|
||||
item.type === "message"
|
||||
? { id, parentId, type: "message", message: item.payload }
|
||||
: {
|
||||
id,
|
||||
parentId,
|
||||
type: "custom",
|
||||
customType: item.customType,
|
||||
...(item.payload === undefined ? {} : { data: item.payload }),
|
||||
};
|
||||
parentId = id;
|
||||
if (item.type === "message" || lane.readConfig().entryProjectors[item.customType] !== undefined) {
|
||||
triggerEntryId = id;
|
||||
}
|
||||
return entry;
|
||||
});
|
||||
const nextPhase: CheckpointPhase =
|
||||
triggerEntryId === undefined
|
||||
? current.phase
|
||||
: {
|
||||
kind: "checkpoint",
|
||||
continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
|
||||
triggerEntryId,
|
||||
skipInboxOnce: true,
|
||||
};
|
||||
const nextState: RunState = {
|
||||
...current,
|
||||
phase: nextPhase,
|
||||
inbox: { ...current.inbox, writes: [] },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
...entries.map((entry) => insertEntry(entry)),
|
||||
...ids.map((id) => deleteValue(pendingEntry(id))),
|
||||
setValue(laneLeaf(lane.name), parentId),
|
||||
setValue(operationStateValue(drive.operationId), nextState),
|
||||
],
|
||||
next: { ...state, leafId: parentId, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
events: (commit) =>
|
||||
entries.flatMap((entry, index) =>
|
||||
lifecycleEvents(
|
||||
{ ...entry, seq: commit.seqs[index]!, timestamp: commit.timestamp } as Entry,
|
||||
lane.name,
|
||||
drive.operationId,
|
||||
),
|
||||
),
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
async function consumeQueuedMessages<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
phase: CheckpointPhase,
|
||||
queue: "steer" | "followUp",
|
||||
): Promise<ProcedureResult> {
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
async (state, reader) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Queue drain reached a non-run operation");
|
||||
}
|
||||
const current = operation.state;
|
||||
if (current.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (current.phase.kind !== "checkpoint" || !sameCheckpoint(current.phase, phase)) {
|
||||
throw new SessionInvariantError("Queue drain lost its checkpoint boundary");
|
||||
}
|
||||
const allIds = current.inbox[queue];
|
||||
const mode = queue === "steer" ? current.settings.steeringMode : current.settings.followUpMode;
|
||||
const ids = mode === "all" ? allIds : allIds.slice(0, 1);
|
||||
if (ids.length === 0) return { kind: "return", result: { kind: "continue" } as const };
|
||||
const readMessages = (queuedIds: readonly string[], description: string) =>
|
||||
Promise.all(
|
||||
queuedIds.map(async (id) => {
|
||||
const value = await reader.getValue(pendingEntry(id), drive.context);
|
||||
if (value?.value.type !== "message") {
|
||||
throw new SessionInvariantError(`${description} ${id} is missing its message payload`);
|
||||
}
|
||||
return { entryId: id, message: value.value.payload };
|
||||
}),
|
||||
);
|
||||
const messages = (await readMessages(ids, `Queued ${queue} entry`)).map(({ entryId, message }) => ({
|
||||
id: entryId,
|
||||
message,
|
||||
}));
|
||||
const remainingSteerIds = queue === "steer" ? allIds.slice(ids.length) : current.inbox.steer;
|
||||
const remainingFollowUpIds = queue === "followUp" ? allIds.slice(ids.length) : current.inbox.followUp;
|
||||
const [remainingSteer, remainingFollowUp, nextRun] = await Promise.all([
|
||||
readMessages(remainingSteerIds, "Steer entry"),
|
||||
readMessages(remainingFollowUpIds, "Follow-up entry"),
|
||||
readMessages(state.pendingNextRun, "Pending next-run entry"),
|
||||
]);
|
||||
let parentId = state.leafId;
|
||||
const entries = messages.map(({ id, message }) => {
|
||||
const entry: NewEntry<MessageEntry> = { id, parentId, type: "message", message };
|
||||
parentId = id;
|
||||
return entry;
|
||||
});
|
||||
const triggerEntryId = entries.at(-1)?.id;
|
||||
if (triggerEntryId === undefined) throw new SessionInvariantError("Queue drain produced no trigger entry");
|
||||
const nextState: RunState = {
|
||||
...current,
|
||||
phase: {
|
||||
kind: "checkpoint",
|
||||
continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
|
||||
triggerEntryId,
|
||||
skipInboxOnce: true,
|
||||
},
|
||||
inbox: { ...current.inbox, [queue]: allIds.slice(ids.length) },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
...entries.map((entry) => insertEntry(entry)),
|
||||
...ids.map((id) => deleteValue(pendingEntry(id))),
|
||||
setValue(laneLeaf(lane.name), triggerEntryId),
|
||||
setValue(operationStateValue(drive.operationId), nextState),
|
||||
],
|
||||
next: {
|
||||
...state,
|
||||
leafId: triggerEntryId,
|
||||
operation: { meta: operation.meta, state: nextState },
|
||||
},
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
events: (commit) => [
|
||||
...entries.flatMap((entry, index) =>
|
||||
lifecycleEvents(
|
||||
{ ...entry, seq: commit.seqs[index]!, timestamp: commit.timestamp },
|
||||
lane.name,
|
||||
drive.operationId,
|
||||
),
|
||||
),
|
||||
{
|
||||
type: "queue_update",
|
||||
lane: lane.name,
|
||||
steer: remainingSteer,
|
||||
followUp: remainingFollowUp,
|
||||
nextRun,
|
||||
},
|
||||
],
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
async function finishRun<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
phase: CheckpointPhase,
|
||||
): Promise<ProcedureResult> {
|
||||
const context = await readRunContext(lane, drive, phase);
|
||||
if (context.kind === "lost_ownership") return context;
|
||||
if (context.kind === "state_changed") return { kind: "continue" };
|
||||
const hook = await lane.hooks.runWithGate(
|
||||
"before_run_end",
|
||||
{ lane: lane.name, runId: drive.operationId, messages: context.value },
|
||||
drive.gate,
|
||||
drive.context,
|
||||
);
|
||||
const followUp = hook?.followUp;
|
||||
const followUpId = followUp === undefined ? undefined : lane.session.idGenerator.next();
|
||||
const followUpMessage: AgentMessage | undefined =
|
||||
followUp === undefined ? undefined : { role: "user", content: followUp, timestamp: Date.now() };
|
||||
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
async (state, reader) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Run finish reached a non-run operation");
|
||||
}
|
||||
const current = operation.state;
|
||||
if (current.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (current.phase.kind !== "checkpoint" || !sameCheckpoint(current.phase, phase)) {
|
||||
throw new SessionInvariantError("Run finish lost its checkpoint boundary");
|
||||
}
|
||||
if (
|
||||
current.phase.continuation.kind !== "may_finish" ||
|
||||
current.inbox.steer.length !== 0 ||
|
||||
current.inbox.followUp.length !== 0 ||
|
||||
current.inbox.writes.length !== 0
|
||||
) {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (followUpId !== undefined && followUpMessage !== undefined) {
|
||||
const nextState: RunState = {
|
||||
...current,
|
||||
phase: {
|
||||
kind: "checkpoint",
|
||||
continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
|
||||
triggerEntryId: followUpId,
|
||||
},
|
||||
};
|
||||
const entry: NewEntry<MessageEntry> = {
|
||||
id: followUpId,
|
||||
parentId: state.leafId,
|
||||
type: "message",
|
||||
message: followUpMessage,
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
insertEntry(entry),
|
||||
setValue(laneLeaf(lane.name), followUpId),
|
||||
setValue(operationStateValue(drive.operationId), nextState),
|
||||
],
|
||||
next: {
|
||||
...state,
|
||||
leafId: followUpId,
|
||||
operation: { meta: operation.meta, state: nextState },
|
||||
},
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
events: (commit) =>
|
||||
lifecycleEvents(
|
||||
{ ...entry, seq: commit.seqs[0]!, timestamp: commit.timestamp },
|
||||
lane.name,
|
||||
drive.operationId,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
if (state.leafId === null) throw new SessionInvariantError("Completed run has no leaf");
|
||||
const finalId = current.phase.continuation.includeFinalAssistant ? current.latestAssistantEntryId : null;
|
||||
if (current.phase.continuation.includeFinalAssistant && finalId === null) {
|
||||
throw new SessionInvariantError("Completed run is missing its final assistant");
|
||||
}
|
||||
let finalEntry: { id: string; message: SettledAssistantMessage } | undefined;
|
||||
if (finalId !== null) {
|
||||
const entry = (await reader.getEntries([finalId], drive.context)).get(finalId);
|
||||
if (entry?.type !== "message") {
|
||||
throw new SessionInvariantError(`Run finish references invalid assistant ${finalId}`);
|
||||
}
|
||||
const message = entry.message;
|
||||
if (!isSettledAssistant(message)) {
|
||||
throw new SessionInvariantError(`Run finish references invalid assistant ${finalId}`);
|
||||
}
|
||||
finalEntry = { id: finalId, message };
|
||||
}
|
||||
const final =
|
||||
finalEntry === undefined ? {} : { finalEntryId: finalEntry.id, finalMessage: finalEntry.message };
|
||||
const outcome: TerminalOperationOutcome = {
|
||||
operation: "run",
|
||||
runId: drive.operationId,
|
||||
kind: "completed",
|
||||
leafId: state.leafId,
|
||||
...final,
|
||||
};
|
||||
const lastResult: LaneLastResult = {
|
||||
operationId: drive.operationId,
|
||||
kind: "run",
|
||||
outcome: "completed",
|
||||
leafId: state.leafId,
|
||||
runCompletion: finalId === null ? "terminated_tools" : "assistant",
|
||||
...(finalId === null ? {} : { finalAssistantEntryId: finalId }),
|
||||
};
|
||||
const cleanup = await operationCleanupWrites(reader, drive.operationId, current, drive.context);
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
...cleanup,
|
||||
setValue(laneLastResultValue(lane.name), lastResult),
|
||||
setValue(laneStateValue(lane.name), {
|
||||
currentOperationId: null,
|
||||
pendingNextRun: state.pendingNextRun,
|
||||
}),
|
||||
],
|
||||
next: { ...state, lastResult, operation: null },
|
||||
materialize: () => ({ kind: "settled", outcome }) as const,
|
||||
events: () => [
|
||||
{
|
||||
type: "run_end",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
leafId: state.leafId,
|
||||
outcome: "completed",
|
||||
...final,
|
||||
},
|
||||
],
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
/** Advance one durable run checkpoint by one visible transition. */
|
||||
export async function runCheckpoint<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
run: RunState,
|
||||
phase: CheckpointPhase,
|
||||
): Promise<ProcedureResult> {
|
||||
if (run.phase.kind !== "checkpoint" || !sameCheckpoint(run.phase, phase)) {
|
||||
throw new SessionInvariantError("runCheckpoint requires its current checkpoint phase");
|
||||
}
|
||||
if (run.control.status === "cancel_requested") return { kind: "continue" };
|
||||
if (phase.skipInboxOnce !== true && run.inbox.writes.length !== 0) {
|
||||
return applyPendingWrites(lane, drive, phase);
|
||||
}
|
||||
if (phase.skipInboxOnce !== true && run.inbox.steer.length !== 0) {
|
||||
return consumeQueuedMessages(lane, drive, phase, "steer");
|
||||
}
|
||||
if (phase.continuation.kind === "need_assistant") {
|
||||
const config = lane.readConfig();
|
||||
const retryPolicy = config.retryPolicy.enabled
|
||||
? { maxAttempts: config.retryPolicy.maxRetries + 1, baseDelayMs: config.retryPolicy.baseDelayMs }
|
||||
: { maxAttempts: 1, baseDelayMs: config.retryPolicy.baseDelayMs };
|
||||
const stepId = lane.session.idGenerator.next();
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
(state) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Generation start reached a non-run operation");
|
||||
}
|
||||
const current = operation.state;
|
||||
if (current.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (
|
||||
current.phase.kind !== "checkpoint" ||
|
||||
!sameCheckpoint(current.phase, phase) ||
|
||||
current.phase.continuation.kind !== "need_assistant"
|
||||
) {
|
||||
throw new SessionInvariantError("Generation start lost its checkpoint boundary");
|
||||
}
|
||||
if (
|
||||
current.phase.skipInboxOnce !== true &&
|
||||
(current.inbox.writes.length !== 0 || current.inbox.steer.length !== 0)
|
||||
) {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
const nextState: RunState = {
|
||||
...current,
|
||||
phase: {
|
||||
kind: "assistant",
|
||||
generation: {
|
||||
status: "ready",
|
||||
context: {
|
||||
stepId,
|
||||
triggerEntryId: current.phase.triggerEntryId,
|
||||
configuration: state.configuration,
|
||||
streamOptions: config.streamOptions,
|
||||
retryPolicy,
|
||||
overflowRecoveryUsed: current.phase.continuation.overflowRecoveryUsed,
|
||||
},
|
||||
nextAttempt: 1,
|
||||
},
|
||||
},
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [setValue(operationStateValue(drive.operationId), nextState)],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
if (run.inbox.followUp.length !== 0) {
|
||||
return consumeQueuedMessages(lane, drive, phase, "followUp");
|
||||
}
|
||||
return finishRun(lane, drive, phase);
|
||||
}
|
||||
678
packages/agent/src/harness/runtime2/drive/generation.ts
Normal file
678
packages/agent/src/harness/runtime2/drive/generation.ts
Normal file
|
|
@ -0,0 +1,678 @@
|
|||
import {
|
||||
type AssistantMessage,
|
||||
type AssistantMessageEvent,
|
||||
assistantMessageEventToFrame,
|
||||
isRetryableAssistantError,
|
||||
type Tool,
|
||||
} from "@earendil-works/pi-ai";
|
||||
import type { AgentMessage } from "../../../types.ts";
|
||||
import type { HarnessEvent } from "../../agent-harness.ts";
|
||||
import { type Context, withAbortSignal } from "../../context.ts";
|
||||
import { streamHarnessAssistant } from "../../execution/assistant.ts";
|
||||
import { applyStreamOptionsPatch } from "../../hooks.ts";
|
||||
import { insertEntry, insertUsage } from "../../session/commit.ts";
|
||||
import { buildSessionContext } from "../../session/context.ts";
|
||||
import { SessionInvariantError } from "../../session/session.ts";
|
||||
import type {
|
||||
Entry,
|
||||
Generation,
|
||||
GenerationContext,
|
||||
JsonValue,
|
||||
MessageEntry,
|
||||
NewEntry,
|
||||
OperationError,
|
||||
RunState,
|
||||
SettledAssistantMessage,
|
||||
ToolCall,
|
||||
UsageRow,
|
||||
} from "../../session/types.ts";
|
||||
import {
|
||||
deleteList,
|
||||
laneLeaf,
|
||||
operationState as operationStateValue,
|
||||
pendingAssistantFrames,
|
||||
setValue,
|
||||
} from "../../session/values.ts";
|
||||
import type { AgentHarnessStreamOptions } from "../../types.ts";
|
||||
import type { Lane } from "../lane.ts";
|
||||
import { openFrameProgress } from "../progress.ts";
|
||||
import type { Drive, LaneCommand, LostOwnership, ProcedureResult } from "../types.ts";
|
||||
|
||||
type LocalResult = { kind: "committed" } | { kind: "state_changed" } | LostOwnership;
|
||||
type ReadResult<T> = { kind: "value"; value: T } | { kind: "state_changed" } | LostOwnership;
|
||||
|
||||
const ZERO_USAGE = {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
};
|
||||
|
||||
function sameReadyGeneration(current: Generation, expected: Extract<Generation, { status: "ready" }>): boolean {
|
||||
return (
|
||||
current.status === "ready" &&
|
||||
current.context.stepId === expected.context.stepId &&
|
||||
current.nextAttempt === expected.nextAttempt
|
||||
);
|
||||
}
|
||||
|
||||
function uuidV7Timestamp(id: string): number {
|
||||
const timestamp = Number.parseInt(id.slice(0, 8) + id.slice(9, 13), 16);
|
||||
if (!Number.isSafeInteger(timestamp)) throw new SessionInvariantError(`Invalid reserved UUIDv7 ${id}`);
|
||||
return timestamp;
|
||||
}
|
||||
|
||||
function configurationError(
|
||||
code: "model_unavailable" | "configured_tools_unavailable",
|
||||
details: JsonValue,
|
||||
): OperationError {
|
||||
return {
|
||||
code,
|
||||
message:
|
||||
code === "model_unavailable"
|
||||
? "The configured model is unavailable in this process"
|
||||
: "One or more configured tools are unavailable in this process",
|
||||
details,
|
||||
};
|
||||
}
|
||||
|
||||
function providerError(message: SettledAssistantMessage): OperationError {
|
||||
return {
|
||||
code: "assistant_error",
|
||||
message: message.errorMessage ?? `Assistant request ended with ${message.stopReason}`,
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeError(message: SettledAssistantMessage, errorMessage: string): SettledAssistantMessage {
|
||||
return { ...message, stopReason: "error", errorMessage };
|
||||
}
|
||||
|
||||
function normalizeAborted(message: SettledAssistantMessage): SettledAssistantMessage {
|
||||
return {
|
||||
...message,
|
||||
stopReason: "aborted",
|
||||
errorMessage: message.errorMessage ?? "Assistant request was cancelled",
|
||||
};
|
||||
}
|
||||
|
||||
function retryDelay(baseDelayMs: number, attempt: number): number {
|
||||
const delay = baseDelayMs * 2 ** Math.max(0, attempt - 1);
|
||||
return Number.isSafeInteger(delay) ? delay : Number.MAX_SAFE_INTEGER;
|
||||
}
|
||||
|
||||
function saturatingAdd(left: number, right: number): number {
|
||||
const sum = left + right;
|
||||
return Number.isSafeInteger(sum) ? sum : Number.MAX_SAFE_INTEGER;
|
||||
}
|
||||
|
||||
function deferredHandleIsValid(
|
||||
message: SettledAssistantMessage,
|
||||
identity: GenerationContext["configuration"]["model"],
|
||||
): boolean {
|
||||
const handle = message.deferred;
|
||||
return (
|
||||
message.stopReason === "deferred" &&
|
||||
handle !== undefined &&
|
||||
handle.id.length !== 0 &&
|
||||
handle.provider === identity.provider &&
|
||||
handle.modelId === identity.modelId &&
|
||||
handle.api === message.api
|
||||
);
|
||||
}
|
||||
|
||||
async function resolveSystemPrompt<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
context: Context,
|
||||
): Promise<string> {
|
||||
const config = lane.readConfig();
|
||||
if (config.systemPrompt === undefined) return "";
|
||||
if (typeof config.systemPrompt === "string") return config.systemPrompt;
|
||||
const source = config.toolContext;
|
||||
const toolContext = typeof source === "function" ? await source(context) : source;
|
||||
return config.systemPrompt(toolContext as TContext, context);
|
||||
}
|
||||
|
||||
async function readRequestMessages<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
generation: Extract<Generation, { status: "ready" }>,
|
||||
): Promise<ReadResult<AgentMessage[]>> {
|
||||
const path = await lane.commandDriveOwned<ReadResult<Entry[]>>(
|
||||
drive,
|
||||
async (state, reader): Promise<LaneCommand<ReadResult<Entry[]>>> => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Generation context reached a non-run operation");
|
||||
}
|
||||
if (operation.state.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "state_changed" } as const };
|
||||
}
|
||||
if (
|
||||
operation.state.phase.kind !== "assistant" ||
|
||||
!sameReadyGeneration(operation.state.phase.generation, generation)
|
||||
) {
|
||||
throw new SessionInvariantError("Generation context lost its ready boundary");
|
||||
}
|
||||
if (state.leafId === null) throw new SessionInvariantError("Assistant generation has no branch leaf");
|
||||
const entries = (
|
||||
await reader.scanBranch(
|
||||
{ start: state.leafId, stopAtType: "compaction", order: "newestFirst" },
|
||||
drive.context,
|
||||
)
|
||||
).reverse();
|
||||
return { kind: "return", result: { kind: "value", value: entries } as const };
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
if (path.kind !== "value") return path;
|
||||
return {
|
||||
kind: "value",
|
||||
value: await buildSessionContext(
|
||||
path.value,
|
||||
{ entryProjectors: lane.readConfig().entryProjectors },
|
||||
drive.context,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
async function commitGenerationIntent<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
expected: Extract<Generation, { status: "ready" }>,
|
||||
generation: Extract<Generation, { status: "effect_pending" }>,
|
||||
): Promise<LocalResult> {
|
||||
return lane.commandDriveOwned<LocalResult>(
|
||||
drive,
|
||||
(state) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Generation intent reached a non-run operation");
|
||||
}
|
||||
const run = operation.state;
|
||||
if (run.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "state_changed" } as const };
|
||||
}
|
||||
if (run.phase.kind !== "assistant" || !sameReadyGeneration(run.phase.generation, expected)) {
|
||||
throw new SessionInvariantError("Generation intent lost its ready boundary");
|
||||
}
|
||||
const nextState: RunState = { ...run, phase: { kind: "assistant", generation } };
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [setValue(operationStateValue(drive.operationId), nextState)],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => ({ kind: "committed" }) as const,
|
||||
events: () =>
|
||||
expected.nextAttempt === 1
|
||||
? [
|
||||
{
|
||||
type: "turn_start",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
turnId: expected.context.stepId,
|
||||
},
|
||||
]
|
||||
: [],
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
/** Commit a non-retryable request-configuration failure before reserving response ids. */
|
||||
export function enterConfigurationFailure<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
expected: Extract<Generation, { status: "ready" }>,
|
||||
error: OperationError,
|
||||
): Promise<ProcedureResult> {
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
(state) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Configuration failure reached a non-run operation");
|
||||
}
|
||||
const run = operation.state;
|
||||
if (run.control.status === "cancel_requested") {
|
||||
return { kind: "return", result: { kind: "continue" } as const };
|
||||
}
|
||||
if (run.phase.kind !== "assistant" || !sameReadyGeneration(run.phase.generation, expected)) {
|
||||
throw new SessionInvariantError("Configuration failure lost its ready boundary");
|
||||
}
|
||||
const nextState: RunState = {
|
||||
...run,
|
||||
phase: { kind: "failure_drain", error, provenance: { kind: "configuration" } },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [setValue(operationStateValue(drive.operationId), nextState)],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
interface SettlementIdentity {
|
||||
responseEntryId: string;
|
||||
usageId: string;
|
||||
stepId: string;
|
||||
}
|
||||
|
||||
/** Classify and atomically settle one complete assistant response under its reserved ids. */
|
||||
export function settleAssistant<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
identity: SettlementIdentity,
|
||||
response: SettledAssistantMessage,
|
||||
options: { recovery?: true } = {},
|
||||
): Promise<ProcedureResult> {
|
||||
return lane.commandDriveOwned<ProcedureResult>(
|
||||
drive,
|
||||
(state) => {
|
||||
const operation = state.operation;
|
||||
if (operation === null || operation.state.kind !== "run") {
|
||||
throw new SessionInvariantError("Assistant settlement reached a non-run operation");
|
||||
}
|
||||
const run = operation.state;
|
||||
if (run.phase.kind !== "assistant" || run.phase.generation.status !== "effect_pending") {
|
||||
throw new SessionInvariantError("Assistant settlement lost its effect-pending boundary");
|
||||
}
|
||||
const generation = run.phase.generation;
|
||||
if (
|
||||
generation.context.stepId !== identity.stepId ||
|
||||
generation.responseEntryId !== identity.responseEntryId ||
|
||||
generation.usageId !== identity.usageId
|
||||
) {
|
||||
throw new SessionInvariantError("Assistant settlement ids contradict the durable request intent");
|
||||
}
|
||||
|
||||
const modelIdentity = generation.context.configuration.model;
|
||||
let committed = response;
|
||||
let nextPhase: RunState["phase"];
|
||||
if (run.control.status === "cancel_requested") {
|
||||
committed = normalizeAborted(response);
|
||||
nextPhase = {
|
||||
kind: "checkpoint",
|
||||
continuation: { kind: "may_finish", includeFinalAssistant: true },
|
||||
triggerEntryId: identity.responseEntryId,
|
||||
};
|
||||
} else if (response.stopReason === "aborted") {
|
||||
throw new SessionInvariantError("Assistant response is aborted while durable control is running");
|
||||
} else if (response.stopReason === "deferred") {
|
||||
if (deferredHandleIsValid(response, modelIdentity)) {
|
||||
nextPhase = {
|
||||
kind: "deferred",
|
||||
deferred: {
|
||||
status: "suspended",
|
||||
stepId: generation.context.stepId,
|
||||
sourceEntryId: identity.responseEntryId,
|
||||
poll: 0,
|
||||
configuration: generation.context.configuration,
|
||||
streamOptions: generation.context.streamOptions,
|
||||
},
|
||||
};
|
||||
} else {
|
||||
committed = normalizeError(response, "Provider returned an invalid deferred handle");
|
||||
nextPhase = {
|
||||
kind: "failure_drain",
|
||||
error: providerError(committed),
|
||||
provenance: { kind: "response", entryId: identity.responseEntryId },
|
||||
};
|
||||
}
|
||||
} else if (
|
||||
response.stopReason === "error" &&
|
||||
(options.recovery === true || isRetryableAssistantError(response)) &&
|
||||
generation.attempt < generation.context.retryPolicy.maxAttempts
|
||||
) {
|
||||
const delayMs = retryDelay(generation.context.retryPolicy.baseDelayMs, generation.attempt);
|
||||
nextPhase = {
|
||||
kind: "assistant",
|
||||
generation: {
|
||||
status: "retry_wait",
|
||||
context: generation.context,
|
||||
nextAttempt: generation.attempt + 1,
|
||||
notBefore: saturatingAdd(Date.now(), delayMs),
|
||||
errorMessage: response.errorMessage ?? "Assistant request failed",
|
||||
},
|
||||
};
|
||||
} else if (response.stopReason === "error") {
|
||||
nextPhase = {
|
||||
kind: "failure_drain",
|
||||
error: providerError(response),
|
||||
provenance: { kind: "response", entryId: identity.responseEntryId },
|
||||
};
|
||||
} else {
|
||||
const calls = response.content.flatMap((content, sourceIndex) =>
|
||||
content.type === "toolCall" ? [{ content, sourceIndex }] : [],
|
||||
);
|
||||
if (calls.length !== 0) {
|
||||
const timestamp = uuidV7Timestamp(identity.responseEntryId);
|
||||
const planned: ToolCall[] = calls.map(({ sourceIndex }) => ({
|
||||
status: "planned",
|
||||
sourceIndex,
|
||||
resultEntryId: lane.session.idGenerator.next(timestamp),
|
||||
}));
|
||||
nextPhase = {
|
||||
kind: "tools",
|
||||
batch: {
|
||||
assistantEntryId: identity.responseEntryId,
|
||||
configuration: generation.context.configuration,
|
||||
turnId: generation.context.stepId,
|
||||
calls: planned,
|
||||
},
|
||||
};
|
||||
} else if (response.stopReason === "toolUse") {
|
||||
committed = normalizeError(response, "Provider reported tool use without any tool calls");
|
||||
nextPhase = {
|
||||
kind: "failure_drain",
|
||||
error: providerError(committed),
|
||||
provenance: { kind: "response", entryId: identity.responseEntryId },
|
||||
};
|
||||
} else {
|
||||
nextPhase = {
|
||||
kind: "checkpoint",
|
||||
continuation: { kind: "may_finish", includeFinalAssistant: true },
|
||||
triggerEntryId: identity.responseEntryId,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const nextState: RunState = {
|
||||
...run,
|
||||
phase: nextPhase,
|
||||
latestAssistantEntryId: identity.responseEntryId,
|
||||
};
|
||||
const responseEntry: NewEntry<MessageEntry> = {
|
||||
id: identity.responseEntryId,
|
||||
parentId: state.leafId,
|
||||
type: "message",
|
||||
message: committed,
|
||||
};
|
||||
const usageRow: Omit<UsageRow, "seq"> = {
|
||||
id: identity.usageId,
|
||||
usage: committed.usage,
|
||||
entryId: identity.responseEntryId,
|
||||
adjustment: false,
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
insertEntry(responseEntry),
|
||||
insertUsage(usageRow),
|
||||
setValue(laneLeaf(lane.name), identity.responseEntryId),
|
||||
deleteList(pendingAssistantFrames(drive.operationId, identity.responseEntryId)),
|
||||
setValue(operationStateValue(drive.operationId), nextState),
|
||||
],
|
||||
next: {
|
||||
...state,
|
||||
leafId: identity.responseEntryId,
|
||||
operation: { meta: operation.meta, state: nextState },
|
||||
},
|
||||
materialize: () => ({ kind: "continue" }) as const,
|
||||
events: (commit) => {
|
||||
const entry: MessageEntry = {
|
||||
...responseEntry,
|
||||
seq: commit.seqs[0]!,
|
||||
timestamp: commit.timestamp,
|
||||
};
|
||||
const events: HarnessEvent[] = [
|
||||
{ type: "entry_added", lane: lane.name, entry, ...options },
|
||||
{
|
||||
type: "usage",
|
||||
lane: lane.name,
|
||||
row: { ...usageRow, seq: commit.seqs[1]! },
|
||||
totals: commit.stats.usage,
|
||||
},
|
||||
];
|
||||
if (
|
||||
options.recovery !== true &&
|
||||
nextPhase.kind === "assistant" &&
|
||||
nextPhase.generation.status === "retry_wait"
|
||||
) {
|
||||
events.push({
|
||||
type: "retry_scheduled",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
step: identity.stepId,
|
||||
attempt: nextPhase.generation.nextAttempt,
|
||||
maxAttempts: nextPhase.generation.context.retryPolicy.maxAttempts,
|
||||
delayMs: retryDelay(generation.context.retryPolicy.baseDelayMs, generation.attempt),
|
||||
errorMessage: nextPhase.generation.errorMessage,
|
||||
});
|
||||
}
|
||||
if (
|
||||
options.recovery !== true &&
|
||||
nextPhase.kind !== "tools" &&
|
||||
!(nextPhase.kind === "assistant" && nextPhase.generation.status === "retry_wait")
|
||||
) {
|
||||
events.push({
|
||||
type: "turn_end",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
turnId: identity.stepId,
|
||||
message: committed,
|
||||
toolResults: [],
|
||||
...options,
|
||||
});
|
||||
}
|
||||
return events;
|
||||
},
|
||||
};
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
}
|
||||
|
||||
/** Execute one ready assistant generation through intent, admitted stream, and settlement. */
|
||||
export async function runGeneration<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
run: RunState,
|
||||
generation: Generation,
|
||||
): Promise<ProcedureResult> {
|
||||
if (generation.status !== "ready") {
|
||||
throw new SessionInvariantError("M3 runGeneration requires a ready generation");
|
||||
}
|
||||
if (run.phase.kind !== "assistant" || !sameReadyGeneration(run.phase.generation, generation)) {
|
||||
throw new SessionInvariantError("runGeneration requires its current ready phase");
|
||||
}
|
||||
if (run.control.status === "cancel_requested") return { kind: "continue" };
|
||||
|
||||
const identity = generation.context.configuration.model;
|
||||
const model = lane.models.getModel(identity.provider, identity.modelId);
|
||||
if (model === undefined) {
|
||||
return enterConfigurationFailure(lane, drive, generation, configurationError("model_unavailable", identity));
|
||||
}
|
||||
const config = lane.readConfig();
|
||||
const toolsByName = new Map(config.tools.map((tool) => [tool.name, tool]));
|
||||
const missingTools = generation.context.configuration.activeToolNames.filter((name) => !toolsByName.has(name));
|
||||
if (missingTools.length !== 0) {
|
||||
return enterConfigurationFailure(
|
||||
lane,
|
||||
drive,
|
||||
generation,
|
||||
configurationError("configured_tools_unavailable", { tools: missingTools }),
|
||||
);
|
||||
}
|
||||
const tools: Tool[] = generation.context.configuration.activeToolNames.map((name) => {
|
||||
const tool = toolsByName.get(name);
|
||||
if (tool === undefined) throw new SessionInvariantError(`Configured tool ${name} disappeared during resolution`);
|
||||
return {
|
||||
name: tool.name,
|
||||
description: tool.description,
|
||||
parameters: tool.parameters,
|
||||
...(tool.constrainedSampling === undefined ? {} : { constrainedSampling: tool.constrainedSampling }),
|
||||
};
|
||||
});
|
||||
|
||||
const messages = await readRequestMessages(lane, drive, generation);
|
||||
if (messages.kind === "lost_ownership") return messages;
|
||||
if (messages.kind === "state_changed") return { kind: "continue" };
|
||||
const systemPrompt = await resolveSystemPrompt(lane, drive.context);
|
||||
const beforeRequest = await lane.hooks.runWithGate(
|
||||
"before_request",
|
||||
{
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
model,
|
||||
step: "assistant",
|
||||
attempt: generation.nextAttempt,
|
||||
streamOptions: generation.context.streamOptions,
|
||||
},
|
||||
drive.gate,
|
||||
drive.context,
|
||||
);
|
||||
const streamOptions: AgentHarnessStreamOptions =
|
||||
beforeRequest?.streamOptions === undefined
|
||||
? generation.context.streamOptions
|
||||
: applyStreamOptionsPatch(generation.context.streamOptions, beforeRequest.streamOptions);
|
||||
const at = Date.now();
|
||||
const responseEntryId = lane.session.idGenerator.next(at);
|
||||
const usageId = lane.session.idGenerator.next(at);
|
||||
const effectPending: Extract<Generation, { status: "effect_pending" }> = {
|
||||
status: "effect_pending",
|
||||
context: generation.context,
|
||||
attempt: generation.nextAttempt,
|
||||
responseEntryId,
|
||||
usageId,
|
||||
intendedOutputLimit: model.maxTokens,
|
||||
contextWindow: model.contextWindow,
|
||||
};
|
||||
const intent = await commitGenerationIntent(lane, drive, generation, effectPending);
|
||||
if (intent.kind === "lost_ownership") return intent;
|
||||
if (intent.kind === "state_changed") return { kind: "continue" };
|
||||
|
||||
const progress = openFrameProgress(lane, drive, responseEntryId);
|
||||
let drained = false;
|
||||
const drainProgress = async () => {
|
||||
if (drained) return;
|
||||
progress.seal();
|
||||
await progress.drain();
|
||||
drained = true;
|
||||
};
|
||||
let response: SettledAssistantMessage;
|
||||
try {
|
||||
response = await streamHarnessAssistant(
|
||||
messages.value,
|
||||
{
|
||||
model,
|
||||
systemPrompt,
|
||||
tools,
|
||||
thinkingLevel: generation.context.configuration.thinkingLevel,
|
||||
streamOptions,
|
||||
transformContext: async (requestContext, context) => {
|
||||
const result = await lane.hooks.runWithGate(
|
||||
"transform_context",
|
||||
{ lane: lane.name, runId: drive.operationId, ...requestContext },
|
||||
drive.gate,
|
||||
context,
|
||||
);
|
||||
return {
|
||||
messages: result?.messages ?? requestContext.messages,
|
||||
systemPrompt: result?.systemPrompt ?? requestContext.systemPrompt,
|
||||
};
|
||||
},
|
||||
toProviderMessages: config.toProviderMessages,
|
||||
beforePayload: async (payload, requestModel, context) => {
|
||||
const result = await lane.hooks.runWithGate(
|
||||
"before_payload",
|
||||
{ lane: lane.name, runId: drive.operationId, model: requestModel, payload },
|
||||
drive.gate,
|
||||
context,
|
||||
);
|
||||
return result?.payload;
|
||||
},
|
||||
afterResponse: async (message, metadata, context) => {
|
||||
await drainProgress();
|
||||
const result = await lane.hooks.runWithGate(
|
||||
"after_response",
|
||||
{ lane: lane.name, runId: drive.operationId, ...metadata, message },
|
||||
drive.gate,
|
||||
context,
|
||||
);
|
||||
return result?.message ?? message;
|
||||
},
|
||||
request: (aiContext, options, context) => {
|
||||
const admitted = withAbortSignal(drive.gate.signal, context);
|
||||
return drive.gate.admit(() =>
|
||||
lane.models.streamSimple(model, aiContext, {
|
||||
...options,
|
||||
signal: admitted.abortSignal,
|
||||
telemetryContext: admitted.telemetryContext,
|
||||
}),
|
||||
);
|
||||
},
|
||||
observer: {
|
||||
start(message, event, context) {
|
||||
const frame = event === undefined ? undefined : assistantMessageEventToFrame(event);
|
||||
if (frame !== undefined) progress.write(frame);
|
||||
return lane.emitBatch(
|
||||
[{ type: "message_start", lane: lane.name, runId: drive.operationId, message }],
|
||||
context,
|
||||
);
|
||||
},
|
||||
update(message, event: AssistantMessageEvent, context) {
|
||||
const frame = assistantMessageEventToFrame(event);
|
||||
if (frame !== undefined) progress.write(frame);
|
||||
return lane.emitBatch(
|
||||
[
|
||||
{
|
||||
type: "message_update",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
message,
|
||||
event,
|
||||
},
|
||||
],
|
||||
context,
|
||||
);
|
||||
},
|
||||
end(message, context) {
|
||||
return lane.emitBatch(
|
||||
[
|
||||
{
|
||||
type: "message_end",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
message,
|
||||
entryId: responseEntryId,
|
||||
},
|
||||
],
|
||||
context,
|
||||
);
|
||||
},
|
||||
},
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
} finally {
|
||||
await drainProgress();
|
||||
}
|
||||
|
||||
return settleAssistant(lane, drive, { responseEntryId, usageId, stepId: generation.context.stepId }, response);
|
||||
}
|
||||
|
||||
export function interruptedAssistantMessage(
|
||||
generation: Extract<Generation, { status: "effect_pending" }>,
|
||||
partial: AssistantMessage | undefined,
|
||||
): SettledAssistantMessage {
|
||||
const warning =
|
||||
"Assistant request was interrupted. The preceding content is the latest committed partial; newer live output may be missing and the external outcome is unknown.";
|
||||
return partial === undefined
|
||||
? {
|
||||
role: "assistant",
|
||||
content: [],
|
||||
api: "unknown",
|
||||
provider: generation.context.configuration.model.provider,
|
||||
model: generation.context.configuration.model.modelId,
|
||||
usage: ZERO_USAGE,
|
||||
stopReason: "error",
|
||||
errorMessage: warning,
|
||||
timestamp: Date.now(),
|
||||
}
|
||||
: { ...partial, usage: ZERO_USAGE, stopReason: "error", errorMessage: warning };
|
||||
}
|
||||
98
packages/agent/src/harness/runtime2/drive/recovery.ts
Normal file
98
packages/agent/src/harness/runtime2/drive/recovery.ts
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
import { type AssistantMessageFrame, reduceAssistantMessageFrames } from "@earendil-works/pi-ai";
|
||||
import { SessionInvariantError } from "../../session/session.ts";
|
||||
import type { Generation, RunState } from "../../session/types.ts";
|
||||
import { pendingAssistantFrames } from "../../session/values.ts";
|
||||
import type { Lane } from "../lane.ts";
|
||||
import type { Drive, ProcedureResult } from "../types.ts";
|
||||
import { interruptedAssistantMessage, settleAssistant } from "./generation.ts";
|
||||
|
||||
/** Settle an orphaned assistant request from its bounded committed frame prefix without another provider call. */
|
||||
export async function recoverAssistantGeneration<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
run: RunState,
|
||||
generation: Extract<Generation, { status: "effect_pending" }>,
|
||||
): Promise<ProcedureResult> {
|
||||
if (
|
||||
run.phase.kind !== "assistant" ||
|
||||
run.phase.generation.status !== "effect_pending" ||
|
||||
run.phase.generation.responseEntryId !== generation.responseEntryId
|
||||
) {
|
||||
throw new SessionInvariantError("Assistant recovery requires its current effect-pending generation");
|
||||
}
|
||||
const read = await lane.commandDriveOwned<{ kind: "frames"; frames: AssistantMessageFrame[] }>(
|
||||
drive,
|
||||
async (state, reader) => {
|
||||
const operation = state.operation;
|
||||
if (
|
||||
operation === null ||
|
||||
operation.state.kind !== "run" ||
|
||||
operation.state.phase.kind !== "assistant" ||
|
||||
operation.state.phase.generation.status !== "effect_pending" ||
|
||||
operation.state.phase.generation.responseEntryId !== generation.responseEntryId
|
||||
) {
|
||||
throw new SessionInvariantError("Assistant recovery lost its effect-pending generation");
|
||||
}
|
||||
const frames: AssistantMessageFrame[] = [];
|
||||
let cursor: { seq: number } | undefined;
|
||||
for (;;) {
|
||||
const page = await reader.readList(
|
||||
pendingAssistantFrames(drive.operationId, generation.responseEntryId),
|
||||
{ order: "asc", limit: 1_000, ...(cursor === undefined ? {} : { cursor }) },
|
||||
drive.context,
|
||||
);
|
||||
frames.push(...page.map(({ value }) => value));
|
||||
if (page.length < 1_000) break;
|
||||
cursor = { seq: page[page.length - 1]!.seq };
|
||||
}
|
||||
return { kind: "return", result: { kind: "frames", frames } as const };
|
||||
},
|
||||
drive.context,
|
||||
);
|
||||
if (read.kind === "lost_ownership") return read;
|
||||
const frames = read.frames;
|
||||
const current = lane.state.operation;
|
||||
if (current?.meta.operationId !== drive.operationId || !lane.isDriveActive(drive)) {
|
||||
return { kind: "lost_ownership" };
|
||||
}
|
||||
if (
|
||||
current.state.kind !== "run" ||
|
||||
current.state.phase.kind !== "assistant" ||
|
||||
current.state.phase.generation.status !== "effect_pending" ||
|
||||
current.state.phase.generation.responseEntryId !== generation.responseEntryId
|
||||
) {
|
||||
throw new SessionInvariantError("Assistant recovery lost its effect-pending generation");
|
||||
}
|
||||
const message = interruptedAssistantMessage(generation, reduceAssistantMessageFrames(frames));
|
||||
await lane.emitBatch(
|
||||
[
|
||||
{
|
||||
type: "message_start",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
message,
|
||||
recovery: true,
|
||||
},
|
||||
{
|
||||
type: "message_end",
|
||||
lane: lane.name,
|
||||
runId: drive.operationId,
|
||||
message,
|
||||
entryId: generation.responseEntryId,
|
||||
recovery: true,
|
||||
},
|
||||
],
|
||||
drive.context,
|
||||
);
|
||||
return settleAssistant(
|
||||
lane,
|
||||
drive,
|
||||
{
|
||||
responseEntryId: generation.responseEntryId,
|
||||
usageId: generation.usageId,
|
||||
stepId: generation.context.stepId,
|
||||
},
|
||||
message,
|
||||
{ recovery: true },
|
||||
);
|
||||
}
|
||||
|
|
@ -86,6 +86,7 @@ describe("streamHarnessAssistant", () => {
|
|||
let receivedContext: { systemPrompt?: string; messages: Message[]; tools?: unknown[] } | undefined;
|
||||
let receivedOptions: SimpleStreamOptions | undefined;
|
||||
let responseMetadata: AssistantResponseMetadata | undefined;
|
||||
let startEventType: string | undefined;
|
||||
|
||||
const result = await streamHarnessAssistant(
|
||||
input,
|
||||
|
|
@ -144,9 +145,10 @@ describe("streamHarnessAssistant", () => {
|
|||
return stream;
|
||||
},
|
||||
observer: {
|
||||
start(message) {
|
||||
start(message, event) {
|
||||
order.push("observer_start");
|
||||
starts.push(message);
|
||||
startEventType = event?.type;
|
||||
},
|
||||
update(message) {
|
||||
order.push("observer_update");
|
||||
|
|
@ -184,6 +186,7 @@ describe("streamHarnessAssistant", () => {
|
|||
});
|
||||
expect(responseMetadata).toEqual({ status: 201, headers: { "request-id": "r1" } });
|
||||
expect(starts).toHaveLength(1);
|
||||
expect(startEventType).toBe("start");
|
||||
expect(starts[0]).not.toBe(updates[0]);
|
||||
expect(updates).toHaveLength(1);
|
||||
expect(ends[0]).toBe(result);
|
||||
|
|
@ -244,6 +247,7 @@ describe("streamHarnessAssistant", () => {
|
|||
const events: string[] = [];
|
||||
const final = assistant("complete");
|
||||
let options: SimpleStreamOptions | undefined;
|
||||
let synthesizedStartEvent: unknown = "unset";
|
||||
const result = await streamHarnessAssistant(
|
||||
[user("prompt")],
|
||||
{
|
||||
|
|
@ -259,8 +263,9 @@ describe("streamHarnessAssistant", () => {
|
|||
return stream;
|
||||
},
|
||||
observer: {
|
||||
start(message) {
|
||||
start(message, event) {
|
||||
events.push(`start:${message.stopReason}`);
|
||||
synthesizedStartEvent = event;
|
||||
},
|
||||
update() {
|
||||
events.push("update");
|
||||
|
|
@ -275,6 +280,7 @@ describe("streamHarnessAssistant", () => {
|
|||
|
||||
expect(result).toBe(final);
|
||||
expect(events).toEqual(["start:stop", "end:stop"]);
|
||||
expect(synthesizedStartEvent).toBeUndefined();
|
||||
expect(options).not.toHaveProperty("reasoning");
|
||||
});
|
||||
|
||||
|
|
|
|||
703
packages/agent/test/harness/runtime2/drive-generation.test.ts
Normal file
703
packages/agent/test/harness/runtime2/drive-generation.test.ts
Normal file
|
|
@ -0,0 +1,703 @@
|
|||
import {
|
||||
type AssistantMessage,
|
||||
createAssistantMessageEventStream,
|
||||
createModels,
|
||||
fauxAssistantMessage,
|
||||
fauxProvider,
|
||||
type MutableModels,
|
||||
type Provider,
|
||||
} from "@earendil-works/pi-ai";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import type { HarnessEvent, WatchHandle } from "../../../src/harness/agent-harness.ts";
|
||||
import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts";
|
||||
import { BACKGROUND_CONTEXT, type Context } from "../../../src/harness/context.ts";
|
||||
import { HookRegistry } from "../../../src/harness/hooks.ts";
|
||||
import { runCheckpoint, startRun } from "../../../src/harness/runtime2/drive/checkpoint.ts";
|
||||
import { runGeneration } from "../../../src/harness/runtime2/drive/generation.ts";
|
||||
import { recoverAssistantGeneration } from "../../../src/harness/runtime2/drive/recovery.ts";
|
||||
import { Lane } from "../../../src/harness/runtime2/lane.ts";
|
||||
import { restoreLane } from "../../../src/harness/runtime2/restore.ts";
|
||||
import { type Config, Drive } from "../../../src/harness/runtime2/types.ts";
|
||||
import { MemoryStorage } from "../../../src/harness/session/memory.ts";
|
||||
import { StorageBackedSession } from "../../../src/harness/session/session.ts";
|
||||
import { InstrumentedStorage } from "../../../src/harness/session/testing/instrumented-storage.ts";
|
||||
import type {
|
||||
Generation,
|
||||
LaneConfiguration,
|
||||
MessageEntry,
|
||||
RunState,
|
||||
Session,
|
||||
Write,
|
||||
} from "../../../src/harness/session/types.ts";
|
||||
import * as storedValues from "../../../src/harness/session/values.ts";
|
||||
import { deferred } from "./test-utils.ts";
|
||||
|
||||
const sessions: Session[] = [];
|
||||
const operationId = "01950000-0000-7000-8000-000000000001";
|
||||
|
||||
class FrameBlockingMemoryStorage extends MemoryStorage {
|
||||
blockNextFrame = false;
|
||||
readonly frameStarted = deferred();
|
||||
readonly releaseFrame = deferred();
|
||||
|
||||
override async commit(writes: Write[], context: Context) {
|
||||
if (
|
||||
this.blockNextFrame &&
|
||||
writes.some(
|
||||
(write) =>
|
||||
write.kind === "list" && write.op === "append" && write.namespace === "pi.pending.assistant_frame",
|
||||
)
|
||||
) {
|
||||
this.blockNextFrame = false;
|
||||
this.frameStarted.resolve();
|
||||
await this.releaseFrame.promise;
|
||||
}
|
||||
return super.commit(writes, context);
|
||||
}
|
||||
}
|
||||
|
||||
interface Fixture {
|
||||
lane: Lane<undefined>;
|
||||
drive: Drive;
|
||||
session: Session;
|
||||
storage: InstrumentedStorage;
|
||||
models: MutableModels;
|
||||
faux: ReturnType<typeof fauxProvider>;
|
||||
hooks: HookRegistry;
|
||||
events: HarnessEvent[];
|
||||
config: Config<undefined>;
|
||||
}
|
||||
|
||||
function unusedWatch<T>(): WatchHandle<T> {
|
||||
throw new Error("watch is not used by generation tests");
|
||||
}
|
||||
|
||||
async function createFixture(backend: MemoryStorage = new MemoryStorage({ now: () => 100 })): Promise<Fixture> {
|
||||
const storage = new InstrumentedStorage(backend);
|
||||
const session = new StorageBackedSession(
|
||||
{ id: `generation-${sessions.length}`, createdAt: 1, storageVersion: 1 },
|
||||
storage,
|
||||
);
|
||||
sessions.push(session);
|
||||
const faux = fauxProvider({ tokenSize: { min: 1, max: 1 } });
|
||||
const model = faux.getModel();
|
||||
const configuration: LaneConfiguration = {
|
||||
model: { provider: model.provider, modelId: model.id },
|
||||
thinkingLevel: "off",
|
||||
activeToolNames: [],
|
||||
};
|
||||
await session.mutate(
|
||||
"main",
|
||||
(mutator) =>
|
||||
mutator.commit(
|
||||
[
|
||||
storedValues.setValue(storedValues.laneLeaf("main"), null),
|
||||
storedValues.setValue(storedValues.laneConfig("main"), configuration),
|
||||
storedValues.setValue(storedValues.laneState("main"), {
|
||||
currentOperationId: null,
|
||||
pendingNextRun: [],
|
||||
}),
|
||||
],
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
BACKGROUND_CONTEXT,
|
||||
);
|
||||
const models = createModels();
|
||||
models.setProvider(faux.provider);
|
||||
const hooks = new HookRegistry(() => {});
|
||||
const events: HarnessEvent[] = [];
|
||||
const config: Config<undefined> = {
|
||||
tools: [],
|
||||
resources: {},
|
||||
streamOptions: {},
|
||||
retryPolicy: { enabled: true, maxRetries: 3, baseDelayMs: 1 },
|
||||
compaction: DEFAULT_COMPACTION_SETTINGS,
|
||||
steeringMode: "all",
|
||||
followUpMode: "all",
|
||||
toolExecution: "parallel",
|
||||
toolContext: undefined,
|
||||
systemPrompt: undefined,
|
||||
toProviderMessages: (messages) =>
|
||||
messages.filter(
|
||||
(message) => message.role === "user" || message.role === "assistant" || message.role === "toolResult",
|
||||
),
|
||||
entryProjectors: {},
|
||||
};
|
||||
const lane = new Lane<undefined>(
|
||||
"main",
|
||||
session,
|
||||
models,
|
||||
hooks,
|
||||
await restoreLane(session, "main", BACKGROUND_CONTEXT),
|
||||
(cause) => (cause instanceof Error ? cause : new Error(String(cause))),
|
||||
(batch) => {
|
||||
events.push(...structuredClone(batch));
|
||||
return Promise.resolve();
|
||||
},
|
||||
unusedWatch,
|
||||
() => config,
|
||||
);
|
||||
const admission = await lane.accept({ kind: "prompt", operationId, prompt: "question" }, BACKGROUND_CONTEXT);
|
||||
if (!admission.ok) throw admission.error;
|
||||
const drive = new Drive({ operationId }, BACKGROUND_CONTEXT);
|
||||
lane.activeDrive = drive;
|
||||
storage.clearCommitAttempts();
|
||||
return { lane, drive, session, storage, models, faux, hooks, events, config };
|
||||
}
|
||||
|
||||
function currentRun(lane: Lane<undefined>): RunState {
|
||||
const operation = lane.state.operation;
|
||||
if (operation?.state.kind !== "run") throw new Error("fixture has no run");
|
||||
return operation.state;
|
||||
}
|
||||
|
||||
function readyGeneration(lane: Lane<undefined>): Extract<Generation, { status: "ready" }> {
|
||||
const run = currentRun(lane);
|
||||
if (run.phase.kind !== "assistant" || run.phase.generation.status !== "ready") {
|
||||
throw new Error("fixture has no ready generation");
|
||||
}
|
||||
return run.phase.generation;
|
||||
}
|
||||
|
||||
async function advanceToReady(fixture: Fixture): Promise<Extract<Generation, { status: "ready" }>> {
|
||||
let run = currentRun(fixture.lane);
|
||||
await startRun(fixture.lane, fixture.drive, run);
|
||||
run = currentRun(fixture.lane);
|
||||
if (run.phase.kind !== "checkpoint") throw new Error("run did not reach checkpoint");
|
||||
await runCheckpoint(fixture.lane, fixture.drive, run, run.phase);
|
||||
return readyGeneration(fixture.lane);
|
||||
}
|
||||
|
||||
async function expectProjectionRestores(fixture: Fixture): Promise<void> {
|
||||
expect(fixture.lane.state).toEqual(await restoreLane(fixture.session, "main", BACKGROUND_CONTEXT));
|
||||
}
|
||||
|
||||
function writesOperationState(writes: readonly Write[], status: string): boolean {
|
||||
return writes.some(
|
||||
(write) =>
|
||||
write.kind === "value" &&
|
||||
write.op === "set" &&
|
||||
write.namespace === "pi.op.state" &&
|
||||
typeof write.value === "object" &&
|
||||
write.value !== null &&
|
||||
"phase" in write.value &&
|
||||
typeof write.value.phase === "object" &&
|
||||
write.value.phase !== null &&
|
||||
"kind" in write.value.phase &&
|
||||
write.value.phase.kind === "assistant" &&
|
||||
"generation" in write.value.phase &&
|
||||
typeof write.value.phase.generation === "object" &&
|
||||
write.value.phase.generation !== null &&
|
||||
"status" in write.value.phase.generation &&
|
||||
write.value.phase.generation.status === status,
|
||||
);
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT);
|
||||
});
|
||||
|
||||
describe("runtime2 generation checkpoint", () => {
|
||||
it("consumes before_run and snapshots a ready generation", async () => {
|
||||
const fixture = await createFixture();
|
||||
fixture.hooks.on("before_run", () => ({
|
||||
messages: [{ role: "user", content: "injected", timestamp: 2 }],
|
||||
}));
|
||||
const starting = currentRun(fixture.lane);
|
||||
|
||||
expect(await startRun(fixture.lane, fixture.drive, starting)).toEqual({ kind: "continue" });
|
||||
let run = currentRun(fixture.lane);
|
||||
if (run.phase.kind !== "checkpoint") throw new Error("missing checkpoint");
|
||||
expect(run.phase.triggerEntryId).toBe(fixture.lane.state.leafId);
|
||||
expect(fixture.events.map((event) => event.type)).toContain("entry_added");
|
||||
await expectProjectionRestores(fixture);
|
||||
|
||||
expect(await runCheckpoint(fixture.lane, fixture.drive, run, run.phase)).toEqual({ kind: "continue" });
|
||||
run = currentRun(fixture.lane);
|
||||
if (run.phase.kind !== "assistant" || run.phase.generation.status !== "ready") {
|
||||
throw new Error("missing ready generation");
|
||||
}
|
||||
expect(run.phase.generation.context).toMatchObject({
|
||||
configuration: fixture.lane.state.configuration,
|
||||
streamOptions: {},
|
||||
retryPolicy: { maxAttempts: 4, baseDelayMs: 1 },
|
||||
overflowRecoveryUsed: false,
|
||||
});
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
|
||||
it("drains mixed pending writes while separating leaf and generation trigger", async () => {
|
||||
const fixture = await createFixture();
|
||||
await startRun(fixture.lane, fixture.drive, currentRun(fixture.lane));
|
||||
const messageId = "01950000-0000-7000-8000-000000000010";
|
||||
const customId = "01950000-0000-7000-8000-000000000011";
|
||||
await fixture.lane.command((state) => {
|
||||
const operation = state.operation;
|
||||
if (operation?.state.kind !== "run") throw new Error("missing run");
|
||||
const nextState: RunState = {
|
||||
...operation.state,
|
||||
inbox: { ...operation.state.inbox, writes: [messageId, customId] },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
storedValues.setValue(storedValues.pendingEntry(messageId), {
|
||||
type: "message",
|
||||
payload: { role: "user", content: "projecting", timestamp: 3 },
|
||||
}),
|
||||
storedValues.setValue(storedValues.pendingEntry(customId), {
|
||||
type: "custom",
|
||||
customType: "display-only",
|
||||
payload: { value: true },
|
||||
}),
|
||||
storedValues.setValue(storedValues.operationState(operationId), nextState),
|
||||
],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => undefined,
|
||||
};
|
||||
}, BACKGROUND_CONTEXT);
|
||||
const run = currentRun(fixture.lane);
|
||||
if (run.phase.kind !== "checkpoint") throw new Error("missing checkpoint");
|
||||
|
||||
await runCheckpoint(fixture.lane, fixture.drive, run, run.phase);
|
||||
|
||||
const drained = currentRun(fixture.lane);
|
||||
if (drained.phase.kind !== "checkpoint") throw new Error("missing drained checkpoint");
|
||||
expect(fixture.lane.state.leafId).toBe(customId);
|
||||
expect(drained.phase.triggerEntryId).toBe(messageId);
|
||||
expect(drained.phase).toMatchObject({
|
||||
continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
|
||||
skipInboxOnce: true,
|
||||
});
|
||||
expect(drained.inbox.writes).toEqual([]);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
});
|
||||
|
||||
describe("runtime2 assistant generation", () => {
|
||||
it("commits intent before provider admission, preserves queued inbox state, and settles reserved ids", async () => {
|
||||
const fixture = await createFixture();
|
||||
const ready = await advanceToReady(fixture);
|
||||
const hookStarted = deferred();
|
||||
const releaseHook = deferred();
|
||||
fixture.hooks.on("before_request", async () => {
|
||||
hookStarted.resolve();
|
||||
await releaseHook.promise;
|
||||
return undefined;
|
||||
});
|
||||
let effectPendingAtProvider = false;
|
||||
fixture.faux.setResponses([
|
||||
async () => {
|
||||
const state = await fixture.session.getValue(storedValues.operationState(operationId), BACKGROUND_CONTEXT);
|
||||
effectPendingAtProvider =
|
||||
state?.value.kind === "run" &&
|
||||
state.value.phase.kind === "assistant" &&
|
||||
state.value.phase.generation.status === "effect_pending";
|
||||
return fauxAssistantMessage("answer", { timestamp: 5 });
|
||||
},
|
||||
]);
|
||||
fixture.storage.clearCommitAttempts();
|
||||
const generating = runGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), ready);
|
||||
await hookStarted.promise;
|
||||
const steerId = "01950000-0000-7000-8000-000000000020";
|
||||
await fixture.lane.command((state) => {
|
||||
const operation = state.operation;
|
||||
if (operation?.state.kind !== "run") throw new Error("missing run");
|
||||
const nextState: RunState = {
|
||||
...operation.state,
|
||||
inbox: { ...operation.state.inbox, steer: [steerId] },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
storedValues.setValue(storedValues.pendingEntry(steerId), {
|
||||
type: "message",
|
||||
payload: { role: "user", content: "late steer", timestamp: 4 },
|
||||
}),
|
||||
storedValues.setValue(storedValues.operationState(operationId), nextState),
|
||||
],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => undefined,
|
||||
};
|
||||
}, BACKGROUND_CONTEXT);
|
||||
releaseHook.resolve();
|
||||
|
||||
expect(await generating).toEqual({ kind: "continue" });
|
||||
expect(effectPendingAtProvider).toBe(true);
|
||||
const settled = currentRun(fixture.lane);
|
||||
expect(settled.inbox.steer).toEqual([steerId]);
|
||||
if (settled.phase.kind !== "checkpoint") throw new Error("response did not reach checkpoint");
|
||||
const responseId = settled.latestAssistantEntryId;
|
||||
if (responseId === null) throw new Error("missing response id");
|
||||
const entry = await fixture.session.getEntry(responseId, BACKGROUND_CONTEXT);
|
||||
expect(entry).toMatchObject({ id: responseId, type: "message", message: { content: [{ text: "answer" }] } });
|
||||
expect(
|
||||
await fixture.session.readList(
|
||||
storedValues.pendingAssistantFrames(operationId, responseId),
|
||||
undefined,
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
).toEqual([]);
|
||||
const attempts = fixture.storage.getCommitAttempts();
|
||||
expect(attempts.some((writes) => writesOperationState(writes, "effect_pending"))).toBe(true);
|
||||
const settlement = attempts.find((writes) => writes.some((write) => write.kind === "entry"));
|
||||
expect(
|
||||
settlement?.map((write) =>
|
||||
write.kind === "value" || write.kind === "list"
|
||||
? `${write.kind}:${write.op}:${write.namespace}`
|
||||
: write.kind,
|
||||
),
|
||||
).toEqual([
|
||||
"entry",
|
||||
"usage",
|
||||
"value:set:pi.lane.leaf",
|
||||
"list:delete:pi.pending.assistant_frame",
|
||||
"value:set:pi.op.state",
|
||||
]);
|
||||
expect(fixture.events.map((event) => event.type)).toEqual(
|
||||
expect.arrayContaining([
|
||||
"turn_start",
|
||||
"message_start",
|
||||
"message_update",
|
||||
"message_end",
|
||||
"entry_added",
|
||||
"usage",
|
||||
"turn_end",
|
||||
]),
|
||||
);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
|
||||
it("does not await frame storage inside the provider event loop and persists frames in order", async () => {
|
||||
const backend = new FrameBlockingMemoryStorage({ now: () => 100 });
|
||||
const fixture = await createFixture(backend);
|
||||
const ready = await advanceToReady(fixture);
|
||||
const base = fixture.faux.provider;
|
||||
const provider: Provider = {
|
||||
id: base.id,
|
||||
name: base.name,
|
||||
auth: base.auth,
|
||||
getModels: () => base.getModels(),
|
||||
stream: (model, context, options) => base.stream(model, context, options),
|
||||
streamSimple: () => {
|
||||
const stream = createAssistantMessageEventStream();
|
||||
const partial: AssistantMessage = {
|
||||
...fauxAssistantMessage([], { timestamp: 5 }),
|
||||
stopReason: "pending",
|
||||
};
|
||||
const text = { type: "text" as const, text: "" };
|
||||
void (async () => {
|
||||
stream.push({ type: "start", partial });
|
||||
await backend.frameStarted.promise;
|
||||
partial.content.push(text);
|
||||
stream.push({ type: "text_start", contentIndex: 0, partial });
|
||||
await Promise.resolve();
|
||||
text.text = "ab";
|
||||
stream.push({ type: "text_delta", contentIndex: 0, delta: "ab", partial });
|
||||
await Promise.resolve();
|
||||
stream.push({ type: "text_end", contentIndex: 0, content: "ab", partial });
|
||||
await Promise.resolve();
|
||||
stream.push({ type: "done", reason: "stop", message: fauxAssistantMessage("ab", { timestamp: 5 }) });
|
||||
})();
|
||||
return stream;
|
||||
},
|
||||
};
|
||||
fixture.models.setProvider(provider);
|
||||
backend.blockNextFrame = true;
|
||||
fixture.storage.clearCommitAttempts();
|
||||
|
||||
const generating = runGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), ready);
|
||||
await backend.frameStarted.promise;
|
||||
await expect
|
||||
.poll(() => fixture.events.filter((event) => event.type === "message_update").length)
|
||||
.toBeGreaterThan(1);
|
||||
backend.releaseFrame.resolve();
|
||||
await generating;
|
||||
|
||||
const frames = fixture.storage
|
||||
.getCommitAttempts()
|
||||
.flatMap((writes) =>
|
||||
writes.flatMap((write) =>
|
||||
write.kind === "list" && write.op === "append" && write.namespace === "pi.pending.assistant_frame"
|
||||
? [write.value]
|
||||
: [],
|
||||
),
|
||||
);
|
||||
expect(
|
||||
frames.map((frame) =>
|
||||
typeof frame === "object" && frame !== null && "type" in frame ? frame.type : undefined,
|
||||
),
|
||||
).toEqual(["start", "text_start", "text_delta", "text_end"]);
|
||||
expect(frames[0]).toMatchObject({ partial: { content: [] } });
|
||||
expect(frames[2]).toMatchObject({ delta: "ab" });
|
||||
});
|
||||
|
||||
it("enters configuration failure without reserving response ids or calling the provider", async () => {
|
||||
const fixture = await createFixture();
|
||||
await fixture.lane.setActiveTools(["missing"], BACKGROUND_CONTEXT);
|
||||
const ready = await advanceToReady(fixture);
|
||||
fixture.storage.clearCommitAttempts();
|
||||
|
||||
expect(await runGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), ready)).toEqual({
|
||||
kind: "continue",
|
||||
});
|
||||
|
||||
const run = currentRun(fixture.lane);
|
||||
expect(run.phase).toMatchObject({
|
||||
kind: "failure_drain",
|
||||
error: { code: "configured_tools_unavailable", details: { tools: ["missing"] } },
|
||||
provenance: { kind: "configuration" },
|
||||
});
|
||||
expect(fixture.faux.state.callCount).toBe(0);
|
||||
expect(
|
||||
fixture.storage
|
||||
.getCommitAttempts()
|
||||
.flat()
|
||||
.some((write) => write.kind === "entry" || write.kind === "usage"),
|
||||
).toBe(false);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
|
||||
it("fails a missing captured model before reserving ids", async () => {
|
||||
const fixture = await createFixture();
|
||||
const ready = await advanceToReady(fixture);
|
||||
fixture.models.deleteProvider(fixture.faux.provider.id);
|
||||
fixture.storage.clearCommitAttempts();
|
||||
|
||||
expect(await runGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), ready)).toEqual({
|
||||
kind: "continue",
|
||||
});
|
||||
|
||||
expect(currentRun(fixture.lane).phase).toMatchObject({
|
||||
kind: "failure_drain",
|
||||
error: { code: "model_unavailable" },
|
||||
provenance: { kind: "configuration" },
|
||||
});
|
||||
expect(fixture.faux.state.callCount).toBe(0);
|
||||
expect(
|
||||
fixture.storage
|
||||
.getCommitAttempts()
|
||||
.flat()
|
||||
.some((write) => write.kind === "entry" || write.kind === "usage"),
|
||||
).toBe(false);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
|
||||
it("declines intent when cancellation or exact-owner replacement wins preparation", async () => {
|
||||
const cancelled = await createFixture();
|
||||
const cancelledReady = await advanceToReady(cancelled);
|
||||
const cancelHookStarted = deferred();
|
||||
const releaseCancelHook = deferred();
|
||||
cancelled.hooks.on("before_request", async () => {
|
||||
cancelHookStarted.resolve();
|
||||
await releaseCancelHook.promise;
|
||||
return undefined;
|
||||
});
|
||||
cancelled.storage.clearCommitAttempts();
|
||||
const cancelling = runGeneration(cancelled.lane, cancelled.drive, currentRun(cancelled.lane), cancelledReady);
|
||||
await cancelHookStarted.promise;
|
||||
await cancelled.lane.command((state) => {
|
||||
const operation = state.operation;
|
||||
if (operation?.state.kind !== "run") throw new Error("missing run");
|
||||
const nextState: RunState = {
|
||||
...operation.state,
|
||||
control: { status: "cancel_requested", requestedAt: 10, drainedSteer: [], drainedFollowUp: [] },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [storedValues.setValue(storedValues.operationState(operationId), nextState)],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => undefined,
|
||||
};
|
||||
}, BACKGROUND_CONTEXT);
|
||||
releaseCancelHook.resolve();
|
||||
expect(await cancelling).toEqual({ kind: "continue" });
|
||||
expect(cancelled.faux.state.callCount).toBe(0);
|
||||
expect(
|
||||
cancelled.storage.getCommitAttempts().some((writes) => writesOperationState(writes, "effect_pending")),
|
||||
).toBe(false);
|
||||
|
||||
const replaced = await createFixture();
|
||||
const replacedReady = await advanceToReady(replaced);
|
||||
const ownerHookStarted = deferred();
|
||||
const releaseOwnerHook = deferred();
|
||||
replaced.hooks.on("before_request", async () => {
|
||||
ownerHookStarted.resolve();
|
||||
await releaseOwnerHook.promise;
|
||||
return undefined;
|
||||
});
|
||||
replaced.storage.clearCommitAttempts();
|
||||
const losing = runGeneration(replaced.lane, replaced.drive, currentRun(replaced.lane), replacedReady);
|
||||
await ownerHookStarted.promise;
|
||||
replaced.lane.activeDrive = new Drive({ operationId }, BACKGROUND_CONTEXT);
|
||||
releaseOwnerHook.resolve();
|
||||
expect(await losing).toEqual({ kind: "lost_ownership" });
|
||||
expect(replaced.faux.state.callCount).toBe(0);
|
||||
expect(replaced.storage.getCommitAttempts()).toEqual([]);
|
||||
});
|
||||
|
||||
it("recovers an orphan with no frames into a zero-usage retry wait", async () => {
|
||||
const fixture = await createFixture();
|
||||
const ready = await advanceToReady(fixture);
|
||||
const responseEntryId = "01950000-0000-7000-8000-000000000028";
|
||||
const usageId = "01950000-0000-7000-8000-000000000029";
|
||||
const pending: Extract<Generation, { status: "effect_pending" }> = {
|
||||
status: "effect_pending",
|
||||
context: { ...ready.context, retryPolicy: { maxAttempts: 2, baseDelayMs: 1 } },
|
||||
attempt: 1,
|
||||
responseEntryId,
|
||||
usageId,
|
||||
intendedOutputLimit: 100,
|
||||
contextWindow: 1_000,
|
||||
};
|
||||
await fixture.lane.command((state) => {
|
||||
const operation = state.operation;
|
||||
if (operation?.state.kind !== "run") throw new Error("missing run");
|
||||
const nextState: RunState = {
|
||||
...operation.state,
|
||||
phase: { kind: "assistant", generation: pending },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [storedValues.setValue(storedValues.operationState(operationId), nextState)],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => undefined,
|
||||
};
|
||||
}, BACKGROUND_CONTEXT);
|
||||
|
||||
await recoverAssistantGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), pending);
|
||||
|
||||
const entry = (await fixture.session.getEntry(responseEntryId, BACKGROUND_CONTEXT)) as MessageEntry;
|
||||
expect(entry.message).toMatchObject({
|
||||
api: "unknown",
|
||||
provider: ready.context.configuration.model.provider,
|
||||
model: ready.context.configuration.model.modelId,
|
||||
content: [],
|
||||
stopReason: "error",
|
||||
usage: { totalTokens: 0 },
|
||||
});
|
||||
expect(currentRun(fixture.lane).phase).toMatchObject({
|
||||
kind: "assistant",
|
||||
generation: { status: "retry_wait", nextAttempt: 2 },
|
||||
});
|
||||
expect(fixture.faux.state.callCount).toBe(0);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
|
||||
it("recovers an orphan from committed frames without a provider or response hook", async () => {
|
||||
const fixture = await createFixture();
|
||||
const ready = await advanceToReady(fixture);
|
||||
const responseEntryId = "01950000-0000-7000-8000-000000000030";
|
||||
const usageId = "01950000-0000-7000-8000-000000000031";
|
||||
const partial: AssistantMessage = {
|
||||
...fauxAssistantMessage([], { timestamp: 6 }),
|
||||
stopReason: "pending",
|
||||
};
|
||||
const pending: Extract<Generation, { status: "effect_pending" }> = {
|
||||
status: "effect_pending",
|
||||
context: { ...ready.context, retryPolicy: { maxAttempts: 1, baseDelayMs: 1 } },
|
||||
attempt: 1,
|
||||
responseEntryId,
|
||||
usageId,
|
||||
intendedOutputLimit: 100,
|
||||
contextWindow: 1_000,
|
||||
};
|
||||
await fixture.lane.command((state) => {
|
||||
const operation = state.operation;
|
||||
if (operation?.state.kind !== "run") throw new Error("missing run");
|
||||
const nextState: RunState = {
|
||||
...operation.state,
|
||||
phase: { kind: "assistant", generation: pending },
|
||||
};
|
||||
return {
|
||||
kind: "commit",
|
||||
writes: [
|
||||
storedValues.setValue(storedValues.operationState(operationId), nextState),
|
||||
storedValues.appendList(storedValues.pendingAssistantFrames(operationId, responseEntryId), {
|
||||
type: "start",
|
||||
partial,
|
||||
}),
|
||||
storedValues.appendList(storedValues.pendingAssistantFrames(operationId, responseEntryId), {
|
||||
type: "text_start",
|
||||
contentIndex: 0,
|
||||
content: { type: "text", text: "" },
|
||||
}),
|
||||
storedValues.appendList(storedValues.pendingAssistantFrames(operationId, responseEntryId), {
|
||||
type: "text_delta",
|
||||
contentIndex: 0,
|
||||
delta: "draft",
|
||||
}),
|
||||
storedValues.appendList(storedValues.pendingAssistantFrames(operationId, responseEntryId), {
|
||||
type: "text_end",
|
||||
contentIndex: 0,
|
||||
content: "corrected",
|
||||
}),
|
||||
],
|
||||
next: { ...state, operation: { meta: operation.meta, state: nextState } },
|
||||
materialize: () => undefined,
|
||||
};
|
||||
}, BACKGROUND_CONTEXT);
|
||||
let afterResponseCalls = 0;
|
||||
fixture.hooks.on("after_response", () => {
|
||||
afterResponseCalls++;
|
||||
return undefined;
|
||||
});
|
||||
|
||||
expect(await recoverAssistantGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), pending)).toEqual({
|
||||
kind: "continue",
|
||||
});
|
||||
|
||||
expect(fixture.faux.state.callCount).toBe(0);
|
||||
expect(afterResponseCalls).toBe(0);
|
||||
const entry = (await fixture.session.getEntry(responseEntryId, BACKGROUND_CONTEXT)) as MessageEntry;
|
||||
expect(entry.message).toMatchObject({
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "corrected" }],
|
||||
stopReason: "error",
|
||||
usage: { input: 0, output: 0 },
|
||||
});
|
||||
expect(currentRun(fixture.lane).phase).toMatchObject({ kind: "failure_drain" });
|
||||
expect(
|
||||
await fixture.session.readList(
|
||||
storedValues.pendingAssistantFrames(operationId, responseEntryId),
|
||||
undefined,
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
).toEqual([]);
|
||||
expect(
|
||||
fixture.events.filter((event) => "recovery" in event && event.recovery).map((event) => event.type),
|
||||
).toEqual(["message_start", "message_end", "entry_added"]);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
|
||||
it("finishes the no-tool run with terminal cleanup and a hydratable final assistant", async () => {
|
||||
const fixture = await createFixture();
|
||||
const ready = await advanceToReady(fixture);
|
||||
fixture.faux.setResponses([fauxAssistantMessage("done", { timestamp: 7 })]);
|
||||
await runGeneration(fixture.lane, fixture.drive, currentRun(fixture.lane), ready);
|
||||
const run = currentRun(fixture.lane);
|
||||
if (run.phase.kind !== "checkpoint") throw new Error("missing terminal checkpoint");
|
||||
|
||||
const result = await runCheckpoint(fixture.lane, fixture.drive, run, run.phase);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
kind: "settled",
|
||||
outcome: { operation: "run", runId: operationId, kind: "completed", finalMessage: { role: "assistant" } },
|
||||
});
|
||||
expect(fixture.lane.state.operation).toBeNull();
|
||||
expect(fixture.lane.state.lastResult).toMatchObject({
|
||||
operationId,
|
||||
kind: "run",
|
||||
outcome: "completed",
|
||||
runCompletion: "assistant",
|
||||
});
|
||||
expect(
|
||||
await fixture.session.getValue(storedValues.operationMeta(operationId), BACKGROUND_CONTEXT),
|
||||
).toBeUndefined();
|
||||
expect(
|
||||
await fixture.session.getValue(storedValues.operationState(operationId), BACKGROUND_CONTEXT),
|
||||
).toBeUndefined();
|
||||
expect(fixture.events.filter((event) => event.type === "run_end")).toHaveLength(1);
|
||||
await expectProjectionRestores(fixture);
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue