forge_server/

boot.rs

//! The boot the server runs: the config dir, the workspace, the subsystem
//! wave, the socket, and the boot spawn.
//!
//! Moved here from the view's `main.rs` so the core comes up under one
//! function for a view and for no view alike, and so no view's removal can
//! take the boot with it.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use forge_workspace::Workspace;

/// Resolve the Claude config directory at the boot boundary: honour
/// `$CLAUDE_CONFIG_DIR` (ignoring empty values), else fall back to
/// `$HOME/.claude`. After this point, the resolved path is threaded as a
/// typed `PathBuf` (via `Workspace::new`, which in turn binds each
/// `Agent::spawn(config_dir)` to its account's path).
///
/// Per hard rule #14 - no cwd-derived fallbacks. When `$CLAUDE_CONFIG_DIR`
/// is unset/empty AND `dirs::home_dir()` returns None, refuse to launch
/// rather than substituting `./.claude`.
pub fn resolve_config_dir() -> anyhow::Result<PathBuf> {
    if let Ok(value) = std::env::var("CLAUDE_CONFIG_DIR") {
        let trimmed = value.trim_end_matches('/');
        if !trimmed.is_empty() {
            return Ok(PathBuf::from(trimmed));
        }
    }
    dirs::home_dir().map(|h| h.join(".claude")).ok_or_else(|| {
        anyhow::anyhow!("$CLAUDE_CONFIG_DIR unset and could not resolve home directory")
    })
}

/// One panic, as the hook records it.
struct PanicRecord {
    thread: String,
    payload: String,
    location: String,
    backtrace: Option<String>,
}

/// Install the process-wide panic hook, once.
///
/// Every panic is one ERROR line in the tracing log, and the previous hook
/// still runs so stderr keeps its behaviour. Installed after the tracing
/// subscriber and before anything is spawned, so a panic in the boot wave is
/// recorded rather than lost.
pub fn install_panic_hook() {
    static INSTALLED: std::sync::Once = std::sync::Once::new();
    INSTALLED.call_once(|| {
        let previous = std::panic::take_hook();
        std::panic::set_hook(Box::new(move |info| {
            let thread = std::thread::current();
            let record =
                panic_record(thread.name().unwrap_or("unnamed"), info.payload(), info.location());
            tracing::error!(
                target: crate::logging::targets::APP_LIFECYCLE,
                event_name = "panic",
                thread = %record.thread,
                payload = %record.payload,
                location = %record.location,
                backtrace = %record.backtrace.as_deref().unwrap_or(""),
                "a thread panicked",
            );
            previous(info);
        }));
    });
}

/// The record one panic writes: the thread, the payload as a string, the
/// file and line, and the backtrace when `RUST_BACKTRACE` is set.
fn panic_record(
    thread: &str,
    payload: &(dyn std::any::Any + Send),
    location: Option<&std::panic::Location<'_>>,
) -> PanicRecord {
    let payload = payload
        .downcast_ref::<&str>()
        .map(|text| (*text).to_owned())
        .or_else(|| payload.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "a panic with no message".to_owned());
    let location = location
        .map_or_else(String::new, |location| format!("{}:{}", location.file(), location.line()));
    let backtrace = std::env::var_os("RUST_BACKTRACE")
        .filter(|value| value.to_str() != Some("0"))
        .map(|_| std::backtrace::Backtrace::force_capture().to_string());
    PanicRecord { thread: thread.to_owned(), payload, location, backtrace }
}

/// Install the process-level rustls crypto provider, before anything can
/// build a TLS client.
///
/// The graph carries both `aws-lc-rs` and `ring`, so rustls will not choose
/// and the first handshake panics. `install_default` returns `Err` when a
/// provider is already in place, which is the idempotent success case: a
/// second boot entry or an embedding caller may run it again.
fn install_crypto_provider() {
    let already_installed =
        rustls::crypto::aws_lc_rs::default_provider().install_default().is_err();
    tracing::debug!(
        target: crate::logging::targets::APP_LIFECYCLE,
        event_name = "rustls_crypto_provider",
        already_installed,
        "the process-level rustls crypto provider is in place",
    );
}

/// Build the workspace and start the whole subsystem wave, in the order the
/// boot has always started it.
///
/// `project` is the CLI's positional, validated here because a mistyped
/// name has to fail the boot the same way for every caller; what each
/// caller then DOES with it differs (the headless server starts that
/// project's sessions).
///
/// Not `async`: every step is a spawn, a store write or the skill install, and the await on
/// anything that actually does I/O belongs to `serve_socket`. Nothing here
/// needs a `LocalSet` either - every task in this wave is a plain
/// `tokio::spawn`, verified by the headless boot test exercising it
/// without one.
pub fn start_core(config_dir: &Path, project: Option<&str>) -> anyhow::Result<Arc<Workspace>> {
    // **The provider first, the hook second.** The graph carries both rustls
    // providers, so the first TLS handshake panics unless one is installed;
    // the hook is what names that panic if it ever happens again.
    install_crypto_provider();
    install_panic_hook();
    let workspace = match Workspace::new(config_dir.to_path_buf()) {
        Ok(w) => Arc::new(w),
        Err(err) => return Err(anyhow::anyhow!("forge: {err}")),
    };
    // The settings document carries the build that made it; this is the one
    // place the server's own version reaches the workspace. The short stamp
    // is the one the home's header draws as `server`.
    workspace.set_settings_version(crate::FORGE_VERSION_SHORT);

    // Validate the CLI's positional `<PROJECT>` arg, if any, before any
    // view or spawn wave. This turns `forge xyz` (where `xyz` isn't in
    // forge.toml) into a clean stderr error + non-zero exit instead of a
    // boot that fails deeper with a worse message.
    if let Some(name) = project
        && let Err(err) = workspace.validate_project_name(name)
    {
        return Err(anyhow::anyhow!("forge: {err}"));
    }

    install_shipped_skills(config_dir);
    start_subsystems(&workspace);
    #[cfg(target_os = "macos")]
    {
        let fleet = install_browser(&workspace)?;
        start_config_watch(&workspace, move |workspace| {
            refresh_browser(workspace, &fleet);
        });
    }
    #[cfg(not(target_os = "macos"))]
    {
        install_browser(&workspace);
        start_config_watch(&workspace, |_workspace| {});
    }

    Ok(workspace)
}

/// Install the skills forge ships into the config dir.
///
/// **forge.toml is not written by this path; these skill directories are.**
/// Each skill is forge's own text, compared byte for byte, so an upgrade
/// refreshes what differs and a boot with identical copies touches nothing.
/// One skill's failure is its own line: the rest still install.
fn install_shipped_skills(config_dir: &Path) {
    for skill in forge_workspace::shipped_skills() {
        let dir = forge_workspace::skill_dir(config_dir, skill.name);
        match forge_workspace::install_skill(config_dir, skill) {
            Ok(outcome) => {
                let Some((action, message)) = install_line(outcome) else { continue };
                tracing::info!(
                    target: "forge_workspace::skills",
                    event_name = skill_installed_event(skill.name),
                    skill = skill.name,
                    path = %dir.display(),
                    action,
                    "{message}",
                );
            }
            Err(error) => {
                tracing::warn!(
                    target: "forge_workspace::skills",
                    event_name = skill_install_failed_event(skill.name),
                    skill = skill.name,
                    path = %dir.display(),
                    consequence = skill_failure_consequence(skill.name),
                    %error,
                    "the skill could not be installed",
                );
            }
        }
    }
}

/// The line an install gets, or `None` when it changed nothing.
///
/// **A sweep is not a write**, so its line says what actually happened
/// rather than claiming the skill was written.
fn install_line(outcome: forge_workspace::SkillInstall) -> Option<(&'static str, &'static str)> {
    match outcome {
        forge_workspace::SkillInstall::Created => {
            Some(("created", "the skill was written at boot"))
        }
        forge_workspace::SkillInstall::Replaced => {
            Some(("replaced", "the skill was written at boot"))
        }
        forge_workspace::SkillInstall::Swept => {
            // **Not "were removed":** a removal can fail and its WARN sits
            // beside this line, so the INFO says what was owed, not what
            // happened.
            Some(("swept", "the skill's leftover files were due for removal at boot"))
        }
        forge_workspace::SkillInstall::Unchanged => None,
    }
}

/// What a failed install leaves behind, per skill.
///
/// **Handoff degrades; the wizard can be left mixed.** A handoff turn
/// carries its three steps in the judgment prompt as a fallback, while the
/// wizard's two files can be split across a failed rename, so the line says
/// which of the two happened rather than one sentence standing for both.
fn skill_failure_consequence(name: &str) -> &'static str {
    match name {
        "handoff" => "a handoff turn falls back to the prompt's own steps",
        _ => "the skill may be missing or partly updated until a later boot writes it",
    }
}

/// The log event names stay per-skill: handoff's pair is the one existing
/// traces and greps know, and a new skill gets its own rather than being
/// folded into handoff's line.
fn skill_installed_event(name: &str) -> &'static str {
    match name {
        "handoff" => "handoff_skill_installed",
        "wizard" => "wizard_skill_installed",
        _ => "skill_installed",
    }
}

fn skill_install_failed_event(name: &str) -> &'static str {
    match name {
        "handoff" => "handoff_skill_install_failed",
        "wizard" => "wizard_skill_install_failed",
        _ => "skill_install_failed",
    }
}

/// Watch `forge.toml` for the process's life and apply each valid edit.
///
/// The watch is a single file (`forge.toml`), never the config directory, so
/// nothing else in that tree can trigger a reload. Each event is a whole-file
/// read: a rejected file leaves the previous config live and logs one WARN,
/// and an applied one logs the tables that moved and the keys that only a
/// restart can take. `apply` runs for the subsystems the boot holds.
fn start_config_watch<F>(workspace: &Arc<Workspace>, apply: F)
where
    F: Fn(&Workspace) + Send + Sync + 'static,
{
    let path = workspace.config_path();
    let weak = Arc::downgrade(workspace);
    let apply = Arc::new(apply);
    // **The first watch is installed before the member starts.** An edit in
    // the window between this call and the driver's first poll is not
    // missed, and a failed install is one WARN rather than a restart
    // ladder.
    let (changed, edits) = tokio::sync::mpsc::unbounded_channel();
    let first = match forge_workspace::env::config_watch::watch(path.clone(), changed) {
        Ok(watch) => Some((watch, edits)),
        Err(error) => {
            tracing::warn!(
                target: crate::logging::targets::APP_LIFECYCLE,
                event_name = "config_watch_failed",
                path = %path.display(),
                error = %error,
                "forge.toml is not being watched; an edit needs a restart this run",
            );
            return;
        }
    };
    let first = Arc::new(std::sync::Mutex::new(first));
    workspace.supervisor().spawn("config_watch", forge_supervise::Policy::backoff(), move || {
        let path = path.clone();
        let weak = weak.clone();
        let apply = Arc::clone(&apply);
        let first = Arc::clone(&first);
        async move {
            let first_watch =
                first.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take();
            let (watch, mut edits) = if let Some(pair) = first_watch {
                pair
            } else {
                // A restart after the first install needs a fresh watch and
                // channel.
                let (changed, edits) = tokio::sync::mpsc::unbounded_channel();
                match forge_workspace::env::config_watch::watch(path.clone(), changed) {
                    Ok(watch) => (watch, edits),
                    Err(error) => {
                        tracing::warn!(
                            target: crate::logging::targets::APP_LIFECYCLE,
                            event_name = "config_watch_failed",
                            path = %path.display(),
                            error = %error,
                            "forge.toml is not being watched; an edit needs a restart this run",
                        );
                        return;
                    }
                }
            };
            // The handle lives in this task, so the watch lives as long
            // as the process (or, in a test, the runtime) without a
            // handle outside.
            let _watch = watch;
            loop {
                let _ = edits.recv().await;
                let Some(workspace) = weak.upgrade() else { break };
                match workspace.reload_config() {
                    Ok(reload) => {
                        // A save that changed nothing is not an apply.
                        if reload.changed.is_empty() && reload.restart_only.is_empty() {
                            continue;
                        }
                        tracing::info!(
                            target: crate::logging::targets::APP_LIFECYCLE,
                            event_name = "config_reloaded",
                            changed = ?reload.changed,
                            restart_only = ?reload.restart_only,
                            "forge.toml applied to the running forge",
                        );
                        apply(&workspace);
                    }
                    Err(error) => {
                        tracing::warn!(
                            target: crate::logging::targets::APP_LIFECYCLE,
                            event_name = "config_reload_rejected",
                            path = %workspace.config_path().display(),
                            error = %error,
                            "forge.toml was not applied; the last good config stays live",
                        );
                    }
                }
            }
        }
    });
}

/// Push the reloaded `[browser]` knobs and profile names into the fleet.
#[cfg(target_os = "macos")]
fn refresh_browser(workspace: &Workspace, fleet: &Arc<forge_browser::service::BrowserService>) {
    let browser = workspace.browser_settings();
    fleet.configure_encoding(browser.fps, browser.quality);
    fleet.configure_idle(browser.idle_minutes);
    configure_fleet_profiles(workspace, fleet);
}

/// Install the server's browser fleet: constructed here, one host and one
/// driver per profile spawned on first use.
///
/// **A missing app-support directory fails the boot**, the same resolution
/// the durable store needs; the vendored driver stack is not required at
/// boot, so a server without it boots and its browser calls name what is
/// missing instead of the whole server refusing to start.
/// **The join between the config and the fleet's resolver.** The workspace
/// resolves `forge.toml`'s org and project names; the fleet resolves a seat
/// against those maps. Both ends have tests, and this is the copy between
/// them, so it is one named call rather than two lines inside a boot.
#[cfg(target_os = "macos")]
fn configure_fleet_profiles(workspace: &Workspace, fleet: &forge_browser::service::BrowserService) {
    let (orgs, projects) = workspace.browser_profile_config();
    fleet.configure_profiles(orgs, projects);
}

#[cfg(target_os = "macos")]
fn install_browser(
    workspace: &Arc<Workspace>,
) -> anyhow::Result<Arc<forge_browser::service::BrowserService>> {
    let fleet = forge_browser::service::BrowserService::installed()
        .map_err(|why| anyhow::anyhow!("forge: {why}"))?;
    // **The encoding and the idle clock are the `[browser]` keys**: a
    // profile nobody has touched closes after the bound.
    let browser = workspace.browser_settings();
    fleet.configure_encoding(browser.fps, browser.quality);
    fleet.configure_idle(browser.idle_minutes);
    // **The profile names are config's, not a client's**: every seat resolves
    // its profile from `forge.toml` here, at boot, rather than naming one.
    configure_fleet_profiles(workspace, &fleet);
    fleet.set_supervisor(Arc::clone(workspace.supervisor()));
    fleet.start_idle();
    fleet.start_pairing();
    // **A host an unclean exit left behind is reaped before any profile is
    // used.** It still holds the profile's singleton lock, and the first
    // call would wedge on it; the reaper checks the process is really this
    // bundle's host, and that its server is gone, before it signals anything.
    //
    // **The prune runs after the reap.** The reaper identifies an orphan by
    // its executable's path, and a prune that unlinked a live host's binary
    // first would take that identity away (`proc_pidpath` returns nothing for
    // an unlinked binary). A peer's live hosts are spared by the reaper's
    // parent check; the prune re-asks whether the machine is shared before
    // every removal, because a peer can take its lock after the boot's first
    // look. Off the boot thread: each prune is a CEF tree.
    let reaper = Arc::clone(&fleet);
    let prune_workspace = Arc::downgrade(workspace);
    let reap = async move { reaper.reap_stale().await };
    tokio::spawn(reap_then_prune(reap, move || {
        forge_browser::host_bundle::prune_other_versions(&move || {
            prune_workspace.upgrade().is_some_and(|workspace| workspace.sole_instance())
        });
    }));
    workspace.install_browser_host(Arc::new(HostedBrowser(Arc::clone(&fleet))));
    Ok(fleet)
}

/// Await the stale-host reap, then run the prune on a blocking thread.
///
/// **The order is the contract.** The reaper identifies an orphan by its
/// executable's path, and `proc_pidpath` answers nothing for an unlinked
/// binary, so a prune that removed a host's bundle first would take the
/// identity away. Off the async workers because each prune is a CEF tree.
#[cfg(target_os = "macos")]
async fn reap_then_prune<F>(reap: impl std::future::Future<Output = ()>, prune: F)
where
    F: FnOnce() + Send + 'static,
{
    reap.await;
    let _ = tokio::task::spawn_blocking(prune).await;
}

/// A platform with no CEF host: the browser tools answer the no-host refusal
/// rather than a server that pretends to have a fleet.
#[cfg(not(target_os = "macos"))]
fn install_browser(_workspace: &Arc<Workspace>) {}

/// The fleet behind the workspace's port.
#[cfg(target_os = "macos")]
struct HostedBrowser(Arc<forge_browser::service::BrowserService>);

#[cfg(target_os = "macos")]
#[async_trait::async_trait]
impl forge_workspace::browser::BrowserHost for HostedBrowser {
    async fn call(
        &self,
        seat: &forge_primitives::SessionSlot,
        tool: &str,
        args: serde_json::Value,
    ) -> Result<Vec<forge_primitives::browser::BrowserPart>, String> {
        self.0.call(seat, tool, args).await
    }

    async fn attach(
        &self,
        viewer: u64,
        seat: &forge_primitives::SessionSlot,
        tab: Option<u16>,
    ) -> Result<forge_workspace::browser::BrowserAttachment, String> {
        let attachment = self.0.attach(viewer, seat, tab).await?;
        Ok(forge_workspace::browser::BrowserAttachment {
            key: attachment.key,
            id: attachment.id,
            tab: attachment.tab,
            frames: attachment.frames,
            view: attachment.view,
            tabs: attachment.tabs,
            nav: attachment.nav,
            failures: attachment.failures,
            files: attachment.files,
            downloads: attachment.downloads,
            audio: attachment.audio,
        })
    }

    async fn attach_page(
        &self,
        viewer: u64,
        seat: &forge_primitives::SessionSlot,
        target: &str,
    ) -> Result<forge_workspace::browser::BrowserAttachment, String> {
        let attachment = self.0.attach_page(viewer, seat, target).await?;
        Ok(forge_workspace::browser::BrowserAttachment {
            key: attachment.key,
            id: attachment.id,
            tab: attachment.tab,
            frames: attachment.frames,
            view: attachment.view,
            tabs: attachment.tabs,
            nav: attachment.nav,
            failures: attachment.failures,
            files: attachment.files,
            downloads: attachment.downloads,
            audio: attachment.audio,
        })
    }

    async fn answer_files(
        &self,
        profile: &str,
        tab: u16,
        id: u64,
        files: Vec<forge_primitives::browser::BrowserUpload>,
    ) -> Result<(), String> {
        self.0.answer_files(profile, tab, id, files).await
    }

    async fn detach(&self, viewer: u64) {
        self.0.detach(viewer).await;
    }

    async fn input(
        &self,
        profile: &str,
        tab: u16,
        viewer: u64,
        seq: u64,
        event: &forge_primitives::browser::BrowserInputEvent,
    ) -> Result<(), String> {
        self.0.input(profile, tab, viewer, seq, event).await
    }

    async fn take_control(&self, profile: &str, tab: u16, viewer: u64) -> Result<(), String> {
        self.0.take_control(profile, tab, viewer).await
    }

    async fn release_control(&self, profile: &str, tab: u16, viewer: u64) {
        self.0.release_control(profile, tab, viewer).await;
    }

    async fn resize(
        &self,
        profile: &str,
        tab: u16,
        viewer: u64,
        width: u32,
        height: u32,
        scale: f64,
    ) {
        self.0.resize(profile, tab, viewer, width, height, scale).await;
    }

    async fn nav(
        &self,
        profile: &str,
        tab: u16,
        viewer: u64,
        back: bool,
    ) -> Result<forge_primitives::browser::BrowserNavState, String> {
        self.0.nav(profile, tab, viewer, back).await
    }

    async fn nav_state(
        &self,
        profile: &str,
        tab: u16,
    ) -> Option<forge_primitives::browser::BrowserNavState> {
        self.0.nav_state(profile, tab).await
    }

    async fn tabs(
        &self,
        seat: &forge_primitives::SessionSlot,
    ) -> Result<forge_primitives::browser::BrowserTabList, String> {
        self.0.tabs(seat).await
    }

    async fn tab_stream(
        &self,
        seat: &forge_primitives::SessionSlot,
    ) -> Result<tokio::sync::watch::Receiver<forge_primitives::browser::BrowserTabList>, String>
    {
        self.0.tab_stream(seat).await
    }

    async fn sweep_sessions(&self, seats: &[forge_primitives::SessionSlot], generation: u64) {
        self.0.sweep_sessions(seats, generation).await;
    }

    async fn replace_seat(&self, seat: &forge_primitives::SessionSlot) {
        self.0.replace_seat(seat).await;
    }

    fn session_ended(&self, seat: &forge_primitives::SessionSlot) {
        self.0.session_ended(seat);
    }

    async fn hand_off_started(
        &self,
        seat: &forge_primitives::SessionSlot,
        target: &str,
    ) -> Result<u16, String> {
        self.0.hand_off_started(seat, target).await
    }

    async fn hand_off_finished(&self, seat: &forge_primitives::SessionSlot, tab: u16) {
        self.0.hand_off_finished(seat, tab).await;
    }

    async fn warm(&self, seat: &forge_primitives::SessionSlot) {
        self.0.warm(seat).await;
    }

    async fn stop(&self) -> Vec<String> {
        self.0.stop_all().await
    }
}

/// The subsystem wave the boot starts, in the order it has always started it.
///
/// A function of its own so a test can run the same list the binary runs: a
/// `start_*` call the wave drops is invisible to any test that starts its
/// subsystem directly.
fn start_subsystems(workspace: &Arc<Workspace>) {
    // Kick off the per-account boot-time loading state machine (one task
    // per `[[accounts]]` entry) in the background. Each task drives its
    // account from `Loading` through to a terminal `Ready` or `Bailed`.
    // The redb store has already seeded the in-memory map at
    // `Workspace::new`.
    workspace.start_account_loading_tasks();

    // The 60 s usage poller, after the loading tasks: they probe each account
    // once at boot and the poller skips its own immediate-fire tick, so this
    // is what refreshes the account snapshots from then on. Without it they
    // hold their boot values for the process's life.
    workspace.start_usage_poller();

    // Rotation cooldowns change what a project row's `would_bind` answers,
    // on a write and again on the clock that thaws them; this is the one
    // watcher that turns both into a roster announcement.
    workspace.start_rotation_announcer();

    // Bind the gateway's inference listener and open the boot gate for
    // it. Every spawned session's base URL names this port: a failed bind
    // holds the gate shut AND spawn attempts are refused at the spawn
    // entries until forge restarts.
    workspace.start_gateway_listener();

    // Load the model catalogue's cache and refresh a stale feed in the
    // background. No-op unless `[dictate] enabled` is set, and never on
    // the boot path: the fetch is what the models page's freshness line
    // reports on.
    workspace.start_dictate_catalogue();

    // Fetch, verify and load the dictation models, on the same screen the
    // accounts appear on. No-op unless `[dictate] enabled` is set: a 3 GB
    // download is opt-in.
    workspace.start_dictate_preflight();

    // Load the speaking engine beside it, so the first `say` pays no model
    // load. No-op unless `[tts] enabled` is set.
    workspace.start_voice_prewarm();

    // Spawn the worker-kick drainer task (#259). Routes every
    // `maybe_kick_worker_on_connected` enqueue through a single
    // `KICK_DISPATCH_INTERVAL`-spaced dispatcher so multi-worker boots
    // don't hit Anthropic's per-IP burst limit. Idempotent - the start
    // helper no-ops a second call.
    workspace.start_kick_dispatcher();

    // Boot catch-up: one tick fires each overdue cron once, advancing
    // past the missed slots (a cron missed while forge was down fires
    // now, not once per skipped tick). Runs after Workspace::new so
    // spawn-to-fire can spawn an asleep project's session.
    workspace.fire_due_crons(std::time::SystemTime::now());

    // Start the durable-cron scheduler: wakes every ~60s and fires
    // crons as they come due, advancing/removing each.
    workspace.start_cron_scheduler();

    // Start the chase sweep: every ~30s the board's misses are derived
    // and the crossings delivered - the pass that makes the task list
    // drive instead of report.
    workspace.start_chase_sweep();

    // Start the idle sweep: every ~60s a lead with nothing in flight is
    // handed the board sweep, once per two hours, with a six-hour
    // fallback for a lead that never goes idle. Delivery never wakes a
    // sleeping session.
    workspace.start_idle_sweep();

    // Start the Gotify subsystem when configured with at least one
    // durable subscription loaded at boot; no-op otherwise.
    workspace.start_gotify_subsystem();

    // Prove each configured `[[slack]]` token with `auth.test`. A
    // failure is logged and that workspace stays dormant; it never
    // stops the boot. No-op with no `[[slack]]` entry.
    workspace.start_slack_verification();

    // Start one Slack pump per workspace with a durable subscription
    // loaded at boot; no-op otherwise. The subscribe tool starts one
    // for a workspace that gains its first.
    workspace.start_slack_subsystem();

    // Drop review state left behind by branches deleted since the
    // last run. Worker teardown only catches a branch already gone at
    // that moment, and a branch usually outlives its worker.
    workspace.start_review_branch_sweep();
}

/// What [`serve_socket`] did.
///
/// Its own type rather than an `Option` so the caller decides what an
/// unbound socket means to it: a headless server with no view serves
/// nothing and treats the failure as fatal.
pub enum SocketServing {
    /// `[server] enabled = false`: nothing serves, by configuration.
    Disabled,
    /// The socket is listening; the serve task ends when the process does.
    Bound,
    /// The bind failed, its log line already written.
    BindFailed,
}

/// Serve the socket on the address `[server]` names, registering the serve
/// task with the workspace's supervisor and returning whether one is
/// serving.
///
/// A bind that fails never fails the boot here: this is not on the boot
/// path, and the failure is logged plus reported as
/// [`SocketServing::BindFailed`] so the caller's boot cannot die of it.
pub async fn serve_socket(workspace: &Arc<Workspace>) -> anyhow::Result<SocketServing> {
    let server = workspace.server_config();
    if !server.enabled {
        tracing::debug!(
            target: crate::logging::targets::APP_LIFECYCLE,
            event_name = "server_socket_disabled",
            "[server] enabled = false; nothing is serving this run",
        );
        return Ok(SocketServing::Disabled);
    }
    let addr = std::net::SocketAddr::new(server.bind, server.port);
    // Built from the workspace so every piece the socket needs is derived
    // where the workspace is in hand - the browser fleet above all, which
    // must be the one the sessions' own tools drive.
    let state = Arc::new(crate::transport::TransportState::for_workspace(workspace));
    match tokio::net::TcpListener::bind(addr).await {
        Ok(listener) => {
            let bound = listener.local_addr().unwrap_or(addr);
            tracing::info!(
                target: crate::logging::targets::APP_LIFECYCLE,
                event_name = "server_socket_listening",
                %bound,
                "the server's socket is listening",
            );
            // **Taken before the serve task is spawned**, so the fold is
            // attached before this returns and the boot wave can spawn a
            // session whose first `Connected` finds a copy to land on.
            let updates = state.surface.subscribe_mirror();
            // The listener and the fold's receiver are not `Clone`, and a
            // `Never` member is started once, so the factory takes them from
            // their slots.
            let listener = Arc::new(std::sync::Mutex::new(Some(listener)));
            let updates = Arc::new(std::sync::Mutex::new(Some(updates)));
            workspace.supervisor().spawn(
                "socket_serve",
                forge_supervise::Policy::Never,
                move || {
                    let listener = Arc::clone(&listener);
                    let updates = Arc::clone(&updates);
                    let state = Arc::clone(&state);
                    async move {
                        let Some(listener) = listener
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .take()
                        else {
                            return;
                        };
                        let Some(updates) = updates
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .take()
                        else {
                            return;
                        };
                        if let Err(error) = crate::transport::serve(state, listener, updates).await
                        {
                            tracing::error!(
                                target: crate::logging::targets::APP_LIFECYCLE,
                                event_name = "server_socket_failed",
                                error = %error,
                                "the socket stopped serving",
                            );
                        }
                    }
                },
            );
            Ok(SocketServing::Bound)
        }
        Err(error) => {
            tracing::error!(
                target: crate::logging::targets::APP_LIFECYCLE,
                event_name = "server_socket_bind_failed",
                error = %error,
                "the socket is not serving",
            );
            Ok(SocketServing::BindFailed)
        }
    }
}

/// How long the boot wave holds for the accounts and the listener before it
/// fires anyway.
///
/// A view can hold for as long as the reader is on its preflight screen - a
/// signal a headless process does not have. A listener that never binds - a
/// port someone else holds is the case - would otherwise make
/// `all_accounts_loaded` false forever and turn the wave into a spawn that
/// never fires, silently.
pub const BOOT_WAVE_WAIT: std::time::Duration = std::time::Duration::from_secs(60);

/// Hold until the accounts and the listener are ready, or `timeout` passes.
///
/// `true` when they settled, `false` at expiry; the caller decides what a
/// hold that ran out means (the wave fires anyway and warns). The bound is
/// the whole point - see [`start_boot_wave`] - so it is a parameter, which
/// is what lets a test pin it at a millisecond instead of a minute.
async fn wait_for_accounts(workspace: &Arc<Workspace>, timeout: std::time::Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;
    while !workspace.all_accounts_loaded() && tokio::time::Instant::now() < deadline {
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
    }
    workspace.all_accounts_loaded()
}

/// The boot spawn, for a process with no view: hold until the accounts
/// settle, then dispatch the wave a view's connect path dispatches -
/// the named project's lead, or every project whose lead row was left
/// open, in forge.toml order. A seat whose last turn was cut off carries
/// the restart note; a parked one is resumed with nothing.
///
/// Settling is terminal rather than `Ready` - a bailed account releases
/// the spawn, because the walk skips it and the pollers keep re-probing
/// it, so holding would wait on nothing.
///
/// **The hold is bounded** (`BOOT_WAVE_WAIT`). On expiry the wave goes
/// out regardless and a WARN names what was not ready: a refused spawn
/// carries its own reason (the spawn entries attach it), which beats a wave
/// that never fires and says nothing.
///
/// Spawned rather than awaited: the boot must not block on accounts, and
/// nothing downstream reads the result.
pub fn start_boot_wave(workspace: Arc<Workspace>, project: Option<String>, force_new: bool) {
    tokio::spawn(async move {
        if !wait_for_accounts(&workspace, BOOT_WAVE_WAIT).await {
            tracing::warn!(
                target: crate::logging::targets::APP_LIFECYCLE,
                event_name = "boot_wave_wait_expired",
                gateway_ready = workspace.gateway_ready(),
                gateway_bind_error = workspace.gateway_bind_error().as_deref().unwrap_or(""),
                "the boot spawn is going out with the accounts or the gateway listener not ready \
                 (a bound listener here means the account map is what held it); a refused spawn \
                 carries its own reason",
            );
        }
        // **The derived open set is read before any spawn runs**, so the
        // `idle` write each spawn lands cannot hide the state it replaced.
        // `--new` makes these leads start fresh instead of resuming (and
        // cascades to their workers); a spawn that does resume takes the
        // restart note as its first turn only when the row said a turn was
        // in flight.
        let targets = workspace.boot_wave_targets(project.as_deref());
        for target in targets {
            let launch_settings = forge_workspace::SessionLaunchSettings {
                force_new,
                resume_kick: target.resume_kick.clone(),
                ..Default::default()
            };
            let project_name = target.project_name.clone();
            let cmd = forge_workspace::Command::SpawnProject {
                project_name: project_name.clone(),
                launch_settings,
            };
            if let Err(err) = workspace.dispatch(cmd) {
                tracing::error!(
                    target: crate::logging::targets::APP_LIFECYCLE,
                    event_name = "boot_wave_dispatch_failed",
                    error = %err,
                    project = %project_name,
                    "boot wave dispatch failed",
                );
            }
        }
    });
}

/// The stop's exit: a teardown that did not finish every session leaves a
/// process tree behind, and a supervisor must be able to tell that from a
/// clean stop.
///
/// # Errors
///
/// A non-empty `unready` is an error naming the sessions that outlasted
/// the wait.
pub fn stop_result(unready: &[forge_primitives::SessionSlot]) -> anyhow::Result<()> {
    if unready.is_empty() {
        return Ok(());
    }
    let named: Vec<String> = unready.iter().map(forge_primitives::SessionSlot::display).collect();
    let (subject, tail) = if unready.len() == 1 {
        ("1 session".to_owned(), "its process tree may outlive")
    } else {
        (format!("{} sessions", unready.len()), "their process trees may outlive")
    };
    anyhow::bail!(
        "forge: {subject} did not finish teardown before the wait ran out ({}); {tail} this exit",
        named.join(", "),
    );
}

#[cfg(test)]
mod tests {
    use super::{
        BOOT_WAVE_WAIT, install_crypto_provider, install_shipped_skills, panic_record,
        start_boot_wave, start_config_watch, start_subsystems, stop_result, wait_for_accounts,
    };
    #[cfg(target_os = "macos")]
    use super::{configure_fleet_profiles, reap_then_prune};
    use std::sync::Arc;
    use std::time::Duration;

    use forge_workspace::Workspace;

    /// **The panic record names the thread and the payload.** The hook
    /// itself is process-global, so the formatter is what a test can drive;
    /// a record that dropped either half would leave a panic unattributable.
    #[test]
    fn the_panic_record_names_the_thread_and_the_payload() {
        let record = panic_record("forge-worker", &"the pump died", None);
        assert_eq!(record.thread, "forge-worker", "the thread is named");
        assert_eq!(record.payload, "the pump died", "the payload is the panic's own words");
    }

    /// The install is idempotent: the second call finds a provider in
    /// place and returns cleanly rather than failing, and the process
    /// carries a default afterwards. Rustls refuses to pick one itself
    /// while both `aws-lc-rs` and `ring` are in the graph, which is the
    /// panic this exists to prevent.
    #[test]
    fn installing_the_rustls_provider_twice_is_clean() {
        install_crypto_provider();

        install_crypto_provider();

        assert!(
            rustls::crypto::CryptoProvider::get_default().is_some(),
            "the boot's install leaves a process-level rustls provider in place",
        );
        // The call a TLS client makes: without a default it panics, which
        // is the Gotify stream task's first handshake.
        let _ = rustls::ClientConfig::builder();
    }

    /// The boot entry installs the provider before it builds anything: a
    /// TLS client can be built from anywhere downstream, and this graph
    /// carries two provider features. An empty config dir refuses at
    /// `Workspace::new`, so the install is the only boot step that runs
    /// and the test never reaches the machine's single-instance lock.
    #[test]
    fn the_boot_entry_installs_the_provider_before_it_builds_anything() {
        let dir = tempfile::tempdir().expect("tempdir");

        let events = crate::test_support::logged(|| {
            let booted = super::start_core(dir.path(), None);
            assert!(booted.is_err(), "an empty config dir does not build a workspace");
        });

        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::DEBUG
                    && fields.contains("event_name=\"rustls_crypto_provider\"")
            }),
            "the boot entry logs the provider install before it builds anything: {events:?}",
        );
    }

    /// **Each skill's install line keeps its own name.** Handoff's pair is
    /// what existing traces and greps know, so it must not move; the wizard's
    /// is its own rather than folded into handoff's line, and a boot with
    /// nothing to write says nothing.
    #[test]
    fn the_boot_names_each_skills_install_event() {
        let dir = tempfile::tempdir().expect("tempdir");
        let events = crate::test_support::logged(|| install_shipped_skills(dir.path()));

        for (skill, event) in
            [("handoff", "handoff_skill_installed"), ("wizard", "wizard_skill_installed")]
        {
            assert!(
                events.iter().any(|(level, fields)| {
                    *level == tracing::Level::INFO
                        && fields.contains(&format!("event_name=\"{event}\""))
                        && fields.contains(&format!("skill=\"{skill}\""))
                        && fields.contains("action=\"created\"")
                }),
                "{skill}'s install line is named {event}: {events:?}",
            );
        }
        assert_eq!(super::skill_install_failed_event("handoff"), "handoff_skill_install_failed");
        assert_eq!(super::skill_install_failed_event("wizard"), "wizard_skill_install_failed");

        let second = crate::test_support::logged(|| install_shipped_skills(dir.path()));
        assert!(
            !second.iter().any(|(_, fields)| fields.contains("_skill_installed")
                || fields.contains("_skill_install_failed")),
            "an unchanged boot writes no install line: {second:?}",
        );
    }

    /// A rewrite is named as one; a boot that only swept a leftover says so
    /// rather than claiming it wrote the skill.
    #[test]
    fn a_replaced_install_names_its_action() {
        let dir = tempfile::tempdir().expect("tempdir");
        install_shipped_skills(dir.path());
        std::fs::write(
            forge_workspace::skill_dir(dir.path(), "wizard").join("SKILL.md"),
            b"edited",
        )
        .expect("edit");

        let events = crate::test_support::logged(|| install_shipped_skills(dir.path()));
        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::INFO
                    && fields.contains("event_name=\"wizard_skill_installed\"")
                    && fields.contains("action=\"replaced\"")
            }),
            "the rewrite is named: {events:?}",
        );
    }

    /// **The sweep has its own records**: each removed path at debug, a count
    /// at info, and an install line that says `swept`, not `written`.
    #[test]
    fn a_sweep_names_what_it_removed() {
        let dir = tempfile::tempdir().expect("tempdir");
        install_shipped_skills(dir.path());
        std::fs::write(
            forge_workspace::skill_dir(dir.path(), "wizard").join("SKILL.md.tmp"),
            "half a write",
        )
        .expect("leftover");

        let events = crate::test_support::logged(|| install_shipped_skills(dir.path()));
        let has = |event: &str, level: tracing::Level| {
            events.iter().any(|(seen, fields)| {
                *seen == level && fields.contains(&format!("event_name=\"{event}\""))
            })
        };
        assert!(has("skill_file_removed", tracing::Level::DEBUG), "the path is named: {events:?}");
        assert!(has("skill_files_swept", tracing::Level::INFO), "the count is logged: {events:?}");
        assert!(
            events.iter().any(|(_, fields)| {
                fields.contains("event_name=\"wizard_skill_installed\"")
                    && fields.contains("action=\"swept\"")
            }),
            "a sweep is not a write: {events:?}",
        );
    }

    /// A removal that fails is a WARN inside the install, and the boot's
    /// install line still reports the sweep.
    #[test]
    fn a_failed_removal_warns_and_the_install_still_reports() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().expect("tempdir");
        install_shipped_skills(dir.path());
        let skill_dir = forge_workspace::skill_dir(dir.path(), "wizard");
        std::fs::write(skill_dir.join("SKILL.md.tmp"), "half a write").expect("leftover");
        std::fs::set_permissions(&skill_dir, std::fs::Permissions::from_mode(0o555)).expect("mode");
        let events = crate::test_support::logged(|| install_shipped_skills(dir.path()));
        std::fs::set_permissions(&skill_dir, std::fs::Permissions::from_mode(0o755)).expect("mode");

        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::WARN
                    && fields.contains("event_name=\"skill_file_remove_failed\"")
            }),
            "the failed removal warns: {events:?}",
        );
        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::INFO
                    && fields.contains("event_name=\"wizard_skill_installed\"")
                    && fields.contains("action=\"swept\"")
            }),
            "the install still reports: {events:?}",
        );
    }

    /// A symlinked skill directory is refused with a warning rather than
    /// written through.
    #[test]
    fn a_symlinked_skill_directory_is_refused_with_a_warning() {
        use std::os::unix::fs::symlink;
        let dir = tempfile::tempdir().expect("tempdir");
        let outside = tempfile::tempdir().expect("outside");
        std::fs::create_dir_all(dir.path().join("skills")).expect("skills dir");
        symlink(outside.path(), forge_workspace::skill_dir(dir.path(), "wizard")).expect("symlink");

        let events = crate::test_support::logged(|| install_shipped_skills(dir.path()));
        assert!(
            events.iter().any(|(level, fields)| {
                *level == tracing::Level::WARN
                    && fields.contains("event_name=\"skill_dir_is_a_symlink\"")
                    && fields.contains("skill=\"wizard\"")
            }),
            "the refusal is a warning naming the skill: {events:?}",
        );
        assert!(!outside.path().join("SKILL.md").exists(), "nothing is written through it");
    }

    /// A stop whose teardown ran out of budget exits non-zero and names
    /// what was still running; a clean stop is a success.
    #[test]
    fn a_teardown_that_did_not_finish_stops_with_a_failure() {
        assert!(stop_result(&[]).is_ok(), "a clean stop is a success");

        let slot = forge_primitives::SessionSlot::new("org", "proj", "slow");
        let error = stop_result(std::slice::from_ref(&slot)).expect_err("a left tree is a failure");
        let rendered = format!("{error}");
        assert!(rendered.contains("org/proj/slow"), "the failure names the session: {rendered}");
        assert!(rendered.contains("its process tree"), "one session reads as one: {rendered}");

        let second = forge_primitives::SessionSlot::new("org", "proj", "slower");
        let both = [slot, second];
        let error = stop_result(&both).expect_err("two left trees are a failure");
        let rendered = format!("{error}");
        assert!(rendered.contains("2 sessions"), "two sessions read as two: {rendered}");
        assert!(rendered.contains("their process trees"), "and take the plural: {rendered}");
    }

    /// A bare boot fixture: one project whose account is base-url'd at a dead
    /// local port (its boot probe fails in milliseconds rather than reaching
    /// an endpoint), dictation off and a gateway port the OS just handed out,
    /// so the wave's other steps no-op or park. The port is picked free and
    /// then handed on - a busy box can take it in between, and the listener's
    /// failed bind is logged rather than read here.
    fn boot_fixture() -> (Arc<Workspace>, tempfile::TempDir) {
        fixture(false)
    }

    /// The same boot, with the socket serving on a port the OS hands out.
    fn socket_fixture() -> (Arc<Workspace>, tempfile::TempDir) {
        fixture(true)
    }

    /// The bare boot over two projects, `forge` first and `notes` second,
    /// so a wave test can tell a seat's state from its position: neither
    /// carries a row until the test seeds one, so nothing spawns by
    /// default.
    fn fixture(serve: bool) -> (Arc<Workspace>, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("tempdir");
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("a free port");
        let port = listener.local_addr().expect("the port the OS handed out").port();
        drop(listener);
        let socket = std::net::TcpListener::bind("127.0.0.1:0").expect("a free port");
        let socket_port = socket.local_addr().expect("the port the OS handed out").port();
        drop(socket);
        let forge = dir.path().join("forge");
        std::fs::create_dir_all(&forge).expect("forge/");
        std::fs::write(
            forge.join("forge.toml"),
            format!(
                "\
[[orgs]]
name = \"TestOrg\"
accounts = [\"Stub\"]

[[orgs.projects]]
name = \"forge\"
path = \"/tmp\"

[[orgs.projects]]
name = \"notes\"
path = \"/tmp\"

[[accounts]]
display_name = \"Stub\"
token = \"t\"
provider = \"openrouter\"
base_url = \"http://127.0.0.1:9/api\"
models = [\"claude-sonnet-5\"]

[server]
enabled = {serve}
port = {socket_port}

[gateway]
port = {port}
",
            ),
        )
        .expect("write forge.toml");
        let workspace = Workspace::new_for_test(dir.path().to_owned()).expect("workspace");
        (Arc::new(workspace), dir)
    }

    /// Drain the intercepted app-level commands until the wave's spawns
    /// arrive. The wave is a spawned task, so the command lands after
    /// `start_boot_wave` returns rather than inside it.
    async fn spawned_projects(
        workspace: &Arc<Workspace>,
    ) -> Vec<(String, forge_workspace::SessionLaunchSettings)> {
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                let spawned: Vec<_> = workspace
                    .drain_test_dispatch_buffer()
                    .into_iter()
                    .filter_map(|command| match command {
                        forge_workspace::Command::SpawnProject {
                            project_name,
                            launch_settings,
                        } => Some((project_name, launch_settings)),
                        _ => None,
                    })
                    .collect();
                if !spawned.is_empty() {
                    return spawned;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the boot wave dispatches its spawns")
    }

    /// **The boot wave is the only production site that asks for the
    /// restart note, and only a seat that was mid-turn earns it.** The
    /// intercept replaces the spawn, so what is read here is the command
    /// the wave would send: both open seats are in the wave, in forge.toml
    /// order, and only the interrupted one is kicked.
    #[tokio::test]
    async fn the_boot_wave_opens_the_seats_that_were_open() {
        use forge_workspace::store::sessions::SeatState;

        let (workspace, _dir) = fixture(false);
        workspace.enable_test_dispatch_intercept();
        workspace.seed_test_ready_account("Stub");
        workspace.seed_test_gateway_ready(true);
        assert!(
            wait_for_accounts(&workspace, Duration::from_secs(30)).await,
            "the fixture's accounts settle before the wave",
        );
        workspace
            .seed_seat_state_for_test("TestOrg", "forge", "lead", SeatState::Idle)
            .expect("seed the parked seat");
        workspace
            .seed_seat_state_for_test("TestOrg", "notes", "lead", SeatState::InTurn)
            .expect("seed the interrupted seat");

        start_boot_wave(Arc::clone(&workspace), None, false);

        let spawned = spawned_projects(&workspace).await;
        assert_eq!(spawned.len(), 2, "both open projects are in the wave: {spawned:?}");
        assert_eq!(spawned[0].0, "forge", "declaration order is the wave's order");
        assert_eq!(spawned[1].0, "notes");
        assert!(!spawned[0].1.force_new, "a normal launch is not a --new one");
        assert_eq!(
            spawned[0].1.resume_kick, None,
            "a seat that was parked is resumed with nothing to do",
        );
        assert_eq!(
            spawned[1].1.resume_kick.as_deref(),
            Some(forge_workspace::LEAD_RESTART_NOTE),
            "the seat whose turn was cut off takes the restart note",
        );
    }

    /// **The close door keeps a project out of the wave.** A closed lead is
    /// not spawned, and its stored workers ride its Connected hook - which
    /// never runs for a lead nothing starts.
    #[tokio::test]
    async fn the_boot_wave_leaves_a_closed_project_shut() {
        use forge_workspace::store::sessions::SeatState;

        let (workspace, _dir) = fixture(false);
        workspace.enable_test_dispatch_intercept();
        workspace.seed_test_ready_account("Stub");
        workspace.seed_test_gateway_ready(true);
        assert!(
            wait_for_accounts(&workspace, Duration::from_secs(30)).await,
            "the fixture's accounts settle before the wave",
        );
        workspace
            .seed_seat_state_for_test("TestOrg", "forge", "lead", SeatState::Idle)
            .expect("seed the open seat");
        workspace
            .seed_seat_state_for_test("TestOrg", "notes", "lead", SeatState::Closed)
            .expect("seed the closed seat");

        start_boot_wave(Arc::clone(&workspace), None, false);

        let dispatched = tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                let drained = workspace.drain_test_dispatch_buffer();
                if drained
                    .iter()
                    .any(|command| matches!(command, forge_workspace::Command::SpawnProject { .. }))
                {
                    return drained;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the boot wave dispatches its spawns");
        let lead_spawns: Vec<&String> = dispatched
            .iter()
            .filter_map(|command| match command {
                forge_workspace::Command::SpawnProject { project_name, .. } => Some(project_name),
                _ => None,
            })
            .collect();
        assert_eq!(lead_spawns, vec!["forge"], "the closed project's lead is not spawned");
    }

    /// `--new` still asks for the note: the spawn's resume gate is what
    /// drops it, so the wave's own contract does not depend on the flag.
    #[tokio::test]
    async fn the_boot_wave_asks_for_the_restart_note_under_new_as_well() {
        use forge_workspace::store::sessions::SeatState;

        let (workspace, _dir) = fixture(false);
        workspace.enable_test_dispatch_intercept();
        workspace.seed_test_ready_account("Stub");
        workspace.seed_test_gateway_ready(true);
        assert!(
            wait_for_accounts(&workspace, Duration::from_secs(30)).await,
            "the fixture's accounts settle before the wave",
        );
        workspace
            .seed_seat_state_for_test("TestOrg", "notes", "lead", SeatState::InTurn)
            .expect("seed the interrupted seat");

        start_boot_wave(Arc::clone(&workspace), None, true);

        let spawned = spawned_projects(&workspace).await;
        assert_eq!(spawned.len(), 1, "only the open project is in the wave: {spawned:?}");
        assert!(spawned[0].1.force_new, "--new reaches the command");
        assert_eq!(
            spawned[0].1.resume_kick.as_deref(),
            Some(forge_workspace::LEAD_RESTART_NOTE),
            "a --new wave still asks for the note; the spawn decides delivery",
        );
    }

    /// **A project named on the command line starts whether or not it is
    /// closed**, because naming it is the user's own act and closing is not
    /// forgetting. A closed seat has no interrupted turn to report, so it
    /// is resumed with nothing.
    #[tokio::test]
    async fn the_boot_wave_starts_a_named_closed_project() {
        use forge_workspace::store::sessions::SeatState;

        let (workspace, _dir) = fixture(false);
        workspace.enable_test_dispatch_intercept();
        workspace.seed_test_ready_account("Stub");
        workspace.seed_test_gateway_ready(true);
        assert!(
            wait_for_accounts(&workspace, Duration::from_secs(30)).await,
            "the fixture's accounts settle before the wave",
        );
        workspace
            .seed_seat_state_for_test("TestOrg", "forge", "lead", SeatState::Closed)
            .expect("seed the closed seat");

        start_boot_wave(Arc::clone(&workspace), Some("forge".to_owned()), false);

        let spawned = spawned_projects(&workspace).await;
        assert_eq!(spawned.len(), 1, "the named project alone: {spawned:?}");
        assert_eq!(spawned[0].0, "forge", "the named project starts though it is closed");
        assert_eq!(
            spawned[0].1.resume_kick, None,
            "a closed seat has no interrupted turn to report",
        );
    }

    /// **`serve_socket` returns with the fold already subscribed.** The boot
    /// wave runs after this returns, so a session spawned there emits its
    /// first `Connected` into a fan-out nobody is holding yet unless the
    /// subscription was taken before the serve task was spawned - and a
    /// dropped `Connected` is a seat whose conversation the transport never
    /// holds. Nothing has connected here: the subscriber is the fold's own.
    #[tokio::test]
    async fn the_socket_returns_with_the_fold_subscribed() {
        let (workspace, _dir) = socket_fixture();
        assert_eq!(workspace.test_subscriber_count(), 0, "nothing is attached before the socket");

        let serving = super::serve_socket(&workspace).await.expect("the socket serves");
        let super::SocketServing::Bound = serving else {
            panic!("the socket binds on a port the OS handed out");
        };
        // Read before anything awaits: the claim is the order the call returns in.
        assert_eq!(
            workspace.test_subscriber_count(),
            1,
            "the fold's subscription is held when serve_socket returns",
        );
        // **And it is a stream something reads.** One subscriber taking the
        // frame is what tells this count from a subscription taken and then
        // dropped; nothing else has attached, so the reader is the fold's.
        assert!(
            workspace.emit_for_test_reported(forge_workspace::SessionUpdate::CatalogLoaded),
            "the fold's own stream takes the frame the boot wave would emit",
        );
        workspace.supervisor().stop("socket_serve");
    }

    /// The wait ends when the watch reports, so this bound is reached when no
    /// report arrives within it: a stuck watch, or a delivery slower still.
    /// It sits under nextest's 120 s kill, so that failure names the property
    /// instead of dying as a bare timeout.
    const CONFIG_EDIT_WEDGE: Duration = Duration::from_secs(100);

    /// A write to the watched file applies to the running workspace, with no
    /// restart and no call from the test: the watch is the only path.
    ///
    /// macOS ignores it locally for the same reason `config_watch`'s delivery
    /// tests are: fsevents stalled a report for 407 s under a full gate. The
    /// macOS CI job runs it with `--run-ignored ignored-only`, and Ubuntu runs
    /// it on inotify.
    #[cfg_attr(target_os = "macos", ignore = "fsevents stalls under a full gate")]
    #[tokio::test]
    async fn a_config_edit_reaches_the_running_workspace() {
        let (workspace, _dir) = boot_fixture();
        let mut updates = workspace.subscribe_answerer();
        start_config_watch(&workspace, |_workspace| {});

        let path = workspace.config_path();
        let raw = std::fs::read_to_string(&path).expect("read");
        std::fs::write(&path, raw.replace("Stub", "Renamed")).expect("write");

        // The wait is the update stream, not a clock: fsevents schedules the
        // delivery, so only a stuck watch, or a delivery slower than the
        // bound, reaches the wedge.
        let applied = tokio::time::timeout(CONFIG_EDIT_WEDGE, async {
            loop {
                let frame = updates.recv().await.expect("the update stream stays open");
                if matches!(&*frame.update, forge_workspace::SessionUpdate::AccountsChanged { .. })
                {
                    break;
                }
            }
        })
        .await;
        assert!(
            applied.is_ok(),
            "the watched file's edit reaches the running workspace within {CONFIG_EDIT_WEDGE:?}",
        );
        assert_eq!(
            workspace.list_projects()[0].accounts,
            vec!["Renamed".to_owned()],
            "the running workspace holds the edited org pin",
        );
    }

    /// **The config's names are the fleet's maps.** The workspace's own
    /// resolution and the fleet's lookup both have tests; this is the boot's
    /// copy between them, so a rename on either side fails here.
    #[cfg(target_os = "macos")]
    #[test]
    fn the_fleets_profile_names_are_the_configs() {
        use forge_primitives::SessionSlot;

        let dir = tempfile::tempdir().expect("tempdir");
        let forge = dir.path().join("forge");
        std::fs::create_dir_all(&forge).expect("forge/");
        std::fs::write(
            forge.join("forge.toml"),
            "\
[[orgs]]
name = \"Busytools\"
accounts = [\"Stub\"]
browser_profile = \"work\"

[[orgs.projects]]
name = \"forge\"
path = \"/tmp\"

[[orgs.projects]]
name = \"apply\"
path = \"/tmp\"
browser_profile = \"job\"

[[accounts]]
display_name = \"Stub\"
token = \"t\"
provider = \"openrouter\"
base_url = \"http://127.0.0.1:9/api\"
models = [\"claude-sonnet-5\"]

[server]
enabled = false

[gateway]
port = 49201
",
        )
        .expect("write forge.toml");
        let workspace = Workspace::new_for_test(dir.path().to_owned()).expect("workspace");
        let fleet = forge_browser::service::BrowserService::at(
            dir.path().join("browser"),
            dir.path().join("browser/stack"),
        );

        configure_fleet_profiles(&workspace, &fleet);

        let key = |project: &str| {
            fleet
                .profile_for(&SessionSlot::lead("Busytools", project))
                .expect("a seat resolves")
                .as_str()
                .to_owned()
        };
        assert_eq!(key("forge"), "Busytools/work", "the org's name arrives");
        assert_eq!(key("apply"), "Busytools/job", "the project's override arrives");
    }

    /// The wave starts the 60 s usage poller. Its first tick is a minute out
    /// and its pass reaches the network, so the swap-once guard is what a
    /// test can read - a wave without it leaves the account snapshots frozen
    /// at their boot values and the session page's 5h/7d bars never move.
    ///
    /// The wave is driven directly here: `start_core` reaches the machine's
    /// single-instance lock, so the one call from it into this function is the
    /// boundary this test does not cover.
    #[tokio::test]
    async fn the_boot_wave_starts_the_usage_poller() {
        let (workspace, _dir) = boot_fixture();
        assert!(
            !workspace.test_usage_poller_started(),
            "the fixture starts with the poller unstarted, or this test proves nothing",
        );

        start_subsystems(&workspace);

        assert!(workspace.test_usage_poller_started(), "the boot wave starts the usage poller");
    }

    /// **The prune runs after the reap.** Reversing them is invisible to
    /// every behavioural test that does not need the identity a prune can
    /// unlink, so the order itself is pinned.
    #[cfg(target_os = "macos")]
    #[tokio::test]
    async fn the_prune_runs_after_the_reap() {
        let order = Arc::new(std::sync::Mutex::new(Vec::new()));
        let reaped = Arc::clone(&order);
        let pruned = Arc::clone(&order);
        reap_then_prune(
            async move {
                reaped.lock().expect("the order").push("reap");
            },
            move || {
                pruned.lock().expect("the order").push("prune");
            },
        )
        .await;
        assert_eq!(*order.lock().expect("the order"), vec!["reap", "prune"]);
    }

    /// **The wave starts the rotation announcer**: the one watcher that turns
    /// a cooldown write and its clock-driven thaw into a roster frame. Dropping
    /// the call leaves a home row's `would_bind` stale with every test green.
    #[tokio::test]
    async fn the_boot_wave_starts_the_rotation_announcer() {
        let (workspace, _dir) = boot_fixture();
        assert!(
            !workspace.test_rotation_announcer_started(),
            "the fixture starts with the announcer unstarted, or this test proves nothing",
        );

        start_subsystems(&workspace);

        assert!(
            workspace.test_rotation_announcer_started(),
            "the boot wave starts the rotation announcer",
        );
    }

    /// **The wave starts the idle sweep**: without the call the feature is
    /// dark in every real boot while every unit test stays green.
    #[tokio::test]
    async fn the_boot_wave_starts_the_idle_sweep() {
        let (workspace, _dir) = boot_fixture();
        assert!(
            !workspace.supervisor().snapshot().iter().any(|member| member.name == "idle_sweep"),
            "the fixture starts with the idle sweep unstarted, or this test proves nothing",
        );

        start_subsystems(&workspace);

        assert!(
            workspace.supervisor().snapshot().iter().any(|member| member.name == "idle_sweep"),
            "the boot wave starts the idle sweep",
        );
    }

    /// The wave starts the dictate prewarm, which is what keeps a take from
    /// paying the model load: started at boot rather than at the first press,
    /// on a task of its own so the fetch, the verify and the load are not on
    /// the boot path. The fixture has dictation switched off, so the guard is
    /// what the test reads; the prewarm's own fetch and load need the real
    /// weights and are exercised by forge-dictate's own gates.
    #[tokio::test]
    async fn the_boot_wave_starts_the_dictate_prewarm() {
        let (workspace, _dir) = boot_fixture();
        assert!(
            !workspace.test_dictate_preflight_started(),
            "the fixture starts with the prewarm unstarted, or this test proves nothing",
        );

        start_subsystems(&workspace);

        assert!(
            workspace.test_dictate_preflight_started(),
            "the boot wave asks for the prewarm; dropping the call leaves every take \
             paying the load",
        );
    }

    /// The wave asks for the speaking engine's prewarm, which is what keeps
    /// the first `say` from paying the model load: started at boot rather
    /// than on the first listener, on a task of its own so the
    /// multi-gigabyte load is not on the boot path. The fixture has `[tts]`
    /// off, so this reads the wave-asked guard only; that the dispatch then
    /// loads the engine is pinned by
    /// `the_boot_prewarm_loads_the_engine_when_tts_is_on` in
    /// forge-workspace, which drives a stub load.
    #[tokio::test]
    async fn the_boot_wave_starts_the_voice_prewarm() {
        let (workspace, _dir) = boot_fixture();
        assert!(
            !workspace.test_voice_prewarm_started(),
            "the fixture starts with the prewarm unstarted, or this test proves nothing",
        );

        start_subsystems(&workspace);

        assert!(
            workspace.test_voice_prewarm_started(),
            "the boot wave asks for the prewarm; dropping the call leaves the first say \
             paying the load",
        );
    }

    /// The hold gives up at its deadline rather than waiting for ever - the
    /// behaviour a listener that never binds would otherwise hang on. Pinned
    /// against the production bound's VALUE as well, so raising it is a
    /// change this test has to be told about.
    #[tokio::test]
    async fn the_boot_wave_hold_expires_rather_than_waiting_for_ever() {
        let (workspace, _dir) = crate::surface::testing::workspace();
        assert!(
            !workspace.all_accounts_loaded(),
            "the fixture's accounts start unsettled, or this test proves nothing",
        );
        assert_eq!(BOOT_WAVE_WAIT, Duration::from_secs(60), "the production bound");

        let settled = wait_for_accounts(&workspace, Duration::from_millis(1)).await;
        assert!(!settled, "the hold gives up at its deadline instead of waiting for ever");
    }

    /// And it releases the moment the accounts settle, without waiting the
    /// bound out.
    #[tokio::test]
    async fn the_boot_wave_hold_releases_once_the_accounts_settle() {
        let (workspace, _dir) = crate::surface::testing::workspace();
        workspace.seed_test_ready_account("Granite");
        workspace.seed_test_ready_account("OpenRouter-TM");
        assert!(workspace.all_accounts_loaded(), "both accounts terminal and the listener ready");

        let started = std::time::Instant::now();
        let settled = wait_for_accounts(&workspace, Duration::from_secs(60)).await;
        assert!(settled, "a settled boot is released at once");
        assert!(
            started.elapsed() < Duration::from_secs(1),
            "and it does not wait the bound out: {:?}",
            started.elapsed(),
        );
    }
}