From 82ac541d922029b1f671a8a475558a133c2f2a31 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Fri, 21 Aug 2026 06:46:39 +0000 Subject: [PATCH] feat(prism): miner Verda BYOK alongside Lium MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live intake accepts Lium or Verda serverless jobs (digest-only image, 1× B200 pin). Sold-out stays queued; miners cannot override image/cmd. --- Cargo.lock | 26 + crates/prism-challenge/Cargo.toml | 1 + crates/prism-challenge/src/api.rs | 54 +- crates/prism-lium-payer/Cargo.toml | 3 + crates/prism-lium-payer/src/dispatch.rs | 180 ++++++ crates/prism-lium-payer/src/lib.rs | 147 ++++- crates/prism-verda/Cargo.toml | 33 + crates/prism-verda/src/image.rs | 133 ++++ crates/prism-verda/src/job_server.py | 111 ++++ crates/prism-verda/src/lib.rs | 720 +++++++++++++++++++++ crates/prism-verda/tests/live_b200_boot.rs | 81 +++ crates/prism-verda/tests/live_dense.rs | 96 +++ docs/PRISM.md | 7 +- docs/external-miner/prism.md | 24 +- xtask/src/external_docs_check.rs | 1 + 15 files changed, 1568 insertions(+), 49 deletions(-) create mode 100644 crates/prism-lium-payer/src/dispatch.rs create mode 100644 crates/prism-verda/Cargo.toml create mode 100644 crates/prism-verda/src/image.rs create mode 100644 crates/prism-verda/src/job_server.py create mode 100644 crates/prism-verda/src/lib.rs create mode 100644 crates/prism-verda/tests/live_b200_boot.rs create mode 100644 crates/prism-verda/tests/live_dense.rs diff --git a/Cargo.lock b/Cargo.lock index e5c7452aa..f2447c32d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3838,6 +3838,7 @@ dependencies = [ "prism-review", "prism-store", "prism-tree", + "prism-verda", "reqwest 0.12.28", "serde", "serde_json", @@ -4007,9 +4008,12 @@ name = "prism-lium-payer" version = "0.1.0" dependencies = [ "chacha20poly1305", + "http", "prism-lium", "prism-recipe", + "prism-verda", "rand 0.8.7", + "serde_json", "sha2 0.10.9", "tempfile", "tracing", @@ -4174,6 +4178,28 @@ dependencies = [ "thiserror 2.0.19", ] +[[package]] +name = "prism-verda" +version = "0.1.0" +dependencies = [ + "async-trait", + "base64 0.22.1", + "lium-rent-pool", + "prism-lium", + "prism-lium-harness", + "prism-lium-types", + "prism-pipeline", + "prism-recipe", + "reqwest 0.12.28", + "serde", + "serde_json", + "sha2 0.10.9", + "tokio", + "tracing", + "wiremock", + "zip", +] + [[package]] name = "prism-zoneb" version = "0.1.0" diff --git a/crates/prism-challenge/Cargo.toml b/crates/prism-challenge/Cargo.toml index 26018fa0d..d1bbf2a4d 100644 --- a/crates/prism-challenge/Cargo.toml +++ b/crates/prism-challenge/Cargo.toml @@ -30,6 +30,7 @@ prism-final = { path = "../prism-final" } prism-intake = { path = "../prism-intake" } prism-lium = { path = "../prism-lium" } prism-lium-payer = { path = "../prism-lium-payer" } +prism-verda = { path = "../prism-verda" } prism-orphan = { path = "../prism-orphan" } prism-pipeline = { path = "../prism-pipeline" } prism-playground = { path = "../prism-playground" } diff --git a/crates/prism-challenge/src/api.rs b/crates/prism-challenge/src/api.rs index 073818f66..ad56fac7a 100644 --- a/crates/prism-challenge/src/api.rs +++ b/crates/prism-challenge/src/api.rs @@ -216,11 +216,15 @@ async fn post_submission( headers: axum::http::HeaderMap, body: bytes::Bytes, ) -> Response { - let miner_lium_key = headers - .get(prism_lium_payer::LIUM_API_KEY_HEADER) - .or_else(|| headers.get("X-Lium-Api-Key")) - .and_then(|v| v.to_str().ok()) - .and_then(prism_lium_payer::normalize_lium_api_key); + if let Ok(v) = serde_json::from_slice::(body.as_ref()) { + if let Some(msg) = prism_lium_payer::miner_image_override_error(&v) { + return json_err(StatusCode::BAD_REQUEST, "miner_image_override", msg); + } + } + let payer_creds = match prism_lium_payer::creds_from_headers(&headers) { + Ok(c) => c, + Err(e) => return json_err(StatusCode::BAD_REQUEST, e.code(), e.message()), + }; let mut req = match parse_submission_body(&headers, body.as_ref()) { Ok(r) => r, Err(e) => return json_err(StatusCode::BAD_REQUEST, "invalid_submission", &e), @@ -272,21 +276,20 @@ async fn post_submission( Ok(b) => b, Err(e) => return json_err(StatusCode::BAD_REQUEST, "tree", &e), }; - // Live Lium: miners fund their own pod via X-Lium-Api-Key (not persisted). - if prism_lium_payer::require_miner_lium(st.backend_mode) && miner_lium_key.is_none() && !exists - { + // Live: miners fund Lium or Verda (not persisted in Postgres). + if prism_lium_payer::require_miner_lium(st.backend_mode) && payer_creds.is_none() && !exists { return json_err( StatusCode::BAD_REQUEST, "missing_lium_api_key", - "live Prism eval requires header X-Lium-Api-Key (miner-funded Lium account)", + "live Prism eval requires X-Lium-Api-Key or Verda BYOK headers", ); } let row = queued_row(&st, &req, id.clone(), tree_blob); // Idempotent no-op duplicate accepted (same id → 200 OK {status:"already-queued"}). match st.store.insert_queued(&row).await { Ok(()) => { - if let (Some(vault), Some(key)) = (&st.payer_vault, miner_lium_key.as_ref()) { - vault.insert(id.clone(), key.clone()); + if let (Some(vault), Some(creds)) = (&st.payer_vault, payer_creds.as_ref()) { + vault.insert_creds(id.clone(), creds); } // Registration finalizes only after the row is queued so intake // failures never consume the miner's single slot. @@ -307,8 +310,8 @@ async fn post_submission( (StatusCode::ACCEPTED, Json(json!({"submission_id": id, "status": "accepted", "note": lium_rent_pool::CAPACITY_POLICY}))).into_response() } Err(StoreError::Backend(e)) if e.contains("duplicate") || e.contains("unique") => { - if let (Some(vault), Some(key)) = (&st.payer_vault, miner_lium_key.as_ref()) { - vault.insert(id.clone(), key.clone()); + if let (Some(vault), Some(creds)) = (&st.payer_vault, payer_creds.as_ref()) { + vault.insert_creds(id.clone(), creds); } ( StatusCode::OK, @@ -445,23 +448,20 @@ async fn post_retry( } // Prefer the request header, else reuse the sealed BYOK vault entry for // this submission_id (auto-/admin-retry must not drop miner Lium keys). - let header_lium_key = headers - .get(prism_lium_payer::LIUM_API_KEY_HEADER) - .or_else(|| headers.get("X-Lium-Api-Key")) - .and_then(|v| v.to_str().ok()) - .and_then(prism_lium_payer::normalize_lium_api_key); - let miner_lium_key = header_lium_key - .clone() - .or_else(|| st.payer_vault.as_ref().and_then(|v| v.get(&id))); + let header_creds = prism_lium_payer::creds_from_headers(&headers) + .ok() + .flatten(); + let miner_creds = + header_creds.or_else(|| st.payer_vault.as_ref().and_then(|v| v.get_creds(&id))); if infra && row.metrics_json.is_none() && prism_lium_payer::require_miner_lium(st.backend_mode) - && miner_lium_key.is_none() + && miner_creds.is_none() { return json_err( StatusCode::BAD_REQUEST, "missing_lium_api_key", - "live Prism retry requires X-Lium-Api-Key (or a sealed payer vault entry) when another GPU run is needed", + "live Prism retry requires X-Lium-Api-Key or Verda BYOK (or a sealed payer vault entry)", ); } if row.retry_count >= st.retry_max && !infra { @@ -479,10 +479,8 @@ async fn post_retry( } match st.store.reset_for_retry(&id, true).await { Ok(_) => { - if let (Some(vault), Some(key)) = (&st.payer_vault, miner_lium_key) { - // Re-seal so TTL covers the new attempt even when the header - // was omitted and we reused the vault entry. - vault.insert(id.clone(), key); + if let (Some(vault), Some(creds)) = (&st.payer_vault, miner_creds.as_ref()) { + vault.insert_creds(id.clone(), creds); } ( StatusCode::ACCEPTED, @@ -523,6 +521,7 @@ async fn get_status(State(st): State>) -> Response { "recent_terminal": done_24h, "recipe_pin": prism_recipe::recipe_pin_hex(), "lium_capacity_note": lium_rent_pool::CAPACITY_POLICY, + "verda_capacity_note": prism_verda::CAPACITY_POLICY, })) .into_response() } @@ -1069,6 +1068,7 @@ mod tests { assert_eq!(s, StatusCode::OK); assert_eq!(v["backend"], "sim"); assert_eq!(v["lium_capacity_note"], lium_rent_pool::CAPACITY_POLICY); + assert_eq!(v["verda_capacity_note"], prism_verda::CAPACITY_POLICY); let (s, v) = call( app.clone(), Request::get("/v1/recipe").body(Body::empty()).unwrap(), diff --git a/crates/prism-lium-payer/Cargo.toml b/crates/prism-lium-payer/Cargo.toml index b13252cc5..8f6666765 100644 --- a/crates/prism-lium-payer/Cargo.toml +++ b/crates/prism-lium-payer/Cargo.toml @@ -10,9 +10,12 @@ publish = false [dependencies] chacha20poly1305 = "0.10" +http = "1" prism-lium = { path = "../prism-lium" } prism-recipe = { path = "../prism-recipe" } +prism-verda = { path = "../prism-verda" } rand = "0.8" +serde_json = "1" sha2 = "0.10" tracing = "0.1" diff --git a/crates/prism-lium-payer/src/dispatch.rs b/crates/prism-lium-payer/src/dispatch.rs new file mode 100644 index 000000000..14710ead1 --- /dev/null +++ b/crates/prism-lium-payer/src/dispatch.rs @@ -0,0 +1,180 @@ +//! Intake header → payer creds (Lium XOR Verda unless `X-Compute-Provider`). + +use serde_json::Value; + +use super::ProviderCreds; + +/// `X-Compute-Provider` when both Lium and Verda headers are complete. +pub const COMPUTE_PROVIDER_HEADER: &str = "x-compute-provider"; +/// Verda OAuth client id. +pub const VERDA_CLIENT_ID_HEADER: &str = "x-verda-client-id"; +/// Verda OAuth client secret. +pub const VERDA_CLIENT_SECRET_HEADER: &str = "x-verda-client-secret"; +/// Verda inference / tasks token. +pub const VERDA_INFERENCE_KEY_HEADER: &str = "x-verda-inference-key"; +/// Alias for the inference token (`X-Verda-Api-Key`). +pub const VERDA_API_KEY_HEADER: &str = "x-verda-api-key"; + +/// Intake payer failure (HTTP 400). +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum IntakePayerError { + /// Both providers complete, no `X-Compute-Provider`. + Ambiguous, + /// Partial Verda triplet. + MissingVerda, +} + +impl IntakePayerError { + /// Stable API error code. + #[must_use] + pub const fn code(&self) -> &'static str { + match self { + Self::Ambiguous => "ambiguous_compute_provider", + Self::MissingVerda => "missing_verda_credentials", + } + } + + /// Miner-facing message. + #[must_use] + pub const fn message(&self) -> &'static str { + match self { + Self::Ambiguous => { + "both Lium and Verda credentials present — set X-Compute-Provider: lium or verda" + } + Self::MissingVerda => { + "Verda BYOK needs X-Verda-Client-Id, X-Verda-Client-Secret, and X-Verda-Inference-Key" + } + } + } +} + +fn nz(s: Option<&str>) -> Option<&str> { + s.map(str::trim).filter(|t| !t.is_empty()) +} + +fn header_str<'a>(headers: &'a http::HeaderMap, name: &str) -> Option<&'a str> { + headers.get(name).and_then(|v| v.to_str().ok()) +} + +/// Resolve miner payer from HTTP headers (case-insensitive names). +/// +/// # Errors +/// Ambiguous dual complete creds, or incomplete Verda triplet. +pub fn creds_from_headers( + headers: &http::HeaderMap, +) -> Result, IntakePayerError> { + let inference = header_str(headers, VERDA_INFERENCE_KEY_HEADER) + .or_else(|| header_str(headers, VERDA_API_KEY_HEADER)) + .or_else(|| header_str(headers, "X-Verda-Inference-Key")) + .or_else(|| header_str(headers, "X-Verda-Api-Key")); + creds_from_parts( + header_str(headers, COMPUTE_PROVIDER_HEADER) + .or_else(|| header_str(headers, "X-Compute-Provider")), + header_str(headers, super::LIUM_API_KEY_HEADER) + .or_else(|| header_str(headers, "X-Lium-Api-Key")), + header_str(headers, VERDA_CLIENT_ID_HEADER) + .or_else(|| header_str(headers, "X-Verda-Client-Id")), + header_str(headers, VERDA_CLIENT_SECRET_HEADER) + .or_else(|| header_str(headers, "X-Verda-Client-Secret")), + inference, + ) +} + +/// Resolve miner payer from already-extracted header values. +/// +/// # Errors +/// Ambiguous dual complete creds, or incomplete Verda triplet. +pub fn creds_from_parts( + provider: Option<&str>, + lium: Option<&str>, + verda_id: Option<&str>, + verda_sec: Option<&str>, + verda_inf: Option<&str>, +) -> Result, IntakePayerError> { + let lium = nz(lium).and_then(super::normalize_lium_api_key); + let id = nz(verda_id); + let sec = nz(verda_sec); + let inf = nz(verda_inf); + let verda_any = id.is_some() || sec.is_some() || inf.is_some(); + let verda = match (id, sec, inf) { + (Some(id), Some(sec), Some(inf)) => Some(ProviderCreds::Verda { + id: id.to_owned(), + sec: sec.to_owned(), + inf: inf.to_owned(), + }), + _ if verda_any => return Err(IntakePayerError::MissingVerda), + _ => None, + }; + let want = nz(provider).map(str::to_ascii_lowercase); + if want.as_deref() == Some("lium") { + if let Some(k) = lium { + return Ok(Some(ProviderCreds::Lium(k))); + } + } + if want.as_deref() == Some("verda") { + return verda.ok_or(IntakePayerError::MissingVerda).map(Some); + } + match (lium, verda) { + (Some(k), None) => Ok(Some(ProviderCreds::Lium(k))), + (None, Some(v)) => Ok(Some(v)), + (None, None) => Ok(None), + (Some(_), Some(_)) => Err(IntakePayerError::Ambiguous), + } +} + +/// JSON keys miners must not use to override the operator image/cmd. +#[must_use] +pub fn miner_image_override_error(v: &Value) -> Option<&'static str> { + const BAD: &[&str] = &[ + "image", + "docker_image", + "image_digest", + "cmd", + "command", + "entrypoint", + "template", + "template_id", + ]; + let obj = v.as_object()?; + BAD.iter() + .find(|k| obj.contains_key(**k)) + .map(|_| "miners cannot set image, template, cmd, or entrypoint — operator pin only") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lium_or_verda_or_ambiguous() { + let l = creds_from_parts(None, Some("sk"), None, None, None).unwrap(); + assert_eq!(l.unwrap().provider(), "lium"); + let v = creds_from_parts(None, None, Some("id"), Some("sec"), Some("inf")).unwrap(); + assert_eq!(v.unwrap().provider(), "verda"); + assert_eq!( + creds_from_parts(None, Some("sk"), Some("id"), Some("sec"), Some("inf")), + Err(IntakePayerError::Ambiguous) + ); + let forced = creds_from_parts( + Some("verda"), + Some("sk"), + Some("id"), + Some("sec"), + Some("inf"), + ) + .unwrap(); + assert_eq!(forced.unwrap().provider(), "verda"); + assert_eq!( + creds_from_parts(None, None, Some("id"), None, Some("inf")), + Err(IntakePayerError::MissingVerda) + ); + } + + #[test] + fn reject_miner_image_fields() { + let v = serde_json::json!({"image": "evil:latest", "miner_hotkey": "aa"}); + assert!(miner_image_override_error(&v).is_some()); + let ok = serde_json::json!({"miner_hotkey": "aa", "zip_base64": "e30="}); + assert!(miner_image_override_error(&ok).is_none()); + } +} diff --git a/crates/prism-lium-payer/src/lib.rs b/crates/prism-lium-payer/src/lib.rs index f028bba45..ec2f28876 100644 --- a/crates/prism-lium-payer/src/lib.rs +++ b/crates/prism-lium-payer/src/lib.rs @@ -1,4 +1,4 @@ -//! Miner-funded Lium keys (BYOK): vault + live-client factory. +//! Miner-funded Lium / Verda keys (BYOK): vault + live-client factory. //! //! Keys are held in process memory and optionally in a TTL-bounded //! encrypted seal directory (`PRISM_PAYER_VAULT_DIR` + key file). Default TTL @@ -9,6 +9,7 @@ #![forbid(unsafe_code)] +mod dispatch; mod sealed; use std::collections::HashMap; @@ -16,6 +17,70 @@ use std::fmt; use std::sync::{Arc, Mutex}; use prism_lium::{EvalJobBackend, LiumClient, LiumError, LiumSshConfig, LIUM_API_BASE_URL}; +use prism_verda::{VerdaClient, VerdaCreds}; + +pub use dispatch::{ + creds_from_headers, creds_from_parts, miner_image_override_error, IntakePayerError, + COMPUTE_PROVIDER_HEADER, VERDA_API_KEY_HEADER, VERDA_CLIENT_ID_HEADER, + VERDA_CLIENT_SECRET_HEADER, VERDA_INFERENCE_KEY_HEADER, +}; + +/// Miner-funded provider credentials (never logged). +#[derive(Clone, PartialEq, Eq)] +pub enum ProviderCreds { + /// Lium REST API key. + Lium(String), + /// Verda OAuth + inference token. + Verda { + /// Cloud API client id. + id: String, + /// Cloud API client secret. + sec: String, + /// Inference / tasks bearer. + inf: String, + }, +} + +impl fmt::Debug for ProviderCreds { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Lium(_) => f.debug_tuple("Lium").field(&"").finish(), + Self::Verda { .. } => f.debug_struct("Verda").finish_non_exhaustive(), + } + } +} + +impl ProviderCreds { + /// Provider label persisted on the eval row / receipt. + #[must_use] + pub const fn provider(&self) -> &'static str { + match self { + Self::Lium(_) => "lium", + Self::Verda { .. } => "verda", + } + } + + fn encode(&self) -> String { + match self { + Self::Lium(k) => k.clone(), + Self::Verda { id, sec, inf } => format!("verda\u{1e}{id}\u{1e}{sec}\u{1e}{inf}"), + } + } + + fn decode(raw: &str) -> Self { + if let Some(rest) = raw.strip_prefix("verda\u{1e}") { + let p: Vec<&str> = rest.split('\u{1e}').collect(); + if p.len() == 3 { + return Self::Verda { + id: p[0].to_owned(), + sec: p[1].to_owned(), + inf: p[2].to_owned(), + }; + } + } + Self::Lium(raw.to_owned()) + } +} pub use sealed::{ expiry_secs, recommended_ttl_secs, SealedVaultConfig, DEFAULT_TTL_SECS, DIR_ENV, @@ -90,10 +155,15 @@ impl PayerKeyVault { n } - /// Insert or replace the payer key for `submission_id`. + /// Insert or replace the Lium payer key for `submission_id`. pub fn insert(&self, submission_id: impl Into, api_key: impl Into) { + self.insert_creds(submission_id, &ProviderCreds::Lium(api_key.into())); + } + + /// Insert Lium or Verda creds (seal stores a tagged blob, never logged). + pub fn insert_creds(&self, submission_id: impl Into, creds: &ProviderCreds) { let id = submission_id.into(); - let key = api_key.into(); + let key = creds.encode(); if key.trim().is_empty() { return; } @@ -107,12 +177,20 @@ impl PayerKeyVault { } } - /// Clone of the stored key, if any (memory, then sealed file). - /// - /// Measure / resume paths rely on this hydrate-from-seal fallback when the - /// in-memory map is empty after a restart (as long as the seal is unexpired). + /// True when a Lium or Verda seal exists for this submission. #[must_use] - pub fn get(&self, submission_id: &str) -> Option { + pub fn has_payer(&self, submission_id: &str) -> bool { + self.get_creds(submission_id).is_some() + } + + /// Decode stored creds (memory, then sealed file). + #[must_use] + pub fn get_creds(&self, submission_id: &str) -> Option { + self.raw_blob(submission_id) + .map(|s| ProviderCreds::decode(&s)) + } + + fn raw_blob(&self, submission_id: &str) -> Option { if let Ok(g) = self.inner.lock() { if let Some(k) = g.get(submission_id) { return Some(k.clone()); @@ -126,12 +204,24 @@ impl PayerKeyVault { Some(key) } + /// Clone of the stored Lium key, if any (memory, then sealed file). + /// + /// Measure / resume paths rely on this hydrate-from-seal fallback when the + /// in-memory map is empty after a restart (as long as the seal is unexpired). + #[must_use] + pub fn get(&self, submission_id: &str) -> Option { + match self.get_creds(submission_id) { + Some(ProviderCreds::Lium(k)) => Some(k), + _ => None, + } + } + /// Re-persist the seal with a fresh TTL window (memory and/or disk). /// /// Call before measure and on heartbeats so full-budget trains cannot /// outlive the on-disk seal. Returns `false` when no key is available. pub fn refresh(&self, submission_id: &str) -> bool { - let Some(key) = self.get(submission_id) else { + let Some(key) = self.raw_blob(submission_id) else { return false; }; if let Some(cfg) = &self.sealed { @@ -205,15 +295,27 @@ impl PayerBackendFactory { submission_id: &str, operator: Arc, ) -> Result, String> { - if let Some(key) = self.vault.get(submission_id) { - let client = LiumClient::with_config(key, self.base_url.clone(), self.ssh.clone()) + match self.vault.get_creds(submission_id) { + Some(ProviderCreds::Lium(key)) => { + let client = LiumClient::with_config(key, self.base_url.clone(), self.ssh.clone()) + .map_err(|e: LiumError| e.to_string())?; + return Ok(Arc::new(client)); + } + Some(ProviderCreds::Verda { id, sec, inf }) => { + let client = VerdaClient::new(VerdaCreds { + client_id: id, + client_secret: sec, + inference_key: inf, + }) .map_err(|e: LiumError| e.to_string())?; - return Ok(Arc::new(client)); + return Ok(Arc::new(client)); + } + None => {} } if self.allow_operator_fallback { return Ok(operator); } - Err("miner Lium API key missing for this submission — resubmit with X-Lium-Api-Key".into()) + Err("miner compute key missing — resubmit with X-Lium-Api-Key or Verda BYOK headers".into()) } } @@ -326,4 +428,23 @@ mod tests { } assert_eq!(v.get("onlyseal").as_deref(), Some("sk_from_disk")); } + + #[test] + fn verda_creds_do_not_look_like_lium_key() { + let v = PayerKeyVault::new(); + v.insert_creds( + "j1", + &ProviderCreds::Verda { + id: "cid".into(), + sec: "csec".into(), + inf: "inf".into(), + }, + ); + assert!(v.get("j1").is_none()); + assert!(v.has_payer("j1")); + let c = v.get_creds("j1").unwrap(); + assert_eq!(c.provider(), "verda"); + let dbg = format!("{v:?}"); + assert!(!dbg.contains("csec")); + } } diff --git a/crates/prism-verda/Cargo.toml b/crates/prism-verda/Cargo.toml new file mode 100644 index 000000000..6b0f34f91 --- /dev/null +++ b/crates/prism-verda/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "prism-verda" +description = "Verda serverless batch-job BYOK client for Prism (EvalJobBackend)" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +publish = false + +[dependencies] +async-trait = "0.1" +lium-rent-pool = { path = "../lium-rent-pool" } +prism-lium = { path = "../prism-lium" } +prism-lium-harness = { path = "../prism-lium-harness" } +prism-lium-types = { path = "../prism-lium-types" } +prism-recipe = { path = "../prism-recipe" } +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.10" +tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } +tracing = "0.1" + +[dev-dependencies] +base64 = "0.22" +prism-pipeline = { path = "../prism-pipeline" } +tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } +wiremock = "0.6" +zip = { version = "2", default-features = false, features = ["deflate"] } + +[lints] +workspace = true diff --git a/crates/prism-verda/src/image.rs b/crates/prism-verda/src/image.rs new file mode 100644 index 000000000..c4aadc9d2 --- /dev/null +++ b/crates/prism-verda/src/image.rs @@ -0,0 +1,133 @@ +//! Digest-only image pin for Verda job deployments (miners cannot override). + +use prism_lium_harness::resolved_pod_image; +use prism_lium_types::LiumError; +use sha2::{Digest, Sha256}; + +/// Embedded operator job server (hash-pinned in create payload tests). +pub const JOB_SERVER_PY: &str = include_str!("job_server.py"); + +/// Health path the replica must expose. +pub const HEALTH_PATH: &str = "/health"; +/// Container listen port (Verda health + job POST). +pub const EXPOSED_PORT: u16 = 8000; + +/// Resolve the operator image. Tags and miner-supplied refs are rejected. +/// +/// # Errors +/// Non-digest `PRISM_POD_IMAGE_REF` / `PRISM_VERDA_IMAGE_REF`. +pub fn pinned_image() -> Result { + if let Some(image) = std::env::var("PRISM_VERDA_IMAGE_REF") + .ok() + .filter(|s| !s.trim().is_empty()) + { + return require_digest(&image); + } + resolved_pod_image().map(|(image, _, _)| image) +} + +/// Fail closed unless `repository@sha256:<64 lowercase hex>`. +/// +/// # Errors +/// Tag or empty digest. +pub fn require_digest(image: &str) -> Result { + let image = image.trim(); + let Some((repo, digest)) = image.rsplit_once("@sha256:") else { + return Err(LiumError::Integrity( + "Verda image must be repository@sha256:<64 lowercase hex>".into(), + )); + }; + if repo.is_empty() + || digest.len() != 64 + || !digest + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + { + return Err(LiumError::Integrity( + "Verda image digest must be 64 lowercase hex".into(), + )); + } + Ok(image.to_owned()) +} + +/// SHA-256 hex of the operator job server (integrity pin). +#[must_use] +pub fn job_server_sha256() -> String { + hex_sha256(JOB_SERVER_PY.as_bytes()) +} + +/// Hex SHA-256. +#[must_use] +pub fn hex_sha256(bytes: &[u8]) -> String { + let d = Sha256::digest(bytes); + format!("{d:x}") +} + +/// Verda rejects each `cmd` argument longer than 253 characters. +pub const CMD_ARG_MAX: usize = 253; +/// Hex payload chunk size for `PRISM_JS_C*` env vars. +pub const JOB_SERVER_ENV_CHUNK: usize = 200; +const HEX: &[char] = &[ + '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f', +]; + +/// Operator `command` list: short bootstrap that `exec`s the env-chunked server. +/// +/// The full `job_server.py` cannot be a `cmd` argument (253-char cap). +#[must_use] +pub fn job_command() -> Vec { + let boot = "import os,binascii;exec(binascii.unhexlify(''.join(os.environ[k] for k in sorted(os.environ) if k.startswith('PRISM_JS_C'))).decode())"; + debug_assert!(boot.len() < CMD_ARG_MAX); + vec!["python3".into(), "-c".into(), boot.into()] +} + +/// Hex-chunked job server for Verda `env` (plain). +#[must_use] +pub fn job_server_env_pairs() -> Vec<(String, String)> { + let mut hex = String::with_capacity(JOB_SERVER_PY.len() * 2); + for b in JOB_SERVER_PY.bytes() { + hex.push(HEX[(b >> 4) as usize]); + hex.push(HEX[(b & 0x0f) as usize]); + } + hex.as_bytes() + .chunks(JOB_SERVER_ENV_CHUNK) + .enumerate() + .map(|(i, chunk)| { + ( + format!("PRISM_JS_C{i:02}"), + String::from_utf8_lossy(chunk).into_owned(), + ) + }) + .collect() +} + +/// Miner-facing sold-out copy (queued, not Score(0)). +pub const CAPACITY_POLICY: &str = "When Verda has no matching 1× B200 (or documented H200/H100 fallback) compute, the job stays queued and retries until a replica can start (sold out is not Score(0)). Bad ZIP, auth, and image-override errors still fail."; + +/// Capacity note for events. +pub const CAPACITY_NOTE: &str = + "B200s are currently out of capacity on Verda; this job is queued until compute appears."; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn digest_ok_and_tag_rejected() { + let ok = + "docker.io/x/y@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + assert_eq!(require_digest(ok).unwrap(), ok); + assert!(require_digest("docker.io/x/y:latest").is_err()); + assert!(require_digest("docker.io/x/y@sha256:abc").is_err()); + } + + #[test] + fn job_server_hash_stable() { + assert_eq!(job_server_sha256().len(), 64); + assert!(JOB_SERVER_PY.contains("do_POST")); + assert!(JOB_SERVER_PY.contains("/health")); + let cmd = job_command(); + assert!(cmd.iter().all(|a| a.len() < CMD_ARG_MAX)); + assert!(!job_server_env_pairs().is_empty()); + } +} diff --git a/crates/prism-verda/src/job_server.py b/crates/prism-verda/src/job_server.py new file mode 100644 index 000000000..e44a17d32 --- /dev/null +++ b/crates/prism-verda/src/job_server.py @@ -0,0 +1,111 @@ +"""Operator-owned Verda batch entry (not miner-supplied). Health + one job then exit.""" +import json +import os +import subprocess +import sys +import tarfile +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +PORT = int(os.environ.get("PRISM_VERDA_PORT", "8000")) +WORKDIR = "/tmp/prism_eval" + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + sys.stderr.write("%s\n" % (fmt % args)) + + def do_GET(self): + if self.path.split("?", 1)[0] in ("/health", "/"): + self.send_response(200) + self.send_header("Content-Type", "text/plain") + self.end_headers() + self.wfile.write(b"ok") + return + self.send_error(404) + + def do_POST(self): + n = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(n) + os.makedirs(WORKDIR, exist_ok=True) + tar = os.path.join(WORKDIR, "upload.tar") + with open(tar, "wb") as f: + f.write(body) + try: + with tarfile.open(tar) as tf: + tf.extractall(WORKDIR) + except (tarfile.TarError, OSError) as e: + self._json(500, {"ok": False, "error": "extract: %s" % e}) + os._exit(1) + timeout = int(os.environ.get("PRISM_VERDA_TIMEOUT_SECS", "18000")) + env = os.environ.copy() + try: + subprocess.check_call( + [ + sys.executable, + "-c", + "import transformers", + ], + cwd=WORKDIR, + env=env, + ) + except subprocess.CalledProcessError: + subprocess.check_call( + [ + sys.executable, + "-m", + "pip", + "install", + "--break-system-packages", + "--root-user-action=ignore", + "transformers==4.44.2", + "datasets==3.0.2", + "pyarrow==17.0.0", + ], + cwd=WORKDIR, + env=env, + ) + try: + proc = subprocess.run( + [sys.executable, "main.py"], + cwd=WORKDIR, + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + code = proc.returncode + captured = (proc.stdout or "") + "\n" + (proc.stderr or "") + except subprocess.TimeoutExpired as e: + code = 124 + captured = "%s\n" % e + log = "" + try: + with open(os.path.join(WORKDIR, "harness.log"), "r", errors="replace") as f: + log = f.read() + except OSError: + log = captured + metrics = "" + mp = os.path.join(WORKDIR, "metrics.json") + if os.path.isfile(mp): + with open(mp, "r", errors="replace") as f: + metrics = f.read() + payload = { + "ok": code == 0, + "code": code, + "log": log[-65536:], + "metrics": metrics, + } + self._json(200 if code == 0 else 500, payload) + os._exit(0 if code == 0 else 1) + + def _json(self, code, obj): + raw = json.dumps(obj).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + +if __name__ == "__main__": + ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever() diff --git a/crates/prism-verda/src/lib.rs b/crates/prism-verda/src/lib.rs new file mode 100644 index 000000000..85a1adad0 --- /dev/null +++ b/crates/prism-verda/src/lib.rs @@ -0,0 +1,720 @@ +//! Verda serverless batch jobs as a Prism [`EvalJobBackend`]. +//! +//! Miners pay with OAuth client credentials + an inference token. The +//! container image and job command are operator-pinned (digest + embedded +//! `job_server.py`). No SSH. + +#![forbid(unsafe_code)] +#![allow(clippy::missing_errors_doc, clippy::doc_markdown)] + +mod image; + +pub use image::{ + hex_sha256, job_command, job_server_env_pairs, job_server_sha256, pinned_image, require_digest, + CAPACITY_NOTE, CAPACITY_POLICY, EXPOSED_PORT, HEALTH_PATH, JOB_SERVER_PY, +}; + +use std::time::Duration; + +use async_trait::async_trait; +use prism_lium::{ + EvalJobBackend, Instance, InstanceSpec, LiumError, Offer, RemoteExecResult, MIN_LIFETIME_HOURS, +}; +use prism_lium_harness::{harness_env_pairs, harness_upload_tar, parse_metrics_output}; +use prism_lium_types::CostGuardrailError; +use serde_json::{json, Value}; + +/// Public API base (OAuth + job-deployments). +pub const VERDA_API_BASE: &str = "https://api.verda.com/v1"; +/// Batch job trigger / poll host. +pub const VERDA_TASKS_BASE: &str = "https://tasks.datacrunch.io"; + +/// Miner Verda credentials (never logged). +#[derive(Clone)] +pub struct VerdaCreds { + /// Cloud API client id. + pub client_id: String, + /// Cloud API client secret. + pub client_secret: String, + /// Inference / tasks bearer. + pub inference_key: String, +} + +impl std::fmt::Debug for VerdaCreds { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("VerdaCreds") + .field("client_id", &"") + .finish_non_exhaustive() + } +} + +/// Live Verda backend (one miner account). +pub struct VerdaClient { + creds: VerdaCreds, + api_base: String, + tasks_base: String, + http: reqwest::Client, + train_hours_cap: f64, +} + +impl std::fmt::Debug for VerdaClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("VerdaClient") + .field("api_base", &self.api_base) + .finish_non_exhaustive() + } +} + +impl VerdaClient { + /// Production endpoints. + pub fn new(creds: VerdaCreds) -> Result { + Self::with_bases(creds, VERDA_API_BASE, VERDA_TASKS_BASE) + } + + /// Test / isolated bases. + pub fn with_bases( + creds: VerdaCreds, + api_base: &str, + tasks_base: &str, + ) -> Result { + if creds.client_id.trim().is_empty() + || creds.client_secret.trim().is_empty() + || creds.inference_key.trim().is_empty() + { + return Err(LiumError::Api("missing_verda_credentials".into())); + } + let http = reqwest::Client::builder() + .timeout(Duration::from_mins(1)) + .user_agent("prism-verda/0.1") + .build() + .map_err(|e| LiumError::Transport(e.to_string()))?; + let train_hours_cap = std::env::var("PRISM_TEST_TRAIN_MINUTES") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|m| *m > 0.0) + .map(|m| m / 60.0) + .or_else(|| { + std::env::var("PRISM_TRAIN_HOURS_CAP") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|h| *h > 0.0) + }) + .unwrap_or(prism_recipe::TRAIN_HOURS_CAP); + Ok(Self { + creds, + api_base: api_base.trim_end_matches('/').to_owned(), + tasks_base: tasks_base.trim_end_matches('/').to_owned(), + http, + train_hours_cap, + }) + } + + #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] + fn deadline_secs(lifetime_hours: f64) -> u64 { + let h = lifetime_hours.max(MIN_LIFETIME_HOURS); + ((h * 3600.0).ceil() as u64).max(600) + } + + async fn access_token(&self) -> Result { + let res = self + .http + .post(format!("{}/oauth2/token", self.api_base)) + .json(&json!({ + "grant_type": "client_credentials", + "client_id": self.creds.client_id, + "client_secret": self.creds.client_secret, + })) + .send() + .await + .map_err(|e| LiumError::Transport(e.to_string()))?; + let status = res.status(); + let body = res.text().await.unwrap_or_default(); + if !status.is_success() { + return Err(LiumError::Api(format!( + "oauth2/token -> {status} {}", + redact(&body) + ))); + } + let v: Value = + serde_json::from_str(&body).map_err(|e| LiumError::Api(format!("oauth json: {e}")))?; + v.get("access_token") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .map(str::to_owned) + .ok_or_else(|| LiumError::Api("oauth2/token missing access_token".into())) + } + + async fn api_json( + &self, + method: reqwest::Method, + path: &str, + body: Option, + ) -> Result<(reqwest::StatusCode, Value), LiumError> { + let token = self.access_token().await?; + let mut req = self + .http + .request(method, format!("{}{path}", self.api_base)) + .bearer_auth(&token); + if let Some(b) = body { + req = req.json(&b); + } + let res = req + .send() + .await + .map_err(|e| LiumError::Transport(e.to_string()))?; + let status = res.status(); + let text = res.text().await.unwrap_or_default(); + let json = if text.trim().is_empty() { + Value::Null + } else { + serde_json::from_str(&text).unwrap_or(Value::String(text.clone())) + }; + if status.as_u16() == 429 || is_capacity_status(status, &text) { + return Err(CostGuardrailError::NoCapacity.into()); + } + Ok((status, json)) + } + + /// Snapshot used by unit tests (digest + operator command hash). + #[must_use] + pub fn deployment_body(name: &str, image: &str, compute_name: &str, deadline: u64) -> Value { + Self::deployment_body_with_train( + name, + image, + compute_name, + deadline, + prism_recipe::TRAIN_HOURS_CAP, + ) + } + + fn deployment_body_with_train( + name: &str, + image: &str, + compute_name: &str, + deadline: u64, + train_hours: f64, + ) -> Value { + let mut env: Vec = harness_env_pairs(train_hours, compute_name, false) + .into_iter() + .map(|(key, value)| { + json!({ + "name": key, + "value_or_reference_to_secret": value, + "type": "plain", + }) + }) + .collect(); + for (key, value) in job_server_env_pairs() { + env.push(json!({ + "name": key, + "value_or_reference_to_secret": value, + "type": "plain", + })); + } + json!({ + "name": name, + "containers": [{ + "image": image, + "exposed_port": EXPOSED_PORT, + "healthcheck": { + "enabled": true, + "port": EXPOSED_PORT, + "path": HEALTH_PATH, + }, + "entrypoint_overrides": { + "enabled": true, + "cmd": job_command(), + }, + "env": env, + }], + "compute": { "name": compute_name, "size": 1 }, + "scaling": { + "max_replica_count": 1, + "queue_message_ttl_seconds": 86400, + "deadline_seconds": deadline, + }, + "container_registry_settings": { "is_private": false }, + }) + } + + fn pick_compute(resources: &[Value]) -> Result<(String, String), LiumError> { + // Dashboard "1× B200 available" is not always `is_available` + `size=1` + // on `/serverless-compute-resources`. Pin B200 by name first; create + // with `compute.size=1`. Only H200/H100 if no B200 SKU exists at all. + let named = |needles: &[&str], require_avail: bool, want_size: Option| { + resources.iter().find_map(|r| { + let name = r.get("name").and_then(Value::as_str)?; + let avail = r + .get("is_available") + .and_then(Value::as_bool) + .unwrap_or(true); + let size = r.get("size").and_then(Value::as_u64); + if require_avail && !avail { + return None; + } + if want_size.is_some() && size != want_size { + return None; + } + let low = name.to_ascii_lowercase(); + needles + .iter() + .any(|n| low.contains(n)) + .then(|| (name.to_owned(), name.to_owned())) + }) + }; + if let Some(want) = std::env::var("PRISM_VERDA_COMPUTE") + .ok() + .filter(|s| !s.trim().is_empty()) + { + let needle = want.to_ascii_lowercase(); + if let Some(hit) = named(&[needle.as_str()], false, Some(1)) + .or_else(|| named(&[needle.as_str()], false, None)) + { + return Ok(hit); + } + return Err(LiumError::from(CostGuardrailError::NoCapacity)); + } + named(&["b200"], true, Some(1)) + .or_else(|| named(&["b200"], true, None)) + .or_else(|| named(&["b200"], false, Some(1))) + .or_else(|| named(&["b200"], false, None)) + .or_else(|| named(&["h200"], true, Some(1))) + .or_else(|| named(&["h100"], true, Some(1))) + .ok_or_else(|| LiumError::from(CostGuardrailError::NoCapacity)) + } + + async fn list_compute(&self) -> Result, LiumError> { + let (status, v) = self + .api_json(reqwest::Method::GET, "/serverless-compute-resources", None) + .await?; + if !status.is_success() { + return Err(LiumError::Api(format!( + "compute-resources -> {status} {}", + redact(&v.to_string()) + ))); + } + match v { + Value::Array(a) => Ok(a), + other => other + .get("compute_resources") + .and_then(Value::as_array) + .cloned() + .ok_or_else(|| LiumError::Api("compute-resources: expected array".into())), + } + } + + async fn wait_running(&self, name: &str) -> Result<(), LiumError> { + for _ in 0..40 { + let (status, v) = self + .api_json( + reqwest::Method::GET, + &format!("/job-deployments/{name}/status"), + None, + ) + .await?; + if status.is_success() { + let st = v + .get("status") + .and_then(Value::as_str) + .unwrap_or("") + .to_ascii_lowercase(); + if matches!(st.as_str(), "running" | "healthy" | "ready") { + return Ok(()); + } + } + tokio::time::sleep(Duration::from_secs(15)).await; + } + Err(LiumError::Api(format!( + "verda job-deployment {name} not running" + ))) + } + + async fn tasks_endpoint(&self, name: &str) -> Result { + let (status, v) = self + .api_json( + reqwest::Method::GET, + &format!("/job-deployments/{name}"), + None, + ) + .await?; + if status.is_success() { + if let Some(u) = v.get("endpoint_base_url").and_then(Value::as_str) { + return Ok(u.trim_end_matches('/').to_owned()); + } + } + Ok(format!("{}/{name}", self.tasks_base)) + } + + async fn post_job(&self, name: &str, tar: &[u8]) -> Result { + let ep = self.tasks_endpoint(name).await?; + let res = self + .http + .post(format!("{ep}/job")) + .bearer_auth(&self.creds.inference_key) + .header("X-Inference-Id", name) + .header("Content-Type", "application/octet-stream") + .body(tar.to_vec()) + .send() + .await + .map_err(|e| LiumError::Transport(e.to_string()))?; + let status = res.status(); + let text = res.text().await.unwrap_or_default(); + if status.as_u16() == 429 || is_capacity_status(status, &text) { + return Err(CostGuardrailError::NoCapacity.into()); + } + if !status.is_success() { + return Err(LiumError::Api(format!( + "tasks job -> {status} {}", + redact(&text) + ))); + } + Ok(name.to_owned()) + } + + async fn poll_result( + &self, + name: &str, + timeout: Duration, + ) -> Result { + let start = std::time::Instant::now(); + loop { + if start.elapsed() > timeout { + return Err(LiumError::Exec("verda job deadline".into())); + } + let res = self + .http + .get(format!("{}/status/{name}", self.tasks_base)) + .bearer_auth(&self.creds.inference_key) + .header("X-Inference-Id", name) + .send() + .await + .map_err(|e| LiumError::Transport(e.to_string()))?; + let text = res.text().await.unwrap_or_default(); + let st = serde_json::from_str::(&text) + .ok() + .and_then(|v| { + v.get("Status") + .or_else(|| v.get("status")) + .and_then(Value::as_str) + .map(str::to_ascii_lowercase) + }) + .unwrap_or_default(); + let inflight = matches!( + st.as_str(), + "queue" + | "queued" + | "initialized" + | "running" + | "inference" + | "starting" + | "pending" + | "" + ); + if inflight { + tokio::time::sleep(Duration::from_secs(20)).await; + continue; + } + let res = self + .http + .get(format!("{}/result/{name}", self.tasks_base)) + .bearer_auth(&self.creds.inference_key) + .header("X-Inference-Id", name) + .send() + .await + .map_err(|e| LiumError::Transport(e.to_string()))?; + let status = res.status(); + let body = res.text().await.unwrap_or_default(); + if status.as_u16() == 404 || body.contains("result not found") { + tokio::time::sleep(Duration::from_secs(20)).await; + continue; + } + return parse_job_result(&body); + } + } +} + +fn parse_job_result(body: &str) -> Result { + if let Ok(v) = serde_json::from_str::(body) { + if let Some(m) = v.get("metrics").and_then(Value::as_str) { + if !m.is_empty() { + return parse_metrics_output(&format!("METRICS_JSON={m}\nEVAL_OK\n"), 0, ""); + } + } + if v.get("bpb").is_some() { + return serde_json::from_value(v) + .map_err(|e| LiumError::Exec(format!("result metrics: {e}"))); + } + let log = v.get("log").and_then(Value::as_str).unwrap_or(body); + return parse_metrics_output(log, 0, ""); + } + parse_metrics_output(body, 0, "") +} + +fn is_capacity_status(status: reqwest::StatusCode, text: &str) -> bool { + let l = text.to_ascii_lowercase(); + status.as_u16() == 503 + || l.contains("no_capacity") + || l.contains("out of capacity") + || l.contains("sold out") + || l.contains("no available") +} + +fn redact(s: &str) -> String { + let mut t = s.to_owned(); + for needle in ["client_secret", "access_token", "inference"] { + if t.to_ascii_lowercase().contains(needle) { + t = format!("<{needle} redacted len={}>", s.len()); + break; + } + } + if t.len() > 400 { + t.truncate(400); + } + t +} + +#[async_trait] +impl EvalJobBackend for VerdaClient { + async fn list_offers(&self, _max_price_per_hour: Option) -> Result, LiumError> { + let rows = self.list_compute().await?; + Ok(rows + .iter() + .filter_map(|r| { + let name = r.get("name").and_then(Value::as_str)?; + Some(Offer { + id: name.to_owned(), + gpu_type: name.to_owned(), + gpu_count: 1, + price_per_hour: 0.0, + provider: "verda".into(), + }) + }) + .collect()) + } + + async fn provision(&self, spec: &InstanceSpec) -> Result { + if spec.max_lifetime_hours <= 0.0 { + return Err(CostGuardrailError::LifetimeMissing.into()); + } + if spec.max_lifetime_hours < MIN_LIFETIME_HOURS { + return Err(CostGuardrailError::LifetimeBelowFloor.into()); + } + let image = if let Some(d) = spec.image_digest.as_deref().filter(|s| !s.is_empty()) { + require_digest(d)? + } else { + pinned_image()? + }; + let compute = self.list_compute().await?; + let (compute_name, gpu_type) = Self::pick_compute(&compute)?; + let deadline = Self::deadline_secs(spec.max_lifetime_hours); + let body = Self::deployment_body_with_train( + &spec.name, + &image, + &compute_name, + deadline, + self.train_hours_cap, + ); + // Never honor miner image/cmd: body is built here only. + debug_assert!(body["containers"][0]["image"] == image); + let (status, v) = self + .api_json(reqwest::Method::POST, "/job-deployments", Some(body)) + .await?; + if !status.is_success() { + let msg = v.to_string(); + if lium_rent_pool::is_no_capacity(&msg) || is_capacity_status(status, &msg) { + return Err(CostGuardrailError::NoCapacity.into()); + } + return Err(LiumError::Api(format!( + "create job-deployment -> {status} {}", + redact(&msg) + ))); + } + Ok(Instance { + id: spec.name.clone(), + status: "PROVISIONING".into(), + provider: "verda".into(), + gpu_type: Some(gpu_type), + ssh_connect_cmd: None, + }) + } + + async fn terminate(&self, instance_id: &str) -> Result<(), LiumError> { + let (status, v) = self + .api_json( + reqwest::Method::DELETE, + &format!("/job-deployments/{instance_id}"), + None, + ) + .await?; + if status.is_success() || status.as_u16() == 404 { + return Ok(()); + } + Err(LiumError::Api(format!( + "delete job-deployment -> {status} {}", + redact(&v.to_string()) + ))) + } + + async fn verify_terminated(&self, instance_id: &str) -> Result { + let (status, _) = self + .api_json( + reqwest::Method::GET, + &format!("/job-deployments/{instance_id}"), + None, + ) + .await?; + Ok(status.as_u16() == 404) + } + + async fn exec_eval( + &self, + instance_id: &str, + architecture_py: &str, + training_py: &str, + tree_blob: Option<&[u8]>, + ) -> Result { + self.wait_running(instance_id).await?; + let tar = harness_upload_tar(architecture_py, training_py, tree_blob)?; + let _ = harness_env_pairs(self.train_hours_cap, "verda", false); + self.post_job(instance_id, &tar).await?; + let timeout = Duration::from_secs(Self::deadline_secs(self.train_hours_cap + 1.5)); + self.poll_result(instance_id, timeout).await + } + + async fn resume_eval(&self, instance_id: &str) -> Result { + let timeout = Duration::from_secs(Self::deadline_secs(self.train_hours_cap + 1.5)); + self.poll_result(instance_id, timeout).await + } + + async fn instance_running(&self, instance_id: &str) -> Result { + let (status, v) = self + .api_json( + reqwest::Method::GET, + &format!("/job-deployments/{instance_id}/status"), + None, + ) + .await?; + if !status.is_success() { + return Ok(false); + } + let st = v + .get("status") + .and_then(Value::as_str) + .unwrap_or("") + .to_ascii_lowercase(); + Ok(matches!(st.as_str(), "running" | "healthy" | "ready")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn create_payload_is_digest_and_operator_cmd() { + let image = + "docker.io/x/y@sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + let body = VerdaClient::deployment_body("prism-abc", image, "B200", 14400); + assert_eq!(body["containers"][0]["image"], image); + assert_eq!(body["compute"]["size"], 1); + assert_eq!(body["compute"]["name"], "B200"); + let cmd = body["containers"][0]["entrypoint_overrides"]["cmd"] + .as_array() + .unwrap(); + assert_eq!(cmd[0], "python3"); + assert!(cmd.iter().all(|a| a.as_str().unwrap().len() < 253)); + assert_eq!( + body["containers"][0]["entrypoint_overrides"]["enabled"], + true + ); + let env = body["containers"][0]["env"].as_array().unwrap(); + assert!(env.iter().any(|e| e["name"] == "PRISM_JS_C00")); + assert_eq!(job_server_sha256().len(), 64); + assert!(require_digest("nginx:latest").is_err()); + } + + #[test] + fn pick_b200_then_fallback() { + let rows = vec![ + json!({"name": "H100", "size": 1, "is_available": true}), + json!({"name": "1x B200", "size": 1, "is_available": true}), + ]; + let (n, _) = VerdaClient::pick_compute(&rows).unwrap(); + assert!(n.to_ascii_lowercase().contains("b200")); + let mixed = vec![ + json!({"name": "B200", "size": 2, "is_available": true}), + json!({"name": "B200", "size": 1, "is_available": true}), + ]; + let (n, _) = VerdaClient::pick_compute(&mixed).unwrap(); + assert!(n.to_ascii_lowercase().contains("b200")); + let queued = vec![json!({"name": "8x B200", "size": 8, "is_available": false})]; + let (n, _) = VerdaClient::pick_compute(&queued).unwrap(); + assert!(n.to_ascii_lowercase().contains("b200")); + let none = vec![json!({"name": "L4", "size": 1, "is_available": true})]; + assert!(VerdaClient::pick_compute(&none).is_err()); + } + + #[test] + fn pick_honors_compute_override() { + let rows = vec![ + json!({"name": "B200", "size": 1, "is_available": true}), + json!({"name": "L40S", "size": 1, "is_available": true}), + ]; + std::env::set_var("PRISM_VERDA_COMPUTE", "L40S"); + let (n, _) = VerdaClient::pick_compute(&rows).unwrap(); + std::env::remove_var("PRISM_VERDA_COMPUTE"); + assert!(n.to_ascii_lowercase().contains("l40")); + } + + #[tokio::test] + async fn create_no_capacity_is_queued_error() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + let api = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/oauth2/token")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "t" + }))) + .mount(&api) + .await; + Mock::given(method("GET")) + .and(path("/serverless-compute-resources")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([]))) + .mount(&api) + .await; + let c = VerdaClient::with_bases( + VerdaCreds { + client_id: "id".into(), + client_secret: "sec".into(), + inference_key: "inf".into(), + }, + &api.uri(), + &api.uri(), + ) + .unwrap(); + let spec = InstanceSpec { + name: "prism-t".into(), + max_lifetime_hours: 2.0, + ..InstanceSpec::default() + }; + let err = c.provision(&spec).await.unwrap_err(); + assert!( + matches!(err, LiumError::Cost(CostGuardrailError::NoCapacity)) + || lium_rent_pool::is_no_capacity(&err.to_string()), + "{err}" + ); + } + + #[test] + fn creds_debug_redacted() { + let c = VerdaCreds { + client_id: "id".into(), + client_secret: "SECRETVALUE".into(), + inference_key: "INF".into(), + }; + let d = format!("{c:?}"); + assert!(!d.contains("SECRETVALUE")); + assert!(!d.contains("INF")); + } +} diff --git a/crates/prism-verda/tests/live_b200_boot.rs b/crates/prism-verda/tests/live_b200_boot.rs new file mode 100644 index 000000000..fd1e45aa3 --- /dev/null +++ b/crates/prism-verda/tests/live_b200_boot.rs @@ -0,0 +1,81 @@ +//! Live proof: create a 1× B200 job-deployment and wait until Running. +//! Does not run the 1h harness. Requires `VERDA_*`. Never logs secrets. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use prism_lium::{EvalJobBackend, InstanceSpec}; +use prism_verda::{VerdaClient, VerdaCreds}; + +fn creds() -> Option { + let client_id = std::env::var("VERDA_CLIENT_ID").ok()?; + let client_secret = std::env::var("VERDA_CLIENT_SECRET").ok()?; + let inference_key = std::env::var("VERDA_INFERENCE_KEY").ok()?; + if client_id.is_empty() || client_secret.is_empty() || inference_key.is_empty() { + return None; + } + Some(VerdaCreds { + client_id, + client_secret, + inference_key, + }) +} + +#[tokio::test] +#[ignore = "live Verda B200 boot; set VERDA_*"] +async fn verda_b200_deployment_reaches_running() { + let creds = creds().expect("VERDA_CLIENT_ID / SECRET / INFERENCE_KEY"); + std::env::set_var("PRISM_VERDA_COMPUTE", "B200"); + std::env::set_var( + "PRISM_VERDA_IMAGE_REF", + "docker.io/pytorch/pytorch@sha256:c8268a92a69bd500f8be0e665b2630ee006dadaf7bfbc24249141b15ff622755", + ); + let client = VerdaClient::new(creds).unwrap(); + let offers = client.list_offers(None).await.expect("list compute"); + assert!( + offers + .iter() + .any(|o| o.gpu_type.to_ascii_lowercase().contains("b200")), + "no B200 SKU in serverless catalog: {:?}", + offers.iter().map(|o| &o.gpu_type).collect::>() + ); + let name = format!( + "prism-b200-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() + ); + eprintln!("verda_b200_boot deployment={name}"); + let spec = InstanceSpec { + name: name.clone(), + max_lifetime_hours: 1.0, + ..InstanceSpec::default() + }; + let inst = match client.provision(&spec).await { + Ok(i) => i, + Err(e) => { + let _ = client.terminate(&name).await; + panic!("provision B200: {e}"); + } + }; + assert_eq!(inst.provider, "verda"); + let gpu = inst.gpu_type.unwrap_or_default(); + assert!( + gpu.to_ascii_lowercase().contains("b200"), + "expected B200 compute, got {gpu}" + ); + let running = client.instance_running(&name).await.unwrap_or(false); + if !running { + // Pull + schedule can take several minutes. + for i in 0..40 { + tokio::time::sleep(std::time::Duration::from_secs(15)).await; + if client.instance_running(&name).await.unwrap_or(false) { + eprintln!("verda_b200_boot running after {}s", (i + 1) * 15); + break; + } + } + } + let ok = client.instance_running(&name).await.unwrap_or(false); + let _ = client.terminate(&name).await; + assert!(ok, "B200 job-deployment {name} never reached running"); +} diff --git a/crates/prism-verda/tests/live_dense.rs b/crates/prism-verda/tests/live_dense.rs new file mode 100644 index 000000000..7b90895f3 --- /dev/null +++ b/crates/prism-verda/tests/live_dense.rs @@ -0,0 +1,96 @@ +//! Live Verda BYOK 1h dense-1b proof. Requires `VERDA_*` and +//! `PRISM_AUTOMODEL_PIN_DIR`. Never logs secrets. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use base64::Engine; +use prism_lium::{EvalJobBackend, InstanceSpec}; +use prism_pipeline::{expand_zip_fields, tree_blob_for, SubmissionRequest}; +use prism_verda::{VerdaClient, VerdaCreds}; + +fn creds() -> Option { + let client_id = std::env::var("VERDA_CLIENT_ID").ok()?; + let client_secret = std::env::var("VERDA_CLIENT_SECRET").ok()?; + let inference_key = std::env::var("VERDA_INFERENCE_KEY").ok()?; + if client_id.is_empty() || client_secret.is_empty() || inference_key.is_empty() { + return None; + } + Some(VerdaCreds { + client_id, + client_secret, + inference_key, + }) +} + +fn dense_zip() -> Vec { + use std::io::Write; + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../docs/external-miner/examples/dense-1b"); + let mut buf = std::io::Cursor::new(Vec::new()); + { + let mut w = zip::ZipWriter::new(&mut buf); + let opts = zip::write::SimpleFileOptions::default(); + for name in ["automodel.base", "automodel.patch", "prism.toml"] { + w.start_file(name, opts).unwrap(); + w.write_all(&std::fs::read(root.join(name)).unwrap()) + .unwrap(); + } + w.finish().unwrap(); + } + buf.into_inner() +} + +#[tokio::test] +#[ignore = "live Verda GPU 1h; set VERDA_*"] +async fn verda_live_dense_eval_1h() { + let creds = creds().expect("VERDA_CLIENT_ID / SECRET / INFERENCE_KEY"); + std::env::set_var( + "PRISM_VERDA_IMAGE_REF", + // Public digest — nvcr.io stays Queue forever on miner Verda. + "docker.io/pytorch/pytorch@sha256:c8268a92a69bd500f8be0e665b2630ee006dadaf7bfbc24249141b15ff622755", + ); + if std::env::var("PRISM_TEST_TRAIN_MINUTES").is_err() { + std::env::set_var("PRISM_TEST_TRAIN_MINUTES", "60"); + } + let client = VerdaClient::new(creds).unwrap(); + let zip = dense_zip(); + let mut req = SubmissionRequest { + miner_hotkey: "dd".repeat(32), + zip_base64: Some(base64::engine::general_purpose::STANDARD.encode(zip)), + ..Default::default() + }; + expand_zip_fields(&mut req).expect("expand dense-1b"); + let tree = tree_blob_for(&req).expect("tree"); + let name = format!( + "prism-1h-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() + ); + eprintln!("verda_live deployment={name}"); + let spec = InstanceSpec { + name: name.clone(), + max_lifetime_hours: 2.0, + ..InstanceSpec::default() + }; + let inst = client.provision(&spec).await.expect("provision"); + assert_eq!(inst.provider, "verda"); + let eval = client + .exec_eval( + &inst.id, + &req.architecture_py, + &req.training_py, + tree.as_deref(), + ) + .await; + let _ = client.terminate(&inst.id).await; + let gone = client.verify_terminated(&inst.id).await.unwrap_or(false); + let eval = eval.expect("exec_eval"); + eprintln!( + "EVAL_METRICS bpb={} tokens_seen={} wall={} notes={:?} gpu={:?}", + eval.bpb, eval.tokens_seen, eval.wall_clock_seconds, eval.notes, eval.gpu_type + ); + assert!(gone, "deployment must be deleted"); + assert!(eval.tokens_seen > 0 || eval.bpb.is_finite()); +} diff --git a/docs/PRISM.md b/docs/PRISM.md index f9ea85f71..4ae86b0a2 100644 --- a/docs/PRISM.md +++ b/docs/PRISM.md @@ -20,8 +20,9 @@ accepted on live. Novel architectures remain allowed when expressed as AutoModel extensions in the patch. Full pin fields and ZIP layout: [`PRISM_RECIPE.md`](PRISM_RECIPE.md). -Each evaluation runs on a Lium GPU pod **funded by the miner** via -`X-Lium-Api-Key` (Sim backend in CI only). Intake applies the patch +Each evaluation runs on a Lium GPU pod **or Verda serverless job funded by +the miner** via `X-Lium-Api-Key` **or** Verda BYOK (`X-Verda-Client-Id` / +Secret / `X-Verda-Inference-Key`) (Sim backend in CI only). Intake applies the patch fail-closed onto the pin; the delta is persisted for `GET /v1/submissions/{id}/diff`. Copy / similarity / agentic review focus on the **miner delta** (touched files / hunks), then the shared **agentic** @@ -1144,7 +1145,7 @@ the harness semantics listed above, and the baseline sources they may reuse. | Dimension | Real | Fallback | |-----------|------|----------| -| Eval backend | Live Lium when not `PRISM_FORCE_SIM` — miners bill via `X-Lium-Api-Key` (operator `LIUM_API_KEY` optional fallback if `PRISM_ALLOW_OPERATOR_LIUM=1`). **Hard pin: 1× NVIDIA B200** (default). Explicit env fallbacks: **4× RTX 5090** (`PRISM_POD_GPU_COUNT=4`) or **2×/8× RTX PRO 6000** (`PRISM_POD_GPU_COUNT=2`/`8`). No silent 8× B200/5090 fallback; non-pin SKUs rejected at rent | `SimLiumBackend` | +| Eval backend | Live Lium **or Verda** when not `PRISM_FORCE_SIM` — miners bill via `X-Lium-Api-Key` or Verda BYOK (operator `LIUM_API_KEY` optional fallback if `PRISM_ALLOW_OPERATOR_LIUM=1`). **Hard pin: 1× NVIDIA B200**. Verda uses serverless job-deployments (no SSH). Sold-out stays queued | `SimLiumBackend` | | Reviewer | `/run/base/openrouter/api_key` exists → OpenRouter LLM | `SimReviewer` (deterministic) | | Agentic | same OpenRouter key → `OpenRouterAgent` | `SimAgent` (AST + metrics heuristics) | | Store | `BASE_DATABASE_URL` set → Postgres w/ migrations | in-memory (dev only) | diff --git a/docs/external-miner/prism.md b/docs/external-miner/prism.md index 3a7164229..1715efdae 100644 --- a/docs/external-miner/prism.md +++ b/docs/external-miner/prism.md @@ -33,7 +33,9 @@ prism.toml # optional — entry / model-config knobs 4. Produce a unified diff against the pin commit, e.g. `git diff > automodel.patch`. 5. Write `automodel.base` as a single line equal to `automodel_pin_id`, pack - the ZIP, and `POST /v1/submissions` with your hotkey + **`X-Lium-Api-Key`**. + the ZIP, and `POST /v1/submissions` with your hotkey + **`X-Lium-Api-Key`** + **or** Verda BYOK (`X-Verda-Client-Id`, `X-Verda-Client-Secret`, + `X-Verda-Inference-Key`). Models must stay in **850M–1B parameters** (total unique; tied embeddings once). A 215M pack is **invalid** for recipe 2.1. Miner **model code** (build/train/ @@ -119,19 +121,29 @@ an example, not a scored baseline. **Diff visibility.** After intake, inspect your applied delta at `GET /v1/submissions/{id}/diff` (full unified diff + diffstat / classification). -Evaluation runs on **miner-funded** Lium GPU pods (you pay the rent). Master -still operates the pod over SSH; you do **not** deploy a miner CVM. CI uses -`SimLiumBackend` and does not need a key. +Evaluation runs on **miner-funded** GPUs (Lium SSH pods or Verda serverless +jobs). You pay the rent. Master still operates the job; you do **not** +deploy a miner CVM. CI uses `SimLiumBackend` and does not need a key. ## Pay for your own GPU (required on live) -Create a [Lium](https://lium.io) account, fund it, and pass your API key on -every live submit: +Create a [Lium](https://lium.io) **or** [Verda](https://verda.com) account, +fund it, and pass **one** provider on every live submit: ```http X-Lium-Api-Key: ``` +```http +X-Verda-Client-Id: +X-Verda-Client-Secret: +X-Verda-Inference-Key: +``` + +`X-Verda-Api-Key` aliases the inference token. If both providers are +complete, set `X-Compute-Provider: lium` or `verda`. You cannot set +`image` / `cmd` / `template` — operator pin only. + The key is held in master memory for that submission and may also land in a **TTL-bounded encrypted seal file** on the master host (default ≥36h; never in Postgres, never logged). Master **re-seals** on measure start and heartbeats diff --git a/xtask/src/external_docs_check.rs b/xtask/src/external_docs_check.rs index 3732a5e1a..716525f00 100644 --- a/xtask/src/external_docs_check.rs +++ b/xtask/src/external_docs_check.rs @@ -31,6 +31,7 @@ const PRISM_AUTOMODEL_PINS: &[(&str, &str)] = &[ ("recipe_json_pin_id", "automodel_pin_id"), ("diff_route", "/v1/submissions/{id}/diff"), ("lium_byok", "X-Lium-Api-Key"), + ("verda_byok", "X-Verda-Client-Id"), ("competition_id", "prism-v2.1"), ("scoring_generation_21", "scoring_generation"), ("new_competition", "new competition"),