car-server-core 0.55.0

use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};

use car_inference::schema::ModelSource;
use car_inference::{InferenceConfig, InferenceEngine};
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio_tungstenite::{accept_async, connect_async, tungstenite::Message};

type Ws =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

const HOST_TOKEN: &str = "model-host-token-cccccccccccccccccccccccc";
const ORDINARY_TOKEN: &str = "model-session-token-dddddddddddddddddddddddd";

/// This integration-test binary exercises registry availability refreshes,
/// which clear the durable Parslee credential verdict. Pin its process-global
/// CAR root once, before any engine is built, so concurrent tests share one
/// stable scratch destination rather than racing per-test environment changes.
fn test_car_home() -> &'static Path {
    static ROOT: OnceLock<PathBuf> = OnceLock::new();
    ROOT.get_or_init(|| {
        let root =
            std::env::temp_dir().join(format!("car-models-surface-gate-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).expect("create process-scoped CAR_HOME");
        // SAFETY: OnceLock serializes the one mutation before any engine in
        // this test binary is constructed; the value never changes afterward.
        unsafe { std::env::set_var(car_home::ENV_VAR, &root) };
        root
    })
}

fn state_with_engine(root: &TempDir) -> (Arc<ServerState>, Arc<InferenceEngine>) {
    let _ = test_car_home();
    let engine = Arc::new(InferenceEngine::new(InferenceConfig {
        state_root: root.path().join("state"),
        models_dir: root.path().join("models"),
        ..InferenceConfig::default()
    }));
    let state = Arc::new(ServerState::with_config(
        ServerStateConfig::new(root.path().join("journal")).with_inference(engine.clone()),
    ));
    state
        .install_host_token(HOST_TOKEN.to_string())
        .expect("install host token");
    (state, engine)
}

async fn spawn_dispatcher(state: Arc<ServerState>) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)))
        .await
        .unwrap();
    let address = listener.local_addr().unwrap();
    tokio::spawn(async move {
        let (stream, peer) = listener.accept().await.unwrap();
        let socket = accept_async(stream).await.unwrap();
        let (write, read) = socket.split();
        let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
    });
    address
}

async fn call(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
    .unwrap();
    loop {
        let text = ws.next().await.unwrap().unwrap().into_text().unwrap();
        let value: serde_json::Value = serde_json::from_str(&text).unwrap();
        if value.get("id").and_then(serde_json::Value::as_str) == Some(id) {
            return value;
        }
    }
}

async fn negotiate(ws: &mut Ws) {
    let response = call(
        ws,
        "handshake",
        "server.handshake",
        serde_json::json!({
            "protocol_version":car_proto::PROTOCOL_VERSION,
            "required_capabilities": car_proto::REQUIRED_CLIENT_CAPABILITIES,
            "optional_capabilities": [],
        }),
    )
    .await;
    assert_eq!(
        response["result"]["protocol_version"],
        car_proto::PROTOCOL_VERSION
    );
}

#[test]
fn harness_pins_credential_state_under_a_process_scratch_root() {
    let root = TempDir::new().unwrap();
    let _ = state_with_engine(&root);
    car_inference::parslee_credential::clear_credential_rejected();
    let credential_state = car_inference::parslee_credential::credential_state_path();
    assert_eq!(car_home::root().as_deref(), Some(test_car_home()));
    assert_eq!(credential_state.parent(), Some(test_car_home()));
    assert!(
        credential_state.exists(),
        "engine construction must exercise the credential-state write under scratch: {}",
        credential_state.display()
    );
}

#[tokio::test]
async fn catalog_snapshot_is_valid_and_list_unified_remains_an_array() {
    let root = TempDir::new().unwrap();
    let (state, _) = state_with_engine(&root);
    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();
    negotiate(&mut ws).await;

    let snapshot_response = call(
        &mut ws,
        "snapshot",
        "models.catalog_snapshot",
        serde_json::json!({}),
    )
    .await;
    let snapshot: car_inference::CatalogSnapshot =
        serde_json::from_value(snapshot_response["result"].clone()).unwrap();
    eprintln!("C1_WS_CATALOG_SNAPSHOT={snapshot_response}");
    snapshot
        .validate()
        .expect("surface snapshot must self-validate");
    assert!(!snapshot.models.is_empty());

    let legacy = call(
        &mut ws,
        "legacy-list",
        "models.list_unified",
        serde_json::json!({}),
    )
    .await;
    eprintln!(
        "C1_WS_LIST_UNIFIED_SHAPE={}",
        if legacy["result"].is_array() {
            "array"
        } else {
            "non-array"
        }
    );
    assert!(
        legacy["result"].is_array(),
        "v2 list_unified result shape must remain a bare array: {legacy}"
    );
}

/// car#1399: every `models.list_unified` row carries the machine-fit
/// annotation, computed by the daemon; clients filter on it. This is the
/// distinguishing test for that change — on a daemon that predates it the
/// rows carry no `fit` key at all.
#[tokio::test]
async fn list_unified_rows_carry_a_fit_annotation_and_search_keeps_family() {
    let root = TempDir::new().unwrap();
    let (state, _) = state_with_engine(&root);
    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();
    negotiate(&mut ws).await;

    let listed = call(
        &mut ws,
        "list",
        "models.list_unified",
        serde_json::json!({}),
    )
    .await;
    let rows = listed["result"]
        .as_array()
        .unwrap_or_else(|| panic!("list_unified must stay a bare array: {listed}"));
    assert!(!rows.is_empty());
    let mut saw_local = false;
    let mut saw_remote = false;
    for row in rows {
        let id = row["id"].as_str().unwrap();
        let fit = row["fit"]
            .as_str()
            .unwrap_or_else(|| panic!("{id}: `fit` must be a string, row was {row}"));
        assert!(
            matches!(fit, "fits" | "too_big" | "unknown"),
            "{id}: fit was {fit}"
        );
        assert!(
            row["platform_compatible"].is_boolean(),
            "{id}: platform_compatible must be a bool, row was {row}"
        );
        assert!(
            row["estimated_peak_mb"].is_null() || row["estimated_peak_mb"].is_u64(),
            "{id}: estimated_peak_mb must be a number or null, row was {row}"
        );
        assert!(
            row["deprecated"].is_boolean(),
            "{id}: deprecated must be a bool, row was {row}"
        );
        // Every pre-existing field is still there.
        for key in [
            "available",
            "is_local",
            "weights_ready",
            "downloads_weights",
            "cost",
            "car_enabled",
            "can_remove",
            "in_use",
        ] {
            assert!(
                !row[key].is_null() || key == "management_evidence",
                "{id}: {key} missing"
            );
        }
        if row["is_local"].as_bool().unwrap() {
            saw_local = true;
            assert!(
                row["family"].is_string(),
                "{id}: local rows publish family: {row}"
            );
        } else {
            saw_remote = true;
            // Remote rows: their memory is not this machine's, and this view
            // carries no upstream identifier for managed aliases.
            assert_eq!(fit, "fits", "{id}: remote rows are fits: {row}");
            assert_eq!(row["platform_compatible"], true, "{id}: {row}");
            assert!(row["estimated_peak_mb"].is_null(), "{id}: {row}");
            assert!(
                row["family"].is_null(),
                "{id}: remote rows publish no family: {row}"
            );
            assert!(
                row["version"].is_null(),
                "{id}: remote rows publish no version: {row}"
            );
        }
    }
    assert!(
        saw_local && saw_remote,
        "the builtin catalog has both kinds of row"
    );

    // `models.search` still names the family of every row (its documented
    // job), now alongside the same fit annotation.
    let searched = call(&mut ws, "search", "models.search", serde_json::json!({})).await;
    let entries = searched["result"]["models"].as_array().unwrap();
    assert_eq!(entries.len(), rows.len());
    for entry in entries {
        let id = entry["id"].as_str().unwrap();
        assert!(
            entry["family"].is_string(),
            "{id}: search entry family: {entry}"
        );
        assert!(
            entry["version"].is_string(),
            "{id}: search entry version: {entry}"
        );
        assert!(entry["fit"].is_string(), "{id}: search entry fit: {entry}");
        assert!(entry["tags"].is_array(), "{id}: search entry tags: {entry}");
    }
}

/// The exact key set of a `models.list_unified` row and of a `models.search`
/// entry, pinned. `ModelSearchEntry` flattens `ModelInfo`, so any field added
/// to the unified row also lands on search — this test makes that a
/// deliberate, reviewed change rather than an incidental one. Note the
/// naming: the flattened row keeps `models.list_unified`'s snake_case
/// (`estimated_peak_mb`, `platform_compatible`); `ModelSearchEntry`'s
/// `camelCase` rename reaches only its own fields, all single words.
#[tokio::test]
async fn list_unified_and_search_entries_carry_exactly_these_keys() {
    use std::collections::BTreeSet;

    const UNIFIED_KEYS: &[&str] = &[
        "id",
        "name",
        "provider",
        "capabilities",
        "param_count",
        "size_mb",
        "context_length",
        "available",
        "is_local",
        "operator_managed_external_runtime",
        "weights_ready",
        "downloads_weights",
        "max_output_tokens",
        "public_benchmarks",
        "cost",
        "car_enabled",
        "can_remove",
        "in_use",
        "management_evidence",
        // car#1399, additive.
        "fit",
        "estimated_peak_mb",
        "platform_compatible",
        "deprecated",
        "family",
        "version",
    ];
    /// Search's own fields. `family` and `version` are NOT listed here: they
    /// reach the search entry through the flattened row, and search
    /// overwrites them for every row.
    const SEARCH_ONLY_KEYS: &[&str] = &["tags", "pullable", "upgrade"];

    let root = TempDir::new().unwrap();
    let (state, _) = state_with_engine(&root);
    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();
    negotiate(&mut ws).await;

    let keys = |row: &serde_json::Value| -> BTreeSet<String> {
        row.as_object()
            .unwrap_or_else(|| panic!("row must be an object: {row}"))
            .keys()
            .cloned()
            .collect()
    };
    let expected_unified: BTreeSet<String> = UNIFIED_KEYS.iter().map(|k| k.to_string()).collect();
    let expected_search: BTreeSet<String> = UNIFIED_KEYS
        .iter()
        .chain(SEARCH_ONLY_KEYS)
        .map(|k| k.to_string())
        .collect();

    let listed = call(
        &mut ws,
        "list",
        "models.list_unified",
        serde_json::json!({}),
    )
    .await;
    let rows = listed["result"].as_array().unwrap();
    for row in rows {
        assert_eq!(
            keys(row),
            expected_unified,
            "list_unified keys for {}",
            row["id"]
        );
    }

    let searched = call(&mut ws, "search", "models.search", serde_json::json!({})).await;
    let entries = searched["result"]["models"].as_array().unwrap();
    assert_eq!(entries.len(), rows.len());
    for entry in entries {
        assert_eq!(
            keys(entry),
            expected_search,
            "search keys for {}",
            entry["id"]
        );
    }

    // One local row and one managed remote alias row, in both views.
    let unified_by_id = |id: &str| {
        rows.iter()
            .find(|row| row["id"] == id)
            .unwrap_or_else(|| panic!("{id} missing from list_unified"))
    };
    let search_by_id = |id: &str| {
        entries
            .iter()
            .find(|entry| entry["id"] == id)
            .unwrap_or_else(|| panic!("{id} missing from search"))
    };
    let local_id = rows
        .iter()
        .find(|row| row["is_local"] == true && row["downloads_weights"] == true)
        .map(|row| row["id"].as_str().unwrap().to_string())
        .expect("the builtin catalog has a downloadable local row");
    let alias_id = rows
        .iter()
        .find(|row| {
            row["id"]
                .as_str()
                .is_some_and(|id| id.starts_with("parslee/openrouter/"))
        })
        .map(|row| row["id"].as_str().unwrap().to_string())
        .expect("the builtin catalog registers the managed parslee/openrouter/* aliases");

    let local = unified_by_id(&local_id);
    assert!(
        local["family"].is_string() && local["version"].is_string(),
        "{local}"
    );
    let alias = unified_by_id(&alias_id);
    assert_eq!(alias["is_local"], false, "{alias}");
    assert!(
        alias["family"].is_null() && alias["version"].is_null(),
        "list_unified publishes no family/version for a managed alias: {alias}"
    );

    for (id, unified) in [(&local_id, local), (&alias_id, alias)] {
        let entry = search_by_id(id);
        assert!(
            entry["family"].is_string(),
            "search names family for {id}: {entry}"
        );
        assert!(
            entry["version"].is_string(),
            "search names version for {id}: {entry}"
        );
        // The fit annotation is the same one list_unified publishes.
        for key in [
            "fit",
            "estimated_peak_mb",
            "platform_compatible",
            "deprecated",
        ] {
            assert_eq!(
                entry[key], unified[key],
                "{id}.{key} differs between search and list_unified"
            );
        }
        assert!(entry["fit"].is_string(), "{id}: {entry}");
        assert!(entry["platform_compatible"].is_boolean(), "{id}: {entry}");
        assert!(entry["deprecated"].is_boolean(), "{id}: {entry}");
    }
    assert_eq!(search_by_id(&local_id)["family"], local["family"]);
    assert_eq!(search_by_id(&local_id)["version"], local["version"]);
}

async fn become_host(ws: &mut Ws) {
    let response = call(
        ws,
        "host",
        "session.auth",
        serde_json::json!({"host_token":HOST_TOKEN}),
    )
    .await;
    assert_eq!(response["result"]["role"], "host", "{response}");
}

#[tokio::test]
async fn distinct_host_auth_envelope_grants_role_and_allows_management_mutation() {
    let root = TempDir::new().unwrap();
    let (state, _) = state_with_engine(&root);
    state
        .install_auth_token(ORDINARY_TOKEN.to_string())
        .expect("install ordinary auth token");
    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();

    let auth = call(
        &mut ws,
        "host-auth",
        "session.auth",
        serde_json::json!({"host_token":HOST_TOKEN}),
    )
    .await;
    assert_eq!(auth["result"]["role"], "host", "{auth}");
    negotiate(&mut ws).await;

    let set = call(
        &mut ws,
        "host-mutation",
        "models.resource_policy.set",
        serde_json::json!({"profile":"custom","custom_max_model_mb":0}),
    )
    .await;
    assert_eq!(set["result"]["policy"]["custom_max_model_mb"], 0, "{set}");
}

/// A post-set engine policy remains the source of truth even if the persisted
/// file changes underneath the running daemon. Before car#1476 list/search used
/// the engine while recommend/setup reopened the file and disagreed.
#[tokio::test]
async fn set_policy_drives_fit_recommend_and_setup_through_one_active_accessor() {
    use car_inference::resource_policy::{
        FileResourcePolicyRepository, ResourcePolicy, ResourcePolicyRepository,
    };

    const MODEL_ID: &str = "qwen/qwen3-4b:q4_k_m";
    let root = TempDir::new().unwrap();
    let (state, engine) = state_with_engine(&root);
    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();
    negotiate(&mut ws).await;
    become_host(&mut ws).await;

    let set = call(
        &mut ws,
        "set-active",
        "models.resource_policy.set",
        serde_json::json!({"profile":"custom","custom_max_model_mb":0}),
    )
    .await;
    assert_eq!(set["result"]["policy"]["custom_max_model_mb"], 0, "{set}");
    assert_eq!(
        engine.active_local_resource_policy().policy,
        ResourcePolicy::custom_gb(0.0).unwrap()
    );

    // Distinguishing seam: resource_policy.set persisted and applied Custom,
    // then an external file replacement makes a fresh repository read say
    // Everyday. The running engine must still answer every fit surface from
    // the already-applied Custom policy.
    let repository = FileResourcePolicyRepository::new(engine.config.state_root.clone());
    repository.save(&ResourcePolicy::everyday()).unwrap();
    assert_eq!(repository.load().unwrap(), ResourcePolicy::everyday());

    let listed = call(
        &mut ws,
        "active-list",
        "models.list_unified",
        serde_json::json!({}),
    )
    .await;
    let row = listed["result"]
        .as_array()
        .unwrap()
        .iter()
        .find(|row| row["id"] == MODEL_ID)
        .expect("cross-platform local fixture row");
    assert_eq!(row["fit"], "too_big", "{row}");

    let recommended = call(
        &mut ws,
        "active-recommend",
        "models.recommend",
        serde_json::json!({"use_case":"assistant","tier":"balanced"}),
    )
    .await;
    assert!(
        recommended["result"]["not_enough_memory"]
            .as_array()
            .unwrap()
            .iter()
            .any(|row| row["model_id"] == MODEL_ID),
        "recommend must use active Custom(0), not persisted Everyday: {recommended}"
    );

    let setup = call(
        &mut ws,
        "active-setup",
        "models.setup_plan",
        serde_json::json!({"use_case":"assistant","tier":"balanced"}),
    )
    .await;
    assert_eq!(setup["result"]["resource_policy"], set["result"]["policy"]);
    assert!(
        setup["result"]["needs_more_memory"]
            .as_array()
            .unwrap()
            .iter()
            .any(|row| row["model_id"] == MODEL_ID),
        "setup must agree with list/recommend after resource_policy.set: {setup}"
    );
}

#[tokio::test]
async fn custom_policy_round_trips_zero_and_storage_roots_are_host_only() {
    let root = TempDir::new().unwrap();
    let (state, engine) = state_with_engine(&root);
    let local_id = engine
        .unified_registry
        .all()
        .find(|schema| matches!(schema.source, ModelSource::Local { .. }))
        .unwrap()
        .id
        .clone();
    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();
    negotiate(&mut ws).await;

    for method in ["models.pull", "models.install"] {
        let denied = call(
            &mut ws,
            &format!("denied-{method}"),
            method,
            serde_json::json!({"name":"missing/model"}),
        )
        .await;
        assert!(
            denied["error"]["message"]
                .as_str()
                .unwrap_or_default()
                .contains("host-management role"),
            "{method} must require host authority before model lookup: {denied}"
        );
    }

    let denied = call(
        &mut ws,
        "denied",
        "models.storage_roots",
        serde_json::json!({}),
    )
    .await;
    assert!(denied["error"]["message"]
        .as_str()
        .unwrap_or_default()
        .contains("host-management role"));
    become_host(&mut ws).await;
    let strict_pull = call(
        &mut ws,
        "strict-pull",
        "models.pull",
        serde_json::json!({"name":"missing/model","path":"/tmp/escape"}),
    )
    .await;
    assert!(
        strict_pull["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .contains("invalid model-management params"),
        "pull must reject unknown fields: {strict_pull}"
    );

    let set = call(
        &mut ws,
        "set",
        "models.resource_policy.set",
        serde_json::json!({"profile":"custom","custom_max_model_mb":0}),
    )
    .await;
    assert_eq!(set["result"]["policy"]["custom_max_model_mb"], 0, "{set}");
    let get = call(
        &mut ws,
        "get",
        "models.resource_policy.get",
        serde_json::json!({}),
    )
    .await;
    assert_eq!(get["result"]["policy"]["custom_max_model_mb"], 0, "{get}");
    let preflight = call(
        &mut ws,
        "preflight",
        "models.preflight",
        serde_json::json!({"model_id":local_id}),
    )
    .await;
    assert_eq!(
        preflight["result"]["verdict"], "disabled_by_policy",
        "{preflight}"
    );
    let roots = call(
        &mut ws,
        "roots",
        "models.storage_roots",
        serde_json::json!({}),
    )
    .await;
    for field in [
        "state_root",
        "models_dir",
        "hf_home",
        "hf_hub",
        "install_receipts_dir",
        "management_state_dir",
    ] {
        assert!(
            roots["result"][field].is_string(),
            "missing {field}: {roots}"
        );
    }
    let unknown = call(
        &mut ws,
        "unknown",
        "models.resource_policy.get",
        serde_json::json!({"path":"/tmp/escape"}),
    )
    .await;
    assert!(
        unknown.get("error").is_some(),
        "unknown fields must fail: {unknown}"
    );
}

#[cfg(unix)]
#[tokio::test]
async fn host_adopt_and_remove_preserve_shared_cache() {
    use std::os::unix::fs::symlink;

    let root = TempDir::new().unwrap();
    let (state, engine) = state_with_engine(&root);
    let schema = engine
        .unified_registry
        .all()
        .find(|schema| matches!(schema.source, ModelSource::Local { .. }))
        .unwrap()
        .clone();
    let shared = root.path().join("hf/snapshot");
    std::fs::create_dir_all(&shared).unwrap();
    std::fs::write(shared.join("model.gguf"), b"weights").unwrap();
    std::fs::write(shared.join("tokenizer.json"), b"{}").unwrap();
    std::fs::write(shared.join("sentinel"), b"preserve").unwrap();
    std::fs::create_dir_all(&engine.config.models_dir).unwrap();
    symlink(&shared, engine.config.models_dir.join(&schema.name)).unwrap();

    let address = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{address}")).await.unwrap();
    negotiate(&mut ws).await;
    become_host(&mut ws).await;
    let adopted = call(
        &mut ws,
        "adopt",
        "models.adopt",
        serde_json::json!({"model_id":schema.id}),
    )
    .await;
    assert_eq!(adopted["result"]["can_remove"], true, "{adopted}");
    let removed = call(
        &mut ws,
        "remove",
        "models.remove",
        serde_json::json!({"model_id":schema.id}),
    )
    .await;
    assert_eq!(removed["result"]["removed_from_car"], true, "{removed}");
    assert!(shared.join("sentinel").exists());
    assert!(!engine.config.models_dir.join(&schema.name).exists());
}