diff --git a/packages/agent/docs/assistant-durability.md b/packages/agent/docs/assistant-durability.md index 9fb2de016..127e5df6e 100644 --- a/packages/agent/docs/assistant-durability.md +++ b/packages/agent/docs/assistant-durability.md @@ -3,7 +3,7 @@ This document specifies durable partial assistant messages for ordinary assistant generation and deferred-response polling. It builds on: - bound typed value/list addresses from `values.md`; -- `AssistantMessageFrame`, `assistantMessageEventToFrame()`, and `reduceAssistantMessageFrames()` from `@earendil-works/pi-ai`; +- `AssistantMessageFrame`, `AssistantMessageFrameEncoder`, and `reduceAssistantMessageFrames()` from `@earendil-works/pi-ai`; - the assistant intent/effect/settlement state machine in `harness.md`. The design persists compact replayable stream frames without making them operation-state authority and without storing a growing full partial message on every update. @@ -59,22 +59,26 @@ Each list element is one `AssistantMessageFrame`. The storage transaction's glob ## Frame contract -For each pi-ai stream event: +Create one pi-ai encoder per provider stream and feed it every event in order: ```ts -const frame = assistantMessageEventToFrame(event); +const encoder = new AssistantMessageFrameEncoder(); +const frame = encoder.encode(event); ``` -- `start` and non-terminal block events produce a frame; -- terminal `done` and `error` return `undefined` because final response settlement is separate; -- later frames omit repeated growing `partial` snapshots; -- text/thinking/tool end frames may contain one authoritative completed block value because some providers, including OpenAI Responses, can supply or correct content only at the end; -- unfinished tool-call JSON is reconstructed by pi-ai's reducer; -- completed tool-call arguments are parsed but remain unvalidated against a tool schema. +`partial` is the provider's shared live response-so-far helper, not an event-time snapshot. The encoder keeps per-open-block counters and trims text/thinking delta prefixes already represented by an advanced block-start snapshot. It temporarily buffers only the raw JSON prefix needed to synchronize an already-advanced tool call, then emits one checkpoint and resumes compact deltas. It never clones the growing full message on every event. + +- `start` produces an empty-content metadata frame; +- a non-terminal event produces zero or one frame; +- covered queued deltas produce no frame; +- terminal `done` and `error` produce no frame because final response settlement is separate; +- a setup `error` before `start` is valid and produces no frames; +- text/thinking/tool end frames contain the authoritative completed block value; +- completed tool-call arguments remain unvalidated against a tool schema. Do not define a second harness frame codec or reducer. Persistence stores the exported pi-ai value directly; hydration calls `reduceAssistantMessageFrames()`. -Provider event blocks may interleave. Storage and reduction rely on `contentIndex`, never block contiguity. +Provider event blocks may interleave. Encoding and reduction rely on `contentIndex`, never block contiguity. Text and ordinary thinking blocks must be empty when `*_start` is published and then append only through matching deltas until end; redacted thinking may be complete at start and emit no deltas. Streaming tool calls start with empty arguments and emit their complete raw JSON through deltas; a provider that starts with complete arguments must emit a cumulative delta prefix that parses to that snapshot at an event boundary before later argument deltas. ## Fresh stream scheduling @@ -83,15 +87,15 @@ The simplest scheduling is deliberate: one list-append transaction per converted For every `start` or update event: ```text -convert event to frame -→ synchronously enqueue invocation-fenced appendList(frames, frame) +encode the event against the per-stream frame encoder +→ when a frame is returned, synchronously enqueue invocation-fenced appendList(frames, frame) → attach the ordinary harness-fault observer to the returned promise → replace process-local latestFrameWrite promise reference → emit and await the existing message_start/message_update event → consume the next provider event ``` -`appendList()` is called synchronously, so all frame mutations enter the lane line in provider-event order. The procedure does not await each write. Every promise is immediately observed for fault propagation before the latest reference replaces its predecessor. Model output limits bound the number and total size of queued frames; a response may queue at most its bounded output plus constant frame/transaction overhead. +`appendList()` is called synchronously for each returned frame, so all frame mutations enter the lane line in provider-event order. The procedure does not await each write. Every promise is immediately observed for fault propagation before the latest reference replaces its predecessor. Covered queued events allocate no durable frame or write. Encoder state is proportional to open block count plus the unsynchronized prefix of an active tool-call JSON stream; it retains no second full assistant message. Model output limits bound queued frame bytes to bounded output plus frame/transaction overhead. The event side retains existing behavior: `AssistantStreamObserver.start/update` awaits `events.emit()` for every event. There is no separate latest event-delivery promise because no assistant event delivery remains outstanding when the provider loop advances. @@ -214,7 +218,7 @@ Reconnect replays no historical `message_start` or `message_update` events. The ## Events -Live event ordering remains: +A started generation retains this live event ordering: ```text message_start @@ -227,6 +231,8 @@ message_start → usage ``` +A request setup failure may produce `error` before `start`; that path emits no `message_start` or frame and proceeds through `after_response`, `message_end`, and ordinary error settlement. Successful `done` and updates before `start` are protocol defects. + Frame commits emit ordinary storage telemetry only. There is no public frame event and no claim that a `message_update` was durable. `entry_added` remains the only proof that the final assistant entry committed. A crash may occur after a live update event but before its asynchronously queued frame append commits. Reconnect then shows the latest committed frame prefix, which may be older than the last live event. @@ -277,7 +283,7 @@ JSONL retains deleted frame bytes until snapshot compaction. Logical deletion is 2. One effect-pending response ID constructs exactly one assistant frame-list address. 3. Every stored element is an exported pi-ai `AssistantMessageFrame`. 4. Terminal `done`/`error` events are never stored as frames. -5. Frame order equals provider event order. +5. Frame order is a subsequence of provider event order; zero-frame covered events do not disturb order. 6. Awaiting the latest frame-write promise at stream settlement implies all accepted appends completed. 7. Frames never establish provider completion or suppress unknown-outcome recovery. 8. Final or synthetic response settlement atomically deletes the exact frame list. @@ -290,8 +296,9 @@ JSONL retains deleted frame bytes until snapshot compaction. Logical deletion is ### Frame integration -- each convertible pi-ai event appends exactly one frame; -- `done`/`error` append nothing; +- one per-stream encoder handles shared live partials and synchronous event bursts without duplicate content; +- each event appends zero or one frame, with covered queued deltas appending nothing; +- `done`/`error` append nothing, including pre-generation error; - interleaved content indexes preserve sequence; - provider loop does not await individual frame writes; - frame appends enqueue synchronously before the next provider event; @@ -345,4 +352,4 @@ Expected runtime areas: - terminal cleanup, forks, and migrations; - instrumented writer and backend conformance tests. -Implement bound typed value/list addresses first, then frame enqueue/settlement, then recovery/snapshots. Update `harness.md` with this complete lifecycle before implementing runtime2 assistant parity. +Implement bound typed value/list addresses first, then frame enqueue/settlement, then recovery/snapshots. Update `harness.md` with this complete lifecycle before implementing runtime assistant parity. diff --git a/packages/agent/docs/harness.md b/packages/agent/docs/harness.md index 8745ff325..3f1342f55 100644 --- a/packages/agent/docs/harness.md +++ b/packages/agent/docs/harness.md @@ -171,7 +171,7 @@ TX[ insert injected messages if any, upsert pi.lane.leaf when needed, TX[ upsert pi.op.state/O = assistant ready (config snapshot) ] 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, +TX[ append pi.pending.assistant_frame/O:n2 += frame ] ← zero or one per non-terminal event, enqueued without awaiting TX[ insert entry n2, insert usage u1, upsert pi.lane.leaf = n2, delete list pi.pending.assistant_frame/O:n2, @@ -246,7 +246,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`, `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. +- `AssistantMessageFrame`, `AssistantMessageFrameEncoder`, 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`. - `Context`, `ContextKey`, `BACKGROUND_CONTEXT`, and the immutable derivation helpers: `packages/agent/src/harness/context.ts`. Context is invocation-scoped propagation and authority, not a durable payload or receiver default. @@ -1609,7 +1609,7 @@ Every settlement row above — tool plan, retry, overflow, deferred, checkpoint, ### Streamed frame persistence -During the effect window, the assistant procedure converts every provider stream event with pi-ai's `assistantMessageEventToFrame`. Terminal `done`/`error` events convert to no frame — final settlement is separate and terminal events are never stored. The converter snapshots the initial partial once in the `start` frame, strips repeated growing `partial` snapshots from later frames, and lets end frames carry one authoritative completed block value where a provider protocol supplies or corrects content only at the end. Interleaved blocks rely on `contentIndex`, never contiguity. Hydration calls `reduceAssistantMessageFrames`; the harness defines no second frame codec or reducer. +During the effect window, the assistant procedure owns one pi-ai `AssistantMessageFrameEncoder` and feeds it every provider stream event in order. Provider `partial` values remain shared live response-so-far helpers. The encoder emits an empty-content start frame, snapshots each block once when its queued start is consumed, trims text/thinking delta prefixes already covered by that snapshot, and uses a bounded raw-JSON checkpoint to synchronize an already-advanced tool call before compact deltas resume. Streaming tool calls begin with empty arguments and emit complete raw JSON through deltas; a provider that begins with complete arguments must emit a cumulative prefix that parses to that snapshot at an event boundary before later deltas. A non-terminal event therefore produces zero or one frame. Terminal `done`/`error` events produce no frame — final settlement is separate — and a setup `error` before `start` produces no frames. End frames carry one authoritative completed block value where a provider protocol supplies or corrects content only at the end. Interleaved blocks rely on `contentIndex`, never contiguity. Hydration calls `reduceAssistantMessageFrames`; the harness defines no second frame codec or reducer. For every convertible event, the procedure: @@ -2921,7 +2921,7 @@ type HarnessEventPayload = // Messages | { type: "message_start"; runId?: string; message: AgentMessage } | { type: "message_update"; runId: string; message: AgentMessage; - event: AssistantMessageEvent } + event: AssistantMessageEvent; frame?: AssistantMessageFrame } | { type: "message_end"; runId?: string; message: AgentMessage; entryId?: string } // Tools @@ -3004,14 +3004,14 @@ interface Events { Ordering for a streamed assistant response, asserted exactly by the conformance tests: ``` -message_start → message_update* (each delivery awaited; its converted frame -append is enqueued without awaiting storage) → stop frame admission and await +message_start → message_update* (each delivery awaited; any encoder-returned +frame append is enqueued without awaiting storage) → stop frame admission and await the latest frame write → after_response hook → message_end (final value, optional reserved id) → atomic response + usage + frame-list delete + classified-state commit → entry_added → usage ``` -Frame commits emit ordinary storage telemetry only: there is no frame-persistence event, and no event claims a `message_update` was durable. A crash may follow a live update whose queued frame append never committed; reconnect then shows the older committed frame prefix. Only `entry_added` proves message durability. Classification is computed before the transaction and becomes durable with it; it is not a separate event. Abort and overflow classification may normalize the committed response after `message_end`, so `entry_added` is authoritative for those two cases. A synthetic settlement performs no provider effect, update, or response hook: `message_start → message_end → atomic commit → entry_added → usage`. +A `message_update` carries the encoder's frame when that event advanced durable reconstruction and omits it when an older queued event was already covered by the live block-start snapshot; remote adapters forward only frame-bearing updates. Frame commits emit ordinary storage telemetry only: there is no frame-persistence event, and no event claims a `message_update` was durable. A crash may follow a live update whose queued frame append never committed; reconnect then shows the older committed frame prefix. Only `entry_added` proves message durability. Classification is computed before the transaction and becomes durable with it; it is not a separate event. Abort and overflow classification may normalize the committed response after `message_end`, so `entry_added` is authoritative for those two cases. A synthetic settlement performs no provider effect, update, or response hook: `message_start → message_end → atomic commit → entry_added → usage`. `tool_update` is always an immediate process-local event, including when its originating callback also requests `checkpoint:true`; no second public event announces checkpoint commit, and no checkpoint write is dropped or coalesced (§3.8). The synchronous callback retains the latest asynchronous delivery promise, and tool-promise settlement awaits it — together with the latest checkpoint-write promise — before `after_tool`, preserving the existing listener order without backpressuring each update. Reconnect snapshots reveal only the latest committed checkpoint. For a real effect, `tool_end` follows finalization but precedes the `outcome_ready` staging commit, in completion order; it is observation, not proof of durability. Finalized result-message lifecycle and `entry_added` occur later in source order when ready values materialize. Synthetic blocked and unsafe-recovery outcomes emit no tool-effect lifecycle. A crash after `tool_end` but before staging may recover or safely replay the still-uncertain call. Historical events are not replayed, though a safely replayed execution emits its own recovery-tagged lifecycle. @@ -3190,7 +3190,7 @@ interface AssistantResponseMetadata { interface AssistantStreamObserver { start(message: AssistantMessage, - event: Extract | undefined, + event: Extract, context: Context): void | Promise; update(message: AssistantMessage, event: AssistantMessageEvent, context: Context): void | Promise; @@ -3240,15 +3240,16 @@ 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, startEvent, context) -→ observer.update(message, event, context)* +→ either observer.start(message, startEvent, context) + → observer.update(message, event, context)* + or a pre-generation error with no start/update callback → settle the stream completely → afterResponse(settled message, captured metadata, context) → observer.end(message, 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` 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`. +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 feeds actual start/update events to one per-stream `AssistantMessageFrameEncoder` and synchronously enqueues each returned invocation-fenced frame append without awaiting storage (§3.7); covered queued events return no frame. `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. A pre-generation `error` emits no synthetic start: the adapter calls only `observer.end` after the response hook. An update or successful `done` before `start`, a duplicate start, or an event after terminal is a provider protocol defect. 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: @@ -3522,7 +3523,7 @@ These rows preserve requirements, not frozen package boundaries. Create their ha | R9 | **Threshold and overflow compaction** | In-run structural decision, durable once-per-trigger threshold marker, continuation preservation, all overflow predicates, atomic response/preparation publication, specified normalization/projection, one overflow recovery flag, bounded second failure, and deterministic tests for added boundaries. | Threshold decline/empty across reopen, all overflow classifier/preparation inputs, no overflow tool plan, genuine length, crash/reopen at every transition. | | R10 | **Navigation** | Add navigation `OperationRequest` acceptance without task installation and compose `navigateTree()` from accept+drive. Implement navigation source/target/summary consumption-time dereference, validation, summarized decision/generation, one final transaction combining move/summary/leaf/label with terminal writes, summary-only navigation hook, and deterministic transition coverage. | Explicit accept+drive versus convenience equivalence; root/current/unknown rejection; invalid durable source/target/summary combinations; accept/close/reopen/drive; summarized/unsummarized paths; final leaf at summary; abort race; exact atomic publication including cleanup. | | R11 | **Schema version and migrations** | Chained migrate-on-open under the writer lease, with total mappings for open `pi.op.state` including `outcome_ready`, staged `pi.pending.entry` results, invocation memos, optional progress checkpoints, and assistant frame lists (paged in sequence order, mapped preserving `seq` or deleted whole-key); address namespace/key/kind migrations; JSONL lenient old-shape replay, mandatory post-migration compaction, refuse-newer. | Version gate; chained crash-idempotent migration; planned/effect-pending/outcome-ready tool states mapped safely; staged result/checkpoint/memo mappings; every legacy frame mapped or the whole list explicitly discarded with no completion inference; resumed operation correctness; old bytes retired. | -| R12 | **Surface completion** | Complete full-session watch, remaining live execution snapshot updates and outcome-ready visibility, event catalog/order/filtering, telemetry, public exports, backend parity, hosted-control/convenience documentation, remove scaffold code including the S4 fake client, rename the sole implementation from `runtime2/` to `runtime/`, and update imports/tests. | Snapshot/event gap; attach during active effect with live/durable divergence for both frames and checkpoints; restored effect-pending exposing reduced `streamingMessage`; live partial precedence over durable reduction; outcome-ready, waiting/terminal states; no replay of old message or tool lifecycle; recovered turn brackets balanced by turn id across durable waits and reopened on the next pass; close before tool-effect admission creates no raw `pi.harness.tool` span; stale-wake fencing; sensitive-event/content-free telemetry; full race/crash matrix on all backends. | +| R12 | **Surface completion** | Complete full-session watch, remaining live execution snapshot updates and outcome-ready visibility, event catalog/order/filtering, telemetry, public exports, backend parity, hosted-control/convenience documentation, remove scaffold code including the S4 fake client, and update imports/tests. | Snapshot/event gap; attach during active effect with live/durable divergence for both frames and checkpoints; restored effect-pending exposing reduced `streamingMessage`; live partial precedence over durable reduction; outcome-ready, waiting/terminal states; no replay of old message or tool lifecycle; recovered turn brackets balanced by turn id across durable waits and reopened on the next pass; close before tool-effect admission creates no raw `pi.harness.tool` span; stale-wake fencing; sensitive-event/content-free telemetry; full race/crash matrix on all backends. | | P1 | **Protocol schemas** | Future work after the internal harness is complete: define shared TypeBox schemas for serializable pi-ai and harness protocol data, derive the corresponding TypeScript types from those schemas, and reuse them across client/server protocol boundaries. Validation runs only on untrusted wire input and never inside Session, Storage, operation procedures, or in-process extensions. | Schema/type parity, accepted and rejected client/server payloads, protocol round trips, and no validators or schema construction on internal storage paths. | # Part 9 — Invariants and tests @@ -3567,7 +3568,7 @@ Operations: 28. Completed tool calls form a source-ordered prefix. A sequential suffix permits at most one effect-pending or outcome-ready call before planned calls; a parallel suffix may mix `planned`, `effect_pending`, and `outcome_ready`. Completion-order outcome staging never extends the prefix; source-ordered materialization does. 29. Every outcome-ready call has exactly one matching finalized `pi.pending.entry`, no immutable result entry, no invocation memos, and no tool-output checkpoint. Outcome-ready and completed calls never execute again. 30. A tool progress checkpoint is an optional bounded complete `AgentToolResult` snapshot, selected with `checkpoint:true`. It never proves completion. Every selected checkpoint synchronously enqueues one invocation-fenced scalar replacement; no write is dropped or coalesced, only the latest write promise reference is retained, and awaiting it implies completion of every earlier write. Staging or terminal cleanup deletes the value and fences late recreation. -31. Scalar assistant/deferred state is the sole restart authority for streamed partials. One effect-pending response id constructs exactly one `pendingAssistantFrames(operationId, responseEntryId)` address; every element is an exported pi-ai `AssistantMessageFrame`; frame order equals provider event order; terminal `done`/`error` events are never stored; frames never establish provider completion or suppress unknown-outcome recovery. +31. Scalar assistant/deferred state is the sole restart authority for streamed partials. One effect-pending response id constructs exactly one `pendingAssistantFrames(operationId, responseEntryId)` address; every element is an exported pi-ai `AssistantMessageFrame`; frame order is a subsequence of provider event order because already-covered queued events produce no frame; terminal `done`/`error` events are never stored; frames never establish provider completion or suppress unknown-outcome recovery. 32. Every final or synthetic response settlement — normal, recovery, cancellation, or external finalization — atomically deletes its exact frame list. Idle forks contain no frame lists. A restored partial may appear in `streamingMessage` but never in `transcript` before settlement. 33. The provider loop never awaits storage per frame; frame appends are enqueued synchronously in provider-event order, and awaiting the latest frame-write promise at stream settlement implies every accepted append completed. 34. Successful attachment publishes only complete lane projections and an open-operation inventory. It resolves no model/tool identity and starts no work. A later drive fences the authoritative owned projection and exact `Drive` identity; storage reads only dereference payloads named by that projection. diff --git a/packages/agent/docs/values.md b/packages/agent/docs/values.md index 13915a3df..e27a8b7e0 100644 --- a/packages/agent/docs/values.md +++ b/packages/agent/docs/values.md @@ -435,7 +435,7 @@ Assistant partial durability is the first built-in list consumer: const frames = pendingAssistantFrames(operationId, responseEntryId); ``` -`AssistantMessageFrame`, `assistantMessageEventToFrame()`, and `reduceAssistantMessageFrames()` come from `@earendil-works/pi-ai`. Do not define a second frame codec or reducer. +`AssistantMessageFrame`, `AssistantMessageFrameEncoder`, and `reduceAssistantMessageFrames()` come from `@earendil-works/pi-ai`. Do not define a second frame codec or reducer. For every convertible non-terminal provider event, the assistant procedure: diff --git a/packages/agent/docs/work-packages/05-direct-durable-drive.md b/packages/agent/docs/work-packages/05-direct-durable-drive.md index 02041aac3..20c2e4cde 100644 --- a/packages/agent/docs/work-packages/05-direct-durable-drive.md +++ b/packages/agent/docs/work-packages/05-direct-durable-drive.md @@ -13,40 +13,40 @@ The review checkpoints at M1, M3, M5, M7 and the final review are mandatory proc Read completely, in this order, before editing anything. - Read **all** of `harness.md`, start to finish. Excerpts, searches, prior conversation, and this handoff are not substitutes. -- Treat the current `harness.md`, this handoff, and the current runtime2/session/execution APIs as the only architectural sources of truth. Do not rely on any external prototype, archive, or prior conversation. -- Implement runtime2 **clean-room**: do not inspect Git history for deleted runtime implementations, and do not recover, copy, adapt, or consult removed runtime1 code as a reference. +- Treat the current `harness.md`, this handoff, and the current runtime/session/execution APIs as the only architectural sources of truth. Do not rely on any external prototype, archive, or prior conversation. +- Implement runtime **clean-room**: do not inspect Git history for deleted runtime implementations, and do not recover, copy, adapt, or consult removed runtime1 code as a reference. - Reread each relevant current source file completely before changing it. 1. `packages/agent/docs/harness.md` — all parts, normative. 2. `packages/agent/src/harness/agent-harness.ts` — public types and interfaces. 3. `packages/agent/src/harness/session/types.ts` — durable types (`OperationState`, `RunState`, `RunPhase`, `Generation`, `Deferred`, `ToolBatch`, `ToolCall`, `StructuralDecision`, `NavigationState`, `LaneLastResult`, `SessionReader`, `Write`, `CommitResult`). 4. `packages/agent/src/harness/session/values.ts` — every typed address and write helper. -5. `packages/agent/src/harness/runtime2/lane.ts` — `Lane`, `Lane.command`, `LaneCommand`, `accept`. -6. `packages/agent/src/harness/runtime2/types.ts` — `Config`, `LaneState`, `LaneCommand`, `Drive`, `ProcedureResult`, and `SliceNotImplemented`. -7. `packages/agent/src/harness/runtime2/harness.ts` — `Harness`, config store, `fault`, `close`. -8. `packages/agent/src/harness/runtime2/restore.ts` — projection restore. +5. `packages/agent/src/harness/runtime/lane.ts` — `Lane`, `Lane.command`, `LaneCommand`, `accept`. +6. `packages/agent/src/harness/runtime/types.ts` — `Config`, `LaneState`, `LaneCommand`, `Drive`, `ProcedureResult`, and `SliceNotImplemented`. +7. `packages/agent/src/harness/runtime/harness.ts` — `Harness`, config store, `fault`, `close`. +8. `packages/agent/src/harness/runtime/restore.ts` — projection restore. 9. `packages/agent/src/harness/execution/effect-gate.ts`, `execution/assistant.ts`, `execution/tools.ts`. 10. `packages/agent/src/harness/hooks.ts`, `events.ts`, `context.ts`, `result.ts`, `telemetry.ts`. 11. `packages/agent/src/harness/types.ts` — `AgentHarnessTool`, `AgentHarnessToolInvocation`. 12. `packages/agent/src/harness/session/testing/storage-decorator.ts`, `instrumented-storage.ts`, `testing/index.ts`. -13. Tests: `test/harness/runtime2/{accept,harness,lane,restore,watch}.test.ts`, `test/harness/runtime2/test-utils.ts`, `test/harness/execution-primitives.test.ts`, `test/harness/types.test.ts`. +13. Tests: `test/harness/runtime/{accept,harness,lane,restore,watch}.test.ts`, `test/harness/runtime/test-utils.ts`, `test/harness/execution-primitives.test.ts`, `test/harness/types.test.ts`. --- ## 1. Module ownership and import direction ``` -agent-harness.ts public types + constructor -> runtime2/harness.ts +agent-harness.ts public types + constructor -> runtime/harness.ts session/** durable storage + session imports no harness runtime module execution/effect-gate Gate / GateControl / AbortRequested execution/assistant provider streaming execution/tools tool phase helpers -runtime2/types.ts Config / LaneState / LaneCommand / Drive / ProcedureResult -runtime2/progress.ts progress channels -> runtime2/types, type-only runtime2/lane -runtime2/drive/*.ts procedures -> runtime2/types, progress, type-only runtime2/lane -runtime2/drive.ts the switch -> runtime2/drive/*, type-only runtime2/lane (M7 only) -runtime2/lane.ts Lane owner + public surfaces -> runtime2/drive.ts (value import, M8) -runtime2/harness.ts Harness -> runtime2/lane.ts +runtime/types.ts Config / LaneState / LaneCommand / Drive / ProcedureResult +runtime/progress.ts progress channels -> runtime/types, type-only runtime/lane +runtime/drive/*.ts procedures -> runtime/types, progress, type-only runtime/lane +runtime/drive.ts the switch -> runtime/drive/*, type-only runtime/lane (M7 only) +runtime/lane.ts Lane owner + public surfaces -> runtime/drive.ts (value import, M8) +runtime/harness.ts Harness -> runtime/lane.ts ``` Rules: @@ -69,7 +69,7 @@ Rules: One installed process-local pass is one `Drive`. `Lane.activeDrive` is the ownership slot. The same object carries the shared completion, effect gate, invocation context, and pass-local wait policy; there is no handle, supervisor, scope, or separate capability object. ```ts -// packages/agent/src/harness/runtime2/types.ts +// packages/agent/src/harness/runtime/types.ts export class Drive { readonly operationId: string; readonly completion: Promise; @@ -116,7 +116,7 @@ async function runGeneration( Exact ownership and cleanup stay on `Lane`: ```ts -// packages/agent/src/harness/runtime2/lane.ts +// packages/agent/src/harness/runtime/lane.ts /** Package-internal so deterministic tests can install and replace exact owners. */ activeDrive: Drive | undefined; @@ -269,7 +269,7 @@ The retained pass context is `withoutAbortSignal(installerContext)`; admitted ef ### 2.3 Lane.drive claim / join / hydration — approved ```ts -// packages/agent/src/harness/runtime2/lane.ts +// packages/agent/src/harness/runtime/lane.ts type DriveClaim = | { kind: "join"; drive: Drive } | { kind: "install"; drive: Drive } @@ -376,10 +376,10 @@ private mismatch(expected: string, currentOperationId: string | null, last: Lane ### 2.4 Drive switch contract — approved -**Compile ordering.** `Drive` lives in `runtime2/types.ts` from M1; M2 replaced the initial `StepResult` with `ProcedureResult`. `runtime2/drive.ts` is created in M7, once every procedure module it imports exists. M2–M6 call procedures directly from tests with a concrete `Lane` and `Drive`; public drive stays guarded until M8. +**Compile ordering.** `Drive` lives in `runtime/types.ts` from M1; M2 replaced the initial `StepResult` with `ProcedureResult`. `runtime/drive.ts` is created in M7, once every procedure module it imports exists. M2–M6 call procedures directly from tests with a concrete `Lane` and `Drive`; public drive stays guarded until M8. ```ts -// packages/agent/src/harness/runtime2/types.ts +// packages/agent/src/harness/runtime/types.ts export type LostOwnership = { kind: "lost_ownership" }; export type ProcedureResult = @@ -399,7 +399,7 @@ export type ProcedureResult = There is no global `advance`/`reload` distinction. A procedure must not return `continue` from an unchanged state it does not understand; that is an invariant defect, not a spin-and-hope path. Whether a particular intent commit landed matters only to the straight-line procedure that is about to start its effect. Such a helper returns a small local result such as `"committed" | "state_changed" | LostOwnership`; it does not become drive-loop vocabulary. Tool children likewise use a tool-local result when the batch must know whether a child staged or materialized anything. ```ts -// packages/agent/src/harness/runtime2/drive.ts — M7 +// packages/agent/src/harness/runtime/drive.ts — M7 export async function runDrive( lane: Lane, drive: Drive, ): Promise { @@ -461,7 +461,7 @@ Procedures therefore do **not** reread `laneState`, `operationMeta`, `operationS Drive-owned commands use one small `Lane` variant so exact-object ownership is checked at commit admission rather than only in an asynchronous planner: ```ts -// packages/agent/src/harness/runtime2/lane.ts +// packages/agent/src/harness/runtime/lane.ts async commandDriveOwned( drive: Drive, plan: (state: LaneState, reader: SessionReader) => @@ -517,7 +517,7 @@ The single-owner contract is explicit: while a harness owns a `Session`, retaine ### 2.6 Generation procedure — approved shape, illustrative bodies ```ts -// packages/agent/src/harness/runtime2/drive/generation.ts +// packages/agent/src/harness/runtime/drive/generation.ts async function runReadyGeneration( lane: Lane, drive: Drive, run: RunState, gen: Extract, @@ -578,7 +578,7 @@ Preparation precedes the intent commit; settlement writes under the reserved ids ### 2.7 Progress helper — approved ```ts -// packages/agent/src/harness/runtime2/progress.ts +// packages/agent/src/harness/runtime/progress.ts export interface ProgressChannel { write(item: T): void; seal(): void; @@ -860,7 +860,7 @@ return lane.commandDriveOwned(drive, async (state, reader) => { Standalone structural and navigation procedures use the same visible suffix after their own result-publication writes. They do not call a generic terminal procedure. -`runtime2/drive/terminal.ts` contains only mechanical helpers: +`runtime/drive/terminal.ts` contains only mechanical helpers: - `operationCleanupWrites(reader, operationId, operationState, context)` scans typed operation prefixes and returns concrete deletes for meta/state, tool arguments, memos, preparations, tool-output checkpoints, the exact live assistant-frame list, and operation-owned pending entries; - `hydrateTerminalOutcome(reader, lastResult, context)` dereferences only entries directly named by the bounded value. @@ -1132,7 +1132,7 @@ Every milestone uses the same subsections. Public drive stays disabled until M8. ### Guard that keeps public drive disabled (M1–M7) -`runtime2/lane.ts` keeps throwing `SliceNotImplemented` from `drive`, `requestAbort`, `prompt`, `skill`, `promptFromTemplate`, `compact`, `navigateTree`, `resume`, `abort`, `steer`, `followUp`, `nextRun`, `cancelQueued`, `recordUsage`, `waitForIdle`, `runWhenIdle`, and from `accept({ kind: "compaction" })` / `accept({ kind: "navigation" })`. `requestAbort` stays guarded through M7 even though its primitive lands there: a durable `cancel_requested` marker on an operation that nothing can drive has no exit. Procedures are exercised with a concrete `Lane` and `Drive` constructed in each focused test against directly constructed durable state. **M8 removes those throws in one commit**, after every phase and reconciliation path exists. +`runtime/lane.ts` keeps throwing `SliceNotImplemented` from `drive`, `requestAbort`, `prompt`, `skill`, `promptFromTemplate`, `compact`, `navigateTree`, `resume`, `abort`, `steer`, `followUp`, `nextRun`, `cancelQueued`, `recordUsage`, `waitForIdle`, `runWhenIdle`, and from `accept({ kind: "compaction" })` / `accept({ kind: "navigation" })`. `requestAbort` stays guarded through M7 even though its primitive lands there: a durable `cancel_requested` marker on an operation that nothing can drive has no exit. Procedures are exercised with a concrete `Lane` and `Drive` constructed in each focused test against directly constructed durable state. **M8 removes those throws in one commit**, after every phase and reconciliation path exists. --- @@ -1143,7 +1143,7 @@ Every milestone uses the same subsections. Public drive stays disabled until M8. **Read completely before editing** - `packages/agent/src/harness/execution/breakpoint.ts` - `packages/agent/src/harness/agent-harness.ts` -- `packages/agent/src/harness/runtime2/{lane,types,harness}.ts` +- `packages/agent/src/harness/runtime/{lane,types,harness}.ts` - `packages/protocol/src/harness.ts`, `packages/protocol/src/codec.ts` - `packages/coding-agent/src/experimental/harness-wire-adapter.ts` - `packages/agent/test/harness/execution-primitives.test.ts`, `test/harness/types.test.ts` @@ -1159,9 +1159,9 @@ Every milestone uses the same subsections. Public drive stays disabled until M8. | Path | Symbols/sections | |---|---| | `src/harness/agent-harness.ts` | `ActionInfo`, `ActionRequired`; `ActionRequired` member of `RunOutcome`/`CompactionOutcome`/`NavigationOutcome`; `action_required` member of `DriveOutcome`; `SettledResumeOutcome`; `Exclude<…, ActionRequired>` in `TerminalOperationOutcome`; `LaneSnapshot.operation.action`; `AgentLane.peekAction`/`executeAction`/`runToCompletion`; `AgentHarnessOptions.drive` | -| `src/harness/runtime2/types.ts` | `Config.drive` | -| `src/harness/runtime2/harness.ts` | `drive: options.drive ?? "automatic"` | -| `src/harness/runtime2/lane.ts` | the three action-method stubs | +| `src/harness/runtime/types.ts` | `Config.drive` | +| `src/harness/runtime/harness.ts` | `drive: options.drive ?? "automatic"` | +| `src/harness/runtime/lane.ts` | the three action-method stubs | | `packages/protocol/src/harness.ts` | `ActionInfoSchema`; `action_required` member of `RunValueSchema`; optional `action` on the operation snapshot schema | | `packages/coding-agent/src/experimental/harness-wire-adapter.ts` | `case "action_required":` | | `packages/agent/docs/harness.md` | sections listed above | @@ -1201,19 +1201,19 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig **Read completely before editing** - `src/harness/execution/effect-gate.ts`, `src/harness/hooks.ts` -- `src/harness/runtime2/{lane,types,harness,restore}.ts` +- `src/harness/runtime/{lane,types,harness,restore}.ts` - `src/harness/session/{values,types,session,commit,in-memory-storage-state,memory}.ts` - `src/harness/session/jsonl/storage.ts` - `src/harness/session/testing/{storage-decorator,instrumented-storage,index}.ts` - `src/harness/session/testing/conformance/storage.ts` -- `packages/agent/src/index.ts` (confirm it exports no `runtime2` module, §2.11) -- `test/harness/runtime2/test-utils.ts`, `test/harness/instrumented-storage.test.ts`, `test/harness/types.test.ts` +- `packages/agent/src/index.ts` (confirm it exports no `runtime` module, §2.11) +- `test/harness/runtime/test-utils.ts`, `test/harness/instrumented-storage.test.ts`, `test/harness/types.test.ts` **Delete** — none. **Create** - `src/harness/session/testing/gating-storage.ts` — `GatingStorage` (§2.10) -- `src/harness/runtime2/progress.ts` — `ProgressChannel`, `openFrameProgress`, `openToolProgress` (§2.7) +- `src/harness/runtime/progress.ts` — `ProgressChannel`, `openFrameProgress`, `openToolProgress` (§2.7) - `test/harness/gating-storage.test.ts` **Modify** @@ -1222,10 +1222,10 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig | `src/harness/execution/effect-gate.ts` | add `Gate`, `GateControl`, `createGate()`, `DriveAbandoned` (§2.2); keep `AbortRequested`; retire `EffectGate.assertOpen` from the procedure-facing surface | | `src/harness/hooks.ts` | `runWithGate`/`runToolWithGate` accept `Gate` and use `admit` | | `src/harness/result.ts`, `src/harness/agent-harness.ts` | keep tagged/runtime errors in the dependency-neutral result module and re-export them from `agent-harness.ts`; this lets `agent-harness.ts` directly wire `createAgentHarness` while concrete Lane imports remain order-independent | -| `src/harness/runtime2/types.ts` | add the `Drive` class (completion, gate control, stripped pass Context, installer signal, wait policy, mutable `deferredPermits`, mutable `finalizedOutcome`) and the initial procedure-result union, simplified in M2; move `LaneCommand` here from `lane.ts` | -| `src/harness/runtime2/lane.ts` | make the class `Lane`; expose package-internal `session`, `models`, `hooks`, `emitBatch`, `command`, `readConfig`, and `mismatch` for direct procedure use; add package-internal `activeDrive: Drive | undefined` and `isDriveActive(drive)`; import `LaneCommand` from `types.ts`; remove the obsolete `AcceptanceConfig` type | -| `src/harness/runtime2/harness.ts` | `Harness extends Lane`; `buildLane` returns `Lane`; pass `HookRegistry` and full `Config` into `Lane` | -| `src/harness/runtime2/restore.ts` | name `Lane` wherever it names `Lane` | +| `src/harness/runtime/types.ts` | add the `Drive` class (completion, gate control, stripped pass Context, installer signal, wait policy, mutable `deferredPermits`, mutable `finalizedOutcome`) and the initial procedure-result union, simplified in M2; move `LaneCommand` here from `lane.ts` | +| `src/harness/runtime/lane.ts` | make the class `Lane`; expose package-internal `session`, `models`, `hooks`, `emitBatch`, `command`, `readConfig`, and `mismatch` for direct procedure use; add package-internal `activeDrive: Drive | undefined` and `isDriveActive(drive)`; import `LaneCommand` from `types.ts`; remove the obsolete `AcceptanceConfig` type | +| `src/harness/runtime/harness.ts` | `Harness extends Lane`; `buildLane` returns `Lane`; pass `HookRegistry` and full `Config` into `Lane` | +| `src/harness/runtime/restore.ts` | name `Lane` wherever it names `Lane` | | `src/harness/session/types.ts` | add `stats: SessionStats` to `CommitResult` (§2.12) | | `src/harness/session/commit.ts` | `PreparedCommit.result` becomes `Omit` | | `src/harness/session/in-memory-storage-state.ts` | `InMemoryStorageState.applyValidated` returns the post-apply `SessionStats` for Memory and JSONL | @@ -1237,7 +1237,7 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig | `test/harness/instrumented-storage.test.ts` | the three fabricated literals at `:58`, `:62`, `:74` gain `stats` (§2.12) | | `src/harness/session/session.ts` | make `idGenerator` injectable through constructor options (**named API change**; alternative: canonical id normalization in tests) | | `src/harness/session/testing/index.ts` | export `GatingStorage`, `CommitDiscarded` | -| `test/harness/runtime2/test-utils.ts` | construct concrete `Lane` and `Drive` objects directly; name concrete contexts wherever a Lane is constructed | +| `test/harness/runtime/test-utils.ts` | construct concrete `Lane` and `Drive` objects directly; name concrete contexts wherever a Lane is constructed | **Behavior/invariants landed.** `admit` is the only gate check; `GateControl` is owner-held; progress `drain()` propagates write failures; `GatingStorage` parks, waits, releases, and discards deterministically, and after `discard()` no commit reaches the backend by any path; `CommitResult.stats` reports exact post-commit session totals seeded from storage state; `Lane`/`Harness` remain generic in `TContext`; `Drive` is context-type independent; no `any`. @@ -1264,19 +1264,19 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig **Read completely before editing** - `src/harness/session/types.ts` (`OperationState`, `LaneLastResult`; `TerminalOperationOutcome` in `agent-harness.ts`) - `src/harness/session/values.ts` (all operation cleanup prefix constructors) -- `src/harness/runtime2/{lane,types,restore,progress}.ts` -- `test/harness/runtime2/{harness,watch,restore}.test.ts` +- `src/harness/runtime/{lane,types,restore,progress}.ts` +- `test/harness/runtime/{harness,watch,restore}.test.ts` **Create** -- `src/harness/runtime2/drive/terminal.ts` — `operationCleanupWrites`, `hydrateTerminalOutcome` (§2.9) -- `test/harness/runtime2/drive-terminal.test.ts` +- `src/harness/runtime/drive/terminal.ts` — `operationCleanupWrites`, `hydrateTerminalOutcome` (§2.9) +- `test/harness/runtime/drive-terminal.test.ts` **Modify** | Path | Change | |---|---| -| `src/harness/runtime2/types.ts` | replaced global `StepResult.advance/reload` with `ProcedureResult.continue`; retained waiting, settled, and lost ownership (§2.4) | -| `src/harness/runtime2/lane.ts` | add `commandDriveOwned`, with exact-Drive identity checked immediately before commit admission (§2.5); document `state` as the live control authority | -| `src/harness/runtime2/progress.ts` | use the planner's current `LaneState` and `commandDriveOwned`; remove durable lane/operation control reads (§2.7) | +| `src/harness/runtime/types.ts` | replaced global `StepResult.advance/reload` with `ProcedureResult.continue`; retained waiting, settled, and lost ownership (§2.4) | +| `src/harness/runtime/lane.ts` | add `commandDriveOwned`, with exact-Drive identity checked immediately before commit admission (§2.5); document `state` as the live control authority | +| `src/harness/runtime/progress.ts` | use the planner's current `LaneState` and `commandDriveOwned`; remove durable lane/operation control reads (§2.7) | | `src/harness/session/session.ts` | document retained mutators as forbidden while a harness owns the Session; no runtime compatibility layer | **Behavior/invariants landed.** A live `Lane.state` is the sole control-flow authority. Process loss restores it from durable values. Runtime reads dereference payloads only. Exact-Drive replacement between planner return and commit admission declines the commit. Cleanup enumerates every operation-owned address through typed prefix scans and concrete deletes; `pendingNextRun` is not cleanup-owned. Hydration reads only the bounded latest-result value and entries it directly names. @@ -1284,10 +1284,10 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/lane.test.ts` | exact owner replaced after an async planner returns but before commit admission → no write; commit admitted first → write and projection publish complete before replacement | -| `test/harness/runtime2/progress.test.ts` | current projection permits matching frame/tool progress; terminal or phase movement first → queued write declines; no durable control reads | -| `test/harness/runtime2/drive-terminal.test.ts` | cleanup completeness for run, compaction, navigation, including leftovers and exact live frame list; `pendingNextRun` excluded; bounded hydration equality for directly constructed latest-result values per kind | -| `test/harness/runtime2/restore.test.ts` | required durable projection absence/contradiction remains attachment corruption; a stored null leaf remains valid | +| `test/harness/runtime/lane.test.ts` | exact owner replaced after an async planner returns but before commit admission → no write; commit admitted first → write and projection publish complete before replacement | +| `test/harness/runtime/progress.test.ts` | current projection permits matching frame/tool progress; terminal or phase movement first → queued write declines; no durable control reads | +| `test/harness/runtime/drive-terminal.test.ts` | cleanup completeness for run, compaction, navigation, including leftovers and exact live frame list; `pendingNextRun` excluded; bounded hydration equality for directly constructed latest-result values per kind | +| `test/harness/runtime/restore.test.ts` | required durable projection absence/contradiction remains attachment corruption; a stored null leaf remains valid | **Commands.** `npm run check` · targeted vitest for every modified test file @@ -1306,30 +1306,30 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig - `test/harness/execution-assistant.test.ts` **Create** -- `src/harness/runtime2/drive/checkpoint.ts` — `startRun`, `runCheckpoint` -- `src/harness/runtime2/drive/generation.ts` — `runGeneration`, `settleAssistant`, `enterConfigurationFailure` (§2.6) -- `src/harness/runtime2/drive/recovery.ts` — synthetic interrupted-assistant settlement from committed frames, owned beside generation classification -- `test/harness/runtime2/drive-generation.test.ts` +- `src/harness/runtime/drive/checkpoint.ts` — `startRun`, `runCheckpoint` +- `src/harness/runtime/drive/generation.ts` — `runGeneration`, `settleAssistant`, `enterConfigurationFailure` (§2.6) +- `src/harness/runtime/drive/recovery.ts` — synthetic interrupted-assistant settlement from committed frames, owned beside generation classification +- `test/harness/runtime/drive-generation.test.ts` -**Do not create `drive.ts` here.** A switch written now would have to import `drive/deferred.ts` (M4), `drive/tools.ts` (M5), `drive/structural.ts` (M6), and `drive/reconcile.ts` (M7), none of which exist — the package would not compile, and stubbing those modules to make it compile would create exactly the half-built graph this plan forbids. `Drive` lives in `runtime2/types.ts` from M1 and `ProcedureResult` from M2, so procedures compile and are tested standalone. `drive.ts` is created once, in M7. +**Do not create `drive.ts` here.** A switch written now would have to import `drive/deferred.ts` (M4), `drive/tools.ts` (M5), `drive/structural.ts` (M6), and `drive/reconcile.ts` (M7), none of which exist — the package would not compile, and stubbing those modules to make it compile would create exactly the half-built graph this plan forbids. `Drive` lives in `runtime/types.ts` from M1 and `ProcedureResult` from M2, so procedures compile and are tested standalone. `drive.ts` is created once, in M7. **Modify** | Path | Change | |---|---| -| `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` | +| `src/harness/execution/assistant.ts` | use provider-facing `Tool` declarations; observer callbacks receive trailing Context, and `start` receives the actual start event; pre-generation errors synthesize no start, while updates or successful completion before start are protocol defects; 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. **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/drive-generation.test.ts` | each procedure invoked directly with a concrete `Lane` and `Drive`, no switch; intent before provider; reserved-id settlement; frame order without storage awaits; orphan recovery without provider call; mixed projecting/unprojected drains (leaf vs trigger); queued inbox mutation before intent → intent commits while preserving it; cancellation before intent → procedure returns `continue` and commits nothing; fence taken by another operation → returns `lost_ownership` and commits nothing; published projection equals fresh restore | +| `test/harness/runtime/drive-generation.test.ts` | each procedure invoked directly with a concrete `Lane` and `Drive`, no switch; intent before provider; reserved-id settlement; frame order without storage awaits; orphan recovery without provider call; mixed projecting/unprojected drains (leaf vs trigger); queued inbox mutation before intent → intent commits while preserving it; cancellation before intent → procedure returns `continue` and commits nothing; fence taken by another operation → returns `lost_ownership` and commits nothing; published projection equals fresh restore | **Commands.** `npm run check` · targeted vitest **Do not implement yet.** `drive.ts`, retry, deferred, tools, structural, reconciliation, public wiring. -**Exit condition / review questions.** Does any planner throw for an expected verification failure? Does any procedure start an effect on a path whose intent did not commit? Does any procedure derive state from a pre-job snapshot? Does `packages/agent/src/harness/runtime2/` contain any module importing a file that does not exist yet? +**Exit condition / review questions.** Does any planner throw for an expected verification failure? Does any procedure start an effect on a path whose intent did not commit? Does any procedure derive state from a pre-job snapshot? Does `packages/agent/src/harness/runtime/` contain any module importing a file that does not exist yet? --- @@ -1343,21 +1343,21 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig - `src/harness/agent-harness.ts` (`DriveOptions.waitForRetry`, `pollDeferred`, `DriveOutcome.waiting`) **Create** -- `src/harness/runtime2/drive/deferred.ts` -- `test/harness/runtime2/drive-retry-deferred.test.ts` +- `src/harness/runtime/drive/deferred.ts` +- `test/harness/runtime/drive-retry-deferred.test.ts` **Modify** | Path | Change | |---|---| -| `src/harness/runtime2/drive/generation.ts` | retry classification → `retry_wait`; `waitForRetry` policy | -| `src/harness/runtime2/drive/deferred.ts` | `drive.deferredPermits` consumption lives with the deferred procedure. **Do not touch `drive.ts`** — it does not exist until M7 | +| `src/harness/runtime/drive/generation.ts` | retry classification → `retry_wait`; `waitForRetry` policy | +| `src/harness/runtime/drive/deferred.ts` | `drive.deferredPermits` consumption lives with the deferred procedure. **Do not touch `drive.ts`** — it does not exist until M7 | **Behavior/invariants landed.** No durable armed poll state: `suspended` never becomes a durable pollable `ready`; the permit is consumed by the **fresh intent commit** that reserves new ids; a crash before that commit leaves `suspended` at the same poll number; an unknown-outcome poll does not consume a poll, deletes the old frame list, and mints fresh ids next poll; an unknown-outcome structural attempt *does* consume an attempt. **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/drive-retry-deferred.test.ts` | waiting outcomes; permit consumption point; crash before intent stays suspended; unknown-outcome poll number and id freshness; pass-local permit never mutates caller options | +| `test/harness/runtime/drive-retry-deferred.test.ts` | waiting outcomes; permit consumption point; crash before intent stays suspended; unknown-outcome poll number and id freshness; pass-local permit never mutates caller options | **Commands.** `npm run check` · targeted vitest @@ -1378,8 +1378,8 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig - `test/harness/execution-tools.test.ts`, `test/harness/tools.test.ts` **Create** -- `src/harness/runtime2/drive/tools.ts` (§2.8) -- `test/harness/runtime2/drive-tools.test.ts` +- `src/harness/runtime/drive/tools.ts` (§2.8) +- `test/harness/runtime/drive-tools.test.ts` **Modify** | Path | Change | @@ -1393,7 +1393,7 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/drive-tools.test.ts` | sequential and parallel two-call batches preserving sibling status; out-of-order completion → source-order placement; safe replay re-executes, unsafe does not; `outcome_ready` never re-executes; memo lifetime; checkpoint cadence; drain before `after_tool`; `addedToolNames` at placement; abort after admission | +| `test/harness/runtime/drive-tools.test.ts` | sequential and parallel two-call batches preserving sibling status; out-of-order completion → source-order placement; safe replay re-executes, unsafe does not; `outcome_ready` never re-executes; memo lifetime; checkpoint cadence; drain before `after_tool`; `addedToolNames` at placement; abort after admission | **Commands.** `npm run check` · targeted vitest @@ -1412,7 +1412,7 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig - `src/harness/session/types.ts` (`StructuralDecision`, `SummaryGeneration`, `SummaryContext`, `NavigationState`) - `test/harness/{compaction,branch-summarization}.test.ts` -**Ownership split (approved).** `runtime2/drive/structural.ts` owns **all** persistence: the preparation row, the decision, request intent rows, usage writes, the result entry, and the terminal transaction. The compaction modules stay pure generators of summary text and keep their existing behavior for their existing callers; they gain one seam so the runtime can wrap each provider call with commit intent → `gate.admit` → commit usage. +**Ownership split (approved).** `runtime/drive/structural.ts` owns **all** persistence: the preparation row, the decision, request intent rows, usage writes, the result entry, and the terminal transaction. The compaction modules stay pure generators of summary text and keep their existing behavior for their existing callers; they gain one seam so the runtime can wrap each provider call with commit intent → `gate.admit` → commit usage. Today `compact()`, `generateSummaryWithUsage()`, and the private `generateTurnPrefixSummary()` each call `completeSimpleWithRetries(models, model, …)` directly, and `generateBranchSummary()` does the same through `GenerateBranchSummaryOptions`. @@ -1460,7 +1460,7 @@ const requestOptions: SimpleStreamOptions = { So the drive runner composes the same four, with the signal and telemetry parent taken from the **admitted** context rather than the ambient one: ```ts -// packages/agent/src/harness/runtime2/drive/structural.ts — sketch +// packages/agent/src/harness/runtime/drive/structural.ts — sketch const request: NestedRequest = async (label, aiContext, options, context) => { const index = nextRequestIndex(label); // 0 for history, 1 for turn_prefix const usageId = lane.session.idGenerator.next(); @@ -1498,7 +1498,7 @@ The difference that matters is what it does **not** hand you. harness.md §3.9 r The default `retryingNestedRequest` keeps calling `completeSimpleWithRetries` unchanged, so existing callers are byte-identical and the two paths cannot diverge. ```ts -// packages/agent/src/harness/runtime2/drive/structural.ts — module-private, never exported. +// packages/agent/src/harness/runtime/drive/structural.ts — module-private, never exported. /** A NestedRequest cannot return without an AssistantMessage, so a declined local intent * unwinds to the owning structural procedure. This signal never escapes the module. */ @@ -1542,16 +1542,16 @@ A test must pin the option composition: assert the options reaching a faux `Mode Split-turn compaction makes two such calls — `"history"` then `"turn_prefix"` — so it produces two intents and two usage rows, which is exactly what harness.md §3.9 requires (`usageIds` accumulates). `AssistantMessage` already carries `usage`, so the runner persists what it needs without the compaction code knowing about storage. **Create** -- `src/harness/runtime2/drive/structural.ts` -- `test/harness/runtime2/drive-structural.test.ts` +- `src/harness/runtime/drive/structural.ts` +- `test/harness/runtime/drive-structural.test.ts` **Modify** | Path | Change | |---|---| -| `src/harness/runtime2/drive/checkpoint.ts` | threshold check keyed by `thresholdCheckedTriggerEntryId` | +| `src/harness/runtime/drive/checkpoint.ts` | threshold check keyed by `thresholdCheckedTriggerEntryId` | | `src/harness/compaction/compaction.ts` | add `NestedRequestLabel`, `NestedRequest`, `retryingNestedRequest`; add an **optional trailing** `request?: NestedRequest` parameter to `compact()`, `generateSummaryWithUsage()`, and the private `generateTurnPrefixSummary()`. Each replaces its `completeSimpleWithRetries(models, model, aiContext, options, retry, callbacks, context)` call with `(request ?? retryingNestedRequest(models, model, retry, callbacks))(label, aiContext, options, context)`, label `"history"` in `generateSummaryWithUsage` and `"turn_prefix"` in `generateTurnPrefixSummary`; `compact()` threads its `request` into both. `models`/`model` stay — they are still needed for `maxTokens`/`contextWindow` budgeting. **No persistence here**: `compaction.ts` never writes `operationPreparation` or any other value | | `src/harness/compaction/branch-summarization.ts` | add optional `request?: NestedRequest` to `GenerateBranchSummaryOptions`, defaulted in the destructuring in `generateBranchSummary` to `retryingNestedRequest(models, model, retry, callbacks)`, label `"branch"` | -| `src/harness/runtime2/drive/structural.ts` | owns **all** persistence: writes `operationPreparation(operationId, taskId)` before the decision hook and revalidates it against its source on resume; commits each request intent and usage; commits the result entry and the terminal transaction | +| `src/harness/runtime/drive/structural.ts` | owns **all** persistence: writes `operationPreparation(operationId, taskId)` before the decision hook and revalidates it against its source on resume; commits each request intent and usage; commits the result entry and the terminal transaction | The parameter is optional and defaulted precisely so existing callers are untouched and keep in-helper retry: `packages/coding-agent/src/core/agent-session.ts` (`compact(…)` at the compaction path and `generateBranchSummary(…)`) and `packages/agent/test/harness/compaction.test.ts` compile and behave identically. Only `drive/structural.ts` passes a `request`, and when it does, `retry` is left `undefined` so no retry can happen below the attempt level. @@ -1560,7 +1560,7 @@ The parameter is optional and defaulted precisely so existing callers are untouc **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/drive-structural.test.ts` | threshold once per boundary; decline/success/crash resume; **one intent and one usage row per provider request**, asserted through `InstrumentedStorage` for a split-turn attempt (two of each, not one); a retryable failure inside an attempt produces **zero** extra provider calls before the attempt-level `retry_wait` commit; crash after request one leaves its usage row and an uncertain attempt; model disappearance between requests fails in band; unknown attempt consumption; preparation written and revalidated by `drive/structural.ts` only; navigation with and without summary | +| `test/harness/runtime/drive-structural.test.ts` | threshold once per boundary; decline/success/crash resume; **one intent and one usage row per provider request**, asserted through `InstrumentedStorage` for a split-turn attempt (two of each, not one); a retryable failure inside an attempt produces **zero** extra provider calls before the attempt-level `retry_wait` commit; crash after request one leaves its usage row and an uncertain attempt; model disappearance between requests fails in band; unknown attempt consumption; preparation written and revalidated by `drive/structural.ts` only; navigation with and without summary | **Commands.** `npm run check` · targeted vitest @@ -1575,27 +1575,27 @@ The parameter is optional and defaulted precisely so existing callers are untouc **Goal.** Abort, close, fault, external finalization. Graph becomes total. **Read completely before editing** -- `src/harness/runtime2/harness.ts` (`fault`, `close`), `src/harness/runtime2/lane.ts` (`seal`) +- `src/harness/runtime/harness.ts` (`fault`, `close`), `src/harness/runtime/lane.ts` (`seal`) - `src/harness/events.ts`, `src/harness/context.ts` - `packages/agent/docs/harness.md` §4.6, §4.7, §4.9 **Create** -- `src/harness/runtime2/drive/reconcile.ts` — reconciliation per phase **and** the package-private cancellation primitive `requestOperationAbort(lane, operationId, context)` -- `src/harness/runtime2/drive.ts` — the total switch: `runDrive`, `hydrateSettled`, `dispatch`, `dispatchRun` (§2.4). Created **here**, not earlier, because this is the first milestone at which every module it imports (`checkpoint`, `generation`, `deferred`, `tools`, `structural`, `reconcile`, `terminal`, `recovery`) exists -- `test/harness/runtime2/drive-reconcile.test.ts` -- `test/harness/runtime2/drive-lifecycle.test.ts` -- `test/harness/runtime2/drive-switch.test.ts` +- `src/harness/runtime/drive/reconcile.ts` — reconciliation per phase **and** the package-private cancellation primitive `requestOperationAbort(lane, operationId, context)` +- `src/harness/runtime/drive.ts` — the total switch: `runDrive`, `hydrateSettled`, `dispatch`, `dispatchRun` (§2.4). Created **here**, not earlier, because this is the first milestone at which every module it imports (`checkpoint`, `generation`, `deferred`, `tools`, `structural`, `reconcile`, `terminal`, `recovery`) exists +- `test/harness/runtime/drive-reconcile.test.ts` +- `test/harness/runtime/drive-lifecycle.test.ts` +- `test/harness/runtime/drive-switch.test.ts` **Modify** | Path | Change | |---|---| -| `src/harness/runtime2/lane.ts` | add package-internal `startDrive`, exact `removeDrive`, `abandonDrive`, finalization notification, and seal behavior; lifecycle tests call these directly with an installed `Drive`. `Lane.requestAbort` and public `drive` keep their `SliceNotImplemented` guards until M8 | -| `src/harness/runtime2/harness.ts` | fault seals owners; close detaches non-cooperative effects | +| `src/harness/runtime/lane.ts` | add package-internal `startDrive`, exact `removeDrive`, `abandonDrive`, finalization notification, and seal behavior; lifecycle tests call these directly with an installed `Drive`. `Lane.requestAbort` and public `drive` keep their `SliceNotImplemented` guards until M8 | +| `src/harness/runtime/harness.ts` | fault seals owners; close detaches non-cooperative effects | **External finalization — owner notification (approved).** harness.md §4.9 requires the in-process finalizer to notify a live owner *immediately after* the terminal commit, not to let it discover the loss later. Discovery-only would let the task keep a provider stream or tool running until its next lane job, and would risk a second end event. ```ts -// packages/agent/src/harness/runtime2/lane.ts +// packages/agent/src/harness/runtime/lane.ts /** Package-internal M7 lifecycle seam; not part of AgentLane. */ interruptOwnerForFinalization( operationId: string, outcome: TerminalOperationOutcome, @@ -1634,11 +1634,11 @@ Both methods run on `Lane`; procedures do not remove ownership. The external fin **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/drive-reconcile.test.ts` | `requestOperationAbort` exercised directly with a concrete `Lane` and `Drive`, not through `Lane.requestAbort`; abort before/after admission at each admit site; mid-batch abort; reconciliation per phase; `aborted` response always under `cancel_requested` | -| `test/harness/runtime2/drive-switch.test.ts` | `before_drive` runs once, after the current owned operation is inspected; `before_drive` is **not** invoked for an already `cancel_requested` operation; `continue` requires a replaced projection and re-dispatches; unchanged `continue` faults instead of spinning; `lost_ownership` ends the pass with zero further commits; external finalization → `hydrateSettled` returns the same outcome; foreign operation on the fence → `OperationMismatch` | -| `test/harness/runtime2/drive-lifecycle.test.ts` | close **before** admission returns `Result.err(Closed)`; close **during** admitted work rejects `HarnessClosed` (these two are asserted separately, never with a shared loose matcher); non-cooperative effect; fault sealing; invocation cancel abandons at `effect_pending` with no durable write | -| `test/harness/runtime2/drive-lifecycle.test.ts` (finalization) | finalize while a **provider stream** is active, and again while a **tool** is active; also let an effect return successfully after committed finalization but before terminal event delivery, and attempt a same-id `drive` during that delivery: the owner's gate closes and its signal fires within committed materialization; the terminal event is delivered before shared completion settles; a cooperative effect rejects with `OperationEnded` and resolves through `drive.finalizedOutcome` **without entering a lane job**, while a successful return racing event delivery reaches `lost_ownership` but still returns `finalizedOutcome`; neither path settles/fails the Drive from the pass continuation or writes after the terminal commit (asserted through `InstrumentedStorage`: zero commit attempts after the terminal one); exactly **one** `run_end` is delivered; `accept` returns `LaneBusy` until both event-delivered settlement and the owner's `finally` have occurred, then succeeds; a non-cooperative effect that ignores the signal delays nothing and produces no unhandled rejection | -| `test/harness/runtime2/drive-lifecycle.test.ts` (catch boundary) | each signal from admitted work reaches its own arm and no other: `OperationEnded` → settled outcome, no second end event; `AbortRequested` → marker awaited, reconciliation runs, `aborted` under `cancel_requested`; `DriveAbandoned` → pass rejects, state stays `effect_pending`, **zero** durable writes, next drive recovers it as an orphan; `HarnessClosed` and `HarnessFault` → propagate unchanged, no synthetic settlement; `OperationMismatch` from `hydrateSettled` → propagates to `Result.err`. Every arm ends with the owner removed exactly once (§2.1 ordering) | +| `test/harness/runtime/drive-reconcile.test.ts` | `requestOperationAbort` exercised directly with a concrete `Lane` and `Drive`, not through `Lane.requestAbort`; abort before/after admission at each admit site; mid-batch abort; reconciliation per phase; `aborted` response always under `cancel_requested` | +| `test/harness/runtime/drive-switch.test.ts` | `before_drive` runs once, after the current owned operation is inspected; `before_drive` is **not** invoked for an already `cancel_requested` operation; `continue` requires a replaced projection and re-dispatches; unchanged `continue` faults instead of spinning; `lost_ownership` ends the pass with zero further commits; external finalization → `hydrateSettled` returns the same outcome; foreign operation on the fence → `OperationMismatch` | +| `test/harness/runtime/drive-lifecycle.test.ts` | close **before** admission returns `Result.err(Closed)`; close **during** admitted work rejects `HarnessClosed` (these two are asserted separately, never with a shared loose matcher); non-cooperative effect; fault sealing; invocation cancel abandons at `effect_pending` with no durable write | +| `test/harness/runtime/drive-lifecycle.test.ts` (finalization) | finalize while a **provider stream** is active, and again while a **tool** is active; also let an effect return successfully after committed finalization but before terminal event delivery, and attempt a same-id `drive` during that delivery: the owner's gate closes and its signal fires within committed materialization; the terminal event is delivered before shared completion settles; a cooperative effect rejects with `OperationEnded` and resolves through `drive.finalizedOutcome` **without entering a lane job**, while a successful return racing event delivery reaches `lost_ownership` but still returns `finalizedOutcome`; neither path settles/fails the Drive from the pass continuation or writes after the terminal commit (asserted through `InstrumentedStorage`: zero commit attempts after the terminal one); exactly **one** `run_end` is delivered; `accept` returns `LaneBusy` until both event-delivered settlement and the owner's `finally` have occurred, then succeeds; a non-cooperative effect that ignores the signal delays nothing and produces no unhandled rejection | +| `test/harness/runtime/drive-lifecycle.test.ts` (catch boundary) | each signal from admitted work reaches its own arm and no other: `OperationEnded` → settled outcome, no second end event; `AbortRequested` → marker awaited, reconciliation runs, `aborted` under `cancel_requested`; `DriveAbandoned` → pass rejects, state stays `effect_pending`, **zero** durable writes, next drive recovers it as an orphan; `HarnessClosed` and `HarnessFault` → propagate unchanged, no synthetic settlement; `OperationMismatch` from `hydrateSettled` → propagates to `Result.err`. Every arm ends with the owner removed exactly once (§2.1 ordering) | **Commands.** `npm run check` · targeted vitest @@ -1654,34 +1654,34 @@ Both methods run on `Lane`; procedures do not remove ownership. The external fin **Read completely before editing** - `src/harness/agent-harness.ts` (`AgentLane`, `AgentHarness`) -- `src/harness/runtime2/lane.ts` (all `SliceNotImplemented` stubs) -- `test/harness/runtime2/{accept,lane,harness,watch}.test.ts` +- `src/harness/runtime/lane.ts` (all `SliceNotImplemented` stubs) +- `test/harness/runtime/{accept,lane,harness,watch}.test.ts` **Create** -- `test/harness/runtime2/drive-ownership.test.ts` -- `test/harness/runtime2/drive-surfaces.test.ts` -- `test/harness/runtime2/drive-crash-matrix.test.ts` +- `test/harness/runtime/drive-ownership.test.ts` +- `test/harness/runtime/drive-surfaces.test.ts` +- `test/harness/runtime/drive-crash-matrix.test.ts` **Modify** | Path | Change | |---|---| -| `src/harness/runtime2/lane.ts` | **ordered**: (1) `accept` accepts `kind: "compaction"` and `kind: "navigation"` — today both throw `SliceNotImplemented` at the top of `accept`, so a standalone structural operation cannot be admitted at all; (2) implement the public `drive` claim/join/hydration surface (§2.3) using M7's internal lifecycle methods; `accept` returns `LaneBusy` while an owner exists; (3) expose `requestAbort` by delegating to the M7 primitive; (4) only then the convenience compositions `prompt`, `skill`, `promptFromTemplate`, `compact`, `navigateTree`, `resume`, `abort`, `steer`, `followUp`, `nextRun`, `cancelQueued`, `recordUsage`, `waitForIdle`, `runWhenIdle`; `inspectExecution` reports `running` when owned | -| `src/harness/runtime2/restore.ts` | restore every phase projection | +| `src/harness/runtime/lane.ts` | **ordered**: (1) `accept` accepts `kind: "compaction"` and `kind: "navigation"` — today both throw `SliceNotImplemented` at the top of `accept`, so a standalone structural operation cannot be admitted at all; (2) implement the public `drive` claim/join/hydration surface (§2.3) using M7's internal lifecycle methods; `accept` returns `LaneBusy` while an owner exists; (3) expose `requestAbort` by delegating to the M7 primitive; (4) only then the convenience compositions `prompt`, `skill`, `promptFromTemplate`, `compact`, `navigateTree`, `resume`, `abort`, `steer`, `followUp`, `nextRun`, `cancelQueued`, `recordUsage`, `waitForIdle`, `runWhenIdle`; `inspectExecution` reports `running` when owned | +| `src/harness/runtime/restore.ts` | restore every phase projection | | `src/harness/telemetry.ts` | drive spans | -| `src/harness/runtime2/types.ts` | `SliceNotImplemented` retained only for `watchSession` | +| `src/harness/runtime/types.ts` | `SliceNotImplemented` retained only for `watchSession` | `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. 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. +**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. Pi-ai's per-stream frame encoder treats provider `partial` as a shared live helper and removes already-covered queued deltas without cloning the growing message per event; the crash matrix pins synchronous shared-partial bursts. `AgentHarness.watchSession` is **explicitly deferred** and keeps its rejection. **Focused tests** | Path | Theme | |---|---| -| `test/harness/runtime2/drive-ownership.test.ts` | join before hydration; same-id join observes delivered events; mismatch ids. Both paths use the same result/error adapter: normal outcome → `Result.ok`; `OperationMismatch` and `Closed` → `Result.err`; `HarnessClosed`, `HarnessFault`, `DriveAbandoned` → rejection. Installer cancellation rejects installer and existing joiners with `DriveAbandoned`; joiner cancellation rejects only that joiner and leaves the Drive installed | -| `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; a provider that publishes start and then mutates one shared partial for block start without yielding still persists a reducible immutable start frame | +| `test/harness/runtime/drive-ownership.test.ts` | join before hydration; same-id join observes delivered events; mismatch ids. Both paths use the same result/error adapter: normal outcome → `Result.ok`; `OperationMismatch` and `Closed` → `Result.err`; `HarnessClosed`, `HarnessFault`, `DriveAbandoned` → rejection. Installer cancellation rejects installer and existing joiners with `DriveAbandoned`; joiner cancellation rejects only that joiner and leaves the Drive installed | +| `test/harness/runtime/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/runtime/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/runtime/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/runtime/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, block start, and deltas synchronously against one shared live partial persists a reducible non-duplicated frame sequence | **Commands.** `npm run check` · `./test.sh` @@ -1699,29 +1699,29 @@ Both methods run on `Lane`; procedures do not remove ownership. The external fin Milestone in parentheses where creation order matters. **Create** -- `packages/agent/src/harness/runtime2/drive.ts` (**M7** — not before; it imports every procedure module) -- `packages/agent/src/harness/runtime2/drive/checkpoint.ts` -- `packages/agent/src/harness/runtime2/drive/generation.ts` -- `packages/agent/src/harness/runtime2/drive/recovery.ts` -- `packages/agent/src/harness/runtime2/drive/deferred.ts` -- `packages/agent/src/harness/runtime2/drive/tools.ts` -- `packages/agent/src/harness/runtime2/drive/structural.ts` -- `packages/agent/src/harness/runtime2/drive/reconcile.ts` -- `packages/agent/src/harness/runtime2/drive/terminal.ts` -- `packages/agent/src/harness/runtime2/progress.ts` +- `packages/agent/src/harness/runtime/drive.ts` (**M7** — not before; it imports every procedure module) +- `packages/agent/src/harness/runtime/drive/checkpoint.ts` +- `packages/agent/src/harness/runtime/drive/generation.ts` +- `packages/agent/src/harness/runtime/drive/recovery.ts` +- `packages/agent/src/harness/runtime/drive/deferred.ts` +- `packages/agent/src/harness/runtime/drive/tools.ts` +- `packages/agent/src/harness/runtime/drive/structural.ts` +- `packages/agent/src/harness/runtime/drive/reconcile.ts` +- `packages/agent/src/harness/runtime/drive/terminal.ts` +- `packages/agent/src/harness/runtime/progress.ts` - `packages/agent/src/harness/session/testing/gating-storage.ts` - `packages/agent/test/harness/gating-storage.test.ts` -- `packages/agent/test/harness/runtime2/drive-terminal.test.ts` -- `packages/agent/test/harness/runtime2/drive-generation.test.ts` -- `packages/agent/test/harness/runtime2/drive-retry-deferred.test.ts` -- `packages/agent/test/harness/runtime2/drive-tools.test.ts` -- `packages/agent/test/harness/runtime2/drive-structural.test.ts` -- `packages/agent/test/harness/runtime2/drive-reconcile.test.ts` -- `packages/agent/test/harness/runtime2/drive-switch.test.ts` -- `packages/agent/test/harness/runtime2/drive-lifecycle.test.ts` -- `packages/agent/test/harness/runtime2/drive-ownership.test.ts` -- `packages/agent/test/harness/runtime2/drive-surfaces.test.ts` -- `packages/agent/test/harness/runtime2/drive-crash-matrix.test.ts` +- `packages/agent/test/harness/runtime/drive-terminal.test.ts` +- `packages/agent/test/harness/runtime/drive-generation.test.ts` +- `packages/agent/test/harness/runtime/drive-retry-deferred.test.ts` +- `packages/agent/test/harness/runtime/drive-tools.test.ts` +- `packages/agent/test/harness/runtime/drive-structural.test.ts` +- `packages/agent/test/harness/runtime/drive-reconcile.test.ts` +- `packages/agent/test/harness/runtime/drive-switch.test.ts` +- `packages/agent/test/harness/runtime/drive-lifecycle.test.ts` +- `packages/agent/test/harness/runtime/drive-ownership.test.ts` +- `packages/agent/test/harness/runtime/drive-surfaces.test.ts` +- `packages/agent/test/harness/runtime/drive-crash-matrix.test.ts` **Modify** - `packages/agent/src/harness/agent-harness.ts` @@ -1735,10 +1735,10 @@ Milestone in parentheses where creation order matters. - `packages/agent/src/harness/compaction/compaction.ts` - `packages/agent/src/harness/compaction/branch-summarization.ts` - `packages/agent/src/harness/tools/bash.ts` -- `packages/agent/src/harness/runtime2/lane.ts` -- `packages/agent/src/harness/runtime2/harness.ts` -- `packages/agent/src/harness/runtime2/types.ts` -- `packages/agent/src/harness/runtime2/restore.ts` +- `packages/agent/src/harness/runtime/lane.ts` +- `packages/agent/src/harness/runtime/harness.ts` +- `packages/agent/src/harness/runtime/types.ts` +- `packages/agent/src/harness/runtime/restore.ts` - `packages/agent/src/harness/session/session.ts` - `packages/agent/src/harness/session/types.ts` - `packages/agent/src/harness/session/commit.ts` @@ -1752,12 +1752,12 @@ Milestone in parentheses where creation order matters. - `packages/session-backends/sqlite-node/test/storage.test.ts` - `packages/agent/test/harness/execution-primitives.test.ts` - `packages/agent/test/harness/types.test.ts` -- `packages/agent/test/harness/runtime2/test-utils.ts` -- `packages/agent/test/harness/runtime2/accept.test.ts` -- `packages/agent/test/harness/runtime2/lane.test.ts` -- `packages/agent/test/harness/runtime2/harness.test.ts` -- `packages/agent/test/harness/runtime2/restore.test.ts` -- `packages/agent/test/harness/runtime2/watch.test.ts` +- `packages/agent/test/harness/runtime/test-utils.ts` +- `packages/agent/test/harness/runtime/accept.test.ts` +- `packages/agent/test/harness/runtime/lane.test.ts` +- `packages/agent/test/harness/runtime/harness.test.ts` +- `packages/agent/test/harness/runtime/restore.test.ts` +- `packages/agent/test/harness/runtime/watch.test.ts` - `packages/agent/docs/harness.md` - `packages/agent/docs/work-packages/02-atomic-run-acceptance.md` - `packages/agent/docs/work-packages/03-remove-drive-deadlines.md` @@ -1797,10 +1797,10 @@ npm run check grep -rn "Breakpoint\|action_required\|ActionInfo\|ActionRequired\|peekAction\|executeAction\|runToCompletion\|SettledResumeOutcome" \ packages/*/src packages/*/test packages/agent/docs/harness.md grep -rn "delete_prefix\|interface Deps\b" packages/agent/src -grep -rn "SliceNotImplemented" packages/agent/src/harness/runtime2 # watchSession only -grep -rn ": any\b\|\|as unknown as\|@ts-expect-error" packages/agent/src/harness/runtime2 -grep -rn 'kind: "advance"\|kind: "reload"' packages/agent/src/harness/runtime2 # must stay empty -grep -n "runtime2" packages/agent/src/index.ts # must stay empty (§2.11) +grep -rn "SliceNotImplemented" packages/agent/src/harness/runtime # watchSession only +grep -rn ": any\b\|\|as unknown as\|@ts-expect-error" packages/agent/src/harness/runtime +grep -rn 'kind: "advance"\|kind: "reload"' packages/agent/src/harness/runtime # must stay empty +grep -n "runtime" packages/agent/src/index.ts # must stay empty (§2.11) # CommitResult.stats reaches all THREE backends, not just the two in packages/agent: grep -rn "firstSeq" packages/agent/src packages/agent/test packages/session-backends grep -rn "prepared.result" packages/agent/src packages/session-backends # each composes stats diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 7a1f0767b..0022d75c8 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -2,6 +2,7 @@ import type { Api, AssistantMessage, AssistantMessageEvent, + AssistantMessageFrame, DeferredHandle, ImageContent, Message, @@ -52,7 +53,7 @@ export { UnknownTemplate, } from "./result.ts"; -import { createAgentHarness } from "./runtime2/harness.ts"; +import { createAgentHarness } from "./runtime/harness.ts"; import type { BranchSummaryEntry, CompactionEntry, @@ -321,6 +322,7 @@ export type HarnessEventPayload = runId: string; message: AgentMessage; event: AssistantMessageEvent; + frame?: AssistantMessageFrame; } | { type: "message_end"; runId?: string; message: AgentMessage; entryId?: string } | { diff --git a/packages/agent/src/harness/execution/assistant.ts b/packages/agent/src/harness/execution/assistant.ts index ca0c5336f..7faad5aac 100644 --- a/packages/agent/src/harness/execution/assistant.ts +++ b/packages/agent/src/harness/execution/assistant.ts @@ -25,7 +25,7 @@ export interface AssistantResponseMetadata { export interface AssistantStreamObserver { start( message: AssistantMessage, - event: Extract | undefined, + event: Extract, context: Context, ): void | Promise; update(message: AssistantMessage, event: AssistantMessageEvent, context: Context): void | Promise; @@ -130,17 +130,18 @@ export async function streamHarnessAssistant( let started = false; for await (const event of stream) { if (event.type === "start") { + if (started) throw new Error("Assistant message stream emitted more than one start event"); started = true; await config.observer.start({ ...event.partial }, event, context); } else if (isUpdateEvent(event)) { + if (!started) throw new Error(`Assistant message stream emitted ${event.type} before start`); await config.observer.update({ ...event.partial }, event, context); + } else if (event.type === "done" && !started) { + throw new Error("Assistant message stream emitted done before start"); } } const settled = (await stream.result()) as SettledAssistantMessage; - if (!started) { - await config.observer.start({ ...settled }, undefined, context); - } let finalMessage = settled; if (config.afterResponse) { try { diff --git a/packages/agent/src/harness/runtime2/drive/checkpoint.ts b/packages/agent/src/harness/runtime/drive/checkpoint.ts similarity index 100% rename from packages/agent/src/harness/runtime2/drive/checkpoint.ts rename to packages/agent/src/harness/runtime/drive/checkpoint.ts diff --git a/packages/agent/src/harness/runtime2/drive/generation.ts b/packages/agent/src/harness/runtime/drive/generation.ts similarity index 98% rename from packages/agent/src/harness/runtime2/drive/generation.ts rename to packages/agent/src/harness/runtime/drive/generation.ts index 6831f93d1..e6117b867 100644 --- a/packages/agent/src/harness/runtime2/drive/generation.ts +++ b/packages/agent/src/harness/runtime/drive/generation.ts @@ -1,7 +1,7 @@ import { type AssistantMessage, type AssistantMessageEvent, - assistantMessageEventToFrame, + AssistantMessageFrameEncoder, isRetryableAssistantError, type Tool, } from "@earendil-works/pi-ai"; @@ -547,6 +547,7 @@ export async function runGeneration( if (intent.kind === "state_changed") return { kind: "continue" }; const progress = openFrameProgress(lane, drive, responseEntryId); + const frameEncoder = new AssistantMessageFrameEncoder(); let drained = false; const drainProgress = async () => { if (drained) return; @@ -608,7 +609,7 @@ export async function runGeneration( }, observer: { start(message, event, context) { - const frame = event === undefined ? undefined : assistantMessageEventToFrame(event); + const frame = frameEncoder.encode(event); if (frame !== undefined) progress.write(frame); return lane.emitBatch( [{ type: "message_start", lane: lane.name, runId: drive.operationId, message }], @@ -616,7 +617,7 @@ export async function runGeneration( ); }, update(message, event: AssistantMessageEvent, context) { - const frame = assistantMessageEventToFrame(event); + const frame = frameEncoder.encode(event); if (frame !== undefined) progress.write(frame); return lane.emitBatch( [ @@ -626,6 +627,7 @@ export async function runGeneration( runId: drive.operationId, message, event, + ...(frame === undefined ? {} : { frame }), }, ], context, diff --git a/packages/agent/src/harness/runtime2/drive/recovery.ts b/packages/agent/src/harness/runtime/drive/recovery.ts similarity index 100% rename from packages/agent/src/harness/runtime2/drive/recovery.ts rename to packages/agent/src/harness/runtime/drive/recovery.ts diff --git a/packages/agent/src/harness/runtime2/drive/terminal.ts b/packages/agent/src/harness/runtime/drive/terminal.ts similarity index 100% rename from packages/agent/src/harness/runtime2/drive/terminal.ts rename to packages/agent/src/harness/runtime/drive/terminal.ts diff --git a/packages/agent/src/harness/runtime2/harness.ts b/packages/agent/src/harness/runtime/harness.ts similarity index 98% rename from packages/agent/src/harness/runtime2/harness.ts rename to packages/agent/src/harness/runtime/harness.ts index 33072b8b0..969df7249 100644 --- a/packages/agent/src/harness/runtime2/harness.ts +++ b/packages/agent/src/harness/runtime/harness.ts @@ -32,7 +32,7 @@ import { type Config, type LaneState, SliceNotImplemented } from "./types.ts"; type GlobalConfigProperty = GlobalConfigEventPayload["property"]; -/** Runtime2 implementation of AgentHarness. The harness is the main lane. */ +/** Runtime implementation of AgentHarness. The harness is the main lane. */ export class Harness extends Lane implements AgentHarness { readonly events: HarnessEventBus; readonly seed: LaneConfiguration; @@ -291,7 +291,7 @@ export class Harness extends Lane } } -/** Attach runtime2 without starting provider, tool, hook, or timer effects. */ +/** Attach runtime without starting provider, tool, hook, or timer effects. */ export async function createAgentHarness( options: AgentHarnessOptions, context: Context, diff --git a/packages/agent/src/harness/runtime2/index.ts b/packages/agent/src/harness/runtime/index.ts similarity index 100% rename from packages/agent/src/harness/runtime2/index.ts rename to packages/agent/src/harness/runtime/index.ts diff --git a/packages/agent/src/harness/runtime2/lane.ts b/packages/agent/src/harness/runtime/lane.ts similarity index 99% rename from packages/agent/src/harness/runtime2/lane.ts rename to packages/agent/src/harness/runtime/lane.ts index 97d7f51f5..bd987e5a5 100644 --- a/packages/agent/src/harness/runtime2/lane.ts +++ b/packages/agent/src/harness/runtime/lane.ts @@ -108,7 +108,7 @@ function capturedModel(operation: Operation): ModelIdentity | undefined { } } -/** Runtime2 implementation of one configured lane. */ +/** Runtime implementation of one configured lane. */ export class Lane implements AgentLane { readonly name: string; readonly sessionTree: SessionTree; diff --git a/packages/agent/src/harness/runtime2/progress.ts b/packages/agent/src/harness/runtime/progress.ts similarity index 100% rename from packages/agent/src/harness/runtime2/progress.ts rename to packages/agent/src/harness/runtime/progress.ts diff --git a/packages/agent/src/harness/runtime2/restore.ts b/packages/agent/src/harness/runtime/restore.ts similarity index 100% rename from packages/agent/src/harness/runtime2/restore.ts rename to packages/agent/src/harness/runtime/restore.ts diff --git a/packages/agent/src/harness/runtime2/types.ts b/packages/agent/src/harness/runtime/types.ts similarity index 100% rename from packages/agent/src/harness/runtime2/types.ts rename to packages/agent/src/harness/runtime/types.ts diff --git a/packages/agent/test/harness/execution-assistant.test.ts b/packages/agent/test/harness/execution-assistant.test.ts index e8a89a786..5129b5ffe 100644 --- a/packages/agent/test/harness/execution-assistant.test.ts +++ b/packages/agent/test/harness/execution-assistant.test.ts @@ -148,7 +148,7 @@ describe("streamHarnessAssistant", () => { start(message, event) { order.push("observer_start"); starts.push(message); - startEventType = event?.type; + startEventType = event.type; }, update(message) { order.push("observer_update"); @@ -243,44 +243,29 @@ describe("streamHarnessAssistant", () => { expect(lifecycle.filter((event) => event === "update").length).toBeGreaterThan(0); }); - it("synthesizes start from the settled response when the stream omits it", async () => { - const events: string[] = []; + it("rejects a successful terminal event before start", async () => { const final = assistant("complete"); let options: SimpleStreamOptions | undefined; - let synthesizedStartEvent: unknown = "unset"; - const result = await streamHarnessAssistant( - [user("prompt")], - { - model: model(), - systemPrompt: "system", - thinkingLevel: "off", - streamOptions: {}, - toProviderMessages, - request: (_context, requestOptions) => { - options = requestOptions; - const stream = createAssistantMessageEventStream(); - queueMicrotask(() => stream.push({ type: "done", reason: "stop", message: final })); - return stream; + await expect( + streamHarnessAssistant( + [user("prompt")], + { + model: model(), + systemPrompt: "system", + thinkingLevel: "off", + streamOptions: {}, + toProviderMessages, + request: (_context, requestOptions) => { + options = requestOptions; + const stream = createAssistantMessageEventStream(); + queueMicrotask(() => stream.push({ type: "done", reason: "stop", message: final })); + return stream; + }, + observer: { start() {}, update() {}, end() {} }, }, - observer: { - start(message, event) { - events.push(`start:${message.stopReason}`); - synthesizedStartEvent = event; - }, - update() { - events.push("update"); - }, - end(message) { - events.push(`end:${message.stopReason}`); - }, - }, - }, - BACKGROUND_CONTEXT, - ); - - expect(result).toBe(final); - expect(events).toEqual(["start:stop", "end:stop"]); - expect(synthesizedStartEvent).toBeUndefined(); + BACKGROUND_CONTEXT, + ), + ).rejects.toThrow("done before start"); expect(options).not.toHaveProperty("reasoning"); }); @@ -297,7 +282,10 @@ describe("streamHarnessAssistant", () => { toProviderMessages, request: () => { const stream = createAssistantMessageEventStream(); - queueMicrotask(() => stream.push({ type: "done", reason: "stop", message: final })); + queueMicrotask(() => { + stream.push({ type: "start", partial: { ...final, content: [], stopReason: "pending" } }); + stream.push({ type: "done", reason: "stop", message: final }); + }); return stream; }, afterResponse: async () => { @@ -335,8 +323,8 @@ describe("streamHarnessAssistant", () => { return stream; }, observer: { - start(message) { - events.push(`start:${message.stopReason}`); + start() { + throw new Error("pre-generation error must not synthesize start"); }, update() { events.push("update"); @@ -350,6 +338,6 @@ describe("streamHarnessAssistant", () => { ); expect(result).toBe(final); - expect(events).toEqual(["start:error", "end:error"]); + expect(events).toEqual(["end:error"]); }); }); diff --git a/packages/agent/test/harness/gating-storage.test.ts b/packages/agent/test/harness/gating-storage.test.ts index 08f6e4597..fa00fff03 100644 --- a/packages/agent/test/harness/gating-storage.test.ts +++ b/packages/agent/test/harness/gating-storage.test.ts @@ -4,7 +4,7 @@ import { MemoryStorage } from "../../src/harness/session/memory.ts"; import { CommitDiscarded, GatingStorage, InstrumentedStorage } from "../../src/harness/session/testing/index.ts"; import type { CommitResult, Write } from "../../src/harness/session/types.ts"; import * as storedValues from "../../src/harness/session/values.ts"; -import { deferred } from "./runtime2/test-utils.ts"; +import { deferred } from "./runtime/test-utils.ts"; class ControlledLandingStorage extends MemoryStorage { readonly started = deferred(); diff --git a/packages/agent/test/harness/runtime2/accept.test.ts b/packages/agent/test/harness/runtime/accept.test.ts similarity index 99% rename from packages/agent/test/harness/runtime2/accept.test.ts rename to packages/agent/test/harness/runtime/accept.test.ts index 0bde9488a..5915a8f25 100644 --- a/packages/agent/test/harness/runtime2/accept.test.ts +++ b/packages/agent/test/harness/runtime/accept.test.ts @@ -4,7 +4,7 @@ import { type HarnessEvent, HarnessFault, type OperationRequest } from "../../.. import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT, createContextKey, withContextValue } from "../../../src/harness/context.ts"; import type { Result } from "../../../src/harness/result.ts"; -import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts"; +import { createAgentHarness, Harness } from "../../../src/harness/runtime/harness.ts"; import { MemoryStorage } from "../../../src/harness/session/memory.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; import { InstrumentedStorage } from "../../../src/harness/session/testing/index.ts"; @@ -61,7 +61,7 @@ async function createHarness( await beforeCreate?.(created.session); const harnessOptions = { ...options(created.session), resources }; const { harness } = await createAgentHarness(harnessOptions, BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("Expected runtime2 Harness"); + if (!(harness instanceof Harness)) throw new Error("Expected runtime Harness"); created.storage.clearCommitAttempts(); return { harness, ...created, models: harnessOptions.models }; } @@ -75,7 +75,7 @@ afterEach(async () => { for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 atomic run acceptance", () => { +describe("runtime atomic run acceptance", () => { it.each([ ["text", { kind: "prompt", prompt: "hello" } satisfies OperationRequest, [{ type: "text", text: "hello" }]], [ @@ -313,7 +313,7 @@ describe("runtime2 atomic run acceptance", () => { BACKGROUND_CONTEXT, ); const { harness } = await createAgentHarness(options(session), BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("Expected runtime2 Harness"); + if (!(harness instanceof Harness)) throw new Error("Expected runtime Harness"); storage.failure = new Error("accept failed"); await expect(harness.accept({ kind: "prompt", prompt: "hello" }, BACKGROUND_CONTEXT)).rejects.toBeInstanceOf( diff --git a/packages/agent/test/harness/runtime2/drive-generation.test.ts b/packages/agent/test/harness/runtime/drive-generation.test.ts similarity index 97% rename from packages/agent/test/harness/runtime2/drive-generation.test.ts rename to packages/agent/test/harness/runtime/drive-generation.test.ts index 8909033dc..389272bd5 100644 --- a/packages/agent/test/harness/runtime2/drive-generation.test.ts +++ b/packages/agent/test/harness/runtime/drive-generation.test.ts @@ -12,12 +12,12 @@ import type { HarnessEvent, WatchHandle } from "../../../src/harness/agent-harne 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 { runCheckpoint, startRun } from "../../../src/harness/runtime/drive/checkpoint.ts"; +import { runGeneration } from "../../../src/harness/runtime/drive/generation.ts"; +import { recoverAssistantGeneration } from "../../../src/harness/runtime/drive/recovery.ts"; +import { Lane } from "../../../src/harness/runtime/lane.ts"; +import { restoreLane } from "../../../src/harness/runtime/restore.ts"; +import { type Config, Drive } from "../../../src/harness/runtime/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"; @@ -197,7 +197,7 @@ afterEach(async () => { for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 generation checkpoint", () => { +describe("runtime generation checkpoint", () => { it("consumes before_run and snapshots a ready generation", async () => { const fixture = await createFixture(); fixture.hooks.on("before_run", () => ({ @@ -274,7 +274,7 @@ describe("runtime2 generation checkpoint", () => { }); }); -describe("runtime2 assistant generation", () => { +describe("runtime 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); @@ -386,19 +386,15 @@ describe("runtime2 assistant generation", () => { stopReason: "pending", }; const text = { type: "text" as const, text: "" }; - void (async () => { + queueMicrotask(() => { 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; }, }; @@ -427,9 +423,9 @@ describe("runtime2 assistant generation", () => { frames.map((frame) => typeof frame === "object" && frame !== null && "type" in frame ? frame.type : undefined, ), - ).toEqual(["start", "text_start", "text_delta", "text_end"]); + ).toEqual(["start", "text_start", "text_end"]); expect(frames[0]).toMatchObject({ partial: { content: [] } }); - expect(frames[2]).toMatchObject({ delta: "ab" }); + expect(frames[1]).toMatchObject({ content: { text: "ab" } }); }); it("enters configuration failure without reserving response ids or calling the provider", async () => { diff --git a/packages/agent/test/harness/runtime2/drive-terminal.test.ts b/packages/agent/test/harness/runtime/drive-terminal.test.ts similarity index 98% rename from packages/agent/test/harness/runtime2/drive-terminal.test.ts rename to packages/agent/test/harness/runtime/drive-terminal.test.ts index 90fb42af9..51e433547 100644 --- a/packages/agent/test/harness/runtime2/drive-terminal.test.ts +++ b/packages/agent/test/harness/runtime/drive-terminal.test.ts @@ -1,7 +1,7 @@ import { fauxAssistantMessage } from "@earendil-works/pi-ai"; import { afterEach, describe, expect, it, vi } from "vitest"; import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; -import { hydrateTerminalOutcome, operationCleanupWrites } from "../../../src/harness/runtime2/drive/terminal.ts"; +import { hydrateTerminalOutcome, operationCleanupWrites } from "../../../src/harness/runtime/drive/terminal.ts"; import { insertEntry } from "../../../src/harness/session/commit.ts"; import { MemoryStorage } from "../../../src/harness/session/memory.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; @@ -100,7 +100,7 @@ afterEach(async () => { for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 terminal cleanup mechanics", () => { +describe("runtime terminal cleanup mechanics", () => { it("deletes every run-owned family and exact live frame list but preserves pendingNextRun", async () => { const { session } = await createSession(); const operationId = "run"; @@ -255,7 +255,7 @@ describe("runtime2 terminal cleanup mechanics", () => { }); }); -describe("runtime2 terminal outcome hydration", () => { +describe("runtime terminal outcome hydration", () => { it("hydrates run, compaction, and navigation outcomes from only their direct entry references", async () => { const { session, storage } = await createSession(); const assistant = fauxAssistantMessage([{ type: "text", text: "done" }]); diff --git a/packages/agent/test/harness/runtime2/harness.test.ts b/packages/agent/test/harness/runtime/harness.test.ts similarity index 99% rename from packages/agent/test/harness/runtime2/harness.test.ts rename to packages/agent/test/harness/runtime/harness.test.ts index ede2493cb..839dad0a5 100644 --- a/packages/agent/test/harness/runtime2/harness.test.ts +++ b/packages/agent/test/harness/runtime/harness.test.ts @@ -5,8 +5,8 @@ import { AgentHarness, HarnessClosed, HarnessFault } from "../../../src/harness/ import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; import type { Result } from "../../../src/harness/result.ts"; -import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts"; -import { Lane } from "../../../src/harness/runtime2/lane.ts"; +import { createAgentHarness, Harness } from "../../../src/harness/runtime/harness.ts"; +import { Lane } from "../../../src/harness/runtime/lane.ts"; import { MemorySessionRepo, type MemoryStorage } from "../../../src/harness/session/memory.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; import type { @@ -59,7 +59,7 @@ async function configureMain(session: Session, configuration: LaneConfiguration async function createStorageSession(storage: MemoryStorage): Promise { const session = new StorageBackedSession( - { id: `runtime2-storage-${standaloneSessions.length}`, createdAt: 1, storageVersion: 1 }, + { id: `runtime-storage-${standaloneSessions.length}`, createdAt: 1, storageVersion: 1 }, storage, ); standaloneSessions.push(session); @@ -143,7 +143,7 @@ afterEach(async () => { for (const session of standaloneSessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 AgentHarness", () => { +describe("runtime AgentHarness", () => { it("is selected by the public AgentHarness constructor", async () => { const session = await createSession(); @@ -608,7 +608,7 @@ describe("runtime2 AgentHarness", () => { it("faults without publishing a lane when creation commit fails", async () => { const { session, storage } = await createFailingSession(); const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("missing runtime2 harness"); + if (!(harness instanceof Harness)) throw new Error("missing runtime harness"); const failure = new Error("create lane failed"); storage.failure = failure; @@ -646,7 +646,7 @@ describe("runtime2 AgentHarness", () => { const storage = new ControlledMemoryStorage(); const session = await createStorageSession(storage); const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("missing runtime2 harness"); + if (!(harness instanceof Harness)) throw new Error("missing runtime harness"); const commitStarted = deferred(); const releaseCommit = deferred(); storage.beforeNextCommit = async () => { @@ -851,7 +851,7 @@ describe("runtime2 AgentHarness", () => { }, ]); const pendingId = await harness.sessionTree.appendCustomEntry("note", { text: "queued" }, BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("missing runtime2 harness"); + if (!(harness instanceof Harness)) throw new Error("missing runtime harness"); const updatedState = harness.state.operation?.state; if (updatedState?.kind !== "run") throw new Error("missing updated run state"); expect(updatedState.inbox.writes).toEqual([pendingId]); @@ -947,10 +947,10 @@ describe("runtime2 AgentHarness", () => { it("faults queued lane work before releasing a failed mutation", async () => { const { session, storage } = await createFailingSession(); const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("missing runtime2 harness"); + if (!(harness instanceof Harness)) throw new Error("missing runtime harness"); unwrap(await harness.createLane("worker", null, BACKGROUND_CONTEXT)); const worker = harness.lanesByName.get("worker"); - if (worker === undefined) throw new Error("missing runtime2 worker lane"); + if (worker === undefined) throw new Error("missing runtime worker lane"); const initialConfiguration = worker.state.configuration; const faultEvent = new Promise((resolve) => harness.events.on("fault", resolve)); const configUpdate = vi.fn(); diff --git a/packages/agent/test/harness/runtime2/lane.test.ts b/packages/agent/test/harness/runtime/lane.test.ts similarity index 97% rename from packages/agent/test/harness/runtime2/lane.test.ts rename to packages/agent/test/harness/runtime/lane.test.ts index f8357e659..92d9cb851 100644 --- a/packages/agent/test/harness/runtime2/lane.test.ts +++ b/packages/agent/test/harness/runtime/lane.test.ts @@ -6,9 +6,9 @@ import { BACKGROUND_CONTEXT, type Context } from "../../../src/harness/context.t import { HarnessEventBus } from "../../../src/harness/events.ts"; import { HookRegistry } from "../../../src/harness/hooks.ts"; import { convertToLlm } from "../../../src/harness/messages.ts"; -import { Lane } from "../../../src/harness/runtime2/lane.ts"; -import { restoreLane } from "../../../src/harness/runtime2/restore.ts"; -import { Drive } from "../../../src/harness/runtime2/types.ts"; +import { Lane } from "../../../src/harness/runtime/lane.ts"; +import { restoreLane } from "../../../src/harness/runtime/restore.ts"; +import { Drive } from "../../../src/harness/runtime/types.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; import type { LaneConfiguration, Session } from "../../../src/harness/session/types.ts"; import * as storedValues from "../../../src/harness/session/values.ts"; @@ -31,7 +31,7 @@ async function createLane( }> { const storage = new ControlledMemoryStorage(); const session = new StorageBackedSession( - { id: `runtime2-lane-${sessions.length}`, createdAt: 1, storageVersion: 1 }, + { id: `runtime-lane-${sessions.length}`, createdAt: 1, storageVersion: 1 }, storage, ); sessions.push(session); @@ -103,7 +103,7 @@ afterEach(async () => { for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 Lane commands", () => { +describe("runtime Lane commands", () => { it("reads and replaces configuration from owned state", async () => { const { lane, model, session } = await createLane(); const activeToolNames = ["read"]; diff --git a/packages/agent/test/harness/runtime2/progress.test.ts b/packages/agent/test/harness/runtime/progress.test.ts similarity index 98% rename from packages/agent/test/harness/runtime2/progress.test.ts rename to packages/agent/test/harness/runtime/progress.test.ts index 1a98dc10a..b25f68c97 100644 --- a/packages/agent/test/harness/runtime2/progress.test.ts +++ b/packages/agent/test/harness/runtime/progress.test.ts @@ -4,9 +4,9 @@ import type { WatchHandle } from "../../../src/harness/agent-harness.ts"; import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT, type Context, withAbortSignal } from "../../../src/harness/context.ts"; import { HookRegistry } from "../../../src/harness/hooks.ts"; -import { Lane } from "../../../src/harness/runtime2/lane.ts"; -import { openFrameProgress, openToolProgress } from "../../../src/harness/runtime2/progress.ts"; -import { type Config, Drive, type LaneState as RuntimeLaneState } from "../../../src/harness/runtime2/types.ts"; +import { Lane } from "../../../src/harness/runtime/lane.ts"; +import { openFrameProgress, openToolProgress } from "../../../src/harness/runtime/progress.ts"; +import { type Config, Drive, type LaneState as RuntimeLaneState } from "../../../src/harness/runtime/types.ts"; import { MemoryStorage } from "../../../src/harness/session/memory.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; import type { RunState, Session, Write } from "../../../src/harness/session/types.ts"; @@ -131,7 +131,7 @@ afterEach(async () => { for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 active drive ownership", () => { +describe("runtime active drive ownership", () => { it("settles or rejects the shared completion without exposing owner controls through the gate", async () => { const settled = new Drive({ operationId }, BACKGROUND_CONTEXT); const outcome = { kind: "waiting", operationId, reason: "retry", notBefore: 10 } as const; @@ -166,7 +166,7 @@ describe("runtime2 active drive ownership", () => { }); }); -describe("runtime2 progress channels", () => { +describe("runtime progress channels", () => { it("enqueues assistant frames in order, seals admission, and exposes the settlement delete", async () => { const responseEntryId = "response"; const { lane, drive } = await createFixture( diff --git a/packages/agent/test/harness/runtime2/restore.test.ts b/packages/agent/test/harness/runtime/restore.test.ts similarity index 99% rename from packages/agent/test/harness/runtime2/restore.test.ts rename to packages/agent/test/harness/runtime/restore.test.ts index 0e1b9a658..ef6ac78e0 100644 --- a/packages/agent/test/harness/runtime2/restore.test.ts +++ b/packages/agent/test/harness/runtime/restore.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; -import { restoreLane, restoreSession } from "../../../src/harness/runtime2/restore.ts"; +import { restoreLane, restoreSession } from "../../../src/harness/runtime/restore.ts"; import { MemorySessionRepo, MemoryStorage } from "../../../src/harness/session/memory.ts"; import type { LaneConfiguration, OperationMeta, RunState, Session } from "../../../src/harness/session/types.ts"; import * as storedValues from "../../../src/harness/session/values.ts"; @@ -50,7 +50,7 @@ afterEach(async () => { for (const repo of repos.splice(0)) await repo.close(BACKGROUND_CONTEXT); }); -describe("runtime2 lane restore", () => { +describe("runtime lane restore", () => { it("restores an idle lane and its latest terminal result", async () => { const session = await createSession(); const result = { diff --git a/packages/agent/test/harness/runtime2/test-utils.ts b/packages/agent/test/harness/runtime/test-utils.ts similarity index 100% rename from packages/agent/test/harness/runtime2/test-utils.ts rename to packages/agent/test/harness/runtime/test-utils.ts diff --git a/packages/agent/test/harness/runtime2/watch.test.ts b/packages/agent/test/harness/runtime/watch.test.ts similarity index 99% rename from packages/agent/test/harness/runtime2/watch.test.ts rename to packages/agent/test/harness/runtime/watch.test.ts index fbffc3f65..8e795b9a4 100644 --- a/packages/agent/test/harness/runtime2/watch.test.ts +++ b/packages/agent/test/harness/runtime/watch.test.ts @@ -3,7 +3,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { type HarnessEvent, HarnessFault, type WatchHandle } from "../../../src/harness/agent-harness.ts"; import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT, type Context, createContextKey, withContextValue } from "../../../src/harness/context.ts"; -import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts"; +import { createAgentHarness, Harness } from "../../../src/harness/runtime/harness.ts"; import * as sessionWrites from "../../../src/harness/session/commit.ts"; import { MemoryStorage } from "../../../src/harness/session/memory.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; @@ -75,7 +75,7 @@ async function attach(session: Session): Promise> { const models = createModels(); models.setProvider(provider.provider); const { harness } = await createAgentHarness({ session, models, model: provider.getModel() }, BACKGROUND_CONTEXT); - if (!(harness instanceof Harness)) throw new Error("Expected runtime2 Harness"); + if (!(harness instanceof Harness)) throw new Error("Expected runtime Harness"); return harness; } @@ -97,7 +97,7 @@ afterEach(async () => { for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); -describe("runtime2 lane watch", () => { +describe("runtime lane watch", () => { it("captures a compaction-bounded transcript and isolates the returned snapshot", async () => { const session = await createSession(); await commit(session, [ diff --git a/packages/ai/README.md b/packages/ai/README.md index b7a00567c..050ef00f8 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -652,6 +652,10 @@ for await (const event of s) { ### Complete Event Reference +Successful generation follows `start → updates* → done`. A failure after generation starts follows `start → updates* → error`. Request setup may fail before generation starts, in which case the stream contains only `error`; `done` and update events are invalid before `start`. + +Every non-terminal event's `partial` is the shared live response-so-far helper. It is intentionally not an event-time snapshot: providers may mutate the same message and content blocks as generation advances, including while older events wait in the stream queue. Inspect it when handling an event instead of retaining it as historical state. Text and ordinary thinking blocks are empty when their `*_start` event is emitted and grow only through matching `*_delta` events until the authoritative `*_end`; redacted thinking may be complete at start and emit no deltas. A streaming tool call starts with empty arguments and emits its full raw JSON through `toolcall_delta`. A provider that starts with complete arguments must emit a cumulative delta prefix that parses to those arguments at an event boundary before later argument deltas. + All streaming events emitted during assistant message generation: | Event Type | Description | Key Properties | @@ -673,22 +677,23 @@ Streaming events for different content blocks are not guaranteed to be contiguou ### Compact Assistant Message Frames -`assistantMessageEventToFrame()` converts stream events to compact, persistable `AssistantMessageFrame` values. It snapshots only public message/block fields, omits the growing `partial` message and provider scratch fields from later frames, and returns `undefined` for terminal `done` and `error` events because final message settlement is separate. End frames retain authoritative completed text, thinking, or tool arguments once; this permits provider end events to correct or supply content without repeated full-partial snapshots. +`AssistantMessageFrameEncoder` converts one stream into compact, persistable `AssistantMessageFrame` values. Create one encoder per stream and feed it every event in order. The encoder understands that `partial` is live: a block-start event consumed after the provider has already queued later deltas snapshots the current block once, and covered queued text/thinking deltas produce no duplicate frame. It retains only per-open-block counters plus, temporarily, the raw prefix needed to synchronize an already-advanced tool call. It never clones the growing full partial per token. -`reduceAssistantMessageFrames()` is the canonical pure reducer for these frames. It reconstructs text, thinking, and tool-call arguments, including interleaved blocks identified by `contentIndex`, and rejects malformed block sequences. It returns `undefined` when there is no start frame to reduce. For an unfinished tool call, the reducer accumulates its JSON deltas and invokes `parseStreamingJson` once after consuming the iterable; a `toolcall_end` frame instead supplies authoritative final arguments. The reducer does not validate arguments against a tool's TypeBox schema. Call `validateToolCall` before execution where schema validation is required. +The start frame contains message metadata with empty content. Text and thinking frames store each generated character at most once before the authoritative end frame. Tool calls that were already advanced when their start event was consumed use one compact JSON checkpoint before ordinary deltas resume. Terminal `done` and `error` events produce no frame because final message settlement is separate. A pre-generation `error` therefore produces no frames. -Deltas reconstruct the latest unfinished block. An end frame replaces that block with the provider's authoritative completed content and metadata. Persist every frame in order; terminal `done`/`error` message settlement remains separate. +`reduceAssistantMessageFrames()` is the canonical pure reducer. It reconstructs text, thinking, and tool-call arguments, including interleaved blocks identified by `contentIndex`, and rejects malformed sequences. It performs a single pass over the iterable and returns `undefined` when there is no start frame. End frames replace blocks with the provider's authoritative completed content and metadata. The reducer does not validate tool arguments against a TypeBox schema; call `validateToolCall` before execution. ```typescript import { - assistantMessageEventToFrame, + AssistantMessageFrameEncoder, reduceAssistantMessageFrames, type AssistantMessageFrame, } from '@earendil-works/pi-ai'; +const encoder = new AssistantMessageFrameEncoder(); const frames: AssistantMessageFrame[] = []; for await (const event of s) { - const frame = assistantMessageEventToFrame(event); + const frame = encoder.encode(event); if (frame) frames.push(frame); } @@ -696,6 +701,8 @@ const reconstructedPartial = reduceAssistantMessageFrames(frames); const finalMessage = await s.result(); // Persist terminal settlement separately. ``` +An encoder rejects duplicate starts, updates before start, `done` before start, events after a terminal event, duplicate block starts, and block-kind mismatches. An `error` before start is valid and returns no frame. + ## Image Input Models with vision capabilities can process images. You can check if a model supports images via the `input` property. If you pass images to a non-vision model, they are silently ignored. @@ -920,7 +927,7 @@ Every `AssistantMessage` includes a `stopReason` field that indicates how the ge ## Error Handling -Request failures never throw out of the stream functions: when a request ends with an error (including aborts and tool call validation errors), the streaming API emits an error event and the final message carries the details: +Request failures never throw out of the stream functions: when a request ends with an error (including aborts and tool call validation errors), the streaming API emits an error event and the final message carries the details. Setup failures may emit `error` without `start`; failures after generation begins emit `start`, any observed updates, then `error`: ```typescript // In streaming diff --git a/packages/ai/src/api/anthropic-messages.ts b/packages/ai/src/api/anthropic-messages.ts index 6c4ff62d0..f224c812f 100644 --- a/packages/ai/src/api/anthropic-messages.ts +++ b/packages/ai/src/api/anthropic-messages.ts @@ -827,8 +827,6 @@ export const streamSimple: StreamFunction<"anthropic-messages", SimpleStreamOpti context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - assertRequestAuth(model.provider, options?.apiKey, options?.headers); - const base = { ...buildBaseOptions(model, context, options, options?.apiKey), toolChoice: options?.toolChoice, diff --git a/packages/ai/src/api/azure-openai-responses.ts b/packages/ai/src/api/azure-openai-responses.ts index 56578805d..e0011696a 100644 --- a/packages/ai/src/api/azure-openai-responses.ts +++ b/packages/ai/src/api/azure-openai-responses.ts @@ -165,10 +165,7 @@ export const streamSimple: StreamFunction<"azure-openai-responses", SimpleStream context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey; - if (!apiKey) { - throw new Error(`No API key for provider: ${model.provider}`); - } + const apiKey = options?.apiKey ?? ""; const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/google-generative-ai.ts b/packages/ai/src/api/google-generative-ai.ts index a0f9f39c9..755bbb2d4 100644 --- a/packages/ai/src/api/google-generative-ai.ts +++ b/packages/ai/src/api/google-generative-ai.ts @@ -299,10 +299,7 @@ export const streamSimple: StreamFunction<"google-generative-ai", SimpleStreamOp context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey; - if (!apiKey) { - throw new Error(`No API key for provider: ${model.provider}`); - } + const apiKey = options?.apiKey ?? ""; const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/mistral-conversations.ts b/packages/ai/src/api/mistral-conversations.ts index 64bb10815..b9d291f8e 100644 --- a/packages/ai/src/api/mistral-conversations.ts +++ b/packages/ai/src/api/mistral-conversations.ts @@ -183,10 +183,7 @@ export const streamSimple: StreamFunction<"mistral-conversations", SimpleStreamO context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey; - if (!apiKey) { - throw new Error(`No API key for provider: ${model.provider}`); - } + const apiKey = options?.apiKey ?? ""; const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/openai-codex-responses.ts b/packages/ai/src/api/openai-codex-responses.ts index 71a917d67..8ee5b0271 100644 --- a/packages/ai/src/api/openai-codex-responses.ts +++ b/packages/ai/src/api/openai-codex-responses.ts @@ -494,10 +494,7 @@ export const streamSimple: StreamFunction<"openai-codex-responses", SimpleStream context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey; - if (!apiKey) { - throw new Error(`No API key for provider: ${model.provider}`); - } + const apiKey = options?.apiKey ?? ""; const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/openai-completions.ts b/packages/ai/src/api/openai-completions.ts index 08b181f23..4b6eb34c7 100644 --- a/packages/ai/src/api/openai-completions.ts +++ b/packages/ai/src/api/openai-completions.ts @@ -464,7 +464,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio type: "toolCall", id: toolCall.id || "", name, - arguments: hasCustomInput ? { [customInputProperty]: "" } : {}, + arguments: {}, partialArgs: hasCustomInput ? undefined : "", customInput: hasCustomInput ? { property: customInputProperty, jsonBuffer: { input: "", started: false, closed: false } } @@ -496,7 +496,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio } if (toolCall.custom && !toolCall.function && !block.customInput) { const customInputProperty = grammarToolInputProperties.get(block.name) ?? "input"; - block.arguments = { [customInputProperty]: "" }; + block.arguments = {}; block.customInput = { property: customInputProperty, jsonBuffer: { input: "", started: false, closed: false }, @@ -685,8 +685,6 @@ export const streamSimple: StreamFunction<"openai-completions", SimpleStreamOpti context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - getClientApiKey(model.provider, options?.apiKey, options?.headers); - const base = { ...buildBaseOptions(model, context, options, options?.apiKey), toolChoice: options?.toolChoice, diff --git a/packages/ai/src/api/openai-responses-shared.ts b/packages/ai/src/api/openai-responses-shared.ts index dca994a6f..b64dde563 100644 --- a/packages/ai/src/api/openai-responses-shared.ts +++ b/packages/ai/src/api/openai-responses-shared.ts @@ -503,12 +503,11 @@ export async function processResponsesStream( } if (item.type === "custom_tool_call") { const inputProperty = options?.grammarToolInputProperties?.get(item.name) ?? "input"; - const input = item.input || ""; const block: StreamingToolCall = { type: "toolCall", id: `${item.call_id}|${item.id}`, name: item.name, - arguments: { [inputProperty]: input }, + arguments: {}, ...(item.namespace !== undefined ? { namespace: item.namespace } : {}), customInput: { property: inputProperty, @@ -523,6 +522,9 @@ export async function processResponsesStream( } satisfies ResponsesOutputSlot; outputSlots.set(outputIndex, slot); stream.push({ type: "toolcall_start", contentIndex: slot.contentIndex, partial: output }); + if (item.input) { + pushToolCallDelta(slot, appendCustomToolCallInput(block, item.input, false)); + } return slot; } return undefined; diff --git a/packages/ai/src/api/openai-responses.ts b/packages/ai/src/api/openai-responses.ts index 6bfb3d65b..074b159aa 100644 --- a/packages/ai/src/api/openai-responses.ts +++ b/packages/ai/src/api/openai-responses.ts @@ -200,8 +200,6 @@ export const streamSimple: StreamFunction<"openai-responses", SimpleStreamOption context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - getClientApiKey(model.provider, options?.apiKey, options?.headers); - const base = { ...buildBaseOptions(model, context, options, options?.apiKey), toolChoice: options?.toolChoice, diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index d35a48b84..1698d1489 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -527,10 +527,19 @@ export interface Context { /** * Event protocol for AssistantMessageEventStream. * - * Streams should emit `start` before partial updates, then terminate with either: - * - `done` carrying the final successful AssistantMessage, or - * - `error` carrying the final AssistantMessage with stopReason "error" or "aborted" - * and errorMessage. + * Successful streams emit `start` before partial updates and terminate with + * `done`. A stream may terminate directly with `error` when request setup fails + * before generation starts; after `start`, failures also terminate with `error`. + * Updates and `done` must never appear before `start`. + * + * `partial` is the shared live response-so-far helper, not an event-time + * snapshot. Text and thinking blocks are empty when their `*_start` event is + * emitted and grow only through their corresponding `*_delta` events until the + * authoritative `*_end`. Redacted thinking may be complete at start and emit no + * deltas. A streaming tool call starts with empty arguments and emits its full + * raw JSON through `toolcall_delta`; a provider that starts with complete + * arguments must emit a cumulative delta prefix that parses to those arguments + * before emitting any later argument delta. */ export type AssistantMessageEvent = | { type: "start"; partial: AssistantMessage } diff --git a/packages/ai/src/utils/assistant-message-frame.ts b/packages/ai/src/utils/assistant-message-frame.ts index f51489cd7..3c22f78e3 100644 --- a/packages/ai/src/utils/assistant-message-frame.ts +++ b/packages/ai/src/utils/assistant-message-frame.ts @@ -2,8 +2,8 @@ import type { AssistantMessage, AssistantMessageEvent, TextContent, ThinkingCont import { parseStreamingJson } from "./json-parse.ts"; /** - * Compact, replayable representation of an assistant-message streaming event. - * Terminal settlement is intentionally excluded and must be persisted separately. + * Compact, replayable assistant-message progress. Terminal settlement is + * intentionally excluded and must be persisted separately. */ export type AssistantMessageFrame = | { type: "start"; partial: AssistantMessage } @@ -20,6 +20,7 @@ export type AssistantMessageFrame = redacted?: boolean; } | { type: "toolcall_start"; contentIndex: number; toolCall: ToolCall } + | { type: "toolcall_checkpoint"; contentIndex: number; json: string } | { type: "toolcall_delta"; contentIndex: number; delta: string } | { type: "toolcall_end"; @@ -31,7 +32,16 @@ export type AssistantMessageFrame = namespace?: string; }; -type BlockState = +type EncoderBlockState = + | { kind: "text" | "thinking"; coveredChars: number; deltaChars: number } + | { + kind: "toolCall"; + caughtUp: boolean; + catchupJson: string; + snapshotArguments: string; + }; + +type ReducerBlockState = | { kind: "text"; ended: boolean } | { kind: "thinking"; ended: boolean } | { kind: "toolCall"; ended: boolean; json: string }; @@ -64,21 +74,10 @@ function cloneToolCall(toolCall: ToolCall): ToolCall { }; } -function cloneContentBlock(content: AssistantMessage["content"][number]): AssistantMessage["content"][number] { - switch (content.type) { - case "text": - return cloneTextContent(content); - case "thinking": - return cloneThinkingContent(content); - case "toolCall": - return cloneToolCall(content); - } -} - -function cloneAssistantMessage(message: AssistantMessage): AssistantMessage { +function cloneStartMessage(message: AssistantMessage): AssistantMessage { return { role: "assistant", - content: message.content.map(cloneContentBlock), + content: [], api: message.api, provider: message.provider, model: message.model, @@ -86,11 +85,7 @@ function cloneAssistantMessage(message: AssistantMessage): AssistantMessage { ...(message.responseId === undefined ? {} : { responseId: message.responseId }), ...(message.diagnostics === undefined ? {} : { diagnostics: structuredClone(message.diagnostics) }), usage: structuredClone(message.usage), - stopReason: message.stopReason, - ...(message.deferred === undefined ? {} : { deferred: structuredClone(message.deferred) }), - ...(message.errorMessage === undefined ? {} : { errorMessage: message.errorMessage }), - ...(message.rawStopReason === undefined ? {} : { rawStopReason: message.rawStopReason }), - ...(message.endTurn === undefined ? {} : { endTurn: message.endTurn }), + stopReason: "pending", timestamp: message.timestamp, }; } @@ -110,99 +105,212 @@ function eventBlock(event: Exclude("")); + +/** + * Encodes one assistant stream. `partial` remains a shared live accumulator; + * the encoder uses per-block offsets to avoid replaying deltas already visible + * when an older queued event is consumed. + */ +export class AssistantMessageFrameEncoder { + private started = false; + private terminal = false; + private readonly blocks = new Map(); + + encode(event: AssistantMessageEvent): AssistantMessageFrame | undefined { + if (this.terminal) throw new Error(`Assistant message event ${event.type} follows a terminal event`); + + switch (event.type) { + case "start": + if (this.started) throw new Error("Assistant message stream contains more than one start event"); + this.started = true; + return { type: "start", partial: cloneStartMessage(event.partial) }; + case "done": + if (!this.started) throw new Error("Assistant message done event appears before start"); + this.terminal = true; + return undefined; + case "error": + this.terminal = true; + return undefined; } - case "text_delta": - return { type: "text_delta", contentIndex: event.contentIndex, delta: event.delta }; - case "text_end": { - const content = eventBlock(event); - if (content.type !== "text") { - throw new Error(`text_end event points to ${content.type} block at index ${event.contentIndex}`); + + if (!this.started) throw new Error(`Assistant message ${event.type} event appears before start`); + + switch (event.type) { + case "text_start": { + const content = eventBlock(event); + if (content.type !== "text") { + throw new Error(`text_start event points to ${content.type} block at index ${event.contentIndex}`); + } + this.startBlock(event.contentIndex, { + kind: "text", + coveredChars: content.text.length, + deltaChars: 0, + }); + return { type: "text_start", contentIndex: event.contentIndex, content: cloneTextContent(content) }; + } + case "text_delta": + return this.encodeTextDelta(event.contentIndex, event.delta, "text"); + case "text_end": { + const content = eventBlock(event); + if (content.type !== "text") { + throw new Error(`text_end event points to ${content.type} block at index ${event.contentIndex}`); + } + this.endBlock(event.contentIndex, "text"); + return { + type: "text_end", + contentIndex: event.contentIndex, + content: event.content, + ...(content.textSignature === undefined ? {} : { textSignature: content.textSignature }), + }; + } + case "thinking_start": { + const content = eventBlock(event); + if (content.type !== "thinking") { + throw new Error(`thinking_start event points to ${content.type} block at index ${event.contentIndex}`); + } + this.startBlock(event.contentIndex, { + kind: "thinking", + coveredChars: content.thinking.length, + deltaChars: 0, + }); + return { + type: "thinking_start", + contentIndex: event.contentIndex, + content: cloneThinkingContent(content), + }; + } + case "thinking_delta": + return this.encodeTextDelta(event.contentIndex, event.delta, "thinking"); + case "thinking_end": { + const content = eventBlock(event); + if (content.type !== "thinking") { + throw new Error(`thinking_end event points to ${content.type} block at index ${event.contentIndex}`); + } + this.endBlock(event.contentIndex, "thinking"); + return { + type: "thinking_end", + contentIndex: event.contentIndex, + content: event.content, + ...(content.thinkingSignature === undefined ? {} : { thinkingSignature: content.thinkingSignature }), + ...(content.redacted === undefined ? {} : { redacted: content.redacted }), + }; + } + case "toolcall_start": { + const content = eventBlock(event); + if (content.type !== "toolCall") { + throw new Error(`toolcall_start event points to ${content.type} block at index ${event.contentIndex}`); + } + const snapshotArguments = serializedArguments(content.arguments); + const caughtUp = snapshotArguments === EMPTY_PARSED_TOOL_ARGUMENTS; + this.startBlock(event.contentIndex, { + kind: "toolCall", + caughtUp, + catchupJson: "", + snapshotArguments: caughtUp ? "" : snapshotArguments, + }); + return { type: "toolcall_start", contentIndex: event.contentIndex, toolCall: cloneToolCall(content) }; + } + case "toolcall_delta": { + const state = this.block(event.contentIndex, "toolCall"); + if (state.kind !== "toolCall") throw new Error("Unreachable tool-call encoder state"); + if (state.caughtUp) { + return event.delta.length === 0 + ? undefined + : { type: "toolcall_delta", contentIndex: event.contentIndex, delta: event.delta }; + } + state.catchupJson += event.delta; + if ( + serializedArguments(parseStreamingJson(state.catchupJson)) !== + state.snapshotArguments + ) { + return undefined; + } + state.caughtUp = true; + state.snapshotArguments = ""; + const json = state.catchupJson; + state.catchupJson = ""; + return json.length === 0 + ? undefined + : { type: "toolcall_checkpoint", contentIndex: event.contentIndex, json }; + } + case "toolcall_end": { + const content = eventBlock(event); + if (content.type !== "toolCall") { + throw new Error(`toolcall_end event points to ${content.type} block at index ${event.contentIndex}`); + } + if (event.toolCall.type !== "toolCall") { + throw new Error(`toolcall_end event has invalid tool call at index ${event.contentIndex}`); + } + this.endBlock(event.contentIndex, "toolCall"); + return { + type: "toolcall_end", + contentIndex: event.contentIndex, + id: event.toolCall.id, + name: event.toolCall.name, + arguments: structuredClone(event.toolCall.arguments), + ...(event.toolCall.thoughtSignature === undefined + ? {} + : { thoughtSignature: event.toolCall.thoughtSignature }), + ...(event.toolCall.namespace === undefined ? {} : { namespace: event.toolCall.namespace }), + }; } - return { - type: "text_end", - contentIndex: event.contentIndex, - content: event.content, - ...(content.textSignature === undefined ? {} : { textSignature: content.textSignature }), - }; } - case "thinking_start": { - const content = eventBlock(event); - if (content.type !== "thinking") { - throw new Error(`thinking_start event points to ${content.type} block at index ${event.contentIndex}`); - } - return { - type: "thinking_start", - contentIndex: event.contentIndex, - content: cloneThinkingContent(content), - }; + } + + private startBlock(contentIndex: number, state: EncoderBlockState): void { + assertContentIndex(contentIndex); + if (this.blocks.has(contentIndex)) { + throw new Error(`Assistant message block ${contentIndex} starts more than once`); } - case "thinking_delta": - return { type: "thinking_delta", contentIndex: event.contentIndex, delta: event.delta }; - case "thinking_end": { - const content = eventBlock(event); - if (content.type !== "thinking") { - throw new Error(`thinking_end event points to ${content.type} block at index ${event.contentIndex}`); - } - return { - type: "thinking_end", - contentIndex: event.contentIndex, - content: event.content, - ...(content.thinkingSignature === undefined ? {} : { thinkingSignature: content.thinkingSignature }), - ...(content.redacted === undefined ? {} : { redacted: content.redacted }), - }; + this.blocks.set(contentIndex, state); + } + + private block(contentIndex: number, kind: EncoderBlockState["kind"]): EncoderBlockState { + assertContentIndex(contentIndex); + const state = this.blocks.get(contentIndex); + if (state === undefined) throw new Error(`Assistant message ${kind} block ${contentIndex} has not started`); + if (state.kind !== kind) { + throw new Error(`Assistant message block ${contentIndex} is ${state.kind}, not ${kind}`); } - case "toolcall_start": { - const content = eventBlock(event); - if (content.type !== "toolCall") { - throw new Error(`toolcall_start event points to ${content.type} block at index ${event.contentIndex}`); - } - return { type: "toolcall_start", contentIndex: event.contentIndex, toolCall: cloneToolCall(content) }; - } - case "toolcall_delta": - return { type: "toolcall_delta", contentIndex: event.contentIndex, delta: event.delta }; - case "toolcall_end": { - const content = eventBlock(event); - if (content.type !== "toolCall") { - throw new Error(`toolcall_end event points to ${content.type} block at index ${event.contentIndex}`); - } - if (event.toolCall.type !== "toolCall") { - throw new Error(`toolcall_end event has invalid tool call at index ${event.contentIndex}`); - } - return { - type: "toolcall_end", - contentIndex: event.contentIndex, - id: event.toolCall.id, - name: event.toolCall.name, - arguments: structuredClone(event.toolCall.arguments), - ...(event.toolCall.thoughtSignature === undefined - ? {} - : { thoughtSignature: event.toolCall.thoughtSignature }), - ...(event.toolCall.namespace === undefined ? {} : { namespace: event.toolCall.namespace }), - }; - } - case "done": - case "error": - return undefined; + return state; + } + + private endBlock(contentIndex: number, kind: EncoderBlockState["kind"]): void { + this.block(contentIndex, kind); + this.blocks.delete(contentIndex); + } + + private encodeTextDelta( + contentIndex: number, + delta: string, + kind: "text" | "thinking", + ): AssistantMessageFrame | undefined { + const state = this.block(contentIndex, kind); + if (state.kind === "toolCall") throw new Error("Unreachable text encoder state"); + const deltaStart = state.deltaChars; + state.deltaChars += delta.length; + const covered = Math.max(0, state.coveredChars - deltaStart); + if (covered >= delta.length) return undefined; + const uncovered = covered === 0 ? delta : delta.slice(covered); + return kind === "text" + ? { type: "text_delta", contentIndex, delta: uncovered } + : { type: "thinking_delta", contentIndex, delta: uncovered }; } } function appendBlock( message: AssistantMessage, - states: Map, + states: Map, contentIndex: number, block: TextContent | ThinkingContent | ToolCall, - state: BlockState, + state: ReducerBlockState, ): void { assertContentIndex(contentIndex); if (contentIndex !== message.content.length) { @@ -215,11 +323,11 @@ function appendBlock( function activeBlock( message: AssistantMessage, - states: Map, + states: Map, contentIndex: number, - expectedKind: BlockState["kind"], + expectedKind: ReducerBlockState["kind"], frameType: AssistantMessageFrame["type"], -): { block: TextContent | ThinkingContent | ToolCall; state: BlockState } { +): { block: TextContent | ThinkingContent | ToolCall; state: ReducerBlockState } { assertContentIndex(contentIndex); const state = states.get(contentIndex); const block = message.content[contentIndex]; @@ -238,27 +346,25 @@ function activeBlock( } /** - * Replay compact frames into an assistant message without mutating the frames. - * Returns undefined when the iterable has no start frame. Frames for different - * content indexes may be interleaved, but every block must have a valid start sequence. + * Replay compact frames without mutating them. Returns `undefined` when the + * iterable contains no start frame. */ export function reduceAssistantMessageFrames(frames: Iterable): AssistantMessage | undefined { - const replayFrames = [...frames]; - if (!replayFrames.some((frame) => frame.type === "start")) return undefined; - let message: AssistantMessage | undefined; - const states = new Map(); + let frameBeforeStart: AssistantMessageFrame["type"] | undefined; + const states = new Map(); - for (const frame of replayFrames) { + for (const frame of frames) { if (frame.type === "start") { - if (message) { - throw new Error("Assistant message frame sequence contains more than one start frame"); - } + if (message) throw new Error("Assistant message frame sequence contains more than one start frame"); + if (frameBeforeStart !== undefined) + throw new Error(`${frameBeforeStart} frame appears before the start frame`); message = structuredClone(frame.partial); continue; } if (!message) { - throw new Error(`${frame.type} frame appears before the start frame`); + frameBeforeStart ??= frame.type; + continue; } switch (frame.type) { @@ -319,6 +425,15 @@ export function reduceAssistantMessageFrames(frames: Iterable(frame.json); + break; + } case "toolcall_delta": { const { block, state } = activeBlock(message, states, frame.contentIndex, "toolCall", frame.type); if (block.type !== "toolCall" || state.kind !== "toolCall") { diff --git a/packages/ai/test/assistant-message-frame.test.ts b/packages/ai/test/assistant-message-frame.test.ts index fd027c5e6..c8fa8c306 100644 --- a/packages/ai/test/assistant-message-frame.test.ts +++ b/packages/ai/test/assistant-message-frame.test.ts @@ -5,7 +5,7 @@ import { type AssistantMessage, type AssistantMessageEvent, type AssistantMessageFrame, - assistantMessageEventToFrame, + AssistantMessageFrameEncoder, type Model, reduceAssistantMessageFrames, } from "../src/index.ts"; @@ -31,8 +31,8 @@ function seed(): AssistantMessage { }; } -function frame(event: AssistantMessageEvent): AssistantMessageFrame { - const converted = assistantMessageEventToFrame(event); +function frame(encoder: AssistantMessageFrameEncoder, event: AssistantMessageEvent): AssistantMessageFrame { + const converted = encoder.encode(event); if (!converted) throw new Error(`Expected ${event.type} event to produce a frame`); return converted; } @@ -40,13 +40,14 @@ function frame(event: AssistantMessageEvent): AssistantMessageFrame { describe("assistant message frames", () => { it("uses authoritative text end content and signature", () => { const partial = seed(); - const frames: AssistantMessageFrame[] = [frame({ type: "start", partial })]; + const encoder = new AssistantMessageFrameEncoder(); + const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial })]; partial.content.push({ type: "text", text: "Hello " }); - frames.push(frame({ type: "text_start", contentIndex: 0, partial })); + frames.push(frame(encoder, { type: "text_start", contentIndex: 0, partial })); partial.content[0] = { type: "text", text: "Hello world", textSignature: "sig-text" }; frames.push( - frame({ type: "text_delta", contentIndex: 0, delta: "incorrect", partial }), - frame({ type: "text_end", contentIndex: 0, content: "Hello world", partial }), + frame(encoder, { type: "text_delta", contentIndex: 0, delta: "incorrect", partial }), + frame(encoder, { type: "text_end", contentIndex: 0, content: "Hello world", partial }), ); expect(frames.at(-1)).toEqual({ @@ -62,21 +63,22 @@ describe("assistant message frames", () => { it("preserves initial and final thinking metadata, including redaction", () => { const partial = seed(); - const frames: AssistantMessageFrame[] = [frame({ type: "start", partial })]; + const encoder = new AssistantMessageFrameEncoder(); + const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial })]; partial.content.push({ type: "thinking", thinking: "[redacted]", thinkingSignature: "encrypted-start", redacted: true, }); - frames.push(frame({ type: "thinking_start", contentIndex: 0, partial })); + frames.push(frame(encoder, { type: "thinking_start", contentIndex: 0, partial })); partial.content[0] = { type: "thinking", thinking: "[redacted]", thinkingSignature: "encrypted-final", redacted: true, }; - frames.push(frame({ type: "thinking_end", contentIndex: 0, content: "[redacted]", partial })); + frames.push(frame(encoder, { type: "thinking_end", contentIndex: 0, content: "[redacted]", partial })); expect(frames.at(-1)).toEqual({ type: "thinking_end", @@ -201,11 +203,12 @@ describe("assistant message frames", () => { for (const event of events) yield event; } - const frames: AssistantMessageFrame[] = [frame({ type: "start", partial: output })]; + const encoder = new AssistantMessageFrameEncoder(); + const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial: output })]; const stream = new AssistantMessageEventStream(); const push = stream.push.bind(stream); stream.push = (event) => { - const converted = assistantMessageEventToFrame(event); + const converted = encoder.encode(event); if (converted) frames.push(converted); push(event); }; @@ -214,6 +217,123 @@ describe("assistant message frames", () => { expect(reduceAssistantMessageFrames(frames)?.content).toEqual(output.content); }); + it("reconciles queued text events against one advanced live partial without duplicate content", () => { + const partial = seed(); + const events: AssistantMessageEvent[] = [{ type: "start", partial }]; + const text = { type: "text" as const, text: "" }; + partial.content.push(text); + events.push({ type: "text_start", contentIndex: 0, partial }); + for (const delta of ["Hel", "lo", " ", "world"]) { + text.text += delta; + events.push({ type: "text_delta", contentIndex: 0, delta, partial }); + } + + const encoder = new AssistantMessageFrameEncoder(); + const frames = events.flatMap((event) => { + const encoded = encoder.encode(event); + return encoded === undefined ? [] : [encoded]; + }); + + expect(frames.map((item) => item.type)).toEqual(["start", "text_start"]); + expect(frames[0]).toMatchObject({ type: "start", partial: { content: [], stopReason: "pending" } }); + expect(reduceAssistantMessageFrames(frames)?.content).toEqual([{ type: "text", text: "Hello world" }]); + }); + + it("trims only the covered prefix when a start snapshot lands inside a delta", () => { + const partial = seed(); + const encoder = new AssistantMessageFrameEncoder(); + const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial })]; + const text = { type: "text" as const, text: "Hel" }; + partial.content.push(text); + frames.push(frame(encoder, { type: "text_start", contentIndex: 0, partial })); + expect(encoder.encode({ type: "text_delta", contentIndex: 0, delta: "He", partial })).toBeUndefined(); + const remainder = encoder.encode({ type: "text_delta", contentIndex: 0, delta: "llo", partial }); + if (remainder === undefined) throw new Error("Expected uncovered text delta"); + frames.push(remainder); + + expect(remainder).toEqual({ type: "text_delta", contentIndex: 0, delta: "lo" }); + expect(reduceAssistantMessageFrames(frames)?.content).toEqual([{ type: "text", text: "Hello" }]); + }); + + it("checkpoints queued tool JSON without replaying covered deltas", () => { + const partial = seed(); + const toolCall = { type: "toolCall" as const, id: "call", name: "write", arguments: {} }; + const events: AssistantMessageEvent[] = [{ type: "start", partial }]; + partial.content.push(toolCall); + events.push({ type: "toolcall_start", contentIndex: 0, partial }); + toolCall.arguments = { path: "README.md" }; + events.push( + { type: "toolcall_delta", contentIndex: 0, delta: '{"path":"READ', partial }, + { type: "toolcall_delta", contentIndex: 0, delta: 'ME.md"}', partial }, + ); + + const encoder = new AssistantMessageFrameEncoder(); + const frames = events.flatMap((event) => { + const encoded = encoder.encode(event); + return encoded === undefined ? [] : [encoded]; + }); + expect(frames.map((item) => item.type)).toEqual(["start", "toolcall_start", "toolcall_checkpoint"]); + expect(frames.at(-1)).toEqual({ + type: "toolcall_checkpoint", + contentIndex: 0, + json: '{"path":"README.md"}', + }); + expect(reduceAssistantMessageFrames(frames)?.content).toEqual([ + { type: "toolCall", id: "call", name: "write", arguments: { path: "README.md" } }, + ]); + }); + + it("streams grammar-tool JSON compactly from an empty argument start", () => { + const partial = seed(); + const encoder = new AssistantMessageFrameEncoder(); + const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial })]; + const toolCall = { type: "toolCall" as const, id: "call", name: "bash", arguments: {} }; + partial.content.push(toolCall); + frames.push(frame(encoder, { type: "toolcall_start", contentIndex: 0, partial })); + toolCall.arguments = { command: "ls -la /tmp" }; + frames.push( + frame(encoder, { + type: "toolcall_delta", + contentIndex: 0, + delta: '{"command":"ls -la /tmp"}', + partial, + }), + ); + + expect(frames.at(-1)).toEqual({ + type: "toolcall_delta", + contentIndex: 0, + delta: '{"command":"ls -la /tmp"}', + }); + expect(reduceAssistantMessageFrames(frames)?.content[0]).toMatchObject({ + type: "toolCall", + arguments: { command: "ls -la /tmp" }, + }); + }); + + it("accepts a pre-generation error but rejects success or updates before start", () => { + const failed = seed(); + failed.stopReason = "error"; + failed.errorMessage = "setup failed"; + expect( + new AssistantMessageFrameEncoder().encode({ type: "error", reason: "error", error: failed }), + ).toBeUndefined(); + + const completed = seed(); + completed.stopReason = "stop"; + expect(() => + new AssistantMessageFrameEncoder().encode({ type: "done", reason: "stop", message: completed }), + ).toThrow("done event appears before start"); + expect(() => + new AssistantMessageFrameEncoder().encode({ + type: "text_delta", + contentIndex: 0, + delta: "x", + partial: seed(), + }), + ).toThrow("text_delta event appears before start"); + }); + it("treats end signature metadata, including absence, as authoritative", () => { const frames: AssistantMessageFrame[] = [ { type: "start", partial: seed() }, @@ -268,7 +388,10 @@ describe("assistant message frames", () => { }; partial.content.push(toolCall); - const end = frame({ type: "toolcall_end", contentIndex: 0, toolCall, partial }); + const encoder = new AssistantMessageFrameEncoder(); + frame(encoder, { type: "start", partial }); + frame(encoder, { type: "toolcall_start", contentIndex: 0, partial }); + const end = frame(encoder, { type: "toolcall_end", contentIndex: 0, toolCall, partial }); expect(end).toEqual({ type: "toolcall_end", contentIndex: 0, @@ -304,28 +427,13 @@ describe("assistant message frames", () => { const partialWithScratch = partial as AssistantMessage & { outputIndex?: number }; partialWithScratch.outputIndex = 3; - const start = frame({ type: "start", partial }); - const textStart = frame({ type: "text_start", contentIndex: 0, partial }); - const thinkingStart = frame({ type: "thinking_start", contentIndex: 1, partial }); - const toolStart = frame({ type: "toolcall_start", contentIndex: 2, partial }); + const encoder = new AssistantMessageFrameEncoder(); + const start = frame(encoder, { type: "start", partial }); + const textStart = frame(encoder, { type: "text_start", contentIndex: 0, partial }); + const thinkingStart = frame(encoder, { type: "thinking_start", contentIndex: 1, partial }); + const toolStart = frame(encoder, { type: "toolcall_start", contentIndex: 2, partial }); - expect(start.type === "start" && start.partial.content).toEqual([ - { type: "text", text: "visible", textSignature: "text-sig" }, - { - type: "thinking", - thinking: "reasoning", - thinkingSignature: "thinking-sig", - redacted: false, - }, - { - type: "toolCall", - id: "call", - name: "run", - arguments: { value: 1 }, - thoughtSignature: "tool-sig", - namespace: "tools", - }, - ]); + expect(start.type === "start" && start.partial.content).toEqual([]); expect(start).not.toHaveProperty("partial.outputIndex"); expect(textStart).not.toHaveProperty("content.index"); expect(thinkingStart).not.toHaveProperty("content.index"); @@ -361,7 +469,8 @@ describe("assistant message frames", () => { it("snapshots mutable event data and keeps reduction pure", () => { const partial = seed(); partial.diagnostics = [{ type: "test", timestamp: 2, details: { value: "original" } }]; - const start = frame({ type: "start", partial }); + const encoder = new AssistantMessageFrameEncoder(); + const start = frame(encoder, { type: "start", partial }); partial.diagnostics[0]!.details!.value = "mutated"; partial.usage.cost.total = 99; @@ -371,7 +480,7 @@ describe("assistant message frames", () => { name: "run", arguments: { nested: { value: "original" } }, }); - const toolStart = frame({ type: "toolcall_start", contentIndex: 0, partial }); + const toolStart = frame(encoder, { type: "toolcall_start", contentIndex: 0, partial }); const sourceTool = partial.content[0]; if (sourceTool?.type !== "toolCall") throw new Error("Expected source tool call"); (sourceTool.arguments.nested as Record).value = "mutated"; @@ -390,11 +499,15 @@ describe("assistant message frames", () => { it("omits terminal events because settlement is separate", () => { const message = seed(); + const completed = new AssistantMessageFrameEncoder(); + completed.encode({ type: "start", partial: message }); message.stopReason = "stop"; - expect(assistantMessageEventToFrame({ type: "done", reason: "stop", message })).toBeUndefined(); + expect(completed.encode({ type: "done", reason: "stop", message })).toBeUndefined(); message.stopReason = "error"; message.errorMessage = "failed"; - expect(assistantMessageEventToFrame({ type: "error", reason: "error", error: message })).toBeUndefined(); + expect( + new AssistantMessageFrameEncoder().encode({ type: "error", reason: "error", error: message }), + ).toBeUndefined(); }); it("returns undefined when there is no start frame", () => { @@ -438,8 +551,10 @@ describe("assistant message frames", () => { it("rejects conversion events whose contentIndex points to the wrong block kind", () => { const partial = seed(); + const encoder = new AssistantMessageFrameEncoder(); + encoder.encode({ type: "start", partial }); partial.content.push({ type: "thinking", thinking: "" }); - expect(() => assistantMessageEventToFrame({ type: "text_start", contentIndex: 0, partial })).toThrow( + expect(() => encoder.encode({ type: "text_start", contentIndex: 0, partial })).toThrow( "text_start event points to thinking block", ); }); diff --git a/packages/ai/test/pre-generation-error.test.ts b/packages/ai/test/pre-generation-error.test.ts new file mode 100644 index 000000000..d30fde904 --- /dev/null +++ b/packages/ai/test/pre-generation-error.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "vitest"; +import { streamSimple as streamAnthropic } from "../src/api/anthropic-messages.ts"; +import { streamSimple as streamAzure } from "../src/api/azure-openai-responses.ts"; +import { streamSimple as streamGoogle } from "../src/api/google-generative-ai.ts"; +import { streamSimple as streamMistral } from "../src/api/mistral-conversations.ts"; +import { streamSimple as streamCodex } from "../src/api/openai-codex-responses.ts"; +import { streamSimple as streamOpenAICompletions } from "../src/api/openai-completions.ts"; +import { streamSimple as streamOpenAIResponses } from "../src/api/openai-responses.ts"; +import type { Api, AssistantMessageEventStream, Model } from "../src/types.ts"; + +function model(api: TApi): Model { + return { + id: "test-model", + name: "Test", + api, + provider: "test-provider", + baseUrl: "https://example.invalid", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 1_000, + maxTokens: 100, + }; +} + +async function expectPreGenerationError(create: () => AssistantMessageEventStream): Promise { + const stream = create(); + const events = []; + for await (const event of stream) events.push(event); + expect(events.map((event) => event.type)).toEqual(["error"]); + expect(await stream.result()).toMatchObject({ stopReason: "error", content: [] }); +} + +describe("direct API pre-generation errors", () => { + it("return an error stream instead of throwing synchronously when auth is missing", async () => { + await expectPreGenerationError(() => streamAnthropic(model("anthropic-messages"), { messages: [] }, {})); + await expectPreGenerationError(() => streamAzure(model("azure-openai-responses"), { messages: [] }, {})); + await expectPreGenerationError(() => streamGoogle(model("google-generative-ai"), { messages: [] }, {})); + await expectPreGenerationError(() => streamMistral(model("mistral-conversations"), { messages: [] }, {})); + await expectPreGenerationError(() => streamCodex(model("openai-codex-responses"), { messages: [] }, {})); + await expectPreGenerationError(() => streamOpenAICompletions(model("openai-completions"), { messages: [] }, {})); + await expectPreGenerationError(() => streamOpenAIResponses(model("openai-responses"), { messages: [] }, {})); + }); +}); diff --git a/packages/coding-agent/src/experimental/harness-wire-adapter.ts b/packages/coding-agent/src/experimental/harness-wire-adapter.ts index 51177006a..e876f8999 100644 --- a/packages/coding-agent/src/experimental/harness-wire-adapter.ts +++ b/packages/coding-agent/src/experimental/harness-wire-adapter.ts @@ -10,17 +10,17 @@ import type { LaneSnapshot as HarnessLaneSnapshot, RunResult as HarnessRunResult, } from "@earendil-works/pi-agent-core"; -import { - type AssistantMessage, - assistantMessageEventToFrame, - type DeferredHandle, - type ImageContent, - type TextContent, - type ThinkingContent, - type ToolCall, - type ToolResultMessage, - type Usage, - type UserMessage, +import type { + AssistantMessageFrame as AiAssistantMessageFrame, + AssistantMessage, + DeferredHandle, + ImageContent, + TextContent, + ThinkingContent, + ToolCall, + ToolResultMessage, + Usage, + UserMessage, } from "@earendil-works/pi-ai"; import { type AssistantMessageFrame, @@ -580,9 +580,7 @@ function toWireToolOutput(result: AgentToolResult): ToolOutput { }; } -function toWireFrame(event: Extract): AssistantMessageFrame { - const frame = assistantMessageEventToFrame(event.event); - if (frame === undefined) throw new TypeError(`Unsupported assistant message event: ${event.event.type}`); +function toWireFrame(frame: AiAssistantMessageFrame): AssistantMessageFrame { switch (frame.type) { case "start": return { type: "start", partial: toWireAssistantMessage(frame.partial) }; @@ -590,6 +588,7 @@ function toWireFrame(event: Extract): return { type: "text_start", contentIndex: frame.contentIndex, content: { ...frame.content } }; case "text_delta": case "thinking_delta": + case "toolcall_checkpoint": case "toolcall_delta": return { ...frame }; case "text_end": @@ -735,7 +734,15 @@ export function toWireLaneEvent(event: HarnessEvent): LaneEvent | undefined { case "message_end": return { ...event, message: toWireMessage(event.message), ...base, lane: event.lane }; case "message_update": - return { type: "message_update", runId: event.runId, frame: toWireFrame(event), ...base, lane: event.lane }; + return event.frame === undefined + ? undefined + : { + type: "message_update", + runId: event.runId, + frame: toWireFrame(event.frame), + ...base, + lane: event.lane, + }; case "tool_start": return { ...event, diff --git a/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts b/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts index 22888488f..f9d717ef3 100644 --- a/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts +++ b/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts @@ -244,6 +244,7 @@ describe("Harness wire adapter", () => { runId: "run-1", message: partial, event: { type: "text_delta", contentIndex: 0, delta: "swer", partial }, + frame: { type: "text_delta", contentIndex: 0, delta: "swer" }, }; const wire = toWireLaneEvent(event); expect(Check(LaneEventSchema, wire)).toBe(true); diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index ccf1195ee..4212c5c68 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -579,7 +579,7 @@ describe("experimental durable server composition", () => { expect(legacyPrompt).not.toHaveBeenCalled(); }); - // Runtime2 cannot complete no-tool runs yet; keep the framed prompt path covered as a sanitized failure above. + // Runtime cannot complete no-tool runs yet; keep the framed prompt path covered as a sanitized failure above. test.skip("streams prompt events from the worker to the client", async ({ onTestFinished }) => { const spawn = vi .spyOn(processRuntime, "spawnInternalProcess") @@ -622,7 +622,7 @@ describe("experimental durable server composition", () => { ); }); - // Re-enable with runtime2 no-tool execution. + // Re-enable with runtime no-tool execution. test.skip("completes and persists a prompt through the worker-owned Harness", async ({ onTestFinished }) => { const spawn = vi .spyOn(processRuntime, "spawnInternalProcess") diff --git a/packages/protocol/src/harness.ts b/packages/protocol/src/harness.ts index 7d9eadbc5..50c27e79e 100644 --- a/packages/protocol/src/harness.ts +++ b/packages/protocol/src/harness.ts @@ -277,6 +277,11 @@ export const AssistantMessageFrameSchema = Type.Union([ contentIndex: Type.Integer({ minimum: 0 }), toolCall: ToolCallSchema, }), + StrictObject({ + type: Type.Literal("toolcall_checkpoint"), + contentIndex: Type.Integer({ minimum: 0 }), + json: Type.String(), + }), StrictObject({ type: Type.Literal("toolcall_delta"), contentIndex: Type.Integer({ minimum: 0 }),