From 79f89165f61340da0e40d2438758aeaac03523d3 Mon Sep 17 00:00:00 2001 From: "fabro-sh-0530[bot]" <281434857+fabro-sh-0530[bot]@users.noreply.github.com> Date: Tue, 5 May 2026 15:34:16 -0400 Subject: [PATCH] Wire end-to-end steering for running agents (#209) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary This makes the advertised mid-run steering path real: users can send append or interrupt steering messages through the API, CLI, and web UI, and the worker delivers them to live API-mode agent sessions or buffers them for the next session. The change adds the control protocol, session interrupt machinery, workflow hub, server route/OpenAPI/client updates, and UI feedback needed for the whole path. ### Plan Summary - Add `SteerKind`/`run.steer` wire protocol and `POST /runs/{id}/steer` - Deliver steers through subprocess JSONL or the in-process `SteeringHub` - Support append and interrupt behavior in agent sessions, with bounded buffering and events - Expose steering in the CLI/web UI and surface SSE toasts ## Flow ```mermaid flowchart TB UI["CLI / Web UI"] --> API["POST /runs/{id}/steer"] API -->|"subprocess transport"| Control["Worker control JSONL"] API -->|"in-process transport"| Hub["SteeringHub"] Control --> Hub Hub -->|"active API sessions"| Session["SessionControlHandle"] Hub -->|"no active session"| Pending["Pending buffer"] Pending -->|"first future API session"| Session Session --> Agent["Session round loop"] Agent --> Events["RunEvent stream"] Events --> UI ``` ## What changed and why - Agent sessions now expose a lightweight `SessionControlHandle`, drain steering at the top of each round, and use a replaceable round cancellation token for interrupts. LLM waits are cancelled promptly, while tool execution observes cancellation cooperatively so every committed `tool_use` still gets a matching `tool_result`. - `SteeringHub` owns active API session registration, broadcast delivery, pending buffering, FIFO queue caps, and steering lifecycle/drop events. A completion coordinator closes the final-response race without introducing a workflow dependency into the agent crate. - The server route replaces the 501 stub, validates run state and best-effort CLI-only steerability, and forwards through either subprocess control JSONL or the in-process hub. OpenAPI and generated clients now include the request type. - The CLI and web UI can send append or interrupt steers. Run detail and board views open the new composer, and shared SSE subscriptions now support per-subscriber event callbacks so invalidation and steering toasts can coexist on one EventSource. ## Review notes - Steering actors stay on top-level `RunEvent.actor`; event props only carry steering kind/drop metadata. - Buffered steers replay as append messages to the first API session that registers after an empty-active period. Per-stage targeting remains out of scope. - CLI-mode agent stages are still not steerable; the server returns a best-effort 409 when all active agent stages are CLI-mode, while the worker hub remains the authoritative safety net. - No persistence or schema migration is required; active and pending steering state is in memory. - New tests focus on protocol round-trips, hub buffering/bounds, session steering-loop behavior, SSE fanout, and basic server rejection paths. ⚒️ Generated with [Fabro](https://fabro.sh) --------- Co-authored-by: Fabro Co-authored-by: Bryan Helmkamp Co-authored-by: Claude Opus 4.7 (1M context) --- .../app/components/steer-composer.tsx | 143 ++++ apps/fabro-web/app/hooks/use-run-toasts.ts | 72 ++ apps/fabro-web/app/lib/mutations.ts | 21 + apps/fabro-web/app/lib/run-events.test.tsx | 42 ++ apps/fabro-web/app/lib/run-events.ts | 26 +- apps/fabro-web/app/lib/sse.ts | 29 +- apps/fabro-web/app/routes/run-detail.tsx | 24 +- apps/fabro-web/app/routes/runs.tsx | 22 +- docs/internal/events.md | 81 ++- docs/public/api-reference/fabro-api.yaml | 129 ++++ docs/public/human-tools/steering.mdx | 10 +- docs/public/reference/cli.mdx | 24 + ...05-05-decouple-interrupts-from-steering.md | 111 +++ lib/crates/fabro-agent/src/lib.rs | 2 +- lib/crates/fabro-agent/src/session.rs | 682 ++++++++++++++++-- lib/crates/fabro-agent/src/types.rs | 9 +- .../fabro-agent/tests/it/parity_matrix.rs | 14 +- .../fabro-api/tests/run_event_round_trip.rs | 30 + lib/crates/fabro-cli/src/args.rs | 24 + lib/crates/fabro-cli/src/commands/run/mod.rs | 2 + .../fabro-cli/src/commands/run/runner.rs | 37 +- .../fabro-cli/src/commands/run/steer.rs | 32 + lib/crates/fabro-cli/tests/it/cmd/fabro.rs | 1 + lib/crates/fabro-client/src/client.rs | 29 + .../fabro-interview/src/control_protocol.rs | 81 +++ lib/crates/fabro-server/src/server.rs | 132 +++- .../fabro-server/src/server/handler/mod.rs | 4 +- .../fabro-server/src/server/handler/steer.rs | 181 +++++ lib/crates/fabro-server/src/server/tests.rs | 359 ++++++++- lib/crates/fabro-store/src/run_state.rs | 68 +- lib/crates/fabro-types/src/lib.rs | 2 +- lib/crates/fabro-types/src/run_event/agent.rs | 48 +- lib/crates/fabro-types/src/run_event/mod.rs | 105 +++ lib/crates/fabro-types/src/run_event/run.rs | 12 + .../fabro-workflow/src/event/convert.rs | 63 +- lib/crates/fabro-workflow/src/event/events.rs | 100 +++ lib/crates/fabro-workflow/src/event/names.rs | 8 + .../fabro-workflow/src/event/stored_fields.rs | 62 +- .../fabro-workflow/src/handler/agent.rs | 25 +- .../fabro-workflow/src/handler/fan_in.rs | 6 + .../src/handler/llm/activation_lease.rs | 249 +++++++ .../fabro-workflow/src/handler/llm/api.rs | 317 +++++++- .../fabro-workflow/src/handler/llm/cli.rs | 4 + .../fabro-workflow/src/handler/llm/mod.rs | 1 + lib/crates/fabro-workflow/src/handler/mod.rs | 10 + .../fabro-workflow/src/handler/prompt.rs | 9 +- lib/crates/fabro-workflow/src/lib.rs | 2 + .../fabro-workflow/src/operations/fork.rs | 24 +- .../fabro-workflow/src/operations/start.rs | 29 +- .../fabro-workflow/src/pipeline/execute.rs | 2 + .../src/pipeline/execute/tests.rs | 9 +- .../fabro-workflow/src/pipeline/initialize.rs | 148 ++-- .../fabro-workflow/src/pipeline/types.rs | 2 + lib/crates/fabro-workflow/src/steering_hub.rs | 511 +++++++++++++ .../src/.openapi-generator/FILES | 1 + .../src/api/human-in-the-loop-api.ts | 159 +++- .../fabro-api-client/src/models/index.ts | 1 + .../src/models/steer-run-request.ts | 29 + 58 files changed, 4118 insertions(+), 241 deletions(-) create mode 100644 apps/fabro-web/app/components/steer-composer.tsx create mode 100644 apps/fabro-web/app/hooks/use-run-toasts.ts create mode 100644 docs/superpowers/plans/2026-05-05-decouple-interrupts-from-steering.md create mode 100644 lib/crates/fabro-cli/src/commands/run/steer.rs create mode 100644 lib/crates/fabro-server/src/server/handler/steer.rs create mode 100644 lib/crates/fabro-workflow/src/handler/llm/activation_lease.rs create mode 100644 lib/crates/fabro-workflow/src/steering_hub.rs create mode 100644 lib/packages/fabro-api-client/src/models/steer-run-request.ts diff --git a/apps/fabro-web/app/components/steer-composer.tsx b/apps/fabro-web/app/components/steer-composer.tsx new file mode 100644 index 000000000..4c5156a35 --- /dev/null +++ b/apps/fabro-web/app/components/steer-composer.tsx @@ -0,0 +1,143 @@ +import { useEffect, useRef, useState } from "react"; + +import { ApiError } from "../lib/api-client"; +import { useSteerRun } from "../lib/mutations"; +import { ErrorMessage } from "./ui"; + +interface SteerComposerProps { + runId: string; + open: boolean; + onClose: () => void; +} + +export function SteerComposer({ runId, open, onClose }: SteerComposerProps) { + const [text, setText] = useState(""); + const [errorMessage, setErrorMessage] = useState(null); + const textareaRef = useRef(null); + const onCloseRef = useRef(onClose); + onCloseRef.current = onClose; + const { trigger, isMutating } = useSteerRun(runId); + + useEffect(() => { + if (open) { + requestAnimationFrame(() => textareaRef.current?.focus()); + } else { + setText(""); + setErrorMessage(null); + } + }, [open]); + + useEffect(() => { + if (!open) return; + const onKey = (e: KeyboardEvent) => { + if (e.key === "Escape") { + e.preventDefault(); + onCloseRef.current(); + } + }; + window.addEventListener("keydown", onKey); + return () => window.removeEventListener("keydown", onKey); + }, [open]); + + if (!open) return null; + + const trimmed = text.trim(); + const canSubmit = trimmed.length > 0 && !isMutating; + + async function send(interrupt: boolean) { + if (!canSubmit) return; + setErrorMessage(null); + try { + await trigger({ text: trimmed, interrupt }); + onClose(); + } catch (err) { + if (err instanceof ApiError) { + // Try to surface the well-known 409 codes inline. + const body = err.body as { code?: string; detail?: string } | null; + if (body?.code === "cli_agent_not_steerable") { + setErrorMessage( + "All running agent stages are CLI-mode and can't be steered.", + ); + } else if (body?.code === "use_answer_endpoint") { + setErrorMessage( + "Run is blocked on a question; answer the question first.", + ); + } else { + setErrorMessage(body?.detail ?? err.message ?? "Steer failed."); + } + } else { + setErrorMessage("Steer failed; try again."); + } + } + } + + function handleKeyDown(e: React.KeyboardEvent) { + if (e.key === "Enter" && !e.shiftKey) { + e.preventDefault(); + void send(false); + } + } + + return ( +
{ + if (e.target === e.currentTarget) onClose(); + }} + > +
+
Steer agent
+