car-server-core 0.24.1

//! Agent run tracing — run lifecycle (U1).
//!
//! Drives `runs.start` / `runs.complete` over a real `run_dispatch`
//! WebSocket session (mirroring `embed_smoke.rs`) and asserts the
//! durable-boundary + terminal-outcome + Incomplete-on-drop behavior:
//!
//! 1. `runs.start` returns a unique `run_id`; two sequential starts get
//!    distinct ids (R1 — durable per-run key).
//! 2. The session's `current_run_id` is set before `runs.start` responds
//!    (KTD3 race guard).
//! 3. `runs.complete` with a `success` outcome records the terminal
//!    `OutcomeStatus`.
//! 4. A healthy run that completes-then-closes is NOT `Incomplete` (ack +
//!    grace window — R5).
//! 5. A mid-run drop (no `runs.complete`) yields an `Incomplete` terminal
//!    record (R5).
//! 6. `runs.start` with no resolvable `agent_id` is rejected; the
//!    `agent_name` fallback path records successfully.

use car_proto::RunTermination;
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use car_memgine::MemgineEngine;
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
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>,
>;

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);
    Arc::new(ServerState::with_config(cfg))
}

/// Spin up a loopback listener running `run_dispatch` against the first
/// connection, sharing the caller's `state` so the test can inspect the
/// `runs` registry and the per-session `current_run_id` after calls.
async fn spawn_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 {
        let (stream, peer) = listener.accept().await.expect("accept");
        let ws = accept_async(stream).await.expect("ws handshake");
        let (write, read) = ws.split();
        let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
    });
    addr
}

/// Send one request and return the full JSON-RPC response (error or
/// success — caller inspects).
async fn call_raw(ws: &mut Ws, id: &str, method: &str, params: serde_json::Value) -> 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");
    let text = ws
        .next()
        .await
        .expect("response frame")
        .expect("response frame ok")
        .into_text()
        .expect("text response")
        .to_string();
    serde_json::from_str(&text).expect("parse response")
}

/// Send a request expected to succeed; returns the `result`.
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} should succeed; full response: {resp}"
    );
    resp["result"].clone()
}

/// The single connected session's `current_run_id`, if any.
async fn only_session_current_run(state: &Arc<ServerState>) -> Option<String> {
    let session = {
        let sessions = state.sessions.lock().await;
        sessions
            .values()
            .next()
            .expect("one connected session")
            .clone()
    };
    let current = session.current_run_id.lock().await.clone();
    current
}

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

#[tokio::test]
async fn runs_start_returns_unique_ids_for_sequential_starts() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    let r1 = call_ok(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "first goal" }),
    )
    .await;
    let r2 = call_ok(
        &mut ws,
        "s2",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "second goal" }),
    )
    .await;

    let id1 = r1["run_id"].as_str().unwrap();
    let id2 = r2["run_id"].as_str().unwrap();
    assert!(!id1.is_empty() && !id2.is_empty());
    assert_ne!(id1, id2, "two sequential starts must mint distinct run_ids");
    assert_eq!(r1["agent_id"], "agent-a");
}

#[tokio::test]
async fn idempotency_key_dedups_runs_start() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    // A start with an idempotency_key adopts it as the run_id.
    let r1 = call_ok(
        &mut ws,
        "k1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "occurrence", "idempotency_key": "occ-42" }),
    )
    .await;
    assert_eq!(r1["run_id"], "occ-42", "the key becomes the run_id");
    assert_eq!(r1["agent_id"], "agent-a");

    // A second start with the SAME key returns the existing run, not a new one.
    let r2 = call_ok(
        &mut ws,
        "k2",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "occurrence again", "idempotency_key": "occ-42" }),
    )
    .await;
    assert_eq!(r2["run_id"], "occ-42", "same key returns the same run");
    assert_eq!(r2["agent_id"], "agent-a");
    assert_eq!(
        state.run_store.list_runs("agent-a").len(),
        1,
        "an idempotent re-start must not open a duplicate run"
    );

    // A different key opens a distinct run.
    let r3 = call_ok(
        &mut ws,
        "k3",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "different", "idempotency_key": "occ-99" }),
    )
    .await;
    assert_eq!(r3["run_id"], "occ-99");
    assert_eq!(
        state.run_store.list_runs("agent-a").len(),
        2,
        "a fresh key is a fresh run"
    );
}

#[tokio::test]
async fn current_run_id_is_set_before_runs_start_responds() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    let r = call_ok(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "do it" }),
    )
    .await;
    let run_id = r["run_id"].as_str().unwrap().to_string();

    // By the time the ack is received, the handler has already tagged the
    // session's current run (set BEFORE responding — KTD3). The recorder
    // a proposal submitted right after the ack would therefore read this.
    let current = only_session_current_run(&state).await;
    assert_eq!(
        current.as_deref(),
        Some(run_id.as_str()),
        "current_run_id must be set before runs.start responds"
    );
}

#[tokio::test]
async fn runs_complete_records_terminal_outcome_status() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    let started = call_ok(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "do it" }),
    )
    .await;
    let run_id = started["run_id"].as_str().unwrap().to_string();

    let ack = call_ok(
        &mut ws,
        "c1",
        "runs.complete",
        serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
    )
    .await;
    assert_eq!(ack["ok"], true);
    assert_eq!(ack["run_id"], run_id);

    // The run is now terminal with the reported outcome.
    let meta = state.run_meta(&run_id).await.expect("run recorded");
    match meta.termination {
        Some(RunTermination::Outcome { status, .. }) => {
            assert_eq!(status, car_ir::OutcomeStatus::Success);
        }
        other => panic!("expected Outcome termination, got {other:?}"),
    }
    // The session's current run was cleared by the complete.
    assert_eq!(only_session_current_run(&state).await, None);
}

#[tokio::test]
async fn completed_then_closed_run_is_not_incomplete() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    let started = call_ok(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "do it" }),
    )
    .await;
    let run_id = started["run_id"].as_str().unwrap().to_string();
    call_ok(
        &mut ws,
        "c1",
        "runs.complete",
        serde_json::json!({ "run_id": run_id, "outcome": success_outcome() }),
    )
    .await;

    // Close the connection AFTER the complete ack — the healthy
    // serve-mode close. Disconnect cleanup (with its grace window) must
    // leave the terminal Success record untouched, not overwrite it with
    // Incomplete.
    ws.close(None).await.ok();
    drop(ws);
    tokio::time::sleep(std::time::Duration::from_millis(600)).await;

    let meta = state.run_meta(&run_id).await.expect("run still recorded");
    match meta.termination {
        Some(RunTermination::Outcome { status, .. }) => {
            assert_eq!(
                status,
                car_ir::OutcomeStatus::Success,
                "a completed run must stay Success after close, not be raced to Incomplete"
            );
        }
        other => panic!("expected the run to remain terminal Outcome, got {other:?}"),
    }
}

#[tokio::test]
async fn mid_run_drop_yields_incomplete() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    let started = call_ok(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-a", "intent": "do it" }),
    )
    .await;
    let run_id = started["run_id"].as_str().unwrap().to_string();

    // Drop mid-run WITHOUT runs.complete. Past the grace window the
    // daemon must write an Incomplete terminal marker.
    ws.close(None).await.ok();
    drop(ws);
    tokio::time::sleep(std::time::Duration::from_millis(600)).await;

    let meta = state.run_meta(&run_id).await.expect("run still recorded");
    assert!(
        matches!(meta.termination, Some(RunTermination::Incomplete)),
        "a mid-run drop must record Incomplete, got {:?}",
        meta.termination
    );
}

#[tokio::test]
async fn runs_start_rejects_unresolvable_agent_but_name_fallback_records() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    // Guard against an ambient CAR_AGENT_ID from the test environment —
    // the "no resolvable id" assertion needs it absent.
    let restore = std::env::var("CAR_AGENT_ID").ok();
    std::env::remove_var("CAR_AGENT_ID");

    // No agent_id, no session.auth binding, no CAR_AGENT_ID, no usable
    // agent_name → rejected.
    let rejected = call_raw(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "intent": "orphan run" }),
    )
    .await;
    assert!(
        rejected.get("error").is_some(),
        "runs.start with no resolvable agent_id must be rejected; got {rejected}"
    );

    // The agent_name fallback synthesizes a deterministic id and records.
    let ok = call_ok(
        &mut ws,
        "s2",
        "runs.start",
        serde_json::json!({ "agent_name": "Bulldozer Agent", "intent": "one-shot run" }),
    )
    .await;
    let agent_id = ok["agent_id"].as_str().unwrap();
    assert_eq!(
        agent_id, "name:bulldozer-agent",
        "agent_name must synthesize a deterministic name-derived id"
    );
    assert!(!ok["run_id"].as_str().unwrap().is_empty());

    if let Some(v) = restore {
        std::env::set_var("CAR_AGENT_ID", v);
    }
}

/// Bind the single connected session to `agent_id` as if a `session.auth
/// { token, agent_id }` handshake had attached it (the supervised-child
/// path). Lets a test exercise the bound-session branch of
/// `resolve_run_agent_id` without standing up a supervisor.
async fn bind_only_session(state: &Arc<ServerState>, agent_id: &str) {
    let session = {
        let sessions = state.sessions.lock().await;
        sessions
            .values()
            .next()
            .expect("one connected session")
            .clone()
    };
    *session.agent_id.lock().await = Some(agent_id.to_string());
}

/// FIX 5 (trace forgery): a session BOUND to agent A must not be able to
/// record a run under a different agent B via `runs.start {agent_id:"B"}`.
/// The bound identity is authoritative — a mismatching param is rejected,
/// a matching param (or none) records under A.
#[tokio::test]
async fn bound_session_cannot_forge_run_under_other_agent() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().join("journals"));
    let addr = spawn_dispatcher(state.clone()).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}")).await.unwrap();

    // Establish the connection (create the session), then bind it to A.
    let started_a = call_ok(
        &mut ws,
        "s0",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-A", "intent": "warm up" }),
    )
    .await;
    assert_eq!(started_a["agent_id"], "agent-A");
    bind_only_session(&state, "agent-A").await;

    // Attempt to forge a run under agent-B from the A-bound session.
    let forged = call_raw(
        &mut ws,
        "s1",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-B", "intent": "forge under B" }),
    )
    .await;
    assert!(
        forged.get("error").is_some(),
        "a session bound to A must NOT be able to start a run as B; got {forged}"
    );
    // And nothing got recorded under B.
    assert!(
        state.run_store.list_runs("agent-B").is_empty(),
        "no run may be attributed to agent-B by the A-bound session"
    );

    // A matching explicit param is fine (redundant) — records under A.
    let matching = call_ok(
        &mut ws,
        "s2",
        "runs.start",
        serde_json::json!({ "agent_id": "agent-A", "intent": "legit, explicit" }),
    )
    .await;
    assert_eq!(matching["agent_id"], "agent-A");

    // No param at all derives the agent from the binding — records under A.
    let derived = call_ok(
        &mut ws,
        "s3",
        "runs.start",
        serde_json::json!({ "intent": "legit, derived" }),
    )
    .await;
    assert_eq!(
        derived["agent_id"], "agent-A",
        "an unspecified agent_id on a bound session derives the bound agent"
    );
}