diff --git a/INVARIANTS.md b/INVARIANTS.md index e2f9013e..25c8f6d5 100644 --- a/INVARIANTS.md +++ b/INVARIANTS.md @@ -18,8 +18,8 @@ materialization, messaging, DING, or presence must preserve them. | **Mutation-only filesystem wakeups** | Supervisor and DING filesystem watchers ignore read/open access events and wake early only for create, modify, rename, or remove events. Their own catalog and inbox reads therefore cannot bypass the bounded timer cadence or form a Linux inotify CPU loop. | `src/watch.rs::only_mutations_wake_watch_loops`; `src/watch.rs::linux_reads_are_silent_but_real_mutations_wake`; `src/ding/mod.rs::idle_ding_does_not_spin_on_its_own_inbox_reads`; `src/run.rs::idle_supervisor_does_not_spin_on_its_own_catalog_reads` | | **Bounded DING PTY probe churn** | An unsafe or active composer retains its FIFO notice but deferred delivery retries use a bounded backoff, so each inbox poll cannot spawn another short-lived PTY probe. | `src/ding/mod.rs::deferred_delivery_backoff_bounds_short_lived_pty_attempts` | | **Agent-declared presence discipline** | The shipped bus contract requires agents to declare `busy` before executing work, use `available` only while yielding or ready, and reserve `dnd` for an explicit hold. Both native harnesses materialize that contract. Busy remains observable but does not suppress DING; fresh `dnd` is the only delivery gate. | `tests/native_only.rs::clean_path_executes_the_maintained_native_authoring_guide`; `src/ding/mod.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry` | -| **Stable roster JSON** | `st2 agents --json [--enrich]` preserves field names, order, null handling, presence, typed desired state and rationale, the retirement compatibility projection, opaque declared Resource descriptors, activity, and inbox counts. Presence remains independent from desired lifecycle. | `src/agents.rs::agents_json_has_stable_wire_shape`; `src/agents.rs::agents_json_preserves_opaque_declared_resource_descriptors`; `tests/status_agents.rs::roster_json_and_human_output_distinguish_retirement_from_presence`; `tests/status_agents.rs::roster_keeps_presence_separate_from_suspended_desired_state` | -| **Agent-declared presence** | Refresh preserves non-DND declared status and only advances liveness; a missing status starts as `available`, while `dnd` is never refreshed and an unrefreshed declaration ages to `unknown`. | `src/status.rs::refresh_preserves_value_and_bumps_mtime`; `src/status.rs::refresh_leaves_dnd_to_age_out`; `src/status.rs::refresh_missing_writes_available_default`; `src/status.rs::stale_mtime_reads_as_unknown_regardless_of_contents` | +| **Stable roster JSON** | `st2 agents --json [--enrich]` preserves field names, order, null handling, presence, typed desired state and rationale, the retirement compatibility projection, opaque declared Resource descriptors, origin-timed activity, and inbox counts. Presence remains independent from desired lifecycle. | `src/agents.rs::agents_json_has_stable_wire_shape`; `src/agents.rs::agents_json_preserves_opaque_declared_resource_descriptors`; `tests/status_agents.rs::roster_json_and_human_output_distinguish_retirement_from_presence`; `tests/status_agents.rs::roster_keeps_presence_separate_from_suspended_desired_state` | +| **Agent-declared presence** | Refresh preserves non-DND declared status and advances the embedded heartbeat; a missing status starts as `available`. Legacy DND migrates without renewing its hold, version 1 DND is never refreshed, and stale, malformed, or implausibly future heartbeats read as `unknown`. | `src/status.rs::refresh_preserves_value_and_changes_heartbeat_bytes`; `src/status.rs::refresh_upgrades_legacy_dnd_without_renewing_the_hold`; `src/status.rs::refresh_missing_writes_available_default`; `src/status.rs::version_1_staleness_and_future_skew_are_bounded`; `src/status.rs::malformed_versioned_record_is_unknown_without_mtime_fallback` | | **Retirement health** | A retired declaration is healthy only after every declared task ID is absent. Any live or dead declared task record reports incomplete retirement; retired declarations do not require presence. Live declarations retain their existing task and presence checks. | `tests/doctor.rs::retired_declaration_is_healthy_when_tasks_and_presence_are_absent`; `tests/doctor.rs::retired_declaration_is_unhealthy_while_a_declared_task_is_alive`; `tests/doctor.rs::retired_declaration_is_unhealthy_while_a_dead_task_record_remains` | | **Suspension health** | A suspended declaration is healthy when no declared task is live and every retained dead record is explicitly keep-pinned. It requires no presence, but this weaker result never proves retirement. Resume preserves ordinary keep and adopt-only policy. | `tests/doctor.rs::suspended_declaration_is_healthy_when_tasks_are_absent_without_presence`; `tests/doctor.rs::suspended_declaration_distinguishes_live_dead_keep_and_dead_nonkeep`; `tests/reconcile.rs::resuming_uses_ordinary_reconcile_and_does_not_override_keep` | | **Crash loops surface** | A task parked by a fail-mode restart policy notifies its supervisor once over the bus. | `tests/run.rs::surface_crash_loop_notifies_the_supervisor_over_the_bus` | diff --git a/docs/vrs/spec.md b/docs/vrs/spec.md index 561aa7ce..5ca542da 100644 --- a/docs/vrs/spec.md +++ b/docs/vrs/spec.md @@ -677,6 +677,150 @@ atomic inbox file → DING attempt → agent reads → archive receipt - **R10:** Fleet identities are agents. General-purpose identity kinds are unsupported. +### Presence record and freshness (R08) + +This section answers the presence part of DQ3. The version 1 implementation +follows this contract. + +#### Version 1 record + +The presence record path is `/status`. + +The status file uses this exact version 1 shape: + +```text +available +v1 1785802653486 +``` + +Line one is one settable state: `offline`, `available`, `busy`, `away`, or +`dnd`. `unknown` remains derived and is never written. + +Line two is `v1`, one ASCII space, and an unsigned base-10 timestamp. The +timestamp counts milliseconds from the Unix epoch. + +The record ends with one newline. It has no other non-empty lines. Both lines +form one atomic record. + +The state remains on line one for old readers. An old reader can ignore line +two and continue to parse the state. + +#### Writers and atomicity + +`st2 status --set` writes the requested state with the current timestamp. A +live DING sidecar refreshes valid non-DND records every five minutes. + +A missing record becomes `available` with the current timestamp. DING does not +refresh `dnd`, `unknown`, or malformed records. + +Every new writer emits version 1. It writes a temporary sibling and atomically +renames the complete record over the target. + +A healthy periodic refresh changes the timestamp bytes. Replication can order +that content change without using the source file mtime. + +#### Clock, freshness, and skew + +The timestamp uses the writer's UTC wall clock. A monotonic clock cannot cross +a process restart or a host boundary. + +Participating hosts must keep their UTC clocks within sixty seconds. A larger +clock error makes cross-host presence unknown. + +The stale interval remains fifteen minutes. A valid record is fresh while its +age is less than fifteen minutes. + +A record becomes `unknown` when its age reaches fifteen minutes. This rule +applies to every settable state, including `offline` and `dnd`. + +A timestamp up to sixty seconds in the reader's future is allowed. The reader +uses zero age for this bounded future value. + +A timestamp more than sixty seconds in the future produces `unknown`. The +reader does not use file mtime as a fallback for malformed version 1. + +Current readers treat a future legacy status mtime as fresh because they cannot +calculate its age. A sufficiently future `dnd` mtime can therefore suppress +delivery until the reader's clock catches up. The version 1 skew rules close +this defect. They clamp only bounded future time and map larger future time to +`unknown`. + +An unrecognized state still produces `offline`. A literal `unknown` produces +`unknown`. A valid state with a malformed version, timestamp, or extra line +also produces `unknown`. + +The sixty-second allowance is smaller than the five-minute refresh margin. It +can extend a fresh DND hold by no more than sixty seconds. + +#### Why readers use origin time + +st2 does not require one catalog transport. Fabric is preferred, and Git over +SSH or a plain copy remains supported. + +Git does not preserve file modification times. A checkout gives files the +checkout time. Therefore, presence freshness lives in record bytes. No +supported transport must preserve file metadata. + +Replica arrival time measures transport delay, not agent activity. The +embedded writer time protects presence freshness, DND expiry, and the status +contribution to `lastActivity`. + +The same reason applies to the context boot freshness check. A replica arrival +must not make old context appear fresh. This proposal does not change the +context record. + +#### DND behavior + +A fresh `dnd` record suppresses DING delivery. The sidecar leaves its timestamp +unchanged, so an abandoned hold ages out. + +A stale or invalid DND record does not suppress delivery. It reads as +`unknown`, which preserves the existing fresh-DND rule. + +Replication delay cannot renew a DND hold. The reader uses the embedded write +time, not the replica materialization time. + +#### Legacy rollout + +A legacy record contains one valid state line and no version line. The first +version 1 reader release uses legacy file mtime for freshness. + +Version 1 writers never emit a legacy record. A live non-DND sidecar upgrades +its legacy record at its next five-minute refresh. + +A version 1 sidecar upgrades a legacy DND record once. It uses the legacy mtime +as the embedded timestamp, so the migration cannot renew the hold. + +After that migration, the sidecar does not refresh DND. If the legacy mtime is +unavailable, the sidecar leaves the record unchanged. + +A malformed two-line record is not legacy. Readers must not hide a bad version +1 record behind the legacy mtime fallback. + +Fallback removal is a separate reviewed change. Removal requires all three +receipts below: + +1. Every supported deployed status writer emits version 1. +2. Two fleet scans, separated by fifteen minutes, find no active legacy record. +3. No supported or retained rollback binary can emit a legacy record. + +After removal, a one-line record produces `unknown`. No presence freshness +decision then depends on status file mtime. + +#### `lastActivity` + +For a version 1 status record, `lastActivity` uses the embedded timestamp. It +does not use the replica materialization mtime. + +The reader clamps an allowed future timestamp to its current time. It omits a +malformed version 1 timestamp from the activity calculation. + +Inbox and archive entries continue to use their local file mtimes. During the +legacy window, a one-line status record also contributes its file mtime. + +This choice reports when the agent wrote its heartbeat. A delayed replica +cannot make an old heartbeat appear to be new agent activity. + ## Provider session-start restoration (R07, R09, R17, R33) ```text @@ -794,11 +938,11 @@ the resident supervisor continues to reconcile the complete local catalog. changes may defer delivery. Resolve the remaining gap with a stronger evented signal or other measured classifier; a small on-device model is an optional experiment, not a required architecture. -- **DQ3 Catalog agent state:** Define the catalog paths, schemas, freshness - rules, and atomic update semantics for presence, activity status, current - plan, and current plan step. Prove that stale state is distinguishable and - that a supervisor can follow plan progress without inspecting a PTY before - adding the shape to `AGENT-SPEC.md`. +- **DQ3 Remaining catalog agent state:** The R08 presence record above + defines the presence path, schema, freshness, and atomic update rules. + Activity status, current plan, and current plan step remain undefined. Prove + their stale-state and supervisor-following behavior before adding their shape + to `AGENT-SPEC.md`. - **DQ4 Relaunch boundary (R29-R30):** Preserve R11's nondisruptive adoption while making launch drift visible. For each declared task, derive the desired launch fingerprint from a deterministic, versioned encoding of only: diff --git a/src/agents.rs b/src/agents.rs index 767d652f..592f23ba 100644 --- a/src/agents.rs +++ b/src/agents.rs @@ -4,7 +4,7 @@ use std::fs; use std::path::{Path, PathBuf}; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::UNIX_EPOCH; use serde::Serialize; @@ -31,8 +31,8 @@ pub struct AgentRow { pub desired_state_reason: Option, /// Typed Resource bindings declared directly by the agent. pub resources: Vec, - /// Newest mtime (unix ms) across the agent's inbox, archive, and status file; `None` if nothing - /// has been touched. `--enrich` only. + /// Newest activity time across inbox, archive, and status. Version 1 status uses its embedded + /// writer timestamp; message files and legacy status use local mtime. `--enrich` only. pub last_activity_ms: Option, /// Count of canonical message files in the agent's inbox. `--enrich` only. pub inbox: usize, @@ -62,7 +62,7 @@ pub fn roster_from_discovered(found: &Discovered, this_host: &str) -> Vec usize { .unwrap_or(0) } -/// Newest mtime (unix ms) across the agent's inbox files, archive files, and status file. `None` if -/// none of those exist. -fn newest_mtime_ms(agent_dir: &Path) -> Option { +/// Newest activity time across status and message state. A version 1 status contributes its origin +/// timestamp. Inbox, archive, and legacy status retain their local-mtime behavior. +fn newest_activity_ms(agent_dir: &Path) -> Option { let mut candidates: Vec = Vec::new(); for dir in [ message::inbox_dir(agent_dir), @@ -161,20 +161,20 @@ fn newest_mtime_ms(agent_dir: &Path) -> Option { candidates.extend(rd.flatten().map(|e| e.path())); } } - candidates.push(status::status_path(agent_dir)); - let mut newest: Option = None; + let mut newest = status::activity_time_ms(&status::status_path(agent_dir)); for p in candidates { if let Ok(m) = fs::metadata(&p) && let Ok(t) = m.modified() - && newest.is_none_or(|n| t > n) + && let Ok(duration) = t.duration_since(UNIX_EPOCH) { - newest = Some(t); + let timestamp = duration.as_secs_f64() * 1000.0; + if newest.is_none_or(|current| timestamp > current) { + newest = Some(timestamp); + } } } newest - .and_then(|t| t.duration_since(UNIX_EPOCH).ok()) - .map(|d| d.as_secs_f64() * 1000.0) } #[cfg(test)] @@ -232,14 +232,7 @@ mod tests { #[test] fn agents_json_preserves_opaque_declared_resource_descriptors() { - let mut resource_row = row( - "hetz.worker", - State::Available, - None, - false, - None, - 0, - ); + let mut resource_row = row("hetz.worker", State::Available, None, false, None, 0); resource_row.resources.push( Resource::new( "work".into(), diff --git a/src/ding/mod.rs b/src/ding/mod.rs index c7d94af4..a817abe8 100644 --- a/src/ding/mod.rs +++ b/src/ding/mod.rs @@ -868,7 +868,7 @@ impl SessionWatch { pub struct DingConfig { /// Fallback poll cadence and liveness-check cadence. pub poll: Duration, - /// Presence mtime refresh cadence while the target session is alive. + /// Presence heartbeat refresh cadence while the target session is alive. pub status_refresh: Duration, } @@ -3110,11 +3110,9 @@ Enter to select · ↑/↓ to navigate · Esc to cancel"; flush_without_catalog(Some(&status_path), &mut pending, &poker); assert_eq!(pending.len(), 1, "fresh dnd suppresses delivery"); - let stale = std::time::SystemTime::now() - status::STATUS_STALE - Duration::from_secs(1); - std::fs::File::open(&status_path) - .unwrap() - .set_modified(stale) - .unwrap(); + let stale_ms = crate::message::now_ms() + - u64::try_from((status::STATUS_STALE + Duration::from_secs(1)).as_millis()).unwrap(); + std::fs::write(&status_path, format!("dnd\nv1 {stale_ms}\n")).unwrap(); flush_without_catalog(Some(&status_path), &mut pending, &poker); assert!( pending.is_empty(), diff --git a/src/status.rs b/src/status.rs index e0650359..4f086609 100644 --- a/src/status.rs +++ b/src/status.rs @@ -1,25 +1,29 @@ //! Native per-agent presence status. //! -//! A `status` file (sibling of `agent.kdl` in the agent's dir) holds exactly one word: one of the -//! settable states. `unknown` is DERIVED, never written — a status whose mtime is older than -//! [`STATUS_STALE`] reads as `unknown` regardless of its contents, so a crashed/gone agent stops -//! reading as its last set value. Reads are permissive (missing → `offline`, corrupt → `offline`). -//! Writes are atomic (tmp + rename) so a concurrent reader never sees a partial file. The ding -//! periodically re-writes an agent's own non-DND status to bump the mtime *preserving the value* so -//! a healthy-but-idle agent never rots to `unknown` (the failure that read the whole fleet `unknown` -//! for 45 min). `dnd` is intentionally not refreshed: an abandoned hold ages to `unknown` after the -//! same stale window. The vocabulary, stale window, and file semantics are stable. +//! A `status` file (sibling of `agent.kdl` in the agent's dir) stores a settable state and an +//! embedded Unix-millisecond heartbeat. `unknown` is DERIVED, never written. A heartbeat older than +//! [`STATUS_STALE`] reads as `unknown`, so a crashed agent stops reading as its last set value. +//! Writes are atomic (tmp + rename), and the DING sidecar refreshes valid non-DND records every +//! [`STATUS_REFRESH`]. Legacy one-line records use mtime during the bounded migration. New writers +//! always emit version 1, whose freshness survives transports that do not preserve file metadata. +//! `dnd` is not refreshed after migration, so an abandoned hold ages to `unknown`. use std::fs; +use std::io::Read as _; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::{Duration, SystemTime}; +use std::time::{Duration, UNIX_EPOCH}; -/// A status file older than this reads as `unknown` no matter its contents. +/// A valid status heartbeat at least this old reads as `unknown`. pub const STATUS_STALE: Duration = Duration::from_secs(15 * 60); /// How often a live agent's status should be refreshed to stay inside the stale window — 5 min gives /// a 3× safety margin (two missed refreshes before `unknown`). pub const STATUS_REFRESH: Duration = Duration::from_secs(5 * 60); +/// Maximum accepted positive difference between a writer's UTC clock and the reader's clock. +pub const STATUS_FUTURE_SKEW: Duration = Duration::from_secs(60); + +const RECORD_VERSION: &str = "v1"; +const RECORD_PREFIX: &str = "v1 "; /// Presence state. `Unknown` is derived from staleness and is never written to disk. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -72,38 +76,45 @@ pub fn status_path(agent_dir: &Path) -> PathBuf { agent_dir.join("status") } -/// Read an agent's effective presence. Order matters: missing file → `offline`; -/// mtime older than [`STATUS_STALE`] → `unknown` (regardless of contents); unreadable → `offline`; -/// else the first line's word if valid, else `offline` (never trust a corrupt file). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ParsedRecord { + Legacy(State), + Version1 { state: State, written_at_ms: u64 }, + LiteralUnknown, + InvalidState, + Malformed, +} + +/// Read an agent's effective presence. Missing or unreadable files produce `offline`. Version 1 +/// uses its embedded timestamp. A valid legacy line uses file mtime during migration. Malformed +/// versioned records fail closed to `unknown` and never fall back to mtime. pub fn read_state(status_path: &Path) -> State { - let meta = match fs::metadata(status_path) { - Ok(m) => m, - Err(_) => return State::Offline, // missing - }; - if let Ok(mtime) = meta.modified() - && let Ok(age) = SystemTime::now().duration_since(mtime) - && age >= STATUS_STALE - { - return State::Unknown; // stale → not trustworthy - } - let raw = match fs::read_to_string(status_path) { - Ok(r) => r, + let (record, legacy_mtime_ms) = match read_record(status_path) { + Ok(record) => record, Err(_) => return State::Offline, }; - let first = raw.lines().next().unwrap_or("").trim(); - State::parse_any(first).unwrap_or(State::Offline) + read_parsed_at(record, legacy_mtime_ms, crate::message::now_ms()) } -/// Set an agent's presence to a settable `state`, atomically (tmp sibling + rename), writing -/// `\n`. Creates the agent dir if missing. +/// Set an agent's presence to a version 1 record with the current timestamp. The write is atomic +/// (temporary sibling + rename) and creates the agent directory when needed. pub fn set_state(status_path: &Path, state: State) -> anyhow::Result<()> { - write_atomic(status_path, state.as_str()) + write_record(status_path, state, crate::message::now_ms()) +} + +/// Timestamp contributed by the status record to `agents --enrich`. Version 1 returns its embedded +/// writer time, clamped to the reader's current time within the allowed future skew. Legacy records +/// return mtime during migration. Invalid records and excessive future skew contribute nothing. +pub fn activity_time_ms(status_path: &Path) -> Option { + let (record, legacy_mtime_ms) = read_record(status_path).ok()?; + activity_time_at(record, legacy_mtime_ms, crate::message::now_ms()) + .map(|timestamp| timestamp as f64) } /// Outcome of a [`refresh`] call. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RefreshOutcome { - /// File present + a valid settable state → re-wrote the same value, mtime bumped. + /// A valid record received a new heartbeat or completed its one-time legacy upgrade. Refreshed, /// File recorded `dnd` → left untouched so an abandoned hold ages out. LeftDnd, @@ -117,43 +128,171 @@ pub enum RefreshOutcome { Error, } -/// Bump an agent's status mtime so a live-but-idle agent never rots to `unknown`, preserving the -/// recorded value. Missing → write `available`. A valid non-DND value → re-write the same value. -/// `dnd`, `unknown`, or corrupt → leave untouched (never renew a hold or invent a value). Atomic. +/// Refresh the embedded heartbeat for a live agent while preserving its state. Missing writes +/// `available`. A legacy non-DND record upgrades with the current time. A legacy DND record upgrades +/// once with its old mtime. Version 1 DND, `unknown`, and malformed records remain untouched. pub fn refresh(status_path: &Path) -> RefreshOutcome { - if !status_path.exists() { - return match write_atomic(status_path, State::Available.as_str()) { - Ok(()) => RefreshOutcome::WroteDefault, - Err(_) => RefreshOutcome::Error, - }; - } - let raw = match fs::read_to_string(status_path) { - Ok(r) => r, + let (record, legacy_mtime_ms) = match read_record(status_path) { + Ok(record) => record, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return match write_record(status_path, State::Available, crate::message::now_ms()) { + Ok(()) => RefreshOutcome::WroteDefault, + Err(_) => RefreshOutcome::Error, + }; + } Err(_) => return RefreshOutcome::Error, }; - let first = raw.lines().next().unwrap_or("").trim(); - if first == "unknown" { - return RefreshOutcome::LeftUnknown; + match record { + ParsedRecord::Version1 { + state: State::Dnd, .. + } => RefreshOutcome::LeftDnd, + ParsedRecord::Version1 { state, .. } => { + match write_record(status_path, state, crate::message::now_ms()) { + Ok(()) => RefreshOutcome::Refreshed, + Err(_) => RefreshOutcome::Error, + } + } + ParsedRecord::Legacy(State::Dnd) => { + let Some(written_at_ms) = legacy_mtime_ms else { + return RefreshOutcome::LeftDnd; + }; + match write_record(status_path, State::Dnd, written_at_ms) { + Ok(()) => RefreshOutcome::Refreshed, + Err(_) => RefreshOutcome::Error, + } + } + ParsedRecord::Legacy(state) => { + match write_record(status_path, state, crate::message::now_ms()) { + Ok(()) => RefreshOutcome::Refreshed, + Err(_) => RefreshOutcome::Error, + } + } + ParsedRecord::LiteralUnknown => RefreshOutcome::LeftUnknown, + ParsedRecord::InvalidState | ParsedRecord::Malformed => RefreshOutcome::LeftCorrupt, + } +} + +fn read_record(path: &Path) -> std::io::Result<(ParsedRecord, Option)> { + let mut file = fs::File::open(path)?; + let mut raw = String::new(); + file.read_to_string(&mut raw)?; + let record = parse_record(&raw); + let legacy_mtime_ms = matches!(record, ParsedRecord::Legacy(_)) + .then(|| file_mtime_ms(&file)) + .flatten(); + Ok((record, legacy_mtime_ms)) +} + +fn parse_record(raw: &str) -> ParsedRecord { + let has_final_newline = raw.ends_with('\n'); + let body = raw.strip_suffix('\n').unwrap_or(raw); + let mut lines = body.split('\n'); + let state = match State::parse_any(lines.next().unwrap_or("")) { + Some(State::Unknown) => return ParsedRecord::LiteralUnknown, + Some(state) => state, + None => return ParsedRecord::InvalidState, + }; + let Some(version_line) = lines.next() else { + return ParsedRecord::Legacy(state); + }; + if !has_final_newline || lines.next().is_some() { + return ParsedRecord::Malformed; } - if first == "dnd" { - return RefreshOutcome::LeftDnd; + let Some(timestamp) = version_line.strip_prefix(RECORD_PREFIX) else { + return ParsedRecord::Malformed; + }; + if timestamp.is_empty() || !timestamp.bytes().all(|byte| byte.is_ascii_digit()) { + return ParsedRecord::Malformed; } - match State::parse_settable(first) { - Some(s) => match write_atomic(status_path, s.as_str()) { - Ok(()) => RefreshOutcome::Refreshed, - Err(_) => RefreshOutcome::Error, + match timestamp.parse::() { + Ok(written_at_ms) => ParsedRecord::Version1 { + state, + written_at_ms, }, - None => RefreshOutcome::LeftCorrupt, + Err(_) => ParsedRecord::Malformed, + } +} + +fn read_parsed_at(record: ParsedRecord, legacy_mtime_ms: Option, now_ms: u64) -> State { + match record { + ParsedRecord::Legacy(state) => legacy_mtime_ms.map_or(State::Unknown, |written_at_ms| { + effective_state_at(state, written_at_ms, now_ms) + }), + ParsedRecord::Version1 { + state, + written_at_ms, + } => effective_state_at(state, written_at_ms, now_ms), + ParsedRecord::LiteralUnknown | ParsedRecord::Malformed => State::Unknown, + ParsedRecord::InvalidState => State::Offline, } } +fn effective_state_at(state: State, written_at_ms: u64, now_ms: u64) -> State { + if written_at_ms > now_ms { + return if written_at_ms - now_ms <= duration_ms(STATUS_FUTURE_SKEW) { + state + } else { + State::Unknown + }; + } + if now_ms - written_at_ms >= duration_ms(STATUS_STALE) { + State::Unknown + } else { + state + } +} + +fn activity_time_at( + record: ParsedRecord, + legacy_mtime_ms: Option, + now_ms: u64, +) -> Option { + let written_at_ms = match record { + ParsedRecord::Legacy(_) => legacy_mtime_ms?, + ParsedRecord::Version1 { written_at_ms, .. } => written_at_ms, + ParsedRecord::LiteralUnknown | ParsedRecord::InvalidState | ParsedRecord::Malformed => { + return None; + } + }; + if written_at_ms > now_ms { + (written_at_ms - now_ms <= duration_ms(STATUS_FUTURE_SKEW)).then_some(now_ms) + } else { + Some(written_at_ms) + } +} + +fn file_mtime_ms(file: &fs::File) -> Option { + file.metadata() + .ok()? + .modified() + .ok()? + .duration_since(UNIX_EPOCH) + .ok() + .and_then(|duration| u64::try_from(duration.as_millis()).ok()) +} + +fn duration_ms(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn write_record(path: &Path, state: State, written_at_ms: u64) -> anyhow::Result<()> { + anyhow::ensure!( + state != State::Unknown, + "unknown is derived and cannot be written" + ); + write_atomic( + path, + &format!("{}\n{RECORD_VERSION} {written_at_ms}\n", state.as_str()), + ) +} + /// Atomic write: a temp sibling + rename, so a concurrent reader sees either the old bytes or the new /// bytes, never a partial file. -fn write_atomic(path: &Path, value: &str) -> anyhow::Result<()> { +fn write_atomic(path: &Path, content: &str) -> anyhow::Result<()> { let dir = path.parent().unwrap_or(Path::new(".")); fs::create_dir_all(dir)?; let tmp = dir.join(tmp_name()); - fs::write(&tmp, format!("{value}\n"))?; + fs::write(&tmp, content)?; // rename over the target — atomic on the same filesystem. if let Err(e) = fs::rename(&tmp, path) { let _ = fs::remove_file(&tmp); // best-effort cleanup @@ -177,7 +316,7 @@ fn tmp_name() -> String { #[cfg(test)] mod tests { use super::*; - use std::time::Duration as Dur; + use std::time::{Duration as Dur, SystemTime}; #[test] fn missing_is_offline() { @@ -196,13 +335,21 @@ mod tests { State::Away, State::Dnd, ] { + let before = crate::message::now_ms(); set_state(&sp, st).unwrap(); + let after = crate::message::now_ms(); assert_eq!(read_state(&sp), st); - // file content is exactly `\n` (wire format). - assert_eq!( - fs::read_to_string(&sp).unwrap(), - format!("{}\n", st.as_str()) - ); + let raw = fs::read_to_string(&sp).unwrap(); + let ParsedRecord::Version1 { + state, + written_at_ms, + } = parse_record(&raw) + else { + panic!("new writer did not emit version 1: {raw:?}"); + }; + assert_eq!(state, st); + assert!((before..=after).contains(&written_at_ms)); + assert_eq!(raw, format!("{}\nv1 {written_at_ms}\n", st.as_str())); } } @@ -211,6 +358,9 @@ mod tests { assert!(State::parse_settable("unknown").is_none()); assert!(State::parse_settable("available").is_some()); assert!(State::parse_settable("bogus").is_none()); + + let tmp = tempfile::tempdir().unwrap(); + assert!(set_state(&status_path(tmp.path()), State::Unknown).is_err()); } #[test] @@ -222,29 +372,101 @@ mod tests { } #[test] - fn stale_mtime_reads_as_unknown_regardless_of_contents() { + fn malformed_versioned_record_is_unknown_without_mtime_fallback() { + let tmp = tempfile::tempdir().unwrap(); + let sp = status_path(tmp.path()); + for raw in [ + "available\nv2 100\n", + "available\nv1 nope\n", + "available\nv1 100", + "available\nv1 100\nextra\n", + "available\n\n", + ] { + fs::write(&sp, raw).unwrap(); + fs::File::open(&sp) + .unwrap() + .set_modified(SystemTime::now()) + .unwrap(); + assert_eq!(read_state(&sp), State::Unknown, "raw: {raw:?}"); + } + } + + #[test] + fn literal_unknown_remains_derived_unknown() { let tmp = tempfile::tempdir().unwrap(); let sp = status_path(tmp.path()); - set_state(&sp, State::Busy).unwrap(); - // Backdate the mtime past the stale window. - let old = SystemTime::now() - STATUS_STALE - Dur::from_secs(60); - let f = fs::File::open(&sp).unwrap(); - f.set_modified(old).unwrap(); + fs::write(&sp, "unknown\n").unwrap(); assert_eq!(read_state(&sp), State::Unknown); } #[test] - fn refresh_preserves_value_and_bumps_mtime() { + fn version_1_uses_embedded_time_instead_of_file_mtime() { + let tmp = tempfile::tempdir().unwrap(); + let sp = status_path(tmp.path()); + let now_ms = crate::message::now_ms(); + write_record(&sp, State::Busy, now_ms).unwrap(); + fs::File::open(&sp) + .unwrap() + .set_modified(SystemTime::now() - STATUS_STALE - Dur::from_secs(60)) + .unwrap(); + assert_eq!(read_state(&sp), State::Busy); + } + + #[test] + fn version_1_staleness_and_future_skew_are_bounded() { + let now_ms = 2_000_000_u64; + let stale_ms = duration_ms(STATUS_STALE); + let skew_ms = duration_ms(STATUS_FUTURE_SKEW); + + for (written_at_ms, expected) in [ + (now_ms - stale_ms + 1, State::Away), + (now_ms - stale_ms, State::Unknown), + (now_ms + skew_ms, State::Away), + (now_ms + skew_ms + 1, State::Unknown), + ] { + let record = ParsedRecord::Version1 { + state: State::Away, + written_at_ms, + }; + assert_eq!(read_parsed_at(record, None, now_ms), expected); + } + } + + #[test] + fn legacy_record_uses_mtime_with_the_same_skew_bound() { let tmp = tempfile::tempdir().unwrap(); let sp = status_path(tmp.path()); - set_state(&sp, State::Busy).unwrap(); - // Backdate → would read unknown… - let old = SystemTime::now() - STATUS_STALE - Dur::from_secs(60); - fs::File::open(&sp).unwrap().set_modified(old).unwrap(); + fs::write(&sp, "busy\n").unwrap(); + + fs::File::open(&sp) + .unwrap() + .set_modified(SystemTime::now() - STATUS_STALE - Dur::from_secs(60)) + .unwrap(); + assert_eq!(read_state(&sp), State::Unknown); + + fs::File::open(&sp) + .unwrap() + .set_modified(SystemTime::now() + STATUS_FUTURE_SKEW + Dur::from_secs(60)) + .unwrap(); assert_eq!(read_state(&sp), State::Unknown); - // …refresh keeps `busy` but bumps mtime, so it reads busy again. + } + + #[test] + fn refresh_preserves_value_and_changes_heartbeat_bytes() { + let tmp = tempfile::tempdir().unwrap(); + let sp = status_path(tmp.path()); + write_record(&sp, State::Busy, 1).unwrap(); assert_eq!(refresh(&sp), RefreshOutcome::Refreshed); assert_eq!(read_state(&sp), State::Busy); + let ParsedRecord::Version1 { + state, + written_at_ms, + } = parse_record(&fs::read_to_string(&sp).unwrap()) + else { + panic!("refreshed record is not version 1"); + }; + assert_eq!(state, State::Busy); + assert!(written_at_ms > 1); } #[test] @@ -253,6 +475,13 @@ mod tests { let sp = status_path(tmp.path()); assert_eq!(refresh(&sp), RefreshOutcome::WroteDefault); assert_eq!(read_state(&sp), State::Available); + assert!(matches!( + parse_record(&fs::read_to_string(&sp).unwrap()), + ParsedRecord::Version1 { + state: State::Available, + .. + } + )); } #[test] @@ -268,12 +497,91 @@ mod tests { assert_eq!(read_state(&sp), State::Dnd); } + #[test] + fn refresh_upgrades_legacy_non_dnd_with_current_heartbeat() { + let tmp = tempfile::tempdir().unwrap(); + let sp = status_path(tmp.path()); + fs::write(&sp, "away\n").unwrap(); + + let before = crate::message::now_ms(); + assert_eq!(refresh(&sp), RefreshOutcome::Refreshed); + let after = crate::message::now_ms(); + let ParsedRecord::Version1 { + state, + written_at_ms, + } = parse_record(&fs::read_to_string(&sp).unwrap()) + else { + panic!("legacy record was not upgraded"); + }; + assert_eq!(state, State::Away); + assert!((before..=after).contains(&written_at_ms)); + } + + #[test] + fn refresh_upgrades_legacy_dnd_without_renewing_the_hold() { + let tmp = tempfile::tempdir().unwrap(); + let sp = status_path(tmp.path()); + fs::write(&sp, "dnd\n").unwrap(); + fs::File::open(&sp) + .unwrap() + .set_modified(SystemTime::now() - STATUS_STALE - Dur::from_secs(60)) + .unwrap(); + let legacy_timestamp = file_mtime_ms(&fs::File::open(&sp).unwrap()).unwrap(); + + assert_eq!(refresh(&sp), RefreshOutcome::Refreshed); + let raw = fs::read_to_string(&sp).unwrap(); + assert_eq!( + parse_record(&raw), + ParsedRecord::Version1 { + state: State::Dnd, + written_at_ms: legacy_timestamp, + } + ); + assert_eq!(read_state(&sp), State::Unknown); + assert_eq!(refresh(&sp), RefreshOutcome::LeftDnd); + assert_eq!(fs::read_to_string(&sp).unwrap(), raw); + } + #[test] fn refresh_leaves_corrupt_untouched() { let tmp = tempfile::tempdir().unwrap(); let sp = status_path(tmp.path()); - fs::write(&sp, "garbage\n").unwrap(); - assert_eq!(refresh(&sp), RefreshOutcome::LeftCorrupt); - assert_eq!(fs::read_to_string(&sp).unwrap(), "garbage\n"); // untouched + for raw in ["garbage\n", "available\nv1 nope\n"] { + fs::write(&sp, raw).unwrap(); + assert_eq!(refresh(&sp), RefreshOutcome::LeftCorrupt); + assert_eq!(fs::read_to_string(&sp).unwrap(), raw); + } + } + + #[test] + fn activity_uses_origin_time_and_clamps_only_allowed_future_skew() { + let now_ms = 2_000_000_u64; + let skew_ms = duration_ms(STATUS_FUTURE_SKEW); + + let record = ParsedRecord::Version1 { + state: State::Available, + written_at_ms: 1234, + }; + assert_eq!(activity_time_at(record, None, now_ms), Some(1234)); + assert_eq!( + activity_time_at(ParsedRecord::Legacy(State::Busy), Some(1234), now_ms), + Some(1234) + ); + + let bounded_future = ParsedRecord::Version1 { + state: State::Available, + written_at_ms: now_ms + skew_ms, + }; + assert_eq!(activity_time_at(bounded_future, None, now_ms), Some(now_ms)); + + let excessive_future = ParsedRecord::Version1 { + state: State::Available, + written_at_ms: now_ms + skew_ms + 1, + }; + assert_eq!(activity_time_at(excessive_future, None, now_ms), None); + assert_eq!( + activity_time_at(ParsedRecord::Malformed, None, now_ms), + None + ); } } diff --git a/tests/catalog_apply.rs b/tests/catalog_apply.rs index c63254fe..9c99ab6c 100644 --- a/tests/catalog_apply.rs +++ b/tests/catalog_apply.rs @@ -37,6 +37,17 @@ fn write_agent_for_host(catalog: &Path, host: &str, identity: &str, retired: boo .unwrap(); } +fn assert_version_1_status(path: &Path, expected_state: &str) { + let raw = fs::read_to_string(path).unwrap(); + let lines: Vec<&str> = raw.lines().collect(); + assert_eq!(lines.len(), 2, "status record: {raw:?}"); + assert_eq!(lines[0], expected_state); + assert!(lines[1].strip_prefix("v1 ").is_some_and(|timestamp| { + !timestamp.is_empty() && timestamp.bytes().all(|byte| byte.is_ascii_digit()) + })); + assert!(raw.ends_with('\n')); +} + fn ensure_external_pty_config(catalog: &Path) { let config = catalog.join("catalog.kdl"); if !config.exists() { @@ -2191,7 +2202,7 @@ fn marker_time_state_routes_existing_orphans_but_never_flat_falls_back_for_new_a .next() .is_some() ); - assert_eq!(fs::read_to_string(old.join("status")).unwrap(), "busy\n"); + assert_version_1_status(&old.join("status"), "busy"); let phantom = send(&catalog, "host.new", "too early"); assert!(!phantom.status.success()); @@ -2452,10 +2463,7 @@ fn marker_time_status_write_remains_bound_to_its_retained_agent_capability() { "{}", String::from_utf8_lossy(&state.stderr) ); - assert_eq!( - fs::read_to_string(retained_host.join("old/status")).unwrap(), - "busy\n" - ); + assert_version_1_status(&retained_host.join("old/status"), "busy"); assert!(!outside.join("old/status").exists()); fs::remove_file(catalog.join("agents/host")).unwrap(); fs::rename(&retained_host, catalog.join("agents/host")).unwrap(); diff --git a/tests/status_agents.rs b/tests/status_agents.rs index 553fa770..682b9eb9 100644 --- a/tests/status_agents.rs +++ b/tests/status_agents.rs @@ -8,7 +8,7 @@ use std::fs; use std::os::unix::fs::PermissionsExt; use std::path::Path; use std::process::Command; -use std::time::{Duration, SystemTime}; +use std::time::Duration; use st2::agents::roster; use st2::message::send_to_inbox; @@ -438,9 +438,9 @@ agent "two" { assert_eq!(selected[0]["name"], "Second Agent Spec"); } -/// A status file older than the stale window projects as `unknown` in the roster, no matter its value. +/// A version 1 heartbeat older than the stale window projects as `unknown` in the roster. #[test] -fn roster_derives_unknown_from_a_stale_status() { +fn roster_derives_unknown_from_a_stale_version_1_heartbeat() { let tmp = tempfile::tempdir().unwrap(); let root = tmp.path(); write(root, "hetz/idle/agent.kdl", &agent_kdl("idle", "hetz")); @@ -450,8 +450,70 @@ fn roster_derives_unknown_from_a_stale_status() { // Fresh → available. assert_eq!(roster(root, "hetz")[0].status, State::Available); - // Backdate the status file past the stale window → unknown. - let old = SystemTime::now() - st2::status::STATUS_STALE - Duration::from_secs(60); - fs::File::open(&sp).unwrap().set_modified(old).unwrap(); + // Backdate the embedded heartbeat past the stale window → unknown. + let stale_ms = st2::message::now_ms() + - u64::try_from((st2::status::STATUS_STALE + Duration::from_secs(60)).as_millis()).unwrap(); + fs::write(&sp, format!("available\nv1 {stale_ms}\n")).unwrap(); assert_eq!(roster(root, "hetz")[0].status, State::Unknown); } + +#[test] +fn roster_uses_version_1_origin_time_for_last_activity() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + write(root, "hetz/idle/agent.kdl", &agent_kdl("idle", "hetz")); + let sp = status_path(&root.join("hetz/idle")); + let heartbeat_ms = st2::message::now_ms() - 1_000; + fs::write(&sp, format!("available\nv1 {heartbeat_ms}\n")).unwrap(); + + let row = &roster(root, "hetz")[0]; + assert_eq!(row.status, State::Available); + assert_eq!(row.last_activity_ms, Some(heartbeat_ms as f64)); +} + +#[test] +fn status_cli_writes_and_reads_the_version_1_record() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + write(root, "h/worker/agent.kdl", &agent_kdl("worker", "h")); + let before = st2::message::now_ms(); + + let set = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["status", "h.worker", "--set", "busy", "--root"]) + .arg(root) + .args(["--host", "h"]) + .output() + .unwrap(); + assert!( + set.status.success(), + "{}", + String::from_utf8_lossy(&set.stderr) + ); + assert_eq!(String::from_utf8(set.stdout).unwrap(), "status: busy\n"); + + let after = st2::message::now_ms(); + let raw = fs::read_to_string(status_path(&root.join("h/worker"))).unwrap(); + let lines: Vec<&str> = raw.lines().collect(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0], "busy"); + let timestamp = lines[1] + .strip_prefix("v1 ") + .unwrap() + .parse::() + .unwrap(); + assert!((before..=after).contains(×tamp)); + assert!(raw.ends_with('\n')); + + let get = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["status", "h.worker", "--root"]) + .arg(root) + .args(["--host", "h"]) + .output() + .unwrap(); + assert!( + get.status.success(), + "{}", + String::from_utf8_lossy(&get.stderr) + ); + assert_eq!(String::from_utf8(get.stdout).unwrap(), "busy\n"); +}