diff --git a/crates/ide-agent/src/conversation.rs b/crates/ide-agent/src/conversation.rs new file mode 100644 index 000000000..1173efda9 --- /dev/null +++ b/crates/ide-agent/src/conversation.rs @@ -0,0 +1,281 @@ +//! A task's conversation and those of its subagents. +//! +//! An agent that supports ACP subagent sessions runs each subagent (Claude Code's Agent +//! tool, Codex's spawned agents) as a child session: `subagent_update` on the parent +//! announces it and reports its state, its own updates carry its session id, and +//! `session_message`s carry what it was asked and what it answered. The parent's +//! timeline shows a card per subagent; each subagent keeps its own timeline. + +use crate::model::AgentEntry; +use crate::timeline::{Timeline, blocks_text, content_text}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// Where an entry lives: the task's own conversation (`None`) or a subagent's session. +pub(crate) type Session = Option; + +#[derive(Default)] +pub(crate) struct Conversation { + pub main: Timeline, + /// In the order they started. + subagents: Vec, +} + +struct Subagent { + id: String, + /// The session that started it. + parent: Session, + timeline: Timeline, +} + +/// A subagent's conversation, as saved with its task. +#[derive(Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct SavedSubagent { + pub id: String, + pub parent: Option, + pub entries: Vec, +} + +impl Conversation { + pub fn restore(main: Vec, subagents: Vec) -> Self { + Self { + main: Timeline::from_entries(main), + subagents: subagents + .into_iter() + .map(|s| Subagent { id: s.id, parent: s.parent, timeline: Timeline::from_entries(s.entries) }) + .collect(), + } + } + + pub fn saved_subagents(&self) -> Vec { + self.subagents + .iter() + .map(|s| SavedSubagent { id: s.id.clone(), parent: s.parent.clone(), entries: s.timeline.entries.clone() }) + .collect() + } + + pub fn timeline(&self, session: &Session) -> Option<&Timeline> { + match session { + None => Some(&self.main), + Some(id) => self.subagents.iter().find(|s| &s.id == id).map(|s| &s.timeline), + } + } + + fn timeline_mut(&mut self, session: &Session) -> Option<&mut Timeline> { + match session { + None => Some(&mut self.main), + Some(id) => self.subagents.iter_mut().find(|s| &s.id == id).map(|s| &mut s.timeline), + } + } + + fn parent_of(&self, id: &str) -> Option<&Session> { + self.subagents.iter().find(|s| s.id == id).map(|s| &s.parent) + } + + /// The session an update for session id `session` belongs to: a known subagent's, else the task's. + pub fn owner(&self, session: Option<&str>) -> Session { + session.filter(|s| self.parent_of(s).is_some()).map(str::to_owned) + } + + /// The parent's title (or description) for subagent `id`. + pub fn label(&self, id: &str) -> Option { + let parent = self.parent_of(id)?; + self.timeline(parent)?.entries.iter().find_map(|e| match e { + AgentEntry::Subagent { id: card, title, description, .. } if card == id => { + title.clone().or_else(|| description.clone()) + } + _ => None, + }) + } + + /// Applies one `session/update` of session id `session` (the task's own when unknown); + /// returns the entries it added or changed. + pub fn apply(&mut self, session: Option<&str>, update: &Value, now: f64) -> Vec<(Session, usize)> { + let owner = self.owner(session); + let kind = update.get("sessionUpdate").and_then(Value::as_str).unwrap_or_default(); + let mut out = Vec::new(); + match kind { + "subagent_update" => { + let Some(id) = update.get("sessionId").and_then(Value::as_str) else { return out }; + if self.parent_of(id).is_none() { + self.subagents.push(Subagent { id: id.to_owned(), parent: owner, timeline: Timeline::default() }); + } + let parent = self.parent_of(id).cloned().flatten(); + let Some(timeline) = self.timeline_mut(&parent) else { return out }; + let (index, stopped) = timeline.subagent(id, update, now); + out.push((parent, index)); + if stopped && let Some(child) = self.timeline_mut(&Some(id.to_owned())) { + let id = Some(id.to_owned()); + out.extend(child.end_turn(None).into_iter().map(|i| (id.clone(), i))); + } + } + "session_message" | "session_message_chunk" => { + let append = kind == "session_message_chunk"; + let text = if append { + content_text(update.get("content")).unwrap_or_default() + } else { + blocks_text(update.get("content")) + }; + let field = |k: &str| update.get(k).and_then(Value::as_str).map(str::to_owned); + let (sender, recipient) = (field("senderSessionId"), field("recipientSessionId")); + let ours = + |s: &Option| s.as_deref().filter(|s| self.parent_of(s) == Some(&owner)).map(str::to_owned); + // Between this session and a subagent of it: the subagent's task or answer, on its card. + let card = ours(&recipient).map(|c| (c, false)).or_else(|| ours(&sender).map(|c| (c, true))); + let Some(timeline) = self.timeline_mut(&owner) else { return out }; + let index = match card { + Some((child, reply)) => timeline.card_message(&child, reply, &text, append), + None => { + let id = field("messageId").unwrap_or_default(); + Some(timeline.message(&id, sender != owner, &text, append)) + } + }; + out.extend(index.map(|i| (owner, i))); + } + _ => { + let Some(timeline) = self.timeline_mut(&owner) else { return out }; + out.extend(timeline.apply(update).changed.into_iter().map(|i| (owner.clone(), i))); + if let Some(id) = owner.as_deref() + && matches!(kind, "tool_call" | "tool_call_update") + { + out.extend(self.refresh_card(id)); + } + } + } + out + } + + /// Updates subagent `id`'s card from its timeline. + fn refresh_card(&mut self, id: &str) -> Option<(Session, usize)> { + let parent = self.parent_of(id)?.clone(); + let child = self.subagents.iter().position(|s| s.id == id)?; + let timeline = std::mem::take(&mut self.subagents[child].timeline); + let changed = self.timeline_mut(&parent).and_then(|p| p.card_activity(id, &timeline)); + self.subagents[child].timeline = timeline; + changed.map(|i| (parent, i)) + } + + /// Settles what an ended turn of the task left open. Subagents work on (in the background). + pub fn end_turn(&mut self, done: Option) -> Vec<(Session, usize)> { + self.main.end_turn(done).into_iter().map(|i| (None, i)).collect() + } + + /// Settles everything left open once the agent is gone, subagents included. + pub fn settle(&mut self, now: f64) -> Vec<(Session, usize)> { + let mut out: Vec<(Session, usize)> = self.main.end_turn(None).into_iter().map(|i| (None, i)).collect(); + out.extend(self.main.stop_subagents(now).into_iter().map(|i| (None, i))); + for s in &mut self.subagents { + let id = Some(s.id.clone()); + let mut changed = s.timeline.end_turn(None); + changed.extend(s.timeline.stop_subagents(now)); + out.extend(changed.into_iter().map(|i| (id.clone(), i))); + } + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::{AgentSubagentStatus, AgentToolStatus}; + use serde_json::json; + + fn card(c: &Conversation, session: &Session, index: usize) -> AgentEntry { + c.timeline(session).unwrap().entries[index].clone() + } + + /// Claude Code's Agent tool, as `claude-agent-acp` reports it to a client with subagent sessions. + #[test] + fn a_subagent_gets_a_card_in_its_parent_and_its_own_timeline() { + let mut c = Conversation::default(); + let sub = Some("child".to_owned()); + let announce = json!({"sessionUpdate": "subagent_update", "sessionId": "child", "title": "Explore", + "description": "Find the parser", "capabilities": {"cancel": {}}, "state": {"state": "running"}}); + assert_eq!(c.apply(Some("root"), &announce, 10.0), vec![(None, 0)]); + // The launch prompt: the parent's outgoing view, then the child's incoming one. + let prompt = json!({"sessionUpdate": "session_message", "messageId": "toolu_1", + "senderSessionId": "root", "recipientSessionId": "child", + "content": [{"type": "text", "text": "Look for it"}]}); + assert_eq!(c.apply(Some("root"), &prompt, 11.0), vec![(None, 0)]); + assert_eq!(c.apply(Some("child"), &prompt, 11.0), vec![(sub.clone(), 0)]); + assert_eq!(card(&c, &sub, 0), AgentEntry::Message { incoming: true, text: "Look for it".into() }); + + // The child's own work stays out of the parent's timeline; its card follows it. + c.apply( + Some("child"), + &json!({"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "hm"}}), + 12.0, + ); + let changed = c.apply( + Some("child"), + &json!({"sessionUpdate": "tool_call", "toolCallId": "t1", "title": "Read lib.rs", "kind": "read", "status": "in_progress"}), + 12.0, + ); + assert_eq!(changed, vec![(sub.clone(), 2), (None, 0)]); + assert_eq!(c.main.entries.len(), 1); + // Settings of the child session don't touch the task. + assert_eq!(c.owner(Some("child")), sub); + assert_eq!(c.owner(Some("root")), None); + + let done = json!({"sessionUpdate": "subagent_update", "sessionId": "child", "state": {"state": "idle", "stopReason": "end_turn"}}); + let changed = c.apply(Some("root"), &done, 20.0); + // Its tool calls left open are settled when it stops. + assert_eq!(changed, vec![(None, 0), (sub.clone(), 2)]); + let reply = json!({"sessionUpdate": "session_message", "messageId": "toolu_1:reply", "senderSessionId": "child", + "recipientSessionId": "root", "content": [{"type": "text", "text": "In "}, {"type": "text", "text": "parser.rs"}]}); + c.apply(Some("root"), &reply, 20.0); + assert_eq!( + c.main.entries[0], + AgentEntry::Subagent { + id: "child".into(), + title: Some("Explore".into()), + description: Some("Find the parser".into()), + status: AgentSubagentStatus::Done, + error: None, + prompt: Some("Look for it".into()), + reply: Some("In \nparser.rs".into()), + tools: 1, + activity: Some("Read lib.rs".into()), + started_ms: 10.0, + ended_ms: Some(20.0), + cancellable: true, + } + ); + assert_eq!(c.label("child").as_deref(), Some("Explore")); + } + + #[test] + fn nested_subagents_and_failures() { + let mut c = Conversation::default(); + c.apply( + None, + &json!({"sessionUpdate": "subagent_update", "sessionId": "a", "state": {"state": "running"}}), + 1.0, + ); + c.apply( + Some("a"), + &json!({"sessionUpdate": "subagent_update", "sessionId": "b", "description": "deeper"}), + 2.0, + ); + let a = Some("a".to_owned()); + assert!(matches!(card(&c, &a, 0), AgentEntry::Subagent { ref id, .. } if id == "b")); + let failed = json!({"sessionUpdate": "subagent_update", "sessionId": "b", + "state": {"state": "idle", "stopReason": "error", "error": {"code": -32603, "message": "boom"}}}); + assert_eq!(c.apply(Some("a"), &failed, 3.0), vec![(a.clone(), 0)]); + assert!(matches!(card(&c, &a, 0), + AgentEntry::Subagent { status: AgentSubagentStatus::Failed, error: Some(ref e), ended_ms: Some(3.0), .. } if e == "boom")); + + // Saved and restored, then the agent is gone: what still works is stopped. + let mut c = Conversation::restore(c.main.entries.clone(), c.saved_subagents()); + c.apply( + Some("a"), + &json!({"sessionUpdate": "tool_call", "toolCallId": "t", "title": "Run", "status": "in_progress"}), + 4.0, + ); + let settled = c.settle(9.0); + assert_eq!(settled, vec![(None, 0), (a.clone(), 1)]); + assert!(matches!(c.main.entries[0], AgentEntry::Subagent { status: AgentSubagentStatus::Cancelled, .. })); + assert!(matches!(card(&c, &a, 1), AgentEntry::Tool { status: AgentToolStatus::Failed, .. })); + } +} diff --git a/crates/ide-agent/src/lib.rs b/crates/ide-agent/src/lib.rs index 72f61e73f..ecf5906cd 100644 --- a/crates/ide-agent/src/lib.rs +++ b/crates/ide-agent/src/lib.rs @@ -8,6 +8,7 @@ //! next prompt when the agent supports `session/load`. mod config; +mod conversation; mod model; mod provider; mod skills; @@ -16,6 +17,7 @@ mod worktree; pub use model::*; +use conversation::{Conversation, SavedSubagent, Session}; use ide_core::{IdeError, IdeResult, IpcPath}; use ide_jsonrpc::{Framing, Handler, Peer, RpcError, codes}; use parking_lot::{Mutex, RwLock}; @@ -26,7 +28,6 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; -use timeline::Timeline; /// Receives the events of one project's tasks. pub type AgentSink = Arc; @@ -39,6 +40,9 @@ const REPLAY_GRACE: Duration = Duration::from_millis(300); /// can trail its `session/prompt` response by this much. const TURN_GRACE: Duration = Duration::from_millis(100); const STDERR_LINES: usize = 20; +/// Updates of a session's state rather than its conversation. +const SESSION_STATE: [&str; 5] = + ["session_info_update", "config_option_update", "current_mode_update", "available_commands_update", "usage_update"]; /// Referenced files up to this size are embedded in the prompt (when the agent takes it). const MAX_EMBED: u64 = 256 * 1024; @@ -155,12 +159,14 @@ struct TaskFile { /// In the project's own checkout: files already changed when the task started. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] before: BTreeMap, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + subagents: Vec, } struct TaskState { task: AgentTask, session_id: Option, - timeline: Timeline, + conversation: Conversation, wanted: Vec, before: Arc>, /// A turn is running or about to (prompts arriving now are queued). @@ -192,7 +198,7 @@ struct Task { replaying: AtomicBool, /// The current turn is being cancelled to send the first queued message right away. send_now: AtomicBool, - dirty: Mutex>, + dirty: Mutex>, flush_scheduled: AtomicBool, stderr: Mutex>, /// Set by `delete`; held while saving. @@ -235,11 +241,9 @@ impl AgentManager { // The agent process did not survive the restart: whatever was in flight is over. if matches!(f.task.status, AgentStatus::Running | AgentStatus::NeedsInput | AgentStatus::Starting) { f.task.status = AgentStatus::Ready; - let mut timeline = Timeline::from_entries(std::mem::take(&mut f.entries)); - timeline.end_turn(None); - f.entries = timeline.entries; } let task = Task::restore(f, shared.clone()); + task.state.lock().conversation.settle(now_ms()); tasks.insert(task.id.clone(), task); } Err(e) => tracing::warn!("skipping agent task {}: {e}", path.display()), @@ -301,8 +305,12 @@ impl AgentManager { self.tasks.read().get(id).cloned().ok_or_else(|| IdeError::not_found(format!("no agent task {id}"))) } - pub fn entries(&self, id: &str) -> IdeResult> { - Ok(self.task(id)?.state.lock().timeline.entries.clone()) + /// The entries of the task's conversation, or of its subagent `subagent`'s. + pub fn entries(&self, id: &str, subagent: Option) -> IdeResult> { + let task = self.task(id)?; + let s = task.state.lock(); + let timeline = s.conversation.timeline(&subagent).ok_or_else(|| IdeError::not_found("no such subagent"))?; + Ok(timeline.entries.clone()) } /// Creates the task (and its worktree) and sends `req.prompt` as its first message. @@ -357,7 +365,14 @@ impl AgentManager { usage: None, queue: Vec::new(), }; - let file = TaskFile { task: task.clone(), session_id: None, entries: Vec::new(), wanted: req.config, before }; + let file = TaskFile { + task: task.clone(), + session_id: None, + entries: Vec::new(), + wanted: req.config, + before, + subagents: Vec::new(), + }; let handle = Task::restore(file, self.shared.clone()); handle.state.lock().busy = true; self.tasks.write().insert(task.id.clone(), handle.clone()); @@ -487,9 +502,17 @@ impl AgentManager { } /// Stops the current turn (`session/cancel`); pending permission requests are cancelled. - /// Queued messages stay queued. - pub fn cancel(&self, id: &str) -> IdeResult<()> { - self.task(id)?.cancel(); + /// Queued messages stay queued. With `subagent`, stops only that subagent's work. + pub fn cancel(&self, id: &str, subagent: Option) -> IdeResult<()> { + let task = self.task(id)?; + match subagent { + None => task.cancel(), + Some(sub) => { + if let Some(live) = task.live.lock().clone() { + let _ = live.peer.notify("session/cancel", json!({ "sessionId": sub })); + } + } + } Ok(()) } @@ -551,6 +574,8 @@ impl AgentManager { pub fn shutdown(&self) { for task in self.tasks.read().values() { task.kill(); + // What arrived after the last save (a subagent's late updates). + task.persist(); } *self.shared.sink.write() = None; } @@ -559,14 +584,14 @@ impl AgentManager { impl Task { fn restore(file: TaskFile, shared: Arc) -> Arc { let task = file.task; - let timeline = Timeline::from_entries(file.entries); + let conversation = Conversation::restore(file.entries, file.subagents); Arc::new(Self { id: task.id.clone(), shared, state: Mutex::new(TaskState { task, session_id: file.session_id, - timeline, + conversation, wanted: file.wanted, before: Arc::new(file.before), busy: false, @@ -623,9 +648,10 @@ impl Task { serde_json::to_vec(&TaskFile { task: s.task.clone(), session_id: s.session_id.clone(), - entries: s.timeline.entries.clone(), + entries: s.conversation.main.entries.clone(), wanted: s.wanted.clone(), before: (*s.before).clone(), + subagents: s.conversation.saved_subagents(), }) }; let path = self.shared.paths.store_dir.join(format!("{}.json", self.id)); @@ -634,7 +660,7 @@ impl Task { } } - fn mark_dirty(self: &Arc, indices: impl IntoIterator) { + fn mark_dirty(self: &Arc, indices: impl IntoIterator) { self.dirty.lock().extend(indices); if !self.flush_scheduled.swap(true, Ordering::AcqRel) { let task = self.clone(); @@ -652,9 +678,9 @@ impl Task { let s = self.state.lock(); indices .into_iter() - .filter_map(|i| { - let entry = s.timeline.entries.get(i)?.clone(); - Some(AgentEvent::Entry { task: self.id.clone(), index: i as u32, entry }) + .filter_map(|(subagent, i)| { + let entry = s.conversation.timeline(&subagent)?.entries.get(i)?.clone(); + Some(AgentEvent::Entry { task: self.id.clone(), subagent, index: i as u32, entry }) }) .collect() }; @@ -664,8 +690,8 @@ impl Task { } fn push_entry(self: &Arc, entry: AgentEntry) -> usize { - let i = self.state.lock().timeline.push(entry); - self.mark_dirty([i]); + let i = self.state.lock().conversation.main.push(entry); + self.mark_dirty([(None, i)]); i } @@ -675,7 +701,7 @@ impl Task { loop { let result = self.turn(&prompt).await; tokio::time::sleep(TURN_GRACE).await; - let changed = self.state.lock().timeline.end_turn(Some(now_ms())); + let changed = self.state.lock().conversation.end_turn(Some(now_ms())); self.mark_dirty(changed); let send_now = self.send_now.swap(false, Ordering::AcqRel); let next = { @@ -789,7 +815,12 @@ impl Task { "initialize", json!({ "protocolVersion": 1, - "clientCapabilities": { "fs": { "readTextFile": false, "writeTextFile": false }, "terminal": false }, + "clientCapabilities": { + "fs": { "readTextFile": false, "writeTextFile": false }, + "terminal": false, + // Subagents as sessions of their own (ACP subagents RFD). + "subagents": {}, + }, "clientInfo": { "name": "OpenIDEA", "version": env!("CARGO_PKG_VERSION") }, }), None, @@ -890,8 +921,16 @@ impl Task { } } - fn on_update(self: &Arc, update: &Value) { + /// Applies a `session/update` of session `session` (the task's, or one of its subagents'). + fn on_update(self: &Arc, session: Option<&str>, update: &Value) { let kind = update.get("sessionUpdate").and_then(Value::as_str).unwrap_or_default(); + // A subagent's title, settings and usage are its own, not the task's. + if self.state.lock().conversation.owner(session).is_some() { + if !SESSION_STATE.contains(&kind) && !self.replaying.load(Ordering::Acquire) { + self.apply(session, kind, update); + } + return; + } if kind == "session_info_update" { // The agent's own title for the session replaces the first line of the prompt. let title = update.get("title").and_then(Value::as_str).map(str::trim).filter(|t| !t.is_empty()); @@ -946,8 +985,16 @@ impl Task { if self.replaying.load(Ordering::Acquire) { return; } - let applied = self.state.lock().timeline.apply(update); - self.mark_dirty(applied.changed); + self.apply(session, kind, update); + } + + fn apply(self: &Arc, session: Option<&str>, kind: &str, update: &Value) { + let applied = self.state.lock().conversation.apply(session, update, now_ms()); + self.mark_dirty(applied); + // A subagent can start or finish after the turn that started it: saved right away. + if kind == "subagent_update" { + self.persist(); + } } /// Blocks the calling (blocking-pool) thread until the user answers. @@ -959,6 +1006,12 @@ impl Task { .filter(|t| !t.is_empty()) .unwrap_or("The agent asks for permission") .to_owned(); + // A subagent's request is answered in the task's conversation, named after the subagent. + let session = params.get("sessionId").and_then(Value::as_str); + let title = match session.and_then(|s| self.state.lock().conversation.label(s)) { + Some(label) => format!("{label}: {title}"), + None => title, + }; let options: Vec = params .get("options") .and_then(Value::as_array) @@ -986,11 +1039,11 @@ impl Task { .map_or_else(|| "Cancelled".to_owned(), |o| o.name.clone()); { let mut s = self.state.lock(); - if let Some(AgentEntry::Permission { chosen, .. }) = s.timeline.entries.get_mut(entry) { + if let Some(AgentEntry::Permission { chosen, .. }) = s.conversation.main.entries.get_mut(entry) { *chosen = Some(name); } } - self.mark_dirty([entry]); + self.mark_dirty([(None, entry)]); let still_asking = !self.permissions.lock().is_empty(); if !still_asking && self.state.lock().task.status == AgentStatus::NeedsInput { self.set_status(AgentStatus::Running, None); @@ -1032,7 +1085,7 @@ impl Handler for Client { if method == "session/update" && let (Some(task), Some(update)) = (self.task.upgrade(), params.get("update")) { - task.on_update(update); + task.on_update(params.get("sessionId").and_then(Value::as_str), update); } } diff --git a/crates/ide-agent/src/model.rs b/crates/ide-agent/src/model.rs index 5ba2707a1..364629ce0 100644 --- a/crates/ide-agent/src/model.rs +++ b/crates/ide-agent/src/model.rs @@ -166,6 +166,18 @@ pub struct AgentToolDiff { pub removed: u32, } +/// What a subagent is doing, as its parent reports it. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, Type)] +#[serde(rename_all = "camelCase")] +pub enum AgentSubagentStatus { + Running, + /// One of its requests (a permission) waits for an answer. + NeedsInput, + Done, + Cancelled, + Failed, +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Type)] #[serde(rename_all = "camelCase")] pub struct AgentPlanItem { @@ -218,6 +230,33 @@ pub enum AgentEntry { Plan { items: Vec, }, + /// A subagent the agent started: its own session, whose entries are listed separately. + Subagent { + /// The subagent's session id. + id: String, + title: Option, + description: Option, + status: AgentSubagentStatus, + error: Option, + /// What it was asked to do, last. + prompt: Option, + /// Its answer. + reply: Option, + /// Tool calls it made. + tools: u32, + /// Title of its latest tool call. + activity: Option, + started_ms: f64, + /// When it last stopped working; `None` while it works. + ended_ms: Option, + /// Whether it can be stopped on its own. + cancellable: bool, + }, + /// A message between sessions in a subagent's conversation: its task (`incoming`) or its report. + Message { + incoming: bool, + text: String, + }, Permission { id: u32, title: String, @@ -279,9 +318,10 @@ pub enum AgentEvent { Removed { id: String, }, - /// Entry `index` of `task` was added or replaced. + /// Entry `index` of `task` (of its subagent `subagent`) was added or replaced. Entry { task: String, + subagent: Option, index: u32, entry: AgentEntry, }, diff --git a/crates/ide-agent/src/tests.rs b/crates/ide-agent/src/tests.rs index 457fa2e46..6a7f0c962 100644 --- a/crates/ide-agent/src/tests.rs +++ b/crates/ide-agent/src/tests.rs @@ -25,6 +25,44 @@ impl FakeAgent { } } +impl FakeAgent { + /// Runs a subagent in session `c1`, as an agent with ACP subagent sessions reports it. + fn delegate(&self, root: &str) { + let sub = |state: Value| json!({"sessionUpdate": "subagent_update", "sessionId": "c1", "state": state}); + let mut announce = sub(json!({"state": "running"})); + announce["title"] = json!("Explore"); + self.update(root, announce); + self.update( + root, + json!({"sessionUpdate": "session_message", "messageId": "m1", "senderSessionId": root, + "recipientSessionId": "c1", "content": [{"type": "text", "text": "Find the lexer"}]}), + ); + self.update("c1", json!({"sessionUpdate": "usage_update", "used": 900, "size": 1000})); + self.update( + "c1", + json!({"sessionUpdate": "tool_call", "toolCallId": "c1-t1", "title": "Read lexer.rs", + "kind": "read", "status": "completed"}), + ); + let peer = self.peer.get().unwrap().clone(); + let _: Value = tokio::runtime::Handle::current() + .block_on(peer.request( + "session/request_permission", + json!({"sessionId": "c1", "toolCall": {"toolCallId": "c1-t2", "title": "Run ls"}, + "options": [{"optionId": "yes", "name": "Allow", "kind": "allow_once"}]}), + None, + )) + .unwrap(); + self.say("c1", "It is in lexer.rs"); + self.update(root, sub(json!({"state": "idle", "stopReason": "end_turn"}))); + self.update( + root, + json!({"sessionUpdate": "session_message", "messageId": "m2", "senderSessionId": "c1", + "recipientSessionId": root, "content": [{"type": "text", "text": "lexer.rs"}]}), + ); + self.say(root, "Found it"); + } +} + impl Handler for FakeAgent { fn on_notification(&self, method: &str, _: Value) { self.calls.lock().push(method.to_owned()); @@ -34,7 +72,12 @@ impl Handler for FakeAgent { self.calls.lock().push(method.to_owned()); let session = params.get("sessionId").and_then(Value::as_str).unwrap_or("s1").to_owned(); match method { - "initialize" => Ok(json!({"protocolVersion": 1, "agentCapabilities": {"loadSession": true}})), + "initialize" => { + if params.pointer("/clientCapabilities/subagents").is_some_and(Value::is_object) { + self.calls.lock().push("subagents".into()); + } + Ok(json!({"protocolVersion": 1, "agentCapabilities": {"loadSession": true}})) + } "session/new" => Ok(json!({"sessionId": "s1", "modes": {"currentModeId": "ask", "availableModes": [{"id": "ask", "name": "Ask"}, {"id": "code", "name": "Code"}]}, "configOptions": [{"id": "model", "category": "model", "type": "select", "currentValue": "big", @@ -53,6 +96,10 @@ impl Handler for FakeAgent { self.say(&session, "second turn"); return Ok(json!({"stopReason": "end_turn"})); } + if text == "delegate" { + self.delegate(&session); + return Ok(json!({"stopReason": "end_turn"})); + } if text == "wait" { // Works until cancelled. while !self.calls.lock().iter().any(|c| c == "session/cancel") { @@ -194,7 +241,7 @@ async fn a_task_runs_asks_for_permission_and_survives_a_restart() { wait_for(&mgr, &task.id, AgentStatus::NeedsInput).await; let permission = mgr - .entries(&task.id) + .entries(&task.id, None) .unwrap() .iter() .find_map(|e| match e { @@ -205,7 +252,7 @@ async fn a_task_runs_asks_for_permission_and_survives_a_restart() { mgr.respond(&task.id, permission, Some("yes".into())).unwrap(); wait_for(&mgr, &task.id, AgentStatus::Ready).await; - let entries = mgr.entries(&task.id).unwrap(); + let entries = mgr.entries(&task.id, None).unwrap(); let kinds: Vec = entries .iter() .map(|e| match e { @@ -261,13 +308,13 @@ async fn a_task_runs_asks_for_permission_and_survives_a_restart() { mgr.shutdown(); let agent2 = Arc::new(FakeAgent::default()); let mgr2 = manager(tmp.path(), agent2.clone()); - assert_eq!(mgr2.entries(&task.id).unwrap().len(), 5); + assert_eq!(mgr2.entries(&task.id, None).unwrap().len(), 5); assert_eq!(mgr2.task(&task.id).unwrap().state.lock().task.title, "Lexer fix"); assert_eq!(mgr2.providers().iter().find(|p| p.id == "claude").unwrap().config.len(), 2); mgr2.prompt(&task.id, text("again"), false).await.unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; wait_for(&mgr2, &task.id, AgentStatus::Ready).await; - let entries = mgr2.entries(&task.id).unwrap(); + let entries = mgr2.entries(&task.id, None).unwrap(); assert_eq!(entries.len(), 7); assert_eq!(entries[6], AgentEntry::Agent { text: "second turn".into() }); assert!(agent2.calls.lock().contains(&"session/load".to_owned())); @@ -303,14 +350,14 @@ async fn messages_sent_while_busy_are_queued_and_send_now_interrupts() { mgr.send_queued_now(&task.id, &queue[0].id).unwrap(); for _ in 0..500 { - if mgr.entries(&task.id).unwrap().iter().any(|e| *e == AgentEntry::Agent { text: "second turn".into() }) { + if mgr.entries(&task.id, None).unwrap().iter().any(|e| *e == AgentEntry::Agent { text: "second turn".into() }) { break; } tokio::time::sleep(Duration::from_millis(10)).await; } wait_for(&mgr, &task.id, AgentStatus::Ready).await; let users: Vec = mgr - .entries(&task.id) + .entries(&task.id, None) .unwrap() .into_iter() .filter_map(|e| match e { @@ -326,7 +373,7 @@ async fn messages_sent_while_busy_are_queued_and_send_now_interrupts() { wait_for(&mgr, &task.id, AgentStatus::Running).await; agent.calls.lock().retain(|c| c != "session/cancel"); mgr.prompt(&task.id, text("later"), false).await.unwrap(); - mgr.cancel(&task.id).unwrap(); + mgr.cancel(&task.id, None).unwrap(); wait_for(&mgr, &task.id, AgentStatus::Ready).await; let queued = mgr.task(&task.id).unwrap().state.lock().task.queue.clone(); assert_eq!(queued.len(), 1); @@ -334,6 +381,68 @@ async fn messages_sent_while_busy_are_queued_and_send_now_interrupts() { assert!(mgr.task(&task.id).unwrap().state.lock().task.queue.is_empty()); } +#[tokio::test(flavor = "multi_thread")] +async fn subagents_get_their_own_conversation() { + let tmp = tempfile::tempdir().unwrap(); + let agent = Arc::new(FakeAgent::default()); + let mgr = manager(tmp.path(), agent.clone()); + let task = mgr + .create(NewAgentTask { + prompt: text("delegate"), + provider: "claude".into(), + config: vec![], + isolation: Isolation::Local, + base: None, + }) + .await + .unwrap(); + // The subagent's permission request is answered in the task's conversation. + wait_for(&mgr, &task.id, AgentStatus::NeedsInput).await; + let (id, title) = mgr + .entries(&task.id, None) + .unwrap() + .into_iter() + .find_map(|e| match e { + AgentEntry::Permission { id, title, .. } => Some((id, title)), + _ => None, + }) + .unwrap(); + assert_eq!(title, "Explore: Run ls"); + mgr.respond(&task.id, id, Some("yes".into())).unwrap(); + wait_for(&mgr, &task.id, AgentStatus::Ready).await; + assert!(agent.calls.lock().iter().any(|c| c == "subagents")); + + let check = |mgr: &AgentManager| { + let main = mgr.entries(&task.id, None).unwrap(); + let card = main.iter().find(|e| matches!(e, AgentEntry::Subagent { .. })).unwrap(); + let AgentEntry::Subagent { title, status, prompt, reply, tools, activity, .. } = card else { unreachable!() }; + assert_eq!(title.as_deref(), Some("Explore")); + assert_eq!(*status, AgentSubagentStatus::Done); + assert_eq!((prompt.as_deref(), reply.as_deref()), (Some("Find the lexer"), Some("lexer.rs"))); + assert_eq!((*tools, activity.as_deref()), (1, Some("Read lexer.rs"))); + assert!(main.contains(&AgentEntry::Agent { text: "Found it".into() })); + assert!(!main.iter().any(|e| matches!(e, AgentEntry::Tool { .. }))); + let sub = mgr.entries(&task.id, Some("c1".into())).unwrap(); + assert!(matches!(&sub[0], AgentEntry::Tool { title, .. } if title == "Read lexer.rs")); + assert_eq!(sub[1], AgentEntry::Agent { text: "It is in lexer.rs".into() }); + }; + // Updates can trail the end of the turn. + for _ in 0..500 { + if mgr.entries(&task.id, None).unwrap().contains(&AgentEntry::Agent { text: "Found it".into() }) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + check(&mgr); + // The subagent's context use is not the task's. + assert_ne!(mgr.task(&task.id).unwrap().state.lock().task.usage.as_ref().map(|u| u.used), Some(900)); + assert!(mgr.entries(&task.id, Some("nope".into())).is_err()); + + // Saved with the task. + mgr.shutdown(); + check(&manager(tmp.path(), Arc::new(FakeAgent::default()))); +} + #[test] fn prompts_become_acp_content_blocks() { let p = AgentPrompt { @@ -381,15 +490,15 @@ async fn cancelling_resolves_the_open_permission_and_delete_forgets_the_task() { .await .unwrap(); wait_for(&mgr, &task.id, AgentStatus::NeedsInput).await; - mgr.cancel(&task.id).unwrap(); + mgr.cancel(&task.id, None).unwrap(); wait_for(&mgr, &task.id, AgentStatus::Ready).await; - let entries = mgr.entries(&task.id).unwrap(); + let entries = mgr.entries(&task.id, None).unwrap(); assert!(entries.iter().any(|e| matches!(e, AgentEntry::Permission { chosen: Some(c), .. } if c == "Cancelled"))); assert_eq!(entries.last(), Some(&AgentEntry::Agent { text: "chose none".into() })); assert_eq!(mgr.task(&task.id).unwrap().state.lock().task.model.as_deref(), Some("big"), "the agent's default"); mgr.delete(&task.id, false).await.unwrap(); - assert!(mgr.entries(&task.id).is_err()); + assert!(mgr.entries(&task.id, None).is_err()); assert!(!tmp.path().join("store").join(format!("{}.json", task.id)).exists()); assert!( mgr.create(NewAgentTask { @@ -452,7 +561,7 @@ async fn real_claude_task_in_a_worktree() { let status = mgr.task(&task.id).unwrap().state.lock().task.status; match status { AgentStatus::NeedsInput => { - let pending = mgr.entries(&task.id).unwrap().into_iter().rev().find_map(|e| match e { + let pending = mgr.entries(&task.id, None).unwrap().into_iter().rev().find_map(|e| match e { AgentEntry::Permission { id, options, chosen: None, .. } => Some((id, options)), _ => None, }); @@ -467,7 +576,7 @@ async fn real_claude_task_in_a_worktree() { } tokio::time::sleep(Duration::from_millis(200)).await; } - for e in mgr.entries(&task.id).unwrap() { + for e in mgr.entries(&task.id, None).unwrap() { eprintln!("{e:?}"); } let t = mgr.task(&task.id).unwrap().state.lock().task.clone(); diff --git a/crates/ide-agent/src/timeline.rs b/crates/ide-agent/src/timeline.rs index 1ab7c96f1..b0c772882 100644 --- a/crates/ide-agent/src/timeline.rs +++ b/crates/ide-agent/src/timeline.rs @@ -1,6 +1,6 @@ //! Folds ACP `session/update` notifications into a task's entries. -use crate::model::{AgentEntry, AgentPlanItem, AgentToolDiff, AgentToolStatus}; +use crate::model::{AgentEntry, AgentPlanItem, AgentSubagentStatus, AgentToolDiff, AgentToolStatus}; use ide_core::IpcPath; use serde_json::Value; use std::collections::HashMap; @@ -13,6 +13,10 @@ pub(crate) struct Timeline { pub entries: Vec, tools: HashMap, plan: Option, + /// Subagent cards, by the subagent's session id. + cards: HashMap, + /// Session messages, by message id. + messages: HashMap, } /// What a `session/update` changed in the timeline. @@ -31,6 +35,9 @@ impl Timeline { t.tools.insert(id.clone(), i); } AgentEntry::Plan { .. } => t.plan = Some(i), + AgentEntry::Subagent { id, .. } => { + t.cards.insert(id.clone(), i); + } _ => {} } } @@ -166,6 +173,119 @@ impl Timeline { } } + /// Adds or updates the card of subagent `id` from its `subagent_update`; returns the + /// card's index and whether the subagent just stopped working. + pub fn subagent(&mut self, id: &str, update: &Value, now: f64) -> (usize, bool) { + let index = match self.cards.get(id) { + Some(&i) => i, + None => { + let i = self.push(AgentEntry::Subagent { + id: id.to_owned(), + title: None, + description: None, + status: AgentSubagentStatus::Running, + error: None, + prompt: None, + reply: None, + tools: 0, + activity: None, + started_ms: now, + ended_ms: None, + cancellable: false, + }); + self.cards.insert(id.to_owned(), i); + i + } + }; + let AgentEntry::Subagent { title, description, status, error, ended_ms, cancellable, .. } = + &mut self.entries[index] + else { + return (index, false); + }; + // Patch semantics: an omitted field is unchanged, `null` clears it. + if let Some(v) = update.get("title") { + *title = v.as_str().map(str::to_owned); + } + if let Some(v) = update.get("description") { + *description = v.as_str().map(str::to_owned); + } + if let Some(v) = update.get("capabilities") { + *cancellable = v.get("cancel").is_some_and(Value::is_object); + } + let mut stopped = false; + if let Some((next, message)) = update.get("state").and_then(subagent_status) { + let working = matches!(next, AgentSubagentStatus::Running | AgentSubagentStatus::NeedsInput); + stopped = !working && ended_ms.is_none(); + *ended_ms = if working { None } else { ended_ms.or(Some(now)) }; + *status = next; + *error = message; + } + (index, stopped) + } + + /// Sets (or with `append`, extends) what subagent `id` was asked or answered, on its card. + pub fn card_message(&mut self, id: &str, reply: bool, text: &str, append: bool) -> Option { + let &index = self.cards.get(id)?; + let AgentEntry::Subagent { prompt, reply: answer, .. } = &mut self.entries[index] else { return None }; + let field = if reply { answer } else { prompt }; + match field { + Some(t) if append => t.push_str(text), + _ => *field = Some(text.to_owned()), + } + Some(index) + } + + /// Adds, replaces or (with `append`) extends session message `id`. + pub fn message(&mut self, id: &str, incoming: bool, text: &str, append: bool) -> usize { + if let Some(&i) = self.messages.get(id) + && let AgentEntry::Message { text: t, .. } = &mut self.entries[i] + { + if append { + t.push_str(text); + } else { + *t = text.to_owned(); + } + return i; + } + let i = self.push(AgentEntry::Message { incoming, text: text.to_owned() }); + self.messages.insert(id.to_owned(), i); + i + } + + /// Updates subagent `id`'s card with what its own timeline `of` shows: how many tool + /// calls it made and its latest one. Returns the card's index when it changed. + pub fn card_activity(&mut self, id: &str, of: &Timeline) -> Option { + let &index = self.cards.get(id)?; + let AgentEntry::Subagent { tools, activity, .. } = &mut self.entries[index] else { return None }; + let calls = of.entries.iter().filter_map(|e| match e { + AgentEntry::Tool { title, kind, .. } => Some(if title.is_empty() { kind } else { title }), + _ => None, + }); + let count = calls.clone().count() as u32; + let latest = calls.clone().next_back().cloned(); + if *tools == count && *activity == latest { + return None; + } + *tools = count; + *activity = latest; + Some(index) + } + + /// Marks the subagents still working as stopped (their agent is gone). + pub fn stop_subagents(&mut self, now: f64) -> Vec { + let mut changed = Vec::new(); + for (i, e) in self.entries.iter_mut().enumerate() { + if let AgentEntry::Subagent { status, ended_ms, .. } = e + && matches!(status, AgentSubagentStatus::Running | AgentSubagentStatus::NeedsInput) + { + *status = AgentSubagentStatus::Cancelled; + *ended_ms = Some(now); + changed.push(i); + } + } + changed + } + /// Settles what an ended turn left open: running tools and pending permissions, /// and records when the turn ended (`done`) on the message that started it. pub fn end_turn(&mut self, done: Option) -> Vec { @@ -198,7 +318,30 @@ fn str_field(v: &Value, key: &str) -> String { v.get(key).and_then(Value::as_str).unwrap_or_default().to_owned() } -fn content_text(content: Option<&Value>) -> Option { +/// The status a subagent `state` reports, with its error message; `None` for an unknown state. +fn subagent_status(state: &Value) -> Option<(AgentSubagentStatus, Option)> { + Some(match state.get("state")?.as_str()? { + "running" => (AgentSubagentStatus::Running, None), + "requires_action" => (AgentSubagentStatus::NeedsInput, None), + "idle" => match state.get("stopReason").and_then(Value::as_str) { + Some("cancelled") => (AgentSubagentStatus::Cancelled, None), + Some("error") => ( + AgentSubagentStatus::Failed, + state.pointer("/error/message").and_then(Value::as_str).map(str::to_owned), + ), + _ => (AgentSubagentStatus::Done, None), + }, + _ => return None, + }) +} + +/// The text of a list of content blocks, one per line. +pub(crate) fn blocks_text(content: Option<&Value>) -> String { + let blocks = content.and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default(); + blocks.iter().filter_map(|b| content_text(Some(b))).collect::>().join("\n") +} + +pub(crate) fn content_text(content: Option<&Value>) -> Option { let c = content?; match c.get("type").and_then(Value::as_str) { Some("text") => c.get("text").and_then(Value::as_str).map(str::to_owned), diff --git a/crates/ide-host/src/ops/agent.rs b/crates/ide-host/src/ops/agent.rs index 020a1fcb2..46ce3f405 100644 --- a/crates/ide-host/src/ops/agent.rs +++ b/crates/ide-host/src/ops/agent.rs @@ -28,8 +28,9 @@ pub async fn commands( host.project(window)?.agents.commands(&provider, task.as_deref()).await } -pub fn entries(host: &Host, window: &str, task: String) -> IdeResult> { - host.project(window)?.agents.entries(&task) +/// Entries of `task`'s conversation, or of its subagent `subagent`'s. +pub fn entries(host: &Host, window: &str, task: String, subagent: Option) -> IdeResult> { + host.project(window)?.agents.entries(&task, subagent) } pub async fn create(host: &Host, window: &str, request: NewAgentTask) -> IdeResult { @@ -56,8 +57,9 @@ pub async fn set_config(host: &Host, window: &str, task: String, option: String, host.project(window)?.agents.set_config(&task, &option, value).await } -pub fn cancel(host: &Host, window: &str, task: String) -> IdeResult<()> { - host.project(window)?.agents.cancel(&task) +/// Stops the task's current turn, or with `subagent` only that subagent's work. +pub fn cancel(host: &Host, window: &str, task: String, subagent: Option) -> IdeResult<()> { + host.project(window)?.agents.cancel(&task, subagent) } /// Answers a permission request; `option: None` cancels it. @@ -87,7 +89,7 @@ pub(crate) fn register(d: &mut Dispatcher) { d.add("agent_subscribe", |c| async move { subscribe(&c.host, &c.window, c.channel("onEvent")?) }); d.add("agent_providers", |c| async move { providers(&c.host, &c.window) }); d.add("agent_commands", |c| async move { commands(&c.host, &c.window, c.arg("provider")?, c.arg("task")?).await }); - d.add("agent_entries", |c| async move { entries(&c.host, &c.window, c.arg("task")?) }); + d.add("agent_entries", |c| async move { entries(&c.host, &c.window, c.arg("task")?, c.arg("subagent")?) }); d.add("agent_create", |c| async move { create(&c.host, &c.window, c.arg("request")?).await }); d.add("agent_prompt", |c| async move { prompt(&c.host, &c.window, c.arg("task")?, c.arg("prompt")?, c.arg("now")?).await @@ -100,7 +102,7 @@ pub(crate) fn register(d: &mut Dispatcher) { d.add("agent_set_config", |c| async move { set_config(&c.host, &c.window, c.arg("task")?, c.arg("option")?, c.arg("value")?).await }); - d.add("agent_cancel", |c| async move { cancel(&c.host, &c.window, c.arg("task")?) }); + d.add("agent_cancel", |c| async move { cancel(&c.host, &c.window, c.arg("task")?, c.arg("subagent")?) }); d.add("agent_respond", |c| async move { respond(&c.host, &c.window, c.arg("task")?, c.arg("permission")?, c.arg("option")?) }); diff --git a/src-tauri/src/commands/agent.rs b/src-tauri/src/commands/agent.rs index 1c69a9b19..6266d5ca4 100644 --- a/src-tauri/src/commands/agent.rs +++ b/src-tauri/src/commands/agent.rs @@ -46,10 +46,16 @@ pub async fn agent_commands( agent::commands(&state.host, window.label(), provider, task).await } +/// Entries of `task`'s conversation, or of its subagent `subagent`'s. #[tauri::command] #[specta::specta] -pub fn agent_entries(window: Window, state: State<'_, AppState>, task: String) -> IdeResult> { - agent::entries(&state.host, window.label(), task) +pub fn agent_entries( + window: Window, + state: State<'_, AppState>, + task: String, + subagent: Option, +) -> IdeResult> { + agent::entries(&state.host, window.label(), task, subagent) } #[tauri::command] @@ -108,10 +114,16 @@ pub async fn agent_set_config( agent::set_config(&state.host, window.label(), task, option, value).await } +/// Stops the task's current turn, or with `subagent` only that subagent's work. #[tauri::command] #[specta::specta] -pub fn agent_cancel(window: Window, state: State<'_, AppState>, task: String) -> IdeResult<()> { - agent::cancel(&state.host, window.label(), task) +pub fn agent_cancel( + window: Window, + state: State<'_, AppState>, + task: String, + subagent: Option, +) -> IdeResult<()> { + agent::cancel(&state.host, window.label(), task, subagent) } /// Answers a permission request; `option: null` cancels it. diff --git a/src/bindings.ts b/src/bindings.ts index acde503a1..6349469d0 100644 --- a/src/bindings.ts +++ b/src/bindings.ts @@ -61,7 +61,8 @@ export const commands = { * before a task exists (skills and custom commands on disk included). */ agentCommands: (provider: string, task: string | null) => typedError(__TAURI_INVOKE("agent_commands", { provider, task })), - agentEntries: (task: string) => typedError(__TAURI_INVOKE("agent_entries", { task })), + /** Entries of `task`'s conversation, or of its subagent `subagent`'s. */ + agentEntries: (task: string, subagent: string | null) => typedError(__TAURI_INVOKE("agent_entries", { task, subagent })), agentCreate: (request: NewAgentTask) => typedError(__TAURI_INVOKE("agent_create", { request })), /** Sends a message; while the agent works it is queued, or with `now` it interrupts the turn. */ agentPrompt: (task: string, prompt: AgentPrompt, now: boolean) => typedError(__TAURI_INVOKE("agent_prompt", { task, prompt, now })), @@ -71,7 +72,8 @@ export const commands = { agentSendQueuedNow: (task: string, queued: string) => typedError(__TAURI_INVOKE("agent_send_queued_now", { task, queued })), /** Changes a session setting (model, mode, thinking effort…). */ agentSetConfig: (task: string, option: string, value: string) => typedError(__TAURI_INVOKE("agent_set_config", { task, option, value })), - agentCancel: (task: string) => typedError(__TAURI_INVOKE("agent_cancel", { task })), + /** Stops the task's current turn, or with `subagent` only that subagent's work. */ + agentCancel: (task: string, subagent: string | null) => typedError(__TAURI_INVOKE("agent_cancel", { task, subagent })), /** Answers a permission request; `option: null` cancels it. */ agentRespond: (task: string, permission: number, option: string | null) => typedError(__TAURI_INVOKE("agent_respond", { task, permission, option })), agentArchive: (task: string) => typedError(__TAURI_INVOKE("agent_archive", { task })), @@ -1001,7 +1003,25 @@ doneMs?: number | null } | { type: "agent"; text: string } | { type: "thought"; /** ACP tool kind: `read`, `edit`, `execute`, `search`, `fetch`, `think`… */ kind: string; status: AgentToolStatus; locations: IpcPath[]; /** Text the tool reported (command output…), truncated. */ -output: string | null; diffs: AgentToolDiff[] } | { type: "plan"; items: AgentPlanItem[] } | { type: "permission"; id: number; title: string; options: AgentPermissionOption[]; +output: string | null; diffs: AgentToolDiff[] } | { type: "plan"; items: AgentPlanItem[] } | +/** A subagent the agent started: its own session, whose entries are listed separately. */ +{ type: "subagent"; +/** The subagent's session id. */ +id: string; title: string | null; description: string | null; status: AgentSubagentStatus; error: string | null; +/** What it was asked to do, last. */ +prompt: string | null; +/** Its answer. */ +reply: string | null; +/** Tool calls it made. */ +tools: number; +/** Title of its latest tool call. */ +activity: string | null; startedMs: number | null; +/** When it last stopped working; `None` while it works. */ +endedMs: number | null; +/** Whether it can be stopped on its own. */ +cancellable: boolean } | +/** A message between sessions in a subagent's conversation: its task (`incoming`) or its report. */ +{ type: "message"; incoming: boolean; text: string } | { type: "permission"; id: number; title: string; options: AgentPermissionOption[]; /** Name of the chosen option, or `None` while pending. Cancelled requests read "Cancelled". */ chosen: string | null } | { type: "error"; message: string }; @@ -1009,8 +1029,8 @@ chosen: string | null } | { type: "error"; message: string }; export type AgentEvent = /** A task was created or its metadata changed. */ { type: "task"; task: AgentTask } | { type: "removed"; id: string } | -/** Entry `index` of `task` was added or replaced. */ -{ type: "entry"; task: string; index: number; entry: AgentEntry }; +/** Entry `index` of `task` (of its subagent `subagent`) was added or replaced. */ +{ type: "entry"; task: string; subagent: string | null; index: number; entry: AgentEntry }; export type AgentImage = { mimeType: string, @@ -1065,6 +1085,11 @@ export type AgentStatus = /** The turn ended; the result can be reviewed or the conversation continued. */ "ready" | "failed" | "archived"; +/** What a subagent is doing, as its parent reports it. */ +export type AgentSubagentStatus = "running" | +/** One of its requests (a permission) waits for an answer. */ +"needsInput" | "done" | "cancelled" | "failed"; + export type AgentTask = { id: string, title: string, diff --git a/src/features/agents/Session.module.css b/src/features/agents/Session.module.css index 1d882ce7f..fb627a2bb 100644 --- a/src/features/agents/Session.module.css +++ b/src/features/agents/Session.module.css @@ -550,3 +550,185 @@ .removed { color: var(--danger); } + +/* Subagents: a card per subagent in its parent's conversation, and the way into its own. */ +.subagent { + display: flex; + flex-direction: column; + gap: 4px; + padding: 6px 8px; + border: 1px solid var(--border); + border-radius: var(--radius-small); + background: var(--bg-elev); +} +.subagentHead { + display: flex; + align-items: center; + gap: 4px; + min-width: 0; +} +.subagentOpen { + flex: 1; + min-width: 0; + display: flex; + align-items: center; + gap: 6px; + padding: 2px 4px; + margin: -2px -4px; + border: 0; + border-radius: var(--radius-row); + background: none; + color: var(--fg); + font: inherit; + text-align: left; + cursor: pointer; +} +.subagentOpen:hover { + background: var(--bg-hover); +} +.subagentTitle { + flex: none; + max-width: 50%; + font-weight: 600; + color: var(--fg-strong); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} +.subagentDescription { + flex: 1 1 0; + min-width: 0; + color: var(--fg-muted); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} +.subagentMeta { + flex: none; + margin-left: auto; + color: var(--fg-subtle); + font-size: var(--fs-small); + font-variant-numeric: tabular-nums; + white-space: nowrap; +} +.subagentActivity { + padding-left: 22px; + color: var(--fg-muted); + font-family: var(--font-mono); + font-size: var(--fs-small); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} +.subagentSection > summary { + padding-left: 22px; + color: var(--fg-muted); + font-size: var(--fs-small); + cursor: pointer; + list-style: none; +} +.subagentSection > summary::-webkit-details-marker { + display: none; +} +.subagentSection > summary:hover { + color: var(--fg); +} +.subagentSection > :not(summary) { + margin: 4px 0 2px 22px; +} +.subagentReply { + max-height: 320px; + overflow: auto; +} +.done { + color: var(--success); +} + +.subagentBar { + flex: none; + display: flex; + align-items: center; + gap: 6px; + height: 30px; + padding: 0 8px 0 6px; + border-bottom: 1px solid var(--border); + min-width: 0; +} +.crumbs { + flex: 1; + min-width: 0; + display: flex; + align-items: center; + gap: 4px; + overflow: hidden; +} +.crumb { + flex: 0 1 auto; + min-width: 0; + padding: 0; + border: 0; + background: none; + color: var(--link); + font: inherit; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; + cursor: pointer; +} +.crumb:hover { + color: var(--link-hover); + text-decoration: underline; +} +.crumbCurrent { + flex: 0 1 auto; + min-width: 0; + font-weight: 600; + color: var(--fg-strong); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} + +.messageFrom { + margin-bottom: 2px; + color: var(--fg-subtle); + font-size: var(--fs-small); +} + +.readOnly { + flex: none; + padding: 10px 16px 14px; + border-top: 1px solid var(--border); + color: var(--fg-muted); +} + +.subagentStrip { + display: flex; + flex-wrap: wrap; + gap: 4px; +} +.subagentChip { + display: inline-flex; + align-items: center; + gap: 6px; + max-width: 240px; + height: 22px; + padding: 0 8px; + border: 1px solid var(--border); + border-radius: 11px; + background: var(--bg-elev); + color: var(--fg-secondary); + font: inherit; + font-size: var(--fs-small); + cursor: pointer; +} +.subagentChip:hover { + background: var(--bg-hover); + color: var(--fg); +} +.subagentChipTitle { + min-width: 0; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} diff --git a/src/features/agents/SessionEditor.tsx b/src/features/agents/SessionEditor.tsx index b6e6351af..2ebd07561 100644 --- a/src/features/agents/SessionEditor.tsx +++ b/src/features/agents/SessionEditor.tsx @@ -1,6 +1,7 @@ /** An agent task as an editor tab: the conversation, a composer and the task's changed files. */ -import { useEffect, useLayoutEffect, useMemo, useRef, useState, type RefObject } from "react"; +import { Fragment, useEffect, useLayoutEffect, useMemo, useRef, useState, type RefObject } from "react"; import clsx from "clsx"; +import { useShallow } from "zustand/react/shallow"; import type { AgentChange, AgentEntry, AgentTask, AgentUsage } from "@/bindings"; import { executeAction } from "@/platform/actions"; import { call, commands, onIdeEvent } from "@/platform/ipc"; @@ -28,7 +29,18 @@ import { canSend, emptyDraft, usageLabel, workTime } from "./composerModel"; import { ConfigPickers } from "./ConfigPickers"; import { findRanges, formatDuration, formatSentAt } from "./conversation"; import { Markdown } from "./Markdown"; -import { branchLabel, isBusy, providerIcon, statusLabel } from "./model"; +import { + branchLabel, + entriesKey, + findSubagent, + isBusy, + providerIcon, + statusLabel, + subagentStatusLabel, + subagentTitle, + subagentWorking, + type SubagentEntry, +} from "./model"; import { archiveTask, cancelTask, @@ -56,11 +68,23 @@ export function SessionEditor({ input }: { input: EditorInput }) { const entries = useAgents((s) => s.entries.get(id)); const changes = useAgents((s) => s.changes.get(id)); const head = useLocalHead(task?.isolation === "local" ? task.status : null); + // The subagent conversation shown, with the subagents that lead to it (empty: the task's own). + const [path, setPath] = useState([]); + const view = path.at(-1) ?? null; + const viewEntries = useAgents((s) => s.entries.get(entriesKey(id, view))); + // Each subagent's card, from its parent's conversation. + const cards = useAgents( + useShallow((s) => path.map((sub, i) => findSubagent(s.entries.get(entriesKey(id, path[i - 1] ?? null)), sub))), + ); useEffect(() => { void loadEntries(id).catch(() => undefined); void refreshChanges(id); return markVisible(id); }, [id]); + useEffect(() => { + if (view) void loadEntries(id, view).catch(() => undefined); + }, [id, view]); + const openSubagent = (sub: string) => setPath((p) => [...p, sub]); const [finding, setFinding] = useState(false); const [findFocus, setFindFocus] = useState(0); const [session] = useState(() => ({ @@ -88,15 +112,25 @@ export function SessionEditor({ input }: { input: EditorInput }) { >
+ {view && } setFinding(false)} /> - + {view ? ( +
+ Subagents take their instructions from the agent that started them. +
+ ) : ( + + )}
@@ -210,16 +244,70 @@ function useNow(interval: number): number { return now; } +/** Where a subagent conversation sits: the way back to the task, and the subagent's state. */ +function SubagentBar({ + task, + path, + cards, + onNavigate, +}: { + task: AgentTask; + path: string[]; + cards: Array; + onNavigate: (path: string[]) => void; +}) { + const tx = useTx(); + const card = cards.at(-1); + return ( +
+ onNavigate(path.slice(0, -1))} /> + + {card && ( + + {subagentStatusLabel(card.status)} + + )} + {card && subagentWorking(card.status) && card.cancellable && ( + void cancelTask(task.id, card.id)} /> + )} +
+ ); +} + interface StreamProps { task: AgentTask; entries: AgentEntry[] | null; + /** The card of the subagent whose conversation this is; `null`: the task's own. */ + subagent: SubagentEntry | null; + onOpenSubagent: (id: string) => void; finding: boolean; /** Bumped by Find in Conversation: focus the find field again. */ findFocus: number; onCloseFind: () => void; } -function Stream({ task, entries, finding, findFocus, onCloseFind }: StreamProps) { +function Stream({ task, entries, subagent, onOpenSubagent, finding, findFocus, onCloseFind }: StreamProps) { const ref = useRef(null); const stick = useRef(true); useLayoutEffect(() => { @@ -255,9 +343,16 @@ function Stream({ task, entries, finding, findFocus, onCloseFind }: StreamProps) {entries === null ? ( ) : ( - entries.map((entry, i) => ) + entries.map((entry, i) => ) )} - {(task.status === "running" || task.status === "starting") && !waiting && ( + {subagent && subagentWorking(subagent.status) && ( +
+ + {subagent.status === "needsInput" ? "Waiting for your answer…" : "Working…"} + {subagent.startedMs != null && } +
+ )} + {!subagent && (task.status === "running" || task.status === "starting") && !waiting && (
{task.status === "starting" ? "Starting the agent…" : "Working…"} @@ -439,7 +534,15 @@ const TOOL_ICONS: Record = { think: "todo", }; -function Entry({ task, entry }: { task: AgentTask; entry: AgentEntry }) { +function Entry({ + task, + entry, + onOpenSubagent, +}: { + task: AgentTask; + entry: AgentEntry; + onOpenSubagent: (id: string) => void; +}) { switch (entry.type) { case "user": return ; @@ -470,6 +573,24 @@ function Entry({ task, entry }: { task: AgentTask; entry: AgentEntry }) { ))}
); + case "subagent": + return onOpenSubagent(entry.id)} />; + case "message": + return entry.incoming ? ( +
+
+ Task from the parent agent +
+
{entry.text}
+
+ ) : ( +
+
+ Report to the parent agent +
+ +
+ ); case "permission": return ; case "error": @@ -568,6 +689,102 @@ function ToolEntry({ task, entry }: { task: AgentTask; entry: Extract; + case "needsInput": + return ; + case "done": + return ; + case "cancelled": + return ; + case "failed": + return ; + } +} + +/** A subagent in its parent's conversation: what it does, its latest step, what it was asked and answered. */ +function SubagentCard({ task, entry, onOpen }: { task: AgentTask; entry: SubagentEntry; onOpen: () => void }) { + const tx = useTx(); + const working = subagentWorking(entry.status); + const now = useNow(working ? 1000 : 60_000); + const time = entry.startedMs == null ? null : (entry.endedMs ?? now) - entry.startedMs; + const title = subagentTitle(entry); + return ( +
+
+ + {working && entry.cancellable && ( + void cancelTask(task.id, entry.id)} /> + )} +
+ {working && entry.activity && ( +
+ {shortTitle(task, entry.activity)} +
+ )} + {entry.error && ( +
+ + {entry.error} +
+ )} + {entry.prompt && ( +
+ + Task + +
{entry.prompt}
+
+ )} + {entry.reply && ( +
+ + Result + +
+ +
+
+ )} +
+ ); +} + +/** The subagents of the conversation still at work, to look in on. */ +function WorkingSubagents({ entries, onOpen }: { entries: AgentEntry[] | null; onOpen: (id: string) => void }) { + const tx = useTx(); + const working = (entries ?? []).filter((e): e is SubagentEntry => e.type === "subagent" && subagentWorking(e.status)); + if (!working.length) return null; + return ( +
+ {working.map((e) => ( + + ))} +
+ ); +} + /** Agent-written titles with the task's own folder dropped from paths. */ const shortTitle = (task: AgentTask, title: string) => title.replaceAll(`${task.cwd}/`, ""); @@ -606,7 +823,15 @@ function PermissionBanner({ task, entries }: { task: AgentTask; entries: AgentEn ); } -function SessionComposer({ task, entries }: { task: AgentTask; entries: AgentEntry[] | null }) { +function SessionComposer({ + task, + entries, + onOpenSubagent, +}: { + task: AgentTask; + entries: AgentEntry[] | null; + onOpenSubagent: (id: string) => void; +}) { const project = useProject((s) => s.project); const composer = useAgents((s) => s.composers.get(task.id)) ?? emptyDraft(); const [sending, setSending] = useState(false); @@ -630,6 +855,7 @@ function SessionComposer({ task, entries }: { task: AgentTask; entries: AgentEnt ); return (
+ {task.queue.length > 0 && } { expect(attentionChange(undefined, "ready")).toBeNull(); }); + it("finds subagent cards and keys their conversations apart", () => { + const card: SubagentEntry = { + type: "subagent", + id: "c1", + title: null, + description: "Find the lexer", + status: "running", + error: null, + prompt: null, + reply: null, + tools: 0, + activity: null, + startedMs: 0, + endedMs: null, + cancellable: false, + }; + expect(findSubagent([{ type: "agent", text: "a" }, card], "c1")).toBe(card); + expect(findSubagent(undefined, "c1")).toBeUndefined(); + expect(subagentTitle(card)).toBe("Find the lexer"); + expect(subagentTitle({ ...card, title: "Explore" })).toBe("Explore"); + expect(entriesKey("t", null)).toBe("t"); + expect(entriesKey("t", "c1")).not.toBe(entriesKey("t", null)); + }); + it("formats labels", () => { expect(branchLabel("agent/fix-lexer")).toBe("fix-lexer"); expect(branchLabel(null)).toBe(""); diff --git a/src/features/agents/model.ts b/src/features/agents/model.ts index 620dd52ee..d9504352f 100644 --- a/src/features/agents/model.ts +++ b/src/features/agents/model.ts @@ -1,5 +1,5 @@ /** Pure helpers: task grouping, entry updates, labels. */ -import type { AgentEntry, AgentStatus, AgentTask } from "@/bindings"; +import type { AgentEntry, AgentStatus, AgentSubagentStatus, AgentTask } from "@/bindings"; export const SESSION_KIND = "agent-session"; export const HOME_KIND = "agent-home"; @@ -8,6 +8,35 @@ export const TOOL_WINDOW = "agents"; export const sessionKey = (task: string) => `agent:${task}`; +/** Key of a conversation in the entries store: the task's own, or its subagent `subagent`'s. */ +export const entriesKey = (task: string, subagent: string | null) => (subagent ? `${task}/${subagent}` : task); + +export type SubagentEntry = Extract; + +/** The card of subagent `id` in its parent's `entries`. */ +export const findSubagent = (entries: AgentEntry[] | undefined, id: string) => + entries?.find((e): e is SubagentEntry => e.type === "subagent" && e.id === id); + +export const subagentWorking = (s: AgentSubagentStatus) => s === "running" || s === "needsInput"; + +/** "Explore", else what it was started for. */ +export const subagentTitle = (e: SubagentEntry) => e.title ?? e.description ?? "Subagent"; + +export function subagentStatusLabel(s: AgentSubagentStatus): string { + switch (s) { + case "running": + return "Running"; + case "needsInput": + return "Needs input"; + case "done": + return "Done"; + case "cancelled": + return "Stopped"; + case "failed": + return "Failed"; + } +} + const PROVIDER_ICONS = new Set(["claude", "codex", "gemini", "copilot", "cursor", "opencode", "qwen", "kimi", "goose", "auggie", "kiro", "vibe"]); /** The agent's logo; custom agents get the generic agents icon. */ diff --git a/src/features/agents/store.ts b/src/features/agents/store.ts index c5d7cb5d0..a3bc43a31 100644 --- a/src/features/agents/store.ts +++ b/src/features/agents/store.ts @@ -15,7 +15,7 @@ import { call, commands, isTauri } from "@/platform/ipc"; import { editorService, notifications } from "@/workbench/contributions"; import { canSend, emptyDraft, toPrompt, type ComposerDraft, type FileRef } from "./composerModel"; import { withQuote } from "./conversation"; -import { HOME_KEY, HOME_KIND, SESSION_KIND, attentionChange, isBusy, sessionKey, upsertEntry } from "./model"; +import { HOME_KEY, HOME_KIND, SESSION_KIND, attentionChange, entriesKey, isBusy, sessionKey, upsertEntry } from "./model"; /** New Task page settings (its text lives in the `HOME` composer). */ export interface Draft { @@ -35,7 +35,7 @@ interface AgentsState { /** Why the tasks could not be loaded. */ error: string | null; tasks: Map; - /** Entries of the tasks opened in this window. */ + /** Entries of the tasks (and subagents) opened in this window, by `entriesKey`. */ entries: Map; changes: Map; providers: AgentProvider[]; @@ -170,11 +170,12 @@ function onEvent(e: AgentEvent) { break; } case "entry": { - const current = s.entries.get(e.task); + const key = entriesKey(e.task, e.subagent); + const current = s.entries.get(key); if (!current) return; const next = upsertEntry(current, e.index, e.entry); - if (next) useAgents.setState({ entries: new Map(s.entries).set(e.task, next) }); - else void loadEntries(e.task); + if (next) useAgents.setState({ entries: new Map(s.entries).set(key, next) }); + else void loadEntries(e.task, e.subagent); if (e.entry.type === "tool" && e.entry.status === "completed" && e.entry.diffs.length) { scheduleChanges(e.task); } @@ -200,9 +201,10 @@ function notifyChange(prev: AgentTask | undefined, task: AgentTask) { }); } -export async function loadEntries(task: string): Promise { - const entries = await call(commands.agentEntries(task)); - useAgents.setState((s) => ({ entries: new Map(s.entries).set(task, entries) })); +/** Loads the entries of the task's conversation, or of its subagent `subagent`'s. */ +export async function loadEntries(task: string, subagent: string | null = null): Promise { + const entries = await call(commands.agentEntries(task, subagent)); + useAgents.setState((s) => ({ entries: new Map(s.entries).set(entriesKey(task, subagent), entries) })); } export async function refreshChanges(task: string): Promise { @@ -373,7 +375,8 @@ export function addReferences(refs: Array<{ path: string; rel: string; mention: else openHome(); } -export const cancelTask = (task: string) => call(commands.agentCancel(task)); +/** Stops the task's turn, or with `subagent` only that subagent's work. */ +export const cancelTask = (task: string, subagent: string | null = null) => call(commands.agentCancel(task, subagent)); export const respond = (task: string, permission: number, option: string | null) => call(commands.agentRespond(task, permission, option)); export const archiveTask = (task: string) => call(commands.agentArchive(task)); diff --git a/src/platform/locales/text/zh.json b/src/platform/locales/text/zh.json index 7848331f9..5286b6a5b 100644 --- a/src/platform/locales/text/zh.json +++ b/src/platform/locales/text/zh.json @@ -3208,6 +3208,17 @@ "New worktree": "新 worktree", "Starting…": "正在启动…", "Waiting for your answer": "等待你的回答", + "Waiting for your answer…": "等待你的回答…", + "Subagent": "子 agent", + "Subagents at work": "工作中的子 agent", + "Stop Subagent": "停止子 agent", + "Open the subagent's conversation": "打开子 agent 的对话", + "Subagents take their instructions from the agent that started them.": "子 agent 只接收启动它的 agent 的指令。", + "Task from the parent agent": "父 agent 交给的任务", + "Report to the parent agent": "交给父 agent 的报告", + "Task": "任务", + "tool call": "次工具调用", + "tool calls": "次工具调用", "The agent stopped": "Agent 已停止", "Not installed": "未安装", "{command} was not found on PATH": "在 PATH 中找不到 {command}",