frame-host 0.4.0

//! Async serving of the console shell, coordinated with the embedded liminal
//! component's liveness and the process shutdown signals.
//!
//! Binding and serving are split into two phases ([`bind`] then
//! [`BoundServer::serve`]) — mirroring [`ShellServer`]'s own `bind`/`serve`
//! split — which is the SERVE-ORDER FIX (2026-07-21 boot-visibility fix,
//! [`crate::truth`]): [`crate::application::Application::serve`] starts the
//! announcer pump draining its retained boot backlog only AFTER [`bind`]
//! succeeds, so the backlog can never drain before `/frame/config.json` is
//! reachable.
//!
//! Once serving, one shutdown trigger, four sources: SIGTERM, SIGINT, the
//! embedder's own stop future, and the liminal liveness monitor. Whichever
//! fires first drains the console HTTP server; the caller
//! ([`crate::application::run_application`]) then drives liminal's own
//! graceful shutdown.

use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use tokio::net::TcpStream;
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::Notify;

use crate::config::FrameConfig;
use crate::embedded::EmbeddedLiminal;
use crate::error::HostError;
use crate::page::PageServer;
use crate::server::{CONFIG_ROUTE, DocumentAdvert, ShellConfig, ShellServer};
use crate::truth::AppTruth;

/// How often the liveness monitor probes the embedded liminal component.
const MONITOR_INTERVAL: Duration = Duration::from_secs(1);

/// How long a single liveness probe waits for a connection before treating the
/// component as gone.
const PROBE_TIMEOUT: Duration = Duration::from_secs(2);

/// Why serving stopped.
pub enum ServeStop {
    /// A requested stop — a shutdown signal (SIGTERM/SIGINT) or the
    /// embedder's own stop trigger. Clean and expected.
    Signal,
    /// The embedded liminal component stopped answering at runtime. A dead
    /// server behind a healthy-looking host is forbidden: this is a loud,
    /// nonzero-exit stop.
    LiminalExited {
        /// Which liveness probe failed and against which address.
        detail: String,
    },
}

/// A bound-but-not-yet-accepting console shell: the listener is bound and
/// the served config contract already validated, but the userspace accept
/// loop has not started. Splitting bind from serve (mirroring
/// [`ShellServer::bind`]/[`ShellServer::serve`]'s own two-phase shape) is
/// the SERVE-ORDER FIX (2026-07-21 boot-visibility fix, [`crate::truth`]):
/// the caller starts the announcer pump only AFTER [`bind`] returns this,
/// so the pump's retained boot backlog can never drain before
/// `/frame/config.json` is reachable.
///
/// A bound `TcpListener` already queues incoming connections at the kernel
/// level the instant it is bound (POSIX `listen(2)` backlog semantics) —
/// [`Self::serve`] only needs to start polling `accept()` promptly, which it
/// does as its first action.
pub struct BoundServer {
    server: ShellServer,
    liminal_health_addr: SocketAddr,
    liminal_tcp_addr: SocketAddr,
    liminal_websocket_addr: SocketAddr,
}

/// Binds the console shell listener, validating the served config contract
/// first (`/frame/config.json`, and the app-truth snapshot at
/// `crate::server::APP_STATUS_ROUTE`) — but does not yet run the accept
/// loop; call [`BoundServer::serve`] for that.
///
/// The served `/frame/config.json` advertises `busEndpoint` (plus the
/// deprecated duplicate `liminalEndpoint` for one compatibility window) —
/// derived from the embedded component's real bound WebSocket address — so
/// the page connects to the bus directly (this host carries no feed bytes).
///
/// # Errors
///
/// Returns a typed failure from binding the shell server.
pub async fn bind(
    config: &FrameConfig,
    liminal: &EmbeddedLiminal,
    bus_endpoint: String,
    truth: &Arc<AppTruth>,
    page: PageServer,
) -> Result<BoundServer, HostError> {
    let page_addr = page.local_addr();
    let server = ShellServer::adopt(
        page.into_listener(),
        ShellConfig {
            bind: page_addr,
            asset_root: config.frame.assets.clone(),
            bus_endpoint,
            auth_token: config.frame.auth_token.clone(),
            channel: config.frame.channel.clone(),
            // The FULL configured channel list of the embedded bus — the one
            // source of truth the served `channels` roster is derived from.
            channels: config
                .bus
                .channels
                .iter()
                .map(|channel| channel.name.clone())
                .collect(),
            // The same-origin health proxy targets the embedded component's REAL
            // bound health address — one source of truth, no drift.
            bus_health: liminal.health_addr(),
            document: config.document.as_ref().map(|section| DocumentAdvert {
                id: section.id.clone(),
                component_id: section.component_id.clone(),
                feed_channel: section.feed_channel.clone(),
                authoring_channel: section.authoring_channel.clone(),
                blink_interval_ms: section.blink_interval_ms,
            }),
            truth: Arc::clone(truth),
        },
    )?;
    tracing::info!(
        addr = %server.local_addr(),
        assets = %config.frame.assets.display(),
        bus_tcp = %liminal.tcp_addr(),
        bus_ws = %liminal.websocket_addr(),
        bus_health = %liminal.health_addr(),
        // The token itself is a credential and never logged.
        auth = if config.frame.auth_token.is_empty() { "open" } else { "bearer" },
        channel = config.frame.channel.as_deref().unwrap_or("(SDK default)"),
        channels = %config
            .bus
            .channels
            .iter()
            .map(|channel| channel.name.as_str())
            .collect::<Vec<_>>()
            .join(", "),
        config_route = CONFIG_ROUTE,
        "frame server bound: console shell reachable; embedded bus live; the page connects to \
         the bus WebSocket directly (D3). The announcer pump (when [frame].channel is declared) \
         starts draining its retained boot backlog only now that this surface is reachable."
    );
    Ok(BoundServer {
        server,
        liminal_health_addr: liminal.health_addr(),
        liminal_tcp_addr: liminal.tcp_addr(),
        liminal_websocket_addr: liminal.websocket_addr(),
    })
}

impl BoundServer {
    /// Serves the console shell until a shutdown signal or a bus liveness
    /// failure, then returns why it stopped.
    ///
    /// Serving drains on the FIRST of three shutdown sources: SIGTERM,
    /// SIGINT, or `external_stop` resolving (the embedder's own trigger;
    /// [`crate::application::run_application`] passes a never-resolving
    /// future so the process signals alone govern it).
    ///
    /// # Errors
    ///
    /// Returns a typed failure from registering the shutdown signal
    /// handlers, the accept loop, or a poisoned reason slot.
    pub async fn serve(
        self,
        external_stop: impl Future<Output = ()> + Send + 'static,
    ) -> Result<ServeStop, HostError> {
        // Register the signal streams up front so a registration failure is
        // a startup error, never a silent inability to shut down.
        let mut sigterm = signal(SignalKind::terminate())
            .map_err(|source| HostError::ShutdownSignal { source })?;
        let mut sigint = signal(SignalKind::interrupt())
            .map_err(|source| HostError::ShutdownSignal { source })?;

        let shutdown = Arc::new(Notify::new());
        let reason = Arc::new(Mutex::new(ServeStop::Signal));

        let signal_shutdown = Arc::clone(&shutdown);
        let signals = tokio::spawn(async move {
            tokio::select! {
                _ = sigterm.recv() => tracing::info!("SIGTERM received; draining frame server"),
                _ = sigint.recv() => tracing::info!("SIGINT received; draining frame server"),
                () = external_stop => tracing::info!("embedder stop requested; draining frame server"),
            }
            // `notify_one` STORES a permit when no waiter is registered yet;
            // `notify_waiters` would drop a stop that fires before the accept
            // loop first polls its shutdown future, leaving a server that can
            // never drain. There is exactly one consumer ([`wait_for`]).
            signal_shutdown.notify_one();
        });

        let monitor = tokio::spawn(liveness_monitor(
            self.liminal_health_addr,
            self.liminal_tcp_addr,
            self.liminal_websocket_addr,
            Arc::clone(&shutdown),
            Arc::clone(&reason),
        ));

        let serve_result = self.server.serve(wait_for(Arc::clone(&shutdown))).await;

        // Serving has ended; stop the background tasks before reading the reason.
        signals.abort();
        monitor.abort();

        serve_result?;
        let stop = std::mem::replace(
            &mut *reason
                .lock()
                .map_err(|_| HostError::SynchronizationPoisoned)?,
            ServeStop::Signal,
        );
        Ok(stop)
    }
}

/// Resolves when shutdown is requested. A `'static` future that owns the
/// notifier, as [`ShellServer::serve`] requires.
async fn wait_for(shutdown: Arc<Notify>) {
    shutdown.notified().await;
}

/// The result of one liveness probe against one liminal listener address.
enum Liveness {
    /// The listener accepted a connection — its accept thread is alive.
    Alive,
    /// The listener REFUSED the connection: a port liminal bound and never
    /// releases while alive now has no listener, so that accept thread is gone.
    /// This is a definitive death signal.
    Dead(String),
    /// The probe failed for a reason that does NOT mean liminal died — a
    /// connect timeout under host load, or local fd/port exhaustion. Retried on
    /// the next tick rather than treated as a death (avoids tearing down a
    /// healthy stack over a transient host condition).
    Transient(String),
}

/// Probes the embedded liminal component on an interval; on the first
/// DEFINITIVE death of any of its listeners (while the host is still meant to be
/// serving) it records the reason and requests shutdown.
///
/// It probes all three ports liminal bound — the health endpoint, the TCP wire
/// listener (native clients, e.g. the demo publisher), and the browser-facing
/// WebSocket listener — because each has its own independently-failable accept
/// thread; a wire-only death would otherwise leave the native data path dead
/// behind a healthy-looking host, exactly what R4 forbids.
///
/// A refused connection is definitive: the kernel completes a handshake from the
/// listen backlog whether or not `accept()` is mid-call, so a live listener
/// never refuses and `ConnectionRefused` means the socket is closed (its thread
/// gone). Non-refusal errors (timeout, fd exhaustion) are NOT liminal death and
/// are retried.
///
/// IDLE COST (flagged for idle-cost sign-off): each tick opens and immediately
/// closes one throwaway TCP connection to each live liminal port, which liminal
/// accepts and tears down (a WS-port probe spawns a short-lived handshake worker;
/// a wire-port probe spawns a beamr connection process that is reaped). Liminal
/// exposes NO push-based liveness/termination signal reachable from an embedding
/// host, so this bounded interval probe is the only honest substitute; a
/// liminal-side embedder liveness signal would let it be dropped entirely.
async fn liveness_monitor(
    health_addr: SocketAddr,
    tcp_addr: SocketAddr,
    websocket_addr: SocketAddr,
    shutdown: Arc<Notify>,
    reason: Arc<Mutex<ServeStop>>,
) {
    let mut ticker = tokio::time::interval(MONITOR_INTERVAL);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // Consume the immediate first tick: probe only after the first interval.
    ticker.tick().await;
    loop {
        ticker.tick().await;
        if let Some(detail) = probe_failure(health_addr, tcp_addr, websocket_addr).await {
            tracing::error!(
                detail,
                "embedded liminal liveness probe found a dead listener; shutting down the frame server loudly"
            );
            match reason.lock() {
                Ok(mut slot) => *slot = ServeStop::LiminalExited { detail },
                Err(_) => {
                    tracing::error!("serve-reason slot poisoned while recording liminal exit");
                }
            }
            // Permit-storing wake (see the signal task): a death detected
            // before the accept loop registers its waiter must still drain it.
            shutdown.notify_one();
            return;
        }
    }
}

/// Returns a description of the first DEFINITIVELY-dead listener, or `None` if
/// every listener answered (a transient probe failure counts as answered — it
/// is logged and retried, never treated as death).
async fn probe_failure(
    health_addr: SocketAddr,
    tcp_addr: SocketAddr,
    websocket_addr: SocketAddr,
) -> Option<String> {
    for (label, addr) in [
        ("health endpoint", health_addr),
        ("TCP wire listener", tcp_addr),
        ("WebSocket listener", websocket_addr),
    ] {
        match probe_connect(addr).await {
            Liveness::Alive => {}
            Liveness::Dead(detail) => {
                return Some(format!("liminal {label} at {addr}: {detail}"));
            }
            Liveness::Transient(detail) => {
                tracing::debug!(
                    label,
                    detail,
                    "liminal liveness probe transient failure; not treated as death, retrying next tick"
                );
            }
        }
    }
    None
}

/// Connects to `addr` and immediately closes, classifying the outcome.
async fn probe_connect(addr: SocketAddr) -> Liveness {
    match tokio::time::timeout(PROBE_TIMEOUT, TcpStream::connect(addr)).await {
        Ok(Ok(stream)) => {
            drop(stream);
            Liveness::Alive
        }
        Ok(Err(error)) if error.kind() == std::io::ErrorKind::ConnectionRefused => {
            Liveness::Dead(format!("refused the connection ({error})"))
        }
        Ok(Err(error)) => Liveness::Transient(format!("connect errored ({error})")),
        Err(_) => Liveness::Transient(format!("connect timed out after {PROBE_TIMEOUT:?}")),
    }
}

#[cfg(test)]
mod tests {
    use super::{Liveness, ServeStop, liveness_monitor, probe_connect, probe_failure};
    use std::net::TcpListener;
    use std::sync::{Arc, Mutex};
    use std::time::Duration;
    use tokio::sync::Notify;

    /// A live listener answers `Alive`; a closed port answers `Dead` (refused).
    #[tokio::test]
    async fn probe_connect_classifies_alive_and_refused() -> Result<(), Box<dyn std::error::Error>>
    {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let alive_addr = listener.local_addr()?;
        assert!(matches!(probe_connect(alive_addr).await, Liveness::Alive));

        // Bind then release to obtain an address nothing listens on.
        let released = TcpListener::bind("127.0.0.1:0")?;
        let dead_addr = released.local_addr()?;
        drop(released);
        assert!(matches!(probe_connect(dead_addr).await, Liveness::Dead(_)));

        // `probe_failure` returns None while all three answer, Some once one dies.
        assert!(
            probe_failure(alive_addr, alive_addr, alive_addr)
                .await
                .is_none()
        );
        assert!(
            probe_failure(alive_addr, dead_addr, alive_addr)
                .await
                .is_some()
        );
        Ok(())
    }

    /// The monitor stays quiet while listeners live, then records `LiminalExited`
    /// and requests shutdown once one dies — the R4 runtime-death path end to end.
    #[tokio::test]
    async fn monitor_fires_liminal_exited_when_a_listener_dies()
    -> Result<(), Box<dyn std::error::Error>> {
        let health = TcpListener::bind("127.0.0.1:0")?;
        let tcp = TcpListener::bind("127.0.0.1:0")?;
        let websocket = TcpListener::bind("127.0.0.1:0")?;
        let (health_addr, tcp_addr, websocket_addr) = (
            health.local_addr()?,
            tcp.local_addr()?,
            websocket.local_addr()?,
        );

        let shutdown = Arc::new(Notify::new());
        let reason = Arc::new(Mutex::new(ServeStop::Signal));
        let monitor = tokio::spawn(liveness_monitor(
            health_addr,
            tcp_addr,
            websocket_addr,
            Arc::clone(&shutdown),
            Arc::clone(&reason),
        ));

        // While all three live, the monitor must not fire.
        tokio::time::sleep(Duration::from_millis(2500)).await;
        assert!(matches!(
            *reason.lock().map_err(|_| "poisoned")?,
            ServeStop::Signal
        ));

        // Kill the wire listener: the monitor must notice and request shutdown.
        drop(tcp);
        let notified = tokio::time::timeout(Duration::from_secs(6), shutdown.notified());
        assert!(
            notified.await.is_ok(),
            "monitor must request shutdown on a dead listener"
        );
        assert!(matches!(
            *reason.lock().map_err(|_| "poisoned")?,
            ServeStop::LiminalExited { .. }
        ));
        monitor.await?;
        Ok(())
    }
}