frame_host/

application.rs

//! The embedding seam's orchestration (design §4.1/§4.2): boot the full
//! stack around an application's [`AppSpec`], serve until a stop, tear the
//! whole stack down in order.
//!
//! Boot order: component runtime → component register/start → the
//! application's readiness proof → embedded bus boot → document binding
//! (when configured) → application-event announcer connect → the
//! application's fact announcements (retained by [`crate::truth::AppTruth`]
//! the instant they are accepted, independent of the bus). Serving BINDS the
//! page server FIRST, then starts the announcer pump draining its retained
//! backlog (2026-07-21 boot-visibility fix — see [`crate::truth`] for the
//! defect this order used to have and the snapshot endpoint that closes the
//! remaining race even so); ordered teardown runs announcer intake close,
//! document binding stop, component ordered stop (whose transitions the
//! still-connected announcer publishes), announcer join, app-truth recorder
//! join, and the bus's graceful drain — every failure typed, logged, and
//! never hidden behind an earlier one.
//!
//! frame-host's own binary is the first consumer of this seam
//! ([`crate::app`]); the generated scaffold is the second.

use std::collections::HashSet;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::thread::JoinHandle;

use frame_core::component::ComponentId;
use frame_core::event::LifecycleSubscription;

use crate::announcer::Announcer;
use crate::config::FrameConfig;
use crate::doc_binding::DocBinding;
use crate::embedded::EmbeddedLiminal;
use crate::error::HostError;
use crate::page::PageServer;
use crate::runtime::{EventLoggerHandle, HostRuntime};
use crate::serve::{ServeStop, bind};
use crate::spec::{AppSpec, ComponentInstall, ReadinessProbe};
use crate::truth::AppTruth;

/// Bounded buffer between the registry's lifecycle stream and the
/// announcer. Overflow drops oldest-first and is reported loudly by the
/// announcer pump — the same posture as the console logger's buffer.
const ANNOUNCER_EVENT_BUFFER: usize = 256;

/// A booted application stack: the component runtime with the application's
/// components running, the embedded bus, the optional document binding, and
/// the application-event announcer (absent — loudly — only when the config
/// declares no `[frame].channel`).
pub struct Application {
    runtime: HostRuntime,
    logger: EventLoggerHandle,
    liminal: EmbeddedLiminal,
    doc_binding: Option<DocBinding>,
    announcer: Option<Announcer>,
    bus_endpoint: String,
    components: HashSet<ComponentId>,
    /// Process-lifetime application-truth recorder (2026-07-21
    /// boot-visibility fix): every lifecycle transition and announced fact
    /// this process has produced, served at
    /// `crate::server::APP_STATUS_ROUTE` so a client joining at any moment
    /// observes the full boot history over plain HTTP — no bus race, no
    /// out-of-band subscribe-before-serve shortcut required. See
    /// [`crate::truth`] for the full defect/fix writeup.
    truth: Arc<AppTruth>,
    /// Join handle for the dedicated recorder thread ([`AppTruth::spawn`]).
    truth_recorder: JoinHandle<()>,
}

impl Application {
    /// Boots the full stack around the application's spec, in order. A boot
    /// failure tears down whatever already booted so no component is left
    /// running; the first typed failure is what gets reported.
    ///
    /// # Errors
    ///
    /// Returns the first typed failure from spec validation, runtime
    /// composition, component install, the application's readiness proof,
    /// the embedded bus boot, the document binding, the announcer connect
    /// (a typed boot failure — never a silently event-less page), or the
    /// application's fact announcements.
    pub fn boot(config: &FrameConfig, spec: AppSpec) -> Result<Self, HostError> {
        if spec.components.is_empty() {
            return Err(HostError::ConfigContract {
                detail: "the application spec installs no components: a host with nothing to \
                         host is a composition error, not an empty stack"
                    .to_owned(),
            });
        }
        let components: HashSet<ComponentId> = spec
            .components
            .iter()
            .map(|install| install.meta.id)
            .collect();

        let mut runtime = HostRuntime::new(spec.policy)?;

        // The announcer's lifecycle subscription is taken BEFORE install so
        // the boot transitions are retained for publication once the pump
        // goes live; the console logger subscribes before install for the
        // same reason.
        let announcer_subscription: Option<LifecycleSubscription> = match &config.frame.channel {
            Some(_) => {
                let capacity = NonZeroUsize::new(ANNOUNCER_EVENT_BUFFER)
                    .ok_or(HostError::SynchronizationPoisoned)?;
                Some(runtime.registry().subscribe(capacity)?)
            }
            None => None,
        };
        let logger = runtime.spawn_event_logger(components.clone())?;

        // The application-truth recorder (2026-07-21 boot-visibility fix):
        // a THIRD dedicated lifecycle subscription, taken here for the same
        // reason as the two above — so the retained boot transitions are
        // never missed. Recording is independent of whether
        // `[frame].channel` is declared: this is host-local truth served
        // over plain HTTP (`/frame/app/status.json`), not a bus publish.
        // See `crate::truth` for the full defect/fix writeup.
        let (truth, truth_recorder) = AppTruth::spawn(runtime.registry(), components.clone())?;

        // Installs every declared component, then runs the application's
        // post-start readiness proof (design §4.2), before the bus boots and
        // before anything serves; either failure best-effort abandons
        // whatever installed so far so a failed boot never leaks a process
        // tree.
        install_and_verify_ready(&mut runtime, spec.components, spec.readiness)?;

        // Boot the embedded bus component in-process (its own threads; no
        // child process). A bind/boot failure exits with the failing
        // component named — never a half-up stack. The components already
        // running are torn down first so they are not abandoned.
        let liminal = match EmbeddedLiminal::boot(&config.bus) {
            Ok(liminal) => liminal,
            Err(error) => {
                if let Err(teardown) = runtime.shutdown(logger) {
                    tracing::error!(
                        %teardown,
                        "frame-core teardown after bus boot failure also failed"
                    );
                }
                join_truth_recorder(truth_recorder, "bus boot failure");
                return Err(error);
            }
        };
        let bus_endpoint = liminal.websocket_endpoint();
        tracing::info!(
            endpoint = %bus_endpoint,
            "embedded bus component booted; /frame/config.json advertises its real bound WebSocket address as busEndpoint"
        );

        // The document-service binding when [document] is configured.
        let doc_binding = match DocBinding::boot(&liminal, config) {
            Ok(binding) => binding,
            Err(error) => {
                teardown_after_boot_failure(
                    None,
                    liminal,
                    runtime,
                    logger,
                    truth_recorder,
                    "document-service",
                );
                return Err(error);
            }
        };

        // The application-event announcer (design §4.3): a SEPARATE native
        // participant connection to the embedded server's real bound TCP
        // address. The page server itself carries zero feed bytes (ruling
        // D3); the announcer is a node on the bus, not a proxy in front of
        // it. A failed connect is a typed boot failure.
        let announcer = match connect_announcer(config, &liminal, announcer_subscription) {
            Ok(announcer) => announcer,
            Err(error) => {
                teardown_after_boot_failure(
                    doc_binding,
                    liminal,
                    runtime,
                    logger,
                    truth_recorder,
                    "announcer connect",
                );
                return Err(error);
            }
        };
        // Mirror every fact this announcer accepts into the app-truth
        // snapshot (2026-07-21 boot-visibility fix) — attached BEFORE the
        // boot fact announcement below, so it is retained here too.
        if let Some(announcer) = &announcer {
            announcer.attach_truth(Arc::clone(&truth));
        }

        // The application's own fact announcements, through the connected
        // announcer. Queued facts publish once the pump goes live at serve
        // time, after the retained boot transitions.
        if let Err(error) = announce_boot_facts(spec.announce, &runtime, announcer.as_ref()) {
            teardown_after_boot_failure(
                doc_binding,
                liminal,
                runtime,
                logger,
                truth_recorder,
                "fact announcement",
            );
            return Err(error);
        }

        Ok(Self {
            runtime,
            logger,
            liminal,
            doc_binding,
            announcer,
            bus_endpoint,
            components,
            truth,
            truth_recorder,
        })
    }

    /// Serves the page until SIGTERM, SIGINT, `external_stop` resolving, or
    /// a bus liveness failure — blocking.
    ///
    /// SERVE-ORDER FIX (2026-07-21 boot-visibility fix, [`crate::truth`]):
    /// the HTTP+WS surface binds FIRST; the announcer pump only starts
    /// draining its retained boot backlog (transitions, then queued facts,
    /// then live events) once that surface is reachable. Before this fix the
    /// pump started as this function's first action — before the tokio
    /// runtime even existed — so the entire backlog drained before
    /// `/frame/config.json` (the only doctrine-sanctioned way a browser
    /// discovers the bus address) was fetchable at all; no real browser
    /// could ever win that race. `/frame/app/status.json` (also served the
    /// instant this binds) closes the remaining gap: a client that still
    /// arrives after the live drain observes the full boot history there
    /// instead of racing the bus.
    ///
    /// # Errors
    ///
    /// Returns typed failures from binding the page server, starting the
    /// announcer pump, building the async runtime, registering signal
    /// handlers, or the accept loop. The caller still owns the stack and
    /// must drive [`Self::shutdown`].
    pub fn serve(
        &self,
        config: &FrameConfig,
        page: PageServer,
        external_stop: impl Future<Output = ()> + Send + 'static,
    ) -> Result<ServeStop, HostError> {
        let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .map_err(|source| HostError::AsyncRuntime { source })?;
        // SERVE-ORDER FIX (2026-07-21 boot-visibility fix): the HTTP+WS
        // surface binds FIRST, inside `serve` below; the announcer pump only
        // starts draining its retained boot backlog AFTER that surface is
        // reachable — moved out of the position this function used to start
        // it in (its former first line, before the tokio runtime even
        // existed). A real browser fetches `/frame/config.json` and only
        // then subscribes; starting the pump before the surface bound meant
        // the entire backlog drained before that fetch was even possible.
        // See `crate::truth` for the snapshot half of this fix, which closes
        // the remaining gap even when a client races the pump regardless.
        tokio_runtime.block_on(async {
            let bound = bind(
                config,
                &self.liminal,
                self.bus_endpoint.clone(),
                &self.truth,
                page,
            )
            .await?;
            if let Some(announcer) = &self.announcer {
                announcer.start(self.components.clone())?;
            }
            bound.serve(external_stop).await
        })
    }

    /// Ordered teardown, always running every step and hiding no failure:
    /// announcer intake closes first, the document binding stops (its pump
    /// rides the bus), the components stop and remove in order (their
    /// transitions publish through the still-connected announcer, whose
    /// pump exits on the final Removed), the announcer joins, then the bus
    /// drains gracefully.
    ///
    /// # Errors
    ///
    /// Returns the first typed teardown failure by stage precedence
    /// (document, components, announcer, bus); every later failure is
    /// still executed and logged before the first is returned.
    pub fn shutdown(self) -> Result<(), HostError> {
        // Design §4.1 teardown order, step 1: the announcer stops taking
        // new facts. Its pump keeps publishing until the final Removed.
        if let Some(announcer) = &self.announcer {
            announcer.close_intake();
        }

        let doc_shutdown = match self.doc_binding {
            None => Ok(()),
            Some(binding) => binding.shutdown(),
        };
        if let Err(ref error) = doc_shutdown {
            tracing::error!(%error, "document-service shutdown failed");
        }

        let host_shutdown = self.runtime.shutdown(self.logger);
        if let Err(ref error) = host_shutdown {
            tracing::error!(%error, "frame-core ordered shutdown failed");
        }

        // Join the announcer only after a completed component shutdown has
        // published every Removed transition (the pump's exit condition);
        // after a FAILED component shutdown the pump may never observe them,
        // so the announcer is dropped un-joined — loudly — instead of
        // hanging the teardown on it.
        let announcer_shutdown = match self.announcer {
            None => Ok(()),
            Some(announcer) => {
                if host_shutdown.is_ok() {
                    match announcer.stop() {
                        Ok(None) => Ok(()),
                        Ok(Some(death)) => {
                            tracing::warn!(
                                death,
                                "the application-event announcer had died at runtime; its \
                                 recorded death detail is reported here at teardown"
                            );
                            Ok(())
                        }
                        Err(error) => Err(error),
                    }
                } else {
                    tracing::error!(
                        "component shutdown failed; abandoning the announcer pump un-joined \
                         rather than hanging teardown on transitions that will never arrive"
                    );
                    drop(announcer);
                    Ok(())
                }
            }
        };
        if let Err(ref error) = announcer_shutdown {
            tracing::error!(%error, "announcer shutdown failed");
        }

        // The app-truth recorder observes the SAME expected-component-Removed
        // exit condition as the console logger, published by the ordered
        // component shutdown above — join it only once that has actually
        // happened; a FAILED component shutdown means those Removed
        // transitions may never arrive, so the recorder is abandoned
        // un-joined — loudly — rather than hanging teardown on it (the exact
        // posture the announcer's own join takes just above).
        let truth_shutdown = if host_shutdown.is_ok() {
            self.truth_recorder
                .join()
                .map_err(|_| HostError::TruthRecorderPanicked)
        } else {
            tracing::error!(
                "component shutdown failed; abandoning the app-truth recorder un-joined rather \
                 than hanging teardown on transitions that will never arrive"
            );
            drop(self.truth_recorder);
            Ok(())
        };
        if let Err(ref error) = truth_shutdown {
            tracing::error!(%error, "app-truth recorder shutdown failed");
        }

        let liminal_shutdown = self.liminal.shutdown();
        if let Err(ref error) = liminal_shutdown {
            tracing::error!(%error, "embedded bus graceful shutdown failed");
        }

        doc_shutdown?;
        host_shutdown?;
        announcer_shutdown?;
        truth_shutdown?;
        liminal_shutdown?;
        Ok(())
    }
}

/// Installs every declared component, then runs the application's readiness
/// proof — extracted from [`Application::boot`] purely to keep that
/// function's line count within the workspace's clippy budget; the boot-time
/// cleanup posture is unchanged: either failure best-effort abandons
/// whatever installed so far (`abandon_after_boot_failure`) before the
/// typed error propagates, exactly as the two checks did inline.
///
/// # Errors
///
/// Returns the runtime's typed install failure, or
/// [`HostError::Application`] carrying the readiness proof's typed refusal.
fn install_and_verify_ready(
    runtime: &mut HostRuntime,
    components: Vec<ComponentInstall>,
    readiness: ReadinessProbe,
) -> Result<(), HostError> {
    if let Err(error) = runtime.install(components) {
        runtime.abandon_after_boot_failure();
        return Err(error);
    }
    if let Err(source) = readiness(runtime.registry()) {
        runtime.abandon_after_boot_failure();
        return Err(HostError::Application {
            stage: "readiness",
            source,
        });
    }
    Ok(())
}

/// Connects the application-event announcer over the embedded server's real
/// bound TCP wire address when `[frame].channel` is declared; otherwise the
/// announcer is disabled LOUDLY (documented: the served page then observes
/// no host events on any channel).
///
/// # Errors
///
/// Returns [`HostError::AnnouncerConnect`] — a typed boot failure, never a
/// silently event-less page.
fn connect_announcer(
    config: &FrameConfig,
    liminal: &EmbeddedLiminal,
    subscription: Option<LifecycleSubscription>,
) -> Result<Option<Announcer>, HostError> {
    let (Some(channel), Some(subscription)) = (&config.frame.channel, subscription) else {
        tracing::warn!(
            "no [frame].channel is declared: the application-event announcer is disabled and \
             the served page will observe no host events on any channel; declare \
             [frame].channel to announce lifecycle transitions, capability denials, and \
             application facts"
        );
        return Ok(None);
    };
    let auth_token = config
        .bus
        .auth
        .as_ref()
        .map(|auth| auth.token.clone().into_bytes());
    Announcer::connect(
        &liminal.tcp_addr().to_string(),
        auth_token.as_deref(),
        channel.clone(),
        subscription,
    )
    .map(Some)
}

/// Runs the application's boot-time fact announcements through the
/// connected announcer, or skips them LOUDLY when no announcer exists.
///
/// # Errors
///
/// Returns [`HostError::Application`] carrying the application's own typed
/// refusal.
fn announce_boot_facts(
    announce: crate::spec::FactAnnouncement,
    runtime: &HostRuntime,
    announcer: Option<&Announcer>,
) -> Result<(), HostError> {
    let Some(announcer) = announcer else {
        tracing::warn!("application facts are not announced: no [frame].channel is declared");
        return Ok(());
    };
    announce(runtime.registry(), announcer).map_err(|source| HostError::Application {
        stage: "announce",
        source,
    })
}

/// Best-effort ordered teardown after a boot-stage failure, each failure
/// logged loudly; the original boot failure is what the caller reports.
fn teardown_after_boot_failure(
    doc_binding: Option<DocBinding>,
    liminal: EmbeddedLiminal,
    runtime: HostRuntime,
    logger: EventLoggerHandle,
    truth_recorder: JoinHandle<()>,
    failed_stage: &str,
) {
    if let Some(binding) = doc_binding
        && let Err(teardown) = binding.shutdown()
    {
        tracing::error!(
            %teardown,
            failed_stage,
            "document-service teardown after boot failure also failed"
        );
    }
    if let Err(teardown) = liminal.shutdown() {
        tracing::error!(
            %teardown,
            failed_stage,
            "embedded bus teardown after boot failure also failed"
        );
    }
    if let Err(teardown) = runtime.shutdown(logger) {
        tracing::error!(
            %teardown,
            failed_stage,
            "frame-core teardown after boot failure also failed"
        );
    }
    join_truth_recorder(truth_recorder, failed_stage);
}

/// Best-effort join of the app-truth recorder thread after a boot-stage
/// failure. The recorder's exit condition (every expected component's
/// Removed transition) is the SAME one the console logger relies on,
/// published by the `runtime.shutdown`/`abandon_after_boot_failure` call
/// just above each of this function's call sites — so by the time this
/// runs, the join is expected to complete promptly. A panicked recorder is
/// logged loudly, never hidden behind the original boot failure the caller
/// reports.
fn join_truth_recorder(truth_recorder: JoinHandle<()>, failed_stage: &str) {
    if truth_recorder.join().is_err() {
        tracing::error!(
            failed_stage,
            "app-truth recorder thread panicked during teardown after a boot failure"
        );
    }
}

/// Wait quantum for the dev management adapter's pump. The effective
/// granularity is the substrate's IO quantum (hardcoded 5 s at the
/// pinned liminal — upstream ASK-2); this value carries the intent for
/// the day the quantum becomes settable. Quiet ticks are EMPTY by ruling
/// (brief §Adapter pump sanction).
const DEV_PUMP_WAIT: std::time::Duration = std::time::Duration::from_secs(1);

/// [`run_application`], started in DEV mode (F-7b): identical boot and
/// serve, plus the management door — the conversation is opened on the
/// embedded bus, the handoff line is printed on stdout, and the adapter
/// thread serves stage-and-activate requests behind the
/// [`crate::dev::require_dev_node`] gate. A dev-client disconnect is an
/// owner-defined stop: the serve future resolves and the ordinary ordered
/// teardown runs — never scheduler-shutdown-as-cleanup.
///
/// # Errors
///
/// Returns the first typed failure from boot, the door open, serving, or
/// ordered shutdown.
pub fn run_dev_application(
    mut config: FrameConfig,
    spec: AppSpec,
    page: PageServer,
    dev: crate::dev::DevWiring,
) -> Result<(), HostError> {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    config.finalize_page_origins(page.local_addr());
    let app = Application::boot(&config, spec)?;

    let endpoint = app.liminal.tcp_addr().to_string();
    let attachment = frame_conv::BusAttachment {
        endpoint: endpoint.clone(),
        session_capacity: 1,
        operation_window_millis: 10,
        inbound_buffer: 16,
        answer_window: std::time::Duration::from_secs(30),
    };
    let (mut conversation, conversation_id) = crate::dev::DevManagementConversation::open(
        &attachment,
        crate::dev::DevResumeStore::default(),
    )
    .map_err(|error| HostError::DevManagement {
        detail: error.to_string(),
    })?;
    // The one handoff line `frame dev` parses from the piped stdout.
    println!(
        "{}",
        crate::dev::format_management_line(&endpoint, conversation_id)
    );

    let mut engine = crate::dev::BarrierEngine::new(crate::dev::NodeMode::Dev, dev.boot);
    let support = app
        .runtime
        .first_support_ref()
        .ok_or_else(|| HostError::DevManagement {
            detail: "dev wiring requires the application's FFI support module".to_owned(),
        })?;
    let mut control = crate::dev::DevNodeControl::new(
        app.runtime.registry_handle(),
        app.runtime.support_swapper(),
        dev.meta,
        support,
        dev.mailbox_proof,
        dev.content_proof,
    );

    let close = Arc::new(AtomicBool::new(false));
    let close_for_adapter = Arc::clone(&close);
    let (exit_tx, exit_rx) = tokio::sync::oneshot::channel::<crate::dev::ServeExit>();
    let adapter = std::thread::Builder::new()
        .name("frame-dev-management".to_owned())
        .spawn(move || {
            let exit = crate::dev::serve(
                &mut conversation,
                &mut engine,
                &mut control,
                &close_for_adapter,
                DEV_PUMP_WAIT,
            );
            let _ = exit_tx.send(exit);
        })
        .map_err(|source| HostError::DevManagement {
            detail: format!("adapter thread spawn: {source}"),
        })?;

    // Serving ends on process signals, bus liveness failure, OR the
    // adapter's exit (dev-client disconnect / connection fate) — each
    // path runs the same ordered teardown below.
    let serve_outcome = app.serve(&config, page, async {
        match exit_rx.await {
            Ok(exit) => {
                tracing::info!(?exit, "dev management adapter exited; ordered stop");
            }
            Err(_closed) => {
                tracing::info!("dev management adapter closed without an exit report");
            }
        }
    });
    // Ordered close-wake for the adapter on OUR teardown paths (signal,
    // bus failure): raise the flag, then join — bounded by the pump wait.
    close.store(true, Ordering::Release);
    if adapter.join().is_err() {
        tracing::error!("dev management adapter thread panicked during teardown");
    }
    let shutdown_outcome = app.shutdown();
    let stop = serve_outcome?;
    if let ServeStop::LiminalExited { detail } = stop {
        return Err(HostError::LiminalExited { detail });
    }
    shutdown_outcome?;
    tracing::info!("frame dev server exited cleanly");
    Ok(())
}

/// Loads nothing itself: boots the full stack around the supplied
/// application spec from an already-loaded config and an already-resolved,
/// already-bound page server ([`PageServer`]), serves until a shutdown signal
/// (or a bus liveness failure), then tears the whole stack down in order
/// (design §4.2's `run_application`).
///
/// The caller resolves and prints the page URL (from
/// [`PageServer::local_addr`]) before calling this, so the loud "serving at"
/// line always states the REAL bound address. This finalizes the derived bus
/// origin allow-list from that same address before the bus binds.
///
/// # Errors
///
/// Returns the first typed failure from boot, serving, or ordered shutdown.
/// A serve failure takes precedence in the report, but no teardown failure
/// is ever hidden — each is logged as it happens. An unexpected bus
/// termination at runtime surfaces as an error after a clean teardown.
pub fn run_application(
    mut config: FrameConfig,
    spec: AppSpec,
    page: PageServer,
) -> Result<(), HostError> {
    // The page server is already resolved and BOUND (prefer-and-walk, or
    // stated-and-exact) by the caller, so its REAL address is known before the
    // bus binds: fold it into the derived origin allow-list now, so the bus's
    // WebSocket acceptor allows the served page's own origin (a stated
    // allow-list is LAW and left untouched — see `finalize_page_origins`).
    config.finalize_page_origins(page.local_addr());
    let app = Application::boot(&config, spec)?;
    // Only the process signals (and the bus liveness monitor) govern the
    // standalone server: the embedder stop future never resolves.
    let serve_outcome = app.serve(&config, page, std::future::pending());
    let shutdown_outcome = app.shutdown();
    let stop = serve_outcome?;
    if let ServeStop::LiminalExited { detail } = stop {
        return Err(HostError::LiminalExited { detail });
    }
    shutdown_outcome?;
    tracing::info!("frame server exited cleanly");
    Ok(())
}