World: Phase C — replay scrubber (Gource-style playback)
ci / gates (push) Has been cancelled
ci / rust (push) Has been cancelled
ci / sandbox-k8s (push) Has been cancelled
ci / frontend (push) Has been cancelled
ci / e2e (push) Has been cancelled

- /api/world/replay?hours= — reconstructs a sorted, timestamped taxonomy timeline
  from the workspace's run history (agent_runs ⋈ sessions) for client playback.
- engine.clearWorldNodes() — drop transient world nodes/targets for a loop restart.
- WorldCanvas: a WorldClock control (Live ⇄ Replay) feeding the same engine —
  play/pause, 1x/2x/4x speed, seek bar, loop. Live subscriptions detach in replay.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Omar Sobh
2026-06-23 21:39:03 -07:00
co-authored by Claude Opus 4.8
parent 05c54de47d
commit 53279e6339
4 changed files with 234 additions and 6 deletions
+43 -2
View File
@@ -15,14 +15,16 @@ use std::collections::HashSet;
use std::convert::Infallible;
use std::time::Duration;
use axum::extract::State;
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::{AppState, Authed};
use crate::{ApiError, AppState, Authed};
fn sse(event: &str, data: Value) -> Result<Event, Infallible> {
Ok(Event::default().event(event).data(data.to_string()))
@@ -134,6 +136,45 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
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() })))
}
// THE NORMALIZE SEAM (future) -------------------------------------------------
// Translate one durable `run_events` row into zero+ taxonomy events, the Rust
// twin of the handoff bridge's normalize(). Wire this into the poll loop above