research: auto-transition processing → reviewing on last run

Hooks the topology_worker's post-terminal path into a new
notify_run_completed repo helper that atomically transitions the topic
processing → reviewing when the completed run has research_topic_id set
AND no siblings for that topic are still queued or running. Guarded on
status='processing' so a retry, a re-fire, or a topic already past
processing are all no-ops. Best-effort at the worker; DB hiccups are
logged and never fail the run.

The manual /submit-review endpoint stays as an escape hatch for topics
that end up parked in processing with nothing to complete (updated the
doc comment).

.sqlx/query-e26955627419e4773f72abb9010fcdcd589d777f0fd20b444a629feed2bf75e4.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE research_topics t\n            SET status = 'reviewing', updated_at = now()\n          WHERE t.id = (\n                    SELECT research_topic_id FROM topology_runs\n                     WHERE id = $1 AND research_topic_id IS NOT NULL\n                )\n            AND t.status = 'processing'\n            AND NOT EXISTS (\n                    SELECT 1 FROM topology_runs\n                     WHERE research_topic_id = t.id\n                       AND id <> $1\n                       AND status IN ('queued', 'running')\n                )\n          RETURNING t.id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "e26955627419e4773f72abb9010fcdcd589d777f0fd20b444a629feed2bf75e4"
+}

.sqlx/query-ebeb5ef13a40b1693425588096a4ec37dd01ecf6cd9c051aaf37399a69f617bf.json

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

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

@@ -244,9 +244,10 @@ pub async fn start_topic(
 }
 
 /// `POST /api/research/:id/submit-review` — flips status `processing → reviewing`.
-/// v1 is caller-driven: the UI hits this when the human is happy with the
+/// Kept as a manual escape hatch: the orchestrator auto-transitions on the
-/// runs' output. A later commit hooks this from the orchestrator on the
+/// last topology_run's completion (see topology_worker's
-/// last topology_run's completion.
+/// `notify_run_completed` hook), so callers only need this when there are
+/// no runs (e.g. a topic parked in `processing` with nothing in flight).
 pub async fn submit_review(
     State(state): State<AppState>,
     Authed(user): Authed,

crates/cm-api/src/topology_worker.rs

@@ -72,6 +72,7 @@ async fn run_job(
                 let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;
             }
         }
+        maybe_transition_research_topic(pool, id).await;
         return;
     }
 
@@ -138,6 +139,21 @@ async fn run_job(
             }
         }
     }
+    maybe_transition_research_topic(pool, id).await;
+}
+
+/// Post-terminal hook: if this run belongs to a research topic and it was
+/// the last sibling in flight, transition the topic `processing → reviewing`.
+/// Best-effort — a DB hiccup here logs but doesn't fail the run.
+async fn maybe_transition_research_topic(pool: &PgPool, id: Uuid) {
+    match cm_db::repo::topology_runs::notify_run_completed(pool, id).await {
+        Ok(true) => {
+            // Left intentionally quiet on success; the UI polls the topic
+            // status. Future: emit a run_event so live viewers see it flip.
+        }
+        Ok(false) => {}
+        Err(e) => eprintln!("topology_worker: notify_run_completed({id}) failed: {e}"),
+    }
 }
 
 /// Drive a graph to completion with the durable per-step checkpoint +

crates/cm-api/tests/topology_jobs.rs

@@ -2,11 +2,51 @@
 //! (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::{loops, topology_runs, users, workspaces};
+use cm_db::repo::{loops, research_topics, topology_runs, users, workspaces};
 use cm_domain::{Role, User, UserId, Workspace, WorkspaceId};
 use serde_json::json;
 use uuid::Uuid;
 
+async fn seed_user(pool: &sqlx::PgPool, ws: WorkspaceId, email: &str) -> UserId {
+    let user = User {
+        id: UserId::new(),
+        workspace_id: ws,
+        email: email.into(),
+        role: Role::Owner,
+        display_name: "Owner".into(),
+        created_at: time::OffsetDateTime::UNIX_EPOCH,
+    };
+    users::insert(pool, &user).await.unwrap();
+    user.id
+}
+
+/// Enqueue a durable topology run with `research_topic_id` set. Kept in the
+/// test file to avoid a production repo helper for the topic-scoped enqueue
+/// path (nothing else in the codebase writes this column yet).
+async fn enqueue_run_with_topic(
+    pool: &sqlx::PgPool,
+    workspace_id: WorkspaceId,
+    task: &str,
+    topic_id: Uuid,
+) -> Uuid {
+    let id = Uuid::now_v7();
+    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n","role":"r"}], "edges": []});
+    sqlx::query!(
+        "INSERT INTO topology_runs
+            (id, workspace_id, task, kind, status, graph, tier, research_topic_id)
+         VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5)",
+        id,
+        workspace_id.as_uuid(),
+        task,
+        graph,
+        topic_id,
+    )
+    .execute(pool)
+    .await
+    .unwrap();
+    id
+}
+
 async fn seed_workspace(pool: &sqlx::PgPool) -> WorkspaceId {
     let ws = Workspace {
         id: WorkspaceId::new(),
@@ -138,15 +178,7 @@ async fn list_by_loop_returns_only_that_loops_iterations_newest_first() {
     let ws = seed_workspace(&pool).await;
 
     // A loop needs a valid `created_by` user in the same workspace.
-    let user = User {
+    let user_id = seed_user(&pool, ws, "[email protected]").await;
-        id: UserId::new(),
-        workspace_id: ws,
-        email: "[email protected]".into(),
-        role: Role::Owner,
-        display_name: "Hop".into(),
-        created_at: time::OffsetDateTime::UNIX_EPOCH,
-    };
-    users::insert(&pool, &user).await.unwrap();
 
     let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
     let triggers = json!({});
@@ -165,7 +197,7 @@ async fn list_by_loop_returns_only_that_loops_iterations_newest_first() {
             next_fire_at: None,
             webhook_token: None,
             webhook_signing_key: None,
-            created_by: user.id.as_uuid(),
+            created_by: user_id.as_uuid(),
         },
     )
     .await
@@ -199,20 +231,13 @@ async fn list_by_loop_returns_only_that_loops_iterations_newest_first() {
             next_fire_at: None,
             webhook_token: None,
             webhook_signing_key: None,
-            created_by: user.id.as_uuid(),
+            created_by: user_id.as_uuid(),
         },
     )
     .await
     .unwrap();
-    let _other_it = loops::enqueue_iteration(
+    let _other_it =
-        &pool,
+        loops::enqueue_iteration(&pool, other_loop, ws.as_uuid(), "other", &graph, 1, None)
-        other_loop,
-        ws.as_uuid(),
-        "other",
-        &graph,
-        1,
-        None,
-    )
             .await
             .unwrap();
 
@@ -239,3 +264,112 @@ async fn list_by_loop_returns_only_that_loops_iterations_newest_first() {
         .unwrap();
     assert!(cross.is_empty(), "loops are scoped by workspace");
 }
+
+#[tokio::test]
+async fn notify_run_completed_transitions_topic_when_no_siblings_in_flight() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let user_id = seed_user(&pool, ws, "[email protected]").await;
+
+    let topic = research_topics::create(
+        &pool,
+        ws.as_uuid(),
+        "Topic",
+        "desc",
+        "spec",
+        user_id.as_uuid(),
+    )
+    .await
+    .unwrap();
+    // The auto-transition only fires while the topic is `processing`.
+    research_topics::set_status(&pool, topic, ws.as_uuid(), "processing")
+        .await
+        .unwrap();
+
+    let run_id = enqueue_run_with_topic(&pool, ws, "task", topic).await;
+    // Flip to a terminal state before calling — mirrors the worker order.
+    let result = json!({"final_output": "done"});
+    topology_runs::complete(&pool, run_id, &result)
+        .await
+        .unwrap();
+
+    let transitioned = topology_runs::notify_run_completed(&pool, run_id)
+        .await
+        .unwrap();
+    assert!(transitioned, "no siblings in flight → topic transitions");
+    let t = research_topics::get(&pool, topic, ws.as_uuid())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(t.status, "reviewing");
+
+    // Idempotent: a second call after the topic has already left `processing`
+    // is a no-op.
+    let again = topology_runs::notify_run_completed(&pool, run_id)
+        .await
+        .unwrap();
+    assert!(
+        !again,
+        "second call is a no-op — topic is no longer processing"
+    );
+}
+
+#[tokio::test]
+async fn notify_run_completed_leaves_topic_processing_when_siblings_in_flight() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let user_id = seed_user(&pool, ws, "[email protected]").await;
+
+    let topic = research_topics::create(
+        &pool,
+        ws.as_uuid(),
+        "Topic",
+        "desc",
+        "spec",
+        user_id.as_uuid(),
+    )
+    .await
+    .unwrap();
+    research_topics::set_status(&pool, topic, ws.as_uuid(), "processing")
+        .await
+        .unwrap();
+
+    let done = enqueue_run_with_topic(&pool, ws, "first", topic).await;
+    let _still_queued = enqueue_run_with_topic(&pool, ws, "second", topic).await;
+    topology_runs::complete(&pool, done, &json!({"final_output": "x"}))
+        .await
+        .unwrap();
+
+    let transitioned = topology_runs::notify_run_completed(&pool, done)
+        .await
+        .unwrap();
+    assert!(!transitioned, "sibling still queued → hold");
+    let t = research_topics::get(&pool, topic, ws.as_uuid())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(t.status, "processing");
+}
+
+#[tokio::test]
+async fn notify_run_completed_ignores_runs_with_no_research_topic() {
+    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":"n","role":"r"}], "edges": []});
+    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
+        .await
+        .unwrap();
+    topology_runs::complete(&pool, id, &json!({}))
+        .await
+        .unwrap();
+
+    let transitioned = topology_runs::notify_run_completed(&pool, id)
+        .await
+        .unwrap();
+    assert!(
+        !transitioned,
+        "no research_topic_id → nothing to transition"
+    );
+}

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

@@ -187,6 +187,37 @@ pub async fn touch(pool: &PgPool, id: Uuid) -> Result<(), DbError> {
     Ok(())
 }
 
+/// If this run belongs to a research topic AND no siblings of that topic
+/// are still queued or running, transition the topic `processing → reviewing`.
+/// Guarded by `status = 'processing'` so a repeat call (e.g. a retry) is a
+/// no-op; a topic already reviewing/publishing/published stays put.
+/// Returns `true` when the topic was transitioned.
+pub async fn notify_run_completed(pool: &PgPool, id: Uuid) -> Result<bool, DbError> {
+    // One statement: subquery locates the topic id, subquery counts siblings
+    // still in flight (excluding *this* run — it's about to be flipped to
+    // completed/failed by the caller, but ordering isn't guaranteed here).
+    let row = sqlx::query!(
+        "UPDATE research_topics t
+            SET status = 'reviewing', updated_at = now()
+          WHERE t.id = (
+                    SELECT research_topic_id FROM topology_runs
+                     WHERE id = $1 AND research_topic_id IS NOT NULL
+                )
+            AND t.status = 'processing'
+            AND NOT EXISTS (
+                    SELECT 1 FROM topology_runs
+                     WHERE research_topic_id = t.id
+                       AND id <> $1
+                       AND status IN ('queued', 'running')
+                )
+          RETURNING t.id",
+        id,
+    )
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.is_some())
+}
+
 /// Mark a job completed and store its final result blob.
 pub async fn complete(pool: &PgPool, id: Uuid, result: &Value) -> Result<(), DbError> {
     sqlx::query!(