From 1be112030e78b32e3ecc87c2fcc6173aca12f5b7 Mon Sep 17 00:00:00 2001 From: daniel Date: Sun, 13 Sep 2026 18:05:01 +0100 Subject: [PATCH 1/2] fix(runtime): increase publication queue capacity --- src/runlet_progress/transport.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/runlet_progress/transport.rs b/src/runlet_progress/transport.rs index f7aeb1e4..7dac31c7 100644 --- a/src/runlet_progress/transport.rs +++ b/src/runlet_progress/transport.rs @@ -12,7 +12,7 @@ use std::{ time::{Duration, Instant}, }; -const CAPACITY: usize = 256; +const CAPACITY: usize = 1024; const MAX_FRAME_BYTES: usize = 16 * 1024; const HEARTBEAT: Duration = Duration::from_millis(500); pub(crate) const LEASE: Duration = Duration::from_secs(5); From 422e50bf1460f67b35361a9bec0d0a2e2cbccac5 Mon Sep 17 00:00:00 2001 From: daniel Date: Sun, 13 Sep 2026 22:19:02 +0100 Subject: [PATCH 2/2] fix(runtime)!: recover scoped observation streams after loss BREAKING CHANGE: Runtime side-channel events now use source-scoped epoch envelopes. Older readers cannot render these payloads; mixed-version loss remains fail-closed. --- docs/runtime-authority.md | 104 ++++ src/acp_child.rs | 279 +++++++++-- src/events.rs | 131 ++++- src/runlet_progress.rs | 1 + src/runlet_progress/authority.rs | 553 +++++++++++++++++++++ src/runlet_progress/transport.rs | 309 ++++++++++-- src/runlet_progress/transport/tests.rs | 645 +++++++++++++++++++------ src/tui/app.rs | 152 +++--- src/tui/progress_tests.rs | 499 +++++++++++++++++-- src/tui/ui.rs | 25 +- 10 files changed, 2329 insertions(+), 369 deletions(-) create mode 100644 docs/runtime-authority.md create mode 100644 src/runlet_progress/authority.rs diff --git a/docs/runtime-authority.md b/docs/runtime-authority.md new file mode 100644 index 00000000..42d95f58 --- /dev/null +++ b/docs/runtime-authority.md @@ -0,0 +1,104 @@ +# Runtime side-channel authority + +Runtime events are private, best-effort stderr observations, not a snapshot, +replay log, or proof of current remote state. This protocol does not change ACP. + +## Wire contract + +New publishers use two marked JSON event shapes: + +- `runtime_boundary`: `source` (a random, worker-lifetime identity), `epoch` + (a monotonically increasing unsigned integer), and `state`: `open`, + `heartbeat`, or `lost`. +- `runtime_scoped`: `source`, `epoch`, and `payload` (another runtime event). + +A worker initially writes `open` at epoch 0. All its authoritative payloads, +including session attachment markers and forwarded events, name its epoch. +Forwarding wraps rather than replaces the original event: the complete chain of +containing publishers remains a prerequisite for descendant authority. Source +identities are not authentication credentials; stderr is owned by the child. + +An accepted `heartbeat` or scoped payload renews an already-live epoch's +five-second receive lease. It never opens an epoch. After explicit loss or local +expiry, same-epoch heartbeats, repeated opens, and payloads are rejected. Recovery +requires an explicit strictly newer `open`, or a newly owned replacement stream. +An ancestor heartbeat cannot restore a descendant. Ancestor loss also +invalidates its dependent descendants; each needs its own newer open. + +The writer attempts a healthy epoch rotation every two seconds, even without +queue overflow. Healthy continuous rotation preserves observations. It does not +clear or reconstruct state. This gives receivers which expired locally a way to +recover without a reverse channel when IO and publication claims make progress. +A different source ID is not, by itself, an exclusive replacement operation: +actual stream replacement must replace the receiver owner and its tracker. + +## Publication ordering and failure + +The queue remains **1024 frames**. Serialization writes into a bounded 16 KiB +buffer; provenance is limited to 16 wrappers. Authoritative publication does not +wait for IO, queue capacity, or a reset acknowledgement. Diagnostics are +best-effort and do not renew or repair authoritative state. + +Producer tokens are captured before encoding. Queue overflow or invalid encoding +closes exactly that token's admission generation using compare-and-exchange. +A delayed failure cannot close a newer generation. Queued frames carrying an +invalidated token are discarded, including a retained token enqueued after +reopening. A frame already selected by the single writer may finish, but its +write is strictly before `lost` and the replacement `open`. New-generation +admission opens only after the complete replacement boundary write succeeds. + +For healthy periodic rotation, producers hold a nonblocking shared publication +claim. The writer tries an exclusive claim, drains previously admitted frames +before the new open, and then advances the epoch. The drain is bounded by queue +capacity. Publication attempted while that claim is held records loss +separately, so a dropped observation cannot disappear in the epoch commit. +That loss is reported by the writer (possibly after the just-written open if it +raced the boundary). Progress requires eventual acquisition of the exclusive +claim; no scheduling fairness bound is claimed. + +Only the writer performs IO. IO error, partial-write error, unwind, epoch +exhaustion, and sender disconnect retire that worker; they cannot restart it +under a reused identity. The worker's terminal guard runs on all exit paths. +The publication-claim lock contains no mutable snapshot. Producers never block +on it, and poison is not recovered. No execution future joins the writer; +cancellation and dropping publishers cannot wait for a blocked stderr sink. + +## Recovery and compatibility limits + +Loss clears retained observations conservatively. Subsequent accepted events +can describe future activity, but an open does not reconstruct lost agent, +child, compaction, storage, or progress snapshots. Existing progress tombstones +and session identity filtering still apply. Loss of one authority may clear +more UI state than necessary; it does not authorize another authority's backlog. +A persistent incomplete-status warning distinguishes accepted future observations +from complete status: another publisher's healthy open cannot hide a descendant's +ongoing loss. A fresh visible session clears this historical warning, not the +owned stream's epoch tombstones. If an attachment marker was itself lost, the +receiver requires a fresh accepted marker; it does not guess session ownership. + +Legacy boolean status lacks provenance and epochs. Its loss/expiry is latched +for its owned stream: an unscoped `true` is not a recovery certificate. Mixed +legacy/scoped input is treated conservatively, not promoted to scoped authority. +Old readers do not recognize new wrappers and therefore cannot render their +payloads; the protocol intentionally fails closed rather than dual-emitting +unscoped copies. Existing legacy event shapes remain parseable. The side channel +is ephemeral, so no persistent artifact rewrite or migration is performed. + +**Freshness is an ordering contract, not an absolute wall-clock guarantee.** A +receiver rejects backlog from epochs it has invalidated. A newer explicit open +may itself have spent time in an OS pipe, forwarding queue, or receiver mailbox. +One-way buffered data cannot prove that this boundary was generated after the +receiver's local expiry, nor that the remote process is still alive at display +time. Buffered newer boundaries and their observations can consequently be +accepted until the next loss/lease expiry. No timestamp or heartbeat can remove +that physical-buffer limitation without an acknowledged challenge or an owned +stream replacement. Eventual recovery assumes IO drains, the runtime continues +executing, and fresh boundaries and observations can traverse every dependency. + +Receiver authority registries retain at most 4096 source identities per owned +stream. They never evict epoch tombstones to admit an unseen identity: at the +limit, new identities fail closed until the stream is replaced. This bounds +registry memory/work but is a deliberate long-lived-stream liveness limit. +The existing stderr line readers still assemble a full line before the parser's +64 KiB limit; arbitrary child output without a newline is not bounded by this +publisher protocol. This change does not claim a bounded physical receive path. diff --git a/src/acp_child.rs b/src/acp_child.rs index e4179ae6..fa112734 100644 --- a/src/acp_child.rs +++ b/src/acp_child.rs @@ -1241,6 +1241,8 @@ fn harness_diagnostic(label: &str, line: &str) -> Option { | crate::events::RuntimeEvent::ChildFinished { .. } | crate::events::RuntimeEvent::RunletProgress { .. } | crate::events::RuntimeEvent::RunletTransport { .. } + | crate::events::RuntimeEvent::RuntimeBoundary { .. } + | crate::events::RuntimeEvent::RuntimeScoped { .. } ) ) { return None; @@ -1944,66 +1946,53 @@ async fn forward_stderr( .map(str::to_owned) .collect::>(); let mut lines = BufReader::new(stderr).lines(); - let mut deadline = None; - let mut unavailable = false; + let mut authority = crate::runlet_progress::authority::Authority::default(); 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(|_| ()) + for event in authority.expire(tokio::time::Instant::now().into_std()) { + output(ForwardedStderr::Cleanup(event)); + } + let next = if let Some(expires) = authority.deadline() { + match tokio::time::timeout_at(expires.into(), lines.next_line()).await { + Ok(next) => next, + Err(_) => continue, } } else { - Ok(lines.next_line().await) + 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; - } + Ok(Some(line)) => line, + _ => break, }; + for event in authority.expire(tokio::time::Instant::now().into_std()) { + output(ForwardedStderr::Cleanup(event)); + } if let Some(event) = crate::events::parse(&line) { - if unavailable { + let admission = authority.observe(&event, tokio::time::Instant::now().into_std()); + if !admission.accepted { 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); + let payload = crate::runlet_progress::authority::payload(&event); + if matches!( + payload, + crate::events::RuntimeEvent::RunletTransport { available: true } + ) { + continue; } - if event.forward_from_child() { + if payload.forward_from_child() + || matches!( + payload, + crate::events::RuntimeEvent::RuntimeBoundary { .. } + | crate::events::RuntimeEvent::RunletTransport { available: false } + ) + { if let crate::events::RuntimeEvent::SubagentStateChanged { parent_id: Some(parent_id), .. - } = event + } = payload { - ancestors.insert(parent_id); + ancestors.insert(parent_id.clone()); } - // Preserve recursively forwarded private events byte-for-byte. + // Publisher wraps the whole event, preserving nested provenance. output(ForwardedStderr::RuntimeLine(line)); continue; } @@ -2072,9 +2061,23 @@ fn prompt_outcome( clippy::disallowed_methods, clippy::disallowed_macros )] -mod test_support { +pub(crate) mod test_support { use super::*; + #[cfg(feature = "tui")] + /// Runs the real owned stderr receiver into a real parent publisher. + pub(crate) async fn forward_runtime( + reader: impl tokio::io::AsyncRead + Unpin, + parent: crate::runlet_progress::transport::Transport, + ) { + super::forward_stderr(reader, "acp.kit", None, |item| match item { + super::ForwardedStderr::RuntimeLine(line) => parent.publish_runtime_line(&line), + super::ForwardedStderr::Cleanup(event) => parent.publish_event(&event), + super::ForwardedStderr::Diagnostic(line) => parent.publish_line(&line), + }) + .await; + } + impl ChildSession { pub(crate) fn closure_probe_for_test() -> (Self, oneshot::Receiver<()>) { let (tx, mut rx) = mpsc::channel(1); @@ -4762,6 +4765,153 @@ for line in sys.stdin: mod forwards_subagent_events { use super::*; + #[tokio::test(start_paused = true)] + async fn nested_reader_emits_scoped_loss_despite_parent_heartbeats() { + use crate::events::{BoundaryState, EVENT_MARKER, RuntimeEvent}; + use tokio::io::AsyncWriteExt; + let (mut writer, reader) = tokio::io::duplex(8192); + 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 boundary = |source: &str, epoch, state| RuntimeEvent::RuntimeBoundary { + source: source.into(), + epoch, + state, + }; + let nested = |epoch, payload| RuntimeEvent::RuntimeScoped { + source: "parent".into(), + epoch, + payload: Box::new(payload), + }; + let line = |event: &RuntimeEvent| { + format!("{EVENT_MARKER}{}\n", serde_json::to_string(event).unwrap()) + }; + for event in [ + boundary("parent", 0, BoundaryState::Open), + nested(0, boundary("child", 0, BoundaryState::Open)), + ] { + writer.write_all(line(&event).as_bytes()).await.unwrap(); + assert_eq!( + rx.recv().await.unwrap(), + ForwardedStderr::RuntimeLine(line(&event).trim_end().into()) + ); + } + tokio::time::advance(crate::runlet_progress::transport::LEASE / 2).await; + let heartbeat = boundary("parent", 0, BoundaryState::Heartbeat); + writer.write_all(line(&heartbeat).as_bytes()).await.unwrap(); + assert!(matches!( + rx.recv().await.unwrap(), + ForwardedStderr::RuntimeLine(_) + )); + tokio::time::advance(crate::runlet_progress::transport::LEASE / 2).await; + assert_eq!( + rx.recv().await.unwrap(), + ForwardedStderr::Cleanup(nested(0, boundary("child", 0, BoundaryState::Lost))) + ); + // Healthy containing epoch rotation cannot revive its expired child. + for event in [ + boundary("parent", 2, BoundaryState::Open), + nested(2, boundary("child", 0, BoundaryState::Heartbeat)), + nested(2, boundary("child", 0, BoundaryState::Open)), + nested(2, boundary("child", 2, BoundaryState::Open)), + ] { + writer.write_all(line(&event).as_bytes()).await.unwrap(); + } + for expected in [ + boundary("parent", 2, BoundaryState::Open), + nested(2, boundary("child", 2, BoundaryState::Open)), + ] { + assert_eq!( + rx.recv().await.unwrap(), + ForwardedStderr::RuntimeLine(line(&expected).trim_end().into()) + ); + } + drop(writer); + forward.await.unwrap(); + assert!(rx.recv().await.is_none()); + } + + #[tokio::test(start_paused = true)] + async fn scoped_reader_expires_backlog_and_recovers_only_new_open() { + use crate::events::{BoundaryState, EVENT_MARKER, RuntimeEvent}; + use tokio::io::AsyncWriteExt; + let (mut writer, reader) = tokio::io::duplex(8192); + 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 boundary = |epoch, state| RuntimeEvent::RuntimeBoundary { + source: "child".into(), + epoch, + state, + }; + let started = |epoch| RuntimeEvent::RuntimeScoped { + source: "child".into(), + epoch, + payload: Box::new(RuntimeEvent::ChildStarted { + call: "parent:compose:0".into(), + tool: "shell".into(), + summary: "working".into(), + at: 1, + }), + }; + let line = |event: &RuntimeEvent| { + format!("{EVENT_MARKER}{}\n", serde_json::to_string(event).unwrap()) + }; + writer + .write_all( + format!( + "{}{}", + line(&boundary(0, BoundaryState::Open)), + line(&started(0)) + ) + .as_bytes(), + ) + .await + .unwrap(); + for _ in 0..2 { + assert!(matches!( + rx.recv().await.unwrap(), + ForwardedStderr::RuntimeLine(_) + )); + } + tokio::time::advance(crate::runlet_progress::transport::LEASE).await; + assert_eq!( + rx.recv().await.unwrap(), + ForwardedStderr::Cleanup(boundary(0, BoundaryState::Lost)) + ); + writer + .write_all( + format!( + "{}{}{}{}{}", + line(&boundary(0, BoundaryState::Heartbeat)), + line(&boundary(0, BoundaryState::Open)), + line(&started(0)), + line(&boundary(2, BoundaryState::Open)), + line(&started(2)) + ) + .as_bytes(), + ) + .await + .unwrap(); + for expected in [boundary(2, BoundaryState::Open), started(2)] { + let ForwardedStderr::RuntimeLine(actual) = rx.recv().await.unwrap() else { + panic!("expected scoped frame") + }; + assert_eq!(crate::events::parse(&actual), Some(expected)); + } + drop(writer); + forward.await.unwrap(); + assert!(rx.recv().await.is_none()); + } + #[tokio::test(start_paused = true)] async fn nested_transport_loss_preserves_diagnostics_not_stale_lifecycle() { use crate::events::{EVENT_MARKER, RuntimeEvent}; @@ -4798,13 +4948,13 @@ for line in sys.stdin: rx.recv().await.unwrap(), ForwardedStderr::RuntimeLine(_) )); + let reset_line = format!( + "{EVENT_MARKER}{}\n", + serde_json::to_string(&RuntimeEvent::RunletTransport { available: false }) + .unwrap() + ); 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(); + writer.write_all(reset_line.as_bytes()).await.unwrap(); } else { tokio::time::advance(crate::runlet_progress::transport::LEASE).await; } @@ -4814,8 +4964,16 @@ for line in sys.stdin: _ => panic!("expected nested invalidation"), }; assert_eq!(reset, RuntimeEvent::RunletTransport { available: false }); + // Neither a heartbeat nor an explicit reset/recovery pair has + // source identity. A healthy intermediate parent must not + // revive failed descendant evidence, including after expiry. writer - .write_all(format!("{heartbeat}{start}later child error\n").as_bytes()) + .write_all( + format!( + "{heartbeat}{start}{reset_line}{heartbeat}{start}later child error\n" + ) + .as_bytes(), + ) .await .unwrap(); assert!( @@ -4824,6 +4982,21 @@ for line in sys.stdin: drop(writer); forward.await.unwrap(); assert!(rx.try_recv().is_err()); + + // Loss belongs to the old stream, not the harness label or + // ancestor. A newly owned stream can forward fresh evidence; + // its heartbeat is still consumed locally, never promoted to + // a certificate for the parent's other descendants. + let fresh = format!("{heartbeat}{start}"); + let mut output = Vec::new(); + forward_stderr(fresh.as_bytes(), "acp.kit", None, |item| { + output.push(item); + }) + .await; + assert_eq!( + output, + vec![ForwardedStderr::RuntimeLine(start.trim_end().to_owned())] + ); } } diff --git a/src/events.rs b/src/events.rs index 363c5fa8..a0792b27 100644 --- a/src/events.rs +++ b/src/events.rs @@ -28,6 +28,15 @@ pub const EVENT_MARKER: &str = "\u{1}kit-runtime\u{1}"; /// Environment variable that turns emission on for a `serve` process. pub const EVENTS_ENV: &str = "KIT_RUNTIME_EVENTS"; +/// Scoped authority control. Only a newer `Open` can recover a lost lease. +#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum BoundaryState { + Open, + Heartbeat, + Lost, +} + /// One runtime event sent privately to the terminal client. /// /// `call` is the compose child call id, shaped `:compose:`, @@ -35,6 +44,18 @@ pub const EVENTS_ENV: &str = "KIT_RUNTIME_EVENTS"; #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] #[serde(tag = "event", rename_all = "snake_case")] pub enum RuntimeEvent { + /// A publisher-owned stream epoch, separate from all descendant publishers. + RuntimeBoundary { + source: String, + epoch: u64, + state: BoundaryState, + }, + /// Authority dependencies are retained recursively when a parent forwards. + RuntimeScoped { + source: String, + epoch: u64, + payload: Box, + }, /// Process-wide progress transport lease/reset, not a source execution. RunletTransport { available: bool }, /// Authoritative, value-free observations owned by an exact compose call. @@ -187,6 +208,29 @@ pub enum GenerationOutcome { } impl RuntimeEvent { + /// Bound recursive provenance and validate progress even inside wrappers. + pub(crate) fn valid_wire_payload(&self) -> bool { + let mut event = self; + for _ in 0..=16 { + match event { + Self::RuntimeScoped { + source, payload, .. + } => { + if source.is_empty() || source.len() > 64 { + return false; + } + event = payload; + } + Self::RuntimeBoundary { source, .. } => { + return !source.is_empty() && source.len() <= 64; + } + Self::RunletProgress { progress } => return progress.bounded(), + _ => return true, + } + } + false + } + /// Whether a parent Kit runtime should forward this child event unchanged. pub(crate) fn forward_from_child(&self) -> bool { matches!( @@ -206,7 +250,9 @@ impl RuntimeEvent { let call = match self { Self::RunletProgress { progress } => return Some(&progress.owner), Self::ChildStarted { call, .. } | Self::ChildFinished { call, .. } => call, - Self::RunletTransport { .. } + Self::RuntimeBoundary { .. } + | Self::RuntimeScoped { .. } + | Self::RunletTransport { .. } | Self::StorageStatus { .. } | Self::SessionStarted { .. } | Self::CompactionStarted { .. } @@ -245,11 +291,9 @@ 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()) + if !event.valid_wire_payload() + || (matches!(event, RuntimeEvent::RunletProgress { .. }) && body.len() > 4096) { return None; } @@ -335,8 +379,8 @@ mod tests { use serde_json::json; use super::{ - EVENT_MARKER, GenerationOutcome, HarnessVendor, RuntimeEvent, SubagentStatus, parse, - summarize_input, summarize_output, test_support::write_event, + BoundaryState, EVENT_MARKER, GenerationOutcome, HarnessVendor, RuntimeEvent, + SubagentStatus, parse, summarize_input, summarize_output, test_support::write_event, }; #[test] @@ -620,6 +664,79 @@ mod tests { ); } + #[test] + fn scoped_wire_roundtrips_without_promoting_legacy_status() { + for state in [ + BoundaryState::Open, + BoundaryState::Heartbeat, + BoundaryState::Lost, + ] { + let event = RuntimeEvent::RuntimeScoped { + source: "parent".into(), + epoch: 4, + payload: Box::new(RuntimeEvent::RuntimeBoundary { + source: "descendant".into(), + epoch: 2, + state, + }), + }; + let line = format!("{EVENT_MARKER}{}", serde_json::to_string(&event).unwrap()); + assert_eq!(parse(&line), Some(event)); + } + let legacy = format!( + "{EVENT_MARKER}{}", + r#"{"event":"runlet_transport","available":true}"# + ); + assert_eq!( + parse(&legacy), + Some(RuntimeEvent::RunletTransport { available: true }) + ); + } + + #[test] + fn scoped_wire_validates_nested_metadata_and_depth() { + let wrap = |payload| RuntimeEvent::RuntimeScoped { + source: "publisher".into(), + epoch: 0, + payload: Box::new(payload), + }; + let parse_event = |event: &RuntimeEvent| { + parse(&format!( + "{EVENT_MARKER}{}", + serde_json::to_string(event).unwrap() + )) + }; + let mut event = RuntimeEvent::StorageStatus { + pending: true, + exhausted: false, + }; + for _ in 0..16 { + event = wrap(event); + assert_eq!(parse_event(&event), Some(event.clone())); + } + assert!(parse_event(&wrap(event)).is_none()); + for source in [String::new(), "x".repeat(65)] { + assert!( + parse_event(&wrap(RuntimeEvent::RuntimeBoundary { + source, + epoch: 0, + state: BoundaryState::Open, + })) + .is_none() + ); + } + let malformed = format!( + "{EVENT_MARKER}{}", + r#"{"event":"runtime_boundary","source":"p","epoch":0,"state":"available"}"# + ); + assert!(parse(&malformed).is_none()); + let malformed = format!( + "{EVENT_MARKER}{}", + r#"{"event":"runtime_scoped","source":"p","epoch":0}"# + ); + assert!(parse(&malformed).is_none()); + } + #[test] fn ignores_ordinary_diagnostics() { assert!(parse("listening on 127.0.0.1:7331").is_none()); diff --git a/src/runlet_progress.rs b/src/runlet_progress.rs index 8f196993..67ea9fd8 100644 --- a/src/runlet_progress.rs +++ b/src/runlet_progress.rs @@ -1,6 +1,7 @@ //! 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 authority; pub(crate) mod transport; use agentkit_tool_compose::{ BackendRun, ComposeOutcome, RunletBackend, RunletProgress, RunletProgressEnd, diff --git a/src/runlet_progress/authority.rs b/src/runlet_progress/authority.rs new file mode 100644 index 00000000..21d1a810 --- /dev/null +++ b/src/runlet_progress/authority.rs @@ -0,0 +1,553 @@ +//! Receiver-owned authority. Transitions are synchronous and require no locks. +//! Source tombstones are never evicted: after MAX_SOURCES identities, unseen +//! publishers fail closed until the containing receiver/stream is replaced. +//! This deliberately bounds memory at the expense of long-lived stream liveness. +use std::{collections::BTreeMap, time::Instant}; + +use super::transport::LEASE; +use crate::events::{BoundaryState, RuntimeEvent}; + +const MAX_SOURCES: usize = 4096; + +#[derive(Default)] +pub(crate) struct Authority { + sources: BTreeMap, + legacy_deadline: Option, + legacy_lost: bool, + scoped_root: bool, +} +struct Scope { + epoch: u64, + live: bool, + legacy_tainted: bool, + deadline: Instant, + parents: Vec<(String, u64)>, +} +#[derive(Default)] +pub(crate) struct Admission { + pub accepted: bool, + #[cfg_attr(not(feature = "tui"), allow(dead_code))] + pub invalidated: bool, + #[cfg_attr(not(feature = "tui"), allow(dead_code))] + pub opened: bool, +} +impl Authority { + pub fn deadline(&self) -> Option { + self.sources + .values() + .filter(|s| s.live) + .map(|s| s.deadline) + .chain(self.legacy_deadline) + .min() + } + fn invalidate(&mut self, source: &str, epoch: u64) -> bool { + let mut changed = false; + for (id, scope) in &mut self.sources { + if (id == source && scope.epoch <= epoch) + || scope + .parents + .iter() + .any(|(id, generation)| id == source && *generation <= epoch) + { + changed |= scope.live; + scope.live = false; + } + } + changed + } + /// Lost outputs retain the exact dependency path of the expired authority. + pub fn expire(&mut self, now: Instant) -> Vec { + let expired: Vec<_> = self + .sources + .iter() + .filter(|(_, s)| s.live && now >= s.deadline) + .map(|(id, s)| (id.clone(), s.epoch, s.parents.clone())) + .collect(); + let mut outputs = Vec::new(); + for (source, epoch, parents) in expired { + self.invalidate(&source, epoch); + let mut event = RuntimeEvent::RuntimeBoundary { + source, + epoch, + state: BoundaryState::Lost, + }; + for (source, epoch) in parents.into_iter().rev() { + event = RuntimeEvent::RuntimeScoped { + source, + epoch, + payload: Box::new(event), + }; + } + outputs.push(event); + } + if self.legacy_deadline.is_some_and(|deadline| now >= deadline) { + self.legacy_deadline = None; + self.legacy_lost = true; + outputs.push(RuntimeEvent::RunletTransport { available: false }); + } + outputs + } + pub fn observe(&mut self, event: &RuntimeEvent, now: Instant) -> Admission { + self.visit(event, now, &mut Vec::new()) + } + fn visit( + &mut self, + event: &RuntimeEvent, + now: Instant, + parents: &mut Vec<(String, u64)>, + ) -> Admission { + match event { + RuntimeEvent::RuntimeScoped { + source, + epoch, + payload, + } => { + let Some(scope) = self.sources.get_mut(source) else { + return Admission::default(); + }; + if !scope.live || scope.epoch != *epoch || scope.parents != *parents { + return Admission::default(); + } + scope.deadline = now + LEASE; + parents.push((source.clone(), *epoch)); + self.visit(payload, now, parents) + } + RuntimeEvent::RuntimeBoundary { + source, + epoch, + state, + } => { + if (self.sources.len() >= MAX_SOURCES && !self.sources.contains_key(source)) + || self.sources.get(source).is_some_and(|s| s.legacy_tainted) + || parents.iter().any(|(id, _)| id == source) + { + return Admission::default(); + } + match state { + BoundaryState::Open => { + if self.sources.get(source).is_some_and(|s| *epoch <= s.epoch) { + return Admission::default(); + } + let continuous = self + .sources + .get(source) + .is_some_and(|s| s.live && s.parents == *parents); + let mut invalidated = if continuous { + // Healthy rotation preserves observations and rebases containment. + for scope in self.sources.values_mut() { + for (id, generation) in &mut scope.parents { + if id == source { + *generation = *epoch; + } + } + } + false + } else { + self.invalidate(source, *epoch) + }; + if parents.is_empty() { + // Once this wire speaks scoped authority, raw legacy + // backlog is never again admitted on the same receiver. + invalidated |= self.legacy_deadline.take().is_some(); + self.scoped_root = true; + } + self.sources.insert( + source.clone(), + Scope { + epoch: *epoch, + live: true, + legacy_tainted: false, + deadline: now + LEASE, + parents: parents.clone(), + }, + ); + Admission { + accepted: true, + invalidated, + opened: true, + } + } + BoundaryState::Heartbeat => { + let accepted = self.sources.get_mut(source).is_some_and(|s| { + if s.live && s.epoch == *epoch && s.parents == *parents { + s.deadline = now + LEASE; + true + } else { + false + } + }); + Admission { + accepted, + ..Admission::default() + } + } + BoundaryState::Lost => { + if self + .sources + .get(source) + .is_some_and(|s| *epoch < s.epoch || s.parents != *parents) + { + return Admission::default(); + } + let invalidated = self.invalidate(source, *epoch); + self.sources.insert( + source.clone(), + Scope { + epoch: *epoch, + live: false, + legacy_tainted: false, + deadline: now, + parents: parents.clone(), + }, + ); + Admission { + accepted: true, + invalidated, + opened: false, + } + } + } + } + RuntimeEvent::RunletTransport { available } if !parents.is_empty() => { + if *available { + // A legacy heartbeat cannot certify or recover any scope. + return Admission::default(); + } + let Some((source, epoch)) = parents.last() else { + return Admission::default(); + }; + let invalidated = self.invalidate(source, *epoch); + // No source provenance exists for this nested legacy loss. Latch + // the containing publisher, even across its healthy rotations. + if let Some(scope) = self.sources.get_mut(source) { + scope.legacy_tainted = true; + } + Admission { + accepted: true, + invalidated, + opened: false, + } + } + RuntimeEvent::RunletTransport { available } => { + if self.legacy_lost || self.scoped_root { + return Admission::default(); + } + if *available { + self.legacy_deadline = Some(now + LEASE); + } else { + self.legacy_lost = true; + self.legacy_deadline = None; + } + Admission { + accepted: true, + invalidated: !available, + opened: false, + } + } + _ => { + let accepted = !parents.is_empty() || (!self.legacy_lost && !self.scoped_root); + if accepted && parents.is_empty() { + self.legacy_deadline = Some(now + LEASE); + } + Admission { + accepted, + ..Admission::default() + } + } + } + } +} + +pub(crate) fn payload(event: &RuntimeEvent) -> &RuntimeEvent { + match event { + RuntimeEvent::RuntimeScoped { payload: inner, .. } => payload(inner), + _ => event, + } +} + +#[cfg(test)] +#[allow(clippy::disallowed_macros)] +mod tests { + use super::*; + fn boundary(source: &str, epoch: u64, state: BoundaryState) -> RuntimeEvent { + RuntimeEvent::RuntimeBoundary { + source: source.into(), + epoch, + state, + } + } + fn scoped(source: &str, epoch: u64, payload: RuntimeEvent) -> RuntimeEvent { + RuntimeEvent::RuntimeScoped { + source: source.into(), + epoch, + payload: Box::new(payload), + } + } + fn data() -> RuntimeEvent { + RuntimeEvent::StorageStatus { + pending: true, + exhausted: false, + } + } + #[test] + fn source_cap_never_evicts_invalidated_generations() { + let mut tracker = Authority::default(); + let now = Instant::now(); + for index in 0..MAX_SOURCES { + let source = format!("source-{index}"); + assert!( + tracker + .observe(&boundary(&source, 0, BoundaryState::Open), now) + .accepted + ); + tracker.observe(&boundary(&source, 0, BoundaryState::Lost), now); + } + assert!( + !tracker + .observe(&boundary("overflow", 0, BoundaryState::Open), now) + .accepted + ); + assert!( + !tracker + .observe(&boundary("source-0", 0, BoundaryState::Open), now) + .accepted + ); + assert!( + tracker + .observe(&boundary("source-0", 2, BoundaryState::Open), now) + .accepted + ); + assert!( + tracker + .observe(&scoped("source-0", 2, data()), now) + .accepted + ); + } + + #[test] + fn scoped_root_rejects_unscoped_backlog_and_legacy_heartbeats() { + let mut tracker = Authority::default(); + let now = Instant::now(); + assert!(tracker.observe(&data(), now).accepted); + assert!( + tracker + .observe(&boundary("root", 0, BoundaryState::Open), now) + .invalidated + ); + assert!(!tracker.observe(&data(), now).accepted); + assert!( + !tracker + .observe(&RuntimeEvent::RunletTransport { available: true }, now) + .accepted + ); + assert!( + !tracker + .observe(&RuntimeEvent::RunletTransport { available: false }, now) + .accepted + ); + assert!(tracker.observe(&scoped("root", 0, data()), now).accepted); + tracker.observe(&boundary("root", 0, BoundaryState::Lost), now); + assert!(!tracker.observe(&data(), now).accepted); + tracker.observe(&boundary("root", 2, BoundaryState::Open), now); + assert!(!tracker.observe(&data(), now).accepted); + assert!(tracker.observe(&scoped("root", 2, data()), now).accepted); + } + + #[test] + fn nested_legacy_loss_taints_containing_source_across_rotations() { + let mut tracker = Authority::default(); + let now = Instant::now(); + tracker.observe(&boundary("root", 0, BoundaryState::Open), now); + tracker.observe( + &scoped("root", 0, boundary("child", 0, BoundaryState::Open)), + now, + ); + let lost = scoped( + "root", + 0, + scoped( + "child", + 0, + RuntimeEvent::RunletTransport { available: false }, + ), + ); + assert!(tracker.observe(&lost, now).invalidated); + assert!( + tracker + .observe(&boundary("root", 2, BoundaryState::Open), now) + .accepted + ); + assert!( + !tracker + .observe( + &scoped("root", 2, boundary("child", 2, BoundaryState::Open)), + now + ) + .accepted + ); + assert!( + !tracker + .observe( + &scoped( + "root", + 2, + scoped( + "child", + 0, + RuntimeEvent::RunletTransport { available: true } + ) + ), + now + ) + .accepted + ); + assert!( + !tracker + .observe(&scoped("root", 2, scoped("child", 0, data())), now) + .accepted + ); + assert!(tracker.observe(&scoped("root", 2, data()), now).accepted); + } + + #[test] + fn rotation_preserves_descendants_but_loss_requires_new_child_open() { + let mut tracker = Authority::default(); + let now = Instant::now(); + assert!( + tracker + .observe(&boundary("parent", 0, BoundaryState::Open), now) + .accepted + ); + assert!( + tracker + .observe( + &scoped("parent", 0, boundary("child", 0, BoundaryState::Open)), + now + ) + .accepted + ); + let rotation = tracker.observe(&boundary("parent", 2, BoundaryState::Open), now); + assert!(rotation.accepted && !rotation.invalidated); + assert!( + tracker + .observe(&scoped("parent", 2, scoped("child", 0, data())), now) + .accepted + ); + assert!( + tracker + .observe(&boundary("parent", 2, BoundaryState::Lost), now) + .invalidated + ); + assert!( + tracker + .observe(&boundary("parent", 4, BoundaryState::Open), now) + .accepted + ); + assert!( + !tracker + .observe(&scoped("parent", 4, scoped("child", 0, data())), now) + .accepted + ); + assert!( + !tracker + .observe( + &scoped("parent", 4, boundary("child", 0, BoundaryState::Open)), + now + ) + .accepted + ); + assert!( + tracker + .observe( + &scoped("parent", 4, boundary("child", 2, BoundaryState::Open)), + now + ) + .accepted + ); + assert!( + tracker + .observe(&scoped("parent", 4, scoped("child", 2, data())), now) + .accepted + ); + } + #[test] + fn healthy_parent_cannot_renew_or_recover_expired_child() { + let mut tracker = Authority::default(); + let now = Instant::now(); + tracker.observe(&boundary("parent", 0, BoundaryState::Open), now); + tracker.observe( + &scoped("parent", 0, boundary("child", 0, BoundaryState::Open)), + now, + ); + tracker.observe( + &boundary("parent", 0, BoundaryState::Heartbeat), + now + LEASE / 2, + ); + let lost = tracker.expire(now + LEASE); + assert_eq!( + lost, + vec![scoped( + "parent", + 0, + boundary("child", 0, BoundaryState::Lost) + )] + ); + for state in [BoundaryState::Heartbeat, BoundaryState::Open] { + assert!( + !tracker + .observe( + &scoped("parent", 0, boundary("child", 0, state)), + now + LEASE + ) + .accepted + ); + } + assert!( + tracker + .observe(&boundary("parent", 2, BoundaryState::Open), now + LEASE) + .accepted + ); + assert!( + !tracker + .observe( + &scoped("parent", 2, scoped("child", 0, data())), + now + LEASE + ) + .accepted + ); + assert!( + tracker + .observe( + &scoped("parent", 2, boundary("child", 2, BoundaryState::Open)), + now + LEASE + ) + .accepted + ); + } + #[test] + fn legacy_loss_is_latched_but_explicit_scopes_can_recover() { + let mut tracker = Authority::default(); + let now = Instant::now(); + tracker.observe(&data(), now); + assert_eq!(tracker.expire(now + LEASE).len(), 1); + assert!( + !tracker + .observe( + &RuntimeEvent::RunletTransport { available: true }, + now + LEASE + ) + .accepted + ); + assert!(!tracker.observe(&data(), now + LEASE).accepted); + assert!( + tracker + .observe(&boundary("new", 0, BoundaryState::Open), now + LEASE) + .accepted + ); + assert!( + tracker + .observe(&scoped("new", 0, data()), now + LEASE) + .accepted + ); + } +} diff --git a/src/runlet_progress/transport.rs b/src/runlet_progress/transport.rs index 7dac31c7..95e31b2b 100644 --- a/src/runlet_progress/transport.rs +++ b/src/runlet_progress/transport.rs @@ -1,12 +1,12 @@ //! 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 crate::events::{BoundaryState, EVENT_MARKER, RuntimeEvent}; use std::{ io::{self, Write}, sync::{ - Arc, OnceLock, - atomic::{AtomicBool, Ordering}, + Arc, OnceLock, RwLock, + atomic::{AtomicBool, AtomicU64, Ordering}, mpsc::{self, Receiver, RecvTimeoutError, SyncSender}, }, time::{Duration, Instant}, @@ -14,20 +14,33 @@ use std::{ const CAPACITY: usize = 1024; const MAX_FRAME_BYTES: usize = 16 * 1024; +const ROTATE: Duration = Duration::from_secs(2); const HEARTBEAT: Duration = Duration::from_millis(500); pub(crate) const LEASE: Duration = Duration::from_secs(5); enum Frame { - Authoritative(Vec), + Authoritative { generation: u64, bytes: Vec }, Diagnostic(Vec), } #[derive(Clone)] pub(crate) struct Transport { + source: String, + gate: Arc, sender: SyncSender, - disabled: Arc, + terminal: Arc, + admission: Arc, diagnostics_lost: Arc, } +// Producers only try_read. The worker only try_write for healthy rotation; +// neither path waits for another lock holder. The unit lock owns publication +// claims, not recoverable data. Poison is terminal, never repaired. +#[derive(Default)] +struct PublicationGate { + active: RwLock<()>, + lost: AtomicBool, +} + impl Transport { /// A concrete owned IO boundary; production uses stderr, tests may use a pipe. pub(crate) fn start( @@ -35,9 +48,21 @@ impl Transport { capacity: usize, runtime_events: bool, ) -> io::Result { + if capacity > CAPACITY { + return Err(io::Error::other("runtime queue exceeds bounded capacity")); + } + // Random stream identity is never reused across workers or restarts. + let mut random = [0; 32]; + getrandom::fill(&mut random).map_err(io::Error::other)?; + let source = blake3::Hash::from_bytes(random).to_hex().to_string(); + let worker_source = source.clone(); + let gate = Arc::new(PublicationGate::default()); + let worker_gate = gate.clone(); let (sender, receiver) = mpsc::sync_channel(capacity); - let disabled = Arc::new(AtomicBool::new(false)); - let guard = WorkerGuard(disabled.clone()); + let terminal = Arc::new(AtomicBool::new(false)); + let guard = WorkerGuard(terminal.clone()); + let admission = Arc::new(AtomicU64::new(0)); + let worker_admission = admission.clone(); let diagnostics_lost = Arc::new(AtomicBool::new(false)); let worker_loss = diagnostics_lost.clone(); std::thread::Builder::new() @@ -46,11 +71,22 @@ impl Transport { // 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); + let _ = write_loop( + writer, + receiver, + &worker_loss, + &worker_admission, + runtime_events, + &worker_source, + &worker_gate, + ); })?; Ok(Self { + source, + gate, sender, - disabled, + terminal, + admission, diagnostics_lost, }) } @@ -77,34 +113,70 @@ impl Transport { /// 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) { + let Ok(_claim) = self.gate.active.try_read() else { + self.gate.lost.store(true, Ordering::Release); return; - } - if line.len() >= MAX_FRAME_BYTES { - self.disabled.store(true, Ordering::Release); + }; + let Some(generation) = self.generation() else { return; + }; + // Capture authority before parsing and re-encoding, preserving every + // descendant's provenance. Malformed marked frames fail closed rather + // than becoming diagnostics, even when a rotation is in progress. + if let Some(event) = crate::events::parse(line) { + match encode_scoped(&self.source, generation, &event) { + Ok(frame) => self.enqueue_authoritative(generation, frame), + Err(_) => self.lose(generation), + } + } else { + self.lose(generation); } - 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); + // Even generations admit; odd generations are closed until the writer has + // written an ordered Lost/Open boundary. Tokens are captured BEFORE encoding. + fn generation(&self) -> Option { + let generation = self.admission.load(Ordering::Acquire); + (!self.terminal.load(Ordering::Acquire) && generation.is_multiple_of(2)) + .then_some(generation) + } + + fn lose(&self, generation: u64) { + // A late failure from an invalidated producer must not close a newer + // generation. Concurrent losses of the same generation coalesce. + let _ = self.admission.compare_exchange( + generation, + generation | 1, + Ordering::AcqRel, + Ordering::Acquire, + ); + } + + fn enqueue_authoritative(&self, generation: u64, bytes: Vec) { + if self + .sender + .try_send(Frame::Authoritative { generation, bytes }) + .is_err() + { + self.lose(generation); } } /// 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) { + let Ok(_claim) = self.gate.active.try_read() else { + // There is no generation token yet. A rotation may be in progress; + // record loss separately so it cannot disappear inside its commit. + self.gate.lost.store(true, Ordering::Release); return; - } - match encode_frame(event) { - Ok(frame) => self.enqueue_authoritative(frame), - Err(_) => self.disabled.store(true, Ordering::Release), + }; + let Some(generation) = self.generation() else { + return; + }; + match encode_scoped(&self.source, generation, event) { + Ok(frame) => self.enqueue_authoritative(generation, frame), + Err(_) => self.lose(generation), } } } @@ -127,67 +199,198 @@ pub(crate) fn global() -> Option<&'static Transport> { fn write_loop( mut writer: impl Write, receiver: Receiver, - disabled: &AtomicBool, diagnostics_lost: &AtomicBool, + admission: &AtomicU64, runtime_events: bool, + source: &str, + gate: &PublicationGate, ) -> io::Result<()> { let mut heartbeat = Instant::now(); - transport_status(&mut writer, runtime_events, true)?; - let mut reset_sent = false; + let mut opened = heartbeat; + // The initial boundary always names epoch zero, even if a producer lost it + // before the worker started. Loss and its replacement follow in wire order. + transport_status(&mut writer, runtime_events, source, 0, BoundaryState::Open)?; 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 gate.lost.swap(false, Ordering::AcqRel) { + admission.fetch_or(1, Ordering::AcqRel); + } + let generation = admission.load(Ordering::Acquire); + // Healthy periodic boundaries need not discard any observations. A + // successful exclusive claim proves all producer claims have completed; + // drain their bounded queue BEFORE changing epochs. Failed claims record + // loss separately (above), including attempts during the boundary write. + let rotation = if generation.is_multiple_of(2) && opened.elapsed() >= ROTATE { + gate.active.try_write().ok() + } else { + None + }; + if rotation.is_some() { + for frame in receiver.try_iter().take(CAPACITY) { + match frame { + Frame::Authoritative { + generation: token, + bytes, + } if token == admission.load(Ordering::Acquire) => writer.write_all(&bytes)?, + Frame::Diagnostic(bytes) => writer.write_all(&bytes)?, + _ => {} + } + } + // Authoritative publication is excluded; diagnostics may continue. + // The production queue is bounded at CAPACITY, so this drain is + // finite even when arbitrary diagnostic producers keep publishing. + if gate.lost.swap(false, Ordering::AcqRel) { + admission.fetch_or(1, Ordering::AcqRel); + } } + let generation = admission.load(Ordering::Acquire); + if !generation.is_multiple_of(2) || rotation.is_some() { + let closed = generation | 1; + if generation.is_multiple_of(2) { + let _ = admission.compare_exchange( + generation, + closed, + Ordering::AcqRel, + Ordering::Acquire, + ); + } else { + transport_status( + &mut writer, + runtime_events, + source, + generation - 1, + BoundaryState::Lost, + )?; + } + if closed == u64::MAX { + return transport_status( + &mut writer, + runtime_events, + source, + closed - 1, + BoundaryState::Lost, + ); + } + transport_status( + &mut writer, + runtime_events, + source, + closed + 1, + BoundaryState::Open, + )?; + let _ = + admission.compare_exchange(closed, closed + 1, Ordering::AcqRel, Ordering::Acquire); + heartbeat = Instant::now(); + opened = heartbeat; + } + // IO above is an external callback under this unit guard. Publication + // reentry only try_reads, so it records loss rather than deadlocking. + // A Write panic poisons the gate and WorkerGuard retires the worker; + // no poison recovery is attempted. Release before receiving more work. + drop(rotation); 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, + Frame::Authoritative { generation, .. } + if generation != admission.load(Ordering::Acquire) => + { + continue; + } + Frame::Authoritative { bytes, .. } | Frame::Diagnostic(bytes) => bytes, }; + // A loss concurrent with this write is ordered AFTER it. The + // single writer cannot let any selected old frame cross reset. 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::Timeout) => {} Err(RecvTimeoutError::Disconnected) => { - return transport_status(&mut writer, runtime_events && !reset_sent, false); + return transport_status( + &mut writer, + runtime_events, + source, + admission.load(Ordering::Acquire) & !1, + BoundaryState::Lost, + ); } } + let generation = admission.load(Ordering::Acquire); + if generation.is_multiple_of(2) && heartbeat.elapsed() >= HEARTBEAT { + transport_status( + &mut writer, + runtime_events, + source, + generation, + BoundaryState::Heartbeat, + )?; + heartbeat = Instant::now(); + } } } -fn transport_status(writer: &mut impl Write, enabled: bool, available: bool) -> io::Result<()> { +fn transport_status( + writer: &mut impl Write, + enabled: bool, + source: &str, + epoch: u64, + state: BoundaryState, +) -> io::Result<()> { if enabled { - write_frame(writer, &RuntimeEvent::RunletTransport { available })?; + write_frame( + writer, + &RuntimeEvent::RuntimeBoundary { + source: source.into(), + epoch, + state, + }, + )?; } Ok(()) } +#[derive(serde::Serialize)] +struct ScopedFrame<'a> { + event: &'static str, + source: &'a str, + epoch: u64, + payload: &'a RuntimeEvent, +} + +fn encode_scoped(source: &str, epoch: u64, event: &RuntimeEvent) -> io::Result> { + let mut inner = event; + let mut depth = 1; // Include the scope this publisher adds. + while let RuntimeEvent::RuntimeScoped { payload, .. } = inner { + depth += 1; + if depth > 16 { + return Err(io::Error::other("runtime provenance is too deeply nested")); + } + inner = payload; + } + if !event.valid_wire_payload() { + return Err(io::Error::other("invalid runtime metadata")); + } + encode_value(&ScopedFrame { + event: "runtime_scoped", + source, + epoch, + payload: event, + }) +} + 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")); + if !event.valid_wire_payload() { + return Err(io::Error::other("invalid runtime metadata")); } + encode_value(event) +} + +fn encode_value(event: &impl serde::Serialize) -> io::Result> { // 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/runlet_progress/transport/tests.rs index 11a161ce..3b1632ab 100644 --- a/src/runlet_progress/transport/tests.rs +++ b/src/runlet_progress/transport/tests.rs @@ -21,169 +21,320 @@ fn progress() -> Progress { } } -#[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}"), +// Commands act at the actual external Write boundary. No production state or +// callbacks are replaced; a partial success makes write_all retry its suffix. +enum Action { + Accept, + Partial, + Error, + Panic, +} +struct GatedWriter { + writes: mpsc::Sender>, + actions: Receiver, +} +impl Write for GatedWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.writes.send(bytes.to_vec()).map_err(io::Error::other)?; + match self.actions.recv().map_err(io::Error::other)? { + Action::Accept => Ok(bytes.len()), + Action::Partial => Ok(bytes.len() / 2), + Action::Error => Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "external failure", + )), + Action::Panic => panic!("external sink panic"), } } - 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()); + fn flush(&mut self) -> io::Result<()> { + Ok(()) } - assert!(transport.disabled.load(Ordering::Acquire)); - reader - .set_read_timeout(Some(Duration::from_secs(5))) +} +struct Gate { + writes: Receiver>, + actions: mpsc::Sender, +} +impl Gate { + fn start(capacity: usize) -> (Transport, Self) { + let (writes, received) = mpsc::channel(); + let (actions, commands) = mpsc::channel(); + let transport = Transport::start( + GatedWriter { + writes, + actions: commands, + }, + capacity, + true, + ) .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); + ( + transport, + Self { + writes: received, + actions, + }, + ) + } + fn bytes(&self) -> Vec { + self.writes.recv_timeout(Duration::from_secs(5)).unwrap() + } + fn event(&self) -> RuntimeEvent { + crate::events::parse(String::from_utf8(self.bytes()).unwrap().trim_end()).unwrap() + } + fn accept(&self) { + self.actions.send(Action::Accept).unwrap(); + } + fn boundary(&self, transport: &Transport, epoch: u64, state: BoundaryState) { + assert_eq!(self.event(), boundary(&transport.source, epoch, state)); + } + // A heartbeat is emitted only after the Open write committed admission. + fn reopen(&self, transport: &Transport, epoch: u64) { + self.boundary(transport, epoch - 2, BoundaryState::Lost); + assert_eq!(transport.generation(), None); + self.accept(); + self.boundary(transport, epoch, BoundaryState::Open); + assert_eq!(transport.generation(), None); + transport.publish(progress()); // Rejected while Open is blocked. + self.accept(); + self.boundary(transport, epoch, BoundaryState::Heartbeat); + assert_eq!(transport.generation(), Some(epoch)); + } + fn finish(&self, transport: Transport, epoch: u64) { + let source = transport.source.clone(); + let terminal = transport.terminal.clone(); + drop(transport); + self.accept(); + assert_eq!(self.event(), boundary(&source, epoch, BoundaryState::Lost)); + self.accept(); + assert!(matches!( + self.writes.recv_timeout(Duration::from_secs(5)), + Err(mpsc::RecvTimeoutError::Disconnected) + )); + wait_terminal(&terminal); + } } - -#[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(()) - } +fn boundary(source: &str, epoch: u64, state: BoundaryState) -> RuntimeEvent { + RuntimeEvent::RuntimeBoundary { + source: source.into(), + epoch, + state, } - 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()); +} +fn scoped(transport: &Transport, epoch: u64, payload: RuntimeEvent) -> RuntimeEvent { + RuntimeEvent::RuntimeScoped { + source: transport.source.clone(), + epoch, + payload: Box::new(payload), + } +} +fn wait_terminal(terminal: &AtomicBool) { + let deadline = Instant::now() + Duration::from_secs(5); + while !terminal.load(Ordering::Acquire) { + assert!(Instant::now() < deadline, "worker did not terminate"); + std::thread::yield_now(); } } -#[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 }))); +fn stalled_writer_loss_discards_queued_frames_and_resumes_new_events() { + let (transport, gate) = Gate::start(2); + gate.boundary(&transport, 0, BoundaryState::Open); + transport.publish(progress()); + transport.publish_event(&RuntimeEvent::SessionStarted { + session_id: "queued".into(), + }); + transport.publish(progress()); // Real bounded queue overflow. + assert_eq!(transport.generation(), None); + gate.accept(); + gate.reopen(&transport, 2); + let event = RuntimeEvent::SessionStarted { + session_id: "fresh".into(), + }; + transport.publish_event(&event); + gate.accept(); + assert_eq!(gate.event(), scoped(&transport, 2, event)); + gate.finish(transport, 2); +} + +#[test] +fn retained_old_token_enqueued_after_reopen_is_discarded_and_failure_is_isolated() { + let (transport, gate) = Gate::start(2); + gate.boundary(&transport, 0, BoundaryState::Open); + let old = transport.generation().unwrap(); + transport.lose(old); + gate.accept(); + gate.reopen(&transport, 2); + let old_bytes = encode_scoped( + &transport.source, + old, + &RuntimeEvent::RunletProgress { + progress: progress(), + }, + ) + .unwrap(); + // Queue is empty and writer is blocked in Heartbeat: this enqueue succeeds. + transport.enqueue_authoritative(old, old_bytes.clone()); + assert_eq!(transport.generation(), Some(2)); + let fresh = RuntimeEvent::SessionStarted { + session_id: "fresh".into(), + }; + transport.publish_event(&fresh); + // Queue is now full: a delayed old-token enqueue fails, then a delayed + // encoding failure loses the same old token. Neither may close epoch two. + transport.enqueue_authoritative(old, old_bytes); + transport.lose(old); + assert_eq!(transport.generation(), Some(2)); + gate.accept(); + assert_eq!(gate.event(), scoped(&transport, 2, fresh)); + gate.finish(transport, 2); } -#[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, +fn loss_during_inflight_authoritative_write_precedes_lost_and_open() { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + let event = RuntimeEvent::RunletProgress { + progress: progress(), }; - 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())) - ); + gate.accept(); + assert_eq!(gate.event(), scoped(&transport, 0, event)); + transport.publish(progress()); // Backlog is invalidated, not in-flight IO. + transport.lose(0); + gate.accept(); + gate.reopen(&transport, 2); + gate.finish(transport, 2); +} - 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)); +#[test] +fn invalid_events_invalidate_all_queued_authoritative_frames() { + for invalid in [ + RuntimeEvent::SessionStarted { + session_id: "x".repeat(MAX_FRAME_BYTES), + }, + RuntimeEvent::RunletProgress { + progress: Progress { + owner: "x".repeat(257), + ..progress() + }, + }, + ] { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + transport.publish(progress()); + transport.publish_event(&invalid); + assert_eq!(transport.generation(), None); + gate.accept(); + gate.reopen(&transport, 2); + gate.finish(transport, 2); + } +} + +#[test] +fn repeated_concurrent_loss_during_blocked_writes_cannot_revive_old_frames() { + let (transport, gate) = Gate::start(16); + gate.boundary(&transport, 0, BoundaryState::Open); + for epoch in [2, 4, 6] { + transport.publish(progress()); + std::thread::scope(|scope| { + for _ in 0..8 { + let transport = &transport; + scope.spawn(move || transport.lose(epoch - 2)); + } + }); + gate.accept(); + gate.reopen(&transport, epoch); + } 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(); + gate.accept(); assert_eq!( - events.last(), - Some(&RuntimeEvent::RunletTransport { available: false }) - ); - assert!( - events - .iter() - .all(|event| matches!(event, RuntimeEvent::RunletTransport { .. })) + gate.event(), + scoped( + &transport, + 6, + RuntimeEvent::RunletProgress { + progress: progress() + } + ) ); + gate.finish(transport, 6); +} + +#[test] +fn worker_errors_and_panics_fail_closed_at_initial_lost_open_and_recovered_payload() { + for stage in 0..4 { + for partial in [false, true] { + for panic in [false, true] { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + if stage > 0 { + transport.lose(0); + gate.accept(); + gate.boundary(&transport, 0, BoundaryState::Lost); + } + if stage > 1 { + gate.accept(); + gate.boundary(&transport, 2, BoundaryState::Open); + } + if stage > 2 { + gate.accept(); + gate.boundary(&transport, 2, BoundaryState::Heartbeat); + transport.publish(progress()); + gate.accept(); + assert_eq!( + gate.event(), + scoped( + &transport, + 2, + RuntimeEvent::RunletProgress { + progress: progress() + } + ) + ); + } + if partial { + gate.actions.send(Action::Partial).unwrap(); + let suffix = gate.bytes(); + assert!(!suffix.starts_with(EVENT_MARKER.as_bytes())); + assert!(suffix.ends_with(b"\n")); + } + gate.actions + .send(if panic { Action::Panic } else { Action::Error }) + .unwrap(); + assert!(matches!( + gate.writes.recv_timeout(Duration::from_secs(5)), + Err(mpsc::RecvTimeoutError::Disconnected) + )); + wait_terminal(&transport.terminal); + assert_eq!(transport.generation(), None); + if stage == 1 || stage == 2 { + assert_eq!(transport.admission.load(Ordering::Acquire), 1); + } + transport.publish(progress()); + assert_eq!(transport.generation(), None); + } + } + } } +#[test] +fn disconnect_while_open_boundary_is_blocked() { + for recovering in [false, true] { + let (transport, gate) = Gate::start(2); + gate.boundary(&transport, 0, BoundaryState::Open); + let epoch = if recovering { + transport.lose(0); + gate.accept(); + gate.boundary(&transport, 0, BoundaryState::Lost); + gate.accept(); + gate.boundary(&transport, 2, BoundaryState::Open); + 2 + } else { + 0 + }; + gate.finish(transport, epoch); + } +} #[cfg(unix)] #[test] fn plain_diagnostics_do_not_enable_runtime_frames() { @@ -220,7 +371,7 @@ fn oversized_diagnostics_truncate_without_disabling_later_errors() { } assert!(line.ends_with(" [truncated]\n")); assert!(line.len() <= MAX_FRAME_BYTES); - assert!(!transport.disabled.load(Ordering::Acquire)); + assert!(!transport.terminal.load(Ordering::Acquire)); transport.publish_line("later child error"); drop(transport); let mut rest = String::new(); @@ -235,23 +386,223 @@ fn diagnostic_queue_overflow_does_not_poison_recovered_publication() { // No sink timing or exact implementation work counts are involved. let (sender, receiver) = mpsc::sync_channel(1); let transport = Transport { + source: String::from("test-source"), + gate: Arc::new(PublicationGate::default()), sender, - disabled: Arc::new(AtomicBool::new(false)), + terminal: Arc::new(AtomicBool::new(false)), + admission: Arc::new(AtomicU64::new(0)), diagnostics_lost: Arc::new(AtomicBool::new(false)), }; transport.publish_line("first"); transport.publish_line("dropped"); - assert!(!transport.disabled.load(Ordering::Acquire)); + assert!(!transport.terminal.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 admission = transport.admission.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(); + write_loop( + &mut wire, + receiver, + &diagnostics_lost, + &admission, + false, + "test-source", + &PublicationGate::default(), + ) + .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)); } + +#[test] +fn healthy_periodic_rotation_drains_admitted_payloads_before_open_without_loss() { + for queued_payloads in [false, true] { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + gate.accept(); + gate.boundary(&transport, 0, BoundaryState::Heartbeat); + let events = [ + RuntimeEvent::SessionStarted { + session_id: "admitted-before-rotation".into(), + }, + RuntimeEvent::RunletProgress { + progress: progress(), + }, + ]; + if queued_payloads { + for event in &events { + transport.publish_event(event); + } + } + assert_eq!(transport.generation(), Some(0)); + // Expire the real rotation clock while IO is stalled, without filling + // the queue. This triggers a state transition, not a speed assertion. + std::thread::sleep(ROTATE); + gate.accept(); + if queued_payloads { + for event in events { + assert_eq!(gate.event(), scoped(&transport, 0, event)); + gate.accept(); + } + } + // Exact wire sequence excludes Lost and proves all admitted old payloads + // precede the new Open. Empty stalled streams must rotate as well. + gate.boundary(&transport, 2, BoundaryState::Open); + assert_eq!(transport.generation(), None); + gate.accept(); + gate.boundary(&transport, 2, BoundaryState::Heartbeat); + assert_eq!(transport.generation(), Some(2)); + transport.publish(progress()); + gate.accept(); + assert_eq!( + gate.event(), + scoped( + &transport, + 2, + RuntimeEvent::RunletProgress { + progress: progress() + } + ) + ); + gate.finish(transport, 2); + } +} + +#[test] +fn rejected_producer_during_periodic_open_does_not_wait_or_lose_loss_at_commit() { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + gate.accept(); + gate.boundary(&transport, 0, BoundaryState::Heartbeat); + std::thread::sleep(ROTATE); + gate.accept(); + gate.boundary(&transport, 2, BoundaryState::Open); + assert_eq!(transport.generation(), None); + // The real worker holds its exclusive claim across this blocked Write. + // Completion before releasing IO proves publication doesn't await the sink + // or lock holder (the same API used by producer cancellation/Drop). + let producer = transport.clone(); + let (finished, completion) = mpsc::channel(); + let publisher = std::thread::spawn(move || { + producer.publish(progress()); + drop(producer); + finished.send(()).unwrap(); + }); + completion.recv_timeout(Duration::from_secs(5)).unwrap(); + publisher.join().unwrap(); + assert!(transport.gate.lost.load(Ordering::Acquire)); + gate.accept(); + // The worker can emit a heartbeat before processing the separately recorded + // failed claim. It must then explicitly retire the just-opened epoch. + gate.boundary(&transport, 2, BoundaryState::Heartbeat); + gate.accept(); + gate.reopen(&transport, 4); + transport.publish(progress()); + gate.accept(); + assert_eq!( + gate.event(), + scoped( + &transport, + 4, + RuntimeEvent::RunletProgress { + progress: progress() + } + ) + ); + gate.finish(transport, 4); +} + +#[test] +fn publication_reserves_its_scope_within_the_wire_wrapper_limit() { + for wrappers in [15, 16] { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + let mut event = RuntimeEvent::RunletProgress { + progress: progress(), + }; + for epoch in 0..wrappers { + event = RuntimeEvent::RuntimeScoped { + source: "upstream".into(), + epoch, + payload: Box::new(event), + }; + } + // Both inputs are independently accepted wire values. Publication must + // reserve one additional wrapper instead of emitting unparseable depth 17. + let input = String::from_utf8(encode_frame(&event).unwrap()).unwrap(); + assert_eq!(crate::events::parse(input.trim_end()), Some(event.clone())); + transport.publish_event(&event); + if wrappers == 15 { + assert_eq!(transport.generation(), Some(0)); + gate.accept(); + assert_eq!(gate.event(), scoped(&transport, 0, event)); + gate.finish(transport, 0); + } else { + assert_eq!(transport.generation(), None); + gate.accept(); + gate.reopen(&transport, 2); + gate.finish(transport, 2); + } + } +} + +#[test] +fn healthy_periodic_open_failure_retires_worker_without_poison_recovery() { + for panic in [false, true] { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + gate.accept(); + gate.boundary(&transport, 0, BoundaryState::Heartbeat); + std::thread::sleep(ROTATE); + gate.accept(); + gate.boundary(&transport, 2, BoundaryState::Open); + assert_eq!(transport.generation(), None); + // Unlike loss recovery, healthy rotation owns the exclusive publication + // claim through this real Write. Exercise a partial write before failure. + gate.actions.send(Action::Partial).unwrap(); + let suffix = gate.bytes(); + assert!(!suffix.starts_with(EVENT_MARKER.as_bytes())); + gate.actions + .send(if panic { Action::Panic } else { Action::Error }) + .unwrap(); + assert!(matches!( + gate.writes.recv_timeout(Duration::from_secs(5)), + Err(mpsc::RecvTimeoutError::Disconnected) + )); + wait_terminal(&transport.terminal); + assert_eq!(transport.gate.active.is_poisoned(), panic); + assert_eq!(transport.admission.load(Ordering::Acquire), 1); + transport.publish(progress()); + assert_eq!(transport.generation(), None); + assert_eq!(transport.gate.active.is_poisoned(), panic); + assert_eq!(transport.admission.load(Ordering::Acquire), 1); + } +} + +#[test] +fn forwarded_frames_keep_provenance_and_malformed_frames_close_admission() { + let (transport, gate) = Gate::start(4); + gate.boundary(&transport, 0, BoundaryState::Open); + let child = RuntimeEvent::RuntimeScoped { + source: "child".into(), + epoch: 12, + payload: Box::new(RuntimeEvent::StorageStatus { + pending: true, + exhausted: false, + }), + }; + let line = format!("{EVENT_MARKER}{}", serde_json::to_string(&child).unwrap()); + transport.publish_runtime_line(&line); + gate.accept(); + assert_eq!(gate.event(), scoped(&transport, 0, child)); + transport.publish_runtime_line(&format!("{EVENT_MARKER}not-json")); + assert_eq!(transport.generation(), None); + gate.accept(); + gate.reopen(&transport, 2); + gate.finish(transport, 2); +} diff --git a/src/tui/app.rs b/src/tui/app.rs index f3914a03..505d7de0 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -803,8 +803,9 @@ pub struct AgentCounts { } pub struct App { - progress_last_frame: Option, + runtime_authority: crate::runlet_progress::authority::Authority, progress_unavailable: bool, + runtime_incomplete: bool, pub root: PathBuf, pub provider: String, pub model: String, @@ -1064,8 +1065,9 @@ 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, + runtime_authority: Default::default(), progress_unavailable: false, + runtime_incomplete: false, root, provider, model, @@ -1369,7 +1371,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_authority.deadline().is_some() || self.working() || !self.transcript_dynamic.is_empty() || self.toast.is_some() @@ -2272,6 +2274,8 @@ impl App { Update::ProcessExited(error) => { // Confirmed process exit can retire known roster rows. A mere // diagnostic gap cannot claim those terminal outcomes. + self.runtime_authority + .expire(Instant::now() + crate::runlet_progress::transport::LEASE); self.invalidate_runtime_status(); self.finish_turn_with_outcome(false, None); self.retire_active_agents_at(crate::events::now_millis()); @@ -2290,15 +2294,13 @@ impl App { } fn invalidate_runtime_status(&mut self) { - if self.progress_unavailable { - return; - } + // Other healthy scopes can contribute fresh state while one scope is + // unavailable. Every subsequent loss must clear that state too. self.progress_unavailable = true; + self.runtime_incomplete = 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(); + // Lifecycle ordering and removal tombstones survive recovery. self.agents_scroll = 0; // Attachment identity survives transport loss; ACP emits it only when // attaching. Keep it distinct from session_id to reject another session. @@ -2319,44 +2321,46 @@ impl App { self.progress_unavailable } + pub(super) fn runtime_incomplete(&self) -> bool { + self.runtime_incomplete + } + /// 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 - }) - { + if !self.runtime_authority.expire(now).is_empty() { 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 let RuntimeEvent::RunletTransport { available } = event { - if available { - if self.progress_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"); - } - self.progress_unavailable = false; - self.progress_last_frame = Some(Instant::now()); - } else { - self.disable_runtime(); + let now = Instant::now(); + self.progress_tick_at(now); + let admission = self.runtime_authority.observe(&event, now); + if admission.invalidated { + self.disable_runtime(); + } + if !admission.accepted { + return; + } + if admission.opened { + if self.progress_unavailable { + self.note("Runtime status resumed; earlier agent, child, compaction and storage state remains unknown"); } + self.progress_unavailable = false; + } + let scoped = matches!(event, RuntimeEvent::RuntimeScoped { .. }); + let mut event = event; + while let RuntimeEvent::RuntimeScoped { payload, .. } = event { + event = *payload; + } + if matches!( + event, + RuntimeEvent::RuntimeBoundary { .. } | RuntimeEvent::RunletTransport { .. } + ) { return; } // Attachment markers identify the stream even during a gap, but do not @@ -2365,12 +2369,15 @@ impl App { self.runtime_session_id = Some(session_id); return; } - if self.runtime_unavailable() { + if self.runtime_unavailable() && !scoped { return; } let parent = event.parent_call().map(str::to_string); let owner_id = match event { - RuntimeEvent::RunletTransport { .. } | RuntimeEvent::SessionStarted { .. } => return, + RuntimeEvent::RunletTransport { .. } + | RuntimeEvent::SessionStarted { .. } + | RuntimeEvent::RuntimeBoundary { .. } + | RuntimeEvent::RuntimeScoped { .. } => return, RuntimeEvent::StorageStatus { pending, exhausted } => { self.storage_pending = pending; self.storage_exhausted = exhausted; @@ -2530,9 +2537,8 @@ impl App { return; } RuntimeEvent::RunletProgress { progress } => { - if self.progress_unavailable { - return; - } + // Admission already verified this event's own dependency scope. + // Another scope's loss must not block fresh healthy progress. 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, @@ -2656,6 +2662,7 @@ impl App { /// history and diagnostics remain useful, while transcript-derived state /// starts empty. pub fn start_session(&mut self, session_id: String) { + self.runtime_incomplete = false; self.model_switch = None; self.cancel_steer_edit(); self.selected_steer = None; @@ -7919,11 +7926,8 @@ 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); assert!(app.needs_redraw_tick()); - if app.needs_redraw_tick() { - app.tick(); - } + app.progress_tick_at(Instant::now() + crate::runlet_progress::transport::LEASE); assert!(app.runtime_unavailable()); assert_eq!(app.agent_counts().total, 0); assert!(!app.storage_pending && !app.storage_exhausted); @@ -7931,34 +7935,60 @@ mod tests { } #[test] - fn healthy_heartbeat_expires_old_state_before_renewing_lease() { + fn legacy_heartbeat_cannot_recover_expired_authority() { let mut app = app(); app.apply(Update::Runtime(RuntimeEvent::StorageStatus { pending: true, exhausted: true, })); - app.progress_last_frame = Some(Instant::now() - crate::runlet_progress::transport::LEASE); + app.progress_tick_at(Instant::now() + crate::runlet_progress::transport::LEASE); app.apply(Update::Runtime(RuntimeEvent::RunletTransport { available: true, })); - assert!(!app.runtime_unavailable()); + assert!(app.runtime_unavailable()); assert!(!app.storage_pending && !app.storage_exhausted); - assert!(app.needs_redraw_tick()); - assert!( - matches!(app.blocks.last(), Some(Block::Notice(text)) if text.contains("state remains unknown")) - ); - - let blocks = app.blocks.len(); - app.apply(Update::Runtime(RuntimeEvent::RunletTransport { - available: true, - })); - assert_eq!(app.blocks.len(), blocks); - app.tick(); - assert!(!app.runtime_unavailable()); + assert!(!app.needs_redraw_tick()); + } - app.progress_last_frame = Some(Instant::now() - crate::runlet_progress::transport::LEASE); - app.tick(); + #[test] + fn scoped_recovery_preserves_healthy_state_but_never_reconstructs_lost_state() { + use crate::events::BoundaryState; + let mut app = app(); + let boundary = |epoch, state| RuntimeEvent::RuntimeBoundary { + source: "runtime".into(), + epoch, + state, + }; + let data = |epoch| RuntimeEvent::RuntimeScoped { + source: "runtime".into(), + epoch, + payload: Box::new(RuntimeEvent::StorageStatus { + pending: true, + exhausted: true, + }), + }; + app.apply(Update::Runtime(boundary(0, BoundaryState::Open))); + app.apply(Update::Runtime(data(0))); + app.apply(Update::Runtime(boundary(2, BoundaryState::Open))); + assert!(app.storage_pending && app.storage_exhausted); + app.progress_tick_at(Instant::now() + crate::runlet_progress::transport::LEASE); assert!(app.runtime_unavailable()); + for event in [ + boundary(2, BoundaryState::Heartbeat), + boundary(2, BoundaryState::Open), + data(2), + ] { + app.apply(Update::Runtime(event)); + } + assert!(app.runtime_unavailable()); + assert!(!app.storage_pending && !app.storage_exhausted); + app.apply(Update::Runtime(boundary(4, BoundaryState::Open))); + assert!(!app.runtime_unavailable()); + assert!(!app.storage_pending && !app.storage_exhausted); + app.apply(Update::Runtime(data(2))); + assert!(!app.storage_pending); + app.apply(Update::Runtime(data(4))); + assert!(app.storage_pending && app.storage_exhausted); } #[test] diff --git a/src/tui/progress_tests.rs b/src/tui/progress_tests.rs index c07a5f21..109df7c5 100644 --- a/src/tui/progress_tests.rs +++ b/src/tui/progress_tests.rs @@ -11,6 +11,58 @@ fn progress_wire(app: &mut App, event: RuntimeEvent) { crate::events::parse(line.trim_end()).unwrap(), )); } +fn scoped_progress_wire(app: &mut App, epoch: u64, event: RuntimeEvent) { + let event = match event { + RuntimeEvent::RunletTransport { available } => RuntimeEvent::RuntimeBoundary { + source: "tui-test".into(), + epoch, + state: if available { + crate::events::BoundaryState::Open + } else { + crate::events::BoundaryState::Lost + }, + }, + event => RuntimeEvent::RuntimeScoped { + source: "tui-test".into(), + epoch, + payload: Box::new(event), + }, + }; + progress_wire(app, event); +} +fn scoped_progress_start(app: &mut App, epoch: u64, source: &str, incarnation: u64, healed: bool) { + scoped_progress_wire( + app, + epoch, + source_event( + "call-1", + incarnation, + 0, + ProgressChange::Started { + digest: crate::tui::progress::source_digest(source), + healed, + }, + ), + ); +} +fn scoped_progress_step( + app: &mut App, + epoch: u64, + incarnation: u64, + sequence: u64, + node: ProgressNode, +) { + scoped_progress_wire( + app, + epoch, + source_event( + "call-1", + incarnation, + sequence, + ProgressChange::Step { node: Some(node) }, + ), + ); +} fn source_event( owner: &str, incarnation: u64, @@ -869,42 +921,71 @@ fn start_progress_session(app: &mut App, session_id: &str) { fn authoritative_progress_recovery_preserves_old_incarnation_tombstones() { for explicit in [true, false] { let mut app = sample(); + let mut epoch = 0; + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); 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( + scoped_progress_wire( &mut app, + epoch, + RuntimeEvent::SessionStarted { + session_id: "session".into(), + }, + ); + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); + scoped_progress_start(&mut app, epoch, SCRIPT, 1, false); + scoped_progress_step( + &mut app, + epoch, 1, 1, progress_node("a", ProgressState::Succeeded, true), ); - progress_wire( + scoped_progress_wire( &mut app, + epoch, 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 }); + scoped_progress_wire( + &mut app, + epoch, + 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 }); + epoch = 2; + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); assert!(!app.runtime_unavailable()); - progress_start(&mut app, SCRIPT, 1, false); - progress_step( + scoped_progress_start(&mut app, epoch, SCRIPT, 1, false); + scoped_progress_step( &mut app, + epoch, 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( + scoped_progress_start(&mut app, epoch, SCRIPT, 2, false); + scoped_progress_step( &mut app, + epoch, 2, 1, progress_node("b", ProgressState::Succeeded, true), @@ -957,7 +1038,10 @@ async fn authoritative_progress_production_publication_completes_and_cancels_wit reader.read_to_string(&mut wire).unwrap(); assert!(wire.lines().any(|line| matches!( crate::events::parse(line), - Some(RuntimeEvent::RunletTransport { available: false }) + Some(RuntimeEvent::RuntimeBoundary { + state: crate::events::BoundaryState::Lost, + .. + }) ))); } } @@ -1012,8 +1096,20 @@ fn authoritative_progress_terminal_conflicts_invalidate_completed_display() { fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { for explicit in [false, true] { let mut app = sample(); + let mut epoch = 0; + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); start_progress_session(&mut app, "session"); - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "session".into() }); + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::SessionStarted { + session_id: "session".into(), + }, + ); let agent = RuntimeEvent::SubagentStateChanged { id: "child-agent".into(), name: "Child worker".into(), @@ -1030,9 +1126,10 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { generation_started_at_unix_ms: 2, generation_finished_at_unix_ms: None, }; - progress_wire(&mut app, agent.clone()); - progress_wire( + scoped_progress_wire(&mut app, epoch, agent.clone()); + scoped_progress_wire( &mut app, + epoch, RuntimeEvent::ChildStarted { call: "call-1:compose:0".into(), tool: "shell".into(), @@ -1040,15 +1137,17 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { at: 1, }, ); - progress_wire( + scoped_progress_wire( &mut app, + epoch, RuntimeEvent::StorageStatus { pending: true, exhausted: true, }, ); - progress_wire( + scoped_progress_wire( &mut app, + epoch, RuntimeEvent::CompactionStarted { reason: "test".into(), at: 1, @@ -1062,7 +1161,11 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { .any(|block| matches!(block, Block::Tool(call) if !call.children.is_empty())) ); if explicit { - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: false }, + ); } else { app.progress_tick_at( std::time::Instant::now() + crate::runlet_progress::transport::LEASE, @@ -1093,7 +1196,7 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { }, agent.clone(), ] { - progress_wire(&mut app, event); + scoped_progress_wire(&mut app, epoch, event); } assert!(app.runtime_unavailable()); assert_eq!(app.agent_counts().total, 0); @@ -1111,7 +1214,12 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { assert!(!frame.contains("compacting context")); assert!(!frame.contains("context compacted")); - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: true }); + epoch = 2; + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); assert!(!app.runtime_unavailable()); assert_eq!(app.agent_counts().total, 0); assert!(!app.storage_pending && !app.storage_exhausted && !app.compacting); @@ -1123,11 +1231,20 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { 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( + // A newer lifecycle observation is accepted; transport recovery alone + // cannot erase per-agent ordering tombstones or replay the old snapshot. + let mut fresh_agent = agent; + if let RuntimeEvent::SubagentStateChanged { generation, .. } = &mut fresh_agent { + *generation += 1; + } + scoped_progress_wire(&mut app, epoch, fresh_agent); + scoped_progress_wire( &mut app, - RuntimeEvent::StorageStatus { pending: true, exhausted: false }, + epoch, + RuntimeEvent::StorageStatus { + pending: true, + exhausted: false, + }, ); assert_eq!(app.agent_counts().working, 1); assert!(app.storage_pending); @@ -1139,43 +1256,339 @@ fn authoritative_progress_loss_invalidates_all_runtime_lifecycle_state() { } } +#[cfg(unix)] +#[tokio::test] +async fn real_publishers_forward_scoped_expiry_and_recovery_into_tui() { + use crate::{events::BoundaryState, runlet_progress::transport::Transport}; + use std::os::unix::net::UnixStream; + use tokio::io::AsyncBufReadExt; + + async fn receive( + lines: &mut tokio::io::Lines>, + app: &mut App, + ) -> RuntimeEvent { + // Deadlock watchdog only; no timing or work-count assertion. + let line = tokio::time::timeout(std::time::Duration::from_secs(10), lines.next_line()) + .await + .unwrap() + .unwrap() + .unwrap(); + let event = crate::events::parse(&line).unwrap(); + app.apply(Update::Runtime(event.clone())); + event + } + let (child_writer, child_reader) = UnixStream::pair().unwrap(); + let (parent_writer, parent_reader) = UnixStream::pair().unwrap(); + child_reader.set_nonblocking(true).unwrap(); + parent_reader.set_nonblocking(true).unwrap(); + let child = Transport::start(child_writer, 64, true).unwrap(); + let parent = Transport::start(parent_writer, 64, true).unwrap(); + let forwarding = tokio::spawn(crate::acp_child::test_support::forward_runtime( + tokio::net::UnixStream::from_std(child_reader).unwrap(), + parent, + )); + let mut lines = + tokio::io::BufReader::new(tokio::net::UnixStream::from_std(parent_reader).unwrap()).lines(); + let mut app = sample(); + let started = |summary: &str| RuntimeEvent::ChildStarted { + call: "call-1:compose:0".into(), + tool: "shell".into(), + summary: summary.into(), + at: 1, + }; + child.publish_event(&started("initial nested work")); + let (child_source, initial_epoch) = loop { + let event = receive(&mut lines, &mut app).await; + if let RuntimeEvent::RuntimeScoped { payload, .. } = event + && let RuntimeEvent::RuntimeScoped { + source, + epoch, + payload, + } = *payload + && matches!(*payload, RuntimeEvent::ChildStarted { .. }) + { + break (source, epoch); + } + }; + assert!( + app.blocks + .iter() + .any(|block| matches!(block, Block::Tool(call) if !call.children.is_empty())) + ); + // Stall the owned child receiver's lease independently of both real writer + // workers. The publisher queues have ample capacity; no overflow is needed. + tokio::time::pause(); + tokio::time::advance(crate::runlet_progress::transport::LEASE).await; + tokio::task::yield_now().await; + tokio::time::resume(); + loop { + let event = receive(&mut lines, &mut app).await; + if matches!(crate::runlet_progress::authority::payload(&event), RuntimeEvent::RuntimeBoundary { + source, epoch, state: BoundaryState::Lost, + } if source == &child_source && *epoch == initial_epoch) + { + break; + } + } + assert!(app.runtime_incomplete()); + assert!( + app.blocks + .iter() + .all(|block| !matches!(block, Block::Tool(call) if !call.children.is_empty())) + ); + // The real child worker periodically opens a newer epoch even when no + // publication overflow happened. Parent rotation alone cannot hide the loss. + loop { + let event = receive(&mut lines, &mut app).await; + assert!(app.runtime_incomplete()); + if matches!(crate::runlet_progress::authority::payload(&event), RuntimeEvent::RuntimeBoundary { + source, epoch, state: BoundaryState::Open, + } if source == &child_source && *epoch > initial_epoch) + { + break; + } + } + assert!( + app.blocks + .iter() + .all(|block| !matches!(block, Block::Tool(call) if !call.children.is_empty())) + ); + child.publish_event(&started("fresh nested work")); + loop { + let event = receive(&mut lines, &mut app).await; + if matches!(crate::runlet_progress::authority::payload(&event), RuntimeEvent::ChildStarted { summary, .. } if summary == "fresh nested work") + { + break; + } + } + assert!( + app.blocks + .iter() + .any(|block| matches!(block, Block::Tool(call) if !call.children.is_empty())) + ); + assert!(render(&mut app, 180, 50).contains("Runtime status incomplete")); + drop(child); + tokio::time::timeout(std::time::Duration::from_secs(10), forwarding) + .await + .unwrap() + .unwrap(); +} + +#[test] +fn healthy_parent_recovery_cannot_hide_descendant_or_historical_loss() { + use crate::events::BoundaryState; + let mut app = sample(); + scoped_progress_wire( + &mut app, + 0, + RuntimeEvent::RunletTransport { available: true }, + ); + scoped_progress_wire( + &mut app, + 0, + RuntimeEvent::RuntimeBoundary { + source: "child".into(), + epoch: 0, + state: BoundaryState::Open, + }, + ); + scoped_progress_start(&mut app, 0, SCRIPT, 1, false); + scoped_progress_step( + &mut app, + 0, + 1, + 1, + progress_node("a", ProgressState::Running, true), + ); + scoped_progress_wire( + &mut app, + 0, + RuntimeEvent::RuntimeBoundary { + source: "child".into(), + epoch: 0, + state: BoundaryState::Lost, + }, + ); + assert!(app.runtime_unavailable()); + // The live parent remains eligible to publish fresh progress even before + // rotating. Global completeness is not an admission gate for its scope. + scoped_progress_start(&mut app, 0, SCRIPT, 2, false); + scoped_progress_step( + &mut app, + 0, + 2, + 1, + progress_node("b", ProgressState::Succeeded, true), + ); + assert!(render(&mut app, 180, 50).contains("# call @")); + scoped_progress_wire( + &mut app, + 2, + RuntimeEvent::RunletTransport { available: true }, + ); + assert!(!app.runtime_unavailable()); + assert!(app.runtime_incomplete()); + assert!(render(&mut app, 180, 50).contains("Runtime status incomplete")); + scoped_progress_wire( + &mut app, + 2, + RuntimeEvent::RuntimeBoundary { + source: "child".into(), + epoch: 2, + state: BoundaryState::Open, + }, + ); + // Even child recovery cannot reconstruct observations lost in the gap. + assert!(render(&mut app, 180, 50).contains("Runtime status incomplete")); + start_progress_session(&mut app, "fresh"); + assert!(!app.runtime_incomplete()); +} + #[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(); + let mut epoch = 0; + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); start_progress_session(&mut app, "old"); - progress_wire(&mut app, RuntimeEvent::SessionStarted { session_id: "old".into() }); + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::SessionStarted { + session_id: "old".into(), + }, + ); if explicit { - progress_wire(&mut app, RuntimeEvent::RunletTransport { available: false }); + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: false }, + ); } else { - app.progress_tick_at(std::time::Instant::now() + crate::runlet_progress::transport::LEASE); + 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() }); + scoped_progress_wire( + &mut app, + epoch, + 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); + let compaction = RuntimeEvent::CompactionStarted { + reason: "test".into(), + at: 1, + }; + scoped_progress_wire(&mut app, epoch, compaction.clone()); + scoped_progress_start(&mut app, epoch, 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)); + epoch = 2; + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); + scoped_progress_wire(&mut app, epoch, compaction.clone()); + scoped_progress_start(&mut app, epoch, SCRIPT, 2, false); + scoped_progress_step( + &mut app, + epoch, + 2, + 1, + progress_node("a", ProgressState::Succeeded, true), + ); + assert!(!app.compacting); + assert!(!render(&mut app, 140, 50).contains("# call @")); + // A new epoch must not guess which session owns the stream. + { + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::SessionStarted { + session_id: "new".into(), + }, + ); + scoped_progress_wire(&mut app, epoch, compaction); + scoped_progress_start(&mut app, epoch, SCRIPT, 3, false); + scoped_progress_step( + &mut app, + epoch, + 3, + 1, + progress_node("a", ProgressState::Succeeded, true), + ); assert!(app.compacting); assert!(render(&mut app, 140, 50).contains("# call @")); } } } } + +#[test] +fn recovered_storage_risk_stays_visible_despite_incomplete_runtime_status() { + use crate::events::BoundaryState; + for exhausted in [false, true] { + for reopen in [false, true] { + let mut app = sample(); + scoped_progress_wire( + &mut app, + 0, + RuntimeEvent::RunletTransport { available: true }, + ); + scoped_progress_wire( + &mut app, + 0, + RuntimeEvent::RuntimeBoundary { + source: "child".into(), + epoch: 0, + state: BoundaryState::Open, + }, + ); + scoped_progress_wire( + &mut app, + 0, + RuntimeEvent::RuntimeBoundary { + source: "child".into(), + epoch: 0, + state: BoundaryState::Lost, + }, + ); + let epoch = if reopen { 2 } else { 0 }; + if reopen { + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::RunletTransport { available: true }, + ); + } + scoped_progress_wire( + &mut app, + epoch, + RuntimeEvent::StorageStatus { + pending: true, + exhausted, + }, + ); + let frame = render(&mut app, 180, 50); + assert!(frame.contains(if exhausted { + "Storage exhausted: shutting down" + } else { + "Memory-only storage: awaiting disk recovery" + })); + assert!(frame.contains("data is at risk") || frame.contains("data at risk on exit")); + assert!(frame.contains("runtime status incomplete")); + assert!(app.runtime_incomplete()); + } + } +} diff --git a/src/tui/ui.rs b/src/tui/ui.rs index e57e7fe8..3a04bf61 100644 --- a/src/tui/ui.rs +++ b/src/tui/ui.rs @@ -141,14 +141,29 @@ 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.runtime_unavailable() || app.storage_pending || app.storage_exhausted { + if app.runtime_unavailable() + || app.runtime_incomplete() + || app.storage_pending + || app.storage_exhausted + { let area = frame.area(); - let warning = if app.runtime_unavailable() { + let incomplete = app.runtime_unavailable() || app.runtime_incomplete(); + let warning = if app.storage_exhausted { + if incomplete { + " Storage exhausted: shutting down; unpersisted data is at risk; runtime status incomplete" + } else { + " Storage exhausted: shutting down; unpersisted data is at risk" + } + } else if app.storage_pending { + if incomplete { + " Memory-only storage: awaiting disk recovery; data at risk on exit; runtime status incomplete" + } else { + " Memory-only storage: awaiting disk recovery; data at risk on exit" + } + } else 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" + " Runtime status incomplete; some observations may be unavailable" }; frame.render_widget( Paragraph::new(warning).style(Style::default().fg(theme::warn_color())),