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.
fn build_provider(config: &AppConfig) -> Result<Arc<dyn LlmProvider>, String> {
    match config.llm.provider {
        LlmProviderKind::Anthropic => {
            let key = std::env::var("ANTHROPIC_API_KEY")
                .map_err(|_| "llm.provider = \"anthropic\" requires ANTHROPIC_API_KEY")?;
            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 {
        match std::env::var(&p.api_key_env) {
            Ok(key) if !key.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,
    );
    // 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(),
        std::time::Duration::from_secs(3),
    );
    // 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));
    // 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));
    // 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));

    // 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_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);
    // 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(())
}