From be2a7b658ced62da3413339c47ae76f31b8a23bd Mon Sep 17 00:00:00 2001 From: Fabro Date: Mon, 4 May 2026 13:51:18 -0400 Subject: [PATCH] =?UTF-8?q?init=20run=20=E2=9A=92=EF=B8=8F=20Generated=20w?= =?UTF-8?q?ith=20[Fabro](https://fabro.sh)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- graph.fabro | 37 ++++ run.json | 521 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 558 insertions(+) create mode 100644 graph.fabro create mode 100644 run.json diff --git a/graph.fabro b/graph.fabro new file mode 100644 index 000000000..bfd5da463 --- /dev/null +++ b/graph.fabro @@ -0,0 +1,37 @@ +digraph ImplementPlan { + graph [ + goal="Implement and simplify", + model_stylesheet=" + * { model: claude-opus-4-7; } + " + ] + rankdir=LR + + start [shape=Mdiamond, label="Start"] + exit [shape=Msquare, label="Exit"] + + toolchain [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0] + preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0] + preflight_lint [label="Preflight Lint", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1", max_retries=0] + fix_lints [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3] + implement [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD."] + simplify_opus [label="Simplify (Opus)", prompt="@prompts/simplify.md"] + simplify_gpt [label="Simplify (GPT-55)", prompt="@prompts/simplify.md", model="gpt-55"] + verify [label="Verify", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1 && cargo dev docs refresh 2>&1 && cargo dev docs check 2>&1", goal_gate=true, retry_target="fixup"] + fixup [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all clippy lint warnings, test failures, and generated docs errors.", max_visits=3] + fmt [label="Format", shape=parallelogram, script="cargo +nightly-2026-04-14 fmt --all 2>&1", max_retries=0] + + start -> toolchain + toolchain -> preflight_compile [condition="outcome=succeeded"] + toolchain -> exit + preflight_compile -> preflight_lint [condition="outcome=succeeded"] + preflight_compile -> exit + preflight_lint -> implement [condition="outcome=succeeded"] + preflight_lint -> fix_lints + fix_lints -> preflight_lint + implement -> simplify_opus -> simplify_gpt -> verify + verify -> fmt [condition="outcome=succeeded"] + verify -> fixup + fixup -> verify + fmt -> exit +} diff --git a/run.json b/run.json new file mode 100644 index 000000000..7000b15d6 --- /dev/null +++ b/run.json @@ -0,0 +1,521 @@ +{ + "spec": { + "run_id": "01KQT1TWWJYWZGDT8F05E29H9D", + "settings": { + "project": { + "name": null, + "description": null, + "directory": ".", + "metadata": {} + }, + "workflow": { + "name": null, + "description": null, + "graph": "workflow.fabro", + "metadata": {} + }, + "run": { + "goal": { + "type": "inline", + "value": "# Plan: end-to-end steering for running agents\n\n## Context\n\n`README.md` advertises \"Steer running agents mid-turn.\" Today the agent core fully supports it (`Session::steer`, `drain_steering`, `Turn::Steering` → user message, `agent.steering.injected` event, parity tests). Everything north of that is missing or stubbed:\n\n- `POST /runs/{id}/steer` is registered as `not_implemented` (501).\n- Endpoint is not in the OpenAPI spec.\n- No CLI command, no web UI wiring (only a placeholder demo-mode-gated \"Steer\" button on the running-runs board with no handler).\n- No bridge from the server's HTTP layer through the worker subprocess into the live `Session`.\n\nTwo flavors are required:\n\n- **Append** — push to the steering queue; agent picks it up at the next turn boundary (existing `Session::steer`).\n- **Interrupt** — cancel in-flight LLM stream and tool calls in the current round, then deliver as the next user turn. New code in `Session`.\n\nSteers can arrive when no agent stage is active, or when only non-agent stages are active — these **buffer** for the next API-mode session. Steers that arrive when only CLI-mode agent stages are active are **rejected** (no steerable target). Mixed runs with at least one API-mode agent active are accepted and broadcast.\n\n## Decisions\n\n- Scope: full stack — wire protocol, agent, worker, server, OpenAPI, CLI, web UI.\n- Parallel stages (`max_parallel = 4`): broadcast to every active API-mode `Session` in the run.\n- Status policy: accept only when run status is `running`. Reject `blocked` with a hint to use the interview-answer endpoint. Reject terminal states with 409.\n- **CLI-mode steerability predicate (target-oriented, best-effort).** Server's view derives from asynchronously consumed events, so the 409 below is best-effort. Stale state can lead to a forwarded steer that the worker hub then buffers (`agent.steer.buffered`) or drops at run end (`agent.steer.dropped { reason: \"run_ended\" }`). UI surfaces both via SSE.\n 1. ≥1 API-mode agent stage active → forward (broadcast).\n 2. No active agent stages at all (between stages, non-agent stage, idle) → forward (worker buffers for next session).\n 3. Active agent stages exist but none are API-mode → **best-effort 409**.\n- Web UI shows the Steer button whenever `status === \"running\"`; rejection reason flows through the 409 response and is surfaced inline.\n- Every steer carries an `actor: Principal` end-to-end (HTTP → envelope → worker → agent). Per `docs/internal/events-strategy.md:83`, `actor` lives only at top-level `RunEvent.actor`; **not** in event-specific props.\n- Both transport variants must work: `RunAnswerTransport::Subprocess` (worker control JSONL) and `RunAnswerTransport::InProcess` (direct call into the in-process hub).\n- **Round-token cancellation is the sole marker for steering interrupts.** No new `InterruptReason::SteerInterrupt` variant. The loop distinguishes terminal cancel from steer-interrupt by which token fired (`cancel_token` vs `round_token`). Existing `interrupt_reason` (used for `WallClockTimeout` / `Cancelled`) is unchanged.\n- **Bounded queues.** Per-session steering queue cap = 32 messages; per-run pending buffer cap = 32 messages. Overflow evicts oldest (FIFO) and emits `agent.steer.dropped { count, reason }`. Sizes are workspace constants in `fabro-workflow`.\n- **Buffered-steer fanout semantics:** buffered steers go to the **first** session that registers after an empty-active period. Sister parallel sessions registering at almost the same time do not replay the buffer. Documented limitation; per-stage targeting (deferred) is the natural future fix.\n\n## Message flow\n\n```\nHTTP POST /runs/{id}/steer { text, interrupt } (auth → actor: Principal)\n → fabro-server handler\n ├─ validates status + steerability predicate from active_api_stages /\n │ active_cli_stages tracked from worker-emitted events\n ├─ Subprocess: WorkerControlEnvelope::steer(text, kind, actor) → control_tx\n │ → pump_worker_control_jsonl → worker stdin → apply_worker_control_line\n │ → SteeringHub.deliver(text, kind, actor)\n └─ InProcess: directly call SteeringHub.deliver(text, kind, actor) on the\n hub stored alongside the in-process interviewer\n → SteeringHub.deliver:\n ├─ active API handles → broadcast: handle.queue.push((text, kind, actor))\n │ + if Interrupt: handle.round_token.cancel()\n └─ none → push to pending Vec\n → Session round loop: top-of-loop drain_steering() emits\n AgentEvent::SteeringInjected { text, kind } with actor flowing through\n internal event metadata; agent_actor_for_event lifts it to RunEvent.actor.\n```\n\n## Implementation\n\n### 1. Wire protocol — extend `WorkerControlEnvelope`\n\n**Files:** `lib/crates/fabro-types/src/lib.rs` (or new `steering.rs`), `lib/crates/fabro-interview/src/control_protocol.rs`\n\nDefine `SteerKind` in `fabro-types` (not `fabro-interview` — `fabro-interview` already depends on `fabro-types` per `control_protocol.rs:1`, so the canonical enum must live in the lower crate to avoid a cycle):\n\n```rust\n// fabro-types\n#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]\n#[serde(rename_all = \"lowercase\")]\npub enum SteerKind { Append, Interrupt }\n```\n\n`fabro-interview` re-exports it and adds the envelope variant:\n\n```rust\n// fabro-interview/src/control_protocol.rs\npub use fabro_types::SteerKind;\n\n#[serde(rename = \"run.steer\")]\nSteer {\n text: String,\n kind: SteerKind,\n actor: Principal, // matches interview.answer\n},\n```\n\nAdd `WorkerControlEnvelope::steer(text, kind, actor)`. Round-trip serde tests for both kinds + actor next to existing tests at line 104+.\n\n### 2. Agent — `interrupt_with` + round-level cancel + control handle\n\n**Files:** `lib/crates/fabro-agent/src/session.rs`, `lib/crates/fabro-agent/src/types.rs`, `lib/crates/fabro-agent/src/error.rs`, `lib/crates/fabro-agent/src/tool_execution.rs`, `lib/crates/fabro-agent/tests/it/parity_matrix.rs`\n\nChanges to `Session`:\n\n- New field `round_token: Arc>` — replaceable per round.\n- Change `steering_queue` element type from `String` to `(String, SteerKind, Option)` so per-message kind+actor survive into the emitted event.\n- New method `interrupt_with(&self, text: String, actor: Option)`: push `(text, Interrupt, actor)` and cancel `round_token`. **Does not** touch `interrupt_reason` — round-token cancellation alone marks the steer-interrupt path.\n- Existing `steer(text)` updated to push `(text, Append, None)`.\n- No new `InterruptReason` variant. Existing `WallClockTimeout` / `Cancelled` semantics are unchanged. The loop disambiguates by inspecting tokens:\n - `cancel_token.is_cancelled()` → terminal close (existing behavior).\n - `round_token.is_cancelled() && !cancel_token.is_cancelled()` → steer interrupt → continue.\n- New method `control_handle(&self) -> SessionControlHandle` returning `Arc` clones of `steering_queue` and `round_token`. The hub stores the *handle*, not the `Session`. This avoids the ownership mismatch with `AgentApiBackend` (Session is owned by value and mutated via `process_input(&mut self)` in `handler/llm/api.rs:444-494`).\n- `SessionControlHandle::steer(text, actor)` and `interrupt_with(text, actor)` thin wrappers — the hub calls these.\n- Existing `AgentEvent::SteeringInjected` props gain `kind: SteerKind` only. Actor flows through internal event metadata (set on the emitted event), then `agent_actor_for_event` lifts it to top-level `RunEvent.actor` per the events strategy.\n\n**Loop changes in `process_input` (lines 603–941):**\n\n- **Move `drain_steering()` to the top of the loop body**, before `compact_if_needed`/`build_request`. Today: line 694 (before loop, once) and line 924 (after tools). After a SteerInterrupt `continue`, neither runs before the next request. Top-of-loop drain fixes this. Remove the line-694 pre-loop call (top-of-loop covers it on iter 1) and the line-924 post-tool call (next iter's top-of-loop covers it).\n- At top of each iteration: if `round_token` is cancelled, replace with a fresh `CancellationToken`. (No `interrupt_reason` to clear — round-token cancellation is the marker.)\n- Build per-round composite token from `cancel_token` (terminal) and `round_token` (per-round).\n- **Cancellation propagation — two distinct strategies:**\n - **LLM stream awaits** (preemptive — safe to drop in-flight): wrap with `tokio::select!` against `composite.cancelled()` at:\n - `open_stream_with_retry(...)` and any internal retry-backoff `tokio::time::sleep`\n - `event_stream.next()` per chunk (so an idle stream that never produces another chunk doesn't pin the loop)\n - **Tool execution** (cooperative — must NOT drop the future): pass the composite token as a parameter to `execute_tool_calls`. It runs to completion, returning \"Cancelled\" entries internally for any in-flight tool (existing path: `tool_execution.rs:80`). Do **not** wrap in `select!` — dropping the future would lose synthesized cancel results and break the `tool_use`↔`tool_result` invariant.\n- After LLM stream and after tools, branch on whether the round was interrupted:\n - **Mid-LLM interrupt** (`record_assistant_turn` at line 859 has not run yet): drop the unrecorded turn. **Also clear visible UI output**: if any `TextDelta` or `ReasoningDelta` was emitted in the dropped round, emit `AgentEvent::AssistantOutputReplace { text: \"\", reasoning: None }` before `continue` (mirrors the existing retry-clears-output pattern at session.rs:828). No tool_results needed because no `tool_use` was committed to history.\n - **Mid-tool interrupt** (assistant turn with `tool_use` blocks already recorded at line 859 before `execute_tool_calls` ran): `execute_tool_calls` runs to completion and returns **one `ToolResult` per `tool_use` block**. Content varies by tool — bash returns `Ok(\"Command cancelled.\\n…\")` (tools.rs:265-266), other tools may return partial output, an error message, or a synthetic Cancelled marker. The Anthropic invariant only requires one-per-block, not a specific content shape. Always push `Turn::ToolResults` with whatever `execute_tool_calls` returned (existing line 909-921 path), then branch on which token fired. Refactor the current `cancel_token.is_cancelled()` branch (lines 907-915) to: append tool_results unconditionally → close+return-Err (if `cancel_token` fired) or `continue` (if only `round_token` fired).\n- **Append-during-final-response fix (race-safe, dependency-safe).** Today line 881-883 unconditionally `break` when `tool_calls.is_empty()`. A naive `if steering_queue.is_empty() { break }` still loses steers that arrive between the empty check and the function return because the hub still considers the session active. The full close-the-door dance (unregister → check → re-register-or-break) crosses a crate boundary the wrong way (`fabro-agent` does not depend on `fabro-workflow`; reverse cycles per `Cargo.toml:22`). Solution: small trait owned by fabro-agent, implemented in fabro-workflow.\n\n ```rust\n // fabro-agent\n pub trait CompletionCoordinator: Send + Sync {\n /// Called at natural completion (tool_calls empty).\n /// Return true to continue the loop (queue is non-empty),\n /// false to break. Implementor coordinates with whatever\n /// owns the steering source.\n fn on_natural_completion(&self) -> bool;\n }\n ```\n\n `Session` gains `completion_coordinator: Option>`, defaulting to `None` (preserves existing behavior — direct `Session::new` callers and tests just break naturally).\n\n Loop:\n\n ```rust\n if tool_calls.is_empty() {\n let should_continue = self.completion_coordinator\n .as_ref()\n .is_some_and(|c| c.on_natural_completion());\n if should_continue { continue; }\n break;\n }\n ```\n\n In fabro-workflow, an adapter implementing the trait holds `(Arc, StageId, SessionControlHandle)` and does:\n\n ```rust\n fn on_natural_completion(&self) -> bool {\n self.hub.unregister(self.stage_id.clone()); // serializes vs hub.deliver\n if self.handle.queue_is_empty() { return false; }\n self.hub.register(self.stage_id.clone(), self.handle.clone());\n true // session's next iteration drains\n }\n ```\n\n `AgentApiBackend::run` builds the adapter, sets it on the session before `process_input`, and removes it after.\n\n **Hub locking discipline (race safety):** `SteeringHub::deliver` holds the `active` *read* lock for the entire push (clone handle + push to queue happen under the read lock). `unregister` takes the *write* lock. RwLock semantics serialize them — once `unregister` returns, no in-flight push can be racing. The post-unregister queue check sees a stable result.\n- `cancel_token.is_cancelled()` (terminal) still returns `Err(interrupted_error())` and closes — unchanged.\n\nTests in `parity_matrix.rs`:\n\n- `steering_interrupt_mid_llm_idle_stream` — fire `interrupt_with` while the LLM stream is open but producing no chunks. Assert interrupt takes effect within ~1s (proves `tokio::select!` is wired around `next()`).\n- `steering_interrupt_mid_llm_streaming` — fire mid-stream after at least one `TextDelta` has been emitted; assert (a) `AssistantOutputReplace { text: \"\", reasoning: None }` is emitted before the next round (clears stale partial output in the UI), (b) next turn includes the steer text, (c) event has `kind: \"interrupt\"`.\n- `steering_interrupt_mid_tool` — fire while a Bash tool is running; assert (a) `Turn::ToolResults` immediately follows the assistant tool-use turn (one ToolResult per tool_use, content unspecified — could be partial output, \"Command cancelled\", or an error message), then (b) `Turn::Steering` with the new text. No dangling `tool_use`. Test asserts shape, not content.\n- `steering_no_dangling_tool_use_invariant` — assert that no `Turn::Steering` immediately follows a `Turn::Assistant` containing `tool_use` blocks without an intervening `Turn::ToolResults`. Asserts shape (one ToolResult per tool_use), not content. Guards against `select!`-around-tools regressions.\n- `steering_append_kind_field` — fire `steer()` between rounds; assert event carries `kind: \"append\"`.\n- `append_during_final_response_triggers_extra_round` — fire `steer()` while LLM is producing a final no-tool response. Assert agent does NOT exit `process_input` after `tool_calls.is_empty()`; instead runs another model turn that incorporates the steer. (Test uses a stub `CompletionCoordinator` impl that returns `true` once when the queue is non-empty — keeps the agent test free of workflow/hub dependencies.)\n\nQueue overflow tests live at the **`SteeringHub` layer in fabro-workflow**, not here. Direct `Session::steer` callers intentionally bypass the cap, so the agent has nothing to test for overflow.\n\n### 3. Worker — `SteeringHub` + control plumbing\n\n**Files:** `lib/crates/fabro-cli/src/commands/run/runner.rs`, `lib/crates/fabro-workflow/src/services.rs`, `lib/crates/fabro-workflow/src/operations/start.rs`, `lib/crates/fabro-workflow/src/handler/llm/api.rs`\n\nNew type (in `fabro-workflow`, alongside `RunServices`):\n\n```rust\n// All locks below are std::sync — methods are sync and never await while holding them.\npub struct SteeringHub {\n active: std::sync::RwLock>,\n pending: std::sync::Mutex>, // bounded, FIFO\n emitter: Arc,\n}\n\nstruct PendingSteer { text: String, kind: SteerKind, actor: Option }\n\nconst PER_SESSION_QUEUE_CAP: usize = 32;\nconst PER_RUN_PENDING_CAP: usize = 32;\n\nimpl SteeringHub {\n pub fn deliver(&self, text: String, kind: SteerKind, actor: Option);\n pub fn register(&self, stage_id: StageId, handle: SessionControlHandle);\n pub fn unregister(&self, stage_id: StageId);\n pub fn drain_pending_at_run_end(&self); // emits agent.steer.dropped { reason: \"run_ended\" } if any\n}\n```\n\n- `register` decides drain-vs-replace based on **current active-map state**, not history:\n - If `stage_id` is **not already in active** → insert + drain pending into this handle as `Append` + emit `agent.steering.attached`. Covers first-register-after-empty AND close-the-door re-register (which closes the gap where steers can buffer between unregister and re-register).\n - If `stage_id` **is already in active** → replace the handle, do **not** drain pending, do **not** re-emit `attached`. Covers failover (handle replaced under the same id without an intervening unregister).\n- `unregister` is **idempotent**: `agent.steering.detached` fires only when `active.remove(stage_id)` returns `Some`. The close-the-door call removes-and-emits once; the RAII guard at function exit becomes a no-op (entry already gone). Prevents double-emit on natural completion.\n- `deliver` broadcasts to active handles **or** pushes to pending — branched **under the active read lock** so the empty/non-empty decision is atomic with the push. Documented lock order: **active first, then queue or pending; never reverse.** All locks are `std::sync::{RwLock, Mutex}`; **no `.await` while holding any of them.** Sync methods make `CompletionCoordinator::on_natural_completion` callable from the agent loop without converting it to async (tokio locks would force `.await`). This makes the close-the-door pattern race-safe end-to-end.\n- Internal helper `enqueue_into_session_queue(handle, item)` is used by both the broadcast path and the pending-flush path (called from `register`), guaranteeing identical cap enforcement and drop-event emission across both code paths.\n- Sister parallel sessions registering immediately after the first don't replay the buffer (it was drained on the first register) — documented limitation; broadcast-to-future-sessions is deferred with the per-stage targeting feature.\n- **Queue bounds enforced at the hub layer.** Before pushing into a session's `steering_queue` via `SessionControlHandle`, the hub checks `len() >= PER_SESSION_QUEUE_CAP` and evicts the front. Before pushing into `pending`, checks against `PER_RUN_PENDING_CAP`. On eviction, emits `agent.steer.dropped { count: 1, reason: \"queue_full\" }`. **Direct callers of `Session::steer` (loop-detection auto-injection at session.rs:931, tests) bypass the cap** — that's intentional; internal one-shot warnings shouldn't trigger user-facing drop events.\n\n**Plumbing (explicit, not \"via the same path\"):**\n\nIn `runner.rs::execute()` (around lines 88–101): construct `let steering_hub = Arc::new(SteeringHub::new(emitter.clone()));` next to `interviewer` and `cancel_token`. Pass it both into:\n\n1. `spawn_worker_control_stream(interviewer, cancel_token, steering_hub.clone())` — extend the function signature to accept the hub.\n2. `StartServices.steering_hub: Arc` — new required field. Threaded through `operations::start` → `RunServices` → `EngineServices` → handler dispatch.\n\nIn `runner.rs::apply_worker_control_line` (lines 226–250): add a match arm:\n\n```rust\nWorkerControlMessage::Steer { text, kind, actor } => {\n steering_hub.deliver(text, kind, Some(actor));\n}\n```\n\nIn `AgentApiBackend::run()` (`handler/llm/api.rs:444-494`):\n\n- Compute `let stage_id = stage_scope.stage_id();` from the existing `stage_scope` at line 476 (`StageScope::stage_id()` returns `StageId::new(node_id, visit)` per `stage_scope.rs:64-65`). Use this `StageId` everywhere — **not** the bare `node.id` string.\n- After the `Session` is built/cached but before `process_input`, call `services.steering_hub.register(stage_id.clone(), session.control_handle())`.\n- Use a `scopeguard`-style RAII guard so `unregister(stage_id.clone())` runs on success, error, and panic.\n- **Failover (lines 527-572):** inside the failover loop, immediately after `session = new_session;` (line 545) and before `session.initialize().await` (line 556), call `services.steering_hub.register(stage_id.clone(), session.control_handle())` again. The hub overwrites the abandoned handle with the new one. The RAII unregister still works because the same `stage_id` is keyed.\n- The hub never holds the `Session` — only the `Arc`-clones in `SessionControlHandle`. Sidesteps the ownership mismatch.\n\n`AgentCliBackend::run()` is **not** modified — it never registers, so the hub's `active` set never includes CLI stages. The server's steerability predicate uses the `agent.steering.attached/detached` and `agent.cli.started/completed` events to know what's active.\n\n**Run-end drain placement (async cleanup pattern).** Inside `operations::start` (`lib/crates/fabro-workflow/src/operations/start.rs`), wrap the pipeline execution into a result-returning block, then drain pending and flush events explicitly **before** propagating:\n\n```rust\nlet result = run_pipeline(...).await; // success or error\nsteering_hub.drain_pending_at_run_end(); // sync emit of agent.steer.dropped\nstore_progress_logger.flush().await; // awaited flush moves them through the sink\nresult?\n```\n\nA `scopeguard` calling `drain_pending_at_run_end()` is **only** a last-ditch panic fallback — it cannot await the flush, so it's not the primary delivery path. The explicit pattern handles both success and error cleanly. Calling drain from the worker's outer wrap-up (after `operations::start` returns) would lose events because `store_progress_logger.flush().await` at line 818 already ran.\n\n### 4. Server — HTTP handler + OpenAPI + per-stage tracking + InProcess support\n\n**Files:** `docs/public/api-reference/fabro-api.yaml`, `lib/crates/fabro-server/src/server/handler/mod.rs`, `lib/crates/fabro-server/src/server.rs` (or new `handler/steer.rs`), `lib/crates/fabro-server/src/server/tests.rs`\n\nOpenAPI: `POST /runs/{id}/steer` with body `SteerRequest { text: string (required, 1..8192), interrupt: boolean (default false) }`. Responses: `202 Accepted`, `400`, `404`, `409`, `503`. Tag: `Human-in-the-Loop`. Authenticated user becomes `Principal` for the envelope.\n\nHandler (mirror cancel at `handler/lifecycle.rs:162`):\n\n1. Look up `ManagedRun` via `AppState.runs`.\n2. Validate, in order:\n - 404 if missing.\n - 409 if status is `blocked` with `code: \"use_answer_endpoint\"`, hint: `POST /runs/{id}/questions/{qid}/answer`.\n - 409 if terminal (`succeeded`/`failed`/`cancelled`/`archived`).\n - 409 if not `running`.\n - 409 if **target-oriented predicate** rejects: `active_api_stages.is_empty() && !active_cli_stages.is_empty()` with `code: \"cli_agent_not_steerable\"`, message: \"All currently running agent stages are CLI-mode and cannot be steered.\"\n - Otherwise: forward.\n3. **Transport branch on `ManagedRun.answer_transport`:**\n - `Subprocess { control_tx }`: send `WorkerControlEnvelope::steer(text, kind, actor)` with the existing 1s timeout pattern. Map `Timeout`/`Closed` to 503.\n - `InProcess { interviewer, steering_hub }`: directly call `steering_hub.deliver(text, kind, Some(actor))`. No envelope, no JSONL hop, no timeout — same hub the in-process worker would use. Requires storing an `Arc` alongside `interviewer` in `RunAnswerTransport::InProcess` (`server.rs:245`). The in-process spawn site `execute_run_in_process` (line 2541) creates and stores both.\n4. Return 202.\n\n**Tracking active-stage modes (server side):** `ManagedRun` gains:\n\n```rust\nactive_api_stages: HashSet, // primary: agent.steering.attached/detached\nactive_cli_stages: HashSet, // primary: agent.cli.started; backstops below\n```\n\nPlain `HashSet` (no inner lock) — `ManagedRun` is already accessed under `state.runs.lock()` (`server.rs:441` AppState definition; mutation pattern at `server.rs:1724, 1735` for the existing `accepted_questions: HashSet` field at `server.rs:196`). Adding inner `Mutex` would be redundant nested locking.\n\nUpdated by the server's existing event-consumption path. **No reuse of `agent.session.started/ended`** — those events do not reliably fire per stage invocation: `Session::initialize()` (and thus `SessionStarted`) is skipped for reused sessions in `api.rs:490`, and `SessionEnded` only fires on explicit `close()`. The hub-emitted `attached/detached` events fire deterministically per `register/unregister` call inside `AgentApiBackend::run`, which is exactly the steerable window.\n\n**Backstops to prevent leaks** (CLI tracking is fragile because `AgentCliStarted` at cli.rs:511 and `AgentCliCompleted` at cli.rs:648 are 137 lines apart with fallible operations between, and the existing CLI cancel bug means many error paths skip the completion emit):\n\n- On `stage.completed` **and** `stage.failed` (any kind): remove the stage_id (read from top-level `RunEvent.stage_id`) from **both** `active_api_stages` and `active_cli_stages`. Both events fire from the workflow lifecycle (`lifecycle/event.rs:153, 220, 271`); covering only `stage.completed` would leak on the failure path — exactly where the existing CLI cancel bug already strands stages.\n- On terminal run events (`run.completed` / `run.failed`): clear both sets entirely. (Cancellation is folded into `run.failed` via its `reason` field — there is no separate `run.cancelled` event in `lib/crates/fabro-types/src/run_event/mod.rs:87-90`.)\n\nImplementation note for a follow-up PR (out of scope here, in the same area as the existing CLI-cancel debt): wrap the CLI backend's `AgentCliCompleted` emission in a scopeguard so it always fires regardless of error path.\n\n**CLI-only rejection is best-effort.** Server consumes events asynchronously through the run-store subscription path (`server.rs:2023`), so its view of `active_api_stages` / `active_cli_stages` lags actual worker state by a small window. A steer that the server forwards based on a stale view will be handled correctly by the worker hub: if no API session is registered by arrival, the steer buffers and emits `agent.steer.buffered`, which the UI surfaces. The 409-on-all-CLI gate is an optimization for the synchronous user-feedback case; the worker-side hub is the authoritative safety net. Authoritative server-side rejection (round-tripping a confirmation back through the worker control plane) is out of scope.\n\n**After OpenAPI changes, regenerate clients (per `CLAUDE.md` API workflow):**\n\n```bash\ncargo build -p fabro-api # regenerates Rust client via build.rs + progenitor\ncd lib/packages/fabro-api-client && bun run generate # regenerates TypeScript Axios client\n```\n\nBoth must run before `bun run typecheck` in `apps/fabro-web` will pass.\n\n### 5. CLI — `fabro steer`\n\n**Files:** `lib/crates/fabro-cli/src/args.rs`, new `lib/crates/fabro-cli/src/commands/steer.rs`, `lib/crates/fabro-cli/src/commands/mod.rs`, `lib/crates/fabro-cli/src/server_client.rs`, `lib/crates/fabro-cli/src/main.rs`\n\nAdd a new top-level `Commands::Steer(SteerArgs)` (sibling to `Commands::RunCmd`, `Commands::Exec`, etc. in `args.rs:1016`). New top-level command from scratch — no existing `fabro cancel` to mirror (cancel today is Ctrl+C in attached or HTTP-direct).\n\n```\nfabro steer [--interrupt]\nfabro steer --text-stdin [--interrupt] # editors / pipes\n```\n\nImplementation calls a new `server_client.steer_run(run_id, text, kind)` via the regenerated typed API client. Error mapping mirrors the cancel HTTP path.\n\n### 6. Web UI\n\n**Files:** `apps/fabro-web/app/components/steer-composer.tsx` (new), `apps/fabro-web/app/components/steer-composer.test.tsx` (new), `apps/fabro-web/app/lib/mutations.ts`, `apps/fabro-web/app/lib/run-events.ts`, `apps/fabro-web/app/hooks/use-run-toasts.ts` (new), `apps/fabro-web/app/routes/runs.tsx`, `apps/fabro-web/app/routes/run-detail.tsx`\n\n**Composer:** new `SteerComposer` component — modal/popover with textarea, primary \"Send\" button, secondary \"Interrupt\" button, same Enter / Shift+Enter affordance as `interview-dock.tsx`. Reuse `ErrorMessage` from `ui.tsx`.\n\n**Mutation:** `useSteerRun(runId)` in `lib/mutations.ts` mirroring `useSubmitInterviewAnswer` (lines 97–117). On 409 with `code: \"cli_agent_not_steerable\"`, surface inline (\"All running agent stages are CLI-mode and can't be steered.\"). Invalidates run detail on success.\n\n**Surfacing:** wire the existing \"Steer\" button in `routes/runs.tsx:42, 365–369`:\n- Remove the demo-mode gate at lines 437–439.\n- Open `SteerComposer` on click.\n\nAdd a \"Steer\" button to `run-detail.tsx` page header (only when `statusKind === \"running\"`), opening the same composer.\n\n**Toast dispatch:** `lib/run-events.ts` only resolves SWR-invalidation keys (line 44) — does not dispatch toasts. Add `useRunToasts(runId)` hook in `app/hooks/use-run-toasts.ts` that subscribes to the same SSE stream (via existing `subscribeToSharedEventSource`) and calls `useToast().push(...)` for steer-related events, deduping by event id. Mount from `run-detail.tsx` next to `useRunEvents`. SSE invalidation in `run-events.ts` still gets the new event names so SWR refetches; toast is the new hook's job.\n\nToast copy:\n- `agent.steering.injected` with `kind: \"append\"` → \"Steer delivered.\"\n- `agent.steering.injected` with `kind: \"interrupt\"` → \"Agent interrupted — your message is the next turn.\"\n- `agent.steer.buffered` → \"Steer queued — will apply when an agent stage runs.\"\n- `agent.steer.dropped` with `reason: \"queue_full\"` → \"Steer rate limit reached; oldest queued steer dropped.\"\n- `agent.steer.dropped` with `reason: \"run_ended\"` → \"Run ended before queued steer(s) could apply.\"\n\nNote: keep `InterviewDock` semantics untouched — `blocked` runs only. Steer composer handles `running` only. Mutually exclusive surfaces.\n\n## Events\n\n- `agent.steering.injected` — **modified**. `AgentSteeringInjectedProps` (in `lib/crates/fabro-types/src/run_event/agent.rs`) gains `kind: \"append\" | \"interrupt\"`. Actor lives at top-level `RunEvent.actor` only — set via `agent_actor_for_event` from the `actor` carried internally on the `AgentEvent::SteeringInjected` variant. Per `docs/internal/events-strategy.md:83`.\n- `agent.steering.attached` / `agent.steering.detached` — **new**. Emitted by `SteeringHub::register` (only when newly inserted, not on replace) / `unregister` (only when active.remove returned Some). Workflow `Event` variants carry `StageId` internally; `stored_event_fields_for_variant` lifts to top-level `RunEvent.stage_id` (mod.rs:36) so generic event consumers see it where they expect. **Props are empty** — no duplicate `stage_id` in props. Distinct names from existing `agent.session.started/ended` to avoid confusion.\n- `agent.steer.buffered` — **new**. Emitted by the worker hub when a steer arrives with no active session and is parked. Carries `{ kind }` in props (no actor in props — top-level only).\n- `agent.steer.dropped` — **new**. Two shapes:\n - `reason: \"queue_full\"`, `count: 1` — single-item drop with a known dropped steer. Carries the dropped item's `actor` internally; `stored_event_fields_for_variant` lifts it to top-level `RunEvent.actor`.\n - `reason: \"run_ended\"`, `count: N` — aggregate; possibly multiple actors. **No user actor** at top level (system actor); aggregation loss documented.\n\nFor each new event: typed props in `lib/crates/fabro-types/src/run_event/agent.rs`, variant on `EventBody` in `mod.rs`, workflow conversion in `fabro-workflow/src/event/convert.rs`, name in `fabro-workflow/src/event/names.rs`.\n\n**Actor lifting (different paths for different emitters):**\n- `agent.steering.injected` is agent-emitted (`AgentEvent::SteeringInjected`). Carry `actor` on the internal variant; lift via `agent_actor_for_event` (`stored_fields.rs:198`).\n- `agent.steer.buffered` is workflow-hub-emitted (`SteeringHub::deliver`, no agent involvement). Add `actor: Option` to its workflow `Event` variant; lift via `stored_event_fields_for_variant` (`stored_fields.rs:57`).\n- `agent.steer.dropped` with `reason: \"queue_full\"`: carry dropped item's `actor` on the workflow `Event` variant; lift via `stored_event_fields_for_variant`.\n- `agent.steer.dropped` with `reason: \"run_ended\"`: system actor, no user actor lifted.\n- `agent.steering.attached/detached`: lifecycle, no user actor.\n\n## Test strategy\n\n- **Unit** — control-protocol round-trip including actor (`fabro-interview`); `SteerKind` round-trip in `fabro-types`; `SteeringHub` buffer + broadcast + register-drains-or-replaces by active-map state + idempotent unregister + `drain_pending_at_run_end` + **per-session queue overflow drops oldest** + **per-run pending overflow drops oldest** (both at the hub layer, asserting `agent.steer.dropped { reason: \"queue_full\" }` carries the correct actor at top level); server's steerability predicate over mixed `(active_api_stages, active_cli_stages)` sets.\n- **Agent integration** — `parity_matrix.rs` scenarios listed above (idle-stream interrupt, mid-streaming interrupt with stale-output clear, mid-tool interrupt with shape-only assertion, no-dangling-tool-use invariant, append kind field, append-during-final-response triggers extra round). Idle-stream guards against `tokio::select!` regression on stream awaits; no-dangling guards against `select!`-around-tools regression. **Queue overflow is exclusively a hub-layer test** — direct `Session::steer` callers intentionally bypass the cap.\n- **Workflow event conversion** — new test that builds an `AgentEvent::SteeringInjected { actor, kind, text }`, runs it through the conversion machinery, and asserts the resulting `RunEvent` has `kind` in props (no `actor` in props) and the user actor at top-level `RunEvent.actor`. Without this test, a future refactor of `agent_actor_for_event` (`stored_fields.rs:198`) could silently regress steering to `None` — it currently falls through for all variants except `AssistantMessage`/`ToolCall*`.\n- **Server** — handler unit tests for the full status + steerability matrix (running OK, blocked → 409, terminal → 409, missing → 404, all-CLI → 409, mixed API+CLI → accept, no-active-agent → accept, in-process transport delivers via direct hub call, subprocess delivers via control_tx). Existing `in_process_answer_transport_cancel_run_cancels_pending_interviews` (tests.rs:1710) is the template for an in-process steer test.\n- **CLI** — happy-path `fabro steer` against a fake server, plus `--text-stdin`.\n- **Web** — `SteerComposer` (textarea + two buttons + disabled-when-empty); `useSteerRun` posts the right body and surfaces 409 inline; `useRunToasts` dispatches expected toasts with dedup.\n\n## Verification\n\n```bash\n# Backend\ncargo build --workspace\ncargo nextest run --workspace\ncargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings\n\n# API client regeneration (after OpenAPI changes)\ncargo build -p fabro-api\ncd lib/packages/fabro-api-client && bun run generate\ncd ../../..\n\n# Web (depends on regenerated TS client above)\ncd apps/fabro-web && bun run typecheck && bun test\n\n# Manual end-to-end (subprocess transport)\nfabro server start # terminal 1\nfabro run repl # terminal 2 — start a run\nfabro steer \"Try a different approach\" # terminal 3 — append\nfabro steer \"Stop, do X instead\" --interrupt # terminal 3 — interrupt\n# In browser: open the run, click Steer, type and Send / Interrupt; verify SSE event arrives and toast renders.\n\n# Manual end-to-end (in-process transport)\n# Use a registry override / test config that selects InProcess transport,\n# repeat steer + interrupt; same observable behavior, no JSONL hop.\n```\n\n## Out of scope\n\n- Per-stage steer targeting in UI/CLI (broadcast only for v1).\n- Persisting unconsumed steers across run resume.\n- Steering of non-agent stages (commands, conditionals, parallel).\n- Steering CLI-mode agent stages (claude/codex/gemini): structurally impossible without changes to those external CLIs. Server returns 409 only when *all* active agent stages are CLI-mode.\n- Fixing the pre-existing CLI-mode cancel bug (RunCancel currently ignored; subprocesses orphan). Tracked separately.\n- Slack-driven steering.\n- Auth/permission model beyond what `cancel` already does — same caller can do both.\n- Worker-side `stdin` protocol versioning beyond the existing `v: 1` envelope.\n" + }, + "working_dir": null, + "metadata": {}, + "inputs": {}, + "model": { + "provider": "anthropic", + "name": "claude-sonnet-4-6", + "fallbacks": [] + }, + "git": { + "author": null + }, + "prepare": { + "commands": [], + "timeout_ms": 300000 + }, + "execution": { + "mode": "normal", + "approval": "prompt", + "retros": true + }, + "checkpoint": { + "exclude_globs": [] + }, + "sandbox": { + "provider": "daytona", + "preserve": false, + "devcontainer": false, + "env": {}, + "local": { + "worktree_mode": "always" + }, + "docker": { + "image": "buildpack-deps:noble", + "network_mode": null, + "memory_limit": 4000000000, + "cpu_quota": 200000, + "env_vars": {}, + "skip_clone": false + }, + "daytona": { + "auto_stop_interval": 30, + "labels": { + "repo": "fabro-sh/fabro" + }, + "snapshot": { + "name": "fabro-v8", + "cpu": 8, + "memory_gb": 16, + "disk_gb": 20, + "dockerfile": { + "type": "inline", + "value": "FROM ubuntu:24.04\n\nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*\n\n# GitHub CLI\nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*\n\n# Rust\nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y\nENV PATH=\"/root/.cargo/bin:${PATH}\"\nRUN rustup toolchain install nightly-2026-04-14 --profile minimal --component clippy,rustfmt\nRUN cargo install cargo-nextest --locked\nENV CARGO_INCREMENTAL=0\n\n# Bun\nRUN curl -fsSL https://bun.sh/install | bash\nENV PATH=\"/root/.bun/bin:${PATH}\"\n\nWORKDIR /root\n" + } + }, + "network": null, + "skip_clone": false + } + }, + "notifications": {}, + "interviews": { + "provider": null, + "slack": null, + "discord": null, + "teams": null + }, + "agent": { + "permissions": null, + "mcps": {} + }, + "hooks": [], + "scm": { + "provider": null, + "owner": null, + "repository": null, + "github": null + }, + "pull_request": { + "enabled": true, + "draft": false, + "auto_merge": false, + "merge_strategy": "squash" + }, + "artifacts": { + "include": [] + } + } + }, + "graph": { + "name": "ImplementPlan", + "nodes": { + "simplify_opus": { + "id": "simplify_opus", + "attrs": { + "prompt": { + "String": "# Simplify: Code Review and Cleanup\n\nReview changes vs. origin for reuse, quality, and efficiency. Fix any issues found.\n\n## Phase 1: Identify Changes\n\nRun git diff (or git diff HEAD if there are staged changes) to see what changed. If there are no git changes, review the most recently modified files that the user mentioned or that you edited earlier in this conversation.\n\n## Phase 2: Launch Three Review Agents in Parallel\n\nUse the Agent tool to launch all three agents concurrently in a single message. Pass each agent the full diff so it has the complete context.\n\n### Agent 1: Code Reuse Review\n\nFor each change:\n\n1. Search for existing utilities and helpers that could replace newly written code. Use Grep to find similar patterns elsewhere in the codebase — common locations are utility directories, shared modules, and files adjacent to the changed ones.\n2. Flag any new function that duplicates existing functionality. Suggest the existing function to use instead.\n3. Flag any inline logic that could use an existing utility — hand-rolled string manipulation, manual path handling, custom environment checks, ad-hoc type guards, and similar patterns are common candidates.\n\nNote: This is a greenfield app, so focus on maximizing simplicity and don't worry about changing things to achieve it.\n\n### Agent 2: Code Quality Review\n\nReview the same changes for hacky patterns:\n\n1. Redundant state: state that duplicates existing state, cached values that could be derived, observers/effects that could be direct calls\n2. Parameter sprawl: adding new parameters to a function instead of generalizing or restructuring existing ones\n3. Copy-paste with slight variation: near-duplicate code blocks that should be unified with a shared abstraction\n4. Leaky abstractions: exposing internal details that should be encapsulated, or breaking existing abstraction boundaries\n5. Stringly-typed code: using raw strings where constants, enums (string unions), or branded types already exist in the codebase\n\nNote: This is a greenfield app, so be aggressive in optimizing quality.\n\n### Agent 3: Efficiency Review\n\nReview the same changes for efficiency:\n\n1. Unnecessary work: redundant computations, repeated file reads, duplicate network/API calls, N+1 patterns\n2. Missed concurrency: independent operations run sequentially when they could run in parallel\n3. Hot-path bloat: new blocking work added to startup or per-request/per-render hot paths\n4. Unnecessary existence checks: pre-checking file/resource existence before operating (TOCTOU anti-pattern) — operate directly and handle the error\n5. Memory: unbounded data structures, missing cleanup, event listener leaks\n6. Overly broad operations: reading entire files when only a portion is needed, loading all items when filtering for one\n\n## Phase 3: Fix Issues\n\nWait for all three agents to complete. Aggregate their findings and fix each issue directly. If a finding is a false positive or not worth addressing, note it and move on — do not argue with the finding, just skip it.\n\nWhen done, briefly summarize what was fixed (or confirm the code was already clean)." + }, + "model": { + "String": "claude-opus-4-7" + }, + "label": { + "String": "Simplify (Opus)" + }, + "provider": { + "String": "anthropic" + } + } + }, + "toolchain": { + "id": "toolchain", + "attrs": { + "label": { + "String": "Toolchain" + }, + "provider": { + "String": "anthropic" + }, + "model": { + "String": "claude-opus-4-7" + }, + "shape": { + "String": "parallelogram" + }, + "script": { + "String": "command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1" + }, + "max_retries": { + "Integer": 0 + } + } + }, + "start": { + "id": "start", + "attrs": { + "shape": { + "String": "Mdiamond" + }, + "model": { + "String": "claude-opus-4-7" + }, + "provider": { + "String": "anthropic" + }, + "label": { + "String": "Start" + } + } + }, + "fixup": { + "id": "fixup", + "attrs": { + "provider": { + "String": "anthropic" + }, + "model": { + "String": "claude-opus-4-7" + }, + "prompt": { + "String": "The verify step failed. Read the build output from context and fix all clippy lint warnings, test failures, and generated docs errors." + }, + "label": { + "String": "Fixup" + }, + "max_visits": { + "Integer": 3 + } + } + }, + "verify": { + "id": "verify", + "attrs": { + "script": { + "String": "cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1 && cargo dev docs refresh 2>&1 && cargo dev docs check 2>&1" + }, + "label": { + "String": "Verify" + }, + "retry_target": { + "String": "fixup" + }, + "goal_gate": { + "Boolean": true + }, + "shape": { + "String": "parallelogram" + }, + "provider": { + "String": "anthropic" + }, + "model": { + "String": "claude-opus-4-7" + } + } + }, + "fmt": { + "id": "fmt", + "attrs": { + "provider": { + "String": "anthropic" + }, + "model": { + "String": "claude-opus-4-7" + }, + "max_retries": { + "Integer": 0 + }, + "label": { + "String": "Format" + }, + "script": { + "String": "cargo +nightly-2026-04-14 fmt --all 2>&1" + }, + "shape": { + "String": "parallelogram" + } + } + }, + "fix_lints": { + "id": "fix_lints", + "attrs": { + "model": { + "String": "claude-opus-4-7" + }, + "provider": { + "String": "anthropic" + }, + "label": { + "String": "Fix Lints" + }, + "max_visits": { + "Integer": 3 + }, + "prompt": { + "String": "The preflight lint step failed. Read the build output from context and fix all clippy lint warnings." + } + } + }, + "exit": { + "id": "exit", + "attrs": { + "model": { + "String": "claude-opus-4-7" + }, + "shape": { + "String": "Msquare" + }, + "label": { + "String": "Exit" + }, + "provider": { + "String": "anthropic" + } + } + }, + "simplify_gpt": { + "id": "simplify_gpt", + "attrs": { + "model": { + "String": "gpt-5.5" + }, + "label": { + "String": "Simplify (GPT-55)" + }, + "provider": { + "String": "openai" + }, + "prompt": { + "String": "# Simplify: Code Review and Cleanup\n\nReview changes vs. origin for reuse, quality, and efficiency. Fix any issues found.\n\n## Phase 1: Identify Changes\n\nRun git diff (or git diff HEAD if there are staged changes) to see what changed. If there are no git changes, review the most recently modified files that the user mentioned or that you edited earlier in this conversation.\n\n## Phase 2: Launch Three Review Agents in Parallel\n\nUse the Agent tool to launch all three agents concurrently in a single message. Pass each agent the full diff so it has the complete context.\n\n### Agent 1: Code Reuse Review\n\nFor each change:\n\n1. Search for existing utilities and helpers that could replace newly written code. Use Grep to find similar patterns elsewhere in the codebase — common locations are utility directories, shared modules, and files adjacent to the changed ones.\n2. Flag any new function that duplicates existing functionality. Suggest the existing function to use instead.\n3. Flag any inline logic that could use an existing utility — hand-rolled string manipulation, manual path handling, custom environment checks, ad-hoc type guards, and similar patterns are common candidates.\n\nNote: This is a greenfield app, so focus on maximizing simplicity and don't worry about changing things to achieve it.\n\n### Agent 2: Code Quality Review\n\nReview the same changes for hacky patterns:\n\n1. Redundant state: state that duplicates existing state, cached values that could be derived, observers/effects that could be direct calls\n2. Parameter sprawl: adding new parameters to a function instead of generalizing or restructuring existing ones\n3. Copy-paste with slight variation: near-duplicate code blocks that should be unified with a shared abstraction\n4. Leaky abstractions: exposing internal details that should be encapsulated, or breaking existing abstraction boundaries\n5. Stringly-typed code: using raw strings where constants, enums (string unions), or branded types already exist in the codebase\n\nNote: This is a greenfield app, so be aggressive in optimizing quality.\n\n### Agent 3: Efficiency Review\n\nReview the same changes for efficiency:\n\n1. Unnecessary work: redundant computations, repeated file reads, duplicate network/API calls, N+1 patterns\n2. Missed concurrency: independent operations run sequentially when they could run in parallel\n3. Hot-path bloat: new blocking work added to startup or per-request/per-render hot paths\n4. Unnecessary existence checks: pre-checking file/resource existence before operating (TOCTOU anti-pattern) — operate directly and handle the error\n5. Memory: unbounded data structures, missing cleanup, event listener leaks\n6. Overly broad operations: reading entire files when only a portion is needed, loading all items when filtering for one\n\n## Phase 3: Fix Issues\n\nWait for all three agents to complete. Aggregate their findings and fix each issue directly. If a finding is a false positive or not worth addressing, note it and move on — do not argue with the finding, just skip it.\n\nWhen done, briefly summarize what was fixed (or confirm the code was already clean)." + } + } + }, + "preflight_compile": { + "id": "preflight_compile", + "attrs": { + "model": { + "String": "claude-opus-4-7" + }, + "script": { + "String": "cargo check -q --workspace 2>&1" + }, + "label": { + "String": "Preflight Compile" + }, + "shape": { + "String": "parallelogram" + }, + "provider": { + "String": "anthropic" + }, + "max_retries": { + "Integer": 0 + } + } + }, + "preflight_lint": { + "id": "preflight_lint", + "attrs": { + "shape": { + "String": "parallelogram" + }, + "model": { + "String": "claude-opus-4-7" + }, + "script": { + "String": "cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1" + }, + "max_retries": { + "Integer": 0 + }, + "label": { + "String": "Preflight Lint" + }, + "provider": { + "String": "anthropic" + } + } + }, + "implement": { + "id": "implement", + "attrs": { + "model": { + "String": "claude-opus-4-7" + }, + "label": { + "String": "Implement" + }, + "provider": { + "String": "anthropic" + }, + "prompt": { + "String": "Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD." + } + } + } + }, + "edges": [ + { + "from": "start", + "to": "toolchain", + "attrs": {} + }, + { + "from": "toolchain", + "to": "preflight_compile", + "attrs": { + "condition": { + "String": "outcome=succeeded" + } + } + }, + { + "from": "toolchain", + "to": "exit", + "attrs": {} + }, + { + "from": "preflight_compile", + "to": "preflight_lint", + "attrs": { + "condition": { + "String": "outcome=succeeded" + } + } + }, + { + "from": "preflight_compile", + "to": "exit", + "attrs": {} + }, + { + "from": "preflight_lint", + "to": "implement", + "attrs": { + "condition": { + "String": "outcome=succeeded" + } + } + }, + { + "from": "preflight_lint", + "to": "fix_lints", + "attrs": {} + }, + { + "from": "fix_lints", + "to": "preflight_lint", + "attrs": {} + }, + { + "from": "implement", + "to": "simplify_opus", + "attrs": {} + }, + { + "from": "simplify_opus", + "to": "simplify_gpt", + "attrs": {} + }, + { + "from": "simplify_gpt", + "to": "verify", + "attrs": {} + }, + { + "from": "verify", + "to": "fmt", + "attrs": { + "condition": { + "String": "outcome=succeeded" + } + } + }, + { + "from": "verify", + "to": "fixup", + "attrs": {} + }, + { + "from": "fixup", + "to": "verify", + "attrs": {} + }, + { + "from": "fmt", + "to": "exit", + "attrs": {} + } + ], + "attrs": { + "goal": { + "String": "# Plan: end-to-end steering for running agents\n\n## Context\n\n`README.md` advertises \"Steer running agents mid-turn.\" Today the agent core fully supports it (`Session::steer`, `drain_steering`, `Turn::Steering` → user message, `agent.steering.injected` event, parity tests). Everything north of that is missing or stubbed:\n\n- `POST /runs/{id}/steer` is registered as `not_implemented` (501).\n- Endpoint is not in the OpenAPI spec.\n- No CLI command, no web UI wiring (only a placeholder demo-mode-gated \"Steer\" button on the running-runs board with no handler).\n- No bridge from the server's HTTP layer through the worker subprocess into the live `Session`.\n\nTwo flavors are required:\n\n- **Append** — push to the steering queue; agent picks it up at the next turn boundary (existing `Session::steer`).\n- **Interrupt** — cancel in-flight LLM stream and tool calls in the current round, then deliver as the next user turn. New code in `Session`.\n\nSteers can arrive when no agent stage is active, or when only non-agent stages are active — these **buffer** for the next API-mode session. Steers that arrive when only CLI-mode agent stages are active are **rejected** (no steerable target). Mixed runs with at least one API-mode agent active are accepted and broadcast.\n\n## Decisions\n\n- Scope: full stack — wire protocol, agent, worker, server, OpenAPI, CLI, web UI.\n- Parallel stages (`max_parallel = 4`): broadcast to every active API-mode `Session` in the run.\n- Status policy: accept only when run status is `running`. Reject `blocked` with a hint to use the interview-answer endpoint. Reject terminal states with 409.\n- **CLI-mode steerability predicate (target-oriented, best-effort).** Server's view derives from asynchronously consumed events, so the 409 below is best-effort. Stale state can lead to a forwarded steer that the worker hub then buffers (`agent.steer.buffered`) or drops at run end (`agent.steer.dropped { reason: \"run_ended\" }`). UI surfaces both via SSE.\n 1. ≥1 API-mode agent stage active → forward (broadcast).\n 2. No active agent stages at all (between stages, non-agent stage, idle) → forward (worker buffers for next session).\n 3. Active agent stages exist but none are API-mode → **best-effort 409**.\n- Web UI shows the Steer button whenever `status === \"running\"`; rejection reason flows through the 409 response and is surfaced inline.\n- Every steer carries an `actor: Principal` end-to-end (HTTP → envelope → worker → agent). Per `docs/internal/events-strategy.md:83`, `actor` lives only at top-level `RunEvent.actor`; **not** in event-specific props.\n- Both transport variants must work: `RunAnswerTransport::Subprocess` (worker control JSONL) and `RunAnswerTransport::InProcess` (direct call into the in-process hub).\n- **Round-token cancellation is the sole marker for steering interrupts.** No new `InterruptReason::SteerInterrupt` variant. The loop distinguishes terminal cancel from steer-interrupt by which token fired (`cancel_token` vs `round_token`). Existing `interrupt_reason` (used for `WallClockTimeout` / `Cancelled`) is unchanged.\n- **Bounded queues.** Per-session steering queue cap = 32 messages; per-run pending buffer cap = 32 messages. Overflow evicts oldest (FIFO) and emits `agent.steer.dropped { count, reason }`. Sizes are workspace constants in `fabro-workflow`.\n- **Buffered-steer fanout semantics:** buffered steers go to the **first** session that registers after an empty-active period. Sister parallel sessions registering at almost the same time do not replay the buffer. Documented limitation; per-stage targeting (deferred) is the natural future fix.\n\n## Message flow\n\n```\nHTTP POST /runs/{id}/steer { text, interrupt } (auth → actor: Principal)\n → fabro-server handler\n ├─ validates status + steerability predicate from active_api_stages /\n │ active_cli_stages tracked from worker-emitted events\n ├─ Subprocess: WorkerControlEnvelope::steer(text, kind, actor) → control_tx\n │ → pump_worker_control_jsonl → worker stdin → apply_worker_control_line\n │ → SteeringHub.deliver(text, kind, actor)\n └─ InProcess: directly call SteeringHub.deliver(text, kind, actor) on the\n hub stored alongside the in-process interviewer\n → SteeringHub.deliver:\n ├─ active API handles → broadcast: handle.queue.push((text, kind, actor))\n │ + if Interrupt: handle.round_token.cancel()\n └─ none → push to pending Vec\n → Session round loop: top-of-loop drain_steering() emits\n AgentEvent::SteeringInjected { text, kind } with actor flowing through\n internal event metadata; agent_actor_for_event lifts it to RunEvent.actor.\n```\n\n## Implementation\n\n### 1. Wire protocol — extend `WorkerControlEnvelope`\n\n**Files:** `lib/crates/fabro-types/src/lib.rs` (or new `steering.rs`), `lib/crates/fabro-interview/src/control_protocol.rs`\n\nDefine `SteerKind` in `fabro-types` (not `fabro-interview` — `fabro-interview` already depends on `fabro-types` per `control_protocol.rs:1`, so the canonical enum must live in the lower crate to avoid a cycle):\n\n```rust\n// fabro-types\n#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]\n#[serde(rename_all = \"lowercase\")]\npub enum SteerKind { Append, Interrupt }\n```\n\n`fabro-interview` re-exports it and adds the envelope variant:\n\n```rust\n// fabro-interview/src/control_protocol.rs\npub use fabro_types::SteerKind;\n\n#[serde(rename = \"run.steer\")]\nSteer {\n text: String,\n kind: SteerKind,\n actor: Principal, // matches interview.answer\n},\n```\n\nAdd `WorkerControlEnvelope::steer(text, kind, actor)`. Round-trip serde tests for both kinds + actor next to existing tests at line 104+.\n\n### 2. Agent — `interrupt_with` + round-level cancel + control handle\n\n**Files:** `lib/crates/fabro-agent/src/session.rs`, `lib/crates/fabro-agent/src/types.rs`, `lib/crates/fabro-agent/src/error.rs`, `lib/crates/fabro-agent/src/tool_execution.rs`, `lib/crates/fabro-agent/tests/it/parity_matrix.rs`\n\nChanges to `Session`:\n\n- New field `round_token: Arc>` — replaceable per round.\n- Change `steering_queue` element type from `String` to `(String, SteerKind, Option)` so per-message kind+actor survive into the emitted event.\n- New method `interrupt_with(&self, text: String, actor: Option)`: push `(text, Interrupt, actor)` and cancel `round_token`. **Does not** touch `interrupt_reason` — round-token cancellation alone marks the steer-interrupt path.\n- Existing `steer(text)` updated to push `(text, Append, None)`.\n- No new `InterruptReason` variant. Existing `WallClockTimeout` / `Cancelled` semantics are unchanged. The loop disambiguates by inspecting tokens:\n - `cancel_token.is_cancelled()` → terminal close (existing behavior).\n - `round_token.is_cancelled() && !cancel_token.is_cancelled()` → steer interrupt → continue.\n- New method `control_handle(&self) -> SessionControlHandle` returning `Arc` clones of `steering_queue` and `round_token`. The hub stores the *handle*, not the `Session`. This avoids the ownership mismatch with `AgentApiBackend` (Session is owned by value and mutated via `process_input(&mut self)` in `handler/llm/api.rs:444-494`).\n- `SessionControlHandle::steer(text, actor)` and `interrupt_with(text, actor)` thin wrappers — the hub calls these.\n- Existing `AgentEvent::SteeringInjected` props gain `kind: SteerKind` only. Actor flows through internal event metadata (set on the emitted event), then `agent_actor_for_event` lifts it to top-level `RunEvent.actor` per the events strategy.\n\n**Loop changes in `process_input` (lines 603–941):**\n\n- **Move `drain_steering()` to the top of the loop body**, before `compact_if_needed`/`build_request`. Today: line 694 (before loop, once) and line 924 (after tools). After a SteerInterrupt `continue`, neither runs before the next request. Top-of-loop drain fixes this. Remove the line-694 pre-loop call (top-of-loop covers it on iter 1) and the line-924 post-tool call (next iter's top-of-loop covers it).\n- At top of each iteration: if `round_token` is cancelled, replace with a fresh `CancellationToken`. (No `interrupt_reason` to clear — round-token cancellation is the marker.)\n- Build per-round composite token from `cancel_token` (terminal) and `round_token` (per-round).\n- **Cancellation propagation — two distinct strategies:**\n - **LLM stream awaits** (preemptive — safe to drop in-flight): wrap with `tokio::select!` against `composite.cancelled()` at:\n - `open_stream_with_retry(...)` and any internal retry-backoff `tokio::time::sleep`\n - `event_stream.next()` per chunk (so an idle stream that never produces another chunk doesn't pin the loop)\n - **Tool execution** (cooperative — must NOT drop the future): pass the composite token as a parameter to `execute_tool_calls`. It runs to completion, returning \"Cancelled\" entries internally for any in-flight tool (existing path: `tool_execution.rs:80`). Do **not** wrap in `select!` — dropping the future would lose synthesized cancel results and break the `tool_use`↔`tool_result` invariant.\n- After LLM stream and after tools, branch on whether the round was interrupted:\n - **Mid-LLM interrupt** (`record_assistant_turn` at line 859 has not run yet): drop the unrecorded turn. **Also clear visible UI output**: if any `TextDelta` or `ReasoningDelta` was emitted in the dropped round, emit `AgentEvent::AssistantOutputReplace { text: \"\", reasoning: None }` before `continue` (mirrors the existing retry-clears-output pattern at session.rs:828). No tool_results needed because no `tool_use` was committed to history.\n - **Mid-tool interrupt** (assistant turn with `tool_use` blocks already recorded at line 859 before `execute_tool_calls` ran): `execute_tool_calls` runs to completion and returns **one `ToolResult` per `tool_use` block**. Content varies by tool — bash returns `Ok(\"Command cancelled.\\n…\")` (tools.rs:265-266), other tools may return partial output, an error message, or a synthetic Cancelled marker. The Anthropic invariant only requires one-per-block, not a specific content shape. Always push `Turn::ToolResults` with whatever `execute_tool_calls` returned (existing line 909-921 path), then branch on which token fired. Refactor the current `cancel_token.is_cancelled()` branch (lines 907-915) to: append tool_results unconditionally → close+return-Err (if `cancel_token` fired) or `continue` (if only `round_token` fired).\n- **Append-during-final-response fix (race-safe, dependency-safe).** Today line 881-883 unconditionally `break` when `tool_calls.is_empty()`. A naive `if steering_queue.is_empty() { break }` still loses steers that arrive between the empty check and the function return because the hub still considers the session active. The full close-the-door dance (unregister → check → re-register-or-break) crosses a crate boundary the wrong way (`fabro-agent` does not depend on `fabro-workflow`; reverse cycles per `Cargo.toml:22`). Solution: small trait owned by fabro-agent, implemented in fabro-workflow.\n\n ```rust\n // fabro-agent\n pub trait CompletionCoordinator: Send + Sync {\n /// Called at natural completion (tool_calls empty).\n /// Return true to continue the loop (queue is non-empty),\n /// false to break. Implementor coordinates with whatever\n /// owns the steering source.\n fn on_natural_completion(&self) -> bool;\n }\n ```\n\n `Session` gains `completion_coordinator: Option>`, defaulting to `None` (preserves existing behavior — direct `Session::new` callers and tests just break naturally).\n\n Loop:\n\n ```rust\n if tool_calls.is_empty() {\n let should_continue = self.completion_coordinator\n .as_ref()\n .is_some_and(|c| c.on_natural_completion());\n if should_continue { continue; }\n break;\n }\n ```\n\n In fabro-workflow, an adapter implementing the trait holds `(Arc, StageId, SessionControlHandle)` and does:\n\n ```rust\n fn on_natural_completion(&self) -> bool {\n self.hub.unregister(self.stage_id.clone()); // serializes vs hub.deliver\n if self.handle.queue_is_empty() { return false; }\n self.hub.register(self.stage_id.clone(), self.handle.clone());\n true // session's next iteration drains\n }\n ```\n\n `AgentApiBackend::run` builds the adapter, sets it on the session before `process_input`, and removes it after.\n\n **Hub locking discipline (race safety):** `SteeringHub::deliver` holds the `active` *read* lock for the entire push (clone handle + push to queue happen under the read lock). `unregister` takes the *write* lock. RwLock semantics serialize them — once `unregister` returns, no in-flight push can be racing. The post-unregister queue check sees a stable result.\n- `cancel_token.is_cancelled()` (terminal) still returns `Err(interrupted_error())` and closes — unchanged.\n\nTests in `parity_matrix.rs`:\n\n- `steering_interrupt_mid_llm_idle_stream` — fire `interrupt_with` while the LLM stream is open but producing no chunks. Assert interrupt takes effect within ~1s (proves `tokio::select!` is wired around `next()`).\n- `steering_interrupt_mid_llm_streaming` — fire mid-stream after at least one `TextDelta` has been emitted; assert (a) `AssistantOutputReplace { text: \"\", reasoning: None }` is emitted before the next round (clears stale partial output in the UI), (b) next turn includes the steer text, (c) event has `kind: \"interrupt\"`.\n- `steering_interrupt_mid_tool` — fire while a Bash tool is running; assert (a) `Turn::ToolResults` immediately follows the assistant tool-use turn (one ToolResult per tool_use, content unspecified — could be partial output, \"Command cancelled\", or an error message), then (b) `Turn::Steering` with the new text. No dangling `tool_use`. Test asserts shape, not content.\n- `steering_no_dangling_tool_use_invariant` — assert that no `Turn::Steering` immediately follows a `Turn::Assistant` containing `tool_use` blocks without an intervening `Turn::ToolResults`. Asserts shape (one ToolResult per tool_use), not content. Guards against `select!`-around-tools regressions.\n- `steering_append_kind_field` — fire `steer()` between rounds; assert event carries `kind: \"append\"`.\n- `append_during_final_response_triggers_extra_round` — fire `steer()` while LLM is producing a final no-tool response. Assert agent does NOT exit `process_input` after `tool_calls.is_empty()`; instead runs another model turn that incorporates the steer. (Test uses a stub `CompletionCoordinator` impl that returns `true` once when the queue is non-empty — keeps the agent test free of workflow/hub dependencies.)\n\nQueue overflow tests live at the **`SteeringHub` layer in fabro-workflow**, not here. Direct `Session::steer` callers intentionally bypass the cap, so the agent has nothing to test for overflow.\n\n### 3. Worker — `SteeringHub` + control plumbing\n\n**Files:** `lib/crates/fabro-cli/src/commands/run/runner.rs`, `lib/crates/fabro-workflow/src/services.rs`, `lib/crates/fabro-workflow/src/operations/start.rs`, `lib/crates/fabro-workflow/src/handler/llm/api.rs`\n\nNew type (in `fabro-workflow`, alongside `RunServices`):\n\n```rust\n// All locks below are std::sync — methods are sync and never await while holding them.\npub struct SteeringHub {\n active: std::sync::RwLock>,\n pending: std::sync::Mutex>, // bounded, FIFO\n emitter: Arc,\n}\n\nstruct PendingSteer { text: String, kind: SteerKind, actor: Option }\n\nconst PER_SESSION_QUEUE_CAP: usize = 32;\nconst PER_RUN_PENDING_CAP: usize = 32;\n\nimpl SteeringHub {\n pub fn deliver(&self, text: String, kind: SteerKind, actor: Option);\n pub fn register(&self, stage_id: StageId, handle: SessionControlHandle);\n pub fn unregister(&self, stage_id: StageId);\n pub fn drain_pending_at_run_end(&self); // emits agent.steer.dropped { reason: \"run_ended\" } if any\n}\n```\n\n- `register` decides drain-vs-replace based on **current active-map state**, not history:\n - If `stage_id` is **not already in active** → insert + drain pending into this handle as `Append` + emit `agent.steering.attached`. Covers first-register-after-empty AND close-the-door re-register (which closes the gap where steers can buffer between unregister and re-register).\n - If `stage_id` **is already in active** → replace the handle, do **not** drain pending, do **not** re-emit `attached`. Covers failover (handle replaced under the same id without an intervening unregister).\n- `unregister` is **idempotent**: `agent.steering.detached` fires only when `active.remove(stage_id)` returns `Some`. The close-the-door call removes-and-emits once; the RAII guard at function exit becomes a no-op (entry already gone). Prevents double-emit on natural completion.\n- `deliver` broadcasts to active handles **or** pushes to pending — branched **under the active read lock** so the empty/non-empty decision is atomic with the push. Documented lock order: **active first, then queue or pending; never reverse.** All locks are `std::sync::{RwLock, Mutex}`; **no `.await` while holding any of them.** Sync methods make `CompletionCoordinator::on_natural_completion` callable from the agent loop without converting it to async (tokio locks would force `.await`). This makes the close-the-door pattern race-safe end-to-end.\n- Internal helper `enqueue_into_session_queue(handle, item)` is used by both the broadcast path and the pending-flush path (called from `register`), guaranteeing identical cap enforcement and drop-event emission across both code paths.\n- Sister parallel sessions registering immediately after the first don't replay the buffer (it was drained on the first register) — documented limitation; broadcast-to-future-sessions is deferred with the per-stage targeting feature.\n- **Queue bounds enforced at the hub layer.** Before pushing into a session's `steering_queue` via `SessionControlHandle`, the hub checks `len() >= PER_SESSION_QUEUE_CAP` and evicts the front. Before pushing into `pending`, checks against `PER_RUN_PENDING_CAP`. On eviction, emits `agent.steer.dropped { count: 1, reason: \"queue_full\" }`. **Direct callers of `Session::steer` (loop-detection auto-injection at session.rs:931, tests) bypass the cap** — that's intentional; internal one-shot warnings shouldn't trigger user-facing drop events.\n\n**Plumbing (explicit, not \"via the same path\"):**\n\nIn `runner.rs::execute()` (around lines 88–101): construct `let steering_hub = Arc::new(SteeringHub::new(emitter.clone()));` next to `interviewer` and `cancel_token`. Pass it both into:\n\n1. `spawn_worker_control_stream(interviewer, cancel_token, steering_hub.clone())` — extend the function signature to accept the hub.\n2. `StartServices.steering_hub: Arc` — new required field. Threaded through `operations::start` → `RunServices` → `EngineServices` → handler dispatch.\n\nIn `runner.rs::apply_worker_control_line` (lines 226–250): add a match arm:\n\n```rust\nWorkerControlMessage::Steer { text, kind, actor } => {\n steering_hub.deliver(text, kind, Some(actor));\n}\n```\n\nIn `AgentApiBackend::run()` (`handler/llm/api.rs:444-494`):\n\n- Compute `let stage_id = stage_scope.stage_id();` from the existing `stage_scope` at line 476 (`StageScope::stage_id()` returns `StageId::new(node_id, visit)` per `stage_scope.rs:64-65`). Use this `StageId` everywhere — **not** the bare `node.id` string.\n- After the `Session` is built/cached but before `process_input`, call `services.steering_hub.register(stage_id.clone(), session.control_handle())`.\n- Use a `scopeguard`-style RAII guard so `unregister(stage_id.clone())` runs on success, error, and panic.\n- **Failover (lines 527-572):** inside the failover loop, immediately after `session = new_session;` (line 545) and before `session.initialize().await` (line 556), call `services.steering_hub.register(stage_id.clone(), session.control_handle())` again. The hub overwrites the abandoned handle with the new one. The RAII unregister still works because the same `stage_id` is keyed.\n- The hub never holds the `Session` — only the `Arc`-clones in `SessionControlHandle`. Sidesteps the ownership mismatch.\n\n`AgentCliBackend::run()` is **not** modified — it never registers, so the hub's `active` set never includes CLI stages. The server's steerability predicate uses the `agent.steering.attached/detached` and `agent.cli.started/completed` events to know what's active.\n\n**Run-end drain placement (async cleanup pattern).** Inside `operations::start` (`lib/crates/fabro-workflow/src/operations/start.rs`), wrap the pipeline execution into a result-returning block, then drain pending and flush events explicitly **before** propagating:\n\n```rust\nlet result = run_pipeline(...).await; // success or error\nsteering_hub.drain_pending_at_run_end(); // sync emit of agent.steer.dropped\nstore_progress_logger.flush().await; // awaited flush moves them through the sink\nresult?\n```\n\nA `scopeguard` calling `drain_pending_at_run_end()` is **only** a last-ditch panic fallback — it cannot await the flush, so it's not the primary delivery path. The explicit pattern handles both success and error cleanly. Calling drain from the worker's outer wrap-up (after `operations::start` returns) would lose events because `store_progress_logger.flush().await` at line 818 already ran.\n\n### 4. Server — HTTP handler + OpenAPI + per-stage tracking + InProcess support\n\n**Files:** `docs/public/api-reference/fabro-api.yaml`, `lib/crates/fabro-server/src/server/handler/mod.rs`, `lib/crates/fabro-server/src/server.rs` (or new `handler/steer.rs`), `lib/crates/fabro-server/src/server/tests.rs`\n\nOpenAPI: `POST /runs/{id}/steer` with body `SteerRequest { text: string (required, 1..8192), interrupt: boolean (default false) }`. Responses: `202 Accepted`, `400`, `404`, `409`, `503`. Tag: `Human-in-the-Loop`. Authenticated user becomes `Principal` for the envelope.\n\nHandler (mirror cancel at `handler/lifecycle.rs:162`):\n\n1. Look up `ManagedRun` via `AppState.runs`.\n2. Validate, in order:\n - 404 if missing.\n - 409 if status is `blocked` with `code: \"use_answer_endpoint\"`, hint: `POST /runs/{id}/questions/{qid}/answer`.\n - 409 if terminal (`succeeded`/`failed`/`cancelled`/`archived`).\n - 409 if not `running`.\n - 409 if **target-oriented predicate** rejects: `active_api_stages.is_empty() && !active_cli_stages.is_empty()` with `code: \"cli_agent_not_steerable\"`, message: \"All currently running agent stages are CLI-mode and cannot be steered.\"\n - Otherwise: forward.\n3. **Transport branch on `ManagedRun.answer_transport`:**\n - `Subprocess { control_tx }`: send `WorkerControlEnvelope::steer(text, kind, actor)` with the existing 1s timeout pattern. Map `Timeout`/`Closed` to 503.\n - `InProcess { interviewer, steering_hub }`: directly call `steering_hub.deliver(text, kind, Some(actor))`. No envelope, no JSONL hop, no timeout — same hub the in-process worker would use. Requires storing an `Arc` alongside `interviewer` in `RunAnswerTransport::InProcess` (`server.rs:245`). The in-process spawn site `execute_run_in_process` (line 2541) creates and stores both.\n4. Return 202.\n\n**Tracking active-stage modes (server side):** `ManagedRun` gains:\n\n```rust\nactive_api_stages: HashSet, // primary: agent.steering.attached/detached\nactive_cli_stages: HashSet, // primary: agent.cli.started; backstops below\n```\n\nPlain `HashSet` (no inner lock) — `ManagedRun` is already accessed under `state.runs.lock()` (`server.rs:441` AppState definition; mutation pattern at `server.rs:1724, 1735` for the existing `accepted_questions: HashSet` field at `server.rs:196`). Adding inner `Mutex` would be redundant nested locking.\n\nUpdated by the server's existing event-consumption path. **No reuse of `agent.session.started/ended`** — those events do not reliably fire per stage invocation: `Session::initialize()` (and thus `SessionStarted`) is skipped for reused sessions in `api.rs:490`, and `SessionEnded` only fires on explicit `close()`. The hub-emitted `attached/detached` events fire deterministically per `register/unregister` call inside `AgentApiBackend::run`, which is exactly the steerable window.\n\n**Backstops to prevent leaks** (CLI tracking is fragile because `AgentCliStarted` at cli.rs:511 and `AgentCliCompleted` at cli.rs:648 are 137 lines apart with fallible operations between, and the existing CLI cancel bug means many error paths skip the completion emit):\n\n- On `stage.completed` **and** `stage.failed` (any kind): remove the stage_id (read from top-level `RunEvent.stage_id`) from **both** `active_api_stages` and `active_cli_stages`. Both events fire from the workflow lifecycle (`lifecycle/event.rs:153, 220, 271`); covering only `stage.completed` would leak on the failure path — exactly where the existing CLI cancel bug already strands stages.\n- On terminal run events (`run.completed` / `run.failed`): clear both sets entirely. (Cancellation is folded into `run.failed` via its `reason` field — there is no separate `run.cancelled` event in `lib/crates/fabro-types/src/run_event/mod.rs:87-90`.)\n\nImplementation note for a follow-up PR (out of scope here, in the same area as the existing CLI-cancel debt): wrap the CLI backend's `AgentCliCompleted` emission in a scopeguard so it always fires regardless of error path.\n\n**CLI-only rejection is best-effort.** Server consumes events asynchronously through the run-store subscription path (`server.rs:2023`), so its view of `active_api_stages` / `active_cli_stages` lags actual worker state by a small window. A steer that the server forwards based on a stale view will be handled correctly by the worker hub: if no API session is registered by arrival, the steer buffers and emits `agent.steer.buffered`, which the UI surfaces. The 409-on-all-CLI gate is an optimization for the synchronous user-feedback case; the worker-side hub is the authoritative safety net. Authoritative server-side rejection (round-tripping a confirmation back through the worker control plane) is out of scope.\n\n**After OpenAPI changes, regenerate clients (per `CLAUDE.md` API workflow):**\n\n```bash\ncargo build -p fabro-api # regenerates Rust client via build.rs + progenitor\ncd lib/packages/fabro-api-client && bun run generate # regenerates TypeScript Axios client\n```\n\nBoth must run before `bun run typecheck` in `apps/fabro-web` will pass.\n\n### 5. CLI — `fabro steer`\n\n**Files:** `lib/crates/fabro-cli/src/args.rs`, new `lib/crates/fabro-cli/src/commands/steer.rs`, `lib/crates/fabro-cli/src/commands/mod.rs`, `lib/crates/fabro-cli/src/server_client.rs`, `lib/crates/fabro-cli/src/main.rs`\n\nAdd a new top-level `Commands::Steer(SteerArgs)` (sibling to `Commands::RunCmd`, `Commands::Exec`, etc. in `args.rs:1016`). New top-level command from scratch — no existing `fabro cancel` to mirror (cancel today is Ctrl+C in attached or HTTP-direct).\n\n```\nfabro steer [--interrupt]\nfabro steer --text-stdin [--interrupt] # editors / pipes\n```\n\nImplementation calls a new `server_client.steer_run(run_id, text, kind)` via the regenerated typed API client. Error mapping mirrors the cancel HTTP path.\n\n### 6. Web UI\n\n**Files:** `apps/fabro-web/app/components/steer-composer.tsx` (new), `apps/fabro-web/app/components/steer-composer.test.tsx` (new), `apps/fabro-web/app/lib/mutations.ts`, `apps/fabro-web/app/lib/run-events.ts`, `apps/fabro-web/app/hooks/use-run-toasts.ts` (new), `apps/fabro-web/app/routes/runs.tsx`, `apps/fabro-web/app/routes/run-detail.tsx`\n\n**Composer:** new `SteerComposer` component — modal/popover with textarea, primary \"Send\" button, secondary \"Interrupt\" button, same Enter / Shift+Enter affordance as `interview-dock.tsx`. Reuse `ErrorMessage` from `ui.tsx`.\n\n**Mutation:** `useSteerRun(runId)` in `lib/mutations.ts` mirroring `useSubmitInterviewAnswer` (lines 97–117). On 409 with `code: \"cli_agent_not_steerable\"`, surface inline (\"All running agent stages are CLI-mode and can't be steered.\"). Invalidates run detail on success.\n\n**Surfacing:** wire the existing \"Steer\" button in `routes/runs.tsx:42, 365–369`:\n- Remove the demo-mode gate at lines 437–439.\n- Open `SteerComposer` on click.\n\nAdd a \"Steer\" button to `run-detail.tsx` page header (only when `statusKind === \"running\"`), opening the same composer.\n\n**Toast dispatch:** `lib/run-events.ts` only resolves SWR-invalidation keys (line 44) — does not dispatch toasts. Add `useRunToasts(runId)` hook in `app/hooks/use-run-toasts.ts` that subscribes to the same SSE stream (via existing `subscribeToSharedEventSource`) and calls `useToast().push(...)` for steer-related events, deduping by event id. Mount from `run-detail.tsx` next to `useRunEvents`. SSE invalidation in `run-events.ts` still gets the new event names so SWR refetches; toast is the new hook's job.\n\nToast copy:\n- `agent.steering.injected` with `kind: \"append\"` → \"Steer delivered.\"\n- `agent.steering.injected` with `kind: \"interrupt\"` → \"Agent interrupted — your message is the next turn.\"\n- `agent.steer.buffered` → \"Steer queued — will apply when an agent stage runs.\"\n- `agent.steer.dropped` with `reason: \"queue_full\"` → \"Steer rate limit reached; oldest queued steer dropped.\"\n- `agent.steer.dropped` with `reason: \"run_ended\"` → \"Run ended before queued steer(s) could apply.\"\n\nNote: keep `InterviewDock` semantics untouched — `blocked` runs only. Steer composer handles `running` only. Mutually exclusive surfaces.\n\n## Events\n\n- `agent.steering.injected` — **modified**. `AgentSteeringInjectedProps` (in `lib/crates/fabro-types/src/run_event/agent.rs`) gains `kind: \"append\" | \"interrupt\"`. Actor lives at top-level `RunEvent.actor` only — set via `agent_actor_for_event` from the `actor` carried internally on the `AgentEvent::SteeringInjected` variant. Per `docs/internal/events-strategy.md:83`.\n- `agent.steering.attached` / `agent.steering.detached` — **new**. Emitted by `SteeringHub::register` (only when newly inserted, not on replace) / `unregister` (only when active.remove returned Some). Workflow `Event` variants carry `StageId` internally; `stored_event_fields_for_variant` lifts to top-level `RunEvent.stage_id` (mod.rs:36) so generic event consumers see it where they expect. **Props are empty** — no duplicate `stage_id` in props. Distinct names from existing `agent.session.started/ended` to avoid confusion.\n- `agent.steer.buffered` — **new**. Emitted by the worker hub when a steer arrives with no active session and is parked. Carries `{ kind }` in props (no actor in props — top-level only).\n- `agent.steer.dropped` — **new**. Two shapes:\n - `reason: \"queue_full\"`, `count: 1` — single-item drop with a known dropped steer. Carries the dropped item's `actor` internally; `stored_event_fields_for_variant` lifts it to top-level `RunEvent.actor`.\n - `reason: \"run_ended\"`, `count: N` — aggregate; possibly multiple actors. **No user actor** at top level (system actor); aggregation loss documented.\n\nFor each new event: typed props in `lib/crates/fabro-types/src/run_event/agent.rs`, variant on `EventBody` in `mod.rs`, workflow conversion in `fabro-workflow/src/event/convert.rs`, name in `fabro-workflow/src/event/names.rs`.\n\n**Actor lifting (different paths for different emitters):**\n- `agent.steering.injected` is agent-emitted (`AgentEvent::SteeringInjected`). Carry `actor` on the internal variant; lift via `agent_actor_for_event` (`stored_fields.rs:198`).\n- `agent.steer.buffered` is workflow-hub-emitted (`SteeringHub::deliver`, no agent involvement). Add `actor: Option` to its workflow `Event` variant; lift via `stored_event_fields_for_variant` (`stored_fields.rs:57`).\n- `agent.steer.dropped` with `reason: \"queue_full\"`: carry dropped item's `actor` on the workflow `Event` variant; lift via `stored_event_fields_for_variant`.\n- `agent.steer.dropped` with `reason: \"run_ended\"`: system actor, no user actor lifted.\n- `agent.steering.attached/detached`: lifecycle, no user actor.\n\n## Test strategy\n\n- **Unit** — control-protocol round-trip including actor (`fabro-interview`); `SteerKind` round-trip in `fabro-types`; `SteeringHub` buffer + broadcast + register-drains-or-replaces by active-map state + idempotent unregister + `drain_pending_at_run_end` + **per-session queue overflow drops oldest** + **per-run pending overflow drops oldest** (both at the hub layer, asserting `agent.steer.dropped { reason: \"queue_full\" }` carries the correct actor at top level); server's steerability predicate over mixed `(active_api_stages, active_cli_stages)` sets.\n- **Agent integration** — `parity_matrix.rs` scenarios listed above (idle-stream interrupt, mid-streaming interrupt with stale-output clear, mid-tool interrupt with shape-only assertion, no-dangling-tool-use invariant, append kind field, append-during-final-response triggers extra round). Idle-stream guards against `tokio::select!` regression on stream awaits; no-dangling guards against `select!`-around-tools regression. **Queue overflow is exclusively a hub-layer test** — direct `Session::steer` callers intentionally bypass the cap.\n- **Workflow event conversion** — new test that builds an `AgentEvent::SteeringInjected { actor, kind, text }`, runs it through the conversion machinery, and asserts the resulting `RunEvent` has `kind` in props (no `actor` in props) and the user actor at top-level `RunEvent.actor`. Without this test, a future refactor of `agent_actor_for_event` (`stored_fields.rs:198`) could silently regress steering to `None` — it currently falls through for all variants except `AssistantMessage`/`ToolCall*`.\n- **Server** — handler unit tests for the full status + steerability matrix (running OK, blocked → 409, terminal → 409, missing → 404, all-CLI → 409, mixed API+CLI → accept, no-active-agent → accept, in-process transport delivers via direct hub call, subprocess delivers via control_tx). Existing `in_process_answer_transport_cancel_run_cancels_pending_interviews` (tests.rs:1710) is the template for an in-process steer test.\n- **CLI** — happy-path `fabro steer` against a fake server, plus `--text-stdin`.\n- **Web** — `SteerComposer` (textarea + two buttons + disabled-when-empty); `useSteerRun` posts the right body and surfaces 409 inline; `useRunToasts` dispatches expected toasts with dedup.\n\n## Verification\n\n```bash\n# Backend\ncargo build --workspace\ncargo nextest run --workspace\ncargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings\n\n# API client regeneration (after OpenAPI changes)\ncargo build -p fabro-api\ncd lib/packages/fabro-api-client && bun run generate\ncd ../../..\n\n# Web (depends on regenerated TS client above)\ncd apps/fabro-web && bun run typecheck && bun test\n\n# Manual end-to-end (subprocess transport)\nfabro server start # terminal 1\nfabro run repl # terminal 2 — start a run\nfabro steer \"Try a different approach\" # terminal 3 — append\nfabro steer \"Stop, do X instead\" --interrupt # terminal 3 — interrupt\n# In browser: open the run, click Steer, type and Send / Interrupt; verify SSE event arrives and toast renders.\n\n# Manual end-to-end (in-process transport)\n# Use a registry override / test config that selects InProcess transport,\n# repeat steer + interrupt; same observable behavior, no JSONL hop.\n```\n\n## Out of scope\n\n- Per-stage steer targeting in UI/CLI (broadcast only for v1).\n- Persisting unconsumed steers across run resume.\n- Steering of non-agent stages (commands, conditionals, parallel).\n- Steering CLI-mode agent stages (claude/codex/gemini): structurally impossible without changes to those external CLIs. Server returns 409 only when *all* active agent stages are CLI-mode.\n- Fixing the pre-existing CLI-mode cancel bug (RunCancel currently ignored; subprocesses orphan). Tracked separately.\n- Slack-driven steering.\n- Auth/permission model beyond what `cancel` already does — same caller can do both.\n- Worker-side `stdin` protocol versioning beyond the existing `v: 1` envelope.\n" + }, + "rankdir": { + "String": "LR" + }, + "model_stylesheet": { + "String": "\n * { model: claude-opus-4-7; }\n " + } + } + }, + "workflow_slug": "implement-plan", + "source_directory": "/Users/bhelmkamp/p/fabro-sh/fabro", + "provenance": { + "server": { + "version": "0.223.0-nightly.0" + }, + "client": { + "user_agent": "fabro-cli/0.223.0-nightly.0", + "name": "fabro-cli", + "version": "0.223.0-nightly.0" + }, + "subject": { + "kind": "user", + "identity": { + "issuer": "https://github.com", + "subject": "19" + }, + "login": "brynary", + "auth_method": "github" + } + }, + "manifest_blob": "723e73d1abd8d1e3158d3e4b5dcc8295fe47d15adbc5b1982a8c68c10e3d9d82", + "definition_blob": "791b2ce7454b6fff8fa26bea4af48533a25d0e56c4cecc44be1ea071680b4f9d", + "git": { + "origin_url": "https://github.com/fabro-sh/fabro", + "branch": "main", + "sha": "8064aa269eb893efca2a1654d9ece4acf8129307", + "dirty": "clean", + "push_outcome": { + "type": "not_attempted" + } + }, + "in_place": false + }, + "graph_source": "digraph ImplementPlan {\n graph [\n goal=\"Implement and simplify\",\n model_stylesheet=\"\n * { model: claude-opus-4-7; }\n \"\n ]\n rankdir=LR\n\n start [shape=Mdiamond, label=\"Start\"]\n exit [shape=Msquare, label=\"Exit\"]\n\n toolchain [label=\"Toolchain\", shape=parallelogram, script=\"command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1\", max_retries=0]\n preflight_compile [label=\"Preflight Compile\", shape=parallelogram, script=\"cargo check -q --workspace 2>&1\", max_retries=0]\n preflight_lint [label=\"Preflight Lint\", shape=parallelogram, script=\"cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1\", max_retries=0]\n fix_lints [label=\"Fix Lints\", prompt=\"The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.\", max_visits=3]\n implement [label=\"Implement\", prompt=\"Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD.\"]\n simplify_opus [label=\"Simplify (Opus)\", prompt=\"@prompts/simplify.md\"]\n simplify_gpt [label=\"Simplify (GPT-55)\", prompt=\"@prompts/simplify.md\", model=\"gpt-55\"]\n verify [label=\"Verify\", shape=parallelogram, script=\"cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1 && cargo dev docs refresh 2>&1 && cargo dev docs check 2>&1\", goal_gate=true, retry_target=\"fixup\"]\n fixup [label=\"Fixup\", prompt=\"The verify step failed. Read the build output from context and fix all clippy lint warnings, test failures, and generated docs errors.\", max_visits=3]\n fmt [label=\"Format\", shape=parallelogram, script=\"cargo +nightly-2026-04-14 fmt --all 2>&1\", max_retries=0]\n\n start -> toolchain\n toolchain -> preflight_compile [condition=\"outcome=succeeded\"]\n toolchain -> exit\n preflight_compile -> preflight_lint [condition=\"outcome=succeeded\"]\n preflight_compile -> exit\n preflight_lint -> implement [condition=\"outcome=succeeded\"]\n preflight_lint -> fix_lints\n fix_lints -> preflight_lint\n implement -> simplify_opus -> simplify_gpt -> verify\n verify -> fmt [condition=\"outcome=succeeded\"]\n verify -> fixup\n fixup -> verify\n fmt -> exit\n}\n", + "start": null, + "status": { + "kind": "starting" + }, + "status_updated_at": "2026-05-04T17:51:04.111847Z", + "pending_control": null, + "checkpoint": null, + "checkpoints": [], + "conclusion": null, + "retro": null, + "retro_prompt": null, + "retro_response": null, + "sandbox": { + "provider": "daytona", + "working_directory": "/home/daytona/workspace", + "identifier": "fabro-01KQT1TWWJYWZGDT8F05E29H9D", + "repo_cloned": true, + "clone_origin_url": "https://github.com/fabro-sh/fabro", + "clone_branch": "main" + }, + "final_patch": null, + "pull_request": null, + "superseded_by": null, + "pending_interviews": {}, + "stages": {} +} \ No newline at end of file