fix(prism): queue submits when lium b200s sell out

Author: echobt

bins/prism-challenge/src/main.rs

@@ -782,7 +782,7 @@ fn spawn_orchestrator(
     spawn_rate_limit_recovery(store, gating);
 }
 
-/// Re-queue last-6h failed rows that died on Lium HTTP 429 (no `retry_count` burn).
+/// Re-queue failed Lium 429 (6h window) and no_capacity / B200 sold-out rows.
 async fn recover_rate_limited(
     store: &Arc<dyn PrismStore>,
     gating: Option<&Arc<dyn GatingStore>>,

crates/lium-rent-pool/src/lib.rs

@@ -2,15 +2,22 @@
 //!
 //! Live Prism eval is miner-funded (`X-Lium-Api-Key`). Each miner key has its
 //! own Lium rate budget — there is **no** process-wide rent serialize queue.
-//! This crate only classifies 429 bodies and decides autonomous recovery.
+//! This crate classifies 429 / no-capacity rent failures and recovery.
 
 #![forbid(unsafe_code)]
 #![allow(clippy::missing_errors_doc)]
 
 /// Autonomous recovery looks back this far for failed 429 submissions.
 pub const RECOVERY_WINDOW_MS: u64 = 6 * 60 * 60 * 1000;
 
-/// True when a failed row should re-enter the rent queue (429 within window).
+/// Miner-facing text when Lium has no matching 1× B200 offer.
+pub const CAPACITY_NOTE: &str =
+    "B200s are currently out of capacity on Lium; this job is queued until an offer appears.";
+
+/// Always-on policy (intake / recipe / `/v1/status`).
+pub const CAPACITY_POLICY: &str = "When Lium has no matching 1× B200 offer, the job stays queued and retries until an offer appears (sold out is not Score(0)). Bad ZIP, auth, and template-permission errors still fail.";
+
+/// True when a failed row should re-enter the rent queue.
 #[must_use]
 pub fn should_recover(error_detail: &str, updated_at_ms: u64, now_ms: u64) -> bool {
     let l = error_detail.to_ascii_lowercase();
@@ -19,6 +26,9 @@ pub fn should_recover(error_detail: &str, updated_at_ms: u64, now_ms: u64) -> bo
     if l.contains("huggingface") || l.contains("fineweb") || l.contains("\"stage\": \"dataset\"") {
         return false;
     }
+    if is_no_capacity(error_detail) {
+        return true;
+    }
     is_rate_limited(error_detail) && now_ms.saturating_sub(updated_at_ms) <= RECOVERY_WINDOW_MS
 }
 
@@ -58,6 +68,36 @@ pub fn is_rate_limited(msg: &str) -> bool {
     l.contains("429") || l.contains("too many requests") || l.contains("rate limit")
 }
 
+/// Auth / template-permission / missing BYOK — never treat as sold-out.
+#[must_use]
+pub fn is_auth_or_permission(msg: &str) -> bool {
+    let l = msg.to_ascii_lowercase();
+    l.contains("permission")
+        || l.contains("unauthorized")
+        || l.contains("forbidden")
+        || l.contains("401")
+        || l.contains("invalid api key")
+        || l.contains("missing_lium_api_key")
+        || l.contains("api key missing")
+}
+
+/// No matching Lium offer / B200 sold out (not miner ZIP, not auth).
+#[must_use]
+pub fn is_no_capacity(msg: &str) -> bool {
+    if is_auth_or_permission(msg) {
+        return false;
+    }
+    let l = msg.to_ascii_lowercase();
+    l.contains("no_capacity")
+        || l.contains("no lium offer")
+        || l.contains("no offer matches")
+        || l.contains("no matching offer")
+        || l.contains("sold out")
+        || l.contains("out of capacity")
+        || l.contains("lack of b200")
+        || (l.contains("b200") && (l.contains("unavailable") || l.contains("no offer")))
+}
+
 #[cfg(test)]
 mod tests {
     #![allow(clippy::unwrap_used)]
@@ -81,5 +121,22 @@ mod tests {
             100,
             100 + 1
         ));
+        assert!(is_no_capacity(
+            "measure: provision: no Lium offer matches GPU preference and price caps (no_capacity)"
+        ));
+        assert!(is_no_capacity("lium rent sold out: no matching offer"));
+        assert!(!is_no_capacity(
+            "lium api: POST /rent -> 400 permission to rent this template"
+        ));
+        assert!(should_recover(
+            "provision: no_capacity (no matching B200 offer)",
+            0,
+            RECOVERY_WINDOW_MS + 1
+        ));
+        assert!(!should_recover(
+            "lium api: 400 permission to rent this template",
+            100,
+            100 + 1
+        ));
     }
 }

crates/prism-challenge/src/api.rs

@@ -304,14 +304,7 @@ async fn post_submission(
                     }
                 }
             }
-            (
-                StatusCode::ACCEPTED,
-                Json(prism_pipeline::SubmissionAccepted {
-                    submission_id: id,
-                    status: "accepted".into(),
-                }),
-            )
-                .into_response()
+            (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()) {
@@ -529,6 +522,7 @@ async fn get_status(State(st): State<Arc<AppState>>) -> Response {
         "queues": {"queued": queued, "active": active},
         "recent_terminal": done_24h,
         "recipe_pin": prism_recipe::recipe_pin_hex(),
+        "lium_capacity_note": lium_rent_pool::CAPACITY_POLICY,
     }))
     .into_response()
 }
@@ -1040,6 +1034,7 @@ mod tests {
         )
         .await;
         assert_eq!(s, StatusCode::ACCEPTED, "{v}");
+        assert_eq!(v["note"], lium_rent_pool::CAPACITY_POLICY);
         let id = v["submission_id"].as_str().unwrap().to_owned();
 
         let (s2, v2) = call(
@@ -1073,6 +1068,7 @@ mod tests {
         .await;
         assert_eq!(s, StatusCode::OK);
         assert_eq!(v["backend"], "sim");
+        assert_eq!(v["lium_capacity_note"], lium_rent_pool::CAPACITY_POLICY);
         let (s, v) = call(
             app.clone(),
             Request::get("/v1/recipe").body(Body::empty()).unwrap(),

crates/prism-challenge/src/orchestrator.rs

@@ -284,38 +284,47 @@ impl<C: ChainClient + Send> Orchestrator<C> {
 
     async fn maybe_auto_retry(&self, row: &SubmissionState, class: &str, msg: &str) -> bool {
         let rate = lium_rent_pool::is_rate_limited(msg);
-        if !rate && row.retry_count >= self.cfg.auto_retry_max {
+        let capacity = lium_rent_pool::is_no_capacity(msg);
+        let skip_burn = rate || capacity;
+        if !skip_burn && row.retry_count >= self.cfg.auto_retry_max {
             return false;
         }
         warn!(
             submission_id = %row.id,
             class,
             rate_limited = rate,
+            no_capacity = capacity,
             attempt = row.retry_count + 1,
             max = self.cfg.auto_retry_max,
             error = %msg,
             "auto-retrying submission after infra failure"
         );
-        let _ = self.store.reset_for_retry(&row.id, !rate).await;
+        let _ = self.store.reset_for_retry(&row.id, !skip_burn).await;
+        let note = capacity.then_some(lium_rent_pool::CAPACITY_NOTE);
         let _ = self
             .store
             .apply(
                 &row.id,
-                &StatePatch::default(),
+                &StatePatch {
+                    error_detail: note.map(str::to_owned),
+                    ..StatePatch::default()
+                },
                 Some(&StageEvent {
                     stage: Stage::Queued,
                     detail: Some(serde_json::json!({
                         "auto_retry": true,
                         "class": class,
                         "rate_limited": rate,
+                        "no_capacity": capacity,
+                        "note": note,
                         "attempt": row.retry_count + 1,
                         "error": msg,
                     })),
                     at_ms: 0,
                 }),
             )
             .await;
-        if !rate {
+        if !skip_burn {
             if let Some(g) = &self.gating {
                 let _ = g
                     .bump_attempt(
@@ -376,8 +385,8 @@ impl<C: ChainClient + Send> Orchestrator<C> {
         // `train_script` training crash) additionally fail terminal under
         // their own class, which grants unbounded resubmit. Every other
         // EVAL_FAIL phase (eval / battery / score) stays the historical
-        // windowed `install` class. Non-EVAL_FAIL failures are Lium infra and
-        // keep `install` + operator-paid auto-retry.
+        // windowed `install` class. Non-EVAL_FAIL failures are Lium infra:
+        // 429 / `no_capacity` requeue without burn; other infra uses auto-retry.
         if msg.contains("EVAL_FAIL") {
             let class = classify_eval_fail(&msg);
             fail_terminal(self.store.as_ref(), self.gating.as_ref(), row, class, &msg).await;

crates/prism-challenge/tests/e2e_orchestrate_sim.rs

@@ -6,6 +6,7 @@
 
 use std::sync::Arc;
 
+use async_trait::async_trait;
 use chain::{AxonInfo, ChainError, Metagraph, WeightsTlockPayload};
 use chain::{ChainClient, FakeChain, FakeChainConfig};
 use challenge_agentic::SimAgent;
@@ -15,7 +16,10 @@ use prism_challenge::{
     example_valid_request, submission_id, GatewayClient, GatewayClientConfig, MemoryPrismStore,
     Orchestrator, OrchestratorConfig, PrismStore, ScoringMode, Stage, StatePatch, SubmissionState,
 };
-use prism_lium::{EvalJobBackend, SimLiumBackend};
+use prism_lium::{
+    CostGuardrailError, EvalJobBackend, Instance, InstanceSpec, LiumError, Offer, RemoteExecResult,
+    SimLiumBackend,
+};
 use prism_review::SimReviewer;
 use std::sync::Mutex;
 
@@ -337,3 +341,162 @@ async fn emit_and_submit_covers_expected_set() {
     assert_eq!(tip.epoch, 7);
     assert_eq!(store.emit_cursor(541).await.unwrap(), Some(7));
 }
+
+/// Provision always fails with a fixed Lium error (no rent).
+struct ProvisionFail(&'static str);
+
+#[async_trait]
+impl EvalJobBackend for ProvisionFail {
+    async fn list_offers(&self, _: Option<f64>) -> Result<Vec<Offer>, LiumError> {
+        Ok(Vec::new())
+    }
+    async fn provision(&self, _: &InstanceSpec) -> Result<Instance, LiumError> {
+        if self.0 == "capacity" {
+            return Err(CostGuardrailError::NoCapacity.into());
+        }
+        Err(LiumError::Api(
+            "POST /executors/x/rent -> 400 You don't have permission to rent this template.".into(),
+        ))
+    }
+    async fn terminate(&self, _: &str) -> Result<(), LiumError> {
+        Ok(())
+    }
+    async fn verify_terminated(&self, _: &str) -> Result<bool, LiumError> {
+        Ok(true)
+    }
+    async fn exec_eval(
+        &self,
+        _: &str,
+        _: &str,
+        _: &str,
+        _: Option<&[u8]>,
+    ) -> Result<RemoteExecResult, LiumError> {
+        Err(LiumError::Exec("unreachable".into()))
+    }
+}
+
+fn row_from_example(id: &str, req: &prism_challenge::SubmissionRequest) -> SubmissionState {
+    SubmissionState {
+        id: id.to_owned(),
+        miner_hotkey: req.miner_hotkey.clone(),
+        miner_coldkey: None,
+        epoch: 7,
+        netuid: 541,
+        status: Stage::Queued,
+        architecture_py: req.architecture_py.clone(),
+        training_py: req.training_py.clone(),
+        tree_blob: None,
+        label: req.label.clone(),
+        pod_id: None,
+        pod_provider: None,
+        receipt: None,
+        metrics_json: None,
+        bpb: None,
+        arch_id: None,
+        review: None,
+        similarity: None,
+        final_score: None,
+        retry_count: 0,
+        error_detail: None,
+        created_at_ms: 1,
+        updated_at_ms: 1,
+    }
+}
+
+fn orch_with_backend(
+    store: &Arc<MemoryPrismStore>,
+    chain: &Arc<LockedFake>,
+    backend: Arc<dyn EvalJobBackend>,
+) -> Orchestrator<LockedFake> {
+    let gateway = Arc::new(
+        GatewayClient::new(GatewayClientConfig {
+            base_url: "dry-run".into(),
+            max_attempts: 1,
+            backoff: std::time::Duration::from_millis(1),
+        })
+        .unwrap(),
+    );
+    Orchestrator::new(
+        OrchestratorConfig {
+            netuid: 541,
+            scoring_mode: ScoringMode::Shadow,
+            auto_retry_max: 0,
+            claim_poll: std::time::Duration::from_millis(10),
+            ..Default::default()
+        },
+        Arc::clone(store) as Arc<dyn PrismStore>,
+        backend,
+        Arc::new(SimReviewer::new()),
+        Arc::new(SimAgent::new()),
+        &gateway,
+        Arc::clone(chain),
+        sk(),
+    )
+}
+
+#[tokio::test]
+async fn no_capacity_requeues_with_b200_note() {
+    let store = Arc::new(MemoryPrismStore::new());
+    let chain = Arc::new(LockedFake(Mutex::new(fake_chain())));
+    let orch = orch_with_backend(
+        &store,
+        &chain,
+        Arc::new(ProvisionFail("capacity")) as Arc<dyn EvalJobBackend>,
+    );
+    let req = example_valid_request();
+    let id = submission_id(&req);
+    store
+        .insert_queued(&row_from_example(&id, &req))
+        .await
+        .unwrap();
+
+    assert!(orch.cycle_once().await.unwrap());
+    let row = store.get(&id).await.unwrap().expect("row");
+    assert_eq!(row.status, Stage::Queued, "{row:?}");
+    assert_eq!(row.retry_count, 0, "sold-out must not burn retry_count");
+    assert!(row.final_score.is_none());
+    let detail = row.error_detail.unwrap_or_default();
+    assert!(
+        detail.contains("B200s are currently out of capacity on Lium"),
+        "{detail}"
+    );
+    let events = store.events(&id).await.unwrap();
+    let queued_note = events.iter().any(|e| {
+        e.stage == Stage::Queued
+            && e.detail.as_ref().is_some_and(|d| {
+                d.get("no_capacity") == Some(&serde_json::json!(true))
+                    && d.get("note")
+                        .and_then(|n| n.as_str())
+                        .is_some_and(|n| n.contains("B200s are currently out of capacity"))
+            })
+    });
+    assert!(queued_note, "events={events:?}");
+
+    assert!(orch.cycle_once().await.unwrap());
+    let row = store.get(&id).await.unwrap().expect("row");
+    assert_eq!(row.status, Stage::Queued, "next tick still queued");
+}
+
+#[tokio::test]
+async fn template_permission_stays_failed() {
+    let store = Arc::new(MemoryPrismStore::new());
+    let chain = Arc::new(LockedFake(Mutex::new(fake_chain())));
+    let orch = orch_with_backend(
+        &store,
+        &chain,
+        Arc::new(ProvisionFail("permission")) as Arc<dyn EvalJobBackend>,
+    );
+    let req = example_valid_request();
+    let id = submission_id(&req);
+    store
+        .insert_queued(&row_from_example(&id, &req))
+        .await
+        .unwrap();
+
+    assert!(orch.cycle_once().await.unwrap());
+    let row = store.get(&id).await.unwrap().expect("row");
+    assert_eq!(row.status, Stage::Failed, "{row:?}");
+    let detail = row.error_detail.unwrap_or_default();
+    assert!(detail.contains("permission"), "{detail}");
+    assert!(!detail.contains("B200s are currently out of capacity"));
+}

crates/prism-store/src/store.rs

@@ -40,7 +40,7 @@ pub trait PrismStore: Send + Sync + std::fmt::Debug {
     /// post-run infra retry resumes without another GPU run; an incomplete
     /// measure attempt is cleared so provisioning starts cleanly.
     /// `bump_retry` increments `retry_count` (manual/auto infra). Pass
-    /// `false` for Lium 429 autonomous requeue (do not burn attempt budget).
+    /// `false` for Lium 429 / `no_capacity` requeue (do not burn attempt budget).
     async fn reset_for_retry(
         &self,
         id: &str,