diff --git a/apps/fabro-web/app/hooks/use-run-toasts.ts b/apps/fabro-web/app/hooks/use-run-toasts.ts index 087b1fb65..67d33e034 100644 --- a/apps/fabro-web/app/hooks/use-run-toasts.ts +++ b/apps/fabro-web/app/hooks/use-run-toasts.ts @@ -41,14 +41,12 @@ function steeringToastMessage(payload: RunEventPayload): string | null { const props = payload.properties ?? {}; switch (payload.event) { - case "agent.steering.injected": { - const kind = props.kind; - if (kind === "append") return "Steer delivered."; - if (kind === "interrupt") { - return "Agent interrupted — your message is the next turn."; - } - return null; - } + case "run.interrupt": + return "Agent interrupted."; + case "run.steer": + return "Steer accepted."; + case "agent.steering.injected": + return "Steer delivered."; case "agent.steer.buffered": return "Steer queued — will apply when an agent stage runs."; case "agent.steer.dropped": { diff --git a/apps/fabro-web/app/lib/run-events.test.tsx b/apps/fabro-web/app/lib/run-events.test.tsx index bf35aad8c..497c3c0d1 100644 --- a/apps/fabro-web/app/lib/run-events.test.tsx +++ b/apps/fabro-web/app/lib/run-events.test.tsx @@ -192,7 +192,7 @@ describe("subscribeToRunEvents", () => { }, }); - source.emit({ id: "evt-1", event: "agent.steer.buffered", properties: { kind: "append" } }); + source.emit({ id: "evt-1", event: "agent.steer.buffered", properties: {} }); expect(seen).toEqual(["agent.steer.buffered"]); expect(keys).toEqual([queryKeys.runs.events("run-shared-payload", 1000)]); diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index 348f97c97..e89f67e10 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/apps/fabro-web/app/lib/run-events.ts @@ -78,6 +78,8 @@ const INTERVIEW_EVENTS = new Set([ "interview.interrupted", ]); const STEERING_EVENTS = new Set([ + "run.interrupt", + "run.steer", "agent.steering.injected", "agent.session.activated", "agent.session.deactivated", diff --git a/docs/internal/events.md b/docs/internal/events.md index 337274272..b2f0c30c5 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -200,6 +200,40 @@ Informational, warning, or error notice emitted during the run. | `code` | string | Machine-readable notice code | | `message` | string | Human-readable message | +### `run.interrupt` + +Emitted after a live worker accepts a run interrupt control operation. The +actor is stored in the top-level `actor` envelope field. Properties are empty. + +```json +{ + "id": "...", "ts": "...", "run_id": "...", + "event": "run.interrupt", + "actor": { "kind": "user", "login": "octocat" }, + "properties": {} +} +``` + +### `run.steer` + +Emitted after a live worker accepts run steering text. The actor is stored in +the top-level `actor` envelope field. + +```json +{ + "id": "...", "ts": "...", "run_id": "...", + "event": "run.steer", + "actor": { "kind": "user", "login": "octocat" }, + "properties": { + "text": "Remember to run tests after changes" + } +} +``` + +| Property | Type | Description | +|----------|------|-------------| +| `text` | string | Accepted steering text | + ### `metadata.snapshot.started` Emitted when Fabro begins a durable metadata snapshot operation. These are product events for Fabro metadata snapshots, not tracing spans for the underlying git or filesystem work. diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index e46c8b23c..9b542118f 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -885,10 +885,10 @@ paths: summary: Steer Run description: | Send a mid-run steering message to the live agent session(s) of a - running run. Set `interrupt=true` to cancel the in-flight LLM stream - and tool calls in the current round and deliver the message as the - next user turn; otherwise the message is appended to the steering - queue and picked up at the next turn boundary. + running run. Set `interrupt=true` to atomically interrupt the active + API-mode agent round first, then deliver this message as the next + user turn. Without `interrupt=true`, the message is appended to the + steering queue and may buffer until the next API-mode agent session. parameters: - $ref: "#/components/parameters/RunId" requestBody: @@ -940,6 +940,52 @@ paths: schema: $ref: "#/components/schemas/ErrorResponse" + /api/v1/runs/{id}/interrupt: + post: + operationId: interruptRun + tags: [Human-in-the-Loop] + summary: Interrupt Run + description: | + Interrupt the active API-mode agent round without sending steering + text. The agent keeps its steering lease and waits for a later steer + message before starting another LLM round. + parameters: + - $ref: "#/components/parameters/RunId" + responses: + "202": + description: Interrupt accepted and forwarded to the worker + "404": + description: Run not found + headers: + x-request-id: + $ref: "#/components/headers/XRequestId" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + "409": + description: | + Run is not currently interruptible. Returned when the run is in a + terminal state, blocked (use the answer endpoint instead), has no + active API-mode agent session, or all currently running agent + stages are CLI-mode. + headers: + x-request-id: + $ref: "#/components/headers/XRequestId" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + "503": + description: Worker control channel unavailable + headers: + x-request-id: + $ref: "#/components/headers/XRequestId" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + /api/v1/runs/{id}/start: post: operationId: startRun @@ -4609,10 +4655,10 @@ components: interrupt: type: boolean description: | - When true, cancel the in-flight LLM stream and tool calls in the - current round before delivering. When false (default), append to - the steering queue and let the agent pick it up at the next - turn boundary. + When true, apply a worker-control interrupt first, then deliver + this text as steering in the same control operation. When false + (default), append to the steering queue and let the agent pick it + up at the next turn boundary. default: false StartRunRequest: @@ -8332,4 +8378,4 @@ components: login: type: string description: User's login identifier (e.g. GitHub username). - example: octocat \ No newline at end of file + example: octocat diff --git a/docs/public/human-tools/steering.mdx b/docs/public/human-tools/steering.mdx index e834889e2..dba4575c3 100644 --- a/docs/public/human-tools/steering.mdx +++ b/docs/public/human-tools/steering.mdx @@ -7,7 +7,7 @@ Steering lets you send guidance to an agent while it's working — without waiti ## How steering works -A steering message is injected into the agent's conversation as a user-role message. The agent sees it on its next LLM turn — after the current tool call finishes — and can adjust its approach immediately. +A steering message is injected into the agent's conversation as a user-role message. The agent sees it on its next LLM turn and can adjust its approach immediately. The delivery flow: @@ -16,16 +16,18 @@ The delivery flow: 3. Before the next LLM call, Fabro drains the queue and injects each message as a `Steering` turn in the conversation history 4. The LLM sees the guidance alongside its existing context and adjusts accordingly -Steering is **asynchronous** — the agent picks up the message at its next natural pause point (between tool calls), not mid-execution. +Steering is **asynchronous** — the agent picks up the message at its next natural pause point. + +When you send steering with `interrupt=true`, Fabro first cancels the active API-mode agent round, then queues the steering text in the same worker-control operation. The agent resumes with that steering text as the next user turn. A standalone interrupt through the API cancels the active round without text and keeps the session waiting until a later steer arrives. ## When steering is delivered Steering messages are drained from the queue at two points during the agent loop: 1. **Before the first LLM call** — any messages queued before the agent starts its first turn -2. **After each tool execution round** — between tool results being collected and the next LLM call +2. **After each interrupted or completed round** — before the next LLM call -This means there is a natural latency between sending a steering message and the agent seeing it. If the agent is in the middle of a long-running shell command, the message waits until that command finishes and the next LLM turn begins. +This means there is a natural latency between sending a plain steering message and the agent seeing it. If the agent is in the middle of a long-running shell command, the message waits until that command finishes and the next LLM turn begins. Use interrupting steering when the current round should stop before the message is delivered. Multiple steering messages sent in quick succession are all delivered together at the next drain point. diff --git a/docs/public/reference/cli.mdx b/docs/public/reference/cli.mdx index f3c83498a..3717137ce 100644 --- a/docs/public/reference/cli.mdx +++ b/docs/public/reference/cli.mdx @@ -1128,7 +1128,7 @@ fabro steer [OPTIONS] [TEXT] | Option | Description | | --- | --- | -| `--interrupt` | Cancel the in-flight LLM stream / tool calls and deliver the message as the next user turn (default: append to the steering queue) | +| `--interrupt` | Interrupt the active API-mode agent round first, then deliver the message as the next user turn (default: append to the steering queue) | | `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | | `--text-stdin` | Read steer text from stdin instead of a positional arg | diff --git a/docs/superpowers/plans/2026-05-05-decouple-interrupts-from-steering.md b/docs/superpowers/plans/2026-05-05-decouple-interrupts-from-steering.md new file mode 100644 index 000000000..29f743bfc --- /dev/null +++ b/docs/superpowers/plans/2026-05-05-decouple-interrupts-from-steering.md @@ -0,0 +1,111 @@ +# Decouple Interrupts From Steering Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Split run interruption from steering message delivery while keeping `steer interrupt=true` as ergonomic sugar. + +**Architecture:** `run.interrupt` becomes a standalone control/event path that cancels the active API-mode agent round and waits for later steering. `run.steer` becomes a plain message-injection path with no delivery kind. The existing combined user flow applies interrupt first, then steer, preserving user convenience without coupling the concepts in queue/event types. + +**Tech Stack:** Rust workspace crates (`fabro-agent`, `fabro-workflow`, `fabro-server`, `fabro-interview`, `fabro-api`, `fabro-client`), OpenAPI, generated TypeScript API client, React web app, SSE run events. + +--- + +## Summary + +Split steering message delivery from agent interruption. `run.steer` becomes a plain "user injected guidance" event with no `kind`; `run.interrupt` becomes a separate control/event path that cancels the active agent round. `POST /runs/{id}/steer { interrupt: true }` remains ergonomic sugar for "interrupt, then steer," but there is no standalone CLI interrupt command. + +Standalone interrupts cancel the current active API-mode agent round, keep the stage's steering lease active, and wait at a steerable point until a later steer resumes the agent. + +## Key Changes + +- Public API: + - Add `POST /api/v1/runs/{id}/interrupt` with `202`, `404`, `409`, and `503` behavior matching steer/cancel conventions. + - Keep `POST /api/v1/runs/{id}/steer`; keep `interrupt?: boolean` as request convenience. + - For `steer interrupt=true`, require an active API-mode agent session and send one atomic worker-control operation that applies interrupt first, then enqueues the steer. + - Non-interrupt steer may still buffer when no API session is active and no CLI agent is running. + - Regenerate Rust and TypeScript API clients after OpenAPI updates. + +- Control protocol and types: + - Replace `WorkerControlMessage::Steer { text, kind, actor }` with `Steer { text, actor }`. + - Add `WorkerControlMessage::Interrupt { actor }`. + - Add `WorkerControlMessage::InterruptThenSteer { text, actor }` for the combined convenience path. The server must send this as one control envelope; do not implement `steer interrupt=true` as two independent enqueue operations. + - Remove `SteerKind` from `fabro-types`, `fabro-agent`, workflow event payloads, and web client assumptions. + - Keep existing CLI `fabro steer --interrupt`; implement it through the existing steer API, not a new CLI command. + +- Workflow/session behavior: + - Split `SessionControlHandle` into plain `steer(text, actor)` and plain `interrupt(actor)`. + - Add an explicit agent-side `waiting_for_steer` state, guarded by the same control state as the steering queue. + - Steering enqueues text, clears `waiting_for_steer`, and wakes the session. + - Interrupt cancels the current round token and, when the steering queue is empty, sets `waiting_for_steer` so the session cannot immediately start another LLM round with unchanged context. + - Duplicate pure interrupts while already `waiting_for_steer` are idempotent: accept them, emit another `run.interrupt`, keep the session waiting, and do not enqueue synthetic steering. + - If an interrupt cancels an LLM stream or tool round without queued steering, the session waits without closing/deactivating its steering lease; terminal run cancellation still wins immediately. + - If interrupt and steer are applied together, the steer resumes the waiting/next round immediately. + - Terminal run cancellation must wake/break any `waiting_for_steer` wait and return the existing cancellation error path. + - Update the natural-completion close-the-door path so the activation lease is kept alive when the steering queue is nonempty or `waiting_for_steer` is true. `CompletionCoordinator`, `ActivationLease::release_if_queue_empty`, and `SessionControlHandle` should expose/use a single "has pending control work" predicate instead of checking queue emptiness alone. + +- Events: + - Add top-level persisted `run.interrupt` with `actor` in the envelope and empty properties. + - Add top-level persisted `run.steer` with `actor` in the envelope and `properties.text`. + - Keep `agent.steering.injected`, but remove `kind`; it means the steer actually entered agent history. + - Keep `agent.steer.buffered` and `agent.steer.dropped`, but remove `kind` from buffered. + - Update run-event conversion, stored fields, docs, SSE invalidation, and toast logic for the simplified payloads. + +- Event ownership and ordering: + - The worker-side control handler / `SteeringHub` is the single source of truth for persisted `run.interrupt` and `run.steer`; API handlers only emit these events indirectly after the live worker accepts the control envelope. + - A failed or timed-out control-channel request must not emit `run.interrupt` or `run.steer`. + - For `InterruptThenSteer`, persisted order must be `run.interrupt`, then `run.steer`, then later `agent.steering.injected` only when the text is drained into agent history. + +## Protocol vs Events + +Worker-control messages are transport commands, not persisted `RunEvent` names. `run.interrupt_then_steer` exists only as a worker-control envelope for atomic delivery of the combined convenience path and must never be emitted as a persisted run event. The only persisted run-level event names introduced here are `run.interrupt` and `run.steer`. + +## API Response Matrix + +| Run state | `POST /steer` | `POST /steer { interrupt: true }` | `POST /interrupt` | +| --- | --- | --- | --- | +| Active API-mode session | `202` accepted; emits `run.steer`; later `agent.steering.injected` | `202` accepted atomically; emits `run.interrupt`, then `run.steer` | `202` accepted; emits `run.interrupt` | +| Active API-mode session already `waiting_for_steer` | `202` accepted; emits `run.steer`; clears wait and later emits `agent.steering.injected` | `202` accepted atomically; emits `run.interrupt`, then `run.steer`; clears wait | `202` accepted idempotently; emits `run.interrupt`; remains waiting | +| No active API session, no active CLI agent | `202` accepted; emits `run.steer`, then `agent.steer.buffered` | `409` `no_active_api_session` | `409` `no_active_api_session` | +| Active CLI-only agent stages | `409` `cli_agent_not_steerable` | `409` `cli_agent_not_steerable` | `409` `cli_agent_not_steerable` | +| Blocked on interview/question | `409` `use_answer_endpoint` | `409` `use_answer_endpoint` | `409` `use_answer_endpoint` | +| Terminal run | `409` `run_not_steerable` | `409` `run_not_steerable` | `409` `run_not_interruptible` | +| Missing live worker channel | `503` `worker_control_unavailable` | `503` `worker_control_unavailable` | `503` `worker_control_unavailable` | +| Archived run | Existing archived-run rejection response from `reject_if_archived` | Existing archived-run rejection response from `reject_if_archived` | Existing archived-run rejection response from `reject_if_archived` | + +## Implementation Checklist + +- [x] OpenAPI/API clients: add `/runs/{id}/interrupt`, keep `SteerRunRequest.interrupt`, regenerate `fabro-api`, `fabro-client` usage, and TypeScript API client models. +- [x] Server route and transport: add interrupt handler, add `RunAnswerTransport::interrupt`, add `RunAnswerTransport::interrupt_then_steer`, enforce the API response matrix, and ensure combined steer uses one control operation. +- [x] Worker protocol: simplify `run.steer`, add `run.interrupt`, add `run.interrupt_then_steer`, and update subprocess/in-process dispatch. +- [x] Agent session state: replace `SteerKind` queue items with plain text+actor, add `waiting_for_steer`, wake on steer, block after pure interrupt, make duplicate interrupts idempotent while waiting, and break wait on terminal cancellation. +- [x] Natural-completion lease safety: update `CompletionCoordinator`, `ActivationLease::release_if_queue_empty`, and `SessionControlHandle` so close-the-door keeps the lease alive when either the queue is nonempty or `waiting_for_steer` is true. +- [x] Steering hub behavior: expose plain `deliver_steer`, `interrupt`, and `interrupt_then_steer`; emit accepted control events in the required order; keep buffering/dropping only for steering text. +- [x] Events/schema/store/web: add `run.interrupt` and `run.steer`, simplify `agent.steering.injected` and `agent.steer.buffered`, update stored fields, docs, run-state projections, SSE invalidation, and toasts. +- [x] Docs: update internal event docs, public API reference, CLI docs for `fabro steer --interrupt`, and steering docs to explain interrupt as separate control flow. +- [x] Verification: run the Rust and web test commands listed below, plus formatting and clippy. + +## Test Plan + +- `fabro-interview` worker-control envelope tests: JSON round-trip tests for transport-only `run.interrupt`, simplified `run.steer`, and transport-only `run.interrupt_then_steer` preserving `text` and `actor`. +- `fabro-types` / `fabro-api` persisted event tests: `run.interrupt` and `run.steer` serialize as persisted `RunEvent`s; no persisted event named `run.interrupt_then_steer` exists. +- `fabro-agent`: unit tests for plain steer injection, LLM-stream interrupt entering `waiting_for_steer`, tool-round interrupt entering `waiting_for_steer`, pure interrupt racing with a no-tool natural completion without releasing the lease, later steer wake-up, duplicate interrupt while waiting, interrupt-plus-steer resuming without an extra wait, and terminal cancel breaking the wait. +- `fabro-workflow`: steering hub tests for buffering plain steers, dropping plain steers, broadcasting interrupts to active sessions, and preserving interrupt-before-steer ordering. +- `fabro-server`: handler tests for `/interrupt`, `steer interrupt=true`, the full API response matrix, missing worker channel, CLI-only conflict, terminal/blocked conflicts, and stale activation/deactivation behavior. +- `fabro-cli` runner/subprocess bridge: worker-control line handler tests for simplified `run.steer`, `run.interrupt`, and `run.interrupt_then_steer`. +- `fabro-store` and run-event tests: projection ignores top-level `run.interrupt`/`run.steer` except where explicitly tracked; `agent.session.activated` remains the provider-used source. +- Web tests: update run-event invalidation and toast assertions for `run.interrupt`, `run.steer`, simplified `agent.steering.injected`, and simplified `agent.steer.buffered`. +- Verification commands: + - `cargo build -p fabro-api` + - `cd lib/packages/fabro-api-client && bun run generate` + - `cargo nextest run --workspace` + - `cd apps/fabro-web && bun test && bun run typecheck` + - `cargo +nightly-2026-04-14 fmt --check --all` + - `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings` + +## Assumptions + +- Event names are exactly `run.interrupt` and `run.steer`. +- Standalone interrupt is API-only for now; no `fabro interrupt` CLI command and no new standalone web button are required. +- A pure interrupt is not a failure, pause, or cancellation of the run; it is a mid-stage wait point that resumes only when steering arrives or the run is terminally cancelled. +- Non-interrupt steering keeps the current buffering behavior, but interrupt steering does not buffer the interrupt portion when no active API session exists. +- Duplicate pure interrupts while already waiting are idempotent `202` responses that emit another persisted `run.interrupt` and leave the wait state unchanged. diff --git a/lib/crates/fabro-agent/src/lib.rs b/lib/crates/fabro-agent/src/lib.rs index 3ba5b81bb..74e1d9e10 100644 --- a/lib/crates/fabro-agent/src/lib.rs +++ b/lib/crates/fabro-agent/src/lib.rs @@ -55,7 +55,7 @@ pub use tools::{ make_shell_tool, make_shell_tool_with_config, make_write_file_tool, register_core_tools, }; pub use truncation::{TruncationMode, truncate_lines, truncate_output, truncate_tool_output}; -pub use types::{AgentEvent, SessionEvent, SessionState, SteerKind, Turn}; +pub use types::{AgentEvent, SessionEvent, SessionState, Turn}; #[cfg(test)] #[allow( diff --git a/lib/crates/fabro-agent/src/session.rs b/lib/crates/fabro-agent/src/session.rs index 71187cab8..ef2e669d1 100644 --- a/lib/crates/fabro-agent/src/session.rs +++ b/lib/crates/fabro-agent/src/session.rs @@ -14,9 +14,9 @@ use fabro_llm::{Error as LlmError, retry}; use fabro_mcp::config::{McpServerSettings, McpTransport}; use fabro_mcp::connection_manager::McpConnectionManager; use fabro_model::Provider; -use fabro_types::{Principal, SteerKind}; +use fabro_types::Principal; use futures::StreamExt; -use tokio::sync::{Mutex as AsyncMutex, broadcast}; +use tokio::sync::{Mutex as AsyncMutex, Notify, broadcast}; use tokio::time; use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; @@ -40,9 +40,15 @@ use crate::subagent::{SubAgentCallbackEvent, SubAgentEventCallback, SubAgentMana use crate::tool_execution::execute_tool_calls; use crate::types::{AgentEvent, SessionEvent, SessionState, Turn}; -/// One queued steering message: text + delivery kind + the principal that -/// authored it (None for direct internal callers like loop-detection). -pub type SteeringItem = (String, SteerKind, Option); +/// One queued steering message: text + the principal that authored it (None +/// for direct internal callers like loop-detection). +pub type SteeringItem = (String, Option); + +#[derive(Default)] +struct ControlState { + queue: VecDeque, + waiting_for_steer: bool, +} /// Trait that lets the workflow layer keep an agent in `process_input` when a /// natural completion (no tool calls) coincides with an unconsumed steering @@ -62,8 +68,9 @@ pub trait CompletionCoordinator: Send + Sync { /// interrupt the current round without holding the session itself. #[derive(Clone)] pub struct SessionControlHandle { - queue: Arc>>, + control: Arc>, round_token: Arc>, + notify: Arc, } impl Default for SessionControlHandle { @@ -80,37 +87,45 @@ impl SessionControlHandle { #[must_use] pub fn new() -> Self { Self { - queue: Arc::new(Mutex::new(VecDeque::new())), + control: Arc::new(Mutex::new(ControlState::default())), round_token: Arc::new(RwLock::new(CancellationToken::new())), + notify: Arc::new(Notify::new()), } } - /// Push an `Append`-kind steering message onto the queue. + /// Push a steering message onto the queue and wake a session waiting + /// after a pure interrupt. pub fn steer(&self, text: String, actor: Option) { - self.enqueue((text, SteerKind::Append, actor)); + self.enqueue((text, actor)); } - /// Push an `Interrupt`-kind steering message onto the queue *and* cancel - /// the current round so the agent loop reacts immediately. - pub fn interrupt_with(&self, text: String, actor: Option) { - self.enqueue((text, SteerKind::Interrupt, actor)); - } - - /// Direct enqueue used by callers that already encoded a kind/actor - /// (e.g. the hub flushing buffered steers as Appends). For - /// `Interrupt`-kind items, also cancels the current round token. - pub fn enqueue(&self, item: SteeringItem) { - let cancel_after = matches!(item.1, SteerKind::Interrupt); - self.queue - .lock() - .expect("steering queue lock poisoned") - .push_back(item); - if cancel_after { - self.round_token - .read() - .expect("round token lock poisoned") - .cancel(); + /// Cancel the current round and, if no steering text is queued, park the + /// session at a steerable wait point. + pub fn interrupt(&self, _actor: Option) { + { + let mut control = self.control.lock().expect("control state lock poisoned"); + if control.queue.is_empty() { + control.waiting_for_steer = true; + } } + self.cancel_round(); + self.notify.notify_waiters(); + } + + /// Atomically apply interrupt semantics, then enqueue steering text. + pub fn interrupt_then_steer(&self, text: String, actor: Option) { + self.interrupt_then_enqueue((text, actor)); + } + + /// Direct enqueue used by callers such as the hub flushing buffered + /// steers. + pub fn enqueue(&self, item: SteeringItem) { + { + let mut control = self.control.lock().expect("control state lock poisoned"); + control.waiting_for_steer = false; + control.queue.push_back(item); + } + self.notify.notify_waiters(); } /// Push `item` while enforcing a FIFO cap: if the queue is at or above @@ -118,39 +133,97 @@ impl SessionControlHandle { /// single lock acquisition. #[must_use] pub fn enqueue_bounded(&self, item: SteeringItem, cap: usize) -> Option { - let cancel_after = matches!(item.1, SteerKind::Interrupt); let evicted = { - let mut q = self.queue.lock().expect("steering queue lock poisoned"); - let evicted = if q.len() >= cap { q.pop_front() } else { None }; - q.push_back(item); + let mut control = self.control.lock().expect("control state lock poisoned"); + let evicted = if control.queue.len() >= cap { + control.queue.pop_front() + } else { + None + }; + control.queue.push_back(item); + control.waiting_for_steer = false; evicted }; - if cancel_after { - self.round_token - .read() - .expect("round token lock poisoned") - .cancel(); - } + self.notify.notify_waiters(); evicted } + /// Interrupt the current round and push `item` while enforcing a FIFO cap. + #[must_use] + pub fn interrupt_then_enqueue_bounded( + &self, + item: SteeringItem, + cap: usize, + ) -> Option { + let evicted = { + let mut control = self.control.lock().expect("control state lock poisoned"); + let evicted = if control.queue.len() >= cap { + control.queue.pop_front() + } else { + None + }; + control.waiting_for_steer = true; + control.queue.push_back(item); + control.waiting_for_steer = false; + evicted + }; + self.cancel_round(); + self.notify.notify_waiters(); + evicted + } + + fn interrupt_then_enqueue(&self, item: SteeringItem) { + { + let mut control = self.control.lock().expect("control state lock poisoned"); + control.waiting_for_steer = true; + control.queue.push_back(item); + control.waiting_for_steer = false; + } + self.cancel_round(); + self.notify.notify_waiters(); + } + + fn cancel_round(&self) { + self.round_token + .read() + .expect("round token lock poisoned") + .cancel(); + } + /// Whether the steering queue currently has no unconsumed messages. #[must_use] pub fn queue_is_empty(&self) -> bool { - self.queue + self.control .lock() - .expect("steering queue lock poisoned") + .expect("control state lock poisoned") + .queue .is_empty() } + /// Whether queue work or an interrupt-induced wait is still pending. + #[must_use] + pub fn has_pending_control_work(&self) -> bool { + let control = self.control.lock().expect("control state lock poisoned"); + !control.queue.is_empty() || control.waiting_for_steer + } + + #[must_use] + pub fn is_waiting_for_steer(&self) -> bool { + self.control + .lock() + .expect("control state lock poisoned") + .waiting_for_steer + } + /// Current queue length. Production callers should generally prefer /// `queue_is_empty` or `enqueue_bounded`'s atomic eviction; this is /// kept for tests and diagnostics. #[must_use] pub fn queue_len(&self) -> usize { - self.queue + self.control .lock() - .expect("steering queue lock poisoned") + .expect("control state lock poisoned") + .queue .len() } } @@ -164,7 +237,8 @@ pub struct Session { llm_client: Client, provider_profile: Arc, sandbox: Arc, - steering_queue: Arc>>, + control_state: Arc>, + control_notify: Arc, followup_queue: Arc>>, cancel_token: CancellationToken, round_token: Arc>, @@ -197,7 +271,8 @@ impl Session { llm_client, provider_profile, sandbox, - steering_queue: Arc::new(Mutex::new(VecDeque::new())), + control_state: Arc::new(Mutex::new(ControlState::default())), + control_notify: Arc::new(Notify::new()), followup_queue: Arc::new(Mutex::new(VecDeque::new())), cancel_token: CancellationToken::new(), round_token: Arc::new(RwLock::new(CancellationToken::new())), @@ -653,16 +728,21 @@ impl Session { self.event_emitter.subscribe() } - /// Push an `Append`-kind steer onto the queue (no actor — internal - /// callers like loop-detection use this). + /// Push a steer onto the queue (no actor — internal callers like + /// loop-detection use this). pub fn steer(&self, message: String) { self.control_handle().steer(message, None); } - /// Push an `Interrupt`-kind steer onto the queue and cancel the current - /// round token so the agent loop reacts mid-round. - pub fn interrupt_with(&self, message: String, actor: Option) { - self.control_handle().interrupt_with(message, actor); + /// Cancel the current round and wait for later steering before starting + /// another LLM round. + pub fn control_interrupt(&self, actor: Option) { + self.control_handle().interrupt(actor); + } + + /// Cancel the current round and deliver the message as the next steer. + pub fn interrupt_then_steer(&self, message: String, actor: Option) { + self.control_handle().interrupt_then_steer(message, actor); } /// Cheap, cloneable handle that lets external coordinators deliver @@ -670,8 +750,9 @@ impl Session { #[must_use] pub fn control_handle(&self) -> SessionControlHandle { SessionControlHandle { - queue: self.steering_queue.clone(), + control: self.control_state.clone(), round_token: self.round_token.clone(), + notify: self.control_notify.clone(), } } @@ -969,9 +1050,19 @@ impl Session { } } + // Terminal cancellation wins even when a control interrupt has + // parked the session waiting for steering. + if self.cancel_token.is_cancelled() { + self.close(); + return Err(self.interrupted_error()); + } + // Drain pending steering messages at the top of every iteration - // so an Interrupt-kind steer pushed mid-round is delivered as the - // first turn of the next round. + // so steering pushed mid-round is delivered as the first turn of + // the next round. A pure interrupt with no queued steer parks the + // session here until a later steer arrives. + self.drain_steering(); + self.wait_for_steer_if_needed().await?; self.drain_steering(); // Check max_tool_rounds_per_input @@ -994,12 +1085,6 @@ impl Session { break; } - // Check cancellation - if self.cancel_token.is_cancelled() { - self.close(); - return Err(self.interrupted_error()); - } - // Snapshot the per-round token; it stays stable for this iteration. let round_token = self .round_token @@ -1348,13 +1433,14 @@ impl Session { } fn drain_steering(&mut self) { - let messages: Vec = self - .steering_queue - .lock() - .expect("steering queue lock poisoned") - .drain(..) - .collect(); - for (text, kind, actor) in messages { + let messages: Vec = { + let mut control = self + .control_state + .lock() + .expect("control state lock poisoned"); + control.queue.drain(..).collect() + }; + for (text, actor) in messages { self.history.push(Turn::Steering { content: text.clone(), timestamp: SystemTime::now(), @@ -1362,12 +1448,36 @@ impl Session { self.event_emitter .emit(self.id.clone(), AgentEvent::SteeringInjected { text, - kind, actor, }); } } + async fn wait_for_steer_if_needed(&mut self) -> Result<(), Error> { + loop { + let notified = self.control_notify.notified(); + let should_wait = { + let control = self + .control_state + .lock() + .expect("control state lock poisoned"); + control.waiting_for_steer && control.queue.is_empty() + }; + if !should_wait { + return Ok(()); + } + + tokio::select! { + biased; + () = self.cancel_token.cancelled() => { + self.close(); + return Err(self.interrupted_error()); + } + () = notified => {} + } + } + } + fn build_request(&self) -> Request { let mut messages = Vec::new(); if !self.system_prompt.trim().is_empty() { @@ -1433,6 +1543,7 @@ async fn kill_mcp_pid(sandbox: &dyn Sandbox, pid: &str) { mod tests { use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; use fabro_llm::error::{ProviderErrorDetail, ProviderErrorKind}; use fabro_llm::provider::{ProviderAdapter, StreamEventStream}; @@ -1440,6 +1551,7 @@ mod tests { ContentPart, ReasoningEffort, Request, Response, Role, StreamEvent, ToolDefinition, }; use futures::stream; + use tokio::time::{sleep, timeout}; use super::*; use crate::config::ToolApprovalAdapter; @@ -1665,45 +1777,73 @@ mod tests { } #[tokio::test] - async fn steer_event_carries_append_kind() { + async fn steer_event_carries_text() { let mut session = make_session(vec![text_response("OK")]).await; let mut rx = session.subscribe(); session.steer("hi there".to_string()); session.process_input("Do something").await.unwrap(); - // Drain events; find the SteeringInjected one and assert kind=Append - let mut found_kind = None; + let mut found_text = None; while let Ok(ev) = rx.try_recv() { - if let AgentEvent::SteeringInjected { kind, .. } = ev.event { - found_kind = Some(kind); + if let AgentEvent::SteeringInjected { text, .. } = ev.event { + found_text = Some(text); break; } } - assert_eq!(found_kind, Some(SteerKind::Append)); + assert_eq!(found_text.as_deref(), Some("hi there")); } #[tokio::test] - async fn interrupt_with_pushes_interrupt_kind_event() { + async fn pure_interrupt_enters_waiting_for_steer_without_queueing_text() { + let handle = SessionControlHandle::new(); + + handle.interrupt(None); + handle.interrupt(None); + + assert!(handle.is_waiting_for_steer()); + assert_eq!(handle.queue_len(), 0); + assert!(handle.has_pending_control_work()); + } + + #[tokio::test] + async fn pure_interrupt_waits_until_later_steer() { + let mut session = make_session(vec![text_response("OK")]).await; + let handle = session.control_handle(); + handle.interrupt(None); + + let wake_handle = handle.clone(); + tokio::spawn(async move { + sleep(Duration::from_millis(10)).await; + wake_handle.steer("resume now".to_string(), None); + }); + + timeout(Duration::from_secs(1), session.process_input("start")) + .await + .expect("session should wake when steering arrives") + .unwrap(); + + let turns = session.history().turns(); + assert!(matches!(&turns[1], Turn::Steering { content, .. } if content == "resume now")); + assert!(!handle.is_waiting_for_steer()); + } + + #[tokio::test] + async fn interrupt_then_steer_injects_steering_text() { let mut session = make_session(vec![text_response("OK")]).await; let mut rx = session.subscribe(); - // Get a control handle, then push an interrupt steer before any - // round runs. Since no LLM call is in flight, the round_token - // cancel just makes the first iteration loop once before - // proceeding — drain_steering at top-of-loop will pick up the - // queued (text, Interrupt) item. let handle = session.control_handle(); - handle.interrupt_with("stop now".to_string(), None); + handle.interrupt_then_steer("stop now".to_string(), None); session.process_input("start").await.unwrap(); - let mut found_kind = None; + let mut found_text = None; while let Ok(ev) = rx.try_recv() { - if let AgentEvent::SteeringInjected { kind, .. } = ev.event { - found_kind = Some(kind); + if let AgentEvent::SteeringInjected { text, .. } = ev.event { + found_text = Some(text); break; } } - assert_eq!(found_kind, Some(SteerKind::Interrupt)); + assert_eq!(found_text.as_deref(), Some("stop now")); } #[tokio::test] diff --git a/lib/crates/fabro-agent/src/types.rs b/lib/crates/fabro-agent/src/types.rs index f7f2cd71b..4e4cb952e 100644 --- a/lib/crates/fabro-agent/src/types.rs +++ b/lib/crates/fabro-agent/src/types.rs @@ -2,7 +2,6 @@ use std::time::SystemTime; use fabro_llm::Error as LlmError; use fabro_llm::types::{ContentPart, ThinkingData, TokenCounts, ToolCall, ToolResult}; -pub use fabro_types::SteerKind; use serde::{Deserialize, Serialize}; use crate::error::Error; @@ -159,7 +158,6 @@ pub enum AgentEvent { }, SteeringInjected { text: String, - kind: SteerKind, /// Principal that authored the steer. Lifted to top-level /// `RunEvent.actor` by the workflow event-conversion layer; never /// serialized into event props. @@ -316,13 +314,8 @@ impl AgentEvent { Self::SkillExpanded { skill_name } => { debug!(session_id, skill = skill_name.as_str(), "Skill expanded"); } - Self::SteeringInjected { text, kind, .. } => { - debug!( - session_id, - text_len = text.len(), - kind = kind.as_str(), - "Steering injected" - ); + Self::SteeringInjected { text, .. } => { + debug!(session_id, text_len = text.len(), "Steering injected"); } Self::CompactionStarted { estimated_tokens, diff --git a/lib/crates/fabro-api/tests/run_event_round_trip.rs b/lib/crates/fabro-api/tests/run_event_round_trip.rs index 3a1b0fb1f..1ac43d302 100644 --- a/lib/crates/fabro-api/tests/run_event_round_trip.rs +++ b/lib/crates/fabro-api/tests/run_event_round_trip.rs @@ -48,6 +48,36 @@ fn run_event_round_trips_run_created_with_web_url() { assert_run_event_round_trip(value); } +#[test] +fn run_event_round_trips_run_interrupt() { + let value = json!({ + "id": "evt_run_interrupt", + "ts": "2026-04-29T12:00:00Z", + "run_id": fixtures::RUN_1, + "event": "run.interrupt", + "actor": { "kind": "system", "system_kind": "engine" }, + "properties": {} + }); + + assert_run_event_round_trip(value); +} + +#[test] +fn run_event_round_trips_run_steer() { + let value = json!({ + "id": "evt_run_steer", + "ts": "2026-04-29T12:00:00Z", + "run_id": fixtures::RUN_1, + "event": "run.steer", + "actor": { "kind": "system", "system_kind": "engine" }, + "properties": { + "text": "try another approach" + } + }); + + assert_run_event_round_trip(value); +} + #[test] fn run_event_round_trips_stage_started() { let value = json!({ diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index d375e7976..0a8545e09 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -257,8 +257,14 @@ async fn apply_worker_control_line( cancel_token.cancel(); interviewer.interrupt_all().await; } - WorkerControlMessage::Steer { text, kind, actor } => { - steering_hub.deliver(text, kind, Some(actor)); + WorkerControlMessage::Steer { text, actor } => { + steering_hub.deliver_steer(text, Some(actor)); + } + WorkerControlMessage::Interrupt { actor } => { + steering_hub.interrupt(Some(&actor)); + } + WorkerControlMessage::InterruptThenSteer { text, actor } => { + steering_hub.interrupt_then_steer(&text, Some(&actor)); } } } diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index 3a205c9ae..9fa4cea5b 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -805,6 +805,14 @@ impl Client { Ok(()) } + pub async fn interrupt_run(&self, run_id: &RunId) -> Result<()> { + self.send_api(|client| async move { + client.interrupt_run().id(run_id.to_string()).send().await + }) + .await?; + Ok(()) + } + pub async fn steer_run(&self, run_id: &RunId, text: String, interrupt: bool) -> Result<()> { let body: types::SteerRunRequest = types::SteerRunRequest::builder() .text(text) diff --git a/lib/crates/fabro-interview/src/control_protocol.rs b/lib/crates/fabro-interview/src/control_protocol.rs index 14f591b17..daef9a7fe 100644 --- a/lib/crates/fabro-interview/src/control_protocol.rs +++ b/lib/crates/fabro-interview/src/control_protocol.rs @@ -1,5 +1,4 @@ use fabro_types::Principal; -pub use fabro_types::SteerKind; use serde::{Deserialize, Serialize}; use crate::{Answer, AnswerSubmission, AnswerValue}; @@ -35,12 +34,30 @@ impl WorkerControlEnvelope { } #[must_use] - pub fn steer(text: impl Into, kind: SteerKind, actor: Principal) -> Self { + pub fn steer(text: impl Into, actor: Principal) -> Self { Self { v: WORKER_CONTROL_PROTOCOL_VERSION, message: WorkerControlMessage::Steer { text: text.into(), - kind, + actor, + }, + } + } + + #[must_use] + pub fn interrupt(actor: Principal) -> Self { + Self { + v: WORKER_CONTROL_PROTOCOL_VERSION, + message: WorkerControlMessage::Interrupt { actor }, + } + } + + #[must_use] + pub fn interrupt_then_steer(text: impl Into, actor: Principal) -> Self { + Self { + v: WORKER_CONTROL_PROTOCOL_VERSION, + message: WorkerControlMessage::InterruptThenSteer { + text: text.into(), actor, }, } @@ -59,11 +76,11 @@ pub enum WorkerControlMessage { #[serde(rename = "run.cancel")] RunCancel, #[serde(rename = "run.steer")] - Steer { - text: String, - kind: SteerKind, - actor: Principal, - }, + Steer { text: String, actor: Principal }, + #[serde(rename = "run.interrupt")] + Interrupt { actor: Principal }, + #[serde(rename = "run.interrupt_then_steer")] + InterruptThenSteer { text: String, actor: Principal }, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -151,9 +168,36 @@ mod tests { #[test] fn steer_append_round_trips_through_json() { - let envelope = WorkerControlEnvelope::steer( - "try again", - SteerKind::Append, + let envelope = WorkerControlEnvelope::steer("try again", fabro_types::Principal::System { + system_kind: fabro_types::SystemActorKind::Engine, + }); + let json = serde_json::to_string(&envelope).unwrap(); + assert_eq!( + json, + r#"{"v":1,"type":"run.steer","text":"try again","actor":{"kind":"system","system_kind":"engine"}}"# + ); + let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed, envelope); + } + + #[test] + fn interrupt_round_trips_through_json() { + let envelope = WorkerControlEnvelope::interrupt(fabro_types::Principal::System { + system_kind: fabro_types::SystemActorKind::Engine, + }); + let json = serde_json::to_string(&envelope).unwrap(); + assert_eq!( + json, + r#"{"v":1,"type":"run.interrupt","actor":{"kind":"system","system_kind":"engine"}}"# + ); + let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed, envelope); + } + + #[test] + fn interrupt_then_steer_round_trips_through_json() { + let envelope = WorkerControlEnvelope::interrupt_then_steer( + "stop, do X instead", fabro_types::Principal::System { system_kind: fabro_types::SystemActorKind::Engine, }, @@ -161,24 +205,9 @@ mod tests { let json = serde_json::to_string(&envelope).unwrap(); assert_eq!( json, - r#"{"v":1,"type":"run.steer","text":"try again","kind":"append","actor":{"kind":"system","system_kind":"engine"}}"# + r#"{"v":1,"type":"run.interrupt_then_steer","text":"stop, do X instead","actor":{"kind":"system","system_kind":"engine"}}"# ); let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); assert_eq!(parsed, envelope); } - - #[test] - fn steer_interrupt_round_trips_through_json() { - let envelope = WorkerControlEnvelope::steer( - "stop, do X instead", - SteerKind::Interrupt, - fabro_types::Principal::System { - system_kind: fabro_types::SystemActorKind::Engine, - }, - ); - let json = serde_json::to_string(&envelope).unwrap(); - assert!(json.contains(r#""kind":"interrupt""#)); - let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); - assert_eq!(parsed, envelope); - } } diff --git a/lib/crates/fabro-interview/src/lib.rs b/lib/crates/fabro-interview/src/lib.rs index 693ed63db..c242cf14a 100644 --- a/lib/crates/fabro-interview/src/lib.rs +++ b/lib/crates/fabro-interview/src/lib.rs @@ -222,7 +222,7 @@ pub use callback::CallbackInterviewer; pub use console::ConsoleInterviewer; pub use control::{ControlInterviewer, SubmitError}; pub use control_protocol::{ - SteerKind, WORKER_CONTROL_PROTOCOL_VERSION, WorkerControlAnswer, WorkerControlEnvelope, + WORKER_CONTROL_PROTOCOL_VERSION, WorkerControlAnswer, WorkerControlEnvelope, WorkerControlMessage, }; pub use queue::QueueInterviewer; diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index e07105556..5ce452d10 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -303,22 +303,53 @@ impl RunAnswerTransport { /// Forward a steer to the worker (subprocess) or directly into the /// in-process steering hub. - async fn steer( - &self, - text: String, - kind: fabro_types::SteerKind, - actor: Principal, - ) -> Result<(), AnswerTransportError> { + async fn steer(&self, text: String, actor: Principal) -> Result<(), AnswerTransportError> { match self { Self::Subprocess { control_tx } => { - let message = WorkerControlEnvelope::steer(text, kind, actor); + let message = WorkerControlEnvelope::steer(text, actor); timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message)) .await .map_err(|_| AnswerTransportError::Timeout)? .map_err(|_| AnswerTransportError::Closed) } Self::InProcess { steering_hub, .. } => { - steering_hub.deliver(text, kind, Some(actor)); + steering_hub.deliver_steer(text, Some(actor)); + Ok(()) + } + } + } + + async fn interrupt(&self, actor: Principal) -> Result<(), AnswerTransportError> { + match self { + Self::Subprocess { control_tx } => { + let message = WorkerControlEnvelope::interrupt(actor); + timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message)) + .await + .map_err(|_| AnswerTransportError::Timeout)? + .map_err(|_| AnswerTransportError::Closed) + } + Self::InProcess { steering_hub, .. } => { + steering_hub.interrupt(Some(&actor)); + Ok(()) + } + } + } + + async fn interrupt_then_steer( + &self, + text: String, + actor: Principal, + ) -> Result<(), AnswerTransportError> { + match self { + Self::Subprocess { control_tx } => { + let message = WorkerControlEnvelope::interrupt_then_steer(text, actor); + timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message)) + .await + .map_err(|_| AnswerTransportError::Timeout)? + .map_err(|_| AnswerTransportError::Closed) + } + Self::InProcess { steering_hub, .. } => { + steering_hub.interrupt_then_steer(&text, Some(&actor)); Ok(()) } } diff --git a/lib/crates/fabro-server/src/server/handler/steer.rs b/lib/crates/fabro-server/src/server/handler/steer.rs index 9b283b297..02cbfa300 100644 --- a/lib/crates/fabro-server/src/server/handler/steer.rs +++ b/lib/crates/fabro-server/src/server/handler/steer.rs @@ -6,7 +6,7 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::post; use fabro_api::types::SteerRunRequest; -use fabro_types::{Principal, SteerKind}; +use fabro_types::Principal; use fabro_workflow::run_status::RunStatus; use super::super::{AnswerTransportError, AppState, parse_run_id_path, reject_if_archived}; @@ -14,7 +14,21 @@ use crate::error::ApiError; use crate::principal_middleware::RequiredUser; pub(super) fn routes() -> axum::Router> { - axum::Router::new().route("/runs/{id}/steer", post(steer_run)) + axum::Router::new() + .route("/runs/{id}/steer", post(steer_run)) + .route("/runs/{id}/interrupt", post(interrupt_run)) +} + +enum RunControlRequest { + Steer { text: String }, + Interrupt, + InterruptThenSteer { text: String }, +} + +impl RunControlRequest { + const fn requires_active_api_session(&self) -> bool { + matches!(self, Self::Interrupt | Self::InterruptThenSteer { .. }) + } } async fn steer_run( @@ -39,12 +53,37 @@ async fn steer_run( if text.trim().is_empty() { return ApiError::bad_request("Steer text must not be empty.").into_response(); } - let kind = if interrupt { - SteerKind::Interrupt + let control = if interrupt { + RunControlRequest::InterruptThenSteer { text } } else { - SteerKind::Append + RunControlRequest::Steer { text } }; + control_run(auth, state, id.to_string(), control).await +} + +async fn interrupt_run( + auth: RequiredUser, + State(state): State>, + Path(id): Path, +) -> Response { + control_run(auth, state, id, RunControlRequest::Interrupt).await +} + +async fn control_run( + auth: RequiredUser, + state: Arc, + id: String, + control: RunControlRequest, +) -> Response { + let id = match parse_run_id_path(&id) { + Ok(id) => id, + Err(response) => return response, + }; + if let Some(response) = reject_if_archived(state.as_ref(), &id).await { + return response; + } + // Status + steerability gate. Take the answer_transport snapshot under // the same lock so we can hand it off without further state races. let answer_transport = { @@ -65,16 +104,29 @@ async fn steer_run( | RunStatus::Queued | RunStatus::Starting | RunStatus::Paused { .. } => { - return ApiError::new(StatusCode::CONFLICT, "Run is not currently running.") - .into_response(); + return ApiError::with_code( + StatusCode::CONFLICT, + "Run is not currently running.", + "run_not_steerable", + ) + .into_response(); } RunStatus::Failed { .. } | RunStatus::Succeeded { .. } | RunStatus::Removing | RunStatus::Dead | RunStatus::Archived { .. } => { - return ApiError::new(StatusCode::CONFLICT, "Run is no longer steerable.") - .into_response(); + let code = if matches!(&control, RunControlRequest::Interrupt) { + "run_not_interruptible" + } else { + "run_not_steerable" + }; + return ApiError::with_code( + StatusCode::CONFLICT, + "Run is no longer steerable.", + code, + ) + .into_response(); } RunStatus::Running => {} } @@ -91,28 +143,47 @@ async fn steer_run( ) .into_response(); } + if managed_run.active_api_stages.is_empty() && control.requires_active_api_session() { + return ApiError::with_code( + StatusCode::CONFLICT, + "Run has no active API-mode agent session.", + "no_active_api_session", + ) + .into_response(); + } managed_run.answer_transport.clone() }; let Some(answer_transport) = answer_transport else { - return ApiError::new( + return ApiError::with_code( StatusCode::SERVICE_UNAVAILABLE, "Run has no live worker control channel.", + "worker_control_unavailable", ) .into_response(); }; let actor = Principal::User(auth.0); - match answer_transport.steer(text, kind, actor).await { + let result = match control { + RunControlRequest::Steer { text } => answer_transport.steer(text, actor).await, + RunControlRequest::Interrupt => answer_transport.interrupt(actor).await, + RunControlRequest::InterruptThenSteer { text } => { + answer_transport.interrupt_then_steer(text, actor).await + } + }; + + match result { Ok(()) => StatusCode::ACCEPTED.into_response(), - Err(AnswerTransportError::Timeout) => ApiError::new( + Err(AnswerTransportError::Timeout) => ApiError::with_code( StatusCode::SERVICE_UNAVAILABLE, "Worker control channel timed out.", + "worker_control_unavailable", ) .into_response(), - Err(AnswerTransportError::Closed) => ApiError::new( + Err(AnswerTransportError::Closed) => ApiError::with_code( StatusCode::SERVICE_UNAVAILABLE, "Worker control channel is closed.", + "worker_control_unavailable", ) .into_response(), } diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index bb3403506..dcd3bd0b2 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -12,14 +12,16 @@ use chrono::{Duration as ChronoDuration, Utc}; use fabro_auth::{AuthCredential, AuthDetails}; use fabro_config::ServerSettingsBuilder; use fabro_config::bind::Bind; -use fabro_interview::{AnswerValue, ControlInterviewer, Interviewer, Question}; +use fabro_interview::{ + AnswerValue, ControlInterviewer, Interviewer, Question, WorkerControlMessage, +}; use fabro_llm::types::{Message as LlmMessage, Request as LlmRequest}; use fabro_model::Provider; use fabro_types::settings::ServerAuthMethod; use fabro_types::{ AttrValue, AuthMethod, CommandTermination, FailureCategory, FailureDetail, Graph, - InterviewQuestionRecord, Outcome, QuestionType, RunBlobId, RunId, RunSpec, SystemActorKind, - fixtures, + InterviewQuestionRecord, Outcome, QuestionType, RunBlobId, RunId, RunSpec, SuccessReason, + SystemActorKind, fixtures, }; use fabro_util::check_report::CheckStatus; use httpmock::Method::{GET, POST}; @@ -1875,6 +1877,63 @@ async fn subprocess_answer_transport_cancel_run_enqueues_cancel_message() { ); } +#[tokio::test] +async fn subprocess_answer_transport_steer_enqueues_plain_steer_message() { + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let transport = RunAnswerTransport::Subprocess { control_tx }; + let actor = Principal::System { + system_kind: SystemActorKind::Engine, + }; + + transport + .steer("try again".to_string(), actor.clone()) + .await + .unwrap(); + + assert_eq!( + control_rx.recv().await, + Some(WorkerControlEnvelope::steer("try again", actor)) + ); +} + +#[tokio::test] +async fn subprocess_answer_transport_interrupt_enqueues_interrupt_message() { + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let transport = RunAnswerTransport::Subprocess { control_tx }; + let actor = Principal::System { + system_kind: SystemActorKind::Engine, + }; + + transport.interrupt(actor.clone()).await.unwrap(); + + assert_eq!( + control_rx.recv().await, + Some(WorkerControlEnvelope::interrupt(actor)) + ); +} + +#[tokio::test] +async fn subprocess_answer_transport_interrupt_then_steer_enqueues_single_combined_message() { + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let transport = RunAnswerTransport::Subprocess { control_tx }; + let actor = Principal::System { + system_kind: SystemActorKind::Engine, + }; + + transport + .interrupt_then_steer("try again".to_string(), actor.clone()) + .await + .unwrap(); + + assert_eq!( + control_rx.recv().await, + Some(WorkerControlEnvelope::interrupt_then_steer( + "try again", + actor + )) + ); +} + #[tokio::test] async fn in_process_answer_transport_cancel_run_cancels_pending_interviews() { let interviewer = Arc::new(ControlInterviewer::new()); @@ -6120,6 +6179,188 @@ async fn steer_empty_text_returns_bad_request() { ); } +fn insert_running_control_run( + state: &Arc, + run_id: RunId, + answer_transport: Option, +) -> tempfile::TempDir { + let temp_dir = tempfile::tempdir().unwrap(); + let mut run = managed_run( + String::new(), + RunStatus::Running, + chrono::Utc::now(), + temp_dir.path().join(run_id.to_string()), + RunExecutionMode::Start, + ); + run.answer_transport = answer_transport; + state + .runs + .lock() + .expect("runs lock poisoned") + .insert(run_id, run); + temp_dir +} + +#[tokio::test] +async fn steer_without_active_api_session_forwards_plain_steer_for_buffering() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let _temp_dir = insert_running_control_run( + &state, + run_id, + Some(RunAnswerTransport::Subprocess { control_tx }), + ); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/steer"))) + .header("content-type", "application/json") + .body(Body::from(r#"{"text":"try again"}"#)) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::ACCEPTED).await; + let envelope = control_rx.recv().await.unwrap(); + assert!(matches!( + envelope.message, + WorkerControlMessage::Steer { ref text, .. } if text == "try again" + )); +} + +#[tokio::test] +async fn steer_interrupt_without_active_api_session_returns_conflict() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let (control_tx, _control_rx) = tokio::sync::mpsc::channel(1); + let _temp_dir = insert_running_control_run( + &state, + run_id, + Some(RunAnswerTransport::Subprocess { control_tx }), + ); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/steer"))) + .header("content-type", "application/json") + .body(Body::from(r#"{"text":"try again","interrupt":true}"#)) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::CONFLICT); + let body = body_json(response.into_body()).await; + assert_eq!(body["errors"][0]["code"], "no_active_api_session"); +} + +#[tokio::test] +async fn interrupt_with_active_api_session_forwards_interrupt() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let stage_id = StageId::new("agent", 1); + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let _temp_dir = insert_running_control_run( + &state, + run_id, + Some(RunAnswerTransport::Subprocess { control_tx }), + ); + { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + runs.get_mut(&run_id) + .unwrap() + .active_api_stages + .insert(stage_id, "session-a".to_string()); + } + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/interrupt"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::ACCEPTED).await; + let envelope = control_rx.recv().await.unwrap(); + assert!(matches!( + envelope.message, + WorkerControlMessage::Interrupt { + actor: Principal::User(_), + } + )); +} + +#[tokio::test] +async fn steer_interrupt_with_active_api_session_forwards_combined_control_message() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let stage_id = StageId::new("agent", 1); + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let _temp_dir = insert_running_control_run( + &state, + run_id, + Some(RunAnswerTransport::Subprocess { control_tx }), + ); + { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + runs.get_mut(&run_id) + .unwrap() + .active_api_stages + .insert(stage_id, "session-a".to_string()); + } + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/steer"))) + .header("content-type", "application/json") + .body(Body::from(r#"{"text":"try again","interrupt":true}"#)) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::ACCEPTED).await; + let envelope = control_rx.recv().await.unwrap(); + assert!(matches!( + envelope.message, + WorkerControlMessage::InterruptThenSteer { ref text, .. } if text == "try again" + )); +} + +#[tokio::test] +async fn interrupt_terminal_run_returns_run_not_interruptible() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let temp_dir = tempfile::tempdir().unwrap(); + { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + runs.insert( + run_id, + managed_run( + String::new(), + RunStatus::Succeeded { + reason: SuccessReason::Completed, + }, + chrono::Utc::now(), + temp_dir.path().join(run_id.to_string()), + RunExecutionMode::Start, + ), + ); + } + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/interrupt"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::CONFLICT); + let body = body_json(response.into_body()).await; + assert_eq!(body["errors"][0]["code"], "run_not_interruptible"); +} + #[test] fn active_api_stage_projection_ignores_stale_deactivation() { let state = test_app_state(); diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index fafb01c05..7b1073bca 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -31,7 +31,6 @@ pub mod stage_completion; pub mod stage_id; pub mod start; pub mod status; -pub mod steering; pub use artifact::ArtifactUpload; pub use auth::{IdpIdentity, IdpIdentityError}; @@ -87,4 +86,3 @@ pub use status::{ BlockedReason, FailureReason, InvalidTransition, ParseFailureReasonError, ParseSuccessReasonError, RunControlAction, RunStatus, SuccessReason, TerminalStatus, }; -pub use steering::SteerKind; diff --git a/lib/crates/fabro-types/src/run_event/agent.rs b/lib/crates/fabro-types/src/run_event/agent.rs index 9ef85d027..da5856167 100644 --- a/lib/crates/fabro-types/src/run_event/agent.rs +++ b/lib/crates/fabro-types/src/run_event/agent.rs @@ -2,7 +2,6 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use super::BilledTokenCounts; -use crate::SteerKind; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct AgentSessionStartedProps { @@ -107,14 +106,15 @@ pub struct AgentTurnLimitReachedProps { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct AgentSteeringInjectedProps { pub text: String, - pub kind: SteerKind, pub visit: u32, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct AgentSteerBufferedProps { - pub kind: SteerKind, -} +#[allow( + clippy::empty_structs_with_brackets, + reason = "This type must serialize as {} rather than null." +)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct AgentSteerBufferedProps {} #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index 9df4a6456..6ac1f331d 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -62,6 +62,10 @@ pub enum EventBody { RunStarting(RunStatusTransitionProps), #[serde(rename = "run.running")] RunRunning(RunStatusTransitionProps), + #[serde(rename = "run.interrupt")] + RunInterrupt(RunInterruptProps), + #[serde(rename = "run.steer")] + RunSteer(RunSteerProps), #[serde(rename = "run.blocked")] RunBlocked(RunBlockedProps), #[serde(rename = "run.unblocked")] @@ -350,6 +354,8 @@ impl EventBody { Self::RunQueued(_) => "run.queued", Self::RunStarting(_) => "run.starting", Self::RunRunning(_) => "run.running", + Self::RunInterrupt(_) => "run.interrupt", + Self::RunSteer(_) => "run.steer", Self::RunBlocked(_) => "run.blocked", Self::RunUnblocked(_) => "run.unblocked", Self::RunRemoving(_) => "run.removing", @@ -493,6 +499,8 @@ fn is_known_event_name(event: &str) -> bool { | "run.queued" | "run.starting" | "run.running" + | "run.interrupt" + | "run.steer" | "run.blocked" | "run.unblocked" | "run.removing" @@ -921,6 +929,58 @@ mod tests { assert_eq!(body.event_name(), "interview.interrupted"); } + #[test] + fn run_interrupt_round_trips_with_empty_properties_and_actor() { + let line = json!({ + "id": "evt_interrupt", + "ts": "2026-04-04T12:00:00Z", + "run_id": fixtures::RUN_1, + "event": "run.interrupt", + "actor": { "kind": "system", "system_kind": "engine" }, + "properties": {} + }); + + let parsed = RunEvent::from_value(line.clone()).unwrap(); + assert!(matches!(parsed.body, EventBody::RunInterrupt(_))); + assert_eq!(parsed.to_value().unwrap(), line); + } + + #[test] + fn run_steer_round_trips_with_text_and_actor() { + let line = json!({ + "id": "evt_steer", + "ts": "2026-04-04T12:00:00Z", + "run_id": fixtures::RUN_1, + "event": "run.steer", + "actor": { "kind": "system", "system_kind": "engine" }, + "properties": { "text": "try another approach" } + }); + + let parsed = RunEvent::from_value(line.clone()).unwrap(); + assert!(matches!( + &parsed.body, + EventBody::RunSteer(props) if props.text == "try another approach" + )); + assert_eq!(parsed.to_value().unwrap(), line); + } + + #[test] + fn run_interrupt_then_steer_is_not_a_known_persisted_event() { + let line = json!({ + "id": "evt_combined", + "ts": "2026-04-04T12:00:00.000Z", + "run_id": fixtures::RUN_1, + "event": "run.interrupt_then_steer", + "properties": { "text": "try another approach" } + }); + + let parsed = RunEvent::from_value(line).unwrap(); + assert!(matches!( + parsed.body, + EventBody::Unknown { ref name, .. } if name == "run.interrupt_then_steer" + )); + } + #[test] fn run_submitted_round_trip_preserves_definition_blob() { let line = json!({ diff --git a/lib/crates/fabro-types/src/run_event/run.rs b/lib/crates/fabro-types/src/run_event/run.rs index 6c8e1ddd1..176260ac6 100644 --- a/lib/crates/fabro-types/src/run_event/run.rs +++ b/lib/crates/fabro-types/src/run_event/run.rs @@ -69,6 +69,18 @@ pub struct RunStatusTransitionProps {} #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] pub struct RunStatusEffectProps {} +#[allow( + clippy::empty_structs_with_brackets, + reason = "This type must serialize as {} rather than null." +)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +pub struct RunInterruptProps {} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct RunSteerProps { + pub text: String, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct RunSubmittedProps { #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/lib/crates/fabro-types/src/steering.rs b/lib/crates/fabro-types/src/steering.rs deleted file mode 100644 index 94515d93f..000000000 --- a/lib/crates/fabro-types/src/steering.rs +++ /dev/null @@ -1,52 +0,0 @@ -use serde::{Deserialize, Serialize}; - -/// Two flavors of mid-run steering messages delivered to a live agent -/// session. -/// -/// - `Append` — push to the steering queue; the agent picks it up at the next -/// turn boundary. -/// - `Interrupt` — cancel the in-flight LLM stream / tool call in the current -/// round, then deliver the message as the next user turn. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "lowercase")] -pub enum SteerKind { - Append, - Interrupt, -} - -impl SteerKind { - #[must_use] - pub const fn as_str(self) -> &'static str { - match self { - Self::Append => "append", - Self::Interrupt => "interrupt", - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn append_round_trips_through_json() { - let json = serde_json::to_string(&SteerKind::Append).unwrap(); - assert_eq!(json, "\"append\""); - let parsed: SteerKind = serde_json::from_str(&json).unwrap(); - assert_eq!(parsed, SteerKind::Append); - } - - #[test] - fn interrupt_round_trips_through_json() { - let json = serde_json::to_string(&SteerKind::Interrupt).unwrap(); - assert_eq!(json, "\"interrupt\""); - let parsed: SteerKind = serde_json::from_str(&json).unwrap(); - assert_eq!(parsed, SteerKind::Interrupt); - } - - #[test] - fn unknown_value_fails_to_deserialize() { - let result: Result = serde_json::from_str("\"unknown\""); - assert!(result.is_err()); - } -} diff --git a/lib/crates/fabro-workflow/src/event/convert.rs b/lib/crates/fabro-workflow/src/event/convert.rs index 7f3c94c21..e5a287443 100644 --- a/lib/crates/fabro-workflow/src/event/convert.rs +++ b/lib/crates/fabro-workflow/src/event/convert.rs @@ -100,6 +100,12 @@ fn event_body_from_event(event: &Event) -> EventBody { Event::RunRunning => { EventBody::RunRunning(fabro_types::RunStatusTransitionProps::default()) } + Event::RunInterrupt { .. } => { + EventBody::RunInterrupt(fabro_types::RunInterruptProps::default()) + } + Event::RunSteer { text, .. } => { + EventBody::RunSteer(fabro_types::RunSteerProps { text: text.clone() }) + } Event::RunBlocked { blocked_reason } => { EventBody::RunBlocked(fabro_types::RunBlockedProps { blocked_reason: *blocked_reason, @@ -597,10 +603,9 @@ fn event_body_from_event(event: &Event) -> EventBody { visit: *visit, }) } - AgentEvent::SteeringInjected { text, kind, .. } => { + AgentEvent::SteeringInjected { text, .. } => { EventBody::AgentSteeringInjected(fabro_types::AgentSteeringInjectedProps { text: text.clone(), - kind: *kind, visit: *visit, }) } @@ -1029,8 +1034,8 @@ fn event_body_from_event(event: &Event) -> EventBody { Event::AgentSessionEnded { .. } => { EventBody::AgentSessionEnded(fabro_types::AgentSessionEndedProps {}) } - Event::AgentSteerBuffered { kind, .. } => { - EventBody::AgentSteerBuffered(fabro_types::AgentSteerBufferedProps { kind: *kind }) + Event::AgentSteerBuffered { .. } => { + EventBody::AgentSteerBuffered(fabro_types::AgentSteerBufferedProps::default()) } Event::AgentSteerDropped { reason, count, .. } => { EventBody::AgentSteerDropped(fabro_types::AgentSteerDroppedProps { diff --git a/lib/crates/fabro-workflow/src/event/events.rs b/lib/crates/fabro-workflow/src/event/events.rs index 04bc998b2..f752b7925 100644 --- a/lib/crates/fabro-workflow/src/event/events.rs +++ b/lib/crates/fabro-workflow/src/event/events.rs @@ -3,7 +3,7 @@ use std::collections::BTreeMap; use ::fabro_types::{ BilledTokenCounts, BlockedReason, CommandTermination, FailureReason, ForkSourceRef, GitContext, ParallelBranchId, Principal, PullRequestRecord, RunBlobId, RunId, RunNoticeLevel, - RunProvenance, StageId, SteerKind, SuccessReason, run_event as fabro_types, + RunProvenance, StageId, SuccessReason, run_event as fabro_types, }; use fabro_agent::{AgentEvent, SandboxEvent}; use serde::{Deserialize, Serialize}; @@ -68,6 +68,15 @@ pub enum Event { RunQueued, RunStarting, RunRunning, + RunInterrupt { + #[serde(default, skip_serializing_if = "Option::is_none")] + actor: Option, + }, + RunSteer { + text: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + actor: Option, + }, RunBlocked { blocked_reason: BlockedReason, }, @@ -562,7 +571,6 @@ pub enum Event { /// A steer arrived with no active session and was parked in the run-wide /// pending buffer. The actor (steer author) is lifted to top-level. AgentSteerBuffered { - kind: SteerKind, #[serde(default, skip_serializing_if = "Option::is_none")] actor: Option, }, @@ -705,6 +713,12 @@ impl Event { Self::RunRunning => { info!("Run running"); } + Self::RunInterrupt { .. } => { + info!("Run interrupt accepted"); + } + Self::RunSteer { text, .. } => { + info!(text_len = text.len(), "Run steer accepted"); + } Self::RunBlocked { blocked_reason } => { info!(?blocked_reason, "Run blocked"); } @@ -1340,8 +1354,8 @@ impl Event { Self::AgentSessionEnded { session_id, .. } => { debug!(session_id, "Agent session ended"); } - Self::AgentSteerBuffered { kind, .. } => { - debug!(kind = kind.as_str(), "Steer buffered (no active session)"); + Self::AgentSteerBuffered { .. } => { + debug!("Steer buffered (no active session)"); } Self::AgentSteerDropped { reason, count, .. } => { warn!(?reason, count, "Steer dropped"); diff --git a/lib/crates/fabro-workflow/src/event/names.rs b/lib/crates/fabro-workflow/src/event/names.rs index fda341bc2..f91290fa2 100644 --- a/lib/crates/fabro-workflow/src/event/names.rs +++ b/lib/crates/fabro-workflow/src/event/names.rs @@ -11,6 +11,8 @@ pub fn event_name(event: &Event) -> &'static str { Event::RunQueued => "run.queued", Event::RunStarting => "run.starting", Event::RunRunning => "run.running", + Event::RunInterrupt { .. } => "run.interrupt", + Event::RunSteer { .. } => "run.steer", Event::RunBlocked { .. } => "run.blocked", Event::RunUnblocked => "run.unblocked", Event::RunRemoving => "run.removing", diff --git a/lib/crates/fabro-workflow/src/event/stored_fields.rs b/lib/crates/fabro-workflow/src/event/stored_fields.rs index 8408af91f..92527aee3 100644 --- a/lib/crates/fabro-workflow/src/event/stored_fields.rs +++ b/lib/crates/fabro-workflow/src/event/stored_fields.rs @@ -63,6 +63,8 @@ fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields { Event::RunCancelRequested { actor } | Event::RunPauseRequested { actor } | Event::RunUnpauseRequested { actor } + | Event::RunInterrupt { actor } + | Event::RunSteer { actor, .. } | Event::RunArchived { actor } | Event::RunUnarchived { actor, .. } | Event::InterviewCompleted { actor, .. } diff --git a/lib/crates/fabro-workflow/src/handler/llm/activation_lease.rs b/lib/crates/fabro-workflow/src/handler/llm/activation_lease.rs index 6d9d089a4..1d2362ec8 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/activation_lease.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/activation_lease.rs @@ -69,13 +69,13 @@ impl ActivationLease { self.hub.detach(&self.stage_id, &self.session_id); } - pub fn release_if_queue_empty(&self, handle: &SessionControlHandle) -> bool { + pub fn release_if_no_pending_control_work(&self, handle: &SessionControlHandle) -> bool { if self.released.load(Ordering::Acquire) { return true; } if !self .hub - .detach_if_queue_empty(&self.stage_id, &self.session_id, handle) + .detach_if_no_pending_control_work(&self.stage_id, &self.session_id, handle) { return false; } @@ -111,7 +111,7 @@ mod tests { use std::sync::{Arc, Mutex}; use fabro_agent::SessionControlHandle; - use fabro_types::{RunId, SteerKind}; + use fabro_types::RunId; use super::*; @@ -153,7 +153,7 @@ mod tests { let stage_id = StageId::new("agent", 1); let handle = SessionControlHandle::new(); - hub.deliver("queued".to_string(), SteerKind::Interrupt, None); + hub.deliver_steer("queued".to_string(), None); let _lease = ActivationLease::activate( options( stage_id.clone(), @@ -167,6 +167,7 @@ mod tests { assert_eq!(handle.queue_len(), 1); assert_eq!(names.lock().unwrap().as_slice(), [ + "run.steer", "agent.steer.buffered", "agent.session.activated" ]); diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 4ffe27cd7..127985b68 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -978,7 +978,7 @@ impl CompletionCoordinator for SteeringCompletionCoordinator { let Some(active_lease) = lease.as_ref() else { return false; }; - if active_lease.release_if_queue_empty(&self.handle) { + if active_lease.release_if_no_pending_control_work(&self.handle) { lease.take(); false } else { diff --git a/lib/crates/fabro-workflow/src/steering_hub.rs b/lib/crates/fabro-workflow/src/steering_hub.rs index db9b33ebd..7ef2b3c97 100644 --- a/lib/crates/fabro-workflow/src/steering_hub.rs +++ b/lib/crates/fabro-workflow/src/steering_hub.rs @@ -13,14 +13,15 @@ //! - `pending` is `std::sync::Mutex` taken under the active read lock. //! - All methods are sync — no `.await` while holding any lock — so the //! `CompletionCoordinator::on_natural_completion` close-the-door dance can -//! call `detach_if_queue_empty(...)` synchronously from the agent loop. +//! call `detach_if_no_pending_control_work(...)` synchronously from the +//! agent loop. use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex, RwLock}; use fabro_agent::{SessionControlHandle, SteeringItem}; use fabro_types::run_event::AgentSteerDroppedReason; -use fabro_types::{Principal, StageId, SteerKind}; +use fabro_types::{Principal, StageId}; use crate::event::{Emitter, Event}; @@ -111,19 +112,16 @@ impl SteeringHub { } } - /// Drain pending run-wide steers into `handle` as `Append` messages. + /// Drain pending run-wide steers into `handle`. pub fn drain_pending_into(&self, stage_id: &StageId, handle: &SessionControlHandle) { let pending: Vec = { let mut pending = self.pending.lock().expect("pending lock poisoned"); pending.drain(..).collect() }; for item in pending { - // Buffered steers always flush as Append — the original - // Interrupt semantics no longer make sense once the round has - // rolled over. Self::enqueue_into_session_queue( handle, - (item.text, SteerKind::Append, item.actor), + (item.text, item.actor), &self.emitter, Some(stage_id), ); @@ -148,7 +146,7 @@ impl SteeringHub { /// is empty and the active session id matches, remove the stage and /// return `true`. If the queue is non-empty, leave the registration /// intact and return `false`. - pub fn detach_if_queue_empty( + pub fn detach_if_no_pending_control_work( &self, stage_id: &StageId, session_id: &str, @@ -158,7 +156,7 @@ impl SteeringHub { let Some(entry) = active.get(stage_id) else { return false; }; - if entry.session_id != session_id || !handle.queue_is_empty() { + if entry.session_id != session_id || handle.has_pending_control_work() { return false; } active.remove(stage_id); @@ -168,7 +166,12 @@ impl SteeringHub { /// Deliver a steer from the HTTP control plane. Broadcasts to every /// active session if any are registered, otherwise parks the message /// in the run-wide pending buffer. - pub fn deliver(&self, text: String, kind: SteerKind, actor: Option) { + pub fn deliver_steer(&self, text: String, actor: Option) { + self.emitter.emit(&Event::RunSteer { + text: text.clone(), + actor: actor.clone(), + }); + // Hold the active read lock for the entire decide-and-dispatch // step so register/unregister cannot race with this push. let active = self.active.read().expect("active lock poisoned"); @@ -196,8 +199,7 @@ impl SteeringHub { visit: None, }); } - self.emitter - .emit(&Event::AgentSteerBuffered { kind, actor }); + self.emitter.emit(&Event::AgentSteerBuffered { actor }); drop(active); return; } @@ -206,13 +208,62 @@ impl SteeringHub { for (stage_id, entry) in active.iter() { Self::enqueue_into_session_queue( &entry.handle, - (text.clone(), kind, actor.clone()), + (text.clone(), actor.clone()), &self.emitter, Some(stage_id), ); } } + /// Interrupt every active API-mode session. Does not buffer when no + /// active session exists. + pub fn interrupt(&self, actor: Option<&Principal>) { + let active = self.active.read().expect("active lock poisoned"); + if active.is_empty() { + return; + } + + self.emitter.emit(&Event::RunInterrupt { + actor: actor.cloned(), + }); + for entry in active.values() { + entry.handle.interrupt(actor.cloned()); + } + } + + /// Atomically apply interrupt semantics, then deliver steering text to + /// every active API-mode session. Emits persisted run events in the same + /// order. + pub fn interrupt_then_steer(&self, text: &str, actor: Option<&Principal>) { + let active = self.active.read().expect("active lock poisoned"); + if active.is_empty() { + return; + } + + self.emitter.emit(&Event::RunInterrupt { + actor: actor.cloned(), + }); + self.emitter.emit(&Event::RunSteer { + text: text.to_string(), + actor: actor.cloned(), + }); + + for (stage_id, entry) in active.iter() { + if let Some((_, evicted_actor)) = entry.handle.interrupt_then_enqueue_bounded( + (text.to_string(), actor.cloned()), + PER_SESSION_QUEUE_CAP, + ) { + self.emitter.emit(&Event::AgentSteerDropped { + reason: AgentSteerDroppedReason::QueueFull, + count: 1, + actor: evicted_actor, + node_id: Some(stage_id.node_id().to_string()), + visit: Some(stage_id.visit()), + }); + } + } + } + /// Drain any unconsumed pending steers and emit a single /// `agent.steer.dropped` event with `reason: run_ended`. Called from /// `operations::start` after the pipeline finishes (success or @@ -244,7 +295,7 @@ impl SteeringHub { emitter: &Emitter, stage_id: Option<&StageId>, ) { - if let Some((.., evicted_actor)) = handle.enqueue_bounded(item, PER_SESSION_QUEUE_CAP) { + if let Some((_, evicted_actor)) = handle.enqueue_bounded(item, PER_SESSION_QUEUE_CAP) { emitter.emit(&Event::AgentSteerDropped { reason: AgentSteerDroppedReason::QueueFull, count: 1, @@ -258,29 +309,48 @@ impl SteeringHub { #[cfg(test)] mod tests { + use std::sync::{Arc, Mutex}; + use fabro_agent::SessionControlHandle; - use fabro_types::{Principal, StageId, SteerKind, SystemActorKind}; + use fabro_types::{Principal, RunId, StageId, SystemActorKind}; use super::SteeringHub; + use crate::event::Emitter; + + fn hub_with_event_names() -> (Arc, Arc>>) { + let emitter = Arc::new(Emitter::new(RunId::new())); + let names = Arc::new(Mutex::new(Vec::new())); + let names_for_listener = Arc::clone(&names); + emitter.on_event(move |event| { + names_for_listener + .lock() + .unwrap() + .push(event.event_name().to_string()); + }); + (Arc::new(SteeringHub::new(emitter)), names) + } #[test] fn deliver_with_no_active_buffers_message() { - let hub = SteeringHub::for_tests(); - hub.deliver( + let (hub, names) = hub_with_event_names(); + hub.deliver_steer( "hi".into(), - SteerKind::Append, Some(Principal::System { system_kind: SystemActorKind::Engine, }), ); assert_eq!(hub.pending_len(), 1); + assert_eq!(names.lock().unwrap().as_slice(), [ + "run.steer", + "agent.steer.buffered" + ]); } #[test] fn drain_pending_at_run_end_clears_buffer() { let hub = SteeringHub::for_tests(); - hub.deliver("a".into(), SteerKind::Append, None); - hub.deliver("b".into(), SteerKind::Append, None); + hub.deliver_steer("a".into(), None); + hub.deliver_steer("b".into(), None); assert_eq!(hub.pending_len(), 2); hub.drain_pending_at_run_end(); assert_eq!(hub.pending_len(), 0); @@ -290,7 +360,7 @@ mod tests { fn pending_buffer_evicts_oldest_at_cap() { let hub = SteeringHub::for_tests(); for i in 0..(super::PER_RUN_PENDING_CAP + 5) { - hub.deliver(format!("msg{i}"), SteerKind::Append, None); + hub.deliver_steer(format!("msg{i}"), None); } assert_eq!(hub.pending_len(), super::PER_RUN_PENDING_CAP); } @@ -306,8 +376,8 @@ mod tests { #[test] fn attach_and_drain_pending_into_first_session() { let hub = SteeringHub::for_tests(); - hub.deliver("queued1".into(), SteerKind::Append, None); - hub.deliver("queued2".into(), SteerKind::Interrupt, None); + hub.deliver_steer("queued1".into(), None); + hub.deliver_steer("queued2".into(), None); assert_eq!(hub.pending_len(), 2); let stage = StageId::new("agent-node", 1); @@ -330,7 +400,7 @@ mod tests { assert!(hub.attach_handle(&stage_a, "session-a", &handle_a)); assert!(hub.attach_handle(&stage_b, "session-b", &handle_b)); - hub.deliver("hello".into(), SteerKind::Append, None); + hub.deliver_steer("hello".into(), None); assert_eq!(handle_a.queue_len(), 1); assert_eq!(handle_b.queue_len(), 1); @@ -343,7 +413,7 @@ mod tests { let stage = StageId::new("a", 1); let handle1 = SessionControlHandle::new(); assert!(hub.attach_handle(&stage, "session-a", &handle1)); - hub.deliver("x".into(), SteerKind::Append, None); + hub.deliver_steer("x".into(), None); assert_eq!(handle1.queue_len(), 1); let handle2 = SessionControlHandle::new(); @@ -359,36 +429,73 @@ mod tests { assert!(hub.attach_handle(&stage, "session-a", &handle)); assert!(!hub.detach(&stage, "session-b")); - hub.deliver("still-active".into(), SteerKind::Append, None); + hub.deliver_steer("still-active".into(), None); assert_eq!(handle.queue_len(), 1); assert_eq!(hub.active_count(), 1); } #[test] - fn detach_if_queue_empty_respects_session_id_and_queue_state() { + fn detach_if_no_pending_control_work_respects_session_id_and_queue_state() { let hub = SteeringHub::for_tests(); let stage = StageId::new("a", 1); let handle = SessionControlHandle::new(); assert!(hub.attach_handle(&stage, "session-a", &handle)); - assert!(!hub.detach_if_queue_empty(&stage, "session-b", &handle)); - hub.deliver("queued".into(), SteerKind::Append, None); - assert!(!hub.detach_if_queue_empty(&stage, "session-a", &handle)); + assert!(!hub.detach_if_no_pending_control_work(&stage, "session-b", &handle)); + hub.deliver_steer("queued".into(), None); + assert!(!hub.detach_if_no_pending_control_work(&stage, "session-a", &handle)); assert_eq!(hub.active_count(), 1); } #[test] - fn detach_if_queue_empty_removes_matching_empty_session() { + fn detach_if_no_pending_control_work_removes_matching_empty_session() { let hub = SteeringHub::for_tests(); let stage = StageId::new("a", 1); let handle = SessionControlHandle::new(); assert!(hub.attach_handle(&stage, "session-a", &handle)); - assert!(hub.detach_if_queue_empty(&stage, "session-a", &handle)); + assert!(hub.detach_if_no_pending_control_work(&stage, "session-a", &handle)); assert_eq!(hub.active_count(), 0); } + #[test] + fn pure_interrupt_marks_active_sessions_waiting_without_queueing_text() { + let (hub, names) = hub_with_event_names(); + let stage = StageId::new("a", 1); + let handle = SessionControlHandle::new(); + assert!(hub.attach_handle(&stage, "session-a", &handle)); + + hub.interrupt(None); + hub.interrupt(None); + + assert!(handle.is_waiting_for_steer()); + assert_eq!(handle.queue_len(), 0); + assert_eq!(hub.pending_len(), 0); + assert_eq!(names.lock().unwrap().as_slice(), [ + "run.interrupt", + "run.interrupt" + ]); + } + + #[test] + fn interrupt_then_steer_cancels_and_queues_text() { + let (hub, names) = hub_with_event_names(); + let stage = StageId::new("a", 1); + let handle = SessionControlHandle::new(); + assert!(hub.attach_handle(&stage, "session-a", &handle)); + + hub.interrupt_then_steer("stop", None); + + assert!(!handle.is_waiting_for_steer()); + assert_eq!(handle.queue_len(), 1); + assert_eq!(hub.pending_len(), 0); + assert_eq!(names.lock().unwrap().as_slice(), [ + "run.interrupt", + "run.steer" + ]); + } + #[test] fn per_session_queue_evicts_oldest_at_cap() { let hub = SteeringHub::for_tests(); @@ -397,7 +504,7 @@ mod tests { assert!(hub.attach_handle(&stage, "session-a", &handle)); for i in 0..(super::PER_SESSION_QUEUE_CAP + 5) { - hub.deliver(format!("m{i}"), SteerKind::Append, None); + hub.deliver_steer(format!("m{i}"), None); } assert_eq!(handle.queue_len(), super::PER_SESSION_QUEUE_CAP); } diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index 2dffdba54..b85a83a9e 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -299,6 +299,7 @@ models/stage-outcome.ts models/stage-projection.ts models/stage-state.ts models/start-run-request.ts +models/steer-run-request.ts models/submit-answer-request.ts models/success-reason.ts models/system-actor-kind.ts diff --git a/lib/packages/fabro-api-client/src/api/human-in-the-loop-api.ts b/lib/packages/fabro-api-client/src/api/human-in-the-loop-api.ts index 9dc565b77..3f804972b 100644 --- a/lib/packages/fabro-api-client/src/api/human-in-the-loop-api.ts +++ b/lib/packages/fabro-api-client/src/api/human-in-the-loop-api.ts @@ -36,6 +36,8 @@ import type { SshAccessRequest } from '../models'; // @ts-ignore import type { SshAccessResponse } from '../models'; // @ts-ignore +import type { SteerRunRequest } from '../models'; +// @ts-ignore import type { SubmitAnswerRequest } from '../models'; /** * HumanInTheLoopApi - axios parameter creator @@ -179,6 +181,46 @@ export const HumanInTheLoopApiAxiosParamCreator = function (configuration?: Conf options: localVarRequestOptions, }; }, + /** + * Interrupt the active API-mode agent round without sending steering text. The agent keeps its steering lease and waits for a later steer message before starting another LLM round. + * @summary Interrupt Run + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + interruptRun: async (id: string, options: RawAxiosRequestConfig = {}): Promise => { + // verify required parameter 'id' is not null or undefined + assertParamExists('interruptRun', 'id', id) + const localVarPath = `/api/v1/runs/{id}/interrupt` + .replace(`{${"id"}}`, encodeURIComponent(String(id))); + // use dummy base URL string because the URL constructor only accepts absolute URLs. + const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL); + let baseOptions; + if (configuration) { + baseOptions = configuration.baseOptions; + } + + const localVarRequestOptions = { method: 'POST', ...baseOptions, ...options}; + const localVarHeaderParameter = {} as any; + const localVarQueryParameter = {} as any; + + // authentication SessionCookie required + + // authentication BearerAuth required + // http bearer authentication required + await setBearerAuthToObject(localVarHeaderParameter, configuration) + + localVarHeaderParameter['Accept'] = 'application/json'; + + setSearchParams(localVarUrlObj, localVarQueryParameter); + let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {}; + localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers}; + + return { + url: toPathString(localVarUrlObj), + options: localVarRequestOptions, + }; + }, /** * Returns pending human-in-the-loop questions for a run. Questions are generated when the workflow needs user input to proceed. * @summary List Run Questions @@ -333,6 +375,51 @@ export const HumanInTheLoopApiAxiosParamCreator = function (configuration?: Conf options: localVarRequestOptions, }; }, + /** + * Send a mid-run steering message to the live agent session(s) of a running run. Set `interrupt=true` to atomically interrupt the active API-mode agent round first, then deliver this message as the next user turn. Without `interrupt=true`, the message is appended to the steering queue and may buffer until the next API-mode agent session. + * @summary Steer Run + * @param {string} id Unique run identifier (ULID). + * @param {SteerRunRequest} steerRunRequest + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + steerRun: async (id: string, steerRunRequest: SteerRunRequest, options: RawAxiosRequestConfig = {}): Promise => { + // verify required parameter 'id' is not null or undefined + assertParamExists('steerRun', 'id', id) + // verify required parameter 'steerRunRequest' is not null or undefined + assertParamExists('steerRun', 'steerRunRequest', steerRunRequest) + const localVarPath = `/api/v1/runs/{id}/steer` + .replace(`{${"id"}}`, encodeURIComponent(String(id))); + // use dummy base URL string because the URL constructor only accepts absolute URLs. + const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL); + let baseOptions; + if (configuration) { + baseOptions = configuration.baseOptions; + } + + const localVarRequestOptions = { method: 'POST', ...baseOptions, ...options}; + const localVarHeaderParameter = {} as any; + const localVarQueryParameter = {} as any; + + // authentication SessionCookie required + + // authentication BearerAuth required + // http bearer authentication required + await setBearerAuthToObject(localVarHeaderParameter, configuration) + + localVarHeaderParameter['Content-Type'] = 'application/json'; + localVarHeaderParameter['Accept'] = 'application/json'; + + setSearchParams(localVarUrlObj, localVarQueryParameter); + let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {}; + localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers}; + localVarRequestOptions.data = serializeDataIfNeeded(steerRunRequest, localVarRequestOptions, configuration) + + return { + url: toPathString(localVarUrlObj), + options: localVarRequestOptions, + }; + }, /** * Submits an answer to a pending question. The answer can be freeform text or a selected option key, depending on the question type. * @summary Submit Run Answer @@ -433,6 +520,19 @@ export const HumanInTheLoopApiFp = function(configuration?: Configuration) { const localVarOperationServerBasePath = operationServerMap['HumanInTheLoopApi.getSandboxFile']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, + /** + * Interrupt the active API-mode agent round without sending steering text. The agent keeps its steering lease and waits for a later steer message before starting another LLM round. + * @summary Interrupt Run + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + async interruptRun(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.interruptRun(id, options); + const localVarOperationServerIndex = configuration?.serverIndex ?? 0; + const localVarOperationServerBasePath = operationServerMap['HumanInTheLoopApi.interruptRun']?.[localVarOperationServerIndex]?.url; + return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); + }, /** * Returns pending human-in-the-loop questions for a run. Questions are generated when the workflow needs user input to proceed. * @summary List Run Questions @@ -478,6 +578,20 @@ export const HumanInTheLoopApiFp = function(configuration?: Configuration) { const localVarOperationServerBasePath = operationServerMap['HumanInTheLoopApi.putSandboxFile']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, + /** + * Send a mid-run steering message to the live agent session(s) of a running run. Set `interrupt=true` to atomically interrupt the active API-mode agent round first, then deliver this message as the next user turn. Without `interrupt=true`, the message is appended to the steering queue and may buffer until the next API-mode agent session. + * @summary Steer Run + * @param {string} id Unique run identifier (ULID). + * @param {SteerRunRequest} steerRunRequest + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + async steerRun(id: string, steerRunRequest: SteerRunRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.steerRun(id, steerRunRequest, options); + const localVarOperationServerIndex = configuration?.serverIndex ?? 0; + const localVarOperationServerBasePath = operationServerMap['HumanInTheLoopApi.steerRun']?.[localVarOperationServerIndex]?.url; + return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); + }, /** * Submits an answer to a pending question. The answer can be freeform text or a selected option key, depending on the question type. * @summary Submit Run Answer @@ -535,6 +649,16 @@ export const HumanInTheLoopApiFactory = function (configuration?: Configuration, getSandboxFile(id: string, path: string, options?: RawAxiosRequestConfig): AxiosPromise { return localVarFp.getSandboxFile(id, path, options).then((request) => request(axios, basePath)); }, + /** + * Interrupt the active API-mode agent round without sending steering text. The agent keeps its steering lease and waits for a later steer message before starting another LLM round. + * @summary Interrupt Run + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + interruptRun(id: string, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.interruptRun(id, options).then((request) => request(axios, basePath)); + }, /** * Returns pending human-in-the-loop questions for a run. Questions are generated when the workflow needs user input to proceed. * @summary List Run Questions @@ -571,6 +695,17 @@ export const HumanInTheLoopApiFactory = function (configuration?: Configuration, putSandboxFile(id: string, path: string, body: File, options?: RawAxiosRequestConfig): AxiosPromise { return localVarFp.putSandboxFile(id, path, body, options).then((request) => request(axios, basePath)); }, + /** + * Send a mid-run steering message to the live agent session(s) of a running run. Set `interrupt=true` to atomically interrupt the active API-mode agent round first, then deliver this message as the next user turn. Without `interrupt=true`, the message is appended to the steering queue and may buffer until the next API-mode agent session. + * @summary Steer Run + * @param {string} id Unique run identifier (ULID). + * @param {SteerRunRequest} steerRunRequest + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + steerRun(id: string, steerRunRequest: SteerRunRequest, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.steerRun(id, steerRunRequest, options).then((request) => request(axios, basePath)); + }, /** * Submits an answer to a pending question. The answer can be freeform text or a selected option key, depending on the question type. * @summary Submit Run Answer @@ -626,6 +761,17 @@ export class HumanInTheLoopApi extends BaseAPI { return HumanInTheLoopApiFp(this.configuration).getSandboxFile(id, path, options).then((request) => request(this.axios, this.basePath)); } + /** + * Interrupt the active API-mode agent round without sending steering text. The agent keeps its steering lease and waits for a later steer message before starting another LLM round. + * @summary Interrupt Run + * @param {string} id Unique run identifier (ULID). + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + public interruptRun(id: string, options?: RawAxiosRequestConfig) { + return HumanInTheLoopApiFp(this.configuration).interruptRun(id, options).then((request) => request(this.axios, this.basePath)); + } + /** * Returns pending human-in-the-loop questions for a run. Questions are generated when the workflow needs user input to proceed. * @summary List Run Questions @@ -665,6 +811,18 @@ export class HumanInTheLoopApi extends BaseAPI { return HumanInTheLoopApiFp(this.configuration).putSandboxFile(id, path, body, options).then((request) => request(this.axios, this.basePath)); } + /** + * Send a mid-run steering message to the live agent session(s) of a running run. Set `interrupt=true` to atomically interrupt the active API-mode agent round first, then deliver this message as the next user turn. Without `interrupt=true`, the message is appended to the steering queue and may buffer until the next API-mode agent session. + * @summary Steer Run + * @param {string} id Unique run identifier (ULID). + * @param {SteerRunRequest} steerRunRequest + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + public steerRun(id: string, steerRunRequest: SteerRunRequest, options?: RawAxiosRequestConfig) { + return HumanInTheLoopApiFp(this.configuration).steerRun(id, steerRunRequest, options).then((request) => request(this.axios, this.basePath)); + } + /** * Submits an answer to a pending question. The answer can be freeform text or a selected option key, depending on the question type. * @summary Submit Run Answer @@ -678,4 +836,3 @@ export class HumanInTheLoopApi extends BaseAPI { return HumanInTheLoopApiFp(this.configuration).submitRunAnswer(id, qid, submitAnswerRequest, options).then((request) => request(this.axios, this.basePath)); } } - diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 7a1c55f6b..5767dbca4 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -297,4 +297,4 @@ export * from './workflow-namespace'; export * from './workflow-reference'; export * from './workflow-settings'; export * from './worktree-mode'; -export * from './write-blob-response'; \ No newline at end of file +export * from './write-blob-response'; diff --git a/lib/packages/fabro-api-client/src/models/steer-run-request.ts b/lib/packages/fabro-api-client/src/models/steer-run-request.ts index 2dcadf9f5..eb1498556 100644 --- a/lib/packages/fabro-api-client/src/models/steer-run-request.ts +++ b/lib/packages/fabro-api-client/src/models/steer-run-request.ts @@ -23,7 +23,7 @@ export interface SteerRunRequest { */ 'text': string; /** - * When true, cancel the in-flight LLM stream and tool calls in the current round before delivering. When false (default), append to the steering queue and let the agent pick it up at the next turn boundary. + * When true, apply a worker-control interrupt first, then deliver this text as steering in the same control operation. When false (default), append to the steering queue and let the agent pick it up at the next turn boundary. */ 'interrupt'?: boolean; }