loops: backend routes + repo + cron scheduler + HMAC webhook

Third commit of the Research + Loops arc. Lights up loops as durable
recurring topology executions:

  GET    /api/loops                 list workspace's loops
  POST   /api/loops                 create — returns webhook_token +
                                    signing_key ONCE when webhook trigger
                                    is enabled; never exposed again
  GET    /api/loops/:id             detail
  PATCH  /api/loops/:id             update definition
  DELETE /api/loops/:id             delete
  POST   /api/loops/:id/run         trigger one iteration NOW
  POST   /api/loops/:id/enable      set enabled=true
  POST   /api/loops/:id/disable     set enabled=false
  POST   /webhooks/loops/:token     public; HMAC-SHA256-verified

Scheduler (cm_runtime::spawn_loop_scheduler) wakes every 10s, queries the
partial index on (next_fire_at) for due loops, enqueues one topology_runs
row per fire with loop_id + iteration + parent_run_id chained back to the
previous iteration. Uses croner via the existing scheduling::next_occurrence
helper. Missed windows fire ONCE and skip the backlog — next_fire_at is
always computed strictly AFTER now(), so a late scheduler doesn't drain a
buildup.

Webhook signatures follow the same pattern as the Stripe billing webhook
(HMAC-SHA256 with constant-time hex compare). Token + signing key are
24-byte OS-RNG values; the URL uses base64-url for the token, and the
signing key is base64-std. Both surface exactly once at create time.

All three fire paths (scheduler, immediate-run, webhook) funnel through
`cm_db::repo::loops::enqueue_iteration` so the invariants stay in one
place. `iters` repeat policy is enforced by the scheduler tick; `until`
and `on_completion` land with the orchestrator hook in commit 4.

Adds cm-llm as a direct cm-api dep, getrandom for the webhook material
generator, and wires the scheduler spawn into the server binary alongside
the resume sweeper and outbox drainer.

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

@@ -0,0 +1,313 @@
+//! Loops — durable recurring topology executions. The row holds the
+//! definition (graph + task_template + triggers + repeat_policy) and a small
+//! amount of scheduler state (enabled, next_fire_at, last_run_id).
+//! Each fire produces a normal `topology_runs` row with loop_id + iteration
+//! + parent_run_id set, so the run driver picks it up like any other job.
+//!
+//! See 0031 migration header for the state semantics and missed-window rule.
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sqlx::PgPool;
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Loop {
+    pub id: Uuid,
+    pub workspace_id: Uuid,
+    pub title: String,
+    pub description: String,
+    pub graph: Value,
+    pub task_template: String,
+    pub triggers: Value,
+    pub repeat_policy: Value,
+    pub enabled: bool,
+    pub next_fire_at: Option<OffsetDateTime>,
+    pub last_run_id: Option<Uuid>,
+    pub webhook_token: Option<String>,
+    pub webhook_signing_key: Option<String>,
+    pub created_by: Uuid,
+    pub created_at: OffsetDateTime,
+    pub updated_at: OffsetDateTime,
+}
+
+/// Minimal fields the scheduler needs when it wakes up.
+#[derive(Debug, Clone)]
+pub struct DueLoop {
+    pub id: Uuid,
+    pub workspace_id: Uuid,
+    pub graph: Value,
+    pub task_template: String,
+    pub triggers: Value,
+    pub repeat_policy: Value,
+    pub last_run_id: Option<Uuid>,
+}
+
+pub struct NewLoop<'a> {
+    pub workspace_id: Uuid,
+    pub title: &'a str,
+    pub description: &'a str,
+    pub graph: &'a Value,
+    pub task_template: &'a str,
+    pub triggers: &'a Value,
+    pub repeat_policy: &'a Value,
+    pub enabled: bool,
+    pub next_fire_at: Option<OffsetDateTime>,
+    pub webhook_token: Option<&'a str>,
+    pub webhook_signing_key: Option<&'a str>,
+    pub created_by: Uuid,
+}
+
+pub async fn create(pool: &PgPool, input: NewLoop<'_>) -> Result<Uuid, DbError> {
+    let id = Uuid::now_v7();
+    sqlx::query!(
+        "INSERT INTO loops
+            (id, workspace_id, title, description, graph, task_template,
+             triggers, repeat_policy, enabled, next_fire_at,
+             webhook_token, webhook_signing_key, created_by)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
+        id,
+        input.workspace_id,
+        input.title,
+        input.description,
+        input.graph,
+        input.task_template,
+        input.triggers,
+        input.repeat_policy,
+        input.enabled,
+        input.next_fire_at,
+        input.webhook_token,
+        input.webhook_signing_key,
+        input.created_by,
+    )
+    .execute(pool)
+    .await?;
+    Ok(id)
+}
+
+pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<Loop>, DbError> {
+    let rows = sqlx::query_as!(
+        Loop,
+        "SELECT id, workspace_id, title, description, graph, task_template,
+                triggers, repeat_policy, enabled, next_fire_at, last_run_id,
+                webhook_token, webhook_signing_key, created_by, created_at, updated_at
+           FROM loops
+          WHERE workspace_id = $1
+          ORDER BY updated_at DESC",
+        workspace_id,
+    )
+    .fetch_all(pool)
+    .await?;
+    Ok(rows)
+}
+
+pub async fn get(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<Option<Loop>, DbError> {
+    let row = sqlx::query_as!(
+        Loop,
+        "SELECT id, workspace_id, title, description, graph, task_template,
+                triggers, repeat_policy, enabled, next_fire_at, last_run_id,
+                webhook_token, webhook_signing_key, created_by, created_at, updated_at
+           FROM loops
+          WHERE id = $1 AND workspace_id = $2",
+        id,
+        workspace_id,
+    )
+    .fetch_optional(pool)
+    .await?;
+    Ok(row)
+}
+
+/// Look up a loop by its webhook token — used only by the webhook receiver,
+/// which has no session context. Returns the minimal shape needed to enqueue
+/// an iteration and verify the HMAC signature.
+pub async fn get_by_webhook_token(
+    pool: &PgPool,
+    token: &str,
+) -> Result<Option<(Uuid, Uuid, String, DueLoop)>, DbError> {
+    let row = sqlx::query!(
+        "SELECT id, workspace_id, graph, task_template, triggers, repeat_policy,
+                last_run_id, webhook_signing_key
+           FROM loops
+          WHERE webhook_token = $1 AND enabled",
+        token,
+    )
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.and_then(|r| {
+        let key = r.webhook_signing_key?;
+        Some((
+            r.id,
+            r.workspace_id,
+            key,
+            DueLoop {
+                id: r.id,
+                workspace_id: r.workspace_id,
+                graph: r.graph,
+                task_template: r.task_template,
+                triggers: r.triggers,
+                repeat_policy: r.repeat_policy,
+                last_run_id: r.last_run_id,
+            },
+        ))
+    }))
+}
+
+pub struct UpdateLoop<'a> {
+    pub title: &'a str,
+    pub description: &'a str,
+    pub graph: &'a Value,
+    pub task_template: &'a str,
+    pub triggers: &'a Value,
+    pub repeat_policy: &'a Value,
+    pub next_fire_at: Option<OffsetDateTime>,
+}
+
+pub async fn update(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: Uuid,
+    input: UpdateLoop<'_>,
+) -> Result<(), DbError> {
+    sqlx::query!(
+        "UPDATE loops
+            SET title = $3, description = $4, graph = $5, task_template = $6,
+                triggers = $7, repeat_policy = $8, next_fire_at = $9,
+                updated_at = now()
+          WHERE id = $1 AND workspace_id = $2",
+        id,
+        workspace_id,
+        input.title,
+        input.description,
+        input.graph,
+        input.task_template,
+        input.triggers,
+        input.repeat_policy,
+        input.next_fire_at,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn set_enabled(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: Uuid,
+    enabled: bool,
+) -> Result<(), DbError> {
+    sqlx::query!(
+        "UPDATE loops SET enabled = $3, updated_at = now()
+          WHERE id = $1 AND workspace_id = $2",
+        id,
+        workspace_id,
+        enabled,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<(), DbError> {
+    sqlx::query!(
+        "DELETE FROM loops WHERE id = $1 AND workspace_id = $2",
+        id,
+        workspace_id,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Loops the scheduler tick should fire NOW. Only reads what the enqueue
+/// path needs, so the tick stays cheap even when the workspace has hundreds
+/// of loops.
+pub async fn due(pool: &PgPool) -> Result<Vec<DueLoop>, DbError> {
+    let rows = sqlx::query!(
+        "SELECT id, workspace_id, graph, task_template, triggers, repeat_policy,
+                last_run_id
+           FROM loops
+          WHERE enabled AND next_fire_at IS NOT NULL AND next_fire_at <= now()",
+    )
+    .fetch_all(pool)
+    .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| DueLoop {
+            id: r.id,
+            workspace_id: r.workspace_id,
+            graph: r.graph,
+            task_template: r.task_template,
+            triggers: r.triggers,
+            repeat_policy: r.repeat_policy,
+            last_run_id: r.last_run_id,
+        })
+        .collect())
+}
+
+/// Post-fire bookkeeping: bump last_run_id + advance next_fire_at (NULL when
+/// the loop has no cron trigger). Called by the scheduler after a successful
+/// enqueue_iteration.
+pub async fn mark_fired(
+    pool: &PgPool,
+    id: Uuid,
+    run_id: Uuid,
+    next_fire_at: Option<OffsetDateTime>,
+) -> Result<(), DbError> {
+    sqlx::query!(
+        "UPDATE loops
+            SET last_run_id = $2, next_fire_at = $3, updated_at = now()
+          WHERE id = $1",
+        id,
+        run_id,
+        next_fire_at,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Next iteration number for a loop (1 if it has never fired).
+pub async fn next_iteration(pool: &PgPool, loop_id: Uuid) -> Result<i32, DbError> {
+    let n: Option<i32> = sqlx::query_scalar!(
+        "SELECT MAX(iteration) FROM topology_runs WHERE loop_id = $1",
+        loop_id,
+    )
+    .fetch_one(pool)
+    .await?;
+    Ok(n.unwrap_or(0) + 1)
+}
+
+/// Enqueue an iteration as a normal `topology_runs` row. The scheduler,
+/// on-completion hook, and webhook receiver all funnel through here so the
+/// invariants (loop_id + iteration + parent_run_id all set together) stay
+/// in one place.
+pub async fn enqueue_iteration(
+    pool: &PgPool,
+    loop_id: Uuid,
+    workspace_id: Uuid,
+    task: &str,
+    graph: &Value,
+    iteration: i32,
+    parent_run_id: Option<Uuid>,
+) -> Result<Uuid, DbError> {
+    let run_id = Uuid::now_v7();
+    sqlx::query!(
+        "INSERT INTO topology_runs
+            (id, workspace_id, task, kind, status, graph, tier,
+             loop_id, iteration, parent_run_id)
+         VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7)",
+        run_id,
+        workspace_id,
+        task,
+        graph,
+        loop_id,
+        iteration,
+        parent_run_id,
+    )
+    .execute(pool)
+    .await?;
+    Ok(run_id)
+}

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

@@ -8,6 +8,7 @@ pub mod credits;
 pub mod files;
 pub mod fleet_beszel;
 pub mod fleet_tailscale;
+pub mod loops;
 pub mod messages;
 pub mod node_metrics;
 pub mod node_rules;