diff --git a/crates/cccc-cli/tests/suite/daemon_self_launch.rs b/crates/cccc-cli/tests/suite/daemon_self_launch.rs index 267bcf1a3..73c43e1ce 100644 --- a/crates/cccc-cli/tests/suite/daemon_self_launch.rs +++ b/crates/cccc-cli/tests/suite/daemon_self_launch.rs @@ -5,6 +5,84 @@ use std::path::Path; use std::process::{Command, Output, Stdio}; use std::time::{Duration, Instant}; +// These black-box cases also run unchanged against the stock daemon as the +// missing-feature reference. Only the CLI's public IPC interface is used. +#[test] +fn host_disk_diagnostic_is_sampled_without_any_group() { + with_disk_daemon(|home| { + std::thread::sleep(Duration::from_millis(100)); + let diagnostic = wait_disk_sample(home); + assert!(diagnostic["space"]["total_bytes"].as_u64().is_some()); + assert!(diagnostic["space"]["available_bytes"].as_u64().is_some()); + assert!(diagnostic["sampled_at"].as_str().is_some()); + assert_eq!(diagnostic["sample_interval_seconds"], 60); + assert!( + std::fs::read_dir(home.join("groups")) + .expect("group directory") + .next() + .is_none(), + "the host diagnostic must not require a Group" + ); + }); +} + +#[test] +fn host_disk_diagnostic_status_polling_does_not_resample() { + with_disk_daemon(|home| { + let initial = wait_disk_sample(home); + for _ in 0..8 { + let current = disk_ping(home); + assert_eq!(current["attempts"], initial["attempts"]); + assert_eq!(current["sampled_at"], initial["sampled_at"]); + assert_eq!(current["space"], initial["space"]); + } + }); +} + +fn with_disk_daemon(check: impl FnOnce(&Path)) { + let temp = tempfile::tempdir().expect("home"); + let home = temp.path().join("home"); + let executable = Path::new(env!("CARGO_BIN_EXE_cccc")); + let start = run(executable, &home, &["daemon", "start"]); + assert!(start.status.success(), "start: {}", detail(&start)); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| check(&home))); + let stop = run(executable, &home, &["daemon", "stop"]); + wait_for_daemon_exit(&home); + assert!(stop.status.success(), "stop: {}", detail(&stop)); + assert_daemon_lock_released(&home); + if let Err(error) = result { + std::panic::resume_unwind(error); + } +} + +fn disk_ping(home: &Path) -> serde_json::Value { + let layout = cccc_core::HomeLayout::from_path(home).expect("home layout"); + let response = tokio::runtime::Runtime::new() + .expect("RPC runtime") + .block_on( + cccc_client::DaemonClient::new(layout).call(&cccc_contracts::DaemonRequest { + v: 1, + op: "ping".into(), + args: Default::default(), + }), + ) + .expect("actual daemon ping"); + assert!(response.ok, "ping succeeds"); + response.result["diagnostics"]["disk"].clone() +} + +fn wait_disk_sample(home: &Path) -> serde_json::Value { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let value = disk_ping(home); + if value["attempts"].as_u64().is_some_and(|count| count > 0) { + return value; + } + assert!(Instant::now() < deadline, "host disk diagnostic is missing"); + std::thread::sleep(Duration::from_millis(10)); + } +} + #[test] fn installed_cli_starts_daemon_without_a_sibling_daemon_binary() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/crates/cccc-daemon/src/dispatch.rs b/crates/cccc-daemon/src/dispatch.rs index 7e4e0297b..8ee2a6118 100644 --- a/crates/cccc-daemon/src/dispatch.rs +++ b/crates/cccc-daemon/src/dispatch.rs @@ -44,7 +44,7 @@ pub(crate) fn resolve_operation(request: &DaemonRequest) -> Option { }) } -fn ping(_home: &HomeLayout, _request: &DaemonRequest) -> OpResult { +fn ping(home: &HomeLayout, _request: &DaemonRequest) -> OpResult { object(json!({ "pid": std::process::id(), "version": env!("CARGO_PKG_VERSION"), @@ -52,6 +52,7 @@ fn ping(_home: &HomeLayout, _request: &DaemonRequest) -> OpResult { "executable": std::env::current_exe().ok(), "ts": cccc_contracts::utc_now(), "ipc_v": 1, + "diagnostics": { "disk": crate::server_host_disk::snapshot(home) }, "capabilities": { "events_stream": true, "remote_access": true, diff --git a/crates/cccc-daemon/src/lib.rs b/crates/cccc-daemon/src/lib.rs index 5891277d9..fdb308ad9 100644 --- a/crates/cccc-daemon/src/lib.rs +++ b/crates/cccc-daemon/src/lib.rs @@ -16,6 +16,7 @@ mod server_automation; mod server_connection; mod server_connections; mod server_events_stream; +mod server_host_disk; mod server_lifecycle; mod server_terminal_attach; diff --git a/crates/cccc-daemon/src/server.rs b/crates/cccc-daemon/src/server.rs index e22c5958a..9ba3dd672 100644 --- a/crates/cccc-daemon/src/server.rs +++ b/crates/cccc-daemon/src/server.rs @@ -19,6 +19,7 @@ use crate::server_actor_activity::ActorActivityService; use crate::server_automation::AutomationScheduler; use crate::server_connection::spawn_connection; use crate::server_connections::ConnectionTasks; +use crate::server_host_disk::HostDiskService; use crate::server_lifecycle::{DaemonLifecycle, cleanup_stale}; type RuntimeRestoreSpawner = fn(HomeLayout, DispatchLocks); @@ -40,6 +41,7 @@ async fn run_with_restore(home: HomeLayout, restore: RuntimeRestoreSpawner) -> R let mut lifecycle = DaemonLifecycle::new(paths, lock); std::fs::write(&lifecycle.paths.pid, format!("{}\n", std::process::id()))?; let (shutdown_tx, shutdown_rx) = watch::channel(false); + let host_disk = HostDiskService::start(lifecycle.paths.home.clone()); let actor_activity = ActorActivityService::start(lifecycle.paths.home.clone()); let dispatch_locks = DispatchLocks::default(); @@ -62,6 +64,7 @@ async fn run_with_restore(home: HomeLayout, restore: RuntimeRestoreSpawner) -> R ) .await }; + host_disk.finish().await; actor_activity.finish().await; lifecycle.finish(result) } diff --git a/crates/cccc-daemon/src/server_host_disk.rs b/crates/cccc-daemon/src/server_host_disk.rs new file mode 100644 index 000000000..ff188bad1 --- /dev/null +++ b/crates/cccc-daemon/src/server_host_disk.rs @@ -0,0 +1,183 @@ +//! Passive, process-owned diagnostics for the volume containing CCCC_HOME. +//! No Group, ledger, NoticeState or delivery policy participates in sampling. +use cccc_core::HomeLayout; +use serde::Serialize; +use std::collections::BTreeMap; +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, OnceLock, Weak}; +use std::time::Duration; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; + +const SAMPLE_INTERVAL: Duration = Duration::from_secs(60); +const LOW_BYTES: u64 = 3 * 1024 * 1024 * 1024; +const RECOVERY_BYTES: u64 = 4 * 1024 * 1024 * 1024; + +#[derive(Clone, Debug, Serialize)] +struct Space { + available_bytes: u64, + total_bytes: u64, +} + +#[derive(Clone, Debug, Serialize)] +pub(crate) struct Diagnostic { + state: &'static str, + sample_interval_seconds: u64, + low_bytes: u64, + recovery_bytes: u64, + attempts: u64, + last_attempt_at: Option, + sampled_at: Option, + space: Option, + error: Option, + #[serde(skip)] + low: bool, +} + +impl Default for Diagnostic { + fn default() -> Self { + Self { + state: "unknown", + sample_interval_seconds: SAMPLE_INTERVAL.as_secs(), + low_bytes: LOW_BYTES, + recovery_bytes: RECOVERY_BYTES, + attempts: 0, + last_attempt_at: None, + sampled_at: None, + space: None, + error: None, + low: false, + } + } +} + +impl Diagnostic { + fn record(&mut self, result: io::Result) { + let now = cccc_contracts::utc_now(); + self.attempts = self.attempts.saturating_add(1); + self.last_attempt_at = Some(now.clone()); + match result { + Ok(space) => { + self.low = space.available_bytes + < if self.low { + self.recovery_bytes + } else { + self.low_bytes + }; + self.state = if self.low { "low" } else { "healthy" }; + self.space = Some(space); + self.sampled_at = Some(now); + self.error = None; + } + Err(error) => { + self.state = "unavailable"; + self.error = Some(error.to_string()); + // Preserve the prior sample, its timestamp and hysteresis. + // A failed observation is neither recovery nor fresh capacity. + } + } + } +} + +type Shared = Arc>; +type Registry = BTreeMap>>; + +fn registry() -> &'static Mutex { + static REGISTRY: OnceLock> = OnceLock::new(); + REGISTRY.get_or_init(Mutex::default) +} + +/// Read the existing sample only. Polling never probes a filesystem. +pub(crate) fn snapshot(home: &HomeLayout) -> Option { + let shared = registry() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(home.root())? + .upgrade()?; + let result = shared + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + Some(result) +} + +fn sample(path: &Path) -> io::Result { + let usage = fs2::statvfs(path)?; + let space = Space { + available_bytes: usage.available_space(), + total_bytes: usage.total_space(), + }; + if space.total_bytes == 0 || space.available_bytes > space.total_bytes { + return Err(io::Error::other("invalid filesystem capacity sample")); + } + Ok(space) +} + +pub(crate) struct HostDiskService { + stop: Option>, + task: JoinHandle<()>, +} + +impl HostDiskService { + pub(crate) fn start(home: HomeLayout) -> Self { + Self::with_interval(home, SAMPLE_INTERVAL) + } + + fn with_interval(home: HomeLayout, period: Duration) -> Self { + let shared: Shared = Arc::new(Mutex::new(Diagnostic::default())); + { + let mut entries = registry() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + entries.retain(|_, state| state.strong_count() > 0); + entries.insert(home.root().to_owned(), Arc::downgrade(&shared)); + } + let (stop, mut stopped) = oneshot::channel(); + let task = tokio::spawn(async move { + let mut interval = tokio::time::interval(period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = &mut stopped => break, + _ = interval.tick() => {}, + } + let path = home.root().to_owned(); + let result = tokio::task::spawn_blocking(move || sample(&path)) + .await + .unwrap_or_else(|error| Err(io::Error::other(error.to_string()))); + shared + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .record(result); + } + }); + Self { + stop: Some(stop), + task, + } + } + + pub(crate) async fn finish(mut self) { + if let Some(stop) = self.stop.take() { + let _ = stop.send(()); + } + if tokio::time::timeout(Duration::from_millis(500), &mut self.task) + .await + .is_err() + { + self.task.abort(); + let _ = (&mut self.task).await; + } + } +} + +impl Drop for HostDiskService { + fn drop(&mut self) { + self.task.abort(); + } +} + +#[cfg(test)] +#[path = "server_host_disk_tests.rs"] +mod tests; diff --git a/crates/cccc-daemon/src/server_host_disk_tests.rs b/crates/cccc-daemon/src/server_host_disk_tests.rs new file mode 100644 index 000000000..80d421faa --- /dev/null +++ b/crates/cccc-daemon/src/server_host_disk_tests.rs @@ -0,0 +1,139 @@ +use super::*; + +fn captured(id: &str) -> Space { + let rows: serde_json::Value = + serde_json::from_str(include_str!("../tests/fixtures/host-disk-samples.json")) + .expect("captured samples"); + let row = rows + .as_array() + .expect("samples array") + .iter() + .find(|row| row["id"] == id) + .expect("sample id"); + let block = row["f_frsize"].as_u64().expect("block size"); + let space = Space { + available_bytes: row["f_bavail"].as_u64().expect("available blocks") * block, + total_bytes: row["f_blocks"].as_u64().expect("total blocks") * block, + }; + assert_eq!(space.available_bytes, row["available_bytes"]); + assert_eq!(space.total_bytes, row["total_bytes"]); + space +} + +#[test] +fn captured_samples_enter_low_and_require_recovery() { + let mut diagnostic = Diagnostic::default(); + assert_eq!(diagnostic.state, "unknown"); + for (sample, expected) in [ + ("volume_a", "healthy"), + ("volume_b", "healthy"), + ("volume_c", "low"), + ("volume_b", "low"), + ("volume_a", "healthy"), + ] { + diagnostic.record(Ok(captured(sample))); + assert_eq!(diagnostic.state, expected); + assert!(diagnostic.error.is_none()); + } + assert_eq!(diagnostic.attempts, 5); +} + +#[test] +fn captured_capacity_at_each_policy_boundary_is_not_low() { + // The observations stay verbatim. Place the policy boundaries at two + // captured values to test strict entry and inclusive recovery precisely. + let mut diagnostic = Diagnostic { + low_bytes: captured("volume_b").available_bytes, + recovery_bytes: captured("volume_a").available_bytes, + ..Default::default() + }; + diagnostic.record(Ok(captured("volume_b"))); + assert_eq!(diagnostic.state, "healthy"); + diagnostic.record(Ok(captured("volume_c"))); + assert_eq!(diagnostic.state, "low"); + diagnostic.record(Ok(captured("volume_b"))); + assert_eq!(diagnostic.state, "low"); + diagnostic.record(Ok(captured("volume_a"))); + assert_eq!(diagnostic.state, "healthy"); +} + +#[test] +fn failed_sample_is_visible_and_cannot_clear_low_hysteresis() { + let temp = tempfile::tempdir().expect("probe directory"); + let mut diagnostic = Diagnostic::default(); + diagnostic.record(Ok(captured("volume_c"))); + let old = diagnostic.clone(); + diagnostic.record(sample(&temp.path().join("missing"))); + assert_eq!(diagnostic.state, "unavailable"); + assert!(diagnostic.error.is_some()); + assert_eq!(diagnostic.sampled_at, old.sampled_at); + assert_eq!( + diagnostic + .space + .as_ref() + .expect("last sample") + .available_bytes, + old.space.as_ref().expect("old sample").available_bytes + ); + diagnostic.record(Ok(captured("volume_b"))); + assert_eq!(diagnostic.state, "low"); + assert!(diagnostic.error.is_none()); + diagnostic.record(Ok(captured("volume_a"))); + assert_eq!(diagnostic.state, "healthy"); +} + +#[test] +fn small_filesystem_probe_succeeds_and_reports_a_missing_path() { + let temp = tempfile::tempdir().expect("probe directory"); + std::fs::write(temp.path().join("one-byte"), b"x").expect("small write"); + let result = sample(temp.path()).expect("real filesystem sample"); + assert!(result.total_bytes > 0); + assert!(result.available_bytes <= result.total_bytes); + assert!(sample(&temp.path().join("missing")).is_err()); +} + +async fn wait_attempts(home: &HomeLayout, attempts: u64) -> Diagnostic { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + if let Some(value) = snapshot(home) + && value.attempts >= attempts + { + return value; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("independent sampler progressed") +} + +#[tokio::test] +async fn host_timer_samples_without_groups_or_notice_ticks() { + let temp = tempfile::tempdir().expect("home"); + let home = HomeLayout::from_path(temp.path()).expect("home layout"); + let service = HostDiskService::with_interval(home.clone(), Duration::from_millis(20)); + // No Group, automation scheduler, notice tick or status poll is needed. + tokio::time::sleep(Duration::from_millis(80)).await; + let diagnostic = wait_attempts(&home, 2).await; + assert!(diagnostic.space.is_some()); + assert!(diagnostic.error.is_none()); + service.finish().await; + assert!(snapshot(&home).is_none(), "shutdown releases cached state"); +} + +#[tokio::test] +async fn status_reads_are_cached_and_home_scoped() { + let temp = tempfile::tempdir().expect("home"); + let home = HomeLayout::from_path(temp.path()).expect("home layout"); + let other = HomeLayout::from_path(temp.path().join("other")).expect("other home"); + let service = HostDiskService::start(home.clone()); + let initial = wait_attempts(&home, 1).await; + for _ in 0..100 { + let current = snapshot(&home).expect("cached status"); + assert_eq!(current.attempts, initial.attempts); + assert_eq!(current.sampled_at, initial.sampled_at); + assert!(snapshot(&other).is_none()); + } + service.finish().await; + assert!(snapshot(&home).is_none()); +} diff --git a/crates/cccc-daemon/tests/fixtures/host-disk-samples.README.md b/crates/cccc-daemon/tests/fixtures/host-disk-samples.README.md new file mode 100644 index 000000000..4bf2623a9 --- /dev/null +++ b/crates/cccc-daemon/tests/fixtures/host-disk-samples.README.md @@ -0,0 +1,13 @@ +# Captured filesystem samples + +Three real Linux filesystem samples captured with `os.statvfs` on +2026-10-22. Filesystem paths are omitted and the records use aliases. +All timestamps, block counts and byte counts are unchanged. +`available_bytes = f_bavail * f_frsize`; `total_bytes = f_blocks * f_frsize`. + +These are policy inputs, not a claimed history of one volume changing size. +The low/recovery tests replay them to exercise hysteresis without filling a +filesystem or requiring spare capacity. Exact-boundary tests place policy +thresholds at captured values, rather than manufacturing capacity samples. +The independent filesystem probe uses a small temporary directory and a +missing child path; it asserts no free-space floor. diff --git a/crates/cccc-daemon/tests/fixtures/host-disk-samples.json b/crates/cccc-daemon/tests/fixtures/host-disk-samples.json new file mode 100644 index 000000000..01daf7f55 --- /dev/null +++ b/crates/cccc-daemon/tests/fixtures/host-disk-samples.json @@ -0,0 +1,32 @@ +[ + { + "id": "volume_a", + "captured_at": "2026-10-22T19:21:11.793841+00:00", + "f_frsize": 4096, + "f_blocks": 78687103, + "f_bavail": 9616834, + "f_bfree": 12827970, + "available_bytes": 39390552064, + "total_bytes": 322302373888 + }, + { + "id": "volume_b", + "captured_at": "2026-10-22T19:21:11.793878+00:00", + "f_frsize": 4096, + "f_blocks": 802229, + "f_bavail": 801974, + "f_bfree": 801974, + "available_bytes": 3284885504, + "total_bytes": 3285929984 + }, + { + "id": "volume_c", + "captured_at": "2026-10-22T19:22:11.226861+00:00", + "f_frsize": 4096, + "f_blocks": 1280, + "f_bavail": 1280, + "f_bfree": 1280, + "available_bytes": 5242880, + "total_bytes": 5242880 + } +] diff --git a/docs/standards/CCCC_DAEMON_IPC_V1.md b/docs/standards/CCCC_DAEMON_IPC_V1.md index 5b7eb154a..e09db1cd2 100644 --- a/docs/standards/CCCC_DAEMON_IPC_V1.md +++ b/docs/standards/CCCC_DAEMON_IPC_V1.md @@ -367,12 +367,42 @@ Result: compatibility?: string; build?: { source_id: string }; executable?: string | null; + diagnostics?: { + disk: { + state: "unknown" | "healthy" | "low" | "unavailable"; + sample_interval_seconds: number; + low_bytes: number; + recovery_bytes: number; + attempts: number; + last_attempt_at: string | null; + sampled_at: string | null; + space: { available_bytes: number; total_bytes: number } | null; + error: string | null; + } | null; + }; } ``` Notes: - SDK-compatible daemons MUST return `ipc_v: 1`; omitting it is interpreted as IPC version `0`. - The bundled daemon MUST identify its implementation as `rust`. +- `diagnostics.disk` is passive host information for the filesystem containing + `CCCC_HOME`, sampled once on daemon startup and every 60 seconds thereafter. + One daemon-owned sampler runs independently of Groups, notice ticks and + status requests; missed intervals are skipped. Reading `ping` never samples. + Available bytes are those allocatable by the daemon's user, excluding + filesystem reservations that user cannot allocate. There is no disk Mail, + foreman notice, notification owner, ledger event or notice-policy change. +- The diagnostic enters `low` below 3 GiB available and returns to `healthy` + at or above 4 GiB. The 1 GiB recovery margin prevents flapping near the low + boundary; these diagnostic constants and the cadence are exposed in every + sample. They impose no write admission or free-space requirement. + A failed probe sets `unavailable` with an error, retaining the last successful + sample, its `sampled_at`, and the previous low/recovery state. Only a successful + sample can establish recovery. `last_attempt_at` and `attempts` include failed + probes. Before the first result the state is `unknown`; timestamps and space + are null. A null disk object means no active sampler in that process/home. + Samples are in memory and reset when the daemon exits. No disk cache is written. - `compatibility`, when present, is an implementation-specific compatibility identity; clients MUST NOT infer compatibility from the implementation name alone. - SDK-compatible daemons MUST return a `capabilities` feature map. The native daemon advertises supported `events_stream`, `remote_access`, optional