From 80c2529d0f48a8f230d126b693cf4b3b1d33273e Mon Sep 17 00:00:00 2001 From: "Codex (Smith Crafts Worker)" Date: Sun, 6 Sep 2026 16:27:16 -0400 Subject: [PATCH] feat(agents): change conversation thinking level at turn boundaries Signed-off-by: Codex (Smith Crafts Worker) --- Cargo.lock | 1 + crates/buzz-acp/Cargo.toml | 1 + crates/buzz-acp/src/lib.rs | 5 + crates/buzz-acp/src/live_effort.rs | 312 +++++++++++ crates/buzz-acp/src/live_effort_real_test.rs | 181 ++++++ crates/buzz-acp/src/live_effort_tests.rs | 529 ++++++++++++++++++ crates/buzz-acp/src/pool.rs | 66 ++- desktop/playwright.config.ts | 10 +- desktop/src/features/agents/AGENTS.md | 24 +- .../agents/lib/conversationEffort.test.mjs | 182 ++++++ .../features/agents/lib/conversationEffort.ts | 150 +++++ .../features/agents/observerEventOrdering.ts | 17 + .../src/features/agents/observerRelayStore.ts | 38 +- .../observerTranscriptRetention.test.mjs | 37 ++ .../agents/ui/ConversationEffortPicker.tsx | 166 ++++++ .../agents/ui/ManagedAgentSessionPanel.tsx | 11 + .../channels/ui/AgentSessionThreadPanel.tsx | 1 + desktop/src/shared/api/agentControl.ts | 14 + desktop/src/shared/api/types.ts | 5 +- desktop/src/testing/e2eBridge.ts | 28 +- desktop/tests/e2e/conversation-effort.spec.ts | 236 ++++++++ desktop/tests/helpers/bridge.ts | 5 +- docs/conversation-effort.md | 71 +++ 23 files changed, 2049 insertions(+), 41 deletions(-) create mode 100644 crates/buzz-acp/src/live_effort.rs create mode 100644 crates/buzz-acp/src/live_effort_real_test.rs create mode 100644 crates/buzz-acp/src/live_effort_tests.rs create mode 100644 desktop/src/features/agents/lib/conversationEffort.test.mjs create mode 100644 desktop/src/features/agents/lib/conversationEffort.ts create mode 100644 desktop/src/features/agents/observerEventOrdering.ts create mode 100644 desktop/src/features/agents/ui/ConversationEffortPicker.tsx create mode 100644 desktop/tests/e2e/conversation-effort.spec.ts create mode 100644 docs/conversation-effort.md diff --git a/Cargo.lock b/Cargo.lock index d5f80a5fc82..0ea0eb231f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -847,6 +847,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..2a85e68cef1 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -79,3 +79,4 @@ nix = { version = "0.31", default-features = false, features = ["signal"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } httparse = "1" +tempfile = "3" diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index ddd594b142d..e55b8d2df15 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4,6 +4,7 @@ mod acp; mod config; mod engram_fetch; mod filter; +mod live_effort; mod observer; mod pi_launcher; mod pool; @@ -1612,6 +1613,9 @@ fn handle_relay_observer_control_event( Some("cancel_turn") => { handle_cancel_turn_control(&payload, pool, observer); } + Some("switch_effort") => { + pool.queue_live_effort(&payload, observer); + } Some("switch_model") => { handle_switch_model_control(&payload, pool, observer); } @@ -3008,6 +3012,7 @@ async fn tokio_main() -> Result<()> { } loop { + pool.apply_pending_effort(observer.as_ref()).await; // Whether buffered work is waiting on a lazy pool. Also gates the // retry-deadline sleep arm below: a `Failed` lifecycle keeps its // (possibly past) `retry_at` until the next wake, so sleeping on it diff --git a/crates/buzz-acp/src/live_effort.rs b/crates/buzz-acp/src/live_effort.rs new file mode 100644 index 00000000000..25819ba5794 --- /dev/null +++ b/crates/buzz-acp/src/live_effort.rs @@ -0,0 +1,312 @@ +//! Owner-requested native effort changes at a response boundary. No turn is +//! cancelled or replayed: the exact ACP session is edited while its worker is +//! idle, before the next claim. Runtime overrides never become saved defaults. + +use std::collections::VecDeque; +use std::time::{Duration, Instant}; + +use serde::Deserialize; +use serde_json::{json, Value}; +use uuid::Uuid; + +use crate::acp::AcpError; +use crate::observer::{ObserverContext, ObserverHandle}; +use crate::pool::{AgentPool, SessionState}; +use crate::scope::SessionScope; + +const CAPACITY: usize = 32; +const RECEIPTS: usize = 256; +const EXPIRY: Duration = Duration::from_secs(300); +const APPLY_TIMEOUT: Duration = Duration::from_secs(5); +const CONFIG_CAPACITY: usize = 128; +const MAX_CONFIG_BYTES: usize = 256 * 1024; + +impl SessionState { + /// Bound native snapshots independently of the adapter's session lifetime. + /// A session without retained configuration cannot advertise live control. + pub(crate) fn remember_effort_config(&mut self, scope: &SessionScope, config: &mut Value) { + config["liveEffortSwitching"] = json!(true); + let retain = (self.configs.contains_key(scope) || self.configs.len() < CONFIG_CAPACITY) + && serde_json::to_vec(config).is_ok_and(|bytes| bytes.len() <= MAX_CONFIG_BYTES); + config["liveEffortSwitching"] = json!(retain); + if retain { + self.configs.insert(scope.clone(), config.clone()); + } else { + self.configs.remove(scope); + } + } +} + +#[cfg(test)] +#[path = "live_effort_tests.rs"] +mod tests; + +#[derive(Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +struct Request { + channel_id: Uuid, + session_id: String, + session_token: Uuid, + effort: String, + request_id: Uuid, +} + +struct Pending { + request: Request, + received: Instant, +} + +#[derive(Default)] +pub(crate) struct LiveEffortQueue { + pending: VecDeque, + // Never reapply a reconnect replay, including after the first edit settled. + receipts: VecDeque<(Request, &'static str, Instant)>, +} + +impl LiveEffortQueue { + pub(crate) fn has_session(&self, channel: Uuid, session: &str) -> bool { + self.pending + .iter() + .any(|p| p.request.channel_id == channel && p.request.session_id == session) + } + + fn finish( + &mut self, + request: &Request, + status: &'static str, + observer: Option<&ObserverHandle>, + ) { + if let Some((_, previous, _)) = self + .receipts + .iter_mut() + .find(|(r, _, _)| r.request_id == request.request_id) + { + *previous = status; + } + emit_result(request, status, observer); + } +} + +fn context(request: &Request) -> ObserverContext { + ObserverContext { + channel_id: Some(request.channel_id.to_string()), + session_id: Some(request.session_id.clone()), + turn_id: None, + started_at: None, + } +} + +fn emit_result(request: &Request, status: &str, observer: Option<&ObserverHandle>) { + if let Some(observer) = observer { + observer.emit( + "control_result", + None, + &context(request), + json!({ + "type": "switch_effort", "requestId": request.request_id, + "sessionId": request.session_id, "sessionToken": request.session_token, "effort": request.effort, "status": status, + }), + ); + } +} + +fn effort_option(config: &Value) -> Option<&Value> { + config + .get("configOptions")? + .as_array()? + .iter() + .find(|option| { + option.get("category").and_then(Value::as_str) == Some("thought_level") + && option.get("type").and_then(Value::as_str) == Some("select") + }) +} + +fn supports_value(options: &Value, desired: &str) -> bool { + options.as_array().is_some_and(|items| { + items.iter().any(|item| { + item.get("value").and_then(Value::as_str) == Some(desired) + || item + .get("options") + .is_some_and(|group| supports_value(group, desired)) + }) + }) +} + +impl AgentPool { + pub(crate) fn queue_live_effort(&mut self, payload: &Value, observer: Option<&ObserverHandle>) { + let Ok(request) = serde_json::from_value::(payload.clone()) else { + return; + }; + if request.session_id.is_empty() + || request.session_id.len() > 256 + || request.effort.is_empty() + || request.effort.len() > 128 + || request + .session_id + .chars() + .chain(request.effort.chars()) + .any(char::is_control) + { + emit_result(&request, "invalid_request", observer); + return; + } + self.live_effort + .receipts + .retain(|(_, _, at)| at.elapsed() <= EXPIRY * 2); + if let Some((original, status, _)) = self + .live_effort + .receipts + .iter() + .find(|(r, _, _)| r.request_id == request.request_id) + { + emit_result(original, status, observer); + return; + } + // Keep all receipts for the full replay window; capacity is backpressure, + // not eviction that would allow an old request to execute a second time. + if self.live_effort.pending.len() >= CAPACITY || self.live_effort.receipts.len() >= RECEIPTS + { + emit_result(&request, "busy", observer); + return; + } + // One outstanding edit per exact session. Later requests must wait for + // its receipt rather than silently replacing an already queued choice. + if self.live_effort.pending.iter().any(|p| { + p.request.channel_id == request.channel_id && p.request.session_id == request.session_id + }) { + emit_result(&request, "busy", observer); + return; + } + self.live_effort + .receipts + .push_back((request.clone(), "queued", Instant::now())); + self.live_effort.pending.push_back(Pending { + request: request.clone(), + received: Instant::now(), + }); + emit_result(&request, "queued", observer); + } + + /// At most one native RPC per main-loop iteration. Bounds control latency + /// and avoids holding the relay loop for N sequential adapter timeouts. + pub(crate) async fn apply_pending_effort(&mut self, observer: Option<&ObserverHandle>) { + let count = self.live_effort.pending.len(); + for _ in 0..count { + let Some(pending) = self.live_effort.pending.pop_front() else { + return; + }; + let request = &pending.request; + if pending.received.elapsed() > EXPIRY { + self.live_effort.finish(request, "expired", observer); + continue; + } + // A busy worker's session IDs are not visible here. Wait until all + // possible owners return before resolving the target: adapters may + // use process-local IDs shared by siblings in the same channel. + if self.effort_target_is_busy(request.channel_id) { + self.live_effort.pending.push_back(pending); + continue; + } + let mut targets = Vec::new(); + for (index, slot) in self.agents_mut().iter().enumerate() { + if let Some(agent) = slot { + for (scope, id) in &agent.state.sessions { + if scope.channel_id() == request.channel_id && id == &request.session_id { + targets.push((index, scope.clone())); + } + } + } + } + if targets.len() > 1 { + // Some adapters use process-local session IDs. Never pick an + // arbitrary worker if the reported channel/session is ambiguous. + self.live_effort.finish(request, "unavailable", observer); + continue; + } + let target = targets.pop(); + let Some((index, scope)) = target else { + self.live_effort.finish(request, "stale_session", observer); + continue; + }; + let Some(agent) = self.agents_mut()[index].as_mut() else { + continue; + }; + let Some(mut config) = agent.state.configs.get(&scope).cloned() else { + self.live_effort.finish(request, "unavailable", observer); + continue; + }; + if config.get("effortSessionToken").and_then(Value::as_str) + != Some(request.session_token.to_string().as_str()) + { + self.live_effort.finish(request, "stale_session", observer); + continue; + } + let Some(option) = effort_option(&config) else { + self.live_effort.finish(request, "unsupported", observer); + continue; + }; + let Some(config_id) = option + .get("configId") + .or_else(|| option.get("id")) + .and_then(Value::as_str) + .map(str::to_owned) + else { + self.live_effort.finish(request, "unavailable", observer); + continue; + }; + if !option + .get("options") + .is_some_and(|options| supports_value(options, &request.effort)) + { + self.live_effort.finish(request, "unsupported", observer); + continue; + } + let result = tokio::time::timeout( + APPLY_TIMEOUT, + agent.acp.session_set_config_option( + &request.session_id, + &config_id, + &request.effort, + ), + ) + .await; + let (status, retire) = match result { + Ok(Ok(response)) => { + if let Some(options) = response.get("configOptions").filter(|v| v.is_array()) { + config["configOptions"] = options.clone(); + let confirmed = effort_option(&config) + .and_then(|o| o.get("currentValue")) + .and_then(Value::as_str) + == Some(request.effort.as_str()); + agent.state.remember_effort_config(&scope, &mut config); + if let Some(observer) = observer { + observer.emit( + "session_config_captured", + Some(index), + &context(request), + config, + ); + } + (if confirmed { "applied" } else { "unconfirmed" }, false) + } else { + ("unconfirmed", false) + } + } + Ok(Err(AcpError::AgentError { .. })) => ("rejected", false), + Ok(Err(_)) | Err(_) => ("unconfirmed", true), + }; + if retire { + // A timed-out/poisoned stream must not serve another turn. The + // existing pool maintenance replaces this slot; never fabricate + // successful application or replay this edit into its successor. + self.agents_mut()[index].take(); + } + self.live_effort.finish(request, status, observer); + return; + } + } +} + +#[cfg(test)] +#[path = "live_effort_real_test.rs"] +mod real_test; diff --git a/crates/buzz-acp/src/live_effort_real_test.rs b/crates/buzz-acp/src/live_effort_real_test.rs new file mode 100644 index 00000000000..f19fb1df1eb --- /dev/null +++ b/crates/buzz-acp/src/live_effort_real_test.rs @@ -0,0 +1,181 @@ +//! Opt-in provider acceptance: two real turns around the production effort queue. +//! Requires BUZZ_TEST_CODEX_ADAPTER and an already-authenticated CODEX_HOME. +use super::*; +use crate::acp::AcpClient; +use crate::pool::{OwnedAgent, SessionState, TaskMeta}; +use crate::scope::SessionScope; + +#[tokio::test] +#[ignore = "requires authenticated Codex; performs two real provider turns"] +async fn real_codex_preserves_conversation_across_queued_effort_change() { + let command = std::env::var("BUZZ_TEST_CODEX_ADAPTER").expect("set BUZZ_TEST_CODEX_ADAPTER"); + let cwd = tempfile::tempdir().unwrap(); + let mut acp = AcpClient::spawn(&command, &[], &[], false).await.unwrap(); + let wire_observer = ObserverHandle::in_process(); + acp.set_observer(Some(wire_observer.clone()), 0); + let init = acp.initialize().await.unwrap(); + let session = acp + .session_new_full( + cwd.path().to_str().unwrap(), + vec![], + None, + Some("Buzz live effort acceptance"), + ) + .await + .unwrap(); + let session_id = session.session_id; + let model = std::env::var("BUZZ_TEST_CODEX_MODEL").unwrap_or_else(|_| "gpt-5.6-sol".into()); + let method = crate::acp::resolve_model_switch_method(&session.raw, &model) + .expect("model advertised by the real adapter"); + let config = match method { + crate::acp::ModelSwitchMethod::ConfigOption { + config_id, + option_value, + } => { + acp.session_set_config_option(&session_id, &config_id, &option_value) + .await + } + crate::acp::ModelSwitchMethod::SetModel { model_id } => { + acp.session_set_model(&session_id, &model_id).await + } + } + .unwrap(); + let option = effort_option(&config).expect("native thought_level option"); + assert!(supports_value(&option["options"], "high")); + let mut low = acp + .session_set_config_option(&session_id, option["id"].as_str().unwrap(), "low") + .await + .unwrap(); + assert_eq!(effort_option(&low).unwrap()["currentValue"], "low"); + let session_token = Uuid::new_v4(); + low["effortSessionToken"] = json!(session_token); + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let mut state = SessionState::default(); + state.sessions.insert(scope.clone(), session_id.clone()); + state.configs.insert(scope.clone(), low); + let mut worker = OwnedAgent { + index: 0, + acp, + state, + model_capabilities: None, + desired_model: None, + model_overridden: false, + desired_model_request_id: None, + desired_model_pending_ack: false, + startup_effort: Some("low".into()), + agent_name: "codex".into(), + goose_system_prompt_supported: None, + protocol_version: 2, + }; + let observer = ObserverHandle::in_process(); + let mut pool = AgentPool::from_slots(vec![None]); + let nonce = Uuid::new_v4().to_string(); + let prompt = format!("Do not use any tools. Remember this verification word for the next turn: {nonce}. Reply with exactly REMEMBERED."); + let first_session = session_id.clone(); + let running = tokio::spawn(async move { + let response = worker + .acp + .session_prompt_with_idle_timeout( + &first_session, + &prompt, + Duration::from_secs(90), + Duration::from_secs(180), + ) + .await + .unwrap(); + (worker, response) + }); + tokio::time::timeout(Duration::from_secs(10), async { + while !wire_observer + .snapshot() + .iter() + .any(|event| event.kind == "acp_write" && event.payload["method"] == "session/prompt") + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("first provider prompt was written"); + assert!( + !running.is_finished(), + "queue while the real response is running" + ); + let task_id = running.id(); + let (control_tx, mut control_rx) = tokio::sync::oneshot::channel(); + pool.task_map_mut().insert( + task_id, + TaskMeta { + agent_index: 0, + channel_id: Some(scope.channel_id()), + scope: Some(scope.clone()), + turn_id: "real-provider-turn".into(), + recoverable_batch: None, + control_tx: Some(control_tx), + steer_tx: None, + successful_steer_deliveries: Default::default(), + }, + ); + let request = json!({"requestId":Uuid::new_v4(),"channelId":scope.channel_id(),"sessionId":session_id,"sessionToken":session_token,"effort":"high"}); + pool.queue_live_effort(&request, Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!( + observer.snapshot().last().unwrap().payload["status"], + "queued" + ); + assert!(matches!( + control_rx.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + )); + let (worker, _) = running.await.unwrap(); + let first_text = response_text(&wire_observer, &session_id, 0); + assert!( + first_text.contains("REMEMBERED"), + "unexpected first response: {first_text}" + ); + pool.task_map_mut().remove(&task_id); + pool.return_agent(worker); + assert!( + pool.try_claim(Some(&scope)).is_none(), + "pending edit fences next dispatch" + ); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!( + observer.snapshot().last().unwrap().payload["status"], + "applied" + ); + let mut worker = pool.try_claim(Some(&scope)).unwrap(); + assert_eq!(worker.state.sessions[&scope], session_id); + assert_eq!( + effort_option(&worker.state.configs[&scope]).unwrap()["currentValue"], + "high" + ); + assert_eq!(worker.startup_effort.as_deref(), Some("low")); + let second_start = wire_observer.snapshot().last().unwrap().seq + 1; + let _ = worker.acp.session_prompt_with_idle_timeout(&session_id, "Do not use tools. Reply with only the verification word I gave you in the previous turn.", Duration::from_secs(90), Duration::from_secs(180)).await.unwrap(); + let second_text = response_text(&wire_observer, &session_id, second_start); + assert!( + second_text.contains(&nonce), + "conversation memory lost: {second_text}" + ); + worker.acp.shutdown().await; + println!("REAL_CODEX_EFFORT_OK adapter={} model={} session={} low->high; busy queued; unchanged session; memory retained; startup default low", init["agentInfo"], model, session_id); +} + +fn response_text(observer: &ObserverHandle, session_id: &str, start_seq: u64) -> String { + observer + .snapshot() + .iter() + .filter_map(|event| { + let msg = &event.payload; + (event.seq >= start_seq + && event.kind == "acp_read" + && msg["method"] == "session/update" + && msg["params"]["sessionId"] == session_id + && msg["params"]["update"]["sessionUpdate"] == "agent_message_chunk") + .then(|| msg["params"]["update"]["content"]["text"].as_str()) + .flatten() + }) + .collect() +} diff --git a/crates/buzz-acp/src/live_effort_tests.rs b/crates/buzz-acp/src/live_effort_tests.rs new file mode 100644 index 00000000000..413ae15ac00 --- /dev/null +++ b/crates/buzz-acp/src/live_effort_tests.rs @@ -0,0 +1,529 @@ +use super::*; +use crate::acp::AcpClient; +use crate::pool::{OwnedAgent, SessionState, TaskMeta}; +use crate::scope::SessionScope; + +fn config(value: &str) -> Value { + json!({"effortSessionToken":"fbf7259f-74df-4859-8b53-c0d8b77fb21e","configOptions":[{"id":"native-reasoning","category":"thought_level","type":"select", + "currentValue":value,"options":[{"value":"low"},{"value":"high"}]}],"relayUrl":"ws://test.invalid"}) +} + +async fn agent(scope: &SessionScope, file: &std::path::Path, behavior: &str) -> OwnedAgent { + let script = r#"import json,sys,time +for line in sys.stdin: + r=json.loads(line) + with open(sys.argv[1],'a') as f: f.write(json.dumps(r)+'\n') + if sys.argv[2]=='hang': time.sleep(60) + if sys.argv[2]=='reject': + print(json.dumps({'jsonrpc':'2.0','id':r['id'],'error':{'code':-32602,'message':'unsupported effort'}}),flush=True) + else: + value=r['params']['value'] if sys.argv[2]=='apply' else 'low' + opts=[{'id':'native-reasoning','category':'thought_level','type':'select','currentValue':value,'options':[{'value':'low'},{'value':'high'}]}] + print(json.dumps({'jsonrpc':'2.0','id':r['id'],'result':{'configOptions':opts}}),flush=True) +"#; + let acp = AcpClient::spawn( + "python3", + &[ + "-u".into(), + "-c".into(), + script.into(), + file.to_string_lossy().to_string(), + behavior.into(), + ], + &[], + false, + ) + .await + .unwrap(); + let mut state = SessionState::default(); + state.sessions.insert(scope.clone(), "same-session".into()); + state.configs.insert(scope.clone(), config("low")); + state.turn_counts.insert(scope.clone(), 3); + OwnedAgent { + index: 0, + acp, + state, + model_capabilities: None, + desired_model: None, + model_overridden: false, + desired_model_request_id: None, + desired_model_pending_ack: false, + startup_effort: Some("low".into()), + agent_name: "effort-test".into(), + goose_system_prompt_supported: None, + protocol_version: 2, + } +} + +fn request(channel: Uuid, session: &str, effort: &str) -> Value { + json!({"type":"switch_effort","requestId":Uuid::new_v4(),"channelId":channel, + "sessionId":session,"sessionToken":"fbf7259f-74df-4859-8b53-c0d8b77fb21e","effort":effort}) +} + +fn status(observer: &ObserverHandle) -> String { + observer + .snapshot() + .iter() + .rev() + .find(|e| e.kind == "control_result") + .unwrap() + .payload["status"] + .as_str() + .unwrap() + .into() +} + +#[tokio::test] +async fn applies_native_rpc_to_same_session_and_preserves_sibling_defaults_and_history() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let channel = Uuid::new_v4(); + let scope = SessionScope::Thread { + channel_id: channel, + root_event_id: "a".repeat(64), + }; + let sibling = SessionScope::Thread { + channel_id: channel, + root_event_id: "b".repeat(64), + }; + let mut worker = agent(&scope, &file, "apply").await; + worker + .state + .sessions + .insert(sibling.clone(), "sibling-session".into()); + worker.state.configs.insert(sibling.clone(), config("low")); + let mut pool = AgentPool::from_slots(vec![Some(worker)]); + let observer = ObserverHandle::in_process(); + let pick = request(channel, "same-session", "high"); + pool.queue_live_effort(&pick, Some(&observer)); + assert_eq!(status(&observer), "queued"); + assert!( + pool.try_claim(Some(&scope)).is_none(), + "next turn must wait for effort" + ); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "applied"); + let worker = pool.try_claim(Some(&scope)).unwrap(); + assert_eq!(worker.state.sessions[&scope], "same-session"); + assert_eq!(worker.state.turn_counts[&scope], 3); + assert_eq!(worker.startup_effort.as_deref(), Some("low")); + assert_eq!( + worker.state.configs[&scope]["configOptions"][0]["currentValue"], + "high" + ); + assert_eq!( + worker.state.configs[&sibling]["configOptions"][0]["currentValue"], + "low" + ); + let captured = observer + .snapshot() + .into_iter() + .find(|e| e.kind == "session_config_captured") + .unwrap(); + assert_eq!(captured.session_id.as_deref(), Some("same-session")); + let rpc: Value = serde_json::from_str(std::fs::read_to_string(&file).unwrap().trim()).unwrap(); + assert_eq!(rpc["method"], "session/set_config_option"); + assert_eq!( + rpc["params"], + json!({"sessionId":"same-session","configId":"native-reasoning","value":"high"}) + ); + pool.return_agent(worker); + pool.queue_live_effort(&pick, Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!( + std::fs::read_to_string(file).unwrap().lines().count(), + 1, + "replay cannot execute twice" + ); +} + +#[tokio::test] +async fn busy_turn_queues_without_cancelling_then_applies_before_next_claim() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let channel = Uuid::new_v4(); + let scope = SessionScope::Conversation { + channel_id: channel, + }; + let worker = agent(&scope, &file, "apply").await; + let mut pool = AgentPool::from_slots(vec![None]); + let observer = ObserverHandle::in_process(); + let task = tokio::spawn(std::future::pending::<()>()); + let id = task.id(); + let (control_tx, mut control_rx) = tokio::sync::oneshot::channel(); + pool.task_map_mut().insert( + id, + TaskMeta { + agent_index: 0, + channel_id: Some(channel), + scope: Some(scope.clone()), + turn_id: "busy".into(), + recoverable_batch: None, + control_tx: Some(control_tx), + steer_tx: None, + successful_steer_deliveries: Default::default(), + }, + ); + pool.queue_live_effort(&request(channel, "same-session", "high"), Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert!( + !file.exists(), + "must not write while a response owns the stream" + ); + assert!( + matches!( + control_rx.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + ), + "must not cancel the response" + ); + pool.task_map_mut().remove(&id); + task.abort(); + pool.return_agent(worker); + assert!(pool.try_claim(Some(&scope)).is_none()); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "applied"); + assert!(pool.try_claim(Some(&scope)).is_some()); +} + +#[tokio::test] +async fn rejects_unsupported_and_stale_targets_without_any_rpc() { + for (session, effort, expected) in [ + ("same-session", "ultra", "unsupported"), + ("expired-session", "high", "stale_session"), + ] { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let channel = Uuid::new_v4(); + let scope = SessionScope::Conversation { + channel_id: channel, + }; + let mut pool = AgentPool::from_slots(vec![Some(agent(&scope, &file, "apply").await)]); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort(&request(channel, session, effort), Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), expected); + assert!(!file.exists()); + assert_eq!( + pool.agents_mut()[0].as_ref().unwrap().state.configs[&scope], + config("low") + ); + } +} + +#[tokio::test] +async fn adapter_rejection_preserves_session_and_mismatched_ack_is_unconfirmed() { + for (behavior, expected) in [("reject", "rejected"), ("mismatch", "unconfirmed")] { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let channel = Uuid::new_v4(); + let scope = SessionScope::Conversation { + channel_id: channel, + }; + let mut pool = AgentPool::from_slots(vec![Some(agent(&scope, &file, behavior).await)]); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort(&request(channel, "same-session", "high"), Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), expected); + let worker = pool.try_claim(Some(&scope)).unwrap(); + assert_eq!(worker.state.sessions[&scope], "same-session"); + assert_eq!( + worker.state.configs[&scope]["configOptions"][0]["currentValue"], + "low" + ); + } +} + +#[tokio::test] +async fn bounds_pending_controls_and_releases_expired_claim_fences() { + let mut pool = AgentPool::from_slots(vec![]); + let observer = ObserverHandle::in_process(); + for _ in 0..CAPACITY { + pool.queue_live_effort(&request(Uuid::new_v4(), "session", "high"), Some(&observer)); + } + pool.queue_live_effort( + &request(Uuid::new_v4(), "overflow", "high"), + Some(&observer), + ); + assert_eq!(status(&observer), "busy"); + assert_eq!(pool.live_effort.pending.len(), CAPACITY); + for item in &mut pool.live_effort.pending { + item.received = Instant::now() - EXPIRY - Duration::from_secs(1); + } + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "expired"); + assert!(pool.live_effort.pending.is_empty()); +} + +#[tokio::test] +async fn timeout_retires_the_stream_without_claiming_application() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let channel = Uuid::new_v4(); + let scope = SessionScope::Conversation { + channel_id: channel, + }; + let mut pool = AgentPool::from_slots(vec![Some(agent(&scope, &file, "hang").await)]); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort(&request(channel, "same-session", "high"), Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "unconfirmed"); + assert!(pool.agents_mut()[0].is_none()); +} + +#[tokio::test] +async fn pending_edit_holds_exact_owner_without_blocking_another_idle_worker() { + let dir = tempfile::tempdir().unwrap(); + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let sibling = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let owner = agent(&scope, &dir.path().join("owner"), "apply").await; + let mut other = agent(&sibling, &dir.path().join("sibling"), "apply").await; + other.index = 1; + let mut pool = AgentPool::from_slots(vec![Some(owner), Some(other)]); + pool.queue_live_effort(&request(scope.channel_id(), "same-session", "high"), None); + assert!( + pool.try_claim(Some(&scope)).is_none(), + "do not fall through and duplicate the held session on another worker" + ); + assert_eq!(pool.try_claim(Some(&sibling)).unwrap().index, 1); + pool.apply_pending_effort(None).await; + assert_eq!(pool.try_claim(Some(&scope)).unwrap().index, 0); + assert!( + !dir.path().join("sibling").exists(), + "never mutate the sibling adapter" + ); +} + +#[tokio::test] +async fn duplicate_adapter_session_ids_do_not_select_an_arbitrary_sibling() { + let dir = tempfile::tempdir().unwrap(); + let channel = Uuid::new_v4(); + let first = SessionScope::Thread { + channel_id: channel, + root_event_id: "a".repeat(64), + }; + let second = SessionScope::Thread { + channel_id: channel, + root_event_id: "b".repeat(64), + }; + let owner = agent(&first, &dir.path().join("first"), "apply").await; + let mut other = agent(&second, &dir.path().join("second"), "apply").await; + other.index = 1; + let mut pool = AgentPool::from_slots(vec![Some(owner), Some(other)]); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort(&request(channel, "same-session", "high"), Some(&observer)); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "unavailable"); + assert!(!dir.path().join("first").exists()); + assert!(!dir.path().join("second").exists()); + assert!(pool.try_claim(Some(&first)).is_some()); + assert!(pool.try_claim(Some(&second)).is_some()); +} + +#[tokio::test] +async fn encrypted_owner_route_rejects_foreign_stale_and_tampered_effort_controls() { + let owner = nostr::Keys::generate(); + let agent = nostr::Keys::generate(); + let outsider = nostr::Keys::generate(); + for case in ["owner", "foreign", "stale", "tampered"] { + let sender = if case == "foreign" { &outsider } else { &owner }; + let pick = request(Uuid::new_v4(), "session", "high"); + let encrypted = + crate::encrypt_observer_payload(sender, &agent.public_key(), &pick).unwrap(); + let mut builder = buzz_sdk::build_agent_observer_frame( + &agent.public_key().to_hex(), + &agent.public_key().to_hex(), + "control", + &encrypted, + ) + .unwrap(); + if case == "stale" { + builder = builder.custom_created_at(nostr::Timestamp::from( + nostr::Timestamp::now().as_secs() - 301, + )); + } + let mut event = builder.sign_with_keys(sender).unwrap(); + if case == "tampered" { + event.content.push('x'); + } + let observer = ObserverHandle::in_process(); + let mut pool = AgentPool::from_slots(vec![]); + let (publisher, _rx) = crate::relay::RelayEventPublisher::test_pair(); + crate::handle_relay_observer_control_event( + &agent, + event, + &mut pool, + Some(&observer), + &owner.public_key().to_hex(), + publisher, + ); + assert_eq!( + pool.live_effort.pending.len(), + usize::from(case == "owner"), + "{case}" + ); + if case == "owner" { + assert_eq!(status(&observer), "queued"); + } else { + assert!(observer.snapshot().is_empty(), "{case}"); + } + } +} + +#[tokio::test] +async fn busy_duplicate_session_cannot_make_an_idle_sibling_look_unique() { + let dir = tempfile::tempdir().unwrap(); + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let sibling = SessionScope::Thread { + channel_id: scope.channel_id(), + root_event_id: "a".repeat(64), + }; + let owner = agent(&scope, &dir.path().join("owner"), "apply").await; + let mut busy = agent(&sibling, &dir.path().join("busy"), "apply").await; + busy.index = 1; + let mut pool = AgentPool::from_slots(vec![Some(owner), None]); + let task = tokio::spawn(std::future::pending::<()>()); + pool.task_map_mut().insert( + task.id(), + TaskMeta { + agent_index: 1, + channel_id: Some(scope.channel_id()), + scope: Some(sibling), + turn_id: "busy".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: Default::default(), + }, + ); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort( + &request(scope.channel_id(), "same-session", "high"), + Some(&observer), + ); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "queued"); + assert!(!dir.path().join("owner").exists()); + pool.task_map_mut().remove(&task.id()); + task.abort(); + pool.return_agent(busy); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "unavailable"); + assert!(!dir.path().join("owner").exists()); + assert!(!dir.path().join("busy").exists()); +} + +#[tokio::test] +async fn native_config_id_and_grouped_values_use_the_reported_wire_shape() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let mut worker = agent(&scope, &file, "apply").await; + let option = &mut worker.state.configs.get_mut(&scope).unwrap()["configOptions"][0]; + option.as_object_mut().unwrap().remove("id"); + option["configId"] = json!("spec-reasoning"); + option["options"] = json!([{"name":"Levels", "options":[{"value":"high"}]}]); + let mut pool = AgentPool::from_slots(vec![Some(worker)]); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort( + &request(scope.channel_id(), "same-session", "high"), + Some(&observer), + ); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "applied"); + let rpc: Value = serde_json::from_str( + std::fs::read_to_string(file) + .unwrap() + .lines() + .next() + .unwrap(), + ) + .unwrap(); + assert_eq!(rpc["params"]["configId"], "spec-reasoning"); +} + +#[test] +fn configuration_snapshots_follow_every_session_invalidation_boundary() { + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + for boundary in ["scope", "channel", "all"] { + let mut state = SessionState::default(); + state.sessions.insert(scope.clone(), "session".into()); + state.configs.insert(scope.clone(), config("high")); + match boundary { + "scope" => { + state.invalidate_scope(&scope); + } + "channel" => { + state.invalidate_channel(&scope.channel_id()); + } + _ => state.invalidate_all(), + } + assert!(state.configs.is_empty(), "{boundary}"); + assert!(state.sessions.is_empty(), "{boundary}"); + } +} + +#[test] +fn native_snapshot_capacity_keeps_existing_targets_and_reopens_after_invalidation() { + let mut state = SessionState::default(); + for _ in 0..CONFIG_CAPACITY { + state.remember_effort_config( + &SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }, + &mut config("low"), + ); + } + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let mut snapshot = config("low"); + state.remember_effort_config(&scope, &mut snapshot); + assert_eq!(snapshot["liveEffortSwitching"], false); + assert!(!state.configs.contains_key(&scope)); + let retained = state.configs.keys().next().unwrap().clone(); + state.remember_effort_config(&retained, &mut config("high")); + assert_eq!( + state.configs[&retained]["configOptions"][0]["currentValue"], + "high" + ); + state.invalidate_scope(&retained); + state.remember_effort_config(&scope, &mut snapshot); + assert_eq!(snapshot["liveEffortSwitching"], true); + assert_eq!(state.configs.len(), CONFIG_CAPACITY); + snapshot["oversized"] = json!("x".repeat(MAX_CONFIG_BYTES)); + state.remember_effort_config(&scope, &mut snapshot); + assert_eq!(snapshot["liveEffortSwitching"], false); + assert!(!state.configs.contains_key(&scope)); +} + +#[tokio::test] +async fn a_reused_adapter_session_id_cannot_receive_an_old_conversations_edit() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("rpc"); + let scope = SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let mut worker = agent(&scope, &file, "apply").await; + worker.state.configs.get_mut(&scope).unwrap()["effortSessionToken"] = json!(Uuid::new_v4()); + let mut pool = AgentPool::from_slots(vec![Some(worker)]); + let observer = ObserverHandle::in_process(); + pool.queue_live_effort( + &request(scope.channel_id(), "same-session", "high"), + Some(&observer), + ); + pool.apply_pending_effort(Some(&observer)).await; + assert_eq!(status(&observer), "stale_session"); + assert!(!file.exists()); + assert!(pool.try_claim(Some(&scope)).is_some()); +} diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 06383d456d3..c8a382eb87a 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -118,6 +118,8 @@ pub struct ChannelDeliveryState { /// spawning a real agent subprocess. #[derive(Default)] pub struct SessionState { + /// Last accepted native configuration, scoped exactly like the ACP session. + pub configs: HashMap, /// session scope → session_id pub sessions: HashMap, pub heartbeat_session: Option, @@ -161,6 +163,7 @@ impl SessionState { /// Invalidate a single session scope's session and turn counter. /// Returns `true` if the scope had an active session. pub fn invalidate_scope(&mut self, scope: &SessionScope) -> bool { + self.configs.remove(scope); self.turn_counts.remove(scope); self.core_sections.remove(scope); self.canvas_sections.remove(scope); @@ -175,6 +178,7 @@ impl SessionState { let scopes: Vec = self .sessions .keys() + .chain(self.configs.keys()) .chain(self.turn_counts.keys()) .chain(self.core_sections.keys()) .chain(self.canvas_sections.keys()) @@ -196,6 +200,7 @@ impl SessionState { /// Invalidate all sessions and turn counters (e.g. after agent exit). pub fn invalidate_all(&mut self) { self.sessions.clear(); + self.configs.clear(); self.turn_counts.clear(); self.heartbeat_session = None; self.heartbeat_turn_count = 0; @@ -325,6 +330,7 @@ impl OwnedAgent { /// (running inside a spawned task). The `task_map` tracks in-flight /// tasks for panic recovery. pub struct AgentPool { + pub(crate) live_effort: crate::live_effort::LiveEffortQueue, agents: Vec>, result_tx: mpsc::UnboundedSender, result_rx: mpsc::UnboundedReceiver, @@ -831,6 +837,7 @@ impl AgentPool { task_map: HashMap::new(), session_owners: HashMap::new(), held_since: HashMap::new(), + live_effort: Default::default(), } } @@ -907,6 +914,14 @@ impl AgentPool { /// /// Returns `None` if all agents are checked out. pub fn try_claim(&mut self, scope: Option<&SessionScope>) -> Option { + // Hold the worker at its response boundary until its exact-session edit + // settles. Other workers remain available to serve their conversations. + let effort_ready = + |agent: &OwnedAgent| { + !agent.state.sessions.iter().any(|(scope, session)| { + self.live_effort.has_session(scope.channel_id(), session) + }) + }; // Pass 1: prefer agent with existing session for this scope. if let Some(scope) = scope { let idx = self.agents.iter().position(|slot| { @@ -915,12 +930,18 @@ impl AgentPool { .unwrap_or(false) }); if let Some(i) = idx { + if !self.agents[i].as_ref().is_some_and(effort_ready) { + return None; + } return self.agents[i].take(); } } // Pass 2: first idle agent. - let idx = self.agents.iter().position(|slot| slot.is_some()); + let idx = self + .agents + .iter() + .position(|slot| slot.as_ref().is_some_and(effort_ready)); idx.map(|i| self.agents[i].take().unwrap()) } @@ -1161,6 +1182,15 @@ impl AgentPool { scopes.any(|scope| scope != first) } + pub(crate) fn effort_target_is_busy(&self, channel_id: Uuid) -> bool { + self.task_map.values().any(|task| { + task.channel_id == Some(channel_id) + || self.session_owners.iter().any(|(scope, owner)| { + scope.channel_id() == channel_id && *owner == task.agent_index + }) + }) + } + /// Idle-path model switch: set `desired_model` on the idle agent for /// `channel_id` and invalidate its exact session scope so the next turn /// re-creates that session under the new model. @@ -1579,10 +1609,13 @@ async fn create_session_and_apply_model( } opts }; - agent.acp.observe( - "session_config_captured", - serde_json::json!({ + let mut captured_config = serde_json::json!({ "configOptions": config_options_for_cache, + "effortSessionToken": Uuid::new_v4(), + "liveEffortSwitching": channel.scope.is_some(), + "conversationLabel": channel.name, + "conversationId": channel.scope.map(|scope| scope.root_event_id().map(str::to_owned) + .unwrap_or_else(|| scope.channel_id().to_string())), "modes": resp.raw.get("modes").cloned().unwrap_or(serde_json::Value::Null), // `models` must come from the SAME snapshot as configOptions — the // post-switch snapshot on a successful switch, session/new otherwise. @@ -1596,8 +1629,29 @@ async fn create_session_and_apply_model( // Pair identity for the desktop session-config cache, which is // keyed by (agent, relay) like the lifecycle frames. "relayUrl": ctx.relay_url, - }), - ); + }); + if let Some(scope) = channel.scope { + agent + .state + .remember_effort_config(scope, &mut captured_config); + } + if let (Some(scope), Some(observer)) = (channel.scope, agent.acp.observer_handle()) { + observer.emit( + "session_config_captured", + Some(agent.index), + &observer::ObserverContext { + channel_id: Some(scope.channel_id().to_string()), + session_id: Some(resp.session_id.clone()), + turn_id: None, + started_at: None, + }, + captured_config, + ); + } else { + agent + .acp + .observe("session_config_captured", captured_config); + } // Apply permission mode if not the agent's built-in default AND the agent // advertises the requested mode in session/new. Agents that don't support diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index aad2580dad0..ee03e0040da 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -1,5 +1,8 @@ import { defineConfig, devices } from "@playwright/test"; +const port = process.env.BUZZ_E2E_PORT ?? "4173"; +const baseURL = `http://127.0.0.1:${port}`; + export default defineConfig({ testDir: "./tests/e2e", timeout: 30_000, @@ -10,7 +13,7 @@ export default defineConfig({ ["html", { open: "never", outputFolder: "playwright-report" }], ], use: { - baseURL: "http://127.0.0.1:4173", + baseURL, screenshot: "only-on-failure", trace: "on-first-retry", video: "retain-on-failure", @@ -149,6 +152,7 @@ export default defineConfig({ "**/deep-link-invite.spec.ts", "**/invite-link-copy.spec.ts", "**/global-agent-config-screenshots.spec.ts", + "**/conversation-effort.spec.ts", "**/doctor-states.spec.ts", "**/onboarding-avatar-skip.spec.ts", "**/onboarding-backup.spec.ts", @@ -211,9 +215,9 @@ export default defineConfig({ }, ], webServer: { - command: "python3 -m http.server 4173 -d dist", + command: `python3 -m http.server ${port} -d dist`, cwd: ".", reuseExistingServer: !process.env.CI, - url: "http://127.0.0.1:4173", + url: baseURL, }, }); diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index dfb9c0ed494..b481dd0cde5 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -221,7 +221,7 @@ with a TypeScript lookup table or an id comparison in a component. hosting location, availability, or permission. Keep all identity surfaces on the shared provenance context, without per-row directory subscriptions. See [the provenance contract](../../../../docs/agent-management-provenance.md). -14. **Thinking effort has two surfaces: a local-only WRITE control and a +14. **Saved thinking effort has a local-only WRITE control and a read-only two-facts DISPLAY.** The write control is `EffortPickerField` (`ui/EffortPickerField.tsx`), a self-contained section component mounted in `AgentInstanceEditDialog` beside the Model block. It is **Save-gated, not @@ -245,17 +245,21 @@ with a TypeScript lookup table or an id comparison in a component. spawn will launch with) and, when a running ACP session differs, `field.overriddenValue` struck through (the live session's current effort). No component owns "configured vs current" logic; the reader's canonical tier - ordering feeds both facts. Do not add a second effort write path or restate + ordering feeds both facts. Do not add a second saved-effort write path or restate the two-facts logic in a component. - **Cut invariant — live mid-conversation effort machinery was deliberately - removed.** Effort is spawn-scoped only: the worker holds one `startup_effort` - read from `BUZZ_ACP_EFFORT_LEVEL` and applies it once at session creation - (`apply_startup_effort` in `buzz-acp/src/pool.rs`); there is no pool-level - effort authority, no live effort switching, and no effort-ack frame. Do not - reintroduce a live effort-switch RPC, a pool effort field, or a - mid-conversation effort control without a plan ruling. The archived live-effort - machinery lives on `archive/claude-config-gaps-live-effort` for reference only. + **Conversation overrides are separate from saved effort.** The activity pane + exposes `ConversationEffortPicker` only to the owner and only when an exact + session snapshot advertises `liveEffortSwitching` and native `thought_level` + options. The owner selects the conversation explicitly when several are + available. The encrypted observer control carries channel, session, request + ID, session token, and effort; it never writes a default. `buzz-acp` queues the edit until + the response ends, fences its worker before the next claim, and reports + `applied` only after the native response confirms the value. A missing result + is unconfirmed; a queued result must not optimistically update the selection. + `startup_effort` remains the sole saved default applied at session creation. + See [the conversation-effort design](../../../../docs/conversation-effort.md) + for the scope of this follow-up to the live control deferred in #4557. 15. **The persona `description` is public display metadata.** It is optional, capped at 280 characters, and validated through the shared visible-text diff --git a/desktop/src/features/agents/lib/conversationEffort.test.mjs b/desktop/src/features/agents/lib/conversationEffort.test.mjs new file mode 100644 index 00000000000..be80c75d6bf --- /dev/null +++ b/desktop/src/features/agents/lib/conversationEffort.test.mjs @@ -0,0 +1,182 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + conversationEfforts, + matchingEffortStatus, + retainSessionConfigs, +} from "./conversationEffort.ts"; + +const native = ( + value = "medium", + options = [ + { value: "low", name: "Low" }, + { value: "medium", name: "Medium" }, + ], +) => ({ + liveEffortSwitching: true, + effortSessionToken: "fbf7259f-74df-4859-8b53-c0d8b77fb21e", + configOptions: [ + { + id: "reasoning_effort", + category: "thought_level", + type: "select", + currentValue: value, + options, + }, + ], +}); +const frame = (patch = {}) => ({ + seq: 1, + timestamp: "2026-09-06T00:00:00Z", + kind: "session_config_captured", + agentIndex: 0, + channelId: "channel", + sessionId: "session", + turnId: null, + payload: native(), + ...patch, +}); + +test("effort inventory is scoped to exact sessions and reads native values", () => { + const choices = conversationEfforts( + [ + frame(), + frame({ sessionId: "sibling", payload: native("low") }), + frame({ channelId: "foreign" }), + ], + "channel", + ); + assert.equal(choices.length, 2); + assert.equal(choices.find((c) => c.sessionId === "session").value, "medium"); + assert.deepEqual( + choices[0].options.map((o) => o.value), + ["low", "medium"], + ); +}); +test("newer unsupported snapshot removes control instead of reviving stale capabilities", () => { + assert.deepEqual( + conversationEfforts( + [frame({ seq: 2, payload: { configOptions: [] } }), frame()], + "channel", + ), + [], + ); +}); +test("replayed snapshots cannot overwrite newer applied value; grouped choices remain exact", () => { + const choices = conversationEfforts( + [ + frame({ + seq: 2, + payload: native("ultra", [ + { name: "Advanced", options: [{ value: "ultra", name: "Ultra" }] }, + ]), + }), + frame(), + ], + "channel", + ); + assert.equal(choices[0].value, "ultra"); + assert.deepEqual(choices[0].options, [{ value: "ultra", label: "Ultra" }]); +}); +test("an acknowledgement must match type, request, channel, session, and effort", () => { + const request = { + requestId: "new", + channelId: "channel", + sessionId: "session", + sessionToken: "current-session", + effort: "high", + }; + const result = { ...request, type: "switch_effort", status: "applied" }; + assert.equal(matchingEffortStatus(result, request), "applied"); + for (const [key, value] of Object.entries({ + requestId: "old", + sessionToken: "previous-session", + channelId: "foreign", + sessionId: "sibling", + effort: "low", + type: "switch_model", + })) { + assert.equal( + matchingEffortStatus({ ...result, [key]: value }, request), + null, + key, + ); + } + assert.equal( + matchingEffortStatus({ ...result, status: "queued" }, request), + "queued", + ); +}); + +test("old harnesses do not advertise a live control", () => { + assert.deepEqual( + conversationEfforts( + [frame({ payload: { ...native(), liveEffortSwitching: undefined } })], + "channel", + ), + [], + ); +}); +test("retained config survives transcript eviction and strips unrelated inventories", () => { + const retained = retainSessionConfigs( + [], + [frame({ payload: { ...native(), models: ["large catalog"] } })], + ); + assert.equal( + retainSessionConfigs(retained, [frame({ kind: "turn_started" })]), + retained, + ); + assert.equal(conversationEfforts(retained, "channel")[0].value, "medium"); + assert.equal(retained[0].payload.models, undefined); +}); +test("rotation supersedes an old target and late replay cannot resurrect it", () => { + const previous = frame({ + payload: { ...native(), conversationId: "thread-root" }, + }); + const next = frame({ + sessionId: "replacement", + seq: 2, + payload: { ...native("low"), conversationId: "thread-root" }, + }); + const retained = retainSessionConfigs(retainSessionConfigs([], [previous]), [ + next, + ]); + const choices = conversationEfforts([...retained, previous], "channel"); + assert.deepEqual( + choices.map((c) => c.sessionId), + ["replacement"], + ); + assert.equal(retainSessionConfigs(retained, [previous]), retained); +}); +test("retained sessions are bounded", () => { + const retained = retainSessionConfigs( + [], + Array.from({ length: 200 }, (_, i) => + frame({ sessionId: `session-${i}`, seq: i }), + ), + ); + assert.equal(retained.length, 128); + assert.equal(retained[0].seq, 72); +}); + +test("profile-wide activity keeps exact channel targets even when adapter IDs coincide", () => { + const choices = conversationEfforts( + [ + frame(), + frame({ + channelId: "other", + payload: { ...native("low"), conversationLabel: "Other channel" }, + }), + ], + null, + ); + assert.equal(choices.length, 2); + assert.deepEqual(choices.map((c) => c.channelId).sort(), [ + "channel", + "other", + ]); + assert.equal( + choices.find((c) => c.channelId === "other").label, + "Other channel", + ); +}); diff --git a/desktop/src/features/agents/lib/conversationEffort.ts b/desktop/src/features/agents/lib/conversationEffort.ts new file mode 100644 index 00000000000..76ed1067954 --- /dev/null +++ b/desktop/src/features/agents/lib/conversationEffort.ts @@ -0,0 +1,150 @@ +import type { ControlResultFrame } from "@/shared/api/types"; +import type { ObserverEvent } from "../ui/agentSessionTypes"; +import { compareObserverEvents } from "../observerEventOrdering"; + +type NativeOption = { value: string; label: string }; +export type ConversationEffort = { + channelId: string; + label: string | null; + sessionId: string; + sessionToken: string; + timestamp: string; + value: string; + options: NativeOption[]; +}; + +function object(value: unknown): Record | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} + +function values(input: unknown): NativeOption[] { + if (!Array.isArray(input)) return []; + return input.flatMap((item) => { + const option = object(item); + if (!option) return []; + if (typeof option.value === "string") { + return [ + { + value: option.value, + label: typeof option.name === "string" ? option.name : option.value, + }, + ]; + } + return values(option.options); + }); +} + +/** Keep small native snapshots independent of the bounded activity transcript. + * A new session for the same conversation supersedes the previous target. */ +export function retainSessionConfigs( + current: readonly ObserverEvent[], + arrivals: readonly ObserverEvent[], +): readonly ObserverEvent[] { + const captures = arrivals.filter( + (event) => + event.kind === "session_config_captured" && + event.channelId && + event.sessionId, + ); + if (captures.length === 0) return current; + const incoming = new Set(captures); + const latest = new Map(); + let changed = false; + for (const event of [...current, ...captures]) { + if ( + !event.channelId || + !event.sessionId || + event.kind !== "session_config_captured" + ) + continue; + const config = object(event.payload); + const key = JSON.stringify([ + event.channelId, + config?.conversationId ?? event.sessionId, + ]); + const previous = latest.get(key); + if (previous && compareObserverEvents(event, previous) <= 0) continue; + latest.set(key, { + ...event, + payload: { + conversationId: config?.conversationId, + effortSessionToken: config?.effortSessionToken, + conversationLabel: config?.conversationLabel, + liveEffortSwitching: config?.liveEffortSwitching, + configOptions: Array.isArray(config?.configOptions) + ? config.configOptions.filter( + (option) => object(option)?.category === "thought_level", + ) + : [], + }, + }); + if (incoming.has(event)) changed = true; + } + return changed + ? [...latest.values()].sort(compareObserverEvents).slice(-128) + : current; +} + +/** Read only exact-session native snapshots; a later empty snapshot removes support. */ +export function conversationEfforts( + events: readonly ObserverEvent[], + channelId: string | null, +): ConversationEffort[] { + return retainSessionConfigs([], events) + .flatMap((event) => { + if (channelId && event.channelId !== channelId) return []; + const config = object(event.payload); + if ( + config?.liveEffortSwitching !== true || + typeof config.effortSessionToken !== "string" || + !config.effortSessionToken + ) + return []; + if (!Array.isArray(config?.configOptions)) return []; + const option = config.configOptions + .map(object) + .find((o) => o?.category === "thought_level" && o.type === "select"); + if (!option || typeof option.currentValue !== "string") return []; + const options = values(option.options); + if (options.length === 0) return []; + return [ + { + channelId: event.channelId as string, + label: + typeof config.conversationLabel === "string" + ? config.conversationLabel + : null, + sessionId: event.sessionId as string, + sessionToken: config.effortSessionToken, + timestamp: event.timestamp, + value: option.currentValue, + options, + }, + ]; + }) + .sort((a, b) => b.timestamp.localeCompare(a.timestamp)); +} + +export type EffortRequest = { + requestId: string; + sessionId: string; + sessionToken: string; + channelId: string; + effort: string; +}; + +export function matchingEffortStatus( + frame: ControlResultFrame, + request: EffortRequest, +): string | null { + return frame.type === "switch_effort" && + frame.requestId === request.requestId && + frame.sessionId === request.sessionId && + frame.sessionToken === request.sessionToken && + frame.channelId === request.channelId && + frame.effort === request.effort + ? frame.status + : null; +} diff --git a/desktop/src/features/agents/observerEventOrdering.ts b/desktop/src/features/agents/observerEventOrdering.ts new file mode 100644 index 00000000000..8fc0e1e14e6 --- /dev/null +++ b/desktop/src/features/agents/observerEventOrdering.ts @@ -0,0 +1,17 @@ +import type { ObserverEvent } from "./ui/agentSessionTypes"; + +export function compareObserverEvents( + left: ObserverEvent, + right: ObserverEvent, +) { + const leftTime = Date.parse(left.timestamp); + const rightTime = Date.parse(right.timestamp); + if (Number.isFinite(leftTime) && Number.isFinite(rightTime)) { + const timeDiff = leftTime - rightTime; + if (timeDiff !== 0) { + return timeDiff; + } + } + + return left.seq - right.seq; +} diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index b945496380f..2b1dd41cecc 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -1,4 +1,7 @@ +import { compareObserverEvents } from "./observerEventOrdering"; +export { compareObserverEvents } from "./observerEventOrdering"; import * as React from "react"; +import { retainSessionConfigs } from "./lib/conversationEffort"; import { subscribeToAgentObserverFrames } from "@/shared/api/observerRelay"; import type { RelayEvent, ManagedAgent } from "@/shared/api/types"; @@ -66,6 +69,12 @@ type AgentObserverStoreListener = (update?: AgentObserverStoreUpdate) => void; const listeners = new Set(); const eventsByAgent = new Map(); +const sessionConfigsByAgent = new Map(); +export function getAgentSessionConfigs( + pubkey: string, +): readonly ObserverEvent[] { + return sessionConfigsByAgent.get(normalizePubkey(pubkey)) ?? EMPTY_EVENTS; +} const transcriptByAgent = new Map(); const snapshotByAgent = new Map(); @@ -280,6 +289,10 @@ function appendAgentEvents( if (added.length === 0) return null; const sortedAdded = [...added].sort(compareObserverEvents); + sessionConfigsByAgent.set( + key, + retainSessionConfigs(getAgentSessionConfigs(key), sortedAdded), + ); const sorted = allAtEnd ? [...current, ...sortedAdded] : [...current, ...sortedAdded].sort(compareObserverEvents); @@ -407,22 +420,6 @@ export function getArchivedChannelEvents( ); } -export function compareObserverEvents( - left: ObserverEvent, - right: ObserverEvent, -) { - const leftTime = Date.parse(left.timestamp); - const rightTime = Date.parse(right.timestamp); - if (Number.isFinite(leftTime) && Number.isFinite(rightTime)) { - const timeDiff = leftTime - rightTime; - if (timeDiff !== 0) { - return timeDiff; - } - } - - return left.seq - right.seq; -} - /** * Returns true if `candidate` sorts strictly after `stored` using the same * two-key ordering as `compareObserverEvents`: later timestamp wins; equal @@ -911,6 +908,14 @@ export function injectObserverEventsForE2E( } } +/** E2E-only: use the live ingestion path, including correlated control results. */ +export function injectLiveObserverEventsForE2E( + agentPubkey: string, + events: ObserverEvent[], +) { + processLiveObserverEvents(agentPubkey, events); +} + /** * Synchronize the observer store with a sorted buffer of events for one agent. * Used by test harnesses and replay bridges that already hold decoded frames. @@ -932,6 +937,7 @@ export function resetAgentObserverStore() { startPromise = null; eventProcessingQueue = Promise.resolve(); eventsByAgent.clear(); + sessionConfigsByAgent.clear(); transcriptByAgent.clear(); evictionFloorByAgent.clear(); snapshotByAgent.clear(); diff --git a/desktop/src/features/agents/observerTranscriptRetention.test.mjs b/desktop/src/features/agents/observerTranscriptRetention.test.mjs index aa7ab427c92..f8d1200f8aa 100644 --- a/desktop/src/features/agents/observerTranscriptRetention.test.mjs +++ b/desktop/src/features/agents/observerTranscriptRetention.test.mjs @@ -23,6 +23,7 @@ import { beforeEach, describe, it } from "node:test"; import { getAgentObserverSnapshot, + getAgentSessionConfigs, getAgentTranscript, resetAgentObserverStore, subscribeAgentObserverStore, @@ -416,3 +417,39 @@ describe("live observer journal — in-order append fast path ordering/dedup", ( ); }); }); + +it("native session config remains available after its transcript frame is evicted", () => { + resetAgentObserverStore(); + syncAgentObserverEvents(AGENT_PUBKEY, [ + { + ...makeEvent(0), + kind: "session_config_captured", + payload: { + liveEffortSwitching: true, + effortSessionToken: "fbf7259f-74df-4859-8b53-c0d8b77fb21e", + configOptions: [ + { + category: "thought_level", + type: "select", + currentValue: "high", + options: [{ value: "high" }], + }, + ], + }, + }, + ]); + fillSequential(MAX_OBSERVER_EVENTS + 1); + assert.equal( + getAgentObserverSnapshot(AGENT_PUBKEY).events.some( + (e) => e.kind === "session_config_captured", + ), + false, + ); + assert.equal( + getAgentSessionConfigs(AGENT_PUBKEY)[0].payload.configOptions[0] + .currentValue, + "high", + ); + resetAgentObserverStore(); + assert.deepEqual(getAgentSessionConfigs(AGENT_PUBKEY), []); +}); diff --git a/desktop/src/features/agents/ui/ConversationEffortPicker.tsx b/desktop/src/features/agents/ui/ConversationEffortPicker.tsx new file mode 100644 index 00000000000..34589a178eb --- /dev/null +++ b/desktop/src/features/agents/ui/ConversationEffortPicker.tsx @@ -0,0 +1,166 @@ +import * as React from "react"; +import { switchManagedAgentEffort } from "@/shared/api/agentControl"; +import { + getAgentSessionConfigs, + subscribeControlResults, +} from "../observerRelayStore"; +import { + conversationEfforts, + matchingEffortStatus, + type EffortRequest, +} from "../lib/conversationEffort"; +import type { ObserverEvent } from "./agentSessionTypes"; +import { PersonaDropdownField } from "./PersonaDropdownField"; + +const RESULT_TEXT: Record = { + applied: "Thinking level applied to this conversation.", + queued: "Queued — applies after the current response, before the next one.", + rejected: "The adapter rejected this level. The previous level is unchanged.", + unsupported: "This model does not support that level. Choose another level.", + stale_session: + "This conversation’s session has ended. Open its latest activity and try again.", + unavailable: "Live thinking levels are unavailable for this conversation.", + busy: "Another change is pending. Wait for its result before trying again.", + expired: "The change expired before it could apply. Choose the level again.", + unconfirmed: + "The agent did not confirm this change. Check its reported level before continuing.", + invalid_request: "The agent could not accept this request.", +}; + +export function ConversationEffortPicker({ + pubkey, + channelId, + events, +}: { + pubkey: string; + channelId: string | null; + events: readonly ObserverEvent[]; +}) { + const retained = getAgentSessionConfigs(pubkey); + const choices = React.useMemo( + () => conversationEfforts([...retained, ...events], channelId), + [retained, events, channelId], + ); + const [selectedSession, setSelectedSession] = React.useState(""); + const [pending, setPending] = React.useState(null); + const [message, setMessage] = React.useState(null); + const selection = + choices.find( + (choice) => `${choice.channelId}:${choice.sessionId}` === selectedSession, + ) ?? (choices.length === 1 ? choices[0] : undefined); + const subscription = React.useRef<() => void>(() => {}); + const id = React.useId(); + React.useEffect(() => () => subscription.current(), []); + React.useEffect(() => { + if ( + pending && + !choices.some( + (choice) => + choice.channelId === pending.channelId && + choice.sessionId === pending.sessionId && + choice.sessionToken === pending.sessionToken, + ) + ) { + subscription.current(); + setPending(null); + setMessage(RESULT_TEXT.stale_session); + } + }, [choices, pending]); + + async function change(effort: string) { + if (!selection || pending) return; + const request = { + requestId: crypto.randomUUID(), + channelId: selection.channelId, + sessionId: selection.sessionId, + sessionToken: selection.sessionToken, + effort, + }; + setPending(request); + setMessage("Requesting thinking level change…"); + let timeout: ReturnType; + let disposed = false; + const deliveryTimeout = setTimeout(() => { + setMessage( + "Waiting for the agent to acknowledge this change. The applied level is still unconfirmed.", + ); + }, 8_000); + const unsubscribe = subscribeControlResults(pubkey, (frame) => { + const status = matchingEffortStatus(frame, request); + if (!status) return; + clearTimeout(deliveryTimeout); + setMessage(RESULT_TEXT[status] ?? RESULT_TEXT.unconfirmed); + if (status !== "queued") { + setPending(null); + subscription.current(); + } + }); + subscription.current = () => { + disposed = true; + unsubscribe(); + clearTimeout(timeout); + clearTimeout(deliveryTimeout); + }; + timeout = setTimeout(() => { + setMessage(RESULT_TEXT.unconfirmed); + setPending(null); + subscription.current(); + }, 310_000); + try { + await switchManagedAgentEffort(pubkey, request); + } catch { + if (disposed) return; + subscription.current(); + setPending(null); + setMessage( + "Could not deliver the change. The applied level is unconfirmed.", + ); + } + } + + if (choices.length === 0) return null; + return ( +
+ {choices.length > 1 ? ( + <> + + ({ + value: `${choice.channelId}:${choice.sessionId}`, + label: `${choice.label ?? `Conversation ${index + 1}`} · ${new Date(choice.timestamp).toLocaleString()}`, + }))} + /> + + ) : null} + + { + void change(value); + }} + options={selection?.options ?? []} + placeholder="Choose a conversation first" + /> +

+ {message ?? + "For this conversation. A change applies after the current response; saved defaults stay unchanged."} +

+
+ ); +} diff --git a/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx b/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx index 4e526692246..84423e633a0 100644 --- a/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx +++ b/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx @@ -1,4 +1,5 @@ import * as React from "react"; +import { ConversationEffortPicker } from "./ConversationEffortPicker"; import { CircleAlert, CircleDot, @@ -39,6 +40,7 @@ import { import { buildTranscriptState } from "./agentSessionTranscript"; type ManagedAgentSessionPanelProps = { + canChangeEffort?: boolean; agent: Pick & { status: ManagedAgent["status"] | "unknown"; avatarUrl?: string | null; @@ -60,6 +62,7 @@ type ManagedAgentSessionPanelProps = { }; export function ManagedAgentSessionPanel({ + canChangeEffort = false, agent, autoTail = false, channelId = null, @@ -146,6 +149,14 @@ export function ManagedAgentSessionPanel({ /> ) : null} + {canChangeEffort && hasObserver ? ( + + ) : null}
{ + await sendAgentObserverControl(pubkey, { type: "switch_effort", ...request }); +} + /** Send a stop request; the harness acknowledges it via control_result. */ export async function cancelManagedAgentTurn( pubkey: string, diff --git a/desktop/src/shared/api/types.ts b/desktop/src/shared/api/types.ts index b988843d60b..7042561a64a 100644 --- a/desktop/src/shared/api/types.ts +++ b/desktop/src/shared/api/types.ts @@ -458,7 +458,10 @@ export type SwitchManagedAgentModelStatus = | "failure"; export type ControlResultFrame = { - type: "cancel_turn" | "switch_model"; + type: "cancel_turn" | "switch_model" | "switch_effort"; + sessionId?: string; + sessionToken?: string; + effort?: string; status: string; modelId?: string; /** Opaque per-pick id echoed from the request; correlates late frames. */ diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 6ddc1111b03..0ea0c44672a 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -43,6 +43,7 @@ import { syncAgentTurnsFromEvents } from "@/features/agents/activeAgentTurnsStor import { recordTimeoutFromRejection } from "@/features/moderation/lib/timeoutStore"; import { injectObserverEventsForE2E, + injectLiveObserverEventsForE2E, syncAgentObserverEvents, } from "@/features/agents/observerRelayStore"; import { @@ -392,7 +393,10 @@ type E2eConfig = { sendMessageErrors?: string[]; /** Test-only observer control requests captured after mock publish. */ observerControlResults?: Array<{ - type: "cancel_turn" | "switch_model"; + type: "cancel_turn" | "switch_model" | "switch_effort"; + sessionId?: string; + sessionToken?: string; + effort?: string; status: string; channelId?: string | null; requestId?: string; @@ -1487,6 +1491,7 @@ declare global { turnId: string; kind?: "turn_started" | "turn_completed"; }) => void; + __BUZZ_E2E_SEED_LIVE_OBSERVER_EVENTS__?: Window["__BUZZ_E2E_SEED_OBSERVER_EVENTS__"]; __BUZZ_E2E_SEED_OBSERVER_EVENTS__?: (input: { agentPubkey: string; events: Array<{ @@ -4960,13 +4965,19 @@ let mockObserverControlSeq = 0; function emitMockObserverControlResult( agentPubkey: string, request: { - type: "cancel_turn" | "switch_model"; + type: "cancel_turn" | "switch_model" | "switch_effort"; + sessionId?: string; + sessionToken?: string; + effort?: string; channelId?: string | null; requestId?: string; modelId?: string; }, result: { - type: "cancel_turn" | "switch_model"; + type: "cancel_turn" | "switch_model" | "switch_effort"; + sessionId?: string; + sessionToken?: string; + effort?: string; status: string; channelId?: string | null; requestId?: string; @@ -4977,6 +4988,9 @@ function emitMockObserverControlResult( const payload = { type: result.type, status: result.status, + sessionId: result.sessionId ?? request.sessionId, + sessionToken: result.sessionToken ?? request.sessionToken, + effort: result.effort ?? request.effort, ...(result.requestId !== undefined ? { requestId: result.requestId } : request.requestId !== undefined @@ -11063,7 +11077,10 @@ function sendToMockSocket(args: { const frame = event.tags.find((tag) => tag[0] === "frame")?.[1]; if (frame === "control") { let payload: { - type: "cancel_turn" | "switch_model"; + type: "cancel_turn" | "switch_model" | "switch_effort"; + sessionId?: string; + sessionToken?: string; + effort?: string; channelId?: string | null; requestId?: string; modelId?: string; @@ -11855,6 +11872,9 @@ export function maybeInstallE2eTauriMocks() { syncAgentTurnsFromEvents(agentPubkey, [event]); syncAgentObserverEvents(agentPubkey, [event]); }; + window.__BUZZ_E2E_SEED_LIVE_OBSERVER_EVENTS__ = ({ agentPubkey, events }) => { + injectLiveObserverEventsForE2E(agentPubkey, events); + }; window.__BUZZ_E2E_SEED_OBSERVER_EVENTS__ = ({ agentPubkey, events }) => { injectObserverEventsForE2E(agentPubkey, events); }; diff --git a/desktop/tests/e2e/conversation-effort.spec.ts b/desktop/tests/e2e/conversation-effort.spec.ts new file mode 100644 index 00000000000..a854d51cf62 --- /dev/null +++ b/desktop/tests/e2e/conversation-effort.spec.ts @@ -0,0 +1,236 @@ +import { expect, test, type Page } from "@playwright/test"; +import { installMockBridge, TEST_IDENTITIES } from "../helpers/bridge"; +import { waitForAnimations } from "../helpers/animations"; + +const AGENT = TEST_IDENTITIES.charlie.pubkey; +const CHANNEL = "94a444a4-c0a3-5966-ab05-530c6ddc2301"; +const SESSION = "codex-existing-conversation"; +const config = (value: string) => ({ + liveEffortSwitching: true, + effortSessionToken: "fbf7259f-74df-4859-8b53-c0d8b77fb21e", + configOptions: [ + { + id: "reasoning_effort", + name: "Reasoning effort", + type: "select", + category: "thought_level", + currentValue: value, + options: [ + { value: "low", name: "Low" }, + { value: "high", name: "High" }, + ], + }, + ], +}); + +async function seed( + page: Page, + kind: string, + payload: unknown, + seq: number, + sessionId = SESSION, +) { + await page.evaluate( + ({ agentPubkey, channelId, sessionId, kind, payload, seq }) => { + window.__BUZZ_E2E_SEED_LIVE_OBSERVER_EVENTS__?.({ + agentPubkey, + events: [ + { + seq, + timestamp: new Date().toISOString(), + kind, + payload, + agentIndex: 0, + channelId, + sessionId, + turnId: null, + }, + ], + }); + }, + { agentPubkey: AGENT, channelId: CHANNEL, sessionId, kind, payload, seq }, + ); +} + +async function openActivity(page: Page, status: string) { + await installMockBridge(page, { + managedAgents: [ + { + pubkey: AGENT, + name: "Codex Colleague", + status: "running", + channelNames: ["agents"], + }, + ], + observerControlResults: [{ type: "switch_effort", status }], + }); + await page.goto(`/#/channels/${CHANNEL}?agentSession=${AGENT}`); + await expect(page.getByTestId("agent-session-thread-panel")).toBeVisible(); + await page.waitForFunction( + () => typeof window.__BUZZ_E2E_SEED_LIVE_OBSERVER_EVENTS__ === "function", + ); + await seed(page, "session_config_captured", config("low"), 1); + await seed( + page, + "acp_read", + { + method: "session/update", + params: { + sessionId: SESSION, + update: { + sessionUpdate: "agent_message_chunk", + content: { + type: "text", + text: "I have the conversation context and am checking the remaining cases.", + }, + }, + }, + }, + 2, + ); + const picker = page.getByTestId("conversation-effort-picker"); + await expect(picker).toBeVisible(); + await expect(page.getByTestId("agent-session-thread-panel")).toContainText( + "I have the conversation context", + ); + return picker; +} + +test("live effort waits for the exact adapter receipt and keeps the conversation", async ({ + page, +}, testInfo) => { + const picker = await openActivity(page, "queued"); + await picker.getByLabel("Thinking level").click(); + await page.getByRole("menuitemradio", { name: "High", exact: true }).click(); + await expect(picker.getByRole("status")).toContainText("Queued"); + await expect(picker.getByLabel("Thinking level")).toContainText("Low"); + await expect(picker.getByLabel("Thinking level")).toBeDisabled(); + await waitForAnimations(page); + await page + .getByTestId("agent-session-thread-panel") + .screenshot({ path: testInfo.outputPath("live-queued.png") }); + const request = await page.evaluate(() => + window.__BUZZ_E2E_OBSERVER_CONTROLS__?.at(-1), + ); + expect(request).toMatchObject({ + agentPubkey: AGENT, + payload: { + type: "switch_effort", + channelId: CHANNEL, + sessionId: SESSION, + effort: "high", + requestId: expect.any(String), + }, + }); + if (!request) throw new Error("Expected a recorded effort request"); + const payload = request.payload as Record; + await seed( + page, + "control_result", + { ...payload, status: "applied", requestId: "old-replayed-pick" }, + 3, + ); + await expect(picker.getByRole("status")).toContainText("Queued"); + await seed(page, "session_config_captured", config("high"), 4); + await seed(page, "control_result", { ...payload, status: "applied" }, 5); + await expect(picker.getByRole("status")).toHaveText( + "Thinking level applied to this conversation.", + ); + await expect(picker.getByLabel("Thinking level")).toContainText("High"); + await expect(picker.getByLabel("Thinking level")).toBeEnabled(); + await waitForAnimations(page); + await page.getByTestId("agent-session-thread-panel").screenshot({ + path: testInfo.outputPath("live-applied.png"), + }); +}); + +test("adapter rejection leaves the reported level unchanged", async ({ + page, +}) => { + const picker = await openActivity(page, "rejected"); + await picker.getByLabel("Thinking level").click(); + await page.getByRole("menuitemradio", { name: "High", exact: true }).click(); + await expect(picker.getByRole("status")).toContainText("rejected"); + await expect(picker.getByLabel("Thinking level")).toContainText("Low"); + await expect(picker.getByLabel("Thinking level")).toBeEnabled(); +}); + +test("multiple reported sessions require a conversation choice", async ({ + page, +}) => { + const picker = await openActivity(page, "queued"); + await seed( + page, + "session_config_captured", + config("low"), + 2, + "sibling-conversation", + ); + await expect( + picker.getByLabel("Conversation", { exact: true }), + ).toContainText("Choose a conversation"); + await expect(picker.getByLabel("Thinking level")).toBeDisabled(); + expect( + await page.evaluate(() => window.__BUZZ_E2E_OBSERVER_CONTROLS__?.length), + ).toBe(0); + await picker.getByLabel("Conversation", { exact: true }).click(); + await page.getByRole("menuitemradio").first().click(); + await picker.getByLabel("Thinking level").click(); + await page.getByRole("menuitemradio", { name: "High", exact: true }).click(); + const control = await page.evaluate(() => + window.__BUZZ_E2E_OBSERVER_CONTROLS__?.at(-1), + ); + expect(control?.payload).toMatchObject({ + channelId: CHANNEL, + sessionId: "sibling-conversation", + effort: "high", + }); +}); + +test("a session without the live capability has no effort control", async ({ + page, +}) => { + await openActivity(page, "queued"); + await seed( + page, + "session_config_captured", + { ...config("low"), liveEffortSwitching: false }, + 20, + ); + await expect(page.getByTestId("conversation-effort-picker")).toHaveCount(0); + expect( + await page.evaluate(() => window.__BUZZ_E2E_OBSERVER_CONTROLS__?.length), + ).toBe(0); +}); + +test("recreated sessions reject late receipts for the prior conversation", async ({ + page, +}) => { + const picker = await openActivity(page, "queued"); + await picker.getByLabel("Thinking level").click(); + await page.getByRole("menuitemradio", { name: "High", exact: true }).click(); + await expect(picker.getByRole("status")).toContainText("Queued"); + const control = await page.evaluate(() => + window.__BUZZ_E2E_OBSERVER_CONTROLS__?.at(-1), + ); + expect(control).toBeTruthy(); + await seed( + page, + "session_config_captured", + { + ...config("low"), + effortSessionToken: "79674911-359b-49ab-ae85-89d6c31d5e0f", + }, + 20, + ); + await expect(picker.getByRole("status")).toContainText("session has ended"); + await expect(picker.getByLabel("Thinking level")).toBeEnabled(); + await seed( + page, + "control_result", + { ...control?.payload, status: "applied" }, + 21, + ); + await expect(picker.getByRole("status")).toContainText("session has ended"); + await expect(picker.getByLabel("Thinking level")).toContainText("Low"); +}); diff --git a/desktop/tests/helpers/bridge.ts b/desktop/tests/helpers/bridge.ts index c3f4ed69f4c..4d0b2131315 100644 --- a/desktop/tests/helpers/bridge.ts +++ b/desktop/tests/helpers/bridge.ts @@ -307,7 +307,10 @@ type MockBridgeOptions = { sendMessageErrors?: string[]; /** Test-only observer control results emitted after mock control publishes. */ observerControlResults?: Array<{ - type: "cancel_turn" | "switch_model"; + type: "cancel_turn" | "switch_model" | "switch_effort"; + sessionId?: string; + sessionToken?: string; + effort?: string; status: string; channelId?: string | null; requestId?: string; diff --git a/docs/conversation-effort.md b/docs/conversation-effort.md new file mode 100644 index 00000000000..998067b0e4c --- /dev/null +++ b/docs/conversation-effort.md @@ -0,0 +1,71 @@ +# Conversation thinking level + +An owner can change the thinking level of an existing conversation from the +agent activity pane. A running response finishes at its current level; the next +response in that same ACP session uses the selected level. Conversation history +and saved defaults are preserved. If several conversations are visible, the +owner chooses one before choosing a level. + +This is the live-control follow-up deferred in #4557. It does not change saved +agent settings, model discovery, or portable exports covered by #5016. + +## Authority and lifecycle + +The harness captures native `thought_level` select options after session setup +and identifies the exact channel and ACP session in the observer envelope. The +snapshot advertises `liveEffortSwitching` only when it fits the worker cache +(128 snapshots, at most 256 KiB each); older harnesses and sessions without +native options have no live picker. The frontend retains at most 128 small +session-config snapshots per agent independently of transcript eviction. A new +session for the same conversation supersedes the previous selection. + +`switch_effort` uses the existing encrypted, signed owner observer-control route +and its five-minute freshness check. Requests include `channelId`, `sessionId`, +`requestId`, `sessionToken` and `effort`. The per-session token comes from the +harness snapshot and changes on recreation, preventing a stale request from +editing a successor when an adapter reuses session IDs. The harness discovers the native configuration ID from +the session snapshot, validates supported values, and sends +`session/set_config_option` only while the owning worker is idle. It never +cancels a prompt, recreates the session, or changes `startup_effort` on success. +A pending edit fences that worker before its next claim. Other workers remain +available. A possible busy sibling must return before target resolution because +some adapters use process-local session IDs; ambiguous targets are rejected. + +The queue allows 32 pending requests, one per exact session. Up to 256 receipts +are retained for ten minutes, covering the full accepted replay window without +evicting receipts early. Edits expire after five minutes; a native RPC has a +five-second deadline, with at most one RPC per main-loop iteration. A transport +failure retires the affected worker so a poisoned stream cannot serve a new +turn; normal pool maintenance replaces it. Its session is then lost and the UI +reports the change as unconfirmed, never applied. + +`queued` means accepted for later application. `applied` requires a returned +native `currentValue` matching the request. Rejection, unsupported or stale +sessions, capacity, expiration, and missing confirmation remain distinct +outcomes. The UI correlates all five request fields and subscribes before +publishing. It retains the reported value while queued and explains missing +acknowledgment or confirmation rather than claiming success from relay delivery. + +## Verification + +`cargo test -p buzz-acp --lib live_effort` exercises the real queue and ACP stdio +boundary. Desktop unit tests cover snapshot retention and receipt correlation; +`conversation-effort.spec.ts` covers the owner activity workflow through the +mock relay bridge. Existing activity-control tests guard adjacent Stop and +model controls. + +The ignored `real_codex_preserves_conversation_across_queued_effort_change` test +performs two actual provider turns. Set `BUZZ_TEST_CODEX_ADAPTER` to an installed +adapter, `CODEX_HOME` to an already-authenticated profile, and optionally +`BUZZ_TEST_CODEX_MODEL` to an advertised model ID supporting Low and High. Run: + +```sh +cargo test -p buzz-acp --lib \ + real_codex_preserves_conversation_across_queued_effort_change -- --ignored --nocapture +``` + +It verifies that a busy edit queues without cancellation, confirms High via the +adapter, and recalls a random word from the first turn in the same session while +the saved startup level remains Low. Provider turn records independently expose +the actual model and effort used by each turn; browser fixtures alone do not +establish provider behavior.