From b83b553be877237327343512341d89b89dfae028 Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 14 Sep 2026 00:07:16 +0100 Subject: [PATCH] chore(runtime): remove Runlet sidechannel progress tracking --- docs/user/compose-and-local-tools.md | 2 +- src/acp_child.rs | 35 +- .../transport.rs => diagnostic_transport.rs} | 14 +- .../tests.rs | 45 +- src/events.rs | 168 +-- src/events/test_support.rs | 5 - src/lib.rs | 2 +- src/protocols/acp/tool_projection.rs | 2 +- src/protocols/acp/tool_projection/terminal.rs | 4 +- src/protocols/acp/tool_projection/tests.rs | 2 +- src/runlet_progress.rs | 272 ---- src/runtime.rs | 6 +- src/tools/observed.rs | 100 +- src/tools/subagent/tests.rs | 1 - src/tui/app.rs | 357 +---- src/tui/mod.rs | 36 +- src/tui/progress.rs | 223 ---- src/tui/progress_tests.rs | 1181 ----------------- src/tui/runtime_health_tests.rs | 155 +++ src/tui/source.rs | 33 + src/tui/ui.rs | 297 +---- 21 files changed, 293 insertions(+), 2647 deletions(-) rename src/{runlet_progress/transport.rs => diagnostic_transport.rs} (93%) rename src/{runlet_progress/transport => diagnostic_transport}/tests.rs (90%) delete mode 100644 src/runlet_progress.rs delete mode 100644 src/tui/progress.rs delete mode 100644 src/tui/progress_tests.rs create mode 100644 src/tui/runtime_health_tests.rs create mode 100644 src/tui/source.rs diff --git a/docs/user/compose-and-local-tools.md b/docs/user/compose-and-local-tools.md index 1988c96d..2cd84534 100644 --- a/docs/user/compose-and-local-tools.md +++ b/docs/user/compose-and-local-tools.md @@ -32,7 +32,7 @@ Retries repeat the body. Do not retry a write unless repeating it is safe or the Background calls no longer hold their originating turn open, and interrupting that turn does not stop them. When a call detaches, the model receives its tool-call ID and can stop it with `close({ call_id: "call_..." })`. Cancellation is delivered through the same result lifecycle as completion, as a failed result reporting that tool execution was cancelled. -The TUI keeps every running call visible. A running compose card shows its Runlet source inline, with live call states, binding resolution, and loop or retry counts. Completion replaces the source with the compose output. Unless the user explicitly opened or closed it, the output collapses when a later tool call or model message arrives and remains available from the tool card. Completion or failure is delivered back to the owning session and wakes the session loop directly without inserting synthetic user content. Background work is process- and session-scoped rather than a durable operating-system job, so closing Kit ends its inspectable lifetime. +The TUI keeps every running call visible. A running compose card groups its canonical ACP child tool calls. Its Runlet source is available as a bounded, neutral view; the display does not infer binding, loop, or retry state. Completion replaces the source with the compose output. Unless the user explicitly opened or closed it, the output collapses when a later tool call or model message arrives and remains available from the tool card. Completion or failure is delivered back to the owning session and wakes the session loop directly without inserting synthetic user content. Background work is process- and session-scoped rather than a durable operating-system job, so closing Kit ends its inspectable lifetime. ## Ordering, dependencies, and concurrency diff --git a/src/acp_child.rs b/src/acp_child.rs index a755a0b1..2ee1e694 100644 --- a/src/acp_child.rs +++ b/src/acp_child.rs @@ -1248,12 +1248,7 @@ struct RunConfig { fn harness_diagnostic(label: &str, line: &str) -> Option { if matches!( crate::events::parse(line), - Some( - crate::events::RuntimeEvent::ChildStarted { .. } - | crate::events::RuntimeEvent::ChildFinished { .. } - | crate::events::RuntimeEvent::RunletProgress { .. } - | crate::events::RuntimeEvent::RunletTransport { .. } - ) + Some(crate::events::RuntimeEvent::RunletTransport { .. }) ) { return None; } @@ -1361,12 +1356,12 @@ async fn run( ancestor_id.as_deref(), |output| match output { ForwardedStderr::RuntimeLine(line) => { - if let Some(transport) = crate::runlet_progress::transport::global() { + if let Some(transport) = crate::diagnostic_transport::global() { transport.publish_runtime_line(&line); } } ForwardedStderr::Diagnostic(line) => { - if let Some(transport) = crate::runlet_progress::transport::global() { + if let Some(transport) = crate::diagnostic_transport::global() { transport.publish_line(&line); } } @@ -2003,9 +1998,8 @@ async fn forward_stderr( } match event { crate::events::RuntimeEvent::RunletTransport { available: true } => { - deadline = Some( - tokio::time::Instant::now() + crate::runlet_progress::transport::LEASE, - ); + deadline = + Some(tokio::time::Instant::now() + crate::diagnostic_transport::LEASE); continue; } crate::events::RuntimeEvent::RunletTransport { available: false } => { @@ -2017,8 +2011,7 @@ async fn forward_stderr( _ => {} } if deadline.is_some() { - deadline = - Some(tokio::time::Instant::now() + crate::runlet_progress::transport::LEASE); + deadline = Some(tokio::time::Instant::now() + crate::diagnostic_transport::LEASE); } if event.forward_from_child() { if let crate::events::RuntimeEvent::SubagentStateChanged { @@ -2560,12 +2553,7 @@ mod tests { #[test] fn nested_runtime_events_are_not_forwarded_as_parent_events() { - let event = crate::events::RuntimeEvent::ChildStarted { - call: "subagent-call:compose:shell".into(), - tool: "shell".into(), - summary: "inspect".into(), - at: 0, - }; + let event = crate::events::RuntimeEvent::RunletTransport { available: true }; let line = format!( "{}{}", crate::events::EVENT_MARKER, @@ -4835,11 +4823,8 @@ for line in sys.stdin: serde_json::to_string(&RuntimeEvent::RunletTransport { available: true }) .unwrap() ); - let started = RuntimeEvent::ChildStarted { - call: "parent:compose:0".into(), - tool: "shell".into(), - summary: "working".into(), - at: 1, + let started = RuntimeEvent::SubagentDescendantsRemoved { + ancestor_id: "parent".into(), }; let start = format!( "{EVENT_MARKER}{}\n", @@ -4861,7 +4846,7 @@ for line in sys.stdin: ); writer.write_all(reset.as_bytes()).await.unwrap(); } else { - tokio::time::advance(crate::runlet_progress::transport::LEASE).await; + tokio::time::advance(crate::diagnostic_transport::LEASE).await; } let reset = match rx.recv().await.unwrap() { ForwardedStderr::RuntimeLine(line) => crate::events::parse(&line).unwrap(), diff --git a/src/runlet_progress/transport.rs b/src/diagnostic_transport.rs similarity index 93% rename from src/runlet_progress/transport.rs rename to src/diagnostic_transport.rs index f7aeb1e4..71a47963 100644 --- a/src/runlet_progress/transport.rs +++ b/src/diagnostic_transport.rs @@ -1,6 +1,5 @@ //! One process-owned blocking writer, never joined by an execution future. //! Publication is bounded and nonblocking, including cancellation/Drop paths. -use super::Progress; use crate::events::{EVENT_MARKER, RuntimeEvent}; use std::{ io::{self, Write}, @@ -41,7 +40,7 @@ impl Transport { let diagnostics_lost = Arc::new(AtomicBool::new(false)); let worker_loss = diagnostics_lost.clone(); std::thread::Builder::new() - .name("runlet-diagnostics".into()) + .name("kit-diagnostics".into()) .spawn(move || { // Guard invalidates publication on success, error, or unwind. No IO // in its destructor and no restart that could reuse stale evidence. @@ -54,10 +53,6 @@ impl Transport { diagnostics_lost, }) } - pub(crate) fn publish(&self, progress: Progress) { - self.publish_event(&RuntimeEvent::RunletProgress { progress }); - } - /// Best-effort diagnostics remain available after authoritative loss. pub(crate) fn publish_line(&self, line: &str) { const TRUNCATED: &str = " [truncated]"; @@ -150,7 +145,7 @@ fn write_loop( Frame::Authoritative(bytes) | Frame::Diagnostic(bytes) => bytes, }; writer.write_all(&bytes)?; - // Busy legacy diagnostic traffic must not starve the lease. + // Busy diagnostic traffic must not starve the lease. if !disabled.load(Ordering::Acquire) && heartbeat.elapsed() >= HEARTBEAT { transport_status(&mut writer, runtime_events, true)?; heartbeat = Instant::now(); @@ -183,11 +178,6 @@ fn write_frame(writer: &mut impl Write, event: &RuntimeEvent) -> io::Result<()> } fn encode_frame(event: &RuntimeEvent) -> io::Result> { - if let RuntimeEvent::RunletProgress { progress } = event - && !progress.bounded() - { - return Err(io::Error::other("invalid progress metadata")); - } // A bounded writer, not an unbounded serialization followed by a size check. let mut frame = vec![0; MAX_FRAME_BYTES]; let len = { diff --git a/src/runlet_progress/transport/tests.rs b/src/diagnostic_transport/tests.rs similarity index 90% rename from src/runlet_progress/transport/tests.rs rename to src/diagnostic_transport/tests.rs index 11a161ce..886c78d0 100644 --- a/src/runlet_progress/transport/tests.rs +++ b/src/diagnostic_transport/tests.rs @@ -6,21 +6,8 @@ clippy::disallowed_macros )] use super::*; -use crate::runlet_progress::Change; use std::io::{BufRead, BufReader, Read}; -fn progress() -> Progress { - Progress { - owner: "parent".into(), - incarnation: 1, - sequence: 0, - change: Change::Started { - digest: "a".repeat(64), - healed: false, - }, - } -} - #[cfg(unix)] #[test] fn stalled_writer_loss_resets_after_drain_and_never_resumes() { @@ -40,14 +27,15 @@ fn stalled_writer_loss_resets_after_drain_and_never_resumes() { writer.set_nonblocking(false).unwrap(); let transport = Transport::start(writer, 2, true).unwrap(); for _ in 0..4 { - transport.publish_event(&RuntimeEvent::ChildFinished { - call: "parent:compose:0".into(), - tool: "shell".into(), + transport.publish_event(&RuntimeEvent::CompactionFinished { + reason: "test".into(), ok: true, - summary: "done".into(), + compacted: true, millis: 1, }); - transport.publish(progress()); + transport.publish_event(&RuntimeEvent::SubagentDescendantsRemoved { + ancestor_id: "parent".into(), + }); } assert!(transport.disabled.load(Ordering::Acquire)); reader @@ -68,7 +56,9 @@ fn stalled_writer_loss_resets_after_drain_and_never_resumes() { crate::events::parse(line.trim_end()), Some(RuntimeEvent::RunletTransport { available: false }) ); - transport.publish(progress()); + transport.publish_event(&RuntimeEvent::SubagentDescendantsRemoved { + ancestor_id: "parent".into(), + }); drop(transport); line.clear(); assert_eq!(reader.read_line(&mut line).unwrap(), 0); @@ -114,7 +104,9 @@ fn writer_error_and_unwind_fail_closed() { diagnostics_lost: Arc::new(AtomicBool::new(false)), }; assert!(transport.disabled.load(Ordering::Acquire)); - transport.publish(progress()); + transport.publish_event(&RuntimeEvent::SubagentDescendantsRemoved { + ancestor_id: "parent".into(), + }); } } @@ -135,12 +127,11 @@ fn last_sender_disconnect_finishes_transport() { #[cfg(unix)] #[test] -fn lifecycle_frames_share_queue_and_oversize_loss_invalidates_progress() { - let event = RuntimeEvent::ChildFinished { - call: "owner:compose:0".into(), - tool: "shell".into(), +fn lifecycle_frames_share_queue_and_oversize_loss_invalidates_lifecycle() { + let event = RuntimeEvent::CompactionFinished { + reason: "test".into(), ok: true, - summary: "done".into(), + compacted: true, millis: 1, }; let (writer, mut reader) = std::os::unix::net::UnixStream::pair().unwrap(); @@ -166,7 +157,9 @@ fn lifecycle_frames_share_queue_and_oversize_loss_invalidates_progress() { session_id: "x".repeat(MAX_FRAME_BYTES), }); assert!(transport.disabled.load(Ordering::Acquire)); - transport.publish(progress()); + transport.publish_event(&RuntimeEvent::SubagentDescendantsRemoved { + ancestor_id: "parent".into(), + }); transport.publish_line("later child error"); drop(transport); wire.clear(); diff --git a/src/events.rs b/src/events.rs index 363c5fa8..b6be0bf3 100644 --- a/src/events.rs +++ b/src/events.rs @@ -1,17 +1,5 @@ -//! Side channel that carries nested tool activity from `kit serve` to the -//! terminal client, along with the id of each persisted ACP session it opens. -//! -//! ACP reports the model-visible `compose` call, but every interesting thing -//! Kit does happens *inside* that call: the Runlet program dispatches shell, -//! edit, subagent, and A2A children concurrently. The terminal client renders -//! that as inline live script state, so it needs the child lifecycle. -//! -//! Rather than fork the ACP surface, the events ride on stderr — Kit's -//! diagnostics channel — as single JSON lines behind a control-character -//! marker. The terminal client owns the `serve` child process and pipes its -//! stderr, so marked lines become script-state updates and everything else -//! becomes log output. Emission is opt-in through `KIT_RUNTIME_EVENTS` so ordinary -//! ACP hosts never see the extra chatter. +//! Ephemeral stderr diagnostics for session attachment, runtime health, and +//! subagent lifecycle. Tool cards are projected through canonical ACP events. use std::{ sync::OnceLock, @@ -19,7 +7,6 @@ use std::{ }; use serde::{Deserialize, Serialize}; -use serde_json::Value; /// Prefix that distinguishes an event line from a diagnostic line. The /// leading control character cannot appear in ordinary log text. @@ -29,37 +16,15 @@ pub const EVENT_MARKER: &str = "\u{1}kit-runtime\u{1}"; pub const EVENTS_ENV: &str = "KIT_RUNTIME_EVENTS"; /// One runtime event sent privately to the terminal client. -/// -/// `call` is the compose child call id, shaped `:compose:`, -/// so a client can attribute every child to the ACP tool call it belongs to. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] #[serde(tag = "event", rename_all = "snake_case")] pub enum RuntimeEvent { - /// Process-wide progress transport lease/reset, not a source execution. + /// Process-wide diagnostic transport lease/reset. RunletTransport { available: bool }, - /// Authoritative, value-free observations owned by an exact compose call. - RunletProgress { - progress: crate::runlet_progress::Progress, - }, /// Process-wide durability state, independent of the active ACP session. StorageStatus { pending: bool, exhausted: bool }, /// A persisted ACP session was opened by the child runtime. SessionStarted { session_id: String }, - /// A nested tool call started running. - ChildStarted { - call: String, - tool: String, - summary: String, - at: u64, - }, - /// A nested tool call finished, successfully or not. - ChildFinished { - call: String, - tool: String, - ok: bool, - summary: String, - millis: u64, - }, /// Automatic transcript compaction started. CompactionStarted { reason: String, at: u64 }, /// Automatic transcript compaction finished. @@ -191,33 +156,12 @@ impl RuntimeEvent { pub(crate) fn forward_from_child(&self) -> bool { matches!( self, - Self::ChildStarted { .. } - | Self::ChildFinished { .. } - | Self::SubagentStateChanged { .. } + Self::SubagentStateChanged { .. } | Self::SubagentUsage { .. } | Self::SubagentActivity { .. } | Self::SubagentDescendantsRemoved { .. } ) } - - /// The ACP tool call this child belongs to, when the id carries one. - #[must_use] - pub fn parent_call(&self) -> Option<&str> { - let call = match self { - Self::RunletProgress { progress } => return Some(&progress.owner), - Self::ChildStarted { call, .. } | Self::ChildFinished { call, .. } => call, - Self::RunletTransport { .. } - | Self::StorageStatus { .. } - | Self::SessionStarted { .. } - | Self::CompactionStarted { .. } - | Self::CompactionFinished { .. } - | Self::SubagentStateChanged { .. } - | Self::SubagentUsage { .. } - | Self::SubagentActivity { .. } - | Self::SubagentDescendantsRemoved { .. } => return None, - }; - call.rsplit_once(":compose:").map(|(parent, _)| parent) - } } /// Whether this process should emit runtime events. @@ -228,12 +172,12 @@ pub fn enabled() -> bool { } /// Enqueues one event without waiting for stderr. Loss disables the transport; -/// its explicit reset (or the client lease on a stalled sink) hides source state. +/// its explicit reset (or the client lease on a stalled sink) invalidates runtime status. pub fn emit(event: &RuntimeEvent) { if !enabled() { return; } - if let Some(transport) = crate::runlet_progress::transport::global() { + if let Some(transport) = crate::diagnostic_transport::global() { transport.publish_event(event); } } @@ -245,15 +189,7 @@ pub fn parse(line: &str) -> Option { if body.len() > 64 * 1024 { return None; } - // Existing diagnostic events retain their historical parser shape. The new - // bounded payload is checked before it can reach retained UI state. - let event: RuntimeEvent = serde_json::from_str(body).ok()?; - if let RuntimeEvent::RunletProgress { progress } = &event - && (body.len() > 4096 || !progress.bounded()) - { - return None; - } - Some(event) + serde_json::from_str(body).ok() } /// Milliseconds since the Unix epoch, saturating at zero on a broken clock. @@ -265,61 +201,6 @@ pub fn now_millis() -> u64 { .unwrap_or_default() } -/// One short line describing what a nested call was asked to do. -#[must_use] -pub fn summarize_input(input: &Value) -> String { - subject( - input, - &[ - "command", "path", "file", "prompt", "task", "query", "url", "message", - ], - ) -} - -/// One short line describing what a nested call produced. -#[must_use] -pub fn summarize_output(output: &Value) -> String { - subject( - output, - &["stdout", "text", "message", "summary", "path", "error"], - ) -} - -/// One short line describing a tool payload. -/// -/// Tool inputs and outputs are small JSON objects whose most descriptive field -/// differs per tool, so the first field that reads like a subject wins, and -/// anything unexpected falls back to compact JSON. -fn subject(value: &Value, keys: &[&str]) -> String { - let named = value.as_object().and_then(|fields| { - keys.iter() - .filter_map(|key| fields.get(*key)) - .find(|field| !matches!(field, Value::String(text) if text.trim().is_empty())) - .map(render_value) - }); - truncate(&named.unwrap_or_else(|| render_value(value)), 160) -} - -fn render_value(value: &Value) -> String { - let text = match value { - Value::String(text) => text.clone(), - other => other.to_string(), - }; - text.lines() - .find(|line| !line.trim().is_empty()) - .unwrap_or_default() - .trim() - .to_string() -} - -fn truncate(text: &str, limit: usize) -> String { - if text.chars().count() <= limit { - return text.to_string(); - } - let kept: String = text.chars().take(limit).collect(); - format!("{kept}…") -} - #[cfg(test)] #[allow( clippy::unwrap_used, @@ -330,13 +211,12 @@ fn truncate(text: &str, limit: usize) -> String { clippy::disallowed_macros )] mod tests { - use std::io::{self, Write}; - use serde_json::json; + use std::io::{self, Write}; use super::{ EVENT_MARKER, GenerationOutcome, HarnessVendor, RuntimeEvent, SubagentStatus, parse, - summarize_input, summarize_output, test_support::write_event, + test_support::write_event, }; #[test] @@ -348,21 +228,18 @@ mod tests { let wire = String::from_utf8(wire).unwrap(); let parsed = parse(wire.trim_end()).unwrap(); assert_eq!(parsed, event); - assert_eq!(parsed.parent_call(), None); } } #[test] fn reads_back_an_emitted_event_line() { - let event = RuntimeEvent::ChildStarted { - call: "call-1:compose:abcdef".into(), - tool: "shell".into(), - summary: "ls".into(), + let event = RuntimeEvent::CompactionStarted { + reason: "test".into(), at: 7, }; let line = format!("{EVENT_MARKER}{}", serde_json::to_string(&event).unwrap()); let parsed = parse(&line).expect("event round trips"); - assert_eq!(parsed.parent_call(), Some("call-1")); + assert_eq!(parsed, event); } #[test] @@ -397,7 +274,6 @@ mod tests { for event in [changed, removed] { let line = format!("{EVENT_MARKER}{}", serde_json::to_string(&event).unwrap()); assert_eq!(parse(&line), Some(event.clone())); - assert_eq!(event.parent_call(), None); } } @@ -451,7 +327,6 @@ mod tests { let line = format!("{EVENT_MARKER}{}", serde_json::to_string(&event).unwrap()); assert_eq!(parse(&line), Some(event.clone())); assert!(event.forward_from_child()); - assert_eq!(event.parent_call(), None); } } @@ -469,7 +344,6 @@ mod tests { let parsed = parse(&line).unwrap(); assert_eq!(parsed, usage); assert!(parsed.forward_from_child()); - assert_eq!(parsed.parent_call(), None); } #[test] @@ -595,8 +469,8 @@ mod tests { }; let line = format!("{EVENT_MARKER}{}", serde_json::to_string(&event).unwrap()); let parsed = parse(&line).expect("event round trips"); + assert_eq!(parsed, event); assert!(matches!(parsed, RuntimeEvent::CompactionStarted { .. })); - assert_eq!(parsed.parent_call(), None); } #[test] @@ -624,22 +498,6 @@ mod tests { fn ignores_ordinary_diagnostics() { assert!(parse("listening on 127.0.0.1:7331").is_none()); } - - #[test] - fn summarizes_by_the_most_descriptive_field() { - assert_eq!( - summarize_input(&json!({ "timeout_seconds": 30, "command": "cargo test" })), - "cargo test" - ); - assert_eq!( - summarize_output(&json!({ "success": true, "stdout": "ok\nrest" })), - "ok" - ); - assert_eq!( - summarize_output(&json!({ "success": true, "stdout": "" })), - "{\"success\":true,\"stdout\":\"\"}" - ); - } } #[cfg(test)] diff --git a/src/events/test_support.rs b/src/events/test_support.rs index d606bdf7..c97bbd1d 100644 --- a/src/events/test_support.rs +++ b/src/events/test_support.rs @@ -61,11 +61,6 @@ pub(crate) fn with_stalled_stderr(test: &str) -> bool { use super::{EVENT_MARKER, RuntimeEvent}; pub(crate) fn write_event(writer: &mut impl std::io::Write, event: &RuntimeEvent) { - if let RuntimeEvent::RunletProgress { progress } = event - && !progress.bounded() - { - return; - } if let Ok(line) = serde_json::to_string(event) { let _ = writeln!(writer, "{EVENT_MARKER}{line}"); } diff --git a/src/lib.rs b/src/lib.rs index b32eb078..c81948d1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod config_editor; #[doc(hidden)] pub mod config_files; mod credentials; +mod diagnostic_transport; pub mod docs; pub mod events; mod fatal; @@ -34,7 +35,6 @@ pub(crate) mod process_tree; pub mod protocols; pub mod provider; pub mod request_budget; -mod runlet_progress; pub mod runtime; /// Shared internal filesystem and process-lifetime recovery controls. pub mod resilient_fs { diff --git a/src/protocols/acp/tool_projection.rs b/src/protocols/acp/tool_projection.rs index b42489f9..b39dab20 100644 --- a/src/protocols/acp/tool_projection.rs +++ b/src/protocols/acp/tool_projection.rs @@ -28,7 +28,7 @@ pub(crate) mod terminal; )] mod diff_tests; -const CAPACITY: usize = crate::runlet_progress::MAX_NODES; +const CAPACITY: usize = 256; const MAX_ID: usize = 256; const MAX_PATH: usize = 4096; diff --git a/src/protocols/acp/tool_projection/terminal.rs b/src/protocols/acp/tool_projection/terminal.rs index 5cf1953f..2e7de650 100644 --- a/src/protocols/acp/tool_projection/terminal.rs +++ b/src/protocols/acp/tool_projection/terminal.rs @@ -1,5 +1,7 @@ //! Agent-owned terminals are v2 only. Frames share the bounded invocation bus. use super::{MAX_ID, Update, publish, v2_bus}; + +pub(super) const MAX_COMMAND_BYTES: usize = 64 * 1024; use agentkit_tools_core::ToolRequest; use base64::{Engine as _, engine::general_purpose::STANDARD}; use serde_json::{Map, Value}; @@ -75,7 +77,7 @@ impl Terminal { ("terminalId".into(), Value::from(output.call.clone())), ])); // Omit oversized metadata rather than retaining or truncating commands. - if command.len() <= crate::runlet_progress::MAX_SOURCE { + if command.len() <= MAX_COMMAND_BYTES { patch["command"] = Value::from(command); } if cwd.is_absolute() diff --git a/src/protocols/acp/tool_projection/tests.rs b/src/protocols/acp/tool_projection/tests.rs index 30e86d14..0c24a507 100644 --- a/src/protocols/acp/tool_projection/tests.rs +++ b/src/protocols/acp/tool_projection/tests.rs @@ -357,7 +357,7 @@ fn terminal_drop_marks_exit_and_bounds_binary_chunks_and_metadata() { let invocation = Invocation::start(&request, None).unwrap(); let terminal = terminal::Terminal::start( &request, - &"x".repeat(crate::runlet_progress::MAX_SOURCE + 1), + &"x".repeat(terminal::MAX_COMMAND_BYTES + 1), Path::new("/tmp"), ); terminal.output().unwrap().chunk(&vec![255; 20_000]); diff --git a/src/runlet_progress.rs b/src/runlet_progress.rs deleted file mode 100644 index 8f196993..00000000 --- a/src/runlet_progress.rs +++ /dev/null @@ -1,272 +0,0 @@ -//! Bounded, value-free adapter. Ownership is captured from the bridge envelope, -//! never from thread-local operation context. The consumer lives inside execute. -use crate::events::RuntimeEvent; -pub(crate) mod transport; -use agentkit_tool_compose::{ - BackendRun, ComposeOutcome, RunletBackend, RunletProgress, RunletProgressEnd, -}; -use serde::{Deserialize, Serialize}; -use serde_json::Value; -use std::{num::NonZeroUsize, time::Duration}; -use transport::Transport; - -pub const MAX_SOURCE: usize = 64 * 1024; -pub const MAX_NODES: usize = 256; -#[cfg(feature = "tui")] -pub const MAX_RUNS: usize = 32; -const MAX_ID: usize = 256; - -#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] -pub struct Node { - pub id: String, - pub call: bool, - pub start: usize, - pub end: usize, - pub state: State, - pub attempt: u32, -} -#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] -#[serde(rename_all = "snake_case")] -pub enum State { - Planned, - Blocked, - Ready, - WaitingForCapacity, - Running, - Succeeded, - Failed, - Cancelling, - Cancelled, - Pruned, -} -#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] -#[serde(tag = "type", rename_all = "snake_case")] -pub enum Change { - Started { - digest: String, - healed: bool, - }, - /// Every upstream sequence is carried, even relationships we do not render. - Step { - node: Option, - }, - Finished { - complete: bool, - }, -} -#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] -pub struct Progress { - pub owner: String, - pub incarnation: u64, - pub sequence: u64, - pub change: Change, -} -impl Progress { - pub fn bounded(&self) -> bool { - self.owner.len() <= MAX_ID - && self.incarnation != 0 - && match &self.change { - Change::Started { digest, .. } => { - digest.len() == 64 && digest.bytes().all(|b| b.is_ascii_hexdigit()) - } - Change::Step { node: Some(n) } => { - n.id.len() <= MAX_ID && n.start <= n.end && n.end <= MAX_SOURCE - } - _ => true, - } - } -} - -/// Pollable diagnostic adapter; each poll returns at most one bounded event. -/// Dropping an unfinished adapter emits invalidation for its captured owner. -pub(crate) struct Active { - receiver: RunletProgress, - transport: Transport, - sequence: u64, - started: bool, - ended: bool, -} -impl Active { - pub(crate) fn new(receiver: RunletProgress, transport: Transport) -> Self { - Self { - receiver, - transport, - sequence: 0, - started: false, - ended: false, - } - } - fn event(&self, change: Change) -> RuntimeEvent { - RuntimeEvent::RunletProgress { - progress: Progress { - owner: self.receiver.parent_call_id.0.clone(), - incarnation: self.receiver.incarnation, - sequence: self.sequence, - change, - }, - } - } - pub(crate) fn poll(&mut self) -> Option { - if self.ended { - return None; - } - if !self.started { - self.started = true; - return Some(self.event(Change::Started { - digest: self.receiver.source_digest.clone(), - healed: self.receiver.healed, - })); - } - let change = match self.receiver.try_recv() { - Ok(Some(event)) => { - self.sequence = event.sequence; - let node = match event.change { - runlet::ProgressChange::NodeAdded(node) - | runlet::ProgressChange::NodeUpdated(node) => Some(Node::from(node)), - runlet::ProgressChange::EdgeAdded(_) => None, - // The bridge suppresses raw runtime completion. Never treat - // it as host-authoritative completion if the contract changes. - runlet::ProgressChange::Finished(_) => { - self.ended = true; - return Some(self.event(Change::Finished { complete: false })); - } - }; - if node - .as_ref() - .is_some_and(|n| n.id.len() > MAX_ID || n.end > MAX_SOURCE) - { - self.ended = true; - Change::Finished { complete: false } - } else { - Change::Step { node } - } - } - Ok(None) => return None, - Err(end) => { - self.ended = true; - Change::Finished { - complete: matches!( - end, - RunletProgressEnd::Succeeded | RunletProgressEnd::Failed - ), - } - } - }; - Some(self.event(change)) - } -} -impl From for Node { - fn from(node: runlet::ProgressNode) -> Self { - Self { - id: node.id, - call: matches!(node.kind, runlet::NodeKind::Call), - start: node.span.start, - end: node.span.end, - state: node.state.into(), - attempt: node.attempt, - } - } -} -impl From for State { - fn from(state: runlet::ProgressState) -> Self { - match state { - runlet::ProgressState::Planned => Self::Planned, - runlet::ProgressState::Blocked => Self::Blocked, - runlet::ProgressState::Ready => Self::Ready, - runlet::ProgressState::WaitingForCapacity => Self::WaitingForCapacity, - runlet::ProgressState::Running => Self::Running, - runlet::ProgressState::Succeeded => Self::Succeeded, - runlet::ProgressState::Failed => Self::Failed, - runlet::ProgressState::Cancelling => Self::Cancelling, - runlet::ProgressState::Cancelled => Self::Cancelled, - runlet::ProgressState::Pruned => Self::Pruned, - } - } -} -impl Drop for Active { - fn drop(&mut self) { - if self.started - && !self.ended - && let RuntimeEvent::RunletProgress { progress } = - self.event(Change::Finished { complete: false }) - { - self.transport.publish(progress); - } - } -} - -pub async fn execute(run: BackendRun) -> Result { - let Some(transport) = transport::global() else { - use agentkit_tool_compose::ComposeBackend; - return RunletBackend.execute(run).await; - }; - execute_observed(run, transport).await -} - -pub(crate) async fn execute_observed( - run: BackendRun, - transport: &Transport, -) -> Result { - let (tx, mut rx) = tokio::sync::mpsc::channel(2); - let execute = RunletBackend.execute_with_progress( - run, - tx, - NonZeroUsize::new(1024).unwrap_or(NonZeroUsize::MIN), - ); - let consume = async move { - let mut active: Vec = Vec::new(); - let mut closed = false; - let mut tick = tokio::time::interval(Duration::from_millis(25)); - loop { - if closed { - tick.tick().await; - } else { - use futures_util::future::{Either, select}; - match select(std::pin::pin!(rx.recv()), std::pin::pin!(tick.tick())).await { - Either::Left((envelope, _)) => { - match envelope { - Some(receiver) => { - if active.len() == 2 { - active.remove(0); - } - active.push(Active::new(receiver, transport.clone())); - } - None => closed = true, - } - continue; - } - Either::Right(_) => {} - } - } - for run in &mut active { - for _ in 0..512 { - let Some(event) = run.poll() else { - break; - }; - if let RuntimeEvent::RunletProgress { progress } = event { - transport.publish(progress); - } - } - } - active.retain(|run| !run.ended); - if closed && active.is_empty() { - break; - } - } - }; - let (result, ()) = futures_util::future::join(execute, consume).await; - result -} - -/// Retain bounded source bytes even when an ACP caller submits oversized input. -#[cfg(feature = "tui")] -pub(crate) fn bounded_source(source: String) -> String { - if source.len() <= MAX_SOURCE { - return source; - } - let mut end = MAX_SOURCE - 32; - while !source.is_char_boundary(end) { - end -= 1; - } - format!("{}\n… source truncated", &source[..end]) -} diff --git a/src/runtime.rs b/src/runtime.rs index cbb7221e..6ff587ec 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -2653,11 +2653,7 @@ impl ComposeBackend for HiddenRunletBackend { async fn execute(&self, mut run: BackendRun) -> Result { run.visible_specs = self.specs(); - if crate::events::enabled() { - crate::runlet_progress::execute(run).await - } else { - RunletBackend.execute(run).await - } + RunletBackend.execute(run).await } } diff --git a/src/tools/observed.rs b/src/tools/observed.rs index ac8aeebd..4b5b358c 100644 --- a/src/tools/observed.rs +++ b/src/tools/observed.rs @@ -1,24 +1,14 @@ -//! Lifecycle reporting for the hidden tools behind `compose`. -//! -//! The wrapper is transparent to the model and to compose: it forwards the -//! spec, permission requests, and invocation untouched, and only publishes -//! start/finish events on the runtime side channel (see [`crate::events`]) so -//! a client can draw what a Runlet program is doing while it runs. +//! Transparent hidden-tool wrapper with canonical ACP lifecycle projection. -use std::{sync::Arc, time::Instant}; +use std::sync::Arc; -use agentkit_core::ToolOutput; use agentkit_tools_core::{ PermissionRequest, Tool, ToolContext, ToolError, ToolExecutionOutcome, ToolRequest, ToolResult, ToolSpec, }; use async_trait::async_trait; -use serde_json::Value; - -use crate::events::{self, RuntimeEvent, summarize_input, summarize_output}; - -/// Wraps a tool so its calls appear on the runtime side channel. +/// Wraps a tool so its calls appear in canonical ACP notifications. pub struct Observed(T, Option); impl Observed { @@ -95,16 +85,12 @@ impl Tool for Observed { request: ToolRequest, context: &mut ToolContext<'_>, ) -> Result { - let display = DisplayInvocation::start(&request); let projection = crate::protocols::acp::tool_projection::Invocation::start(&request, self.1.as_deref()); let outcome = self.0.invoke(request, context).await; if let Some(projection) = projection { projection.finish(outcome.as_ref().is_ok_and(|result| !result.result.is_error)); } - if let Some(display) = display { - display.finish(outcome.as_ref()); - } outcome } @@ -113,79 +99,16 @@ impl Tool for Observed { request: ToolRequest, context: &mut ToolContext<'_>, ) -> ToolExecutionOutcome { - let display = DisplayInvocation::start(&request); let projection = crate::protocols::acp::tool_projection::Invocation::start(&request, self.1.as_deref()); let outcome = self.0.invoke_outcome(request, context).await; if let Some(projection) = projection { projection.finish(matches!(&outcome, ToolExecutionOutcome::Completed(result) if !result.result.is_error)); } - if let Some(display) = display { - match &outcome { - ToolExecutionOutcome::Completed(result) => display.finish(Ok(result)), - ToolExecutionOutcome::Failed(error) - | ToolExecutionOutcome::FailedBeforeInvocation(error) => display.finish(Err(error)), - // An approval interruption is not a completed invocation. - ToolExecutionOutcome::Interrupted(_) => {} - } - } outcome } } -struct DisplayInvocation { - call: String, - tool: String, - started: Instant, -} - -impl DisplayInvocation { - fn start(request: &ToolRequest) -> Option { - if !events::enabled() { - return None; - } - let call = request.call_id.0.clone(); - let tool = request.tool_name.0.to_string(); - events::emit(&RuntimeEvent::ChildStarted { - call: call.clone(), - tool: tool.clone(), - summary: summarize_input(&request.input), - at: events::now_millis(), - }); - Some(Self { - call, - tool, - started: Instant::now(), - }) - } - - fn finish(self, result: Result<&ToolResult, &ToolError>) { - let (ok, summary) = match result { - Ok(result) => ( - !result.result.is_error, - summarize_output(&output_value(&result.result.output)), - ), - Err(error) => (false, summarize_output(&Value::from(error.to_string()))), - }; - events::emit(&RuntimeEvent::ChildFinished { - call: self.call, - tool: self.tool, - ok, - summary, - millis: u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX), - }); - } -} - -fn output_value(output: &ToolOutput) -> Value { - match output { - ToolOutput::Text(text) => Value::from(text.clone()), - ToolOutput::Structured(value) => value.clone(), - ToolOutput::Parts(parts) => Value::from(format!("{} parts", parts.len())), - ToolOutput::Files(files) => Value::from(format!("{} files", files.len())), - } -} - #[cfg(test)] #[allow( clippy::unwrap_used, @@ -197,28 +120,13 @@ fn output_value(output: &ToolOutput) -> Value { )] mod tests { use super::*; - use agentkit_core::{MetadataMap, SessionId, ToolCallId, ToolResultPart, TurnId}; + use agentkit_core::{MetadataMap, SessionId, ToolCallId, ToolOutput, ToolResultPart, TurnId}; use agentkit_tools_core::{ AllowAllPermissions, ApprovalReason, ApprovalRequest, OwnedToolContext, ToolInterruption, ToolName, }; use serde_json::json; - #[test] - fn output_summaries_preserve_strings_values_and_counts() { - assert_eq!( - output_value(&ToolOutput::Text("quoted \"text\"\n".into())), - json!("quoted \"text\"\n"), - ); - let structured = json!({"number": 42, "null": null, "array": [true, "text"]}); - assert_eq!( - output_value(&ToolOutput::structured(structured.clone())), - structured - ); - assert_eq!(output_value(&ToolOutput::Parts(vec![])), json!("0 parts")); - assert_eq!(output_value(&ToolOutput::Files(vec![])), json!("0 files")); - } - #[derive(Clone, Copy, Debug)] enum Mode { Completed, diff --git a/src/tools/subagent/tests.rs b/src/tools/subagent/tests.rs index ed6b8c38..aed917a3 100644 --- a/src/tools/subagent/tests.rs +++ b/src/tools/subagent/tests.rs @@ -1110,7 +1110,6 @@ mod lifecycle_events { (SubagentStatus::Removed, Some(GenerationOutcome::Failed)), ] ); - assert!(emitted.iter().all(|event| event.parent_call().is_none())); let generations = emitted .iter() .filter_map(|event| match event { diff --git a/src/tui/app.rs b/src/tui/app.rs index cb0e1d09..41bea611 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -448,29 +448,6 @@ pub enum Action { Quit, } -/// One nested tool dispatch inside a compose run. -pub struct Child { - pub call: String, - pub tool: String, - pub summary: String, - pub result: String, - pub started: Instant, - pub millis: Option, - pub ok: bool, -} - -impl Child { - pub fn running(&self) -> bool { - self.millis.is_none() - } - - pub fn elapsed(&self) -> u64 { - self.millis.unwrap_or_else(|| { - u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX) - }) - } -} - #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum ComposeView { #[default] @@ -491,8 +468,6 @@ pub struct ToolCall { pub finished: Option, /// Runlet source shown inline while this compose call is running. pub script: String, - pub(super) progress: Box, - pub children: Vec, /// Raw tool output, kept whole but folded away until asked for. pub output: Vec, /// Typed tool-result images sharing the user-image retention and decode budgets. @@ -537,54 +512,11 @@ impl ToolCall { } fn finalize_terminal_state(&mut self) { - self.progress.parent_finished(); if self.is_compose() && !self.expansion_explicit { self.expanded = false; self.compose_view = ComposeView::Output; } self.finished = Some(Instant::now()); - self.finish_running_children(); - } - - /// Records a dispatch whose parent call ID matches this tool call. - /// - /// Events do not carry source locations. Keep lifecycle state correlated - /// by call ID without guessing which script expression owns a dispatch. - pub fn attach(&mut self, call: String, tool: String, summary: String) { - self.children.push(Child { - call, - tool, - summary, - result: String::new(), - started: Instant::now(), - millis: None, - ok: true, - }); - } - - pub fn finish_child(&mut self, call: &str, ok: bool, summary: String, millis: u64) { - if let Some(child) = self - .children - .iter_mut() - .rev() - .find(|child| child.running() && child.call == call) - { - child.millis = Some(millis); - child.ok = ok; - child.result = summary; - } - } - - pub fn running_children(&self) -> usize { - self.children.iter().filter(|child| child.running()).count() - } - - fn finish_running_children(&mut self) { - let ok = self.status != ToolCallStatus::Failed; - for child in self.children.iter_mut().filter(|child| child.running()) { - child.millis = Some(child.elapsed()); - child.ok = ok; - } } } @@ -809,8 +741,8 @@ pub struct AgentCounts { } pub struct App { - progress_last_frame: Option, - progress_unavailable: bool, + runtime_last_frame: Option, + runtime_status_unavailable: bool, pub root: PathBuf, pub provider: String, pub model: String, @@ -1075,8 +1007,8 @@ fn agent_status_rank(status: SubagentStatus) -> u8 { impl App { pub fn new(root: PathBuf, provider: String, model: String, a2a: String) -> Self { Self { - progress_last_frame: None, - progress_unavailable: false, + runtime_last_frame: None, + runtime_status_unavailable: false, root, provider, model, @@ -1277,7 +1209,7 @@ impl App { fn block_is_dynamic(block: &Block) -> bool { match block { Block::Thought { millis, .. } => millis.is_none(), - Block::Tool(call) => call.running() || call.running_children() > 0, + Block::Tool(call) => call.running(), _ => false, } } @@ -1391,7 +1323,7 @@ impl App { /// Whether periodic polling must advance animations or expire a runtime lease. pub fn needs_redraw_tick(&self) -> bool { - (!self.progress_unavailable && self.progress_last_frame.is_some()) + (!self.runtime_status_unavailable && self.runtime_last_frame.is_some()) || self.working() || !self.transcript_dynamic.is_empty() || self.toast.is_some() @@ -1409,7 +1341,7 @@ impl App { /// Advances animations and removes expired transient state. pub fn tick(&mut self) { - self.progress_tick_at(Instant::now()); + self.runtime_tick_at(Instant::now()); self.tick_at(crate::events::now_millis()); } @@ -2176,9 +2108,7 @@ impl App { status: ToolCallStatus::Pending, started: Instant::now(), finished: None, - script: crate::runlet_progress::bounded_source(script.unwrap_or_default()), - progress: Box::default(), - children: Vec::new(), + script: super::source::bounded_source(script.unwrap_or_default()), output: Vec::new(), images: Vec::new(), intent: None, @@ -2269,10 +2199,7 @@ impl App { call.intent = intent; } if let Some(script) = script { - let script = crate::runlet_progress::bounded_source(script); - if call.script != script { - call.progress.invalidate(); - } + let script = super::source::bounded_source(script); call.script = script; } if let Some(output) = output { @@ -2356,10 +2283,10 @@ impl App { } fn invalidate_runtime_status(&mut self) { - if self.progress_unavailable { + if self.runtime_status_unavailable { return; } - self.progress_unavailable = true; + self.runtime_status_unavailable = true; // All these fields depend on the same lossy side channel. Absence is // unknown, not idle/success/healthy; the UI exposes unavailability. self.agent_versions.clear(); @@ -2371,35 +2298,27 @@ impl App { self.compacting = false; self.storage_pending = false; self.storage_exhausted = false; - for index in 0..self.blocks.len() { - if let Block::Tool(call) = &mut self.blocks[index] { - call.progress.invalidate(); - call.children.clear(); - } - self.mark_block_dirty(index); - self.reclassify_dynamic(index); - } } pub(super) fn runtime_unavailable(&self) -> bool { - self.progress_unavailable + self.runtime_status_unavailable } /// Monotonic transport deadline, also checked before accepting new traffic. - pub(super) fn progress_tick_at(&mut self, now: Instant) { - if !self.progress_unavailable - && self.progress_last_frame.is_some_and(|last| { - now.saturating_duration_since(last) >= crate::runlet_progress::transport::LEASE + pub(super) fn runtime_tick_at(&mut self, now: Instant) { + if !self.runtime_status_unavailable + && self.runtime_last_frame.is_some_and(|last| { + now.saturating_duration_since(last) >= crate::diagnostic_transport::LEASE }) { self.disable_runtime(); } } - fn progress_activity(&mut self) { + fn runtime_activity(&mut self) { let now = Instant::now(); - self.progress_tick_at(now); - if !self.progress_unavailable { - self.progress_last_frame = Some(now); + self.runtime_tick_at(now); + if !self.runtime_status_unavailable { + self.runtime_last_frame = Some(now); } } @@ -2409,17 +2328,17 @@ impl App { fn apply_runtime_at(&mut self, event: RuntimeEvent, now_unix_ms: u64) { // Check expiry before any frame can refresh the lease or revive a - // lifecycle map. Loss applies to all runtime events, not only progress. - self.progress_activity(); + // lifecycle map. Loss applies to every diagnostic lifecycle event. + self.runtime_activity(); if let RuntimeEvent::RunletTransport { available } = event { if available { - if self.progress_unavailable { + if self.runtime_status_unavailable { // A heartbeat restores transport, not the observations lost - // during the gap. Keep cleared state and progress tombstones. - self.note("Runtime status resumed; earlier agent, child, compaction and storage state remains unknown"); + // during the gap. Keep cleared state. + self.note("Runtime status resumed; earlier agent, compaction and storage state remains unknown"); } - self.progress_unavailable = false; - self.progress_last_frame = Some(Instant::now()); + self.runtime_status_unavailable = false; + self.runtime_last_frame = Some(Instant::now()); } else { self.disable_runtime(); } @@ -2434,18 +2353,15 @@ impl App { if self.runtime_unavailable() { return; } - let parent = event.parent_call().map(str::to_string); - let owner_id = match event { - RuntimeEvent::RunletTransport { .. } | RuntimeEvent::SessionStarted { .. } => return, + match event { + RuntimeEvent::RunletTransport { .. } | RuntimeEvent::SessionStarted { .. } => (), RuntimeEvent::StorageStatus { pending, exhausted } => { self.storage_pending = pending; self.storage_exhausted = exhausted; - return; } - _ if self.session_id.is_some() && self.runtime_session_id != self.session_id => return, + _ if self.session_id.is_some() && self.runtime_session_id != self.session_id => (), RuntimeEvent::CompactionStarted { .. } => { self.compacting = true; - return; } RuntimeEvent::CompactionFinished { ok, compacted, .. } => { self.compacting = false; @@ -2453,7 +2369,6 @@ impl App { self.usage = None; self.note("context compacted"); } - return; } RuntimeEvent::SubagentStateChanged { id, @@ -2541,7 +2456,6 @@ impl App { ); } self.clamp_agents_scroll(); - return; } RuntimeEvent::SubagentActivity { id, activity } => { if let Some(row) = self.agents.get_mut(&id) @@ -2553,7 +2467,6 @@ impl App { { row.activity.apply(activity); } - return; } RuntimeEvent::SubagentUsage { id, @@ -2570,7 +2483,6 @@ impl App { if let Some(row) = self.agents.get_mut(&id) { row.usage = Some(ContextUsage { used, size }); } - return; } RuntimeEvent::SubagentDescendantsRemoved { ancestor_id } => { let mut removed = HashSet::new(); @@ -2593,88 +2505,10 @@ impl App { self.cleaned_agent_ancestors.insert(ancestor_id); self.cleaned_agent_ids.extend(removed); self.clamp_agents_scroll(); - return; - } - RuntimeEvent::RunletProgress { progress } => { - if self.progress_unavailable { - return; - } - let Some(owner) = self.blocks.iter().rev().find_map(|b| match b { - Block::Tool(c) if c.id == progress.owner && c.is_compose() => Some(c), - _ => None, - }) else { - return; - }; - let needs_slot = owner.running() && !owner.progress.retained(); - // Retain at most MAX_RUNS maps across the transcript. Eviction - // preserves incarnation tombstones so stale replay cannot revive it. - let retained = self - .blocks - .iter() - .filter(|b| matches!(b, Block::Tool(c) if c.progress.retained())) - .count(); - if needs_slot - && retained >= crate::runlet_progress::MAX_RUNS - && let Some(id) = self.blocks.iter().find_map(|b| match b { - Block::Tool(c) if c.id != progress.owner && c.progress.retained() => { - Some(c.id.clone()) - } - _ => None, - }) - && let Some(call) = self.call_mut(&id) - { - call.progress.invalidate(); - } - let Some(call) = self.runtime_call_mut(parent.as_deref()) else { - return; - }; - if !call.is_compose() { - return; - } - if call.running() { - call.progress.apply(&progress, &call.script); - } else { - call.progress.apply_terminal(&progress, &call.script); - } - call.id.clone() } - RuntimeEvent::ChildStarted { - call: child_call, - tool, - summary, - .. - } => { - let Some(call) = self.runtime_call_mut(parent.as_deref()) else { - return; - }; - call.attach(child_call, tool, summary); - call.id.clone() - } - RuntimeEvent::ChildFinished { - call: child_call, - ok, - summary, - millis, - .. - } => { - let Some(call) = self.runtime_call_mut(parent.as_deref()) else { - return; - }; - call.finish_child(&child_call, ok, summary, millis); - call.id.clone() - } - }; - if let Some(index) = self.call_index(&owner_id) { - self.reclassify_dynamic(index); } } - fn runtime_call_mut(&mut self, parent: Option<&str>) -> Option<&mut ToolCall> { - // Descendant events can name an unseen compose call. Without its - // owner, the event cannot update another call's counts. - self.call_mut(parent?) - } - fn call_index(&self, id: &str) -> Option { self.tool_indices.get(id).copied() } @@ -5327,15 +5161,6 @@ mod tests { }); } - fn child(call: &str, tool: &str) -> RuntimeEvent { - RuntimeEvent::ChildStarted { - call: call.into(), - tool: tool.into(), - summary: "ls".into(), - at: 0, - } - } - #[test] fn surfaces_compaction_lifecycle_without_a_tool_call() { let mut app = app(); @@ -5437,122 +5262,6 @@ mod tests { assert!(!app.compacting); } - #[test] - fn attributes_nested_calls_to_the_owning_tool_call() { - let mut app = app(); - compose(&mut app, "a = shell({ command: \"ls\" })\nreturn a"); - app.apply(Update::Runtime(child("call-1:compose:abc", "shell"))); - let Some(Block::Tool(call)) = app.blocks.last() else { - panic!("expected a tool block"); - }; - assert_eq!(call.children.len(), 1); - assert_eq!(call.children[0].call, "call-1:compose:abc"); - assert_eq!(call.running_children(), 1); - } - - #[test] - fn unknown_child_owner_cannot_change_a_visible_child_lifecycle() { - let mut app = app(); - // No visible owner is a valid event-stream boundary, not a tool variant. - app.apply(Update::Runtime(child("unseen:compose:before", "shell"))); - assert!(app.blocks.is_empty()); - compose(&mut app, "return shell({ command: \"ls\" })"); - app.apply(Update::Runtime(child("call-1:compose:known", "shell"))); - app.note("intervening non-tool block"); - app.apply(Update::Runtime(child("unseen:compose:abc", "shell"))); - app.apply(Update::Runtime(RuntimeEvent::ChildFinished { - call: "unseen:compose:abc".into(), - tool: "shell".into(), - ok: true, - summary: "done".into(), - millis: 12, - })); - let Block::Tool(call) = &app.blocks[0] else { - panic!("expected the visible compose call"); - }; - assert_eq!(call.children.len(), 1); - assert_eq!(call.running_children(), 1); - assert_eq!(call.children[0].call, "call-1:compose:known"); - assert!(call.children[0].running()); - } - - #[test] - fn terminal_parent_finishes_children_missing_completion_events() { - let mut app = app(); - compose(&mut app, "a = shell({ command: \"sleep 60\" })\nreturn a"); - app.apply(Update::Runtime(child("call-1:compose:shell", "shell"))); - - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: Some(ToolCallStatus::Failed), - script: None, - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - - let Some(Block::Tool(call)) = app.blocks.last() else { - panic!("expected a tool block"); - }; - assert_eq!(call.status, ToolCallStatus::Failed); - assert_eq!(call.running_children(), 0); - assert!(call.children[0].millis.is_some()); - assert!(!call.children[0].ok); - } - - #[test] - fn repeated_dispatches_keep_distinct_call_id_lifecycles() { - let mut app = app(); - compose( - &mut app, - "a = shell({ command: \"one\" })\nb = shell({ command: \"two\" })\nreturn [a, b]", - ); - app.apply(Update::Runtime(child("call-1:compose:a", "shell"))); - app.apply(Update::Runtime(child("call-1:compose:b", "shell"))); - let Some(Block::Tool(call)) = app.blocks.last() else { - panic!("expected a tool block"); - }; - assert_eq!(call.running_children(), 2); - app.apply(Update::Runtime(RuntimeEvent::ChildFinished { - call: "call-1:compose:b".into(), - tool: "shell".into(), - ok: true, - summary: "two".into(), - millis: 10, - })); - let Some(Block::Tool(call)) = app.blocks.last() else { - panic!("expected a tool block"); - }; - assert_eq!(call.running_children(), 1); - assert!(call.children[0].running()); - assert!(!call.children[1].running()); - } - - #[test] - fn unknown_parent_events_do_not_attach_to_the_visible_call() { - let mut app = app(); - compose(&mut app, "a = subagent({prompt: input.prompt})\nreturn a"); - for id in ["descendant:compose:one", "unscoped"] { - app.apply(Update::Runtime(child(id, "subagent"))); - app.apply(Update::Runtime(RuntimeEvent::ChildFinished { - call: id.into(), - tool: "subagent".into(), - ok: true, - summary: "done".into(), - millis: 10, - })); - } - let Some(Block::Tool(call)) = app.blocks.last() else { - panic!("expected a tool block"); - }; - assert!(call.children.is_empty()); - assert_eq!(call.running_children(), 0); - } - #[test] fn closes_running_calls_when_the_turn_ends() { let mut app = app(); @@ -8163,7 +7872,7 @@ mod tests { // Advance the lease age without sleeping, then use the same scheduling // predicate and tick entry point as both event loops. No new traffic. - app.progress_last_frame = Some(Instant::now() - crate::runlet_progress::transport::LEASE); + app.runtime_last_frame = Some(Instant::now() - crate::diagnostic_transport::LEASE); assert!(app.needs_redraw_tick()); if app.needs_redraw_tick() { app.tick(); @@ -8181,7 +7890,7 @@ mod tests { pending: true, exhausted: true, })); - app.progress_last_frame = Some(Instant::now() - crate::runlet_progress::transport::LEASE); + app.runtime_last_frame = Some(Instant::now() - crate::diagnostic_transport::LEASE); app.apply(Update::Runtime(RuntimeEvent::RunletTransport { available: true, })); @@ -8200,7 +7909,7 @@ mod tests { app.tick(); assert!(!app.runtime_unavailable()); - app.progress_last_frame = Some(Instant::now() - crate::runlet_progress::transport::LEASE); + app.runtime_last_frame = Some(Instant::now() - crate::diagnostic_transport::LEASE); app.tick(); assert!(app.runtime_unavailable()); } diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 3fd41f8d..26c862a8 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -15,7 +15,7 @@ mod image; #[cfg(all(test, unix))] mod keyboard_tests; mod markdown; -mod progress; +mod source; mod theme; mod ui; mod wrap; @@ -67,7 +67,7 @@ use wire::{ }; use crate::{ - events::{self, EVENTS_ENV, RuntimeEvent}, + events::{self, EVENTS_ENV}, protocols::acp::{ FileSearchRequest, MODEL_CONFIG_ID, REASONING_EFFORT_CONFIG_ID, model_switch, }, @@ -2759,11 +2759,6 @@ impl std::error::Error for Failure {} /// still drive runtime availability, subagent status, storage, and failure reports. fn stderr_update(line: String) -> Option { match events::parse(&line) { - Some( - RuntimeEvent::ChildStarted { .. } - | RuntimeEvent::ChildFinished { .. } - | RuntimeEvent::RunletProgress { .. }, - ) => None, Some(event) => Some(Update::Runtime(event)), None if line.starts_with("A2A listening on ") => Some(Update::A2aAddress( line.trim_start_matches("A2A listening on ").to_string(), @@ -4038,7 +4033,7 @@ mod tests { let heartbeat = || line(RuntimeEvent::RunletTransport { available: true }); app.apply(stderr_update(heartbeat()).unwrap()); assert!(!app.runtime_unavailable()); - app.progress_tick_at(std::time::Instant::now() + crate::runlet_progress::transport::LEASE); + app.runtime_tick_at(std::time::Instant::now() + crate::diagnostic_transport::LEASE); assert!(app.runtime_unavailable()); app.apply(stderr_update(heartbeat()).unwrap()); assert!( @@ -4060,31 +4055,6 @@ mod tests { matches!(stderr_update(line(event.clone())), Some(Update::Runtime(actual)) if actual == event) ); } - for event in [ - RuntimeEvent::ChildStarted { - call: "child".into(), - tool: "shell".into(), - summary: "running".into(), - at: 1, - }, - RuntimeEvent::ChildFinished { - call: "child".into(), - tool: "shell".into(), - ok: true, - summary: "done".into(), - millis: 1, - }, - RuntimeEvent::RunletProgress { - progress: crate::runlet_progress::Progress { - owner: "root".into(), - incarnation: 1, - sequence: 1, - change: crate::runlet_progress::Change::Finished { complete: true }, - }, - }, - ] { - assert!(stderr_update(line(event)).is_none()); - } assert!( matches!(stderr_update("ordinary diagnostic".into()), Some(Update::Log(line)) if line == "ordinary diagnostic") ); diff --git a/src/tui/progress.rs b/src/tui/progress.rs deleted file mode 100644 index 5c341a19..00000000 --- a/src/tui/progress.rs +++ /dev/null @@ -1,223 +0,0 @@ -//! Single-writer transient source observations. Missing information is unknown. -use crate::runlet_progress::{Change, MAX_NODES, MAX_SOURCE, Node, Progress, State}; -use sha2::{Digest, Sha256}; -use std::collections::{BTreeMap, HashMap}; - -#[derive(Default)] -pub(super) struct ScriptProgress { - incarnation: u64, - sequence: u64, - digest: String, - started: Option, - last: Option<(u64, Change)>, - terminal: Option<(u64, bool)>, - known: bool, - ended: bool, - nodes: HashMap, -} -impl ScriptProgress { - pub fn parent_finished(&mut self) { - if !self.ended { - self.invalidate(); - } - } - pub fn invalidate(&mut self) { - self.known = false; - self.nodes = HashMap::new(); - } - pub fn retained(&self) -> bool { - !self.nodes.is_empty() - } - pub fn apply_terminal(&mut self, event: &Progress, source: &str) { - if !self.ended || event.incarnation != self.incarnation { - self.invalidate(); - return; - } - self.apply(event, source); - } - pub fn apply(&mut self, event: &Progress, source: &str) { - if !event.bounded() { - self.invalidate(); - return; - } - if event.incarnation < self.incarnation { - self.invalidate(); - return; - } - if let Change::Started { digest, healed } = &event.change { - if event.incarnation == self.incarnation { - if event.sequence != 0 || self.started.as_ref() != Some(&event.change) { - self.invalidate(); - } - return; - } - self.invalidate(); - self.incarnation = event.incarnation; - self.sequence = 0; - self.started = Some(event.change.clone()); - self.last = None; - self.terminal = None; - self.ended = false; - self.digest.clone_from(digest); - self.known = event.sequence == 0 - && !healed - && source.len() <= MAX_SOURCE - && source_digest(source) == *digest; - return; - } - if event.incarnation != self.incarnation { - self.invalidate(); - self.incarnation = event.incarnation; - return; - } - if matches!(event.change, Change::Finished { complete: false }) { - self.invalidate(); - self.ended = true; - return; - } - if !self.known { - return; - } - if self.ended { - let duplicate = match &event.change { - Change::Step { .. } => { - self.last.as_ref() == Some(&(event.sequence, event.change.clone())) - } - Change::Finished { complete } => self.terminal == Some((event.sequence, *complete)), - Change::Started { .. } => false, - }; - if !duplicate { - self.invalidate(); - } - return; - } - if source.len() > MAX_SOURCE { - self.invalidate(); - return; - } - match &event.change { - Change::Step { node } => { - // Duplicate delivery is idempotent. Any other reordering or gap - // permanently invalidates this incarnation (there is no replay). - if event.sequence == self.sequence { - if self.last.as_ref() != Some(&(event.sequence, event.change.clone())) { - self.invalidate(); - } - return; - } - if event.sequence != self.sequence.saturating_add(1) { - self.invalidate(); - return; - } - self.sequence = event.sequence; - self.last = Some((event.sequence, event.change.clone())); - if let Some(node) = node { - if let Some(previous) = self.nodes.get(&node.id) - && previous.call - && matches!( - previous.state, - State::Succeeded | State::Failed | State::Cancelled | State::Pruned - ) - && previous.state != node.state - { - self.invalidate(); - return; - } - if source.get(node.start..node.end).is_none() { - self.invalidate(); - return; - } - if !self.nodes.contains_key(&node.id) && self.nodes.len() == MAX_NODES { - self.invalidate(); - return; - } - self.nodes.insert(node.id.clone(), node.clone()); - } - } - Change::Finished { complete } => { - self.ended = true; - self.terminal = Some((event.sequence, *complete)); - if !complete || event.sequence != self.sequence { - self.invalidate(); - } - } - Change::Started { .. } => {} - } - } - /// Annotations keyed by one-based source start line. Columns count Unicode - /// scalar values, not UTF-8 bytes; ranges have an exclusive end position. - pub fn labels(&self, source: &str) -> BTreeMap> { - if !self.known || source.len() > MAX_SOURCE || source_digest(source) != self.digest { - return BTreeMap::new(); - } - let mut groups: BTreeMap<(usize, usize), BTreeMap> = BTreeMap::new(); - for node in self.nodes.values() { - // Structural activity cannot establish that a call has been spawned. - if !node.call { - continue; - } - if self.ended - && !matches!( - node.state, - State::Succeeded | State::Failed | State::Cancelled | State::Pruned - ) - { - continue; - } - *groups - .entry((node.start, node.end)) - .or_default() - .entry(node.state) - .or_default() += 1; - } - let mut labels: BTreeMap> = BTreeMap::new(); - for ((start, end), states) in groups { - let (line, column) = source_position(source, start); - let (end_line, end_column) = source_position(source, end); - let counts = states - .into_iter() - .map(|(state, count)| { - let label = match state { - State::Planned => "planned", - State::Blocked => "blocked", - State::Ready => "ready", - State::WaitingForCapacity => "waiting for capacity", - State::Running => "running", - State::Succeeded => "succeeded", - State::Failed => "failed", - State::Cancelling => "cancelling", - State::Cancelled => "cancelled", - State::Pruned => "pruned", - }; - format!("{count} {label}") - }) - .collect::>() - .join(", "); - labels.entry(line).or_default().push(format!( - "# call @L{line}:C{column}..L{end_line}:C{end_column}: {counts}" - )); - } - labels - } -} -pub(super) fn source_digest(source: &str) -> String { - Sha256::digest(source.as_bytes()) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect() -} - -fn source_position(source: &str, byte: usize) -> (usize, usize) { - // Accepted nodes have already been checked against these exact source bytes. - let prefix = &source[..byte]; - ( - prefix.bytes().filter(|b| *b == b'\n').count() + 1, - prefix - .rsplit('\n') - .next() - .unwrap_or_default() - .chars() - .count() - + 1, - ) -} diff --git a/src/tui/progress_tests.rs b/src/tui/progress_tests.rs deleted file mode 100644 index c07a5f21..00000000 --- a/src/tui/progress_tests.rs +++ /dev/null @@ -1,1181 +0,0 @@ -// Included only in ui's test module; exercises the real wire/app/render APIs. -use crate::runlet_progress::{ - Change as ProgressChange, Node as ProgressNode, Progress, State as ProgressState, -}; - -fn progress_wire(app: &mut App, event: RuntimeEvent) { - let mut bytes = Vec::new(); - crate::events::test_support::write_event(&mut bytes, &event); - let line = String::from_utf8(bytes).unwrap(); - app.apply(Update::Runtime( - crate::events::parse(line.trim_end()).unwrap(), - )); -} -fn source_event( - owner: &str, - incarnation: u64, - sequence: u64, - change: ProgressChange, -) -> RuntimeEvent { - RuntimeEvent::RunletProgress { - progress: Progress { - owner: owner.into(), - incarnation, - sequence, - change, - }, - } -} -fn progress_start(app: &mut App, source: &str, incarnation: u64, healed: bool) { - progress_wire( - app, - source_event( - "call-1", - incarnation, - 0, - ProgressChange::Started { - digest: crate::tui::progress::source_digest(source), - healed, - }, - ), - ); -} -fn progress_node(id: &str, state: ProgressState, call: bool) -> ProgressNode { - ProgressNode { - id: id.into(), - call, - start: 8, - end: 34, - state, - attempt: 0, - } -} -fn progress_step(app: &mut App, incarnation: u64, sequence: u64, node: ProgressNode) { - progress_wire( - app, - source_event( - "call-1", - incarnation, - sequence, - ProgressChange::Step { node: Some(node) }, - ), - ); -} - -#[test] -fn authoritative_progress_distinct_iterations_and_structural_evaluation() { - let mut app = sample(); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Succeeded, true), - ); - let mut retry = progress_node("b", ProgressState::Running, true); - retry.attempt = 1; - progress_step(&mut app, 1, 2, retry.clone()); - progress_step(&mut app, 1, 2, retry); // duplicate - progress_step( - &mut app, - 1, - 3, - progress_node("scope", ProgressState::Running, false), - ); - let frame = render(&mut app, 140, 50); - assert!(frame.contains("1 running"), "{frame}"); - assert!(frame.contains("1 succeeded"), "{frame}"); - assert!(!frame.contains("evaluating"), "{frame}"); - assert!(!frame.contains("resolved")); - // Unknown descendant cannot mutate this call's observations or agent roster. - progress_wire( - &mut app, - source_event( - "descendant", - 99, - 0, - ProgressChange::Started { - digest: crate::tui::progress::source_digest(SCRIPT), - healed: false, - }, - ), - ); - assert!(render(&mut app, 140, 50).contains("1 running")); -} - -#[test] -fn authoritative_progress_gap_reorder_stale_and_replay_are_conservative() { - for sequence in [0, 3] { - let mut app = sample(); - progress_start(&mut app, SCRIPT, 2, false); - progress_step( - &mut app, - 2, - 1, - progress_node("a", ProgressState::Running, true), - ); - progress_step( - &mut app, - 2, - sequence, - progress_node("a", ProgressState::Succeeded, true), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); - progress_start(&mut app, SCRIPT, 3, false); - progress_step( - &mut app, - 3, - 1, - progress_node("new", ProgressState::Running, true), - ); - assert!(render(&mut app, 140, 50).contains("1 running")); - progress_step( - &mut app, - 2, - 2, - progress_node("old", ProgressState::Succeeded, true), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); - } -} - -#[test] -fn authoritative_progress_healed_mismatch_and_unicode_spans_stay_neutral() { - for (source, healed) in [(SCRIPT, true), ("return 1", false)] { - let mut app = sample(); - progress_start(&mut app, source, 1, healed); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Running, true), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); - } - let mut app = sample(); - let script = "return \"🦀\""; - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(script.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - progress_start(&mut app, script, 1, false); - let mut node = progress_node("bad", ProgressState::Running, true); - node.end = 9; - progress_step(&mut app, 1, 1, node); - assert!(!render(&mut app, 140, 50).contains("# call @")); -} - -#[test] -fn authoritative_progress_incomplete_and_parent_completion_never_fabricate_success() { - let mut app = sample(); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Running, true), - ); - progress_wire( - &mut app, - source_event("call-1", 1, 1, ProgressChange::Finished { complete: false }), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); - progress_start(&mut app, SCRIPT, 2, false); - progress_step( - &mut app, - 2, - 1, - progress_node("a", ProgressState::Running, true), - ); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: Some(agent_client_protocol::schema::v2::ToolCallStatus::Completed), - script: None, - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - let call = app - .blocks - .iter() - .find_map(|b| match b { - Block::Tool(c) => Some(c), - _ => None, - }) - .unwrap(); - assert!(call.progress.labels(&call.script).is_empty()); -} - -#[test] -fn authoritative_progress_node_bound_invalidates_without_partial_success() { - let mut app = sample(); - progress_start(&mut app, SCRIPT, 1, false); - for i in 0..=crate::runlet_progress::MAX_NODES { - progress_step( - &mut app, - 1, - i as u64 + 1, - progress_node(&format!("node-{i}"), ProgressState::Succeeded, true), - ); - } - assert!(!render(&mut app, 140, 50).contains("# call @")); -} - -async fn real_bridge_events(source: &str, capacity: usize, cancel: bool) -> Vec { - real_bridge_events_with_transport(source, capacity, cancel, None).await -} - -async fn real_bridge_events_with_transport( - source: &str, - capacity: usize, - cancel: bool, - transport: Option, -) -> Vec { - use agentkit_core::{MetadataMap, SessionId, ToolCallId, TurnId}; - use agentkit_tool_compose::{ - BackendRun, ComposeBackend, ComposeConfig, ComposeOutcome, ComposeTool, RunletBackend, - RunletProgress, - }; - use agentkit_tools_core::{ - AllowAllPermissions, BasicToolExecutor, OwnedToolContext, ToolExecutionScope, ToolExecutor, - ToolName, ToolRegistry, ToolRequest, - }; - use std::{num::NonZeroUsize, sync::Arc}; - struct ObservedBackend( - tokio::sync::mpsc::Sender, - usize, - Option, - ); - #[async_trait::async_trait] - impl ComposeBackend for ObservedBackend { - fn name(&self) -> &'static str { - RunletBackend.name() - } - fn description(&self, specs: Option<&[agentkit_tools_core::ToolSpec]>) -> String { - RunletBackend.description(specs) - } - fn script_description(&self) -> &'static str { - RunletBackend.script_description() - } - async fn execute(&self, run: BackendRun) -> Result { - if let Some(transport) = &self.2 { - return crate::runlet_progress::execute_observed(run, transport).await; - } - RunletBackend - .execute_with_progress(run, self.0.clone(), NonZeroUsize::new(self.1).unwrap()) - .await - } - } - struct Gate { - spec: agentkit_tools_core::ToolSpec, - entered: Arc, - release: Arc, - finished: Arc, - } - #[async_trait::async_trait] - impl agentkit_tools_core::Tool for Gate { - fn spec(&self) -> &agentkit_tools_core::ToolSpec { - &self.spec - } - async fn invoke( - &self, - request: ToolRequest, - _: &mut agentkit_tools_core::ToolContext<'_>, - ) -> Result { - self.entered.notify_one(); - self.release.notified().await; - self.finished.notify_one(); - Ok(agentkit_tools_core::ToolResult::new( - agentkit_core::ToolResultPart::success( - request.call_id, - agentkit_core::ToolOutput::Structured(serde_json::json!(1)), - ), - )) - } - } - let entered = Arc::new(tokio::sync::Notify::new()); - let release = Arc::new(tokio::sync::Notify::new()); - let finished = Arc::new(tokio::sync::Notify::new()); - let gate = Gate { - spec: agentkit_tools_core::ToolSpec::new( - "progress_gate", - "real host boundary", - serde_json::json!({"type":"object"}), - ), - entered: entered.clone(), - release: release.clone(), - finished: finished.clone(), - }; - let transported = transport.is_some(); - let (tx, mut rx) = tokio::sync::mpsc::channel(2); - let compose = ComposeTool::new(ComposeConfig::default()) - .with_backend(ObservedBackend(tx, capacity, transport)); - let executor: Arc = Arc::new(BasicToolExecutor::from_registry( - ToolRegistry::new().with(compose).with(gate), - )); - let session_id = SessionId::new("session"); - let turn_id = TurnId::new("turn"); - let permissions = Arc::new(AllowAllPermissions); - let resources: Arc = Arc::new(()); - let owned = OwnedToolContext { - session_id: session_id.clone(), - turn_id: turn_id.clone(), - metadata: MetadataMap::new(), - permissions: permissions.clone(), - resources: resources.clone(), - cancellation: None, - approved_request: None, - execution_scope: Some(ToolExecutionScope { - executor: executor.clone(), - session_id: session_id.clone(), - turn_id: turn_id.clone(), - permissions, - resources, - cancellation: None, - }), - }; - { - let mut context = owned.borrowed(); - let execute = executor.execute( - ToolRequest { - call_id: ToolCallId::new("call-1"), - tool_name: ToolName::new("compose"), - input: serde_json::json!({"script":source}), - session_id, - turn_id, - metadata: MetadataMap::new(), - }, - &mut context, - ); - if cancel { - use futures_util::future::{Either, select}; - let execute = std::pin::pin!(execute); - let entered = std::pin::pin!(entered.notified()); - let result = select(execute, entered).await; - assert!( - matches!(result, Either::Right(_)), - "host is held pending cancellation" - ); - } else { - let result = execute.await; - assert!( - matches!( - result, - agentkit_tools_core::ToolExecutionOutcome::Completed(_) - ), - "{result:?}" - ); - } - } - if cancel { - release.notify_one(); - finished.notified().await; - } - if transported { - return Vec::new(); - } - let mut stream = crate::runlet_progress::Active::new( - rx.try_recv().unwrap(), - crate::runlet_progress::transport::Transport::start(std::io::sink(), 16, true).unwrap(), - ); - let mut events = Vec::new(); - while let Some(event) = stream.poll() { - events.push(event); - } - events -} - -#[tokio::test] -async fn authoritative_progress_real_compose_bridge_to_diagnostic_app_and_render() { - let source = "a = text.upper(\"é\")\nb = text.lower(a)\nreturn b"; - let events = real_bridge_events(source, 1024, false).await; - assert!(events.iter().any(|event| matches!( - event, - RuntimeEvent::RunletProgress { - progress: Progress { - change: ProgressChange::Finished { complete: true }, - .. - } - } - ))); - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - for event in events { - progress_wire(&mut app, event); - } - let frame = render(&mut app, 140, 50); - assert!(frame.contains("succeeded"), "{frame}"); - assert!(!frame.contains("1 running"), "{frame}"); -} - -#[tokio::test] -async fn authoritative_progress_real_bridge_lag_invalidates_all_observations() { - let source = "return text.upper(\"é\")"; - let events = real_bridge_events(source, 1, false).await; - assert!(events.iter().any(|event| matches!( - event, - RuntimeEvent::RunletProgress { - progress: Progress { - change: ProgressChange::Finished { complete: false }, - .. - } - } - ))); - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - for event in events { - progress_wire(&mut app, event); - } - assert!(!render(&mut app, 140, 50).contains("# call @")); -} - -#[test] -fn authoritative_progress_two_subagents_dependency_is_not_descendant_completion() { - let source = "implementation = subagent({prompt: \"implement\"})\nreview = subagent({prompt: implementation.output})\nreturn review"; - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - progress_start(&mut app, source, 1, false); - let implementation = ProgressNode { - id: "implementation-attempt".into(), - call: true, - start: source.find("subagent").unwrap(), - end: source.find('\n').unwrap(), - state: ProgressState::Running, - attempt: 0, - }; - progress_step(&mut app, 1, 1, implementation); - for descendant in [ - "child:compose:storage", - "child:compose:tests", - "child:compose:review", - ] { - progress_wire( - &mut app, - RuntimeEvent::ChildFinished { - call: descendant.into(), - tool: "subagent".into(), - ok: true, - summary: "done".into(), - millis: 1, - }, - ); - } - let frame = render(&mut app, 150, 50); - assert!(frame.contains("review = subagent"), "{frame}"); - let implementation_line = frame - .lines() - .find(|line| line.contains("implementation = subagent")) - .unwrap(); - assert!( - implementation_line.contains("# call @L1:C") && implementation_line.contains("1 running"), - "{frame}" - ); - let review_line = frame - .lines() - .find(|line| line.contains("review = subagent")) - .unwrap(); - assert!( - !review_line.contains("# call") && !review_line.contains("running"), - "{frame}" - ); - assert!(frame.contains("1 running"), "{frame}"); - assert!(!frame.contains("succeeded"), "{frame}"); - assert!(!frame.contains("blocked"), "{frame}"); // no event has established it - let review = ProgressNode { - id: "review-attempt".into(), - call: true, - start: source.rfind("subagent").unwrap(), - end: source.rfind('\n').unwrap(), - state: ProgressState::Blocked, - attempt: 0, - }; - progress_step(&mut app, 1, 2, review); - assert!(render(&mut app, 150, 50).contains("1 blocked")); -} - -#[test] -fn authoritative_progress_retained_run_bound_and_source_bound() { - let mut app = sample(); - for i in 0..=crate::runlet_progress::MAX_RUNS { - let owner = format!("owner-{i}"); - app.apply(Update::ToolStarted { - id: owner.clone(), - title: "compose".into(), - kind: ToolKind::Other, - script: Some(SCRIPT.into()), - backgrounded: false, - }); - progress_wire( - &mut app, - source_event( - &owner, - i as u64 + 1, - 0, - ProgressChange::Started { - digest: crate::tui::progress::source_digest(SCRIPT), - healed: false, - }, - ), - ); - progress_wire( - &mut app, - source_event( - &owner, - i as u64 + 1, - 1, - ProgressChange::Step { - node: Some(progress_node("n", ProgressState::Running, true)), - }, - ), - ); - } - assert!( - app.blocks - .iter() - .filter(|b| matches!(b, Block::Tool(c) if c.progress.retained())) - .count() - <= crate::runlet_progress::MAX_RUNS - ); - let oldest = app - .blocks - .iter() - .find_map(|b| match b { - Block::Tool(c) if c.id == "owner-0" => Some(c), - _ => None, - }) - .unwrap(); - assert!(oldest.progress.labels(&oldest.script).is_empty()); - let huge = "🦀".repeat(crate::runlet_progress::MAX_SOURCE); - app.apply(Update::ToolStarted { - id: "huge".into(), - title: "compose".into(), - kind: ToolKind::Other, - script: Some(huge), - backgrounded: false, - }); - let Block::Tool(last) = app.blocks.last().unwrap() else { - panic!("tool") - }; - assert!(last.script.len() <= crate::runlet_progress::MAX_SOURCE); - assert!(last.script.ends_with("source truncated")); -} - -#[tokio::test] -async fn authoritative_progress_real_bridge_cancellation_invalidates() { - let source = "return progress_gate({})"; - let events = real_bridge_events(source, 1024, true).await; - assert!(events.iter().any(|event| matches!( - event, - RuntimeEvent::RunletProgress { - progress: Progress { - change: ProgressChange::Finished { complete: false }, - .. - } - } - ))); - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - for event in events { - progress_wire(&mut app, event); - } - assert!(!render(&mut app, 140, 50).contains("# call @")); -} - -#[tokio::test] -async fn authoritative_progress_real_bridge_healing_stays_neutral() { - let source = "if true { x = 1 }\nreturn 2"; - let events = real_bridge_events(source, 1024, false).await; - assert!(events.iter().any(|event| matches!( - event, - RuntimeEvent::RunletProgress { - progress: Progress { - change: ProgressChange::Started { healed: true, .. }, - .. - } - } - ))); - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - for event in events { - progress_wire(&mut app, event); - } - assert!(!render(&mut app, 140, 50).contains("# call @")); -} - -#[tokio::test] -async fn authoritative_progress_real_iterations_remain_distinct() { - let source = "values = for x in [\"a\", \"b\"] { return text.upper(x) }\nreturn values"; - let events = real_bridge_events(source, 1024, false).await; - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - for event in events { - progress_wire(&mut app, event); - } - assert!(render(&mut app, 140, 50).contains("2 succeeded")); -} - -#[test] -fn authoritative_progress_inline_nested_multiline_and_unicode_ranges() { - let source = - "first = text.upper(text.lower(\"🦀\"))\nsecond = text.upper(\n first\n)\nreturn second"; - let mut app = sample(); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: None, - script: Some(source.into()), - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - progress_start(&mut app, source, 1, false); - let first_end = source.find('\n').unwrap(); - let outer = ProgressNode { - id: "outer".into(), - call: true, - start: 8, - end: first_end, - state: ProgressState::Blocked, - attempt: 0, - }; - let inner = ProgressNode { - id: "inner-old".into(), - call: true, - start: source.find("text.lower").unwrap(), - end: first_end - 1, - state: ProgressState::Failed, - attempt: 0, - }; - progress_step(&mut app, 1, 1, outer); - progress_step(&mut app, 1, 2, inner.clone()); - progress_step( - &mut app, - 1, - 3, - ProgressNode { - id: "inner-retry".into(), - state: ProgressState::Running, - attempt: 1, - ..inner - }, - ); - progress_step( - &mut app, - 1, - 4, - ProgressNode { - id: "multiline".into(), - call: true, - start: source.rfind("text.upper").unwrap(), - end: source.rfind("\nreturn").unwrap(), - state: ProgressState::WaitingForCapacity, - attempt: 0, - }, - ); - progress_step( - &mut app, - 1, - 5, - ProgressNode { - id: "root".into(), - call: false, - start: 0, - end: source.len(), - state: ProgressState::Running, - attempt: 0, - }, - ); - let frame = render(&mut app, 240, 50); - let first = frame - .lines() - .find(|line| line.contains("first = text.upper")) - .unwrap(); - assert!( - first.contains("# call @L1:C9..L1:C36: 1 blocked"), - "{frame}" - ); - assert!( - first.contains("# call @L1:C20..L1:C35: 1 running, 1 failed"), - "{frame}" - ); - let second = frame - .lines() - .find(|line| line.contains("second = text.upper")) - .unwrap(); - assert!( - second.contains("# call @L2:C10..L4:C2: 1 waiting for capacity"), - "{frame}" - ); - assert!(!frame.contains("bytes "), "{frame}"); - assert!(!frame.contains("evaluating"), "{frame}"); - assert!( - !frame - .lines() - .find(|line| line.contains("return second")) - .unwrap() - .contains("# call") - ); -} - -#[test] -fn authoritative_progress_conflicting_duplicates_fail_neutral() { - for conflict in 0..5 { - let mut app = sample(); - progress_start(&mut app, SCRIPT, 1, false); - let node = progress_node("a", ProgressState::Succeeded, true); - progress_step(&mut app, 1, 1, node.clone()); - progress_step(&mut app, 1, 1, node.clone()); - progress_start(&mut app, SCRIPT, 1, false); - assert!(render(&mut app, 140, 50).contains("1 succeeded")); - match conflict { - 0 => progress_step( - &mut app, - 1, - 1, - ProgressNode { - state: ProgressState::Failed, - ..node - }, - ), - 1 => progress_start(&mut app, "return 1", 1, false), - 2 => progress_start(&mut app, SCRIPT, 1, true), - 3 => progress_wire( - &mut app, - source_event( - "call-1", - 1, - 2, - ProgressChange::Started { - digest: crate::tui::progress::source_digest(SCRIPT), - healed: false, - }, - ), - ), - _ => progress_step( - &mut app, - 1, - 2, - ProgressNode { - state: ProgressState::Failed, - ..node - }, - ), - } - assert!(!render(&mut app, 140, 50).contains("# call @")); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( - &mut app, - 1, - 2, - progress_node("a", ProgressState::Succeeded, true), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); - } -} - -fn start_progress_session(app: &mut App, session_id: &str) { - app.start_session(session_id.into()); - app.apply(Update::ToolStarted { - id: "call-1".into(), - title: "compose".into(), - kind: ToolKind::Other, - script: Some(SCRIPT.into()), - backgrounded: false, - }); -} - -#[test] -fn authoritative_progress_recovery_preserves_old_incarnation_tombstones() { - for explicit in [true, false] { - let mut app = sample(); - start_progress_session(&mut app, "session"); - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "session".into() }); - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Succeeded, true), - ); - progress_wire( - &mut app, - source_event("call-1", 1, 1, ProgressChange::Finished { complete: true }), - ); - assert!(render(&mut app, 140, 50).contains("1 succeeded")); - if explicit { - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); - } else { - app.progress_tick_at( - std::time::Instant::now() + crate::runlet_progress::transport::LEASE, - ); - } - assert!(!render(&mut app, 140, 50).contains("# call @")); - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); - assert!(!app.runtime_unavailable()); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Succeeded, true), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); - progress_start(&mut app, SCRIPT, 2, false); - progress_step( - &mut app, - 2, - 1, - progress_node("b", ProgressState::Succeeded, true), - ); - assert!(render(&mut app, 140, 50).contains("# call @")); - } -} - -#[cfg(unix)] -#[tokio::test] -async fn authoritative_progress_production_publication_completes_and_cancels_with_stalled_reader() { - use std::{ - io::{Read, Write}, - os::unix::net::UnixStream, - }; - for cancel in [false, true] { - let (mut writer, mut reader) = UnixStream::pair().unwrap(); - writer.set_nonblocking(true).unwrap(); - let mut filled = 0; - loop { - match writer.write(&[b'x'; 4096]) { - Ok(n) => filled += n, - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break, - Err(e) => panic!("fill: {e}"), - } - } - writer.set_nonblocking(false).unwrap(); - let transport = - crate::runlet_progress::transport::Transport::start(writer, 2, true).unwrap(); - let source = if cancel { - "return progress_gate({})" - } else { - "return text.upper(\"é\")" - }; - // Deadlock watchdog, not a timing/performance assertion. Reader remains - // completely stalled until the real production execute/drop finishes. - tokio::time::timeout( - std::time::Duration::from_secs(5), - real_bridge_events_with_transport(source, 1024, cancel, Some(transport.clone())), - ) - .await - .unwrap(); - drop(transport); - reader - .set_read_timeout(Some(std::time::Duration::from_secs(5))) - .unwrap(); - let mut prefix = vec![0; filled]; - reader.read_exact(&mut prefix).unwrap(); - let mut wire = String::new(); - reader.read_to_string(&mut wire).unwrap(); - assert!(wire.lines().any(|line| matches!( - crate::events::parse(line), - Some(RuntimeEvent::RunletTransport { available: false }) - ))); - } -} - -#[test] -fn authoritative_progress_terminal_conflicts_invalidate_completed_display() { - let mut app = sample(); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Succeeded, true), - ); - progress_wire( - &mut app, - source_event("call-1", 1, 1, ProgressChange::Finished { complete: true }), - ); - app.apply(Update::ToolPatched { - id: "call-1".into(), - title: None, - kind: None, - status: Some(agent_client_protocol::schema::v2::ToolCallStatus::Completed), - script: None, - output: None, - images: None, - append_output: false, - intent: None, - backgrounded: false, - }); - app.handle_key(KeyEvent::new(KeyCode::Char('o'), KeyModifiers::CONTROL)); - app.handle_key(KeyEvent::new(KeyCode::Char('o'), KeyModifiers::CONTROL)); - progress_step( - &mut app, - 1, - 1, - progress_node("a", ProgressState::Succeeded, true), - ); - progress_wire( - &mut app, - source_event("call-1", 1, 1, ProgressChange::Finished { complete: true }), - ); - assert!(render(&mut app, 140, 50).contains("1 succeeded")); - progress_wire( - &mut app, - source_event("call-1", 1, 1, ProgressChange::Finished { complete: false }), - ); - assert!(!render(&mut app, 140, 50).contains("# call @")); -} - -#[test] -fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { - for explicit in [false, true] { - let mut app = sample(); - start_progress_session(&mut app, "session"); - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "session".into() }); - let agent = RuntimeEvent::SubagentStateChanged { - id: "child-agent".into(), - name: "Child worker".into(), - status: SubagentStatus::Working, - outcome: None, - generation: 1, - task: "task".into(), - parent_id: Some("parent-agent".into()), - parent_name: Some("Parent".into()), - harness: "acp.kit".into(), - vendor: crate::events::HarnessVendor::Kit, - model: None, - created_at_unix_ms: 1, - generation_started_at_unix_ms: 2, - generation_finished_at_unix_ms: None, - }; - progress_wire(&mut app, agent.clone()); - progress_wire( - &mut app, - RuntimeEvent::ChildStarted { - call: "call-1:compose:0".into(), - tool: "shell".into(), - summary: "working child".into(), - at: 1, - }, - ); - progress_wire( - &mut app, - RuntimeEvent::StorageStatus { - pending: true, - exhausted: true, - }, - ); - progress_wire( - &mut app, - RuntimeEvent::CompactionStarted { - reason: "test".into(), - at: 1, - }, - ); - assert_eq!(app.agent_counts().working, 1); - assert!(app.storage_pending && app.storage_exhausted && app.compacting); - assert!( - app.blocks - .iter() - .any(|block| matches!(block, Block::Tool(call) if !call.children.is_empty())) - ); - if explicit { - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); - } else { - app.progress_tick_at( - std::time::Instant::now() + crate::runlet_progress::transport::LEASE, - ); - } - // Actual completion/cleanup can be lost, delayed or followed by buffered - // starts. None can make the incomplete stream trustworthy again. - for event in [ - RuntimeEvent::ChildFinished { - call: "call-1:compose:0".into(), - tool: "shell".into(), - ok: true, - summary: "done".into(), - millis: 2, - }, - RuntimeEvent::SubagentDescendantsRemoved { - ancestor_id: "parent-agent".into(), - }, - RuntimeEvent::StorageStatus { - pending: false, - exhausted: false, - }, - RuntimeEvent::CompactionFinished { - reason: "test".into(), - ok: true, - compacted: true, - millis: 2, - }, - agent.clone(), - ] { - progress_wire(&mut app, event); - } - assert!(app.runtime_unavailable()); - assert_eq!(app.agent_counts().total, 0); - assert!(!app.storage_pending && !app.storage_exhausted && !app.compacting); - assert!( - app.blocks - .iter() - .all(|block| !matches!(block, Block::Tool(call) if !call.children.is_empty())) - ); - let frame = render(&mut app, 140, 50); - assert!(frame.contains("Runtime status unavailable")); - assert!(frame.contains("storage state unknown")); - assert!(!frame.contains("Child worker")); - assert!(!frame.contains("working child")); - assert!(!frame.contains("compacting context")); - assert!(!frame.contains("context compacted")); - - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); - assert!(!app.runtime_unavailable()); - assert_eq!(app.agent_counts().total, 0); - assert!(!app.storage_pending && !app.storage_exhausted && !app.compacting); - // Leave room for the recovery note alongside the automatically opened roster. - let frame = render(&mut app, 200, 50); - assert!(!frame.contains("Runtime status unavailable")); - assert!(frame.contains("Runtime status resumed")); - assert!(frame.contains("state remains unknown")); - assert!(!frame.contains("Child worker")); - assert!(!frame.contains("working child")); - - // Fresh observations are accepted without reviving cleared state. - progress_wire(&mut app, agent); - progress_wire( - &mut app, - RuntimeEvent::StorageStatus { pending: true, exhausted: false }, - ); - assert_eq!(app.agent_counts().working, 1); - assert!(app.storage_pending); - assert!(app.needs_redraw_tick()); - app.progress_tick_at(std::time::Instant::now() + crate::runlet_progress::transport::LEASE); - assert!(app.runtime_unavailable()); - assert_eq!(app.agent_counts().total, 0); - assert!(!app.storage_pending); - } -} - -#[test] -fn runtime_recovery_keeps_session_filtering_across_attachment_gaps() { - for explicit in [false, true] { - for attach_during_gap in [false, true] { - let mut app = sample(); - start_progress_session(&mut app, "old"); - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "old".into() }); - if explicit { - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); - } else { - app.progress_tick_at(std::time::Instant::now() + crate::runlet_progress::transport::LEASE); - } - start_progress_session(&mut app, "new"); - if attach_during_gap { - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "new".into() }); - } - let compaction = RuntimeEvent::CompactionStarted { reason: "test".into(), at: 1 }; - progress_wire(&mut app, compaction.clone()); - progress_start(&mut app, SCRIPT, 1, false); - assert!(app.runtime_unavailable()); - assert!(!app.compacting); - assert!(!render(&mut app, 140, 50).contains("# call @")); - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); - progress_wire(&mut app, compaction.clone()); - progress_start(&mut app, SCRIPT, 2, false); - progress_step(&mut app, 2, 1, progress_node("a", ProgressState::Succeeded, true)); - assert_eq!(app.compacting, attach_during_gap); - assert_eq!(render(&mut app, 140, 50).contains("# call @"), attach_during_gap); - // A heartbeat must not guess that the stream belongs to the selected session. - if !attach_during_gap { - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "new".into() }); - progress_wire(&mut app, compaction); - progress_start(&mut app, SCRIPT, 3, false); - progress_step(&mut app, 3, 1, progress_node("a", ProgressState::Succeeded, true)); - assert!(app.compacting); - assert!(render(&mut app, 140, 50).contains("# call @")); - } - } - } -} diff --git a/src/tui/runtime_health_tests.rs b/src/tui/runtime_health_tests.rs new file mode 100644 index 00000000..c957896c --- /dev/null +++ b/src/tui/runtime_health_tests.rs @@ -0,0 +1,155 @@ +// Included in ui tests; exercises real diagnostic wire, app, and render APIs. +fn runtime_wire(app: &mut App, event: RuntimeEvent) { + let mut bytes = Vec::new(); + crate::events::test_support::write_event(&mut bytes, &event); + let line = String::from_utf8(bytes).unwrap(); + app.apply(Update::Runtime( + crate::events::parse(line.trim_end()).unwrap(), + )); +} +fn start_runtime_session(app: &mut App, session_id: &str) { + app.start_session(session_id.into()); + app.apply(Update::ToolStarted { + id: "call-1".into(), + title: "compose".into(), + kind: ToolKind::Other, + script: Some(SCRIPT.into()), + backgrounded: false, + }); +} + +#[test] +fn runtime_loss_invalidates_all_runtime_lifecycle_state() { + for explicit in [false, true] { + let mut app = sample(); + start_runtime_session(&mut app, "session"); + runtime_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "session".into() }); + let agent = RuntimeEvent::SubagentStateChanged { + id: "child-agent".into(), + name: "Child worker".into(), + status: SubagentStatus::Working, + outcome: None, + generation: 1, + task: "task".into(), + parent_id: Some("parent-agent".into()), + parent_name: Some("Parent".into()), + harness: "acp.kit".into(), + vendor: crate::events::HarnessVendor::Kit, + model: None, + created_at_unix_ms: 1, + generation_started_at_unix_ms: 2, + generation_finished_at_unix_ms: None, + }; + runtime_wire(&mut app, agent.clone()); + runtime_wire( + &mut app, + RuntimeEvent::StorageStatus { + pending: true, + exhausted: true, + }, + ); + runtime_wire( + &mut app, + RuntimeEvent::CompactionStarted { + reason: "test".into(), + at: 1, + }, + ); + assert_eq!(app.agent_counts().working, 1); + assert!(app.storage_pending && app.storage_exhausted && app.compacting); + if explicit { + runtime_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); + } else { + app.runtime_tick_at( + std::time::Instant::now() + crate::diagnostic_transport::LEASE, + ); + } + // Actual completion/cleanup can be lost, delayed or followed by buffered + // starts. None can make the incomplete stream trustworthy again. + for event in [ + RuntimeEvent::SubagentDescendantsRemoved { + ancestor_id: "parent-agent".into(), + }, + RuntimeEvent::StorageStatus { + pending: false, + exhausted: false, + }, + RuntimeEvent::CompactionFinished { + reason: "test".into(), + ok: true, + compacted: true, + millis: 2, + }, + agent.clone(), + ] { + runtime_wire(&mut app, event); + } + assert!(app.runtime_unavailable()); + assert_eq!(app.agent_counts().total, 0); + assert!(!app.storage_pending && !app.storage_exhausted && !app.compacting); + let frame = render(&mut app, 140, 50); + assert!(frame.contains("Runtime status unavailable")); + assert!(frame.contains("storage state unknown")); + assert!(!frame.contains("Child worker")); + assert!(!frame.contains("compacting context")); + assert!(!frame.contains("context compacted")); + + runtime_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); + assert!(!app.runtime_unavailable()); + assert_eq!(app.agent_counts().total, 0); + assert!(!app.storage_pending && !app.storage_exhausted && !app.compacting); + // Leave room for the recovery note alongside the automatically opened roster. + let frame = render(&mut app, 200, 50); + assert!(!frame.contains("Runtime status unavailable")); + assert!(frame.contains("Runtime status resumed")); + assert!(frame.contains("state remains unknown")); + assert!(!frame.contains("Child worker")); + + // Fresh observations are accepted without reviving cleared state. + runtime_wire(&mut app, agent); + runtime_wire( + &mut app, + RuntimeEvent::StorageStatus { pending: true, exhausted: false }, + ); + assert_eq!(app.agent_counts().working, 1); + assert!(app.storage_pending); + assert!(app.needs_redraw_tick()); + app.runtime_tick_at(std::time::Instant::now() + crate::diagnostic_transport::LEASE); + assert!(app.runtime_unavailable()); + assert_eq!(app.agent_counts().total, 0); + assert!(!app.storage_pending); + } +} + +#[test] +fn runtime_recovery_keeps_session_filtering_across_attachment_gaps() { + for explicit in [false, true] { + for attach_during_gap in [false, true] { + let mut app = sample(); + start_runtime_session(&mut app, "old"); + runtime_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "old".into() }); + if explicit { + runtime_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); + } else { + app.runtime_tick_at(std::time::Instant::now() + crate::diagnostic_transport::LEASE); + } + start_runtime_session(&mut app, "new"); + if attach_during_gap { + runtime_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "new".into() }); + } + let compaction = RuntimeEvent::CompactionStarted { reason: "test".into(), at: 1 }; + runtime_wire(&mut app, compaction.clone()); + assert!(app.runtime_unavailable()); + assert!(!app.compacting); + runtime_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); + runtime_wire(&mut app, compaction.clone()); + assert_eq!(app.compacting, attach_during_gap); + // A heartbeat must not guess that the stream belongs to the selected session. + if !attach_during_gap { + runtime_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "new".into() }); + runtime_wire(&mut app, compaction); + assert!(app.compacting); + } + } + } +} diff --git a/src/tui/source.rs b/src/tui/source.rs new file mode 100644 index 00000000..ac86f354 --- /dev/null +++ b/src/tui/source.rs @@ -0,0 +1,33 @@ +//! Bounded, neutral display of compose source. + +const MAX_SOURCE: usize = 64 * 1024; + +pub(crate) fn bounded_source(source: String) -> String { + if source.len() <= MAX_SOURCE { + return source; + } + let mut end = MAX_SOURCE - 32; + while !source.is_char_boundary(end) { + end -= 1; + } + format!("{}\n… source truncated", &source[..end]) +} + +#[cfg(test)] +#[allow(clippy::disallowed_macros)] +mod tests { + use super::*; + + #[test] + fn bounded_source_preserves_small_input_and_utf8_at_the_limit() { + for source in [String::new(), "return 🦀".into(), "x".repeat(MAX_SOURCE)] { + assert_eq!(bounded_source(source.clone()), source); + } + let source = "🦀".repeat(MAX_SOURCE); + let bounded = bounded_source(source.clone()); + assert!(bounded.len() <= MAX_SOURCE); + let suffix = "\n… source truncated"; + assert!(bounded.ends_with(suffix)); + assert!(source.starts_with(&bounded[..bounded.len() - suffix.len()])); + } +} diff --git a/src/tui/ui.rs b/src/tui/ui.rs index 8fa92869..1e4bb0d6 100644 --- a/src/tui/ui.rs +++ b/src/tui/ui.rs @@ -21,7 +21,7 @@ use crate::events::{GenerationOutcome, SubagentStatus}; use super::{ app::{ AgentPart, AgentTreeRow, App, Block, CachedTranscriptBlock, CachedTranscriptImage, - CachedTranscriptRow, Child, CodeHit, ComposeView, EffortDialog, FilePickerDialog, + CachedTranscriptRow, CodeHit, ComposeView, EffortDialog, FilePickerDialog, FilePickerStatus, ModelDialog, Phase, SessionRename, ToolCall, UserMessage, }, command, @@ -1168,7 +1168,7 @@ fn refresh_transcript_cache_with_images(app: &mut App, images: &mut ImageRuntime for block_index in dirty { let dynamic = match &app.blocks[block_index] { Block::Thought { millis, .. } => millis.is_none(), - Block::Tool(call) => call.running() || call.running_children() > 0, + Block::Tool(call) => call.running(), _ => false, }; let revision = app.transcript_revisions[block_index]; @@ -1624,28 +1624,12 @@ fn tool_lines(app: &App, call: &ToolCall, active: bool) -> Vec> { && !call.script.is_empty())) { lines.extend(script_lines(call)); - } else if let Some(child) = call.children.iter().rev().find(|child| child.running()) { - lines.push(Line::from(vec![ - Span::styled(" ↳ ", theme::faint()), - Span::styled(child.summary.clone(), theme::dim()), - ])); } if call.expanded { lines.extend(output_lines(call)); } return lines; } - if !compose { - for child in call.children.iter().take(6) { - lines.push(Line::from(child_spans(app, child, " "))); - } - if call.children.len() > 6 { - lines.push(Line::from(Span::styled( - format!(" … {} more calls", call.children.len() - 6), - theme::faint(), - ))); - } - } if compose { lines.extend(completed_compose_lines(call)); } else { @@ -1654,25 +1638,17 @@ fn tool_lines(app: &App, call: &ToolCall, active: bool) -> Vec> { lines } -/// Source lines stay neutral. Exact runtime call spans add qualified counts at -/// their source start line; annotations never assert whole-line/binding state. +/// Source lines stay neutral; ACP owns tool lifecycle display. fn script_lines(call: &ToolCall) -> Vec> { - let annotations = call.progress.labels(&call.script); let mut lines: Vec<_> = call .script .lines() .take(MAX_OUTPUT_ROWS) - .enumerate() - .map(|(index, source)| { - let mut spans = vec![ + .map(|source| { + let spans = vec![ Span::styled(" │ ", theme::faint()), Span::styled(source.to_string(), theme::dim()), ]; - if let Some(labels) = annotations.get(&(index + 1)) { - for label in labels { - spans.push(Span::styled(format!(" {label}"), theme::dim())); - } - } Line::from(spans) }) .collect(); @@ -1688,36 +1664,7 @@ fn script_lines(call: &ToolCall) -> Vec> { fn completed_compose_lines(call: &ToolCall) -> Vec> { if !call.expanded { - let mut counts = std::collections::HashMap::<&str, usize>::new(); - for child in &call.children { - *counts.entry(child.tool.as_str()).or_default() += 1; - } - let mut counts = counts.into_iter().collect::>(); - counts.sort_by(|(left_name, left_count), (right_name, right_count)| { - right_count - .cmp(left_count) - .then_with(|| left_name.cmp(right_name)) - }); - if counts.is_empty() { - return output_lines(call); - } - let summary = counts - .into_iter() - .take(4) - .map(|(name, count)| { - if count > 1 { - format!("{name} x {count}") - } else { - name.to_string() - } - }) - .collect::>() - .join(" · "); - return vec![Line::from(vec![ - Span::styled(" ▸ ", theme::dim()), - Span::styled(summary, theme::dim()), - Span::styled(" click or ^o to open", theme::faint()), - ])]; + return output_lines(call); } let (output_style, script_style, hint) = match call.compose_view { @@ -1832,18 +1779,6 @@ fn tool_header(app: &App, call: &ToolCall, active: bool) -> Vec> { spans.push(Span::styled(" · ^k kill", theme::accent())); } } - let running = call.running_children(); - if running > 0 { - spans.push(Span::styled( - format!(" · {running} in flight"), - Style::default().fg(theme::running_color()), - )); - } else if (call.expanded || call.running() || !call.is_compose()) && !call.children.is_empty() { - spans.push(Span::styled( - format!(" · {} calls", call.children.len()), - theme::faint(), - )); - } spans } @@ -1863,33 +1798,6 @@ fn kind_label(kind: &ToolKind) -> &'static str { } } -fn child_spans(app: &App, child: &Child, indent: &str) -> Vec> { - let (glyph, style) = if child.running() { - ( - theme::pulse(theme::Pulse::Child, app.tick).to_string(), - Style::default().fg(theme::running_color()), - ) - } else if child.ok { - ("✓".into(), Style::default().fg(theme::success_color())) - } else { - ("✗".into(), Style::default().fg(theme::error_color())) - }; - let detail = if child.running() || child.result.is_empty() { - child.summary.clone() - } else { - child.result.clone() - }; - vec![ - Span::styled(format!("{indent}{glyph} "), style), - Span::styled(format!("{:<8}", child.tool), theme::dim()), - Span::styled( - format!("{:>7} ", theme::duration(child.elapsed())), - theme::faint(), - ), - Span::styled(detail, theme::dim()), - ] -} - fn working_line(app: &App) -> Line<'static> { let label = match app.phase { Phase::Cancelling => "stopping", @@ -3538,7 +3446,7 @@ mod tests { assert_eq!(right, 79); } - include!("progress_tests.rs"); + include!("runtime_health_tests.rs"); const SCRIPT: &str = "files = shell({ command: \"ls src\" })\n\ checked = for file in files.lines {\n\ @@ -3564,25 +3472,6 @@ mod tests { script: Some(SCRIPT.into()), backgrounded: false, }); - app.apply(Update::Runtime(RuntimeEvent::ChildStarted { - call: "call-1:compose:one".into(), - tool: "shell".into(), - summary: "ls src".into(), - at: 0, - })); - app.apply(Update::Runtime(RuntimeEvent::ChildFinished { - call: "call-1:compose:one".into(), - tool: "shell".into(), - ok: true, - summary: "main.rs".into(), - millis: 120, - })); - app.apply(Update::Runtime(RuntimeEvent::ChildStarted { - call: "call-1:compose:two".into(), - tool: "shell".into(), - summary: "cargo check".into(), - at: 0, - })); app } @@ -4051,7 +3940,7 @@ mod tests { } #[test] - fn compose_script_stays_neutral_with_running_and_successful_calls() { + fn compose_source_stays_neutral() { let mut app = sample(); let frame = render(&mut app, 120, 40); @@ -4060,7 +3949,6 @@ mod tests { frame.contains("files = shell({ command: \"ls src\" })"), "{frame}" ); - assert!(frame.contains("1 in flight"), "{frame}"); assert!(!frame.contains(" # "), "{frame}"); assert!(!frame.contains("resolved"), "{frame}"); assert!(!frame.contains("iteration 1 running"), "{frame}"); @@ -4068,95 +3956,6 @@ mod tests { assert!(app.transcript_width > 100, "{}", app.transcript_width); } - #[test] - fn compose_script_does_not_infer_failure_retry_or_waiting_state() { - let script = "value = boundary retry 2 {\n\ - return shell({ command: \"false\" })\n\ - } catch err {\n\ - return fail(\"FAILED\", err.message)\n\ - }\n\ - later = docs({ query: \"next\" })\n\ - return value"; - let mut app = App::new( - PathBuf::from("/Users/dev/projects/kit"), - "openai-subscription".into(), - "gpt-5.4".into(), - "127.0.0.1:7331".into(), - ); - app.apply(Update::ToolStarted { - id: "call-1".into(), - title: "compose".into(), - kind: ToolKind::Other, - script: Some(script.into()), - backgrounded: false, - }); - app.apply(Update::Runtime(RuntimeEvent::ChildStarted { - call: "call-1:compose:failed".into(), - tool: "shell".into(), - summary: "false".into(), - at: 0, - })); - app.apply(Update::Runtime(RuntimeEvent::ChildFinished { - call: "call-1:compose:failed".into(), - tool: "shell".into(), - ok: false, - summary: "exit code 1".into(), - millis: 10, - })); - - let frame = render(&mut app, 100, 30); - - assert!(frame.contains("value = boundary retry 2 {"), "{frame}"); - assert!(!frame.contains(" # "), "{frame}"); - assert!(!frame.contains("value failed"), "{frame}"); - assert!(!frame.contains("attempt 1"), "{frame}"); - assert!(!frame.contains("shell failure"), "{frame}"); - assert!(!frame.contains("later waiting"), "{frame}"); - } - - #[test] - fn compose_script_does_not_attribute_descendants_to_a_dependent_review() { - let mut app = App::new( - PathBuf::from("/Users/dev/projects/kit"), - "openai-subscription".into(), - "gpt-5.4".into(), - "127.0.0.1:7331".into(), - ); - let script = "a = subagent({name: \"implementation\", prompt: input.task})\n\ - r = subagent({name: \"review\", prompt: json.encode(a.output)})\n\ - return r"; - app.apply(Update::ToolStarted { - id: "root".into(), - title: "compose".into(), - kind: ToolKind::Other, - script: Some(script.into()), - backgrounded: true, - }); - for call in [ - "root:compose:implementation", - "child:compose:storage", - "child:compose:backfill", - "child:compose:transport", - "child:compose:tests", - ] { - app.apply(Update::Runtime(RuntimeEvent::ChildStarted { - call: call.into(), - tool: "subagent".into(), - summary: "working".into(), - at: 0, - })); - } - let frame = render(&mut app, 160, 30); - assert!(frame.contains("1 in flight"), "{frame}"); - let source_rows: Vec<_> = frame - .lines() - .filter_map(|line| line.split_once("│ ").map(|(_, source)| source.trim_end())) - .collect(); - assert_eq!(source_rows, script.lines().collect::>()); - assert!(!frame.contains("subagent: 2 running"), "{frame}"); - assert!(!frame.contains("subagent: 3 running"), "{frame}"); - } - #[test] fn compose_title_uses_trimmed_intent_or_running_tools_fallback() { let mut app = sample(); @@ -4181,43 +3980,6 @@ mod tests { assert!(!intent.contains("Running tools."), "{intent}"); } - #[test] - fn collapsed_compose_groups_sorts_and_caps_child_tool_names() { - let mut app = sample(); - let Block::Tool(call) = app.blocks.last_mut().expect("compose call") else { - panic!("last block was not a tool"); - }; - for (index, tool) in [ - "shell", "docs", "shell", "alpha", "edit", "docs", "fork", "shell", "alpha", - ] - .into_iter() - .enumerate() - { - call.attach(format!("extra-{index}"), tool.into(), tool.into()); - } - app.apply(Update::ToolPatched { - title: None, - kind: None, - images: None, - append_output: false, - intent: None, - id: "call-1".into(), - status: Some(agent_client_protocol::schema::v2::ToolCallStatus::Completed), - script: None, - output: Some(vec!["done".into()]), - backgrounded: false, - }); - - let frame = render(&mut app, 120, 35); - let shell = frame.find("shell x 5").expect("shell summary"); - let alpha = frame.find("alpha x 2").expect("alpha summary"); - let docs = frame.find("docs x 2").expect("docs summary"); - let edit = frame.find("edit").expect("edit summary"); - assert!(shell < alpha && alpha < docs && docs < edit, "{frame}"); - assert!(!frame.contains("edit x 1"), "{frame}"); - assert!(!frame.contains("fork"), "{frame}"); - } - #[test] fn completed_compose_stays_collapsed_when_its_title_arrives_late() { let mut app = App::new( @@ -4273,7 +4035,7 @@ mod tests { }); let collapsed = render(&mut app, 100, 30); - assert!(collapsed.contains("shell x 2"), "{collapsed}"); + assert!(collapsed.contains("1 line of output"), "{collapsed}"); assert!(!collapsed.contains("compose result"), "{collapsed}"); assert!(!collapsed.contains("files = shell"), "{collapsed}"); @@ -4293,7 +4055,10 @@ mod tests { app.toggle_last_output(); let collapsed_again = render(&mut app, 100, 30); - assert!(collapsed_again.contains("shell x 2"), "{collapsed_again}"); + assert!( + collapsed_again.contains("1 line of output"), + "{collapsed_again}" + ); assert!(!collapsed_again.contains("Output"), "{collapsed_again}"); } @@ -4368,42 +4133,6 @@ mod tests { assert!(previous.is_some_and(|call| !call.expanded)); } - #[test] - fn non_compose_child_summary_requires_a_matching_parent() { - let mut app = App::new( - PathBuf::from("/Users/dev/projects/kit"), - "openai-subscription".into(), - "gpt-5.4".into(), - "127.0.0.1:7331".into(), - ); - app.apply(Update::ToolStarted { - id: "call-1".into(), - title: "shell".into(), - kind: ToolKind::Execute, - script: None, - backgrounded: false, - }); - app.apply(Update::Runtime(RuntimeEvent::ChildStarted { - call: "call-1:child".into(), - tool: "shell".into(), - summary: "cargo check".into(), - at: 0, - })); - - let frame = render(&mut app, 80, 20); - - assert!(!frame.contains("↳ cargo check"), "{frame}"); - - app.apply(Update::Runtime(RuntimeEvent::ChildStarted { - call: "call-1:compose:child".into(), - tool: "shell".into(), - summary: "cargo check".into(), - at: 0, - })); - let frame = render(&mut app, 80, 20); - assert!(frame.contains("↳ cargo check"), "{frame}"); - } - #[test] fn only_the_focused_call_shows_the_kill_hint() { let mut app = App::new(