research: publish approval gate + explicit state transitions

Fourth commit of the Research + Loops arc. Completes the state machine
for research topics with the publish approval gate the spec asked for.

Migration 0032 — research_publish_approvals
  Dedicated small table (id, workspace_id, topic_id, requested_by,
  status, decided_by/at, created_at). Keeping it separate from the
  existing `approvals` table (0001) because that one is tightly coupled
  to gated tool calls inside an agent run — session_key + run_id +
  action_type + category + payload + preview + requested_by_agent, all
  NOT NULL. Forcing those nullable would ripple through cm_safety;
  cleaner to give publish approvals their own two-transition state
  machine.

New endpoints
  POST /api/research/:id/submit-review               processing → reviewing
                                                     (v1 caller-driven; the
                                                     orchestrator hook comes
                                                     when we wire actual runs)
  POST /api/research/:id/request-publish             creates a pending
                                                     approval. Rejects with
                                                     409 if the topic already
                                                     has one open.
  GET  /api/research/publish-approvals               list workspace's pending
  POST /api/research/publish-approvals/:id/approve   flips approval to
                                                     approved + transitions
                                                     the topic
                                                     reviewing → publishing
                                                     (which stamps
                                                     published_at)
  POST /api/research/publish-approvals/:id/reject    stays in reviewing; new
                                                     requests allowed

The approve/reject write is an atomic UPDATE ... WHERE status = 'pending';
the decide() repo function returns whether the caller won the race so
concurrent double-approves collapse to a single topic transition.

State machine after this commit:
  standby ─POST /start─▶ processing ─POST /submit-review─▶ reviewing
    ─POST /request-publish + approve─▶ publishing ─(future: artifact
    assembly)─▶ published

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

@@ -16,6 +16,7 @@ pub mod node_tools;
 pub mod nodes;
 pub mod orgs;
 pub mod outbox;
+pub mod research_publish_approvals;
 pub mod research_topics;
 pub mod routine_runs;
 pub mod routines;

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

@@ -0,0 +1,126 @@
+//! Publish approval gate for research topics — see 0032 migration header.
+//! Small table with a small state machine (pending → approved | rejected).
+//! One pending row per topic at a time; enforced at the route layer by
+//! looking up `pending_for_topic` before create.
+
+use serde::{Deserialize, Serialize};
+use sqlx::PgPool;
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PublishApproval {
+    pub id: Uuid,
+    pub workspace_id: Uuid,
+    pub topic_id: Uuid,
+    pub requested_by: Uuid,
+    pub status: String,
+    pub decided_by: Option<Uuid>,
+    pub decided_at: Option<OffsetDateTime>,
+    pub created_at: OffsetDateTime,
+}
+
+pub async fn create(
+    pool: &PgPool,
+    workspace_id: Uuid,
+    topic_id: Uuid,
+    requested_by: Uuid,
+) -> Result<Uuid, DbError> {
+    let id = Uuid::now_v7();
+    sqlx::query!(
+        "INSERT INTO research_publish_approvals
+            (id, workspace_id, topic_id, requested_by, status)
+         VALUES ($1, $2, $3, $4, 'pending')",
+        id,
+        workspace_id,
+        topic_id,
+        requested_by,
+    )
+    .execute(pool)
+    .await?;
+    Ok(id)
+}
+
+/// Pending approval for a topic, if any. The route layer uses this to
+/// short-circuit before writing a duplicate request.
+pub async fn pending_for_topic(
+    pool: &PgPool,
+    topic_id: Uuid,
+) -> Result<Option<PublishApproval>, DbError> {
+    let row = sqlx::query_as!(
+        PublishApproval,
+        "SELECT id, workspace_id, topic_id, requested_by, status,
+                decided_by, decided_at, created_at
+           FROM research_publish_approvals
+          WHERE topic_id = $1 AND status = 'pending'
+          LIMIT 1",
+        topic_id,
+    )
+    .fetch_optional(pool)
+    .await?;
+    Ok(row)
+}
+
+pub async fn get(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: Uuid,
+) -> Result<Option<PublishApproval>, DbError> {
+    let row = sqlx::query_as!(
+        PublishApproval,
+        "SELECT id, workspace_id, topic_id, requested_by, status,
+                decided_by, decided_at, created_at
+           FROM research_publish_approvals
+          WHERE id = $1 AND workspace_id = $2",
+        id,
+        workspace_id,
+    )
+    .fetch_optional(pool)
+    .await?;
+    Ok(row)
+}
+
+pub async fn list_pending(
+    pool: &PgPool,
+    workspace_id: Uuid,
+) -> Result<Vec<PublishApproval>, DbError> {
+    let rows = sqlx::query_as!(
+        PublishApproval,
+        "SELECT id, workspace_id, topic_id, requested_by, status,
+                decided_by, decided_at, created_at
+           FROM research_publish_approvals
+          WHERE workspace_id = $1 AND status = 'pending'
+          ORDER BY created_at DESC",
+        workspace_id,
+    )
+    .fetch_all(pool)
+    .await?;
+    Ok(rows)
+}
+
+/// Atomically flip a pending row to approved/rejected. Returns whether the
+/// caller was the one who won the race — false when the row was already
+/// decided (idempotent).
+pub async fn decide(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: Uuid,
+    decided_by: Uuid,
+    approve: bool,
+) -> Result<bool, DbError> {
+    let new_status = if approve { "approved" } else { "rejected" };
+    let result = sqlx::query!(
+        "UPDATE research_publish_approvals
+            SET status = $4, decided_by = $3, decided_at = now()
+          WHERE id = $1 AND workspace_id = $2 AND status = 'pending'",
+        id,
+        workspace_id,
+        decided_by,
+        new_status,
+    )
+    .execute(pool)
+    .await?;
+    Ok(result.rows_affected() > 0)
+}