diff --git a/crates/pa-core/src/models/registry/tests.rs b/crates/pa-core/src/models/registry/tests.rs index 888f71985..7aaa4bcae 100644 --- a/crates/pa-core/src/models/registry/tests.rs +++ b/crates/pa-core/src/models/registry/tests.rs @@ -164,8 +164,28 @@ fn header_precedence_model_over_provider() { ); } -#[test] -fn live_scope_keyed_cache_merges_over_the_compiled_prime_inference_models() { +async fn serve_catalog_once(payload: serde_json::Value) -> (String, tokio::task::JoinHandle<()>) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/models", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + let mut chunk = [0; 1024]; + let read = socket.read(&mut chunk).await.unwrap(); + assert!(read > 0); + request.extend_from_slice(&chunk[..read]); + } + let body = payload.to_string(); + socket.write_all(format!("HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + }); + (url, task) +} + +#[tokio::test] +#[allow(clippy::too_many_lines)] +async fn live_scope_keyed_cache_merges_over_the_compiled_prime_inference_models() { let dir = tempfile::tempdir().unwrap(); let models_path = dir.path().join("models.json"); std::fs::write( @@ -177,14 +197,14 @@ fn live_scope_keyed_cache_merges_over_the_compiled_prime_inference_models() { } } }"#, ) .unwrap(); - // A live-catalog snapshot scoped to these credentials: one compiled - // model repriced, one new entry, well past the coverage floor so - // the build accepts it. The scope-keyed cache serves this scope's - // snapshot only; another credential never sees it. - let auth = auth_without_env(&serde_json::json!({ - "prime-inference": { "type": "api_key", "key": "live-key", - "primeTeam": { "teamId": "team-1", "name": "Team 1" } } - })); + // Reprice one compiled model and admit one new entry through current + // scoped discovery. Another credential must not see this metadata. + let auth = || { + auth_without_env(&serde_json::json!({ + "prime-inference": { "type": "api_key", "key": "live-key", + "primeTeam": { "teamId": "team-1", "name": "Team 1" } } + })) + }; let compiled = pa_models::transports::prime_inference_offline_entries(); let repriced = &compiled[0]; let mut entries = Vec::new(); @@ -214,20 +234,31 @@ fn live_scope_keyed_cache_merges_over_the_compiled_prime_inference_models() { "modalities": { "input": ["text"], "output": ["text"] }, }, })); - let scope = pa_models::prime_inference::scope_key("live-key", "team-1"); - let snapshot = serde_json::json!({ - "url": format!("{}/models", super::super::prime_inference::PRIME_INFERENCE_BASE_URL), - "scope": scope, - "fetchedAt": 1, - "payload": { "data": entries }, - }); - std::fs::create_dir_all(dir.path().join("models")).unwrap(); - std::fs::write( - dir.path().join("models/prime-inference-models-cache.json"), - serde_json::to_vec(&snapshot).unwrap(), - ) - .unwrap(); - let mut registry = ModelRegistry::create(auth, &models_path); + let (url, response) = serve_catalog_once(serde_json::json!({ "data": entries })).await; + let catalog = || { + Arc::new(pa_models::ModelCatalog::with_discovery_urls( + Some(dir.path().join("models")), + None, + pa_models::fetch::MODEL_CATALOG_URL, + super::super::prime_inference::PRIME_INFERENCE_BASE_URL, + &std::collections::HashMap::from([("prime-inference".to_string(), url.clone())]), + pa_models::ollama_cloud::BASE_URL, + )) + }; + super::super::catalog_chain::install_catalog(&models_path, catalog()); + let mut writer = ModelRegistry::create(auth(), &models_path); + let report = writer + .refresh_catalog_sources(Some(&["prime-inference-public".to_string()])) + .await; + assert_eq!(report.sources.len(), 1); + assert_eq!( + report.sources[0].status, + pa_types::model_refresh::ModelRefreshStatus::Updated + ); + response.await.unwrap(); + // A new catalog must read the public facade's durable scoped metadata. + super::super::catalog_chain::install_catalog(&models_path, catalog()); + let mut registry = ModelRegistry::create(auth(), &models_path); let all = registry.get_all().to_vec(); // The live repriced model replaced its compiled template (other // providers may serve the same id; match the provider too). @@ -272,15 +303,10 @@ fn live_scope_keyed_cache_merges_over_the_compiled_prime_inference_models() { } #[test] -fn a_missing_or_corrupt_cache_falls_back_to_the_bundled_catalog() { +fn invalid_models_config_keeps_the_bundled_catalog() { let dir = tempfile::tempdir().unwrap(); let models_path = dir.path().join("models.json"); std::fs::write(&models_path, "{ not json").unwrap(); - std::fs::write( - dir.path().join("prime-inference-models-cache.json"), - "{ not json", - ) - .unwrap(); let registry = ModelRegistry::create(auth_with(&serde_json::json!({})), &models_path); assert!(registry .get_all() diff --git a/crates/pa-core/tests/offline_catalog_startup.rs b/crates/pa-core/tests/offline_catalog_startup.rs index 7511dc60c..945808e7f 100644 --- a/crates/pa-core/tests/offline_catalog_startup.rs +++ b/crates/pa-core/tests/offline_catalog_startup.rs @@ -7,7 +7,9 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; -use pa_models::{DiscoveryCredential, DiscoveryCredentials, ModelCatalog, PrimeCredentials}; +use pa_models::{ + CatalogFormat, CatalogInputs, ModelCatalog, RefreshPolicy, RefreshRequest, ResolvedSource, +}; use serde_json::json; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -26,6 +28,88 @@ fn files(dir: &Path) -> BTreeMap> { .collect() } +fn fixture_catalog(dir: &Path, base: &str) -> ModelCatalog { + let urls = [ + "openrouter", + "openai", + "anthropic", + "google", + "openai-codex", + "prime-inference", + ] + .into_iter() + .map(|provider| (provider.to_string(), format!("{base}/{provider}"))) + .collect(); + ModelCatalog::with_discovery_urls( + Some(dir.into()), + None, + &format!("{base}/catalog"), + pa_models::prime_inference::PRIME_INFERENCE_BASE_URL, + &urls, + base, + ) +} + +fn fixture_inputs(base: &str, configured: bool) -> CatalogInputs { + let mut sources = vec![ResolvedSource { + id: "prime-inference-public".into(), + provider: "prime-inference".into(), + format: CatalogFormat::PrimeInference, + url: format!("{base}/prime-inference"), + headers: if configured { + vec![ + ("authorization".into(), "Bearer fixture-key".into()), + ("x-prime-team-id".into(), "fixture-team".into()), + ] + } else { + Vec::new() + }, + defaults: pa_models::transports::prime_inference_offline_entries() + .into_iter() + .next(), + is_current: None, + }]; + if configured { + for (provider, format) in [ + ("openai", CatalogFormat::OpenAi), + ("openrouter", CatalogFormat::OpenRouter), + ("anthropic", CatalogFormat::Anthropic), + ("google", CatalogFormat::Google), + ("openai-codex", CatalogFormat::Codex), + ("ollama-cloud", CatalogFormat::Ollama), + ] { + sources.push(ResolvedSource { + id: provider.into(), + provider: provider.into(), + format, + url: format!("{base}/{provider}"), + headers: vec![("authorization".into(), "Bearer fixture-key".into())], + defaults: None, + is_current: None, + }); + } + let mut private = sources[0].clone(); + private.id = "prime-inference-private".into(); + private.format = CatalogFormat::PrimeInferencePrivate; + sources.push(private); + } + CatalogInputs { + sources, + seeds: Vec::new(), + } +} + +fn prime_payload() -> serde_json::Value { + json!({"data": pa_models::transports::prime_inference_offline_entries() + .iter().map(|model| json!({ + "id": model.id, "display_name": model.name, + "pricing": {"input_usd_per_mtok": 0, "output_usd_per_mtok": 0}, + "specs": {"context_window": model.context_window, "max_output_tokens": model.max_tokens, + "supports_reasoning": model.reasoning, + "modalities": {"input": ["text"], "output": ["text"]}}, + })).collect::>()}) +} + // Keep the isolated child actions together so the env-cleared entry point is explicit. #[allow(clippy::too_many_lines)] #[test] @@ -39,90 +123,95 @@ fn startup_child() { .unwrap(); runtime.block_on(async { match action.as_str() { - "models" => { - let dir = tempfile::tempdir().unwrap(); + "models-warm" | "models" => { + let base = std::env::var("PA_OFFLINE_LOOPBACK_URL").unwrap(); + let dir = std::path::PathBuf::from(std::env::var_os("HOME").unwrap()) + .join("agent/models"); + std::fs::create_dir_all(&dir).unwrap(); + let catalog = Arc::new(fixture_catalog(&dir, &base)); + let inputs = fixture_inputs(&base, true); + if action == "models-warm" { + let publication = catalog + .refresh( + &inputs, + &RefreshRequest { + policy: RefreshPolicy::Force, + sources: Some(vec![ + "curated".into(), + "prime-inference-public".into(), + ]), + }, + ) + .await; + assert_eq!(publication.sources.len(), 2); + assert!(publication.sources.iter().all(|report| report.status + == pa_types::model_refresh::ModelRefreshStatus::Updated)); + return; + } + let before = files(&dir); + let resolved = catalog.snapshot(&inputs).models; let model = pa_models::transports::compiled_models() .iter() .find(|model| model.provider == "anthropic") .unwrap() .clone(); - let payload = json!({"schemaVersion": 1, "models": [model]}); - let cache = dir.path().join("provider-model-catalog.v1.json"); - std::fs::write( - &cache, - serde_json::to_vec(&json!({ - "url": pa_models::fetch::MODEL_CATALOG_URL, - "scope": "public", "fetchedAt": 1, "checkedAt": 2, - "payload": payload, - })) - .unwrap(), - ) - .unwrap(); - let prime_payload: Vec<_> = pa_models::transports::prime_inference_offline_entries() - .iter() - .map(|model| json!({ - "id": model.id, "display_name": model.name, - "pricing": {"input_usd_per_mtok": 0, "output_usd_per_mtok": 0}, - "specs": {"context_window": model.context_window, - "max_output_tokens": model.max_tokens, - "supports_reasoning": model.reasoning, - "modalities": {"input": ["text"], "output": ["text"]}}, - })) - .collect(); - std::fs::write(dir.path().join("prime-inference-models-cache.json"), - serde_json::to_vec(&json!({ - "url": format!("{}/models", pa_models::prime_inference::PRIME_INFERENCE_BASE_URL), - "scope": pa_models::prime_inference::scope_key("fixture-key", "fixture-team"), - "fetchedAt": 1, "checkedAt": 2, "payload": {"data": prime_payload}, - })).unwrap()).unwrap(); - let before = files(dir.path()); - let catalog = Arc::new(ModelCatalog::new(Some(dir.path().into()))); - let resolved = catalog.resolve(None); assert!(resolved.iter().any(|item| item.id == model.id)); - let prime = PrimeCredentials { - api_key: "fixture-key".into(), - team_id: Some("fixture-team".into()), - }; - let credentials: DiscoveryCredentials = - ["openai", "anthropic", "google", "ollama-cloud"] - .into_iter() - .map(|id| { - ( - id.into(), - DiscoveryCredential { - secret: "fixture-key".into(), - is_current: None, - }, - ) - }) + for inputs in [fixture_inputs(&base, false), inputs] { + let expected: std::collections::BTreeSet<_> = inputs + .sources + .iter() + .map(|source| source.id.clone()) + .chain(["curated".into(), "default".into()]) .collect(); - // Both public/no-auth startup and configured discovery are covered. - for credentials in [&DiscoveryCredentials::new(), &credentials] { - let reports = catalog - .refresh_catalog_sources(true, Some(&prime), credentials, None) + let publication = catalog + .refresh( + &inputs, + &RefreshRequest { + policy: RefreshPolicy::Force, + sources: None, + }, + ) .await; - assert_eq!(reports.len(), 8, "production defaults include every source"); assert_eq!( - reports.iter().find(|report| report.source == "prime-inference-public") - .unwrap().model_count, - Some(prime_payload.len()), - "credential-scoped cache remains usable without renewing freshness" + publication + .sources + .iter() + .map(|report| report.source.clone()) + .collect::>(), + expected ); + let prime = publication + .sources + .iter() + .find(|report| report.source == "prime-inference-public") + .unwrap(); + if inputs.sources[0].headers.is_empty() { + assert_eq!( + prime.model_count, None, + "public startup cannot use credential-scoped metadata" + ); + } else { + assert_eq!( + prime.model_count, + Some(prime_payload()["data"].as_array().unwrap().len()), + "credential-scoped cache remains usable without renewing freshness" + ); + } assert!( - reports.iter().all(|report| matches!( + publication.sources.iter().all(|report| matches!( report.reason, Some( pa_types::model_refresh::ModelRefreshReason::Offline | pa_types::model_refresh::ModelRefreshReason::Unconfigured ) )), - "{reports:?}" + "{:?}", + publication.sources ); } - catalog.trigger_refresh_with_credentials( - pa_models::RefreshTrigger::Startup, - Some(prime), - ); + let path = dir.parent().unwrap().join("models.json"); + pa_core::models::install_catalog(&path, Arc::clone(&catalog)); + let _ = pa_core::models::startup_refresh(dir.parent().unwrap()); tokio::time::timeout(Duration::from_secs(5), async { while runtime.metrics().num_alive_tasks() != 0 { tokio::task::yield_now().await; @@ -130,13 +219,16 @@ fn startup_child() { }) .await .expect("background refresh settled"); - assert_eq!(catalog.resolve(None), resolved); assert_eq!( - files(dir.path()), + catalog.snapshot(&fixture_inputs(&base, true)).models, + resolved + ); + assert_eq!( + files(&dir), before, "offline never renews freshness or publishes auth snapshots" ); - // The same offline chain still serves bundled and compiled data. + let payload = json!({"schemaVersion": 1, "models": [model.clone()]}); let bundled = tempfile::tempdir().unwrap(); std::fs::write( bundled.path().join("models.bundled.json"), @@ -144,11 +236,15 @@ fn startup_child() { ) .unwrap(); let bundled = ModelCatalog::with_bundled_dir(None, Some(bundled.path().into())); - assert!(bundled.resolve(None).iter().any(|item| item.id == model.id)); + assert!(bundled + .snapshot(&CatalogInputs::default()) + .models + .iter() + .any(|item| item.id == model.id)); let missing = tempfile::tempdir().unwrap(); let fallback = ModelCatalog::with_bundled_dir(None, Some(missing.path().into())); assert_eq!( - fallback.resolve(None), + fallback.snapshot(&CatalogInputs::default()).models, pa_models::transports::compiled_models() ); } @@ -166,12 +262,23 @@ fn startup_child() { } "online" => { let base = std::env::var("PA_OFFLINE_LOOPBACK_URL").unwrap(); - let catalog = ModelCatalog::with_urls(None, None, &base, &base); - let models = catalog - .refresh(true) - .await - .expect("explicit online loopback response"); - assert!(models + let catalog = ModelCatalog::with_urls(None, None, &format!("{base}/online"), &base); + let publication = catalog + .refresh( + &CatalogInputs::default(), + &RefreshRequest { + policy: RefreshPolicy::Force, + sources: Some(vec!["curated".into()]), + }, + ) + .await; + assert_eq!( + publication.sources[0].status, + pa_types::model_refresh::ModelRefreshStatus::Updated + ); + assert!(publication + .snapshot + .models .iter() .any(|model| model.id == "offline-fixture-model")); } @@ -187,6 +294,62 @@ fn startup_child() { }); } +async fn run_child(action: &str, root: &Path, url: &str, endpoint: &str) -> std::process::Output { + let mut child = tokio::process::Command::new(std::env::current_exe().unwrap()); + child + .env_clear() + .kill_on_drop(true) + .args(["--exact", "startup_child", "--nocapture"]) + .env(CHILD, action) + .env("HOME", root) + .env("PI_PACKAGE_DIR", root) + .env("PRIME_AGENT_CODING_AGENT_DIR", root.join("agent")) + .env("PRIME_AGENT_KERNEL_VENV", root.join("kernel")) + .env("RLM_SESSION_DIR", root.join("sessions")) + .env("RLM_HARNESS_STATE_DIR", root.join("harness")) + .env("RLM_GLOBAL_HARNESS_STATE_DIR", root.join("global-harness")) + .env( + "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_SOCKET", + root.join("supervisor.sock"), + ) + .env("TMPDIR", root) + .env( + "PI_OFFLINE", + if matches!(action, "online" | "fence" | "models-warm") { + "0" + } else { + "1" + }, + ) + .env("PA_OFFLINE_LOOPBACK_URL", url); + for name in [ + "XDG_CACHE_HOME", + "XDG_CONFIG_HOME", + "XDG_DATA_HOME", + "XDG_STATE_HOME", + "XDG_RUNTIME_DIR", + ] { + child.env(name, root.join(name)); + } + for name in [ + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + "http_proxy", + "https_proxy", + "all_proxy", + ] { + child.env(name, endpoint); + } + for name in ["NO_PROXY", "no_proxy"] { + child.env(name, "127.0.0.1,localhost,::1"); + } + tokio::time::timeout(Duration::from_secs(15), child.output()) + .await + .expect("finite child") + .unwrap() +} + // The fixture owns both listeners and all child lifetimes in one place. #[allow(clippy::too_many_lines)] async fn fenced_children(actions: &[&str]) { @@ -212,87 +375,57 @@ async fn fenced_children(actions: &[&str]) { } }); let local = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let url = format!("http://{}/catalog", local.local_addr().unwrap()); + let url = format!("http://{}", local.local_addr().unwrap()); let counter = Arc::clone(&served); let local_task = tokio::spawn(async move { - let (mut socket, _) = local.accept().await.unwrap(); - let mut buffer = [0; 4096]; - tokio::time::timeout(Duration::from_secs(2), socket.read(&mut buffer)) - .await - .unwrap() - .unwrap(); - counter.fetch_add(1, Ordering::SeqCst); - let mut model = pa_models::transports::compiled_models() - .iter() - .find(|model| model.provider == "anthropic") - .unwrap() - .clone(); - model.id = "offline-fixture-model".into(); - let body = json!({"schemaVersion": 1, "models": [model]}).to_string(); - let response = format!( - "HTTP/1.1 200 OK\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", - body.len() - ); - socket.write_all(response.as_bytes()).await.unwrap(); + loop { + let (mut socket, _) = local.accept().await.unwrap(); + let mut request = Vec::new(); + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + let mut chunk = [0; 1024]; + let read = socket.read(&mut chunk).await.unwrap(); + assert!(read > 0); + request.extend_from_slice(&chunk[..read]); + } + counter.fetch_add(1, Ordering::SeqCst); + let request = String::from_utf8(request).unwrap(); + let path = request.split_whitespace().nth(1).unwrap(); + let payload = match path { + "/prime-inference" => prime_payload(), + "/catalog" | "/online" => { + let mut model = pa_models::transports::compiled_models() + .iter() + .find(|model| model.provider == "anthropic") + .unwrap() + .clone(); + if path == "/online" { + model.id = "offline-fixture-model".into(); + } + json!({"schemaVersion": 1, "models": [model]}) + } + _ => panic!("unexpected fixture request: {path}"), + }; + let body = payload.to_string(); + socket.write_all(format!("HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + } }); for &action in actions { let root = tempfile::tempdir().unwrap(); - let mut child = tokio::process::Command::new(std::env::current_exe().unwrap()); - child - .env_clear() - .kill_on_drop(true) - .args(["--exact", "startup_child", "--nocapture"]) - .env(CHILD, action) - .env("HOME", root.path()) - .env("PI_PACKAGE_DIR", root.path()) - .env("PRIME_AGENT_CODING_AGENT_DIR", root.path().join("agent")) - .env("PRIME_AGENT_KERNEL_VENV", root.path().join("kernel")) - .env("RLM_SESSION_DIR", root.path().join("sessions")) - .env("RLM_HARNESS_STATE_DIR", root.path().join("harness")) - .env( - "RLM_GLOBAL_HARNESS_STATE_DIR", - root.path().join("global-harness"), - ) - .env( - "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_SOCKET", - root.path().join("supervisor.sock"), - ) - .env("TMPDIR", root.path()) - .env( - "PI_OFFLINE", - if action == "online" || action == "fence" { - "0" - } else { - "1" - }, - ) - .env("PA_OFFLINE_LOOPBACK_URL", &url); - for name in [ - "XDG_CACHE_HOME", - "XDG_CONFIG_HOME", - "XDG_DATA_HOME", - "XDG_STATE_HOME", - "XDG_RUNTIME_DIR", - ] { - child.env(name, root.path().join(name)); - } - for name in [ - "HTTP_PROXY", - "HTTPS_PROXY", - "ALL_PROXY", - "http_proxy", - "https_proxy", - "all_proxy", - ] { - child.env(name, &endpoint); - } - for name in ["NO_PROXY", "no_proxy"] { - child.env(name, "127.0.0.1,localhost,::1"); + if action == "models" { + let warm = run_child("models-warm", root.path(), &url, &endpoint).await; + assert!( + warm.status.success(), + "online fixture warm: {}{}", + String::from_utf8_lossy(&warm.stdout), + String::from_utf8_lossy(&warm.stderr) + ); + assert_eq!( + denied.load(Ordering::SeqCst), + 0, + "fixture warm stayed on loopback" + ); } - let output = tokio::time::timeout(Duration::from_secs(15), child.output()) - .await - .expect("finite child") - .unwrap(); + let output = run_child(action, root.path(), &url, &endpoint).await; assert_eq!( denied.load(Ordering::SeqCst), usize::from(action == "fence"), @@ -309,14 +442,10 @@ async fn fenced_children(actions: &[&str]) { } assert_eq!( served.load(Ordering::SeqCst), - usize::from(actions.contains(&"online")), - "online opt-in bypasses the proxy only for loopback" + usize::from(actions.contains(&"online")) + 2 * usize::from(actions.contains(&"models")), + "only explicit online fixture warms and online control reach loopback" ); - if actions.contains(&"online") { - local_task.await.unwrap(); - } else { - local_task.abort(); - } + local_task.abort(); proxy_task.abort(); } diff --git a/crates/pa-daemon/src/rlm_child_model.rs b/crates/pa-daemon/src/rlm_child_model.rs index 4a3ebc8d7..2d4509236 100644 --- a/crates/pa-daemon/src/rlm_child_model.rs +++ b/crates/pa-daemon/src/rlm_child_model.rs @@ -413,8 +413,8 @@ mod tests { /// The child-propagation regression: a spawned child resolves an /// entitled private `internal/*` model through the worker's same disk caches. - #[test] - fn a_spawned_child_resolves_an_entitled_private_model() { + #[tokio::test] + async fn a_spawned_child_resolves_an_entitled_private_model() { let dir = tempfile::TempDir::new().unwrap(); std::fs::create_dir_all(dir.path()).unwrap(); // The worker-path auth (auth.json): one Prime Inference key+team. @@ -430,42 +430,56 @@ mod tests { .to_string(), ) .unwrap(); - // The authorized private model: the compiled fallback lacks it, the - // fingerprint-scoped disk cache carries it. - let mut auth = pa_core::auth::AuthStorage::create(dir.path()); - let api_key = auth.get_api_key("prime-inference").expect("api key"); - // The stored team selection survives ambient env credentials (a - // dogfood box's ambient `PRIME_API_KEY` changes the pair); the - // cache below is written for whatever pair resolves. - let team_id = auth - .get_provider_headers("prime-inference") - .and_then(|headers| headers.get("X-Prime-Team-ID").cloned()) - .expect("stored team selection"); - let fingerprint = - pa_core::models::private_prime_authorization_fingerprint(&api_key, &team_id); - pa_core::models::write_private_prime_authorization_cache( - &dir.path().join("models.json"), - &pa_core::models::PrivatePrimeAuthorizationCache { - fingerprint, - // The durable wire items (an `/models` payload shape): the - // read path parses and builds the model from them. - items: vec![json!({ - "id": "internal/glm-5.3-fast", - "display_name": "GLM 5.3 Fast", - "pricing": { - "input_usd_per_mtok": 0.42, "output_usd_per_mtok": 2.1, - "cache_read_usd_per_mtok": 0.0, "cache_write_usd_per_mtok": 0.0, - }, - "specs": { - "context_window": 400_000, "max_output_tokens": 131_072, - "supports_reasoning": true, - "modalities": { "input": ["text"], "output": ["text"] }, - }, - })], - models: Vec::new(), - refreshed_at: 1, - }, + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/models", listener.local_addr().unwrap()); + let response = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + let mut chunk = [0; 1024]; + let read = socket.read(&mut chunk).await.unwrap(); + assert!(read > 0); + request.extend_from_slice(&chunk[..read]); + } + let body = json!({"data": [{ + "id": "internal/glm-5.3-fast", "display_name": "GLM 5.3 Fast", + "pricing": { "input_usd_per_mtok": 0.42, "output_usd_per_mtok": 2.1, + "cache_read_usd_per_mtok": 0.0, "cache_write_usd_per_mtok": 0.0 }, + "specs": { "context_window": 400_000, "max_output_tokens": 131_072, + "supports_reasoning": true, + "modalities": { "input": ["text"], "output": ["text"] } } + }]}) + .to_string(); + socket.write_all(format!("HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + }); + let catalog = || { + std::sync::Arc::new(pa_models::ModelCatalog::with_discovery_urls( + Some(dir.path().join("models")), + None, + pa_models::fetch::MODEL_CATALOG_URL, + pa_models::prime_inference::PRIME_INFERENCE_BASE_URL, + &std::collections::HashMap::from([("prime-inference".to_string(), url.clone())]), + pa_models::ollama_cloud::BASE_URL, + )) + }; + let path = dir.path().join("models.json"); + pa_core::models::install_catalog(&path, catalog()); + let mut writer = pa_core::models::ModelRegistry::create( + pa_core::auth::AuthStorage::create(dir.path()), + &path, + ); + let report = writer + .refresh_catalog_sources(Some(&["prime-inference-private".to_string()])) + .await; + assert_eq!(report.sources.len(), 1); + assert_eq!( + report.sources[0].status, + pa_types::model_refresh::ModelRefreshStatus::Updated ); + response.await.unwrap(); + // Fresh child registries must adopt durable metadata, not the writer's memory. + pa_core::models::install_catalog(&path, catalog()); assert!(catalog_models(dir.path()) .iter()