diff --git a/docs/user/subagents-and-acp-harnesses.md b/docs/user/subagents-and-acp-harnesses.md index 3b80d17a..b2ade669 100644 --- a/docs/user/subagents-and-acp-harnesses.md +++ b/docs/user/subagents-and-acp-harnesses.md @@ -176,7 +176,9 @@ Without `output_schema`, `output` is text. A turn that emits selected non-text a } ``` -Text-only turns omit `updates`. Capture is limited to 64 update objects and 64 KiB of serialized update data per turn. Excess or oversized updates are omitted and set `updates.truncated` to `true`. When an ACP tool update's rendered content is only a JSON copy of its structured `rawOutput`, Kit retains only `rawOutput` in the parent-visible update. Kit also captures usage (including reported cumulative session cost), session info, available commands, notices, compaction lifecycle updates, and compaction summary chunks. These remain separate from the final answer text. Usage values describe context occupancy, not billable token deltas; cumulative costs must not be summed across updates. Kit does not add child costs to parent billing totals. Child thoughts, user-message echoes, modes, and configuration are not exposed through this value. Session titles and context usage also feed the live agent roster independently of these capture limits; thoughts are not forwarded as activity text. +Text-only turns omit `updates`. Capture is limited to 64 update objects and 64 KiB of serialized update data per turn. Excess or oversized updates are omitted and set `updates.truncated` to `true`. When an ACP tool update's rendered content is only a JSON copy of its structured `rawOutput`, Kit retains `rawOutput` and replaces the duplicate content with an empty array. This preserves the update's explicit replacement of earlier rich content. Kit also captures usage (including reported cumulative session cost), session info, available commands, notices, compaction lifecycle updates, and compaction summary chunks. These remain separate from the final answer text. Usage values describe context occupancy, not billable token deltas; cumulative costs must not be summed across updates. Kit does not add child costs to parent billing totals. Child thoughts, user-message echoes, modes, and configuration are not exposed through this value. Session titles and context usage also feed the live agent roster independently of these capture limits; thoughts are not forwarded as activity text. + +After a successful child turn, ACP clients receive the child's final diff and terminal content on the existing `subagent`, `prompt`, or `fork` tool card. Content updates replace earlier snapshots; content chunks append. V1 file snapshots also map to v2 file changes. Native v2 diffs without complete before/after text remain v2-only, as do agent-owned terminals. Child terminal output is replayed at completion, not streamed live; terminal IDs are scoped to the parent invocation. Kit does not execute captured terminal commands. Missing exit information remains unknown, and incomplete terminal captures carry `kit/outputIncomplete` metadata. Display is limited to 64 rich content items, with the same metadata marking additional display truncation. Raw child output and captured terminal IDs are not rewritten by this display projection. Failed or cancelled child turns do not publish a completion snapshot. ## Choose the built-in `acp.kit` harness diff --git a/fixtures/mock-acp.py b/fixtures/mock-acp.py index 68dc815f..13f4cafb 100644 --- a/fixtures/mock-acp.py +++ b/fixtures/mock-acp.py @@ -211,6 +211,28 @@ def prompt(request): if "MOCK_REFUSAL" in text: respond(request["id"], {"stopReason": "refusal"}) return + if text == "MOCK_TOOL_CONTENT": + diff = ({"type": "diff", "changes": [{"operation": "modify", "path": "/tmp/child.txt"}], + "patch": {"format": "git_patch", "text": "-old\n+new\n"}} + if "--v2" in sys.argv else + {"type": "diff", "path": "/tmp/child.txt", "oldText": "old\n", "newText": "new\n"}) + raw = {"stdout": "child output"} + updates = [ + {"sessionUpdate": "tool_call_update", "toolCallId": "call-1", + "content": [{"type": "content", "content": {"type": "text", "text": json.dumps(raw)}}, diff], + "rawOutput": raw}, + ] + if "--v2" in sys.argv: + updates += [ + {"sessionUpdate": "terminal_update", "terminalId": "terminal-1", "command": "echo child output"}, + {"sessionUpdate": "tool_call_update", "toolCallId": "call-2", + "content": [{"type": "terminal", "terminalId": "terminal-1"}]}, + {"sessionUpdate": "terminal_output_chunk", "terminalId": "terminal-1", "data": "Y2hpbGQgb3V0cHV0Cg=="}, + {"sessionUpdate": "terminal_update", "terminalId": "terminal-1", "exitStatus": {"exitCode": 0}}, + ] + for update in updates: + send({"jsonrpc": "2.0", "method": "session/update", "params": {"sessionId": session_id, "update": update}}) + text = "tool content done" if "MOCK_RICH_OUTPUT" in text: updates = [ { diff --git a/src/acp_child.rs b/src/acp_child.rs index e4179ae6..59d5f065 100644 --- a/src/acp_child.rs +++ b/src/acp_child.rs @@ -862,6 +862,16 @@ impl ChildOutput { self.capture_value(value); } + /// Preserve the original rich wire content: v1 decoding skips v2 diffs and + /// cannot represent agent-owned terminal notifications. The existing shared + /// count/byte budget applies before these values can reach a parent result. + fn record_tool_value(&mut self, mut value: Value) { + if value["sessionUpdate"] == "tool_call_update" { + deduplicate_tool_output(&mut value); + } + self.capture_value(value); + } + fn capture_value(&mut self, value: Value) { if self.updates.len() >= MAX_CAPTURED_UPDATES { self.updates_truncated = true; @@ -893,7 +903,9 @@ fn deduplicate_tool_output(update: &mut Value) { if rendered_text_only(content).is_some_and(|text| { serde_json::from_str::(text).is_ok_and(|value| value == *raw_output) }) { - object.remove("content"); + // Keep the explicit replacement: omission would resurrect earlier rich + // content when a parent folds these updates into a final snapshot. + object.insert("content".into(), Value::Array(Vec::new())); } } @@ -1401,6 +1413,19 @@ async fn run( if let Ok(mut output) = route.output.lock() { output.record_message(update); } return Ok(()); } + // Capture rich tool values without round-tripping through the v1 + // schema, which silently drops v2 diff entries and terminal events. + // Build the owned value before locking; recording has no callbacks, + // wakeups, or awaits and shares ChildOutput's existing byte budget. + let tool_update = matches!(params["update"]["sessionUpdate"].as_str(), + Some("tool_call" | "tool_call_update" | "tool_call_content_chunk" + | "terminal_update" | "terminal_output_chunk")); + if tool_update { + let value = params["update"].clone(); + if let Ok(mut output) = route.output.lock() { + output.record_tool_value(value); + } + } // Reuse normalized configuration identifiers for captured output. let Some(notification) = notification else { return Ok(()); }; if !route.owner.is_empty() && let Some(activity) = roster_activity(¬ification.update) { @@ -1418,7 +1443,7 @@ async fn run( }), }); } - if let Ok(mut output) = route.output.lock() { + if !tool_update && let Ok(mut output) = route.output.lock() { output.record(notification.update); } Ok(()) @@ -2825,7 +2850,7 @@ mod tests { } #[test] - fn captured_tool_updates_drop_content_that_duplicates_raw_output() { + fn captured_tool_updates_clear_content_that_duplicates_raw_output() { let raw = json!({"exit_code": 0, "stdout": "done", "stderr": "", "success": true}); let mut output = ChildOutput::default(); output.record(update(json!({ @@ -2840,10 +2865,39 @@ mod tests { }))); assert_eq!(output.updates.len(), 1); - assert!(output.updates[0].get("content").is_none()); + assert_eq!(output.updates[0]["content"], json!([])); assert_eq!(output.updates[0]["rawOutput"]["stdout"], "done"); } + #[test] + fn captured_v2_tool_values_keep_explicit_rich_content_replacement() { + let mut output = ChildOutput::default(); + output.record_tool_value(json!({ + "sessionUpdate": "tool_call_update", "toolCallId": "a", + "content": [{"type": "diff", "changes": [{"operation": "modify", "path": "/file"}]}] + })); + let raw = json!({"stdout": "done"}); + output.record_tool_value(json!({ + "sessionUpdate": "tool_call_update", "toolCallId": "a", + "content": [{"type": "content", "content": {"type": "text", "text": raw.to_string()}}], + "rawOutput": raw + })); + assert_eq!( + output.updates[0]["content"][0]["changes"][0]["path"], + "/file" + ); + assert_eq!(output.updates[1]["content"], json!([])); + assert_eq!(output.updates[1]["rawOutput"], raw); + assert_eq!( + output.update_bytes, + output + .updates + .iter() + .map(|value| serde_json::to_vec(value).unwrap().len()) + .sum::() + ); + } + #[test] fn captured_tool_updates_keep_distinct_content_and_raw_output() { let mut output = ChildOutput::default(); diff --git a/src/protocols/acp/tool_projection.rs b/src/protocols/acp/tool_projection.rs index 7bb2bfd8..b42489f9 100644 --- a/src/protocols/acp/tool_projection.rs +++ b/src/protocols/acp/tool_projection.rs @@ -17,6 +17,7 @@ use tokio::sync::broadcast; )] mod tests; +pub(crate) mod child; pub(crate) mod terminal; #[cfg(test)] @@ -221,19 +222,11 @@ pub(crate) fn diff(request: &ToolRequest, path: &Path, old: Option<&str>, new: O || request.session_id.0.len() > MAX_ID || request.call_id.0.len() > MAX_ID || !request.call_id.0.contains(":compose:") - || old - .map_or(0, str::len) - .saturating_add(new.map_or(0, str::len)) - > MAX_DIFF_TEXT - || location_value(path, None).is_none() { return; } - let operation = match (old, new) { - (None, Some(_)) => "add", - (Some(_), None) => "delete", - (Some(_), Some(_)) => "modify", - (None, None) => return, + let Some(content) = diff_content(path, old, new) else { + return; }; publish(Update { session: request.session_id.0.clone(), @@ -241,28 +234,44 @@ pub(crate) fn diff(request: &ToolRequest, path: &Path, old: Option<&str>, new: O start: None, patch: Some(Value::Object(Map::from_iter([ ("toolCallId".into(), Value::from(request.call_id.0.clone())), - ( - "content".into(), - Value::Array(vec![Value::Object(Map::from_iter([ - ("type".into(), Value::from("diff")), - ("path".into(), Value::from(path.to_str())), - ("oldText".into(), Value::from(old)), - ("newText".into(), Value::from(new.unwrap_or_default())), - ( - "changes".into(), - Value::Array(vec![Value::Object(Map::from_iter([ - ("operation".into(), Value::from(operation)), - ("path".into(), Value::from(path.to_str())), - ("fileType".into(), Value::from("text")), - ]))]), - ), - ]))]), - ), + ("content".into(), Value::Array(vec![content])), ]))), ok: false, }); } +/// Shared conversion for local committed edits and captured v1 child snapshots. +fn diff_content(path: &Path, old: Option<&str>, new: Option<&str>) -> Option { + if old + .map_or(0, str::len) + .saturating_add(new.map_or(0, str::len)) + > MAX_DIFF_TEXT + || location_value(path, None).is_none() + { + return None; + } + let operation = match (old, new) { + (None, Some(_)) => "add", + (Some(_), None) => "delete", + (Some(_), Some(_)) => "modify", + (None, None) => return None, + }; + Some(Value::Object(Map::from_iter([ + ("type".into(), Value::from("diff")), + ("path".into(), Value::from(path.to_str())), + ("oldText".into(), Value::from(old)), + ("newText".into(), Value::from(new.unwrap_or_default())), + ( + "changes".into(), + Value::Array(vec![Value::Object(Map::from_iter([ + ("operation".into(), Value::from(operation)), + ("path".into(), Value::from(path.to_str())), + ("fileType".into(), Value::from("text")), + ]))]), + ), + ]))) +} + fn location_value(path: &Path, line: Option) -> Option { if !path.is_absolute() || path.as_os_str().len() > MAX_PATH { return None; @@ -453,7 +462,7 @@ fn forward_event( event: Result, receiver: &mut broadcast::Receiver, session: &str, - active: &mut HashMap, + active: &mut HashMap>, budget: &mut terminal::Budget, send: &impl Fn(Update) -> bool, ) -> Result<(), agentkit_acp::AcpRuntimeError> { @@ -463,16 +472,43 @@ fn forward_event( if active.len() >= CAPACITY || active.contains_key(&update.call) { return Ok(()); } - active.insert(update.call.clone(), terminal::State::default()); + active.insert(update.call.clone(), HashMap::new()); } else if let Some(patch) = &update.patch { - let Some(state) = active.get_mut(&update.call) else { + let Some(terminals) = active.get_mut(&update.call) else { return Ok(()); }; - if patch.get("sessionUpdate").and_then(Value::as_str) == Some("terminal_update") { - state.running = patch.get("exitStatus").is_none(); - } - if !budget.admit(&mut update, state) { - return Ok(()); + let kind = patch.get("sessionUpdate").and_then(Value::as_str); + if matches!(kind, Some("terminal_update" | "terminal_output_chunk")) { + let Some(id) = patch + .get("terminalId") + .and_then(Value::as_str) + .filter(|id| id.len() <= MAX_ID) + else { + return Ok(()); + }; + if !terminals.contains_key(id) + && (terminals.len() >= CAPACITY / 4 + || kind == Some("terminal_output_chunk")) + { + return Ok(()); + } + let state = terminals + .entry(id.to_owned()) + .or_insert_with(terminal::State::running); + if kind == Some("terminal_update") + && let Some(exit) = patch.get("exitStatus") + { + // Omission preserves the last state; null explicitly clears + // an exit, and only an object declares the terminal exited. + if exit.is_null() { + state.running = true; + } else if exit.is_object() { + state.running = false; + } + } + if !budget.admit(&mut update, state) { + return Ok(()); + } } } else if active.remove(&update.call).is_none() { return Ok(()); @@ -483,30 +519,33 @@ fn forward_event( } Ok(_) => {} Err(error) => { - for (call, state) in active.drain() { - // Loss invalidates the stream as well as its card. Do not leave - // an editor waiting for a terminal exit frame that was dropped. - if state.running - && !send(Update { + for (call, terminals) in active.drain() { + // Child replay terminals have independent IDs. Invalidate each + // affected stream, without asserting a child process has exited. + for (id, _) in terminals.into_iter().filter(|(_, state)| state.running) { + let mut patch = Map::from_iter([ + ("sessionUpdate".into(), Value::from("terminal_update")), + ("terminalId".into(), Value::from(id.clone())), + ( + "_meta".into(), + Value::Object(Map::from_iter([( + "kit/outputIncomplete".into(), + Value::from(true), + )])), + ), + ]); + if id == call { + patch.insert("exitStatus".into(), Value::Object(Map::new())); + } + if !send(Update { session: session.into(), call: call.clone(), start: None, - patch: Some(Value::Object(Map::from_iter([ - ("sessionUpdate".into(), Value::from("terminal_update")), - ("terminalId".into(), Value::from(call.clone())), - ("exitStatus".into(), Value::Object(Map::new())), - ( - "_meta".into(), - Value::Object(Map::from_iter([( - "kit/outputIncomplete".into(), - Value::from(true), - )])), - ), - ]))), + patch: Some(Value::Object(patch)), ok: false, - }) - { - return Err(delivery_error()); + }) { + return Err(delivery_error()); + } } if !send(Update { session: session.into(), diff --git a/src/protocols/acp/tool_projection/child.rs b/src/protocols/acp/tool_projection/child.rs new file mode 100644 index 00000000..386b12b0 --- /dev/null +++ b/src/protocols/acp/tool_projection/child.rs @@ -0,0 +1,277 @@ +//! Completion-time display of bounded child content on its existing parent card. +//! Raw child results remain unchanged; no child terminal command is executed. +use std::collections::{BTreeMap, BTreeSet}; + +use agentkit_tools_core::ToolRequest; +use serde_json::{Map, Value}; + +use super::{CAPACITY, MAX_ID, Update, buses, subscribers}; + +const MAX_ITEMS: usize = CAPACITY / 4; +const MAX_BYTES: usize = 64 * 1024; + +pub(crate) fn publish(request: &ToolRequest, items: &[Value], truncated: bool) { + if subscribers() == 0 + || request.session_id.0.len() > MAX_ID + || request.call_id.0.len() > MAX_ID + || !request.call_id.0.contains(":compose:") + { + return; + } + for patch in patches(request, items, truncated) { + buses().publish(Update { + session: request.session_id.0.clone(), + call: request.call_id.0.clone(), + start: None, + patch: Some(patch), + ok: false, + }); + } +} + +fn patches(request: &ToolRequest, items: &[Value], truncated: bool) -> Vec { + // Defend this boundary as well as ChildOutput's shared capture budget. + if items.len() > MAX_ITEMS + || items.iter().fold(0usize, |bytes, item| { + bytes + .saturating_add(serde_json::to_vec(item).map_or(MAX_BYTES + 1, |value| value.len())) + }) > MAX_BYTES + { + return Vec::new(); + } + let mut calls = BTreeMap::<&str, Vec<&Value>>::new(); + for item in items { + let Some(id) = item["toolCallId"].as_str().filter(|id| id.len() <= MAX_ID) else { + continue; + }; + match item["sessionUpdate"].as_str() { + Some("tool_call" | "tool_call_update") => { + if let Some(content) = item.get("content") { + if content.is_null() { + calls.insert(id, Vec::new()); + } else if let Some(content) = content.as_array() { + calls.insert(id, content.iter().collect()); + } + } + } + Some("tool_call_content_chunk") => { + let content = calls.entry(id).or_default(); + content.push(&item["content"]); + } + _ => {} + } + } + let mut content = Vec::new(); + let mut legacy = Vec::new(); + let mut terminals = BTreeMap::<&str, String>::new(); + let mut incomplete = truncated; + for entry in calls.values().flatten() { + if !matches!(entry["type"].as_str(), Some("diff" | "terminal")) { + continue; + } + if content.len() >= MAX_ITEMS { + incomplete = true; + break; + } + match entry["type"].as_str() { + Some("diff") => { + let snapshot = legacy_diff(entry); + // Dual-format content can carry a more precise native operation + // (for example delete) and a patch. Do not replace it with the + // legacy snapshot's inferred operation in the v2 projection. + let native = + serde_json::from_value::((*entry).clone()) + .ok() + .filter(|diff| !diff.changes.is_empty()) + .and_then(|diff| serde_json::to_value(diff).ok()); + if let Some(mut diff) = native { + diff["type"] = Value::from("diff"); + content.push(diff); + } else if let Some(diff) = &snapshot { + content.push(diff.clone()); + } + if let Some(diff) = snapshot { + legacy.push(diff); + } + } + Some("terminal") => { + if let Some(id) = entry["terminalId"].as_str().filter(|id| id.len() <= MAX_ID) { + let mapped = terminal_id(request, id); + if terminals.insert(id, mapped.clone()).is_none() { + content.push(object([ + ("type", Value::from("terminal")), + ("terminalId", Value::from(mapped)), + ])); + } + } + } + _ => {} + } + } + if content.is_empty() { + return Vec::new(); + } + let mut patches = Vec::new(); + let mut exited = BTreeSet::new(); + // Explicit upserts make even a truncated capture's references well-defined. + // A missing real exit is marked incomplete below, never invented. + for mapped in terminals.values() { + patches.push(object([ + ("sessionUpdate", Value::from("terminal_update")), + ("terminalId", Value::from(mapped.clone())), + ])); + } + for item in items { + let Some(id) = item["terminalId"].as_str() else { + continue; + }; + let Some(mapped) = terminals.get(id) else { + continue; + }; + let kind = item["sessionUpdate"].as_str(); + if !matches!(kind, Some("terminal_update" | "terminal_output_chunk")) { + continue; + } + // Validate terminal wire fields; do not forward arbitrary child events. + let Ok(update) = + serde_json::from_value::(item.clone()) + else { + continue; + }; + let Ok(mut value) = serde_json::to_value(update) else { + continue; + }; + // Keep only bounded display metadata. Raw child metadata remains in the + // result, but must not bypass the terminal subscription's payload budget. + let source_incomplete = value["_meta"]["kit/outputIncomplete"] == true; + if let Some(object) = value.as_object_mut() { + object.remove("_meta"); + } + for field in ["output", "exitStatus"] { + if let Some(object) = value.get_mut(field).and_then(Value::as_object_mut) { + object.remove("_meta"); + } + } + let oversized_signal = value["exitStatus"]["signal"] + .as_str() + .is_some_and(|signal| signal.len() > 128); + if oversized_signal + && let Some(exit) = value.get_mut("exitStatus").and_then(Value::as_object_mut) + { + exit.remove("signal"); + } + if source_incomplete || oversized_signal { + value["_meta"] = object([("kit/outputIncomplete", Value::from(true))]); + } + value["terminalId"] = Value::from(mapped.clone()); + if kind == Some("terminal_update") + && let Some(status) = value.get("exitStatus") + { + if status.is_object() { + exited.insert(id); + } else if status.is_null() { + exited.remove(id); + } + } + patches.push(value); + } + for (id, mapped) in terminals { + if !exited.contains(id) || incomplete { + patches.push(object([ + ("sessionUpdate", Value::from("terminal_update")), + ("terminalId", Value::from(mapped)), + ( + "_meta", + object([("kit/outputIncomplete", Value::Bool(true))]), + ), + ])); + } + } + // Legacy file snapshots are representable on both protocols. Native v2 diffs + // without old/new text cannot be truthfully turned into a v1 file snapshot. + let metadata = if incomplete { + Some(object([ + ("kit/outputIncomplete", Value::Bool(true)), + ( + "kit/parentToolCallId", + Value::from( + request + .call_id + .0 + .rsplit_once(":compose:") + .map(|(owner, _)| owner), + ), + ), + ])) + } else { + None + }; + if !legacy.is_empty() { + let mut patch = object([ + ("toolCallId", Value::from(request.call_id.0.clone())), + ("content", Value::Array(legacy)), + ]); + if let Some(metadata) = &metadata { + patch["_meta"] = metadata.clone(); + } + patches.push(patch); + } + let mut final_content = object([ + ("sessionUpdate", Value::from("tool_call_update")), + ("toolCallId", Value::from(request.call_id.0.clone())), + ("content", Value::Array(content)), + ]); + if let Some(metadata) = metadata { + final_content["_meta"] = metadata; + } + patches.push(final_content); + patches +} + +fn legacy_diff(entry: &Value) -> Option { + let diff: agent_client_protocol::schema::v1::Diff = + serde_json::from_value(entry.clone()).ok()?; + super::diff_content(&diff.path, diff.old_text.as_deref(), Some(&diff.new_text)) +} + +fn object(fields: [(&str, Value); N]) -> Value { + Value::Object(Map::from_iter( + fields.into_iter().map(|(key, value)| (key.into(), value)), + )) +} + +fn terminal_id(request: &ToolRequest, id: &str) -> String { + let mut hash = blake3::Hasher::new(); + for part in [&request.session_id.0, &request.call_id.0, id] { + hash.update(&(part.len() as u64).to_le_bytes()); + hash.update(part.as_bytes()); + } + format!("child-{}", hash.finalize().to_hex()) +} + +#[cfg(test)] +#[allow( + clippy::unwrap_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +mod tests; + +#[cfg(test)] +#[allow( + clippy::unwrap_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +mod integration_tests; + +#[cfg(test)] +#[allow( + clippy::unwrap_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +mod lifecycle_tests; diff --git a/src/protocols/acp/tool_projection/child/integration_tests.rs b/src/protocols/acp/tool_projection/child/integration_tests.rs new file mode 100644 index 00000000..6da99cd0 --- /dev/null +++ b/src/protocols/acp/tool_projection/child/integration_tests.rs @@ -0,0 +1,129 @@ +use super::super::*; +use agentkit_core::{MetadataMap, SessionId, ToolCallId, ToolOutput, TurnId}; +use agentkit_tools_core::{AllowAllPermissions, OwnedToolContext, ToolName, ToolSource}; +use serde_json::json; +use std::{collections::BTreeMap, sync::Arc}; + +#[tokio::test] +async fn successful_subagent_prompt_and_fork_project_before_completion_on_both_protocols() { + for child_v2 in [false, true] { + let root = tempfile::tempdir().unwrap(); + let runtime = crate::Runtime::new(root.path(), "gpt-5.4").unwrap(); + let mut args = vec![format!( + "{}/fixtures/mock-acp.py", + env!("CARGO_MANIFEST_DIR") + )]; + if child_v2 { + args.push("--v2".into()); + } + let harnesses = crate::AcpHarnesses::new(BTreeMap::from([( + "rich".into(), + crate::AcpHarnessProfile { + command: "python3".into(), + args, + permissions: Default::default(), + }, + )])) + .unwrap(); + let runtime = + crate::Runtime::with_acp_harnesses(runtime, harnesses, "acp.rich".into()).unwrap(); + let session = if child_v2 { + "real-child-v2" + } else { + "real-child-v1" + }; + let (send_v1, mut recv_v1) = tokio::sync::mpsc::unbounded_channel(); + let (send_v2, mut recv_v2) = tokio::sync::mpsc::unbounded_channel(); + let v1 = Subscription::start(session.into(), move |update| send_v1.send(update).is_ok()); + let v2 = Subscription::start_v2(session.into(), move |update| send_v2.send(update).is_ok()); + let context = OwnedToolContext { + session_id: SessionId::new(session), + turn_id: TurnId::new("turn"), + metadata: MetadataMap::new(), + permissions: Arc::new(AllowAllPermissions), + resources: Arc::new(()), + cancellation: None, + execution_scope: None, + approved_request: None, + }; + let tools = runtime.compose(0); + let mut prior = Value::Null; + let mut terminal_ids = Vec::new(); + for name in ["subagent", "prompt", "fork"] { + let mut input = json!({"prompt": "MOCK_TOOL_CONTENT"}); + if name != "subagent" { + input["subagent"] = prior.clone(); + } + let request = ToolRequest { + session_id: SessionId::new(session), + turn_id: TurnId::new("turn"), + call_id: ToolCallId::new(format!("outer:compose:{name}")), + tool_name: ToolName::new(name), + input, + metadata: MetadataMap::new(), + }; + let tool = tools.get(&ToolName::new(name)).unwrap(); + let result = tool + .invoke(request.clone(), &mut context.borrowed()) + .await + .unwrap(); + let ToolOutput::Structured(value) = result.result.output else { + panic!("expected handle"); + }; + prior = value; + assert_eq!(prior["output"], "tool content done"); + v1.drain().await.unwrap(); + v2.drain().await.unwrap(); + let first = std::iter::from_fn(|| recv_v1.try_recv().ok()) + .map(|update| serde_json::to_value(update.v1().unwrap()).unwrap()) + .collect::>(); + let second = std::iter::from_fn(|| recv_v2.try_recv().ok()) + .map(|update| serde_json::to_value(update.v2().unwrap()).unwrap()) + .collect::>(); + for events in [&first, &second] { + assert_eq!(events[0]["toolCallId"], request.call_id.0); + assert_eq!(events.last().unwrap()["status"], "completed"); + assert_eq!(events.last().unwrap()["toolCallId"], request.call_id.0); + } + if !child_v2 { + let content = first + .iter() + .find_map(|update| update.get("content")) + .unwrap(); + assert_eq!(content[0]["oldText"], "old\n"); + assert_eq!(content[0]["newText"], "new\n"); + } else { + assert!( + first.iter().all(|update| update.get("terminalId").is_none() + && update.get("content").is_none()) + ); + let chunk = second + .iter() + .position(|update| update["sessionUpdate"] == "terminal_output_chunk") + .unwrap(); + let content = second + .iter() + .position(|update| update.get("content").is_some()) + .unwrap(); + assert!(chunk < content && content < second.len() - 1); + let id = second[chunk]["terminalId"].as_str().unwrap().to_owned(); + assert!(!terminal_ids.contains(&id)); + assert_eq!(second[content]["content"][1]["terminalId"], id); + terminal_ids.push(id); + assert_eq!( + prior["updates"]["items"][2]["content"][0]["terminalId"], + "terminal-1" + ); + } + let rich = second + .iter() + .rev() + .find_map(|update| update.get("content")) + .unwrap(); + assert_eq!(rich[0]["changes"][0]["operation"], "modify"); + if child_v2 { + assert_eq!(rich[0]["patch"]["text"], "-old\n+new\n"); + } + } + } +} diff --git a/src/protocols/acp/tool_projection/child/lifecycle_tests.rs b/src/protocols/acp/tool_projection/child/lifecycle_tests.rs new file mode 100644 index 00000000..a745fa23 --- /dev/null +++ b/src/protocols/acp/tool_projection/child/lifecycle_tests.rs @@ -0,0 +1,113 @@ +use super::super::*; +use serde_json::json; + +fn update(patch: Value) -> Update { + Update { + session: "child-lifecycle".into(), + call: "outer:compose:child".into(), + start: None, + patch: Some(patch), + ok: false, + } +} + +#[test] +fn terminal_budget_markers_preserve_each_child_terminal_identity() { + let mut budget = terminal::Budget::default(); + let mut first = terminal::State::running(); + for _ in 0..128 { + let mut value = update( + json!({"sessionUpdate": "terminal_output_chunk", "terminalId": "child-first", "data": "YQ=="}), + ); + assert!(budget.admit(&mut value, &mut first)); + assert_eq!(value.value()["sessionUpdate"], "terminal_output_chunk"); + } + for (id, state) in [ + ("child-first", &mut first), + ("child-second", &mut terminal::State::running()), + ] { + let mut value = update( + json!({"sessionUpdate": "terminal_output_chunk", "terminalId": id, "data": "YQ=="}), + ); + assert!(budget.admit(&mut value, state)); + assert_eq!(value.value()["terminalId"], id); + assert_eq!(value.value()["_meta"]["kit/outputIncomplete"], true); + assert!(value.value().get("exitStatus").is_none()); + } + for metadata in [Value::Null, json!({"child": "replacement"})] { + let mut exit = update( + json!({"sessionUpdate": "terminal_update", "terminalId": "child-first", + "exitStatus": {"exitCode": 0}, "_meta": metadata}), + ); + assert!(budget.admit(&mut exit, &mut first)); + assert_eq!(exit.value()["_meta"]["kit/outputIncomplete"], true); + assert_eq!(exit.value()["exitStatus"]["exitCode"], 0); + } + let mut snapshot = update( + json!({"sessionUpdate": "terminal_update", "terminalId": "child-third", + "output": {"data": "YQ==", "truncated": false}, "exitStatus": {"exitCode": 0}}), + ); + assert!(budget.admit(&mut snapshot, &mut terminal::State::running())); + assert!(snapshot.value().get("output").is_none()); + assert_eq!(snapshot.value()["exitStatus"]["exitCode"], 0); + assert_eq!(snapshot.value()["_meta"]["kit/outputIncomplete"], true); +} + +#[test] +fn lag_marks_only_live_child_terminal_ids_without_fabricated_exits() { + let (sender, mut receiver) = broadcast::channel(8); + let (send, mut receive) = tokio::sync::mpsc::unbounded_channel(); + let sink = |update| send.send(update).is_ok(); + let mut active = HashMap::new(); + let mut budget = terminal::Budget::default(); + let mut start = update(Value::Null); + start.patch = None; + start.start = Some(json!({"toolCallId": start.call, "title": "Child"})); + forward_event( + Ok(start), + &mut receiver, + "child-lifecycle", + &mut active, + &mut budget, + &sink, + ) + .unwrap(); + for patch in [ + json!({"sessionUpdate": "terminal_update", "terminalId": "live"}), + json!({"sessionUpdate": "terminal_update", "terminalId": "exited", "exitStatus": {"exitCode": 0}}), + json!({"sessionUpdate": "terminal_update", "terminalId": "exited", "command": "metadata only"}), + ] { + forward_event( + Ok(update(patch)), + &mut receiver, + "child-lifecycle", + &mut active, + &mut budget, + &sink, + ) + .unwrap(); + } + while receive.try_recv().is_ok() {} + for _ in 0..9 { + sender.send(update(Value::Null)).unwrap(); + } + let Err(broadcast::error::TryRecvError::Lagged(lost)) = receiver.try_recv() else { + panic!("expected real channel lag"); + }; + forward_event( + Err(broadcast::error::RecvError::Lagged(lost)), + &mut receiver, + "child-lifecycle", + &mut active, + &mut budget, + &sink, + ) + .unwrap(); + let marker = receive.try_recv().unwrap().value(); + assert_eq!(marker["terminalId"], "live"); + assert_eq!(marker["_meta"]["kit/outputIncomplete"], true); + assert!(marker.get("exitStatus").is_none()); + assert_eq!(receive.try_recv().unwrap().value()["status"], "failed"); + assert!(receive.try_recv().is_err()); + assert!(active.is_empty()); +} diff --git a/src/protocols/acp/tool_projection/child/tests.rs b/src/protocols/acp/tool_projection/child/tests.rs new file mode 100644 index 00000000..2581fcc7 --- /dev/null +++ b/src/protocols/acp/tool_projection/child/tests.rs @@ -0,0 +1,202 @@ +use super::*; +use serde_json::json; + +fn request(session: &str, name: &str, input: Value) -> ToolRequest { + ToolRequest { + session_id: agentkit_core::SessionId::new(session), + turn_id: agentkit_core::TurnId::new("turn"), + call_id: agentkit_core::ToolCallId::new("parent:compose:node"), + tool_name: agentkit_tools_core::ToolName::new(name), + input, + metadata: agentkit_core::MetadataMap::new(), + } +} + +fn diff(path: &str) -> Value { + json!({"type": "diff", "path": path, "oldText": "before", "newText": "after"}) +} + +#[test] +fn replacements_clears_and_chunks_form_one_final_snapshot() { + let request = request("snapshots", "subagent", json!({})); + let items = vec![ + json!({"sessionUpdate": "tool_call", "toolCallId": "a", "content": [diff("/old")]}), + json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": [diff("/new")]}), + json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "status": "completed"}), + json!({"sessionUpdate": "tool_call", "toolCallId": "b", "content": [diff("/cleared")]}), + json!({"sessionUpdate": "tool_call_update", "toolCallId": "b", "content": null}), + json!({"sessionUpdate": "tool_call_content_chunk", "toolCallId": "b", "content": diff("/chunk")}), + ]; + let patches = patches(&request, &items, false); + let content = patches.last().unwrap()["content"].as_array().unwrap(); + assert_eq!(content.len(), 2); + assert_eq!(content[0]["path"], "/new"); + assert_eq!(content[1]["path"], "/chunk"); + assert_eq!(patches[0]["content"], patches[1]["content"]); + assert_eq!(patches[0]["toolCallId"], request.call_id.0); +} + +#[test] +fn native_v2_diffs_do_not_invent_v1_file_snapshots() { + let request = request("native", "prompt", json!({})); + let content = json!({"type": "diff", "changes": [{"operation": "modify", "path": "/file"}], + "patch": {"format": "git_patch", "text": "-old\n+new\n"}}); + let items = + [json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": [content]})]; + let patches = patches(&request, &items, false); + assert_eq!(patches.len(), 1); + assert_eq!(patches[0]["sessionUpdate"], "tool_call_update"); + let decoded: agentkit_acp::v2::wire::SessionUpdate = + serde_json::from_value(patches[0].clone()).unwrap(); + assert_eq!( + serde_json::to_value(decoded).unwrap()["content"][0]["patch"]["text"], + "-old\n+new\n" + ); +} + +#[test] +fn terminal_replay_is_namespaced_ordered_and_does_not_change_raw_values() { + let request = request("terminals", "fork", json!({})); + let items = vec![ + json!({"sessionUpdate": "terminal_update", "terminalId": "t", "command": "echo hi"}), + json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": [{"type": "terminal", "terminalId": "t"}]}), + json!({"sessionUpdate": "terminal_output_chunk", "terminalId": "t", "data": "aGkK"}), + json!({"sessionUpdate": "terminal_update", "terminalId": "t", "exitStatus": {"exitCode": 0}}), + ]; + let original = items.clone(); + let patches = patches(&request, &items, false); + assert_eq!(items, original); + let id = terminal_id(&request, "t"); + assert_eq!(patches[0]["terminalId"], id); + assert_eq!(patches[1]["command"], "echo hi"); + assert_eq!(patches[2]["data"], "aGkK"); + assert_eq!(patches[3]["exitStatus"]["exitCode"], 0); + assert_eq!(patches.last().unwrap()["content"][0]["terminalId"], id); + let mut other = request.clone(); + other.session_id = agentkit_core::SessionId::new("other"); + assert_ne!(terminal_id(&other, "t"), id); + other = request.clone(); + other.call_id = agentkit_core::ToolCallId::new("parent:compose:next-generation"); + assert_ne!(terminal_id(&other, "t"), id); + for patch in patches { + let _: agentkit_acp::v2::wire::SessionUpdate = serde_json::from_value(patch).unwrap(); + } +} + +#[test] +fn incomplete_terminal_capture_does_not_invent_process_exit() { + let request = request("incomplete", "subagent", json!({})); + let items = [ + json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": [ + {"type": "terminal", "terminalId": "missing"} + ]}), + ]; + let patches = patches(&request, &items, true); + assert_eq!(patches[0]["sessionUpdate"], "terminal_update"); + assert!(patches[1].get("exitStatus").is_none()); + assert_eq!(patches[1]["_meta"]["kit/outputIncomplete"], true); + assert!(patches[1]["exitStatus"].get("exitCode").is_none()); +} + +#[test] +fn capture_byte_accounting_excludes_array_overhead() { + let request = request("exact-bound", "subagent", json!({})); + let mut item = json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", + "content": [diff("/file")], "padding": ""}); + let padding = MAX_BYTES - serde_json::to_vec(&item).unwrap().len(); + item["padding"] = Value::from("x".repeat(padding)); + assert_eq!(serde_json::to_vec(&item).unwrap().len(), MAX_BYTES); + assert!(!patches(&request, &[item], false).is_empty()); +} + +#[test] +fn display_limit_filters_plain_content_first_and_marks_rich_truncation() { + let request = request("content-limit", "subagent", json!({})); + let mut content = vec![ + json!({"type": "content", "content": {"type": "text", "text": "plain"}}); + MAX_ITEMS + 1 + ]; + content.push(diff("/after-plain")); + let item = json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": content}); + let result = patches(&request, &[item], false); + assert_eq!(result.last().unwrap()["content"][0]["path"], "/after-plain"); + let item = json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", + "content": vec![diff("/many"); MAX_ITEMS + 1]}); + let result = patches(&request, &[item], false); + assert_eq!( + result.last().unwrap()["content"].as_array().unwrap().len(), + MAX_ITEMS + ); + assert_eq!( + result.last().unwrap()["_meta"]["kit/outputIncomplete"], + true + ); +} + +#[test] +fn cleared_terminal_exit_remains_unknown_after_replay() { + let request = request("exit-clear", "subagent", json!({})); + let items = [ + json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": [{"type": "terminal", "terminalId": "t"}]}), + json!({"sessionUpdate": "terminal_update", "terminalId": "t", "exitStatus": {"exitCode": 0}}), + json!({"sessionUpdate": "terminal_update", "terminalId": "t", "exitStatus": null}), + ]; + let result = patches(&request, &items, false); + let marker = &result[result.len() - 2]; + assert_eq!(marker["_meta"]["kit/outputIncomplete"], true); + assert!(marker.get("exitStatus").is_none()); +} + +#[test] +fn terminal_presentation_strips_unbudgeted_metadata_and_oversized_signals() { + let request = request("metadata", "subagent", json!({})); + let items = [ + json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", + "content": [{"type": "terminal", "terminalId": "t"}]}), + json!({"sessionUpdate": "terminal_update", "terminalId": "t", + "exitStatus": {"signal": "s".repeat(200), "_meta": {"secret": "nested"}}, + "_meta": {"secret": "x".repeat(1024)}}), + ]; + let result = patches(&request, &items, false); + let exit = &result[1]; + assert!(exit["exitStatus"].get("signal").is_none()); + assert!(exit["exitStatus"].get("_meta").is_none()); + assert_eq!(exit["_meta"]["kit/outputIncomplete"], true); + assert!(!serde_json::to_string(&result).unwrap().contains("secret")); + assert!(items[1]["_meta"].get("secret").is_some()); +} + +#[test] +fn count_and_byte_limits_reject_oversized_snapshots() { + let request = request("bounds", "subagent", json!({})); + assert!(patches(&request, &vec![json!({}); MAX_ITEMS + 1], false).is_empty()); + assert!(patches(&request, &[json!({"huge": "x".repeat(MAX_BYTES)})], false).is_empty()); +} + +#[test] +fn dual_format_diff_preserves_native_operation_and_patch() { + let request = request("dual-format", "subagent", json!({})); + let item = json!({"sessionUpdate": "tool_call_update", "toolCallId": "a", "content": [{ + "type": "diff", "path": "/deleted", "oldText": "before", "newText": "", + "changes": [{"operation": "delete", "path": "/deleted"}], + "patch": {"format": "git_patch", "text": "-before"} + }]}); + let result = patches(&request, &[item], false); + assert_eq!(result.len(), 2); + assert_eq!(result[0]["content"][0]["oldText"], "before"); + assert_eq!(result[1]["content"][0]["changes"][0]["operation"], "delete"); + assert_eq!(result[1]["content"][0]["patch"]["text"], "-before"); +} + +#[test] +fn legacy_snapshots_use_parent_validation_and_text_budget() { + assert!(legacy_diff(&diff("relative")).is_none()); + let mut oversized = diff("/large"); + oversized["newText"] = Value::from("x".repeat(super::super::MAX_DIFF_TEXT)); + assert!(legacy_diff(&oversized).is_none()); + let empty = json!({"type": "diff", "path": "/empty", "newText": ""}); + assert_eq!( + legacy_diff(&empty).unwrap()["changes"][0]["operation"], + "add" + ); +} diff --git a/src/protocols/acp/tool_projection/terminal/budget.rs b/src/protocols/acp/tool_projection/terminal/budget.rs index fca80f09..7892f68f 100644 --- a/src/protocols/acp/tool_projection/terminal/budget.rs +++ b/src/protocols/acp/tool_projection/terminal/budget.rs @@ -13,6 +13,15 @@ pub(in super::super) struct State { incomplete: bool, } +impl State { + pub(in super::super) fn running() -> Self { + Self { + running: true, + incomplete: false, + } + } +} + pub(in super::super) struct Budget { bytes: usize, chunks: usize, @@ -50,7 +59,10 @@ impl Budget { // notice; suppress the rest without dropping the exit. *patch = Map::from_iter([ ("sessionUpdate".into(), Value::from("terminal_update")), - ("terminalId".into(), Value::from(update.call.clone())), + ( + "terminalId".into(), + patch.get("terminalId").cloned().unwrap_or(Value::Null), + ), ( "_meta".into(), Value::Object(Map::from_iter([( @@ -62,6 +74,29 @@ impl Budget { } } Some("terminal_update") => { + // Child terminals can carry replacement output snapshots, not + // just chunks. Charge those against the same lifetime budget. + if let Some(bytes) = patch + .get("output") + .and_then(|output| output.get("data")) + .and_then(Value::as_str) + .map(str::len) + { + if !state.incomplete && self.chunks > 0 && bytes <= self.bytes { + self.bytes -= bytes; + self.chunks -= 1; + } else { + patch.remove("output"); + state.incomplete = true; + patch.insert( + "_meta".into(), + Value::Object(Map::from_iter([( + "kit/outputIncomplete".into(), + Value::from(true), + )])), + ); + } + } // Command/cwd are optional, variable-size metadata. Charge the // worst-case JSON escape expansion too, across all shell calls. for key in ["command", "cwd"] { @@ -76,6 +111,19 @@ impl Budget { patch.remove(key); } } + // ACP metadata replaces the prior object. Once output admission + // loses bytes, later clears/exit metadata cannot hide that fact. + if state.incomplete { + let metadata = patch + .entry("_meta") + .or_insert_with(|| Value::Object(Map::new())); + if !metadata.is_object() { + *metadata = Value::Object(Map::new()); + } + if let Some(metadata) = metadata.as_object_mut() { + metadata.insert("kit/outputIncomplete".into(), Value::from(true)); + } + } } _ => {} } diff --git a/src/tools/subagent.rs b/src/tools/subagent.rs index 6528d05b..75a4df3f 100644 --- a/src/tools/subagent.rs +++ b/src/tools/subagent.rs @@ -2018,12 +2018,19 @@ fn result( ToolError::ExecutionFailed(error) } })?; - let value = serde_json::to_value(value).map_err(|error| { + let serialized = serde_json::to_value(&value).map_err(|error| { ToolError::ExecutionFailed(format!("failed to serialize subagent value: {error}")) })?; + if let Some(updates) = &value.updates { + crate::protocols::acp::tool_projection::child::publish( + &request, + &updates.items, + updates.truncated, + ); + } Ok(ToolResult::new(ToolResultPart::success( request.call_id, - ToolOutput::structured(value), + ToolOutput::structured(serialized), ))) } diff --git a/tests/runtime.rs b/tests/runtime.rs index 421dd5d2..2694468c 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -645,6 +645,59 @@ return { output: child.output, updates: child.updates }"#, ); } +#[tokio::test] +async fn subagent_preserves_v1_and_v2_diff_and_terminal_wire_content() { + for v2 in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let runtime = kit::Runtime::new(directory.path(), "gpt-5.4").unwrap(); + let mut args = vec![format!( + "{}/fixtures/mock-acp.py", + env!("CARGO_MANIFEST_DIR") + )]; + if v2 { + args.push("--v2".into()); + } + let harnesses = kit::AcpHarnesses::new(BTreeMap::from([( + "rich".into(), + kit::AcpHarnessProfile { + command: "python3".into(), + args, + permissions: Default::default(), + }, + )])) + .unwrap(); + let runtime = + kit::Runtime::with_acp_harnesses(runtime, harnesses, "acp.rich".into()).unwrap(); + let outcome = execute_compose( + &runtime, + r#"return subagent({prompt: "MOCK_TOOL_CONTENT"})"#, + ) + .await; + let ToolExecutionOutcome::Completed(result) = outcome else { + panic!("child failed: {outcome:?}"); + }; + let ToolOutput::Structured(value) = result.result.output else { + panic!("expected JSON"); + }; + assert_eq!(value["output"], "tool content done"); + assert_eq!(value["updates"]["truncated"], false); + let items = value["updates"]["items"].as_array().unwrap(); + assert_eq!(items[0]["rawOutput"], json!({"stdout": "child output"})); + let diff = &items[0]["content"][1]; + assert_eq!(diff["type"], "diff"); + if v2 { + assert_eq!(diff["changes"][0]["path"], "/tmp/child.txt"); + assert_eq!(items[1]["terminalId"], "terminal-1"); + assert_eq!(items[2]["content"][0]["terminalId"], "terminal-1"); + assert_eq!(items[3]["data"], "Y2hpbGQgb3V0cHV0Cg=="); + assert_eq!(items[4]["exitStatus"]["exitCode"], 0); + } else { + assert_eq!(diff["oldText"], "old\n"); + assert_eq!(diff["newText"], "new\n"); + } + } +} + #[tokio::test] async fn compose_dispatches_bundled_docs_search_through_runlet() { let directory = tempfile::tempdir().unwrap();