Durable topology jobs (1/4): job-table schema + repo

Evolve topology_runs into a durable-job table (migration 0009): status state
machine (queued/running/completed/failed/cancelled), kind, input graph,
per-step checkpoint, error, last_event_id, timestamps. comparison becomes
nullable (the result blob, absent until completion). Back-compat: existing
rows default to completed/compare.

cm-db repo gains the durable-job ops: enqueue_run, claim_next_queued (CAS via
FOR UPDATE SKIP LOCKED), checkpoint, complete, fail, requeue_stale (resume
sweep), and status(). Regenerated .sqlx cache. Also fix two pre-existing test
RuntimeConfig literals missing the providers field (from the registry work).

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

.sqlx/query-23469c6fe735b664081f5773d755fe27c26ba86615a8ed53e43c8c2dfdf0abb7.json

@@ -0,0 +1,77 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, task, kind, status, error, checkpoint, comparison, last_event_id,\n                created_at, updated_at\n         FROM topology_runs WHERE id = $1 AND workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "task",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "kind",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "status",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "error",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 5,
+        "name": "checkpoint",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 6,
+        "name": "comparison",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 7,
+        "name": "last_event_id",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 8,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 9,
+        "name": "updated_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "23469c6fe735b664081f5773d755fe27c26ba86615a8ed53e43c8c2dfdf0abb7"
+}

.sqlx/query-5fcbd4d6adbf02489051e2fa63d1df670863bf90e55c1c0ac0ab011759cbd272.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE topology_runs\n         SET checkpoint = $2, last_event_id = $3, updated_at = now()\n         WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Jsonb",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5fcbd4d6adbf02489051e2fa63d1df670863bf90e55c1c0ac0ab011759cbd272"
+}

.sqlx/query-7298995b5b58aed46888bb9e5c8d331483aee162fc6bcf1e53232d2afc7c3e62.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE topology_runs\n         SET status = 'queued', updated_at = now()\n         WHERE status = 'running' AND updated_at < now() - make_interval(secs => $1)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Float8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "7298995b5b58aed46888bb9e5c8d331483aee162fc6bcf1e53232d2afc7c3e62"
+}

.sqlx/query-b1da590a84e9d2c84fe3e6683c7566a94be4496117c2458a21428b356e45e911.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE topology_runs\n         SET status = 'running', started_at = COALESCE(started_at, now()), updated_at = now()\n         WHERE id = (\n             SELECT id FROM topology_runs\n             WHERE status = 'queued'\n             ORDER BY created_at\n             FOR UPDATE SKIP LOCKED\n             LIMIT 1\n         )\n         RETURNING id, workspace_id, task, graph, checkpoint, last_event_id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 2,
+        "name": "task",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "graph",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 4,
+        "name": "checkpoint",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 5,
+        "name": "last_event_id",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      true,
+      false
+    ]
+  },
+  "hash": "b1da590a84e9d2c84fe3e6683c7566a94be4496117c2458a21428b356e45e911"
+}

.sqlx/query-b4f2ebe304fa3b4722584bfe6c3df389b46d52724fc9b137dbe83b3d4fce2983.json

@@ -33,7 +33,7 @@
     "nullable": [
       false,
       false,
-      false,
+      true,
       false
     ]
   },

.sqlx/query-c2ea3efe8d13800dc95fc21984a770281dce7d518cdc05941736e3bf628c4876.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT id, task, created_at FROM topology_runs\n         WHERE workspace_id = $1 ORDER BY created_at DESC LIMIT $2",
+  "query": "SELECT id, task, status, kind, created_at FROM topology_runs\n         WHERE workspace_id = $1 ORDER BY created_at DESC LIMIT $2",
   "describe": {
     "columns": [
       {
@@ -15,6 +15,16 @@
       },
       {
         "ordinal": 2,
+        "name": "status",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "kind",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
         "name": "created_at",
         "type_info": "Timestamptz"
       }
@@ -26,10 +36,12 @@
       ]
     },
     "nullable": [
+      false,
+      false,
       false,
       false,
       false
     ]
   },
-  "hash": "8675cedae691e400bc935dde9b05b1ee71fe7494003dbd36c5a801c77c3d97ac"
+  "hash": "c2ea3efe8d13800dc95fc21984a770281dce7d518cdc05941736e3bf628c4876"
 }

.sqlx/query-d7b8f1004dc027b02001c0927f1695a3d8a423ae9243b2a91fc27d8887630238.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE topology_runs\n         SET status = 'failed', error = $2, finished_at = now(), updated_at = now()\n         WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d7b8f1004dc027b02001c0927f1695a3d8a423ae9243b2a91fc27d8887630238"
+}

.sqlx/query-ddde1145d4952d2c9093009ee571679a0bb65949cec547438f8757af8419daa7.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE topology_runs\n         SET status = 'completed', comparison = $2, finished_at = now(), updated_at = now()\n         WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ddde1145d4952d2c9093009ee571679a0bb65949cec547438f8757af8419daa7"
+}

.sqlx/query-fafbe27a2854e8884efec77567bcf258da8d9c1b825b8e8dcfdc2c90efea31b1.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO topology_runs (id, workspace_id, task, kind, status, graph)\n         VALUES ($1, $2, $3, 'run', 'queued', $4)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Uuid",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "fafbe27a2854e8884efec77567bcf258da8d9c1b825b8e8dcfdc2c90efea31b1"
+}

crates/cm-api/tests/slack_inbound.rs

@@ -69,6 +69,7 @@ async fn mention_round_trip_verifies_runs_and_gates_the_reply() {
             browser: None,
             broker_socket: Some(socket.clone()),
             slack_base_url: "http://127.0.0.1:1".into(), // never reached here
+            providers: Default::default(),
         },
     );
     let app = cm_api::router(AppState::new(pool.clone(), runtime).with_broker(socket.clone()));

crates/cm-db/src/repo/topology_runs.rs

@@ -1,4 +1,6 @@
-//! Persistence for saved multi-topology comparison runs.
+//! Persistence for topology runs — both saved "compare" results and durable
+//! single-topology "run" jobs (queued → running → completed/failed), the latter
+//! checkpointed per step for crash-resumable long-horizon execution.
 
 use cm_domain::WorkspaceId;
 use serde_json::Value;
@@ -12,6 +14,8 @@ use crate::DbError;
 pub struct TopologyRunSummary {
     pub id: Uuid,
     pub task: String,
+    pub status: String,
+    pub kind: String,
     pub created_at: OffsetDateTime,
 }
 
@@ -23,7 +27,35 @@ pub struct TopologyRun {
     pub created_at: OffsetDateTime,
 }
 
-/// Save a comparison run for a workspace.
+/// A durable topology job claimed by the background worker (queued or resumed).
+pub struct ClaimedTopologyRun {
+    pub id: Uuid,
+    pub workspace_id: Uuid,
+    pub task: String,
+    /// Input graph (present for `kind = 'run'` jobs).
+    pub graph: Option<Value>,
+    /// Completed-step outputs persisted by the last checkpoint, for resume.
+    pub checkpoint: Option<Value>,
+    /// Event-journal offset reached so far.
+    pub last_event_id: i64,
+}
+
+/// Lifecycle status + progress for a durable run (status endpoint).
+pub struct TopologyRunStatus {
+    pub id: Uuid,
+    pub task: String,
+    pub kind: String,
+    pub status: String,
+    pub error: Option<String>,
+    pub checkpoint: Option<Value>,
+    /// The final result blob (the `RunRecord`/`Comparison`) once completed.
+    pub result: Option<Value>,
+    pub last_event_id: i64,
+    pub created_at: OffsetDateTime,
+    pub updated_at: OffsetDateTime,
+}
+
+/// Save a finished comparison run for a workspace (synchronous path).
 pub async fn insert(
     pool: &PgPool,
     id: Uuid,
@@ -31,6 +63,7 @@ pub async fn insert(
     task: &str,
     comparison: &Value,
 ) -> Result<(), DbError> {
+    // status/kind default to 'completed'/'compare' for the synchronous path.
     sqlx::query!(
         "INSERT INTO topology_runs (id, workspace_id, task, comparison)
          VALUES ($1, $2, $3, $4)",
@@ -44,6 +77,120 @@ pub async fn insert(
     Ok(())
 }
 
+/// Enqueue a durable single-topology run job (`kind = 'run'`, `status = 'queued'`).
+/// The background worker claims and executes it; the result lands in `comparison`.
+pub async fn enqueue_run(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: WorkspaceId,
+    task: &str,
+    graph: &Value,
+) -> Result<(), DbError> {
+    sqlx::query!(
+        "INSERT INTO topology_runs (id, workspace_id, task, kind, status, graph)
+         VALUES ($1, $2, $3, 'run', 'queued', $4)",
+        id,
+        workspace_id.as_uuid(),
+        task,
+        graph,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Atomically claim the oldest queued job, flipping it to `running`. Uses
+/// `FOR UPDATE SKIP LOCKED` so multiple workers never claim the same job.
+/// Returns `None` when the queue is empty.
+pub async fn claim_next_queued(pool: &PgPool) -> Result<Option<ClaimedTopologyRun>, DbError> {
+    let row = sqlx::query!(
+        "UPDATE topology_runs
+         SET status = 'running', started_at = COALESCE(started_at, now()), updated_at = now()
+         WHERE id = (
+             SELECT id FROM topology_runs
+             WHERE status = 'queued'
+             ORDER BY created_at
+             FOR UPDATE SKIP LOCKED
+             LIMIT 1
+         )
+         RETURNING id, workspace_id, task, graph, checkpoint, last_event_id",
+    )
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|r| ClaimedTopologyRun {
+        id: r.id,
+        workspace_id: r.workspace_id,
+        task: r.task,
+        graph: r.graph,
+        checkpoint: r.checkpoint,
+        last_event_id: r.last_event_id,
+    }))
+}
+
+/// Persist mid-run progress: the completed-step checkpoint + journal offset.
+/// Touches `updated_at` so the stale-run sweeper treats the job as alive.
+pub async fn checkpoint(
+    pool: &PgPool,
+    id: Uuid,
+    checkpoint: &Value,
+    last_event_id: i64,
+) -> Result<(), DbError> {
+    sqlx::query!(
+        "UPDATE topology_runs
+         SET checkpoint = $2, last_event_id = $3, updated_at = now()
+         WHERE id = $1",
+        id,
+        checkpoint,
+        last_event_id,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Mark a job completed and store its final result blob.
+pub async fn complete(pool: &PgPool, id: Uuid, result: &Value) -> Result<(), DbError> {
+    sqlx::query!(
+        "UPDATE topology_runs
+         SET status = 'completed', comparison = $2, finished_at = now(), updated_at = now()
+         WHERE id = $1",
+        id,
+        result,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Mark a job failed with an error message.
+pub async fn fail(pool: &PgPool, id: Uuid, error: &str) -> Result<(), DbError> {
+    sqlx::query!(
+        "UPDATE topology_runs
+         SET status = 'failed', error = $2, finished_at = now(), updated_at = now()
+         WHERE id = $1",
+        id,
+        error,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Resume sweep: requeue `running` jobs whose worker went silent (no checkpoint
+/// touch within `older_than_secs`). The next claim resumes them from checkpoint.
+/// Returns how many were requeued.
+pub async fn requeue_stale(pool: &PgPool, older_than_secs: f64) -> Result<u64, DbError> {
+    let result = sqlx::query!(
+        "UPDATE topology_runs
+         SET status = 'queued', updated_at = now()
+         WHERE status = 'running' AND updated_at < now() - make_interval(secs => $1)",
+        older_than_secs,
+    )
+    .execute(pool)
+    .await?;
+    Ok(result.rows_affected())
+}
+
 /// The most recent runs for a workspace, newest first.
 pub async fn list_recent(
     pool: &PgPool,
@@ -51,7 +198,7 @@ pub async fn list_recent(
     limit: i64,
 ) -> Result<Vec<TopologyRunSummary>, DbError> {
     let rows = sqlx::query!(
-        "SELECT id, task, created_at FROM topology_runs
+        "SELECT id, task, status, kind, created_at FROM topology_runs
          WHERE workspace_id = $1 ORDER BY created_at DESC LIMIT $2",
         workspace_id.as_uuid(),
         limit,
@@ -63,12 +210,14 @@ pub async fn list_recent(
         .map(|r| TopologyRunSummary {
             id: r.id,
             task: r.task,
+            status: r.status,
+            kind: r.kind,
             created_at: r.created_at,
         })
         .collect())
 }
 
-/// A single saved run, scoped to its workspace.
+/// A single saved comparison/run result, scoped to its workspace.
 pub async fn get(
     pool: &PgPool,
     id: Uuid,
@@ -86,7 +235,38 @@ pub async fn get(
     Ok(TopologyRun {
         id: row.id,
         task: row.task,
-        comparison: row.comparison,
+        // Nullable since the durable path: a queued/running run has no result yet.
+        comparison: row.comparison.unwrap_or(Value::Null),
         created_at: row.created_at,
     })
 }
+
+/// Lifecycle status + progress for a durable run (status endpoint), workspace-scoped.
+pub async fn status(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: WorkspaceId,
+) -> Result<TopologyRunStatus, DbError> {
+    let row = sqlx::query!(
+        "SELECT id, task, kind, status, error, checkpoint, comparison, last_event_id,
+                created_at, updated_at
+         FROM topology_runs WHERE id = $1 AND workspace_id = $2",
+        id,
+        workspace_id.as_uuid(),
+    )
+    .fetch_optional(pool)
+    .await?
+    .ok_or(DbError::NotFound)?;
+    Ok(TopologyRunStatus {
+        id: row.id,
+        task: row.task,
+        kind: row.kind,
+        status: row.status,
+        error: row.error,
+        checkpoint: row.checkpoint,
+        result: row.comparison,
+        last_event_id: row.last_event_id,
+        created_at: row.created_at,
+        updated_at: row.updated_at,
+    })
+}

crates/cm-runtime/tests/slack_tool.rs

@@ -144,6 +144,7 @@ async fn slack_post_blocks_then_the_broker_executes_exactly_once() {
             browser: None,
             broker_socket: Some(socket),
             slack_base_url: sink_url,
+            providers: Default::default(),
         },
     );
     let session = cm_db::repo::sessions::create(&pool, agent.id, ws.id, "Slack")

migrations/0009_durable_topology_runs.sql

@@ -0,0 +1,26 @@
+-- Evolve topology_runs from a write-once result store into a DURABLE JOB table.
+-- A topology run becomes a first-class job with a lifecycle (queued -> running ->
+-- completed/failed/cancelled), a checkpoint of completed-step outputs (for
+-- crash-resumable long-horizon execution), and an event-journal offset. This
+-- reuses the agent-run lifecycle pattern (RunState + checkpoint CAS + run_events
+-- journal) without coupling to the chat-session FK that agent_runs requires.
+--
+-- Back-compat: existing rows are finished "compare" results, so the new columns
+-- default to status='completed', kind='compare'. The result blob keeps living in
+-- `comparison`, which becomes nullable (a queued/running run has no result yet).
+
+ALTER TABLE topology_runs
+    ALTER COLUMN comparison DROP NOT NULL,
+    ADD COLUMN kind          TEXT NOT NULL DEFAULT 'compare',
+    ADD COLUMN status        TEXT NOT NULL DEFAULT 'completed'
+        CHECK (status IN ('queued', 'running', 'completed', 'failed', 'cancelled')),
+    ADD COLUMN graph         JSONB,
+    ADD COLUMN checkpoint    JSONB,
+    ADD COLUMN error         TEXT,
+    ADD COLUMN last_event_id BIGINT NOT NULL DEFAULT 0,
+    ADD COLUMN started_at    TIMESTAMPTZ,
+    ADD COLUMN finished_at   TIMESTAMPTZ,
+    ADD COLUMN updated_at    TIMESTAMPTZ NOT NULL DEFAULT now();
+
+-- The background worker claims the oldest queued job; index that scan.
+CREATE INDEX topology_runs_status_idx ON topology_runs (status, created_at);