//! Daemon `permission.*` WS namespace — the per-session permission-tier
//! gate (survey "Code as Agent Harness" §3.4.3/§5.2.5) exercised over a
//! real `run_dispatch` WebSocket session.
//!
//! Asserts: classify maps actions to tiers; evaluate escalates a
//! full_access action to `needs_approval`; an approve (by fingerprint)
//! flips a later evaluate to `allow` and is durable for the connection; and
//! set_tier/get_tier round-trip.

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::{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)));
    // Point the SHARED approval ledger at a per-test path — 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))
}

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
}

async fn call_ok(
    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");
    let text = ws
        .next()
        .await
        .expect("frame")
        .expect("frame ok")
        .into_text()
        .expect("text")
        .to_string();
    let resp: serde_json::Value = serde_json::from_str(&text).expect("parse");
    assert!(resp.get("error").is_none(), "{method} failed: {resp}");
    resp["result"].clone()
}

fn proposal() -> serde_json::Value {
    serde_json::json!({
        "id": "p1", "source": "test",
        "actions": [
            { "id": "a1", "type": "state_read" },
            { "id": "a2", "type": "tool_call", "tool": "deploy_service" }
        ]
    })
}

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

    // Default standing tier is read_only (least privilege).
    let tier = call_ok(&mut ws, "1", "permission.get_tier", serde_json::json!({})).await;
    assert_eq!(tier["granted_tier"], "read_only");

    // classify: state_read is read_only, deploy_service is full_access.
    let cls = call_ok(
        &mut ws,
        "2",
        "permission.classify",
        serde_json::json!({ "proposal": proposal() }),
    )
    .await;
    let rows = cls["classifications"].as_array().unwrap();
    assert_eq!(rows[0]["required_tier"], "read_only");
    assert_eq!(rows[1]["required_tier"], "full_access");

    // evaluate: deploy needs approval (mandatory HITL at full_access).
    let ev = call_ok(
        &mut ws,
        "3",
        "permission.evaluate",
        serde_json::json!({ "proposal": proposal() }),
    )
    .await;
    let decisions = ev["decisions"].as_array().unwrap();
    assert_eq!(decisions[0]["decision"], "allow"); // read within sandbox_edit
    assert_eq!(decisions[1]["decision"], "needs_approval");
    let fingerprint = decisions[1]["fingerprint"].as_str().unwrap().to_string();

    // pending surfaces exactly the deploy action.
    let pend = call_ok(
        &mut ws,
        "4",
        "permission.pending",
        serde_json::json!({ "proposal": proposal() }),
    )
    .await;
    assert_eq!(pend["pending"].as_array().unwrap().len(), 1);

    // approve by fingerprint — durable for the connection.
    let rec = call_ok(
        &mut ws,
        "5",
        "permission.approve",
        serde_json::json!({
            "fingerprint": fingerprint,
            "required_tier": "full_access",
            "reviewer": "matt",
            "reason": "reviewed the deploy plan",
        }),
    )
    .await;
    assert_eq!(rec["decision"], "approved");

    // Re-evaluate: the deploy is now allowed.
    let ev2 = call_ok(
        &mut ws,
        "6",
        "permission.evaluate",
        serde_json::json!({ "proposal": proposal() }),
    )
    .await;
    assert_eq!(ev2["decisions"].as_array().unwrap()[1]["decision"], "allow");

    // set_tier round-trips.
    let set = call_ok(
        &mut ws,
        "7",
        "permission.set_tier",
        serde_json::json!({ "tier": "full_access" }),
    )
    .await;
    assert_eq!(set["granted_tier"], "full_access");
}

/// neo review #6a: an under-scoped approval (by fingerprint, read_only) for
/// a full_access action must NOT let a later evaluate return `allow` — the
/// core gate's tier-scoped guard holds over the WS surface too.
#[tokio::test]
async fn permission_ws_underscoped_approval_does_not_widen_access() {
    let tmp = TempDir::new().unwrap();
    let state = loopback_state(tmp.path().to_path_buf());
    let addr = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect");

    let ev = call_ok(
        &mut ws,
        "1",
        "permission.evaluate",
        serde_json::json!({ "proposal": proposal() }),
    )
    .await;
    let fingerprint = ev["decisions"].as_array().unwrap()[1]["fingerprint"]
        .as_str()
        .unwrap()
        .to_string();

    // Approve, but claim only read_only — below the action's true full_access.
    call_ok(
        &mut ws,
        "2",
        "permission.approve",
        serde_json::json!({
            "fingerprint": fingerprint,
            "required_tier": "read_only",
            "reviewer": "ignored-server-stamps-this",
            "reason": "under-scoped",
        }),
    )
    .await;

    // Re-evaluate: still needs approval — the stale low-tier approval can't
    // bypass the full_access gate.
    let ev2 = call_ok(
        &mut ws,
        "3",
        "permission.evaluate",
        serde_json::json!({ "proposal": proposal() }),
    )
    .await;
    assert_eq!(
        ev2["decisions"].as_array().unwrap()[1]["decision"],
        "needs_approval"
    );
}
