//! `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()
}

/// 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()
}

/// Active research topics + their assigned agents. Each returned row is a
/// `(topic_id, title, agent_id, repo_workspace_path)` — one row per
/// (topic, agent) pair. Emitted from the SSE loop as `repo:<topic_id>`
/// project orbs so the World shows a clickable, labeled landmark for
/// every in-flight R&D initiative — no need for a file touch to land
/// first. `repo_workspace_path` (when non-null) is the on-disk clone
/// location; the SSE loop uses it to pre-seed the repo tree.
async fn active_research_topics(
    pool: &PgPool,
    ws: WorkspaceId,
) -> Vec<(String, String, String, Option<String>)> {
    let rows = sqlx::query(
        "SELECT t.id::text AS topic_id,
                t.title AS title,
                t.repo_workspace_path AS repo_path,
                ra.agent_id::text AS agent_id
           FROM research_topics t
           JOIN research_topic_agents ra ON ra.topic_id = t.id
          WHERE t.workspace_id = $1
            AND t.status IN ('processing', 'reviewing', 'publishing')",
    )
    .bind(ws.as_uuid())
    .fetch_all(pool)
    .await
    .unwrap_or_default();
    rows.into_iter()
        .map(|r| {
            (
                r.get::<String, _>("topic_id"),
                r.get::<String, _>("title"),
                r.get::<String, _>("agent_id"),
                r.try_get::<Option<String>, _>("repo_path").unwrap_or(None),
            )
        })
        .collect()
}

/// Cap on pre-seeded file entries per repo. Large repos surface only the
/// top N so the SSE payload stays bounded — a subsequent tool call
/// exercising a specific path will fill in additional nodes on demand.
const REPO_PRESEED_CAP: usize = 200;

/// Read the top-level file list of a topic's cloned repo via `git ls-files`
/// so the SSE loop can pre-seed dir:/file: nodes in the client engine.
/// Bounded by `REPO_PRESEED_CAP`. Returns an empty vec on any failure
/// (missing clone, git not on PATH, empty repo) — a missing pre-seed
/// degrades gracefully to the pre-V3 behavior (tree builds as agents
/// touch files).
async fn preseed_repo_paths(clone_path: &str) -> Vec<String> {
    let path = std::path::Path::new(clone_path);
    if !path.join(".git").exists() {
        return Vec::new();
    }
    let out = tokio::process::Command::new("git")
        .arg("-C")
        .arg(path)
        .arg("ls-files")
        .output()
        .await;
    let Ok(out) = out else { return Vec::new() };
    if !out.status.success() {
        return Vec::new();
    }
    String::from_utf8_lossy(&out.stdout)
        .lines()
        .filter(|l| !l.trim().is_empty())
        .take(REPO_PRESEED_CAP)
        .map(|s| s.to_string())
        .collect()
}

/// Enabled scheduled loops + their assigned agents. Same shape as
/// `active_research_topics` — `(loop_id, title, agent_id)` per (loop, agent).
/// Emitted as `loop:<loop_id>` landmark orbs so recurring/scheduled work is
/// visible in the World at all times, not just while a run is mid-flight.
/// Contrast with research topics (transient statuses processing/reviewing/
/// publishing) — loops are persistent landmarks the user can click.
async fn active_loops(pool: &PgPool, ws: WorkspaceId) -> Vec<(String, String, String)> {
    let rows = sqlx::query(
        "SELECT l.id::text AS loop_id, l.title AS title, la.agent_id::text AS agent_id
           FROM loops l
           JOIN loop_agents la ON la.loop_id = l.id
          WHERE l.workspace_id = $1
            AND l.enabled = TRUE",
    )
    .bind(ws.as_uuid())
    .fetch_all(pool)
    .await
    .unwrap_or_default();
    rows.into_iter()
        .map(|r| {
            (
                r.get::<String, _>("loop_id"),
                r.get::<String, _>("title"),
                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 }));
                }
            }

            // Active research topics → landmark project orbs. One `repo:<id>`
            // per topic, labeled with the topic title so users can click it
            // and drop into the repo-focus (Gource) view before any files are
            // touched. Assigned agents gently converge on their topic's orb
            // so the affinity is visible even in idle windows.
            let research = active_research_topics(&pool, ws).await;
            let mut seen_topics = std::collections::HashSet::new();
            for (topic_id, title, agent_id, repo_path) in &research {
                let node_id = format!("repo:{topic_id}");
                if seen_topics.insert(topic_id.clone()) {
                    yield sse(
                        "node.activity",
                        json!({ "nodeId": node_id, "label": title, "kind": "service", "heat": 0.0 }),
                    );
                    // Pre-seed the repo tree (V3). One-shot on first sight
                    // of the topic per SSE client. Each file emits with
                    // heat=0 so the tree is quiet-solid at rest — activity
                    // still hot-swaps as agents touch files. Bounded to
                    // REPO_PRESEED_CAP so payload stays reasonable.
                    if let Some(clone_path) = repo_path {
                        for p in preseed_repo_paths(clone_path).await {
                            let leaf = std::path::Path::new(&p)
                                .file_name()
                                .and_then(|s| s.to_str())
                                .unwrap_or(&p)
                                .to_string();
                            yield sse(
                                "node.activity",
                                json!({
                                    "nodeId": format!("file:{p}"),
                                    "label": leaf,
                                    "kind": "service",
                                    "heat": 0.0,
                                }),
                            );
                        }
                    }
                }
                yield sse(
                    "world.touch",
                    json!({ "agentId": agent_id, "nodeId": node_id, "kind": "service", "weight": 0.15 }),
                );
            }

            // Scheduled loops → landmark orbs, symmetric to research topics.
            // Persistent landmarks: emitted whenever a loop is enabled, so a
            // loop between fires still reads as an in-flight project. When
            // a loop actually runs, the topology_worker journals events
            // which the run-cursor block below picks up and heats the orb.
            let loops = active_loops(&pool, ws).await;
            let mut seen_loops = std::collections::HashSet::new();
            for (loop_id, title, agent_id) in &loops {
                let node_id = format!("loop:{loop_id}");
                if seen_loops.insert(loop_id.clone()) {
                    yield sse(
                        "node.activity",
                        json!({ "nodeId": node_id, "label": title, "kind": "service", "heat": 0.0 }),
                    );
                }
                yield sse(
                    "world.touch",
                    json!({ "agentId": agent_id, "nodeId": node_id, "kind": "service", "weight": 0.15 }),
                );
            }

            // 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() }),
    ))
}

// 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
// once the runner emits node targets:
//   turn.started   -> agent.status(working) [+ agent.task.update]
//   turn.token     -> agent.reasoning.delta
//   tool.invoked   -> agent.tool.call [+ world.touch if a nodeId is present]
//   door.requested -> door.request ; door.resolved -> door.resolve
//   agent.message  -> agent.message ; runner.telemetry -> telemetry
#[allow(dead_code)]
fn normalize(_event_type: &str, _payload: &Value) -> Vec<(&'static str, Value)> {
    Vec::new()
}