diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 3e2361a111d..c291eb6d01b 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -144,6 +144,28 @@ pub struct AcpClient { observer_agent_index: Option, /// Best-effort context attached to raw ACP wire events. observer_context: ObserverContext, + /// Goose-specific: most recently observed `_meta.goose.activeRunId` from + /// a `session/update` notification of kind `session_info_update`. + /// + /// Goose emits this whenever it starts or clears an active prompt run + /// (`crates/goose/src/acp/server.rs:2277` `send_active_run_update`). + /// Required as `expectedRunId` when calling the non-standard + /// `_goose/unstable/session/steer` method to inject a message into an + /// in-flight turn without cancelling it. + /// + /// `None` until the first `session_info_update` arrives, or after the + /// run clears (goose emits `activeRunId: null` at end of turn). Other + /// agents will simply never populate this — readers must treat `None` + /// as "no active run to steer into" and fall back to cancel+merge. + active_run_id: Option, + /// Per-turn channel for receiving goose-native non-cancelling steer + /// requests from the main loop. Installed by + /// [`install_steer_rx`](Self::install_steer_rx) at dispatch and + /// consumed (via `take()`) by `session_prompt_with_idle_timeout` so it + /// is dropped at scope exit alongside the turn it served. `None` + /// outside of a goose-native turn — the read loop's steer arm is + /// disabled in that case. + steer_rx: Option>, } impl AcpClient { @@ -233,6 +255,8 @@ impl AcpClient { observer: None, observer_agent_index: None, observer_context: ObserverContext::default(), + active_run_id: None, + steer_rx: None, }) } @@ -421,7 +445,7 @@ impl AcpClient { } let result = self - .read_until_response_with_idle_timeout(id, idle_timeout, hard_deadline) + .read_until_response_with_idle_timeout(session_id, id, idle_timeout, hard_deadline) .await; // On timeout errors, leave current_hard_deadline set so cancel_with_cleanup @@ -463,6 +487,41 @@ impl AcpClient { self.last_prompt_id.is_some() } + /// Most recently observed goose `_meta.goose.activeRunId` from a + /// `session_info_update`, if any. + /// + /// Goose-only: other agents leave this `None` for the lifetime of the + /// client. Read directly by `read_until_response_with_idle_timeout`'s + /// steer arm at write time (see [`crate::pool::SteerRequest`] for + /// why the read loop owns this); production callers do not need this + /// accessor. Kept as `pub` so tests can introspect the field. + #[cfg_attr(not(test), allow(dead_code))] + pub fn active_run_id(&self) -> Option<&str> { + self.active_run_id.as_deref() + } + + /// Install a per-turn steer request channel for goose-native + /// non-cancelling mid-turn delivery. + /// + /// Called by the dispatch path immediately before + /// [`session_prompt_with_idle_timeout`] for all prompt tasks. + /// The matching `Sender` is stored in `TaskMeta.steer_tx` for the + /// main loop's mode-gate fork to drive. + /// + /// Panics if a receiver is already installed — there is exactly one + /// turn per `AcpClient` at a time, and stacking receivers would + /// silently misroute steer requests across turns. The previous + /// turn's receiver must have been consumed by the read loop and + /// dropped at scope exit before the next turn dispatches. + pub fn install_steer_rx(&mut self, rx: tokio::sync::mpsc::Receiver) { + assert!( + self.steer_rx.is_none(), + "install_steer_rx: previous turn's receiver was not consumed — \ + stacking receivers would misroute steer requests across turns" + ); + self.steer_rx = Some(rx); + } + /// Cancel a turn cleanly, handling any pending permission request first. /// /// Steps: @@ -557,7 +616,12 @@ impl AcpClient { // but ignore cancellation. let cleanup_idle = std::time::Duration::from_secs(30); let result = self - .read_until_response_with_idle_timeout(prompt_id, cleanup_idle, hard_deadline) + .read_until_response_with_idle_timeout( + session_id, + prompt_id, + cleanup_idle, + hard_deadline, + ) .await?; self.parse_stop_reason(&result) } @@ -786,14 +850,54 @@ impl AcpClient { /// `hard_deadline` is an absolute `Instant` (pre-computed by the caller) so /// that `cancel_with_cleanup` can inherit the remaining budget from the /// original turn rather than starting a fresh timer. + /// Read agent messages until the response with `expected_id` arrives, or + /// either of two timeouts fires. Returns `Result`. + /// + /// - `idle_timeout`: silent-agent guard, **reset on every line of valid + /// JSON** (and explicitly on `session/update` notifications). + /// - `hard_deadline`: absolute wall-clock cap on the whole call, passed + /// in so that `cancel_with_cleanup` can inherit the remaining budget + /// from the original turn rather than starting a fresh timer. + /// + /// While reading, the loop interleaves goose-native non-cancelling steer + /// requests via `tokio::select!`. The select uses `biased` for + /// reader-first throughput, with a pre-select deadline check at the top + /// of every loop iteration so a continuously-ready reader arm cannot + /// starve the hard deadline (Max's review gate). The steer arm is + /// guarded by `pending_steer.is_none()` so at most one steer is in + /// flight at a time; a successful steer response is routed to the + /// caller's oneshot ack instead of being returned as the prompt result. + /// + /// `session_id` is threaded in lexically by callers so the goose-native + /// steer arm can complete `sessionId` in the steer JSON-RPC params at + /// write time without needing access to outer state. See + /// [`crate::pool::SteerRequest`] for why params are built here and not + /// in the main loop. async fn read_until_response_with_idle_timeout( &mut self, + session_id: &str, expected_id: u64, idle_timeout: std::time::Duration, hard_deadline: tokio::time::Instant, ) -> Result { use tokio::time::Instant; + // Take the per-turn steer receiver into a local so it can be + // borrowed independently of `self.reader` inside `select!`. + // Dropped at scope exit (return paths drain `pending_steer` first + // so the ack_tx oneshot is never leaked silently). + let mut steer_rx = self.steer_rx.take(); + + // Tracks the in-flight steer write: `(request_id, ack_tx)`. While + // `Some`, the steer arm is gated off so we don't stack writes, + // and a response matching `id` is routed to the ack_tx instead + // of being treated as the prompt result. Drained on every return + // path with `PromptCompletedNeutral` so callers are never left + // hanging. + let mut pending_steer: Option<(u64, tokio::sync::oneshot::Sender)> = + None; + let mut idle_deadline = Instant::now() + idle_timeout; loop { @@ -805,23 +909,157 @@ impl AcpClient { } else { hard_deadline }; - let remaining = next_deadline.saturating_duration_since(Instant::now()); + + // Pre-select deadline check — required by Max's review. Under + // `biased`, a continuously-ready reader arm wins every poll and + // `sleep_until(next_deadline)` is never reached, silently + // defeating the hard-deadline guarantee for agents that keep + // producing output (see `acp.rs:608` for why the hard deadline + // exists). Check the classified deadline here so a steady- + // stream agent is still bounded. + if Instant::now() >= next_deadline { + if let Some((_, ack_tx)) = pending_steer.take() { + // Prompt is timing out — release the withheld event via + // PromptCompletedNeutral (no fallback signal: there is + // no in-flight turn to signal once we return, and + // normal dispatch handles redelivery). + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } + if idle_fires_first { + tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); + return Err(AcpError::IdleTimeout(idle_timeout)); + } else { + tracing::warn!("hard turn timeout exceeded"); + return Err(AcpError::HardTimeout); + } + } // LinesCodec::new_with_max_length enforces MAX_LINE_SIZE at the // read level — the buffer never grows beyond the limit. - let read_result = tokio::time::timeout(remaining, self.reader.next()).await; + let read_result = tokio::select! { + biased; + read_result = self.reader.next() => Some(read_result), + // Steer arm: gated off whenever a steer write is already in + // flight so we don't stack two writes against the same + // process. The `async { steer_rx.as_mut()?.recv().await }` + // wrapper produces `None` when no receiver is installed, + // which mismatches the `Some(req)` pattern and disables the + // branch for that iteration (no busy loop). Cancel-safe: + // `mpsc::Receiver::recv` does not lose messages on drop. + Some(req) = async { + match steer_rx.as_mut() { + Some(rx) => rx.recv().await, + None => None, + } + }, if pending_steer.is_none() => { + // Selected: build steer params at write time using the + // lexical `session_id` and the freshest `active_run_id`. + // + // `active_run_id` is updated by `session/update` + // notifications inside this very loop; reading it here + // (rather than snapshotting at dispatch) guarantees the + // value matches what goose's run-id check will compare + // against. If it's `None`, no `session/update` has + // arrived yet so we cannot form a valid `expectedRunId` + // — ack `ExpectedRunIdMissing` and drop the request + // without writing anything. The main loop maps this to + // the universal cancel+merge `Steer` fallback. + match self.active_run_id.clone() { + None => { + tracing::warn!( + "goose-native steer: no active_run_id at write time \ + (no session/update seen yet) — falling back to cancel+merge" + ); + let _ = req.ack_tx.send(crate::pool::SteerAck::Err( + crate::pool::SteerError::ExpectedRunIdMissing, + )); + } + Some(run_id) => { + let id = self.next_id; + self.next_id += 1; + let prompt_block_refs: Vec<&str> = + req.prompt_blocks.iter().map(String::as_str).collect(); + let params = + build_steer_params(session_id, &run_id, &prompt_block_refs); + let msg = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "method": "_goose/unstable/session/steer", + "params": params, + }); + tracing::debug!( + target: "acp::wire", + "→ {}", + serde_json::to_string(&msg).unwrap_or_default() + ); + match self.write_ndjson(&msg).await { + Ok(()) => { + pending_steer = Some((id, req.ack_tx)); + } + Err(e) => { + tracing::warn!( + "goose-native steer write failed: {e} — releasing withheld event" + ); + let _ = req.ack_tx.send(crate::pool::SteerAck::Err( + crate::pool::SteerError::Transport(e.to_string()), + )); + } + } + } + } + // Loop back to the next iteration without consuming a + // reader line; we'll wait for either the prompt + // response or the steer response next. + None + } + _ = tokio::time::sleep_until(next_deadline) => { + // The pre-select check at the top of the next iteration + // would catch this anyway, but firing the deadline arm + // here makes the wakeup immediate (no extra reader poll + // round-trip when stdout is idle). + if let Some((_, ack_tx)) = pending_steer.take() { + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } + if idle_fires_first { + tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); + return Err(AcpError::IdleTimeout(idle_timeout)); + } else { + tracing::warn!("hard turn timeout exceeded"); + return Err(AcpError::HardTimeout); + } + } + }; + + // Steer arm fired (or the select selected nothing read-side this + // iteration): no reader frame to process, loop to re-evaluate + // deadlines and arm the next select. + let read_result = match read_result { + Some(r) => r, + None => continue, + }; match read_result { - Ok(None) => return Err(AcpError::AgentExited), - Ok(Some(Err(LinesCodecError::MaxLineLengthExceeded))) => { + None => { + if let Some((_, ack_tx)) = pending_steer.take() { + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } + return Err(AcpError::AgentExited); + } + Some(Err(LinesCodecError::MaxLineLengthExceeded)) => { + if let Some((_, ack_tx)) = pending_steer.take() { + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } return Err(AcpError::Protocol( "agent stdout line exceeded 10MB limit".into(), )); } - Ok(Some(Err(e))) => { + Some(Err(e)) => { + if let Some((_, ack_tx)) = pending_steer.take() { + let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } return Err(AcpError::Io(std::io::Error::other(e))); } - Ok(Some(Ok(line))) => { + Some(Ok(line)) => { let trimmed = line.trim(); if trimmed.is_empty() { continue; @@ -852,14 +1090,50 @@ impl AcpClient { // Malformed lines (skipped above) don't count as real agent activity. idle_deadline = Instant::now() + idle_timeout; - // Check for matching response (has matching id AND no `method` - // field — a `method` field means agent-initiated request, not response). + // Steer response routing must come BEFORE the prompt + // response check: a steer response is a regular + // JSON-RPC response (id + result/error, no method), + // so the matcher must disambiguate by id. Both checks + // share the `no method` guard. if let Some(id) = msg.get("id") { - if *id == serde_json::json!(expected_id) && msg.get("method").is_none() { - if let Some(error) = msg.get("error") { - return Err(AcpError::AgentError(error.to_string())); + if msg.get("method").is_none() { + if let Some((steer_id, _)) = pending_steer.as_ref() { + if *id == serde_json::json!(*steer_id) { + // Take the ack_tx out and route the + // response. We do not return — keep + // reading until the prompt response + // arrives. + let (_, ack_tx) = pending_steer.take().expect("just checked"); + let ack = if let Some(error) = msg.get("error") { + let code = error + .get("code") + .and_then(|c| c.as_i64()) + .unwrap_or(-1); + let message = error.to_string(); + crate::pool::SteerAck::Err( + crate::pool::SteerError::AgentError { code, message }, + ) + } else { + crate::pool::SteerAck::Success + }; + let _ = ack_tx.send(ack); + continue; + } + } + if *id == serde_json::json!(expected_id) { + if let Some(error) = msg.get("error") { + if let Some((_, ack_tx)) = pending_steer.take() { + let _ = ack_tx + .send(crate::pool::SteerAck::PromptCompletedNeutral); + } + return Err(AcpError::AgentError(error.to_string())); + } + if let Some((_, ack_tx)) = pending_steer.take() { + let _ = + ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral); + } + return Ok(msg["result"].clone()); } - return Ok(msg["result"].clone()); } } @@ -897,17 +1171,6 @@ impl AcpClient { } } } - Err(_elapsed) => { - // Classification was determined before sleeping — not - // affected by scheduler jitter between deadline and wakeup. - if idle_fires_first { - tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity"); - return Err(AcpError::IdleTimeout(idle_timeout)); - } else { - tracing::warn!("hard turn timeout exceeded"); - return Err(AcpError::HardTimeout); - } - } } } } @@ -918,7 +1181,12 @@ impl AcpClient { /// Returns `true` if the update indicates a tool call started, signaling that /// the idle clock should be explicitly reset (the agent will be silent while /// the tool executes). - fn handle_session_update(&self, msg: &serde_json::Value) -> bool { + /// + /// Takes `&mut self` (not `&self`) because some updates carry agent state + /// the client must observe — notably goose's `session_info_update` with + /// `_meta.goose.activeRunId`, which seeds [`active_run_id`](Self::active_run_id) + /// so callers can target `_goose/unstable/session/steer` at the correct run. + fn handle_session_update(&mut self, msg: &serde_json::Value) -> bool { let update = &msg["params"]["update"]; let update_type = update .get("sessionUpdate") @@ -978,6 +1246,42 @@ impl AcpClient { ); false } + "session_info_update" => { + // Goose-only as of writing: `_meta.goose.activeRunId` carries + // the id of the currently-active prompt run, or `null` when + // the run has cleared. Other agents don't emit this field; + // for them `active_run_id` stays `None` and steer callers + // will fall back to cancel+merge. + // + // Per the ACP `SessionInfoUpdate` schema, `_meta` is a field + // on the update object itself — nested inside `update`, not + // alongside it at the params level. Goose and buzz-agent both + // emit it at `params.update._meta.goose.activeRunId`. + let meta = msg["params"]["update"] + .get("_meta") + .and_then(|m| m.get("goose")); + if let Some(goose_meta) = meta { + match goose_meta.get("activeRunId") { + Some(serde_json::Value::String(run_id)) => { + tracing::debug!( + target: "acp::update", + "session_info_update: activeRunId={run_id}" + ); + self.active_run_id = Some(run_id.clone()); + } + Some(serde_json::Value::Null) => { + tracing::debug!( + target: "acp::update", + "session_info_update: activeRunId cleared" + ); + self.active_run_id = None; + } + // Missing or non-string/null — leave state untouched. + _ => {} + } + } + false + } "keepalive" => false, other => { tracing::debug!(target: "acp::update", "session/update: {other}"); @@ -1094,6 +1398,34 @@ fn build_prompt_params(session_id: &str, prompt_blocks: &[&str]) -> serde_json:: }) } +/// Build `_goose/unstable/session/steer` params from one or more text +/// content blocks plus the freshest `expectedRunId`. +/// +/// Wire shape: +/// ```json +/// { "sessionId": "...", "expectedRunId": "...", "prompt": [{"type":"text","text":"..."}, ...] } +/// ``` +/// +/// Called from the read-loop steer arm at write time so `expectedRunId` +/// matches goose's *current* run (it advances on each `session/update`). +/// See [`crate::pool::SteerRequest`] for why this is the read loop's job +/// and not the main loop's. +fn build_steer_params( + session_id: &str, + expected_run_id: &str, + prompt_blocks: &[&str], +) -> serde_json::Value { + let blocks: Vec = prompt_blocks + .iter() + .map(|text| serde_json::json!({ "type": "text", "text": text })) + .collect(); + serde_json::json!({ + "sessionId": session_id, + "expectedRunId": expected_run_id, + "prompt": blocks, + }) +} + /// Build a JSON-RPC permission response with `outcome: "selected"`. fn permission_response_selected(id: &serde_json::Value, option_id: &str) -> serde_json::Value { serde_json::json!({ @@ -1787,6 +2119,7 @@ mod tests { let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30); let result = client .read_until_response_with_idle_timeout( + "test", 999, std::time::Duration::from_millis(100), hard_deadline, @@ -1805,6 +2138,7 @@ mod tests { tokio::time::sleep(std::time::Duration::from_millis(5)).await; let result = client .read_until_response_with_idle_timeout( + "test", 999, std::time::Duration::from_secs(60), hard_deadline, @@ -1828,6 +2162,7 @@ mod tests { let start = std::time::Instant::now(); let result = client .read_until_response_with_idle_timeout( + "test", 999, std::time::Duration::from_millis(200), hard_deadline, @@ -1848,6 +2183,7 @@ mod tests { let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); let result = client .read_until_response_with_idle_timeout( + "test", 42, std::time::Duration::from_secs(2), hard_deadline, @@ -1864,6 +2200,7 @@ mod tests { let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); let result = client .read_until_response_with_idle_timeout( + "test", 999, std::time::Duration::from_secs(2), hard_deadline, @@ -1891,6 +2228,7 @@ mod tests { let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); let result = client .read_until_response_with_idle_timeout( + "test", 0, std::time::Duration::from_secs(3), hard_deadline, @@ -1906,7 +2244,7 @@ mod tests { let idle = std::time::Duration::from_millis(100); let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); let result = client - .read_until_response_with_idle_timeout(999, idle, hard_deadline) + .read_until_response_with_idle_timeout("test", 999, idle, hard_deadline) .await; assert!( matches!(result, Err(AcpError::IdleTimeout(_))), @@ -1914,6 +2252,61 @@ mod tests { ); } + /// Hard-deadline starvation regression (Max's review gate, Eva's required test). + /// + /// When the read-loop became a `tokio::select!` with `biased; reader → + /// steer → sleep_until`, a continuously-ready reader arm could win every + /// poll and starve the timer arm — silently defeating the hard-deadline + /// guarantee. The fix is a pre-select deadline check at the top of every + /// loop iteration; this test pins that behavior. + /// + /// Setup: agent emits a **gapless** stream of valid JSON `session/update` + /// notifications (no `sleep` between lines) so the reader arm is + /// continuously ready. Each line is valid JSON, so it resets the idle + /// clock — and we set idle ≫ hard so idle cannot fire first. With + /// `biased; reader → steer → sleep_until`, the reader arm would win + /// every poll and `sleep_until` would never be reached. Only the + /// pre-select deadline check at the top of the loop can stop us. + /// + /// Without the pre-select check, this test hangs against the infinite + /// bash subprocess until the test harness's own outer timeout, and the + /// returned error would never be `HardTimeout`. + #[tokio::test] + async fn hard_deadline_fires_under_continuous_valid_json_stream() { + // Truly infinite, gapless stream of valid JSON. No `sleep` between + // echoes — the reader arm is continuously ready, which is the + // exact starvation scenario the pre-select check guards against. + // `while :; do echo ...; done` (not a fixed-count `for`) so the + // subprocess never naturally exits before the hard deadline, + // regardless of how fast the host drains bash output. Without + // this, fast hardware drains a bounded loop in < hard_deadline + // and the reader hits EOF (`AgentExited`) before the timer fires, + // masking whether the pre-select check actually works. + let mut client = spawn_script( + r#"while :; do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"agent_message_chunk","content":{"text":"x"}}}}'; done"#, + ) + .await; + let hard = std::time::Duration::from_millis(300); + let hard_deadline = tokio::time::Instant::now() + hard; + let idle = std::time::Duration::from_secs(60); // idle ≫ hard + let start = std::time::Instant::now(); + let result = client + .read_until_response_with_idle_timeout("test", 999, idle, hard_deadline) + .await; + let elapsed = start.elapsed(); + assert!( + matches!(result, Err(AcpError::HardTimeout)), + "expected HardTimeout under gapless valid-JSON stream, got {result:?} (elapsed {elapsed:?})" + ); + // Must fire close to the hard deadline, not late. Without the + // pre-select check the reader arm starves sleep_until and elapsed + // tracks the bash subprocess lifetime instead. + assert!( + elapsed < std::time::Duration::from_secs(2), + "HardTimeout fired late ({elapsed:?}); reader arm may be starving sleep_until" + ); + } + /// Same as `agent_request_with_matching_id_not_consumed_as_response` but /// exercises the non-idle `read_until_response` path (via `send_request`). #[tokio::test] @@ -1957,6 +2350,7 @@ mod tests { let start = std::time::Instant::now(); let result = client .read_until_response_with_idle_timeout( + "test", 999, std::time::Duration::from_millis(100), hard_deadline, @@ -1990,6 +2384,7 @@ mod tests { let start = std::time::Instant::now(); let result = client .read_until_response_with_idle_timeout( + "test", 999, std::time::Duration::from_millis(200), hard_deadline, @@ -2068,4 +2463,254 @@ mod tests { "systemPrompt should NOT be in params when value is None" ); } + + // ── Goose-native steer scaffold (PR follow-up to #1160) ────────────── + + /// Helper: spawn an inert `cat` subprocess so we have a real AcpClient + /// to drive `handle_session_update` against. `cat` never writes back, + /// which is fine — these tests don't read from the agent, they just + /// feed JSON into the parser. + async fn spawn_inert_client() -> AcpClient { + AcpClient::spawn("cat", &[], &[]) + .await + .expect("spawn cat as inert client") + } + + /// Build a `session/update` JSON-RPC notification carrying a + /// `session_info_update` with the given `_meta.goose.activeRunId` value. + /// Pass `None` to omit the `activeRunId` field entirely. + /// + /// `_meta` is nested inside the `update` object (per the ACP + /// `SessionInfoUpdate` schema), matching what goose and buzz-agent + /// emit on the wire. + fn session_info_update_msg(active_run_id: Option) -> serde_json::Value { + let mut goose = serde_json::Map::new(); + if let Some(v) = active_run_id { + goose.insert("activeRunId".to_string(), v); + } + let mut meta = serde_json::Map::new(); + meta.insert("goose".to_string(), serde_json::Value::Object(goose)); + serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "session_info_update", + "_meta": serde_json::Value::Object(meta), + }, + } + }) + } + + #[tokio::test] + async fn active_run_id_sets_on_string() { + let mut client = spawn_inert_client().await; + assert!(client.active_run_id().is_none(), "starts as None"); + + let msg = session_info_update_msg(Some(serde_json::json!("run-abc-123"))); + let _ = client.handle_session_update(&msg); + + assert_eq!(client.active_run_id(), Some("run-abc-123")); + } + + #[tokio::test] + async fn active_run_id_clears_on_null() { + let mut client = spawn_inert_client().await; + // Set it first + let set_msg = session_info_update_msg(Some(serde_json::json!("run-xyz"))); + let _ = client.handle_session_update(&set_msg); + assert_eq!(client.active_run_id(), Some("run-xyz")); + + // Then clear with explicit null + let clear_msg = session_info_update_msg(Some(serde_json::Value::Null)); + let _ = client.handle_session_update(&clear_msg); + assert!( + client.active_run_id().is_none(), + "explicit null must clear active_run_id" + ); + } + + #[tokio::test] + async fn active_run_id_untouched_when_missing() { + // Field absent entirely — must NOT clear existing state (only an + // explicit null clears; missing means "no new info this update"). + let mut client = spawn_inert_client().await; + let set_msg = session_info_update_msg(Some(serde_json::json!("run-stable"))); + let _ = client.handle_session_update(&set_msg); + assert_eq!(client.active_run_id(), Some("run-stable")); + + // session_info_update with no activeRunId field — leave state alone. + let missing_msg = session_info_update_msg(None); + let _ = client.handle_session_update(&missing_msg); + assert_eq!( + client.active_run_id(), + Some("run-stable"), + "missing activeRunId must leave state untouched" + ); + } + + #[tokio::test] + async fn active_run_id_untouched_on_wrong_type() { + // A number or object in activeRunId is malformed — neither set nor clear. + let mut client = spawn_inert_client().await; + let set_msg = session_info_update_msg(Some(serde_json::json!("run-stable"))); + let _ = client.handle_session_update(&set_msg); + assert_eq!(client.active_run_id(), Some("run-stable")); + + let wrong_type_msg = session_info_update_msg(Some(serde_json::json!(42))); + let _ = client.handle_session_update(&wrong_type_msg); + assert_eq!( + client.active_run_id(), + Some("run-stable"), + "non-string/non-null activeRunId must leave state untouched" + ); + } + + // ── Goose-native steer arm tests ────────────────────────────────────── + // + // These exercise the seam between `install_steer_rx` and the read + // loop's steer arm, isolated from `AgentPool` / `EventQueue` / + // dispatch. They prove the locked Option-X contract at the read-loop + // boundary: + // 1. With `active_run_id == None`, the steer arm acks + // `Err(ExpectedRunIdMissing)` and writes nothing — the main + // loop's "Err-before-pending" fallback path is reachable. + // 2. With `active_run_id` set, the steer arm writes the JSON-RPC + // request with the matching `expectedRunId` and routes the + // response to the ack oneshot as `Success`. + // + // We don't test the full mode-gate fork here — that lives in lib.rs + // and is covered by goose e2e (Eva's lane). + + /// Steer with no `active_run_id` set acks `ExpectedRunIdMissing` + /// without writing anything. The read loop continues normally and + /// eventually hits the idle timeout (which is fine — we just need to + /// observe the ack). + #[tokio::test] + async fn native_steer_with_no_active_run_id_acks_expected_run_id_missing() { + // Quiet process: never emits anything, so the read loop has only + // the steer arm and the idle timeout to consider. + let mut client = spawn_script("sleep 10").await; + assert!( + client.active_run_id().is_none(), + "precondition: active_run_id starts as None" + ); + + let (steer_tx, steer_rx) = tokio::sync::mpsc::channel::(1); + client.install_steer_rx(steer_rx); + + // Fire-and-forget: send a SteerRequest from a separate task so + // the read loop picks it up via the select! arm. + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); + let send_task = tokio::spawn(async move { + steer_tx + .send(crate::pool::SteerRequest { + prompt_blocks: vec!["test steer body".into()], + ack_tx, + }) + .await + .expect("steer_tx send should succeed"); + }); + + // Drive the read loop with short idle timeout so the test + // doesn't hang. The expected_id is intentionally never going to + // be matched (the script writes nothing); the read loop will + // exit via IdleTimeout shortly after the steer arm fires. + let idle = std::time::Duration::from_millis(500); + let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + let read_result = client + .read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline) + .await; + send_task.await.expect("send_task should complete"); + + // Read loop exit shape: IdleTimeout (no agent activity). + assert!( + matches!(read_result, Err(AcpError::IdleTimeout(_))), + "expected IdleTimeout once steer was acked + script stayed silent, got {read_result:?}" + ); + + // Ack must be ExpectedRunIdMissing — the steer arm bailed out + // without writing because active_run_id was None at write time. + let ack = ack_rx + .await + .expect("ack oneshot must have received a SteerAck"); + match ack { + crate::pool::SteerAck::Err(crate::pool::SteerError::ExpectedRunIdMissing) => {} + other => panic!("expected SteerAck::Err(ExpectedRunIdMissing), got {other:?}"), + } + } + + /// Steer with `active_run_id` set writes the JSON-RPC request and + /// routes the matching response to the ack oneshot as `Success`. + /// Verifies the wire shape (`sessionId` + `expectedRunId` + `prompt`) + /// indirectly: the bash script emits a response keyed by the steer + /// id (0), and `Success` only fires if the read loop matched that + /// id to its `pending_steer` entry. + #[tokio::test] + async fn native_steer_with_active_run_id_routes_response_to_ack() { + // Script: pause briefly so the test task can install the steer + // and we can be sure the response doesn't race ahead of the + // write — then emit the steer response (id=0 because next_id + // starts at 0 and the steer is the first request the read loop + // writes), then idle. This is a JSON-RPC success response with + // a `stopReason` payload (matching the shape goose uses for + // steer responses in fake_llm.rs). + let script = "sleep 0.5; \ + echo '{\"jsonrpc\":\"2.0\",\"id\":0,\"result\":{\"stopReason\":\"end_turn\"}}'; \ + sleep 10"; + let mut client = spawn_script(script).await; + + // Set active_run_id via a synthesized session_info_update so the + // steer arm has a non-None value to read at write time. + let update = session_info_update_msg(Some(serde_json::json!("run-42"))); + let _ = client.handle_session_update(&update); + assert_eq!(client.active_run_id(), Some("run-42")); + + let (steer_tx, steer_rx) = tokio::sync::mpsc::channel::(1); + client.install_steer_rx(steer_rx); + + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); + let send_task = tokio::spawn(async move { + steer_tx + .send(crate::pool::SteerRequest { + prompt_blocks: vec!["test steer body".into()], + ack_tx, + }) + .await + .expect("steer_tx send should succeed"); + }); + + // Drive the read loop. Expected_id 999 will never be emitted by + // the script so the read loop exits via idle timeout after the + // steer response is routed to ack. + let idle = std::time::Duration::from_secs(2); + let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + let read_result = client + .read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline) + .await; + send_task.await.expect("send_task should complete"); + + // Read loop exit: IdleTimeout (no further activity after the + // routed steer response). AgentExited would also be a valid + // exit if the bash script terminated early; either is fine — + // what matters is the ack. + assert!( + matches!( + read_result, + Err(AcpError::IdleTimeout(_)) | Err(AcpError::AgentExited) + ), + "expected IdleTimeout or AgentExited after steer ack, got {read_result:?}" + ); + + // Ack must be Success: the steer response (id=0) was routed to + // pending_steer.ack_tx. + let ack = ack_rx + .await + .expect("ack oneshot must have received a SteerAck"); + match ack { + crate::pool::SteerAck::Success => {} + other => panic!("expected SteerAck::Success, got {other:?}"), + } + } } diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index b60fe4fcc51..88345331db2 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -57,8 +57,16 @@ pub enum MultipleEventHandling { /// Queue new events while a turn is in-flight. Deliver after current turn /// completes. Existing behavior — zero code change in this path. Queue, + /// Cancel the in-flight turn and re-dispatch a merged prompt that frames + /// the new events as a **steering message** — one that arrived while the + /// agent was working, to be woven into the in-progress task rather than + /// treated as a replacement. Fires for any author the inbound author gate + /// admits (owner ∪ allowlist ∪ siblings). This is the default mid-turn + /// delivery path. Requires DedupMode::Queue. + Steer, /// Cancel the in-flight turn and re-dispatch a merged prompt combining - /// the original events with the new ones, for ANY new @mention. + /// the original events with the new ones, framed as a **supersede** (the + /// new request replaces the old), for ANY new @mention. /// Requires DedupMode::Queue. Interrupt, /// Cancel the in-flight turn only when the new @mention is from the agent @@ -290,12 +298,15 @@ pub struct CliArgs { pub dedup: DedupMode, /// How to handle new @mentions while a turn is already in-flight. - /// queue: events wait (default). interrupt: cancel+re-prompt on any mention. - /// owner-interrupt: cancel only for agent owner's mentions. + /// steer (default): cancel+re-prompt, framing the new mention as a message + /// that arrived mid-task — the agent keeps working and weaves it in. + /// queue: events wait until the current turn completes. + /// interrupt: cancel+re-prompt framed as a supersede (new replaces old). + /// owner-interrupt: interrupt only for the agent owner's mentions. #[arg( long, env = "BUZZ_ACP_MULTIPLE_EVENT_HANDLING", - default_value = "queue", + default_value = "steer", value_enum )] pub multiple_event_handling: MultipleEventHandling, @@ -503,6 +514,33 @@ fn validate_allowlist(entries: &[String]) -> Result, ConfigError Ok(validated) } +/// Validate the `--multiple-event-handling` / `--dedup` combination. +/// +/// Every mid-turn cancel mode (`Steer`, `Interrupt`, `OwnerInterrupt`) requires +/// `DedupMode::Queue`: `DedupMode::Drop` discards events during the cancel drain +/// window, which would produce incomplete merged prompts. `Queue` handling +/// imposes no constraint. +fn validate_multiple_event_handling( + handling: MultipleEventHandling, + dedup: DedupMode, +) -> Result<(), ConfigError> { + let is_cancel_mode = matches!( + handling, + MultipleEventHandling::Steer + | MultipleEventHandling::Interrupt + | MultipleEventHandling::OwnerInterrupt + ); + if is_cancel_mode && matches!(dedup, DedupMode::Drop) { + return Err(ConfigError::ConfigFile( + "--multiple-event-handling=steer (or interrupt/owner-interrupt) requires \ + --dedup=queue. DedupMode::Drop discards events during the cancel drain window, \ + producing incomplete merged prompts." + .into(), + )); + } + Ok(()) +} + fn normalize_agent_command_identity(command: &str) -> String { let normalized = command.trim().replace('\\', "/"); let trimmed = normalized.trim_end_matches('/'); @@ -898,18 +936,7 @@ impl Config { } let model = args.model.or(persona_model); - if matches!( - args.multiple_event_handling, - MultipleEventHandling::Interrupt | MultipleEventHandling::OwnerInterrupt - ) && matches!(args.dedup, DedupMode::Drop) - { - return Err(ConfigError::ConfigFile( - "--multiple-event-handling=interrupt (or owner-interrupt) requires --dedup=queue. \ - DedupMode::Drop discards events during the cancel drain window, \ - producing incomplete merged prompts." - .into(), - )); - } + validate_multiple_event_handling(args.multiple_event_handling, args.dedup)?; let config = Config { keys, @@ -2336,6 +2363,61 @@ channels = "ALL" assert!(result.is_empty()); } + // ── Multiple-event-handling validation + default ────────────────────────── + + #[test] + fn test_multiple_event_handling_default_is_steer() { + // Parse a minimal arg set; the default for --multiple-event-handling + // must be `steer` (steering is the default mid-turn delivery path). + let args = CliArgs::parse_from(["buzz-acp", "--private-key", &"0".repeat(64)]); + assert_eq!(args.multiple_event_handling, MultipleEventHandling::Steer); + // Dedup default must remain `queue` so steering's requirement is met. + assert!(matches!(args.dedup, DedupMode::Queue)); + } + + #[test] + fn test_validate_steer_requires_queue_dedup() { + // Steer + Drop is rejected (drain window would drop events). + let err = validate_multiple_event_handling(MultipleEventHandling::Steer, DedupMode::Drop) + .unwrap_err(); + assert!( + err.to_string().contains("requires"), + "expected a dedup-requirement error, got: {err}" + ); + // Steer + Queue is accepted. + assert!( + validate_multiple_event_handling(MultipleEventHandling::Steer, DedupMode::Queue) + .is_ok() + ); + } + + #[test] + fn test_validate_queue_handling_allows_any_dedup() { + // The non-cancel `Queue` handling imposes no dedup constraint. + assert!( + validate_multiple_event_handling(MultipleEventHandling::Queue, DedupMode::Drop).is_ok() + ); + assert!( + validate_multiple_event_handling(MultipleEventHandling::Queue, DedupMode::Queue) + .is_ok() + ); + } + + #[test] + fn test_validate_interrupt_modes_still_require_queue() { + for mode in [ + MultipleEventHandling::Interrupt, + MultipleEventHandling::OwnerInterrupt, + ] { + assert!( + validate_multiple_event_handling(mode, DedupMode::Drop).is_err(), + "{mode:?} + Drop should be rejected" + ); + } + } + + // ── Idle timeout constant + guard (PR #935) ─────────────────────────────── + #[test] fn default_idle_timeout_is_900_seconds() { // Lock the constant value so accidental changes are caught. diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index e4b68dffe08..80c9e9d1591 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -32,7 +32,7 @@ use pool::{ AgentPool, ControlSignal, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource, SessionState, }; -use queue::{EventQueue, QueuedEvent, ThreadTags}; +use queue::{CancelReason, EventQueue, QueuedEvent, ThreadTags}; use relay::{HarnessRelay, RelayEventPublisher}; use tokio::sync::{mpsc, watch}; use tracing_subscriber::EnvFilter; @@ -879,7 +879,30 @@ fn any_respawn_in_flight(crash_history: &[SlotCircuit]) -> bool { /// Result of a background respawn task. struct RespawnResult { index: usize, - result: Result<(AcpClient, u32)>, + /// Tuple: (initialized client, protocol version, supports_goose_steer). + /// The third element is always `true` — the supervisor uses + /// try-and-tolerate for the steer extension. + result: Result<(AcpClient, u32, bool)>, +} + +/// Outcome of a non-cancelling steer attempt, forwarded from a per-attempt +/// watcher task (which awaits the `SteerRequest.ack_tx` oneshot) back to +/// the main loop's `select!`. The main loop drives queue side-effects from +/// this — it cannot await the oneshot itself without blocking the relay +/// stream. +/// +/// Carries enough identity to operate on the right withheld event in +/// `EventQueue::withheld_native_steer`: `channel_id` is the routing key, +/// `event_id` is the hex id of the single event the steer carried. +struct SteerAckEvent { + channel_id: Uuid, + event_id: String, + /// `Ok` if the read loop sent any of the locked `SteerAck` variants. + /// `Err` if the oneshot was dropped without a send — should not happen + /// under the current read-loop drains, but if it ever does the main + /// loop treats it as `PromptCompletedNeutral` (release withheld, no + /// fallback signal) to avoid leaking the withheld event. + ack: std::result::Result, } /// RAII guard that ensures a `RespawnResult` is sent even if the task panics. @@ -903,7 +926,7 @@ impl RespawnGuard { /// Send the result and disarm the guard. Uses `try_send` (sync) so there /// is no await boundary between marking `sent` and actually enqueueing — /// cancellation cannot slip between the two. - fn send(mut self, result: Result<(AcpClient, u32)>) { + fn send(mut self, result: Result<(AcpClient, u32, bool)>) { // Invariant: try_send succeeds because the channel capacity equals the // slot count, and respawn_in_flight guarantees at most one outstanding // result per slot. If this ever fails, the channel sizing or the @@ -1022,6 +1045,16 @@ async fn tokio_main() -> Result<()> { tracing::info!(agent = i, "agent initialized: {init_result}"); let protocol_version = init_result["protocolVersion"].as_u64().unwrap_or(1) as u32; + tracing::info!( + agent = i, + name = init_result + .get("agentInfo") + .or_else(|| init_result.get("serverInfo")) + .and_then(|info| info.get("name")) + .and_then(|v| v.as_str()) + .unwrap_or("unknown"), + "agent initialized — non-cancelling steer enabled (try-and-tolerate)" + ); acp.observe( "agent_initialized", serde_json::json!({ @@ -1331,6 +1364,19 @@ async fn tokio_main() -> Result<()> { // JoinSet for respawn tasks so shutdown can abort them. let mut respawn_tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new(); + // Channel for non-cancelling steer ack watchers to forward outcomes back + // to the main loop. Each `pool.send_steer(...) == Ok(())` spawns a + // short-lived task that awaits the `SteerRequest.ack_tx` oneshot and + // forwards a `SteerAckEvent`. Unbounded because: + // 1. The producer count is bounded by in-flight goose turns + // (`agents` slots, capacity-1 `steer_tx` each), so the channel + // cannot legitimately back up under steady state. + // 2. We must never drop a steer outcome — losing an ack would leak a + // withheld event in `EventQueue::withheld_native_steer` until + // `IN_FLIGHT_DEADLINE_SECS` expires. + let (steer_ack_tx, mut steer_ack_rx) = mpsc::unbounded_channel::(); + + // ── Step 7: Shutdown signal ─────────────────────────────────────────────── let (shutdown_tx, mut shutdown_rx) = watch::channel(()); let tx = shutdown_tx.clone(); @@ -1402,6 +1448,7 @@ async fn tokio_main() -> Result<()> { enum PoolEvent { Result(Box), Panic(tokio::task::JoinError), + SteerAck(SteerAckEvent), } loop { @@ -1448,7 +1495,7 @@ async fn tokio_main() -> Result<()> { while let Ok(rr) = respawn_rx.try_recv() { crash_history[rr.index].respawn_in_flight = false; match rr.result { - Ok((acp, protocol_version)) => { + Ok((acp, protocol_version, _)) => { let agent = OwnedAgent { index: rr.index, acp, @@ -1496,6 +1543,14 @@ async fn tokio_main() -> Result<()> { Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => { Some(PoolEvent::Panic(e)) } + // Goose-native steer ack from a watcher task. Outcomes drive + // queue side-effects (drop / release withheld event) and + // optionally the cancel+merge fallback signal. See the + // `Some(PoolEvent::SteerAck(...))` match arm below for the + // locked semantics (Eva + Max + Perci). + Some(ack_event) = steer_ack_rx.recv() => { + Some(PoolEvent::SteerAck(ack_event)) + } control_event = async { match relay_observer_control_rx.as_mut() { Some(rx) => rx.recv().await, @@ -1784,6 +1839,18 @@ async fn tokio_main() -> Result<()> { // buzz_event.event (needed for mode gate below). let author_hex = buzz_event.event.pubkey.to_hex(); let event_id_hex = buzz_event.event.id.to_hex(); + // Clone for the non-cancelling steer fork, which + // needs the event to render the steer body. The + // clone is unconditional because we don't know + // yet whether the mode gate will demand a steer + // — checking `multiple_event_handling` here + // would couple the queueing path to the mode + // and break the existing invariant that every + // accepted event goes through `queue.push` + // first. `nostr::Event::clone` is cheap (Arc- + // backed payload) so the cost is negligible. + let event_for_steer = buzz_event.event.clone(); + let prompt_tag_for_steer = prompt_tag.clone(); let accepted = queue.push(QueuedEvent { channel_id: buzz_event.channel_id, event: buzz_event.event, @@ -1797,29 +1864,53 @@ async fn tokio_main() -> Result<()> { // cosmetic stale 👀. Acceptable — see ReactionGuard docs. if accepted { let rc = ctx.rest_client.clone(); + let eid = event_id_hex.clone(); tokio::spawn(async move { - pool::reaction_add(&rc, &event_id_hex, "👀").await; + pool::reaction_add(&rc, &eid, "👀").await; }); } // Event is already queued. If mode requires it AND - // the channel has an in-flight task, fire cancel. + // the channel has an in-flight task, fire cancel — + // OR take the non-cancelling (ACP steer) fork for Steer signals. if accepted && queue.is_channel_in_flight(buzz_event.channel_id) { - let should_cancel = match config.multiple_event_handling { - MultipleEventHandling::Queue => false, - MultipleEventHandling::Interrupt => true, - MultipleEventHandling::OwnerInterrupt => { - match owner_cache.get() { - Some(o) => author_hex == *o, - None => false, - } + // Author eligibility (owner ∪ allowlist ∪ siblings) + // is already enforced by the inbound author gate + // above, so the mid-turn signal fires for every + // event that reaches here. + let signal = mode_gate_signal( + config.multiple_event_handling, + &author_hex, + owner_cache.get(), + ); + if let Some(signal) = signal { + // Try-and-tolerate fork: when the mode + // wants a Steer, attempt the non-cancelling + // path first for any agent. On accept, + // withhold the queued event and spawn an + // ack watcher; the main loop's + // `PoolEvent::SteerAck` arm decides + // success/release/fallback. On reject + // (including `-32601 method_not_found` + // from agents that don't implement the + // extension), fall through to the universal + // cancel+merge `Steer` signal so the event + // still reaches the agent. + let native_attempted = matches!(signal, ControlSignal::Steer) + && try_native_steer( + &mut pool, + &mut queue, + buzz_event.channel_id, + event_for_steer, + prompt_tag_for_steer, + &steer_ack_tx, + ); + if !native_attempted { + signal_in_flight_task( + &mut pool, + buzz_event.channel_id, + signal, + ); } - }; - if should_cancel { - signal_in_flight_task( - &mut pool, - buzz_event.channel_id, - ControlSignal::Interrupt, - ); } } for (channel_id, thread_tags) in @@ -1973,6 +2064,123 @@ async fn tokio_main() -> Result<()> { typing_channels.insert(channel_id, thread_tags); } } + Some(PoolEvent::SteerAck(SteerAckEvent { + channel_id, + event_id, + ack, + })) => { + // Goose-native steer attempt resolved. Locked semantics + // (Eva + Max + Perci, unanimous on Option X): + // + // Success + // The agent received the steer via the non-cancelling + // path. Drop the withheld event so normal dispatch + // never redelivers it. + // + // Err(_) where the write never landed (Transport / + // ExpectedRunIdMissing): + // Delivery state of the underlying message is "never + // attempted on the wire". Release withheld back to the + // queue front AND issue the cancel+merge fallback so + // the message still reaches the agent. + // + // Err(AgentError { code: -32601, .. }) + // The agent returned method_not_found — it does not + // implement the steer extension. Release withheld AND + // fire the cancel+merge fallback so the message still + // reaches the agent via the universal path. + // + // Err(AgentError { code: other, .. }) + // The write landed and the agent returned a JSON-RPC + // error at the application level (e.g. wrong run id). + // The agent's turn is still running (or just completed). + // Release withheld for normal dispatch; do NOT fire the + // fallback signal — the agent already saw the steer + // attempt. If the turn is still running, normal dispatch + // re-delivers when it completes. If the turn already + // ended, there is nothing to cancel. + // + // PromptCompletedNeutral + // The read loop wrote the steer (or was preparing to) + // but the prompt completed before the response landed. + // Delivery state is unknown — but the prompt completing + // means there is no in-flight turn to signal anymore. + // Release withheld for normal dispatch; do NOT fire + // the fallback signal (it would target a turn that + // just ended; normal dispatch already handles + // redelivery via the released queue entry). + // + // Err(PromptCompleted) + // `SteerError::PromptCompleted` is returned synchronously + // by `pool::send_steer` when no task is in flight (handled + // in `try_native_steer`'s Err branch, which falls through + // to cancel+merge). It is never routed through the ack + // channel, so this variant never appears in `SteerAckEvent`. + // + // Watcher Err (oneshot dropped) + // Should not happen — the read loop drains + // pending_steer on every return path. If it does, + // treat as PromptCompletedNeutral to avoid leaking + // the withheld event in `withheld_native_steer`. + let (release_withheld, drop_withheld, signal_fallback) = match &ack { + Ok(pool::SteerAck::Success) => (false, true, false), + // -32601 = method_not_found: agent does not implement the + // steer extension. Fire cancel+merge so the message still + // reaches the agent. + Ok(pool::SteerAck::Err(pool::SteerError::AgentError { code, .. })) + if *code == -32601 => + { + (true, false, true) + } + // AgentError: write landed, agent rejected it at the + // application level (e.g. wrong run id). Release for + // normal dispatch; no fallback signal (the turn is still + // running or just ended — either way there is nothing to + // cancel). + Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => { + (true, false, false) + } + // Transport / ExpectedRunIdMissing: write never landed. + // Release and fire the cancel+merge fallback so the + // message still reaches the agent. + Ok(pool::SteerAck::Err(_)) => (true, false, true), + Ok(pool::SteerAck::PromptCompletedNeutral) => (true, false, false), + Err(_recv_err) => (true, false, false), + }; + tracing::info!( + channel = %channel_id, + event_id = %event_id, + ?ack, + release_withheld, + drop_withheld, + signal_fallback, + "non-cancelling steer ack received" + ); + if drop_withheld { + queue.remove_event(channel_id, &event_id); + } + if release_withheld { + queue.release_native_steer(channel_id, &event_id); + } + if signal_fallback { + // Universal cancel+merge fallback. Note: the + // queued event has already been released to the + // front of `queues[channel_id]`, so the cancel + // will pick it up as part of the merged batch and + // re-prompt the agent. + signal_in_flight_task(&mut pool, channel_id, ControlSignal::Steer); + } + // After releasing a withheld event, give dispatch a chance + // to re-flush. If the prompt is still in flight, the + // channel stays `in_flight_channels` and `flush_next` + // skips it — but a Steer fallback signal sent above will + // tear down the in-flight task; on its completion the + // queue drains. We still try here in case the in-flight + // task has already returned. + for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) { + typing_channels.insert(channel_id, thread_tags); + } + } None => {} // relay/heartbeat/shutdown branches handled inline above } } @@ -2040,7 +2248,7 @@ async fn tokio_main() -> Result<()> { // Drain any respawn results that completed before the abort. Explicitly // shut down returned agents instead of relying on AcpClient::Drop. while let Ok(rr) = respawn_rx.try_recv() { - if let Ok((mut acp, _)) = rr.result { + if let Ok((mut acp, _, _)) = rr.result { acp.shutdown().await; tracing::debug!(agent = rr.index, "reaped respawned agent on shutdown"); } @@ -2101,6 +2309,34 @@ fn is_owner_control_command( && event_mentions_agent(event, agent_pubkey_hex) } +// ── signal_in_flight_task ───────────────────────────────────────────────────── + +/// Decide which [`ControlSignal`] (if any) to send to an in-flight turn when a +/// new, already-author-gated event arrives for that channel. +/// +/// Returns `None` to leave the in-flight turn untouched (the event waits in the +/// queue and is delivered when the turn completes). Author eligibility — owner +/// ∪ allowlist ∪ siblings — is enforced upstream by the inbound author gate, so +/// `Steer`/`Interrupt` apply to every event that reaches this point; only +/// `OwnerInterrupt` re-checks authorship (owner-only) here. +/// +/// `owner` is the resolved owner pubkey hex, if known. +fn mode_gate_signal( + handling: MultipleEventHandling, + author_hex: &str, + owner: Option<&str>, +) -> Option { + match handling { + MultipleEventHandling::Queue => None, + MultipleEventHandling::Steer => Some(ControlSignal::Steer), + MultipleEventHandling::Interrupt => Some(ControlSignal::Interrupt), + MultipleEventHandling::OwnerInterrupt => match owner { + Some(o) if author_hex == o => Some(ControlSignal::Interrupt), + _ => None, + }, + } +} + /// Send a control signal to the in-flight task for `channel_id`. /// Returns `true` if a signal was sent, `false` if no in-flight task was found. fn signal_in_flight_task( @@ -2123,6 +2359,115 @@ fn signal_in_flight_task( false } +/// Attempt the non-cancelling (ACP) steer for a freshly-queued event. +/// +/// Caller invariants: +/// - `event` has already been pushed into `EventQueue::queues[channel_id]` +/// via [`EventQueue::push`] — its `event.id` must still be locatable +/// there so [`EventQueue::mark_native_steer_pending`] can move it to the +/// side table. +/// - `multiple_event_handling` resolved to `ControlSignal::Steer`; this +/// function is the non-cancelling fork of that signal. +/// +/// Returns `true` if the native attempt was accepted by the read loop +/// (capacity-1 mpsc `try_send` succeeded, event withheld synchronously, +/// ack watcher spawned). On `true` the caller MUST NOT issue the +/// universal cancel+merge `ControlSignal::Steer` fallback — the watcher +/// will issue it from the ack arm if the native attempt fails. +/// +/// Returns `false` if `pool.send_steer` failed (no in-flight task, +/// `steer_tx` already full from a prior in-flight steer, or read loop +/// torn down). The caller MUST fall through to +/// `signal_in_flight_task(channel_id, ControlSignal::Steer)` so the +/// event still reaches the agent via the universal path. +/// +/// The withheld event is NOT released here on `false` because no withhold +/// was established: `mark_native_steer_pending` only runs on `Ok(())`. +fn try_native_steer( + pool: &mut AgentPool, + queue: &mut EventQueue, + channel_id: uuid::Uuid, + event: nostr::Event, + prompt_tag: String, + steer_ack_tx: &mpsc::UnboundedSender, +) -> bool { + // Build the steer body: framing strings come from + // `queue::native_steer_framing()` (Eva's drift-proof requirement — + // native and cancel+merge fallback share these so the agent gets the + // same orientation regardless of transport). The single event block + // is rendered by `queue::format_event_block`, the same function + // `queue::format_prompt` uses internally for `[Buzz event: …]` + // sections, so the rendering also cannot drift. + // + // Passing `None` for `channel_info` / `profile_lookup` is intentional: + // native steer is a *delta* into a live turn — the agent already saw + // channel context and the actor's profile in the original prompt, + // duplicating it here would defeat the point of non-cancelling + // steering (which is to inject only what's new). + let (header, closing) = queue::native_steer_framing(); + let event_id_hex = event.id.to_hex(); + let be = queue::BatchEvent { + event, + prompt_tag: prompt_tag.clone(), + received_at: std::time::Instant::now(), + }; + let event_block = queue::format_event_block(channel_id, None, &be, None); + let body = format!("{header}\n\n[Buzz event: {prompt_tag}]\n{event_block}\n\n{closing}"); + + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); + let request = pool::SteerRequest { + prompt_blocks: vec![body], + ack_tx, + }; + + match pool.send_steer(channel_id, request) { + Ok(()) => { + // Withhold the queued event synchronously BEFORE spawning + // the watcher: this closes the race where `mark_complete` + // clears `in_flight_channels` and a stray `flush_next` could + // re-deliver the event via normal dispatch. See + // `EventQueue::mark_native_steer_pending` docs at queue.rs:606. + let withheld = queue.mark_native_steer_pending(channel_id, &event_id_hex); + if !withheld { + // Race: the event was already drained out of the queue + // before we got here (e.g. a concurrent flush picked it + // up). The steer is on the wire; if it succeeds the + // agent gets it via the native path AND normal + // dispatch — duplicate delivery is benign (agent gets + // the same message twice). Log so this is visible if it + // ever happens in production. + tracing::warn!( + channel = %channel_id, + event_id = %event_id_hex, + "native steer accepted by read loop but event was not in queue to withhold \ + — possible duplicate delivery if steer succeeds" + ); + } + let ack_tx_clone = steer_ack_tx.clone(); + let event_id_for_watcher = event_id_hex.clone(); + tokio::spawn(async move { + let ack = ack_rx.await; + let _ = ack_tx_clone.send(SteerAckEvent { + channel_id, + event_id: event_id_for_watcher, + ack, + }); + }); + true + } + Err(e) => { + tracing::info!( + channel = %channel_id, + error = ?e, + "non-cancelling steer not accepted — falling back to cancel+merge" + ); + false + } + } +} + +// ── dispatch_pending ────────────────────────────────────────────────────────── + /// Flush queued work to available agents. fn dispatch_pending( pool: &mut AgentPool, @@ -2142,7 +2487,7 @@ fn dispatch_pending( .map(|event| queue::parse_thread_tags(&event.event)) .unwrap_or_default(); let affinity_hit = pool.has_session_for(channel_id); - let agent = match pool.try_claim(Some(channel_id)) { + let mut agent = match pool.try_claim(Some(channel_id)) { Some(a) => a, None => { let pending = queue.pending_channels(); @@ -2163,6 +2508,19 @@ fn dispatch_pending( let ctx_clone = Arc::clone(ctx); let agent_index = agent.index; + // Goose-native non-cancelling steer seam: snapshot capability before + // the agent moves into `run_prompt_task`, and install the per-turn + // steer receiver on the read loop so the main loop's mode-gate fork + // (see the `if accepted && queue.is_channel_in_flight(...)` block + // in the relay event branch of the main `select!` loop) can drive + // it via the matching sender stored in `TaskMeta.steer_tx`. + // Install the steer channel for every prompt task — the supervisor + // uses try-and-tolerate: it attempts the steer for any agent and + // treats `-32601 method_not_found` as "fall back to cancel+merge". + let (tx, rx) = tokio::sync::mpsc::channel::(1); + agent.acp.install_steer_rx(rx); + let steer_tx = Some(tx); + // Prompt text is now built inside run_prompt_task (needs async for // context fetching). Pass None for prompt_text; batch carries the data. let (control_tx, control_rx) = tokio::sync::oneshot::channel::(); @@ -2186,6 +2544,7 @@ fn dispatch_pending( channel_id: Some(channel_id), recoverable_batch, control_tx: Some(control_tx), + steer_tx, }, ); dispatched_channels.push((channel_id, typing_scope)); @@ -2229,8 +2588,14 @@ fn handle_prompt_result( if matches!(result.outcome, PromptOutcome::Cancelled) { // Cancel re-prompt: store as cancelled events so flush_next() // merges them into the next FlushBatch.cancelled_events, - // enabling the annotated merged-prompt format. - queue.requeue_as_cancelled(batch); + // enabling the annotated merged-prompt format. The batch's + // cancel_reason (set by the pool task per the control signal) + // selects steer vs interrupt framing. It is always set on this + // path; if somehow unset, fall back to the gentler Steer framing + // — consistent with MergeFraming::for_reason(None) and the + // system default — rather than telling the agent to supersede. + let reason = batch.cancel_reason.unwrap_or(CancelReason::Steer); + queue.requeue_as_cancelled(batch, reason); } else { queue.requeue(batch); } @@ -2561,6 +2926,7 @@ fn dispatch_heartbeat( channel_id: None, recoverable_batch: None, control_tx: None, + steer_tx: None, }, ); *heartbeat_in_flight = true; @@ -2643,6 +3009,7 @@ fn spawn_respawn_task( true } +// ── spawn_and_init ──────────────────────────────────────────────────────────── /// Spawn an agent subprocess and run the MCP `initialize` handshake. /// /// Takes owned args so it can run in a background `tokio::spawn` task without @@ -2653,7 +3020,7 @@ async fn spawn_and_init( extra_env: &[(String, String)], agent_index: usize, observer: Option, -) -> Result<(AcpClient, u32)> { +) -> Result<(AcpClient, u32, bool)> { let mut acp = AcpClient::spawn(command, args, extra_env) .await .map_err(|e| anyhow::anyhow!("failed to spawn agent: {e}"))?; @@ -2670,7 +3037,7 @@ async fn spawn_and_init( "initializeResult": init_result, }), ); - Ok((acp, protocol_version)) + Ok((acp, protocol_version, true)) } Err(e) => { // Explicitly shut down the spawned child to prevent zombie/leak. @@ -2947,6 +3314,46 @@ mod owner_control_command_tests { )); } + #[test] + fn mode_gate_signal_maps_handling_to_control_signal() { + let owner = "a".repeat(64); + let other = "b".repeat(64); + + // Queue: never signals — events wait for the turn to finish. + assert!(mode_gate_signal(MultipleEventHandling::Queue, &owner, Some(&owner)).is_none()); + + // Steer: always steers (eligibility already enforced upstream). + assert!(matches!( + mode_gate_signal(MultipleEventHandling::Steer, &other, Some(&owner)), + Some(ControlSignal::Steer) + )); + // Steer even when owner is unknown — gate doesn't re-check authorship. + assert!(matches!( + mode_gate_signal(MultipleEventHandling::Steer, &other, None), + Some(ControlSignal::Steer) + )); + + // Interrupt: always interrupts for any eligible author. + assert!(matches!( + mode_gate_signal(MultipleEventHandling::Interrupt, &other, Some(&owner)), + Some(ControlSignal::Interrupt) + )); + + // OwnerInterrupt: interrupts only for the owner. + assert!(matches!( + mode_gate_signal(MultipleEventHandling::OwnerInterrupt, &owner, Some(&owner)), + Some(ControlSignal::Interrupt) + )); + assert!( + mode_gate_signal(MultipleEventHandling::OwnerInterrupt, &other, Some(&owner)).is_none(), + "owner-interrupt must not fire for a non-owner author" + ); + assert!( + mode_gate_signal(MultipleEventHandling::OwnerInterrupt, &owner, None).is_none(), + "owner-interrupt must not fire when the owner is unknown" + ); + } + #[tokio::test] async fn signal_in_flight_task_sends_rotate_once() { let mut pool = AgentPool::from_slots(vec![]); @@ -2962,6 +3369,7 @@ mod owner_control_command_tests { channel_id: Some(channel_id), recoverable_batch: None, control_tx: Some(control_tx), + steer_tx: None, }, ); @@ -3103,6 +3511,44 @@ mod author_gate_tests { "the owner must always be accepted under Allowlist" ); } + + // The default `respond-to` is OwnerOnly. Under steering, "an ineligible + // author must NOT steer" is enforced *here* — author_allowed drops the + // event before it reaches the mode gate — not in the gate itself. These + // pin that invariant against the default mode. + #[tokio::test] + async fn test_owner_only_rejects_stranger_so_no_steer() { + let cache = cache_with_sibling(); + assert!( + !author_allowed( + &RespondTo::OwnerOnly, + &HashSet::new(), + STRANGER, + &cache, + &dummy_rest_client() + ) + .await, + "under the default OwnerOnly, a stranger must be dropped — so it can never reach the mode gate to steer" + ); + } + + #[tokio::test] + async fn test_owner_only_admits_owner_and_sibling_to_steer() { + let cache = cache_with_sibling(); + for (who, label) in [(OWNER, "owner"), (SIBLING, "sibling")] { + assert!( + author_allowed( + &RespondTo::OwnerOnly, + &HashSet::new(), + who, + &cache, + &dummy_rest_client() + ) + .await, + "under default OwnerOnly, the {label} must be admitted so steering can fire" + ); + } + } } #[cfg(test)] @@ -3445,6 +3891,7 @@ mod error_outcome_emission_tests { channel_id: None, recoverable_batch: None, control_tx: None, + steer_tx: None, }, ); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 523d5cf7455..9a5a872b49a 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -35,8 +35,8 @@ use crate::acp::{ use crate::config::{DedupMode, PermissionMode}; use crate::observer; use crate::queue::{ - ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, PromptProfile, - PromptProfileLookup, + CancelReason, ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, + PromptProfile, PromptProfileLookup, }; use crate::relay::{ChannelInfo, RestClient}; @@ -52,6 +52,13 @@ pub struct TaskMeta { /// Control signal for the in-flight prompt task. /// `None` for heartbeat tasks (not controllable) and after signal is consumed. pub control_tx: Option>, + /// Steer request channel for non-cancelling mid-turn delivery. + /// Capacity-1; `try_send` from the main loop fails on `Full`/`Closed`, + /// in which case the caller must fall back to the universal + /// `ControlSignal::Steer` cancel+merge path. `None` for heartbeat + /// tasks only — all prompt tasks install a steer channel regardless + /// of the agent's name. + pub steer_tx: Option>, } /// Agent-level model capabilities. Populated on first session creation. @@ -185,13 +192,127 @@ fn apply_completed_before_control_signal( pub enum ControlSignal { /// Stop the current turn and drop its triggering batch. Cancel, - /// Stop the current turn and requeue its triggering batch for a merged re-prompt. + /// Stop the current turn and requeue its triggering batch for a merged + /// re-prompt framed as a **supersede**: the new request replaces the old. Interrupt, + /// Stop the current turn and requeue its triggering batch for a merged + /// re-prompt framed as a **steer**: a message arrived while the agent was + /// working; it should continue its work and incorporate the message if + /// relevant, not treat it as a replacement task. This is the default + /// mid-turn delivery path (see [`MultipleEventHandling::Steer`]). + Steer, /// Stop the current turn and drop its triggering batch. The session is /// invalidated just like cancel; the next turn creates a fresh session. Rotate, } +/// Goose-native non-cancelling steer request, sent from the main loop to an +/// in-flight prompt task's read loop via a capacity-1 mpsc channel. +/// +/// The read loop owns the `AcpClient`'s reader/writer for the duration of the +/// turn, so we cannot drive a steer write from the main thread directly. The +/// main loop carries the steer prompt body (already framed by +/// `queue::native_steer_framing()` + `queue::format_event_block`); the read +/// loop completes `sessionId` (lexical) and `expectedRunId` +/// (`AcpClient::active_run_id` at write time) when it actually emits the +/// JSON-RPC request. The main loop awaits a `SteerAck` on the `ack_tx` +/// oneshot. +/// +/// ## Why the read loop fills params, not the main loop +/// +/// `expectedRunId` is a *moving target*: the read loop updates +/// `self.active_run_id` as goose emits `session/update` notifications, and +/// the steer is rejected if the supplied id doesn't match the *current* run. +/// A snapshot taken at dispatch (or at mode-gate time) can be stale by the +/// time the read loop actually writes the steer line. Filling params at +/// write time uses the freshest possible run id and is correct-by- +/// construction on the one field whose freshness the protocol checks. +/// `sessionId` is in lexical scope inside the read loop's caller +/// (`session_prompt_blocks_with_idle_timeout`), so no plumbing is required +/// for that — only a function parameter pass-through. +/// +/// If `active_run_id` is `None` at write time (no `session/update` seen yet +/// — e.g. agents that never emit run-id metadata), the steer cannot form a +/// valid `expectedRunId` and the read loop acks +/// [`SteerError::ExpectedRunIdMissing`]. The main loop maps this to the +/// "Err-before-pending" bucket: no withhold/mark was established at +/// `pool::send_steer` time because the request was rejected before any +/// write, so the watcher only needs to release nothing and fall back to the +/// universal `ControlSignal::Steer` cancel+merge path. +pub struct SteerRequest { + /// Prompt body text blocks. Each entry becomes one `text` content + /// block in `params.prompt`. Built by the main loop via + /// `queue::native_steer_framing()` + `queue::format_event_block` so + /// the wording cannot drift from the cancel+merge fallback path. + pub prompt_blocks: Vec, + /// Oneshot for the read loop to report the outcome. + pub ack_tx: tokio::sync::oneshot::Sender, +} + +/// Why a goose-native steer failed. +/// +/// String and integer fields are intentionally `Debug`-only — read by +/// `tracing` macros in the main loop's `PoolEvent::SteerAck` arm via +/// `?ack`. The dead-code lint can't see that path because it doesn't +/// trace through `Debug` derives, hence the `#[allow]`. +#[allow(dead_code)] +#[derive(Debug)] +pub enum SteerError { + /// The agent returned a JSON-RPC error response to the steer request. + /// + /// `code` is the JSON-RPC error code: + /// - `-32601` (`method_not_found`): the agent does not implement the + /// steer extension. The main loop should fire the cancel+merge + /// fallback so the message still reaches the agent. + /// - Any other code: the write landed and the agent rejected it at the + /// application level (e.g. wrong run id). Release the withheld event + /// for normal dispatch; do NOT fire the fallback — the turn is still + /// running or just ended. + AgentError { code: i64, message: String }, + /// Transport-level failure: write error, read EOF, JSON-RPC framing + /// violation, etc. The string carries the underlying `AcpError`'s display. + Transport(String), + /// At steer-write time `AcpClient::active_run_id` was `None`, so the + /// read loop couldn't form a valid `expectedRunId`. The read loop drops + /// the request without writing anything; the main loop should release + /// any withheld event and fall back to the universal cancel+merge + /// `ControlSignal::Steer` path. This is in the same "Err-before-pending" + /// bucket as `Transport` write failures: no in-process state was + /// established, so no in-process cleanup is needed. + ExpectedRunIdMissing, + /// The read loop never got to dispatch the steer because the prompt + /// completed first. Delivery state for the underlying message is + /// unknown after prompt completion — the main loop must treat this as + /// "release the withheld event so normal dispatch handles it" with no + /// claims that the agent did or did not incorporate it. + /// + /// Returned synchronously by `send_steer` when no task is in flight + /// for the channel. Never sent through the ack channel — the ack + /// watcher is only spawned on `send_steer` success. + PromptCompleted, +} + +/// Outcome of a goose-native steer, sent from the read loop back to the +/// main loop's ack watcher. +#[derive(Debug)] +pub enum SteerAck { + /// The agent returned a successful response to the steer request. + /// The main loop must drop the withheld event (`remove_event`) — it + /// has been delivered via the non-cancelling path. + Success, + /// The steer was attempted but failed. Delivery state for the + /// underlying message is unknown after prompt completion; the main + /// loop must release the withheld event and fall back to the + /// universal `Steer` cancel+merge path so the message still reaches + /// the agent. + Err(SteerError), + /// The prompt completed before the read loop selected the steer arm. + /// Treated as a benign no-op: release the withheld event for normal + /// dispatch. Do not fire the fallback `Steer` signal — there is no + /// in-flight turn to signal, and normal dispatch handles delivery. + PromptCompletedNeutral, +} + /// Outcome of a prompt task. #[allow(dead_code)] pub enum PromptOutcome { @@ -342,6 +463,46 @@ impl AgentPool { &mut self.task_map } + /// Try to send a goose-native steer request to the in-flight task for + /// `channel_id`. + /// + /// Returns `Ok(())` if the request was accepted by the read loop's + /// receiver (capacity-1 mpsc; one slot is the single in-flight steer + /// write). Returns `Err(SteerError::Transport(_))` on `Full`/`Closed` + /// (already-in-flight write, or read loop torn down). Callers must + /// fall back to the universal `ControlSignal::Steer` cancel+merge path + /// on `Err`. + /// + /// This does **not** spawn the ack watcher — the caller owns the + /// oneshot `ack_tx` inside `SteerRequest` and is responsible for + /// awaiting it and applying the locked Success / Err / PromptCompletedNeutral + /// semantics. Caller is also responsible for the synchronous + /// `queue.mark_native_steer_pending(...)` *before* spawning the + /// watcher, to close the result-vs-ack race. + /// + /// Returns `Err(SteerError::PromptCompleted)` if no task is in flight + /// for `channel_id` (the prompt completed between the mode-gate check + /// and this call, or the channel was never in flight). This is + /// semantically a soft no-op — the caller should release any withheld + /// event and let normal dispatch handle delivery. + pub fn send_steer( + &mut self, + channel_id: Uuid, + request: SteerRequest, + ) -> Result<(), SteerError> { + let meta = self + .task_map + .values_mut() + .find(|m| m.channel_id == Some(channel_id)) + .ok_or(SteerError::PromptCompleted)?; + let tx = meta + .steer_tx + .as_ref() + .ok_or_else(|| SteerError::Transport("steer_tx not installed".into()))?; + tx.try_send(request) + .map_err(|e| SteerError::Transport(e.to_string())) + } + pub fn result_tx(&self) -> mpsc::UnboundedSender { self.result_tx.clone() } @@ -1218,10 +1379,8 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); agent.state.invalidate(&source); - let retry_batch = match control_signal { - ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch), - ControlSignal::Cancel | ControlSignal::Rotate => None, - }; + let retry_batch = + requeue_cancelled_batch(&ctx, control_signal, batch); let _ = result_tx.send(PromptResult { agent, source, @@ -1232,10 +1391,8 @@ pub async fn run_prompt_task( } Err(AcpError::AgentExited) => { agent.state.invalidate_all(); - let retry_batch = match control_signal { - ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch), - ControlSignal::Cancel | ControlSignal::Rotate => None, - }; + let retry_batch = + requeue_cancelled_batch(&ctx, control_signal, batch); let _ = result_tx.send(PromptResult { agent, source, @@ -1247,10 +1404,8 @@ pub async fn run_prompt_task( Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout) => { // Cancel drain timed out — agent state uncertain. agent.state.invalidate(&source); - let retry_batch = match control_signal { - ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch), - ControlSignal::Cancel | ControlSignal::Rotate => None, - }; + let retry_batch = + requeue_cancelled_batch(&ctx, control_signal, batch); let _ = result_tx.send(PromptResult { agent, source, @@ -1261,10 +1416,8 @@ pub async fn run_prompt_task( } Err(e) => { agent.state.invalidate(&source); - let retry_batch = match control_signal { - ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch), - ControlSignal::Cancel | ControlSignal::Rotate => None, - }; + let retry_batch = + requeue_cancelled_batch(&ctx, control_signal, batch); let _ = result_tx.send(PromptResult { agent, source, @@ -2035,6 +2188,29 @@ fn requeue_batch_if_queue(ctx: &PromptContext, batch: Option) -> Opt } } +/// Map a cancelling [`ControlSignal`] to the [`CancelReason`] that should frame +/// the merged re-prompt, then requeue the batch (in `Queue` dedup mode) with +/// that reason stamped onto [`FlushBatch::cancel_reason`]. `Cancel`/`Rotate` +/// drop the batch entirely. The reason is consumed by the main loop at requeue +/// time (`requeue_as_cancelled`) and ultimately by `format_prompt`. +#[inline] +fn requeue_cancelled_batch( + ctx: &PromptContext, + signal: ControlSignal, + batch: Option, +) -> Option { + let reason = match signal { + ControlSignal::Steer => CancelReason::Steer, + ControlSignal::Interrupt => CancelReason::Interrupt, + // Cancel/Rotate discard the batch — no merged re-prompt. + ControlSignal::Cancel | ControlSignal::Rotate => return None, + }; + requeue_batch_if_queue(ctx, batch).map(|mut b| { + b.cancel_reason = Some(reason); + b + }) +} + /// Log a stop reason at the appropriate tracing level. fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) { let label = match source { @@ -2761,6 +2937,7 @@ mod tests { received_at: std::time::Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let context = ConversationContext::Thread { messages: vec![ContextMessage { diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index a5da30d43f7..2fb1acfd227 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -56,6 +56,19 @@ pub struct BatchEvent { pub received_at: Instant, } +/// Why a batch's prior turn was cancelled — controls how `format_prompt` +/// frames the merged re-prompt. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CancelReason { + /// A new request should **supersede** the interrupted work + /// (`MultipleEventHandling::Interrupt`). + Interrupt, + /// A message arrived while the agent was working; it should **continue** + /// and incorporate the message if relevant + /// (`MultipleEventHandling::Steer`, the default mid-turn path). + Steer, +} + /// A batch of events to prompt the agent with. #[derive(Debug, Clone)] pub struct FlushBatch { @@ -63,8 +76,14 @@ pub struct FlushBatch { pub events: Vec, /// Events from a cancelled batch that triggered this re-prompt. /// Empty for normal (non-cancel) batches. When non-empty, `format_prompt()` - /// produces a merged prompt with annotated sections. + /// produces a merged prompt with annotated sections, framed per + /// [`cancel_reason`](Self::cancel_reason). pub cancelled_events: Vec, + /// How the prior turn was cancelled, when [`cancelled_events`] is non-empty. + /// `None` for normal (non-merge) batches; falls back to the gentler + /// [`Steer`](CancelReason::Steer) framing if a merge somehow lacks a reason + /// (see [`MergeFraming::for_reason`]). + pub cancel_reason: Option, } /// Per-channel event queue with per-channel in-flight enforcement. @@ -127,6 +146,21 @@ pub struct EventQueue { /// `FlushBatch` for that channel as `cancelled_events` so `format_prompt()` /// can produce annotated "[Previous request — interrupted]" sections. cancelled_batches: HashMap>, + /// Why each channel's cancelled batch was cancelled (steer vs interrupt). + /// Set by `requeue_as_cancelled`, consumed by `flush_next` to set + /// `FlushBatch::cancel_reason`. Keyed by channel, cleared on flush. + cancel_reasons: HashMap, + /// Events withheld from `queues` while a goose-native steer is in flight + /// for that event. Invisible to `flush_next` / `has_flushable_work` / + /// `drain` (the events have been moved out of `queues`), so the queue's + /// no-double-deliver invariant holds without any change to the hot drain + /// path. Populated by [`mark_native_steer_pending`]; drained back to the + /// queue front by [`release_native_steer`] (preserving original + /// `received_at` fairness, same discipline as `requeue_preserve_timestamps` + /// at line 453). Bulk recovery on `IN_FLIGHT_DEADLINE_SECS` expiry is + /// performed by `flush_next` / `has_flushable_work` (recover, not + /// log-and-drop — the events were never delivered to the agent). + withheld_native_steer: HashMap>, } impl EventQueue { @@ -141,6 +175,8 @@ impl EventQueue { retry_counts: HashMap::new(), dedup_mode, cancelled_batches: HashMap::new(), + cancel_reasons: HashMap::new(), + withheld_native_steer: HashMap::new(), } } @@ -201,6 +237,12 @@ impl EventQueue { ); self.in_flight_channels.remove(&id); self.in_flight_deadlines.remove(&id); + // Recover any withheld goose-native steer events for the expired + // channel back to the queue front so normal dispatch delivers + // them. Unlike the in-flight batch above (already delivered to a + // now-hung prompt — nothing to recover), these events were never + // delivered to the agent. + self.recover_withheld_for_expired_channel(id); } // Find the channel whose head event has the oldest received_at, @@ -232,6 +274,7 @@ impl EventQueue { // Move cancelled events into the regular events slot. // No new events to merge — re-dispatch the original batch. let cancelled = self.cancelled_batches.remove(&id).unwrap_or_default(); + let cancel_reason = self.cancel_reasons.remove(&id); self.in_flight_channels.insert(id); self.in_flight_deadlines .insert(id, now + Duration::from_secs(IN_FLIGHT_DEADLINE_SECS)); @@ -240,6 +283,7 @@ impl EventQueue { channel_id: id, events: cancelled, cancelled_events: vec![], + cancel_reason, }); } None => return None, @@ -276,11 +320,18 @@ impl EventQueue { .cancelled_batches .remove(&channel_id) .unwrap_or_default(); + let cancel_reason = if cancelled_events.is_empty() { + self.cancel_reasons.remove(&channel_id); + None + } else { + self.cancel_reasons.remove(&channel_id) + }; Some(FlushBatch { channel_id, events, cancelled_events, + cancel_reason, }) } @@ -435,14 +486,19 @@ impl EventQueue { /// in the next `FlushBatch` for this channel (enabling the annotated /// merged-prompt format in `format_prompt()`). /// + /// `reason` records why the turn was cancelled (steer vs interrupt) so the + /// merged prompt is framed correctly. On a double-cancel, the most recent + /// reason wins. + /// /// Unlike `requeue_preserve_timestamps`, events are NOT pushed back into /// the generic queue — they are stored separately and merged by /// `flush_next()`. No retry throttle, no backoff. - pub fn requeue_as_cancelled(&mut self, batch: FlushBatch) { + pub fn requeue_as_cancelled(&mut self, batch: FlushBatch, reason: CancelReason) { let entry = self.cancelled_batches.entry(batch.channel_id).or_default(); // Preserve any already-cancelled events from a prior cancel (double-cancel). entry.extend(batch.cancelled_events); entry.extend(batch.events); + self.cancel_reasons.insert(batch.channel_id, reason); } /// Returns `true` if any channel has pending events that are not in-flight @@ -472,6 +528,10 @@ impl EventQueue { ); self.in_flight_channels.remove(&id); self.in_flight_deadlines.remove(&id); + // Symmetric with the flush_next expiry block: recover withheld + // goose-native steer events for the expired channel so they are + // not permanently orphaned in the side table. + self.recover_withheld_for_expired_channel(id); } self.queues.iter().any(|(id, q)| { @@ -509,6 +569,8 @@ impl EventQueue { self.retry_after.remove(&channel_id); self.retry_counts.remove(&channel_id); self.cancelled_batches.remove(&channel_id); + self.cancel_reasons.remove(&channel_id); + self.withheld_native_steer.remove(&channel_id); // Preserve in_flight_channels AND in_flight_deadlines: the in-flight // task will eventually complete (calling mark_complete) or the deadline // will expire (auto-cleaning the channel). Removing deadlines without @@ -522,6 +584,148 @@ impl EventQueue { self.in_flight_channels.contains(&channel_id) } + // ── Goose-native steer withhold (side table) ────────────────────────── + // + // While a goose-native `_goose/unstable/session/steer` write is in flight + // for a specific queued event, that event is moved out of `queues` into + // `withheld_native_steer` so `flush_next` / `has_flushable_work` / the + // contiguous drain at line 285 cannot see it — closing the race window + // between `mark_complete` (which clears `in_flight_channels`) and the + // ack arriving on the main loop. On `Success` the event is consumed + // (`remove_event`); on `Err` / `PromptCompletedNeutral` it is released + // back to the queue front (`release_native_steer`), preserving its + // original `received_at` for FIFO fairness. + + /// Move a queued event out of `queues[channel_id]` into the side table + /// to withhold it from `flush_next` while a goose-native steer is in + /// flight. + /// + /// Returns `true` if the event was found and withheld, `false` if the + /// event id was not present in `queues[channel_id]` (race-safe no-op: + /// the event may have already been drained, removed, or never queued). + /// + /// Must be called synchronously from the mode-gate fork immediately + /// after `pool.send_steer` returns `Ok(())` and before any watcher task + /// is spawned, so the withhold is established before `mark_complete` / + /// any subsequent `flush_next` tick can run. + pub fn mark_native_steer_pending(&mut self, channel_id: Uuid, event_id: &str) -> bool { + let Some(q) = self.queues.get_mut(&channel_id) else { + return false; + }; + let Some(pos) = q.iter().position(|qe| qe.event.id.to_hex() == event_id) else { + return false; + }; + let qe = q + .remove(pos) + .expect("position came from iter so remove must succeed"); + if q.is_empty() { + self.queues.remove(&channel_id); + } + self.withheld_native_steer + .entry(channel_id) + .or_default() + .push(qe); + true + } + + /// Release a single withheld event back to the front of + /// `queues[channel_id]`, preserving its original `received_at`. + /// + /// Called on `SteerAck::Err(_)` and `SteerAck::PromptCompletedNeutral` + /// (delivery unknown after prompt completion; restoring queued event + /// for normal dispatch). Idempotent: a no-op if the event was already + /// removed or never withheld. + /// + /// Push-to-front matches the discipline of `requeue_preserve_timestamps` + /// at line 453, preserving fairness across channels. + pub fn release_native_steer(&mut self, channel_id: Uuid, event_id: &str) { + let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) else { + return; + }; + let Some(pos) = entries + .iter() + .position(|qe| qe.event.id.to_hex() == event_id) + else { + return; + }; + let qe = entries.remove(pos); + if entries.is_empty() { + self.withheld_native_steer.remove(&channel_id); + } + // Push to FRONT so original `received_at` keeps the event at the head + // of the channel's queue. Per-channel cap is enforced below in case + // a flood of events arrived during the ack window. + let queue = self.queues.entry(channel_id).or_default(); + queue.push_front(qe); + while queue.len() > MAX_PENDING_PER_CHANNEL { + queue.pop_back(); + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "release_native_steer overflow — dropped newest event to enforce cap" + ); + } + } + + /// Drop a specific event by id from both the side table and the main + /// queue. + /// + /// Called on `SteerAck::Success` — the agent received the steer, so the + /// event has been "delivered" via the non-cancelling path and must not + /// be redelivered via normal dispatch. Idempotent across both stores. + pub fn remove_event(&mut self, channel_id: Uuid, event_id: &str) { + if let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) { + entries.retain(|qe| qe.event.id.to_hex() != event_id); + if entries.is_empty() { + self.withheld_native_steer.remove(&channel_id); + } + } + if let Some(q) = self.queues.get_mut(&channel_id) { + q.retain(|qe| qe.event.id.to_hex() != event_id); + if q.is_empty() { + self.queues.remove(&channel_id); + } + } + } + + /// Bulk-release every withheld event for `channel_id` back to the queue + /// front, preserving relative FIFO order. + /// + /// Called from the `IN_FLIGHT_DEADLINE_SECS` expiry blocks in + /// `flush_next` and `has_flushable_work` — if a steer ack never arrives + /// (read loop hung, watcher never posted), the withheld events would + /// otherwise be permanently orphaned. Recover, do not log-and-drop: the + /// events were never delivered to the agent, so normal dispatch must + /// have a chance to deliver them. + /// + /// Iterates the stored entries in reverse so per-entry `push_front` + /// composes to original-FIFO order at the queue front (same discipline + /// as `requeue_preserve_timestamps` at line 453). + fn recover_withheld_for_expired_channel(&mut self, channel_id: Uuid) { + let Some(entries) = self.withheld_native_steer.remove(&channel_id) else { + return; + }; + let n = entries.len(); + let queue = self.queues.entry(channel_id).or_default(); + for qe in entries.into_iter().rev() { + queue.push_front(qe); + } + while queue.len() > MAX_PENDING_PER_CHANNEL { + queue.pop_back(); + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "withheld-steer recovery overflow — dropped newest event to enforce cap" + ); + } + tracing::warn!( + channel_id = %channel_id, + recovered = n, + "in-flight expiry recovered withheld steer event(s) — \ + steer ack never arrived; normal dispatch will deliver" + ); + } + /// Compact expired metadata entries to prevent unbounded map growth. /// /// Removes `retry_after` entries whose deadline has already passed, and @@ -802,7 +1006,12 @@ fn format_prompt_actor(pubkey: &str, profile_lookup: Option<&PromptProfileLookup /// /// Includes: event_id, channel (name + UUID), kind, sender (hex + npub), /// time, content, all tags (never stripped), and parsed structural fields. -fn format_event_block( +/// +/// Reused by the goose-native steer path (lib.rs mode-gate) to render the +/// single withheld event for delivery via `_goose/unstable/session/steer`, +/// without paying for the batch-level context blocks the in-flight turn +/// already has. +pub(crate) fn format_event_block( channel_id: Uuid, channel_info: Option<&PromptChannelInfo>, be: &BatchEvent, @@ -1201,9 +1410,18 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec) -> Vec) -> Vec) -> Vec) -> Self { + match reason { + // Default to steer framing if a merge somehow lacks a reason: the + // gentler "continue your work" wording is the safer fallback. + None | Some(CancelReason::Steer) => MergeFraming { + // We never capture the agent's partial work — session/cancel is + // terminal and returns nothing — so this section holds the + // *original request*, not a transcript. The header must not + // overclaim preserved state (per Dawn's framing review). + prior_header: "[What you were working on]", + new_header_single: "[New message — arrived while you were working]", + new_header_multi_prefix: "[New messages — arrived while you were working", + closing_note: "Note: A new message arrived while you were working. Continue your \ + in-progress work and incorporate the new message if it's relevant; if it's \ + unrelated, you may briefly acknowledge it and carry on.", + }, + Some(CancelReason::Interrupt) => MergeFraming { + prior_header: "[Previous request — interrupted before completion]", + new_header_single: "[New request — supersedes previous]", + new_header_multi_prefix: "[New request — supersedes previous", + closing_note: "Note: The previous request was interrupted. Please address the new \ + request.\nIf the new request is unrelated to the previous one, you may \ + briefly acknowledge the interruption.", + }, + } + } +} + +/// Framing strings for the goose-native steer path (lib.rs mode-gate), +/// pulled from the same source-of-truth as the cancel+merge fallback +/// (`MergeFraming::for_reason(Some(CancelReason::Steer))`). +/// +/// Returns `(new_header_single, closing_note)`. Native-steer renders only +/// the new-message header + the single event block + the closing note — +/// no `prior_header`, no original-request section, because the in-flight +/// goose turn already has all of that in context. The two paths share +/// these strings so an agent receiving either transport gets the same +/// "weave it in, don't abandon your work" orientation (Eva's drift-proof +/// requirement: native and fallback must not diverge in UX). +pub(crate) fn native_steer_framing() -> (&'static str, &'static str) { + let framing = MergeFraming::for_reason(Some(CancelReason::Steer)); + (framing.new_header_single, framing.closing_note) +} + #[cfg(test)] mod tests { use super::*; @@ -1468,6 +1744,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -1485,6 +1762,235 @@ mod tests { assert!(!prompt.contains("--- Event 1 ---")); } + /// Helper: build a merged (cancel + re-prompt) batch with one cancelled + /// event and one new event, framed by `reason`. + fn make_merged_batch(reason: Option) -> FlushBatch { + let ch = Uuid::new_v4(); + FlushBatch { + channel_id: ch, + events: vec![BatchEvent { + event: make_event("the new message"), + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }], + cancelled_events: vec![BatchEvent { + event: make_event("the original task"), + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }], + cancel_reason: reason, + } + } + + #[test] + fn test_format_prompt_steer_framing() { + let batch = make_merged_batch(Some(CancelReason::Steer)); + let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); + + // Steer framing: the new message "arrived while you were working" and + // the agent should "continue" — NOT supersede framing. + assert!( + prompt.contains("arrived while you were working"), + "steer prompt should frame the new message as arriving mid-task: {prompt}" + ); + assert!( + prompt.contains("Continue your"), + "steer prompt should instruct the agent to continue its work: {prompt}" + ); + assert!( + !prompt.contains("supersedes"), + "steer prompt must NOT use supersede framing: {prompt}" + ); + // Both the original and new content must survive the merge. + assert!(prompt.contains("the original task")); + assert!(prompt.contains("the new message")); + } + + #[test] + fn test_format_prompt_interrupt_framing() { + let batch = make_merged_batch(Some(CancelReason::Interrupt)); + let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); + + // Interrupt framing: the new request supersedes the previous one. + assert!( + prompt.contains("supersedes previous"), + "interrupt prompt should use supersede framing: {prompt}" + ); + assert!( + prompt.contains("interrupted before completion"), + "interrupt prompt should label the prior work as interrupted: {prompt}" + ); + assert!( + !prompt.contains("arrived while you were working"), + "interrupt prompt must NOT use steer framing: {prompt}" + ); + } + + #[test] + fn test_format_prompt_no_reason_defaults_to_steer_framing() { + // A merged batch with no recorded reason falls back to the gentler + // steer framing (the safer default — see MergeFraming::for_reason). + let batch = make_merged_batch(None); + let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); + assert!( + prompt.contains("arrived while you were working"), + "unset reason should default to steer framing: {prompt}" + ); + assert!(!prompt.contains("supersedes")); + } + + /// Full steering path, queue mechanics through to rendered prompt. + /// + /// The framing tests above hand-build a `FlushBatch`; this one drives the + /// *real* queue output through the *real* renderer so a regression in how + /// `flush_next` assembles the merged batch (which events land where, whether + /// the reason rides through) is caught against the actual prompt string — + /// the seam the split unit tests don't cover on their own. + #[test] + fn test_steer_end_to_end_queue_to_rendered_prompt() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + // Original turn is in flight: push the work, flush it into a batch. + q.push(make_queued(ch, "draft the migration plan")); + let batch = q.flush_next().unwrap(); + assert!(any_in_flight(&q)); + + // A steering-eligible mention arrives mid-turn. + q.push(make_queued(ch, "actually scope it to v2 only")); + + // The mode gate fires Steer → cancel → requeue as cancelled, carrying + // the steer reason (exactly the lib.rs requeue path). + q.requeue_as_cancelled(batch, CancelReason::Steer); + q.mark_complete(ch); + + // The re-prompt the agent actually receives. + let merged = q.flush_next().unwrap(); + assert_eq!(merged.cancel_reason, Some(CancelReason::Steer)); + let prompt = format_prompt(&merged, &FormatPromptArgs::default()).join("\n\n"); + + // Steer framing — "arrived while you were working" / "Continue", never + // supersede — survives the full queue→render path. + assert!( + prompt.contains("arrived while you were working"), + "end-to-end steer prompt must carry steer framing: {prompt}" + ); + assert!( + prompt.contains("Continue your"), + "end-to-end steer prompt must instruct continue: {prompt}" + ); + assert!( + !prompt.contains("supersedes"), + "end-to-end steer prompt must NOT supersede: {prompt}" + ); + // The honest prior header (no overclaimed partial-work capture). + assert!( + prompt.contains("[What you were working on]"), + "steer prior header must be the honest variant: {prompt}" + ); + // Both the original work and the steering message survive the merge. + assert!(prompt.contains("draft the migration plan")); + assert!(prompt.contains("actually scope it to v2 only")); + } + + #[test] + fn test_format_prompt_steer_framing_multi_event() { + // Multi-event header path must also branch on reason. + let ch = Uuid::new_v4(); + let batch = FlushBatch { + channel_id: ch, + events: vec![ + BatchEvent { + event: make_event("new one"), + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }, + BatchEvent { + event: make_event("new two"), + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }, + ], + cancelled_events: vec![BatchEvent { + event: make_event("original"), + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }], + cancel_reason: Some(CancelReason::Steer), + }; + let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); + assert!(prompt.contains("New messages — arrived while you were working — 2 events]")); + assert!(!prompt.contains("supersedes")); + } + + /// Cross-thread steering: original work in thread A (cancelled), steering + /// message in thread B (new). Pins Perci's edge — the reply instruction + /// targets the *steering* message (the one the agent is responding to, where + /// the mentioner is waiting), while the steer framing still says "continue + /// your in-progress work." This is intended behavior, not a mismatch. + #[test] + fn test_steer_cross_thread_reply_targets_steering_message() { + let ch = Uuid::new_v4(); + let thread_a = "a".repeat(64); + let thread_b = "b".repeat(64); + + let original = make_event_with_tags( + "@bot keep working on thread A", + vec![vec![ + "e".into(), + thread_a.clone(), + "".into(), + "reply".into(), + ]], + ); + let steering = make_event_with_tags( + "@bot note from thread B", + vec![vec![ + "e".into(), + thread_b.clone(), + "".into(), + "reply".into(), + ]], + ); + let _steering_id = steering.id.to_hex(); + + let batch = FlushBatch { + channel_id: ch, + events: vec![BatchEvent { + event: steering, + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }], + cancelled_events: vec![BatchEvent { + event: original, + prompt_tag: "@mention".into(), + received_at: Instant::now(), + }], + cancel_reason: Some(CancelReason::Steer), + }; + + let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); + + // Reply instruction points at the thread root of the steering message + // (thread_b), not the steering event's own id — this matches the + // human-aware reply anchoring from PR #1281: for human-facing turns in + // a thread, the anchor is always the thread root. + assert!( + prompt.contains(&format!("--reply-to {thread_b}")), + "reply instruction should target the steering thread root: {prompt}" + ); + assert!( + !prompt.contains(&format!("--reply-to {thread_a}")), + "reply instruction must NOT target the original thread: {prompt}" + ); + // Steer framing still frames the original as in-progress work to continue. + assert!(prompt.contains("[What you were working on]")); + assert!(prompt.contains("arrived while you were working")); + assert!(!prompt.contains("supersedes")); + } + + // ── Test 9b: requeue preserves events ──────────────────────────────────── + #[test] fn test_requeue_preserves_events() { let mut queue = EventQueue::new(DedupMode::Queue); @@ -1560,6 +2066,7 @@ mod tests { }, ], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -1587,6 +2094,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -1609,6 +2117,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let core = "[Agent Memory — core]\nbe helpful"; let prompt = format_prompt( @@ -1640,6 +2149,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt( &batch, @@ -1669,6 +2179,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let core = "[Agent Memory — core]\nbe helpful"; let prompt = format_prompt( @@ -1695,6 +2206,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; // format_prompt no longer accepts or emits base_prompt/system_prompt. @@ -1718,6 +2230,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let core = "[Agent Memory — core]\nremember this"; @@ -1773,6 +2286,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt( @@ -1810,6 +2324,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ctx = ConversationContext::Thread { @@ -2296,6 +2811,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "engineering".into(), @@ -2326,6 +2842,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "DM".into(), @@ -2363,6 +2880,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -2390,6 +2908,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ctx = ConversationContext::Thread { messages: vec![ @@ -2433,6 +2952,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "DM".into(), @@ -2481,6 +3001,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ctx = ConversationContext::Thread { messages: vec![ContextMessage { @@ -2687,6 +3208,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "DM".into(), @@ -2743,6 +3265,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "DM".into(), @@ -2782,6 +3305,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -2805,6 +3329,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -2827,6 +3352,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -2999,7 +3525,7 @@ mod tests { q.push(make_queued(ch, "new-1")); // Cancel the original batch and release the channel. - q.requeue_as_cancelled(batch); + q.requeue_as_cancelled(batch, CancelReason::Interrupt); q.mark_complete(ch); // flush_next should merge: events=[new-1], cancelled_events=[old-1, old-2]. @@ -3012,6 +3538,64 @@ mod tests { ); } + #[test] + fn test_requeue_as_cancelled_propagates_reason() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + // Merge path (new event present): reason rides on FlushBatch. + q.push(make_queued(ch, "old")); + let batch = q.flush_next().unwrap(); + q.push(make_queued(ch, "new")); + q.requeue_as_cancelled(batch, CancelReason::Steer); + q.mark_complete(ch); + let merged = q.flush_next().unwrap(); + assert_eq!( + merged.cancel_reason, + Some(CancelReason::Steer), + "steer reason should reach the merged batch" + ); + q.mark_complete(ch); + + // Fallback path (no new event): reason still rides through. + q.push(make_queued(ch, "only")); + let batch = q.flush_next().unwrap(); + q.requeue_as_cancelled(batch, CancelReason::Interrupt); + q.mark_complete(ch); + let fallback = q.flush_next().unwrap(); + assert_eq!( + fallback.cancel_reason, + Some(CancelReason::Interrupt), + "interrupt reason should reach the re-dispatched batch" + ); + q.mark_complete(ch); + + // A normal (non-cancel) flush carries no reason. + q.push(make_queued(ch, "plain")); + let plain = q.flush_next().unwrap(); + assert_eq!(plain.cancel_reason, None); + } + + #[test] + fn test_double_cancel_latest_reason_wins() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + q.push(make_queued(ch, "orig")); + let batch1 = q.flush_next().unwrap(); + q.push(make_queued(ch, "new-1")); + q.requeue_as_cancelled(batch1, CancelReason::Interrupt); + q.mark_complete(ch); + let batch2 = q.flush_next().unwrap(); + // Second cancel with a different reason — the latest reason wins. + q.requeue_as_cancelled(batch2, CancelReason::Steer); + q.push(make_queued(ch, "new-2")); + q.mark_complete(ch); + let batch3 = q.flush_next().unwrap(); + assert_eq!(batch3.cancel_reason, Some(CancelReason::Steer)); + } + + // ── Test: requeue_as_cancelled fallback (no new events) ────────────────── + #[test] fn test_requeue_as_cancelled_no_new_events_fallback() { let mut q = EventQueue::new(DedupMode::Queue); @@ -3022,7 +3606,7 @@ mod tests { let batch = q.flush_next().unwrap(); // Cancel the batch (no new events pushed) and release the channel. - q.requeue_as_cancelled(batch); + q.requeue_as_cancelled(batch, CancelReason::Interrupt); q.mark_complete(ch); // Fallback path: cancelled events become regular events, cancelled_events is empty. @@ -3046,7 +3630,7 @@ mod tests { // Push, flush, cancel — no new events queued. q.push(make_queued(ch, "msg")); let batch = q.flush_next().unwrap(); - q.requeue_as_cancelled(batch); + q.requeue_as_cancelled(batch, CancelReason::Interrupt); q.mark_complete(ch); // Channel has only cancelled events — should still be considered flushable. @@ -3064,7 +3648,7 @@ mod tests { // Push, flush, cancel. q.push(make_queued(ch, "msg")); let batch = q.flush_next().unwrap(); - q.requeue_as_cancelled(batch); + q.requeue_as_cancelled(batch, CancelReason::Interrupt); q.mark_complete(ch); // drain_channel should clear cancelled_batches for the channel. @@ -3092,7 +3676,7 @@ mod tests { q.push(make_queued(ch, "new-1")); // First cancel: store 2 cancelled events. - q.requeue_as_cancelled(batch1); + q.requeue_as_cancelled(batch1, CancelReason::Interrupt); q.mark_complete(ch); // Second flush: events=[new-1], cancelled_events=[orig-1, orig-2]. @@ -3102,7 +3686,7 @@ mod tests { // Second cancel: requeue_as_cancelled should accumulate all 3 events // (2 from cancelled_events + 1 from events). - q.requeue_as_cancelled(batch2); + q.requeue_as_cancelled(batch2, CancelReason::Interrupt); // Push 1 more new event and release channel. q.push(make_queued(ch, "new-2")); @@ -3134,6 +3718,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; // No profile lookup → sender treated as human → human-facing thread @@ -3175,6 +3760,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "DM".into(), @@ -3208,6 +3794,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; // Top-level human message (no lookup → human): the reply opens a new @@ -3236,6 +3823,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let ci = PromptChannelInfo { name: "DM".into(), @@ -3277,6 +3865,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; // Human-facing (no lookup) deep reply: anchor to the thread ROOT to @@ -3312,6 +3901,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, }; let prompt = format_prompt(&batch, &FormatPromptArgs::default()).join("\n\n"); @@ -3353,6 +3943,7 @@ mod tests { }, ], cancelled_events: vec![], + cancel_reason: None, }; // Scope derives from the last (threaded) event; human-facing → anchor @@ -3389,6 +3980,7 @@ mod tests { }, ], cancelled_events: vec![], + cancel_reason: None, }; // Last event is top-level and human-facing → opens a new thread @@ -3414,6 +4006,7 @@ mod tests { received_at: Instant::now(), }], cancelled_events: vec![], + cancel_reason: None, } } @@ -3506,4 +4099,190 @@ mod tests { None ); } + + // ── Goose-native steer withhold tests ─────────────────────────────────── + // + // Side-table semantics: `mark_native_steer_pending` moves an event out of + // `queues` into `withheld_native_steer`, making it invisible to + // `flush_next` / `has_flushable_work` / contiguous drain. `Success` ack + // drops it via `remove_event`; `Err` / `PromptCompletedNeutral` ack + // restores it to the queue front via `release_native_steer`. The + // `IN_FLIGHT_DEADLINE_SECS` expiry bulk-recovers withheld events so they + // are never permanently orphaned. + + /// A channel whose only queued event has been withheld for a goose-native + /// steer must be invisible to both `flush_next` and `has_flushable_work`. + /// The withhold is the whole point of the side table — it must close the + /// `mark_complete` → ack race window. + #[test] + fn test_native_steer_withhold_only_channel_not_flushable() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + let qe = make_queued(ch, "hello"); + let event_id = qe.event.id.to_hex(); + q.push(qe); + + assert!(q.mark_native_steer_pending(ch, &event_id)); + + assert!( + q.flush_next().is_none(), + "withheld-only channel must not be flushable" + ); + assert!( + !q.has_flushable_work(), + "withheld-only channel must not register as flushable work" + ); + assert_eq!(pending_count(&q), 0); + assert_eq!(q.withheld_native_steer.get(&ch).map(|v| v.len()), Some(1)); + } + + /// Earlier events on the same channel must flush normally during the + /// steer ack window. Only the specific withheld event is invisible. + /// After `release_native_steer`, the released event sits at the queue + /// front (push-to-front preserves original `received_at` FIFO) and is + /// delivered by the next `flush_next`. + #[test] + fn test_native_steer_earlier_events_flush_during_ack_window() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + // Three events arrive in order: e1, e2 (already queued), then e3 + // (the latest mid-turn mention being steered). + let e1 = make_queued_at(ch, "e1", Duration::from_millis(30)); + let e2 = make_queued_at(ch, "e2", Duration::from_millis(20)); + let e3 = make_queued_at(ch, "e3", Duration::from_millis(10)); + let e1_id = e1.event.id.to_hex(); + let e2_id = e2.event.id.to_hex(); + let e3_id = e3.event.id.to_hex(); + q.push(e1); + q.push(e2); + q.push(e3); + + // Steer in flight for e3 — withhold it from normal dispatch. + assert!(q.mark_native_steer_pending(ch, &e3_id)); + + // Earlier events flush as a normal batch; e3 is invisible. + let batch = q + .flush_next() + .expect("e1+e2 should flush during ack window"); + assert_eq!(batch.channel_id, ch); + assert_eq!(batch.events.len(), 2); + assert_eq!(batch.events[0].event.id.to_hex(), e1_id); + assert_eq!(batch.events[1].event.id.to_hex(), e2_id); + + // Earlier batch completes; channel is no longer in flight. + q.mark_complete(ch); + + // Ack arrives as Err or PromptCompletedNeutral → release e3. + q.release_native_steer(ch, &e3_id); + + let next = q.flush_next().expect("released e3 should now flush"); + assert_eq!(next.channel_id, ch); + assert_eq!(next.events.len(), 1); + assert_eq!(next.events[0].event.id.to_hex(), e3_id); + + assert_eq!(pending_count(&q), 0); + assert!(q.withheld_native_steer.is_empty()); + } + + /// If the steer ack never arrives — read loop hung, watcher never posted — + /// the `IN_FLIGHT_DEADLINE_SECS` auto-expiry block must bulk-recover the + /// withheld events back to the queue front so normal dispatch can deliver + /// them. Recover, not log-and-drop: the events were never seen by the + /// agent. + #[test] + fn test_native_steer_expiry_recovers_withheld() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + let qe = make_queued(ch, "withheld event"); + let event_id = qe.event.id.to_hex(); + q.push(qe); + + // Simulate a prompt in flight for `ch`, then withhold the queued + // event for an in-flight goose-native steer. + q.in_flight_channels.insert(ch); + q.in_flight_deadlines.insert(ch, Instant::now()); + q.in_flight_batch_sizes.insert(ch, 1); + assert!(q.mark_native_steer_pending(ch, &event_id)); + + // Force the in-flight deadline to be in the past, simulating the + // steer ack never arriving and the read loop hanging long enough + // for `IN_FLIGHT_DEADLINE_SECS` to elapse. Same expiry-simulation + // trick used by `test_retry_throttle_blocks_requeue_channel`. + q.in_flight_deadlines + .insert(ch, Instant::now() - Duration::from_secs(1)); + + // `has_flushable_work` runs the expiry block first; it must recover + // the withheld event so the channel registers as flushable. + assert!( + q.has_flushable_work(), + "expired channel with withheld event must register as flushable after recovery" + ); + + // The withheld event has been moved back to `queues[ch]`. + assert!(q.withheld_native_steer.is_empty()); + assert_eq!(pending_count(&q), 1); + + // Normal dispatch delivers it. + let batch = q + .flush_next() + .expect("recovered event should flush via normal dispatch"); + assert_eq!(batch.channel_id, ch); + assert_eq!(batch.events.len(), 1); + assert_eq!(batch.events[0].event.id.to_hex(), event_id); + } + + /// Bulk-release on expiry must preserve original FIFO. The + /// implementation iterates the side-table entries in reverse and + /// `push_front`s each — composing to original-FIFO at the queue front. + /// Test ≥2 withheld entries (3 here) with staggered `received_at`. + #[test] + fn test_native_steer_bulk_release_preserves_fifo() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + // Three events with staggered ages — e1 oldest, e3 newest. + let e1 = make_queued_at(ch, "e1", Duration::from_millis(30)); + let e2 = make_queued_at(ch, "e2", Duration::from_millis(20)); + let e3 = make_queued_at(ch, "e3", Duration::from_millis(10)); + let e1_id = e1.event.id.to_hex(); + let e2_id = e2.event.id.to_hex(); + let e3_id = e3.event.id.to_hex(); + q.push(e1); + q.push(e2); + q.push(e3); + + // Withhold all three in FIFO arrival order (e1, e2, e3 → side table). + // This simulates a pathological repeated-steer flow; the more + // realistic case (one withhold at a time) is covered by the other + // tests. What matters here is that the bulk-recovery path + // (reverse iter + push_front) composes to original FIFO at the + // queue front. + assert!(q.mark_native_steer_pending(ch, &e1_id)); + assert!(q.mark_native_steer_pending(ch, &e2_id)); + assert!(q.mark_native_steer_pending(ch, &e3_id)); + assert_eq!(pending_count(&q), 0); + assert_eq!(q.withheld_native_steer.get(&ch).map(|v| v.len()), Some(3)); + + // Trigger expiry → bulk-release path. + q.in_flight_channels.insert(ch); + q.in_flight_deadlines + .insert(ch, Instant::now() - Duration::from_secs(1)); + q.in_flight_batch_sizes.insert(ch, 3); + assert!(q.has_flushable_work()); + + // After recovery, the queue front-to-back order must match the + // original FIFO: e1, e2, e3. + let recovered: Vec = q + .queues + .get(&ch) + .expect("queue restored") + .iter() + .map(|qe| qe.event.id.to_hex()) + .collect(); + assert_eq!(recovered, vec![e1_id, e2_id, e3_id]); + assert!(q.withheld_native_steer.is_empty()); + } } diff --git a/crates/buzz-agent/src/agent.rs b/crates/buzz-agent/src/agent.rs index a955e959b24..54cb4393995 100644 --- a/crates/buzz-agent/src/agent.rs +++ b/crates/buzz-agent/src/agent.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use serde_json::json; -use tokio::sync::{watch, Semaphore}; +use tokio::sync::{mpsc, watch, Semaphore}; use tokio::task::JoinSet; use crate::builtin; @@ -31,6 +31,11 @@ pub struct RunCtx<'a> { pub skills: &'a [SkillEntry], pub wire: &'a WireSender, pub cancel: &'a mut watch::Receiver, + /// Mid-turn steer queue. Drained at each round boundary (before the next + /// LLM call): queued messages are appended to history as user turns so the + /// model sees them on its next request, without restarting the turn. Fed by + /// the `_goose/unstable/session/steer` handler. + pub steer: &'a mut mpsc::UnboundedReceiver>, pub history: &'a mut Vec, pub original_task: &'a mut Option, pub handoff_count: &'a mut usize, @@ -78,6 +83,11 @@ impl RunCtx<'_> { if *self.cancel.borrow() { return Ok(StopReason::Cancelled); } + // Round boundary: fold in any steer messages queued since the last + // round. They land as user turns so the model incorporates them on + // its next request — the turn continues, it is not restarted. Drain + // non-blocking; an empty queue is the common case. + self.drain_steers(); match self.maybe_handoff().await { HandoffOutcome::Cancelled => return Ok(StopReason::Cancelled), // Context was just reset — the prior request's token count no @@ -227,6 +237,27 @@ impl RunCtx<'_> { } } + /// Non-blocking drain of the steer queue. Each queued steer is appended to + /// history as a user turn so the model picks it up on its next request. A + /// steer whose blocks all fail to render (e.g. unsupported content) is + /// skipped rather than aborting the turn — steering is best-effort + /// augmentation, not a hard input contract like the initial prompt. + fn drain_steers(&mut self) { + while let Ok(blocks) = self.steer.try_recv() { + match prompt_to_text(blocks) { + Ok(text) if !text.trim().is_empty() => { + self.history.push(HistoryItem::User(text)); + } + Ok(_) => { + tracing::debug!("dropping empty steer message"); + } + Err(e) => { + tracing::warn!("dropping unrenderable steer message: {e}"); + } + } + } + } + /// Unified tool-call execution. Three phases: /// 1. Preflight (sequential): emit `pending`; unknown tools fail fast /// with a synthetic result. Cancel here fills every still-empty diff --git a/crates/buzz-agent/src/lib.rs b/crates/buzz-agent/src/lib.rs index 953d66f5c2c..71f448b1e32 100644 --- a/crates/buzz-agent/src/lib.rs +++ b/crates/buzz-agent/src/lib.rs @@ -23,10 +23,11 @@ use crate::config::{Config, MAX_SYSTEM_PROMPT_BYTES, PROTOCOL_VERSION}; use crate::hints::SkillEntry; use crate::llm::Llm; use crate::mcp::McpRegistry; -use crate::types::HistoryItem; +use crate::types::{ContentBlock, HistoryItem}; use crate::wire::{ classify, Inbound, InitializeParams, SessionCancelParams, SessionNewParams, - SessionPromptParams, WireMsg, WireSender, INVALID_PARAMS, METHOD_NOT_FOUND, PARSE_ERROR, + SessionPromptParams, SessionSteerParams, WireMsg, WireSender, INVALID_PARAMS, METHOD_NOT_FOUND, + PARSE_ERROR, }; struct App { @@ -43,6 +44,16 @@ struct Session { history: Vec, cancel_tx: watch::Sender, busy: bool, + /// Run id of the in-flight prompt, set when a prompt starts and cleared + /// when it ends. `None` means no active run — a steer request targeting + /// this session is rejected. Steer-capable clients learn this value from + /// the `params.update._meta.goose.activeRunId` field on `session/update`. + active_run_id: Option, + /// Sender for mid-turn steer messages. Created fresh per prompt (like + /// `cancel_tx`); the running prompt loop holds the matching receiver and + /// drains queued steers at round boundaries. `None` when no prompt is in + /// flight. + steer_tx: Option>>, original_task: Option, handoff_count: usize, stop_rejections: u32, @@ -194,6 +205,13 @@ async fn handle_request( cancel_session(app, params).await; wire::send(wire_tx, wire::ok(id, Value::Null)).await; } + // goose-compatible non-standard extension: inject user input into the + // currently active prompt without starting a new one. Mirrors goose's + // `_goose/unstable/session/steer` wire contract so a single client-side + // delivery path serves both agents. + "_goose/unstable/session/steer" => { + steer_session(app, id, params, wire_tx).await; + } _ => { wire::send( wire_tx, @@ -334,6 +352,8 @@ async fn session_new(app: &Arc, id: Value, params: Value, wire_tx: &WireSen history: Vec::new(), cancel_tx, busy: false, + active_run_id: None, + steer_tx: None, original_task: None, handoff_count: 0, stop_rejections: 0, @@ -362,6 +382,85 @@ async fn cancel_session(app: &Arc, params: Value) { } } +/// Handle `_goose/unstable/session/steer`: queue user input into the in-flight +/// prompt. Validation mirrors goose's `on_steer_session`: +/// - empty prompt → `invalid_params` +/// - no active run (no prompt in flight) → `invalid_params` +/// - `expectedRunId` mismatch → `invalid_params` (caller is steering a turn +/// that already ended or rotated; it must fall back to cancel+merge) +/// +/// On success the message is queued for pickup at the next round boundary and +/// we reply `{ runId, messageId }`, then emit a `queuedSteer` session/update so +/// the client can correlate the accepted steer with its eventual pickup. +async fn steer_session(app: &Arc, id: Value, params: Value, wire_tx: &WireSender) { + let p: SessionSteerParams = match decode(params, "_goose/unstable/session/steer") { + Ok(p) => p, + Err(m) => return reject(wire_tx, id, INVALID_PARAMS, &m).await, + }; + if p.prompt.is_empty() { + return reject( + wire_tx, + id, + INVALID_PARAMS, + "steer: prompt must not be empty", + ) + .await; + } + if p.expected_run_id.is_empty() { + return reject( + wire_tx, + id, + INVALID_PARAMS, + "steer: expectedRunId must not be empty", + ) + .await; + } + let message_id = format!("steer_{}", session_token().unwrap_or_else(|_| "x".into())); + let run_id = { + let sessions = app.sessions.lock().await; + let Some(s) = sessions.get(&p.session_id) else { + return reject(wire_tx, id, INVALID_PARAMS, "steer: unknown session").await; + }; + let Some(active) = s.active_run_id.as_deref() else { + return reject(wire_tx, id, INVALID_PARAMS, "steer: no active run to steer").await; + }; + if active != p.expected_run_id { + return reject( + wire_tx, + id, + INVALID_PARAMS, + &format!( + "steer: expected active run id `{}` but found `{active}`", + p.expected_run_id + ), + ) + .await; + } + // A live run always has a steer_tx; if the channel is gone the run is + // tearing down — treat as no active run rather than queue into the void. + match &s.steer_tx { + Some(tx) if tx.send(p.prompt).is_ok() => active.to_owned(), + _ => return reject(wire_tx, id, INVALID_PARAMS, "steer: no active run to steer").await, + } + }; + wire::send( + wire_tx, + wire::ok(id, json!({ "runId": run_id, "messageId": message_id })), + ) + .await; + // Best-effort correlation hint for the client; mirrors goose's + // `send_queued_steer_update`. Not load-bearing for delivery. + wire::send( + wire_tx, + wire::session_update_with_goose_meta( + &p.session_id, + json!({ "sessionUpdate": "session_info_update" }), + json!({ "queuedSteer": { "messageId": message_id, "runId": run_id } }), + ), + ) + .await; +} + fn spawn_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender) { tokio::spawn(async move { run_prompt(app, id, params, wire_tx).await }); } @@ -383,6 +482,8 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender mut last_request_history_bytes, mut cancel_rx, effective_system_prompt, + run_id, + mut steer_rx, ) = match acquire_session(&app, &p.session_id).await { Ok(v) => v, Err(reason) => { @@ -395,6 +496,17 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender .await } }; + // Advertise the active run id so steer-capable clients can target this turn + // via `expectedRunId`. Mirrors goose's `send_active_run_update`. + wire::send( + &wire_tx, + wire::session_update_with_goose_meta( + &sid, + json!({ "sessionUpdate": "session_info_update" }), + json!({ "activeRunId": run_id }), + ), + ) + .await; let mut ctx = RunCtx { cfg: &app.cfg, session_id: &sid, @@ -404,6 +516,7 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender skills: &skills, wire: &wire_tx, cancel: &mut cancel_rx, + steer: &mut steer_rx, history: &mut history, original_task: &mut original_task, handoff_count: &mut handoff_count, @@ -414,6 +527,9 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender let result = ctx.run(p.prompt).await; if let Some(s) = app.sessions.lock().await.get_mut(&sid) { s.busy = false; + // Clear run state so a late steer can't queue into a finished turn. + s.active_run_id = None; + s.steer_tx = None; s.history = history; s.original_task = original_task; s.handoff_count = handoff_count; @@ -449,6 +565,8 @@ async fn acquire_session( Option, watch::Receiver, Arc, + String, + mpsc::UnboundedReceiver>, ), &'static str, > { @@ -463,6 +581,13 @@ async fn acquire_session( // Skills are read-only after session creation; clone the Vec so RunCtx // can hold a reference without holding the sessions lock. let skills = s.skills.clone(); + // Fresh run id + steer channel for this turn. The run id lets steer-capable + // clients target *this* turn (rejecting steers aimed at a turn that already + // ended); the channel carries mid-turn injections to the run loop. + let run_id = format!("run_{}", session_token().unwrap_or_else(|_| "x".into())); + s.active_run_id = Some(run_id.clone()); + let (steer_tx, steer_rx) = mpsc::unbounded_channel(); + s.steer_tx = Some(steer_tx); Ok(( s.id.clone(), s.mcp.clone(), @@ -475,6 +600,8 @@ async fn acquire_session( s.last_request_history_bytes, rx, Arc::clone(&s.effective_system_prompt), + run_id, + steer_rx, )) } diff --git a/crates/buzz-agent/src/wire.rs b/crates/buzz-agent/src/wire.rs index 6250d0e63f3..7c164724db7 100644 --- a/crates/buzz-agent/src/wire.rs +++ b/crates/buzz-agent/src/wire.rs @@ -66,6 +66,20 @@ pub struct SessionCancelParams { pub session_id: String, } +/// Params for goose's non-standard `_goose/unstable/session/steer` request: +/// inject user input into the *currently active* prompt without starting a new +/// one. `expected_run_id` must match the run id buzz-agent advertised via +/// `params.update._meta.goose.activeRunId` on a `session/update`, so a steer +/// can't race a turn that already ended or hasn't started. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionSteerParams { + pub session_id: String, + #[serde(default)] + pub prompt: Vec, + pub expected_run_id: String, +} + pub fn classify(msg: &Value) -> Inbound { if !msg.is_object() || msg.get("jsonrpc").and_then(Value::as_str) != Some("2.0") { return Inbound::Invalid { @@ -112,6 +126,25 @@ pub fn session_update(sid: &str, update: Value) -> Value { }) } +/// A `session/update` notification carrying a `update._meta.goose.` field. +/// Used to advertise `activeRunId` (so steer-capable clients can target the +/// in-flight run) and `queuedSteer` (so they can correlate an accepted steer +/// with the chunk that later picks it up) — matching goose's wire layout where +/// `_meta` is nested inside the `update` object (per the ACP `SessionInfoUpdate` +/// schema), not alongside it at the params level. +pub fn session_update_with_goose_meta(sid: &str, update: Value, goose_meta: Value) -> Value { + let mut update = update; + update["_meta"] = json!({ "goose": goose_meta }); + json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": sid, + "update": update, + }, + }) +} + pub async fn send(wire: &WireSender, msg: Value) { let _ = wire.send(WireMsg::Notify(msg)).await; } diff --git a/crates/buzz-agent/tests/fake_llm.rs b/crates/buzz-agent/tests/fake_llm.rs index 32459c697b3..a1791e7d619 100644 --- a/crates/buzz-agent/tests/fake_llm.rs +++ b/crates/buzz-agent/tests/fake_llm.rs @@ -574,3 +574,214 @@ async fn system_prompt_absent_no_canary() { h.shutdown().await; } + +// ─── Steering (_goose/unstable/session/steer) ─────────────────────────────── + +/// Wait for the `activeRunId` advert buzz-agent emits at prompt start and +/// return the run id, so a steer can target the live turn. +async fn recv_active_run_id(h: &mut Harness) -> String { + let v = h + .recv_until(|v| { + v.get("method") == Some(&json!("session/update")) + && v["params"]["update"]["_meta"]["goose"]["activeRunId"].is_string() + }) + .await; + v["params"]["update"]["_meta"]["goose"]["activeRunId"] + .as_str() + .unwrap() + .to_owned() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn steer_folds_into_active_turn_without_cancelling() { + // A two-round turn (tool call → text). A steer sent once the run is live + // must (a) be accepted with the matching runId, (b) NOT cancel the turn — + // it still ends with end_turn — and (c) reach the provider as a user turn. + let (url, captures) = spawn_capturing_fake_llm(vec![ + openai_tool_call("call_steer", "fake__noop", json!({})), + openai_text("acknowledged the steer"), + ]) + .await; + let mut h = Harness::spawn(&url).await; + let sid = init_session(&mut h).await; + + let p_id = h + .send( + "session/prompt", + json!({ + "sessionId": sid, + "prompt": [{"type":"text","text":"work on the original task"}], + }), + ) + .await; + + // Learn the run id, then steer into it before the turn finishes. + let run_id = recv_active_run_id(&mut h).await; + let steer_text = "STEER-CANARY: also consider the edge case"; + let s_id = h + .send( + "_goose/unstable/session/steer", + json!({ + "sessionId": sid, + "expectedRunId": run_id, + "prompt": [{"type":"text","text": steer_text}], + }), + ) + .await; + + // Steer is accepted and echoes the run id it landed in. + let mut steer_ok = false; + let mut end_turn = false; + for _ in 0..40 { + let v = h.recv().await; + if v["id"] == json!(s_id) { + assert_eq!( + v["result"]["runId"], + json!(run_id), + "steer ran into the live turn" + ); + assert!( + v["result"]["messageId"] + .as_str() + .is_some_and(|m| m.starts_with("steer_")), + "steer reply carries a messageId" + ); + steer_ok = true; + } else if v["id"] == json!(p_id) { + // The turn was NOT cancelled — it completed normally. + assert_eq!(v["result"]["stopReason"], "end_turn"); + end_turn = true; + } + if steer_ok && end_turn { + break; + } + } + assert!(steer_ok, "steer request was not accepted"); + assert!(end_turn, "turn did not complete with end_turn after steer"); + + // The steered text reached the provider as a user message in some round. + let reqs = captures.lock().await; + let saw_steer = reqs.iter().any(|req| { + req["messages"].as_array().is_some_and(|msgs| { + msgs.iter().any(|m| { + m["role"] == "user" + && m["content"] + .as_str() + .is_some_and(|c| c.contains(steer_text)) + }) + }) + }); + assert!( + saw_steer, + "steered text never reached the provider; captured requests: {reqs:#?}" + ); + drop(reqs); + h.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn steer_rejected_when_no_active_run() { + // No prompt in flight → no active run → invalid_params. + let url = spawn_fake_llm(vec![]).await; + let mut h = Harness::spawn(&url).await; + let sid = init_session(&mut h).await; + + let s_id = h + .send( + "_goose/unstable/session/steer", + json!({ + "sessionId": sid, + "expectedRunId": "run_does_not_exist", + "prompt": [{"type":"text","text":"hello?"}], + }), + ) + .await; + let v = h.recv_until(|v| v["id"] == json!(s_id)).await; + assert_eq!(v["error"]["code"], -32602, "expected invalid_params"); + h.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn steer_rejected_on_run_id_mismatch() { + // A live run, but the caller targets a stale/wrong run id → invalid_params, + // so the client falls back to cancel+merge instead of injecting blind. + let (url, _captures) = spawn_capturing_fake_llm(vec![ + openai_tool_call("call_x", "fake__noop", json!({})), + openai_text("done"), + ]) + .await; + let mut h = Harness::spawn(&url).await; + let sid = init_session(&mut h).await; + + let p_id = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}), + ) + .await; + let _live_run = recv_active_run_id(&mut h).await; + + let s_id = h + .send( + "_goose/unstable/session/steer", + json!({ + "sessionId": sid, + "expectedRunId": "run_stale_mismatch", + "prompt": [{"type":"text","text":"too late"}], + }), + ) + .await; + + let mut saw_reject = false; + for _ in 0..40 { + let v = h.recv().await; + if v["id"] == json!(s_id) { + assert_eq!( + v["error"]["code"], -32602, + "mismatched runId must be rejected" + ); + saw_reject = true; + } else if v["id"] == json!(p_id) { + // Turn finishes normally regardless of the rejected steer. + break; + } + } + assert!(saw_reject, "run-id mismatch was not rejected"); + h.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn steer_rejected_on_empty_prompt() { + let (url, _captures) = spawn_capturing_fake_llm(vec![ + openai_tool_call("call_x", "fake__noop", json!({})), + openai_text("done"), + ]) + .await; + let mut h = Harness::spawn(&url).await; + let sid = init_session(&mut h).await; + let p_id = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}), + ) + .await; + let run_id = recv_active_run_id(&mut h).await; + let s_id = h + .send( + "_goose/unstable/session/steer", + json!({"sessionId": sid, "expectedRunId": run_id, "prompt": []}), + ) + .await; + let mut saw_reject = false; + for _ in 0..40 { + let v = h.recv().await; + if v["id"] == json!(s_id) { + assert_eq!(v["error"]["code"], -32602, "empty prompt must be rejected"); + saw_reject = true; + } else if v["id"] == json!(p_id) { + break; + } + } + assert!(saw_reject, "empty steer prompt was not rejected"); + h.shutdown().await; +} diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index 46db488e95a..6274c79a1c4 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -1654,7 +1654,7 @@ pub fn spawn_agent_child( .unwrap_or(super::types::DEFAULT_AGENT_MAX_TURN_DURATION_SECONDS); command.env("BUZZ_ACP_MAX_TURN_DURATION", max_dur.to_string()); command.env("BUZZ_ACP_AGENTS", record.parallelism.to_string()); - command.env("BUZZ_ACP_MULTIPLE_EVENT_HANDLING", "owner-interrupt"); + command.env("BUZZ_ACP_MULTIPLE_EVENT_HANDLING", "steer"); command.env("BUZZ_ACP_DEDUP", "queue"); if let Some(meta) = runtime_meta { for (key, value) in meta.default_env { diff --git a/desktop/src/features/agents/ui/agentSessionTranscript.ts b/desktop/src/features/agents/ui/agentSessionTranscript.ts index dec0ee320ea..8e1cee499c0 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscript.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscript.ts @@ -357,6 +357,36 @@ export function processTranscriptEvent( ); } } + } else if ( + event.kind === "acp_write" && + method === "_goose/unstable/session/steer" + ) { + const promptText = extractPromptText(payload); + if (promptText) { + const parsedPrompt = parsePromptText(promptText); + if (parsedPrompt.userText) { + upsertMessage( + d, + `steer:${ch}:${event.turnId ?? event.seq}`, + "user", + parsedPrompt.userTitle, + parsedPrompt.userText, + event.timestamp, + channelId, + parsedPrompt.userPubkey, + ); + } + if (parsedPrompt.sections.length > 0) { + upsertMetadata( + d, + `steer-context:${ch}:${event.turnId ?? event.seq}`, + "Prompt context", + parsedPrompt.sections, + event.timestamp, + channelId, + ); + } + } } else if (event.kind === "acp_read" && method === "session/update") { const params = asRecord(payload.params); const update = asRecord(params.update); @@ -375,15 +405,20 @@ export function processTranscriptEvent( channelId, ); } else if (updateType === "user_message_chunk") { - upsertMessage( - d, - `user:${ch}:${messageId ?? turnKey}`, - "user", - "User", - extractContentText(update.content), - event.timestamp, - channelId, - ); + // Suppress user_message_chunk echo when a steer already rendered + // the user message for this turn (Goose echoes steered content back). + const steerKey = `steer:${ch}:${event.turnId ?? event.seq}`; + if (!d.itemsById.has(steerKey)) { + upsertMessage( + d, + `user:${ch}:${messageId ?? turnKey}`, + "user", + "User", + extractContentText(update.content), + event.timestamp, + channelId, + ); + } } else if (updateType === "agent_thought_chunk") { upsertTextItem( d, diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts index 417e95a5b32..857c24c635a 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts @@ -19,7 +19,9 @@ export function parsePromptText(text: string): { userTitle: string; userPubkey: string | null; } { - const sections = parsePromptSections(text); + const sections = parsePromptSections(text).filter( + (s) => s.body.trim().length > 0, + ); if (sections.length === 0) { return { sections: [],