//! `runs.record_turns` — WS-only client-narrated turn append (feedback-
//! agent A2UI/runs plan, U1).
//!
//! Drives an *owning* harness connection (`runs.start` →
//! `runs.record_turns` → `runs.complete`) over real `run_dispatch`
//! WebSocket sessions sharing one `ServerState`, plus subscriber /
//! intruder connections, and asserts the U1 contract:
//!
//!  1. Happy 3-turn batch → `ok:true, base_index:0, count:3`; a second
//!     batch gets `base_index:3`; `runs.get_trace` returns them in order.
//!  2. Index ownership: a client-sent `index: 999` is overwritten by the
//!     daemon-stamped position.
//!  3. Fanout: a host-token subscriber receives one `runs.trace.event`
//!     per turn with contiguous cursors.
//!  4. Authz — wrong agent: a session bound to a DIFFERENT agent is
//!     rejected (uniform not-found, nothing appended).
//!  5. Authz — host token: a host-role session (no agent binding) is
//!     rejected (a write must not be forgeable via the read credential).
//!  6. Authz — unbound session: no `session.auth {agent_id}` → rejected.
//!  7. Size cap: a 1 MB `output` is truncated daemon-side (marker present,
//!     persisted line bounded) — not stored verbatim.
//!  8. Turn ceiling: a batch past the per-run ceiling → `ok:false,
//!     dropped:"run_turn_limit"`; the trace stops growing.
//!  9. Terminal run: `runs.complete` then `record_turns` → `ok:false`
//!     with the `run_terminal` drop reason; the trace is unchanged.
//! 10. Unknown run_id → `ok:false, dropped:"run_not_found"`; nothing
//!     persisted.
//! 11. Persistence: record turns, simulate a daemon restart (fresh state
//!     over the same `~/.car`) → `runs.get_trace` returns them from disk.
//! 12. Validation: missing `run_id`, empty `turns`, and a malformed turn
//!     each return the standard "requires {...}" error frame.
//! 13. Aggregate cap: a turn whose `parameters` is an array of MANY
//!     sub-field-cap strings (aggregate > 256 KiB) and a turn with a
//!     multi-MB `tool` scalar are both bounded on disk — the per-field cap
//!     alone can't see either shape.
//! 14. Ceiling is a hard cap: a batch that would STRADDLE the per-run
//!     ceiling (1999 + 5) is refused whole (`run_turn_limit`); a batch
//!     that lands exactly ON the ceiling (1999 + 1) is accepted.
//! 15. Disk-only owner fallback: a run that exists ONLY on disk (no
//!     in-memory `RunMeta`) takes the `agent_for_run` branch and returns
//!     the structured non-fatal drop — no panic, nothing appended.
//! 16. Pipelined ceiling (ADV-1): 8 × 100-turn batches sent before reading
//!     any response, against a run seeded to 1500, NEVER breach 2000 — the
//!     under-lock ceiling caps it; five land, three are refused.
//! 17. Near-cap re-stamp (ADV-4): a turn within a few bytes of the 256 KiB
//!     cap, recorded at a high (4-digit) re-stamped index, still persists a
//!     line ≤ 256 KiB.
//! 18. Ignored index (ADV-6): turns with `index: -1`, `index: "x"`, and
//!     `index: 1.5` are accepted (not validation-rejected) and re-stamped to
//!     contiguous positions.
//!
//! These run over the real dispatch path so the index re-stamp + fanout +
//! disk persist that `record_run_turns` owns are exercised end to end.

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

/// Per-launch host token a subscriber installs to take the host-management
/// role (the credential the READ path requires, Parslee-ai/car#254). The
/// WRITE path (`runs.record_turns`) deliberately does NOT honor it.
const TEST_HOST_TOKEN: &str = "test-host-token-0123456789abcdef0123456789";

/// `ServerState` whose journal dir is `<car_dir>/journals` — the run store
/// roots at the sibling `<car_dir>/runs`, exactly as the daemon resolves
/// it. A fresh state over the same `car_dir` reads the same disk store
/// (the restart simulation).
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
}

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` (an owning harness + subscriber/intruder 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 until the response to `id` arrives; push any
/// `runs.trace.event` notifications seen along the way onto `events`.
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 → full JSON-RPC response (error or success — caller
/// inspects). No callbacks, no event interleaving.
async fn call_raw(
    ws: &mut Ws,
    id: &str,
    method: &str,
    params: serde_json::Value,
) -> serde_json::Value {
    send(ws, id, method, params).await;
    next_json(ws).await
}

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

async fn auth_host(ws: &mut Ws) {
    let resp = call_raw(
        ws,
        "ha",
        "session.auth",
        serde_json::json!({ "host_token": TEST_HOST_TOKEN }),
    )
    .await;
    assert!(
        resp.get("error").is_none(),
        "host auth must succeed: {resp}"
    );
    assert_eq!(
        resp["result"]["role"], "host",
        "host auth grants host role: {resp}"
    );
}

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

/// Bind the session whose `current_run_id` is `run_id` to `agent_id`, as
/// if a `session.auth {token, agent_id}` handshake had attached it (the
/// supervised-child path). Lets a test exercise the owning-binding branch
/// of `runs.record_turns` without standing up a supervisor — the same
/// shortcut `run_lifecycle.rs::bind_only_session` uses, but keyed to the
/// right connection in a multi-session dispatcher.
async fn bind_session_for_run(state: &Arc<ServerState>, run_id: &str, agent_id: &str) {
    let sessions: Vec<_> = {
        let map = state.sessions.lock().await;
        map.values().cloned().collect()
    };
    for s in sessions {
        let cur = s.current_run_id.lock().await.clone();
        if cur.as_deref() == Some(run_id) {
            *s.agent_id.lock().await = Some(agent_id.to_string());
            return;
        }
    }
    panic!("no session owns run {run_id}");
}

/// Bind the most-recently-connected session that has NO current run to
/// `agent_id` — used to put a *different* agent on a fresh connection for
/// the wrong-agent authz test.
async fn bind_unrun_session(state: &Arc<ServerState>, agent_id: &str) {
    let sessions: Vec<_> = {
        let map = state.sessions.lock().await;
        map.values().cloned().collect()
    };
    for s in sessions {
        if s.current_run_id.lock().await.is_none() && s.agent_id.lock().await.is_none() {
            *s.agent_id.lock().await = Some(agent_id.to_string());
            return;
        }
    }
    panic!("no unbound, run-less session to bind");
}

/// One client-narrated turn: a tool turn with a name, params, and output.
/// Omits `verifier_verdict` (the daemon defaults it to `not_run`) and
/// carries a deliberately-wrong client `index` the daemon must re-stamp.
fn narrated_turn(client_index: usize, tool: &str, output: &str) -> serde_json::Value {
    serde_json::json!({
        "index": client_index,
        "tool": tool,
        "parameters": { "cmd": format!("run {tool}") },
        "output": { "text": output }
    })
}

async fn record_turns(
    ws: &mut Ws,
    id: &str,
    run_id: &str,
    turns: serde_json::Value,
) -> serde_json::Value {
    call_raw(
        ws,
        id,
        "runs.record_turns",
        serde_json::json!({ "run_id": run_id, "turns": turns }),
    )
    .await
}

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

/// Decode a `runs.trace.event` whose `record` is a Turn → (turn, cursor).
fn event_turn(ev: &serde_json::Value) -> Option<(RunTurn, usize)> {
    let cursor = ev["cursor"].as_u64()? as usize;
    let rec: RunRecord = serde_json::from_value(ev["record"].clone()).ok()?;
    match rec {
        RunRecord::Turn(t) => Some((t, cursor)),
        _ => None,
    }
}

// === Test 1: happy 3-turn batch + a second batch's base_index. ===========

#[tokio::test]
async fn happy_batch_stamps_base_index_and_count_and_persists_in_order() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "narrate me").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // First batch of three.
    let r1 = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([
            narrated_turn(0, "Read", "first"),
            narrated_turn(0, "Bash", "second"),
            narrated_turn(0, "Edit", "third"),
        ]),
    )
    .await;
    assert!(r1.get("error").is_none(), "first batch: {r1}");
    assert_eq!(r1["result"]["ok"], true);
    assert_eq!(
        r1["result"]["base_index"].as_u64().unwrap(),
        0,
        "first batch base_index"
    );
    assert_eq!(
        r1["result"]["count"].as_u64().unwrap(),
        3,
        "first batch count"
    );
    assert!(
        r1["result"].get("dropped").is_none(),
        "no drop on a healthy append"
    );

    // Second batch — base_index continues from 3.
    let r2 = record_turns(
        &mut harness,
        "b2",
        &run_id,
        serde_json::json!([narrated_turn(0, "Grep", "fourth")]),
    )
    .await;
    assert_eq!(r2["result"]["ok"], true);
    assert_eq!(
        r2["result"]["base_index"].as_u64().unwrap(),
        3,
        "second batch continues the index"
    );
    assert_eq!(r2["result"]["count"].as_u64().unwrap(), 1);

    // Replay: four turns in order with monotonic indices 0..4.
    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    assert_eq!(turns.len(), 4, "all four turns persisted");
    assert_eq!(
        turns.iter().map(|t| t.index).collect::<Vec<_>>(),
        vec![0, 1, 2, 3]
    );
    assert_eq!(turns[0].tool.as_deref(), Some("Read"));
    assert_eq!(turns[3].tool.as_deref(), Some("Grep"));
    assert_eq!(
        turns[0].output,
        Some(serde_json::json!({ "text": "first" }))
    );
    // verifier_verdict defaulted to not_run when absent on the wire.
    assert_eq!(
        turns[0].verifier_verdict,
        car_proto::VerifierVerdict::NotRun
    );
}

// === Test 2: client `index: 999` is overwritten by the stamped index. ====

#[tokio::test]
async fn client_index_is_overwritten_by_daemon_stamped_position() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "ignore my index").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Both turns claim index 999 — the daemon must stamp 0 then 1.
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([
            narrated_turn(999, "Read", "x"),
            narrated_turn(999, "Edit", "y")
        ]),
    )
    .await;
    assert_eq!(r["result"]["ok"], true);
    assert_eq!(r["result"]["base_index"].as_u64().unwrap(), 0);

    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    assert_eq!(turns.len(), 2);
    assert_eq!(turns[0].index, 0, "stamped, not 999");
    assert_eq!(turns[1].index, 1, "stamped, not 999");
}

// === Test 3: fanout to a host-token subscriber, contiguous cursors. ======

#[tokio::test]
async fn batch_fans_out_one_event_per_turn_with_contiguous_cursors() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "fan me out").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // A separate host-token subscriber.
    let mut sub = connect(addr).await;
    let mut sub_events = Vec::new();
    auth_host(&mut sub).await;
    call_collecting_events(
        &mut sub,
        "hs",
        "host.subscribe",
        serde_json::json!({}),
        &mut sub_events,
    )
    .await;
    let snap = call_collecting_events(
        &mut sub,
        "sub",
        "runs.subscribe",
        serde_json::json!({ "run_id": run_id }),
        &mut sub_events,
    )
    .await;
    assert_eq!(
        snap["result"]["cursor"].as_u64().unwrap(),
        0,
        "subscribe at turn 0"
    );

    // Record three turns in one batch on the owning harness.
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([
            narrated_turn(0, "Read", "t0"),
            narrated_turn(0, "Bash", "t1"),
            narrated_turn(0, "Edit", "t2"),
        ]),
    )
    .await;
    assert_eq!(r["result"]["count"].as_u64().unwrap(), 3);

    // The subscriber receives exactly three turn events, cursors 1,2,3.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while sub_events
        .iter()
        .filter(|e| event_turn(e).is_some())
        .count()
        < 3
    {
        if tokio::time::Instant::now() >= deadline {
            panic!("timed out waiting for 3 fanned-out turns");
        }
        if let Ok(msg) = tokio::time::timeout(Duration::from_secs(2), next_json(&mut sub)).await {
            if msg.get("method").and_then(|m| m.as_str()) == Some("runs.trace.event") {
                sub_events.push(msg["params"].clone());
            }
        }
    }
    let turns: Vec<(RunTurn, usize)> = sub_events.iter().filter_map(event_turn).collect();
    assert_eq!(turns.len(), 3, "exactly three fanned-out turns");
    assert_eq!(turns[0].1, 1, "contiguous cursors 1,2,3");
    assert_eq!(turns[1].1, 2);
    assert_eq!(turns[2].1, 3);
    assert_eq!(turns[0].0.index, 0);
    assert_eq!(turns[2].0.index, 2);
    assert_eq!(turns[1].0.tool.as_deref(), Some("Bash"));
}

// === Test 4: authz — wrong agent is rejected (uniform not-found). ========

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

    // Owner starts the run as agent-a.
    let mut owner = connect(addr).await;
    let run_id = start_run(&mut owner, "agent-a", "owned by a").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // A DIFFERENT connection bound to agent-b tries to write agent-a's run.
    let mut intruder = connect(addr).await;
    // Touch the connection so its session exists, then bind it to agent-b.
    call_ok(&mut intruder, "k", "state.keys", serde_json::json!({})).await;
    bind_unrun_session(&state, "agent-b").await;

    let r = record_turns(
        &mut intruder,
        "b1",
        &run_id,
        serde_json::json!([narrated_turn(0, "Read", "forged")]),
    )
    .await;
    // Uniform not-found — no error frame, no owner leak, nothing appended.
    assert!(
        r.get("error").is_none(),
        "wrong-agent write must NOT be a distinguishable error: {r}"
    );
    assert_eq!(r["result"]["ok"], false);
    assert_eq!(r["result"]["dropped"], "run_not_found");
    assert!(
        !r.to_string().contains("agent-a"),
        "must not leak the owning agent_id: {r}"
    );
    assert_eq!(
        state.run_turn_count(&run_id).await,
        0,
        "nothing appended by the wrong agent"
    );
}

// === Test 5: authz — host token is rejected for writes. ==================

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

    let mut owner = connect(addr).await;
    let run_id = start_run(&mut owner, "agent-a", "owned by a").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // A host-role connection (the READ credential) must NOT be able to
    // write turns — that would let CarHost forge another agent's trace.
    let mut host = connect(addr).await;
    auth_host(&mut host).await;
    let r = record_turns(
        &mut host,
        "b1",
        &run_id,
        serde_json::json!([narrated_turn(0, "Read", "forged-by-host")]),
    )
    .await;
    assert!(
        r.get("error").is_none(),
        "host write rejection is non-fatal, not an error: {r}"
    );
    assert_eq!(r["result"]["ok"], false);
    assert_eq!(
        r["result"]["dropped"], "run_not_found",
        "host token must NOT authorize a write"
    );
    assert_eq!(
        state.run_turn_count(&run_id).await,
        0,
        "host token appended nothing"
    );
}

// === Test 6: authz — unbound session is rejected. ========================

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

    let mut owner = connect(addr).await;
    let run_id = start_run(&mut owner, "agent-a", "owned by a").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // A bare, unbound connection (no session.auth {agent_id}, no host
    // token) tries to write — rejected with uniform not-found.
    let mut anon = connect(addr).await;
    let r = record_turns(
        &mut anon,
        "b1",
        &run_id,
        serde_json::json!([narrated_turn(0, "Read", "forged-anon")]),
    )
    .await;
    assert!(
        r.get("error").is_none(),
        "unbound write rejection is non-fatal: {r}"
    );
    assert_eq!(r["result"]["ok"], false);
    assert_eq!(r["result"]["dropped"], "run_not_found");
    assert_eq!(
        state.run_turn_count(&run_id).await,
        0,
        "unbound session appended nothing"
    );
}

// === Test 7: size cap — a 1 MB output is truncated daemon-side. ==========

#[tokio::test]
async fn oversized_output_is_truncated_daemon_side() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "truncate me").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    let huge = "x".repeat(1024 * 1024); // 1 MB
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([{
            "tool": "Bash",
            "parameters": { "cmd": "cat bigfile" },
            "output": { "text": huge }
        }]),
    )
    .await;
    assert_eq!(
        r["result"]["ok"], true,
        "an oversized turn is accepted (truncated), not rejected: {r}"
    );
    assert_eq!(r["result"]["count"].as_u64().unwrap(), 1);

    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    assert_eq!(turns.len(), 1);
    let stored = turns[0].output.as_ref().unwrap()["text"].as_str().unwrap();
    assert!(
        stored.len() < huge.len(),
        "the stored output is bounded, not the full 1 MB"
    );
    assert!(
        stored.len() <= 17 * 1024,
        "bounded near the 16 KB field cap, got {} bytes",
        stored.len()
    );
    assert!(
        stored.ends_with("…[truncated]"),
        "carries the truncation marker, got tail {:?}",
        &stored[stored.len().saturating_sub(20)..]
    );
    // The persisted JSONL line is bounded (not ~1 MB).
    let line = serde_json::to_string(&trace["records"]).unwrap();
    assert!(
        line.len() < 64 * 1024,
        "the persisted trace is bounded, got {} bytes",
        line.len()
    );
}

// === Test 8: per-run turn ceiling → run_turn_limit; trace stops. =========

#[tokio::test]
async fn batch_past_run_ceiling_is_dropped_with_run_turn_limit() {
    // Reaching the real ~2000 ceiling over the wire would be slow; instead
    // we seed the in-memory run buffer directly to just under and at the
    // ceiling via `record_run_turns`, then assert the handler's gate over a
    // real WS write. This exercises the SAME `RECORD_TURNS_RUN_CEILING`
    // branch the handler checks.
    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;
    let run_id = start_run(&mut harness, "agent-a", "runaway loop").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Seed 2000 turns straight into the buffer (bypasses the handler — this
    // is the runaway-loop precondition, not the code under test).
    let mut seed = Vec::with_capacity(2000);
    for i in 0..2000usize {
        seed.push(RunRecord::Turn(RunTurn {
            index: i,
            prompt: None,
            tool: Some("Bash".into()),
            parameters: serde_json::json!({}),
            output: Some(serde_json::json!({ "text": "seed" })),
            cli_outcome: None,
            verifier_verdict: car_proto::VerifierVerdict::NotRun,
            policy_rejected: None,
        }));
    }
    let total = state.record_run_turns(&run_id, seed).await;
    assert_eq!(
        total,
        car_server_core::RecordRunTurnsOutcome::Appended { new_total: 2000 },
        "seeded the run to the ceiling"
    );

    // Now a real WS write must be refused whole with run_turn_limit.
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([narrated_turn(0, "Read", "over the line")]),
    )
    .await;
    assert!(
        r.get("error").is_none(),
        "ceiling drop is non-fatal, not an error: {r}"
    );
    assert_eq!(r["result"]["ok"], false);
    assert_eq!(r["result"]["dropped"], "run_turn_limit");
    // The trace did not grow past the ceiling.
    assert_eq!(
        state.run_turn_count(&run_id).await,
        2000,
        "nothing appended past the ceiling"
    );
}

// === Test 9: terminal run → run_terminal; trace unchanged. ===============

#[tokio::test]
async fn write_against_terminal_run_is_dropped_with_run_terminal() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "complete then write").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // One real turn, then complete the run.
    let pre = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([narrated_turn(0, "Read", "before complete")]),
    )
    .await;
    assert_eq!(pre["result"]["ok"], true);
    call_ok(
        &mut harness,
        "done",
        "runs.complete",
        serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
    )
    .await;

    // Writing after completion → ok:false, run_terminal, nothing appended.
    let post = record_turns(
        &mut harness,
        "b2",
        &run_id,
        serde_json::json!([narrated_turn(0, "Edit", "after complete")]),
    )
    .await;
    assert!(
        post.get("error").is_none(),
        "post-terminal write is non-fatal: {post}"
    );
    assert_eq!(post["result"]["ok"], false);
    assert_eq!(post["result"]["dropped"], "run_terminal");

    // The persisted trace still carries exactly the one pre-complete turn.
    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    assert_eq!(turns.len(), 1, "the terminal run's trace is unchanged");
    assert_eq!(
        turns[0].output,
        Some(serde_json::json!({ "text": "before complete" }))
    );
}

// === Test 10: unknown run_id → run_not_found; nothing persisted. =========

#[tokio::test]
async fn unknown_run_id_is_dropped_with_run_not_found() {
    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;
    // Bind the connection to an agent (so it's a legitimate writer), but
    // target a run id that was never started.
    call_ok(&mut harness, "k", "state.keys", serde_json::json!({})).await;
    bind_unrun_session(&state, "agent-a").await;

    let unknown = "00000000-0000-0000-0000-000000000000";
    let r = record_turns(
        &mut harness,
        "b1",
        unknown,
        serde_json::json!([narrated_turn(0, "Read", "nowhere")]),
    )
    .await;
    assert!(
        r.get("error").is_none(),
        "unknown run is a non-fatal drop, not an error: {r}"
    );
    assert_eq!(r["result"]["ok"], false);
    assert_eq!(r["result"]["dropped"], "run_not_found");
    // Nothing was created on disk for this id.
    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": unknown }),
    )
    .await;
    assert_eq!(
        trace["not_found"], true,
        "no trace was persisted for the unknown run"
    );
}

// === Test 11: persistence across a simulated daemon restart. ==============

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

    // First daemon: record turns, complete the run, then go away.
    let run_id = {
        let state = state_over(&car_dir);
        let addr = spawn_multi_dispatcher(state.clone()).await;
        let mut harness = connect(addr).await;
        let run_id = start_run(&mut harness, "agent-a", "persist narrated turns").await;
        bind_session_for_run(&state, &run_id, "agent-a").await;
        let r = record_turns(
            &mut harness,
            "b1",
            &run_id,
            serde_json::json!([
                narrated_turn(0, "Read", "disk-1"),
                narrated_turn(0, "Edit", "disk-2"),
            ]),
        )
        .await;
        assert_eq!(r["result"]["count"].as_u64().unwrap(), 2);
        call_ok(
            &mut harness,
            "done",
            "runs.complete",
            serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
        )
        .await;
        harness.close(None).await.ok();
        drop(harness);
        tokio::time::sleep(Duration::from_millis(400)).await;
        run_id
        // `state` + listener drop — the daemon is "gone".
    };

    // Restart: fresh ServerState (empty memory) over the same ~/.car. The
    // trace must serve from disk.
    let state2 = state_over(&car_dir);
    let addr2 = spawn_multi_dispatcher(state2.clone()).await;
    let mut host = connect(addr2).await;
    auth_host(&mut host).await;
    let trace = call_ok(
        &mut host,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    assert_eq!(trace["agent_id"], "agent-a");
    let turns = turns_of(&trace["records"]);
    assert_eq!(
        turns.len(),
        2,
        "both recorded turns replay from disk after restart"
    );
    assert_eq!(
        turns[0].output,
        Some(serde_json::json!({ "text": "disk-1" }))
    );
    assert_eq!(
        turns[1].output,
        Some(serde_json::json!({ "text": "disk-2" }))
    );
    assert_eq!(turns[0].index, 0);
    assert_eq!(turns[1].index, 1);
}

// === Test 12: validation — missing run_id / empty turns / malformed. =====

#[tokio::test]
async fn validation_rejects_missing_run_id_empty_and_malformed_turns() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "validate").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Missing run_id.
    let no_run = call_raw(
        &mut harness,
        "v1",
        "runs.record_turns",
        serde_json::json!({ "turns": [narrated_turn(0, "Read", "x")] }),
    )
    .await;
    assert!(
        no_run.get("error").is_some(),
        "missing run_id is an error: {no_run}"
    );
    assert!(
        no_run["error"]["message"]
            .as_str()
            .unwrap()
            .contains("runs.record_turns requires"),
        "standard requires-error: {no_run}"
    );

    // Empty turns array.
    let empty = call_raw(
        &mut harness,
        "v2",
        "runs.record_turns",
        serde_json::json!({ "run_id": run_id, "turns": [] }),
    )
    .await;
    assert!(
        empty.get("error").is_some(),
        "empty turns is an error: {empty}"
    );
    assert!(
        empty["error"]["message"]
            .as_str()
            .unwrap()
            .contains("non-empty"),
        "names the empty-turns problem: {empty}"
    );

    // Malformed turn: `output` is a string-typed primitive where a turn
    // requires an object; here we send a turn that is itself a bare number.
    let bad = call_raw(
        &mut harness,
        "v3",
        "runs.record_turns",
        serde_json::json!({ "run_id": run_id, "turns": [42] }),
    )
    .await;
    assert!(
        bad.get("error").is_some(),
        "a malformed turn is an error: {bad}"
    );
    assert!(
        bad["error"]["message"]
            .as_str()
            .unwrap()
            .contains("runs.record_turns requires"),
        "standard requires-error: {bad}"
    );

    // Nothing got appended by any of the rejected calls.
    assert_eq!(
        state.run_turn_count(&run_id).await,
        0,
        "no rejected call appended a turn"
    );
}

// === Test 13: aggregate per-turn cap — bounded regardless of leaf shape. ==

#[tokio::test]
async fn aggregate_oversize_turn_is_bounded_regardless_of_leaf_distribution() {
    // Two shapes the per-field 16 KiB cap CANNOT bound on its own:
    //  - `parameters` as an array of many sub-cap strings (every leaf is
    //    under the field cap, the SUM is ~512 KiB), and
    //  - a multi-MB `tool` scalar (a field the per-field pass never
    //    visits).
    // The aggregate 256 KiB per-turn cap must bound both persisted lines.
    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;
    let run_id = start_run(&mut harness, "agent-a", "bound my aggregate").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // 64 strings × 8 KiB = 512 KiB aggregate, each leaf under the 16 KiB
    // field cap (so per-field truncation leaves all of them alone).
    let leaves: Vec<String> = (0..64).map(|_| "y".repeat(8 * 1024)).collect();
    let huge_tool = "t".repeat(3 * 1024 * 1024); // 3 MiB scalar
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([
            {
                "tool": "Bash",
                "parameters": leaves,
                "output": { "text": "many leaves" }
            },
            {
                "tool": huge_tool,
                "parameters": { "cmd": "who am i" },
                "output": { "text": "huge tool name" }
            }
        ]),
    )
    .await;
    assert_eq!(
        r["result"]["ok"], true,
        "oversized turns are accepted (bounded), not rejected: {r}"
    );
    assert_eq!(r["result"]["count"].as_u64().unwrap(), 2);

    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    assert_eq!(turns.len(), 2);

    // Each persisted turn honors the 256 KiB aggregate invariant.
    for (i, t) in turns.iter().enumerate() {
        let n = serde_json::to_vec(t).unwrap().len();
        assert!(
            n <= 256 * 1024,
            "turn {i} persisted within the aggregate cap, got {n} bytes"
        );
    }
    // The leaf-array turn's `parameters` was replaced whole with the
    // aggregate-cap marker (no partial salvage).
    assert_eq!(
        turns[0].parameters,
        serde_json::json!("…[truncated: turn exceeded 256 KiB]"),
        "many-leaves `parameters` replaced with the oversize marker"
    );
    // The huge `tool` scalar was field-capped (the big-three replacement
    // alone couldn't get this turn under the cap).
    let stored_tool = turns[1].tool.as_deref().unwrap();
    assert!(
        stored_tool.len() <= 17 * 1024,
        "tool bounded near the field cap, got {} bytes",
        stored_tool.len()
    );
    assert!(
        stored_tool.ends_with("…[truncated]"),
        "tool carries the truncation marker"
    );
    // The whole persisted JSONL trace is bounded (mirror of test 7).
    let line = serde_json::to_string(&trace["records"]).unwrap();
    assert!(
        line.len() < 64 * 1024,
        "the persisted trace is bounded, got {} bytes",
        line.len()
    );
}

// === Test 14: ceiling straddle — refused whole; landing ON it is fine. ===

#[tokio::test]
async fn batch_straddling_run_ceiling_is_refused_whole() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "straddle the line").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Seed 1999 turns straight into the buffer (the runaway precondition,
    // not the code under test — same shortcut as the ceiling test above).
    let mut seed = Vec::with_capacity(1999);
    for i in 0..1999usize {
        seed.push(RunRecord::Turn(RunTurn {
            index: i,
            prompt: None,
            tool: Some("Bash".into()),
            parameters: serde_json::json!({}),
            output: Some(serde_json::json!({ "text": "seed" })),
            cli_outcome: None,
            verifier_verdict: car_proto::VerifierVerdict::NotRun,
            policy_rejected: None,
        }));
    }
    assert_eq!(
        state.record_run_turns(&run_id, seed).await,
        car_server_core::RecordRunTurnsOutcome::Appended { new_total: 1999 },
        "seeded just under the ceiling"
    );

    // A 5-turn batch would land at 2004 — past the hard cap. Refused
    // WHOLE: no partial admit of the first turn.
    let straddle = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([
            narrated_turn(0, "Read", "s1"),
            narrated_turn(0, "Bash", "s2"),
            narrated_turn(0, "Edit", "s3"),
            narrated_turn(0, "Grep", "s4"),
            narrated_turn(0, "Write", "s5"),
        ]),
    )
    .await;
    assert!(
        straddle.get("error").is_none(),
        "straddle drop is non-fatal, not an error: {straddle}"
    );
    assert_eq!(straddle["result"]["ok"], false);
    assert_eq!(straddle["result"]["dropped"], "run_turn_limit");
    assert_eq!(
        state.run_turn_count(&run_id).await,
        1999,
        "the straddling batch appended NOTHING"
    );

    // A 1-turn batch lands exactly ON the ceiling (1999 + 1 = 2000) — the
    // cap is `>`, not `>=`, so it is accepted.
    let exact = record_turns(
        &mut harness,
        "b2",
        &run_id,
        serde_json::json!([narrated_turn(0, "Read", "the 2000th")]),
    )
    .await;
    assert_eq!(
        exact["result"]["ok"], true,
        "landing exactly on the ceiling is allowed: {exact}"
    );
    assert_eq!(exact["result"]["base_index"].as_u64().unwrap(), 1999);
    assert_eq!(state.run_turn_count(&run_id).await, 2000);
}

// === Test 15: disk-only owner fallback — structured drop, no panic. ======

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

    // A run that exists ONLY on disk: write its `RunStarted` through the
    // store directly — this process's in-memory `runs` map never hears
    // about it (the shape of a run a PRIOR daemon process opened).
    let run_id = "disk-only-run-0001";
    state
        .run_store
        .write_started(&car_proto::RunStarted {
            run_id: run_id.to_string(),
            agent_id: "agent-disk".to_string(),
            intent: "exist only on disk".to_string(),
            outcome_description: None,
            started_at: chrono::Utc::now(),
        })
        .expect("seed the disk-only run");
    assert!(
        state.run_meta(run_id).await.is_none(),
        "precondition: absent from the in-memory map"
    );

    // A connection legitimately bound to the OWNING agent writes to it —
    // the handler resolves the owner via the disk fallback
    // (`agent_for_run`), passes authz, then the shared append finds no
    // live in-memory run.
    let mut harness = connect(addr).await;
    call_ok(&mut harness, "k", "state.keys", serde_json::json!({})).await;
    bind_unrun_session(&state, "agent-disk").await;

    let r = record_turns(
        &mut harness,
        "b1",
        run_id,
        serde_json::json!([narrated_turn(0, "Read", "to disk")]),
    )
    .await;
    // The structured non-fatal drop — NOT a panic, NOT an error frame,
    // NOT a bogus ok:true. (`record_run_turns` only appends to live
    // in-memory runs; the zero-count surfaces as `run_terminal`.)
    assert!(
        r.get("error").is_none(),
        "disk-only write is a non-fatal drop: {r}"
    );
    assert_eq!(r["result"]["ok"], false);
    assert_eq!(r["result"]["dropped"], "run_terminal");
    // Nothing was appended to the on-disk trace.
    let records = state
        .run_store
        .get_run_trace(run_id)
        .expect("the disk trace still resolves");
    assert!(
        !records.iter().any(|rec| matches!(rec, RunRecord::Turn(_))),
        "no turn was appended to the disk-only run"
    );
}

// === Test 16: pipelined batches CANNOT breach the ceiling (ADV-1). =======
//
// The dispatcher spawns a task per frame, so several `runs.record_turns`
// frames sent before any response is read each read a SUB-ceiling
// `run_header` snapshot and pass the handler's lock-free pre-check before
// any of them appends. Only the under-lock ceiling check in
// `record_run_turns` is atomic with the append, so it is the one that caps
// the run. This test seeds a run to 1500, pipelines 8 × 100-turn batches on
// ONE authed connection (sending all 8 request frames before reading a
// single response), and asserts the persisted turn count NEVER exceeds 2000.
//
// WITHOUT fix #1's under-lock check this fails: all 8 batches pass the
// handler pre-check (1500 + 100 = 1600 ≤ 2000 for each) and append, landing
// the run at 2300. WITH it, exactly five batches land (1500 → 2000) and the
// rest are refused whole with `run_turn_limit`.
#[tokio::test]
async fn pipelined_batches_cannot_breach_the_ceiling() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "pipeline the ceiling").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Seed the run to 1500 straight into the buffer (the runaway precondition,
    // not the code under test — same shortcut as tests 8 and 14).
    let mut seed = Vec::with_capacity(1500);
    for i in 0..1500usize {
        seed.push(RunRecord::Turn(RunTurn {
            index: i,
            prompt: None,
            tool: Some("Bash".into()),
            parameters: serde_json::json!({}),
            output: Some(serde_json::json!({ "text": "seed" })),
            cli_outcome: None,
            verifier_verdict: car_proto::VerifierVerdict::NotRun,
            policy_rejected: None,
        }));
    }
    assert_eq!(
        state.record_run_turns(&run_id, seed).await,
        car_server_core::RecordRunTurnsOutcome::Appended { new_total: 1500 },
        "seeded the run to 1500"
    );

    // Build a 100-turn batch (well under RECORD_TURNS_MAX_BATCH = 256).
    let batch_of_100 = || -> serde_json::Value {
        let turns: Vec<serde_json::Value> = (0..100)
            .map(|n| narrated_turn(0, "Read", &format!("pipelined {n}")))
            .collect();
        serde_json::Value::Array(turns)
    };

    // Pipeline: send ALL 8 request frames before reading any response, so the
    // per-frame handler tasks all observe a sub-ceiling header snapshot and
    // race into the append. (1500 + 8 × 100 = 2300 if the cap were soft.)
    const N: usize = 8;
    for k in 0..N {
        send(
            &mut harness,
            &format!("p{k}"),
            "runs.record_turns",
            serde_json::json!({ "run_id": run_id, "turns": batch_of_100() }),
        )
        .await;
    }

    // Drain the 8 responses (ignoring any interleaved trace events). Tally
    // how many landed vs were refused with run_turn_limit.
    let mut ok_count = 0usize;
    let mut limited = 0usize;
    let mut received = 0usize;
    while received < N {
        let msg = next_json(&mut harness).await;
        if msg.get("method").and_then(|m| m.as_str()) == Some("runs.trace.event") {
            continue;
        }
        if msg
            .get("id")
            .and_then(|v| v.as_str())
            .is_some_and(|s| s.starts_with('p'))
        {
            received += 1;
            assert!(
                msg.get("error").is_none(),
                "ceiling handling is non-fatal, not an error: {msg}"
            );
            let res = &msg["result"];
            if res["ok"] == true {
                ok_count += 1;
            } else {
                assert_eq!(
                    res["dropped"], "run_turn_limit",
                    "the only non-fatal drop here is the ceiling: {msg}"
                );
                limited += 1;
            }
        }
    }
    assert_eq!(
        ok_count + limited,
        N,
        "every pipelined frame got a response"
    );

    // The load-bearing assertion: the persisted turn count NEVER exceeded the
    // ceiling, regardless of how the 8 racing tasks interleaved.
    let final_count = state.run_turn_count(&run_id).await;
    assert!(
        final_count <= 2000,
        "the ceiling held under pipelining: persisted {final_count} turns (must be <= 2000)"
    );
    // And it landed exactly on the ceiling — five 100-turn batches admitted
    // (1500 → 2000), the rest refused whole.
    assert_eq!(
        final_count, 2000,
        "exactly five batches admitted to the ceiling, no partial admits"
    );
    assert_eq!(ok_count, 5, "five batches landed (1500 + 5×100 = 2000)");
    assert_eq!(
        limited, 3,
        "the three straddling batches were refused with run_turn_limit"
    );

    // The on-disk trace matches the in-memory count (lock-step).
    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    assert_eq!(
        turns_of(&trace["records"]).len(),
        2000,
        "disk trace also capped at the ceiling"
    );
}

// === Test 17: near-cap turn at a high re-stamped index stays ≤ 256 KiB. ===
//
// ADV-4: the byte cap is MEASURED with the placeholder `index: 0`, but the
// turn is PERSISTED with the daemon's re-stamped index (up to 4 digits near
// the ceiling). A turn sized within a few bytes of the cap, recorded at a
// high index, must still persist a line ≤ 256 KiB — the measurement leaves
// `RECORD_TURN_REINDEX_HEADROOM` so the index growth can't push it over.
#[tokio::test]
async fn near_cap_turn_at_high_index_stays_within_persisted_cap() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "near the cap").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Seed the run to 1999 so the NEXT recorded turn re-stamps to index 1999
    // (4 digits — the worst-case re-stamp growth from the placeholder `0`).
    let mut seed = Vec::with_capacity(1999);
    for i in 0..1999usize {
        seed.push(RunRecord::Turn(RunTurn {
            index: i,
            prompt: None,
            tool: Some("x".into()),
            parameters: serde_json::Value::Null,
            output: None,
            cli_outcome: None,
            verifier_verdict: car_proto::VerifierVerdict::NotRun,
            policy_rejected: None,
        }));
    }
    assert_eq!(
        state.record_run_turns(&run_id, seed).await,
        car_server_core::RecordRunTurnsOutcome::Appended { new_total: 1999 },
        "seeded to 1999 so the next turn re-stamps to a 4-digit index"
    );

    // Build an `output` whose ENCODED turn — at the placeholder index 0 —
    // lands just 2 bytes BELOW the full 256 KiB cap. We can't use one huge
    // string (the per-field 16 KiB pass would clamp it first); instead use an
    // ARRAY of sub-16-KiB leaves (each untouched by the per-field cap) whose
    // SUM is near the cap — test 13's shape, sized to the boundary.
    //
    // This is the precise regression the headroom guards (ADV-4): the turn is
    // MEASURED with the placeholder index 0 but PERSISTED with the daemon's
    // re-stamped index. At index 0 this turn encodes to `cap - 2`; re-stamped
    // to the 4-digit index 1999 it would encode to `cap + 1` — OVER the cap.
    //   - WITHOUT the headroom (if the cap were measured at the full
    //     `RECORD_TURN_MAX_BYTES`): `cap - 2 ≤ cap`, so the turn is admitted
    //     UNTOUCHED and the persisted line is `cap + 1` > 256 KiB — the bug.
    //   - WITH the headroom (`RECORD_TURN_MEASURE_CAP = cap - 16`): `cap - 2`
    //     exceeds the measurement cap, so the oversize marker fires and the
    //     persisted line is bounded. Either way the persisted line is ≤ cap.
    let cap = 256 * 1024;
    // 32 × 8000-char leaves + a 5978-char tail leaf → encoded `cap - 2` at
    // index 0, `cap + 1` at index 1999. Every leaf is < the 16 KiB per-field
    // cap, so the per-field pass leaves them all alone.
    let mut leaves: Vec<String> = (0..32).map(|_| "z".repeat(8000)).collect();
    leaves.push("z".repeat(5978));
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([{
            "index": -7,                         // ignored + re-stamped to 1999
            "tool": "Bash",
            "output": leaves
        }]),
    )
    .await;
    assert!(
        r.get("error").is_none(),
        "a near-cap turn is accepted, not an error: {r}"
    );
    assert_eq!(r["result"]["ok"], true, "near-cap turn admitted: {r}");
    assert_eq!(
        r["result"]["base_index"].as_u64().unwrap(),
        1999,
        "re-stamped to the high index"
    );

    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    let last = turns.last().expect("the near-cap turn persisted");
    assert_eq!(
        last.index, 1999,
        "persisted at the re-stamped 4-digit index"
    );
    // THE load-bearing assertion: the persisted line for this turn —
    // re-stamped to the 4-digit index — is ≤ 256 KiB. Without the ADV-4
    // re-index headroom this is `cap + 1` and FAILS; with it, the measurement
    // caught the turn at `cap - 2 > cap - 16`, fired the marker, and bounded
    // the line.
    let line_len = serde_json::to_vec(last).unwrap().len();
    assert!(
        line_len <= cap,
        "the re-stamped near-cap turn stays within 256 KiB: persisted {line_len} bytes (cap {cap})"
    );
    // The headroom path engaged: this turn was at `cap - 2` (> the measure
    // cap), so the oversize marker replaced its free-form `output`. (This is
    // how the line ended up bounded — the proof the headroom measurement, not
    // luck, kept the persisted line under the cap.)
    assert_eq!(
        last.output,
        Some(serde_json::json!("…[truncated: turn exceeded 256 KiB]")),
        "the near-cap output was replaced by the aggregate-cap marker (measurement headroom engaged)"
    );
}

// === Test 18: a present-but-INVALID client index is ignored, not rejected. =
//
// ADV-6: the protocol doc says the client `index` is "ignored", but the old
// code used `entry().or_insert` — so a present-but-invalid value (`-1`,
// `"x"`, `1.5`) survived into the typed `usize` decode and failed the WHOLE
// batch as a validation error. The handler now overwrites `index` with 0
// unconditionally BEFORE the decode, so any junk index is accepted and the
// turn is re-stamped to its live append position.
#[tokio::test]
async fn invalid_client_index_is_ignored_and_restamped() {
    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;
    let run_id = start_run(&mut harness, "agent-a", "ignore my index").await;
    bind_session_for_run(&state, &run_id, "agent-a").await;

    // Two turns with indices that would each FAIL a typed `usize` decode if
    // they reached it: a negative number and a non-numeric string. (A float
    // `1.5` is covered too — all three are overwritten before decode.)
    let r = record_turns(
        &mut harness,
        "b1",
        &run_id,
        serde_json::json!([
            { "index": -1, "tool": "Read", "output": { "text": "neg index" } },
            { "index": "x", "tool": "Edit", "output": { "text": "string index" } },
            { "index": 1.5, "tool": "Bash", "output": { "text": "float index" } }
        ]),
    )
    .await;
    assert!(
        r.get("error").is_none(),
        "invalid client indices must NOT fail the batch: {r}"
    );
    assert_eq!(r["result"]["ok"], true, "the batch is accepted: {r}");
    assert_eq!(
        r["result"]["count"].as_u64().unwrap(),
        3,
        "all three turns accepted"
    );
    assert_eq!(r["result"]["base_index"].as_u64().unwrap(), 0);

    // They were re-stamped to contiguous 0,1,2 — the client's junk indices
    // were truly ignored, not honored.
    let trace = call_ok(
        &mut harness,
        "gt",
        "runs.get_trace",
        serde_json::json!({ "run_id": run_id }),
    )
    .await;
    let turns = turns_of(&trace["records"]);
    assert_eq!(turns.len(), 3);
    assert_eq!(turns[0].index, 0);
    assert_eq!(turns[1].index, 1);
    assert_eq!(turns[2].index, 2);
    assert_eq!(
        turns[0].output,
        Some(serde_json::json!({ "text": "neg index" }))
    );
    assert_eq!(
        turns[2].output,
        Some(serde_json::json!({ "text": "float index" }))
    );
}

// Keep `loopback_state` referenced (parity with sibling test modules that
// expose both a `state_over` and a journal-only constructor); the suite
// uses `state_over` for the disk-rooted runs store.
#[allow(dead_code)]
fn _loopback_state_in_scope() -> Arc<ServerState> {
    loopback_state(std::env::temp_dir().join("car-record-turns-unused"))
}
