//! Agent run tracing — replay read RPCs `runs.list` / `runs.get_trace`
//! (U5).
//!
//! Drives `runs.start` / `proposal.submit` / `runs.complete` over real
//! `run_dispatch` WebSocket sessions, then exercises the two WS-only
//! replay reads — `runs.list {agent_id}` and `runs.get_trace {run_id}` —
//! reading the disk-backed `RunStore` (U3), not live session state. The
//! reads are authorized by the same `authorize_run_access` gate U4 uses
//! for `runs.subscribe`. Covered (mapping to the U5 plan scenarios):
//!
//! 1. `runs.list` returns an agent's runs newest-first with correct
//!    statuses (R9 happy path).
//! 2. `runs.get_trace` returns a completed run's full ordered trail
//!    identical to what the live `runs.subscribe` stream delivered (R9 —
//!    live/replay parity).
//! 3. A caller not entitled to the `agent_id` / `run_id` is rejected, not
//!    served (R16 — the authorization guard), for BOTH reads.
//! 4. Both reads succeed after a simulated daemon restart — a fresh
//!    `ServerState` with empty memory over the same `~/.car` (R4).
//! 5. `runs.get_trace` of an `Incomplete` (mid-run-drop) run returns the
//!    partial trail + the terminal `Incomplete` marker, not an error
//!    (edge / R5 / R11).
//! 6. `runs.list` for an agent with zero runs returns an empty list (the
//!    empty state — R13).
//! 7. `runs.get_trace` with an unknown `run_id` returns a clear not-found
//!    marker (`{ not_found: true }`), not an error frame (error path).
//!
//! The reads are WS-only (no FFI surface — like `runs.subscribe`), so the
//! contract is the JSON-RPC shape exercised here.

use car_memgine::MemgineEngine;
use car_proto::{RunRecord, RunTurn};
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::{accept_async, connect_async, tungstenite::Message};

type Ws =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

/// Build a `ServerState` whose journal dir is `<car_dir>/journals` — so
/// the run store roots at the sibling `<car_dir>/runs`, exactly as the
/// daemon resolves it. A fresh `ServerState` built over the same `car_dir`
/// reads the same on-disk run store (the restart simulation).
/// Per-launch host token the harness installs so `become_host_client`
/// can take the host-management role — the credential
/// `authorize_run_access` now requires (Parslee-ai/car#254). A "restart"
/// (fresh `ServerState` over the same `car_dir`) reinstalls the SAME
/// token so the host path keeps working across the simulated restart.
const TEST_HOST_TOKEN: &str = "test-host-token-0123456789abcdef0123456789";

fn state_over(car_dir: &Path) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cfg = ServerStateConfig::new(car_dir.join("journals")).with_shared_memgine(engine);
    let state = Arc::new(ServerState::with_config(cfg));
    let _ = state.install_host_token(TEST_HOST_TOKEN.to_string());
    state
}

/// Spin up a listener that accepts MANY connections, each driven by
/// `run_dispatch` against the shared `state` (a harness, a host-client
/// reader, an intruder — distinct live connections).
async fn spawn_multi_dispatcher(state: Arc<ServerState>) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(
        Ipv4Addr::new(127, 0, 0, 1),
        0,
    )))
    .await
    .expect("bind loopback");
    let addr = listener.local_addr().expect("local_addr");
    tokio::spawn(async move {
        loop {
            let (stream, peer) = match listener.accept().await {
                Ok(v) => v,
                Err(_) => break,
            };
            let state = state.clone();
            tokio::spawn(async move {
                let ws = match accept_async(stream).await {
                    Ok(ws) => ws,
                    Err(_) => return,
                };
                let (write, read) = ws.split();
                let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
            });
        }
    });
    addr
}

async fn connect(addr: SocketAddr) -> Ws {
    let (ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();
    ws
}

async fn send(ws: &mut Ws, id: &str, method: &str, params: serde_json::Value) {
    ws.send(Message::Text(
        serde_json::json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })
            .to_string()
            .into(),
    ))
    .await
    .expect("send request");
}

async fn next_json(ws: &mut Ws) -> serde_json::Value {
    let text = ws
        .next()
        .await
        .expect("frame")
        .expect("frame ok")
        .into_text()
        .expect("text")
        .to_string();
    serde_json::from_str(&text).expect("parse json")
}

/// Read frames on `ws` until the response to `id` arrives, returning it.
/// Any `runs.trace.event` notifications seen along the way are pushed onto
/// `events` (a subscriber's stream is interleaved with its RPC acks).
async fn call_collecting_events(
    ws: &mut Ws,
    id: &str,
    method: &str,
    params: serde_json::Value,
    events: &mut Vec<serde_json::Value>,
) -> serde_json::Value {
    send(ws, id, method, params).await;
    loop {
        let msg = next_json(ws).await;
        if msg.get("method").and_then(|m| m.as_str()) == Some("runs.trace.event") {
            events.push(msg["params"].clone());
            continue;
        }
        if msg.get("id").and_then(|v| v.as_str()) == Some(id) {
            return msg;
        }
    }
}

/// Plain request expecting a `result`, no callbacks.
async fn call_ok(
    ws: &mut Ws,
    id: &str,
    method: &str,
    params: serde_json::Value,
) -> serde_json::Value {
    send(ws, id, method, params).await;
    let resp = next_json(ws).await;
    assert!(resp.get("error").is_none(), "{method} failed: {resp}");
    resp["result"].clone()
}

/// Service a `proposal.submit`, replying to the daemon's inbound
/// `tools.execute` callback with `tool_output`. Returns the response.
async fn submit_servicing_callbacks(
    ws: &mut Ws,
    id: &str,
    params: serde_json::Value,
    tool_output: &serde_json::Value,
) -> serde_json::Value {
    send(ws, id, "proposal.submit", params).await;
    loop {
        let msg = next_json(ws).await;
        if msg.get("method").and_then(|m| m.as_str()) == Some("tools.execute") {
            let cb_id = msg["id"].as_str().expect("callback id").to_string();
            ws.send(Message::Text(
                serde_json::json!({ "jsonrpc": "2.0", "id": cb_id, "result": tool_output })
                    .to_string()
                    .into(),
            ))
            .await
            .expect("send callback reply");
            continue;
        }
        if msg.get("id").and_then(|v| v.as_str()) == Some(id) {
            return msg;
        }
    }
}

async fn register_tools(ws: &mut Ws, names: &[&str]) {
    let tools: Vec<serde_json::Value> = names
        .iter()
        .map(|n| serde_json::json!({ "name": n, "description": "", "parameters": {} }))
        .collect();
    send(ws, "reg", "tools.register", serde_json::json!(tools)).await;
    let resp = next_json(ws).await;
    assert!(resp.get("error").is_none(), "tools.register failed: {resp}");
}

async fn start_run(ws: &mut Ws, agent_id: &str, intent: &str) -> String {
    let r = call_ok(
        ws,
        "start",
        "runs.start",
        serde_json::json!({ "agent_id": agent_id, "intent": intent }),
    )
    .await;
    r["run_id"].as_str().unwrap().to_string()
}

async fn complete_run(ws: &mut Ws, run_id: &str) {
    let r = call_ok(
        ws,
        "done",
        "runs.complete",
        serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
    )
    .await;
    assert_eq!(r["ok"], true);
}

/// One `drive_cli` proposal with a unique action_id + prompt.
fn drive_proposal(action_id: &str, prompt: &str) -> serde_json::Value {
    serde_json::json!({
        "proposal": {
            "id": format!("p-{action_id}"),
            "source": "test",
            "actions": [{
                "id": action_id,
                "type": "tool_call",
                "tool": "drive_cli",
                "parameters": { "cli": "claude", "prompt": prompt }
            }]
        }
    })
}

fn drive_output() -> serde_json::Value {
    serde_json::json!({
        "cli": "claude", "exit_code": 0, "timed_out": false, "output_tail": "ok"
    })
}

fn success_outcome() -> serde_json::Value {
    serde_json::json!({
        "status": "success",
        "summary": "did it",
        "evidence": [],
        "metrics": {
            "turns": 1, "tool_calls": 1, "duration_ms": 1.0,
            "retries": 0, "actions_succeeded": 1, "actions_failed": 0
        },
        "timestamp": chrono::Utc::now().to_rfc3339()
    })
}

/// Make `ws` a CarHost host-client (the management surface entitled to any
/// managed agent's runs). Mirrors the U4 live-trace tests.
///
/// #254: host-role now requires the per-launch host token — merely
/// calling `host.subscribe` no longer grants run-trace access. We present
/// the token (the CarHost path) and keep the `host.subscribe` call for
/// fidelity with how the real client also subscribes to host events.
async fn become_host_client(ws: &mut Ws) {
    let auth = call_ok(
        ws,
        "ha",
        "session.auth",
        serde_json::json!({ "host_token": TEST_HOST_TOKEN }),
    )
    .await;
    assert_eq!(
        auth["role"], "host",
        "host token must grant the host role: {auth}"
    );
    let _ = call_ok(ws, "hs", "host.subscribe", serde_json::json!({})).await;
}

/// Extract the ordered `RunTurn`s from a `get_trace` `records` array.
fn turns_of(records: &serde_json::Value) -> Vec<RunTurn> {
    records
        .as_array()
        .expect("records is an array")
        .iter()
        .filter_map(
            |r| match serde_json::from_value::<RunRecord>(r.clone()).ok()? {
                RunRecord::Turn(t) => Some(t),
                _ => None,
            },
        )
        .collect()
}

// --- Test 1: runs.list newest-first with correct statuses. -------------

#[tokio::test]
async fn list_returns_runs_newest_first_with_statuses() {
    let tmp = TempDir::new().unwrap();
    let state = state_over(tmp.path());
    let addr = spawn_multi_dispatcher(state.clone()).await;

    // Two completed runs on the same agent, started in order.
    let mut harness = connect(addr).await;
    register_tools(&mut harness, &["drive_cli"]).await;

    let run1 = start_run(&mut harness, "agent-a", "first goal").await;
    submit_servicing_callbacks(
        &mut harness,
        "p1",
        drive_proposal("a1", "do-1"),
        &drive_output(),
    )
    .await;
    complete_run(&mut harness, &run1).await;

    // A tiny gap so started_at strictly orders run2 after run1.
    tokio::time::sleep(Duration::from_millis(5)).await;
    let run2 = start_run(&mut harness, "agent-a", "second goal").await;
    submit_servicing_callbacks(
        &mut harness,
        "p2",
        drive_proposal("a2", "do-2a"),
        &drive_output(),
    )
    .await;
    submit_servicing_callbacks(
        &mut harness,
        "p3",
        drive_proposal("a3", "do-2b"),
        &drive_output(),
    )
    .await;
    complete_run(&mut harness, &run2).await;

    // A host-client lists the agent's runs.
    let mut host = connect(addr).await;
    become_host_client(&mut host).await;
    let resp = call_ok(
        &mut host,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "agent-a" }),
    )
    .await;

    assert_eq!(resp["agent_id"], "agent-a");
    let runs = resp["runs"].as_array().unwrap();
    assert_eq!(runs.len(), 2, "both runs listed");
    // Newest first: run2 (started later) before run1.
    assert_eq!(runs[0]["run_id"], run2);
    assert_eq!(runs[1]["run_id"], run1);
    assert_eq!(runs[0]["status"], "completed");
    assert_eq!(runs[1]["status"], "completed");
    assert_eq!(runs[0]["intent"], "second goal");
    assert_eq!(runs[1]["intent"], "first goal");
    assert_eq!(
        runs[0]["turn_count"].as_u64().unwrap(),
        2,
        "run2 has two turns"
    );
    assert_eq!(
        runs[1]["turn_count"].as_u64().unwrap(),
        1,
        "run1 has one turn"
    );
}

// --- Test 2: get_trace == the live stream's ordered trail. --------------

#[tokio::test]
async fn get_trace_matches_live_stream_trail() {
    let tmp = TempDir::new().unwrap();
    let state = state_over(tmp.path());
    let addr = spawn_multi_dispatcher(state.clone()).await;

    let mut harness = connect(addr).await;
    register_tools(&mut harness, &["drive_cli"]).await;
    let run_id = start_run(&mut harness, "agent-a", "trace me").await;

    // A host-role client subscribes live and captures the streamed turns.
    let mut sub = connect(addr).await;
    let mut live = Vec::new();
    become_host_client(&mut sub).await; // #254: token-gated host role
    call_collecting_events(
        &mut sub,
        "sub",
        "runs.subscribe",
        serde_json::json!({ "run_id": run_id }),
        &mut live,
    )
    .await;

    // Drive three turns, then complete.
    for (i, p) in ["alpha", "beta", "gamma"].iter().enumerate() {
        submit_servicing_callbacks(
            &mut harness,
            &format!("p{i}"),
            drive_proposal(&format!("a{i}"), p),
            &drive_output(),
        )
        .await;
    }
    // Collect the three live turn events.
    while live
        .iter()
        .filter(|e| {
            matches!(
                serde_json::from_value::<RunRecord>(e["record"].clone()),
                Ok(RunRecord::Turn(_))
            )
        })
        .count()
        < 3
    {
        let frame = tokio::time::timeout(Duration::from_secs(2), next_json(&mut sub))
            .await
            .expect("live turn event");
        if frame.get("method").and_then(|m| m.as_str()) == Some("runs.trace.event") {
            live.push(frame["params"].clone());
        }
    }
    complete_run(&mut harness, &run_id).await;

    let live_turns: Vec<RunTurn> = live
        .iter()
        .filter_map(
            |e| match serde_json::from_value::<RunRecord>(e["record"].clone()).ok()? {
                RunRecord::Turn(t) => Some(t),
                _ => None,
            },
        )
        .collect();
    assert_eq!(live_turns.len(), 3, "live stream delivered three turns");

    // Now replay via get_trace — the persisted trail must carry the same
    // ordered turns the live stream delivered (R9 live/replay parity).
    // The `sub` connection still receives `runs.trace.event` notifications
    // (e.g. the terminal Ended), so read past them to the RPC response.
    let trace_resp = call_collecting_events(
        &mut sub,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
        &mut live,
    )
    .await;
    let trace = trace_resp["result"].clone();
    assert_eq!(trace["run_id"], run_id);
    assert_eq!(trace["agent_id"], "agent-a");
    assert_eq!(
        trace["cursor"].as_u64().unwrap(),
        0,
        "full fetch starts at cursor 0"
    );

    // The persisted trail is bracketed Started ... Ended.
    let records = trace["records"].as_array().unwrap();
    assert!(matches!(
        serde_json::from_value::<RunRecord>(records.first().unwrap().clone()).unwrap(),
        RunRecord::Started(_)
    ));
    assert!(matches!(
        serde_json::from_value::<RunRecord>(records.last().unwrap().clone()).unwrap(),
        RunRecord::Ended(_)
    ));

    let replay_turns = turns_of(&trace["records"]);
    assert_eq!(replay_turns.len(), 3, "replay carries all three turns");
    for (live, replay) in live_turns.iter().zip(replay_turns.iter()) {
        assert_eq!(live.index, replay.index, "turn index parity");
        assert_eq!(live.prompt, replay.prompt, "turn prompt parity");
        assert_eq!(
            live, replay,
            "the replayed turn is identical to the streamed turn"
        );
    }

    // Paging: cursor past the Started record drops it.
    let paged_resp = call_collecting_events(
        &mut sub,
        "gt2",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id, "cursor": 1 }),
        &mut live,
    )
    .await;
    let paged = paged_resp["result"].clone();
    assert_eq!(paged["cursor"].as_u64().unwrap(), 1);
    let paged_recs = paged["records"].as_array().unwrap();
    assert_eq!(
        paged_recs.len(),
        records.len() - 1,
        "cursor 1 skips one record"
    );
    assert!(!matches!(
        serde_json::from_value::<RunRecord>(paged_recs.first().unwrap().clone()).unwrap(),
        RunRecord::Started(_)
    ));
}

// --- Test 3: unentitled caller rejected for BOTH reads (R16). ----------

#[tokio::test]
async fn unentitled_caller_is_rejected_on_both_reads() {
    let tmp = TempDir::new().unwrap();
    let state = state_over(tmp.path());
    let addr = spawn_multi_dispatcher(state.clone()).await;

    let mut harness = connect(addr).await;
    register_tools(&mut harness, &["drive_cli"]).await;
    let run_id = start_run(&mut harness, "agent-a", "private").await;
    submit_servicing_callbacks(
        &mut harness,
        "p1",
        drive_proposal("a1", "secret"),
        &drive_output(),
    )
    .await;
    complete_run(&mut harness, &run_id).await;

    // A bare authenticated connection — NEITHER the owning agent (no
    // session.auth {agent_id} == agent-a) NOR a host-client (no
    // host.subscribe). Both reads must be rejected.
    let mut intruder = connect(addr).await;

    send(
        &mut intruder,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "agent-a" }),
    )
    .await;
    let list_resp = next_json(&mut intruder).await;
    assert!(
        list_resp.get("error").is_some(),
        "runs.list by an unentitled caller must be rejected, got {list_resp}"
    );
    assert!(
        list_resp["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .contains("not authorized"),
        "rejection should cite authorization: {list_resp}"
    );

    send(
        &mut intruder,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let trace_resp = next_json(&mut intruder).await;
    assert!(
        trace_resp.get("error").is_some(),
        "runs.get_trace by an unentitled caller must be rejected, got {trace_resp}"
    );
    assert!(
        trace_resp["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .contains("not authorized"),
        "rejection should cite authorization: {trace_resp}"
    );
    // Crucially, the secret prompt was NOT leaked in the rejection.
    assert!(
        !trace_resp.to_string().contains("secret"),
        "an unauthorized rejection must not leak the run's prompt"
    );

    // #254 regression: self-elevation via host.subscribe (WITHOUT the host
    // token) must NOT unlock another agent's run traces. This is the exact
    // headline of the issue — "runs.list even returns every run summary for
    // any caller-named agent_id."
    let mut faux_host = connect(addr).await;
    let _ = call_ok(
        &mut faux_host,
        "hs",
        "host.subscribe",
        serde_json::json!({}),
    )
    .await;
    send(
        &mut faux_host,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "agent-a" }),
    )
    .await;
    let faux_list = next_json(&mut faux_host).await;
    assert!(
        faux_list.get("error").is_some()
            && faux_list["error"]["message"]
                .as_str()
                .unwrap_or_default()
                .contains("not authorized"),
        "host.subscribe without the host token must NOT authorize runs.list \
         (Parslee-ai/car#254 self-elevation), got {faux_list}"
    );

    // A genuine host-role client (presented the host token) IS entitled.
    let mut host = connect(addr).await;
    become_host_client(&mut host).await;
    let ok_list = call_ok(
        &mut host,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "agent-a" }),
    )
    .await;
    assert_eq!(
        ok_list["runs"].as_array().unwrap().len(),
        1,
        "host-client lists the run"
    );
    let ok_trace = call_ok(
        &mut host,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    assert_eq!(ok_trace["run_id"], run_id, "host-client fetches the trace");
}

// --- Test 4: both reads survive a simulated daemon restart (R4). -------

#[tokio::test]
async fn reads_survive_daemon_restart() {
    let tmp = TempDir::new().unwrap();
    let car_dir = tmp.path().to_path_buf();

    // --- First daemon: a completed run, then it goes away. ---
    let run_id = {
        let state = state_over(&car_dir);
        let addr = spawn_multi_dispatcher(state.clone()).await;
        let mut harness = connect(addr).await;
        register_tools(&mut harness, &["drive_cli"]).await;
        let run_id = start_run(&mut harness, "agent-a", "persist me").await;
        submit_servicing_callbacks(
            &mut harness,
            "p1",
            drive_proposal("a1", "write to disk"),
            &drive_output(),
        )
        .await;
        complete_run(&mut harness, &run_id).await;
        harness.close(None).await.ok();
        drop(harness);
        // Let the disconnect sweep settle (must NOT clobber the terminal).
        tokio::time::sleep(Duration::from_millis(400)).await;
        run_id
        // `state` and its listener drop here — the daemon is "gone".
    };

    // --- Restart: a brand-new ServerState with empty memory over the
    // same ~/.car. The reads must serve entirely from disk. ---
    let state2 = state_over(&car_dir);
    let addr2 = spawn_multi_dispatcher(state2.clone()).await;

    let mut host = connect(addr2).await;
    become_host_client(&mut host).await;

    let listed = call_ok(
        &mut host,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "agent-a" }),
    )
    .await;
    let runs = listed["runs"].as_array().unwrap();
    assert_eq!(
        runs.len(),
        1,
        "the completed run lists from disk after restart"
    );
    assert_eq!(runs[0]["run_id"], run_id);
    assert_eq!(runs[0]["status"], "completed");

    let trace = call_ok(
        &mut host,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    assert_eq!(trace["run_id"], run_id);
    assert_eq!(trace["agent_id"], "agent-a");
    let replay_turns = turns_of(&trace["records"]);
    assert_eq!(
        replay_turns.len(),
        1,
        "the persisted turn replays after restart"
    );
    assert_eq!(
        replay_turns[0].prompt.as_deref(),
        Some("write to disk"),
        "the exact prompt survives the restart"
    );
}

// --- Test 5: get_trace of an Incomplete run → partial trail + marker. --

#[tokio::test]
async fn get_trace_of_incomplete_run_returns_partial_trail_with_marker() {
    let tmp = TempDir::new().unwrap();
    let car_dir = tmp.path().to_path_buf();

    // A run with one turn that drops mid-run (no runs.complete). The
    // disconnect sweep writes an `Incomplete` terminal marker to disk.
    let run_id = {
        let state = state_over(&car_dir);
        let addr = spawn_multi_dispatcher(state.clone()).await;
        let mut harness = connect(addr).await;
        register_tools(&mut harness, &["drive_cli"]).await;
        let run_id = start_run(&mut harness, "agent-b", "interrupted").await;
        submit_servicing_callbacks(
            &mut harness,
            "p1",
            drive_proposal("a1", "got this far"),
            &drive_output(),
        )
        .await;
        // Drop WITHOUT runs.complete.
        harness.close(None).await.ok();
        drop(harness);
        tokio::time::sleep(Duration::from_millis(500)).await;
        run_id
    };

    // Restart and read — the Incomplete run is NOT an error: it returns
    // its partial trail (the one turn) plus the terminal Incomplete marker.
    let state2 = state_over(&car_dir);
    let addr2 = spawn_multi_dispatcher(state2.clone()).await;
    let mut host = connect(addr2).await;
    become_host_client(&mut host).await;

    // It lists as Incomplete.
    let listed = call_ok(
        &mut host,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "agent-b" }),
    )
    .await;
    assert_eq!(listed["runs"][0]["status"], "incomplete");

    let resp = call_collecting_events(
        &mut host,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
        &mut Vec::new(),
    )
    .await;
    assert!(
        resp.get("error").is_none(),
        "an Incomplete run must not be an error: {resp}"
    );
    let trace = &resp["result"];
    let replay_turns = turns_of(&trace["records"]);
    assert_eq!(
        replay_turns.len(),
        1,
        "the partial trail's one turn is present"
    );
    assert_eq!(replay_turns[0].prompt.as_deref(), Some("got this far"));

    // The terminal record is an `Ended` carrying the `Incomplete`
    // termination (not an `Outcome`).
    let records = trace["records"].as_array().unwrap();
    let ended = match serde_json::from_value::<RunRecord>(records.last().unwrap().clone()).unwrap()
    {
        RunRecord::Ended(e) => e,
        other => panic!("expected a terminal Ended record, got {other:?}"),
    };
    assert!(
        matches!(ended.termination, car_proto::RunTermination::Incomplete),
        "the terminal marker is Incomplete, not Outcome"
    );
}

// --- Test 6: runs.list for a zero-run agent → empty list (R13). --------

#[tokio::test]
async fn list_for_agent_with_zero_runs_is_empty() {
    let tmp = TempDir::new().unwrap();
    let state = state_over(tmp.path());
    let addr = spawn_multi_dispatcher(state.clone()).await;

    // A host-client may list any managed agent; an agent that has never
    // run yields an empty list (not an error).
    let mut host = connect(addr).await;
    become_host_client(&mut host).await;
    let resp = call_ok(
        &mut host,
        "ls",
        "runs.list",
        serde_json::json!({ "agent_id": "never-ran" }),
    )
    .await;
    assert_eq!(resp["agent_id"], "never-ran");
    assert!(
        resp["runs"].as_array().unwrap().is_empty(),
        "an agent with no runs returns an empty list"
    );
}

// --- Test 7: get_trace of an unknown run_id → clear not-found. ---------

#[tokio::test]
async fn get_trace_of_unknown_run_is_not_found() {
    let tmp = TempDir::new().unwrap();
    let state = state_over(tmp.path());
    let addr = spawn_multi_dispatcher(state.clone()).await;

    // A host-client (entitled to manage any agent) fetches a run_id that
    // has never existed. The response is a clear not-found MARKER, not an
    // error frame — a UI distinguishes "no such run" from a transport
    // failure.
    let mut host = connect(addr).await;
    become_host_client(&mut host).await;
    let resp = call_collecting_events(
        &mut host,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": "00000000-0000-0000-0000-000000000000" }),
        &mut Vec::new(),
    )
    .await;
    assert!(
        resp.get("error").is_none(),
        "an unknown run is a not-found marker, not an error frame: {resp}"
    );
    assert_eq!(
        resp["result"]["not_found"], true,
        "the response clearly marks the run not-found"
    );
    assert_eq!(
        resp["result"]["run_id"],
        "00000000-0000-0000-0000-000000000000"
    );
}
