car-server-core 0.35.0

//! `cascade.run` — U-Mem Slice 5 live evolve loop, end-to-end through
//! `run_dispatch`. Verifies the daemon walks the cascade on caller-supplied
//! observed confidence (escalating cheapest-first), runs the real tier
//! mechanics, and — critically — gates the `human_expert` tier on the durable
//! `ApprovalLedger`, never on caller-attested confidence.

use car_memgine::MemgineEngine;
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
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::tungstenite::Message;
use tokio_tungstenite::{accept_async, connect_async, MaybeTlsStream, WebSocketStream};

fn shared_state(journal_dir: std::path::PathBuf) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    // Per-test approval journal — never the real user's ~/.car/approvals.jsonl.
    let approvals = journal_dir.join("approvals.jsonl");
    let cfg = ServerStateConfig::new(journal_dir)
        .with_shared_memgine(engine)
        .with_approval_journal(approvals);
    Arc::new(ServerState::with_config(cfg))
}

/// Open one connection to a fresh one-shot dispatcher bound to `state`. The
/// caller drives multiple frames on the SAME connection (so per-session state —
/// e.g. the permission ledger — persists across requests).
async fn connect(
    state: &Arc<ServerState>,
) -> WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>> {
    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");
    let st = state.clone();
    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(), st).await;
    });
    let url = format!("ws://{}", addr);
    let (ws, _resp) = connect_async(&url).await.expect("ws client connect");
    ws
}

async fn send_recv(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
    request: serde_json::Value,
) -> serde_json::Value {
    let body = serde_json::to_string(&request).expect("request to_string");
    ws.send(Message::Text(body.into())).await.expect("send");
    let resp = ws.next().await.expect("frame").expect("frame ok");
    let text = match resp {
        Message::Text(t) => t.to_string(),
        other => panic!("expected Text frame, got {:?}", other),
    };
    serde_json::from_str(&text).expect("parse response JSON")
}

/// One request on its own connection.
async fn rpc(state: &Arc<ServerState>, request: serde_json::Value) -> serde_json::Value {
    let mut ws = connect(state).await;
    send_recv(&mut ws, request).await
}

fn three_tier_policy(target: f64, budget: f64) -> serde_json::Value {
    serde_json::json!({
        "confidence_target": target,
        "budget": budget,
        "tiers": [
            { "tier": "self_reflect", "cost": 1.0, "expected_confidence": 0.0 },
            { "tier": "tool_verify",  "cost": 5.0, "expected_confidence": 0.0 },
            { "tier": "human_expert", "cost": 50.0, "expected_confidence": 0.0 }
        ]
    })
}

#[tokio::test]
async fn cascade_escalates_to_tool_verify_on_observed_confidence() {
    let tmp = TempDir::new().unwrap();
    let state = shared_state(tmp.path().join("journals"));

    let resp = rpc(
        &state,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "c1", "method": "cascade.run",
            "params": {
                "current_confidence": 0.2,
                "policy": three_tier_policy(0.8, 100.0),
                "observed": {
                    "self_reflect": { "confidence": 0.4 },
                    "tool_verify":  { "confidence": 0.85, "knowledge": "verified via search" }
                },
                "claim": "Paris is the capital of France"
            }
        }),
    )
    .await;

    let result = resp.get("result").expect("result");
    let run = result.get("run").expect("run");
    assert_eq!(
        run.get("accepted_tier").and_then(|v| v.as_str()),
        Some("tool_verify")
    );
    assert_eq!(
        run.get("steps").and_then(|v| v.as_array()).map(|a| a.len()),
        Some(2)
    );
    assert_eq!(
        run.get("knowledge").and_then(|v| v.as_str()),
        Some("verified via search")
    );
    // human_expert never ran, so no pending approval.
    assert!(result.get("pending_approval").is_none());
}

/// The load-bearing gate: a caller cannot self-certify the human tier. Even with
/// `human_expert` observed confidence at/above the target, an empty ledger must
/// NOT accept it — the run holds at pending.
#[tokio::test]
async fn cascade_human_tier_is_gated_on_ledger_not_caller_confidence() {
    let tmp = TempDir::new().unwrap();
    let state = shared_state(tmp.path().join("journals"));

    let resp = rpc(
        &state,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "c2", "method": "cascade.run",
            "params": {
                "current_confidence": 0.1,
                "policy": three_tier_policy(0.8, 100.0),
                "observed": {
                    "self_reflect": { "confidence": 0.3 },
                    "tool_verify":  { "confidence": 0.4 },
                    // Caller claims full confidence for the human tier — but no
                    // human has approved, so this must NOT win.
                    "human_expert": { "confidence": 0.99 }
                },
                "claim": "Our Q3 revenue exceeded plan"
            }
        }),
    )
    .await;

    let result = resp.get("result").expect("result");
    let run = result.get("run").expect("run");
    // NOT accepted on the caller's fabricated 0.99 — the ledger is empty.
    assert!(
        run.get("accepted_tier")
            .map(|v| v.is_null())
            .unwrap_or(true),
        "human_expert must not be accepted without a ledger decision: {run}"
    );
    // ...and the run surfaces the pending approval instead.
    let pending = result
        .get("pending_approval")
        .expect("pending_approval present");
    assert_eq!(
        pending.get("fingerprint").and_then(|v| v.as_str()),
        Some("cascade:Our Q3 revenue exceeded plan")
    );
}

/// A `human_expert` tier with a blank claim is rejected — otherwise every
/// claimless cascade would share the global `cascade:` fingerprint.
#[tokio::test]
async fn cascade_rejects_blank_claim_with_human_tier() {
    let tmp = TempDir::new().unwrap();
    let state = shared_state(tmp.path().join("journals"));

    let resp = rpc(
        &state,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "c3", "method": "cascade.run",
            "params": {
                "current_confidence": 0.1,
                "policy": three_tier_policy(0.99, 100.0),
                "observed": { "human_expert": { "confidence": 0.99 } }
                // no "claim"
            }
        }),
    )
    .await;

    assert!(
        resp.get("result").is_none(),
        "expected an error, got {resp}"
    );
    assert!(
        resp.get("error").is_some(),
        "expected a JSON-RPC error: {resp}"
    );
}

/// The positive path: once a human approval is recorded in the session ledger
/// (via the existing `permission.approve` flow), a re-run accepts the
/// `human_expert` tier. Driven on ONE connection so the per-session ledger
/// persists between the approve and the run.
#[tokio::test]
async fn cascade_accepts_human_after_ledger_approval() {
    let tmp = TempDir::new().unwrap();
    let state = shared_state(tmp.path().join("journals"));
    let mut ws = connect(&state).await;

    // Record a human approval for the claim's fingerprint.
    let approve = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "a1", "method": "permission.approve",
            "params": { "fingerprint": "cascade:gravity is real", "reason": "obvious" }
        }),
    )
    .await;
    assert!(approve.get("result").is_some(), "approve failed: {approve}");

    // Now the cascade reaches human_expert and the ledger says Approved → accept.
    let resp = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "r1", "method": "cascade.run",
            "params": {
                "current_confidence": 0.1,
                "policy": three_tier_policy(0.95, 100.0),
                "observed": {
                    "self_reflect": { "confidence": 0.3 },
                    "tool_verify":  { "confidence": 0.4 },
                    "human_expert": { "confidence": 0.99 }
                },
                "claim": "gravity is real"
            }
        }),
    )
    .await;

    let result = resp.get("result").expect("result");
    let run = result.get("run").expect("run");
    assert_eq!(
        run.get("accepted_tier").and_then(|v| v.as_str()),
        Some("human_expert"),
        "approved human tier should be accepted: {run}"
    );
    // Accepted, so no pending approval.
    assert!(result.get("pending_approval").is_none());
}

/// `dry_run` walks the cascade on observed confidence without running the
/// `self_reflect` side effect.
#[tokio::test]
async fn cascade_dry_run_skips_reflect_side_effect() {
    let tmp = TempDir::new().unwrap();
    let state = shared_state(tmp.path().join("journals"));

    let resp = rpc(
        &state,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "d1", "method": "cascade.run",
            "params": {
                "current_confidence": 0.2,
                "dry_run": true,
                "policy": three_tier_policy(0.8, 100.0),
                "observed": { "self_reflect": { "confidence": 0.85 } },
                "claim": "x"
            }
        }),
    )
    .await;

    let run = resp.get("result").and_then(|r| r.get("run")).expect("run");
    assert_eq!(
        run.get("accepted_tier").and_then(|v| v.as_str()),
        Some("self_reflect")
    );
    assert_eq!(
        run.get("knowledge").and_then(|v| v.as_str()),
        Some("self_reflect: dry_run (no reflect)")
    );
}