//! Agent run tracing — in-memory turn eviction on terminal transition
//! (heap-hygiene follow-up).
//!
//! A completed/incomplete run's per-turn buffer (`RunMeta.turns`) holds
//! the full prompt + CLI output of every turn — potentially MBs each — and
//! used to stay resident in the process-wide `ServerState::runs` map for
//! the daemon's whole lifetime. The disk store is GC'd; the in-memory map
//! was not. The fix drops `RunMeta.turns` at the END of `complete_run` /
//! `mark_run_incomplete` (after the disk flush and terminal fanout), while
//! keeping the lightweight `RunMeta` header (agent_id, status, started_at,
//! termination) resident. To stay correct, `runs.subscribe` re-sources a
//! terminal-and-evicted run's snapshot from the disk store (the same
//! source `runs.get_trace` reads), so a late subscriber to a just-completed
//! run still gets the full ordered trace.
//!
//! These run over the real `run_dispatch` WebSocket path so the eviction +
//! disk-fallback are exercised end to end, then assert directly against the
//! shared `ServerState` (`run_meta`) to confirm the in-memory buffer is
//! actually released:
//!
//! a. After a run completes, its in-memory `RunMeta.turns` is empty, BUT a
//!    fresh `runs.subscribe` returns the full ordered trace (disk-sourced)
//!    AND `runs.get_trace` returns it.
//! b. A run that completes, THEN a late subscriber joins, gets the full
//!    snapshot from disk.
//! c. An IN-PROGRESS run's live snapshot is unaffected — turns are still
//!    held in memory and served from there.

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::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>>;

/// Per-launch host token the harness installs so a subscriber can take
/// the host-management role required for cross-agent run reads
/// (Parslee-ai/car#254).
const TEST_HOST_TOKEN: &str = "test-host-token-0123456789abcdef0123456789";

fn loopback_state(journal_dir: std::path::PathBuf) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cfg = ServerStateConfig::new(journal_dir).with_shared_memgine(engine);
    let state = Arc::new(ServerState::with_config(cfg));
    let _ = state.install_host_token(TEST_HOST_TOKEN.to_string());
    state
}

/// Accept many connections, each driven by `run_dispatch` against the
/// shared `state` (a harness producing turns + a late subscriber).
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. 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;
        }
    }
}

/// Service a `proposal.submit` on the harness, replying to the daemon's
/// inbound `tools.execute` callback with `tool_output`.
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) -> String {
    send(
        ws,
        "start",
        "runs.start",
        serde_json::json!({ "agent_id": agent_id, "intent": "do the thing" }),
    )
    .await;
    let resp = next_json(ws).await;
    assert!(resp.get("error").is_none(), "runs.start failed: {resp}");
    resp["result"]["run_id"].as_str().unwrap().to_string()
}

/// 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 },
                "preconditions": [],
                "expected_effects": {},
                "state_dependencies": [],
                "idempotent": false,
                "max_retries": 0,
                "failure_behavior": "abort",
                "metadata": {}
            }],
            "timestamp": chrono::Utc::now().to_rfc3339(),
            "context": {}
        }
    })
}

fn drive_output() -> serde_json::Value {
    serde_json::json!({ "cli": "claude", "exit_code": 0, "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()
    })
}

/// Extract the ordered `(index, prompt)` of the `RunRecord::Turn`s in a
/// `runs.subscribe` snapshot's `turns_so_far`.
fn snapshot_turns(snap: &serde_json::Value) -> Vec<(usize, String)> {
    snap["turns_so_far"]
        .as_array()
        .expect("turns_so_far is an array")
        .iter()
        .map(|t| {
            let turn: RunTurn = serde_json::from_value(t.clone()).expect("decode RunTurn");
            (turn.index, turn.prompt.clone().unwrap_or_default())
        })
        .collect()
}

/// Drive a host-client subscriber connection: `host.subscribe` then
/// `runs.subscribe {run_id}`, returning the subscribe response.
async fn host_subscribe(sub: &mut Ws, run_id: &str) -> serde_json::Value {
    let mut events = Vec::new();
    // #254: take the host role via the host token — host.subscribe alone
    // no longer authorizes cross-agent run reads.
    let ha = call_collecting_events(
        sub,
        "ha",
        "session.auth",
        serde_json::json!({ "host_token": TEST_HOST_TOKEN }),
        &mut events,
    )
    .await;
    assert_eq!(
        ha["result"]["role"], "host",
        "host auth must grant host role: {ha}"
    );
    let hs = call_collecting_events(
        sub,
        "hs",
        "host.subscribe",
        serde_json::json!({}),
        &mut events,
    )
    .await;
    assert!(hs.get("error").is_none(), "host.subscribe: {hs}");
    call_collecting_events(
        sub,
        "sub",
        "runs.subscribe",
        serde_json::json!({ "run_id": run_id }),
        &mut events,
    )
    .await
}

// --- (a) Completed run: in-memory turns released, snapshot disk-sourced,
// get_trace still works. ---

#[tokio::test]
async fn completed_run_evicts_memory_turns_but_subscribe_and_get_trace_still_serve_full_trace() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    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").await;

    // Drive three turns with distinct prompts.
    for (i, prompt) in ["alpha", "beta", "gamma"].iter().enumerate() {
        let r = submit_servicing_callbacks(
            &mut harness,
            &format!("p{i}"),
            drive_proposal(&format!("a{i}"), prompt),
            &drive_output(),
        )
        .await;
        assert!(r.get("error").is_none(), "submit {i}: {r}");
    }

    // Sanity: while in progress the turns are held in memory.
    let mid = state.run_meta(&run_id).await.expect("run recorded");
    assert_eq!(mid.turns.len(), 3, "in-progress run holds turns in memory");
    assert!(mid.termination.is_none());

    // Complete the run.
    send(
        &mut harness,
        "done",
        "runs.complete",
        serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
    )
    .await;
    let ack = next_json(&mut harness).await;
    assert_eq!(ack["result"]["ok"], true, "runs.complete acked: {ack}");

    // The lightweight RunMeta header survives, but its turn buffer is
    // released — this is the heap-hygiene contract.
    let after = state
        .run_meta(&run_id)
        .await
        .expect("run header still resident");
    assert!(
        after.turns.is_empty(),
        "completed run's in-memory turns are dropped (got {} entries)",
        after.turns.len()
    );
    assert!(after.is_terminal(), "header still reflects terminal state");
    assert_eq!(after.agent_id, "agent-a", "header keeps agent_id");

    // The capacity is freed, not just the length — `Vec::new()` releases
    // the backing allocation that holds the heavy per-turn payloads.
    assert_eq!(
        after.turns.capacity(),
        0,
        "evicted turn buffer releases its capacity, not just its length"
    );

    // A fresh `runs.subscribe` to the completed run must still return the
    // full ordered trace — re-sourced from disk.
    let mut sub = connect(addr).await;
    let snap_resp = host_subscribe(&mut sub, &run_id).await;
    assert!(
        snap_resp.get("error").is_none(),
        "subscribe to completed run: {snap_resp}"
    );
    let snap = &snap_resp["result"];
    assert_eq!(
        snap["status"], "completed",
        "snapshot reports terminal status"
    );
    assert_eq!(
        snap["cursor"].as_u64().unwrap(),
        3,
        "disk-sourced snapshot cursor equals the turn count"
    );
    let turns = snapshot_turns(snap);
    assert_eq!(
        turns,
        vec![
            (0, "alpha".to_string()),
            (1, "beta".to_string()),
            (2, "gamma".to_string())
        ],
        "completed-run subscribe returns the full ordered trace from disk"
    );

    // `runs.get_trace` still returns the full trace (Started + turns +
    // Ended) — unchanged by the eviction.
    send(
        &mut sub,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let gt = loop {
        let msg = next_json(&mut sub).await;
        if msg.get("id").and_then(|v| v.as_str()) == Some("gt") {
            break msg;
        }
    };
    assert!(gt.get("error").is_none(), "get_trace: {gt}");
    let records: Vec<RunRecord> =
        serde_json::from_value(gt["result"]["records"].clone()).expect("decode records");
    let prompts: Vec<String> = records
        .iter()
        .filter_map(|r| match r {
            RunRecord::Turn(t) => Some(t.prompt.clone().unwrap_or_default()),
            _ => None,
        })
        .collect();
    assert_eq!(
        prompts,
        vec!["alpha".to_string(), "beta".to_string(), "gamma".to_string()],
        "runs.get_trace still returns the full ordered turn trace"
    );
    assert!(
        records.iter().any(|r| matches!(r, RunRecord::Ended(_))),
        "get_trace includes the terminal Ended record"
    );
}

// --- (b) Complete first, THEN a late subscriber joins → full disk
// snapshot. (Distinct from (a): no subscriber exists at completion time,
// so the only source is disk.) ---

#[tokio::test]
async fn late_subscriber_after_complete_gets_full_disk_snapshot() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    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").await;

    for (i, prompt) in ["one", "two"].iter().enumerate() {
        let r = submit_servicing_callbacks(
            &mut harness,
            &format!("p{i}"),
            drive_proposal(&format!("a{i}"), prompt),
            &drive_output(),
        )
        .await;
        assert!(r.get("error").is_none(), "submit {i}: {r}");
    }

    // Complete with NO subscriber attached.
    send(
        &mut harness,
        "done",
        "runs.complete",
        serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
    )
    .await;
    let ack = next_json(&mut harness).await;
    assert_eq!(ack["result"]["ok"], true);

    let after = state.run_meta(&run_id).await.expect("header resident");
    assert!(after.turns.is_empty(), "turns evicted on complete");

    // Late subscriber joins AFTER completion — its snapshot can only come
    // from disk.
    let mut late = connect(addr).await;
    let snap_resp = host_subscribe(&mut late, &run_id).await;
    let snap = &snap_resp["result"];
    assert_eq!(snap["cursor"].as_u64().unwrap(), 2);
    assert_eq!(snap["status"], "completed");
    assert_eq!(
        snapshot_turns(snap),
        vec![(0, "one".to_string()), (1, "two".to_string())],
        "late subscriber gets the full ordered trace from disk"
    );
}

// --- (c) An in-progress run's live snapshot is unaffected — turns are
// still held and served from memory. ---

#[tokio::test]
async fn in_progress_run_snapshot_still_served_from_memory() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    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").await;

    for (i, prompt) in ["live-0", "live-1"].iter().enumerate() {
        let r = submit_servicing_callbacks(
            &mut harness,
            &format!("p{i}"),
            drive_proposal(&format!("a{i}"), prompt),
            &drive_output(),
        )
        .await;
        assert!(r.get("error").is_none(), "submit {i}: {r}");
    }

    // The run is NOT terminal — turns remain resident in memory.
    let meta = state.run_meta(&run_id).await.expect("run recorded");
    assert!(meta.termination.is_none(), "run is still in progress");
    assert_eq!(meta.turns.len(), 2, "in-progress turns held in memory");

    // Subscribe while still in progress: the snapshot is the live in-memory
    // buffer (status in_progress), exactly as before this change.
    let mut sub = connect(addr).await;
    let snap_resp = host_subscribe(&mut sub, &run_id).await;
    let snap = &snap_resp["result"];
    assert_eq!(
        snap["status"], "in_progress",
        "live snapshot reports in_progress"
    );
    assert_eq!(
        snap["cursor"].as_u64().unwrap(),
        2,
        "live cursor is the turn count"
    );
    assert_eq!(
        snapshot_turns(snap),
        vec![(0, "live-0".to_string()), (1, "live-1".to_string())],
        "in-progress snapshot is served from the in-memory buffer"
    );

    // Drive one more turn — the live stream must still flow (the in-progress
    // path is untouched). The subscriber sees the new turn stream in.
    let r = submit_servicing_callbacks(
        &mut harness,
        "p2",
        drive_proposal("a2", "live-2"),
        &drive_output(),
    )
    .await;
    assert!(r.get("error").is_none(), "submit 2: {r}");

    let mut saw_live_turn = false;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    while tokio::time::Instant::now() < deadline {
        let frame = tokio::time::timeout(Duration::from_secs(2), next_json(&mut sub)).await;
        let Ok(msg) = frame else { break };
        if msg.get("method").and_then(|m| m.as_str()) == Some("runs.trace.event") {
            let rec: RunRecord = serde_json::from_value(msg["params"]["record"].clone()).unwrap();
            if let RunRecord::Turn(t) = rec {
                if t.prompt.as_deref() == Some("live-2") {
                    assert_eq!(msg["params"]["cursor"].as_u64().unwrap(), 3);
                    saw_live_turn = true;
                    break;
                }
            }
        }
    }
    assert!(
        saw_live_turn,
        "the in-progress live stream still delivers post-snapshot turns from memory"
    );

    // Memory still holds all three turns — no eviction for a live run.
    let meta = state.run_meta(&run_id).await.expect("run recorded");
    assert_eq!(
        meta.turns.len(),
        3,
        "in-progress run keeps all turns in memory"
    );
}
