fix(agent_loop): post-phase-5 code review — R1 drain collision, R3 signal threading, #2 provider name

Three fixes from the phase-4.6/phase-5 code review.

**R1**: `prepare_next_turn_from_plugin_manager` was draining
`harness-next-model` alongside `harness-next-thinking-level`.
The model slot has pre-existing dirge semantics — read by the
UI at end-of-run (`ui/mod.rs:2359`) to spawn a fresh agent
against the new model. With the prepareNextTurn hook draining
it first, the UI's consumer saw None and `harness/set-next-model`
silently failed.

Fix: only drain the thinking-level slot in the hook. The model
slot stays for the UI consumer. Mid-run model swap isn't
supported anyway (run.rs already logs a warning when
TurnUpdate.model is set — code review #3); a separate API for
real mid-run model swap waits on rig API growth.

**R3**: `opts.signal` in StreamOptions was silently ignored by
the rig stream adapter. Mid-stream cancellation against the
rig request had no effect — signal only took effect at the
next turn boundary. (Old runner.rs path had the same
limitation; not a regression but a real gap.)

Fix: thread `Option<AbortSignal>` into wrap_streamed_assistant.
Per-chunk pre-poll check: if signal is cancelled, emit an
Error event with "aborted" substring and exit. Mid-LLM-call
cancel now actually stops the rig request promptly.

**#2**: `get_api_key` hook was called with `""` instead of the
provider name. Pi contract: `getApiKey(provider: string) =>
key`. Provider-aware hooks couldn't dispatch.

Fix: add `provider_name: Option<String>` to LoopConfig (and
LoopSpawnConfig, threaded through). `AnyAgent::provider_name()`
returns the canonical name per variant ("anthropic", "glm",
etc.). spawn_runner sets it. stream_assistant_response passes
it to the hook.

**Tests** (3 new):
  - prepare_next_turn_does_not_drain_next_model_slot (R1)
  - signal_cancels_stream_mid_flight (R3)
  - signal_none_does_not_affect_stream (R3 negative case)
  - test_get_api_key_receives_provider_name (#2)

**Not fixed in this commit** (documented as deferred):
  - R2: opts.api_key silently ignored by rig adapter (rig's
    client carries the key at construction; per-request
    override would need a rig API change)
  - R4: opts.request_timeout silently ignored (same reason)
  - Per-provider reasoning mappers (separate larger commit)

Gates:
  - cargo build (default)            clean
  - cargo build --features plugin    clean
  - cargo build --all-features       clean
  - cargo test                       849 passed; 6 ignored
  - cargo fmt                        clean

Author: Yogthos

src/agent/agent_loop/h7_smoke.rs

@@ -176,6 +176,7 @@ async fn h7_scenario_1_simple_text() {
         steering_queue: None,
         tool_execution: crate::agent::agent_loop::types::ToolExecutionMode::Parallel,
         event_channel_capacity: 256,
+        provider_name: None,
     };
     let runner = spawn_loop_runner(cfg).into_agent_runner();
     let (events, response) = drain_to_done(runner).await;
@@ -239,6 +240,7 @@ async fn h7_scenario_2_turn_boundaries() {
         steering_queue: None,
         tool_execution: crate::agent::agent_loop::types::ToolExecutionMode::Parallel,
         event_channel_capacity: 256,
+        provider_name: None,
     };
     let runner = spawn_loop_runner(cfg).into_agent_runner();
     let (events, response) = drain_to_done(runner).await;
@@ -336,6 +338,7 @@ async fn h7_scenario_5_auth_error_surfaces() {
         steering_queue: None,
         tool_execution: crate::agent::agent_loop::types::ToolExecutionMode::Parallel,
         event_channel_capacity: 256,
+        provider_name: None,
     };
     let runner = spawn_loop_runner(cfg).into_agent_runner();
     let (events, _) = drain_to_done(runner).await;
@@ -485,6 +488,7 @@ async fn h7_scenario_3_tool_dispatch() {
         steering_queue: None,
         tool_execution: crate::agent::agent_loop::types::ToolExecutionMode::Sequential,
         event_channel_capacity: 256,
+        provider_name: None,
     };
     let runner = spawn_loop_runner(cfg).into_agent_runner();
     let (events, response) = drain_to_done(runner).await;
@@ -587,6 +591,7 @@ async fn h7_glm_scenario_1_simple_text() {
         steering_queue: None,
         tool_execution: crate::agent::agent_loop::types::ToolExecutionMode::Parallel,
         event_channel_capacity: 256,
+        provider_name: None,
     };
     let runner = spawn_loop_runner(cfg).into_agent_runner();
     let (events, response) = drain_to_done(runner).await;
@@ -707,6 +712,7 @@ async fn h7_glm_scenario_3_tool_dispatch() {
         steering_queue: None,
         tool_execution: crate::agent::agent_loop::types::ToolExecutionMode::Sequential,
         event_channel_capacity: 256,
+        provider_name: None,
     };
     let runner = spawn_loop_runner(cfg).into_agent_runner();
     let (events, response) = drain_to_done(runner).await;

src/agent/agent_loop/integration.rs

@@ -321,6 +321,12 @@ pub struct LoopSpawnConfig {
     /// Channel capacity for the AgentEvent output. 256 matches
     /// the existing `runner::spawn_agent` choice.
     pub event_channel_capacity: usize,
+
+    /// Provider name forwarded to `LoopConfig.provider_name` so
+    /// the `getApiKey` hook receives the canonical provider
+    /// identifier. Code review #2 — was missing; hook used to
+    /// receive empty string.
+    pub provider_name: Option<String>,
 }
 
 impl LoopSpawnConfig {
@@ -335,6 +341,7 @@ impl LoopSpawnConfig {
             history: Vec::new(),
             initial_prompt: prompt.into(),
             tools: Vec::new(),
+            provider_name: None,
             #[cfg(feature = "plugin")]
             plugin_mgr: None,
             steering_queue: None,
@@ -379,6 +386,7 @@ pub fn spawn_loop_runner(cfg: LoopSpawnConfig) -> LoopRunner {
         headers: std::collections::HashMap::new(),
         metadata: std::collections::HashMap::new(),
         request_timeout: None,
+        provider_name: cfg.provider_name.clone(),
     };
 
     #[cfg(feature = "plugin")]

src/agent/agent_loop/plugin_hooks.rs

@@ -232,36 +232,35 @@ fn flatten_text(content: &[serde_json::Value]) -> String {
 // ============================================================
 
 /// Build a `PrepareNextTurnFn` that reads the
-/// `harness-next-thinking-level` slot (and `harness-next-model`)
-/// from the plugin manager. Plugins set these slots via
-/// `harness/set-next-thinking-level` / `harness/set-next-model`
+/// `harness-next-thinking-level` slot from the plugin manager.
+/// Plugins set the slot via `harness/set-next-thinking-level`
 /// inside `on-tool-end` (or any hook firing between turns).
 ///
-/// Returns `Some(TurnUpdate)` with the requested fields when any
-/// slot was set; `None` otherwise. Context is never mutated by
-/// this factory — separate hooks compose if a plugin wants to
-/// rewrite the transcript.
+/// **Does NOT drain `harness-next-model`** (code review bug R1).
+/// That slot has pre-existing dirge semantics: read by the UI
+/// at end-of-run (`ui/mod.rs:2359`) to spawn a fresh agent
+/// against the new model. Mid-run model swap isn't supported
+/// today (rig's stream can't pivot mid-flight, and even
+/// `run_loop` only logs a warning when `TurnUpdate.model` is
+/// set — see code review #3). Draining the slot here would
+/// steal it from the UI consumer and break the existing
+/// `/model` swap flow.
 ///
-/// Locking pattern matches before/after_hook_from_plugin_manager:
-/// acquire-read-release synchronously per call.
+/// Returns `Some(TurnUpdate)` with the requested thinking
+/// level when the slot was set; `None` otherwise.
 pub fn prepare_next_turn_from_plugin_manager(pm: Arc<Mutex<PluginManager>>) -> PrepareNextTurnFn {
     Arc::new(move |_ctx| {
         let pm = pm.clone();
         Box::pin(async move {
-            let (thinking, model) = {
+            let thinking = {
                 let mut mgr = pm.lock().unwrap_or_else(|e| e.into_inner());
-                let t = mgr.take_pending_next_thinking_level();
-                let m = mgr.take_pending_next_model();
-                (t, m)
+                mgr.take_pending_next_thinking_level()
             };
-            let thinking_level = thinking.and_then(parse_thinking_level);
-            if thinking_level.is_none() && model.is_none() {
-                return None;
-            }
+            let thinking_level = thinking.and_then(parse_thinking_level)?;
             Some(TurnUpdate {
                 context: None,
-                model,
-                thinking_level,
+                model: None,
+                thinking_level: Some(thinking_level),
             })
         })
     })
@@ -696,6 +695,36 @@ mod tests {
         }
     }
 
+    /// R1 regression: `prepare_next_turn_from_plugin_manager`
+    /// MUST NOT drain `harness-next-model`. That slot is owned
+    /// by the UI's end-of-run handler (`ui/mod.rs::2359`).
+    /// Earlier versions of phase 5 drained both slots in the
+    /// hook, which silently broke `harness/set-next-model`
+    /// because whichever consumer fired first stole the value.
+    #[tokio::test]
+    async fn prepare_next_turn_does_not_drain_next_model_slot() {
+        let Some(pm) = try_pm() else { return };
+        {
+            let mut mgr = pm.lock().unwrap();
+            mgr.eval(r#"(defn swap [_ctx] (harness/set-next-model "gpt-5"))"#)
+                .unwrap();
+            mgr.register("on-tool-end", "swap");
+            mgr.dispatch_tool_hook("on-tool-end", "@{:tool \"t\" :output \"x\"}")
+                .unwrap();
+        }
+        let hook = prepare_next_turn_from_plugin_manager(pm.clone());
+        let result = hook(turn_ctx()).await;
+        // prepareNextTurn returns None because thinking_level
+        // wasn't set. The model slot remains intact.
+        assert!(
+            result.is_none(),
+            "prepare_next_turn should ignore model slot",
+        );
+        // Critical: the UI's end-of-run consumer can still read it.
+        let pending = pm.lock().unwrap().take_pending_next_model();
+        assert_eq!(pending, Some("gpt-5".to_string()));
+    }
+
     /// Unknown thinking-level strings get filtered out — a
     /// plugin typo doesn't crash the run.
     #[tokio::test]

src/agent/agent_loop/rig_stream.rs

@@ -60,11 +60,12 @@ use super::message::{AssistantMessage, ContentBlock, DeltaPhase, StopReason, Str
 pub fn wrap_rig_stream<R>(
     rig_stream: StreamingCompletionResponse<R>,
     chunk_timeout: Option<std::time::Duration>,
+    signal: Option<crate::agent::agent_loop::tool::AbortSignal>,
 ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>>
 where
     R: Clone + Unpin + Send + GetTokenUsage + 'static,
 {
-    wrap_streamed_assistant(Box::pin(rig_stream), chunk_timeout)
+    wrap_streamed_assistant(Box::pin(rig_stream), chunk_timeout, signal)
 }
 
 /// Lower-level variant: wrap any `Stream<Result<StreamedAssistantContent<R>,
@@ -87,6 +88,7 @@ pub fn wrap_streamed_assistant<R>(
         Box<dyn Stream<Item = Result<StreamedAssistantContent<R>, CompletionError>> + Send>,
     >,
     chunk_timeout: Option<std::time::Duration>,
+    signal: Option<crate::agent::agent_loop::tool::AbortSignal>,
 ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>>
 where
     R: Clone + Unpin + Send + 'static,
@@ -107,6 +109,23 @@ where
             std::collections::HashMap::new();
 
         loop {
+            // Code review R3: honor AbortSignal between chunks.
+            // The loop / tools already check signal at their
+            // boundaries; here we add a per-chunk check so a
+            // mid-stream cancel actually stops the rig request
+            // rather than waiting for the next turn boundary.
+            // Pre-poll check covers the case where signal was
+            // cancelled BEFORE the first chunk arrived; the
+            // post-await check catches cancellation that
+            // happened DURING the chunk wait.
+            if let Some(sig) = signal.as_ref()
+                && sig.is_cancelled()
+            {
+                yield StreamEvent::Error {
+                    error: "stream aborted by cancellation signal".to_string(),
+                };
+                return;
+            }
             // Apply per-chunk timeout if configured. The yield
             // pattern below mirrors `while let Some(...)` exactly
             // for the non-timeout path.
@@ -441,7 +460,7 @@ mod tests {
                 text: " world".to_string(),
             })),
         ]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         let labels: Vec<_> = events.iter().map(label).collect();
         assert_eq!(
             labels,
@@ -481,7 +500,7 @@ mod tests {
             },
             internal_call_id: "internal_1".to_string(),
         })]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         let labels: Vec<_> = events.iter().map(label).collect();
         assert_eq!(
             labels,
@@ -532,7 +551,7 @@ mod tests {
                 content: ToolCallDeltaContent::Delta("th\":\"/tmp/x\"}".to_string()),
             }),
         ]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         let labels: Vec<_> = events.iter().map(label).collect();
         assert_eq!(
             labels,
@@ -605,7 +624,7 @@ mod tests {
                 internal_call_id: "internal_x".to_string(),
             }),
         ]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         let final_msg = events
             .iter()
             .rev()
@@ -696,7 +715,7 @@ mod tests {
                 reasoning: " about this".to_string(),
             }),
         ]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         let labels: Vec<_> = events.iter().map(label).collect();
         assert_eq!(
             labels,
@@ -727,7 +746,7 @@ mod tests {
         let raw = raw_stream(vec![Ok(StreamedAssistantContent::Reasoning(
             Reasoning::new("All thinking"),
         ))]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         assert!(matches!(events[0], StreamEvent::Start { .. }));
         assert!(matches!(
             events[1],
@@ -751,7 +770,7 @@ mod tests {
                 text: " more text".to_string(),
             })),
         ]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         assert!(matches!(events.last(), Some(StreamEvent::Error { .. })));
         let dones = events
             .iter()
@@ -776,7 +795,7 @@ mod tests {
                 text: "done".to_string(),
             })),
         ]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         let final_msg = events
             .iter()
             .rev()
@@ -844,7 +863,7 @@ mod tests {
         let raw = raw_stream(vec![Ok(StreamedAssistantContent::Text(Text {
             text: "ok".to_string(),
         }))]);
-        let events = drain(wrap_streamed_assistant(raw, None)).await;
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
         // Normal completion — no Error.
         assert!(events.iter().any(|e| matches!(e, StreamEvent::Done { .. })));
         assert!(
@@ -863,7 +882,12 @@ mod tests {
         tokio::time::pause();
         let raw = stalling_stream();
         let drain_task = tokio::spawn(async move {
-            drain(wrap_streamed_assistant(raw, Some(Duration::from_secs(5)))).await
+            drain(wrap_streamed_assistant(
+                raw,
+                Some(Duration::from_secs(5)),
+                None,
+            ))
+            .await
         });
         tokio::time::advance(Duration::from_secs(10)).await;
         let events = drain_task.await.unwrap();
@@ -887,6 +911,66 @@ mod tests {
         );
     }
 
+    /// R3 regression: AbortSignal cancellation between chunks
+    /// produces an Error event and stops the stream. Earlier
+    /// versions silently ignored opts.signal at the rig
+    /// adapter level — mid-stream cancel had no effect until
+    /// the next turn boundary.
+    #[tokio::test]
+    async fn signal_cancels_stream_mid_flight() {
+        use crate::agent::agent_loop::tool::AbortSignal;
+        let raw = raw_stream(vec![
+            Ok(StreamedAssistantContent::Text(Text {
+                text: "first".to_string(),
+            })),
+            Ok(StreamedAssistantContent::Text(Text {
+                text: " second".to_string(),
+            })),
+        ]);
+        let signal = AbortSignal::new();
+        signal.cancel();
+        let events = drain(wrap_streamed_assistant(raw, None, Some(signal))).await;
+        // Pre-loop signal check fires before the first chunk
+        // poll. Expect: Start, Error (no Text deltas).
+        let kinds: Vec<&str> = events
+            .iter()
+            .map(|e| match e {
+                StreamEvent::Start { .. } => "start",
+                StreamEvent::Delta { .. } => "delta",
+                StreamEvent::Done { .. } => "done",
+                StreamEvent::Error { .. } => "error",
+            })
+            .collect();
+        assert_eq!(kinds, vec!["start", "error"]);
+        match events.last().unwrap() {
+            StreamEvent::Error { error } => {
+                assert!(
+                    error.contains("aborted"),
+                    "expected 'aborted' in error message; got: {error}"
+                );
+            }
+            _ => panic!("expected Error last"),
+        }
+    }
+
+    /// R3: signal=None means the cancellation check is skipped.
+    /// Pre-R3 behavior preserved when callers don't supply a
+    /// signal (e.g. ad-hoc tests).
+    #[tokio::test]
+    async fn signal_none_does_not_affect_stream() {
+        let raw = raw_stream(vec![Ok(StreamedAssistantContent::Text(Text {
+            text: "ok".to_string(),
+        }))]);
+        let events = drain(wrap_streamed_assistant(raw, None, None)).await;
+        // Normal completion — no Error.
+        assert!(events.iter().any(|e| matches!(e, StreamEvent::Done { .. })));
+        assert!(
+            !events
+                .iter()
+                .any(|e| matches!(e, StreamEvent::Error { .. }))
+        );
+    }
+
     /// Fast stream + tight timeout still completes normally —
     /// timeout only fires when a chunk takes longer than the
     /// deadline, not when the whole stream does. (Per-chunk
@@ -906,6 +990,7 @@ mod tests {
         let events = drain(wrap_streamed_assistant(
             raw,
             Some(Duration::from_millis(10)),
+            None,
         ))
         .await;
         assert!(events.iter().any(|e| matches!(e, StreamEvent::Done { .. })));

src/agent/agent_loop/rig_stream_factory.rs

@@ -191,7 +191,7 @@ where
         // 4. Call model.stream; wrap result or emit error.
         match model.stream(request).await {
             Ok(response) => {
-                let mut wrapped = wrap_rig_stream(response, chunk_timeout);
+                let mut wrapped = wrap_rig_stream(response, chunk_timeout, Some(opts.signal.clone()));
                 use futures::stream::StreamExt;
                 while let Some(evt) = wrapped.next().await {
                     yield evt;