diff --git a/Cargo.lock b/Cargo.lock index 76a09c08..9370d3a1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -342,9 +342,9 @@ dependencies = [ [[package]] name = "agentkit-tool-compose" -version = "0.10.10" +version = "0.10.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddfa6bc961e33850391795f78557d86d581d85267fbb797c2846be7df27dbf80" +checksum = "2bd493e7dd9ed206d203ec37c4ff5c65bb97e5a77b5f0f385a366d5e75064c0c" dependencies = [ "agentkit-core", "agentkit-tools-core", @@ -2674,6 +2674,7 @@ dependencies = [ "ratatui-image", "reqwest", "rmcp", + "runlet", "serde", "serde_json", "sha2 0.11.0", @@ -3855,9 +3856,9 @@ dependencies = [ [[package]] name = "runlet" -version = "0.5.0" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "927f293f868ee447cfc55b1e5220f1717a907c566f45d836c8810e5155b3b6c8" +checksum = "057e0864428c5a79a68683942d3750d05e9ffae541aa0185fde9ae1c87eca100" dependencies = [ "hex", "regex", diff --git a/Cargo.toml b/Cargo.toml index 46be5476..97d27635 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ agentkit-plugins = "=0.10.7" agentkit-provider-openai = "=0.10.8" agentkit-provider-openrouter = "=0.10.8" agentkit-task-manager = "=0.10.7" -agentkit-tool-compose = { version = "=0.10.10", default-features = false, features = ["runlet"] } +agentkit-tool-compose = { version = "=0.10.11", default-features = false, features = ["runlet"] } agentkit-tool-skills = "=0.10.8" agentkit-tools-core = "=0.10.5" async-trait = "=0.1.92" @@ -56,6 +56,7 @@ rmcp = { version = "=3.2.0", default-features = false, features = ["auth", "clie serde = { version = "=1.0.229", features = ["derive"] } serde_json = "=1.0.151" shlex = "=2.0.1" +runlet = "=0.6.0" sha2 = "=0.11.0" subtle = "=2.6.1" tar = { version = "=0.4.46", default-features = false } diff --git a/src/acp_child.rs b/src/acp_child.rs index 291baeba..23781eba 100644 --- a/src/acp_child.rs +++ b/src/acp_child.rs @@ -840,6 +840,8 @@ fn harness_diagnostic(label: &str, line: &str) -> Option { Some( crate::events::RuntimeEvent::ChildStarted { .. } | crate::events::RuntimeEvent::ChildFinished { .. } + | crate::events::RuntimeEvent::RunletProgress { .. } + | crate::events::RuntimeEvent::RunletTransport { .. } ) ) { return None; @@ -898,8 +900,15 @@ async fn run( &label, ancestor_id.as_deref(), |output| match output { - ForwardedStderr::RuntimeLine(line) | ForwardedStderr::Diagnostic(line) => { - eprintln!("{line}"); + ForwardedStderr::RuntimeLine(line) => { + if let Some(transport) = crate::runlet_progress::transport::global() { + transport.publish_runtime_line(&line); + } + } + ForwardedStderr::Diagnostic(line) => { + if let Some(transport) = crate::runlet_progress::transport::global() { + transport.publish_line(&line); + } } ForwardedStderr::Cleanup(event) => crate::events::emit(&event), }, @@ -1256,20 +1265,71 @@ async fn forward_stderr( .map(str::to_owned) .collect::>(); let mut lines = BufReader::new(stderr).lines(); - while let Ok(Some(line)) = lines.next_line().await { - if let Some(event) = crate::events::parse(&line) - && event.forward_from_child() - { - if let crate::events::RuntimeEvent::SubagentStateChanged { - parent_id: Some(parent_id), - .. - } = event - { - ancestors.insert(parent_id); + let mut deadline = None; + let mut unavailable = false; + loop { + // A nested Kit transport has its own lease. The parent's healthy + // heartbeat cannot certify a stalled or failed descendant publisher. + let next = if let Some(expires) = deadline { + if tokio::time::Instant::now() >= expires { + Err(()) + } else { + tokio::time::timeout_at(expires, lines.next_line()) + .await + .map_err(|_| ()) + } + } else { + Ok(lines.next_line().await) + }; + let line = match next { + Ok(Ok(Some(line))) => line, + Ok(_) => break, + Err(()) => { + unavailable = true; + deadline = None; + output(ForwardedStderr::Cleanup( + crate::events::RuntimeEvent::RunletTransport { available: false }, + )); + continue; + } + }; + if let Some(event) = crate::events::parse(&line) { + if unavailable { + continue; + } + match event { + crate::events::RuntimeEvent::RunletTransport { available: true } => { + deadline = Some( + tokio::time::Instant::now() + crate::runlet_progress::transport::LEASE, + ); + continue; + } + crate::events::RuntimeEvent::RunletTransport { available: false } => { + unavailable = true; + deadline = None; + output(ForwardedStderr::RuntimeLine(line)); + continue; + } + _ => {} + } + if deadline.is_some() { + deadline = + Some(tokio::time::Instant::now() + crate::runlet_progress::transport::LEASE); + } + if event.forward_from_child() { + if let crate::events::RuntimeEvent::SubagentStateChanged { + parent_id: Some(parent_id), + .. + } = event + { + ancestors.insert(parent_id); + } + // Preserve recursively forwarded private events byte-for-byte. + output(ForwardedStderr::RuntimeLine(line)); + continue; } - // Preserve recursively forwarded private runtime events byte-for-byte. - output(ForwardedStderr::RuntimeLine(line)); - } else if let Some(line) = harness_diagnostic(label, &line) { + } + if let Some(line) = harness_diagnostic(label, &line) { output(ForwardedStderr::Diagnostic(line)); } } @@ -1408,6 +1468,27 @@ mod tests { ) } + #[tokio::test] + async fn close_with_stalled_stderr() { + if !crate::events::test_support::with_stalled_stderr( + "acp_child::tests::close_with_stalled_stderr", + ) { + return; + } + let (mut session, mut requests) = admission_test_session(); + session.descendant_parent = Some("ancestor".into()); + session.capabilities.session_capabilities.close = Some(Default::default()); + let actor = async { + let Some(Request::Close(close)) = requests.recv().await else { + panic!("expected actual child close request"); + }; + assert_eq!(close.session_id.to_string(), "test"); + close.reply.send(Ok(())).unwrap(); + }; + let (result, ()) = tokio::join!(session.close(), actor); + result.unwrap(); + } + #[tokio::test] async fn actor_events_share_ready_request_fatal_and_completion_backlogs() { for enabled in [0b011u8, 0b101, 0b110, 0b111] { @@ -2671,6 +2752,71 @@ mod tests { mod forwards_subagent_events { use super::*; + #[tokio::test(start_paused = true)] + async fn nested_transport_loss_preserves_diagnostics_not_stale_lifecycle() { + use crate::events::{EVENT_MARKER, RuntimeEvent}; + use tokio::io::AsyncWriteExt; + for explicit in [false, true] { + let (mut writer, reader) = tokio::io::duplex(4096); + let (tx, mut rx) = mpsc::unbounded_channel(); + let forward = tokio::spawn(async move { + forward_stderr(reader, "acp.kit", None, |item| { + tx.send(item).unwrap(); + }) + .await; + }); + let heartbeat = format!( + "{EVENT_MARKER}{}\n", + 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 start = format!( + "{EVENT_MARKER}{}\n", + serde_json::to_string(&started).unwrap() + ); + writer + .write_all(format!("{heartbeat}{start}").as_bytes()) + .await + .unwrap(); + assert!(matches!( + rx.recv().await.unwrap(), + ForwardedStderr::RuntimeLine(_) + )); + if explicit { + let reset = format!( + "{EVENT_MARKER}{}\n", + serde_json::to_string(&RuntimeEvent::RunletTransport { available: false }) + .unwrap() + ); + writer.write_all(reset.as_bytes()).await.unwrap(); + } else { + tokio::time::advance(crate::runlet_progress::transport::LEASE).await; + } + let reset = match rx.recv().await.unwrap() { + ForwardedStderr::RuntimeLine(line) => crate::events::parse(&line).unwrap(), + ForwardedStderr::Cleanup(event) => event, + _ => panic!("expected nested invalidation"), + }; + assert_eq!(reset, RuntimeEvent::RunletTransport { available: false }); + writer + .write_all(format!("{heartbeat}{start}later child error\n").as_bytes()) + .await + .unwrap(); + assert!( + matches!(rx.recv().await.unwrap(), ForwardedStderr::Diagnostic(line) if line.contains("later child error")) + ); + drop(writer); + forward.await.unwrap(); + assert!(rx.try_recv().is_err()); + } + } + #[tokio::test] async fn preserves_nested_roster_event_lines_exactly() { let event = crate::events::RuntimeEvent::SubagentStateChanged { diff --git a/src/events.rs b/src/events.rs index 4c38fa2f..709bbbfb 100644 --- a/src/events.rs +++ b/src/events.rs @@ -14,7 +14,6 @@ //! ACP hosts never see the extra chatter. use std::{ - io::Write, sync::OnceLock, time::{SystemTime, UNIX_EPOCH}, }; @@ -36,6 +35,12 @@ pub const EVENTS_ENV: &str = "KIT_RUNTIME_EVENTS"; #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "event", rename_all = "snake_case")] pub enum RuntimeEvent { + /// Process-wide progress transport lease/reset, not a source execution. + 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. @@ -116,8 +121,10 @@ impl RuntimeEvent { #[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::StorageStatus { .. } + Self::RunletTransport { .. } + | Self::StorageStatus { .. } | Self::SessionStarted { .. } | Self::CompactionStarted { .. } | Self::CompactionFinished { .. } @@ -135,25 +142,33 @@ pub fn enabled() -> bool { *ENABLED.get_or_init(|| std::env::var_os(EVENTS_ENV).is_some()) } -/// Writes one event to stderr when emission is enabled. +/// 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. pub fn emit(event: &RuntimeEvent) { if !enabled() { return; } - let mut stderr = std::io::stderr().lock(); - write_event(&mut stderr, event); -} - -fn write_event(writer: &mut impl Write, event: &RuntimeEvent) { - if let Ok(line) = serde_json::to_string(event) { - let _ = writeln!(writer, "{EVENT_MARKER}{line}"); + if let Some(transport) = crate::runlet_progress::transport::global() { + transport.publish_event(event); } } /// Parses one stderr line, returning an event when the line carries one. #[must_use] pub fn parse(line: &str) -> Option { - serde_json::from_str(line.strip_prefix(EVENT_MARKER)?).ok() + let body = line.strip_prefix(EVENT_MARKER)?; + 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) } /// Milliseconds since the Unix epoch, saturating at zero on a broken clock. @@ -236,7 +251,7 @@ mod tests { use super::{ EVENT_MARKER, GenerationOutcome, RuntimeEvent, SubagentStatus, parse, summarize_input, - summarize_output, write_event, + summarize_output, test_support::write_event, }; #[test] @@ -394,3 +409,6 @@ mod tests { ); } } + +#[cfg(test)] +pub(crate) mod test_support; diff --git a/src/events/test_support.rs b/src/events/test_support.rs new file mode 100644 index 00000000..d606bdf7 --- /dev/null +++ b/src/events/test_support.rs @@ -0,0 +1,72 @@ +//! Isolate the process-wide stderr lock, event opt-in, and singleton transport. +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +use std::{ + io::Write, + process::{Command, Stdio}, + sync::mpsc, + time::{Duration, Instant}, +}; + +/// Parent runs the exact test with an unread stderr pipe and a deadlock watchdog. +/// Child holds Rust's actual stderr lock on a detached writer filling that pipe. +/// Returning true means the caller must exercise its real execution/cleanup API. +pub(crate) fn with_stalled_stderr(test: &str) -> bool { + const CHILD: &str = "KIT_STDERR_CONTENTION_TEST"; + if std::env::var(CHILD).as_deref() == Ok(test) { + let (ready, locked) = mpsc::channel(); + std::thread::spawn(move || { + let mut stderr = std::io::stderr().lock(); + ready.send(()).unwrap(); + loop { + stderr.write_all(&[b'x'; 65536]).unwrap(); + } + }); + locked.recv().unwrap(); + assert!(super::enabled()); + // Start the actual process singleton behind the contended stderr lock. + super::emit(&super::RuntimeEvent::SessionStarted { + session_id: "test".into(), + }); + return true; + } + let mut child = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", test, "--nocapture"]) + .env(CHILD, test) + .env(super::EVENTS_ENV, "1") + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if let Some(status) = child.try_wait().unwrap() { + assert!(status.success(), "contended subprocess failed: {status}"); + return false; + } + if Instant::now() >= deadline { + child.kill().unwrap(); + child.wait().unwrap(); + panic!("execution/cleanup blocked behind process stderr lock"); + } + std::thread::sleep(Duration::from_millis(10)); + } +} + +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 6a545b9c..80d5ab5b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -31,6 +31,7 @@ pub mod plugins; pub(crate) mod process_tree; pub mod protocols; pub mod provider; +mod runlet_progress; pub mod runtime; /// Shared internal filesystem and process-lifetime recovery controls. pub mod resilient_fs { diff --git a/src/runlet_progress.rs b/src/runlet_progress.rs new file mode 100644 index 00000000..9b66bd22 --- /dev/null +++ b/src/runlet_progress.rs @@ -0,0 +1,270 @@ +//! 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; +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. +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/runlet_progress/transport.rs b/src/runlet_progress/transport.rs new file mode 100644 index 00000000..f7aeb1e4 --- /dev/null +++ b/src/runlet_progress/transport.rs @@ -0,0 +1,205 @@ +//! 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}, + sync::{ + Arc, OnceLock, + atomic::{AtomicBool, Ordering}, + mpsc::{self, Receiver, RecvTimeoutError, SyncSender}, + }, + time::{Duration, Instant}, +}; + +const CAPACITY: usize = 256; +const MAX_FRAME_BYTES: usize = 16 * 1024; +const HEARTBEAT: Duration = Duration::from_millis(500); +pub(crate) const LEASE: Duration = Duration::from_secs(5); + +enum Frame { + Authoritative(Vec), + Diagnostic(Vec), +} + +#[derive(Clone)] +pub(crate) struct Transport { + sender: SyncSender, + disabled: Arc, + diagnostics_lost: Arc, +} +impl Transport { + /// A concrete owned IO boundary; production uses stderr, tests may use a pipe. + pub(crate) fn start( + writer: impl Write + Send + 'static, + capacity: usize, + runtime_events: bool, + ) -> io::Result { + let (sender, receiver) = mpsc::sync_channel(capacity); + let disabled = Arc::new(AtomicBool::new(false)); + let guard = WorkerGuard(disabled.clone()); + let diagnostics_lost = Arc::new(AtomicBool::new(false)); + let worker_loss = diagnostics_lost.clone(); + std::thread::Builder::new() + .name("runlet-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. + let _guard = guard; + let _ = write_loop(writer, receiver, &_guard.0, &worker_loss, runtime_events); + })?; + Ok(Self { + sender, + disabled, + 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]"; + let mut frame = Vec::with_capacity(line.len().saturating_add(1).min(MAX_FRAME_BYTES)); + if line.len() >= MAX_FRAME_BYTES { + let end = line.floor_char_boundary(MAX_FRAME_BYTES - TRUNCATED.len() - 1); + frame.extend_from_slice(&line.as_bytes()[..end]); + frame.extend_from_slice(TRUNCATED.as_bytes()); + } else { + frame.extend_from_slice(line.as_bytes()); + } + frame.push(b'\n'); + if self.sender.try_send(Frame::Diagnostic(frame)).is_err() { + self.diagnostics_lost.store(true, Ordering::Release); + } + } + + /// Forwarded lifecycle frames must never silently cross a loss gap. + pub(crate) fn publish_runtime_line(&self, line: &str) { + if self.disabled.load(Ordering::Acquire) { + return; + } + if line.len() >= MAX_FRAME_BYTES { + self.disabled.store(true, Ordering::Release); + return; + } + let mut frame = Vec::with_capacity(line.len() + 1); + frame.extend_from_slice(line.as_bytes()); + frame.push(b'\n'); + self.enqueue_authoritative(frame); + } + + fn enqueue_authoritative(&self, frame: Vec) { + if self.sender.try_send(Frame::Authoritative(frame)).is_err() { + self.disabled.store(true, Ordering::Release); + } + } + + /// Loss of any lifecycle frame invalidates authoritative observation too. + /// Never wait for the sink, queue capacity, or a reset acknowledgement. + pub(crate) fn publish_event(&self, event: &RuntimeEvent) { + if self.disabled.load(Ordering::Acquire) { + return; + } + match encode_frame(event) { + Ok(frame) => self.enqueue_authoritative(frame), + Err(_) => self.disabled.store(true, Ordering::Release), + } + } +} +struct WorkerGuard(Arc); +impl Drop for WorkerGuard { + fn drop(&mut self) { + self.0.store(true, Ordering::Release); + } +} + +pub(crate) fn global() -> Option<&'static Transport> { + static TRANSPORT: OnceLock> = OnceLock::new(); + TRANSPORT + .get_or_init(|| { + Transport::start(std::io::stderr(), CAPACITY, crate::events::enabled()).ok() + }) + .as_ref() +} + +fn write_loop( + mut writer: impl Write, + receiver: Receiver, + disabled: &AtomicBool, + diagnostics_lost: &AtomicBool, + runtime_events: bool, +) -> io::Result<()> { + let mut heartbeat = Instant::now(); + transport_status(&mut writer, runtime_events, true)?; + let mut reset_sent = false; + loop { + if disabled.load(Ordering::Acquire) && !reset_sent { + // A blocked or failed reset is covered by the client lease. + transport_status(&mut writer, runtime_events, false)?; + reset_sent = true; + } + if diagnostics_lost.swap(false, Ordering::AcqRel) { + writer.write_all(b"kit: some child diagnostics were dropped\n")?; + } + match receiver.recv_timeout(HEARTBEAT) { + Ok(frame) => { + let bytes = match frame { + Frame::Authoritative(_) if disabled.load(Ordering::Acquire) => continue, + Frame::Authoritative(bytes) | Frame::Diagnostic(bytes) => bytes, + }; + writer.write_all(&bytes)?; + // Busy legacy 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(); + } + } + Err(RecvTimeoutError::Timeout) => { + if disabled.load(Ordering::Acquire) { + continue; + } + transport_status(&mut writer, runtime_events, true)?; + heartbeat = Instant::now(); + } + Err(RecvTimeoutError::Disconnected) => { + return transport_status(&mut writer, runtime_events && !reset_sent, false); + } + } + } +} + +fn transport_status(writer: &mut impl Write, enabled: bool, available: bool) -> io::Result<()> { + if enabled { + write_frame(writer, &RuntimeEvent::RunletTransport { available })?; + } + Ok(()) +} + +fn write_frame(writer: &mut impl Write, event: &RuntimeEvent) -> io::Result<()> { + // One whole-frame stderr lock, owned only by the detached worker. + writer.write_all(&encode_frame(event)?) +} + +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 = { + let mut cursor = io::Cursor::new(frame.as_mut_slice()); + cursor.write_all(EVENT_MARKER.as_bytes())?; + serde_json::to_writer(&mut cursor, event).map_err(io::Error::other)?; + cursor.write_all(b"\n")?; + cursor.position() as usize + }; + frame.truncate(len); + Ok(frame) +} + +#[cfg(test)] +mod tests; diff --git a/src/runlet_progress/transport/tests.rs b/src/runlet_progress/transport/tests.rs new file mode 100644 index 00000000..11a161ce --- /dev/null +++ b/src/runlet_progress/transport/tests.rs @@ -0,0 +1,257 @@ +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + 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() { + use std::os::unix::net::UnixStream; + let (mut writer, reader) = UnixStream::pair().unwrap(); + // Fill a real pipe before handing it to the production writer. Nonblocking + // is confined to this owned test socket, never the process stderr flags. + 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() == io::ErrorKind::WouldBlock => break, + Err(e) => panic!("fill: {e}"), + } + } + 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(), + ok: true, + summary: "done".into(), + millis: 1, + }); + transport.publish(progress()); + } + assert!(transport.disabled.load(Ordering::Acquire)); + reader + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut reader = BufReader::new(reader); + let mut initial = vec![0; filled]; + reader.read_exact(&mut initial).unwrap(); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!( + crate::events::parse(line.trim_end()), + Some(RuntimeEvent::RunletTransport { available: true }) + ); + line.clear(); + reader.read_line(&mut line).unwrap(); + assert_eq!( + crate::events::parse(line.trim_end()), + Some(RuntimeEvent::RunletTransport { available: false }) + ); + transport.publish(progress()); + drop(transport); + line.clear(); + assert_eq!(reader.read_line(&mut line).unwrap(), 0); +} + +#[test] +fn writer_error_and_unwind_fail_closed() { + struct Broken(bool); + impl Write for Broken { + fn write(&mut self, _: &[u8]) -> io::Result { + if self.0 { + panic!("external sink panic"); + } + Err(io::Error::new(io::ErrorKind::BrokenPipe, "disconnected")) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + for panic in [false, true] { + let disabled = Arc::new(AtomicBool::new(false)); + let (sender, receiver) = mpsc::sync_channel(2); + let copy = disabled.clone(); + let worker = std::thread::spawn(move || { + let _guard = WorkerGuard(copy); + write_loop( + Broken(panic), + receiver, + &_guard.0, + &AtomicBool::new(false), + true, + ) + }); + let outcome = worker.join(); + assert!(if panic { + outcome.is_err() + } else { + outcome.unwrap().is_err() + }); + let transport = Transport { + sender, + disabled, + diagnostics_lost: Arc::new(AtomicBool::new(false)), + }; + assert!(transport.disabled.load(Ordering::Acquire)); + transport.publish(progress()); + } +} + +#[cfg(unix)] +#[test] +fn last_sender_disconnect_finishes_transport() { + let (writer, mut reader) = std::os::unix::net::UnixStream::pair().unwrap(); + reader + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let transport = Transport::start(writer, 2, true).unwrap(); + drop(transport); + let mut text = String::new(); + reader.read_to_string(&mut text).unwrap(); + assert!(text.lines().any(|line| crate::events::parse(line) + == Some(RuntimeEvent::RunletTransport { available: false }))); +} + +#[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(), + ok: true, + summary: "done".into(), + millis: 1, + }; + let (writer, mut reader) = std::os::unix::net::UnixStream::pair().unwrap(); + reader + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let transport = Transport::start(writer, 2, true).unwrap(); + transport.publish_event(&event); + drop(transport); + let mut wire = String::new(); + reader.read_to_string(&mut wire).unwrap(); + assert!( + wire.lines() + .any(|line| crate::events::parse(line) == Some(event.clone())) + ); + + let (writer, mut reader) = std::os::unix::net::UnixStream::pair().unwrap(); + reader + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let transport = Transport::start(writer, 2, true).unwrap(); + transport.publish_event(&RuntimeEvent::SessionStarted { + session_id: "x".repeat(MAX_FRAME_BYTES), + }); + assert!(transport.disabled.load(Ordering::Acquire)); + transport.publish(progress()); + transport.publish_line("later child error"); + drop(transport); + wire.clear(); + reader.read_to_string(&mut wire).unwrap(); + assert!(wire.contains("later child error\n")); + let events: Vec<_> = wire.lines().filter_map(crate::events::parse).collect(); + assert_eq!( + events.last(), + Some(&RuntimeEvent::RunletTransport { available: false }) + ); + assert!( + events + .iter() + .all(|event| matches!(event, RuntimeEvent::RunletTransport { .. })) + ); +} + +#[cfg(unix)] +#[test] +fn plain_diagnostics_do_not_enable_runtime_frames() { + let (writer, mut reader) = std::os::unix::net::UnixStream::pair().unwrap(); + reader + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let transport = Transport::start(writer, 2, false).unwrap(); + transport.publish_line("child diagnostic"); + drop(transport); + let mut wire = String::new(); + reader.read_to_string(&mut wire).unwrap(); + assert_eq!(wire, "child diagnostic\n"); +} + +#[cfg(unix)] +#[test] +fn oversized_diagnostics_truncate_without_disabling_later_errors() { + for runtime_events in [false, true] { + let (writer, reader) = std::os::unix::net::UnixStream::pair().unwrap(); + reader + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let transport = Transport::start(writer, 4, runtime_events).unwrap(); + transport.publish_line(&"Ć©".repeat(MAX_FRAME_BYTES)); + let mut reader = BufReader::new(reader); + let mut line = String::new(); + loop { + line.clear(); + reader.read_line(&mut line).unwrap(); + if crate::events::parse(line.trim_end()).is_none() { + break; + } + } + assert!(line.ends_with(" [truncated]\n")); + assert!(line.len() <= MAX_FRAME_BYTES); + assert!(!transport.disabled.load(Ordering::Acquire)); + transport.publish_line("later child error"); + drop(transport); + let mut rest = String::new(); + reader.read_to_string(&mut rest).unwrap(); + assert!(rest.contains("later child error\n")); + } +} + +#[test] +fn diagnostic_queue_overflow_does_not_poison_recovered_publication() { + // Exercise actual bounded admission at a full queue, then release capacity. + // No sink timing or exact implementation work counts are involved. + let (sender, receiver) = mpsc::sync_channel(1); + let transport = Transport { + sender, + disabled: Arc::new(AtomicBool::new(false)), + diagnostics_lost: Arc::new(AtomicBool::new(false)), + }; + transport.publish_line("first"); + transport.publish_line("dropped"); + assert!(!transport.disabled.load(Ordering::Acquire)); + assert!(transport.diagnostics_lost.load(Ordering::Acquire)); + assert!(matches!(receiver.recv().unwrap(), Frame::Diagnostic(_))); + transport.publish_line("later child error"); + let disabled = transport.disabled.clone(); + let diagnostics_lost = transport.diagnostics_lost.clone(); + drop(transport); + let mut wire = Vec::new(); + write_loop(&mut wire, receiver, &disabled, &diagnostics_lost, false).unwrap(); + let wire = String::from_utf8(wire).unwrap(); + assert!(wire.contains("some child diagnostics were dropped\n")); + assert!(wire.contains("later child error\n")); + assert!(!wire.contains(EVENT_MARKER)); +} diff --git a/src/runtime.rs b/src/runtime.rs index 0d071139..19286c2c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -2532,7 +2532,11 @@ impl ComposeBackend for HiddenRunletBackend { async fn execute(&self, mut run: BackendRun) -> Result { run.visible_specs = self.specs(); - RunletBackend.execute(run).await + if crate::events::enabled() { + crate::runlet_progress::execute(run).await + } else { + RunletBackend.execute(run).await + } } } diff --git a/src/tools/observed.rs b/src/tools/observed.rs index c2877c3e..16da1bea 100644 --- a/src/tools/observed.rs +++ b/src/tools/observed.rs @@ -260,8 +260,66 @@ mod tests { } } + #[tokio::test] + async fn completion_with_stalled_stderr() { + if !crate::events::test_support::with_stalled_stderr( + "tools::observed::tests::completion_with_stalled_stderr", + ) { + return; + } + preserve_native_outcomes().await; + let tool = Observed::new(DirectTool { + spec: ToolSpec::new(ToolName::new("direct"), "direct", json!({})), + }); + let context = OwnedToolContext { + session_id: SessionId::new("session"), + turn_id: TurnId::new("turn"), + metadata: MetadataMap::new(), + permissions: Arc::new(AllowAllPermissions), + resources: Arc::new(()), + cancellation: None, + execution_scope: None, + approved_request: None, + }; + let request = ToolRequest::new( + ToolCallId::new("call"), + ToolName::new("direct"), + json!({}), + context.session_id.clone(), + context.turn_id.clone(), + ); + let result = tool.invoke(request, &mut context.borrowed()).await.unwrap(); + assert_eq!(result.result.output, ToolOutput::text("done")); + } + + struct DirectTool { + spec: ToolSpec, + } + + #[async_trait] + impl Tool for DirectTool { + fn spec(&self) -> &ToolSpec { + &self.spec + } + + async fn invoke( + &self, + request: ToolRequest, + _: &mut ToolContext<'_>, + ) -> Result { + Ok(ToolResult::new(ToolResultPart::success( + request.call_id, + ToolOutput::text("done"), + ))) + } + } + #[tokio::test] async fn both_wrappers_preserve_native_outcomes() { + preserve_native_outcomes().await; + } + + async fn preserve_native_outcomes() { for mode in [ Mode::Completed, Mode::Failed, diff --git a/src/tui/app.rs b/src/tui/app.rs index b4868de2..60835c19 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -358,6 +358,7 @@ 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, @@ -401,6 +402,7 @@ 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; @@ -587,6 +589,8 @@ pub struct AgentCounts { } pub struct App { + progress_last_frame: Option, + progress_unavailable: bool, pub root: PathBuf, pub provider: String, pub model: String, @@ -836,6 +840,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, root, provider, model, @@ -1127,9 +1133,10 @@ impl App { } } - /// Whether the periodic animation clock can change anything on screen. + /// Whether periodic polling must advance animations or expire a runtime lease. pub fn needs_redraw_tick(&self) -> bool { - self.working() + (!self.progress_unavailable && self.progress_last_frame.is_some()) + || self.working() || !self.transcript_dynamic.is_empty() || self.toast.is_some() || self.agents.values().any(|row| match row.status { @@ -1146,6 +1153,7 @@ impl App { /// Advances animations and removes expired transient state. pub fn tick(&mut self) { + self.progress_tick_at(Instant::now()); self.tick_at(crate::events::now_millis()); } @@ -1706,7 +1714,8 @@ impl App { status: ToolCallStatus::Pending, started: Instant::now(), finished: None, - script: script.unwrap_or_default(), + script: crate::runlet_progress::bounded_source(script.unwrap_or_default()), + progress: Box::default(), children: Vec::new(), output: Vec::new(), intent: None, @@ -1757,6 +1766,10 @@ 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(); + } call.script = script; } if let Some(output) = output { @@ -1813,6 +1826,9 @@ impl App { _ => {} }, Update::ProcessExited(error) => { + // Confirmed process exit can retire known roster rows. A mere + // diagnostic gap cannot claim those terminal outcomes. + self.invalidate_runtime_status(); self.finish_turn_with_outcome(false, None); self.retire_active_agents_at(crate::events::now_millis()); self.push_block(Block::Error(error)); @@ -1823,13 +1839,79 @@ impl App { } } + fn disable_runtime(&mut self) { + self.agents.clear(); + self.invalidate_runtime_status(); + } + + fn invalidate_runtime_status(&mut self) { + if self.progress_unavailable { + return; + } + self.progress_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(); + self.cleaned_agent_ids.clear(); + self.cleaned_agent_ancestors.clear(); + self.agents_scroll = 0; + self.runtime_session_id = None; + 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 + } + + /// 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 + }) + { + self.disable_runtime(); + } + } + fn progress_activity(&mut self) { + let now = Instant::now(); + self.progress_tick_at(now); + if !self.progress_unavailable { + self.progress_last_frame = Some(now); + } + } + fn apply_runtime(&mut self, event: RuntimeEvent) { self.apply_runtime_at(event, crate::events::now_millis()); } 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(); + if self.runtime_unavailable() { + return; + } let parent = event.parent_call().map(str::to_string); let owner_id = match event { + RuntimeEvent::RunletTransport { available } => { + if available { + self.progress_activity(); + } else { + self.disable_runtime(); + } + return; + } RuntimeEvent::StorageStatus { pending, exhausted } => { self.storage_pending = pending; self.storage_exhausted = exhausted; @@ -1943,6 +2025,49 @@ impl App { 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, @@ -6285,6 +6410,51 @@ mod tests { ); } + #[test] + fn idle_runtime_lease_keeps_timer_scheduled_until_state_is_invalidated() { + use crate::events::{GenerationOutcome, SubagentStatus}; + + let mut app = app(); + assert!(!app.needs_redraw_tick()); + app.apply(Update::Runtime(RuntimeEvent::RunletTransport { + available: true, + })); + app.apply(Update::Runtime(agent_event( + "idle", + "Completed worker", + SubagentStatus::Idle, + Some(GenerationOutcome::Success), + 1, + None, + (10, 20, Some(30)), + ))); + app.apply(Update::Runtime(RuntimeEvent::StorageStatus { + pending: true, + exhausted: true, + })); + assert!(!app.working()); + assert!(app.transcript_dynamic.is_empty()); + assert!(app.toast.is_none()); + assert!(!app.runtime_unavailable()); + assert_eq!(app.agent_counts().total, 1); + assert!(app.needs_redraw_tick()); + app.tick(); + assert!(!app.runtime_unavailable()); + assert_eq!(app.agent_counts().total, 1); + + // 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); + assert!(app.needs_redraw_tick()); + if app.needs_redraw_tick() { + app.tick(); + } + assert!(app.runtime_unavailable()); + assert_eq!(app.agent_counts().total, 0); + assert!(!app.storage_pending && !app.storage_exhausted); + assert!(!app.needs_redraw_tick()); + } + #[test] fn redraw_ticks_only_while_time_dependent_ui_is_visible() { let mut app = app(); @@ -6574,7 +6744,8 @@ mod tests { assert!(app.needs_redraw_tick()); app.tick_at(5_000); assert!(!app.agents.contains_key("failed")); - assert!(!app.needs_redraw_tick()); + // Runtime traffic established a lease even after the animation ends. + assert!(app.needs_redraw_tick()); } #[test] diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 6bfa3947..ca5553a7 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -11,6 +11,7 @@ mod command; mod editor; mod image; mod markdown; +mod progress; mod theme; mod ui; mod wrap; diff --git a/src/tui/progress.rs b/src/tui/progress.rs new file mode 100644 index 00000000..5c341a19 --- /dev/null +++ b/src/tui/progress.rs @@ -0,0 +1,223 @@ +//! 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 new file mode 100644 index 00000000..a536e1bb --- /dev/null +++ b/src/tui/progress_tests.rs @@ -0,0 +1,1080 @@ +// 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, + 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, + 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, + 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, + 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, + 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, + 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, + 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, + 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, + 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 @")); + } +} + +#[test] +fn authoritative_progress_reset_and_expired_lease_cannot_be_revived() { + for explicit in [true, false] { + let mut app = sample(); + 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 }); + 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, + 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(); + 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(), + 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, + }, + RuntimeEvent::RunletTransport { available: true }, + agent, + ] { + 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")); + } +} diff --git a/src/tui/ui.rs b/src/tui/ui.rs index c11c01ba..2a97dfee 100644 --- a/src/tui/ui.rs +++ b/src/tui/ui.rs @@ -131,9 +131,11 @@ pub fn draw(frame: &mut Frame<'_>, app: &mut App, images: &mut ImageRuntime) { } // Durability stays visible on the start screen and over session pickers. // Pending data belongs to the process, not the currently selected session. - if app.storage_pending || app.storage_exhausted { + if app.runtime_unavailable() || app.storage_pending || app.storage_exhausted { let area = frame.area(); - let warning = if app.storage_exhausted { + let warning = if app.runtime_unavailable() { + " Runtime status unavailable: agent, child, compaction and storage state unknown" + } else if app.storage_exhausted { " Storage exhausted: shutting down; unpersisted data is at risk" } else { " Memory-only storage: awaiting disk recovery; data at risk on exit" @@ -1348,18 +1350,36 @@ fn tool_lines(app: &App, call: &ToolCall, active: bool) -> Vec> { lines } -/// Source text is not execution telemetry. Runtime events identify calls, -/// not source expressions, so script lines carry no inferred state. +/// Source lines stay neutral. Exact runtime call spans add qualified counts at +/// their source start line; annotations never assert whole-line/binding state. fn script_lines(call: &ToolCall) -> Vec> { - call.script + let annotations = call.progress.labels(&call.script); + let mut lines: Vec<_> = call + .script .lines() - .map(|source| { - Line::from(vec![ + .take(MAX_OUTPUT_ROWS) + .enumerate() + .map(|(index, source)| { + let mut 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() + .collect(); + let count = call.script.lines().count(); + if count > MAX_OUTPUT_ROWS { + lines.push(Line::from(Span::styled( + format!(" ā”‚ ā€¦ {} more lines", count - MAX_OUTPUT_ROWS), + theme::faint(), + ))); + } + lines } fn completed_compose_lines(call: &ToolCall) -> Vec> { @@ -1418,19 +1438,7 @@ fn completed_compose_lines(call: &ToolCall) -> Vec> { match call.compose_view { ComposeView::Output => lines.extend(expanded_output_lines(call)), ComposeView::Script => { - let count = call.script.lines().count(); - lines.extend(call.script.lines().take(MAX_OUTPUT_ROWS).map(|source| { - Line::from(vec![ - Span::styled(" ā”‚ ", theme::faint()), - Span::styled(source.to_string(), theme::dim()), - ]) - })); - if count > MAX_OUTPUT_ROWS { - lines.push(Line::from(Span::styled( - format!(" ā”‚ ā€¦ {} more lines", count - MAX_OUTPUT_ROWS), - theme::faint(), - ))); - } + lines.extend(script_lines(call)); } } lines @@ -2754,6 +2762,8 @@ mod tests { assert_eq!(right, 79); } + include!("progress_tests.rs"); + const SCRIPT: &str = "files = shell({ command: \"ls src\" })\n\ checked = for file in files.lines {\n\ return shell({ command: \"cargo check\" })\n\