feat(LAC-3073): wire orchestrated-query SpecializedAgent path through parallel-session registry

src-tauri/src/agents/base_agent.rs

@@ -32,6 +32,12 @@ pub struct Task {
     pub dependencies: Vec<String>,
     pub timeout: Option<Duration>,
     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<String>,
 }
 
 /// Result of task execution by an agent
@@ -171,3 +177,133 @@ 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);
+    }
+
+    /// Log of (task_id, session_id) pairs observed by handle_task.
+    type SeenLog = Arc<Mutex<Vec<(String, Option<String>)>>>;
+
+    /// 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: SeenLog,
+    }
+
+    #[async_trait]
+    impl SpecializedAgent for RecordingAgent {
+        fn agent_type(&self) -> AgentType {
+            AgentType::Desktop
+        }
+
+        fn get_capabilities(&self) -> Vec<AgentCapability> {
+            vec![]
+        }
+
+        async fn can_handle_task(&self, _task: &Task) -> bool {
+            true
+        }
+
+        async fn handle_task(&self, task: Task) -> Result<TaskResult, AgentError> {
+            // 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<dyn SpecializedAgent> = 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"
+            );
+        }
+    }
+}

src-tauri/src/agents/desktop_agent.rs

@@ -42,26 +42,28 @@ impl DesktopAgent {
         })
     }
 
-    /// Execute a desktop-related tool call
-    async fn execute_desktop_tool(&self, tool_call: &ToolCall) -> Result<ToolResult, AgentError> {
+    /// 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<ToolResult, AgentError> {
         let state = self.app_handle.state::<AppState>();
 
         match tool_call.name.as_str() {
             // REMOVED: dev_left_click, desktop_click - Use computer tool with action: "left_click" instead
             "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;

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;

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);
             }

src-tauri/src/agents/session.rs

@@ -477,6 +477,44 @@ pub async fn broadcast_sessions_updated(app: &AppHandle, registry: &Arc<AgentSes
     }
 }
 
+/// Register a new run in the parallel-session registry and announce it.
+///
+/// Shared entry sequence for every execution path that participates in the
+/// session roster (`execute_agent_internal` and the orchestrated-query path,
+/// LAC-3073): create the session, mark it `Running`, emit the discrete
+/// `started` lifecycle event, and broadcast the roster snapshot. Returns the
+/// RAII [`SessionHandle`] whose drop removes the row on any exit.
+///
+/// Returns `None` when the parallel cap is hit — callers proceed without
+/// session tracking rather than failing the run, matching the pre-LAC-1432
+/// behavior where the roster did not exist.
+pub async fn begin_session_run(
+    registry: &Arc<AgentSessionRegistry>,
+    agent_name: &str,
+    app_handle: &AppHandle,
+) -> Option<SessionHandle> {
+    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

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, &registry).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.