loops: kind column + initial_burst + on_artifact_update trigger fan-out

Foundation for folding research into loops as a first-class kind.
This commit ships the plumbing; the research-kind dispatch itself
lands next. Behavior for existing exec-kind loops is unchanged unless
they opt into the new trigger fields.

Migration 0044:
- kind TEXT NOT NULL DEFAULT 'exec' CHECK ('exec' | 'research'). New
  research-kind will run the research pipeline each iteration (next
  commit); 'exec' preserves today's behavior.
- initial_burst_remaining INT NOT NULL DEFAULT 0 — countdown for the
  triggers.initial_burst quota. Decremented CAS-safely on each
  completion until it hits 0.
- Two partial indexes: (kind, source_research_topic_id) for kind-
  aware lookups, and (source_research_topic_id) filtered on
  on_artifact_update=true + enabled=true for the fan-out hook.

Trigger schema extended with two optional fields:
- initial_burst: N — fire N iterations back-to-back at create time.
  create_loop enqueues the first iteration inline (subject to
  empty-roster gate), sets remaining=N-1, and the completion hook
  continues the chain until exhausted.
- on_artifact_update: true — when a bound research_outcomes row is
  inserted for the source topic, wake up one iteration of this loop.
  Coalesced against has_active_run so a burst of rapid revisions
  doesn't queue duplicates.

Backend:
- cm_db::repo::loops helpers (all dynamic sqlx, no offline cache
  regen needed):
  - set_initial_burst_remaining
  - take_initial_burst_slot (CAS UPDATE returning prev value; 0 on
    exhausted or race loss)
  - loops_awaiting_topic (fan-out query: kind=exec + enabled +
    on_artifact_update=true bound to the given topic)
  - has_active_run (queued|running iteration existence check)
  - get_any_workspace (bypasses the workspace scope guard; used by
    the completion hook where the run row is authoritative)
- routes/loops::compose_iteration_task made pub so the completion
  hook can build the same enriched task string as run_now.
- topology_worker::freeze_research_outcome now fans out to awakened
  loops after the outcome insert, using compose_iteration_task and
  coalescing on has_active_run.
- topology_worker::continue_initial_burst runs on every completion:
  · take_initial_burst_slot (CAS) — no-op if already exhausted
  · has_active_run coalesce guard
  · re-fetches the loop via get_any_workspace + compose_iteration_task
  · enqueues via loops::enqueue_iteration with parent_run_id set

Follow-ups already queued:
- kind='research' dispatch in run_job — build the research
  coordinator task from the topic config, run the research pipeline
  each iteration. Requires factoring start_topic's task-build.
- ResearchWizard "When should this run?" step (Just once / Nightly /
  Manual) creating the topic + paired research-kind loop.
- LoopsWizard trigger UI matching the design proposal (burst count,
  cron, on-artifact checkbox).

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

@@ -55,7 +55,7 @@ fn loop_state_root() -> std::path::PathBuf {
 ///     <task_template>
 ///   The topology_worker's completion hook (P3) parses the COMPLETED
 ///   marker to update `consumed_int_ids`.
-async fn compose_iteration_task(pool: &PgPool, loop_id: Uuid, task_template: &str) -> String {
+pub async fn compose_iteration_task(pool: &PgPool, loop_id: Uuid, task_template: &str) -> String {
     let ctx = cm_db::repo::loops::source_research_context(pool, loop_id)
         .await
         .unwrap_or(None);
@@ -183,6 +183,24 @@ struct Triggers {
     on_completion: bool,
     #[serde(default)]
     webhook_enabled: bool,
+    /// NEW — number of iterations to fire back-to-back at loop-create
+    /// time. Enqueue path: fire once immediately, then chain each
+    /// subsequent one via on_completion until the burst quota is
+    /// exhausted (tracked in run metadata). Defaults to 0 for existing
+    /// loops (no auto-fire); new wizards typically set 1 (D1: every
+    /// runnable thing runs at least once).
+    #[serde(default)]
+    initial_burst: u32,
+    /// NEW — when this loop is bound to a source_research_topic and
+    /// that topic gets a fresh research_outcomes row (via
+    /// freeze_research_outcome), enqueue one iteration on this loop.
+    /// Coalesced with any in-flight run (D3: coordinator resolves;
+    /// no race, just one wake per artifact update). Read directly
+    /// from the loops.triggers jsonb by loops_awaiting_topic — no
+    /// need for the Rust parser to hold it after the fact.
+    #[serde(default)]
+    #[allow(dead_code)]
+    on_artifact_update: bool,
 }
 
 fn make_webhook_material() -> (String, String) {
@@ -276,6 +294,52 @@ pub async fn create_loop(
         }
     }
 
+    // initial_burst — fire the first iteration inline, so every loop
+    // runs at least once when the trigger requests it (D1). The burst
+    // continues via the on_completion trigger up to `initial_burst`
+    // (tracked in loops.initial_burst_remaining, decremented on each
+    // completion). Best-effort: a docker/DB failure on the FIRST fire
+    // logs but the loop still lives with next_fire_at set for its
+    // cron.
+    let parsed = parse_triggers(&body.triggers);
+    let initial_burst = parsed.as_ref().map(|t| t.initial_burst).unwrap_or(0);
+    let chain_on_completion = parsed.as_ref().map(|t| t.on_completion).unwrap_or(false);
+    if initial_burst > 0 {
+        ensure_loop_container(&state.pool, user.workspace_id.as_uuid(), id).await;
+        let iter = cm_db::repo::loops::next_iteration(&state.pool, id)
+            .await
+            .unwrap_or(0);
+        let task = compose_iteration_task(&state.pool, id, body.task_template.trim()).await;
+        // Best-effort — we don't want to abort the loop-create response
+        // just because Docker didn't answer the daemon health check.
+        match cm_db::repo::loops::enqueue_iteration(
+            &state.pool,
+            id,
+            user.workspace_id.as_uuid(),
+            &task,
+            &body.graph,
+            iter,
+            None,
+        )
+        .await
+        {
+            Ok(run_id) => {
+                let _ = cm_db::repo::loops::mark_fired(&state.pool, id, run_id, next_fire_at).await;
+                // Log the burst remaining so on_completion can advance it.
+                // Stored on loops.initial_burst_remaining (added by the
+                // same migration as loops.kind). Falls back silently on
+                // schema absence so we don't break dev DBs mid-rollout.
+                let remaining = initial_burst.saturating_sub(1) as i32;
+                if remaining > 0 || chain_on_completion {
+                    let _ =
+                        cm_db::repo::loops::set_initial_burst_remaining(&state.pool, id, remaining)
+                            .await;
+                }
+            }
+            Err(e) => eprintln!("loops::create: initial_burst enqueue failed: {e:?}"),
+        }
+    }
+
     Ok((
         StatusCode::CREATED,
         Json(LoopCreated {

crates/cm-api/src/topology_worker.rs

@@ -171,6 +171,7 @@ async fn run_job(
             }
             freeze_research_outcome(pool, id, &record.final_output).await;
             advance_loop_after_completion(pool, id, &record.final_output).await;
+            continue_initial_burst(pool, id).await;
         }
         Err(e) => {
             // Don't clobber a cancellation (or any already-terminal state) with `failed`.
@@ -207,6 +208,94 @@ async fn freeze_research_outcome(pool: &PgPool, run_id: Uuid, final_output: &str
         cm_db::repo::research_outcomes::insert(pool, topic_id, final_output, Some(run_id)).await
     {
         eprintln!("topology_worker: research_outcomes::insert({run_id}) failed: {e}");
+        return;
+    }
+    // Fan-out: any exec-kind loop bound to this topic with the
+    // on_artifact_update trigger enabled wakes up now. Coalesced —
+    // if a loop already has a queued/running run we skip (D3 fallback:
+    // the coordinator sees the fresh artifact on its next iteration
+    // anyway). Best-effort per loop; one loop's Docker/DB hiccup
+    // doesn't affect the others.
+    let awakened = cm_db::repo::loops::loops_awaiting_topic(pool, topic_id)
+        .await
+        .unwrap_or_default();
+    for (loop_id, workspace_id, task_template, graph) in awakened {
+        if cm_db::repo::loops::has_active_run(pool, loop_id)
+            .await
+            .unwrap_or(false)
+        {
+            continue; // Coalesce.
+        }
+        let iter = cm_db::repo::loops::next_iteration(pool, loop_id)
+            .await
+            .unwrap_or(0);
+        // Build the enriched task with the freshly-inserted artifact
+        // (compose_iteration_task reads latest, which is what we just
+        // wrote).
+        let task =
+            crate::routes::loops::compose_iteration_task(pool, loop_id, &task_template).await;
+        if let Err(e) = cm_db::repo::loops::enqueue_iteration(
+            pool,
+            loop_id,
+            workspace_id,
+            &task,
+            &graph,
+            iter,
+            Some(run_id),
+        )
+        .await
+        {
+            eprintln!("topology_worker: on_artifact_update enqueue({loop_id}) failed: {e}");
+        }
+    }
+}
+
+/// If the just-completed run was a loop iteration with
+/// initial_burst_remaining > 0, enqueue the next iteration and
+/// decrement the counter (CAS-safe via take_initial_burst_slot).
+/// No-op for non-loop runs and for loops whose burst is exhausted.
+async fn continue_initial_burst(pool: &PgPool, run_id: Uuid) {
+    let loop_id = match cm_db::repo::topology_runs::loop_id_for_run(pool, run_id).await {
+        Ok(Some(id)) => id,
+        _ => return,
+    };
+    // If another worker races us, only ONE gets the slot; the other
+    // sees 0 (no-op).
+    let prev = cm_db::repo::loops::take_initial_burst_slot(pool, loop_id)
+        .await
+        .unwrap_or(0);
+    if prev == 0 {
+        return;
+    }
+    // Coalesce with a concurrently-in-flight iteration (a webhook
+    // arriving during a burst, say).
+    if cm_db::repo::loops::has_active_run(pool, loop_id)
+        .await
+        .unwrap_or(false)
+    {
+        return;
+    }
+    // Fetch the loop so we have the workspace + graph + task_template
+    // to enqueue the next iteration.
+    let Ok(Some(l)) = cm_db::repo::loops::get_any_workspace(pool, loop_id).await else {
+        return;
+    };
+    let iter = cm_db::repo::loops::next_iteration(pool, loop_id)
+        .await
+        .unwrap_or(0);
+    let task = crate::routes::loops::compose_iteration_task(pool, loop_id, &l.task_template).await;
+    if let Err(e) = cm_db::repo::loops::enqueue_iteration(
+        pool,
+        loop_id,
+        l.workspace_id,
+        &task,
+        &l.graph,
+        iter,
+        Some(run_id),
+    )
+    .await
+    {
+        eprintln!("topology_worker: continue_initial_burst enqueue failed: {e}");
     }
 }
 

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

@@ -104,6 +104,44 @@ pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<Loop>, DbErro
     Ok(rows)
 }
 
+/// Cross-workspace fetch used by internal callers (topology_worker
+/// completion hooks) where the run row is authoritative for the
+/// workspace binding — no need for a second scoping check. Returns
+/// None if the loop was deleted between run enqueue and completion.
+pub async fn get_any_workspace(pool: &PgPool, id: Uuid) -> Result<Option<Loop>, DbError> {
+    // Dynamic query so this callsite doesn't require an offline sqlx
+    // cache regen — used from the completion hook, not on the hot path.
+    use sqlx::Row;
+    let row: Option<sqlx::postgres::PgRow> = sqlx::query(
+        "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",
+    )
+    .bind(id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|r| Loop {
+        id: r.get("id"),
+        workspace_id: r.get("workspace_id"),
+        title: r.get("title"),
+        description: r.get("description"),
+        graph: r.get("graph"),
+        task_template: r.get("task_template"),
+        triggers: r.get("triggers"),
+        repeat_policy: r.get("repeat_policy"),
+        enabled: r.get("enabled"),
+        next_fire_at: r.try_get("next_fire_at").ok().flatten(),
+        last_run_id: r.try_get("last_run_id").ok().flatten(),
+        webhook_token: r.try_get("webhook_token").ok().flatten(),
+        webhook_signing_key: r.try_get("webhook_signing_key").ok().flatten(),
+        created_by: r.get("created_by"),
+        created_at: r.get("created_at"),
+        updated_at: r.get("updated_at"),
+    }))
+}
+
 pub async fn get(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<Option<Loop>, DbError> {
     let row = sqlx::query_as!(
         Loop,
@@ -254,6 +292,96 @@ pub async fn set_zeroclaw_container(
 /// downstream mini-timeline can show WHEN the plan was adjusted and
 /// WHY. Idempotent: appending a duplicate text/run_id combo is allowed
 /// (rare — indicates the parser matched twice on the same line).
+/// Set the countdown for the triggers.initial_burst quota. On
+/// create_loop we set this to `burst - 1` after firing the first
+/// iteration inline; on each subsequent completion we decrement and,
+/// while it's > 0, enqueue another iteration. Dynamic query so the
+/// new column doesn't need an offline sqlx cache regen.
+pub async fn set_initial_burst_remaining(
+    pool: &PgPool,
+    loop_id: Uuid,
+    remaining: i32,
+) -> Result<(), DbError> {
+    sqlx::query("UPDATE loops SET initial_burst_remaining = $2 WHERE id = $1")
+        .bind(loop_id)
+        .bind(remaining)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+/// Atomically decrement initial_burst_remaining, returning the value
+/// BEFORE decrement. Zero is a no-op (returns 0). Used by the
+/// completion hook: caller enqueues a new iteration if the returned
+/// value is > 0. CAS-safe: two workers can't race and both enqueue.
+pub async fn take_initial_burst_slot(pool: &PgPool, loop_id: Uuid) -> Result<i32, DbError> {
+    use sqlx::Row;
+    let row: Option<sqlx::postgres::PgRow> = sqlx::query(
+        "UPDATE loops
+            SET initial_burst_remaining = GREATEST(initial_burst_remaining - 1, 0)
+          WHERE id = $1 AND initial_burst_remaining > 0
+        RETURNING initial_burst_remaining + 1 AS prev",
+    )
+    .bind(loop_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row
+        .and_then(|r| r.try_get::<i32, _>("prev").ok())
+        .unwrap_or(0))
+}
+
+/// Enumerate loops that should wake up when a research topic gets a
+/// new outcome. Filters to kind='exec', enabled, and
+/// triggers.on_artifact_update = true. Called by the completion hook
+/// after freeze_research_outcome inserts a new row. Returns
+/// (loop_id, workspace_id, task_template, graph) so the caller can
+/// enqueue directly without a second fetch.
+pub async fn loops_awaiting_topic(
+    pool: &PgPool,
+    topic_id: Uuid,
+) -> Result<Vec<(Uuid, Uuid, String, Value)>, DbError> {
+    use sqlx::Row;
+    let rows = sqlx::query(
+        "SELECT id, workspace_id, task_template, graph
+           FROM loops
+          WHERE source_research_topic_id = $1
+            AND kind = 'exec'
+            AND enabled = true
+            AND (triggers ->> 'on_artifact_update')::boolean = true",
+    )
+    .bind(topic_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| {
+            (
+                r.get::<Uuid, _>("id"),
+                r.get::<Uuid, _>("workspace_id"),
+                r.get::<String, _>("task_template"),
+                r.get::<Value, _>("graph"),
+            )
+        })
+        .collect())
+}
+
+/// Truthy when the loop currently has a queued or running iteration.
+/// Used by triggers (on_artifact_update, on_completion chain) to
+/// coalesce — no point enqueuing another iteration while one is
+/// already pending.
+pub async fn has_active_run(pool: &PgPool, loop_id: Uuid) -> Result<bool, DbError> {
+    use sqlx::Row;
+    let row: Option<sqlx::postgres::PgRow> = sqlx::query(
+        "SELECT 1 AS one FROM topology_runs
+          WHERE loop_id = $1 AND status IN ('queued', 'running')
+          LIMIT 1",
+    )
+    .bind(loop_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.is_some())
+}
+
 /// Read the reorder_events array for a loop, newest-first, capped at
 /// `limit`. Used by the progress endpoint to surface a compact recent
 /// history on the sidebar card. Empty array for standalone loops or

migrations/0044_loops_kind_and_triggers.sql

@@ -0,0 +1,39 @@
+-- Folds research into loops as a "kind" of loop. Two kinds exist:
+--
+--   'exec'     — today's behavior. Each iteration runs the coordinator
+--                task_template (optionally prepended with a research
+--                artifact + focus instructions for INT-consumption
+--                loops).
+--   'research' — the fire runs the research pipeline against a bound
+--                topic config, appending a versioned research_outcomes
+--                row each time. Used for nightly paper-survey loops
+--                (D1 answer: every runnable thing is a loop; even a
+--                one-shot topic gets its own kind='research' loop with
+--                initial_burst=1).
+--
+-- Default 'exec' + CHECK constraint keeps every existing row valid.
+-- The paired research + coding pattern (nightly research produces
+-- outcome, coding loop wakes on on_artifact_update to consume next INT)
+-- composes two loops of different kinds sharing a source topic id.
+ALTER TABLE loops
+    ADD COLUMN kind TEXT NOT NULL DEFAULT 'exec'
+        CHECK (kind IN ('exec', 'research')),
+    -- Countdown for the triggers.initial_burst quota. On create_loop
+    -- we enqueue the first iteration inline and set this to burst-1;
+    -- on each completion the topology_worker checks and, when > 0,
+    -- enqueues the next iteration + decrements. Once it hits 0 the
+    -- burst is exhausted (further iterations happen only through
+    -- cron / webhook / on_completion / on_artifact_update).
+    ADD COLUMN initial_burst_remaining INT NOT NULL DEFAULT 0;
+
+CREATE INDEX loops_kind_source_idx
+    ON loops (kind, source_research_topic_id)
+    WHERE source_research_topic_id IS NOT NULL;
+
+-- Fast lookup for the on_artifact_update hook: given a topic_id, find
+-- all exec-kind loops bound to it with the trigger enabled.
+CREATE INDEX loops_on_artifact_update_idx
+    ON loops (source_research_topic_id)
+    WHERE source_research_topic_id IS NOT NULL
+      AND (triggers ->> 'on_artifact_update')::boolean = true
+      AND enabled = true;