world view: restore in-flight landmarks as mission orbs

Slice 9 cleanup removed active_research_topics + active_loops and the
repo:{topic_id} / loop:{id} project orbs they emitted. The World SSE
loop was left with only agents + active runs — no persistent pin for
"this is what the team is working on right now."

Replaces those with a mission-era equivalent: one mission:{id} orb per
running mission, plus world.touch beams from every assigned team
member. Missions outlive individual runs, so the orb persists even
when no run is claimed — matches the UX intent of the legacy
landmarks without the retired research/loops plumbing.

Query: missions ⋈ team_members where m.status='running', grouped by
mission_id for the orb + fanned out per member for the touches.

Closes task #21.

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

@@ -44,6 +44,36 @@ async fn working_agents(pool: &PgPool, ws: WorkspaceId) -> HashSet<String> {
     rows.into_iter().map(|r| r.get::<String, _>("id")).collect()
 }
 
+/// Active missions (status='running') with their assigned team members —
+/// one row per (mission, agent) pair. The World SSE loop emits each as
+/// a `mission:<id>` landmark orb + `world.touch` beams from every team
+/// member. Replaces the retired research/loops landmarks (commit
+/// fdb8cfe) with the missions-era equivalent.
+async fn active_missions(pool: &PgPool, ws: WorkspaceId) -> Vec<(String, String, String)> {
+    let rows = sqlx::query(
+        "SELECT m.id::text        AS mission_id,
+                m.title            AS title,
+                tm.claw_id::text   AS agent_id
+           FROM missions m
+           JOIN team_members tm ON tm.team_id = m.team_id
+          WHERE m.workspace_id = $1
+            AND m.status = 'running'",
+    )
+    .bind(ws.as_uuid())
+    .fetch_all(pool)
+    .await
+    .unwrap_or_default();
+    rows.into_iter()
+        .map(|r| {
+            (
+                r.get::<String, _>("mission_id"),
+                r.get::<String, _>("title"),
+                r.get::<String, _>("agent_id"),
+            )
+        })
+        .collect()
+}
+
 /// Currently-running runs in the workspace as (run_id, agent_id) — each is a
 /// real "this agent is converging on its active work" signal (Gource).
 async fn active_runs(pool: &PgPool, ws: WorkspaceId) -> Vec<(String, String)> {
@@ -361,6 +391,29 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
                 }
             }
 
+            // Mission landmarks: one `mission:<id>` orb per running mission,
+            // with `world.touch` beams from every assigned team member. Missions
+            // outlive individual runs, so the orb gives the World a persistent
+            // pin for "this is what the team is working on right now" even when
+            // no run is claimed. Replaces the retired repo:{topic}/loop:{id}
+            // landmarks after commit fdb8cfe.
+            let missions = active_missions(&pool, ws).await;
+            let mut seen_missions: HashSet<String> = HashSet::new();
+            for (mission_id, title, agent_id) in &missions {
+                if seen_missions.insert(mission_id.clone()) {
+                    let node_id = format!("mission:{mission_id}");
+                    yield sse(
+                        "node.activity",
+                        json!({ "nodeId": node_id, "label": title, "kind": "mission", "heat": 0.75 }),
+                    );
+                }
+                let node_id = format!("mission:{mission_id}");
+                yield sse(
+                    "world.touch",
+                    json!({ "agentId": agent_id, "nodeId": node_id, "kind": "mission", "weight": 0.6 }),
+                );
+            }
+
             // Real convergence: each running agent beams toward its active-run node.
             for (run_id, agent_id) in &runs {
                 let node_id = format!("run:{}", &run_id[..run_id.len().min(8)]);