World: real-data convergence — pawns roam the live tree; runs drive world.touch

- engine: when there are no service/event touch-nodes (a quiet feed), pawns roam
  the real org/company/team structure so the view is always alive with zero
  synthetic data; real touch-nodes take priority when present.
- /api/world/live: emit world.touch + node.activity per currently-running run
  (agent_runs joined to sessions, state='running') so each working agent visibly
  converges on its active-run node; fold running agents into working status;
  telemetry.loops = active run count.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

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

@@ -42,6 +42,23 @@ async fn working_agents(pool: &PgPool, ws: WorkspaceId) -> HashSet<String> {
     rows.into_iter().map(|r| r.get::<String, _>("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)> {
+    let rows = sqlx::query(
+        "SELECT ar.id::text AS run_id, s.agent_id::text AS agent_id
+         FROM agent_runs ar JOIN sessions s ON s.id = ar.session_id
+         WHERE s.workspace_id = $1 AND ar.state = 'running'",
+    )
+    .bind(ws.as_uuid())
+    .fetch_all(pool)
+    .await
+    .unwrap_or_default();
+    rows.into_iter()
+        .map(|r| (r.get::<String, _>("run_id"), r.get::<String, _>("agent_id")))
+        .collect()
+}
+
 /// Count of doors (approvals) awaiting a decision in the workspace.
 async fn doors_pending(pool: &PgPool, ws: WorkspaceId) -> i64 {
     sqlx::query_scalar::<_, i64>(
@@ -67,7 +84,11 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
                 Ok(r) => r,
                 Err(_) => break,
             };
-            let working = working_agents(&pool, ws).await;
+            let mut working = working_agents(&pool, ws).await;
+            let runs = active_runs(&pool, ws).await;
+            for (_, agent_id) in &runs {
+                working.insert(agent_id.clone());
+            }
 
             if first {
                 // Seed the world graph with the workspace's agents as nodes.
@@ -90,9 +111,19 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
                 }
             }
 
+            // 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)]);
+                yield sse(
+                    "node.activity",
+                    json!({ "nodeId": node_id, "label": "active run", "kind": "event", "heat": 0.85 }),
+                );
+                yield sse("world.touch", json!({ "agentId": agent_id, "nodeId": node_id, "kind": "event" }));
+            }
+
             yield sse(
                 "telemetry",
-                json!({ "doorsPending": doors_pending(&pool, ws).await }),
+                json!({ "doorsPending": doors_pending(&pool, ws).await, "loops": runs.len() }),
             );
 
             first = false;

frontend/src/components/world/engine.ts

@@ -269,11 +269,16 @@ export class WorldEngine {
   private stepPawns(dt: number) {
     const pawns = [...this.pawns.values()];
     const worldNodes = [...this.nodes.values()].filter((n) => n.tier === "service" || n.tier === "event");
+    // Prefer real touch-targets (services/events); when there are none (a quiet
+    // live feed), let pawns roam the real org/company/team structure so the view
+    // is always alive without any synthetic data.
+    const targetPool = worldNodes.length
+      ? worldNodes
+      : [...this.nodes.values()].filter((n) => n.tier !== "root");
     for (const p of pawns) {
-      // keep alive offline: drift to a random world node when untargeted
       p.retime -= dt;
-      if (!p.targetId && p.retime <= 0 && worldNodes.length) {
-        p.targetId = worldNodes[Math.floor(Math.random() * worldNodes.length)].id;
+      if (!p.targetId && p.retime <= 0 && targetPool.length) {
+        p.targetId = targetPool[Math.floor(Math.random() * targetPool.length)].id;
         p.retime = 1.5 + Math.random() * 2.5;
       }
       const target = p.targetId ? this.nodes.get(p.targetId) : undefined;