From ad38844a5d334b0e83a991ab2ed70a5a89403590 Mon Sep 17 00:00:00 2001 From: lucascouts Date: Sat, 3 Oct 2026 22:56:05 -0300 Subject: [PATCH] agent: hold the agent's task feed and let the thread stop or background a task 0029 taught the sidebar that work can outlive the turn that launched it, but it only ever learned a number: `_meta["_claude/backgroundTasks"].count`. A card or a history panel needs the tasks themselves -- what each one is, how long it has run, how it ended -- and a way to act on one without typing a prompt. This patch is the model and the actions; nothing renders them yet. Stories 013 (the card, the tray) and 014 (the history panel) consume it. The contract, carried by claude-agent-acp-plus since the release that added it (story 012): - `session_info_update._meta["_claude/tasks"]` = `{ tasks: [...] }`, each entry `{ id, toolCallId, type, description, background, depth, status, startedAt, endedAt, usage, lastTool, summary }`, status one of running, completed, failed, stopped, interrupted. It is a SNAPSHOT, not a delta: every update carries every live task, so a lost update is repaired by the next one. An ended task appears in exactly one snapshot, then drops out. - `_claude/tasks/stop { sessionId, taskId } -> {}` and `_claude/tasks/background { sessionId, toolCallId } -> { backgrounded }`, two JSON-RPC requests sent through `UntypedMessage`. - The `initialize` response advertises both as `agentCapabilities._meta["_claude/tasks"] = { version: 1 }`. What this adds: - `AcpThread::tasks()`, replaced wholesale on each snapshot, and two events: `TasksUpdated` when the list differs, `TaskEnded { task_id, status }` once per ending. The announced set forgets an id when it leaves the list or comes back running, because a resumed subagent restarts under its agent id and its second ending is news too. - A payload that does not parse logs a warning and leaves the list alone; an update without the key (the adapter's title-only updates ride the same notification) changes nothing; a PRESENT empty list clears it. - A running task missing from the next snapshot, never seen ending, is kept once as `interrupted` and announced. When the agent process exits, or the session is closed through `AgentConnection::close_session`, every running task becomes `interrupted`, once. `AcpConnection` had no `close_session` override; the one added waits for an in-flight load or release, closes once, then interrupts. `supports_close_session()` stays false, so in practice only the exit path is live today. - `AgentConnection::{stop_task, background_task}`, defaulting to "not supported". `AcpConnection` sends them only when the capability above is advertised -- never keyed on the entry id or the agent's name, which differ between the published adapter and a local fork. - The retention clause: `is_idle_for_retention()` is false while any task is running. A turn that ended with background work is idle by ThreadStatus, so without it `cleanup_retained_threads` dropped exactly the threads a task card exists to show. `TasksUpdated` also re-runs that cleanup, so a thread kept for its tasks is released when the last one ends rather than at the next unrelated event. What it deliberately does not do: - It leaves 0029's `background_task_count` and its key alone instead of deriving the count from the new list. An adapter that sends only the old key still lights the sidebar, and the two cannot disagree visibly because the adapter publishes both from the same table. - It routes nothing through ThreadMetadataStore, for the reason 0029 gives. - It marks nothing on a thread that is being dropped: the thread is destroyed with its tasks, so there is no observer to tell. The wire shape is pinned by a fixture shared byte for byte with the adapter (`crates/acp_thread/src/test_data/claude_tasks_snapshot.json`, parsed by `test_tasks_contract_fixture_parses`). No other agent sends the key or the capability, so for them nothing changes. Rides the claude-agent-acp-plus flag BY DEPENDENCY, like 0011, 0013 and 0029. Test added 2026-10-09 (0.2.2_p20261009): with no retained-thread budget, a thread whose background task still runs stays loaded after its turn ends, and the snapshot that ends the task releases it. Fails with either half removed -- the running-task clause in `is_idle_for_retention`, or TasksUpdated in agent_panel's match arm. a4b39c38 review: (2026-11-01, story 072) Upstream #66255 keeps native subagent outcomes in their own state (subagent_outcomes), apart from handle_session_update's arms. The _claude/tasks snapshot this patch reads stays in its arm and does not overlap. Claude Agent (Plus) emits no SubagentUpdate, so for it this feed is still the only task channel. Moving Zeo's tasks onto the native subagent surfaces is story 064's work, not this bump's. Story 064 (2026-11-02): the row source for Zed's own subagent registry lives here, in acp_thread::subagent_rows, because every Zeo task surface reads it and this is the patch that gave acp_thread a task model. subagent_rows() turns a parent thread's registered children into SubagentRows: label from the child's title, else the announcement, else the session id; status from the child's ForegroundActivity, with RequiresAction read as Waiting and never Running; a child replayed on load with no live report reads Unknown, its title kept and its Stop withdrawn. any_child_active() answers None, Running or Waiting, Waiting first. Zed's subagent card now takes its idle-reason table (idle_outcome) from here instead of keeping a second copy. Two children with the same title stay two rows: identity is the child session id. 13 tests (subagent_rows_tests.rs, subagent_rows_identity_tests.rs). --- diff --git a/crates/acp_thread/src/acp_thread.rs b/crates/acp_thread/src/acp_thread.rs index 92fb860..e104f7b 100644 --- a/crates/acp_thread/src/acp_thread.rs +++ b/crates/acp_thread/src/acp_thread.rs @@ -9,6 +9,7 @@ pub mod elicitation; mod mention; pub mod notices; pub mod prompt_capabilities; +pub mod subagent_rows; pub mod subagents; mod submission; mod terminal; @@ -47,7 +48,7 @@ use std::fmt::{Formatter, Write}; use std::ops::Range; use std::process::ExitStatus; use std::rc::Rc; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use std::{fmt::Display, mem, path::PathBuf, sync::Arc}; pub use submission::*; use task::{Shell, ShellBuilder}; @@ -3949,6 +3950,134 @@ const CLAUDE_RATE_LIMIT_META_KEY: &str = "_claude/rateLimit"; /// `background_tasks_changed` level here, as `{ "count": n, "tasks": [...] }`. const CLAUDE_BACKGROUND_TASKS_META_KEY: &str = "_claude/backgroundTasks"; +/// The `_meta` key the ACP adapter writes the full task snapshot under, on a +/// `session_info_update`: every running task plus each ended one until it has been +/// published once. A level, not a stream — each snapshot replaces the previous one. +/// +/// Kept apart from [`CLAUDE_BACKGROUND_TASKS_META_KEY`] so an adapter that only sends +/// the count keeps patch `0029`'s background status working. +const CLAUDE_TASKS_META_KEY: &str = "_claude/tasks"; + +pub type AgentTaskId = SharedString; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AgentTaskStatus { + Running, + Completed, + Failed, + Stopped, + Interrupted, +} + +impl AgentTaskStatus { + pub fn is_running(&self) -> bool { + matches!(self, AgentTaskStatus::Running) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AgentTaskUsage { + pub tokens: u64, + pub tool_uses: u64, + pub duration: Duration, +} + +/// A unit of agent work that can outlive the turn that started it: a background shell, +/// a subagent, a monitor. +#[derive(Debug, Clone, PartialEq)] +pub struct AgentTask { + pub id: AgentTaskId, + pub tool_call_id: Option, + pub kind: SharedString, + pub description: SharedString, + pub background: bool, + pub depth: u32, + pub status: AgentTaskStatus, + pub started_at: SystemTime, + pub ended_at: Option, + pub usage: Option, + pub last_tool: Option, + pub summary: Option, +} + +#[derive(Deserialize)] +struct WireTasksSnapshot { + tasks: Vec, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct WireTask { + id: String, + tool_call_id: Option, + #[serde(rename = "type")] + kind: String, + description: String, + background: bool, + depth: u32, + status: WireTaskStatus, + started_at: u64, + ended_at: Option, + usage: Option, + last_tool: Option, + summary: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "lowercase")] +enum WireTaskStatus { + Running, + Completed, + Failed, + Stopped, + Interrupted, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct WireTaskUsage { + tokens: u64, + tool_uses: u64, + duration_ms: u64, +} + +/// `SystemTime + Duration` panics on overflow (reachable on Windows with a +/// garbage timestamp), so an unrepresentable time falls back to the epoch. +fn system_time_from_millis(millis: u64) -> SystemTime { + UNIX_EPOCH + .checked_add(Duration::from_millis(millis)) + .unwrap_or(UNIX_EPOCH) +} + +impl From for AgentTask { + fn from(task: WireTask) -> Self { + AgentTask { + id: task.id.into(), + tool_call_id: task.tool_call_id.map(acp_v2::ToolCallId::new), + kind: task.kind.into(), + description: task.description.into(), + background: task.background, + depth: task.depth, + status: match task.status { + WireTaskStatus::Running => AgentTaskStatus::Running, + WireTaskStatus::Completed => AgentTaskStatus::Completed, + WireTaskStatus::Failed => AgentTaskStatus::Failed, + WireTaskStatus::Stopped => AgentTaskStatus::Stopped, + WireTaskStatus::Interrupted => AgentTaskStatus::Interrupted, + }, + started_at: system_time_from_millis(task.started_at), + ended_at: task.ended_at.map(system_time_from_millis), + usage: task.usage.map(|usage| AgentTaskUsage { + tokens: usage.tokens, + tool_uses: usage.tool_uses, + duration: Duration::from_millis(usage.duration_ms), + }), + last_tool: task.last_tool.map(SharedString::from), + summary: task.summary.map(SharedString::from), + } + } +} + pub const TOKEN_USAGE_WARNING_THRESHOLD: f32 = 0.8; impl TokenUsage { @@ -4056,6 +4185,10 @@ pub struct AcpThread { token_usage: Option, account_usage: AccountUsage, background_task_count: usize, + tasks: Vec, + // Ids whose ending has been announced with `TaskEnded`, so a repeated snapshot does + // not announce it again. An id is forgotten once it leaves `tasks`. + announced_task_endings: HashSet, cost: Option, prompt_capabilities: acp_v2::PromptCapabilities, available_commands: Vec, @@ -4212,6 +4345,11 @@ pub enum AcpThreadEvent { TokenUsageUpdated(OutputTokensUpdate), AccountUsageUpdated, BackgroundTasksUpdated, + TasksUpdated, + TaskEnded { + task_id: AgentTaskId, + status: AgentTaskStatus, + }, EntryUpdated(usize), EntriesRemoved(Range), ToolAuthorizationRequested(PermissionRequestId), @@ -4370,6 +4508,8 @@ impl AcpThread { | AcpThreadEvent::TokenUsageUpdated(_) | AcpThreadEvent::AccountUsageUpdated | AcpThreadEvent::BackgroundTasksUpdated + | AcpThreadEvent::TasksUpdated + | AcpThreadEvent::TaskEnded { .. } | AcpThreadEvent::EntryUpdated(_) | AcpThreadEvent::Retry(_) | AcpThreadEvent::SubagentSpawned(_) @@ -4426,6 +4566,8 @@ impl AcpThread { token_usage: None, account_usage: AccountUsage::default(), background_task_count: 0, + tasks: Vec::new(), + announced_task_endings: HashSet::default(), cost: None, prompt_capabilities, available_commands: Vec::new(), @@ -4935,6 +5077,9 @@ impl AcpThread { } _ => false, }) + // A turn that left background work running reads `Idle`, yet releasing the + // thread would drop the only view of that work. + && !self.tasks.iter().any(|task| task.status.is_running()) } pub fn forget_submission(&mut self, id: SubmissionId, cx: &mut Context) { @@ -5266,6 +5411,89 @@ impl AcpThread { self.background_task_count } + /// The agent's tasks as of its last `_claude/tasks` snapshot, in start order. Empty + /// for an agent that does not publish them. + pub fn tasks(&self) -> &[AgentTask] { + &self.tasks + } + + pub fn stop_task(&self, task_id: &str, cx: &mut App) -> Task> { + self.connection + .clone() + .stop_task(self.session_id(), task_id, cx) + } + + /// Resolves to `false` when the agent had nothing to move to the background. + pub fn background_task(&self, tool_call_id: &str, cx: &mut App) -> Task> { + self.connection + .clone() + .background_task(self.session_id(), tool_call_id, cx) + } + + fn apply_tasks_snapshot(&mut self, snapshot: WireTasksSnapshot, cx: &mut Context) { + let mut tasks: Vec = snapshot.tasks.into_iter().map(AgentTask::from).collect(); + + // A running task whose ending never reached us is gone without a word: keep it + // once, as interrupted, so its ending is still shown and announced. + let now = SystemTime::now(); + for previous in &self.tasks { + if previous.status.is_running() + && !self.announced_task_endings.contains(&previous.id) + && !tasks.iter().any(|task| task.id == previous.id) + { + let mut interrupted = previous.clone(); + interrupted.status = AgentTaskStatus::Interrupted; + interrupted.ended_at = Some(now); + tasks.push(interrupted); + } + } + tasks.sort_by_key(|task| task.started_at); + + // Forget an ending once its id leaves the list, or comes back running: + // a resumed subagent restarts under the same id, and its next ending + // is news again. + self.announced_task_endings.retain(|id| { + tasks + .iter() + .any(|task| &task.id == id && !task.status.is_running()) + }); + if tasks != self.tasks { + self.tasks = tasks; + cx.emit(AcpThreadEvent::TasksUpdated); + } + self.announce_task_endings(cx); + } + + fn announce_task_endings(&mut self, cx: &mut Context) { + for task in &self.tasks { + if !task.status.is_running() && self.announced_task_endings.insert(task.id.clone()) { + cx.emit(AcpThreadEvent::TaskEnded { + task_id: task.id.clone(), + status: task.status, + }); + } + } + } + + /// Marks every running task interrupted and announces each ending once. For the + /// paths where the agent can no longer report on its tasks: its process exited, or + /// the session was closed. + pub fn interrupt_running_tasks(&mut self, cx: &mut Context) { + let now = SystemTime::now(); + let mut changed = false; + for task in &mut self.tasks { + if task.status.is_running() { + task.status = AgentTaskStatus::Interrupted; + task.ended_at = Some(now); + changed = true; + } + } + if changed { + cx.emit(AcpThreadEvent::TasksUpdated); + self.announce_task_endings(cx); + } + } + pub fn cost(&self) -> Option<&SessionCost> { self.cost.as_ref() } @@ -5457,6 +5685,22 @@ impl AcpThread { self.background_task_count = count; cx.emit(AcpThreadEvent::BackgroundTasksUpdated); } + // Absent from the adapter's title-only and count-only updates: a missing + // key is "no change", never "no tasks", or the list would blank. + if let Some(payload) = info_update + .meta + .as_ref() + .and_then(|meta| meta.get(CLAUDE_TASKS_META_KEY)) + { + match serde_json::from_value::(payload.clone()) { + Ok(snapshot) => self.apply_tasks_snapshot(snapshot, cx), + Err(err) => log::warn!( + "malformed _claude/tasks for session {}: {}", + self.session_id(), + err + ), + } + } self.update_session_info(info_update, cx); } acp_v1::SessionUpdate::AvailableCommandsUpdate(acp_v1::AvailableCommandsUpdate { @@ -8905,6 +9149,9 @@ impl AcpThread { } pub fn emit_load_error(&mut self, error: LoadError, cx: &mut Context) { + // Every caller reports the agent process exiting, so no task can end normally + // from here on. + self.interrupt_running_tasks(cx); cx.emit(AcpThreadEvent::LoadError(error)); } @@ -27699,4 +27946,7 @@ mod tests { request.await.expect("turn should complete"); assert_eq!(cx.active_idle_sleep_preventions(), 0); } + + include!("acp_thread_tasks_tests.rs"); + include!("acp_thread_tasks_review_tests.rs"); } diff --git a/crates/acp_thread/src/acp_thread_tasks_review_tests.rs b/crates/acp_thread/src/acp_thread_tasks_review_tests.rs new file mode 100644 index 0000000..253e2c4 --- /dev/null +++ b/crates/acp_thread/src/acp_thread_tasks_review_tests.rs @@ -0,0 +1,52 @@ +// Story 012 — regressions found by the tech review of the task model, beyond the +// pre-authored tests. Included after `acp_thread_tasks_tests.rs` (it reuses that +// file's helpers). + +#[gpui::test] +async fn test_claude_tasks_restarted_task_is_announced_again(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + for status in ["running", "completed", "running", "completed"] { + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("x", status, 1_000)]), + cx, + ); + } + + assert_eq!( + events.borrow().ended, + vec![ + ("x".to_string(), "completed"), + ("x".to_string(), "completed") + ] + ); +} + +#[gpui::test] +async fn test_claude_tasks_restarted_task_that_vanishes_ends_as_interrupted( + cx: &mut TestAppContext, +) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + for status in ["running", "completed", "running"] { + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("x", status, 1_000)]), + cx, + ); + } + tasks_send(&thread, tasks_snapshot(vec![]), cx); + + assert_eq!( + events.borrow().ended, + vec![ + ("x".to_string(), "completed"), + ("x".to_string(), "interrupted") + ] + ); + assert_eq!( + tasks_ids(&thread, cx), + vec![("x".to_string(), "interrupted")] + ); +} diff --git a/crates/acp_thread/src/acp_thread_tasks_tests.rs b/crates/acp_thread/src/acp_thread_tasks_tests.rs new file mode 100644 index 0000000..5062ed9 --- /dev/null +++ b/crates/acp_thread/src/acp_thread_tasks_tests.rs @@ -0,0 +1,848 @@ +// Story 012 / patch 0032 — the agent task model in `AcpThread` (R4.1–R4.9) and the +// `_claude/tasks` contract fixture (Q12). +// +// Included at the end of `mod tests` in `acp_thread.rs`: +// +// include!("acp_thread_tasks_tests.rs"); +// +// so it shares that module's imports and `FakeAgentConnection`. Every test name +// carries `tasks` (filter `cargo test -p acp_thread tasks`), the connection actions +// carry `task_actions`, and the fixture test carries `contract`. + +const TASKS_KEY: &str = "_claude/tasks"; +const CONTRACT_FIXTURE: &str = include_str!("test_data/claude_tasks_snapshot.json"); + +fn tasks_ms(ms: u64) -> std::time::SystemTime { + std::time::UNIX_EPOCH + std::time::Duration::from_millis(ms) +} + +/// The wire word for a status: exhaustive, so a sixth variant fails to compile here. +fn tasks_status_word(status: &AgentTaskStatus) -> &'static str { + match status { + AgentTaskStatus::Running => "running", + AgentTaskStatus::Completed => "completed", + AgentTaskStatus::Failed => "failed", + AgentTaskStatus::Stopped => "stopped", + AgentTaskStatus::Interrupted => "interrupted", + } +} + +/// One wire task. `status` is the wire word; `ended_at` is set for ended statuses. +fn tasks_wire_task(id: &str, status: &str, started_at: u64) -> serde_json::Value { + let ended = status != "running"; + serde_json::json!({ + "id": id, + "toolCallId": format!("toolu_{id}"), + "type": "shell", + "description": format!("Task {id}"), + "background": true, + "depth": 0, + "status": status, + "startedAt": started_at, + "endedAt": if ended { Some(started_at + 1_000) } else { None }, + "usage": null, + "lastTool": null, + "summary": null, + }) +} + +fn tasks_update(payload: serde_json::Value) -> acp_v1::SessionInfoUpdate { + let mut meta = serde_json::Map::new(); + meta.insert(TASKS_KEY.to_string(), payload); + acp_v1::SessionInfoUpdate::new().meta(meta) +} + +fn tasks_snapshot(tasks: Vec) -> acp_v1::SessionInfoUpdate { + tasks_update(serde_json::json!({ "tasks": tasks })) +} + +#[derive(Default)] +struct TasksEvents { + updated: usize, + ended: Vec<(String, &'static str)>, +} + +async fn tasks_thread( + connection: Rc, + cx: &mut TestAppContext, +) -> (Entity, Rc>) { + init_test(cx); + let fs = FakeFs::new(cx.executor()); + let project = Project::test(fs, [], cx).await; + let thread = cx + .update(|cx| { + connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx) + }) + .await + .unwrap(); + + let events = Rc::new(RefCell::new(TasksEvents::default())); + let sink = events.clone(); + thread.update(cx, |_thread, cx| { + cx.subscribe( + &thread, + move |_thread, _event_thread, event: &AcpThreadEvent, _cx| match event { + AcpThreadEvent::TasksUpdated => sink.borrow_mut().updated += 1, + AcpThreadEvent::TaskEnded { task_id, status } => sink + .borrow_mut() + .ended + .push((task_id.to_string(), tasks_status_word(status))), + _ => {} + }, + ) + .detach(); + }); + (thread, events) +} + +fn tasks_send( + thread: &Entity, + update: acp_v1::SessionInfoUpdate, + cx: &mut TestAppContext, +) { + thread + .update(cx, |thread, cx| { + thread.handle_session_update(acp_v1::SessionUpdate::SessionInfoUpdate(update), cx) + }) + .expect("session info update should succeed"); +} + +fn tasks_ids(thread: &Entity, cx: &mut TestAppContext) -> Vec<(String, &'static str)> { + thread.read_with(cx, |thread, _| { + thread + .tasks() + .iter() + .map(|task| (task.id.to_string(), tasks_status_word(&task.status))) + .collect() + }) +} + +// ---- R4.1 ------------------------------------------------------------------------- + +#[gpui::test] +async fn test_claude_tasks_snapshot_replaces_the_list_and_notifies(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("t1", "running", 1_000), + tasks_wire_task("done", "completed", 2_000), + ]), + cx, + ); + assert_eq!( + tasks_ids(&thread, cx), + vec![ + ("t1".to_string(), "running"), + ("done".to_string(), "completed") + ] + ); + assert_eq!(events.borrow().updated, 1); + + // A level, not a stream: the next snapshot replaces the list, it does not merge. + // The ended task the adapter no longer sends is gone; t1's fields are the new ones. + let mut t1 = tasks_wire_task("t1", "running", 1_000); + t1["description"] = serde_json::json!("Renamed task"); + tasks_send( + &thread, + tasks_snapshot(vec![t1, tasks_wire_task("t3", "running", 3_000)]), + cx, + ); + assert_eq!( + tasks_ids(&thread, cx), + vec![("t1".to_string(), "running"), ("t3".to_string(), "running")] + ); + thread.read_with(cx, |thread, _| { + assert_eq!(thread.tasks()[0].description.to_string(), "Renamed task") + }); + assert_eq!(events.borrow().updated, 2); +} + +#[gpui::test] +async fn test_claude_tasks_fields_reach_the_model(cx: &mut TestAppContext) { + let (thread, _events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + tasks_send( + &thread, + tasks_snapshot(vec![serde_json::json!({ + "id": "a1", + "toolCallId": null, + "type": "subagent", + "description": "Explore", + "background": false, + "depth": 2, + "status": "running", + "startedAt": 1_759_500_000_000u64, + "endedAt": null, + "usage": { "tokens": 10, "toolUses": 1, "durationMs": 500 }, + "lastTool": "Read", + "summary": "Reading", + })]), + cx, + ); + + thread.read_with(cx, |thread, _| { + let task = &thread.tasks()[0]; + assert_eq!(task.tool_call_id, None); + assert_eq!(task.kind.to_string(), "subagent"); + assert_eq!(task.description.to_string(), "Explore"); + assert!(!task.background); + assert_eq!(task.depth, 2); + assert_eq!(task.started_at, tasks_ms(1_759_500_000_000)); + assert_eq!(task.ended_at, None); + assert!(task.usage.is_some()); + assert_eq!( + task.last_tool.as_ref().map(|s| s.to_string()), + Some("Read".to_string()) + ); + assert_eq!( + task.summary.as_ref().map(|s| s.to_string()), + Some("Reading".to_string()) + ); + }); +} + +// ---- R4.2 — hostile half first: an absent key must not read as "no tasks" ---------- + +#[gpui::test] +async fn test_claude_tasks_absent_key_leaves_the_list_and_present_empty_clears_it( + cx: &mut TestAppContext, +) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("r1", "running", 1_000), + tasks_wire_task("d1", "completed", 2_000), + ]), + cx, + ); + assert_eq!(events.borrow().updated, 1); + assert_eq!(events.borrow().ended, vec![("d1".to_string(), "completed")]); + + // A title-only update rides the same notification. + tasks_send( + &thread, + acp_v1::SessionInfoUpdate::new().title("Renamed"), + cx, + ); + // So does an update carrying only patch 0029's key. + let mut only_background = serde_json::Map::new(); + only_background.insert( + CLAUDE_BACKGROUND_TASKS_META_KEY.to_string(), + serde_json::json!({ "count": 0, "tasks": [] }), + ); + tasks_send( + &thread, + acp_v1::SessionInfoUpdate::new().meta(only_background), + cx, + ); + // And an update with unrelated meta. + let mut unrelated = serde_json::Map::new(); + unrelated.insert( + "_claude/other".to_string(), + serde_json::json!({ "tasks": [] }), + ); + tasks_send( + &thread, + acp_v1::SessionInfoUpdate::new().meta(unrelated), + cx, + ); + + // Nothing changed: the running task did not "vanish" (no R4.10 interrupt either). + assert_eq!( + tasks_ids(&thread, cx), + vec![ + ("r1".to_string(), "running"), + ("d1".to_string(), "completed") + ] + ); + assert_eq!(events.borrow().updated, 1, "no change, no notification"); + assert_eq!(events.borrow().ended, vec![("d1".to_string(), "completed")]); + + // The converse: a PRESENT empty list is real news and clears the list. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("r1", "completed", 1_000)]), + cx, + ); + tasks_send(&thread, tasks_snapshot(vec![]), cx); + assert_eq!(tasks_ids(&thread, cx), vec![]); + assert_eq!(events.borrow().updated, 3); +} + +// ---- R4.3 ------------------------------------------------------------------------- + +#[gpui::test] +async fn test_claude_tasks_malformed_payload_keeps_the_previous_list(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("t1", "running", 1_000)]), + cx, + ); + + for payload in [ + serde_json::json!("not an object"), + serde_json::json!({ "tasks": "not a list" }), + serde_json::json!({ "tasks": [{ "id": 5 }] }), + serde_json::json!({ "tasks": [{ "id": "t9", "status": "running" }] }), + ] { + tasks_send(&thread, tasks_update(payload), cx); + assert_eq!(tasks_ids(&thread, cx), vec![("t1".to_string(), "running")]); + } + assert_eq!(events.borrow().updated, 1); + assert!(events.borrow().ended.is_empty()); +} + +// ---- R4.4 — once per task, in both directions ------------------------------------- + +#[gpui::test] +async fn test_claude_tasks_ended_is_announced_once_per_task(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("t1", "running", 1_000)]), + cx, + ); + assert!( + events.borrow().ended.is_empty(), + "a running task has not ended" + ); + + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("t1", "completed", 1_000)]), + cx, + ); + // Hostile half: the same ending seen again (a repeated level) must not split + // into a second announcement. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("t1", "completed", 1_000)]), + cx, + ); + assert_eq!(events.borrow().ended, vec![("t1".to_string(), "completed")]); +} + +#[gpui::test] +async fn test_claude_tasks_two_endings_are_two_announcements(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("t1", "running", 1_000), + tasks_wire_task("t2", "running", 2_000), + tasks_wire_task("t3", "running", 3_000), + ]), + cx, + ); + // Converse hostile half: two DIFFERENT tasks ending in one snapshot must not + // collapse into one announcement. + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("t1", "completed", 1_000), + tasks_wire_task("t2", "stopped", 2_000), + tasks_wire_task("t3", "running", 3_000), + ]), + cx, + ); + // A third, later ending is still news after the first two were announced. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("t3", "interrupted", 3_000)]), + cx, + ); + + let mut ended = events.borrow().ended.clone(); + ended.sort(); + assert_eq!( + ended, + vec![ + ("t1".to_string(), "completed"), + ("t2".to_string(), "stopped"), + ("t3".to_string(), "interrupted"), + ] + ); +} + +#[gpui::test] +async fn test_claude_tasks_first_seen_ended_is_announced_and_forgotten_once_gone( + cx: &mut TestAppContext, +) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + // Started and ended between two snapshots: first seen already ended. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("quick", "failed", 1_000)]), + cx, + ); + assert_eq!(events.borrow().ended, vec![("quick".to_string(), "failed")]); + + // The adapter omits it from the next snapshot; the announced set forgets it, so + // an id that comes back ended later is a new ending, not a duplicate. + tasks_send(&thread, tasks_snapshot(vec![]), cx); + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("quick", "completed", 5_000)]), + cx, + ); + assert_eq!( + events.borrow().ended, + vec![ + ("quick".to_string(), "failed"), + ("quick".to_string(), "completed") + ] + ); +} + +// ---- R4.8 ------------------------------------------------------------------------- + +#[gpui::test] +async fn test_claude_tasks_and_background_count_keep_separate_keys(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + // An adapter that sends only 0029's key still lights the background status. + let mut only_background = serde_json::Map::new(); + only_background.insert( + CLAUDE_BACKGROUND_TASKS_META_KEY.to_string(), + serde_json::json!({ "count": 2, "tasks": [] }), + ); + tasks_send( + &thread, + acp_v1::SessionInfoUpdate::new().meta(only_background), + cx, + ); + thread.read_with(cx, |thread, _| { + assert_eq!(thread.background_task_count(), 2); + assert!(thread.tasks().is_empty()); + }); + assert_eq!(events.borrow().updated, 0); + + // And a tasks-only update does not reset the count. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("t1", "running", 1_000)]), + cx, + ); + thread.read_with(cx, |thread, _| { + assert_eq!(thread.background_task_count(), 2) + }); +} + +// ---- R4.9 — running tasks keep the thread; ended ones do not ---------------------- + +#[gpui::test] +async fn test_claude_tasks_running_task_keeps_the_thread_out_of_idle_retention( + cx: &mut TestAppContext, +) { + let (thread, _events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + thread.read_with(cx, |thread, _| { + assert_eq!(thread.status(), ThreadStatus::Idle); + assert!(thread.is_idle_for_retention()); + }); + + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("done", "completed", 1_000), + tasks_wire_task("live", "running", 2_000), + ]), + cx, + ); + thread.read_with(cx, |thread, _| assert!(!thread.is_idle_for_retention())); + + // Hostile half: ended tasks alone must not pin the thread — "has tasks" is not + // "has running tasks". + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("live", "completed", 2_000), + tasks_wire_task("other", "failed", 3_000), + ]), + cx, + ); + thread.read_with(cx, |thread, _| assert!(thread.is_idle_for_retention())); +} + +// ---- R4.10 — a running task that vanishes ends once as Interrupted ----------------- + +#[gpui::test] +async fn test_claude_tasks_vanished_running_task_ends_once_as_interrupted(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("r1", "running", 1_000), + tasks_wire_task("r2", "running", 2_000), + ]), + cx, + ); + + // r1's ending snapshot never arrived: it is simply absent from the next one. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("r2", "running", 2_000)]), + cx, + ); + thread.read_with(cx, |thread, _| { + let r1 = thread + .tasks() + .iter() + .find(|task| task.id.as_ref() == "r1") + .expect("the vanished task is kept once, as interrupted"); + assert_eq!(tasks_status_word(&r1.status), "interrupted"); + assert!(r1.ended_at.is_some()); + }); + assert_eq!( + events.borrow().ended, + vec![("r1".to_string(), "interrupted")] + ); + + // Once only: later snapshots without r1 announce nothing more and never revive it. + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("r2", "running", 2_000)]), + cx, + ); + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("r2", "running", 2_000)]), + cx, + ); + assert_eq!( + events.borrow().ended, + vec![("r1".to_string(), "interrupted")] + ); + let r1_running = tasks_ids(&thread, cx) + .into_iter() + .any(|(id, status)| id == "r1" && status == "running"); + assert!(!r1_running); +} + +#[gpui::test] +async fn test_claude_tasks_published_ending_that_leaves_is_not_reannounced( + cx: &mut TestAppContext, +) { + // Hostile half: the adapter omits an ended task after publishing it once (R1.6). + // That absence is NOT a vanished running task and must not become Interrupted. + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("a", "running", 1_000)]), + cx, + ); + tasks_send( + &thread, + tasks_snapshot(vec![tasks_wire_task("a", "completed", 1_000)]), + cx, + ); + tasks_send(&thread, tasks_snapshot(vec![]), cx); + tasks_send(&thread, tasks_snapshot(vec![]), cx); + + assert_eq!(events.borrow().ended, vec![("a".to_string(), "completed")]); + assert_eq!(tasks_ids(&thread, cx), vec![]); +} + +// ---- R4.11 — the agent process exits --------------------------------------------- + +#[gpui::test] +async fn test_claude_tasks_agent_exit_interrupts_running_tasks_once(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + tasks_send( + &thread, + tasks_snapshot(vec![ + tasks_wire_task("r1", "running", 1_000), + tasks_wire_task("r2", "running", 2_000), + tasks_wire_task("d1", "failed", 3_000), + ]), + cx, + ); + assert_eq!(events.borrow().ended, vec![("d1".to_string(), "failed")]); + + // The path agent_servers takes for every session when the agent process exits. + let exited = || LoadError::Exited { + status: std::process::ExitStatus::default(), + stderr: None, + }; + thread.update(cx, |thread, cx| thread.emit_load_error(exited(), cx)); + thread.update(cx, |thread, cx| thread.emit_load_error(exited(), cx)); + + assert_eq!( + tasks_ids(&thread, cx), + vec![ + ("r1".to_string(), "interrupted"), + ("r2".to_string(), "interrupted"), + ("d1".to_string(), "failed"), + ] + ); + let mut ended = events.borrow().ended.clone(); + ended.sort(); + assert_eq!( + ended, + vec![ + ("d1".to_string(), "failed"), + ("r1".to_string(), "interrupted"), + ("r2".to_string(), "interrupted"), + ] + ); + thread.read_with(cx, |thread, _| assert!(thread.is_idle_for_retention())); +} + +// ---- Q12 — the shared fixture ------------------------------------------------------ + +#[gpui::test] +async fn test_tasks_contract_fixture_parses(cx: &mut TestAppContext) { + let (thread, events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + let payload: serde_json::Value = + serde_json::from_str(CONTRACT_FIXTURE).expect("fixture is JSON"); + + tasks_send(&thread, tasks_update(payload), cx); + + thread.read_with(cx, |thread, _| { + let tasks = thread.tasks(); + assert_eq!(tasks.len(), 3, "every fixture task parses"); + + let shell = &tasks[0]; + assert_eq!(shell.id.to_string(), "b7k2m9x1q"); + assert_eq!( + shell.tool_call_id, + Some(acp_v2::ToolCallId::new("toolu_01ShellSleep20")) + ); + assert_eq!(shell.kind.to_string(), "shell"); + assert_eq!( + shell.description.to_string(), + "Run sleep 20 in the background" + ); + assert!(shell.background); + assert_eq!(shell.depth, 0); + assert_eq!(tasks_status_word(&shell.status), "running"); + assert_eq!(shell.started_at, tasks_ms(1_759_500_000_000)); + assert_eq!(shell.ended_at, None); + assert!(shell.usage.is_none()); + assert!(shell.last_tool.is_none()); + assert!(shell.summary.is_none()); + + let subagent = &tasks[1]; + assert_eq!(subagent.id.to_string(), "a3f8c2d4e5b6"); + assert_eq!(subagent.kind.to_string(), "subagent"); + assert!(!subagent.background); + assert_eq!(subagent.depth, 1); + assert!(subagent.usage.is_some()); + assert_eq!( + subagent.last_tool.as_ref().map(|s| s.to_string()), + Some("Grep".to_string()) + ); + assert_eq!( + subagent.summary.as_ref().map(|s| s.to_string()), + Some("Reading the SDK task message types".to_string()) + ); + + let ended = &tasks[2]; + assert_eq!(ended.id.to_string(), "c9d1e2f3g"); + assert_eq!(tasks_status_word(&ended.status), "completed"); + assert_eq!(ended.started_at, tasks_ms(1_759_500_002_000)); + assert_eq!(ended.ended_at, Some(tasks_ms(1_759_500_005_000))); + assert!( + ended + .summary + .as_ref() + .is_some_and(|s| s.to_string().contains("exit code 0")) + ); + }); + assert_eq!(events.borrow().updated, 1); + assert_eq!( + events.borrow().ended, + vec![("c9d1e2f3g".to_string(), "completed")] + ); +} + +// ---- R4.5, R4.6, R4.7 — the thread's task actions --------------------------------- + +/// A connection that overrides the two task actions and records what it was asked. +struct TaskActionsConnection { + sessions: Arc>>>, + stop_calls: Rc>>, + background_calls: Rc>>, + stop_error: Option<&'static str>, + backgrounded: bool, +} + +impl TaskActionsConnection { + fn new() -> Self { + Self { + sessions: Arc::default(), + stop_calls: Rc::default(), + background_calls: Rc::default(), + stop_error: None, + backgrounded: true, + } + } +} + +impl AgentConnection for TaskActionsConnection { + fn agent_id(&self) -> AgentId { + AgentId::new("task-actions") + } + + fn telemetry_id(&self) -> SharedString { + "task-actions".into() + } + + fn auth_methods(&self) -> &[acp_v2::AuthMethod] { + &[] + } + + fn new_session( + self: Rc, + project: Entity, + work_dirs: PathList, + cx: &mut App, + ) -> Task>> { + let session_id = acp_v2::SessionId::new("task-actions-session"); + let action_log = cx.new(|_| ActionLog::new(project.clone())); + let thread = cx.new(|cx| { + AcpThread::new( + None, + None, + Some(work_dirs), + self.clone(), + project, + action_log, + session_id.clone(), + watch::Receiver::constant(acp_v2::PromptCapabilities::new()), + cx, + ) + }); + self.sessions.lock().insert(session_id, thread.downgrade()); + Task::ready(Ok(thread)) + } + + fn authenticate(&self, _method: acp_v2::AuthMethodId, _cx: &mut App) -> Task> { + Task::ready(Ok(())) + } + + fn prompt( + &self, + _params: acp_v2::PromptRequest, + _cx: &mut App, + ) -> Task> { + Task::ready(Ok(acp_v1::PromptResponse::new(acp_v1::StopReason::EndTurn))) + } + + fn cancel(&self, _session_id: &acp_v2::SessionId, _cx: &mut App) -> gpui::Result<()> { + Ok(()) + } + + fn stop_task( + self: Rc, + session_id: &acp_v2::SessionId, + task_id: &str, + _cx: &mut App, + ) -> Task> { + self.stop_calls + .borrow_mut() + .push((session_id.clone(), task_id.to_string())); + match self.stop_error { + Some(message) => Task::ready(Err(anyhow::anyhow!( + "_claude/tasks/stop failed for {task_id}: {message}" + ))), + None => Task::ready(Ok(())), + } + } + + fn background_task( + self: Rc, + session_id: &acp_v2::SessionId, + tool_call_id: &str, + _cx: &mut App, + ) -> Task> { + self.background_calls + .borrow_mut() + .push((session_id.clone(), tool_call_id.to_string())); + Task::ready(Ok(self.backgrounded)) + } + + fn into_any(self: Rc) -> Rc { + self + } +} + +#[gpui::test] +async fn test_task_actions_thread_delegates_to_its_connection(cx: &mut TestAppContext) { + let connection = Rc::new(TaskActionsConnection::new()); + let stop_calls = connection.stop_calls.clone(); + let background_calls = connection.background_calls.clone(); + let (thread, _events) = tasks_thread(connection, cx).await; + let session_id = acp_v2::SessionId::new("task-actions-session"); + + thread + .update(cx, |thread, cx| thread.stop_task("b1", cx)) + .await + .expect("stop succeeds"); + let moved = thread + .update(cx, |thread, cx| thread.background_task("toolu_fg", cx)) + .await + .expect("background succeeds"); + + assert!(moved); + assert_eq!( + *stop_calls.borrow(), + vec![(session_id.clone(), "b1".to_string())] + ); + assert_eq!( + *background_calls.borrow(), + vec![(session_id, "toolu_fg".to_string())] + ); +} + +#[gpui::test] +async fn test_task_actions_thread_returns_the_agents_error_and_false(cx: &mut TestAppContext) { + let mut connection = TaskActionsConnection::new(); + connection.stop_error = Some("task b1 is not running"); + connection.backgrounded = false; + let (thread, _events) = tasks_thread(Rc::new(connection), cx).await; + + let error = thread + .update(cx, |thread, cx| thread.stop_task("b1", cx)) + .await + .expect_err("the agent's error reaches the caller"); + assert!( + error.to_string().contains("task b1 is not running"), + "{error}" + ); + + let moved = thread + .update(cx, |thread, cx| thread.background_task("toolu_none", cx)) + .await + .expect("an answer of false is not an error"); + assert!(!moved); +} + +#[gpui::test] +async fn test_task_actions_unsupported_for_an_agent_without_them(cx: &mut TestAppContext) { + // FakeAgentConnection does not override the actions: the trait defaults answer. + let (thread, _events) = tasks_thread(Rc::new(FakeAgentConnection::new()), cx).await; + + thread.read_with(cx, |thread, _| assert!(thread.tasks().is_empty())); + let stop = thread + .update(cx, |thread, cx| thread.stop_task("b1", cx)) + .await + .expect_err("stop is unsupported"); + let background = thread + .update(cx, |thread, cx| thread.background_task("toolu_fg", cx)) + .await + .expect_err("background is unsupported"); + for error in [stop, background] { + let text = error.to_string().to_lowercase(); + assert!( + text.contains("not supported") || text.contains("unsupported"), + "{text}" + ); + } +} diff --git a/crates/acp_thread/src/connection.rs b/crates/acp_thread/src/connection.rs index bc4e819..5b0f0b2 100644 --- a/crates/acp_thread/src/connection.rs +++ b/crates/acp_thread/src/connection.rs @@ -164,6 +164,31 @@ pub trait AgentConnection { Task::ready(Err(anyhow::Error::msg("Closing sessions is not supported"))) } + /// Ask the agent to stop one of the session's tasks (`AcpThread::tasks`). + fn stop_task( + self: Rc, + _session_id: &acp_v2::SessionId, + _task_id: &str, + _cx: &mut App, + ) -> Task> { + Task::ready(Err(anyhow::Error::msg( + "Stopping agent tasks is not supported", + ))) + } + + /// Ask the agent to move the foreground tool call `tool_call_id` to the background. + /// `Ok(false)` means the agent answered but had nothing to move. + fn background_task( + self: Rc, + _session_id: &acp_v2::SessionId, + _tool_call_id: &str, + _cx: &mut App, + ) -> Task> { + Task::ready(Err(anyhow::Error::msg( + "Backgrounding agent tasks is not supported", + ))) + } + /// Whether this agent supports resuming existing sessions without loading history. fn supports_resume_session(&self) -> bool { false diff --git a/crates/acp_thread/src/subagent_rows.rs b/crates/acp_thread/src/subagent_rows.rs new file mode 100644 index 0000000..75fb83f --- /dev/null +++ b/crates/acp_thread/src/subagent_rows.rs @@ -0,0 +1,155 @@ +//! Zeo's task surfaces read ACP subagents from Zed's own registry (`AcpThread::subagents()`), +//! so a child is listed whichever adapter spawned it. Story 064; the rows replace nothing +//! upstream draws — the inline card keeps reading the child directly. + +use std::time::{Instant, SystemTime}; + +use agent_client_protocol::v2::schema as acp_v2; +use gpui::{App, SharedString}; + +use crate::{AcpThread, ForegroundActivity}; + +/// Six outcomes, because the card's four (`Running`, `Completed`, `Failed`, `Canceled`) fold +/// `RequiresAction` into Running and an unconfirmed state into the idle default — and both +/// the sidebar and the tray have to tell those apart (R1.3). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SubagentRowStatus { + Running, + Waiting, + Done, + Failed, + Canceled, + Unknown, +} + +impl SubagentRowStatus { + pub fn is_active(self) -> bool { + matches!(self, Self::Running | Self::Waiting) + } + + pub fn is_terminal(self) -> bool { + matches!(self, Self::Done | Self::Failed | Self::Canceled) + } +} + +#[derive(Clone, Debug)] +pub struct SubagentRow { + pub child_session_id: acp_v2::SessionId, + pub label: SharedString, + pub status: SubagentRowStatus, + /// The agent's own cancel grant (`capabilities.cancel`); Zed never forces a stop. + pub can_cancel: bool, + pub error: Option, + /// Wall-clock, like `AgentTask::started_at`, so merged rows sort and persist alike. + pub started_at: Option, +} + +/// What a thread's children add to its own state in the sidebar. Waiting outranks Running: +/// the thread waits for an answer while any child does (R4.2). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ChildActivity { + None, + Running, + Waiting, +} + +/// One row per registered child, in registration order. Forks and sub-threads name a parent +/// too, but are never registered, so they never appear here. +pub fn subagent_rows(parent: &AcpThread, cx: &App) -> Vec { + parent + .subagents() + .map(|child| subagent_row(child.read(cx))) + .collect() +} + +pub fn any_child_active(parent: &AcpThread, cx: &App) -> ChildActivity { + let mut activity = ChildActivity::None; + for child in parent.subagents() { + match subagent_row_status(child.read(cx)).0 { + SubagentRowStatus::Waiting => return ChildActivity::Waiting, + SubagentRowStatus::Running => activity = ChildActivity::Running, + _ => {} + } + } + activity +} + +fn subagent_row(child: &AcpThread) -> SubagentRow { + let association = child.subagent_association(); + // The parent's label describes the child in the parent's view, so it wins (R1.2). + let label = association + .and_then(|association| association.title.clone()) + .or_else(|| child.title()) + .unwrap_or_else(|| SharedString::new_static("Subagent")); + let (status, error) = subagent_row_status(child); + SubagentRow { + child_session_id: child.session_id().clone(), + label, + status, + can_cancel: association.is_some_and(|association| association.can_cancel), + error, + started_at: child.activity_started_at().map(wall_clock), + } +} + +fn subagent_row_status(child: &AcpThread) -> (SubagentRowStatus, Option) { + match child.foreground_activity() { + ForegroundActivity::Running => (SubagentRowStatus::Running, None), + ForegroundActivity::RequiresAction => (SubagentRowStatus::Waiting, None), + // A replayed or not-yet-reported state is history, not the child's current one (R1.7). + ForegroundActivity::Unconfirmed | ForegroundActivity::Unknown => { + (SubagentRowStatus::Unknown, None) + } + ForegroundActivity::Idle => match child.foreground_state() { + Some(acp_v2::StateUpdate::Idle(idle)) => idle_outcome(idle), + _ => (SubagentRowStatus::Unknown, None), + }, + } +} + +fn wall_clock(instant: Instant) -> SystemTime { + SystemTime::now() + .checked_sub(instant.elapsed()) + .unwrap_or(SystemTime::UNIX_EPOCH) +} + +/// How an ACP subagent's work ended, with the error to show for a failure. Shared with the +/// inline subagent card so the two never disagree about an outcome. Idle with no stop reason, +/// or with one this version does not know, is a normal end, as it is for any ACP session. +pub fn idle_outcome(idle: &acp_v2::IdleStateUpdate) -> (SubagentRowStatus, Option) { + match &idle.stop_reason { + Some(acp_v2::StopReason::Cancelled) => (SubagentRowStatus::Canceled, None), + Some(acp_v2::StopReason::Error(error)) => ( + SubagentRowStatus::Failed, + error + .error + .as_ref() + .map(|error| error.message.trim()) + .filter(|message| !message.is_empty()) + .map(|message| message.to_string().into()), + ), + Some(acp_v2::StopReason::MaxTokens) => ( + SubagentRowStatus::Failed, + Some("The subagent reached its token limit.".into()), + ), + Some(acp_v2::StopReason::MaxTurnRequests) => ( + SubagentRowStatus::Failed, + Some("The subagent reached its limit of model requests.".into()), + ), + Some(acp_v2::StopReason::Refusal) => ( + SubagentRowStatus::Failed, + Some("The subagent's model refused to continue.".into()), + ), + _ => (SubagentRowStatus::Done, None), + } +} + +#[cfg(test)] +#[path = "subagent_rows_identity_tests.rs"] +mod identity_tests; +#[cfg(test)] +#[path = "subagent_rows_test_support.rs"] +mod test_support; +#[cfg(test)] +#[path = "subagent_rows_tests.rs"] +mod tests; diff --git a/crates/acp_thread/src/subagent_rows_identity_tests.rs b/crates/acp_thread/src/subagent_rows_identity_tests.rs new file mode 100644 index 0000000..d358529 --- /dev/null +++ b/crates/acp_thread/src/subagent_rows_identity_tests.rs @@ -0,0 +1,259 @@ +// Story 064, task 1.1 (second half) — labels and identity (R1.2), replay (R1.7) and the +// children's combined activity (`any_child_active`, read by R4.1/R4.2). Contract, assumptions and +// wiring: see the header of `subagent_rows_tests.rs`. Filter: `cargo test -p acp_thread +// subagent_rows`. + +use action_log::ActionLog; +use agent_client_protocol::v2::schema as acp_v2; +use gpui::{AppContext as _, TestAppContext}; + +use super::test_support::*; +use super::{ChildActivity, SubagentRowStatus}; +use crate::AcpThread; + +// ── R1.2: labels and identity ────────────────────────────────────────────── + +// The parent's label for the child describes it in the parent's view and wins over the +// child's own title. +#[gpui::test] +async fn subagent_rows_association_title_wins_over_the_childs_own_title(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let child = subagent_rows_child(&rig, "c-1", Some("Child's own title"), cx); + subagent_rows_associate( + &child, + serde_json::json!({ "title": "Explore the parser" }), + cx, + ); + + let row = subagent_rows_find(&subagent_rows_of(&rig, cx), "c-1"); + assert_eq!(row.label.as_ref(), "Explore the parser"); +} + +// Two distinct children that share a title are two rows, each named by its own session. +#[gpui::test] +async fn subagent_rows_two_children_with_one_title_are_two_rows(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let first = subagent_rows_child(&rig, "c-a", Some("Same"), cx); + let second = subagent_rows_child(&rig, "c-b", Some("Same"), cx); + for child in [&first, &second] { + subagent_rows_associate(child, serde_json::json!({ "title": "Explore" }), cx); + } + subagent_rows_report(&first, subagent_rows_running(), cx); + subagent_rows_report(&second, subagent_rows_requires_action(), cx); + + let rows = subagent_rows_of(&rig, cx); + let ids: Vec = rows + .iter() + .map(|row| row.child_session_id.to_string()) + .collect(); + assert_eq!( + ids, + ["c-a", "c-b"], + "one row per child, in registration order" + ); + assert!(rows.iter().all(|row| row.label.as_ref() == "Explore")); + assert_eq!( + subagent_rows_find(&rows, "c-a").status, + SubagentRowStatus::Running + ); + assert_eq!( + subagent_rows_find(&rows, "c-b").status, + SubagentRowStatus::Waiting, + "each row reads its own child, never its namesake's" + ); +} + +// Fallbacks: a cleared association title gives the child's title; neither gives "Subagent". +#[gpui::test] +async fn subagent_rows_label_falls_back_to_child_title_then_subagent(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let titled = subagent_rows_child(&rig, "c-titled", Some("Child's own title"), cx); + subagent_rows_associate( + &titled, + serde_json::json!({ "title": "Parent's label" }), + cx, + ); + subagent_rows_associate(&titled, serde_json::json!({ "title": null }), cx); + let _bare = subagent_rows_child(&rig, "c-bare", None, cx); + + let rows = subagent_rows_of(&rig, cx); + assert_eq!( + subagent_rows_find(&rows, "c-titled").label.as_ref(), + "Child's own title", + "a cleared association title falls back to the child's title" + ); + assert_eq!( + subagent_rows_find(&rows, "c-bare").label.as_ref(), + "Subagent" + ); +} + +// The Stop grant is the association's `can_cancel`, never assumed. +#[gpui::test] +async fn subagent_rows_can_cancel_follows_the_association(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let granted = subagent_rows_child(&rig, "c-granted", None, cx); + let _refused = subagent_rows_child(&rig, "c-refused", None, cx); + subagent_rows_associate( + &granted, + serde_json::json!({ "capabilities": { "cancel": {} } }), + cx, + ); + + let rows = subagent_rows_of(&rig, cx); + assert!(subagent_rows_find(&rows, "c-granted").can_cancel); + assert!(!subagent_rows_find(&rows, "c-refused").can_cancel); +} + +// Only registered children are rows: a thread that names the parent but was never registered +// (a fork, 0070, or a sub-thread) is not a subagent. +#[gpui::test] +async fn subagent_rows_unregistered_threads_are_not_rows(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + assert!( + subagent_rows_of(&rig, cx).is_empty(), + "no children, no rows" + ); + assert_eq!(subagent_rows_activity(&rig, cx), ChildActivity::None); + + let (parent_session_id, project) = rig.parent.read_with(cx, |thread, _| { + (thread.session_id().clone(), thread.project().clone()) + }); + let connection = rig.connection.clone(); + let _stray = cx.update(|cx| { + let action_log = cx.new(|_| ActionLog::new(project.clone())); + cx.new(|cx| { + AcpThread::new( + Some(parent_session_id), + Some("Not registered".into()), + None, + connection, + project, + action_log, + acp_v2::SessionId::new("c-stray"), + watch::Receiver::constant(acp_v2::PromptCapabilities::new()), + cx, + ) + }) + }); + let registered = subagent_rows_child(&rig, "c-registered", None, cx); + subagent_rows_report(®istered, subagent_rows_running(), cx); + + let ids: Vec = subagent_rows_of(&rig, cx) + .iter() + .map(|row| row.child_session_id.to_string()) + .collect(); + assert_eq!(ids, ["c-registered"]); +} + +// ── R1.7: replay ─────────────────────────────────────────────────────────── + +// A load replays the children's states as history: afterwards a child the replay left Done or +// Running reads Unknown, keeps its title, and cannot be stopped until the agent reports again. +#[gpui::test] +async fn subagent_rows_replayed_children_read_unknown_never_done_or_running( + cx: &mut TestAppContext, +) { + let rig = subagent_rows_rig(cx).await; + let done = subagent_rows_child(&rig, "c-done", None, cx); + let running = subagent_rows_child(&rig, "c-running", None, cx); + for (child, title) in [(&done, "Replayed done"), (&running, "Replayed running")] { + subagent_rows_associate( + child, + serde_json::json!({ "title": title, "capabilities": { "cancel": {} } }), + cx, + ); + subagent_rows_report(child, subagent_rows_running(), cx); + } + subagent_rows_report(&done, subagent_rows_idle(None), cx); + let before = subagent_rows_of(&rig, cx); + assert_eq!( + subagent_rows_find(&before, "c-done").status, + SubagentRowStatus::Done, + "precondition: the replay left one child Done" + ); + assert_eq!( + subagent_rows_find(&before, "c-running").status, + SubagentRowStatus::Running, + "precondition: and one Running" + ); + + rig.parent.update(cx, |parent, cx| { + parent.invalidate_replayed_subagent_state(cx) + }); + + let after = subagent_rows_of(&rig, cx); + for (session_id, title) in [ + ("c-done", "Replayed done"), + ("c-running", "Replayed running"), + ] { + let row = subagent_rows_find(&after, session_id); + assert_eq!( + row.status, + SubagentRowStatus::Unknown, + "{session_id}: a replayed state is history, not the child's current state" + ); + assert_eq!(row.label.as_ref(), title, "{session_id}: the title stays"); + assert!( + !row.can_cancel, + "{session_id}: a replayed cancel grant is withdrawn" + ); + } + assert_eq!( + subagent_rows_activity(&rig, cx), + ChildActivity::None, + "replayed children never keep the thread in Background" + ); +} + +// ── any_child_active (feeds R4.1, R4.2) ─────────────────────────────────── + +// Waiting must not collapse into Running when both are present: the thread is waiting for an +// answer while any child is. +#[gpui::test] +async fn subagent_rows_any_child_active_waiting_wins_over_running(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let runner = subagent_rows_child(&rig, "c-run", None, cx); + let waiter = subagent_rows_child(&rig, "c-wait", None, cx); + subagent_rows_report(&runner, subagent_rows_running(), cx); + subagent_rows_report(&waiter, subagent_rows_running(), cx); + subagent_rows_report(&waiter, subagent_rows_requires_action(), cx); + + assert_eq!(subagent_rows_activity(&rig, cx), ChildActivity::Waiting); +} + +// Only Running and Waiting children count: Done, Failed, Canceled and Unknown ones do not, so an +// idle parent with only those reads None (R4.4). +#[gpui::test] +async fn subagent_rows_any_child_active_ignores_ended_and_unknown_children( + cx: &mut TestAppContext, +) { + let rig = subagent_rows_rig(cx).await; + let done = subagent_rows_child(&rig, "c-done", None, cx); + let failed = subagent_rows_child(&rig, "c-failed", None, cx); + let canceled = subagent_rows_child(&rig, "c-canceled", None, cx); + let _unknown = subagent_rows_child(&rig, "c-unknown", None, cx); + for child in [&done, &failed, &canceled] { + subagent_rows_report(child, subagent_rows_running(), cx); + } + subagent_rows_report(&done, subagent_rows_idle(None), cx); + subagent_rows_report( + &failed, + subagent_rows_idle(Some(subagent_rows_error("boom"))), + cx, + ); + subagent_rows_report( + &canceled, + subagent_rows_idle(Some(acp_v2::StopReason::Cancelled)), + cx, + ); + assert_eq!(subagent_rows_activity(&rig, cx), ChildActivity::None); + + let late = subagent_rows_child(&rig, "c-late", None, cx); + subagent_rows_report(&late, subagent_rows_running(), cx); + assert_eq!( + subagent_rows_activity(&rig, cx), + ChildActivity::Running, + "one running child is enough, however many have ended" + ); +} diff --git a/crates/acp_thread/src/subagent_rows_test_support.rs b/crates/acp_thread/src/subagent_rows_test_support.rs new file mode 100644 index 0000000..269e378 --- /dev/null +++ b/crates/acp_thread/src/subagent_rows_test_support.rs @@ -0,0 +1,165 @@ +// Story 064, task 1.1 — shared rig of the `subagent_rows` tests (see `subagent_rows_tests.rs` for +// the contract). Wired at the end of `crates/acp_thread/src/subagent_rows.rs` as +// +// #[cfg(test)] +// #[path = "subagent_rows_test_support.rs"] +// mod test_support; +// +// Every item is `pub(super)`, so the two test modules beside it reach it as +// `super::test_support::*`. + +use std::{path::Path, rc::Rc}; + +use action_log::ActionLog; +use agent_client_protocol::v2::schema as acp_v2; +use gpui::{AppContext as _, Entity, SharedString, TestAppContext}; +use project::{FakeFs, Project}; +use settings::SettingsStore; +use util::path; +use util::path_list::PathList; + +use super::{ChildActivity, SubagentRow, SubagentRowStatus, any_child_active, subagent_rows}; +use crate::{AcpThread, AgentConnection as _, StubAgentConnection}; + +// ── Rig ───────────────────────────────────────────────────────────────────── + +pub(super) struct SubagentRowsRig { + pub(super) parent: Entity, + pub(super) connection: Rc, +} + +pub(super) async fn subagent_rows_rig(cx: &mut TestAppContext) -> SubagentRowsRig { + cx.update(|cx| { + let mut settings_store = SettingsStore::test(cx); + settings_store.register_setting::(); + cx.set_global(settings_store); + }); + let project = Project::test(FakeFs::new(cx.executor()), [], cx).await; + let connection = Rc::new(StubAgentConnection::new()); + let parent = cx + .update(|cx| { + connection + .clone() + .new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx) + }) + .await + .expect("the parent session should be created"); + SubagentRowsRig { parent, connection } +} + +/// An ACP subagent of the rig's parent, registered as the parent's `subagent_update` does: +/// agent-owned, so its activity comes only from the agent's reports (Unconfirmed until one). +pub(super) fn subagent_rows_child( + rig: &SubagentRowsRig, + session_id: &str, + own_title: Option<&str>, + cx: &mut TestAppContext, +) -> Entity { + let (parent_session_id, project) = rig.parent.read_with(cx, |thread, _| { + (thread.session_id().clone(), thread.project().clone()) + }); + let connection = rig.connection.clone(); + let session_id = acp_v2::SessionId::new(session_id); + let own_title = own_title.map(|title| SharedString::from(title.to_string())); + let child = cx.update(|cx| { + let action_log = cx.new(|_| ActionLog::new(project.clone())); + cx.new(|cx| { + AcpThread::new( + Some(parent_session_id), + own_title, + None, + connection, + project, + action_log, + session_id, + watch::Receiver::constant(acp_v2::PromptCapabilities::new()), + cx, + ) + .agent_owned() + }) + }); + cx.update(|cx| rig.connection.track_session(&child, cx)); + rig.parent + .update(cx, |parent, cx| parent.register_subagent(child.clone(), cx)); + child +} + +/// What the parent's `subagent_update` reports about the child, as the wire carries it. +pub(super) fn subagent_rows_associate( + child: &Entity, + association: serde_json::Value, + cx: &mut TestAppContext, +) { + let session_id = child.read_with(cx, |thread, _| thread.session_id().to_string()); + let mut update = association; + update["sessionId"] = serde_json::json!(session_id); + let update: acp_v2::SubagentUpdate = + serde_json::from_value(update).expect("a well-formed subagent update"); + child.update(cx, |thread, cx| { + thread + .update_subagent_association(&update, cx) + .expect("the child is an ACP subagent") + }); +} + +pub(super) fn subagent_rows_report( + child: &Entity, + state: acp_v2::StateUpdate, + cx: &mut TestAppContext, +) { + child.update(cx, |thread, cx| { + thread + .update_session_state(state, cx) + .expect("an agent-owned child reports its state") + }); +} + +pub(super) fn subagent_rows_running() -> acp_v2::StateUpdate { + acp_v2::StateUpdate::Running(acp_v2::RunningStateUpdate::new()) +} + +pub(super) fn subagent_rows_requires_action() -> acp_v2::StateUpdate { + acp_v2::StateUpdate::RequiresAction(acp_v2::RequiresActionStateUpdate::new()) +} + +pub(super) fn subagent_rows_idle(stop_reason: Option) -> acp_v2::StateUpdate { + let idle = acp_v2::IdleStateUpdate::new(); + acp_v2::StateUpdate::Idle(match stop_reason { + Some(stop_reason) => idle.stop_reason(stop_reason), + None => idle, + }) +} + +pub(super) fn subagent_rows_error(message: &str) -> acp_v2::StopReason { + let mut error = acp_v2::Error::internal_error(); + error.message = message.to_string(); + acp_v2::StopReason::Error(acp_v2::ErrorStopReason::new().error(error)) +} + +pub(super) fn subagent_rows_of(rig: &SubagentRowsRig, cx: &mut TestAppContext) -> Vec { + rig.parent + .read_with(cx, |parent, cx| subagent_rows(parent, cx)) +} + +pub(super) fn subagent_rows_activity( + rig: &SubagentRowsRig, + cx: &mut TestAppContext, +) -> ChildActivity { + rig.parent + .read_with(cx, |parent, cx| any_child_active(parent, cx)) +} + +pub(super) fn subagent_rows_find(rows: &[SubagentRow], session_id: &str) -> SubagentRow { + rows.iter() + .find(|row| row.child_session_id.to_string() == session_id) + .unwrap_or_else(|| panic!("child {session_id} has a row: {rows:?}")) + .clone() +} + +pub(super) fn subagent_rows_status( + rig: &SubagentRowsRig, + session_id: &str, + cx: &mut TestAppContext, +) -> SubagentRowStatus { + subagent_rows_find(&subagent_rows_of(rig, cx), session_id).status +} diff --git a/crates/acp_thread/src/subagent_rows_tests.rs b/crates/acp_thread/src/subagent_rows_tests.rs new file mode 100644 index 0000000..7cf8d08 --- /dev/null +++ b/crates/acp_thread/src/subagent_rows_tests.rs @@ -0,0 +1,190 @@ +// Story 064, task 1.1 — native subagent rows: labels, status mapping, replay (R1.2, R1.3, R1.7) +// and the children's combined activity the sidebar reads (R4.1, R4.2 via `any_child_active`). +// +// Wired at the end of `crates/acp_thread/src/subagent_rows.rs` (new, fixup of 0034) as +// +// #[cfg(test)] +// #[path = "subagent_rows_test_support.rs"] +// mod test_support; +// #[cfg(test)] +// #[path = "subagent_rows_tests.rs"] +// mod tests; +// #[cfg(test)] +// #[path = "subagent_rows_identity_tests.rs"] +// mod identity_tests; +// +// This file holds the status mapping (R1.3); `subagent_rows_identity_tests.rs` holds labels and +// identity (R1.2), replay (R1.7) and `any_child_active`. +// +// Validation filter: `cargo test -p acp_thread subagent_rows`. +// +// Contract (design.md §SubagentRowSource, §Data Models): +// - `pub fn subagent_rows(parent: &AcpThread, cx: &App) -> Vec`, one row per +// registered child (`AcpThread::subagents()`), in registration order; +// - `pub fn any_child_active(parent: &AcpThread, cx: &App) -> ChildActivity`, +// `ChildActivity::{None, Running, Waiting}`; +// - `SubagentRow { child_session_id, label, status, can_cancel, error, started_at }`. +// Assumed beyond design.md (adjust HERE, not in each test): +// - design.md types `status` as `TaskActivity`, a name `agent_ui` already uses for its +// history event (and which `acp_thread` cannot reach). This file names the status enum +// `SubagentRowStatus { Running, Waiting, Done, Failed, Canceled, Unknown }` — R1.3's words — +// declared in `subagent_rows.rs`, deriving `Clone, Copy, Debug, PartialEq, Eq`; +// - `ChildActivity` derives `Debug, PartialEq, Eq`; `SubagentRow` derives `Clone, Debug`; +// - field types: `child_session_id: acp_v2::SessionId`, `label: SharedString`, +// `can_cancel: bool`, `error: Option`. `started_at` is not read here. +// - When `ChildActivity` has both a Running and a Waiting child, it reads Waiting: R4.2 shows +// the thread as waiting for an answer while ANY child is Waiting. +// - An `Unknown` child counts as neither Running nor Waiting (R4.1 and R4.4 name only those +// two). +// +// Hostile halves first, per requirement: +// - R1.3: RequiresAction must not collapse into Running (Zed's own card does exactly that), +// Idle-with-error / limit / refusal must not collapse into Done, Idle-cancelled must not +// collapse into Failed or Done, Unconfirmed / Unknown must not read Done (the idle default) +// nor Running; only then the plain Running and Done rows. +// - R1.2: the association title wins over the child's own title; two children sharing a +// title stay two rows, each with its own session id; then the fallbacks. +// - R1.7: a child the replay left Done or Running reads Unknown afterwards, never Done or +// Running, and loses its Stop grant; its title stays. +// - a thread that names the parent but is not registered (a fork, 0070) is never a row. + +use agent_client_protocol::v2::schema as acp_v2; +use gpui::TestAppContext; + +use super::SubagentRowStatus; +use super::test_support::*; + +// ── R1.3: status mapping, hostile halves first ───────────────────────────── + +// Zed's inline card reads RequiresAction as Running (subagent_card.rs `from_reported_state`): +// the row must not. +#[gpui::test] +async fn subagent_rows_requires_action_reads_waiting_never_running(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let child = subagent_rows_child(&rig, "c-wait", None, cx); + subagent_rows_report(&child, subagent_rows_running(), cx); + assert_eq!( + subagent_rows_status(&rig, "c-wait", cx), + SubagentRowStatus::Running, + "precondition: a running child reads Running" + ); + + subagent_rows_report(&child, subagent_rows_requires_action(), cx); + assert_eq!( + subagent_rows_status(&rig, "c-wait", cx), + SubagentRowStatus::Waiting, + "requires_action is Waiting, distinct from Running" + ); +} + +// Idle is not one outcome: an error, a limit or a refusal is Failed, never Done. +#[gpui::test] +async fn subagent_rows_idle_with_error_limit_or_refusal_reads_failed(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let cases = [ + ("c-error", subagent_rows_error("Authentication required")), + ("c-max-tokens", acp_v2::StopReason::MaxTokens), + ("c-max-turns", acp_v2::StopReason::MaxTurnRequests), + ("c-refusal", acp_v2::StopReason::Refusal), + ]; + for (session_id, stop_reason) in cases.clone() { + let child = subagent_rows_child(&rig, session_id, None, cx); + subagent_rows_report(&child, subagent_rows_running(), cx); + subagent_rows_report(&child, subagent_rows_idle(Some(stop_reason)), cx); + } + + let rows = subagent_rows_of(&rig, cx); + for (session_id, _) in cases { + let row = subagent_rows_find(&rows, session_id); + assert_eq!( + row.status, + SubagentRowStatus::Failed, + "{session_id}: idle with an error, a limit or a refusal is Failed" + ); + assert!( + row.error.is_some(), + "{session_id}: a Failed row says why: {row:?}" + ); + } + let error = subagent_rows_find(&rows, "c-error").error; + assert!( + error + .as_ref() + .is_some_and(|message| message.contains("Authentication required")), + "the agent's own error message is the row's error: {error:?}" + ); +} + +// Cancelled is its own outcome: neither Failed nor Done, and it carries no error. +#[gpui::test] +async fn subagent_rows_idle_cancelled_reads_canceled_not_failed_or_done(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let child = subagent_rows_child(&rig, "c-cancel", None, cx); + subagent_rows_report(&child, subagent_rows_running(), cx); + subagent_rows_report( + &child, + subagent_rows_idle(Some(acp_v2::StopReason::Cancelled)), + cx, + ); + + let row = subagent_rows_find(&subagent_rows_of(&rig, cx), "c-cancel"); + assert_eq!(row.status, SubagentRowStatus::Canceled); + assert_eq!(row.error, None, "a cancellation is not an error"); +} + +// A child that has reported nothing (Unconfirmed), or reported that it cannot tell (Unknown), +// is Unknown: not Done (the session's idle default) and not Running. +#[gpui::test] +async fn subagent_rows_unconfirmed_and_unknown_read_unknown(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let _silent = subagent_rows_child(&rig, "c-silent", None, cx); + let unsure = subagent_rows_child(&rig, "c-unsure", None, cx); + subagent_rows_report(&unsure, subagent_rows_running(), cx); + subagent_rows_report( + &unsure, + acp_v2::StateUpdate::Unknown(acp_v2::UnknownStateUpdate::new()), + cx, + ); + + let rows = subagent_rows_of(&rig, cx); + assert_eq!( + subagent_rows_find(&rows, "c-silent").status, + SubagentRowStatus::Unknown, + "no report yet: Unknown, not the idle default" + ); + assert_eq!( + subagent_rows_find(&rows, "c-unsure").status, + SubagentRowStatus::Unknown, + "reported unknown after running: Unknown, not still Running" + ); +} + +// Benign half: Running, and Idle with no stop reason or `end_turn`, which is Done. +#[gpui::test] +async fn subagent_rows_running_and_plain_idle_read_running_and_done(cx: &mut TestAppContext) { + let rig = subagent_rows_rig(cx).await; + let running = subagent_rows_child(&rig, "c-run", None, cx); + let bare = subagent_rows_child(&rig, "c-bare", None, cx); + let ended = subagent_rows_child(&rig, "c-end", None, cx); + subagent_rows_report(&running, subagent_rows_running(), cx); + for child in [&bare, &ended] { + subagent_rows_report(child, subagent_rows_running(), cx); + } + subagent_rows_report(&bare, subagent_rows_idle(None), cx); + subagent_rows_report( + &ended, + subagent_rows_idle(Some(acp_v2::StopReason::EndTurn)), + cx, + ); + + let rows = subagent_rows_of(&rig, cx); + assert_eq!( + subagent_rows_find(&rows, "c-run").status, + SubagentRowStatus::Running + ); + for session_id in ["c-bare", "c-end"] { + let row = subagent_rows_find(&rows, session_id); + assert_eq!(row.status, SubagentRowStatus::Done, "{session_id}"); + assert_eq!(row.error, None, "{session_id}: Done carries no error"); + } +} diff --git a/crates/acp_thread/src/test_data/claude_tasks_snapshot.json b/crates/acp_thread/src/test_data/claude_tasks_snapshot.json new file mode 100644 index 0000000..3b4d6ad --- /dev/null +++ b/crates/acp_thread/src/test_data/claude_tasks_snapshot.json @@ -0,0 +1,50 @@ +{ + "tasks": [ + { + "id": "b7k2m9x1q", + "toolCallId": "toolu_01ShellSleep20", + "type": "shell", + "description": "Run sleep 20 in the background", + "background": true, + "depth": 0, + "status": "running", + "startedAt": 1759500000000, + "endedAt": null, + "usage": null, + "lastTool": null, + "summary": null + }, + { + "id": "a3f8c2d4e5b6", + "toolCallId": "toolu_01ExploreFeed", + "type": "subagent", + "description": "Explore the task feed", + "background": false, + "depth": 1, + "status": "running", + "startedAt": 1759500001000, + "endedAt": null, + "usage": { + "tokens": 18234, + "toolUses": 7, + "durationMs": 42000 + }, + "lastTool": "Grep", + "summary": "Reading the SDK task message types" + }, + { + "id": "c9d1e2f3g", + "toolCallId": "toolu_01RunLinter", + "type": "shell", + "description": "Run the linter", + "background": true, + "depth": 0, + "status": "completed", + "startedAt": 1759500002000, + "endedAt": 1759500005000, + "usage": null, + "lastTool": null, + "summary": "Background command \"Run the linter\" completed (exit code 0)" + } + ] +} diff --git a/crates/agent_servers/src/acp/v1.rs b/crates/agent_servers/src/acp/v1.rs index 8586a4a..bf421ab 100644 --- a/crates/agent_servers/src/acp/v1.rs +++ b/crates/agent_servers/src/acp/v1.rs @@ -14,7 +14,7 @@ use acp_thread::{ use action_log::ActionLog; use agent_client_protocol::schema::{MaybeUndefined, ProtocolVersion}; use agent_client_protocol::{ - Agent, Client, HandleDispatchFrom, JsonRpcRequest, Lines, + Agent, Client, HandleDispatchFrom, JsonRpcRequest, Lines, UntypedMessage, v1::{Builder, ConnectionTo, Responder}, v2::Builder as V2Builder, }; @@ -24,7 +24,7 @@ use agent_client_protocol::{ }; use anyhow::anyhow; use async_channel; -use collections::HashMap; +use collections::{HashMap, HashSet}; use feature_flags::{AcpBetaFeatureFlag, FeatureFlagAppExt as _}; use futures::channel::mpsc; use futures::future::Shared; @@ -55,6 +55,9 @@ use crate::{CURSOR_ID, GEMINI_ID}; pub const GEMINI_TERMINAL_AUTH_METHOD_ID: &str = "spawn-gemini-cli"; const PARAMETERIZED_MODEL_PICKER_META_KEY: &str = "parameterizedModelPicker"; +const CLAUDE_TASKS_CAPABILITY: &str = "_claude/tasks"; +const CLAUDE_TASKS_STOP_METHOD: &str = "_claude/tasks/stop"; +const CLAUDE_TASKS_BACKGROUND_METHOD: &str = "_claude/tasks/background"; async fn close_session_and_drain( connection: &ConnectionTo, session_id: acp_v2::SessionId, @@ -73,6 +76,25 @@ async fn close_session_and_drain( drain_foreground_queue(dispatch_tx).await } +/// Sends one of the `_claude/tasks/*` extension requests. `id` names the task or tool +/// call the request is about, so the caller's error says which one failed. +async fn send_task_request( + connection: ConnectionTo, + method: &'static str, + params: serde_json::Value, + id: String, +) -> Result { + let failed = |error: agent_client_protocol::Error| { + anyhow!("{method} failed for {id}: {}", error.message) + }; + let request = UntypedMessage::new(method, params).map_err(failed)?; + connection + .send_request(request) + .block_task() + .await + .map_err(failed) +} + pub struct AcpConnection { id: AgentId, telemetry_id: SharedString, @@ -80,6 +102,9 @@ pub struct AcpConnection { connection: ConnectionTo, sessions: Rc>>, pending_sessions: RefCell>, + /// Sessions an explicit `close_session` is closing. The release path skips them, + /// so dropping the thread mid-close does not send a second close request. + closing_sessions: RefCell>, auth_methods: Vec, agent_server_store: WeakEntity, agent_capabilities: acp::AgentCapabilities, @@ -735,6 +760,7 @@ impl AcpConnection { agent_version, sessions, pending_sessions: RefCell::new(HashMap::default()), + closing_sessions: RefCell::new(HashSet::default()), agent_capabilities: response.agent_capabilities, accepts_subagent_updates, request_elicitations, @@ -780,6 +806,7 @@ impl AcpConnection { connection, sessions, pending_sessions: RefCell::new(HashMap::default()), + closing_sessions: RefCell::new(HashSet::default()), auth_methods: vec![], agent_server_store, agent_capabilities, @@ -892,6 +919,7 @@ impl AcpConnection { cx: &mut App, ) { if self.pending_sessions.borrow().contains_key(&session_id) + || self.closing_sessions.borrow().contains(&session_id) || !self .sessions .borrow() @@ -938,6 +966,18 @@ impl AcpConnection { self.agent_capabilities.session_capabilities.close.is_some() } + /// Keyed on the capability alone: the entry id is whatever the user named it, and + /// the agent's self-reported name differs between the fork and the published package. + fn agent_supports_tasks(&self) -> bool { + self.agent_capabilities + .meta + .as_ref() + .and_then(|meta| meta.get(CLAUDE_TASKS_CAPABILITY)) + .and_then(|capability| capability.get("version")) + .and_then(serde_json::Value::as_u64) + .is_some_and(|version| version >= 1) + } + /// `rpc_call` gets a receiver that resolves once its response has been applied, and /// holds the agent's later messages until then (see `ordered_response`). With /// `replays_history`, the response ends a history replay. @@ -2017,6 +2057,130 @@ impl AgentConnection for AcpConnection { Ok(()) } + fn close_session( + self: Rc, + session_id: &acp_v2::SessionId, + cx: &mut App, + ) -> Task> { + let Some(thread) = self + .sessions + .borrow() + .get(session_id) + .map(|session| session.thread.clone()) + else { + return Task::ready(Err(anyhow!("unknown session {session_id}"))); + }; + if !self + .closing_sessions + .borrow_mut() + .insert(session_id.clone()) + { + return Task::ready(Err(anyhow!("session {session_id} is already closing"))); + } + let pending_load = self + .pending_sessions + .borrow() + .get(session_id) + .map(|pending| pending.task.clone()); + let connection = self.connection.clone(); + let dispatch_tx = self.dispatch_tx.clone(); + let supports_close = self.agent_supports_session_close(); + let session_id = session_id.clone(); + let this = Rc::downgrade(&self); + cx.spawn(async move |cx| { + if let Some(pending_load) = pending_load { + pending_load.await.ok(); + } + // A release that was already closing the session (the pending entry + // awaited above) finished the job; a second close would only be refused. + let still_open = this.upgrade().is_some_and(|this| { + this.sessions + .borrow() + .get(&session_id) + .is_some_and(|session| session.thread.entity_id() == thread.entity_id()) + }); + let result = if still_open { + close_session_and_drain( + &connection, + session_id.clone(), + supports_close, + &dispatch_tx, + ) + .await + } else { + Ok(()) + }; + // Interrupt only after the drain: a task snapshot the agent queued before + // closing must not re-open a task already marked interrupted. + thread + .update(cx, |thread, cx| thread.interrupt_running_tasks(cx)) + .ok(); + if let Some(this) = this.upgrade() { + this.remove_session(&session_id, thread.entity_id()); + this.closing_sessions.borrow_mut().remove(&session_id); + } + result + }) + } + + fn stop_task( + self: Rc, + session_id: &acp_v2::SessionId, + task_id: &str, + cx: &mut App, + ) -> Task> { + if !self.agent_supports_tasks() { + return Task::ready(Err(anyhow!( + "Stopping agent tasks is not supported: the agent does not advertise \ + {CLAUDE_TASKS_CAPABILITY}" + ))); + } + let params = serde_json::json!({ "sessionId": session_id, "taskId": task_id }); + let request = send_task_request( + self.connection.clone(), + CLAUDE_TASKS_STOP_METHOD, + params, + task_id.to_string(), + ); + cx.foreground_executor() + .spawn(async move { request.await.map(|_| ()) }) + } + + fn background_task( + self: Rc, + session_id: &acp_v2::SessionId, + tool_call_id: &str, + cx: &mut App, + ) -> Task> { + if !self.agent_supports_tasks() { + return Task::ready(Err(anyhow!( + "Backgrounding agent tasks is not supported: the agent does not advertise \ + {CLAUDE_TASKS_CAPABILITY}" + ))); + } + let params = serde_json::json!({ "sessionId": session_id, "toolCallId": tool_call_id }); + let tool_call_id = tool_call_id.to_string(); + let request = send_task_request( + self.connection.clone(), + CLAUDE_TASKS_BACKGROUND_METHOD, + params, + tool_call_id.clone(), + ); + cx.foreground_executor().spawn(async move { + let response = request.await?; + // A missing field is a protocol mismatch, not "nothing was moved". + response + .get("backgrounded") + .and_then(serde_json::Value::as_bool) + .with_context(|| { + format!( + "{CLAUDE_TASKS_BACKGROUND_METHOD} for {tool_call_id}: \ + response has no boolean `backgrounded`: {response}" + ) + }) + }) + } + fn request_elicitations(&self) -> Option> { Some(self.request_elicitations.clone()) } @@ -2347,6 +2511,26 @@ pub mod test_support { self.inner.cancel(session_id, cx) } + fn stop_task( + self: Rc, + session_id: &acp_v2::SessionId, + task_id: &str, + cx: &mut App, + ) -> Task> { + self.inner.clone().stop_task(session_id, task_id, cx) + } + + fn background_task( + self: Rc, + session_id: &acp_v2::SessionId, + tool_call_id: &str, + cx: &mut App, + ) -> Task> { + self.inner + .clone() + .background_task(session_id, tool_call_id, cx) + } + fn request_elicitations(&self) -> Option> { self.inner.request_elicitations() } @@ -10303,6 +10487,8 @@ exit 7 cx.update(|_| ()); cx.run_until_parked(); } + + include!("../acp_task_actions_tests.rs"); } fn mcp_servers_for_project(project: &Entity, cx: &App) -> Vec { diff --git a/crates/agent_servers/src/acp_task_actions_tests.rs b/crates/agent_servers/src/acp_task_actions_tests.rs new file mode 100644 index 0000000..86f4c71 --- /dev/null +++ b/crates/agent_servers/src/acp_task_actions_tests.rs @@ -0,0 +1,505 @@ +// Story 012 / patch 0032 — `AcpConnection::{stop_task, background_task}` against a +// fake ACP peer (R4.5, R4.6, R4.7, R2.7) and session close interrupting tasks (R4.11). +// +// R4.7's gate is the `_claude/tasks` capability the agent advertises in +// `initialize` (`agentCapabilities._meta["_claude/tasks"].version >= 1`), never the +// entry id or the agent's self-reported name. +// +// Included at the end of `mod tests` in `acp.rs`: +// +// include!("acp_task_actions_tests.rs"); +// +// so it shares that module's imports and may set `AcpConnection`'s private `id` and +// `telemetry_id` (the agent's self-reported name). Filter: `cargo test -p agent_servers +// task_actions`. + +type TaskActionsReply = Arc< + dyn Fn(&str, &serde_json::Value) -> Result + + Send + + Sync, +>; + +struct TaskActionsPeer { + connection: Rc, + project: Entity, + requests: Arc>>, + _keep_agent_alive: Task>, +} + +/// A fake agent that answers initialize, session/new and session/close, records +/// every other request, and answers it with `reply`. `agent_id` is the id Zed +/// registered the agent under; `agent_name` is what the agent reports about itself; +/// `tasks_capability` is the value it advertises under `_claude/tasks`, if any. +async fn connect_task_actions_peer( + agent_id: &str, + agent_name: &str, + tasks_capability: Option, + reply: TaskActionsReply, + cx: &mut gpui::TestAppContext, +) -> TaskActionsPeer { + cx.update(|cx| { + let store = settings::SettingsStore::test(cx); + cx.set_global(store); + }); + + let fs = fs::FakeFs::new(cx.executor()); + fs.insert_tree("/", serde_json::json!({ "a": {} })).await; + let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await; + + let requests: Arc>> = Arc::default(); + let (client_transport, agent_transport) = agent_client_protocol::Channel::duplex(); + let sessions: Rc>> = + Rc::new(RefCell::new(HashMap::default())); + let client_session_list: Rc>>> = Rc::new(RefCell::new(None)); + + let mut capabilities = acp::AgentCapabilities::default().session_capabilities( + acp::SessionCapabilities::default().close(acp::SessionCloseCapabilities::new()), + ); + if let Some(value) = tasks_capability { + let mut meta = serde_json::Map::new(); + meta.insert("_claude/tasks".to_string(), value); + capabilities = capabilities.meta(meta); + } + let agent_info = acp::Implementation::new(agent_name.to_string(), "1.0.0".to_string()); + let agent_future = Agent + .v1() + .name("task-actions-agent") + .on_receive_request( + async move |req: acp::InitializeRequest, responder, _cx| { + responder.respond( + acp::InitializeResponse::new(req.protocol_version) + .agent_capabilities(capabilities.clone()) + .agent_info(agent_info.clone()), + ) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |_req: acp::CloseSessionRequest, responder, _cx| { + responder.respond(acp::CloseSessionResponse::new()) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |_req: acp::NewSessionRequest, responder, _cx| { + responder.respond(acp::NewSessionResponse::new(acp::SessionId::new( + "task-session", + ))) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let requests = requests.clone(); + async move |req: agent_client_protocol::UntypedMessage, responder, _cx| { + requests + .lock() + .expect("requests mutex") + .push((req.method().to_string(), req.params().clone())); + match reply(req.method(), req.params()) { + Ok(value) => responder.respond(value), + Err(error) => responder.respond_with_error(error), + } + } + }, + agent_client_protocol::on_receive_request!(), + ) + .connect_with(agent_transport, async move |_agent| { + futures::future::pending::>().await + }); + let agent_io_task = cx.background_spawn(agent_future); + + let (dispatch_tx, dispatch_rx) = mpsc::unbounded::(); + let (connection_tx, connection_rx) = futures::channel::oneshot::channel(); + let client_future = connect_client_future( + "zed-test", + client_transport, + dispatch_tx.clone(), + connection_tx, + futures::future::pending(), + ); + let client_io_task = cx.background_spawn(async move { + client_future.await.ok(); + }); + let client_conn: ConnectionTo = connection_rx + .await + .expect("failed to receive ACP connection handle"); + let response = client_conn + .send_request(acp::InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await + .expect("failed to initialize ACP connection"); + + let request_elicitations = cx.new(|_| ElicitationStore::default()); + let dispatch_context = ClientContext { + sessions: sessions.clone(), + session_list: client_session_list.clone(), + request_elicitations: request_elicitations.clone(), + active_request_scopes: None, + pending_session_updates: None, + accepts_subagent_updates: false, + }; + let dispatch_task = cx.spawn({ + let mut dispatch_rx = dispatch_rx; + move |cx| async move { + let mut cx = cx; + while let Some(work) = dispatch_rx.next().await { + work.run(&mut cx, &dispatch_context); + } + } + }); + let agent_server_store = + project.read_with(cx, |project, _| project.agent_server_store().downgrade()); + + let mut connection = cx.update(|cx| { + AcpConnection::new_for_test( + client_conn, + sessions, + response.agent_capabilities, + false, + request_elicitations, + agent_server_store, + client_io_task, + dispatch_tx, + dispatch_task, + cx, + ) + }); + connection.id = AgentId::new(agent_id.to_string()); + connection.telemetry_id = SharedString::from(agent_name.to_string()); + + let keep_agent_alive = cx.background_spawn(async move { + agent_io_task.await.ok(); + anyhow::Ok(()) + }); + + TaskActionsPeer { + connection: Rc::new(connection), + project, + requests, + _keep_agent_alive: keep_agent_alive, + } +} + +fn tasks_capability_v(version: u64) -> Option { + Some(serde_json::json!({ "version": version })) +} + +/// Zeo's working entry: a custom id, and the published package's own name. +const PLUS_ENTRY: &str = "Claude Agent (Plus)"; +const PLUS_NAME: &str = "@lucascouts/claude-agent-acp-plus"; + +async fn task_actions_session( + peer: &TaskActionsPeer, + cx: &mut gpui::TestAppContext, +) -> (Entity, acp_v2::SessionId) { + let thread = cx + .update(|cx| { + peer.connection.clone().new_session( + peer.project.clone(), + PathList::new(&[std::path::Path::new("/a")]), + cx, + ) + }) + .await + .expect("new session"); + let session_id = thread.read_with(cx, |thread, _| thread.session_id().clone()); + (thread, session_id) +} + +fn task_actions_requests(peer: &TaskActionsPeer) -> Vec<(String, serde_json::Value)> { + peer.requests.lock().expect("requests mutex").clone() +} + +fn task_actions_ok() -> TaskActionsReply { + Arc::new(|_method, _params| Ok(serde_json::json!({}))) +} + +#[gpui::test] +async fn test_task_actions_stop_sends_the_request_and_succeeds(cx: &mut gpui::TestAppContext) { + let peer = connect_task_actions_peer( + PLUS_ENTRY, + PLUS_NAME, + tasks_capability_v(1), + task_actions_ok(), + cx, + ) + .await; + let (_thread, session_id) = task_actions_session(&peer, cx).await; + + cx.update(|cx| peer.connection.clone().stop_task(&session_id, "b1", cx)) + .await + .expect("stop succeeds"); + + assert_eq!( + task_actions_requests(&peer), + vec![( + "_claude/tasks/stop".to_string(), + serde_json::json!({ "sessionId": session_id.to_string(), "taskId": "b1" }), + )] + ); +} + +#[gpui::test] +async fn test_task_actions_stop_returns_the_agents_error(cx: &mut gpui::TestAppContext) { + let peer = connect_task_actions_peer( + PLUS_ENTRY, + PLUS_NAME, + tasks_capability_v(1), + Arc::new(|_method, _params| { + Err(agent_client_protocol::Error::new( + -32602, + "task b1 is not running", + )) + }), + cx, + ) + .await; + let (_thread, session_id) = task_actions_session(&peer, cx).await; + + let error = cx + .update(|cx| peer.connection.clone().stop_task(&session_id, "b1", cx)) + .await + .expect_err("the agent's JSON-RPC error reaches the caller"); + let text = format!("{error:#}"); + assert!(text.contains("_claude/tasks/stop failed for b1"), "{text}"); + assert!(text.contains("task b1 is not running"), "{text}"); +} + +#[gpui::test] +async fn test_task_actions_background_reads_backgrounded(cx: &mut gpui::TestAppContext) { + let peer = connect_task_actions_peer( + PLUS_ENTRY, + PLUS_NAME, + tasks_capability_v(1), + Arc::new(|_method, params| { + let moved = params.get("toolCallId").and_then(|v| v.as_str()) == Some("toolu_fg"); + Ok(serde_json::json!({ "backgrounded": moved })) + }), + cx, + ) + .await; + let (_thread, session_id) = task_actions_session(&peer, cx).await; + + let moved = cx + .update(|cx| { + peer.connection + .clone() + .background_task(&session_id, "toolu_fg", cx) + }) + .await + .expect("background succeeds"); + let not_moved = cx + .update(|cx| { + peer.connection + .clone() + .background_task(&session_id, "toolu_other", cx) + }) + .await + .expect("false is an answer, not an error"); + + assert!(moved); + assert!(!not_moved); + assert_eq!( + task_actions_requests(&peer)[0], + ( + "_claude/tasks/background".to_string(), + serde_json::json!({ "sessionId": session_id.to_string(), "toolCallId": "toolu_fg" }), + ) + ); +} + +#[gpui::test] +async fn test_task_actions_background_missing_field_is_an_error_not_false( + cx: &mut gpui::TestAppContext, +) { + let peer = connect_task_actions_peer( + PLUS_ENTRY, + PLUS_NAME, + tasks_capability_v(1), + task_actions_ok(), + cx, + ) + .await; + let (_thread, session_id) = task_actions_session(&peer, cx).await; + + let result = cx + .update(|cx| { + peer.connection + .clone() + .background_task(&session_id, "toolu_fg", cx) + }) + .await; + assert!( + result.is_err(), + "a response without `backgrounded` must not read as false" + ); +} + +/// Both actions answer "unsupported" and nothing reaches the agent. +async fn assert_task_actions_unsupported(peer: &TaskActionsPeer, cx: &mut gpui::TestAppContext) { + let (_thread, session_id) = task_actions_session(peer, cx).await; + let stop = cx + .update(|cx| peer.connection.clone().stop_task(&session_id, "b1", cx)) + .await; + let background = cx + .update(|cx| { + peer.connection + .clone() + .background_task(&session_id, "toolu_fg", cx) + }) + .await; + assert!(stop.is_err()); + assert!(background.is_err()); + for error in [stop.unwrap_err(), background.unwrap_err()] { + let text = error.to_string().to_lowercase(); + assert!( + text.contains("not supported") || text.contains("unsupported"), + "{text}" + ); + } + assert!( + task_actions_requests(peer).is_empty(), + "nothing reached the agent" + ); +} + +// R4.7, hostile half first: an agent that LOOKS like the Claude adapter — the +// registry's Claude entry id, the published package's own name — but does not +// advertise the capability is not sent a `_claude/*` request. +#[gpui::test] +async fn test_task_actions_unsupported_without_the_capability_even_for_claude_names( + cx: &mut gpui::TestAppContext, +) { + let peer = connect_task_actions_peer( + crate::CLAUDE_AGENT_ID, + PLUS_NAME, + None, + Arc::new(|_method, _params| Ok(serde_json::json!({ "backgrounded": true }))), + cx, + ) + .await; + assert_task_actions_unsupported(&peer, cx).await; +} + +#[gpui::test] +async fn test_task_actions_unsupported_for_capability_version_zero(cx: &mut gpui::TestAppContext) { + let peer = connect_task_actions_peer( + PLUS_ENTRY, + PLUS_NAME, + tasks_capability_v(0), + task_actions_ok(), + cx, + ) + .await; + assert_task_actions_unsupported(&peer, cx).await; +} + +// R4.7, the converse: any entry id and any name, with the capability, is supported — +// the fork and the published package report different names. +#[gpui::test] +async fn test_task_actions_supported_for_any_name_advertising_the_capability( + cx: &mut gpui::TestAppContext, +) { + for (entry, name, version) in [ + ("some-entry", "@lucascouts/claude-agent-acp-fork", 1), + ("another-entry", "an-unrelated-name", 2), + ] { + let peer = connect_task_actions_peer( + entry, + name, + tasks_capability_v(version), + task_actions_ok(), + cx, + ) + .await; + let (_thread, session_id) = task_actions_session(&peer, cx).await; + cx.update(|cx| peer.connection.clone().stop_task(&session_id, "b1", cx)) + .await + .unwrap_or_else(|error| panic!("{entry}/{name} v{version}: {error:#}")); + assert_eq!(task_actions_requests(&peer).len(), 1); + } +} + +// R4.11, session close: running tasks become Interrupted and are announced once; +// ended tasks are left alone. +#[gpui::test] +async fn test_task_actions_close_session_interrupts_running_tasks_once( + cx: &mut gpui::TestAppContext, +) { + let peer = connect_task_actions_peer( + PLUS_ENTRY, + PLUS_NAME, + tasks_capability_v(1), + task_actions_ok(), + cx, + ) + .await; + let (thread, session_id) = task_actions_session(&peer, cx).await; + + let ended: Rc>> = Rc::default(); + let sink = ended.clone(); + thread.update(cx, |_thread, cx| { + cx.subscribe( + &thread, + move |_thread, _event_thread, event: &acp_thread::AcpThreadEvent, _cx| { + if let acp_thread::AcpThreadEvent::TaskEnded { task_id, status } = event { + sink.borrow_mut().push(( + task_id.to_string(), + matches!(status, acp_thread::AgentTaskStatus::Interrupted), + )); + } + }, + ) + .detach(); + }); + + let task = |id: &str, status: &str| { + serde_json::json!({ + "id": id, "toolCallId": null, "type": "shell", "description": id, + "background": true, "depth": 0, "status": status, "startedAt": 1_000, + "endedAt": if status == "running" { None } else { Some(2_000) }, + "usage": null, "lastTool": null, "summary": null, + }) + }; + let mut meta = serde_json::Map::new(); + meta.insert( + "_claude/tasks".to_string(), + serde_json::json!({ "tasks": [task("r1", "running"), task("d1", "completed")] }), + ); + thread + .update(cx, |thread, cx| { + thread.handle_session_update( + acp::SessionUpdate::SessionInfoUpdate(acp::SessionInfoUpdate::new().meta(meta)), + cx, + ) + }) + .expect("snapshot applies"); + assert_eq!(*ended.borrow(), vec![("d1".to_string(), false)]); + + cx.update(|cx| peer.connection.clone().close_session(&session_id, cx)) + .await + .expect("close succeeds"); + cx.run_until_parked(); + + thread.read_with(cx, |thread, _| { + let statuses: Vec<(String, bool)> = thread + .tasks() + .iter() + .map(|t| { + ( + t.id.to_string(), + matches!(t.status, acp_thread::AgentTaskStatus::Interrupted), + ) + }) + .collect(); + assert_eq!( + statuses, + vec![("r1".to_string(), true), ("d1".to_string(), false)] + ); + }); + assert_eq!( + *ended.borrow(), + vec![("d1".to_string(), false), ("r1".to_string(), true)] + ); +} diff --git a/crates/agent_ui/src/agent_diff.rs b/crates/agent_ui/src/agent_diff.rs index 63b7ec7..2a2aa9d 100644 --- a/crates/agent_ui/src/agent_diff.rs +++ b/crates/agent_ui/src/agent_diff.rs @@ -1490,6 +1490,8 @@ impl AgentDiff { | AcpThreadEvent::TokenUsageUpdated(_) | AcpThreadEvent::AccountUsageUpdated | AcpThreadEvent::BackgroundTasksUpdated + | AcpThreadEvent::TasksUpdated + | AcpThreadEvent::TaskEnded { .. } | AcpThreadEvent::NoticesUpdated | AcpThreadEvent::SubagentSpawned(_) | AcpThreadEvent::SubagentAssociationUpdated diff --git a/crates/agent_ui/src/agent_panel.rs b/crates/agent_ui/src/agent_panel.rs index 45b8ff2..2c51f9c 100644 --- a/crates/agent_ui/src/agent_panel.rs +++ b/crates/agent_ui/src/agent_panel.rs @@ -4315,6 +4315,9 @@ impl AgentPanel { | AcpThreadEvent::SubmissionUpdated(_) | AcpThreadEvent::ToolAuthorizationReceived(_) | AcpThreadEvent::ElicitationResponded(_) + // A thread kept loaded for its running tasks becomes + // releasable when the last one ends (story 012, R4.9). + | AcpThreadEvent::TasksUpdated ) && acp_thread.read(cx).is_idle_for_retention() { this.cleanup_retained_threads(cx); @@ -12274,6 +12277,72 @@ mod tests { }); } + // 0034: a turn that leaves background work running reads Idle, yet unloading the + // thread would drop the only view of that work. With no retained-thread budget, the + // thread stays loaded while a task runs, and is released by the snapshot that ends it. + #[gpui::test] + async fn test_retained_thread_with_a_running_task_survives_cleanup_until_it_ends( + cx: &mut TestAppContext, + ) { + let (panel, mut cx) = setup_panel(cx).await; + let connection = StubAgentConnection::new() + .with_supports_load_session(true) + .with_agent_id("loadable-stub".into()) + .with_telemetry_id("loadable-stub".into()); + let (session_id, thread_id) = + open_generating_thread_with_loadable_connection(&panel, &connection, &mut cx); + open_generating_thread_with_loadable_connection(&panel, &connection, &mut cx); + cx.update(|_window, cx| { + SettingsStore::update_global(cx, |store, cx| { + store + .set_user_settings(r#"{ "agent": { "max_idle_retained_threads": 0 } }"#, cx) + .expect("user settings should load"); + }); + }); + + let tasks = |status: &str| { + let mut meta = serde_json::Map::new(); + meta.insert( + "_claude/tasks".to_string(), + serde_json::json!({ "tasks": [{ + "id": "bg-1", + "toolCallId": "toolu_bg-1", + "type": "shell", + "description": "Background build", + "background": true, + "depth": 0, + "status": status, + "startedAt": 1_000, + "endedAt": if status == "running" { None } else { Some(2_000) }, + "usage": null, + "lastTool": null, + "summary": null, + }]}), + ); + acp::SessionUpdate::SessionInfoUpdate(acp::SessionInfoUpdate::new().meta(meta)) + }; + cx.update(|_, cx| connection.send_update(session_id.clone(), tasks("running"), cx)); + cx.run_until_parked(); + + connection.end_turn(session_id.clone(), acp::StopReason::EndTurn); + cx.run_until_parked(); + panel.read_with(&cx, |panel, _cx| { + assert!( + panel.retained_threads.contains_key(&thread_id), + "a thread whose background task still runs must stay loaded" + ); + }); + + cx.update(|_, cx| connection.send_update(session_id.clone(), tasks("completed"), cx)); + cx.run_until_parked(); + panel.read_with(&cx, |panel, _cx| { + assert!( + !panel.retained_threads.contains_key(&thread_id), + "the snapshot that ends the last task releases the thread" + ); + }); + } + #[gpui::test] async fn test_cleanup_retained_threads_preserves_idle_non_loadable_threads( cx: &mut TestAppContext, diff --git a/crates/agent_ui/src/conversation_view.rs b/crates/agent_ui/src/conversation_view.rs index bd03f82..77a775c 100644 --- a/crates/agent_ui/src/conversation_view.rs +++ b/crates/agent_ui/src/conversation_view.rs @@ -367,6 +367,8 @@ impl Conversation { | AcpThreadEvent::TokenUsageUpdated(_) | AcpThreadEvent::AccountUsageUpdated | AcpThreadEvent::BackgroundTasksUpdated + | AcpThreadEvent::TasksUpdated + | AcpThreadEvent::TaskEnded { .. } | AcpThreadEvent::EntryUpdated(_) | AcpThreadEvent::EntriesRemoved(_) | AcpThreadEvent::Retry(_) @@ -814,6 +816,8 @@ fn affects_thread_metadata(event: &AcpThreadEvent) -> bool { | AcpThreadEvent::TokenUsageUpdated(_) | AcpThreadEvent::AccountUsageUpdated | AcpThreadEvent::BackgroundTasksUpdated + | AcpThreadEvent::TasksUpdated + | AcpThreadEvent::TaskEnded { .. } | AcpThreadEvent::PromptCapabilitiesUpdated | AcpThreadEvent::AvailableCommandsUpdated(_) | AcpThreadEvent::ModeUpdated(_) @@ -2345,6 +2349,8 @@ impl ConversationView { // sidebar: the agent panel observes every conversation view, active or not, // and asks the sidebar to rebuild, which re-reads the thread's status. AcpThreadEvent::BackgroundTasksUpdated => {} + // The task card (story 013) reads `tasks()`; the notify below redraws it. + AcpThreadEvent::TasksUpdated | AcpThreadEvent::TaskEnded { .. } => {} AcpThreadEvent::AccountUsageUpdated => { // A redraw is all this needs: the account section is built from the thread // when the tooltip opens, so there is no per-turn state to recompute. diff --git a/crates/agent_ui/src/conversation_view/subagent_card.rs b/crates/agent_ui/src/conversation_view/subagent_card.rs index bc5436d..28a8c49 100644 --- a/crates/agent_ui/src/conversation_view/subagent_card.rs +++ b/crates/agent_ui/src/conversation_view/subagent_card.rs @@ -1,3 +1,4 @@ +use acp_thread::subagent_rows::SubagentRowStatus; use acp_thread::{ AcpThread, AgentThreadEntry, MessageIdentity, SessionMessage, SessionMessageSide, ToolCall, ToolCallContent, ToolCallStatus, @@ -129,31 +130,15 @@ impl SubagentStatus { acp_v2::StateUpdate::Running(_) | acp_v2::StateUpdate::RequiresAction(_) => { (Self::Running, None) } - acp_v2::StateUpdate::Idle(idle) => match &idle.stop_reason { - Some(acp_v2::StopReason::Cancelled) => (Self::Canceled, None), - Some(acp_v2::StopReason::Error(error)) => ( - Self::Failed, - error - .error - .as_ref() - .map(|error| error.message.trim()) - .filter(|message| !message.is_empty()) - .map(|message| message.to_string().into()), - ), - Some(acp_v2::StopReason::MaxTokens) => ( - Self::Failed, - Some("The subagent reached its token limit.".into()), - ), - Some(acp_v2::StopReason::MaxTurnRequests) => ( - Self::Failed, - Some("The subagent reached its limit of model requests.".into()), - ), - Some(acp_v2::StopReason::Refusal) => ( - Self::Failed, - Some("The subagent's model refused to continue.".into()), - ), - _ => (Self::Completed, None), - }, + acp_v2::StateUpdate::Idle(idle) => { + let (outcome, error) = acp_thread::subagent_rows::idle_outcome(idle); + let status = match outcome { + SubagentRowStatus::Canceled => Self::Canceled, + SubagentRowStatus::Failed => Self::Failed, + _ => Self::Completed, + }; + (status, error) + } _ => (Self::Unknown, None), } }