clawmates/crates/bins/clawmates-server/src/main.rs

//! Clawmates server: REST API and streaming gateway (later phases add the
//! scheduler and safety worker) composed into one binary.

mod e2e;

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

use cm_config::{AppConfig, LlmProviderKind};
use cm_llm::{AnthropicProvider, LlmProvider, OpenAiCompatProvider, ScriptedProvider};
use cm_runtime::{Runtime, RuntimeConfig};

#[tokio::main]
async fn main() -> ExitCode {
    match run().await {
        Ok(()) => ExitCode::SUCCESS,
        Err(message) => {
            eprintln!("clawmates-server: {message}");
            ExitCode::FAILURE
        }
    }
}

/// Instantiates the configured LLM provider. The Anthropic key comes from
/// the environment until the secret broker lands in P2.
///
/// The **subscription wins** when both credentials are present. This is the
/// structural half of the fix that `cm_api::subscription` does per-call: a bare
/// model name resolves to whatever this function returns, so making that the
/// subscription means no server-side call can reach the metered key by
/// accident — by construction, rather than by a source-grep test that has
/// already missed four call sites once. The metered key stays usable as a
/// fallback for deployments that have credit; ours does not, which is what
/// made the ordering matter.
fn build_provider(config: &AppConfig) -> Result<Arc<dyn LlmProvider>, String> {
    match config.llm.provider {
        LlmProviderKind::Anthropic => {
            if let Some(provider) = cm_api::subscription::provider() {
                println!(
                    "clawmates-server: default LLM provider = Claude Code subscription \
                     (bare model names bill no metered key)"
                );
                return Ok(Arc::new(provider));
            }
            let key = std::env::var("ANTHROPIC_API_KEY").map_err(|_| {
                "llm.provider = \"anthropic\" needs a credential: either \
                 ANTHROPIC_OAUTH_TOKEN / CLAUDE_CODE_OAUTH_TOKEN (sk-ant-oat…, \
                 the Claude Code subscription, preferred) or ANTHROPIC_API_KEY \
                 (sk-ant-api…, metered)"
                    .to_string()
            })?;
            // A subscription OAuth token pasted where an API key belongs
            // authenticates nothing here and fails on the first model call,
            // far from the mistake. Both start `sk-ant-`, so the confusion is
            // easy to make and hard to spot.
            if key.starts_with("sk-ant-oat") {
                return Err("ANTHROPIC_API_KEY looks like a subscription OAuth token \
                            (sk-ant-oat…), not a Console API key (sk-ant-api…). Set it \
                            as ANTHROPIC_OAUTH_TOKEN instead — that slot understands \
                            bearer auth and is what the phase evaluator reads."
                    .to_string());
            }
            eprintln!(
                "clawmates-server: WARNING — no subscription token; the default LLM \
                 provider is the METERED ANTHROPIC_API_KEY and every bare model name \
                 bills it"
            );
            Ok(Arc::new(AnthropicProvider::new(key)))
        }
        LlmProviderKind::OpenAiCompat => {
            let base_url = config.llm.base_url.clone().expect("validated by cm-config");
            Ok(Arc::new(OpenAiCompatProvider::new(
                base_url,
                std::env::var("CLAWMATES_LLM_API_KEY").ok(),
            )))
        }
        LlmProviderKind::Scripted => {
            let path = config
                .llm
                .scenario_path
                .clone()
                .expect("validated by cm-config");
            let provider = ScriptedProvider::from_path(Path::new(&path))
                .map_err(|e| format!("scenario load failed: {e}"))?;
            Ok(Arc::new(provider))
        }
    }
}

/// Builds the extra named-provider registry (GLM, Kimi, …) from
/// `[[llm.providers]]`. Each is OpenAI-compatible; its key comes from the named
/// env var. A provider whose key env is unset is skipped with a warning, so a
/// missing key degrades that one selector rather than failing boot.
fn build_provider_registry(config: &AppConfig) -> cm_runtime::ProviderRegistry {
    let mut map = std::collections::HashMap::new();
    for p in &config.llm.providers {
        // A provider may legitimately need no key. A model running on our own
        // hardware has nothing to authenticate to, and requiring a variable
        // whose value is ignored is a step that can only ever fail — silently,
        // since an unset key SKIPS the provider and the first symptom is a
        // fallback chain quietly one link shorter than it reads.
        let key = match std::env::var(&p.api_key_env) {
            Ok(k) if !k.is_empty() => Ok(k),
            other if p.api_key_env.trim().is_empty() => {
                let _ = other;
                Ok(String::new())
            }
            other => other,
        };
        match key {
            Ok(key) if !key.is_empty() || p.api_key_env.trim().is_empty() => {
                let provider: Arc<dyn LlmProvider> = match p.format.as_str() {
                    "anthropic" => Arc::new(cm_llm::AnthropicProvider::with_base_url(
                        key,
                        p.base_url.clone(),
                    )),
                    _ => Arc::new(OpenAiCompatProvider::new(p.base_url.clone(), Some(key))),
                };
                map.insert(p.name.clone(), provider);
                println!(
                    "clawmates-server: registered LLM provider '{}' ({})",
                    p.name, p.format
                );
            }
            _ => eprintln!(
                "clawmates-server: provider '{}' skipped — {} is unset",
                p.name, p.api_key_env
            ),
        }
    }
    cm_runtime::ProviderRegistry(map)
}

async fn run() -> Result<(), String> {
    let config_path = PathBuf::from(
        std::env::var("CLAWMATES_CONFIG").unwrap_or_else(|_| "clawmates.toml".into()),
    );
    let config = AppConfig::load_from(&config_path).map_err(|e| e.to_string())?;
    let _telemetry = cm_telemetry::init(
        "clawmates-server",
        config.telemetry.otlp_endpoint.as_deref(),
    )
    .map_err(|e| format!("telemetry: {e}"))?;

    let pool = cm_db::connect(&config.database.url, config.database.max_connections)
        .await
        .map_err(|e| format!("database connection failed: {e}"))?;
    cm_db::MIGRATOR
        .run(&pool)
        .await
        .map_err(|e| format!("migrations failed: {e}"))?;

    if e2e::enabled() {
        e2e::seed(&pool).await?;
    }

    // First-owner bootstrap (self-hosted local-auth installs). Provisions
    // the initial workspace + Owner from env exactly once; a no-op on
    // every later boot. CLAWMATES_BOOTSTRAP_OWNER_PASSWORD is the trigger.
    if let Some(password) = std::env::var("CLAWMATES_BOOTSTRAP_OWNER_PASSWORD")
        .ok()
        .filter(|p| !p.is_empty())
    {
        let email = std::env::var("CLAWMATES_BOOTSTRAP_OWNER_EMAIL")
            .unwrap_or_else(|_| "[email protected]".into());
        let workspace = std::env::var("CLAWMATES_BOOTSTRAP_WORKSPACE")
            .unwrap_or_else(|_| "My Workspace".into());
        let credits = std::env::var("CLAWMATES_BOOTSTRAP_CREDITS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(1250);
        let created = cm_auth::bootstrap_owner(&pool, &workspace, &email, &password, credits)
            .await
            .map_err(|e| format!("bootstrap owner: {e}"))?;
        if created {
            println!("clawmates-server: bootstrapped first owner {email} in '{workspace}'");
        }
    }

    let provider = build_provider(&config)?;
    let provider_registry = build_provider_registry(&config);
    let blob: std::sync::Arc<dyn cm_files::BlobStore> = match config.storage.backend {
        cm_config::StorageBackend::Local => std::sync::Arc::new(cm_files::LocalBlobStore::new(
            PathBuf::from(&config.storage.data_dir),
        )),
        cm_config::StorageBackend::S3 => std::sync::Arc::new(
            cm_files::S3BlobStore::connect(
                config.storage.s3_endpoint.as_deref().expect("validated"),
                config.storage.s3_bucket.as_deref().expect("validated"),
                config.storage.s3_access_key.as_deref().unwrap_or_default(),
                config.storage.s3_secret_key.as_deref().unwrap_or_default(),
            )
            .map_err(|e| format!("s3 storage: {e}"))?,
        ),
    };
    // The fleet node hub (live daemon channels) is shared between the sandbox
    // placement provider and the API routes, so it must exist before the managers.
    let node_hub = std::sync::Arc::new(cm_api::fleet::NodeHub::new());
    // Environment tools need a container engine; absence is tolerated
    // (shell.exec reports it per-call) so the API still serves.
    let (sandboxes, browser, terminals) = if config.sandbox.enabled {
        match cm_sandbox::DockerDriver::connect() {
            Ok(driver) => {
                let driver: std::sync::Arc<dyn cm_sandbox::SandboxDriver> =
                    std::sync::Arc::new(driver);
                let agents = std::sync::Arc::new(
                    cm_runtime::SandboxManager::new(
                        driver.clone(),
                        pool.clone(),
                        "local",
                        &config.sandbox.image,
                    )
                    .with_node_provider(std::sync::Arc::new(
                        cm_api::fleet::HubDriverProvider::new(node_hub.clone()),
                    )),
                );
                let browser = std::sync::Arc::new(
                    cm_runtime::SandboxManager::new(
                        driver.clone(),
                        pool.clone(),
                        "local",
                        &config.sandbox.browser_image,
                    )
                    .with_egress(),
                );
                // Themed interactive terminal containers (zsh + oh-my-zsh + p10k)
                // for the Terminal computer app. On the Local storage backend the
                // file-drive volume is mounted (per-agent subpath) at ~/drives.
                let drives = if config.storage.backend == cm_config::StorageBackend::Local {
                    Some(cm_runtime::DriveConfig {
                        volume: config.sandbox.terminal_drive_volume.clone(),
                        data_dir: PathBuf::from(&config.storage.data_dir),
                    })
                } else {
                    None
                };
                let terminals = std::sync::Arc::new(
                    cm_runtime::TerminalManager::new(
                        driver,
                        pool.clone(),
                        "local",
                        &config.sandbox.terminal_image,
                        config.sandbox.terminal_egress,
                        drives,
                    )
                    // An agent placed on a fleet node runs its terminal there too
                    // (beside its sandbox), sharing the agent's node-local drives.
                    .with_node_provider(std::sync::Arc::new(
                        cm_api::fleet::HubDriverProvider::new(node_hub.clone()),
                    )),
                );
                // Boot reconciliation: any sandbox the engine still holds is an
                // orphan from a dead process (we track none yet) — remove them
                // before warming so a crash/redeploy can't leak containers.
                agents.reconcile_orphans(std::time::Duration::ZERO).await;
                browser.reconcile_orphans(std::time::Duration::ZERO).await;
                terminals.reconcile_orphans().await;
                let agents = if config.sandbox.warm_pool > 0 {
                    agents.warm(config.sandbox.warm_pool)
                } else {
                    agents
                };
                // Periodic reaper: catch leaks while we run (idle > 10 min and
                // owned by no live handle), checked every 5 min.
                agents.clone().spawn_reaper(
                    std::time::Duration::from_secs(300),
                    std::time::Duration::from_secs(600),
                );
                browser.clone().spawn_reaper(
                    std::time::Duration::from_secs(300),
                    std::time::Duration::from_secs(600),
                );
                // Terminals: reap idle (no live session for ~2 h, so a resumable
                // tmux session survives normal navigation gaps) + orphans.
                terminals.clone().spawn_reaper(
                    std::time::Duration::from_secs(300),
                    std::time::Duration::from_secs(7200),
                );
                (Some(agents), Some(browser), Some(terminals))
            }
            Err(error) => {
                eprintln!("clawmates-server: sandbox engine unavailable: {error}");
                (None, None, None)
            }
        }
    } else {
        (None, None, None)
    };
    // Keep handles for the graceful-shutdown drain (SIGTERM) — the managers
    // themselves are moved into the runtime config below.
    let drain_agents = sandboxes.clone();
    let drain_browser = browser.clone();
    let drain_terminals = terminals.clone();
    let runtime = Runtime::with_blob_store(
        pool.clone(),
        provider,
        RuntimeConfig {
            model: config.llm.model.clone(),
            max_tokens: 4096,
            broker_socket: Some(PathBuf::from(&config.broker.socket_path)),
            slack_base_url: config.slack.base_url.clone(),
            sandboxes,
            browser,
            terminals,
            providers: provider_registry,
        },
        blob.clone(),
    );
    // Durable §15 path: expires overdue approvals and resumes decided runs
    // even if the deciding request's process died mid-flight.
    runtime.spawn_resume_sweeper(std::time::Duration::from_secs(2));
    // Routine firings (§7.6).
    cm_scheduler::Scheduler::new(pool.clone(), runtime.clone())
        .spawn(std::time::Duration::from_secs(5));
    // Durable topology run jobs: claim queued runs, drive + checkpoint per step,
    // resume stale ones after a crash. Long-horizon topologies run here, not in
    // the HTTP request.
    cm_api::topology_worker::spawn(
        pool.clone(),
        runtime.clone(),
        // The composed tier (`microvm_graph`) runs each graph node as a VM on a
        // fleet node, so the worker needs the same hub the phase runner uses.
        node_hub.clone(),
        std::time::Duration::from_secs(3),
    );
    // Boot-time content loaders — skills first, then team templates
    // (Slice 3.5d): the team_template_loader binds template_role_skills
    // by looking up skills by name, so the skills catalog must be
    // populated first. Both are idempotent per boot.
    {
        let pool = pool.clone();
        tokio::spawn(async move {
            let n_skills = cm_api::skills_loader::load_builtins(&pool).await;
            eprintln!("skills_loader: loaded {n_skills} builtin skill(s)");
            let n_tpl = cm_api::team_template_loader::load_builtins(&pool).await;
            eprintln!("team_template_loader: loaded {n_tpl} builtin team template(s)");
        });
    }
    // Task-card parser worker (Slice 5): scans recent topology_runs
    // for INT-XX markers in event payloads and upserts mission_tasks
    // rows so the canvas renders a live status timeline.
    cm_api::task_card_worker::spawn(pool.clone());
    // Agents apply their own skill drafts. Announced at boot by the spawner
    // itself, because this flips an approval gate that existed since the
    // feature shipped — and a safety gate whose state is invisible is one
    // nobody notices has changed.
    cm_api::skill_self_authoring::spawn(pool.clone());
    // Load the workflow recipes now rather than lazily on first mission
    // create, so a malformed TOML shows up in the boot log instead of
    // silently yielding a mission with no phase config.
    {
        let recipes = cm_api::workflow_registry::load();
        eprintln!("workflow_registry: {} recipe(s) available", recipes.len());
    }
    // Announce how mission runtimes authenticate. Subscription mode is only
    // legitimate for a single-operator deployment — a consumer subscription
    // credential must never serve another person's work — and the mode is
    // otherwise invisible until it shows up on a bill, so state it at boot.
    {
        let mode = cm_api::mission_runtime::runtime_auth_mode();
        eprintln!(
            "mission_runtime: auth mode = {} (CLAWMATES_RUNTIME_AUTH)",
            mode.as_str()
        );
        if mode == cm_api::mission_runtime::RuntimeAuth::Subscription {
            match cm_db::repo::users::count_all(&pool).await {
                Ok(n) if n > 1 => eprintln!(
                    "mission_runtime: WARNING — subscription auth with {n} users in this \
                     deployment. A consumer subscription credential may only run the \
                     account holder's own work; move the runtime back to \
                     CLAWMATES_RUNTIME_AUTH=api_key before other people use it."
                ),
                Ok(_) => {}
                Err(e) => eprintln!("mission_runtime: user count check skipped: {e}"),
            }
        }
    }
    // The other half of runtime_preflight's question: the runtime has the TOOLS,
    // but can the independent JUDGE be reached? A dead validator makes every
    // done_when phase unmeetable, and without this the first symptom is a
    // mission failing after its VMs have already run.
    cm_api::validator_preflight::report_at_boot(runtime.clone());
    // Every link of the model fallback chain, probed through the real call path.
    // A chain is the one piece of infrastructure nobody looks at until the day it
    // has to work, so it is checked on the days it does not.
    cm_api::subscription::report_at_boot(runtime.clone());
    cm_api::phase_runner::spawn(pool.clone(), runtime.clone(), node_hub.clone());
    // Scheduled missions. `missions.schedule` has collected a cron from the
    // wizard since 0047 and NOTHING read it back — every scheduled mission ever
    // created sat in `draft` forever while the UI said it was on a schedule.
    // 60s matches the finest cron granularity; the sweep claims atomically and
    // records each occurrence in `mission_fires`, so replicas and restarts
    // cannot double-launch a container.
    // Render finished Continuous Research missions into episodes. A sweep, not
    // a phase step: rendering is not the agents' work and must not be able to
    // fail a phase that succeeded, and a transient API error simply retries on
    // the next tick.
    // Every 2 minutes, NOT 5. The mission checkout that holds script.md is
    // deleted 30 minutes after the mission reaches a terminal state, so this
    // sweep is racing a reaper. Two minutes leaves ~15 attempts inside that
    // window; a slower sweep loses the episode permanently.
    cm_api::podcast::spawn(
        pool.clone(),
        Some(blob.clone()),
        std::time::Duration::from_secs(2 * 60),
    );
    cm_api::mission_schedule::spawn(
        pool.clone(),
        Some(node_hub.clone()),
        Some(blob.clone()),
        std::time::Duration::from_secs(60),
    );
    // Per-mission runtime container sweeper (C3): tears down mission
    // runtime containers 30 min after the mission reaches a terminal
    // state so operators have a window to pull final artifacts.
    cm_api::mission_runtime::spawn_sweeper(pool.clone(), std::time::Duration::from_secs(30 * 60));
    // Phase completion summarizer: reads terminal-state phases and
    // asks Claude Opus 4.8 to synthesize a "what got done" card that
    // the UI renders under the phase.
    cm_api::phase_summarizer::spawn(pool.clone(), runtime.clone());
    // Outbound-email delivery: drains the §15-gated `outbox` over SMTP. Inert
    // until CLAWMATES_SMTP_* is set, so it ships safely before credentials exist.
    cm_runtime::spawn_drainer(pool.clone(), std::time::Duration::from_secs(10));
    // Loop scheduler: fires cron-triggered loop iterations. Missed windows
    // fire ONCE and skip the backlog (see cm_runtime::loops for details).
    // Expiry/retention sweep: expires stale auth/oauth rows and prunes old
    // journal/audit rows hourly so unbounded tables don't accumulate.
    cm_api::cleanup_sweeper::spawn(pool.clone(), std::time::Duration::from_secs(3600));
    // Its filesystem counterpart. `cleanup_sweeper` prunes ROWS, and deleting a
    // row has never deleted a directory — which is why the gateway, the smallest
    // disk in the fleet, accumulates mission trees that nothing reclaims.
    cm_api::mission_gc::spawn(pool.clone(), std::time::Duration::from_secs(3600));
    // Agent lifecycle: reap crews whose missions finished (after a 24h grace so
    // the results view can still show who did the work) and crews left bound to
    // nothing. Never touches an agent without an `agent_template_link` row —
    // that is the operator's own staff, which looks identical to an orphan if
    // you judge by team membership alone.
    cm_api::agent_lifecycle::spawn(
        pool.clone(),
        runtime.clone(),
        std::time::Duration::from_secs(3600),
    );
    // Fleet backstop: a node whose heartbeats stop (without a clean channel
    // close) goes offline within ~28s even if its control channel hangs.
    cm_api::fleet::spawn_node_sweeper(pool.clone(), std::time::Duration::from_secs(8), 20);
    // Beszel: poll each workspace's monitoring hub for rich per-node metrics.
    cm_api::beszel::spawn_poller(pool.clone(), std::time::Duration::from_secs(15));
    // Fleet automation: evaluate metric-threshold rules → drain/undrain/alert.
    cm_api::node_rules::spawn_evaluator(pool.clone(), std::time::Duration::from_secs(20));
    // Nightly: check upstream for newer dev-tool releases (claude/kimi/ollama).
    cm_api::tool_versions::spawn_latest_checker(
        pool.clone(),
        std::time::Duration::from_secs(86_400),
    );

    // Hosted identity (Clerk / OIDC): pin the issuer and load its JWKS.
    let auth_verifier = match config.auth.mode {
        cm_config::AuthMode::Clerk | cm_config::AuthMode::Oidc => {
            let issuer = config
                .auth
                .issuer_url
                .as_deref()
                .expect("validated by config");
            Some(std::sync::Arc::new(
                cm_auth::JwtVerifier::discover(issuer)
                    .await
                    .map_err(|e| format!("identity issuer: {e}"))?,
            ))
        }
        cm_config::AuthMode::Local => None,
    };

    let mut app = cm_api::router(
        cm_api::AppState::new(pool, runtime)
            .with_node_hub(node_hub.clone())
            .with_broker(PathBuf::from(&config.broker.socket_path))
            .with_oauth(config.oauth.clone())
            .with_billing(config.billing.clone())
            .with_blobs(blob.clone())
            .with_file_root(
                (config.storage.backend == cm_config::StorageBackend::Local)
                    .then(|| PathBuf::from(&config.storage.data_dir)),
            )
            .with_per_signup_workspace(config.auth.per_signup_workspace)
            .pipe_auth_verifier(auth_verifier),
    );
    if e2e::enabled() {
        app = app.merge(e2e::slack_sink_router());
    }
    let listener = tokio::net::TcpListener::bind(config.listen_addr)
        .await
        .map_err(|e| format!("bind {} failed: {e}", config.listen_addr))?;
    println!("clawmates-server listening on {}", config.listen_addr);
    // Say plainly whether the mission runtime carries the tools we invoke in
    // it. The image on the host silently fell behind its Dockerfile once, and
    // every consequence — an ungated test suite, a scan that scanned nothing —
    // looked like a normal result rather than a broken deployment.
    cm_api::runtime_preflight::report_at_boot();
    // And whether the gateway those missions drive is configured at all. Both
    // of its variables are read at FIRST USE, so a deployment missing them
    // boots clean and fails on the first phase someone runs.
    cm_api::gateway_preflight::report_at_boot();
    // Graceful shutdown: on SIGTERM/Ctrl-C, stop accepting, finish in-flight
    // requests, then DRAIN the sandbox managers so no container is left running.
    let shutdown = async move {
        let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("install SIGTERM handler");
        tokio::select! {
            _ = term.recv() => {}
            _ = tokio::signal::ctrl_c() => {}
        }
        eprintln!("clawmates-server: shutdown signal received");
    };
    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown)
        .await
        .map_err(|e| format!("server error: {e}"))?;
    eprintln!("clawmates-server: draining sandboxes…");
    if let Some(agents) = drain_agents {
        agents.shutdown().await;
    }
    if let Some(browser) = drain_browser {
        browser.shutdown().await;
    }
    if let Some(terminals) = drain_terminals {
        terminals.shutdown().await;
    }
    Ok(())
}