frame_host/

truth.rs

//! The application-truth snapshot (2026-07-21 boot-visibility fix).
//!
//! ## The defect this fixes
//!
//! `Application::boot()` queues every boot-time lifecycle transition and
//! fact announcement in-process; `Application::serve()` used to start the
//! announcer pump as its FIRST action — before the HTTP listener bound — so
//! the entire backlog drained onto the bus before `/frame/config.json` (the
//! only doctrine-sanctioned way a browser discovers the bus address, D3) was
//! even reachable. A real browser can never win that race: fetch config,
//! THEN subscribe is the only sequence a real client can perform, and by the
//! time it completes the boot backlog is already gone (liminal 0.3.1 ships
//! no replay reachable by a plain, non-participant subscribe — confirmed at
//! the bytes: `subscribe_response`,
//! `liminal-server-0.3.1/src/server/connection/apply.rs` ~L436, registers a
//! fresh subscription going forward only, `durable` or not).
//!
//! ## The fix
//!
//! frame-host now maintains its own CURRENT TRUTH for this process,
//! independent of the bus: every lifecycle transition and every announced
//! fact, retained here and served as plain JSON at
//! `/frame/app/status.json` (`crate::server::APP_STATUS_ROUTE`) on the SAME
//! HTTP surface as `/frame/config.json`. A client joining at any moment gets
//! everything so far from this endpoint, then follows the live announcer
//! channel for what happens next — the snapshot-then-stream shape
//! `crate::doc_binding`'s resync path already uses for document content,
//! applied here to lifecycle + facts instead. `crate::application` also
//! fixes the other half of the race: the HTTP+WS surface now binds BEFORE
//! the announcer pump starts draining, so the live channel itself no longer
//! loses the boot backlog to a startup race either.
//!
//! ## Recording is independent of `[frame].channel`
//!
//! This recorder runs on its OWN dedicated lifecycle subscription, taken
//! before component install (same convention as the console logger in
//! `crate::runtime` and the announcer in `crate::announcer`), so it never
//! misses a boot transition — regardless of whether `[frame].channel` is
//! even declared. Facts are recorded by `crate::announcer::Announcer` the
//! moment it ACCEPTS one (`Announcer::attach_truth` + the hook inside
//! `announce_fact`), before its pump goes live, so a snapshot taken between
//! boot and serve already carries the boot fact.
//!
//! ## No cap on the retained history
//!
//! The 2026-07-21 ruling that ordered this fix is explicit: "the
//! process-lifetime fact list is what it is" — no invented bound. This
//! module honours that: [`AppTruth`] retains every transition and fact for
//! the life of the process. The ONLY bound anywhere in this module is
//! the private `TRUTH_EVENT_BUFFER` const, which caps the RELAY queue between one publish
//! and this thread's next drain (the same bounded-with-loud-overflow
//! posture the console logger and announcer already use) — never the
//! retained snapshot itself.

use std::collections::{HashMap, HashSet};
use std::num::NonZeroUsize;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use frame_core::component::ComponentId;
use frame_core::event::{
    EventReceiveError, LifecycleEventKind, LifecycleState, LifecycleSubscription,
};
use frame_core::registry::ComponentRegistry;
use serde::Serialize;
use serde_json::Value;

use crate::error::HostError;

/// Bounded relay buffer between the registry's lifecycle stream and this
/// recorder — the SAME bound and overflow posture as the console logger
/// (`crate::runtime::LIFECYCLE_EVENT_BUFFER`) and the announcer
/// (`crate::announcer::ANNOUNCER_EVENT_BUFFER`). This caps only how many
/// events may be in flight between a publish and this thread's next drain;
/// it is NOT a cap on the retained snapshot (see the module docs).
const TRUTH_EVENT_BUFFER: usize = 256;

/// Wait quantum for the recorder's blocking receive — the same posture and
/// value as the console logger and announcer pump: not a polling interval,
/// only a bound on how long one idle block lasts before checking for a
/// closed stream.
const TRUTH_WAIT_QUANTUM: Duration = Duration::from_secs(3600);

/// One recorded lifecycle transition, timestamped at the moment this
/// recorder observed it.
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RecordedTransition {
    /// Registry-wide monotonic sequence number, matching the announcer's own
    /// published `sequence` field.
    pub sequence: u64,
    /// The component this transition belongs to.
    pub component_id: String,
    /// State before this transition; absent for initial registration.
    pub from: Option<LifecycleState>,
    /// State after this transition.
    pub to: LifecycleState,
    /// Host wall-clock time this recorder observed the transition,
    /// milliseconds since the Unix epoch.
    pub at_epoch_ms: u128,
}

/// One recorded application fact, timestamped at the moment the announcer
/// ACCEPTED it (`Announcer::announce_fact`) — before its pump goes live, so
/// a snapshot taken between boot and serve already carries a boot fact.
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RecordedFact {
    /// Host wall-clock time the fact was accepted, milliseconds since the
    /// Unix epoch.
    pub at_epoch_ms: u128,
    /// The fact body, exactly as announced (unwrapped — the `{"kind":"fact",
    /// "body": ...}` wire envelope is an announcer-publish concern, not a
    /// truth-recording one).
    pub body: Value,
}

/// The JSON shape served at [`crate::server::APP_STATUS_ROUTE`]: CURRENT
/// TRUTH at request time.
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AppTruthSnapshot {
    /// Host wall-clock time this snapshot was assembled, milliseconds since
    /// the Unix epoch.
    pub generated_at_epoch_ms: u128,
    /// Every installed component's LATEST observed lifecycle state, keyed by
    /// component id — a convenience fold over `transitions` below.
    pub current: HashMap<String, LifecycleState>,
    /// Every lifecycle transition observed so far, in order.
    pub transitions: Vec<RecordedTransition>,
    /// Every application fact announced so far, in order.
    pub facts: Vec<RecordedFact>,
}

#[derive(Debug, Default)]
struct TruthState {
    transitions: Vec<RecordedTransition>,
    current: HashMap<ComponentId, LifecycleState>,
    facts: Vec<RecordedFact>,
}

/// The process's current-truth recorder: every lifecycle transition and
/// announced fact since boot, retained for the life of the process (no cap
/// — see the module docs for the bound this deliberately is NOT).
#[derive(Debug)]
pub struct AppTruth {
    state: Mutex<TruthState>,
}

impl AppTruth {
    /// An empty truth with no attached recorder thread. Used internally by
    /// [`Self::spawn`] and, publicly, by fixtures/tests that construct a
    /// [`crate::server::ShellConfig`]/[`crate::server::ShellServer`] without
    /// a live component registry to record from (pure HTTP-serving tests).
    #[must_use]
    pub fn empty() -> Arc<Self> {
        Arc::new(Self {
            state: Mutex::new(TruthState::default()),
        })
    }

    /// Spawns the dedicated recorder thread on its OWN lifecycle
    /// subscription, taken here — BEFORE component install, the same
    /// convention the console logger and the announcer both follow — so the
    /// boot transitions (Registered → Starting → Running) are never missed.
    /// Recording runs independent of whether `[frame].channel`/the announcer
    /// exists: this is host-local truth served over plain HTTP, not a bus
    /// publish.
    ///
    /// The recorder thread's exit condition mirrors the console logger's
    /// exactly: it exits once it has observed the Removed transition of
    /// every component in `expected` (published by ordered shutdown), or on
    /// stream closure.
    ///
    /// # Errors
    ///
    /// Returns a typed failure for a poisoned subscribe or a thread spawn
    /// refusal.
    pub fn spawn(
        registry: &ComponentRegistry,
        expected: HashSet<ComponentId>,
    ) -> Result<(Arc<Self>, JoinHandle<()>), HostError> {
        let capacity =
            NonZeroUsize::new(TRUTH_EVENT_BUFFER).ok_or(HostError::SynchronizationPoisoned)?;
        let subscription = registry.subscribe(capacity)?;
        let truth = Self::empty();
        let recorder = Arc::clone(&truth);
        let join = thread::Builder::new()
            .name("frame-host-app-truth".to_owned())
            .spawn(move || record_stream(&recorder, &subscription, &expected))
            .map_err(|source| HostError::TruthRecorderSpawn { source })?;
        Ok((truth, join))
    }

    /// Records one accepted application fact. Called by
    /// [`crate::announcer::Announcer::announce_fact`] the moment a fact is
    /// accepted — before the pump goes live.
    pub fn record_fact(&self, body: Value) {
        let at_epoch_ms = epoch_ms();
        match self.state.lock() {
            Ok(mut state) => state.facts.push(RecordedFact { at_epoch_ms, body }),
            Err(_) => {
                tracing::error!(
                    "app-truth synchronization poisoned while recording an announced fact; the \
                     fact is real (already accepted by the announcer, and still published on the \
                     bus) but will not appear in the /frame/app/status.json snapshot"
                );
            }
        }
    }

    /// Assembles the current snapshot: every transition and fact recorded so
    /// far, in order, plus each component's latest observed state.
    #[must_use]
    pub fn snapshot(&self) -> AppTruthSnapshot {
        let generated_at_epoch_ms = epoch_ms();
        let Ok(state) = self.state.lock() else {
            tracing::error!(
                "app-truth synchronization poisoned while assembling a snapshot; serving an \
                 empty snapshot rather than a fabricated one"
            );
            return AppTruthSnapshot {
                generated_at_epoch_ms,
                current: HashMap::new(),
                transitions: Vec::new(),
                facts: Vec::new(),
            };
        };
        AppTruthSnapshot {
            generated_at_epoch_ms,
            current: state
                .current
                .iter()
                .map(|(id, lifecycle_state)| (id.to_string(), *lifecycle_state))
                .collect(),
            transitions: state.transitions.clone(),
            facts: state.facts.clone(),
        }
    }

    fn record_transition(
        &self,
        sequence: u64,
        component_id: ComponentId,
        from: Option<LifecycleState>,
        to: LifecycleState,
    ) {
        let at_epoch_ms = epoch_ms();
        match self.state.lock() {
            Ok(mut state) => {
                state.transitions.push(RecordedTransition {
                    sequence,
                    component_id: component_id.to_string(),
                    from,
                    to,
                    at_epoch_ms,
                });
                state.current.insert(component_id, to);
            }
            Err(_) => {
                tracing::error!(
                    component = %component_id,
                    ?to,
                    "app-truth synchronization poisoned while recording a lifecycle transition; \
                     it is real (already published on the registry's own stream) but will not \
                     appear in the /frame/app/status.json snapshot"
                );
            }
        }
    }
}

/// Drains the lifecycle stream into `truth` until every expected
/// component's Removed transition (or stream closure) — the exact posture
/// of `crate::runtime::log_lifecycle_stream`, duplicated here rather than
/// shared because this recorder writes into `AppTruth` instead of
/// `tracing`, and because a third subscriber independent of the console
/// logger and the announcer is what proves this fix (a single shared
/// subscription would reintroduce exactly the kind of hidden coupling the
/// underlying defect came from).
fn record_stream(
    truth: &AppTruth,
    subscription: &LifecycleSubscription,
    expected: &HashSet<ComponentId>,
) {
    let mut reported_lag = 0;
    let mut removed: HashSet<ComponentId> = HashSet::new();
    loop {
        let event = match subscription.recv_timeout(TRUTH_WAIT_QUANTUM) {
            Ok(event) => event,
            Err(EventReceiveError::Timeout) => continue,
            Err(EventReceiveError::Closed) => {
                tracing::info!("lifecycle event stream closed; app-truth recorder exiting");
                return;
            }
            Err(EventReceiveError::Poisoned) => {
                tracing::error!(
                    "lifecycle event stream synchronization poisoned; app-truth recorder exiting"
                );
                return;
            }
        };
        let lagged = subscription.lagged_events();
        if lagged > reported_lag {
            tracing::warn!(
                dropped = lagged - reported_lag,
                total_dropped = lagged,
                "app-truth recorder overflowed its bounded relay buffer; oldest events were \
                 dropped (the retained snapshot itself has no cap — only this in-flight relay \
                 does)"
            );
            reported_lag = lagged;
        }
        if let LifecycleEventKind::Transition { from, to } = event.kind {
            truth.record_transition(event.sequence, event.component_id, from, to);
            if to == LifecycleState::Removed {
                removed.insert(event.component_id);
                if removed.is_superset(expected) {
                    tracing::info!(
                        "every application component removed; app-truth recorder exiting"
                    );
                    return;
                }
            }
        }
        // Capability denials are not part of this snapshot's documented
        // scope (lifecycle transitions + announced facts, per the
        // 2026-07-21 ruling); the announcer still publishes them live on
        // the bus exactly as before. Flagged for the coordinator as a
        // deliberate scope choice, not an oversight.
    }
}

/// Host wall-clock time, milliseconds since the Unix epoch. A clock error
/// before the epoch (never expected on a real host) yields `0`, logged
/// loudly rather than fabricated silently — never `.unwrap()`/`.expect()`.
fn epoch_ms() -> u128 {
    let Ok(duration) = SystemTime::now().duration_since(UNIX_EPOCH) else {
        tracing::error!(
            "host wall clock reports a time before the Unix epoch; recording 0 rather than \
             fabricating a timestamp"
        );
        return 0;
    };
    duration.as_millis()
}

#[cfg(test)]
mod tests {
    use super::{AppTruth, epoch_ms};
    use frame_core::component::ComponentId;
    use frame_core::event::LifecycleState;

    fn id() -> ComponentId {
        ComponentId::derive("frame.host", "app-truth-test")
    }

    #[test]
    fn epoch_ms_is_monotonic_enough_to_order_two_calls() {
        let first = epoch_ms();
        std::thread::sleep(std::time::Duration::from_millis(2));
        let second = epoch_ms();
        assert!(
            second >= first,
            "wall-clock reads must not run backwards in a tight loop"
        );
    }

    #[test]
    fn empty_snapshot_has_no_transitions_or_facts() {
        // Constructed directly (not via `spawn`) to unit-test the pure
        // recording/snapshot logic without a live registry.
        let truth = AppTruth::empty();
        let snapshot = truth.snapshot();
        assert!(snapshot.transitions.is_empty());
        assert!(snapshot.facts.is_empty());
        assert!(snapshot.current.is_empty());
    }

    #[test]
    fn recorded_transition_updates_current_and_appends_history() {
        let truth = AppTruth::empty();
        truth.record_transition(0, id(), None, LifecycleState::Registered);
        truth.record_transition(
            1,
            id(),
            Some(LifecycleState::Registered),
            LifecycleState::Starting,
        );
        let snapshot = truth.snapshot();
        assert_eq!(snapshot.transitions.len(), 2);
        assert_eq!(snapshot.transitions[0].sequence, 0);
        assert!(snapshot.transitions[0].from.is_none());
        assert_eq!(snapshot.transitions[1].sequence, 1);
        assert_eq!(
            snapshot.current.get(&id().to_string()),
            Some(&LifecycleState::Starting),
            "current must fold to the LATEST transition, not the first"
        );
    }

    #[test]
    fn recorded_fact_is_retained_verbatim() {
        let truth = AppTruth::empty();
        truth.record_fact(serde_json::json!({ "entity": "e-1" }));
        let snapshot = truth.snapshot();
        assert_eq!(snapshot.facts.len(), 1);
        assert_eq!(snapshot.facts[0].body["entity"], "e-1");
    }
}