From d0f9e6bffd58dfb4776e908d7799333f81c8583d Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Sat, 15 Aug 2026 22:20:29 +0200 Subject: [PATCH 1/2] supervisor: bounded per-module terminal ring so a missed exit is countable A snapshot alone omitted an exit timestamp, retained only one terminal, and could not count missed terminals. The per-module bounded history reports every retained exit with its chosen disposition; eviction increments dropped so a returning observer knows it has only a suffix. daemon_started_at_ms marks the in-memory daemon incarnation, distinguishing a quiet daemon from a restarted one with no retained history. --- Cargo.lock | 4 +- crates/subc-client-rs/Cargo.toml | 2 +- crates/subc-control/Cargo.toml | 2 +- crates/subc-control/src/lib.rs | 59 +++++++ .../client_control_response_catalog_list.json | 3 +- ...ient_control_response_server_describe.json | 3 +- ...esponse_server_describe_with_counters.json | 3 +- ...ient_control_response_supervisor_list.json | 1 + ...control_response_supervisor_terminals.json | 17 ++ .../tests/golden/supervisor_entry.json | 1 + crates/subc-control/tests/golden_json.rs | 2 + crates/subc-core/Cargo.toml | 2 +- crates/subc-core/src/bin/ck.rs | 75 +++++++- crates/subc-core/src/control.rs | 144 ++++++++++++++- crates/subc-core/src/lib.rs | 1 + crates/subc-core/src/supervise.rs | 167 +++++++++++++++++- crates/subc-core/src/terminal_ring.rs | 130 ++++++++++++++ crates/subc-core/tests/closure.rs | 1 + 18 files changed, 597 insertions(+), 20 deletions(-) create mode 100644 crates/subc-control/tests/golden/client_control_response_supervisor_terminals.json create mode 100644 crates/subc-core/src/terminal_ring.rs diff --git a/Cargo.lock b/Cargo.lock index c1b8ef8a..51eb7b68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -703,7 +703,7 @@ dependencies = [ [[package]] name = "subc-control" -version = "0.1.1" +version = "0.2.0" dependencies = [ "serde", "serde_json", @@ -712,7 +712,7 @@ dependencies = [ [[package]] name = "subc-core" -version = "0.3.1" +version = "0.4.0" dependencies = [ "cortexkit-paths", "fs4", diff --git a/crates/subc-client-rs/Cargo.toml b/crates/subc-client-rs/Cargo.toml index 23e45072..abed81ca 100644 --- a/crates/subc-client-rs/Cargo.toml +++ b/crates/subc-client-rs/Cargo.toml @@ -11,7 +11,7 @@ repository = "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/cortexkit/subconscious" async-trait = "0.1" serde = { version = "1", features = ["derive"] } serde_json = "1" -subc-control = { path = "../subc-control", version = "0.1" } +subc-control = { path = "../subc-control", version = "0.2" } subc-protocol = { path = "../subc-protocol", version = "0.11" } subc-transport = { path = "../subc-transport", version = "0.5" } tokio = { version = "1", features = ["io-util", "macros", "net", "rt", "sync", "time"] } diff --git a/crates/subc-control/Cargo.toml b/crates/subc-control/Cargo.toml index 2957f2f0..f2f30c40 100644 --- a/crates/subc-control/Cargo.toml +++ b/crates/subc-control/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "subc-control" -version = "0.1.1" +version = "0.2.0" edition = "2021" publish = true description = "Client-facing subc control-plane wire shapes." diff --git a/crates/subc-control/src/lib.rs b/crates/subc-control/src/lib.rs index 590c0f28..10bdeb49 100644 --- a/crates/subc-control/src/lib.rs +++ b/crates/subc-control/src/lib.rs @@ -46,6 +46,7 @@ pub mod ops { pub const SUPERVISOR_HEALTH_PROBE: &str = "supervisor.health_probe"; pub const SUPERVISOR_HEALTH: &str = "supervisor.health"; pub const SUPERVISOR_STDERR_TAIL: &str = "supervisor.stderr_tail"; + pub const SUPERVISOR_TERMINALS: &str = "supervisor.terminals"; } /// Client-originated channel-0 control RPC body. @@ -150,6 +151,20 @@ pub enum ClientControlRequest { #[serde(default, skip_serializing_if = "Option::is_none")] max_bytes: Option, }, + /// Retained terminal exits for one module. + /// + /// This stays separate from `supervisor.list`: a history grows with every + /// incident, while the list is a current-state read most callers issue often. + /// + /// The read MUST stay off the supervisor command channel — it reads the + /// module's shared ring directly. This is a requirement, not an + /// optimisation: when the supervision task itself dies, every + /// command-channel op returns `CommandClosed`, and that is precisely the + /// moment an operator needs the exit history most. A history reachable only + /// through the machinery whose death you are diagnosing is unreachable when + /// it matters. Proven failure mode, not a hypothetical. + #[serde(rename = "supervisor.terminals")] + SupervisorTerminals { module_id: String }, } /// subc's channel-0 response body for client control RPCs. @@ -230,6 +245,12 @@ pub enum ClientControlResponse { #[serde(flatten)] tail: StderrTail, }, + #[serde(rename = "supervisor.terminals")] + SupervisorTerminals { + module_id: String, + #[serde(flatten)] + terminals: TerminalHistory, + }, } /// A module's retained stderr, oldest entry first. @@ -290,6 +311,39 @@ fn is_zero_u64(value: &u64) -> bool { *value == 0 } +/// Bounded terminal history for one module, oldest retained record first. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct TerminalHistory { + /// Unix milliseconds at daemon start. The history is in-memory and resets with + /// the daemon, so an empty list only means "no exits during this incarnation". + pub daemon_started_at_ms: u64, + pub entries: Vec, + /// Earlier terminal exits evicted by the bounded ring. + #[serde(default, skip_serializing_if = "is_zero_u64")] + pub dropped: u64, +} + +/// One terminal child exit and the supervisor action it selected. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct TerminalEntry { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_signal: Option, + pub at_ms: u64, + pub disposition: TerminalDisposition, +} + +/// The supervisor disposition selected after observing a terminal exit. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TerminalDisposition { + Stopped, + Disabled, + Failed, + Restarting, +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum PollKind { @@ -400,6 +454,11 @@ pub struct SupervisorEntry { /// (often a panic-abort). Survives respawn. #[serde(default, skip_serializing_if = "Option::is_none")] pub last_exit_signal: Option, + /// Unix milliseconds when the most recent child exit was observed. Present + /// even when the terminal ring is not queried, so existing list readers can + /// order their latest observed exit against events they already received. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_exit_ms: Option, /// Replacement processes spawned for this module so far, against the budget /// that disables it. /// diff --git a/crates/subc-control/tests/golden/client_control_response_catalog_list.json b/crates/subc-control/tests/golden/client_control_response_catalog_list.json index 8a0e34d1..0faf51c7 100644 --- a/crates/subc-control/tests/golden/client_control_response_catalog_list.json +++ b/crates/subc-control/tests/golden/client_control_response_catalog_list.json @@ -91,6 +91,7 @@ "supervisor.set_enabled", "supervisor.health_probe", "supervisor.health", - "supervisor.stderr_tail" + "supervisor.stderr_tail", + "supervisor.terminals" ] } diff --git a/crates/subc-control/tests/golden/client_control_response_server_describe.json b/crates/subc-control/tests/golden/client_control_response_server_describe.json index ae4aaab2..effd0931 100644 --- a/crates/subc-control/tests/golden/client_control_response_server_describe.json +++ b/crates/subc-control/tests/golden/client_control_response_server_describe.json @@ -17,6 +17,7 @@ "supervisor.set_enabled", "supervisor.health_probe", "supervisor.health", - "supervisor.stderr_tail" + "supervisor.stderr_tail", + "supervisor.terminals" ] } diff --git a/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json b/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json index 22bc36b2..468d6577 100644 --- a/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json +++ b/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json @@ -28,6 +28,7 @@ "supervisor.set_enabled", "supervisor.health_probe", "supervisor.health", - "supervisor.stderr_tail" + "supervisor.stderr_tail", + "supervisor.terminals" ] } diff --git a/crates/subc-control/tests/golden/client_control_response_supervisor_list.json b/crates/subc-control/tests/golden/client_control_response_supervisor_list.json index 2f85eab8..32fa27cf 100644 --- a/crates/subc-control/tests/golden/client_control_response_supervisor_list.json +++ b/crates/subc-control/tests/golden/client_control_response_supervisor_list.json @@ -4,6 +4,7 @@ { "enabled": true, "health": "degraded", + "last_exit_ms": 1700000000123, "last_probe_ms": 1700000000000, "live": true, "max_restarts": 3, diff --git a/crates/subc-control/tests/golden/client_control_response_supervisor_terminals.json b/crates/subc-control/tests/golden/client_control_response_supervisor_terminals.json new file mode 100644 index 00000000..7c45ee64 --- /dev/null +++ b/crates/subc-control/tests/golden/client_control_response_supervisor_terminals.json @@ -0,0 +1,17 @@ +{ + "daemon_started_at_ms": 1700000000000, + "entries": [ + { + "at_ms": 1700000000001, + "disposition": "restarting", + "exit_code": 23 + }, + { + "at_ms": 1700000000002, + "disposition": "failed", + "exit_code": 23 + } + ], + "module_id": "terminal-golden", + "op": "supervisor.terminals" +} diff --git a/crates/subc-control/tests/golden/supervisor_entry.json b/crates/subc-control/tests/golden/supervisor_entry.json index e149a2a4..555c75f2 100644 --- a/crates/subc-control/tests/golden/supervisor_entry.json +++ b/crates/subc-control/tests/golden/supervisor_entry.json @@ -1,6 +1,7 @@ { "enabled": true, "health": "degraded", + "last_exit_ms": 1700000000123, "last_probe_ms": 1700000000000, "live": true, "max_restarts": 3, diff --git a/crates/subc-control/tests/golden_json.rs b/crates/subc-control/tests/golden_json.rs index 6a6abaf6..a2712463 100644 --- a/crates/subc-control/tests/golden_json.rs +++ b/crates/subc-control/tests/golden_json.rs @@ -346,6 +346,7 @@ fn thin_core_ops() -> Vec { "supervisor.health_probe".to_string(), "supervisor.health".to_string(), "supervisor.stderr_tail".to_string(), + "supervisor.terminals".to_string(), ] } @@ -376,6 +377,7 @@ fn supervisor_entry() -> SupervisorEntry { last_probe_ms: Some(1_700_000_000_000), last_exit_code: None, last_exit_signal: None, + last_exit_ms: Some(1_700_000_000_123), // Non-equal and non-zero so the pin would catch the two fields being // swapped, which equal values or a zeroed count could not. restart_count: Some(2), diff --git a/crates/subc-core/Cargo.toml b/crates/subc-core/Cargo.toml index f182283b..3d6a3c4c 100644 --- a/crates/subc-core/Cargo.toml +++ b/crates/subc-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "subc-core" -version = "0.3.1" +version = "0.4.0" edition = "2021" publish = false description = "subc daemon core: loopback TCP transport and opaque-byte splice router." diff --git a/crates/subc-core/src/bin/ck.rs b/crates/subc-core/src/bin/ck.rs index 27a02d42..65ee6571 100644 --- a/crates/subc-core/src/bin/ck.rs +++ b/crates/subc-core/src/bin/ck.rs @@ -41,7 +41,7 @@ const PROD_CONNECTION_RELATIVE_PATH: &[&str] = const QUOTA_MODULE_ID: &str = "insula"; const CK_HARNESS: &str = "ck"; -const TOP_HELP_BASE: &str = "ck — CortexKit operator CLI\n\nusage:\n ck [--subc ] [--json] [] []\n\ndomains:\n module supervised modules: list, status, stderr, restart, stop, start, rescan\n health one-line health for every supervised module\n quota AI-provider quota and usage windows\n daemon daemon version, uptime, and connection info"; +const TOP_HELP_BASE: &str = "ck — CortexKit operator CLI\n\nusage:\n ck [--subc ] [--json] [] []\n\ndomains:\n module supervised modules: list, status, stderr, terminals, restart, stop, start, rescan\n health one-line health for every supervised module\n quota AI-provider quota and usage windows\n daemon daemon version, uptime, and connection info"; const TOP_HELP_TAIL: &str = "flags:\n --subc use a specific connection file (default: auto-discover)\n --json raw JSON output instead of tables\n\nrun 'ck ' with no verb to see that domain's commands"; @@ -106,7 +106,7 @@ fn discover_external_domains() -> Vec { } const MODULE_HELP: &str = "ck module — inspect and control supervised modules\n\nusage: ck [--json] module []\n\nverbs:\n ck module list all modules with state and health\n ck module status one module in detail - ck module stderr retained stderr for a module (-n to limit)\n ck module restart drain-restart a module\n ck module stop disable and stop a module (persists until start)\n ck module start enable and spawn a module\n ck module rescan re-read subc.jsonc and reconcile the module set\n ck module rescan --dry-run show what a rescan would change, without changing it"; + ck module stderr retained stderr for a module (-n to limit)\n ck module terminals retained terminal exits for a module\n ck module restart drain-restart a module\n ck module stop disable and stop a module (persists until start)\n ck module start enable and spawn a module\n ck module rescan re-read subc.jsonc and reconcile the module set\n ck module rescan --dry-run show what a rescan would change, without changing it"; const QUOTA_HELP: &str = "ck quota - AI-provider quota and usage windows\n\nusage: ck [--json] quota [--verbose] []\n\n ck quota connected providers and their usage windows\n ck quota --verbose all tracked providers, including unavailable ones\n ck quota claude one provider's windows and status in detail"; @@ -151,6 +151,9 @@ async fn run(argv: impl IntoIterator) -> Result<(), CkError> { module_id, max_lines, }) => module_stderr_tail(&mut client, &module_id, max_lines, args.json).await, + Command::Module(ModuleCommand::Terminals { module_id }) => { + module_terminals(&mut client, &module_id, args.json).await + } Command::Module(ModuleCommand::Restart { module_id }) => { module_restart(&mut client, &module_id, args.json).await } @@ -241,6 +244,9 @@ enum ModuleCommand { module_id: String, max_lines: Option, }, + Terminals { + module_id: String, + }, } struct ResolvedConnection { @@ -598,6 +604,56 @@ async fn module_stderr_tail( Ok(()) } +async fn module_terminals( + client: &mut CkClient, + module_id: &str, + json_output: bool, +) -> Result<(), CkError> { + let response = client + .rpc_value(ClientControlRequest::SupervisorTerminals { + module_id: module_id.to_string(), + }) + .await?; + + if json_output { + print_json(&response)?; + return Ok(()); + } + + let daemon_started_at_ms = response + .get("daemon_started_at_ms") + .and_then(Value::as_u64) + .map(|value| value.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + let dropped = response.get("dropped").and_then(Value::as_u64).unwrap_or(0); + let entries = response + .get("entries") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + println!( + "daemon_started_at_ms={daemon_started_at_ms} · {} terminal record(s) · {dropped} dropped", + entries.len() + ); + for entry in entries { + let at_ms = entry.get("at_ms").and_then(Value::as_u64).unwrap_or(0); + let disposition = entry + .get("disposition") + .and_then(Value::as_str) + .unwrap_or("unknown"); + let exit = match ( + entry.get("exit_signal").and_then(Value::as_i64), + entry.get("exit_code").and_then(Value::as_i64), + ) { + (Some(signal), _) => format!("signal {signal}"), + (None, Some(code)) => format!("code {code}"), + (None, None) => "unknown exit".to_string(), + }; + println!("{at_ms} {exit} → {disposition}"); + } + Ok(()) +} + async fn module_restart( client: &mut CkClient, module_id: &str, @@ -2735,6 +2791,7 @@ fn parse_command(domain: &str, tail: &[OsString]) -> Result { module_id: id(1)?, max_lines: parse_tail_count(tail)?, }, + "terminals" => ModuleCommand::Terminals { module_id: id(1)? }, "restart" => ModuleCommand::Restart { module_id: id(1)? }, "stop" => ModuleCommand::Stop { module_id: id(1)? }, "start" => ModuleCommand::Start { module_id: id(1)? }, @@ -3573,4 +3630,18 @@ mod tests { )); assert!(QUOTA_HELP.contains("--verbose")); } + + #[test] + fn module_terminals_is_parsed_and_documented() { + let command = parse_command( + "module", + &[OsString::from("terminals"), OsString::from("aft")], + ) + .unwrap(); + assert!(matches!( + command, + Command::Module(ModuleCommand::Terminals { module_id }) if module_id == "aft" + )); + assert!(MODULE_HELP.contains("ck module terminals ")); + } } diff --git a/crates/subc-core/src/control.rs b/crates/subc-core/src/control.rs index a9f9adee..fc651314 100644 --- a/crates/subc-core/src/control.rs +++ b/crates/subc-core/src/control.rs @@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize}; use subc_control::{ ops, CatalogEntry, ClientControlRequest, ClientControlResponse, ConsumerIdentity, PollKind, StderrCaptureState, StderrTail, StderrTailEntry, SupervisorEntry, SupervisorHealthEntry, - SupervisorRescanResult, + SupervisorRescanResult, TerminalEntry, TerminalHistory, }; use subc_protocol::{ manifest::{Concurrency, ModuleManifest, ProviderRole}, @@ -64,6 +64,7 @@ const SUBC_CONTROL_OPS: &[&str] = &[ ops::SUPERVISOR_HEALTH_PROBE, ops::SUPERVISOR_HEALTH, ops::SUPERVISOR_STDERR_TAIL, + ops::SUPERVISOR_TERMINALS, ]; const MODULE_TO_SUBC_CONTROL_OPS: &[&str] = &[MODULE_TO_SUBC_OP_CATALOG_UPDATE]; @@ -846,6 +847,9 @@ impl ControlHandler { max_lines, max_bytes, } => self.handle_supervisor_stderr_tail(frame, module_id, max_lines, max_bytes), + ClientControlRequest::SupervisorTerminals { module_id } => { + self.handle_supervisor_terminals(frame, module_id) + } } } @@ -1359,6 +1363,7 @@ impl ControlHandler { last_probe_ms: status.health.last_probe_ms, last_exit_code: status.last_exit.as_ref().and_then(|e| e.code), last_exit_signal: status.last_exit.as_ref().and_then(|e| e.signal), + last_exit_ms: status.last_exit.as_ref().map(|e| e.at_ms), restart_count: Some(status.restart_count), max_restarts: Some(status.max_restarts), }) @@ -1427,6 +1432,44 @@ impl ControlHandler { )?]) } + fn handle_supervisor_terminals( + &self, + frame: Frame, + module_id: String, + ) -> Result, RouterError> { + let Some(module) = self.supervisor.get(&module_id) else { + return Ok(vec![control_error_frame( + &frame, + "unknown_module", + format!("module_id '{module_id}' is not supervised"), + )?]); + }; + + let snapshot = module.terminal_history(); + let response = ClientControlResponse::SupervisorTerminals { + module_id, + terminals: TerminalHistory { + daemon_started_at_ms: snapshot.daemon_started_at_ms, + entries: snapshot + .entries + .into_iter() + .map(|entry| TerminalEntry { + exit_code: entry.exit_code, + exit_signal: entry.exit_signal, + at_ms: entry.at_ms, + disposition: entry.disposition, + }) + .collect(), + dropped: snapshot.dropped, + }, + }; + Ok(vec![control_response_body_frame( + &frame, + &response, + "ClientControlResponse::SupervisorTerminals", + )?]) + } + fn handle_supervisor_health(&self, frame: Frame) -> Result, RouterError> { let generation = self .registry @@ -2845,11 +2888,20 @@ mod tests { /// /// `CARGO_BIN_EXE_*` (compile-time `env!` and runtime `std::env::var` alike) /// is only populated for `tests/*.rs` integration test binaries -- this file - /// compiles as part of the library target, which gets neither. Cargo still - /// builds every `[[bin]]` target before running the library's tests, so the - /// binary is on disk; this test's own executable path is - /// `//deps/subc_core-`, and the sibling binary - /// lives two directories up at `//fake-aft-stub`. + /// compiles as part of the library target, which gets neither. This test's + /// own executable path is `//deps/subc_core-`, + /// and the sibling binary lives two directories up at + /// `//fake-aft-stub`. + /// + /// THE BINARY IS NOT ALWAYS THERE, and the existence check below is why. + /// `cargo test -p subc-core` builds every target including `[[bin]]`, so the + /// stub is on disk; `cargo test -p subc-core --lib` builds ONLY the library + /// test and leaves the stub unbuilt. A bare spawn then fails with a raw + /// `NotFound`, which reads as a broken test rather than an unbuilt + /// dependency -- so state the cause and the remedy instead. Deliberately a + /// panic and not a silent skip: a test that quietly passes when it could not + /// run is worse than one that fails, because it reports health it never + /// verified. fn fake_aft_stub_path() -> PathBuf { let mut path = std::env::current_exe().expect("current_exe available in tests"); path.pop(); // .../deps/ @@ -2859,6 +2911,12 @@ mod tests { } else { "fake-aft-stub" }); + assert!( + path.exists(), + "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \ + [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)", + path.display() + ); path } @@ -3423,6 +3481,80 @@ mod tests { assert!(*truncated); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn supervisor_terminals_golden_is_generated_through_the_real_handler() { + let registry = Arc::new(Registry::default()); + let supervisor_handle = SupervisorHandle::new(); + let supervisor = + Supervisor::new(Arc::clone(®istry), RestartPolicy::new(1, Duration::ZERO)) + .with_handle(supervisor_handle.clone()); + let module = supervisor + .spawn(ModuleSpec { + module_id: "terminal-golden".to_string(), + program: fake_aft_stub_path(), + args: Vec::new(), + env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())], + reserved: false, + reserved_prefixes: Vec::new(), + }) + .unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + while module.terminal_history().entries.len() != 2 { + assert!( + Instant::now() < deadline, + "module did not retain two terminal exits: {:?}", + module.terminal_history() + ); + sleep(Duration::from_millis(10)).await; + } + + let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor_handle); + let request = ClientControlRequest::SupervisorTerminals { + module_id: "terminal-golden".to_string(), + }; + let frame = Frame::build( + FrameType::Request, + control_flags(), + 0, + 0, + 1, + serde_json::to_vec(&request).unwrap(), + ) + .unwrap(); + let (ctx, _egress) = route_ctx(ConnectionId::new(1)); + let responses = handler.handle_control_frame(&ctx, frame).await.unwrap(); + let response: ClientControlResponse = serde_json::from_slice(&responses[0].body).unwrap(); + let ClientControlResponse::SupervisorTerminals { terminals, .. } = &response else { + panic!("expected supervisor.terminals response"); + }; + assert_eq!(terminals.entries.len(), 2); + assert_eq!(terminals.dropped, 0); + + let mut rendered = serde_json::to_value(response).unwrap(); + // Wall-clock fields are the observation contract, but not stable fixture + // bytes; normalize only them after the real handler has shaped the response. + rendered["daemon_started_at_ms"] = json!(1_700_000_000_000u64); + for (index, entry) in rendered["entries"] + .as_array_mut() + .expect("terminal response entries array") + .iter_mut() + .enumerate() + { + entry["at_ms"] = json!(1_700_000_000_001u64 + index as u64); + } + + let golden_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../subc-control/tests/golden/client_control_response_supervisor_terminals.json"); + let serialized = serde_json::to_string_pretty(&rendered).unwrap() + "\n"; + if std::env::var_os("UPDATE_GOLDEN").is_some() { + std::fs::write(&golden_path, &serialized).unwrap(); + } + let expected: Value = + serde_json::from_str(&std::fs::read_to_string(&golden_path).unwrap()).unwrap(); + assert_eq!(rendered, expected); + } + #[test] fn hello_registers_manifest_and_returns_ack() { let registry = Arc::new(Registry::default()); diff --git a/crates/subc-core/src/lib.rs b/crates/subc-core/src/lib.rs index 3dad197c..d196876d 100644 --- a/crates/subc-core/src/lib.rs +++ b/crates/subc-core/src/lib.rs @@ -19,6 +19,7 @@ pub mod router; pub mod server; pub mod stderr_tail; pub mod supervise; +pub mod terminal_ring; pub mod watchdog; #[cfg(feature = "bench-harness")] diff --git a/crates/subc-core/src/supervise.rs b/crates/subc-core/src/supervise.rs index 5e926870..def2f599 100644 --- a/crates/subc-core/src/supervise.rs +++ b/crates/subc-core/src/supervise.rs @@ -9,7 +9,7 @@ use std::{ }; use serde_json::Value; -use subc_control::SupervisorHealthStatus; +use subc_control::{SupervisorHealthStatus, TerminalDisposition}; use subc_protocol::{ session::{HealthReport, HealthStatus, ModuleControlRequest, MODULE_CONTROL_OP_HEALTH_CHECK}, Flags, FrameType, Priority, SUBC_LAUNCH_NONCE_ENV, SUBC_MODULE_ID_ENV, @@ -29,6 +29,7 @@ use crate::{ }, registry::RegistryError, stderr_tail::{pump_stderr, StderrRing, StderrTailConfig, StderrTailSnapshot}, + terminal_ring::{TerminalHistorySnapshot, TerminalRecord, TerminalRing, TerminalRingConfig}, Frame, Registry, }; @@ -297,6 +298,7 @@ pub struct ExitReport { pub kind: ExitKind, pub code: Option, pub signal: Option, + pub at_ms: u64, } /// Point-in-time module status answerable by subc without forwarding to the @@ -431,6 +433,7 @@ struct SupervisorRuntimeConfig { /// are written BEFORE that exit, so a ring recreated per process would be empty /// exactly when it is asked for. stderr_ring: Arc>, + terminal_ring: Arc>, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -710,6 +713,7 @@ pub struct Supervisor { process_liveness: Arc, supervisor_handle: Option, health: HealthConfig, + daemon_started_at_ms: u64, } impl Supervisor { @@ -723,6 +727,7 @@ impl Supervisor { process_liveness: Arc::new(SupervisorProcessLiveness::default()), supervisor_handle: None, health: HealthConfig::default(), + daemon_started_at_ms: unix_ms_now(), } } @@ -885,6 +890,10 @@ impl Supervisor { forwarding: self.forwarding.clone(), supervisor_handle: self.supervisor_handle.clone(), stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))), + terminal_ring: Arc::new(Mutex::new(TerminalRing::new( + TerminalRingConfig::default(), + self.daemon_started_at_ms, + ))), } } @@ -900,6 +909,7 @@ impl Supervisor { health: runtime.health, })); let stderr_ring = Arc::clone(&runtime.stderr_ring); + let terminal_ring = Arc::clone(&runtime.terminal_ring); let (tx, rx) = mpsc::channel(4); let monitor = tokio::spawn(supervise_loop( spec.clone(), @@ -919,6 +929,7 @@ impl Supervisor { snapshot, configuration, stderr_ring, + terminal_ring, commands: tx, monitor: Mutex::new(Some(monitor)), max_restarts: self.restart_policy.max_restarts, @@ -950,6 +961,7 @@ struct SupervisedModuleInner { snapshot: SharedSnapshot, configuration: Arc>, stderr_ring: Arc>, + terminal_ring: Arc>, commands: mpsc::Sender, monitor: Mutex>>, /// Copied from the supervisor's runtime config at spawn so `status()` can @@ -1008,6 +1020,18 @@ impl SupervisedModule { .snapshot(max_lines, max_bytes) } + /// The module's bounded terminal history, oldest retained exit first. + /// + /// The daemon-start stamp distinguishes a quiet supervisor from a replacement + /// daemon whose in-memory history was necessarily reset. + pub fn terminal_history(&self) -> TerminalHistorySnapshot { + self.inner + .terminal_ring + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .snapshot() + } + pub fn status(&self) -> Result { let snapshot = lock_snapshot(&self.inner.snapshot)?.clone(); let registration_active = self @@ -1930,6 +1954,7 @@ async fn health_restart_child( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Disabled, @@ -1961,6 +1986,7 @@ async fn health_restart_child( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Restarting, @@ -2129,6 +2155,7 @@ async fn supervise_loop( runtime.restart_policy, ®istry, &snapshot, + &runtime.terminal_ring, exit_report, ).await { NextAction::Stop { registration_released } => { @@ -2238,6 +2265,7 @@ async fn handle_supervisor_command( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Stopped, @@ -2265,6 +2293,7 @@ async fn handle_supervisor_command( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Stopped, @@ -2343,6 +2372,7 @@ async fn restart_child( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Restarting, @@ -2389,6 +2419,7 @@ async fn reload_child( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Restarting, @@ -2559,6 +2590,7 @@ async fn set_child_enabled( &spec.module_id, registry, snapshot, + &runtime.terminal_ring, child, runtime.drain_timeout, ModuleState::Disabled, @@ -2575,6 +2607,7 @@ async fn on_child_exit( policy: RestartPolicy, registry: &Registry, snapshot: &SharedSnapshot, + terminal_ring: &Arc>, exit_report: ExitReport, ) -> NextAction { match exit_report.kind { @@ -2589,10 +2622,11 @@ async fn on_child_exit( state.state = ModuleState::Stopped; state.process_alive = false; state.pid = None; - state.last_exit = Some(exit_report); + state.last_exit = Some(exit_report.clone()); }) { error!(module_id = %spec.module_id, error = %err, "failed to record clean module exit"); } + record_terminal(terminal_ring, &exit_report, TerminalDisposition::Stopped); let registration_released = match wait_for_registration_release( registry, &spec.module_id, @@ -2618,18 +2652,22 @@ async fn on_child_exit( "supervised module exited abnormally (crash)" ); let mut should_restart = false; + let mut disposition = TerminalDisposition::Disabled; if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| { state.process_alive = false; state.pid = None; - state.last_exit = Some(exit_report); + state.last_exit = Some(exit_report.clone()); if !state.enabled { state.state = ModuleState::Disabled; + disposition = TerminalDisposition::Disabled; } else if state.restart_count >= policy.max_restarts { state.state = ModuleState::Failed; + disposition = TerminalDisposition::Failed; } else { state.restart_count += 1; state.state = ModuleState::Restarting; should_restart = true; + disposition = TerminalDisposition::Restarting; } }) { error!(module_id = %spec.module_id, error = %err, "failed to record crashed module exit"); @@ -2637,6 +2675,7 @@ async fn on_child_exit( registration_released: false, }; } + record_terminal(terminal_ring, &exit_report, disposition); if should_restart { NextAction::Restart @@ -2662,6 +2701,22 @@ async fn on_child_exit( } } +fn record_terminal( + terminal_ring: &Arc>, + exit_report: &ExitReport, + disposition: TerminalDisposition, +) { + terminal_ring + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(TerminalRecord { + exit_code: exit_report.code, + exit_signal: exit_report.signal, + at_ms: exit_report.at_ms, + disposition, + }); +} + fn untrack_if_registration_released( process_liveness: &SupervisorProcessLiveness, registry: &Registry, @@ -3074,6 +3129,7 @@ async fn handle_reload_child_registration_failure( runtime.restart_policy, registry, snapshot, + &runtime.terminal_ring, exit_report, ) .await @@ -3176,10 +3232,12 @@ fn control_flags() -> Flags { Flags::new(false, Priority::Passive, false) } +#[allow(clippy::too_many_arguments)] async fn drain_optional_child( module_id: &str, registry: &Registry, snapshot: &SharedSnapshot, + terminal_ring: &Arc>, child: &mut Option, drain_timeout: Duration, final_state: ModuleState, @@ -3190,6 +3248,7 @@ async fn drain_optional_child( module_id, registry, snapshot, + terminal_ring, child, drain_timeout, final_state, @@ -3209,10 +3268,12 @@ async fn drain_optional_child( } } +#[allow(clippy::too_many_arguments)] async fn drain_child_to_state( module_id: &str, registry: &Registry, snapshot: &SharedSnapshot, + terminal_ring: &Arc>, mut child: SupervisedChild, drain_timeout: Duration, final_state: ModuleState, @@ -3254,13 +3315,33 @@ async fn drain_child_to_state( } state.process_alive = false; state.pid = None; - state.last_exit = Some(exit_report); + state.last_exit = Some(exit_report.clone()); })?; + record_terminal( + terminal_ring, + &exit_report, + terminal_disposition(final_state), + ); child.drain_stderr(module_id).await; wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await } +fn terminal_disposition(final_state: ModuleState) -> TerminalDisposition { + match final_state { + ModuleState::Stopped => TerminalDisposition::Stopped, + ModuleState::Disabled => TerminalDisposition::Disabled, + ModuleState::Restarting => TerminalDisposition::Restarting, + ModuleState::Failed => TerminalDisposition::Failed, + ModuleState::Starting + | ModuleState::Running + | ModuleState::Unresponsive + | ModuleState::Draining => { + unreachable!("terminal exits only finish in terminal or restarting states") + } + } +} + async fn wait_for_registration_release( registry: &Registry, module_id: &str, @@ -3308,6 +3389,7 @@ fn classify_exit(status: &ExitStatus) -> ExitReport { }, code: status.code(), signal: exit_signal(status), + at_ms: unix_ms_now(), } } @@ -3375,6 +3457,83 @@ fn lock_snapshot( .map_err(|_| SuperviseError::StatePoisoned { module_id: None }) } +#[cfg(test)] +mod terminal_history_tests { + use std::{ + path::PathBuf, + sync::Arc, + time::{Duration, Instant}, + }; + + use tokio::time::sleep; + + use super::{ModuleSpec, ModuleState, RestartPolicy, Supervisor}; + use crate::registry::Registry; + + /// See the twin in `control.rs` for why this derives the path from + /// `current_exe()` and why the existence check is here: `--lib` alone does + /// not build `[[bin]]` targets, and a bare spawn then fails with a raw + /// `NotFound` that reads as a broken test rather than an unbuilt dependency. + fn fake_aft_stub_path() -> PathBuf { + let mut path = std::env::current_exe().expect("current_exe available in tests"); + path.pop(); + path.pop(); + path.push(if cfg!(windows) { + "fake-aft-stub.exe" + } else { + "fake-aft-stub" + }); + assert!( + path.exists(), + "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \ + [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)", + path.display() + ); + path + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn terminal_history_survives_respawn_and_keeps_both_crashes_in_order() { + let registry = Arc::new(Registry::default()); + let supervisor = + Supervisor::new(Arc::clone(®istry), RestartPolicy::new(1, Duration::ZERO)); + let module = supervisor + .spawn(ModuleSpec { + module_id: "terminal-history".to_string(), + program: fake_aft_stub_path(), + args: Vec::new(), + env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())], + reserved: false, + reserved_prefixes: Vec::new(), + }) + .unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let history = module.terminal_history(); + if history.entries.len() == 2 { + assert_eq!(module.status().unwrap().state, ModuleState::Failed); + assert_eq!(history.dropped, 0); + assert_eq!( + history + .entries + .iter() + .map(|entry| entry.exit_code) + .collect::>(), + vec![Some(23), Some(23)] + ); + assert!(history.entries[0].at_ms <= history.entries[1].at_ms); + return; + } + assert!( + Instant::now() < deadline, + "module did not retain two terminal exits: {history:?}" + ); + sleep(Duration::from_millis(10)).await; + } + } +} + #[cfg(test)] mod health_evidence_tests { use super::{HealthProbeError, HealthProbeEvidence}; diff --git a/crates/subc-core/src/terminal_ring.rs b/crates/subc-core/src/terminal_ring.rs new file mode 100644 index 00000000..fbaa7f2e --- /dev/null +++ b/crates/subc-core/src/terminal_ring.rs @@ -0,0 +1,130 @@ +use std::collections::VecDeque; + +use subc_control::TerminalDisposition; + +const DEFAULT_MAX_ENTRIES: usize = 32; + +/// Fixed-size retention policy for one module's terminal exits. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TerminalRingConfig { + max_entries: usize, +} + +impl TerminalRingConfig { + /// A zero-sized history cannot answer whether an exit was observed, so clamp it + /// to one record instead of constructing a ring that lies by omission. + pub const fn new(max_entries: usize) -> Self { + Self { + max_entries: if max_entries == 0 { 1 } else { max_entries }, + } + } +} + +impl Default for TerminalRingConfig { + fn default() -> Self { + Self::new(DEFAULT_MAX_ENTRIES) + } +} + +/// One observed child exit and the disposition chosen by its supervisor. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TerminalRecord { + pub exit_code: Option, + pub exit_signal: Option, + pub at_ms: u64, + pub disposition: TerminalDisposition, +} + +/// The retained terminal suffix for one module. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TerminalHistorySnapshot { + pub daemon_started_at_ms: u64, + pub entries: Vec, + pub dropped: u64, +} + +/// Bounded terminal-exit history for one supervised module. +/// +/// This belongs to the module rather than an individual child so a replacement +/// process retains the exits that caused it to exist. +#[derive(Debug)] +pub struct TerminalRing { + config: TerminalRingConfig, + daemon_started_at_ms: u64, + entries: VecDeque, + dropped: u64, +} + +impl TerminalRing { + pub fn new(config: TerminalRingConfig, daemon_started_at_ms: u64) -> Self { + Self { + config, + daemon_started_at_ms, + entries: VecDeque::new(), + dropped: 0, + } + } + + pub fn push(&mut self, entry: TerminalRecord) { + self.entries.push_back(entry); + while self.entries.len() > self.config.max_entries { + self.entries.pop_front(); + self.dropped = self.dropped.saturating_add(1); + } + } + + pub fn snapshot(&self) -> TerminalHistorySnapshot { + TerminalHistorySnapshot { + daemon_started_at_ms: self.daemon_started_at_ms, + entries: self.entries.iter().cloned().collect(), + dropped: self.dropped, + } + } +} + +#[cfg(test)] +mod tests { + use super::{TerminalRecord, TerminalRing, TerminalRingConfig}; + use subc_control::TerminalDisposition; + + fn record(at_ms: u64) -> TerminalRecord { + TerminalRecord { + exit_code: Some(1), + exit_signal: None, + at_ms, + disposition: TerminalDisposition::Restarting, + } + } + + #[test] + fn the_ring_evicts_oldest_exits_and_counts_them() { + let mut ring = TerminalRing::new(TerminalRingConfig::new(2), 10); + ring.push(record(11)); + ring.push(record(12)); + ring.push(record(13)); + + let snapshot = ring.snapshot(); + assert_eq!(snapshot.daemon_started_at_ms, 10); + assert_eq!(snapshot.dropped, 1); + assert_eq!( + snapshot + .entries + .iter() + .map(|entry| entry.at_ms) + .collect::>(), + vec![12, 13] + ); + } + + #[test] + fn an_incoherent_zero_capacity_keeps_one_terminal() { + let mut ring = TerminalRing::new(TerminalRingConfig::new(0), 10); + ring.push(record(11)); + ring.push(record(12)); + + let snapshot = ring.snapshot(); + assert_eq!(snapshot.dropped, 1); + assert_eq!(snapshot.entries.len(), 1); + assert_eq!(snapshot.entries[0].at_ms, 12); + } +} diff --git a/crates/subc-core/tests/closure.rs b/crates/subc-core/tests/closure.rs index e6367c94..e8f33fb5 100644 --- a/crates/subc-core/tests/closure.rs +++ b/crates/subc-core/tests/closure.rs @@ -701,6 +701,7 @@ fn thin_core_ops() -> BTreeSet<&'static str> { ops::SUPERVISOR_HEALTH_PROBE, ops::SUPERVISOR_HEALTH, ops::SUPERVISOR_STDERR_TAIL, + ops::SUPERVISOR_TERMINALS, ]) } From 15fba643833c1165b4a3de765d5c01f081e0dd37 Mon Sep 17 00:00:00 2001 From: ualtinok Date: Sun, 16 Aug 2026 15:30:52 +0200 Subject: [PATCH 2/2] twin: retrigger CI (synchronize event lost for d0f9e6bf)