hive-runtime, hive-agent: clear the ACP picker's pending model on a refusal

When an ACP agent answered session/set_config_option with an error,
choose() only logged it, so the session's current value never moved and
offered_pickers() kept treating the refused model as pending: the picker
showed the refused model as current and hid the effort picker.

hive-runtime now keeps the refused value per category on Choices, exposed
as Choice::refused, set on the RPC error and cleared when the agent next
accepts a value for that category. offered_pickers() treats a refused
value as not settable, so the picker shows the session's actual model and
its effort levels.

The fake ACP agent gains REFUSE, which errors the first
session/set_config_option to that value.

Refs #4832 (ACP half only)

hive-runtime/src/acp/mod.rs

@@ -8,7 +8,7 @@
 mod rpc;
 mod stream;
 
-use std::collections::VecDeque;
+use std::collections::{HashMap, VecDeque};
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -159,6 +159,8 @@ impl Canceller {
 pub struct Choice {
     pub current: String,
     pub values: Vec<String>,
+    /// The value the agent last refused to set, until it accepts one.
+    pub refused: Option<String>,
 }
 
 /// The model and effort the loaded session lets the client pick, from its
@@ -173,17 +175,26 @@ pub struct SessionChoices {
 /// Reads the [`SessionChoices`] of an [`AcpRuntime`]'s session from outside
 /// the call driving it. Cheap to clone.
 #[derive(Clone, Default)]
-pub struct Choices(Arc<std::sync::Mutex<Vec<stream::ConfigOption>>>);
+pub struct Choices(Arc<std::sync::Mutex<Offered>>);
+
+#[derive(Default)]
+struct Offered {
+    options: Vec<stream::ConfigOption>,
+    /// Per category, the value the agent last refused to set.
+    refused: HashMap<&'static str, String>,
+}
 
 impl Choices {
     /// What the session offers now. Empty until a session is attached, and
     /// after the agent is respawned until one is again.
     #[must_use]
     pub fn get(&self) -> SessionChoices {
+        let offered = self.lock();
         let choice = |category| {
-            self.find(category).map(|o| Choice {
-                current: o.current,
-                values: o.values,
+            find(&offered.options, category).map(|o| Choice {
+                current: o.current.clone(),
+                values: o.values.clone(),
+                refused: offered.refused.get(category).cloned(),
             })
         };
         SessionChoices {
@@ -193,16 +204,33 @@ impl Choices {
     }
 
     fn find(&self, category: &str) -> Option<stream::ConfigOption> {
-        let options = self.0.lock().unwrap_or_else(PoisonError::into_inner);
-        options
-            .iter()
-            .find(|o| o.category.as_deref() == Some(category))
-            .cloned()
+        find(&self.lock().options, category).cloned()
     }
 
     fn replace(&self, options: Vec<stream::ConfigOption>) {
-        *self.0.lock().unwrap_or_else(PoisonError::into_inner) = options;
+        self.lock().options = options;
     }
+
+    fn refuse(&self, category: &'static str, value: &str) {
+        self.lock().refused.insert(category, value.to_owned());
+    }
+
+    fn accept(&self, category: &str) {
+        self.lock().refused.remove(category);
+    }
+
+    fn lock(&self) -> std::sync::MutexGuard<'_, Offered> {
+        self.0.lock().unwrap_or_else(PoisonError::into_inner)
+    }
+}
+
+fn find<'a>(
+    options: &'a [stream::ConfigOption],
+    category: &str,
+) -> Option<&'a stream::ConfigOption> {
+    options
+        .iter()
+        .find(|o| o.category.as_deref() == Some(category))
 }
 
 /// Marks a turn in flight for [`Canceller::cancel`] while it lives.
@@ -803,12 +831,13 @@ async fn choose_model_and_effort(
 }
 
 /// Set `session`'s config option of `category` to `wanted`, if the session
-/// offers that value and holds another. A refusal is reported to `sink`, and
+/// offers that value and holds another. A refusal is reported to `sink` and
+/// kept as the choice's [`Choice::refused`] until the agent accepts a value;
 /// the turn goes ahead on the value the session holds.
 async fn choose(
     live: &mut Live,
     session: &str,
-    category: &str,
+    category: &'static str,
     wanted: Option<&str>,
     sink: &impl Sink,
 ) -> std::result::Result<(), AcpError> {
@@ -823,6 +852,7 @@ async fn choose(
     let params = json!({ "sessionId": session, "configId": option.id, "value": wanted });
     match live.conn.request("session/set_config_option", params).await {
         Ok(response) => {
+            live.choices.accept(category);
             if let Some(options) = stream::config_options(&response) {
                 live.offer(options);
             }
@@ -830,6 +860,7 @@ async fn choose(
         Err(e @ AcpError::Rpc { .. }) => {
             tracing::warn!(error = %e, category, wanted, "ACP session/set_config_option failed");
             sink.on_stderr_line(&format!("ACP agent refused {category} {wanted}: {e}"));
+            live.choices.refuse(category, wanted);
         }
         Err(e) => return Err(e),
     }
@@ -931,7 +962,8 @@ mod tests {
     /// With `OPTIONS` set, its sessions offer models `m/think` (the default)
     /// and `m/plain`, and effort levels `low` (the default) and `high` on
     /// `m/think` only; it appends each `session/set_config_option` to
-    /// `<log>.sets` as `id=value`.
+    /// `<log>.sets` as `id=value`. With `REFUSE` set, it answers the first
+    /// `session/set_config_option` to that value with an error.
     const AGENT: &str = r#"
 n=0 prompt= model=m/think effort=low
 printf 'start\n' >> "$1.methods"
@@ -976,6 +1008,11 @@ while IFS= read -r line; do
       cid=${line#*\"configId\":\"}; cid=${cid%%\"*}
       val=${line#*\"value\":\"}; val=${val%%\"*}
       printf '%s=%s\n' "$cid" "$val" >> "$1.sets"
+      if [ -n "$REFUSE" ] && [ "$val" = "$REFUSE" ] && [ ! -e "$1.refused" ]; then
+        : > "$1.refused"
+        printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32602,"message":"unknown value"}}\n' "$id"
+        continue
+      fi
       case $cid in model) model=$val ;; effort) effort=$val ;; esac
       result "$id" ;;
     session/prompt)
@@ -1460,6 +1497,7 @@ done
         Choice {
             current: current.to_owned(),
             values: values.iter().map(|v| (*v).to_owned()).collect(),
+            refused: None,
         }
     }
 
@@ -1511,4 +1549,33 @@ done
         runtime.run(&config, "three", &NoopSink).await.unwrap();
         assert_eq!(sets(dir.path()).len(), 2);
     }
+
+    #[tokio::test]
+    async fn a_refused_value_is_kept_until_the_agent_accepts_one() {
+        let dir = tempfile::tempdir().unwrap();
+        let env = [("OPTIONS", "1"), ("REFUSE", "m/plain")];
+        let runtime = agent(dir.path(), "ok", &env, policy(0));
+        let choices = runtime.choices().unwrap();
+        let config = Config {
+            model: Some("m/plain".into()),
+            ..config(dir.path())
+        };
+
+        runtime.run(&config, "one", &NoopSink).await.unwrap();
+        assert_eq!(
+            choices.get().model,
+            Some(Choice {
+                refused: Some("m/plain".into()),
+                ..choice("m/think", &["m/think", "m/plain"])
+            })
+        );
+
+        // The next turn asks again, and the agent accepts.
+        runtime.run(&config, "two", &NoopSink).await.unwrap();
+        assert_eq!(sets(dir.path()), ["model=m/plain", "model=m/plain"]);
+        assert_eq!(
+            choices.get().model,
+            Some(choice("m/plain", &["m/think", "m/plain"]))
+        );
+    }
 }