use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
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";
fn state_with_engine(root: &TempDir) -> (Arc<ServerState>, Arc<InferenceEngine>) {
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}),
)
.await;
assert_eq!(
response["result"]["protocol_version"],
car_proto::PROTOCOL_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}");
}
#[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());
}