diff --git a/crates/cm-api/src/routes/world.rs b/crates/cm-api/src/routes/world.rs index 7411f02..cad4a3e 100644 --- a/crates/cm-api/src/routes/world.rs +++ b/crates/cm-api/src/routes/world.rs @@ -42,6 +42,23 @@ async fn working_agents(pool: &PgPool, ws: WorkspaceId) -> HashSet { rows.into_iter().map(|r| r.get::("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::("run_id"), r.get::("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, 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, 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; diff --git a/frontend/src/components/world/engine.ts b/frontend/src/components/world/engine.ts index 210b427..4c6301e 100644 --- a/frontend/src/components/world/engine.ts +++ b/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;