Omar Sobh
d8c8793c4a
CI on 6ffbe97 failed on two auto-fixable gates. Both fixed:
* cargo fmt --all — rustfmt applied across the surface touched
by the last ~20 commits (world.rs, security_scan.rs,
routes/{missions,nodes,terminal}.rs, fleet_herdr.rs,
mission_workspace.rs, benchmark_runner.rs, mission_refiner.rs,
lib.rs, tests/mission_orchestrator.rs, cm-db/repo/{missions,teams}.rs,
bins/clawmates-node/src/main.rs)
* eslint apostrophe escapes in HerdrSessions + MissionWizard
* eslint max-lines: extracted EditMissionModal + RefineDiffModal
(each ~200 LoC) into their own files. MissionCanvas drops from
1424 to 1026, comfortably under both the 1250 eslint cap and the
1500 CI budget.
New files:
frontend/src/components/dashboard/EditMissionModal.tsx (211 LoC)
frontend/src/components/dashboard/RefineDiffModal.tsx (208 LoC)
Verified locally: cargo fmt --check clean, cargo check clean,
mission_orchestrator test 3/3 pass, tsc + eslint --quiet both silent.
589 lines
24 KiB
Rust
589 lines
24 KiB
Rust
//! `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 crate::{ApiError, AppState, Authed};
|
||
|
||
fn sse(event: &str, data: Value) -> Result<Event, Infallible> {
|
||
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<String> {
|
||
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",
|
||
)
|
||
.bind(ws.as_uuid())
|
||
.fetch_all(pool)
|
||
.await
|
||
.unwrap_or_default();
|
||
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)> {
|
||
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()
|
||
}
|
||
|
||
/// 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:<path>` 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::<Vec<_>>()
|
||
})
|
||
.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<String, AgentTele> {
|
||
let mut m: std::collections::HashMap<String, AgentTele> = std::collections::HashMap::new();
|
||
let id_of = |r: &sqlx::postgres::PgRow| r.get::<uuid::Uuid, _>("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
|
||
}
|
||
|
||
/// `GET /api/world/live` — the taxonomy SSE feed.
|
||
pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) -> impl IntoResponse {
|
||
let pool = state.pool.clone();
|
||
let ws = user.workspace_id;
|
||
|
||
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<String, String> = std::collections::HashMap::new();
|
||
// Per-run journal cursor so we stream only NEW run_events each poll.
|
||
let mut cursors: std::collections::HashMap<String, i64> = 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;
|
||
// 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<String, u64> = std::collections::HashMap::new();
|
||
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<Value> = 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: 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)]);
|
||
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.
|
||
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),
|
||
}));
|
||
}
|
||
|
||
// 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<uuid::Uuid> = 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_<id> 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;
|
||
tokio::time::sleep(Duration::from_secs(2)).await;
|
||
}
|
||
};
|
||
|
||
Sse::new(stream).keep_alive(KeepAlive::default())
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
pub struct ReplayQuery {
|
||
hours: Option<i64>,
|
||
}
|
||
|
||
/// `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<AppState>,
|
||
Authed(user): Authed,
|
||
Query(q): Query<ReplayQuery>,
|
||
) -> Result<Json<Value>, 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<Value> = 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() }),
|
||
))
|
||
}
|