clawmates/crates/cm-api/src/routes/topology.rs

//! Topology endpoints: the catalog, structural classification, and building a
//! canonical graph from a kind + roles. Stateless (no DB, no provider) — these
//! back the topology builder UI. Running/comparing topologies is a later,
//! provider-backed endpoint.

use std::convert::Infallible;
use std::time::Duration;

use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::IntoResponse;
use axum::Json;
use cm_orchestrator::{compare, Comparison, JudgeScorer, ProviderExecutor};
use cm_topology::{build, classify, heuristics, Classification, TopologyGraph, TopologyKind};
use serde::{Deserialize, Serialize};
use time::format_description::well_known::Rfc3339;
use uuid::Uuid;

use crate::{ApiError, AppState, Authed};

/// A role and its suggested share of the team.
#[derive(Serialize)]
pub struct RoleWeight {
    pub role: String,
    pub weight: f32,
}

/// One supported topology kind, with its description and default role mix.
#[derive(Serialize)]
pub struct CatalogEntry {
    pub kind: TopologyKind,
    pub name: String,
    pub description: String,
    pub role_distribution: Vec<RoleWeight>,
    /// The execution pattern this kind actually runs as. Twelve kinds map onto
    /// five patterns, so this differs from `name` for the aliased ones.
    pub executes_as: String,
    /// False when the kind is an alias — its description promises semantics the
    /// engine does not implement (Market never auctions, Ring never cycles).
    /// A UI should not offer these as if they behaved differently.
    pub distinct_at_execution: bool,
}

/// `GET /api/topologies` — the catalog of supported topology kinds.
pub async fn catalog(_auth: Authed) -> Json<Vec<CatalogEntry>> {
    let entries = TopologyKind::ALL
        .iter()
        .map(|&kind| {
            let h = heuristics(kind);
            CatalogEntry {
                kind,
                name: kind.as_str().to_string(),
                description: kind.description().to_string(),
                role_distribution: h
                    .role_distribution
                    .iter()
                    .map(|(role, weight)| RoleWeight {
                        role: (*role).to_string(),
                        weight: *weight,
                    })
                    .collect(),
                executes_as: kind.execution_pattern().as_str().to_string(),
                distinct_at_execution: kind.is_distinct_at_execution(),
            }
        })
        .collect();
    Json(entries)
}

/// `POST /api/topologies/classify` — infer a topology kind from a graph.
pub async fn classify_graph(
    _auth: Authed,
    Json(graph): Json<TopologyGraph>,
) -> Json<Classification> {
    Json(classify(&graph))
}

/// Request body for building a canonical topology.
#[derive(Deserialize)]
pub struct BuildRequest {
    pub kind: TopologyKind,
    pub roles: Vec<String>,
}

/// `POST /api/topologies/build` — build a canonical graph from a kind + roles.
pub async fn build_graph(
    _auth: Authed,
    Json(req): Json<BuildRequest>,
) -> Result<Json<TopologyGraph>, ApiError> {
    let roles: Vec<&str> = req.roles.iter().map(String::as_str).collect();
    let graph = build(req.kind, &roles).map_err(|_| ApiError::BadRequest)?;
    Ok(Json(graph))
}

/// Request body for running a multi-topology comparison.
#[derive(Deserialize)]
pub struct CompareRequest {
    pub task: String,
    pub graphs: Vec<TopologyGraph>,
}

/// `POST /api/topologies/compare` — run a task across the given topologies and
/// return a leaderboard + quality/cost Pareto front. Uses the configured
/// provider for both execution (tool-free turns) and the LLM judge.
pub async fn compare_topologies(
    State(state): State<AppState>,
    Authed(user): Authed,
    Json(req): Json<CompareRequest>,
) -> Result<Json<Comparison>, ApiError> {
    // Execution turns run on the exec model (default = configured model, e.g.
    // sonnet); the judge uses the judge model (cm_runtime::judge_model). Either
    // can name a registry provider as "<name>:<model>" (e.g. "glm:glm-4.6",
    // "kimi:kimi-k2") to run on GLM/Kimi instead.
    let exec_spec = std::env::var("CLAWMATES_TOPOLOGY_EXEC_MODEL")
        .unwrap_or_else(|_| state.runtime.model().to_string());
    let (exec_provider, exec_model) = state.runtime.resolve_provider(&exec_spec);
    // `cm_runtime::judge_model()`, not a second read of the same variable: this
    // line and that function disagreed on the default (opus-4-8 vs opus-5), so
    // an unconfigured deployment scored topology comparisons on a different
    // model than the door governor and nothing recorded which.
    let judge_spec = cm_runtime::judge_model();
    let (judge_provider, judge_model) = state.runtime.resolve_provider(&judge_spec);
    let executor = ProviderExecutor::new(exec_provider, exec_model, state.runtime.max_tokens());
    let scorer = JudgeScorer::new(judge_provider, judge_model, 16);
    let cmp = compare(&req.graphs, &req.task, &executor, &scorer)
        .await
        .map_err(|_| ApiError::Internal)?;

    // Best-effort persistence: never lose the (expensive) result on a DB hiccup.
    if let Ok(value) = serde_json::to_value(&cmp) {
        let _ = cm_db::repo::topology_runs::insert(
            &state.pool,
            Uuid::now_v7(),
            user.workspace_id,
            &req.task,
            &value,
        )
        .await;
    }
    Ok(Json(cmp))
}

/// Request body for executing a single topology on a real agent container.
#[derive(Deserialize)]
pub struct RunRequest {
    pub task: String,
    pub graph: TopologyGraph,
}

/// Response to an accepted async run job.
#[derive(Serialize)]
pub struct RunAccepted {
    pub run_id: String,
    pub status: String,
}

/// `POST /api/topologies/run` — ENQUEUE a durable single-topology run and return
/// its id immediately (202). The background worker (`topology_worker`) claims it,
/// drives the ZeroClaw role-agents turn-by-turn (orchestrator owns the graph;
/// agents are tool-free behind the §15 MCP door), and checkpoints per step so a
/// long-horizon run survives restarts. Poll `GET /api/topology-runs/{id}` for
/// status and the final result. This async model is what makes minutes-to-hours
/// runs viable — the work no longer lives inside the HTTP request.
pub async fn run_topology(
    State(state): State<AppState>,
    Authed(user): Authed,
    Json(req): Json<RunRequest>,
) -> Result<(StatusCode, Json<RunAccepted>), ApiError> {
    let graph = serde_json::to_value(&req.graph).map_err(|_| ApiError::Internal)?;
    let id = Uuid::now_v7();
    cm_db::repo::topology_runs::enqueue_run(&state.pool, id, user.workspace_id, &req.task, &graph)
        .await?;
    Ok((
        StatusCode::ACCEPTED,
        Json(RunAccepted {
            run_id: id.to_string(),
            status: "queued".into(),
        }),
    ))
}

/// `POST /api/swarm/run` — ENQUEUE a self-verifying swarm run (tier `swarm`): Opus
/// plans tasks → a worker swarm executes → Opus verifies each against the checklist
/// → failures requeue → loop until clean. Streams into the Runs view like any run.
#[derive(serde::Deserialize)]
pub struct SwarmRunRequest {
    pub goal: String,
    #[serde(default)]
    pub checklist: Vec<String>,
    #[serde(default)]
    pub task_count: Option<usize>,
    #[serde(default)]
    pub worker_model: String,
}

pub async fn run_swarm(
    State(state): State<AppState>,
    Authed(user): Authed,
    Json(req): Json<SwarmRunRequest>,
) -> Result<(StatusCode, Json<RunAccepted>), ApiError> {
    if req.goal.trim().is_empty() {
        return Err(ApiError::BadRequest);
    }
    let id = Uuid::now_v7();
    let config = serde_json::json!({
        "goal": req.goal,
        "checklist": req.checklist,
        "task_count": req.task_count,
        "worker_model": req.worker_model,
    });
    cm_db::repo::topology_runs::enqueue_run_tier(
        &state.pool,
        id,
        user.workspace_id,
        &req.goal,
        &config,
        "swarm",
    )
    .await?;
    Ok((
        StatusCode::ACCEPTED,
        Json(RunAccepted {
            run_id: id.to_string(),
            status: "queued".into(),
        }),
    ))
}

/// A saved/queued run, summarized (now includes lifecycle status + kind).
/// `iteration` + `finished_at` populate for loop iterations / terminal runs;
/// they're skipped from the JSON when null to keep the compares path compact.
#[derive(Serialize)]
pub struct RunSummary {
    pub id: String,
    pub task: String,
    pub status: String,
    pub kind: String,
    pub created_at: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub finished_at: Option<String>,
}

/// Query params for `GET /api/topology-runs`.
#[derive(Deserialize)]
pub struct ListRunsQuery {
    #[serde(default)]
    pub limit: Option<i64>,
}

/// `GET /api/topology-runs` — recent runs for the workspace (compares + durable
/// run jobs), newest first.
pub async fn list_runs(
    State(state): State<AppState>,
    Authed(user): Authed,
    Query(q): Query<ListRunsQuery>,
) -> Result<Json<Vec<RunSummary>>, ApiError> {
    let limit = q.limit.filter(|n| *n > 0 && *n <= 200).unwrap_or(20);
    let rows =
        cm_db::repo::topology_runs::list_recent(&state.pool, user.workspace_id, limit).await?;
    let out = rows
        .into_iter()
        .map(|r| RunSummary {
            id: r.id.to_string(),
            task: r.task,
            status: r.status,
            kind: r.kind,
            created_at: r.created_at.format(&Rfc3339).unwrap_or_default(),
            finished_at: r.finished_at.and_then(|t| t.format(&Rfc3339).ok()),
        })
        .collect();
    Ok(Json(out))
}

/// A single run with lifecycle status + progress. `comparison` is the result
/// blob (a `Comparison` for compares, a `RunRecord` for run jobs) and is `null`
/// until the job completes; pollers watch `status` and read `comparison` when it
/// flips to `completed`. `checkpoint` exposes mid-run progress for live views.
#[derive(Serialize)]
pub struct RunDetail {
    pub id: String,
    pub task: String,
    pub kind: String,
    pub status: String,
    pub error: Option<String>,
    pub created_at: String,
    pub updated_at: String,
    pub comparison: serde_json::Value,
    pub checkpoint: Option<serde_json::Value>,
}

/// `GET /api/topology-runs/{id}/events` — Server-Sent Events stream of live run
/// progress. Tails the durable per-step `checkpoint` the worker writes: emits a
/// `step` event per newly-completed step (replaying all so far on connect, so a
/// reload/reconnect re-attaches), then a terminal `done` event with the final
/// output (or error). Each `step` carries the step index as its SSE id, so the
/// browser's automatic `Last-Event-ID` on reconnect resumes without duplicates.
/// This gives the UI live long-horizon progress with no client polling.
pub async fn run_events_sse(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
    headers: HeaderMap,
) -> impl IntoResponse {
    let pool = state.pool.clone();
    let ws = user.workspace_id;
    // Resume after the last step the client already saw (SSE Last-Event-ID).
    let mut sent: usize = headers
        .get("last-event-id")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse::<usize>().ok())
        .map(|n| n + 1)
        .unwrap_or(0);

    // Bytes of `checkpoint.log` already sent. The step cursor above counts
    // RECORDS; this counts BYTES, because a log grows continuously rather than
    // in discrete entries. Two sources, two cursors.
    let mut log_sent: usize = 0;
    let stream = async_stream::stream! {
        loop {
            match cm_db::repo::topology_runs::status(&pool, id, ws).await {
                Ok(st) => {
                    if let Some(records) = st
                        .checkpoint
                        .as_ref()
                        .and_then(|c| c.get("records"))
                        .and_then(|r| r.as_array())
                    {
                        while sent < records.len() {
                            yield Ok::<Event, Infallible>(Event::default()
                                .id(sent.to_string())
                                .event("step")
                                .data(records[sent].to_string()));
                            sent += 1;
                        }
                    }
                    // Live stdout/stderr from a microVM turn, appended by the
                    // node over the fleet WebSocket (`Uplink::VmOut`). Emitted
                    // as `step` so the existing reader renders it with no
                    // frontend change — it already reads `data.text`.
                    if let Some(log) = st
                        .checkpoint
                        .as_ref()
                        .and_then(|c| c.get("log"))
                        .and_then(|v| v.as_str())
                    {
                        if log.len() > log_sent {
                            let fresh = &log[log_sent..];
                            log_sent = log.len();
                            yield Ok::<Event, Infallible>(Event::default().event("step").data(
                                serde_json::json!({ "kind": "output", "text": fresh }).to_string(),
                            ));
                        }
                    }
                    if matches!(st.status.as_str(), "completed" | "failed" | "cancelled") {
                        let done = serde_json::json!({
                            "status": st.status,
                            "error": st.error,
                            "final_output": st.result.as_ref().and_then(|r| r.get("final_output")),
                            "totals": st.result.as_ref().and_then(|r| r.get("totals")),
                        });
                        yield Ok(Event::default().event("done").data(done.to_string()));
                        break;
                    }
                }
                // Unknown id / wrong workspace / gone: end the stream.
                Err(_) => break,
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}

/// A small, JSON-safe view of what a run actually produced. The full
/// `checkpoint` blob can be hundreds of KB per run; this endpoint
/// returns just the counters + trimmed output previews so mission
/// phase cards can render "what did this run do" without dragging the
/// whole checkpoint through the wire on every 3-second poll.
#[derive(Serialize)]
pub struct RunOutput {
    pub status: String,
    pub turns: u64,
    pub tokens: u64,
    pub records_count: usize,
    /// Each entry is a truncated slice of `checkpoint.outputs[i]`
    /// (typically the concatenated agent text output for one turn).
    pub outputs: Vec<RunOutputSlice>,
    /// Error text if the run failed; empty otherwise.
    pub error: Option<String>,
}

#[derive(Serialize)]
pub struct RunOutputSlice {
    pub preview: String,
    pub truncated: bool,
    pub full_len: usize,
}

const OUTPUT_PREVIEW_MAX: usize = 6_000;
const OUTPUT_LIST_MAX: usize = 12;

/// `GET /api/topology-runs/{id}/output` — trimmed summary of what the
/// run produced (per-turn output previews + totals). Cheap enough for
/// the mission page to fetch inline on-demand for any completed run.
pub async fn get_run_output(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<Json<RunOutput>, ApiError> {
    let run = cm_db::repo::topology_runs::status(&state.pool, id, user.workspace_id).await?;
    let cp = run.checkpoint.unwrap_or(serde_json::Value::Null);
    let totals = cp.get("totals").cloned().unwrap_or(serde_json::Value::Null);
    let turns = totals.get("turns").and_then(|v| v.as_u64()).unwrap_or(0);
    let tokens = totals.get("tokens").and_then(|v| v.as_u64()).unwrap_or(0);
    let records_count = cp
        .get("records")
        .and_then(|v| v.as_array())
        .map(|a| a.len())
        .unwrap_or(0);
    let outputs_raw = cp
        .get("outputs")
        .and_then(|v| v.as_array())
        .cloned()
        .unwrap_or_default();
    let outputs = outputs_raw
        .into_iter()
        .take(OUTPUT_LIST_MAX)
        .map(|v| {
            let s = match v {
                serde_json::Value::String(s) => s,
                other => other.to_string(),
            };
            let full_len = s.chars().count();
            let truncated = full_len > OUTPUT_PREVIEW_MAX;
            let preview = if truncated {
                s.chars().take(OUTPUT_PREVIEW_MAX).collect()
            } else {
                s
            };
            RunOutputSlice {
                preview,
                truncated,
                full_len,
            }
        })
        .collect();
    Ok(Json(RunOutput {
        status: run.status,
        turns,
        tokens,
        records_count,
        outputs,
        error: run.error,
    }))
}

/// `POST /api/topology-runs/{id}/cancel` — request cancellation of a queued or
/// running job; the worker stops at its next step boundary. 409 if the run is
/// already terminal or unknown.
pub async fn cancel_run(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, ApiError> {
    let cancelled = cm_db::repo::topology_runs::cancel(&state.pool, id, user.workspace_id).await?;
    if cancelled {
        Ok(StatusCode::OK)
    } else {
        Err(ApiError::Conflict)
    }
}

/// `GET /api/topology-runs/{id}` — a single run with status + result.
pub async fn get_run(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<Json<RunDetail>, ApiError> {
    let run = cm_db::repo::topology_runs::status(&state.pool, id, user.workspace_id).await?;
    Ok(Json(RunDetail {
        id: run.id.to_string(),
        task: run.task,
        kind: run.kind,
        status: run.status,
        error: run.error,
        created_at: run.created_at.format(&Rfc3339).unwrap_or_default(),
        updated_at: run.updated_at.format(&Rfc3339).unwrap_or_default(),
        comparison: run.result.unwrap_or(serde_json::Value::Null),
        checkpoint: run.checkpoint,
    }))
}