slice 5: task-card parser + background worker

Watches topology_runs' event stream for the INT-XX marker protocol
(see skills/foundation/int-xx-marker-protocol.md) and materializes
mission_tasks rows with typed status so the canvas Tasks tab renders
a live timeline instead of raw agent chatter.

Migration 0051 adds mission_id + mission_phase_id columns to
topology_runs (nullable) so runs enqueued by a mission phase can be
attributed. Populated by future phase executors; NULL for legacy
research/loops runs (parser skips them cleanly).

New Rust surface:
  - task_card_parser::parse(text) — line-scanner over TASK/WORK/
    HANDOFF/TEST_PASS/TEST_FAIL/REVIEW_APPROVE/REVIEW_BLOCK/COMPLETED
    markers. Strict: exact kind + colon + INT- prefix, no in-prose
    matches, no bold/code-fence wrappers.
  - task_card_parser::apply_for_run(pool, run_id) — reads the run's
    mission binding, walks its event payloads, extracts text/output/
    content/message string fields (matching every ZeroClaw event
    shape we see), parses markers, UPSERTs mission_tasks via the
    (phase_id, external_id) unique key from Slice 1.
  - task_card_worker::spawn — 15s poller over runs updated in the
    last 5 minutes. Idempotent + generous window survives server
    restarts + task-scheduling jitter.

Boot wires the worker after the content loaders. Silent no-op when
mission wiring isn't populated yet.

MarkerKind → status mapping (monotonic-forward):
  TASK           → created
  WORK           → working
  HANDOFF        → validating
  TEST_PASS      → validating
  TEST_FAIL      → failed
  REVIEW_APPROVE → validating
  REVIEW_BLOCK   → failed
  COMPLETED      → complete

Follow-ups:
  - Wire phase executor to populate topology_runs.mission_id +
    mission_phase_id (Slice 6/7/8 work)
  - Assign assigned_agent_id via the event's producing agent alias
    (currently always None)
  - SSE stream on /api/missions/{id}/tasks for live canvas updates
    (currently the canvas polls via mission GET)

Co-Authored-By: Claude Opus 4.7 <[email protected]>

crates/bins/clawmates-server/src/main.rs

@@ -284,6 +284,10 @@ async fn run() -> Result<(), String> {
             eprintln!("team_template_loader: loaded {n_tpl} builtin team template(s)");
         });
     }
+    // Task-card parser worker (Slice 5): scans recent topology_runs
+    // for INT-XX markers in event payloads and upserts mission_tasks
+    // rows so the canvas renders a live status timeline.
+    cm_api::task_card_worker::spawn(pool.clone());
     // Outbound-email delivery: drains the §15-gated `outbox` over SMTP. Inert
     // until CLAWMATES_SMTP_* is set, so it ships safely before credentials exist.
     cm_runtime::spawn_drainer(pool.clone(), std::time::Duration::from_secs(10));

crates/cm-api/src/lib.rs

@@ -17,6 +17,8 @@ mod routes;
 mod runtime_provision;
 pub mod skills_loader;
 pub mod swarm;
+pub mod task_card_parser;
+pub mod task_card_worker;
 pub mod team_template_loader;
 pub mod tool_versions;
 mod topology_exec;

crates/cm-api/src/task_card_parser.rs

@@ -0,0 +1,253 @@
+//! Task-card parser — Slice 5.
+//!
+//! Watches `run_events` payloads for the INT-XX marker protocol
+//! (see skills/foundation/int-xx-marker-protocol.md) and materializes
+//! `mission_tasks` rows with typed state so the mission canvas Tasks
+//! tab can render a status timeline.
+//!
+//! Recognized markers (case-sensitive, on their own line):
+//!   TASK: INT-NN — <title>       → status='created'
+//!   WORK: INT-NN                 → status='working'
+//!   HANDOFF: INT-NN              → status='validating'
+//!   TEST_PASS: INT-NN            → status='validating' (unchanged if already validating+)
+//!   TEST_FAIL: INT-NN — <reason> → status='failed'
+//!   REVIEW_APPROVE: INT-NN       → status='validating'
+//!   REVIEW_BLOCK: INT-NN — <reason> → status='failed'
+//!   COMPLETED: INT-NN            → status='complete'
+//!
+//! Uses UPSERT keyed on (phase_id, external_id) so the same INT
+//! progressing through states updates a single row; state transitions
+//! are monotonic where reasonable (once complete, only failed can
+//! demote — but the loop-once semantics of INT items mean this rarely
+//! matters in practice).
+
+use serde_json::Value;
+use sqlx::PgPool;
+use sqlx::Row;
+use uuid::Uuid;
+
+use cm_db::repo::missions::UpsertTask;
+
+/// A parsed marker from a single run-event payload.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Marker {
+    pub int_id: String, // "INT-05"
+    pub kind: MarkerKind,
+    pub title: Option<String>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MarkerKind {
+    Task,
+    Work,
+    Handoff,
+    TestPass,
+    TestFail,
+    ReviewApprove,
+    ReviewBlock,
+    Completed,
+}
+
+impl MarkerKind {
+    /// The status this marker implies. Preferences monotonic forward
+    /// motion — the UPSERT layer may still overwrite prior states.
+    pub fn status(&self) -> &'static str {
+        match self {
+            MarkerKind::Task => "created",
+            MarkerKind::Work => "working",
+            MarkerKind::Handoff | MarkerKind::TestPass | MarkerKind::ReviewApprove => "validating",
+            MarkerKind::TestFail | MarkerKind::ReviewBlock => "failed",
+            MarkerKind::Completed => "complete",
+        }
+    }
+}
+
+/// Line-scanner over the raw text of a run's output. Cheap enough to
+/// re-run per event; the UPSERT layer collapses duplicates.
+pub fn parse(text: &str) -> Vec<Marker> {
+    let mut out = Vec::new();
+    for line in text.lines() {
+        let trimmed = line.trim();
+        if let Some(m) = parse_line(trimmed) {
+            out.push(m);
+        }
+    }
+    out
+}
+
+fn parse_line(line: &str) -> Option<Marker> {
+    // Match `<KIND>: INT-NN` (rest optional). Strict on the colon and
+    // the INT- prefix — anything laxer starts matching prose.
+    let (kind_str, rest) = line.split_once(':')?;
+    let kind = match kind_str.trim() {
+        "TASK" => MarkerKind::Task,
+        "WORK" => MarkerKind::Work,
+        "HANDOFF" => MarkerKind::Handoff,
+        "TEST_PASS" => MarkerKind::TestPass,
+        "TEST_FAIL" => MarkerKind::TestFail,
+        "REVIEW_APPROVE" => MarkerKind::ReviewApprove,
+        "REVIEW_BLOCK" => MarkerKind::ReviewBlock,
+        "COMPLETED" => MarkerKind::Completed,
+        _ => return None,
+    };
+    let rest = rest.trim();
+    let (id_tok, tail) = match rest.split_once(char::is_whitespace) {
+        Some((a, b)) => (a, Some(b.trim())),
+        None => (rest, None),
+    };
+    if !id_tok.starts_with("INT-") {
+        return None;
+    }
+    let int_id = id_tok.trim_end_matches(&[',', ';', '.'][..]).to_string();
+    // Title: after the id + any of ` — / – / - ` separators
+    let title = tail.and_then(|t| {
+        let t = t.trim_start_matches(['—', '–', '-', ':'].as_slice()).trim();
+        if t.is_empty() {
+            None
+        } else {
+            Some(t.to_string())
+        }
+    });
+    Some(Marker {
+        int_id,
+        kind,
+        title,
+    })
+}
+
+/// Scan every event of a topology_run that has a mission_id set;
+/// parse markers out of the event payload's `text` / `output` fields;
+/// UPSERT each into `mission_tasks`.
+///
+/// Idempotent: re-running against the same run's events collapses
+/// into the current-truth state (last marker per INT wins per pass).
+pub async fn apply_for_run(pool: &PgPool, run_id: Uuid) -> Result<usize, String> {
+    // Load the run + its mission binding. Bail cheaply if no mission.
+    let run = sqlx::query(
+        "SELECT mission_id, mission_phase_id
+         FROM topology_runs WHERE id = $1",
+    )
+    .bind(run_id)
+    .fetch_optional(pool)
+    .await
+    .map_err(|e| format!("load run: {e}"))?;
+    let Some(run) = run else {
+        return Ok(0);
+    };
+    let mission_id: Option<Uuid> = run.try_get("mission_id").ok().flatten();
+    let phase_id: Option<Uuid> = run.try_get("mission_phase_id").ok().flatten();
+    let (Some(mission_id), Some(phase_id)) = (mission_id, phase_id) else {
+        return Ok(0);
+    };
+
+    // Read event payloads in order.
+    let rows = sqlx::query(
+        "SELECT payload
+         FROM run_events WHERE run_id = $1
+         ORDER BY seq ASC",
+    )
+    .bind(run_id)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| format!("load events: {e}"))?;
+
+    let mut all_markers: Vec<Marker> = Vec::new();
+    for r in rows {
+        let payload: Value = r.get("payload");
+        for text in extract_text_fields(&payload) {
+            all_markers.extend(parse(&text));
+        }
+    }
+
+    let mut applied = 0usize;
+    for m in &all_markers {
+        let title = m.title.clone().unwrap_or_else(|| m.int_id.clone());
+        cm_db::repo::missions::upsert_task(
+            pool,
+            UpsertTask {
+                mission_id,
+                phase_id,
+                external_id: &m.int_id,
+                title: &title,
+                assigned_agent_id: None,
+                status: m.kind.status(),
+                run_id: Some(run_id),
+            },
+        )
+        .await
+        .map_err(|e| format!("upsert_task {}: {e}", m.int_id))?;
+        applied += 1;
+    }
+    Ok(applied)
+}
+
+/// Walk a payload's JSON tree and yield every string field named
+/// `text`, `output`, `content`, or `message`. Handles the shapes we
+/// see across ZeroClaw events (chunk / tool_result / done / final).
+fn extract_text_fields(v: &Value) -> Vec<String> {
+    fn walk(v: &Value, out: &mut Vec<String>) {
+        match v {
+            Value::Object(map) => {
+                for (k, child) in map {
+                    if matches!(k.as_str(), "text" | "output" | "content" | "message") {
+                        if let Value::String(s) = child {
+                            out.push(s.clone());
+                        }
+                    }
+                    walk(child, out);
+                }
+            }
+            Value::Array(items) => {
+                for item in items {
+                    walk(item, out);
+                }
+            }
+            _ => {}
+        }
+    }
+    let mut out = Vec::new();
+    walk(v, &mut out);
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_completed_marker() {
+        let out = parse("COMPLETED: INT-05");
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].int_id, "INT-05");
+        assert_eq!(out[0].kind, MarkerKind::Completed);
+    }
+
+    #[test]
+    fn parses_task_with_title() {
+        let out = parse("TASK: INT-12 — wire the loader");
+        assert_eq!(out[0].kind, MarkerKind::Task);
+        assert_eq!(out[0].title.as_deref(), Some("wire the loader"));
+    }
+
+    #[test]
+    fn parses_test_fail_with_reason() {
+        let out = parse("TEST_FAIL: INT-03 — flake in async_setup");
+        assert_eq!(out[0].kind, MarkerKind::TestFail);
+        assert_eq!(out[0].title.as_deref(), Some("flake in async_setup"));
+    }
+
+    #[test]
+    fn ignores_marker_inside_prose() {
+        // Not on its own line + not starting with the exact kind:
+        let out = parse("we should not TASK: INT-05 like this");
+        assert!(out.is_empty());
+    }
+
+    #[test]
+    fn scans_multiline_batch() {
+        let src = "TASK: INT-01 — foo\nWORK: INT-01\nother text\nCOMPLETED: INT-01\n";
+        let out = parse(src);
+        assert_eq!(out.len(), 3);
+        assert_eq!(out[2].kind, MarkerKind::Completed);
+    }
+}

crates/cm-api/src/task_card_worker.rs

@@ -0,0 +1,60 @@
+//! Background worker that reruns `task_card_parser::apply_for_run`
+//! on recently-active topology_runs. Slice 5.
+//!
+//! Cadence: every 15s, scan runs updated in the last 5 minutes that
+//! have mission_id set. Deliberately generous — the parser is
+//! idempotent, and paying a tiny bit of extra CPU beats missing a
+//! marker because a run completed between polls.
+
+use sqlx::PgPool;
+use sqlx::Row;
+use std::time::Duration;
+use uuid::Uuid;
+
+const POLL_INTERVAL: Duration = Duration::from_secs(15);
+/// Look back this far when picking candidate runs. Wider than the
+/// poll interval to survive server restarts + task-scheduling jitter.
+const LOOKBACK_SECS: i64 = 300;
+
+/// Spawn the polling task. Silent no-op when the mission wiring
+/// isn't populated (no runs with mission_id set — the pre-Slice-9
+/// legacy paths keep working untouched).
+pub fn spawn(pool: PgPool) {
+    tokio::spawn(async move {
+        // Small initial delay so we don't fight the migration on boot.
+        tokio::time::sleep(Duration::from_secs(5)).await;
+        let mut ticker = tokio::time::interval(POLL_INTERVAL);
+        // Drop the first tick — interval fires immediately by default.
+        ticker.tick().await;
+        loop {
+            ticker.tick().await;
+            if let Err(e) = sweep_once(&pool).await {
+                eprintln!("task_card_worker: sweep failed: {e}");
+            }
+        }
+    });
+}
+
+async fn sweep_once(pool: &PgPool) -> Result<(), String> {
+    let rows = sqlx::query(
+        "SELECT id
+         FROM topology_runs
+         WHERE mission_id IS NOT NULL
+           AND created_at > now() - make_interval(secs => $1::float)
+         ORDER BY created_at DESC
+         LIMIT 200",
+    )
+    .bind(LOOKBACK_SECS as f64)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| format!("query candidates: {e}"))?;
+
+    for r in rows {
+        let id: Uuid = r.get("id");
+        if let Err(e) = crate::task_card_parser::apply_for_run(pool, id).await {
+            // Per-run failure shouldn't abort the whole sweep.
+            eprintln!("task_card_worker: apply_for_run({id}) failed: {e}");
+        }
+    }
+    Ok(())
+}

migrations/0051_topology_runs_missions.sql

@@ -0,0 +1,17 @@
+-- Slice 5 — wire topology_runs to missions so the task-card parser
+-- can attribute agent output to the right mission + phase.
+--
+-- Populated by the phase executor (Slices 4/5/6/7/8) when it enqueues
+-- a topology_run on behalf of a mission phase. Left NULL for legacy
+-- runs and the existing research/loops paths that pre-date missions.
+--
+-- The parser (task_card_parser) skips runs where mission_id IS NULL —
+-- keeps the surface additive during the Slice 9 transition.
+
+ALTER TABLE topology_runs
+    ADD COLUMN mission_id       UUID REFERENCES missions(id) ON DELETE SET NULL,
+    ADD COLUMN mission_phase_id UUID REFERENCES mission_phases(id) ON DELETE SET NULL;
+
+CREATE INDEX topology_runs_mission_idx
+    ON topology_runs (mission_id, created_at DESC)
+    WHERE mission_id IS NOT NULL;