Durable topology jobs (3+4/4): background worker + async run API

POST /api/topologies/run now ENQUEUES a durable job and returns 202
{run_id, status:queued} instead of executing inside the HTTP request — the
prerequisite for long-horizon runs (no client/proxy/LB timeout, survives
restarts).

topology_worker: a spawned loop that requeues stale running jobs, claims the
next queued one (CAS via FOR UPDATE SKIP LOCKED), drives it through
execute_resumable, and checkpoints RunProgress after every step; on crash the
stale sweep requeues it and the next claim resumes from the last checkpoint.
Wired into server startup beside the scheduler + resume sweeper.

GET /api/topology-runs/{id} now reports lifecycle status/kind/error/checkpoint
+ the result blob (kept the `comparison` field name for back-compat with the
compare UI; null until completed). list_runs includes status + kind.

Tests: durable lifecycle (enqueue→claim→checkpoint→complete) + stale-requeue
resume, both green; p0 endpoints (compare path) unchanged. 13 + 2 tests pass,
clippy clean.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

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

@@ -188,6 +188,10 @@ async fn run() -> Result<(), String> {
     // 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(), std::time::Duration::from_secs(3));
 
     // Hosted identity (Clerk / OIDC): pin the issuer and load its JWKS.
     let auth_verifier = match config.auth.mode {

crates/cm-api/src/lib.rs

@@ -5,6 +5,7 @@ mod extract;
 mod mcp_door;
 mod routes;
 mod topology_exec;
+pub mod topology_worker;
 
 use axum::routing::{delete, get, patch, post};
 use axum::Router;

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

@@ -4,8 +4,9 @@
 //! provider-backed endpoint.
 
 use axum::extract::{Path, State};
+use axum::http::StatusCode;
 use axum::Json;
-use cm_orchestrator::{compare, execute, Comparison, JudgeScorer, ProviderExecutor, RunRecord};
+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;
@@ -128,44 +129,50 @@ pub struct RunRequest {
     pub graph: TopologyGraph,
 }
 
-/// `POST /api/topologies/run` — execute one topology by driving real ZeroClaw
+/// Response to an accepted async run job.
-/// role-agents (in a container) for each turn, returning the run journal. The
+#[derive(Serialize)]
-/// orchestrator owns the graph; agents are tool-free behind the Clawmates MCP
+pub struct RunAccepted {
-/// door, so §15 holds by construction. Result is persisted best-effort to the
+    pub run_id: String,
-/// existing `topology_runs` table.
+    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<Json<RunRecord>, ApiError> {
+) -> Result<(StatusCode, Json<RunAccepted>), ApiError> {
-    let executor =
+    let graph = serde_json::to_value(&req.graph).map_err(|_| ApiError::Internal)?;
-        crate::topology_exec::ZeroClawDriveExecutor::from_env().map_err(|_| ApiError::Internal)?;
+    let id = Uuid::now_v7();
-    let record = execute(&req.graph, &req.task, &executor)
+    cm_db::repo::topology_runs::enqueue_run(&state.pool, id, user.workspace_id, &req.task, &graph)
-        .await
+        .await?;
-        .map_err(|_| ApiError::Internal)?;
+    Ok((
-
+        StatusCode::ACCEPTED,
-    if let Ok(value) = serde_json::to_value(&record) {
+        Json(RunAccepted {
-        let _ = cm_db::repo::topology_runs::insert(
+            run_id: id.to_string(),
-            &state.pool,
+            status: "queued".into(),
-            Uuid::now_v7(),
+        }),
-            user.workspace_id,
+    ))
-            &req.task,
-            &value,
-        )
-        .await;
-    }
-    Ok(Json(record))
 }
 
-/// A saved comparison run, summarized.
+/// A saved/queued run, summarized (now includes lifecycle status + kind).
 #[derive(Serialize)]
 pub struct RunSummary {
     pub id: String,
     pub task: String,
+    pub status: String,
+    pub kind: String,
     pub created_at: String,
 }
 
-/// `GET /api/topology-runs` — recent saved comparison runs for the workspace.
+/// `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,
@@ -176,32 +183,47 @@ pub async fn list_runs(
         .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(),
         })
         .collect();
     Ok(Json(out))
 }
 
-/// A full saved comparison run.
+/// 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}` — a single saved comparison run.
+/// `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::get(&state.pool, id, user.workspace_id).await?;
+    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(),
-        comparison: run.comparison,
+        updated_at: run.updated_at.format(&Rfc3339).unwrap_or_default(),
+        comparison: run.result.unwrap_or(serde_json::Value::Null),
+        checkpoint: run.checkpoint,
     }))
 }

crates/cm-api/src/topology_worker.rs

@@ -0,0 +1,100 @@
+//! Background worker that drains durable topology run jobs.
+//!
+//! A `POST /api/topologies/run` enqueues a job (`topology_runs` row, status
+//! `queued`); this loop claims it, drives the topology turn-by-turn via the
+//! ZeroClaw runtime, and checkpoints the [`RunProgress`] after every step. If
+//! the worker (or the whole server) dies mid-run, the row is left `running`;
+//! the stale sweep requeues it and the next claim resumes it from the last
+//! checkpointed step — so long-horizon runs survive restarts.
+//!
+//! This reuses the agent-run durability pattern (claim CAS, checkpoint, resume
+//! sweep) without coupling topology runs to the chat-session schema.
+
+use std::time::Duration;
+
+use cm_orchestrator::{execute_resumable, RunProgress};
+use cm_topology::TopologyGraph;
+use sqlx::PgPool;
+
+use crate::topology_exec::ZeroClawDriveExecutor;
+
+/// Requeue a `running` job whose worker hasn't checkpointed within this window.
+const STALE_AFTER_SECS: f64 = 180.0;
+
+/// Spawn the durable topology job worker. Polls for queued jobs every `poll`
+/// interval; runs each to completion (or failure), checkpointing per step.
+pub fn spawn(pool: PgPool, poll: Duration) {
+    tokio::spawn(async move {
+        loop {
+            // Recover jobs orphaned by a dead worker before claiming new ones.
+            if let Err(e) = cm_db::repo::topology_runs::requeue_stale(&pool, STALE_AFTER_SECS).await
+            {
+                eprintln!("topology_worker: requeue_stale failed: {e}");
+            }
+            match cm_db::repo::topology_runs::claim_next_queued(&pool).await {
+                Ok(Some(job)) => run_job(&pool, job).await,
+                Ok(None) => tokio::time::sleep(poll).await,
+                Err(e) => {
+                    eprintln!("topology_worker: claim failed: {e}");
+                    tokio::time::sleep(poll).await;
+                }
+            }
+        }
+    });
+}
+
+/// Drive one claimed job to a terminal state, persisting checkpoints as it goes.
+async fn run_job(pool: &PgPool, job: cm_db::repo::topology_runs::ClaimedTopologyRun) {
+    let id = job.id;
+
+    let Some(graph) = job
+        .graph
+        .as_ref()
+        .and_then(|g| serde_json::from_value::<TopologyGraph>(g.clone()).ok())
+    else {
+        let _ = cm_db::repo::topology_runs::fail(pool, id, "missing or invalid graph").await;
+        return;
+    };
+
+    // Resume from the last checkpoint, or start fresh.
+    let progress: RunProgress = job
+        .checkpoint
+        .and_then(|c| serde_json::from_value(c).ok())
+        .unwrap_or_default();
+
+    let executor = match ZeroClawDriveExecutor::from_env() {
+        Ok(e) => e,
+        Err(e) => {
+            let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;
+            return;
+        }
+    };
+
+    let pool_cb = pool.clone();
+    let result = execute_resumable(&graph, &job.task, &executor, progress, move |snap| {
+        let pool = pool_cb.clone();
+        async move {
+            // Best-effort checkpoint: a failed write just means we re-run the
+            // step on resume (idempotent — topology turns are pure reads here).
+            if let Ok(v) = serde_json::to_value(&snap) {
+                let _ =
+                    cm_db::repo::topology_runs::checkpoint(&pool, id, &v, snap.completed as i64)
+                        .await;
+            }
+            Ok(())
+        }
+    })
+    .await;
+
+    match result {
+        Ok(record) => {
+            let value = serde_json::to_value(&record).unwrap_or(serde_json::Value::Null);
+            if let Err(e) = cm_db::repo::topology_runs::complete(pool, id, &value).await {
+                eprintln!("topology_worker: complete({id}) failed: {e}");
+            }
+        }
+        Err(e) => {
+            let _ = cm_db::repo::topology_runs::fail(pool, id, &format!("{e}")).await;
+        }
+    }
+}

crates/cm-api/tests/topology_jobs.rs

@@ -0,0 +1,87 @@
+//! The durable topology-job lifecycle at the persistence layer: enqueue → claim
+//! (CAS) → checkpoint → complete, plus the stale-run resume sweep. This is the
+//! foundation that lets long-horizon topology runs survive worker restarts.
+
+use cm_db::repo::{topology_runs, workspaces};
+use cm_domain::{Workspace, WorkspaceId};
+use serde_json::json;
+use uuid::Uuid;
+
+async fn seed_workspace(pool: &sqlx::PgPool) -> WorkspaceId {
+    let ws = Workspace {
+        id: WorkspaceId::new(),
+        name: "Acme".into(),
+        plan: "team".into(),
+    };
+    workspaces::insert(pool, &ws).await.unwrap();
+    ws.id
+}
+
+#[tokio::test]
+async fn durable_run_lifecycle_enqueue_claim_checkpoint_complete() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+
+    let id = Uuid::now_v7();
+    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
+    topology_runs::enqueue_run(&pool, id, ws, "write a haiku", &graph)
+        .await
+        .unwrap();
+
+    // Status starts queued, no result yet.
+    let st = topology_runs::status(&pool, id, ws).await.unwrap();
+    assert_eq!(st.status, "queued");
+    assert_eq!(st.kind, "run");
+    assert!(st.result.is_none());
+
+    // Claim flips it to running and returns the job + its graph.
+    let claimed = topology_runs::claim_next_queued(&pool)
+        .await
+        .unwrap()
+        .expect("a queued job to claim");
+    assert_eq!(claimed.id, id);
+    assert!(claimed.graph.is_some());
+    assert!(claimed.checkpoint.is_none());
+
+    // A second claim finds nothing (the job is no longer queued).
+    assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_none());
+
+    // Checkpoint mid-run progress.
+    let progress = json!({"completed": 1, "outputs": ["draft"], "records": [], "totals": {}});
+    topology_runs::checkpoint(&pool, id, &progress, 1).await.unwrap();
+    let st = topology_runs::status(&pool, id, ws).await.unwrap();
+    assert_eq!(st.status, "running");
+    assert_eq!(st.last_event_id, 1);
+    assert!(st.checkpoint.is_some());
+
+    // Complete with a result blob.
+    let result = json!({"final_output": "a haiku", "totals": {"turns": 1}});
+    topology_runs::complete(&pool, id, &result).await.unwrap();
+    let st = topology_runs::status(&pool, id, ws).await.unwrap();
+    assert_eq!(st.status, "completed");
+    assert_eq!(st.result.unwrap()["final_output"], "a haiku");
+}
+
+#[tokio::test]
+async fn stale_running_jobs_are_requeued_for_resume() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+
+    let id = Uuid::now_v7();
+    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
+    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
+        .await
+        .unwrap();
+    // Claim it → running.
+    topology_runs::claim_next_queued(&pool).await.unwrap().unwrap();
+
+    // Not stale yet (just claimed) → sweep is a no-op.
+    assert_eq!(topology_runs::requeue_stale(&pool, 60.0).await.unwrap(), 0);
+
+    // With a zero threshold the running job counts as stale and is requeued,
+    // so the next claim picks it up again (resume from checkpoint).
+    assert_eq!(topology_runs::requeue_stale(&pool, 0.0).await.unwrap(), 1);
+    let st = topology_runs::status(&pool, id, ws).await.unwrap();
+    assert_eq!(st.status, "queued");
+    assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_some());
+}