//! `GET /api/world/live` — the Clawmates Event Taxonomy as a Server-Sent Events //! feed for the World + Observe visualizations. Read-only and workspace-scoped. //! //! Phase 1 emits the *real* shape of the workspace: a `topology.update` of its //! agents, an `agent.status` per agent (working when it holds a live container, //! else idle), and a `telemetry` snapshot — polled, mirroring `run_events_sse`. //! //! The richer `world.touch` / `agent.tool.call` events (an agent converging on //! the node it acts upon — the Gource centerpiece) come from normalizing the //! durable runner's `run_events`; that is the extension seam (see `normalize`), //! filled in as the runner emits node targets. Until then the client's synthetic //! fallback supplies that motion. use std::collections::HashSet; use std::convert::Infallible; use std::time::Duration; use axum::extract::{Query, State}; use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::IntoResponse; use axum::Json; use cm_domain::WorkspaceId; use serde::Deserialize; use serde_json::{json, Value}; use sqlx::{PgPool, Row}; use uuid::Uuid; use crate::{ApiError, AppState, Authed}; fn sse(event: &str, data: Value) -> Result { Ok(Event::default().event(event).data(data.to_string())) } /// Agents that currently hold a live container (any kind) → "working". async fn working_agents(pool: &PgPool, ws: WorkspaceId) -> HashSet { let rows = sqlx::query( "SELECT DISTINCT a.id::text AS id FROM agents a JOIN agent_containers ac ON ac.agent_id = a.id WHERE a.workspace_id = $1 AND a.deleted_at IS NULL", ) .bind(ws.as_uuid()) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter().map(|r| r.get::("id")).collect() } /// One row per (mission, agent) pair for every mission the World should draw. /// /// Three bugs were fixed here at once, and each hid the next: /// /// 1. **The join was on the wrong column.** It read /// `JOIN team_members tm ON tm.team_id = m.team_id`, but `missions.team_id` /// is a legacy pointer at the FIRST minted team — `0056_mission_teams.sql` /// superseded it with the `mission_teams(mission_id, team_id, purpose)` /// junction, which is what everything else (including `/api/workforce`) /// joins through. A multi-team mission showed only its first team. /// /// 2. **It was an INNER JOIN, and microVM missions have no team at all.** /// `mission_orchestrator::on_launch` deliberately mints none for them, so /// the platform's primary execution tier was dropped by the join and the /// World has been showing nothing whatsoever for it. LEFT JOIN, and /// `agent_id` is `None` for those — a phase that ran with no platform /// agents draws no pawns, which is the truth rather than a gap. /// /// 3. **Only `running` missions were selected.** Missions finish in minutes, /// so the World was empty almost always. Recently-finished ones come back /// too, carrying `status` so the client can draw a finished map instead of /// animating a corpse. struct MissionRow { mission_id: String, title: String, status: String, template_kind: String, completed_at: Option, /// `None` for a mission with no team — see (2) above. agent_id: Option, } async fn world_missions(pool: &PgPool, ws: WorkspaceId, only: Option) -> Vec { let rows = sqlx::query( "SELECT m.id::text AS mission_id, m.title AS title, m.status AS status, m.template_kind AS template_kind, to_char(m.completed_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS completed_at, tm.claw_id::text AS agent_id FROM missions m LEFT JOIN mission_teams mt ON mt.mission_id = m.id LEFT JOIN team_members tm ON tm.team_id = mt.team_id WHERE m.workspace_id = $1 AND ( m.status = 'running' OR ( m.status IN ('completed', 'failed') AND m.completed_at > now() - interval '24 hours' ) ) AND ($2::uuid IS NULL OR m.id = $2) ORDER BY m.created_at DESC", ) .bind(ws.as_uuid()) .bind(only) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .map(|r| MissionRow { mission_id: r.get::("mission_id"), title: r.get::("title"), status: r.get::("status"), template_kind: r.get::("template_kind"), completed_at: r.get::, _>("completed_at"), agent_id: r.get::, _>("agent_id"), }) .collect() } /// Every phase of the given missions, in plan order, with the agents that /// execute it. /// /// The whole plan is emitted — including `pending` phases that have not /// started — because the World draws the mission's shape upfront and lights it /// as it progresses. A phase list that only appeared as phases began would make /// a five-phase mission indistinguishable from a one-phase mission until it was /// nearly over. struct PhaseRow { mission_id: String, phase_id: String, kind: String, order_idx: i32, status: String, iteration: i32, started_at: Option, completed_at: Option, } async fn world_phases(pool: &PgPool, ws: WorkspaceId, only: Option) -> Vec { let rows = sqlx::query( "SELECT mp.mission_id::text AS mission_id, mp.id::text AS phase_id, mp.kind AS kind, mp.order_idx AS order_idx, mp.status AS status, COALESCE(mp.iteration, 0) AS iteration, to_char(mp.started_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS started_at, to_char(mp.completed_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS completed_at FROM mission_phases mp JOIN missions m ON m.id = mp.mission_id WHERE m.workspace_id = $1 AND ( m.status = 'running' OR ( m.status IN ('completed', 'failed') AND m.completed_at > now() - interval '24 hours' ) ) AND ($2::uuid IS NULL OR m.id = $2) ORDER BY mp.mission_id, mp.order_idx", ) .bind(ws.as_uuid()) .bind(only) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .map(|r| PhaseRow { mission_id: r.get::("mission_id"), phase_id: r.get::("phase_id"), kind: r.get::("kind"), order_idx: r.get::("order_idx"), status: r.get::("status"), iteration: r.get::("iteration"), started_at: r.get::, _>("started_at"), completed_at: r.get::, _>("completed_at"), }) .collect() } /// Files a phase touched, from the `code_diff` artifact captured at delivery. /// /// This is recorded fact, not inference: `mission_delivery` writes the path /// list with the same revision and excludes it uses for `files_changed`, so the /// orbs the World draws are the files git says changed. /// /// It is end-of-phase detail — the capture runs when a phase finishes — so a /// running phase shows its station lit but no files until it lands. Live /// per-tool file touches are a separate, structured source. struct FileRow { mission_id: String, phase_id: String, path: String, status: String, } async fn world_files(pool: &PgPool, ws: WorkspaceId, only: Option) -> Vec { let rows = sqlx::query( "SELECT ma.mission_id::text AS mission_id, ma.phase_id::text AS phase_id, f->>'path' AS path, f->>'status' AS status FROM mission_artifacts ma JOIN missions m ON m.id = ma.mission_id CROSS JOIN LATERAL jsonb_array_elements( COALESCE(ma.metadata->'files', '[]'::jsonb)) AS f WHERE m.workspace_id = $1 AND ma.kind = 'code_diff' AND ma.phase_id IS NOT NULL AND ( m.status = 'running' OR ( m.status IN ('completed', 'failed') AND m.completed_at > now() - interval '24 hours' ) ) AND ($2::uuid IS NULL OR m.id = $2) LIMIT 2000", ) .bind(ws.as_uuid()) .bind(only) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .filter_map(|r| { Some(FileRow { mission_id: r.get::("mission_id"), phase_id: r.get::("phase_id"), path: r.get::, _>("path")?, status: r.get::, _>("status").unwrap_or_default(), }) }) .collect() } /// One row of `mission_events` — what an agent actually did. struct ActRow { id: i64, mission_id: String, phase_id: Option, agent_id: Option, kind: String, target: String, } /// Structured mission activity since `after`, oldest first. /// /// `after < 0` means "the first pass has not run yet": everything is returned /// so the caller can seed its cursor and draw the backlog as settled history. async fn world_acts(pool: &PgPool, ws: WorkspaceId, only: Option, after: i64) -> Vec { let rows = sqlx::query( "SELECT e.id, e.mission_id::text AS mission_id, e.phase_id::text AS phase_id, e.agent_id::text AS agent_id, e.kind, e.target FROM mission_events e JOIN missions m ON m.id = e.mission_id WHERE m.workspace_id = $1 AND e.kind IN ('tool.call', 'file.touch') AND e.target IS NOT NULL AND e.id > $2 AND ( m.status = 'running' OR ( m.status IN ('completed', 'failed') AND m.completed_at > now() - interval '24 hours' ) ) AND ($3::uuid IS NULL OR m.id = $3) ORDER BY e.id LIMIT 500", ) .bind(ws.as_uuid()) .bind(after) .bind(only) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .filter_map(|r| { Some(ActRow { id: r.get::("id"), mission_id: r.get::("mission_id"), phase_id: r.get::, _>("phase_id"), agent_id: r.get::, _>("agent_id"), kind: r.get::("kind"), target: r.get::, _>("target")?, }) }) .collect() } /// One security finding, as the scanner recorded it. struct FindingRow { mission_id: String, phase_id: String, task_id: String, title: String, } /// Findings raised by a security phase. /// /// They are `mission_tasks` rows — the scanner's own store — not a new table. /// There is deliberately NO severity field here: severity, file and line are /// substrings inside `title` (see `security_scan.rs`), and a severity parsed /// out of prose and then encoded as an orb's RADIUS would be an invented fact /// rendered as a measurement. The title is shown as written. async fn world_findings(pool: &PgPool, ws: WorkspaceId, only: Option) -> Vec { let rows = sqlx::query( "SELECT t.mission_id::text AS mission_id, t.phase_id::text AS phase_id, t.id::text AS task_id, t.title FROM mission_tasks t JOIN mission_phases p ON p.id = t.phase_id JOIN missions m ON m.id = t.mission_id WHERE m.workspace_id = $1 AND p.kind = 'security_scan' AND ( m.status = 'running' OR ( m.status IN ('completed', 'failed') AND m.completed_at > now() - interval '24 hours' ) ) AND ($2::uuid IS NULL OR m.id = $2) LIMIT 300", ) .bind(ws.as_uuid()) .bind(only) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .map(|r| FindingRow { mission_id: r.get::("mission_id"), phase_id: r.get::("phase_id"), task_id: r.get::("task_id"), title: r.get::("title"), }) .collect() } /// A benchmark phase's result, summarised. struct BenchRow { mission_id: String, phase_id: String, note: String, } /// Benchmark deltas, as an annotation on the phase — not as nodes. /// /// `delta` is a `Record` with no schema: `compute_delta` /// produces `{kind:"bencher_diff", samples:[…]}` when the before/after shapes /// are structurally comparable, and `{kind:"opaque"}` when they are not. Only /// the shape that can be parsed is formatted; the rest is COUNTED, never /// guessed at, so a driver whose output we do not understand reports "3 /// samples" rather than an invented improvement. async fn world_benchmarks(pool: &PgPool, ws: WorkspaceId, only: Option) -> Vec { let rows = sqlx::query( "SELECT DISTINCT ON (b.phase_id) b.mission_id::text AS mission_id, b.phase_id::text AS phase_id, b.delta FROM benchmark_snapshots b JOIN missions m ON m.id = b.mission_id WHERE m.workspace_id = $1 AND b.delta IS NOT NULL AND ( m.status = 'running' OR ( m.status IN ('completed', 'failed') AND m.completed_at > now() - interval '24 hours' ) ) AND ($2::uuid IS NULL OR m.id = $2) ORDER BY b.phase_id, b.iteration DESC", ) .bind(ws.as_uuid()) .bind(only) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .filter_map(|r| { let note = benchmark_note(&r.get::("delta"))?; Some(BenchRow { mission_id: r.get::("mission_id"), phase_id: r.get::("phase_id"), note, }) }) .collect() } /// One line describing a benchmark delta, or `None` if there is nothing /// truthful to say about it. fn benchmark_note(delta: &serde_json::Value) -> Option { let samples = delta.get("samples").and_then(|s| s.as_array())?; if samples.is_empty() { return None; } let mut improved = 0usize; let mut regressed = 0usize; // Best (most negative) percent change, since that is the one claim the // shape actually supports. let mut best: Option = None; for s in samples { match s.get("direction").and_then(|d| d.as_str()) { Some("improved") => improved += 1, Some("regressed") => regressed += 1, // Counted in the total but claimed for neither side. _ => {} } if let Some(pct) = s.get("delta_pct").and_then(serde_json::Value::as_f64) { best = Some(best.map_or(pct, |b: f64| b.min(pct))); } } let mut parts = vec![format!("{} sample(s)", samples.len())]; if improved > 0 { parts.push(format!("{improved} faster")); } if regressed > 0 { parts.push(format!("{regressed} slower")); } if let Some(b) = best.filter(|b| *b < 0.0) { parts.push(format!("best {:.1}%", b)); } Some(parts.join(", ")) } /// Agents that execute a phase of this kind, via the purposes the phase runner /// itself uses. Returns empty for a teamless (microVM) mission. async fn phase_agents(pool: &PgPool, mission_id: &str, kind: &str) -> Vec { let Ok(mid) = Uuid::parse_str(mission_id) else { return Vec::new(); }; // `purposes_for` is the runner's own mapping, shared rather than copied — // see its doc comment. let purposes: Vec = crate::phase_runner::purposes_for(kind) .iter() .map(|s| s.to_string()) .collect(); let rows = sqlx::query( "SELECT DISTINCT tm.claw_id::text AS agent_id FROM mission_teams mt JOIN team_members tm ON tm.team_id = mt.team_id WHERE mt.mission_id = $1 AND mt.purpose = ANY($2)", ) .bind(mid) .bind(&purposes) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter() .map(|r| r.get::("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)> { 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() } /// A short human label for a tool's input (for the tool-call target). fn summarize_input(input: &Value) -> String { for k in ["target", "path", "url", "query", "name", "file", "command"] { if let Some(s) = input.get(k).and_then(|v| v.as_str()) { return s.chars().take(48).collect(); } } String::new() } /// Normalize one durable `run_events` row into taxonomy events — the Rust twin of /// the handoff bridge's normalize(). The runner already journals these, so the /// live view shows REAL reasoning, tool-convergence and doors with no runner edit. fn normalize_run_event( agent_id: &str, event_type: &str, payload: &Value, ) -> Vec<(&'static str, Value)> { let mut out = Vec::new(); match event_type { "text_delta" => { if let Some(delta) = payload.get("delta").and_then(|v| v.as_str()) { out.push(( "agent.reasoning.delta", json!({ "agentId": agent_id, "text": delta }), )); } } "step_started" => { let tool = payload .get("tool") .and_then(|v| v.as_str()) .unwrap_or("tool"); let node_id = format!("tool:{tool}"); let target = payload .get("input") .map(summarize_input) .unwrap_or_default(); // File/project I/O gets an explosive burst (high touch weight). let lower = tool.to_lowercase(); let file_op = [ "file", "read", "write", "drive", "vault", "obsidian", "edit", "fs", "save", ] .iter() .any(|k| lower.contains(k)); let weight = if file_op { 1.0 } else { 0.4 }; out.push(( "agent.tool.call", json!({ "agentId": agent_id, "tool": tool, "target": target }), )); out.push(("node.activity", json!({ "nodeId": node_id, "label": tool, "kind": "service", "heat": if file_op { 1.0 } else { 0.9 } }))); // the agent converges on the tool it's using (the Gource beam) out.push(("world.touch", json!({ "agentId": agent_id, "nodeId": node_id, "kind": "service", "weight": weight }))); // File-op tools ALSO emit a `file:` touch so the repo // detail view can build the tree from real events. The path // comes from the input's `path` / `target` / `file` / `url` // keys — same lookup summarize_input does but we keep the // full string so the client can build the dir hierarchy. if file_op { if let Some(input) = payload.get("input") { for k in ["path", "target", "file", "filename", "url"] { if let Some(p) = input.get(k).and_then(|v| v.as_str()) { let cleaned = p.trim().trim_start_matches("./"); if !cleaned.is_empty() { let file_node = format!("file:{cleaned}"); out.push(( "node.activity", json!({ "nodeId": file_node, "label": cleaned, "kind": "service", "heat": 1.0 }), )); out.push(( "world.touch", json!({ "agentId": agent_id, "nodeId": file_node, "kind": "service", "weight": 1.0 }), )); break; } } } } } } "approval_required" => { let action = payload .get("action_type") .and_then(|v| v.as_str()) .unwrap_or("action"); let category = payload .get("category") .and_then(|v| v.as_str()) .unwrap_or(""); let door_id = payload .get("approval_id") .and_then(|v| v.as_str()) .unwrap_or(""); out.push(( "door.request", json!({ "doorId": door_id, "agentId": agent_id, "action": action, "target": category, "summary": action }), )); } "agent_message" => { let s = |k: &str| { payload .get(k) .and_then(|v| v.as_str()) .unwrap_or("") .to_owned() }; out.push(( "agent.message", json!({ "fromAgentId": agent_id, "toAgentId": s("to_agent_id"), "toName": s("to_name"), "text": s("text"), "threadId": s("thread_id"), }), )); } "room_message" => { let s = |k: &str| { payload .get(k) .and_then(|v| v.as_str()) .unwrap_or("") .to_owned() }; let participant_ids = payload .get("participant_ids") .and_then(|v| v.as_array()) .map(|a| { a.iter() .filter_map(|x| x.as_str().map(|s| s.to_owned())) .collect::>() }) .unwrap_or_default(); out.push(( "room.message", json!({ "fromAgentId": agent_id, "threadId": s("thread_id"), "subject": s("subject"), "text": s("text"), "participantIds": participant_ids, }), )); } "a2a_invoked" => { // An external A2A caller started a turn on this agent (a new ingress). out.push(("a2a.invoked", json!({ "agentId": agent_id }))); } _ => {} } out } /// Count of doors (approvals) awaiting a decision in the workspace. async fn doors_pending(pool: &PgPool, ws: WorkspaceId) -> i64 { sqlx::query_scalar::<_, i64>( "SELECT count(*) FROM approvals WHERE workspace_id = $1 AND status = 'pending'", ) .bind(ws.as_uuid()) .fetch_one(pool) .await .unwrap_or(0) } /// Live per-agent metrics for the agent command center's metric band. #[derive(Default)] struct AgentTele { tokens_per_min: i64, cost_per_hr: f64, loops: i64, doors: i64, } /// Per-agent telemetry for every agent in the workspace, in 4 grouped queries /// (not N×4): tokens over the last minute, credits over the last hour, active /// routines, and pending approvals — all keyed by agent id. async fn agent_telemetry( pool: &PgPool, ws: WorkspaceId, ) -> std::collections::HashMap { let mut m: std::collections::HashMap = std::collections::HashMap::new(); let id_of = |r: &sqlx::postgres::PgRow| r.get::("agent_id").to_string(); // tokens in the last minute ≈ tokens/min. for r in sqlx::query( "SELECT agent_id, SUM(tokens_in + tokens_out)::bigint AS v FROM usage_events WHERE workspace_id = $1 AND agent_id IS NOT NULL AND created_at > now() - interval '1 minute' GROUP BY agent_id", ) .bind(ws.as_uuid()) .fetch_all(pool) .await .unwrap_or_default() { m.entry(id_of(&r)).or_default().tokens_per_min = r.get("v"); } // credits spent in the last hour ≈ spend/hr. for r in sqlx::query( "SELECT agent_id, SUM(credits)::float8 AS v FROM usage_events WHERE workspace_id = $1 AND agent_id IS NOT NULL AND created_at > now() - interval '1 hour' GROUP BY agent_id", ) .bind(ws.as_uuid()) .fetch_all(pool) .await .unwrap_or_default() { m.entry(id_of(&r)).or_default().cost_per_hr = r.get("v"); } // active routines = loops. for r in sqlx::query( "SELECT r.agent_id AS agent_id, count(*)::bigint AS v FROM routines r JOIN agents a ON a.id = r.agent_id WHERE a.workspace_id = $1 AND r.status = 'active' GROUP BY r.agent_id", ) .bind(ws.as_uuid()) .fetch_all(pool) .await .unwrap_or_default() { m.entry(id_of(&r)).or_default().loops = r.get("v"); } // pending approvals = doors. for r in sqlx::query( "SELECT requested_by_agent AS agent_id, count(*)::bigint AS v FROM approvals WHERE workspace_id = $1 AND status = 'pending' GROUP BY requested_by_agent", ) .bind(ws.as_uuid()) .fetch_all(pool) .await .unwrap_or_default() { m.entry(id_of(&r)).or_default().doors = r.get("v"); } m } /// Query for `GET /api/world/live`. #[derive(serde::Deserialize)] pub struct LiveQuery { /// Scope the feed to one mission. The client already filters events by /// mission, but doing it here means the server stops querying and sending /// what the viewer will discard. pub mission: Option, } /// `GET /api/world/live` — the taxonomy SSE feed. pub async fn world_live( State(state): State, Authed(user): Authed, Query(q): Query, ) -> impl IntoResponse { let pool = state.pool.clone(); let ws = user.workspace_id; let only_mission = q.mission; let stream = async_stream::stream! { let mut first = true; // Remember last status per agent so we only push deltas after the seed. let mut last: std::collections::HashMap = std::collections::HashMap::new(); // Per-run journal cursor so we stream only NEW run_events each poll. let mut cursors: std::collections::HashMap = std::collections::HashMap::new(); // Last emitted signature per mission / per phase, so the plan is sent // once and then only when something actually moves. let mut last_mission: std::collections::HashMap = std::collections::HashMap::new(); let mut last_phase: std::collections::HashMap = std::collections::HashMap::new(); // (phase, path) pairs already announced — a delivered file is a fact // that happened once, not a recurring event. let mut last_file: HashSet = HashSet::new(); // `mission_events` cursor. -1 until the first pass seeds it, which is // what separates BACKFILL from MOTION: everything already in the table // when a subscriber arrives is history and is drawn as a settled map, // and only what lands afterwards is animated. Without the distinction, // opening a finished mission would replay an hour of tool calls as a // burst storm and read as a mission that just did all of it at once. let mut event_cursor: i64 = -1; // Findings already announced. A finding is raised once. let mut last_finding: HashSet = HashSet::new(); // Last benchmark summary per phase; a looping phase produces a new one. let mut last_bench: std::collections::HashMap = std::collections::HashMap::new(); // Audit-log cursor for edge-initiated inter-agent events (delegation, // A2A) that bypass the run loop. -1 until seeded on the first pass. let mut audit_cursor: i64 = -1; // Cursor over mission_events for the per-agent LIVE column. Starts at // -1 and jumps to the current max on first sight, so a page load // streams forward instead of replaying every past turn. let mut agent_ev_cursor: i64 = -1; // Track the brain-file size we last announced per agent so we only // emit `agent.memory` when the file has actually grown (or shrunk). // On first seed we still emit — the World engine needs the initial // scale for every pawn. let mut last_bytes: std::collections::HashMap = std::collections::HashMap::new(); // Push channel for frames that cannot wait for the 2s poll — an agent's // reasoning arrives token by token. Subscribed BEFORE the first poll so // nothing produced during the initial queries is missed. let mut live_rx = crate::live_bus::global().subscribe(); loop { let roster = match cm_db::repo::agents::roster(&pool, ws).await { Ok(r) => r, Err(_) => break, }; 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. let nodes: Vec = roster .iter() .map(|a| json!({ "id": a.id.to_string(), "tier": "agent", "label": a.name })) .collect(); yield sse("topology.update", json!({ "formation": "live", "nodes": nodes })); } for a in &roster { let id = a.id.to_string(); let status = if working.contains(&id) { "working" } else { "idle" }; if last.get(&id).map(|s| s != status).unwrap_or(true) { last.insert(id.clone(), status.to_string()); yield sse( "agent.status", json!({ "agentId": id, "status": status, "role": a.job_title }), ); } // Cheap brain-file stat — just reads the inode metadata, no // HDF5 open. Missing file (never provisioned) → treat as 0 // so the pawn stays at its base size. Only emit on change. let brain_path = crate::routes::claws::brain_dir() .join(format!("claw_{}.h5", id)); let bytes = std::fs::metadata(&brain_path).ok().map(|m| m.len()).unwrap_or(0); if last_bytes.get(&id).map(|&b| b != bytes).unwrap_or(true) { last_bytes.insert(id.clone(), bytes); yield sse("agent.memory", json!({ "agentId": id, "bytes": bytes })); } } // Mission landmarks, and the whole phase plan behind them. // // The orb pins "this is the work"; the phases give it shape. Every // phase is emitted, including ones that have not started, because // the World draws the plan upfront and lights it as it progresses — // a phase list that appeared only as phases began would make a // five-phase mission indistinguishable from a one-phase mission // until it was nearly over. let missions = world_missions(&pool, ws, only_mission).await; let mut seen_missions: HashSet = HashSet::new(); for m in &missions { if seen_missions.insert(m.mission_id.clone()) { // Delta-guarded like every other stateful event here. Without // this it re-sent the same mission on every poll — harmless // to a client that keys on missionId, and pure noise on the // wire that hides the events that DID change. let sig = format!("{}|{}", m.status, m.completed_at.as_deref().unwrap_or("")); if last_mission.get(&m.mission_id).map(|s| s != &sig).unwrap_or(true) { last_mission.insert(m.mission_id.clone(), sig); yield sse( "mission.update", json!({ "missionId": m.mission_id, "title": m.title, "status": m.status, "templateKind": m.template_kind, "completedAt": m.completed_at, }), ); } // A terminal mission is a map to read, not a scene to // animate: it still gets an orb, but no heat. let running = m.status == "running"; yield sse( "node.activity", json!({ "nodeId": format!("mission:{}", m.mission_id), "label": m.title, "kind": "mission", "heat": if running { 0.75 } else { 0.0 }, }), ); } // `agent_id` is None for a teamless microVM mission. No pawn // beams at it, which is accurate — nothing on this platform ran // that phase except a VM — and is why the join had to become a // LEFT JOIN rather than being left to drop the mission entirely. if let (Some(agent_id), true) = (&m.agent_id, m.status == "running") { yield sse( "world.touch", json!({ "agentId": agent_id, "nodeId": format!("mission:{}", m.mission_id), "kind": "mission", "weight": 0.6, }), ); } } for p in world_phases(&pool, ws, only_mission).await { // Re-emit only on change. `iteration` and `completed_at` are in // the signature because a phase that loops re-enters `running` // from `running` — status alone would hide the retry. let sig = format!( "{}|{}|{}", p.status, p.iteration, p.completed_at.as_deref().unwrap_or("") ); let changed = last_phase.get(&p.phase_id).map(|s| s != &sig).unwrap_or(true); if !changed { continue; } last_phase.insert(p.phase_id.clone(), sig); let agents = phase_agents(&pool, &p.mission_id, &p.kind).await; yield sse( "mission.phase", json!({ "missionId": p.mission_id, "phaseId": p.phase_id, "kind": p.kind, "orderIdx": p.order_idx, "status": p.status, "iteration": p.iteration, "startedAt": p.started_at, "completedAt": p.completed_at, "agentIds": agents, }), ); yield sse( "node.activity", json!({ "nodeId": format!("phase:{}", p.phase_id), "label": p.kind, "kind": "phase", // The client owns the lit/dim encoding via `status`; // heat here is only the arrival pulse. "heat": if p.status == "running" { 0.8 } else { 0.0 }, }), ); } // Files come AFTER the phases they belong to, deliberately: the // client hangs a file tree under the coding station, and parenting // there is first-write-wins. A file that arrived before its plan // would pin its whole directory tree at the origin. // // Files, once each. A file is emitted when its phase's diff was // captured and never again — the World keeps the node alive itself, // and re-sending would re-burst the orb every poll as though the // file had just been touched again. for f in world_files(&pool, ws, only_mission).await { let key = format!("{}|{}", f.phase_id, f.path); if last_file.contains(&key) { continue; } last_file.insert(key); yield sse( "mission.file", json!({ "missionId": f.mission_id, "phaseId": f.phase_id, "path": f.path, "status": f.status, "source": "diff", }), ); } // Benchmark results, as an annotation on the station. Re-emitted // only when the summary changes: a phase that loops produces a new // snapshot per iteration, and that IS news. for b in world_benchmarks(&pool, ws, only_mission).await { if last_bench.get(&b.phase_id).map(|n| n == &b.note).unwrap_or(false) { continue; } last_bench.insert(b.phase_id.clone(), b.note.clone()); yield sse( "mission.benchmark", json!({ "missionId": b.mission_id, "phaseId": b.phase_id, "note": b.note, }), ); } // Findings, once each, hanging off the security station. Same rule // as files: a finding is a fact that was raised once, and // re-sending it would pop the orb again on every poll. for f in world_findings(&pool, ws, only_mission).await { if !last_finding.insert(f.task_id.clone()) { continue; } yield sse( "mission.finding", json!({ "missionId": f.mission_id, "phaseId": f.phase_id, "findingId": f.task_id, "title": f.title, }), ); } // Structured activity — the motion channel. Tool calls and file // touches recorded at the source by the container tap and the // microVM `PostToolUse` hook. Never parsed from prose: a tool name // in a log is indistinguishable from an agent TALKING about a tool. { let backfill = event_cursor < 0; for a in world_acts(&pool, ws, only_mission, event_cursor.max(0)).await { event_cursor = event_cursor.max(a.id); match a.kind.as_str() { "file.touch" => { let key = format!("{}|{}", a.phase_id.as_deref().unwrap_or(""), a.target); if !last_file.insert(key) { continue; } yield sse( "mission.file", json!({ "missionId": a.mission_id, "phaseId": a.phase_id, "agentId": a.agent_id, "path": a.target, // Backlog is history: it draws the file and // stops there. Only what lands while someone // is watching is motion. "source": if backfill { "diff" } else { "tool" }, }), ); } // A tool call with an agent is a pawn leaving its // station and coming back; without one (a microVM phase // has no platform agent) it is only a node lighting up, // which is the truth rather than an invented traveller. _ if !backfill => { let node_id = format!("tool:{}", a.target); match &a.agent_id { Some(agent) => yield sse( "world.touch", json!({ "agentId": agent, "nodeId": node_id, "kind": "event", "weight": 0.45, }), ), None => yield sse( "node.activity", json!({ "nodeId": node_id, "label": a.target, "kind": "event", "heat": 0.5, }), ), } } // A backlogged tool call draws nothing: the orb would be // a tool nobody is using, permanently lit on a map of // work that already finished. _ => {} } } } // 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" })); // Tail the run's journal for richer real events (reasoning, tool // convergence, doors). On first sight, jump the cursor to the // current max so we stream forward without replaying the backlog. // // THIS BLOCK IS THE a2a/door LAYER ONLY, and it works: those // runs are the ones that write `run_events` (`run_events::append` // is reached from `routes/a2a.rs` and `mcp_door.rs`). // // A sibling block used to read `topology_runs.checkpoint.records` // here, on the theory that it covered missions. It could never // have: `run_id` comes from `active_runs`, which selects // `agent_runs.id`, and mission phases live in `topology_runs` // under independently generated ids. The query ran every poll and // matched nothing, for every mission, forever — which is why the // World has never shown a mission's tool or file activity. // Removed rather than repointed: pointing it at `topology_runs` // would resurrect a path whose only per-step content is the // agent's own prose output, and a tool name in prose is // indistinguishable from an agent talking about a tool. Mission // detail arrives as structured `mission_events` instead. if let Some(&after) = cursors.get(run_id) { let rows = sqlx::query( "SELECT seq, event_type, payload FROM run_events WHERE run_id = $1::uuid AND seq > $2 ORDER BY seq ASC LIMIT 200", ) .bind(run_id) .bind(after) .fetch_all(&pool) .await .unwrap_or_default(); let mut maxseq = after; for row in &rows { let seq: i64 = row.get("seq"); let et: String = row.get("event_type"); let payload: Value = row.get("payload"); for (t, d) in normalize_run_event(agent_id, &et, &payload) { yield sse(t, d); } if seq > maxseq { maxseq = seq; } } cursors.insert(run_id.clone(), maxseq); } else { let maxseq: i64 = sqlx::query_scalar( "SELECT coalesce(max(seq), -1) FROM run_events WHERE run_id = $1::uuid", ) .bind(run_id) .fetch_one(&pool) .await .unwrap_or(-1); cursors.insert(run_id.clone(), maxseq); } } // Per-agent telemetry → the command-center metric band (real data: // usage_events tokens/credits, active routines, pending approvals). let tele = agent_telemetry(&pool, ws).await; for a in &roster { let id = a.id.to_string(); let t = tele.get(&id); yield sse("telemetry", json!({ "agentId": id, "tokensPerMin": t.map(|x| x.tokens_per_min).unwrap_or(0), "costPerHr": t.map(|x| x.cost_per_hr).unwrap_or(0.0), "loops": t.map(|x| x.loops).unwrap_or(0), "doorsPending": t.map(|x| x.doors).unwrap_or(0), })); } // Per-agent LIVE column: REASONING STREAM + tool lines. // // These two taxonomy types were declared and listened for since the // command centre shipped, and NOTHING ever emitted them — the cards // could not populate no matter what an agent did. mission_events is // the durable source: `reasoning` rows carry the agent's own step // output, `tool.call` rows its actions. // // NOTE: on the container tier `tool.call` legitimately stays empty — // those agents are tool-free behind the §15 door. Reasoning flows on // every tier; tool lines appear where agents actually hold tools. if agent_ev_cursor < 0 { agent_ev_cursor = sqlx::query_scalar( "SELECT coalesce(max(id), 0) FROM mission_events", ) .fetch_one(&pool) .await .unwrap_or(0); } else { let rows = sqlx::query( "SELECT e.id, e.agent_id, e.kind, e.target, e.detail FROM mission_events e JOIN missions m ON m.id = e.mission_id WHERE m.workspace_id = $1 AND e.id > $2 AND e.agent_id IS NOT NULL -- 'reasoning' is NOT read here: live_bus pushes those -- as the runtime emits them, and emitting from both -- delivered every turn twice — once pushed, once polled -- ~2s later. The row is still written, as the durable -- record; this feed just is not its second mouth. AND e.kind = 'tool.call' ORDER BY e.id LIMIT 200", ) .bind(ws.as_uuid()) .bind(agent_ev_cursor) .fetch_all(&pool) .await .unwrap_or_default(); for r in &rows { let eid: i64 = r.get("id"); agent_ev_cursor = agent_ev_cursor.max(eid); let agent_id: Option = r.get("agent_id"); let Some(agent_id) = agent_id else { continue }; let kind: String = r.get("kind"); let detail: serde_json::Value = r.get("detail"); let target: Option = r.get("target"); debug_assert_eq!(kind, "tool.call", "query filters to tool.call"); let _ = kind; yield sse("agent.tool.call", json!({ "agentId": agent_id.to_string(), "tool": target.clone().unwrap_or_default(), "target": detail.get("path").and_then(|v| v.as_str()), })); } } // WORKING ON NOW — the third card that was declared, listened for, // and never emitted. `agent.task.update` has no producer anywhere in // the backend, so the card read "idle" for an agent mid-turn. // // Derived rather than newly instrumented: an agent is working on its // crew's RUNNING mission, and that mission's phases are the steps. // Nothing is emitted for an agent with no running mission, so "idle" // stays truthful instead of becoming a stale last-known task. { let rows = sqlx::query( "SELECT tm.claw_id AS agent_id, m.id AS mission_id, m.title, p.kind, p.status, p.order_idx FROM team_members tm JOIN mission_teams mt ON mt.team_id = tm.team_id JOIN missions m ON m.id = mt.mission_id JOIN mission_phases p ON p.mission_id = m.id WHERE m.workspace_id = $1 AND m.status = 'running' ORDER BY tm.claw_id, p.order_idx", ) .bind(ws.as_uuid()) .fetch_all(&pool) .await .unwrap_or_default(); let mut cur: Option<(uuid::Uuid, uuid::Uuid, String)> = None; let mut steps: Vec = Vec::new(); let mut flush = |cur: &Option<(uuid::Uuid, uuid::Uuid, String)>, steps: &Vec| -> Option { let (agent_id, mission_id, title) = cur.as_ref()?; Some(json!({ "agentId": agent_id.to_string(), "taskId": mission_id.to_string(), "title": title, "steps": steps, })) }; for r in &rows { let agent_id: uuid::Uuid = r.get("agent_id"); let mission_id: uuid::Uuid = r.get("mission_id"); let title: String = r.get("title"); if cur.as_ref().map(|c| c.0) != Some(agent_id) { if let Some(v) = flush(&cur, &steps) { yield sse("agent.task.update", v); } steps = Vec::new(); cur = Some((agent_id, mission_id, title)); } let status: String = r.get("status"); let kind: String = r.get("kind"); steps.push(json!({ "label": kind, "state": match status.as_str() { "completed" | "skipped" => "done", "running" | "evaluating" => "active", _ => "pending", }, })); } if let Some(v) = flush(&cur, &steps) { yield sse("agent.task.update", v); } } // Workspace-wide telemetry (top-bar pills / Observe system strip). yield sse( "telemetry", json!({ "doorsPending": doors_pending(&pool, ws).await, "loops": runs.len() }), ); // Edge-initiated inter-agent events (gated delegation, A2A ingress) // bypass the run loop, so surface them from the append-only audit log. // On first sight jump the cursor to the current max so we stream // forward instead of replaying history. if audit_cursor < 0 { audit_cursor = sqlx::query_scalar( "SELECT coalesce(max(id), 0) FROM audit_log WHERE workspace_id = $1", ) .bind(ws.as_uuid()) .fetch_one(&pool) .await .unwrap_or(0); } else { let rows = sqlx::query( "SELECT id, actor_id, event_type, subject_id, detail FROM audit_log WHERE workspace_id = $1 AND id > $2 AND event_type IN ('delegation.invoked', 'a2a.invoked') ORDER BY id ASC LIMIT 100", ) .bind(ws.as_uuid()) .bind(audit_cursor) .fetch_all(&pool) .await .unwrap_or_default(); for row in &rows { let id: i64 = row.get("id"); let et: String = row.get("event_type"); let actor: Option = row.get("actor_id"); let subject: String = row.get("subject_id"); let detail: Value = row.get("detail"); match et.as_str() { "delegation.invoked" => { yield sse("agent.delegate", json!({ "fromAgentId": actor.map(|u| u.to_string()).unwrap_or_default(), "toAgentId": detail.get("to_id").and_then(|v| v.as_str()).unwrap_or(""), "toName": subject, "task": detail.get("task").and_then(|v| v.as_str()).unwrap_or(""), })); } "a2a.invoked" => { // subject_id is the claw_ alias → surface the target agent. let agent_id = subject .strip_prefix("claw_") .and_then(|h| uuid::Uuid::parse_str(h).ok()) .map(|u| u.to_string()) .unwrap_or_else(|| subject.clone()); yield sse("a2a.invoked", json!({ "agentId": agent_id })); } _ => {} } if id > audit_cursor { audit_cursor = id; } } } first = false; // Forward pushed frames until the next poll is due, instead of // sleeping through them. This is what makes the reasoning stream // token-level: a chunk reaches the browser as the runtime emits it, // while everything queryable keeps its 2s cadence. let deadline = tokio::time::Instant::now() + Duration::from_secs(2); loop { let left = deadline.saturating_duration_since(tokio::time::Instant::now()); if left.is_zero() { break; } match tokio::time::timeout(left, live_rx.recv()).await { Ok(Ok(ev)) => { if ev.workspace_id == ws.as_uuid() { yield sse(&ev.kind, ev.data); } } // Lagged: this subscriber fell behind and frames were // dropped for it. Keep going — a live view that skips is // right, and blocking the producer would be wrong. Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue, Ok(Err(_)) => break, Err(_) => break, } } } }; Sse::new(stream).keep_alive(KeepAlive::default()) } #[derive(Deserialize)] pub struct ReplayQuery { hours: Option, } /// `GET /api/world/replay?hours=24` — a Gource-style timeline reconstructed from /// the workspace's run history: a sorted list of timestamped taxonomy events the /// client's WorldClock plays back into the same engine (live + replay). pub async fn world_replay( State(state): State, Authed(user): Authed, Query(q): Query, ) -> Result, ApiError> { let hours = q.hours.unwrap_or(24).clamp(1, 720); let rows = sqlx::query( "SELECT ar.id::text AS run_id, s.agent_id::text AS agent_id, extract(epoch FROM ar.created_at)::float8 AS started FROM agent_runs ar JOIN sessions s ON s.id = ar.session_id WHERE s.workspace_id = $1 AND ar.created_at > now() - ($2 * interval '1 hour') ORDER BY ar.created_at ASC", ) .bind(user.workspace_id.as_uuid()) .bind(hours) .fetch_all(&state.pool) .await?; let mut events: Vec = Vec::new(); for r in &rows { let run_id: String = r.get("run_id"); let agent_id: String = r.get("agent_id"); let started: f64 = r.get("started"); let node_id = format!("run:{}", &run_id[..run_id.len().min(8)]); events.push(json!({ "t": started, "type": "agent.status", "data": { "agentId": agent_id, "status": "working" }})); events.push(json!({ "t": started, "type": "node.activity", "data": { "nodeId": node_id, "label": "run", "kind": "event", "heat": 0.85 }})); events.push(json!({ "t": started, "type": "world.touch", "data": { "agentId": agent_id, "nodeId": node_id, "kind": "event" }})); } Ok(Json( json!({ "events": events, "hours": hours, "count": rows.len() }), )) } #[cfg(test)] mod mission_feed_tests { /// The mission query must not re-acquire any of the three bugs it fixed. /// /// All three were invisible: an INNER JOIN does not error when it drops a /// row, the legacy `missions.team_id` is a real column that resolves fine, /// and `status = 'running'` is a defensible-looking filter. The World simply /// showed less than the truth and nothing anywhere said so. A source walk is /// the only guard available — these are runtime `sqlx::query` calls, so the /// compiler checks nothing about them. #[test] fn the_mission_query_keeps_teamless_missions() { let src = include_str!("world.rs"); let q = src .split("async fn world_missions") .nth(1) .and_then(|s| s.split("async fn").next()) .expect("world_missions body"); assert!( q.contains("LEFT JOIN mission_teams"), "must LEFT JOIN mission_teams: a microVM mission has no team rows at \ all, and an INNER JOIN silently drops the platform's primary \ execution tier" ); assert!( q.contains("LEFT JOIN team_members"), "the second join must be LEFT too, or the mission reappears and its \ agents take it back out" ); assert!( !q.contains("tm.team_id = m.team_id"), "must not join on the legacy missions.team_id — 0056 superseded it \ with mission_teams, and it points at only the FIRST minted team" ); assert!( q.contains("m.completed_at >"), "recently-finished missions must be included or the World is empty \ almost always: missions finish in minutes" ); assert!( q.contains("m.status") && q.contains("m.template_kind"), "status and template_kind must be carried: one decides finished-map \ vs live, the other the palette" ); } /// The checkpoint tail read `topology_runs` keyed by an `agent_runs` id and /// could never match. If it comes back, missions silently lose their tool /// and file detail again — and the query looks entirely reasonable. #[test] fn the_dead_checkpoint_tail_stays_dead() { let src = include_str!("world.rs"); // Split so this assertion's own literal is not what the source walk // finds. Written whole, the test fails on itself — which it did, and // which is the same self-match that makes `pkill -f ` kill the // shell carrying the pattern. let needle = concat!( "SELECT checkpoint FROM topology_", "runs WHERE id = $1::uuid" ); assert!( !src.contains(needle), "the checkpoint tail is keyed on an agent_runs id and cannot match a \ topology_runs row; mission detail comes from structured events" ); } /// A benchmark delta is described only as far as its shape allows. /// /// `compute_delta` emits `{kind:"opaque"}` whenever the before/after /// metrics were not structurally comparable — which is most drivers. The /// temptation is to say something anyway; the result would be a performance /// claim the picture makes and the data does not support. #[test] fn a_benchmark_is_described_only_as_far_as_its_shape_allows() { use serde_json::json; assert_eq!( super::benchmark_note(&json!({ "kind": "bencher_diff", "samples": [ {"name": "a", "delta_pct": -12.5, "direction": "improved"}, {"name": "b", "delta_pct": 3.0, "direction": "regressed"}, // No direction and no percentage: counted, claimed for // neither side. {"name": "c"}, ], })) .as_deref(), Some("3 sample(s), 1 faster, 1 slower, best -12.5%") ); // Nothing comparable happened, so nothing is said. assert_eq!( super::benchmark_note(&json!({ "kind": "opaque", "note": "not comparable" })), None ); assert_eq!( super::benchmark_note(&json!({ "kind": "bencher_diff", "samples": [] })), None ); // Everything got slower: no "best" claim at all. assert_eq!( super::benchmark_note(&json!({ "samples": [{"name": "a", "delta_pct": 8.0, "direction": "regressed"}], })) .as_deref(), Some("1 sample(s), 1 slower") ); } /// Agent→phase attribution must come from the runner's own mapping. #[test] fn phase_attribution_reuses_the_runners_mapping() { let src = include_str!("world.rs"); assert!( src.contains("crate::phase_runner::purposes_for"), "a second copy of the kind→purpose mapping would let the picture \ disagree with the machine about who is working on what" ); } /// Files must be emitted after the phases they hang under. /// /// The client parents a file's directory chain to the coding station, and /// parenting is first-write-wins — so a file that reaches the browser /// before its plan pins its whole tree at the origin permanently. Nothing /// errors: the tree renders, in the wrong place, and reads as a layout /// choice. Swapping the two loops back is a one-line-looking edit, which is /// exactly why it needs a guard. #[test] fn files_are_emitted_after_the_phases_they_hang_under() { let src = include_str!("world.rs"); let phases = src .find("for p in world_phases(") .expect("the phase emission loop"); let files = src .find("for f in world_files(") .expect("the file emission loop"); assert!( phases < files, "the phase loop must run before the file loop; files parented \ before their coding station stay at the origin forever" ); } }