diff --git a/CLAUDE.md b/CLAUDE.md index 1fb67b9c..3a33f950 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -38,7 +38,7 @@ A run executes in one of **three execution modes** behind the one `LLMProvider` engine is identical across all three. See [ADR-0012](docs/decisions/0012-managed-inference-dual-mode.md) to [ADR-0015](docs/decisions/0015-managed-mode-data-handling-and-compliance.md) and [docs/architecture/managed-inference.md](docs/architecture/managed-inference.md). -**Status: Phase 1 in progress — milestone M1 (LLM seam proven) reached (PR #9, 2026-06-07); the `FallbackChain` runner (1.K) landed, completing 1.m2 with the cost tracker (PR #13, 2026-06-11); the run loop (1.N — `WorkflowEngine` + `RunEventBus`) landed (PR #17, 2026-06-13) **completing 1.m3** (parse → DAG → run loop emits the canonical event stream), with the built-in `ToolRegistry` (1.T, a 1.m4 component) landing alongside it as the other `AgentRunner` (1.O) join prerequisite; the **`AgentRunner` (1.O) — per-node LLM execution behind the seam — landed (PR #18, 2026-06-14)**; and the **node-type handlers (1.P) — the six non-agent `NodeExecutor` arms (condition / transform / fan_out / fan_in / input / output) behind a dispatching executor — landed (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the human gate (1.Q) landed (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process `resumeFromCheckpoint`, and the `human_in_the_loop` gate with the one-shot timeout port. Node retry (1.S) is next, toward M2.** +**Status: Phase 1 in progress — milestone M1 (LLM seam proven) reached (PR #9, 2026-06-07); the `FallbackChain` runner (1.K) landed, completing 1.m2 with the cost tracker (PR #13, 2026-06-11); the run loop (1.N — `WorkflowEngine` + `RunEventBus`) landed (PR #17, 2026-06-13) **completing 1.m3** (parse → DAG → run loop emits the canonical event stream), with the built-in `ToolRegistry` (1.T, a 1.m4 component) landing alongside it as the other `AgentRunner` (1.O) join prerequisite; the **`AgentRunner` (1.O) — per-node LLM execution behind the seam — landed (PR #18, 2026-06-14)**; and the **node-type handlers (1.P) — the six non-agent `NodeExecutor` arms (condition / transform / fan_out / fan_in / input / output) behind a dispatching executor — landed (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the human gate (1.Q) landed (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process `resumeFromCheckpoint`, and the `human_in_the_loop` gate with the one-shot timeout port; and **node retry (1.S) — the above-chain whole-node retry budget ([ADR-0040](docs/decisions/0040-node-retry-budget-above-the-chain.md), amending ADR-0038) — landed (PR #24, 2026-06-15)**, re-dispatching a whole node on a retryable failure up to `retry.max` attempts (with `node:retrying`, abort-aware backoff, and `retry_on` filtering), with retry-from-node (ADR-0040 Part B) deferred to Phase-2. The pre-egress budget governor (1.AC) is next, toward M2.** Phase 0 (M0, 2026-06-04) landed the monorepo, strict toolchain + CI, `@relavium/shared` (the full Zod contract set), the no-vendor-type seam fence, and `@relavium/db`. Phase 1 has since landed `@relavium/llm` — the `LLMProvider` seam + all three adapters (Anthropic, OpenAI/DeepSeek, @@ -60,8 +60,7 @@ executor-only with a `secretInputNames` masking gate on `NodeExecContext`), and `run_events` log, no checkpoint table — ADR-0003) + cross-process `resumeFromCheckpoint` with idempotent re-delivery and a `workflow_mismatch` identity guard, and the `human_in_the_loop` gate's suspend/resume plus the one-shot `setTimer` timeout port — `approve` auto-resolves, `reject` fails with `run_timeout`). -Active work is now the remaining 1.m4 lane — node retry (1.S) and the pre-egress budget governor -(1.AC) — toward **M2**; see +Active work is now the last 1.m4 workstream — the pre-egress budget governor (1.AC) — toward **M2**; see [docs/roadmap/current.md](docs/roadmap/current.md). See [README.md](README.md) for the public overview. ## Non-negotiable rules for AI agents diff --git a/docs/reference/contracts/agent-session-spec.md b/docs/reference/contracts/agent-session-spec.md index 7451e37e..0eb95291 100644 --- a/docs/reference/contracts/agent-session-spec.md +++ b/docs/reference/contracts/agent-session-spec.md @@ -52,10 +52,28 @@ stateDiagram-v2 | **export** | Serialize the session to a `.relavium.yaml` scaffold ([export](#export-to-workflow)). | The turn loop, tool dispatch, streaming, and fallback are the **same** code paths a workflow `agent` -node uses; the session is a thin wrapper over the `AgentRunner` that manages conversation state and -context. The lifecycle emits the `session:*` event namespace — defined, with the run namespace, in -[sse-event-schema.md](sse-event-schema.md#session-event-namespace) (this spec does not enumerate event -names). +node uses: the session is a thin wrapper over the **correlation-agnostic turn core** — the `runAgentTurn` +path the `AgentRunner` (1.O) also wraps for a workflow node — managing conversation state and context. +*(1.V drives that turn core directly; it does **not** route through the run-only `NodeExecutor` the +`AgentRunner` exposes. "Same `AgentRunner` path" means the shared turn-core execution, not the +`NodeExecutor` surface.)* The lifecycle emits the `session:*` event namespace — defined, with the run +namespace, in [sse-event-schema.md](sse-event-schema.md#session-event-namespace) (this spec does not +enumerate event names). 1.V keeps the conversation **in-memory** (the in-flight `LlmMessage`/`ContentPart` +form) and emits session events through an injected sink; wiring that sink onto the shared `RunEventBus` +(per-session `sequenceNumber` + gap/resync) is **1.W**, and the durable [`SessionMessage`](#session-messages) +schema + persistence is **1.X**. + +### Hard turn cap + +A session carries a **hard turn cap** — a finite DoS fail-safe on the number of turns it will run (engine +default **50**, overridable at construction; **0/absent ⇒ the default**). It is **distinct** from two other +limits and must not be conflated with either: `[chat].max_messages` (a history-**trim** threshold that +silently *continues* the session — [config-spec.md](config-spec.md)) and the turn core's **within-turn** +`maxToolTurns` tool-loop guard. A `sendMessage` past the cap ends **loudly, with no egress**: +`session:turn_completed` carries `stopReason: 'error'` + `error.code: 'turn_limit'` +([sse-event-schema.md](sse-event-schema.md#error-code-taxonomy)) — never a silent stop; the within-turn +`maxToolTurns` guard surfaces the same `turn_limit` code through the same event. The cap is an **engine-API +knob** in 1.V (a surface maps its `[chat]` default onto it); it is **not** a new `[chat]` field in Phase 1. ## Session context diff --git a/docs/reference/contracts/sse-event-schema.md b/docs/reference/contracts/sse-event-schema.md index c4b7c63c..cd110ac7 100644 --- a/docs/reference/contracts/sse-event-schema.md +++ b/docs/reference/contracts/sse-event-schema.md @@ -159,6 +159,27 @@ export interface HumanGatePausedEvent extends BaseEvent { timeoutAction?: 'approve' | 'reject'; // on-timeout policy (present only with timeoutMs); lets a surface show how the gate auto-resolves and a Phase-2 crash-resume re-arm the timer from the log expiresAt?: string; } + +export interface BudgetWarningEvent extends BaseEvent { + type: 'budget:warning'; + spentMicrocents: number; + limitMicrocents: number; + thresholdPct: number; // 0–100, rounded from spent/limit at the pre-egress check point +} + +export interface BudgetPausedEvent extends BaseEvent { + type: 'budget:paused'; + nodeId: string; // the agent node whose next LLM call would exceed the cap + spentMicrocents: number; + limitMicrocents: number; + gateId: string; // stable id of the budget gate; required by engine.resume(runId, gateId, decision) +} + +export interface RunTimeoutEvent extends BaseEvent { + type: 'run:timeout'; + elapsedMs: number; + timeoutMs: number; +} ``` ### Security: event payloads never carry secrets @@ -236,8 +257,8 @@ Within a turn, the conversational work reuses the **same** `agent:token` / `agen | `type` | Meaning | Key payload fields | | --- | --- | --- | -| `budget:warning` | Spend crossed the warning threshold. | `spentMicrocents`, `limitMicrocents`, `thresholdPct` | -| `budget:paused` | Spend would exceed the cap with `on_exceed: pause_for_approval`; the run suspends like a human gate and is resumed via the `resume_budget` IPC command. | `spentMicrocents`, `limitMicrocents` | +| `budget:warning` | Pre-egress worst-case cost estimate would exceed the configured cap, and `on_exceed: warn` is set. Emitted once per run before the capped egress; execution continues. `thresholdPct` is `clamp(round(spent / limit * 100), 0, 100)` observed at the pre-egress check point. | `spentMicrocents`, `limitMicrocents`, `thresholdPct` | +| `budget:paused` | Pre-egress estimate would exceed the cap with `on_exceed: pause_for_approval`; the run suspends like a human gate and is resumed via `engine.resume(runId, gateId, decision)`. `decision: approved` continues; `rejected` closes the run with `run:failed{code: budget_exceeded}`. | `nodeId`, `spentMicrocents`, `limitMicrocents`, `gateId` | | `run:timeout` | The run hit its `timeout_ms`. | `elapsedMs`, `timeoutMs` | These three (and `run:paused` / `human_gate:paused`) are **non-terminal** — they signal a governance/suspension state, not the run's end. A run that cannot continue past a timeout or budget cap still closes with **exactly one** `run:failed` carrying `code: run_timeout` / `budget_exceeded`. The exactly-one-terminal-event invariant (`run:completed | run:failed | run:cancelled`) and its precedence are owned by [ADR-0036](../../decisions/0036-run-loop-substrate-event-bus-and-execution-host.md). diff --git a/docs/roadmap/current.md b/docs/roadmap/current.md index 208dd1bb..470031ed 100644 --- a/docs/roadmap/current.md +++ b/docs/roadmap/current.md @@ -151,9 +151,12 @@ loser-cancel deferred), and a pre-merge BLOCKER secret-leak (the `input` handler the derived `Checkpointer` (state folded from the `run_events` log; no checkpoint table — ADR-0003) + cross-process `resumeFromCheckpoint` (idempotent re-delivery, `workflow_mismatch` identity guard), and the `human_in_the_loop` gate (suspend → notify → resume, plus the one-shot `setTimer` timeout port — `approve` -auto-resolves, `reject` fails with `run_timeout`). The lane now continues at the remaining **1.m4** -workstreams toward **M2** — **node retry (1.S)** and the **pre-egress budget governor (1.AC)** — and the -agent-first sub-spine (**1.V–1.AA**, Lane C) opens now that 1.O exists. +auto-resolves, `reject` fails with `run_timeout`). **Node retry (1.S) is ✅ Done (PR #24, 2026-06-15)** — the above-chain whole-node retry budget +([ADR-0040](../decisions/0040-node-retry-budget-above-the-chain.md) Part A: re-dispatch a whole node on a +retryable, `retry_on`-admitted failure up to `retry.max` attempts with abort-aware backoff and the non-terminal +`node:retrying`, `node:failed` staying the single terminal; the user-triggered retry-from-node Part B is +deferred to Phase-2). The lane now continues at the last **1.m4** workstream toward **M2** — the **pre-egress +budget governor (1.AC)** — and the agent-first sub-spine (**1.V–1.AA**, Lane C) is open now that 1.O exists. > **Multimodal I/O — the shape is landed (1.AD ✅ Done, PR #11, 2026-06-10).** First-class > image/audio/video I/O (input **and** output, incl. generate-media-by-rule) was decided on 2026-06-08: @@ -181,8 +184,10 @@ agent-first sub-spine (**1.V–1.AA**, Lane C) opens now that 1.O exists. > **1.N (`WorkflowEngine` + `RunEventBus`) and 1.T (the built-in `ToolRegistry`) are ✅ Done (PR #17, > merged 2026-06-13)** — **1.N closes 1.m3** (its last component); **1.T** (a 1.m4 component) is the > other 1.O join prerequisite; **the `AgentRunner` join (1.O) is ✅ Done (PR #18, 2026-06-14)**; and the -> **node-type handlers (1.P) are ✅ Done (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the -> human gate (1.Q) are ✅ Done (PR #22, 2026-06-15)**. **Node retry (1.S)** is the next workstream. +> **node-type handlers (1.P) are ✅ Done (PR #20, 2026-06-14)**; **checkpoint/resume (1.R) + the +> human gate (1.Q) are ✅ Done (PR #22, 2026-06-15)**; and **node retry (1.S) is ✅ Done (PR #24, 2026-06-15)** +> (ADR-0040 Part A; the user-triggered retry-from-node Part B is deferred to Phase-2). The **pre-egress budget +> governor (1.AC)** is the next workstream. Carry-over hardening is tracked in [deferred-tasks.md](deferred-tasks.md) — pick items up as Phase 1 first touches each file. diff --git a/docs/roadmap/deferred-tasks.md b/docs/roadmap/deferred-tasks.md index 305c2f6b..3da1f894 100644 --- a/docs/roadmap/deferred-tasks.md +++ b/docs/roadmap/deferred-tasks.md @@ -188,6 +188,14 @@ Severity is the review's verified rating. Check an item off in the PR that resol question: does the estimate need provider-accurate token counting (from the seam's model meta / usage feedback) to avoid systematic over/under-blocking, or is the declared estimate enough? No change now — re-evaluate with real 1.AC telemetry. *(1.AC; ADR-0028)* +- [ ] **Configurable sub-100% budget warning threshold (ADR-0028 amendment).** `budget:warning` + today is emitted only when a pre-egress estimate would already exceed the cap, on the `on_exceed: warn` + path (`thresholdPct` reports the observed spent/limit fraction at that point). A user-facing + early-warning threshold (e.g. `warn_at_pct: 80`) requires amending ADR-0028 to add both a + config default and a per-workflow `budget.warn_at_pct` field, plus a decision on whether it + throttles/queues subsequent egresses or only surfaces a one-time advisory event. Deferred until + there is concrete surface demand or telemetry showing operators need an earlier signal. + *(1.AC; ADR-0028; config-spec.md; workflow-yaml-spec.md)* ## Interpolation engine (1.L2) follow-ups diff --git a/docs/roadmap/phases/phase-1-engine-and-llm.md b/docs/roadmap/phases/phase-1-engine-and-llm.md index f54a102e..47970fbb 100644 --- a/docs/roadmap/phases/phase-1-engine-and-llm.md +++ b/docs/roadmap/phases/phase-1-engine-and-llm.md @@ -25,8 +25,10 @@ > gate (1.Q) are ✅ Done (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process > `resumeFromCheckpoint` (idempotent re-delivery, `workflow_mismatch` identity guard) and the > `human_in_the_loop` gate (suspend/resume + the one-shot `setTimer` timeout port: `approve` auto-resolves, -> `reject` fails with `run_timeout`). The lane now continues at the remaining 1.m4 workstreams (**1.S** node -> retry, **1.AC** budget governor) toward **M2**, and Lane C (1.V–1.AA) opens. *(Session persistence, +> `reject` fails with `run_timeout`). **Node retry (1.S) is ✅ Done (PR #24, 2026-06-15)** — the above-chain +> whole-node retry budget ([ADR-0040](../../decisions/0040-node-retry-budget-above-the-chain.md), Part A; the +> user-triggered retry-from-node Part B is deferred to Phase-2). The lane now continues at the last 1.m4 +> workstream — **1.AC** budget governor — toward **M2**, and Lane C (1.V–1.AA) opens. *(Session persistence, > 1.X/1.Z, must exclude the reasoning signature — non-persisting.)* > > **Multimodal I/O decided (2026-06-08).** First-class image/audio/video I/O (input **and** output) is a @@ -748,23 +750,30 @@ idempotent; gap detection triggers a resync rather than trusting a partial view; resume against a non-matching workflow snapshot is refused with a typed error; and a forced checkpoint-write failure is visible as a typed error/event, not a silent skip. -### 1.S — Retry + fallback wiring (node budget above the chain) — *critical path* +### 1.S — Retry + fallback wiring (node budget above the chain) · ✅ **Done (PR #24, 2026-06-15)** — *critical path* Layer node-level retry above the provider fallback chain so reliability composes -correctly. +correctly. Landed as [ADR-0040](../../decisions/0040-node-retry-budget-above-the-chain.md) +Part A (the in-run budget); ADR-0040 amends ADR-0038 (the primary chain entry no longer +consumes `node.retry`). **Tasks:** -- Implement node-level retry from `retry_config` (`max_attempts`, `backoff_ms`, - `backoff_strategy`, `retry_on[]`), with backoff and optional input adjustment, never - silently skipping a required node. -- Wire the ordering: the engine considers a node failed only after the **whole +- ✅ Implement node-level retry from `retry_config` (`max_attempts` ← `retry.max`, + `backoff_ms`, `backoff_strategy`, `retry_on[]`), with backoff, never silently skipping + a required node. *(Optional input adjustment deferred — ADR-0040 A.9.)* +- ✅ Wire the ordering: the engine considers a node failed only after the **whole fallback chain** (1.K) is exhausted within an attempt, then applies the node retry budget above that. -- Implement retry-from-node (re-run from any node) using the 1.R idempotency key. +- ⏳ Implement retry-from-node (re-run from any node) using the 1.R idempotency key — + **deferred to Phase-2** (ADR-0040 Part B; same-runId re-run conflicts with the + exactly-one-terminal invariant — see [deferred-tasks.md](../deferred-tasks.md)). -**Acceptance:** a node whose provider chain is exhausted retries per its budget and -then fails with `node:failed`; a transient failure recovers within budget; cost is -recorded for every attempt across both layers. +**Acceptance:** ✅ Met (PR #24). A node whose provider chain is exhausted retries per its +budget and then fails with `node:failed`; a transient failure recovers within budget; cost +is recorded for every attempt across both layers. The engine re-dispatches the whole node +on a retryable, `retry_on`-admitted failure up to `retry.max` total attempts with +abort-aware backoff (cancel/sibling-abort wins), emitting the non-terminal `node:retrying`; +`node:failed` stays the single terminal per node, so the 1.R Checkpointer fold is untouched. ### 1.T — Built-in `ToolRegistry` + dispatch · ✅ **Done (PR #17, 2026-06-13)** @@ -834,8 +843,10 @@ reproduces the same final output — **M2 achieved**. These build the `AgentSession` entry point ([ADR-0024](../../decisions/0024-agent-first-entry-point-agentsession.md)). They run **parallel** to 1.L–1.U and do **not** feed the 1.U workflow harness — each is proven by its own harness (1.AA). The `WorkflowEngine` is unchanged; `AgentSession` is an additional entry point on the same substrate. - **1.V — `AgentSession` entry point.** Wrap `AgentRunner` in a multi-turn session (session context, one bound agent + its fallback chain). This workstream also settles the session's **hard turn/round cap** knob — deliberately distinct from `[chat].max_messages`, which is a history-**trim** threshold ([config-spec.md](../../reference/contracts/config-spec.md)) that continues the session. A session that reaches the hard cap ends **loudly**: `session:turn_completed` carries `error.code: 'turn_limit'` ([sse-event-schema.md](../../reference/contracts/sse-event-schema.md#error-code-taxonomy)) — never a silent stop — and the behavior is pinned by a dedicated regression test (a refactor of the turn loop must not be able to silently drop the cap signal). Context compaction (when it lands, later phases) is **append-only by principle**: the persisted transcript is never rewritten or trimmed in place; `agent:context_compacted` is the reserved signal ([sse-event-schema.md](../../reference/contracts/sse-event-schema.md#workflow-governance-and-reserved-events)). *Acceptance:* a session runs a multi-turn conversation with a tool round-trip through the same `AgentRunner` path a workflow agent node uses; a session driven to its hard turn cap emits the `turn_limit`-coded event, regression-pinned. + - *1.V scope notes (decided 2026-06-15, [agent-session-spec.md](../../reference/contracts/agent-session-spec.md)):* the driver reuses the correlation-agnostic turn core (`runAgentTurn`) directly via an **injected `SessionEventSink`** (so the bus wiring is **1.W** — see below); the hard cap is an **engine-API knob** (finite default **50**), with the `[chat]` surface-default mapping deferred to the surface phases (it is **not** a Phase-1 `[chat]` field). **Deferred within 1.V, to be picked up later:** the in-memory transcript is **content-only** — the turn core discards the within-turn tool-call/tool-result messages (`agent-turn.ts` copies its input and returns only final content), so faithful tool-message capture is **revisited when 1.X persistence / 1.Z export needs it** (and once `agent-turn.ts` is free of the concurrent 1.AC edits); **per-session tool narrowing** (ADR-0029 narrow-only — 1.V grants `agent.tools` as-is, no per-session narrow); and **session `output_schema`** (chat is free-form text — structured output stays a workflow concern). - **1.W — `session:*` event namespace.** Emit session lifecycle events on the shared `RunEventBus` with the same `sequenceNumber` gap/resync logic ([sse-event-schema.md](../../reference/contracts/sse-event-schema.md)). *Acceptance:* session events are disjoint from `run:*` and gap-detected identically. -- **1.X — Session persistence.** `agent_sessions` + `session_messages` via `@relavium/db` into `history.db` ([database-schema.md](../../reference/desktop/database-schema.md)). *Acceptance:* a session round-trips to the DB and resumes. **Note:** adding these two tables requires a regenerated Drizzle migration snapshot (the schema-migration drift CI gate). **ADR-0030 ephemerality:** a `reasoning` part's `signature`/`redacted` continuity token must **not** be persisted to `session_messages` — strip it (keep reasoning *text* if a transcript needs it, drop the opaque signature). *Acceptance also asserts:* a round-tripped session row carries no reasoning `signature`. + - *Also owns (from the 1.V injected-sink split, recorded 2026-06-15):* the `SessionEventSink → RunEventBus` adapter; a **`SessionHandle`** (the async-iterable session stream + `cancel`, mirroring `createRunHandle`); and **reconciling the bus session-event gate** — `RunEventBus` validates against `RunEventSchema` and its `RunEventDraft` type both **exclude** the five `session:*` variants (they live only in the separate `SessionEventSchema`), so the bus needs an injected/combined Run+Session schema **and** the matching draft-type widening before it can carry the session lifecycle events 1.V emits. +- **1.X — Session persistence.** `agent_sessions` + `session_messages` via `@relavium/db` into `history.db` ([database-schema.md](../../reference/desktop/database-schema.md)). **Authors the durable `SessionMessageSchema`** in `@relavium/shared` (deferred from 1.V, which runs on the in-flight `LlmMessage` form): `{ id, sessionId, sequenceNumber, role, content: DurableContentPart[], modelId?, timestamp }` — the persisted transcript type these tables store. *Acceptance:* a session round-trips to the DB and resumes. **Note:** adding these two tables requires a regenerated Drizzle migration snapshot (the schema-migration drift CI gate). **ADR-0030 ephemerality:** a `reasoning` part's `signature`/`redacted` continuity token must **not** be persisted to `session_messages` — strip it (keep reasoning *text* if a transcript needs it, drop the opaque signature). *Acceptance also asserts:* a round-tripped session row carries no reasoning `signature`. - **1.Y — Session checkpoint/resume.** Reuse the idempotency-key logic so a session resumes after a restart. - **1.Z — Export-to-workflow serializer.** Session → `.relavium.yaml` **linear-chain scaffold + transcript** ([ADR-0026](../../decisions/0026-session-export-to-workflow.md)). Includes a **`WorkflowDefinition` → YAML emitter** (deterministic key ordering, the `metadata` transcript block, secret exclusion) — 1.L is parse-only, so this workstream owns serialization. *Acceptance:* an exported session parses as a valid workflow whose agent nodes mirror the turns; **parse → serialize round-trips** (including `metadata`); no `secret` value is serialized; and **no reasoning `signature` is serialized** (ADR-0030 ephemerality — the signature is a transient same-provider token, never written to a committable artifact, same exclusion as `secret`). - **1.AA — Node-harness chat regression.** The session counterpart of 1.U: a multi-turn chat with a tool call and an export, run green in CI. @@ -1095,13 +1106,13 @@ flowchart LR | 1.L2 | B | 1.L | 1.M | ✅ — **Done (PR #15)** | | 1.M | B | 1.L2 | 1.N | ✅ — **Done (PR #16)** | | 1.N | B | 1.M | 1.O, 1.R, 1.W | ✅ — **Done (PR #17)** | -| 1.R | B | 1.N | 1.S, 1.Q, 1.Y | ✅ | +| 1.R | B | 1.N | 1.S, 1.Q, 1.Y | ✅ — **Done (PR #22)** | | 1.T | B | 1.E | 1.O, 1.U | ⬤ — **Done (PR #17)** | | 1.AB | B | package scaffold (perf spike first) | 1.P | ✅ folds into 1.P — **Done (PR #16)** | | 1.O | B | **1.K, 1.N, 1.T** (the join), 1.AD (media shape) | 1.P, 1.S, 1.AC, 1.V | ✅ — **Done (PR #18)** | -| 1.P | B | 1.O, 1.AB | 1.Q, 1.U | ✅ | -| 1.S | B | 1.O, 1.R | 1.U | ✅ | -| 1.Q | B | 1.P, 1.R | 1.AC, 1.U | ⬤ | +| 1.P | B | 1.O, 1.AB | 1.Q, 1.U | ✅ — **Done (PR #20)** | +| 1.S | B | 1.O, 1.R | 1.U | ✅ — **Done (PR #24)** | +| 1.Q | B | 1.P, 1.R | 1.AC, 1.U | ✅ — **Done (PR #22)** | | 1.AC | B | 1.O, 1.Q | 1.U | ✅ folds into 1.O | | 1.U | B | 1.P, 1.S, 1.Q, 1.R, 1.T, 1.AC | **M2** | ✅ | | 1.V | C | 1.O | 1.W, 1.X, 1.Z | ◇ | diff --git a/packages/core/src/dag.ts b/packages/core/src/dag.ts index 051b290b..ae5ac674 100644 --- a/packages/core/src/dag.ts +++ b/packages/core/src/dag.ts @@ -146,6 +146,8 @@ export function buildRunPlan(def: Workflow, opts?: BuildRunPlanOptions): RunPlan order, vertices, ...(spec.max_parallel === undefined ? {} : { maxParallel: spec.max_parallel }), + ...(spec.budget === undefined ? {} : { budget: spec.budget }), + ...(spec.timeout_ms === undefined ? {} : { timeoutMs: spec.timeout_ms }), }; } diff --git a/packages/core/src/engine/agent-runner.e2e.test.ts b/packages/core/src/engine/agent-runner.e2e.test.ts index e03d861b..e9268de0 100644 --- a/packages/core/src/engine/agent-runner.e2e.test.ts +++ b/packages/core/src/engine/agent-runner.e2e.test.ts @@ -1,13 +1,14 @@ import type { CapabilityFlags, LlmProvider, ProviderId, StreamChunk } from '@relavium/llm'; import type { RunEvent } from '@relavium/shared'; -import { describe, expect, it } from 'vitest'; +import { beforeAll, describe, expect, it } from 'vitest'; import { parseWorkflow } from '../parser.js'; -import type { ToolRegistry, ToolResultPart } from '../tools/types.js'; +import { createExpressionSandbox, type ExpressionSandbox } from '../expression/sandbox.js'; +import type { ToolDef as CoreToolDef, ToolRegistry, ToolResultPart } from '../tools/types.js'; import { markUntrusted } from '../tools/untrusted.js'; -import { createAgentNodeExecutor } from './agent-runner.js'; import { WorkflowEngine } from './engine.js'; import { createInMemoryHost } from './execution-host.js'; +import { createStandardNodeExecutor } from './node-handlers/dispatcher.js'; import type { RunHandle } from './run-handle.js'; const CAPS: CapabilityFlags = { @@ -77,6 +78,17 @@ const echoRegistry: ToolRegistry = { }, }; +/** The matching LLM-visible tool def for the echo registry. */ +const echoToolDef: CoreToolDef = { + id: 'echo', + source: 'builtin', + description: 'echo', + parseArgs: (raw) => raw, + llmVisibleParams: { type: 'object' }, + policy: { fsScoped: false, spawnsProcess: false, requiresGateApproval: false }, + dispatch: () => Promise.reject(new Error('echoToolDef dispatch is not used directly')), +}; + const WORKFLOW = parseWorkflow( `schema_version: '1.0' workflow: @@ -109,22 +121,22 @@ function assertGapFreeSeq(events: readonly RunEvent[]): void { seqs.forEach((seq, index) => expect(seq).toBe(index)); } +let sandbox: ExpressionSandbox; + +beforeAll(async () => { + sandbox = await createExpressionSandbox(); +}); + describe('AgentRunner end-to-end through the WorkflowEngine', () => { it('runs an agent node, streaming tokens + cost, to run:completed', async () => { const engine = new WorkflowEngine({ host: createInMemoryHost(), - executor: createAgentNodeExecutor({ - resolveProvider: () => - provider([ - { type: 'text_delta', text: 'a summary' }, - { type: 'stop', stopReason: 'stop', usage: { inputTokens: 8, outputTokens: 4 } }, - ]), - registry: noToolRegistry, - tools: [], - keyFor: () => 'key', - sleep: () => Promise.resolve(), - now: () => 1, - }), + executor: agentExecutor(() => + provider([ + { type: 'text_delta', text: 'a summary' }, + { type: 'stop', stopReason: 'stop', usage: { inputTokens: 8, outputTokens: 4 } }, + ]), + ), }); const events = await drain( @@ -185,14 +197,7 @@ workflow: ); const engine = new WorkflowEngine({ host: createInMemoryHost(), - executor: createAgentNodeExecutor({ - resolveProvider: (id) => (id === 'anthropic' ? primary : fallback), - registry: noToolRegistry, - tools: [], - keyFor: () => 'key', - sleep: () => Promise.resolve(), - now: () => 1, - }), + executor: agentExecutor((id) => (id === 'anthropic' ? primary : fallback)), }); const events = await drain(engine.start({ workflow: wf })); @@ -222,8 +227,8 @@ workflow: ); const engine = new WorkflowEngine({ host: createInMemoryHost(), - executor: createAgentNodeExecutor({ - resolveProvider: () => + executor: agentExecutor( + () => scriptedProvider([ // turn 1: a tool call [ @@ -237,12 +242,9 @@ workflow: { type: 'stop', stopReason: 'stop', usage: { inputTokens: 1, outputTokens: 1 } }, ], ]), - registry: echoRegistry, - tools: [], - keyFor: () => 'key', - sleep: () => Promise.resolve(), - now: () => 1, - }), + echoRegistry, + [echoToolDef], + ), }); const events = await drain(engine.start({ workflow: wf })); @@ -282,19 +284,13 @@ workflow: ); const engine = new WorkflowEngine({ host: createInMemoryHost(), - executor: createAgentNodeExecutor({ + executor: agentExecutor(() => // One executor instance serves both concurrent nodes; each builds its own chain + CostTracker. - resolveProvider: () => - provider([ - { type: 'text_delta', text: 'done' }, - { type: 'stop', stopReason: 'stop', usage: { inputTokens: 1, outputTokens: 1 } }, - ]), - registry: noToolRegistry, - tools: [], - keyFor: () => 'k', - sleep: () => Promise.resolve(), - now: () => 1, - }), + provider([ + { type: 'text_delta', text: 'done' }, + { type: 'stop', stopReason: 'stop', usage: { inputTokens: 1, outputTokens: 1 } }, + ]), + ), }); const events = await drain(engine.start({ workflow: wf })); @@ -349,14 +345,7 @@ workflow: const host = createInMemoryHost(); const engine = new WorkflowEngine({ host, - executor: createAgentNodeExecutor({ - resolveProvider: () => scripted, - registry: noToolRegistry, - tools: [], - keyFor: () => 'k', - sleep: () => Promise.resolve(), - now: () => 1, - }), + executor: agentExecutor(() => scripted), }); const events: RunEvent[] = []; for await (const event of engine.start({ workflow: wf, inputs: { text: 'x' } }).events) { @@ -387,3 +376,170 @@ workflow: assertGapFreeSeq(events); }); }); + +function agentExecutor( + resolveProvider: (id: ProviderId) => LlmProvider | undefined, + registry: ToolRegistry = noToolRegistry, + tools: CoreToolDef[] = [], +) { + return createStandardNodeExecutor({ + sandbox, + agent: { + resolveProvider, + registry, + tools, + keyFor: () => 'k', + sleep: () => Promise.resolve(), + now: () => 1, + }, + }); +} + +describe('AgentRunner resource governance end-to-end (ADR-0028, 1.AC)', () => { + function budgetWorkflow(onExceed: string): ReturnType { + return parseWorkflow( + `schema_version: '1.0' +workflow: + id: e2e-budget + budget: + max_cost_microcents: 1 + on_exceed: ${onExceed} + agents: + - id: a + model: claude-opus-4-8 + provider: anthropic + system_prompt: hi + nodes: + - id: n + type: agent + agent_ref: a + prompt_template: 'go' + edges: [] +`, + ); + } + + function cheapProvider(): LlmProvider { + return provider([ + { type: 'text_delta', text: 'ok' }, + { type: 'stop', stopReason: 'stop', usage: { inputTokens: 1, outputTokens: 1 } }, + ]); + } + + it('warns and continues when on_exceed is warn', async () => { + const engine = new WorkflowEngine({ + host: createInMemoryHost(), + executor: agentExecutor(() => cheapProvider()), + }); + const events = await drain( + engine.start({ workflow: budgetWorkflow('warn'), inputs: { text: 'x' } }), + ); + expect(events.at(-1)?.type).toBe('run:completed'); + const warning = events.find((e) => e.type === 'budget:warning'); + expect(warning).toBeDefined(); + // The warning fires before any spend, so the observed spent/limit fraction is 0%. + expect(warning?.type === 'budget:warning' && warning.thresholdPct).toBe(0); + }); + + it('fails the run when on_exceed is fail', async () => { + const engine = new WorkflowEngine({ + host: createInMemoryHost(), + executor: agentExecutor(() => cheapProvider()), + }); + const events = await drain( + engine.start({ workflow: budgetWorkflow('fail'), inputs: { text: 'x' } }), + ); + expect(events.at(-1)?.type).toBe('run:failed'); + const terminal = events.at(-1); + expect(terminal?.type === 'run:failed' && terminal.error.code).toBe('budget_exceeded'); + expect(events.some((e) => e.type === 'node:failed')).toBe(true); + }); + + it('pauses and resumes when on_exceed is pause_for_approval', async () => { + const engine = new WorkflowEngine({ + host: createInMemoryHost(), + executor: agentExecutor(() => cheapProvider()), + }); + const handle = engine.start({ + workflow: budgetWorkflow('pause_for_approval'), + inputs: { text: 'x' }, + }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'budget:paused') { + await engine.resume(handle.runId, event.gateId, { + decision: 'approved', + decidedBy: 'user-1', + }); + } + } + expect(events.at(-1)?.type).toBe('run:completed'); + expect(events.some((e) => e.type === 'budget:paused')).toBe(true); + // run:paused is emitted, but a synchronous resume may settle the run before the event surfaces, + // so we only assert the budget gate fired and the run ultimately completed. + }); + + it('fails the run when a paused budget gate is rejected', async () => { + const engine = new WorkflowEngine({ + host: createInMemoryHost(), + executor: agentExecutor(() => cheapProvider()), + }); + const handle = engine.start({ + workflow: budgetWorkflow('pause_for_approval'), + inputs: { text: 'x' }, + }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'budget:paused') { + await engine.resume(handle.runId, event.gateId, { + decision: 'rejected', + decidedBy: 'user-1', + }); + } + } + expect(events.at(-1)?.type).toBe('run:failed'); + const terminal = events.at(-1); + expect(terminal?.type === 'run:failed' && terminal.error.code).toBe('budget_exceeded'); + }); + + it('fails the run when timeout_ms elapses', async () => { + const base = createInMemoryHost(); + const host = { + ...base, + // Fire any timeout immediately, deterministically. + setTimer: (_ms: number, onFire: () => void) => { + onFire(); + return () => {}; + }, + }; + const engine = new WorkflowEngine({ + host, + executor: agentExecutor(() => cheapProvider()), + }); + const wf = parseWorkflow( + `schema_version: '1.0' +workflow: + id: e2e-timeout + timeout_ms: 1 + agents: + - id: a + model: claude-opus-4-8 + provider: anthropic + system_prompt: hi + nodes: + - id: n + type: agent + agent_ref: a + prompt_template: 'go' + edges: [] +`, + ); + const events = await drain(engine.start({ workflow: wf, inputs: { text: 'x' } })); + expect(events.some((e) => e.type === 'run:timeout')).toBe(true); + expect(events.at(-1)?.type).toBe('run:failed'); + const terminal = events.at(-1); + expect(terminal?.type === 'run:failed' && terminal.error.code).toBe('run_timeout'); + }); +}); diff --git a/packages/core/src/engine/agent-runner.ts b/packages/core/src/engine/agent-runner.ts index 077a927b..afc9f1e4 100644 --- a/packages/core/src/engine/agent-runner.ts +++ b/packages/core/src/engine/agent-runner.ts @@ -40,6 +40,7 @@ import { type ChainCapabilities, type PreEgressHook, } from './agent-turn.js'; +import { BudgetPauseError } from './budget-governor.js'; import type { NodeExecContext, NodeExecutor, NodeOutcome } from './node-executor.js'; type AgentNode = AgentPlanConfig['node']; @@ -169,6 +170,9 @@ async function executeAgent( }); } catch (err) { if (err instanceof AgentTurnError) return failed(err.code, err.message, err.retryable); + if (err instanceof BudgetPauseError) { + return { kind: 'paused', gate: err.toGateRequest() }; + } throw err; // unexpected — the engine's catch-all maps it to a single `internal` failure } diff --git a/packages/core/src/engine/agent-session.test.ts b/packages/core/src/engine/agent-session.test.ts new file mode 100644 index 00000000..1deba57a --- /dev/null +++ b/packages/core/src/engine/agent-session.test.ts @@ -0,0 +1,312 @@ +import type { CapabilityFlags, LlmProvider, ProviderId, StreamChunk } from '@relavium/llm'; +import { + AgentSchema, + SessionContextSchema, + type Agent, + type SessionContext, +} from '@relavium/shared'; +import { describe, expect, it } from 'vitest'; + +import type { ToolRegistry, ToolResultPart } from '../tools/types.js'; +import { markUntrusted } from '../tools/untrusted.js'; +import { + AgentSession, + DEFAULT_SESSION_MAX_TURNS, + SessionStateError, + type SessionDeps, + type SessionStreamEvent, +} from './agent-session.js'; +import { createAbortController } from './execution-host.js'; + +const CAPS: CapabilityFlags = { + tools: true, + streaming: true, + parallelToolCalls: true, + vision: false, + promptCache: false, + reasoning: true, + media: { + input: { image: false, audio: false, video: false, document: false }, + outputCombinations: [], + }, +}; + +async function* streamOf(chunks: readonly StreamChunk[]): AsyncGenerator { + await Promise.resolve(); + for (const c of chunks) yield c; +} + +/** A provider that replays a different chunk list per `stream()` call (call N → scripts[N]). */ +function scriptedProvider(scripts: StreamChunk[][], id: ProviderId = 'anthropic'): LlmProvider { + let call = 0; + return { + id, + supports: CAPS, + generate: () => { + throw new Error('unused'); + }, + stream: () => streamOf(scripts[call++] ?? []), + }; +} + +/** A registry returning a sanitized echo outcome (for the tool round-trip). */ +const echoRegistry: ToolRegistry = { + has: () => true, + list: () => ['echo'], + dispatch: (call) => { + const result: ToolResultPart = { type: 'tool_result', toolCallId: call.id, result: 'TOOL-OK' }; + return Promise.resolve({ + output: 'TOOL-OK', + toolResult: markUntrusted(result), + truncated: false, + events: { + call: { toolId: call.name, toolInput: {} }, + result: { toolId: call.name, success: true, outputSummary: 'TOOL-OK' }, + }, + }); + }, +}; + +const noToolRegistry: ToolRegistry = { + has: () => false, + list: () => [], + dispatch: () => Promise.reject(new Error('no tool dispatch expected')), +}; + +const AGENT: Agent = AgentSchema.parse({ + id: 'chatter', + model: 'claude-opus-4-8', + provider: 'anthropic', + system_prompt: 'You are a concise chat agent.', +}); + +const TOOL_AGENT: Agent = AgentSchema.parse({ + id: 'tool-chatter', + model: 'claude-opus-4-8', + provider: 'anthropic', + system_prompt: 'You may call echo.', + tools: ['echo'], +}); + +const CONTEXT: SessionContext = SessionContextSchema.parse({ + workingDir: '/tmp/session', + fsScopeTier: 'sandboxed', +}); + +const textTurn = (text: string): StreamChunk[] => [ + { type: 'text_delta', text }, + { type: 'stop', stopReason: 'stop', usage: { inputTokens: 5, outputTokens: 3 } }, +]; + +const toolUseTurn = (id: string): StreamChunk[] => [ + { type: 'tool_call_start', id, name: 'echo' }, + { type: 'tool_call_end', id }, + { type: 'stop', stopReason: 'tool_use', usage: { inputTokens: 4, outputTokens: 2 } }, +]; + +interface Harness { + readonly deps: SessionDeps; + readonly events: SessionStreamEvent[]; +} + +function harness( + scripts: StreamChunk[][], + overrides: Partial = {}, + registry: ToolRegistry = noToolRegistry, +): Harness { + const events: SessionStreamEvent[] = []; + const provider = scriptedProvider(scripts); + const deps: SessionDeps = { + resolveProvider: () => provider, + registry, + tools: [], + keyFor: () => 'key', + sleep: () => Promise.resolve(), + newAbortController: createAbortController, + emit: (event) => { + events.push(event); + }, + ...overrides, + }; + return { deps, events }; +} + +const session = (deps: SessionDeps, agent: Agent = AGENT): AgentSession => + new AgentSession({ sessionId: 'sess-1', agentRef: agent.id, agent, context: CONTEXT, deps }); + +const typesOf = (events: readonly SessionStreamEvent[]): readonly string[] => + events.map((e) => e.type); + +describe('AgentSession (1.V) — multi-turn entry point over the shared turn core', () => { + it('runs a multi-turn conversation with a tool round-trip through the same turn core', async () => { + // Turn 1 calls echo then answers (2 stream() calls); turn 2 is a plain answer (1 stream() call). + const { deps, events } = harness( + [toolUseTurn('c1'), textTurn('first answer'), textTurn('second answer')], + {}, + echoRegistry, + ); + const s = session(deps, TOOL_AGENT); + s.start(); + await s.sendMessage('hello'); + await s.sendMessage('again'); + + const types = typesOf(events); + expect(types[0]).toBe('session:started'); + // Two turns, each bracketed by turn_started / turn_completed. + expect(types.filter((t) => t === 'session:turn_started')).toHaveLength(2); + expect(types.filter((t) => t === 'session:turn_completed')).toHaveLength(2); + // The tool round-trip surfaced through the reused core. + expect(types).toContain('agent:tool_call'); + expect(types).toContain('agent:tool_result'); + // Both turns completed successfully (no error payload). + const completes = events.filter((e) => e.type === 'session:turn_completed'); + for (const e of completes) { + expect(e.type === 'session:turn_completed' && e.error).toBeUndefined(); + expect(e.type === 'session:turn_completed' && e.stopReason).toBe('stop'); + } + // The two assistant answers streamed. + const tokens = events.filter((e) => e.type === 'agent:token'); + expect(tokens.map((e) => (e.type === 'agent:token' ? e.token : ''))).toEqual([ + 'first answer', + 'second answer', + ]); + }); + + it('accumulates a session-wide cumulative cost across turns (overwrites the turn-core placeholder)', async () => { + const { deps, events } = harness([textTurn('a'), textTurn('b')]); + const s = session(deps); + s.start(); + await s.sendMessage('one'); + await s.sendMessage('two'); + + const costs = events.filter((e) => e.type === 'cost:updated'); + expect(costs.length).toBeGreaterThanOrEqual(2); + // Each cumulative equals the running sum of per-event costMicrocents (never the 0 placeholder). + let running = 0; + for (const e of costs) { + if (e.type !== 'cost:updated') continue; + running += e.costMicrocents; + expect(e.cumulativeCostMicrocents).toBe(running); + } + }); + + it('emits session:turn_completed{turn_limit} — loudly, no egress — when driven past the hard cap', async () => { + // MANDATORY regression: a turn-loop refactor must not be able to silently drop the cap signal. + // maxTurns 1: turn 1 runs; turn 2 is blocked with turn_limit and never reaches the provider. + const { deps, events } = harness([textTurn('only answer')], { maxTurns: 1 }); + const s = session(deps); + s.start(); + await s.sendMessage('first'); + events.length = 0; // isolate the blocked turn's events + await s.sendMessage('second — over the cap'); + + const completed = events.find((e) => e.type === 'session:turn_completed'); + expect(completed?.type === 'session:turn_completed' ? completed.error?.code : undefined).toBe( + 'turn_limit', + ); + expect(completed?.type === 'session:turn_completed' ? completed.stopReason : undefined).toBe( + 'error', + ); + // No egress for the blocked turn: no streamed token / tool / cost. + expect(typesOf(events)).not.toContain('agent:token'); + expect(typesOf(events)).not.toContain('cost:updated'); + }); + + it('maps a within-turn turn_limit (maxToolTurns exceeded) to session:turn_completed{turn_limit}', async () => { + // maxToolTurns 0: a tool_use turn exceeds the within-turn loop guard → AgentTurnError('turn_limit'). + const { deps, events } = harness( + [toolUseTurn('c1')], + { limits: { maxToolTurns: 0, maxToolCorrections: 3 } }, + echoRegistry, + ); + const s = session(deps, TOOL_AGENT); + s.start(); + await s.sendMessage('loop please'); + + const completed = events.find((e) => e.type === 'session:turn_completed'); + expect(completed?.type === 'session:turn_completed' ? completed.error?.code : undefined).toBe( + 'turn_limit', + ); + }); + + it('completes a turn with an error (stopReason error) when the provider chain is exhausted', async () => { + const { deps, events } = harness([ + [ + { + type: 'error', + error: { kind: 'bad_request', retryable: false, provider: 'anthropic', message: 'nope' }, + }, + ], + ]); + const s = session(deps); + s.start(); + await s.sendMessage('go'); + + const completed = events.find((e) => e.type === 'session:turn_completed'); + expect(completed?.type === 'session:turn_completed' ? completed.stopReason : undefined).toBe( + 'error', + ); + expect(completed?.type === 'session:turn_completed' ? completed.error?.code : undefined).toBe( + 'validation', + ); + }); + + it('cancel() between turns emits session:cancelled and blocks further messages', async () => { + const { deps, events } = harness([textTurn('answer')]); + const s = session(deps); + s.start(); + await s.sendMessage('hi'); + s.cancel(); + + expect(typesOf(events)).toContain('session:cancelled'); + await expect(s.sendMessage('after cancel')).rejects.toBeInstanceOf(SessionStateError); + }); + + it('cancel() during an in-flight turn wins: session:cancelled is the terminal, no turn_completed', async () => { + const { deps, events } = harness([textTurn('streaming…')]); + const sink = deps.emit; + // The cancelling deps must reference the running session, which is constructed after them — a holder. + const ref: { session?: AgentSession } = {}; + let cancelled = false; + const cancellingDeps: SessionDeps = { + ...deps, + emit: (event) => { + sink(event); + // Cancel the moment the first token streams (the turn is in flight inside runAgentTurn). + if (event.type === 'agent:token' && !cancelled) { + cancelled = true; + ref.session?.cancel(); + } + }, + }; + const s = new AgentSession({ + sessionId: 'sess-2', + agentRef: AGENT.id, + agent: AGENT, + context: CONTEXT, + deps: cancellingDeps, + }); + ref.session = s; + s.start(); + await s.sendMessage('stream then cancel'); + + expect(typesOf(events)).toContain('session:cancelled'); + // The cancelled turn must NOT also emit a turn_completed (cancel owns the terminal). + expect(events.filter((e) => e.type === 'session:turn_completed')).toHaveLength(0); + }); + + it('guards the lifecycle: sendMessage before start, and double start, throw SessionStateError', async () => { + const { deps } = harness([textTurn('x')]); + const s = session(deps); + await expect(s.sendMessage('too early')).rejects.toBeInstanceOf(SessionStateError); + s.start(); + expect(() => { + s.start(); + }).toThrow(SessionStateError); + }); + + it('exposes a finite default hard cap', () => { + expect(DEFAULT_SESSION_MAX_TURNS).toBeGreaterThan(0); + expect(Number.isInteger(DEFAULT_SESSION_MAX_TURNS)).toBe(true); + }); +}); diff --git a/packages/core/src/engine/agent-session.ts b/packages/core/src/engine/agent-session.ts new file mode 100644 index 00000000..97dc6de7 --- /dev/null +++ b/packages/core/src/engine/agent-session.ts @@ -0,0 +1,406 @@ +/** + * `AgentSession` (1.V) — Relavium's **agent-first entry point**: an ongoing, multi-turn conversation + * bound to **one** agent + its fallback chain, a first-class peer of a workflow run that **reuses the + * same execution substrate** rather than a parallel implementation + * ([ADR-0024](../../../../docs/decisions/0024-agent-first-entry-point-agentsession.md), + * [agent-session-spec.md](../../../../docs/reference/contracts/agent-session-spec.md)). Each + * `sendMessage` drives **one turn through the same turn core** a workflow `agent` node uses + * ({@link runAgentTurn} — the correlation-agnostic core the `AgentRunner` (1.O) also wraps), so a + * session's streaming, tool-call loop, and provider fallback are identical to a node's; only the entry + * point and lifetime differ. + * + * **Scope (1.V).** This is the in-memory session DRIVER: the lifecycle (`start` / `sendMessage` / + * `cancel`), the conversation accumulator, the session-wide cost total, the hard turn cap, and mapping a + * classified {@link AgentTurnError} to a `session:turn_completed` error. It emits session events through + * an **injected {@link SessionEventSink}** — wiring that sink onto the shared `RunEventBus` (per-session + * `sequenceNumber` + gap/resync + a `SessionHandle`) is **1.W**; DB persistence + the durable + * `SessionMessage` schema is **1.X**; resume is **1.Y**; the export serializer is **1.Z**. 1.V keeps the + * transcript **in-memory** (`LlmMessage[]`, the in-flight content form) and persists nothing. + * + * The hard turn cap is the session's loud DoS fail-safe — distinct from `[chat].max_messages` (a + * history-**trim** threshold that silently continues; later phases) and from the turn core's within-turn + * `maxToolTurns` tool-loop guard. A `sendMessage` past the cap surfaces **loudly** with **no egress**: + * `session:turn_completed` carrying `error.code: 'turn_limit'` — never a silent stop. + */ + +import type { + Agent, + AbortSignalLike, + ErrorCode, + SessionContext, + SessionEvent, + StopReason, + ToolPolicy, +} from '@relavium/shared'; +import { + ToolDefSchema, + type FallbackPlanEntry, + type LlmMessage, + type LlmProvider, + type ProviderId, + type ToolDef as LlmToolDef, +} from '@relavium/llm'; + +import type { ToolDef, ToolDispatchContext, ToolRegistry } from '../tools/types.js'; +import { + AgentTurnError, + DEFAULT_AGENT_TURN_LIMITS, + runAgentTurn, + type AgentTurnLimits, + type AgentTurnResult, + type ChainCapabilities, + type PreEgressHook, +} from './agent-turn.js'; +import type { AbortControllerLike } from './execution-host.js'; +import type { NodeStreamEvent } from './node-executor.js'; + +/** The default hard turn cap when {@link SessionDeps.maxTurns} is omitted — a finite DoS fail-safe. */ +export const DEFAULT_SESSION_MAX_TURNS = 50; + +/** Distribute `Omit` across each union member so the discriminated union (and `.type` narrowing) survives. */ +type DistributiveOmit = T extends unknown ? Omit : never; + +/** + * A session lifecycle event body minus the envelope (`sessionId` / `timestamp` / `sequenceNumber`) the + * transport stamps — the session counterpart of {@link NodeStreamEvent}. 1.V produces the body; 1.W's + * bus wiring attaches the correlation key + the per-session monotonic sequence. + */ +export type SessionLifecycleEvent = DistributiveOmit< + SessionEvent, + 'sessionId' | 'timestamp' | 'sequenceNumber' +>; + +/** + * Everything a session emits, envelope-less: the five `session:*` lifecycle bodies plus the four + * dual-envelope in-turn bodies the turn core produces (`agent:token` / `agent:tool_call` / + * `agent:tool_result` / `cost:updated`). The injected sink receives these; 1.W routes them onto the bus. + */ +export type SessionStreamEvent = SessionLifecycleEvent | NodeStreamEvent; + +/** The injected emission port. 1.V emits through it; 1.W implements it over the shared `RunEventBus`. */ +export type SessionEventSink = (event: SessionStreamEvent) => void; + +/** + * The session's injected dependencies — **platform capabilities only**, mirroring `AgentRunnerDeps` + * (1.O). `resolveProvider` keeps an adapter from ever being imported by core; `keyFor` / `sleep` / + * `now` / `onAuthError` forward into the per-turn `FallbackChain`; `preEgress` is the ADR-0028 budget + * hook (no-op default; 1.AC supplies the estimator for both entry points through the same seam). + */ +export interface SessionDeps { + /** Resolve an authored provider id to its concrete adapter instance; `undefined` ⇒ a host-wiring gap. */ + readonly resolveProvider: (providerId: ProviderId) => LlmProvider | undefined; + /** The shared tool registry (1.T) the agent dispatches through (ADR-0037). */ + readonly registry: ToolRegistry; + /** The registry's tool defs — the source of the LLM-visible schema for the agent's granted tools. */ + readonly tools: readonly ToolDef[]; + /** Host credential resolver — forwarded into the chain; never logged / stored / inspected by core. */ + readonly keyFor: ChainCapabilities['keyFor']; + /** Host delay primitive (the engine has no ambient `setTimeout`). */ + readonly sleep: ChainCapabilities['sleep']; + /** Optional injectable clock for the chain's cooldown bookkeeping. */ + readonly now?: ChainCapabilities['now']; + /** Optional single out-of-band credential refresh (host-owned). */ + readonly onAuthError?: ChainCapabilities['onAuthError']; + /** Create a fresh abort controller per turn — injected so core never names the ambient global. */ + readonly newAbortController: () => AbortControllerLike; + /** The emission port — 1.V emits session/in-turn bodies here; 1.W wires it onto the `RunEventBus`. */ + readonly emit: SessionEventSink; + /** The workflow-wide tool policy threaded into dispatch (default `{}` ⇒ deny-all for gated tools). */ + readonly toolPolicy?: ToolPolicy; + /** Within-turn tool-loop bounds passed to the turn core (default {@link DEFAULT_AGENT_TURN_LIMITS}). */ + readonly limits?: AgentTurnLimits; + /** The session hard turn cap (default {@link DEFAULT_SESSION_MAX_TURNS}); 0/absent ⇒ the default. */ + readonly maxTurns?: number; + /** Pre-egress budget hook (default no-op; 1.AC fills it — ADR-0028). */ + readonly preEgress?: PreEgressHook; +} + +/** Construction params: a caller-minted `sessionId`, the bound agent + its `agentRef`, the context, deps. */ +export interface AgentSessionParams { + /** Process-unique id (the caller mints it via the host id source); carried on every session event. */ + readonly sessionId: string; + /** The agent's authored id — recorded on `session:started` and used as the in-turn event `nodeId`. */ + readonly agentRef: string; + /** The resolved agent config the session binds for its whole lifetime (no mid-session switching). */ + readonly agent: Agent; + /** The workspace situation the session runs against (fs-scope tier, working dir, ctx variables). */ + readonly context: SessionContext; + readonly deps: SessionDeps; +} + +type SessionStatus = 'created' | 'idle' | 'running' | 'cancelled'; + +/** A typed, secret-free session API-misuse error (never a raw string throw — error-handling.md). */ +export class SessionStateError extends Error { + override readonly name = 'SessionStateError'; + constructor( + readonly code: 'not_started' | 'already_started' | 'not_active', + message: string, + ) { + super(message); + } +} + +type PlanResult = + | { readonly ok: true; readonly entries: readonly FallbackPlanEntry[] } + | { readonly ok: false; readonly message: string }; + +/** + * Drive a multi-turn agent conversation over the shared turn core. Construct with a caller-minted + * `sessionId`, call {@link start} once, then {@link sendMessage} per user turn; {@link cancel} aborts an + * in-flight turn. Events flow through the injected {@link SessionEventSink}. + */ +export class AgentSession { + readonly sessionId: string; + + readonly #agentRef: string; + readonly #agent: Agent; + readonly #context: SessionContext; + readonly #deps: SessionDeps; + readonly #maxTurns: number; + readonly #limits: AgentTurnLimits; + + /** The in-memory conversation (in-flight `ContentPart` form); the turn core copies it, never mutates it. */ + readonly #messages: LlmMessage[] = []; + /** Turns that engaged the provider (success or failure) — the hard cap counts these. */ + #turnCount = 0; + /** Session-wide running cost total, authoritatively stamped onto every `cost:updated`. */ + #cumulativeCostMicrocents = 0; + #status: SessionStatus = 'created'; + /** The in-flight turn's controller, so {@link cancel} can abort it; `undefined` between turns. */ + #abort: AbortControllerLike | undefined; + /** Memoized provider fallback plan (the agent binding is fixed for the session). */ + #plan: PlanResult | undefined; + + constructor(params: AgentSessionParams) { + this.sessionId = params.sessionId; + this.#agentRef = params.agentRef; + this.#agent = params.agent; + this.#context = params.context; + this.#deps = params.deps; + const max = params.deps.maxTurns; + this.#maxTurns = max === undefined || max <= 0 ? DEFAULT_SESSION_MAX_TURNS : max; + this.#limits = params.deps.limits ?? DEFAULT_AGENT_TURN_LIMITS; + } + + /** Open the session: emit `session:started` and move to idle. Idempotent-guarded (one start per session). */ + start(): void { + if (this.#status !== 'created') { + throw new SessionStateError( + 'already_started', + `session ${this.sessionId} is already started`, + ); + } + this.#status = 'idle'; + this.#deps.emit({ + type: 'session:started', + agentRef: this.#agentRef, + model: this.#agent.model, + context: this.#context, + }); + } + + /** + * Run one user turn end to end: append the user message, drive the turn core (streaming + tool loop + + * fallback), append the assistant reply, and emit `session:turn_started` → `session:turn_completed`. + * A turn past the hard cap is blocked **loudly** with `turn_limit` and **no egress**. A classified + * turn failure still **completes** — with `stopReason: 'error'` and the mapped error code. Resolves + * when the turn settles; a cancel mid-turn resolves quietly (the terminal is `session:cancelled`). + */ + async sendMessage(text: string): Promise { + if (this.#status === 'created') { + throw new SessionStateError('not_started', `session ${this.sessionId}: call start() first`); + } + if (this.#status !== 'idle') { + throw new SessionStateError( + 'not_active', + `session ${this.sessionId} is ${this.#status}; cannot send a message`, + ); + } + + this.#status = 'running'; + this.#deps.emit({ type: 'session:turn_started' }); + + // Hard turn cap — checked AFTER turn_started (the turn was attempted) but BEFORE any egress: the + // blocked turn completes loudly with turn_limit and never calls a provider. + if (this.#turnCount >= this.#maxTurns) { + this.#status = 'idle'; + this.#emitTurnCompleted( + 'error', + { input: 0, output: 0 }, + { + code: 'turn_limit', + message: `session reached its hard cap of ${this.#maxTurns} turns`, + retryable: false, + }, + ); + return; + } + + this.#messages.push({ role: 'user', content: [{ type: 'text', text }] }); + const abort = this.#deps.newAbortController(); + this.#abort = abort; + try { + const result = await this.#runTurn(abort.signal); + // A cancel landed mid-turn — the cancel path owns the terminal session:cancelled; stay quiet. + if (this.#statusIs('cancelled')) return; + this.#turnCount += 1; + this.#messages.push({ role: 'assistant', content: [...result.content] }); + this.#emitTurnCompleted(result.stopReason, { + input: result.usage.input, + output: result.usage.output, + model: result.model, + }); + } catch (err) { + if (this.#statusIs('cancelled')) return; // cancel-during-turn: session:cancelled is the terminal + if (err instanceof AgentTurnError) { + this.#turnCount += 1; // the turn engaged a provider — it counts toward the cap + this.#emitTurnCompleted( + 'error', + { input: 0, output: 0 }, + { + code: err.code, + message: err.message, + retryable: err.retryable, + }, + ); + } else { + throw err; // unexpected — let it surface; the session is left in a running state for the caller + } + } finally { + this.#abort = undefined; + if (this.#statusIs('running')) this.#status = 'idle'; + } + } + + /** + * Read the current status WITHOUT control-flow narrowing — `#status` can change across an `await` in + * {@link sendMessage} (a {@link cancel} fired from the emit sink), which TS cannot model after the literal + * `this.#status = 'running'` assignment. Routing post-await reads through a call keeps them sound. + */ + #statusIs(status: SessionStatus): boolean { + return this.#status === status; + } + + /** Abort an in-flight turn (if any) and end the session, emitting `session:cancelled`. Idempotent. */ + cancel(): void { + if (this.#status === 'cancelled') return; + this.#status = 'cancelled'; + this.#abort?.abort(); + this.#abort = undefined; + this.#deps.emit({ type: 'session:cancelled' }); + } + + /** Build (memoized) the fallback plan and drive one turn through the shared core. */ + async #runTurn(signal: AbortSignalLike): Promise { + const plan = this.#resolvePlan(); + if (!plan.ok) { + // A host-wiring gap (a provider was not resolved) — a classified, non-retryable internal failure. + throw new AgentTurnError('internal', plan.message, false); + } + const grantedToolIds = new Set(this.#agent.tools ?? []); + const dispatchContext: Omit = { + nodeId: this.#agentRef, + grantedToolIds, + config: {}, // an agent-invoked tool carries no per-tool config block in v1.0 + toolPolicy: this.#deps.toolPolicy ?? {}, + fsScope: this.#context.fsScopeTier, + gateApproved: false, // a chat loop provides no human gate — git_commit stays denied (parity with 1.O) + }; + const llmTools = buildLlmTools(this.#deps.tools, grantedToolIds); + return runAgentTurn({ + system: this.#agent.system_prompt, + messages: this.#messages, + ...(llmTools.length > 0 ? { tools: llmTools } : {}), + planEntries: plan.entries, + chainCapabilities: this.#chainCapabilities(), + ...(this.#agent.temperature === undefined ? {} : { temperature: this.#agent.temperature }), + ...(this.#agent.max_tokens === undefined ? {} : { maxTokens: this.#agent.max_tokens }), + nodeId: this.#agentRef, + emit: (event) => { + this.#onTurnEmit(event); + }, + signal, + registry: this.#deps.registry, + dispatchContext, + limits: this.#limits, + ...(this.#deps.preEgress === undefined ? {} : { preEgress: this.#deps.preEgress }), + }); + } + + /** Forward an in-turn streaming body, stamping the session-wide running total onto `cost:updated`. */ + #onTurnEmit(event: NodeStreamEvent): void { + if (event.type === 'cost:updated') { + this.#cumulativeCostMicrocents += event.costMicrocents; + this.#deps.emit({ ...event, cumulativeCostMicrocents: this.#cumulativeCostMicrocents }); + return; + } + this.#deps.emit(event); + } + + #emitTurnCompleted( + stopReason: StopReason, + tokensUsed: { input: number; output: number; model?: string }, + error?: { code: ErrorCode; message: string; retryable: boolean }, + ): void { + this.#deps.emit({ + type: 'session:turn_completed', + stopReason, + tokensUsed, + ...(error === undefined ? {} : { error }), + }); + } + + /** Build the ordered fallback plan once: primary (maxAttempts 1, ADR-0040) + each authored fallback. */ + #resolvePlan(): PlanResult { + if (this.#plan !== undefined) return this.#plan; + const agent = this.#agent; + const primary = this.#deps.resolveProvider(agent.provider); + if (primary === undefined) { + this.#plan = { ok: false, message: `no provider wired for '${agent.provider}'` }; + return this.#plan; + } + // The primary entry does NOT consume a node/agent retry budget — ADR-0040 makes node retry the + // engine's ABOVE-chain budget (a session has no such loop in 1.V); the primary is a single attempt. + const entries: FallbackPlanEntry[] = [ + { provider: primary, model: agent.model, maxAttempts: 1 }, + ]; + for (const entry of agent.fallback_chain ?? []) { + const provider = this.#deps.resolveProvider(entry.provider); + if (provider === undefined) { + this.#plan = { ok: false, message: `no provider wired for fallback '${entry.provider}'` }; + return this.#plan; + } + entries.push({ provider, model: entry.model, maxAttempts: entry.max_attempts }); + } + this.#plan = { ok: true, entries }; + return this.#plan; + } + + /** Forward only the platform-level chain capabilities the host supplies (mirrors the 1.O runner). */ + #chainCapabilities(): ChainCapabilities { + const deps = this.#deps; + return { + keyFor: deps.keyFor, + sleep: deps.sleep, + ...(deps.now === undefined ? {} : { now: deps.now }), + ...(deps.onAuthError === undefined ? {} : { onAuthError: deps.onAuthError }), + }; + } +} + +/** The agent's granted tools as LLM-visible defs, validated through the seam schema (no unsafe cast). */ +function buildLlmTools(defs: readonly ToolDef[], granted: ReadonlySet): LlmToolDef[] { + const out: LlmToolDef[] = []; + for (const def of defs) { + if (!granted.has(def.id)) continue; + out.push( + ToolDefSchema.parse({ + name: def.id, + ...(def.description.length > 0 ? { description: def.description } : {}), + parameters: def.llmVisibleParams, + }), + ); + } + return out; +} diff --git a/packages/core/src/engine/agent-turn.ts b/packages/core/src/engine/agent-turn.ts index 87c27229..6b6fe721 100644 --- a/packages/core/src/engine/agent-turn.ts +++ b/packages/core/src/engine/agent-turn.ts @@ -44,6 +44,7 @@ import { import { ToolDispatchError } from '../tools/errors.js'; import type { ToolCallPart, ToolDispatchContext, ToolRegistry } from '../tools/types.js'; import { unwrapUntrusted } from '../tools/untrusted.js'; +import { BudgetExceededError, BudgetPauseError } from './budget-governor.js'; import type { NodeStreamEvent } from './node-executor.js'; /** Loop bounds for one agent turn. The authored hard cap + the `turn_limit` surfacing is the 1.V knob. */ @@ -65,7 +66,10 @@ export const DEFAULT_AGENT_TURN_LIMITS: AgentTurnLimits = { * called immediately before every seam call. In 1.O the default is a no-op that always permits; 1.AC * replaces it with the estimator and may throw an {@link AgentTurnError} (`budget_exceeded`) to halt. */ -export type PreEgressHook = (info: { readonly model: string }) => void | Promise; +export type PreEgressHook = (info: { + readonly model: string; + readonly maxTokens?: number; +}) => void | Promise; /** The chain capabilities the host supplies (the platform-level subset of {@link FallbackChainOptions}). */ export type ChainCapabilities = Pick< @@ -265,6 +269,17 @@ async function streamOneTurn( for await (const chunk of chain.stream(buildRequest(messages, params))) { foldChunk(chunk, acc, params, getModel); if (chunk.type === 'error') { + // A pre-egress budget hook may throw its own AgentTurnError or Budget*Error; preserve it + // rather than remapping the wrapped LlmError to a generic internal code. + if (chunk.error.cause instanceof AgentTurnError) { + throw chunk.error.cause; + } + if (chunk.error.cause instanceof BudgetExceededError) { + throw new AgentTurnError('budget_exceeded', chunk.error.cause.message, false); + } + if (chunk.error.cause instanceof BudgetPauseError) { + throw chunk.error.cause; + } throw new AgentTurnError( codeForLlmError(chunk.error), chunk.error.message, @@ -467,6 +482,9 @@ export async function runAgentTurn(params: AgentTurnParams): Promise ({ @@ -484,7 +502,22 @@ export async function runAgentTurn(params: AgentTurnParams): Promise { + const budget: Budget = { max_cost_microcents: 1_000_000, on_exceed: 'warn' }; + + function makeGovernor( + overrides: { budget?: Budget; defaultMaxTokensEstimate?: number } = {}, + ): { + governor: BudgetGovernor; + warnings: Omit, 'runId'>[]; + } { + const warnings: Omit, 'runId'>[] = []; + const governor = new BudgetGovernor({ + budget: overrides.budget ?? budget, + ...(overrides.defaultMaxTokensEstimate === undefined + ? {} + : { defaultMaxTokensEstimate: overrides.defaultMaxTokensEstimate }), + emit: (event) => { + warnings.push(event); + return Promise.resolve(); + }, + }); + return { governor, warnings }; + } + + it('allows a call whose estimate stays within the cap', async () => { + const { governor, warnings } = makeGovernor(); + governor.updateCost(0); + await expect(governor.checkPreEgress('claude-haiku-4-5', 1000)).resolves.toBeUndefined(); + expect(warnings).toHaveLength(0); + }); + + it('warns once when the projected cost exceeds the cap', async () => { + const { governor, warnings } = makeGovernor(); + governor.updateCost(900_000); + await governor.checkPreEgress('claude-sonnet-4-6', 10_000); + expect(warnings).toHaveLength(1); + expect(warnings[0]?.thresholdPct).toBe(90); + // A second check does not emit another warning. + await governor.checkPreEgress('claude-sonnet-4-6', 10_000); + expect(warnings).toHaveLength(1); + }); + + it('fails when on_exceed is fail', async () => { + const { governor } = makeGovernor({ budget: { ...budget, on_exceed: 'fail' } }); + governor.updateCost(900_000); + await expect(governor.checkPreEgress('claude-sonnet-4-6', 10_000)).rejects.toBeInstanceOf( + BudgetExceededError, + ); + }); + + it('pauses when on_exceed is pause_for_approval', async () => { + const { governor } = makeGovernor({ budget: { ...budget, on_exceed: 'pause_for_approval' } }); + governor.updateCost(900_000); + const err = await governor + .checkPreEgress('claude-sonnet-4-6', 10_000) + .catch((e: unknown) => e); + expect(err).toBeInstanceOf(BudgetPauseError); + const gate = (err as BudgetPauseError).toGateRequest(); + expect(gate.gateType).toBe('approval'); + expect(gate.message).toContain('budget cap'); + expect(gate.spentMicrocents).toBe(900_000); + expect(gate.limitMicrocents).toBe(1_000_000); + }); + + it('uses the default max_tokens_estimate when maxTokens is omitted', async () => { + const { governor, warnings } = makeGovernor({ + budget: { ...budget, on_exceed: 'warn' }, + defaultMaxTokensEstimate: 1, + }); + // At exactly the cap minus the default estimate, the call is allowed. + governor.updateCost(1_000_000 - 500); // haiku output is 500_000_000 micro-cents/MTok + await governor.checkPreEgress('claude-haiku-4-5', undefined); + expect(warnings).toHaveLength(0); + }); + + it('clamps thresholdPct to [0, 100]', async () => { + const { governor, warnings } = makeGovernor(); + governor.updateCost(2_000_000); + await governor.checkPreEgress('claude-sonnet-4-6', 1000); + expect(warnings[0]?.thresholdPct).toBe(100); + }); +}); diff --git a/packages/core/src/engine/budget-governor.ts b/packages/core/src/engine/budget-governor.ts new file mode 100644 index 00000000..cd3e95a8 --- /dev/null +++ b/packages/core/src/engine/budget-governor.ts @@ -0,0 +1,170 @@ +import { estimateMaxNextCost } from '@relavium/llm'; +import type { Budget } from '@relavium/shared'; + +import type { RunEventDraft } from './event-bus.js'; +import type { GateRequest } from './node-executor.js'; + +/** + * Default per-call output-token estimate used by the pre-egress budget governor when neither the + * node/session nor the host config supplies `max_tokens_estimate` (ADR-0028). The canonical value + * is deliberately conservative: it is a safety rail, not a performance target. + */ +export const DEFAULT_MAX_TOKENS_ESTIMATE = 4096; + +/** + * Thrown when a pre-egress check would exceed the configured cost cap and `on_exceed: fail` is set. + * The turn/run adapter maps this to the `budget_exceeded` `ErrorCode`. + */ +export class BudgetExceededError extends Error { + override readonly name = 'BudgetExceededError'; + constructor( + readonly spentMicrocents: number, + readonly limitMicrocents: number, + readonly projectedMicrocents: number, + ) { + super( + `pre-egress budget check failed: projected ${projectedMicrocents} micro-cents exceeds ` + + `the cap of ${limitMicrocents} micro-cents (spent ${spentMicrocents})`, + ); + } +} + +/** + * Thrown when a pre-egress check would exceed the configured cost cap and + * `on_exceed: pause_for_approval` is set. The agent runner maps this to a `paused` node outcome + * (reusing the human-gate seam) so the run can be resumed via `engine.resume(runId, gateId, decision)`. + */ +export class BudgetPauseError extends Error { + override readonly name = 'BudgetPauseError'; + constructor( + readonly spentMicrocents: number, + readonly limitMicrocents: number, + readonly thresholdPct: number, + ) { + super( + `pre-egress budget check would exceed the cap of ${limitMicrocents} micro-cents ` + + `(spent ${spentMicrocents}); run paused for approval`, + ); + } + + /** + * Build a `GateRequest` the engine can park like a human gate. The engine assigns the stable + * `gateId` when it persists `budget:paused`. + */ + toGateRequest(): GateRequest { + return { + gateType: 'approval', + message: `The next LLM call would push the run past its budget cap of ${this.limitMicrocents} micro-cents ` + + `(already spent ${this.spentMicrocents}). Approve to continue anyway; reject to fail the run with budget_exceeded.`, + spentMicrocents: this.spentMicrocents, + limitMicrocents: this.limitMicrocents, + isBudgetGate: true, + }; + } +} + +/** What the governor decided at a pre-egress check. */ +export type BudgetCheckResult = + | { readonly kind: 'allow' } + | { readonly kind: 'warn'; readonly spentMicrocents: number; readonly limitMicrocents: number; readonly thresholdPct: number } + | { readonly kind: 'fail'; readonly error: BudgetExceededError } + | { readonly kind: 'pause'; readonly error: BudgetPauseError }; + +/** + * The pre-egress budget governor (ADR-0028, 1.AC). It is stateful per run: it tracks the current + * cumulative cost, emits at most one `budget:warning` event per run, and throws a typed error for + * `fail` / `pause_for_approval`. All cost figures are integer micro-cents. + */ +export class BudgetGovernor { + readonly #budget: Budget; + readonly #defaultMaxTokensEstimate: number; + readonly #emit: (event: Omit, 'runId'>) => Promise; + #cumulativeCostMicrocents = 0; + #warningEmitted = false; + + constructor(params: { + readonly budget: Budget; + readonly defaultMaxTokensEstimate?: number; + readonly emit: (event: Omit, 'runId'>) => Promise; + }) { + this.#budget = params.budget; + this.#defaultMaxTokensEstimate = params.defaultMaxTokensEstimate ?? DEFAULT_MAX_TOKENS_ESTIMATE; + this.#emit = params.emit; + } + + /** Update the governor with the engine's authoritative running cumulative cost. */ + updateCost(cumulativeCostMicrocents: number): void { + this.#cumulativeCostMicrocents = cumulativeCostMicrocents; + } + + /** + * Evaluate one prospective LLM call. Returns a result description (synchronous, no side effects); + * callers apply the action by throwing the supplied error or, for `warn`, emitting the event. + */ + evaluatePreEgress(model: string, maxTokens: number | undefined): BudgetCheckResult { + const estimate = estimateMaxNextCost(model, maxTokens ?? this.#defaultMaxTokensEstimate); + const projected = this.#cumulativeCostMicrocents + estimate; + if (projected <= this.#budget.max_cost_microcents) { + return { kind: 'allow' }; + } + + const thresholdPct = clampPct( + Math.round((this.#cumulativeCostMicrocents / this.#budget.max_cost_microcents) * 100), + ); + + switch (this.#budget.on_exceed) { + case 'warn': + return { + kind: 'warn', + spentMicrocents: this.#cumulativeCostMicrocents, + limitMicrocents: this.#budget.max_cost_microcents, + thresholdPct, + }; + case 'fail': + return { + kind: 'fail', + error: new BudgetExceededError( + this.#cumulativeCostMicrocents, + this.#budget.max_cost_microcents, + projected, + ), + }; + case 'pause_for_approval': + return { + kind: 'pause', + error: new BudgetPauseError( + this.#cumulativeCostMicrocents, + this.#budget.max_cost_microcents, + thresholdPct, + ), + }; + } + } + + /** + * Apply the pre-egress check: emits a one-time `budget:warning` on the warn path, returns on allow, + * and throws `BudgetExceededError` / `BudgetPauseError` for fail / pause. Async because the warning + * event is durable. + */ + async checkPreEgress(model: string, maxTokens: number | undefined): Promise { + const result = this.evaluatePreEgress(model, maxTokens); + if (result.kind === 'allow') return; + if (result.kind === 'warn') { + if (!this.#warningEmitted) { + this.#warningEmitted = true; + await this.#emit({ + type: 'budget:warning', + spentMicrocents: result.spentMicrocents, + limitMicrocents: result.limitMicrocents, + thresholdPct: result.thresholdPct, + }); + } + return; + } + throw result.error; + } +} + +function clampPct(value: number): number { + return Math.max(0, Math.min(100, value)); +} diff --git a/packages/core/src/engine/checkpoint.test.ts b/packages/core/src/engine/checkpoint.test.ts index 331602d1..fd68c361 100644 --- a/packages/core/src/engine/checkpoint.test.ts +++ b/packages/core/src/engine/checkpoint.test.ts @@ -97,7 +97,7 @@ describe('reconstructCheckpointState', () => { ]); expect(state?.runStatus).toBe('paused'); expect(state?.nodeStates.get('gate')).toEqual({ status: 'paused' }); - expect(state?.pendingGates).toEqual([{ gateId: 'g1', nodeId: 'gate' }]); + expect(state?.pendingGates).toEqual([{ gateId: 'g1', nodeId: 'gate', isBudgetGate: false }]); }); it('a resumed gate clears the pending gate + records the decision as the node output', () => { diff --git a/packages/core/src/engine/checkpoint.ts b/packages/core/src/engine/checkpoint.ts index 406adaf8..e3ff3c92 100644 --- a/packages/core/src/engine/checkpoint.ts +++ b/packages/core/src/engine/checkpoint.ts @@ -37,6 +37,8 @@ export interface CheckpointNodeState { export interface CheckpointPendingGate { readonly gateId: string; readonly nodeId: string; + /** True for a budget gate, so a rejected decision can still fail the run on resume. */ + readonly isBudgetGate: boolean; } /** The derived state a rehydrating run is rebuilt from — never a persisted blob (reconstructed from rows). */ @@ -86,7 +88,7 @@ interface ReconAccumulator { totalOutputTokens: number; cumulativeCostMicrocents: number; readonly nodeStates: Map; - readonly pendingGates: Map; // gateId → nodeId + readonly pendingGates: Map; readonly resolvedGateIds: Set; } @@ -144,11 +146,14 @@ function applyNodeEvent(acc: ReconAccumulator, event: RunEvent): void { } } -/** Human-gate lifecycle: park a pending gate, or resolve it (decision becomes the gate vertex output). */ +/** Human-gate / budget-gate lifecycle: park a pending gate, or resolve it (decision becomes the gate vertex output). */ function applyGateEvent(acc: ReconAccumulator, event: RunEvent): void { - if (event.type === 'human_gate:paused') { + if (event.type === 'human_gate:paused' || event.type === 'budget:paused') { acc.nodeStates.set(event.nodeId, { status: 'paused' }); - acc.pendingGates.set(event.gateId, event.nodeId); + acc.pendingGates.set(event.gateId, { + nodeId: event.nodeId, + isBudgetGate: event.type === 'budget:paused', + }); return; } if (event.type !== 'human_gate:resumed') { @@ -161,7 +166,7 @@ function applyGateEvent(acc: ReconAccumulator, event: RunEvent): void { }); // Collect this gate's pending ids first, then mutate — never delete while iterating the Map. const resolvedForNode = [...acc.pendingGates] - .filter(([, nodeId]) => nodeId === event.nodeId) + .filter(([, entry]) => entry.nodeId === event.nodeId) .map(([gateId]) => gateId); for (const gateId of resolvedForNode) { acc.pendingGates.delete(gateId); @@ -215,7 +220,11 @@ export function reconstructCheckpointState( startedAtMs: acc.startedAtMs, nodeStates: acc.nodeStates, completedNodeIds, - pendingGates: [...acc.pendingGates].map(([gateId, nodeId]) => ({ gateId, nodeId })), + pendingGates: [...acc.pendingGates].map(([gateId, entry]) => ({ + gateId, + nodeId: entry.nodeId, + isBudgetGate: entry.isBudgetGate, + })), resolvedGateIds: [...acc.resolvedGateIds], lastSequenceNumber: acc.lastSequenceNumber, totalInputTokens: acc.totalInputTokens, diff --git a/packages/core/src/engine/engine.ts b/packages/core/src/engine/engine.ts index 497ed340..2bd21855 100644 --- a/packages/core/src/engine/engine.ts +++ b/packages/core/src/engine/engine.ts @@ -49,6 +49,7 @@ import type { PlanVertex, RunPlan } from '../run-plan.js'; import type { WorkflowDefinition } from '../parser.js'; import { EngineStateError } from './errors.js'; import { RunEventBus, type RunEventDraft } from './event-bus.js'; +import { BudgetGovernor, DEFAULT_MAX_TOKENS_ESTIMATE } from './budget-governor.js'; import type { CheckpointState } from './checkpoint.js'; import type { AbortControllerLike, ExecutionHost } from './execution-host.js'; import type { @@ -150,6 +151,11 @@ export interface WorkflowEngineDeps { * in a `context:` value injects it here — the same purity seam the node handlers use; omit ⇒ no `read_file`. */ readonly resolverCapabilities?: ResolverCapabilities; + /** + * Per-call output-token default the pre-egress budget governor uses when a node/session omits + * `maxTokens` (ADR-0028). Not the model's absolute max, which would over-block. + */ + readonly maxTokensEstimate?: number; } function maskInputs( @@ -188,16 +194,21 @@ class RunExecution { readonly #bus: RunEventBus; readonly #onSettled: (runId: string) => void; readonly #resolverCapabilities: ResolverCapabilities; + readonly #maxTokensEstimate: number; /** The resolved workflow `context:` (`ctx.*`), folded once at run start (or re-resolved on resume). */ #resolvedContext: Readonly> = {}; readonly #abort: AbortControllerLike; readonly #states = new Map(); - readonly #pendingGates = new Map(); + readonly #pendingGates = new Map(); /** Gate ids whose decision was already applied — a re-delivery is an idempotent no-op (1.R). */ readonly #resolvedGates = new Set(); /** Disarm callbacks for armed gate-timeout timers, by gateId — disarmed on resume / settle (1.Q). */ readonly #gateTimers = new Map void>(); + /** The run-level wall-clock timeout timer, when a `timeout_ms` is configured (ADR-0028). */ + #runTimeoutDisarm: (() => void) | undefined; + /** The pre-egress budget governor, when a workflow `budget` is configured (ADR-0028, 1.AC). */ + #budgetGovernor: BudgetGovernor | undefined; #workflowId = ''; #settled = false; @@ -225,6 +236,7 @@ class RunExecution { capacity: number; onSettled: (runId: string) => void; resolverCapabilities: ResolverCapabilities; + maxTokensEstimate?: number; /** When present, the run is REHYDRATED from this checkpoint (resume) rather than started fresh (1.R). */ checkpoint?: CheckpointState; }) { @@ -247,6 +259,14 @@ class RunExecution { ); this.#secretInputNames = secretNames; this.#maskedInputs = maskInputs(params.inputs, secretNames); + this.#maxTokensEstimate = params.maxTokensEstimate ?? DEFAULT_MAX_TOKENS_ESTIMATE; + if (params.plan.budget !== undefined) { + this.#budgetGovernor = new BudgetGovernor({ + budget: params.plan.budget, + defaultMaxTokensEstimate: this.#maxTokensEstimate, + emit: (draft) => this.#emitDurable({ ...draft, runId: this.runId }), + }); + } if (params.checkpoint === undefined) { for (const id of params.plan.vertices.keys()) { @@ -278,6 +298,7 @@ class RunExecution { async begin(): Promise { this.#startEpochMs = Date.parse(this.#host.clock.now()); + this.#armRunTimeout(); try { this.#workflowId = await this.#host.store.resolveWorkflowId(this.#workflow.workflow.id); await this.#emitDurable({ @@ -302,6 +323,53 @@ class RunExecution { this.#schedule(); } + /** + * Arm the run-level `timeout_ms` timer (ADR-0028). Idempotent: disarms any prior timer first. The + * timer fires once and fails the run with `run_timeout` if it elapses before the run settles. + */ + #armRunTimeout(): void { + this.#disarmRunTimeout(); + const timeoutMs = this.#plan.timeoutMs; + if (timeoutMs === undefined) { + return; + } + this.#runTimeoutDisarm = this.#host.setTimer(timeoutMs, () => { + void this.#onRunTimeout(timeoutMs); + }); + } + + #disarmRunTimeout(): void { + if (this.#runTimeoutDisarm !== undefined) { + this.#runTimeoutDisarm(); + this.#runTimeoutDisarm = undefined; + } + } + + async #onRunTimeout(timeoutMs: number): Promise { + if (this.#settled) { + return; + } + this.#disarmRunTimeout(); + const elapsedMs = this.#elapsedMs(); + await this.#emitDurable({ + type: 'run:timeout', + runId: this.runId, + elapsedMs, + timeoutMs, + }); + if (!this.#settled) { + this.#failure = { + error: { + code: 'run_timeout', + message: `the run exceeded its ${timeoutMs} ms timeout`, + retryable: false, + }, + }; + this.#abort.abort(); + this.#schedule(); + } + } + /** * Resolve the workflow `context:` map into the frozen `ctx.*` namespace (the spec's eager-once context), * threaded to every node via {@link NodeExecContext.ctx}. Returns `false` on failure: a cancel mid-resolve @@ -356,7 +424,10 @@ class RunExecution { // applied immediately. Re-arming a *remaining* gate's deadline is deferred to the Phase-2 // crash-reconciliation that re-arms from persisted policy + a real clock (shared-core-engine.md) — // the data it needs (timeoutAction + expiresAt) is now carried on `human_gate:paused`, so no backfill. - this.#pendingGates.set(gate.gateId, { vertexId: gate.nodeId }); + this.#pendingGates.set(gate.gateId, { + vertexId: gate.nodeId, + isBudgetGate: gate.isBudgetGate, + }); } for (const gateId of cp.resolvedGateIds) { this.#resolvedGates.add(gateId); @@ -390,6 +461,7 @@ class RunExecution { gateAlreadyResolved: boolean, ): Promise { // #startEpochMs was seeded from the checkpoint in #seedFromCheckpoint (preserves total durationMs). + this.#armRunTimeout(); if (!gateAlreadyResolved) { this.#assertGatePending(gateId); // fail fast on a bad gateId, before any context side effect } @@ -419,7 +491,7 @@ class RunExecution { } /** The pending gate for `gateId`, or throw the typed misuse (`run_not_paused` / `unknown_gate`). */ - #assertGatePending(gateId: string): { readonly vertexId: string } { + #assertGatePending(gateId: string): { readonly vertexId: string; readonly isBudgetGate: boolean } { if (this.#pendingGates.size === 0) { throw new EngineStateError('run_not_paused', 'the run has no pending gate to resume', { runId: this.runId, @@ -454,6 +526,35 @@ class RunExecution { this.#pendingGates.delete(gateId); this.#disarmTimer(gateId); // a decision arrived before the timeout — cancel the armed timer (1.Q) this.#pauseEpisode = false; // a later idle-with-gates re-emits run:paused for the remaining gates + + // A rejected budget gate is a run-level budget failure, not a completed gate vertex. + if (gate.isBudgetGate && decision.decision === 'rejected') { + const state = this.#states.get(gate.vertexId); + if (state !== undefined) { + state.status = 'failed'; + } + if (this.#failure === undefined && !this.#cancelling) { + this.#failure = { + nodeId: gate.vertexId, + error: { + code: 'budget_exceeded', + message: 'the budget gate was rejected', + retryable: false, + }, + }; + this.#abort.abort(); + } + await this.#emitDurable({ + type: 'human_gate:resumed', + runId: this.runId, + nodeId: gate.vertexId, + decision: 'rejected', + decidedBy: decision.decidedBy, + }); + this.#schedule(); + return; + } + // Mark the gate vertex completed SYNCHRONOUSLY before the await — mirroring #settleCompleted — so a // concurrent #step (e.g. a sibling gate's timeout firing during this persist) never sees this gate as // still `paused` while it is already out of #pendingGates, which would mis-read the run as stalled. @@ -667,9 +768,23 @@ class RunExecution { } } + /** + * Build the pre-egress hook for this run, when a budget is configured. The hook is stateful: + * it sees the run's current cumulative cost and may emit a one-time `budget:warning` or throw + * `BudgetExceededError` / `BudgetPauseError` for `fail` / `pause_for_approval`. + */ + #makePreEgressHook(): import('./agent-turn.js').PreEgressHook | undefined { + if (this.#budgetGovernor === undefined) { + return undefined; + } + const governor = this.#budgetGovernor; + return (info) => governor.checkPreEgress(info.model, info.maxTokens); + } + /** Run one attempt of a vertex; returns its outcome (an uncaught handler throw → a single `internal`). */ async #runAttempt(vertex: PlanVertex, attemptNumber: number): Promise { try { + const preEgress = this.#makePreEgressHook(); const ctx: NodeExecContext = { vertex, runOutputs: this.#completedOutputs(), @@ -682,6 +797,7 @@ class RunExecution { }, signal: this.#abort.signal, attemptNumber, + ...(preEgress === undefined ? {} : { preEgress }), }; return await this.#executor.execute(ctx); } catch { @@ -851,11 +967,12 @@ class RunExecution { /** A `paused` outcome: park the gate, arm its timeout timer (1.Q), and emit `human_gate:paused`. */ async #settlePaused(vertex: PlanVertex, gate: GateRequest): Promise { const gateId = gate.gateId ?? this.#host.ids.newId(); + const isBudgetGate = gate.isBudgetGate === true; const state = this.#states.get(vertex.id); if (state !== undefined) { state.status = 'paused'; } - this.#pendingGates.set(gateId, { vertexId: vertex.id }); + this.#pendingGates.set(gateId, { vertexId: vertex.id, isBudgetGate }); // Compute the wall-clock deadline from the host clock (the handler has none) and arm a one-shot timer // (1.Q). On fire, an `approve` action auto-resolves the gate; a `reject` (the safe default) fails the // run with run_timeout. The timer is disarmed on resume / terminal settle so it never fires twice. @@ -876,6 +993,16 @@ class RunExecution { }); this.#gateTimers.set(gateId, disarm); } + if (gate.spentMicrocents !== undefined && gate.limitMicrocents !== undefined) { + await this.#emitDurable({ + type: 'budget:paused', + runId: this.runId, + nodeId: vertex.id, + gateId, + spentMicrocents: gate.spentMicrocents, + limitMicrocents: gate.limitMicrocents, + }); + } await this.#emitDurable({ type: 'human_gate:paused', runId: this.runId, @@ -975,6 +1102,7 @@ class RunExecution { disarm(); } this.#gateTimers.clear(); + this.#disarmRunTimeout(); const durationMs = Math.max(0, this.#elapsedMs()); let draft: RunEventDraft; if (type === 'run:completed') { @@ -1146,6 +1274,7 @@ class RunExecution { return; case 'cost:updated': this.#cumulativeCostMicrocents += event.costMicrocents; + this.#budgetGovernor?.updateCost(this.#cumulativeCostMicrocents); this.#bus.emit({ ...event, runId, @@ -1219,6 +1348,7 @@ export class WorkflowEngine { readonly #validateEvents: boolean; readonly #capacity: number; readonly #resolverCapabilities: ResolverCapabilities; + readonly #maxTokensEstimate: number; readonly #runs = new Map(); constructor(deps: WorkflowEngineDeps) { @@ -1227,6 +1357,7 @@ export class WorkflowEngine { this.#validateEvents = deps.validateEvents ?? true; this.#capacity = deps.eventBufferCapacity ?? 256; this.#resolverCapabilities = deps.resolverCapabilities ?? {}; + this.#maxTokensEstimate = deps.maxTokensEstimate ?? DEFAULT_MAX_TOKENS_ESTIMATE; } /** @@ -1254,6 +1385,7 @@ export class WorkflowEngine { host may prune them on a TTL — out of 1.N scope. */ }, resolverCapabilities: this.#resolverCapabilities, + maxTokensEstimate: this.#maxTokensEstimate, }); this.#runs.set(runId, execution); void execution.begin(); @@ -1352,6 +1484,7 @@ export class WorkflowEngine { /* retained like a started run (see start) */ }, resolverCapabilities: this.#resolverCapabilities, + maxTokensEstimate: this.#maxTokensEstimate, checkpoint, }); this.#runs.set(input.runId, execution); diff --git a/packages/core/src/engine/execution-host.test.ts b/packages/core/src/engine/execution-host.test.ts index 46ae91b9..496adba7 100644 --- a/packages/core/src/engine/execution-host.test.ts +++ b/packages/core/src/engine/execution-host.test.ts @@ -144,6 +144,8 @@ describe('InMemoryRunStore', () => { runId: 'r1', timestamp: at, sequenceNumber: 1, + nodeId: 'n1', + gateId: 'bgid', spentMicrocents: 100, limitMicrocents: 50, }, diff --git a/packages/core/src/engine/node-executor.ts b/packages/core/src/engine/node-executor.ts index 99900b63..8e5eef02 100644 --- a/packages/core/src/engine/node-executor.ts +++ b/packages/core/src/engine/node-executor.ts @@ -78,6 +78,17 @@ export interface GateRequest { readonly timeoutAction?: 'approve' | 'reject'; /** The wall-clock deadline; the engine computes it from `timeoutMs` against its clock when omitted. */ readonly expiresAt?: string; + /** + * Budget-gate only (ADR-0028, 1.AC): the figures carried on the paired `budget:paused` event so a + * surface can show spent/limit before it decides whether to continue. + */ + readonly spentMicrocents?: number; + readonly limitMicrocents?: number; + /** + * Set by the budget governor so the engine can treat a `rejected` decision as a run-level + * `budget_exceeded` failure rather than completing the gate vertex. + */ + readonly isBudgetGate?: boolean; } /** @@ -138,6 +149,12 @@ export interface NodeExecContext { readonly signal: AbortSignalLike; /** 1-based attempt number for this dispatch (always 1 in 1.N; node-level retry is 1.S). */ readonly attemptNumber: number; + /** + * Pre-egress budget hook (ADR-0028, 1.AC). Supplied by the run loop when a workflow `budget` is + * configured; the agent runner forwards it into the turn core / fallback chain so every provider + * attempt is gated before egress. + */ + readonly preEgress?: import('./agent-turn.js').PreEgressHook; } /** The injected per-vertex executor. 1.O (`AgentRunner`) and 1.P (node handlers) implement it. */ diff --git a/packages/core/src/engine/node-handlers/dispatcher.ts b/packages/core/src/engine/node-handlers/dispatcher.ts index 76cb499b..3e8eb4df 100644 --- a/packages/core/src/engine/node-handlers/dispatcher.ts +++ b/packages/core/src/engine/node-handlers/dispatcher.ts @@ -53,14 +53,32 @@ export interface StandardNodeExecutorDeps { * absent — they fail loud until their workstream lands. */ export function createStandardNodeExecutor(deps: StandardNodeExecutorDeps): NodeExecutor { - return createDispatchingNodeExecutor({ - ...(deps.agent === undefined ? {} : { agent: createAgentNodeExecutor(deps.agent) }), - condition: createConditionNodeExecutor({ sandbox: deps.sandbox }), - transform: createTransformNodeExecutor({ sandbox: deps.sandbox }), - fan_in: createFanInNodeExecutor({ sandbox: deps.sandbox }), - fan_out: createFanOutNodeExecutor(), - human_in_the_loop: createHumanGateNodeExecutor(deps.humanGate ?? {}), - input: createInputNodeExecutor(), - output: createOutputNodeExecutor(), - }); + return { + execute(ctx) { + const handlers: NodeExecutorMap = { + ...(deps.agent === undefined + ? {} + : { + agent: createAgentNodeExecutor({ + ...deps.agent, + ...(ctx.preEgress === undefined ? {} : { preEgress: ctx.preEgress }), + }), + }), + condition: createConditionNodeExecutor({ sandbox: deps.sandbox }), + transform: createTransformNodeExecutor({ sandbox: deps.sandbox }), + fan_in: createFanInNodeExecutor({ sandbox: deps.sandbox }), + fan_out: createFanOutNodeExecutor(), + human_in_the_loop: createHumanGateNodeExecutor(deps.humanGate ?? {}), + input: createInputNodeExecutor(), + output: createOutputNodeExecutor(), + }; + const handler = handlers[ctx.vertex.type]; + if (handler === undefined) { + return Promise.resolve( + failed('internal', `no executor is registered for node type '${ctx.vertex.type}'`, false), + ); + } + return handler.execute(ctx); + }, + }; } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d5500698..2c9e3a83 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -143,6 +143,24 @@ export type { AgentRunnerDeps } from './engine/agent-runner.js'; export { DEFAULT_AGENT_TURN_LIMITS } from './engine/agent-turn.js'; export type { AgentTurnLimits, PreEgressHook } from './engine/agent-turn.js'; +// AgentSession (1.V) — the agent-first entry point: a multi-turn conversation bound to one agent, +// driving the SAME turn core a workflow `agent` node uses ([ADR-0024], agent-session-spec.md). A surface +// mints a sessionId and constructs it with host capabilities + an injected SessionEventSink; wiring that +// sink onto the shared RunEventBus (per-session sequenceNumber + gap/resync + a SessionHandle) is 1.W, +// and persistence is 1.X. The hard turn cap fails loud with `turn_limit` — never a silent stop. +export { + AgentSession, + DEFAULT_SESSION_MAX_TURNS, + SessionStateError, +} from './engine/agent-session.js'; +export type { + SessionDeps, + AgentSessionParams, + SessionEventSink, + SessionStreamEvent, + SessionLifecycleEvent, +} from './engine/agent-session.js'; + // Node-type handlers (1.P) — the six non-agent NodeExecutor arms (condition / transform / fan_out / // fan_in / input / output) plus the dispatcher that composes them (and the 1.O agent arm) into the one // executor the engine holds. `createStandardNodeExecutor` is the convenience wirer; the per-type diff --git a/packages/core/src/run-plan.ts b/packages/core/src/run-plan.ts index a4b77ab5..7ccbc79d 100644 --- a/packages/core/src/run-plan.ts +++ b/packages/core/src/run-plan.ts @@ -19,7 +19,7 @@ * settled upstream results — not evaluated inputs. */ -import type { Agent, FallbackChainEntry, EngineNodeType, WorkflowNode } from '@relavium/shared'; +import type { Agent, Budget, FallbackChainEntry, EngineNodeType, WorkflowNode } from '@relavium/shared'; import type { ReferenceSite } from './interpolation/collect.js'; @@ -168,4 +168,8 @@ export interface RunPlan { readonly vertices: ReadonlyMap; /** The run-wide concurrency cap (`workflow.max_parallel`), when declared. */ readonly maxParallel?: number; + /** The run-wide pre-egress budget cap (`workflow.budget`), when declared. */ + readonly budget?: Budget; + /** The run-wide wall-clock timeout (`workflow.timeout_ms`), when declared. */ + readonly timeoutMs?: number; } diff --git a/packages/llm/src/budget-estimator.test.ts b/packages/llm/src/budget-estimator.test.ts new file mode 100644 index 00000000..2242e7a9 --- /dev/null +++ b/packages/llm/src/budget-estimator.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from 'vitest'; + +import { estimateMaxNextCost } from './budget-estimator.js'; +import { MODEL_PRICING } from './pricing.js'; + +describe('estimateMaxNextCost', () => { + it('estimates output-only worst case at maxTokens', () => { + const model = MODEL_PRICING['claude-sonnet-4-6']; + // 10_000 output tokens @ $15/MTok = 150_000 micro-cents + expect(estimateMaxNextCost('claude-sonnet-4-6', 10_000)).toBe( + Math.round((10_000 * model.outputPerMtokMicrocents) / 1_000_000), + ); + }); + + it('returns 0 for non-positive maxTokens', () => { + expect(estimateMaxNextCost('claude-sonnet-4-6', 0)).toBe(0); + expect(estimateMaxNextCost('claude-sonnet-4-6', -1)).toBe(0); + }); + + it('throws UnknownModelError for an unlisted model', () => { + expect(() => estimateMaxNextCost('not-a-real-model', 1_000)).toThrow('unknown model id'); + }); + + it('uses the cheaper mini rate for a mini model', () => { + const model = MODEL_PRICING['gpt-5.4-mini']; + expect(estimateMaxNextCost('gpt-5.4-mini', 100_000)).toBe( + Math.round((100_000 * model.outputPerMtokMicrocents) / 1_000_000), + ); + }); +}); diff --git a/packages/llm/src/budget-estimator.ts b/packages/llm/src/budget-estimator.ts new file mode 100644 index 00000000..40dad51b --- /dev/null +++ b/packages/llm/src/budget-estimator.ts @@ -0,0 +1,20 @@ +import { priceModel } from './cost-tracker.js'; + +const TOKENS_PER_MTOK = 1_000_000; + +/** + * Pre-egress worst-case cost estimate for a single LLM call. + * + * The estimate prices only the **output** side at the model's declared `maxTokens` + * (or a configured default), because the engine does not tokenize the prompt locally. + * This is intentionally conservative: it may block slightly early rather than overshoot. + * + * All figures are integer micro-cents. + */ +export function estimateMaxNextCost(modelId: string, maxOutputTokens: number): number { + const p = priceModel(modelId); + if (maxOutputTokens <= 0) { + return 0; + } + return Math.round((maxOutputTokens * p.outputPerMtokMicrocents) / TOKENS_PER_MTOK); +} diff --git a/packages/llm/src/fallback-chain.ts b/packages/llm/src/fallback-chain.ts index ac91c51e..947e9a4b 100644 --- a/packages/llm/src/fallback-chain.ts +++ b/packages/llm/src/fallback-chain.ts @@ -102,6 +102,17 @@ export interface AttemptRecord { readonly skipReason?: string; } +/** + * Hook called immediately before a provider attempt is actually dispatched (after skip checks, + * after the attempt record is allocated, but before credential resolution / the seam call). + * In 1.AC this is where the pre-egress budget governor runs; a rejected hook aborts the attempt + * and is surfaced as a fatal chain error. + */ +export type PreAttemptHook = (info: { + readonly model: string; + readonly maxTokens?: number; +}) => void | Promise; + /** Dependencies injected into a {@link FallbackChain} — all timing is injectable so tests are deterministic. */ export interface FallbackChainOptions { /** @@ -117,6 +128,11 @@ export interface FallbackChainOptions { readonly costTracker?: CostTracker; /** Visibility hook fired once per attempt (succeeded / failed / skipped). */ readonly onAttempt?: (record: AttemptRecord) => void; + /** + * Pre-egress hook called before every real provider attempt (not called for skipped entries). + * See {@link PreAttemptHook}. + */ + readonly preAttempt?: PreAttemptHook; /** * The delay primitive used for backoff between same-entry retries. **Required and host-injected**: * the seam is platform-free (no ambient `setTimeout`), so the host supplies the timer — a @@ -392,6 +408,11 @@ export class FallbackChain { ): Promise { const record = run.next(entry); try { + const maxTokens = entryReq.maxTokens; + await this.#options.preAttempt?.({ + model: entry.model, + ...(maxTokens === undefined ? {} : { maxTokens }), + }); const key = await this.#resolveKey(entry.provider.id); const result = await entry.provider.generate(entryReq, key); this.#emitSuccess(record, entry.model, result.usage); @@ -417,6 +438,11 @@ export class FallbackChain { ): AsyncGenerator { let usage: Usage | undefined; try { + const maxTokens = entryReq.maxTokens; + await this.#options.preAttempt?.({ + model: entry.model, + ...(maxTokens === undefined ? {} : { maxTokens }), + }); const key = await this.#resolveKey(entry.provider.id); for await (const chunk of entry.provider.stream(entryReq, key)) { if (chunk.type === 'error') { diff --git a/packages/llm/src/index.ts b/packages/llm/src/index.ts index 4a5feeef..b5b1c86c 100644 --- a/packages/llm/src/index.ts +++ b/packages/llm/src/index.ts @@ -123,6 +123,7 @@ export { MODEL_PRICING, KNOWN_MODEL_IDS } from './pricing.js'; export type { ModelPricing, CanonicalModelId } from './pricing.js'; export { priceModel, cost, CostTracker } from './cost-tracker.js'; export type { CostUpdate } from './cost-tracker.js'; +export { estimateMaxNextCost } from './budget-estimator.js'; // FallbackChain runner — fallback policy outside the adapters (1.K). export { FallbackChain, withFallback, stripReasoningParts } from './fallback-chain.js'; @@ -132,6 +133,7 @@ export type { AttemptRecord, AttemptOutcome, BackoffStrategy, + PreAttemptHook, } from './fallback-chain.js'; // ToolNormalizer (1.E). diff --git a/packages/shared/src/run-event.test.ts b/packages/shared/src/run-event.test.ts index 6df41c07..36dd932f 100644 --- a/packages/shared/src/run-event.test.ts +++ b/packages/shared/src/run-event.test.ts @@ -133,8 +133,10 @@ const valid: Record> = { 'budget:paused': { type: 'budget:paused', ...env, + nodeId: 'n', spentMicrocents: 1000, limitMicrocents: 1000, + gateId: 'budget-gate-1', }, }; @@ -256,6 +258,7 @@ const reject: Record> = { 'run:timeout (negative elapsedMs)': { ...valid['run:timeout'], elapsedMs: -1 }, 'budget:warning (thresholdPct > 100)': { ...valid['budget:warning'], thresholdPct: 101 }, 'budget:paused (negative spentMicrocents)': { ...valid['budget:paused'], spentMicrocents: -1 }, + 'budget:paused (missing gateId)': { ...valid['budget:paused'], gateId: undefined }, }; describe('RunEvent union — every variant', () => { diff --git a/packages/shared/src/run-event.ts b/packages/shared/src/run-event.ts index 442d3d85..9fe8a6a0 100644 --- a/packages/shared/src/run-event.ts +++ b/packages/shared/src/run-event.ts @@ -347,8 +347,10 @@ export const BudgetWarningEventSchema = z.object({ export const BudgetPausedEventSchema = z.object({ type: z.literal('budget:paused'), ...runBase, + nodeId: nonEmptyString, // the agent node whose next LLM call would exceed the cap spentMicrocents: nonNegativeInt, limitMicrocents: nonNegativeInt, + gateId: nonEmptyString, // stable id of the budget gate; required by engine.resume(runId, gateId, decision) }); /** The run-event variants, discriminated on `type` (exposed via `RunEventSchema.innerType()`). */