loops: kind='research' dispatch — research runs as loops

Delivers the research/loop fold: kind='research' loops run the
research pipeline each iteration, appending a new research_outcomes
version. The paired on_artifact_update fan-out then wakes up any
kind='exec' loops bound to the same topic to consume new INTs. Every
runnable thing is now a loop (D1).

Backend — DB helpers:
- cm_db::repo::loops::kind_and_binding — reads (kind, source_topic,
  task_template) so callers can dispatch without hydrating the whole
  Loop struct.
- cm_db::repo::loops::enqueue_iteration_with_topic — new variant that
  sets research_topic_id on topology_runs alongside loop_id, so the
  completion hook's freeze_research_outcome writes a new outcome
  version for research-kind iterations.
- cm_db::repo::research_topics::get_any_workspace — cross-workspace
  fetch used by the research task builder (the loop row is
  authoritative for the workspace binding via kind_and_binding).

Backend — dispatch:
- routes::loops::compose_research_iteration_task — builds the
  coordinator prompt for a research iteration: topic title +
  description + outcome_kind + prior version pointer + refresh
  instructions (survey new sources, preserve stable INT ids, mark
  superseded items as deprecated rather than delete). The completion
  hook writes the resulting synthesis as research_outcomes v(prior+1).
- routes::loops::compose_and_enqueue_iteration — one-shot dispatch:
  reads the kind, picks compose_iteration_task (exec) or
  compose_research_iteration_task (research), enqueues with or
  without research_topic_id set.

All four enqueue callsites now route through compose_and_enqueue:
- create_loop (initial_burst)
- run_now
- webhook_receive
- topology_worker::continue_initial_burst
- topology_worker::freeze_research_outcome (on_artifact_update fan-out)

Research-kind loops naturally form the "nightly refresh" side of a
paired research + coding loop: research writes a fresh outcome
version → fan-out wakes exec loops with on_artifact_update →
coding loops consume the next INT (which the research loop may have
just added). D2 answer (inherit repo binding): repo lives on the
topic; both loops sharing the source topic id read from the same
context, no duplication. D3 answer (coordinator resolves): the
research iteration prompt tells the team to preserve stable INT ids
and mark deprecations rather than delete, so coding loops' consumed
lists stay valid across versions.

Follow-up (next commit): ResearchWizard schedule step — "Just once /
Nightly / Manual" that creates the paired research-kind loop with
initial_burst=1 (just once) or cron 0 3 * * * (nightly) + optional
paired coding loop with on_artifact_update.

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

@@ -55,6 +55,112 @@ 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`.
+/// One-shot compose + enqueue for a loop iteration. Reads the loop's
+/// kind from the DB and dispatches: kind='exec' uses
+/// compose_iteration_task (INT-consumption prepend); kind='research'
+/// uses compose_research_iteration_task AND sets research_topic_id on
+/// the topology_run so freeze_research_outcome writes a new outcome
+/// version at completion. Returns the run id. Standalone exec loops
+/// (no source topic) still work — the compose helper returns the
+/// task_template verbatim.
+pub async fn compose_and_enqueue_iteration(
+    pool: &sqlx::PgPool,
+    loop_id: Uuid,
+    workspace_id: Uuid,
+    graph: &Value,
+    parent_run_id: Option<Uuid>,
+    task_template_override: Option<&str>,
+) -> Result<Uuid, cm_db::DbError> {
+    // Kind is the source of truth — task_template alone isn't enough
+    // to know whether to write to research_outcomes.
+    let (kind, source_topic, template) =
+        match cm_db::repo::loops::kind_and_binding(pool, loop_id).await? {
+            Some(t) => t,
+            None => return Err(cm_db::DbError::NotFound),
+        };
+    let template_ref = task_template_override.unwrap_or(&template);
+    let iter = cm_db::repo::loops::next_iteration(pool, loop_id).await?;
+    if kind == "research" {
+        // Research-kind requires a bound topic (schema-level constraint
+        // isn't enforced yet — surface the misconfiguration explicitly).
+        let Some(topic_id) = source_topic else {
+            return Err(cm_db::DbError::NotFound);
+        };
+        let task = compose_research_iteration_task(pool, topic_id, template_ref).await;
+        cm_db::repo::loops::enqueue_iteration_with_topic(
+            pool,
+            loop_id,
+            workspace_id,
+            &task,
+            graph,
+            iter,
+            parent_run_id,
+            Some(topic_id),
+        )
+        .await
+    } else {
+        let task = compose_iteration_task(pool, loop_id, template_ref).await;
+        cm_db::repo::loops::enqueue_iteration(
+            pool,
+            loop_id,
+            workspace_id,
+            &task,
+            graph,
+            iter,
+            parent_run_id,
+        )
+        .await
+    }
+}
+
+/// Build the coordinator prompt for a kind='research' loop iteration.
+/// Wraps the topic's description + outcome_kind + prior artifact
+/// version pointer into an instruction that asks the team to refresh
+/// the plan (survey new sources, revise existing INTs, add new ones)
+/// and emit the updated artifact using the same section shape. The
+/// completion hook's `freeze_research_outcome` will insert a new
+/// versioned row automatically because the topology_run carries
+/// research_topic_id.
+pub async fn compose_research_iteration_task(
+    pool: &PgPool,
+    topic_id: Uuid,
+    task_template: &str,
+) -> String {
+    let (title, description, outcome_kind, prior_version) =
+        match cm_db::repo::research_topics::get_any_workspace(pool, topic_id).await {
+            Ok(Some(t)) => {
+                let prior = cm_db::repo::research_outcomes::latest(pool, topic_id)
+                    .await
+                    .unwrap_or(None)
+                    .map(|o| o.version)
+                    .unwrap_or(0);
+                (t.title, t.description, t.outcome_kind, prior)
+            }
+            _ => return task_template.to_string(),
+        };
+    format!(
+        "RESEARCH LOOP ITERATION\n\
+         =======================\n\
+         Topic: {title}\n\
+         Outcome kind: {outcome_kind}\n\
+         Prior artifact version: v{prior_version} (0 = fresh)\n\n\
+         DESCRIPTION:\n{description}\n\n\
+         YOUR JOB THIS ITERATION:\n\
+         - Refresh the research — pull in any new papers / findings since v{prior_version}.\n\
+         - Update the artifact using the SAME section structure the outcome_kind\n\
+           requires (e.g. integrations kind = executive summary + INT-XX cards).\n\
+         - Preserve stable ids (INT-01 stays INT-01 across versions). If an item\n\
+           is superseded, mark it {{deprecated: <reason>}} rather than deleting so\n\
+           downstream coding loops that already consumed it don't lose context.\n\
+         - Add NEW items with new ids continuing from the last used number.\n\
+         - Cite every claim; never fabricate sources or repo paths.\n\n\
+         The workspace's final synthesis is captured as research_outcomes v{}. \
+         Downstream on_artifact_update loops will wake up on this write.\n\n\
+         LOOP OPERATOR NOTES:\n{task_template}\n",
+        prior_version + 1
+    )
+}
+
 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
@@ -306,20 +412,16 @@ pub async fn create_loop(
     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)
+        // Kind-aware compose + enqueue — research-kind loops get a
-            .await
+        // research prompt and research_topic_id set on the run.
-            .unwrap_or(0);
+        let template = body.task_template.trim();
-        let task = compose_iteration_task(&state.pool, id, body.task_template.trim()).await;
+        match compose_and_enqueue_iteration(
-        // 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,
+            Some(template),
         )
         .await
         {
@@ -629,18 +731,18 @@ pub async fn run_now(
     // resolves its gateway URL when it picks up the run. Best-effort;
     // never blocks the enqueue on Docker being unreachable.
     ensure_loop_container(&state.pool, l.workspace_id, l.id).await;
+    // Kind-aware — research loops write to research_outcomes.
     let iter = cm_db::repo::loops::next_iteration(&state.pool, l.id).await?;
-    let task = compose_iteration_task(&state.pool, l.id, &l.task_template).await;
+    let run_id = compose_and_enqueue_iteration(
-    let run_id = cm_db::repo::loops::enqueue_iteration(
         &state.pool,
         l.id,
         l.workspace_id,
-        &task,
         &l.graph,
-        iter,
         l.last_run_id,
+        Some(&l.task_template),
     )
-    .await?;
+    .await
+    .map_err(|_| ApiError::Internal)?;
     cm_db::repo::loops::mark_fired(&state.pool, l.id, run_id, l.next_fire_at).await?;
     Ok(Json(RunTriggered {
         run_id,
@@ -681,15 +783,13 @@ pub async fn webhook_receive(
         Ok(n) => n,
         Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),
     };
-    let task = compose_iteration_task(&state.pool, l.id, &l.task_template).await;
+    let run_id = match compose_and_enqueue_iteration(
-    let run_id = match cm_db::repo::loops::enqueue_iteration(
         &state.pool,
         l.id,
         l.workspace_id,
-        &task,
         &l.graph,
-        iter,
         l.last_run_id,
+        Some(&l.task_template),
     )
     .await
     {

crates/cm-api/src/topology_worker.rs

@@ -226,26 +226,20 @@ async fn freeze_research_outcome(pool: &PgPool, run_id: Uuid, final_output: &str
         {
             continue; // Coalesce.
         }
-        let iter = cm_db::repo::loops::next_iteration(pool, loop_id)
+        // Fan-outs always target kind='exec' (filter enforced in
-            .await
+        // loops_awaiting_topic). compose_and_enqueue_iteration takes
-            .unwrap_or(0);
+        // the exec path and prepends the freshly-inserted artifact.
-        // Build the enriched task with the freshly-inserted artifact
+        if let Err(e) = crate::routes::loops::compose_and_enqueue_iteration(
-        // (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),
+            Some(&task_template),
         )
         .await
         {
-            eprintln!("topology_worker: on_artifact_update enqueue({loop_id}) failed: {e}");
+            eprintln!("topology_worker: on_artifact_update enqueue({loop_id}) failed: {e:?}");
         }
     }
 }
@@ -275,27 +269,24 @@ async fn continue_initial_burst(pool: &PgPool, run_id: Uuid) {
     {
         return;
     }
-    // Fetch the loop so we have the workspace + graph + task_template
+    // Fetch the loop so we have the workspace + graph. The kind-aware
-    // to enqueue the next iteration.
+    // dispatcher pulls task_template + kind from the same helper it
+    // uses at first fire, so bursts across an exec + research pair
+    // behave identically.
     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)
+    if let Err(e) = crate::routes::loops::compose_and_enqueue_iteration(
-        .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),
+        None,
     )
     .await
     {
-        eprintln!("topology_worker: continue_initial_burst enqueue failed: {e}");
+        eprintln!("topology_worker: continue_initial_burst enqueue failed: {e:?}");
     }
 }
 

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

@@ -104,6 +104,36 @@ pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<Loop>, DbErro
     Ok(rows)
 }
 
+/// Kind + source_research_topic_id + task_template — the minimum a
+/// caller needs to compose the right iteration for a loop without
+/// hydrating the whole Loop struct. Kind='exec' preserves today's
+/// behavior; kind='research' builds a research prompt bound to the
+/// source topic so freeze_research_outcome writes a new outcome
+/// version.
+pub async fn kind_and_binding(
+    pool: &PgPool,
+    loop_id: Uuid,
+) -> Result<Option<(String, Option<Uuid>, String)>, DbError> {
+    use sqlx::Row;
+    let row: Option<sqlx::postgres::PgRow> = sqlx::query(
+        "SELECT kind, source_research_topic_id, task_template
+           FROM loops
+          WHERE id = $1",
+    )
+    .bind(loop_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|r| {
+        (
+            r.get::<String, _>("kind"),
+            r.try_get::<Option<Uuid>, _>("source_research_topic_id")
+                .ok()
+                .flatten(),
+            r.get::<String, _>("task_template"),
+        )
+    }))
+}
+
 /// 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
@@ -370,7 +400,6 @@ pub async fn loops_awaiting_topic(
 /// 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')
@@ -718,20 +747,53 @@ pub async fn enqueue_iteration(
     iteration: i32,
     parent_run_id: Option<Uuid>,
 ) -> Result<Uuid, DbError> {
-    let run_id = Uuid::now_v7();
+    enqueue_iteration_with_topic(
-    sqlx::query!(
+        pool,
-        "INSERT INTO topology_runs
+        loop_id,
-            (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,
+        None,
     )
+    .await
+}
+
+/// Variant of `enqueue_iteration` that also sets `research_topic_id` on
+/// the topology_runs row. Used by kind='research' loops so
+/// `freeze_research_outcome` writes a new outcome version each
+/// iteration, and by any future flow that binds a run to both a loop
+/// and a research topic.
+pub async fn enqueue_iteration_with_topic(
+    pool: &PgPool,
+    loop_id: Uuid,
+    workspace_id: Uuid,
+    task: &str,
+    graph: &Value,
+    iteration: i32,
+    parent_run_id: Option<Uuid>,
+    research_topic_id: Option<Uuid>,
+) -> Result<Uuid, DbError> {
+    let run_id = Uuid::now_v7();
+    // Dynamic query so the new column combination (loop_id +
+    // research_topic_id on the same row) doesn't require an offline
+    // sqlx cache regen — the enqueue path only runs on user actions,
+    // not the tight worker loop.
+    sqlx::query(
+        "INSERT INTO topology_runs
+            (id, workspace_id, task, kind, status, graph, tier,
+             loop_id, iteration, parent_run_id, research_topic_id)
+         VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7, $8)",
+    )
+    .bind(run_id)
+    .bind(workspace_id)
+    .bind(task)
+    .bind(graph)
+    .bind(loop_id)
+    .bind(iteration)
+    .bind(parent_run_id)
+    .bind(research_topic_id)
     .execute(pool)
     .await?;
     Ok(run_id)

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

@@ -153,6 +153,43 @@ pub async fn list(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<ResearchTopic
     Ok(rows)
 }
 
+/// Cross-workspace fetch used by internal callers (topology_worker
+/// completion hooks, kind='research' loop iteration builders) where
+/// the caller already has an authoritative workspace binding from the
+/// linked loop row. Skip the workspace scope filter to avoid a second
+/// hop.
+pub async fn get_any_workspace(pool: &PgPool, id: Uuid) -> Result<Option<ResearchTopic>, DbError> {
+    use sqlx::Row;
+    let row: Option<sqlx::postgres::PgRow> = sqlx::query(
+        "SELECT id, workspace_id, title, description, outcome_kind, status,
+                created_by, created_at, updated_at, published_at, topology_kind,
+                repo_id, repo_workspace_path,
+                zeroclaw_container_name, zeroclaw_gateway_url
+           FROM research_topics
+          WHERE id = $1",
+    )
+    .bind(id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|r| ResearchTopic {
+        id: r.get("id"),
+        workspace_id: r.get("workspace_id"),
+        title: r.get("title"),
+        description: r.get("description"),
+        outcome_kind: r.get("outcome_kind"),
+        status: r.get("status"),
+        created_by: r.get("created_by"),
+        created_at: r.get("created_at"),
+        updated_at: r.get("updated_at"),
+        published_at: r.try_get("published_at").ok().flatten(),
+        topology_kind: r.get("topology_kind"),
+        repo_id: r.try_get("repo_id").ok().flatten(),
+        repo_workspace_path: r.try_get("repo_workspace_path").ok().flatten(),
+        zeroclaw_container_name: r.try_get("zeroclaw_container_name").ok().flatten(),
+        zeroclaw_gateway_url: r.try_get("zeroclaw_gateway_url").ok().flatten(),
+    }))
+}
+
 pub async fn get(
     pool: &PgPool,
     id: Uuid,