frame-host 0.4.0

//! The application-event announcer (design §4.3): an in-process NATIVE bus
//! participant that publishes the host authority's real events onto
//! `[frame].channel` as JSON — component lifecycle transitions and capability
//! denials (from the same registry lifecycle subscription that feeds the
//! console logger) plus application-announced facts.
//!
//! ## Doctrine check (browser-participant ruling, frame D3)
//!
//! The page server carries ZERO feed bytes and never proxies the browser's
//! connection. The announcer does not touch that path: it is a SEPARATE
//! first-class participant with its own TCP connection to the embedded
//! server's real bound wire address — the host is a node on the bus, not a
//! proxy in front of it. [`crate::server::ShellServer`] still serves only
//! assets + config.
//!
//! ## What this is not
//!
//! It is not the BEAM component speaking bus protocol. The component still
//! exchanges only mailbox integers with the host; the component-to-
//! conversation surface remains `frame-conv`'s future. The announcer is
//! host-authority telemetry, real end to end.
//!
//! ## Failure posture
//!
//! - A failed connect at boot is a TYPED boot failure
//!   ([`HostError::AnnouncerConnect`]) — never a silently event-less page.
//! - A death at runtime (the bus connection failing under a live announcer)
//!   is LOUD in the host log, recorded with its detail, and surfaced in the
//!   served page's connection state by absence of events — documented, not
//!   masked. The pump exits on the first publish failure; a direct fact
//!   announcement returns [`HostError::AnnouncerPublish`].
//! - The buffer between the registry's lifecycle stream and the pump is
//!   bounded and every discarded event is reported (the exact posture of the
//!   console lifecycle logger in [`crate::runtime`]).

use std::collections::{HashSet, VecDeque};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::Duration;

use frame_core::component::ComponentId;
use frame_core::event::{
    EventReceiveError, EventTryReceiveError, LifecycleEvent, LifecycleEventKind, LifecycleState,
    LifecycleSubscription,
};
use liminal_sdk::remote::PushClient;

use crate::error::HostError;
use crate::truth::AppTruth;

/// Bound on facts queued before the pump goes live. Overflow drops
/// oldest-first and is reported loudly — never silent (the lifecycle-logger
/// posture).
const FACT_QUEUE_BOUND: usize = 256;

/// Wait quantum for the pump's blocking receive. Delivery is event-driven
/// (condvar wakeups); this bound only caps how long one idle block lasts and
/// is not a polling interval — the pump exits when it has observed the
/// Removed transition of every expected component (published by ordered
/// shutdown before the join), or on a publish failure, or on stream closure.
const PUMP_WAIT_QUANTUM: Duration = Duration::from_secs(3600);

/// Facts waiting for the pump to go live, bounded and overflow-counted.
struct PendingFacts {
    queue: VecDeque<serde_json::Value>,
    dropped: usize,
    /// Once true, fact announcements publish inline instead of queueing.
    live: bool,
}

/// The announcer's recorded runtime death: set once, loudly, with detail.
struct DeathSlot {
    dead: AtomicBool,
    detail: Mutex<Option<String>>,
}

/// State shared between the announcer handle and its pump thread.
struct Shared {
    client: Mutex<PushClient>,
    channel: String,
    pending: Mutex<PendingFacts>,
    death: DeathSlot,
}

/// A connected application-event announcer: one native participant
/// connection publishing host-authority events on the application channel.
pub struct Announcer {
    shared: Arc<Shared>,
    /// The registry lifecycle subscription, taken by [`Self::start`]. It is
    /// subscribed BEFORE component install so the boot transitions
    /// (Registered → Starting → Running) are retained in its bounded queue
    /// and published once the pump goes live.
    subscription: Mutex<Option<LifecycleSubscription>>,
    pump: Mutex<Option<JoinHandle<()>>>,
    /// Set by [`Self::close_intake`] at teardown start: subsequent fact
    /// announcements are refused with a typed error.
    intake_closed: AtomicBool,
    /// The process's application-truth recorder (2026-07-21
    /// boot-visibility fix), attached by [`Self::attach_truth`]. Optional:
    /// an announcer with no truth attached still announces facts on the bus
    /// exactly as before, it just is not mirrored into a
    /// `/frame/app/status.json` snapshot (the posture every pre-fix
    /// `Announcer::connect` call site — including this module's own
    /// tests — keeps by construction).
    truth: Mutex<Option<Arc<AppTruth>>>,
}

impl Announcer {
    /// Connects the announcer's own participant connection to the embedded
    /// server's real bound TCP wire address, carrying the bus auth token
    /// when the bus is gated.
    ///
    /// # Errors
    ///
    /// Returns [`HostError::AnnouncerConnect`] with the exact SDK failure —
    /// at boot this is a typed boot failure, never a silently event-less
    /// page.
    pub fn connect(
        address: &str,
        auth_token: Option<&[u8]>,
        channel: String,
        subscription: LifecycleSubscription,
    ) -> Result<Self, HostError> {
        let connected = match auth_token {
            Some(token) => PushClient::connect_with_auth(address, token),
            None => PushClient::connect(address),
        };
        let client = connected.map_err(|error| HostError::AnnouncerConnect {
            address: address.to_owned(),
            detail: error.to_string(),
        })?;
        Ok(Self {
            shared: Arc::new(Shared {
                client: Mutex::new(client),
                channel,
                pending: Mutex::new(PendingFacts {
                    queue: VecDeque::new(),
                    dropped: 0,
                    live: false,
                }),
                death: DeathSlot {
                    dead: AtomicBool::new(false),
                    detail: Mutex::new(None),
                },
            }),
            subscription: Mutex::new(Some(subscription)),
            pump: Mutex::new(None),
            intake_closed: AtomicBool::new(false),
            truth: Mutex::new(None),
        })
    }

    /// Attaches the process's application-truth recorder so every fact this
    /// announcer accepts is ALSO retained for the `/frame/app/status.json`
    /// snapshot, immediately on acceptance — before the pump goes live (see
    /// [`crate::truth`] for why that ordering matters).
    pub fn attach_truth(&self, truth: Arc<AppTruth>) {
        match self.truth.lock() {
            Ok(mut slot) => *slot = Some(truth),
            Err(_) => {
                tracing::error!(
                    "announcer synchronization poisoned while attaching the app-truth recorder; \
                     facts will still publish on the bus but will not appear in the \
                     /frame/app/status.json snapshot"
                );
            }
        }
    }

    /// Starts the pump: publishes the retained boot transitions in order,
    /// then the facts queued before start, then live events as they arrive.
    /// The pump exits after observing the Removed transition of every
    /// component in `expected` (published by ordered shutdown), on stream
    /// closure, or loudly on the first publish failure.
    ///
    /// # Errors
    ///
    /// Returns a typed failure when started twice, when `expected` is empty
    /// (a pump with no exit condition is a composition error), when the
    /// thread cannot spawn, or when the subscription slot is poisoned.
    pub fn start(&self, expected: HashSet<ComponentId>) -> Result<(), HostError> {
        if expected.is_empty() {
            return Err(HostError::ConfigContract {
                detail: "the application-event announcer needs at least one expected component; \
                         an empty set would leave its pump with no ordered exit"
                    .to_owned(),
            });
        }
        let subscription = self
            .subscription
            .lock()
            .map_err(|_| HostError::SynchronizationPoisoned)?
            .take()
            .ok_or(HostError::AnnouncerAlreadyStarted)?;
        let shared = Arc::clone(&self.shared);
        let join = std::thread::Builder::new()
            .name("frame-host-announcer".to_owned())
            .spawn(move || pump(&shared, &subscription, &expected))
            .map_err(|source| HostError::AnnouncerSpawn { source })?;
        *self
            .pump
            .lock()
            .map_err(|_| HostError::SynchronizationPoisoned)? = Some(join);
        Ok(())
    }

    /// Announces one application fact as published JSON
    /// (`{"kind":"fact","body":…}`). Before the pump goes live the fact is
    /// queued (bounded, overflow dropped oldest-first and reported); after,
    /// it publishes inline on the announcer's own connection.
    ///
    /// # Errors
    ///
    /// Returns [`HostError::AnnouncerIntakeClosed`] once teardown has begun
    /// and [`HostError::AnnouncerPublish`] when the live publish fails (the
    /// death is also recorded loudly).
    pub fn announce_fact(&self, fact: serde_json::Value) -> Result<(), HostError> {
        if self.intake_closed.load(Ordering::Acquire) {
            return Err(HostError::AnnouncerIntakeClosed);
        }
        // Mirror into the app-truth snapshot the instant the fact is
        // accepted — before the live/queued decision below, so a snapshot
        // taken between boot and serve already carries a boot fact even
        // though the pump has not gone live yet.
        match self.truth.lock() {
            Ok(slot) => {
                if let Some(truth) = slot.as_ref() {
                    truth.record_fact(fact.clone());
                }
            }
            Err(_) => {
                tracing::error!(
                    "announcer synchronization poisoned while checking for an attached app-truth \
                     recorder; the fact is still announced on the bus"
                );
            }
        }
        let live = {
            let mut pending = self
                .shared
                .pending
                .lock()
                .map_err(|_| HostError::SynchronizationPoisoned)?;
            if !pending.live {
                if pending.queue.len() == FACT_QUEUE_BOUND {
                    let _discarded = pending.queue.pop_front();
                    pending.dropped += 1;
                    tracing::warn!(
                        total_dropped = pending.dropped,
                        "announcer fact queue overflowed its bound before the pump went live; \
                         the oldest fact was dropped"
                    );
                }
                pending.queue.push_back(fact);
                return Ok(());
            }
            pending.live
        };
        debug_assert!(live);
        publish_json(&self.shared, &fact_json(&fact))
    }

    /// Closes the fact intake at teardown start (design §4.1: the announcer
    /// stops accepting new facts first; its pump keeps publishing the
    /// ordered-stop transitions until the final Removed).
    pub fn close_intake(&self) {
        self.intake_closed.store(true, Ordering::Release);
    }

    /// Joins the pump and reports the recorded runtime death, if any. Call
    /// only after ordered component shutdown has published every expected
    /// Removed transition (or after a recorded death) — those are the pump's
    /// exit conditions.
    ///
    /// # Errors
    ///
    /// Returns [`HostError::AnnouncerPanicked`] for a panicked pump and a
    /// typed poison failure for a poisoned slot.
    pub fn stop(self) -> Result<Option<String>, HostError> {
        let join = self
            .pump
            .lock()
            .map_err(|_| HostError::SynchronizationPoisoned)?
            .take();
        if let Some(join) = join {
            join.join().map_err(|_| HostError::AnnouncerPanicked)?;
        }
        let detail = self
            .shared
            .death
            .detail
            .lock()
            .map_err(|_| HostError::SynchronizationPoisoned)?
            .clone();
        Ok(detail)
    }
}

/// Publishes one JSON value on the announcer's connection; on failure the
/// death is recorded loudly and a typed error returned.
fn publish_json(shared: &Shared, value: &serde_json::Value) -> Result<(), HostError> {
    let payload = value.to_string().into_bytes();
    let outcome = {
        let client = shared
            .client
            .lock()
            .map_err(|_| HostError::SynchronizationPoisoned)?;
        client.publish(&shared.channel, payload)
    };
    outcome.map_err(|error| {
        let detail = error.to_string();
        record_death(shared, &detail);
        HostError::AnnouncerPublish {
            channel: shared.channel.clone(),
            detail,
        }
    })
}

/// Records the announcer's runtime death once, loudly. The served page shows
/// this state by absence of events — documented, never masked.
fn record_death(shared: &Shared, detail: &str) {
    if !shared.death.dead.swap(true, Ordering::AcqRel) {
        tracing::error!(
            channel = %shared.channel,
            detail,
            "application-event announcer DIED at runtime: its bus connection failed; the served \
             page will observe the death as an absence of events"
        );
        match shared.death.detail.lock() {
            Ok(mut slot) => *slot = Some(detail.to_owned()),
            Err(_) => {
                tracing::error!("announcer death-detail slot poisoned while recording the death");
            }
        }
    }
}

/// The announcer pump: retained boot transitions first, then queued facts,
/// then live events until every expected component is Removed.
fn pump(shared: &Shared, subscription: &LifecycleSubscription, expected: &HashSet<ComponentId>) {
    let mut removed: HashSet<ComponentId> = HashSet::new();
    let mut reported_lag = 0;

    // 1. The retained backlog: the boot transitions the subscription captured
    //    before the bus existed.
    loop {
        match subscription.try_recv() {
            Ok(event) => {
                if !announce_event(shared, subscription, &mut reported_lag, &event) {
                    return;
                }
                track_removed(&event, &mut removed);
            }
            Err(EventTryReceiveError::Empty) => break,
            Err(EventTryReceiveError::Closed) => {
                tracing::info!("lifecycle event stream closed; announcer pump exiting");
                return;
            }
            Err(EventTryReceiveError::Poisoned) => {
                record_death(shared, "lifecycle event stream synchronization poisoned");
                return;
            }
        }
    }

    // 2. Facts queued before the pump went live, in announcement order.
    let Ok(mut pending) = shared.pending.lock() else {
        record_death(shared, "announcer fact queue synchronization poisoned");
        return;
    };
    pending.live = true;
    if pending.dropped > 0 {
        tracing::warn!(
            dropped = pending.dropped,
            "announcer fact queue dropped facts before the pump went live"
        );
    }
    let queued = std::mem::take(&mut pending.queue);
    drop(pending);
    for fact in queued {
        if publish_json(shared, &fact_json(&fact)).is_err() {
            // publish_json already recorded the death loudly.
            return;
        }
    }
    if removed.is_superset(expected) {
        tracing::info!("every expected component is removed; announcer pump exiting");
        return;
    }

    // 3. Live events until every expected component is Removed.
    loop {
        let event = match subscription.recv_timeout(PUMP_WAIT_QUANTUM) {
            Ok(event) => event,
            Err(EventReceiveError::Timeout) => continue,
            Err(EventReceiveError::Closed) => {
                tracing::info!("lifecycle event stream closed; announcer pump exiting");
                return;
            }
            Err(EventReceiveError::Poisoned) => {
                record_death(shared, "lifecycle event stream synchronization poisoned");
                return;
            }
        };
        if !announce_event(shared, subscription, &mut reported_lag, &event) {
            return;
        }
        track_removed(&event, &mut removed);
        if removed.is_superset(expected) {
            tracing::info!("every expected component is removed; announcer pump exiting");
            return;
        }
    }
}

/// Publishes one lifecycle event (reporting any subscription overflow
/// first); returns `false` when the pump must exit on a recorded death.
fn announce_event(
    shared: &Shared,
    subscription: &LifecycleSubscription,
    reported_lag: &mut usize,
    event: &LifecycleEvent,
) -> bool {
    let lagged = subscription.lagged_events();
    if lagged > *reported_lag {
        tracing::warn!(
            dropped = lagged - *reported_lag,
            total_dropped = lagged,
            "announcer overflowed its bounded lifecycle buffer; oldest events were dropped"
        );
        *reported_lag = lagged;
    }
    publish_json(shared, &event_json(event)).is_ok()
}

/// Notes a Removed transition toward the pump's ordered exit.
fn track_removed(event: &LifecycleEvent, removed: &mut HashSet<ComponentId>) {
    if let LifecycleEventKind::Transition {
        to: LifecycleState::Removed,
        ..
    } = event.kind
    {
        removed.insert(event.component_id);
    }
}

/// The published JSON for one lifecycle stream item.
fn event_json(event: &LifecycleEvent) -> serde_json::Value {
    match &event.kind {
        LifecycleEventKind::Transition { from, to } => serde_json::json!({
            "kind": "lifecycle",
            "sequence": event.sequence,
            "componentId": event.component_id.to_string(),
            "from": from,
            "to": to,
        }),
        LifecycleEventKind::CapabilityDenied(denial) => serde_json::json!({
            "kind": "capability-denied",
            "sequence": event.sequence,
            "componentId": event.component_id.to_string(),
            "denial": denial,
        }),
        LifecycleEventKind::FragmentContentUpdated { key } => serde_json::json!({
            "kind": "fragment-content-updated",
            "sequence": event.sequence,
            "componentId": event.component_id.to_string(),
            "fragmentId": key.fragment_id.as_str(),
        }),
    }
}

/// The published JSON envelope for one application-announced fact.
fn fact_json(body: &serde_json::Value) -> serde_json::Value {
    serde_json::json!({ "kind": "fact", "body": body })
}

#[cfg(test)]
mod tests {
    use super::{event_json, fact_json};
    use frame_core::capability::{CapabilityDenied, CapabilityKind, CapabilityScope};
    use frame_core::component::ComponentId;
    use frame_core::event::{LifecycleEvent, LifecycleEventKind, LifecycleState};

    fn id() -> ComponentId {
        ComponentId::derive("frame.host", "announcer-wire-proof")
    }

    #[test]
    fn lifecycle_transition_encodes_states_and_component() {
        let value = event_json(&LifecycleEvent {
            sequence: 7,
            component_id: id(),
            kind: LifecycleEventKind::Transition {
                from: Some(LifecycleState::Registered),
                to: LifecycleState::Starting,
            },
        });
        assert_eq!(value["kind"], "lifecycle");
        assert_eq!(value["sequence"], 7);
        assert_eq!(value["componentId"], id().to_string());
        assert_eq!(value["from"], "Registered");
        assert_eq!(value["to"], "Starting");
    }

    #[test]
    fn initial_registration_encodes_null_from() {
        let value = event_json(&LifecycleEvent {
            sequence: 0,
            component_id: id(),
            kind: LifecycleEventKind::Transition {
                from: None,
                to: LifecycleState::Registered,
            },
        });
        assert!(value["from"].is_null());
        assert_eq!(value["to"], "Registered");
    }

    #[test]
    fn capability_denial_encodes_the_full_denial() {
        let value = event_json(&LifecycleEvent {
            sequence: 3,
            component_id: id(),
            kind: LifecycleEventKind::CapabilityDenied(CapabilityDenied {
                component_id: id(),
                kind: CapabilityKind::Network,
                scope: CapabilityScope::Host("bus.example".to_owned()),
                declared: false,
            }),
        });
        assert_eq!(value["kind"], "capability-denied");
        assert_eq!(value["sequence"], 3);
        assert_eq!(value["componentId"], id().to_string());
        assert_eq!(value["denial"]["declared"], false);
    }

    #[test]
    fn facts_are_wrapped_verbatim() {
        let value = fact_json(&serde_json::json!({ "entity": "e-1" }));
        assert_eq!(value["kind"], "fact");
        assert_eq!(value["body"]["entity"], "e-1");
    }
}