Omar SobhandClaude Opus 5
107f0dbced
POST /api/library/runs harvests now; GET /api/library/items lists what the library holds. Thin wrappers — the work stays in crate::library — so a run can be started by a person, a schedule or the UI rather than only from an integration test. The response reports `healthy` explicitly rather than leaving a caller to infer it from an empty `shelved` list. A quiet week and a broken run both shelve zero papers, and collapsing those two is the exact ambiguity that cost most of this week. Failure reasons go to the log, not the response body: they can carry the remote URL and raw git stderr. AppState gains an optional blob store (the shelf), wired from the server binary where storage is already constructed. Optional because AppState::new is used by tests that never touch blobs; a route that needs it fails loudly rather than the constructor demanding it everywhere. 393 tests, clippy clean. Co-Authored-By: Claude Opus 5 <[email protected]>
438 lines
20 KiB
Rust
438 lines
20 KiB
Rust
//! 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")?;
|
|
// 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());
|
|
}
|
|
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.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(),
|
|
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());
|
|
// 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}"),
|
|
}
|
|
}
|
|
}
|
|
cm_api::phase_runner::spawn(pool.clone(), runtime.clone());
|
|
// 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());
|
|
// PDF renderer worker (Slice 6): watches mission_artifacts for
|
|
// MD entries with render_pdf_status='pending', calls the
|
|
// configured LLM (default Gemini 2.5 Flash) for styled HTML,
|
|
// prints to PDF via chromium --headless. No-op-friendly when
|
|
// GEMINI_API_KEY / chromium binary aren't configured.
|
|
cm_api::pdf_renderer::spawn(pool.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));
|
|
// 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();
|
|
// 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(())
|
|
}
|