From 9fb387473e3be473e7dd1f2b51303e5230d287c2 Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Sat, 25 Jul 2026 16:46:53 -0400 Subject: [PATCH 1/2] feat(LAC-3073): wire orchestrated-query path through parallel-session registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Registers submit_orchestrated_query / create_orchestrator_task runs in AgentSessionRegistry via a new shared begin_session_run helper (also adopted by execute_agent_internal), and threads the session id per-task — Task.session_id, never on the shared specialist instance — down to execute_computer_tool so roster current_action updates during orchestrated DesktopAgent runs. Subtask splitting propagates the parent's session id; queued/benchmark/ legacy paths pass None. Removes the TODO(LAC-3073) marker. Regression tests: legacy Task JSON without session_id deserializes to None; concurrent tasks through one shared agent instance keep their own session ids. Co-Authored-By: Claude Fable 5 --- src-tauri/src/agents/base_agent.rs | 133 +++++++++++++++++++++++++ src-tauri/src/agents/desktop_agent.rs | 27 +++-- src-tauri/src/agents/mod.rs | 4 +- src-tauri/src/agents/orchestrator.rs | 6 ++ src-tauri/src/agents/session.rs | 38 +++++++ src-tauri/src/anthropic.rs | 34 +------ src-tauri/src/commands/orchestrator.rs | 59 ++++++++++- 7 files changed, 252 insertions(+), 49 deletions(-) diff --git a/src-tauri/src/agents/base_agent.rs b/src-tauri/src/agents/base_agent.rs index 8c165612..973f531d 100644 --- a/src-tauri/src/agents/base_agent.rs +++ b/src-tauri/src/agents/base_agent.rs @@ -32,6 +32,12 @@ pub struct Task { pub dependencies: Vec, pub timeout: Option, pub metadata: serde_json::Value, + /// Parallel-session id for roster attribution (LAC-3073). Travels on the + /// task — never on the long-lived shared agent instance — so concurrent + /// orchestrated runs cannot leak identity into each other. `None` for + /// callers outside the session registry (queued/benchmark/legacy paths). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_id: Option, } /// Result of task execution by an agent @@ -171,3 +177,130 @@ pub trait SpecializedAgent: Send + Sync { true // Default implementation - can be overridden } } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use tokio::sync::Mutex; + + fn task_with_session(id: &str, session_id: Option<&str>) -> Task { + Task { + id: id.to_string(), + description: "test task".to_string(), + tool_calls: vec![], + agent_type: AgentType::Desktop, + priority: TaskPriority::Normal, + dependencies: vec![], + timeout: None, + metadata: serde_json::Value::Null, + session_id: session_id.map(|s| s.to_string()), + } + } + + /// Tasks serialized before LAC-3073 (or built by the frontend without a + /// session) must still deserialize — `session_id` defaults to `None`. + #[test] + fn task_without_session_id_deserializes_to_none() { + let json = serde_json::json!({ + "id": "t1", + "description": "legacy task", + "tool_calls": [], + "agent_type": "Desktop", + "priority": "Normal", + "dependencies": [], + "timeout": null, + "metadata": {} + }); + let task: Task = serde_json::from_value(json).expect("legacy task deserializes"); + assert_eq!(task.session_id, None); + } + + /// Mock agent that mirrors the shared-instance shape of DesktopAgent: + /// one Arc'd instance handles tasks from many concurrent runs. It records + /// which session id each handle_task invocation observed. + struct RecordingAgent { + seen: Arc)>>>, + } + + #[async_trait] + impl SpecializedAgent for RecordingAgent { + fn agent_type(&self) -> AgentType { + AgentType::Desktop + } + + fn get_capabilities(&self) -> Vec { + vec![] + } + + async fn can_handle_task(&self, _task: &Task) -> bool { + true + } + + async fn handle_task(&self, task: Task) -> Result { + // Yield so concurrent invocations interleave — a session id + // stored on the shared instance (the LAC-3073 anti-pattern) + // would be observed by the wrong task here. + tokio::task::yield_now().await; + self.seen + .lock() + .await + .push((task.id.clone(), task.session_id.clone())); + Ok(TaskResult { + task_id: task.id, + success: true, + output: serde_json::Value::Null, + error: None, + execution_time: Duration::from_millis(0), + agent_type: AgentType::Desktop, + metadata: serde_json::Value::Null, + }) + } + + async fn get_status(&self) -> AgentStatus { + AgentStatus { + agent_type: AgentType::Desktop, + is_available: true, + current_tasks: 0, + total_completed: 0, + success_rate: 1.0, + average_execution_time: Duration::from_millis(0), + capabilities: vec![], + } + } + } + + /// Regression test for LAC-3073 per-task attribution: concurrent tasks + /// routed through ONE shared agent instance must each carry their own + /// session id — identity travels on the `Task`, not the agent. + #[tokio::test] + async fn concurrent_tasks_keep_their_own_session_ids() { + let seen = Arc::new(Mutex::new(Vec::new())); + let agent: Arc = Arc::new(RecordingAgent { seen: seen.clone() }); + + let mut handles = Vec::new(); + for i in 0..8 { + let agent = agent.clone(); + let task = task_with_session(&format!("task-{i}"), Some(&format!("session-{i}"))); + handles.push(tokio::spawn(async move { agent.handle_task(task).await })); + } + for handle in handles { + let result = handle.await.expect("join ok").expect("task ok"); + assert!(result.success); + } + + let seen = seen.lock().await; + assert_eq!(seen.len(), 8); + for (task_id, session_id) in seen.iter() { + let index = task_id + .strip_prefix("task-") + .expect("task id shape") + .to_string(); + assert_eq!( + session_id.as_deref(), + Some(format!("session-{index}").as_str()), + "task {task_id} observed a session id from another run" + ); + } + } +} diff --git a/src-tauri/src/agents/desktop_agent.rs b/src-tauri/src/agents/desktop_agent.rs index afd92d35..750866fd 100644 --- a/src-tauri/src/agents/desktop_agent.rs +++ b/src-tauri/src/agents/desktop_agent.rs @@ -42,8 +42,17 @@ impl DesktopAgent { }) } - /// Execute a desktop-related tool call - async fn execute_desktop_tool(&self, tool_call: &ToolCall) -> Result { + /// Execute a desktop-related tool call. + /// + /// `session_id` is the parallel-session identity of the run that issued + /// this task (LAC-3073). It arrives per-call from the `Task` — this + /// instance is shared across concurrent runs, so it must never be stored + /// on `self`. + async fn execute_desktop_tool( + &self, + tool_call: &ToolCall, + session_id: Option<&str>, + ) -> Result { let state = self.app_handle.state::(); match tool_call.name.as_str() { @@ -51,17 +60,10 @@ impl DesktopAgent { "computer" => { // Delegate to the official Anthropic Computer Use tool implementation // This handles all computer actions: click, type, scroll, screenshot, etc. - // - // TODO(LAC-3073): session_id is None because the SpecializedAgent - // path (AgentFactory → handle_task) is not wired through the - // AgentSessionRegistry — no session id exists here yet, and this - // instance is shared across runs so it must not store one. - // Consequence: roster `current_action` doesn't update during - // orchestrated DesktopAgent runs (input arbitration is unaffected). match crate::agent::tools::anthropic_computer_use::execute_computer_tool( &self.app_handle, tool_call.input.clone(), - None, + session_id, ) .await { @@ -378,7 +380,10 @@ impl SpecializedAgent for DesktopAgent { // Execute all tool calls in the task for tool_call in &task.tool_calls { - match self.execute_desktop_tool(tool_call).await { + match self + .execute_desktop_tool(tool_call, task.session_id.as_deref()) + .await + { Ok(result) => results.push(result), Err(e) => { has_error = true; diff --git a/src-tauri/src/agents/mod.rs b/src-tauri/src/agents/mod.rs index c67941f6..7794a80d 100644 --- a/src-tauri/src/agents/mod.rs +++ b/src-tauri/src/agents/mod.rs @@ -13,7 +13,7 @@ pub use browser_agent::BrowserAgent; pub use desktop_agent::DesktopAgent; pub use orchestrator::{Orchestrator, OrchestratorConfig}; pub use session::{ - broadcast_sessions_updated, color_for_slot, AgentSession, AgentSessionId, AgentSessionInfo, - AgentSessionRegistry, AgentSessionStatus, SessionHandle, SESSION_COLOR_SLOTS, + begin_session_run, broadcast_sessions_updated, color_for_slot, AgentSession, AgentSessionId, + AgentSessionInfo, AgentSessionRegistry, AgentSessionStatus, SessionHandle, SESSION_COLOR_SLOTS, }; pub use system_agent::SystemAgent; diff --git a/src-tauri/src/agents/orchestrator.rs b/src-tauri/src/agents/orchestrator.rs index 1fa764b9..28789dc7 100644 --- a/src-tauri/src/agents/orchestrator.rs +++ b/src-tauri/src/agents/orchestrator.rs @@ -204,6 +204,7 @@ impl Orchestrator { "created_at": chrono::Utc::now().to_rfc3339(), "user_input": user_input }), + session_id: None, }; match self.delegate_task(task).await { @@ -248,6 +249,7 @@ impl Orchestrator { "message_index": i, "role": message.role }), + session_id: None, }; tasks.push(task); } @@ -1332,6 +1334,8 @@ impl Orchestrator { "subtask_index": i, "total_subtasks": split_info.len() + 1 }), + // Subtasks run on behalf of the parent's session. + session_id: task.session_id.clone(), }; subtasks.push(subtask); } @@ -1365,6 +1369,8 @@ impl Orchestrator { "subtask_index": split_info.len(), "total_subtasks": split_info.len() + 1 }), + // Subtasks run on behalf of the parent's session. + session_id: task.session_id.clone(), }; subtasks.push(final_subtask); } diff --git a/src-tauri/src/agents/session.rs b/src-tauri/src/agents/session.rs index 6878ae54..c9ce6828 100644 --- a/src-tauri/src/agents/session.rs +++ b/src-tauri/src/agents/session.rs @@ -477,6 +477,44 @@ pub async fn broadcast_sessions_updated(app: &AppHandle, registry: &Arc, + agent_name: &str, + app_handle: &AppHandle, +) -> Option { + match registry.create(agent_name.to_string()).await { + Ok(session) => { + session.set_status(AgentSessionStatus::Running).await; + let handle = SessionHandle::new(registry.clone(), session, app_handle.clone()); + let focused = handle.is_focused(); + let snapshot = handle.session().snapshot(focused).await; + if let Err(e) = app_handle.emit(events::agent_sessions::STARTED, &snapshot) { + warn!("Failed to emit agent-session-started: {}", e); + } + broadcast_sessions_updated(app_handle, registry).await; + Some(handle) + } + Err(e) => { + warn!( + "Failed to register agent session in parallel registry: {} — proceeding without session tracking", + e + ); + None + } + } +} + /// RAII guard that removes an agent session from the registry on drop. /// /// `execute_agent_internal` has ~8 explicit `return Err` paths plus a diff --git a/src-tauri/src/anthropic.rs b/src-tauri/src/anthropic.rs index eadb9603..553b53f5 100644 --- a/src-tauri/src/anthropic.rs +++ b/src-tauri/src/anthropic.rs @@ -512,37 +512,9 @@ async fn execute_agent_internal( // If the parallel cap is hit we log and continue — the queue guarantees // at most one run at a time today, so the cap should never actually bite // until LAC-1432 lifts the queue serialization. - let session_handle = { - let registry = state.agent_sessions(); - match registry.create("orchestrator".to_string()).await { - Ok(session) => { - session - .set_status(crate::agents::AgentSessionStatus::Running) - .await; - let handle = crate::agents::SessionHandle::new( - registry.clone(), - session, - app_handle.clone(), - ); - let focused = handle.is_focused(); - let snapshot = handle.session().snapshot(focused).await; - if let Err(e) = - app_handle.emit(crate::constants::events::agent_sessions::STARTED, &snapshot) - { - warn!("Failed to emit agent-session-started: {}", e); - } - crate::agents::broadcast_sessions_updated(&app_handle, ®istry).await; - Some(handle) - } - Err(e) => { - warn!( - "Failed to register agent session in parallel registry: {} — proceeding without session tracking", - e - ); - None - } - } - }; + let session_handle = + crate::agents::begin_session_run(&state.agent_sessions(), "orchestrator", &app_handle) + .await; // Identity context handed to the computer-use tool registration so the // overlay cursor is keyed by session id and drawn in the session color. diff --git a/src-tauri/src/commands/orchestrator.rs b/src-tauri/src/commands/orchestrator.rs index b07d5bea..7a911773 100644 --- a/src-tauri/src/commands/orchestrator.rs +++ b/src-tauri/src/commands/orchestrator.rs @@ -194,16 +194,25 @@ pub async fn submit_orchestrated_query( use_mcp_tools: Some(true), }; - let result = create_and_execute_task(&orchestrator_guard, task_request).await?; + let result = create_and_execute_task(&orchestrator_guard, task_request, &app_handle).await?; Ok(format!("Orchestrated task completed: {}", result)) } -/// Create and execute a task with the orchestrator +/// Create and execute a task with the orchestrator. +/// +/// Registers the run in the parallel-session registry (LAC-3073) so the +/// roster/switcher UI shows it alongside `submit_query` runs, and stamps the +/// session id onto the `Task` so specialist agents (DesktopAgent) attribute +/// their computer-use actions to this run. The id travels on the task — +/// specialist instances are shared across concurrent runs and must never +/// store it. async fn create_and_execute_task( orchestrator: &Orchestrator, request: TaskCreationRequest, + app_handle: &tauri::AppHandle, ) -> Result { use crate::agents::{AgentType, Task}; + use tauri::Manager; use uuid::Uuid; // Determine agent type intelligently @@ -233,6 +242,21 @@ async fn create_and_execute_task( TaskPriority::Normal }; + // Register this run in the parallel-session registry. On cap overflow we + // proceed without tracking (same policy as execute_agent_internal). The + // RAII handle removes the roster row on every exit path. + let agent_name = match agent_type { + AgentType::Browser => "browser", + AgentType::Desktop => "desktop", + AgentType::System => "system", + AgentType::Orchestrator => "orchestrator", + }; + let registry = app_handle.state::().agent_sessions(); + let session_handle = crate::agents::begin_session_run(®istry, agent_name, app_handle).await; + let session_id = session_handle + .as_ref() + .map(|handle| handle.session().id().to_string()); + // Create task let task = Task { id: Uuid::new_v4().to_string(), @@ -247,9 +271,27 @@ async fn create_and_execute_task( "context": request.context, "use_mcp_tools": request.use_mcp_tools.unwrap_or(false) }), + // If the task ends up queued (orchestrator at capacity) this id + // outlives the session row; execute_computer_tool's registry lookup + // then finds nothing and safely skips attribution. + session_id, }; - match orchestrator.delegate_task(task).await { + let result = orchestrator.delegate_task(task).await; + + // Give the roster a final status snapshot before the RAII drop removes + // the row. A queued result reports success=true and terminates the + // session immediately — matching the command's own early return. + if let Some(ref handle) = session_handle { + use crate::agents::AgentSessionStatus; + let terminal = match &result { + Ok(task_result) if task_result.success => AgentSessionStatus::Finished, + _ => AgentSessionStatus::Failed, + }; + handle.mark_terminal(terminal).await; + } + + match result { Ok(result) => { if result.success { Ok(result @@ -357,11 +399,14 @@ pub async fn configure_orchestrator(config: OrchestratorConfigDTO) -> Result<(), /// Create a new task with enhanced parameters #[tauri::command] -pub async fn create_orchestrator_task(request: TaskCreationRequest) -> Result { +pub async fn create_orchestrator_task( + request: TaskCreationRequest, + app_handle: tauri::AppHandle, +) -> Result { let orchestrator = get_orchestrator().await?; let orchestrator_guard = orchestrator.lock().await; - let result = create_and_execute_task(&orchestrator_guard, request).await?; + let result = create_and_execute_task(&orchestrator_guard, request, &app_handle).await?; Ok(result) } @@ -741,6 +786,7 @@ pub async fn execute_intelligent_parallel_tasks( "context": context, "intelligent_execution": true }), + session_id: None, }; tasks.push(task); } @@ -782,6 +828,7 @@ pub async fn intelligent_task_splitting( "context": context, "task_splitting": true }), + session_id: None, }; match orchestrator_guard @@ -831,6 +878,7 @@ pub async fn execute_optimized_workflow( "enable_splitting": enable_task_splitting, "context": context }), + session_id: None, }; // Step 1: Intelligent task splitting if enabled @@ -958,6 +1006,7 @@ pub async fn benchmark_orchestrator_performance( "benchmark": true, "optimizations_enabled": enable_optimizations }), + session_id: None, }; test_tasks.push(task); } From a79722990ba8d230bee24db3b1e7a1509fca37e4 Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Sat, 25 Jul 2026 17:04:31 -0400 Subject: [PATCH 2/2] fix(LAC-3073): add SeenLog type alias to fix clippy type_complexity Extracts `Arc)>>>` into a named type alias `SeenLog` so clippy's type_complexity lint no longer fires on the RecordingAgent test helper in base_agent.rs. Co-Authored-By: Claude Sonnet 4.6 --- src-tauri/src/agents/base_agent.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src-tauri/src/agents/base_agent.rs b/src-tauri/src/agents/base_agent.rs index 973f531d..305f767a 100644 --- a/src-tauri/src/agents/base_agent.rs +++ b/src-tauri/src/agents/base_agent.rs @@ -216,11 +216,14 @@ mod tests { assert_eq!(task.session_id, None); } + /// Log of (task_id, session_id) pairs observed by handle_task. + type SeenLog = Arc)>>>; + /// Mock agent that mirrors the shared-instance shape of DesktopAgent: /// one Arc'd instance handles tasks from many concurrent runs. It records /// which session id each handle_task invocation observed. struct RecordingAgent { - seen: Arc)>>>, + seen: SeenLog, } #[async_trait]