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))
        }
    }
}

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 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}"))?,
        ),
    };
    // Environment tools need a container engine; absence is tolerated
    // (shell.exec reports it per-call) so the API still serves.
    let (sandboxes, browser) = 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(),
                    &config.sandbox.image,
                ));
                let agents = if config.sandbox.warm_pool > 0 {
                    agents.warm(config.sandbox.warm_pool)
                } else {
                    agents
                };
                (
                    Some(agents),
                    Some(std::sync::Arc::new(
                        cm_runtime::SandboxManager::new(driver, &config.sandbox.browser_image)
                            .with_egress(),
                    )),
                )
            }
            Err(error) => {
                eprintln!("clawmates-server: sandbox engine unavailable: {error}");
                (None, None)
            }
        }
    } else {
        (None, None)
    };
    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,
        },
        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));

    // 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_broker(PathBuf::from(&config.broker.socket_path))
            .with_oauth(config.oauth.clone())
            .with_billing(config.billing.clone())
            .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);
    axum::serve(listener, app)
        .await
        .map_err(|e| format!("server error: {e}"))
}