Merge pull request #171 from BaseIntelligence/fix/prism-prepod-orphan-requeue

fix(prism): requeue pre-pod orphans and unblock infra retry

Author: echobt

crates/prism-challenge/src/api.rs

@@ -19,7 +19,7 @@ use submission_gating::{infra_resubmit_allowed, GatingState, GatingStore, Metagr
 
 use prism_recipe::{BASELINE_ARCHITECTURE_PY, BASELINE_TRAINING_PY};
 
-use crate::CHALLENGE_ID;
+use crate::{NoScoreReasonCode, CHALLENGE_ID};
 use prism_eval_store::{detail_view, eval_json, list_view};
 use prism_intake::{
     coldkey_of, json_err, map_submission_err, map_zip_err, materialize_arch, metagraph_uid, now_ms,
@@ -182,6 +182,13 @@ async fn gate_one_max(
     Ok(None)
 }
 
+fn is_challenge_internal(row: &SubmissionState) -> bool {
+    matches!(
+        row.final_score,
+        Some(FinalScore::NoScore(c)) if c == NoScoreReasonCode::ChallengeInternal as u8
+    )
+}
+
 /// [`prism_pipeline::queued_row`] with the coldkey resolved from the live
 /// metagraph snapshot (`None` off-chain).
 fn queued_row(
@@ -239,11 +246,20 @@ async fn post_submission(
     let id = prism_pipeline::submission_id(&req);
     let gate_challenge = prism_pipeline::gating_key(req.arch_id.as_deref());
 
-    // Idempotent duplicate: identical contract bytes never conflict gating.
-    let exists = match st.store.get(&id).await {
-        Ok(r) => r.is_some(),
+    // Identical bytes never conflict gating. A failed infra row is recovered
+    // here (same path as `/retry`) so miners are not stuck on `already-queued`
+    // while the slot stays `failed` + `blocked`.
+    let existing = match st.store.get(&id).await {
+        Ok(r) => r,
         Err(e) => return json_err(StatusCode::INTERNAL_SERVER_ERROR, "store", &e.to_string()),
     };
+    if existing
+        .as_ref()
+        .is_some_and(|r| r.status == Stage::Failed && is_challenge_internal(r))
+    {
+        return post_retry(State(st), Path(id), headers).await;
+    }
+    let exists = existing.is_some();
     let uid = if exists {
         None
     } else {
@@ -426,17 +442,9 @@ async fn post_retry(
         );
     }
     let gate_key = prism_pipeline::gating_key(row.arch_id.as_deref());
-    let mut infra = matches!(row.final_score, Some(FinalScore::NoScore(6)));
-    if infra {
-        if let Some(g) = &st.gating {
-            infra = g
-                .get(&gate_key, &row.miner_hotkey)
-                .await
-                .ok()
-                .flatten()
-                .is_some_and(|gr| infra_resubmit_allowed(&gr, now_ms()));
-        }
-    }
+    // ChallengeInternal is always miner-retryable (the 30m window only lets a
+    // *different* ZIP through intake while blocked).
+    let infra = is_challenge_internal(&row);
     if !infra {
         if let Err(resp) = verify_bearer(&st.admin_token_hashes, "admin", &headers) {
             return resp;
@@ -1288,6 +1296,103 @@ mod tests {
         assert_eq!(s, StatusCode::ACCEPTED, "{v}");
     }
 
+    #[tokio::test]
+    async fn challenge_internal_reposts_and_retries_after_window() {
+        let (st, gating) = gated_state(&[[0x11; 32]]);
+        let app = submission_router(Arc::clone(&st));
+        let body = serde_json::to_vec(&crate::example_valid_request()).unwrap();
+        let (s, v) = call(
+            app.clone(),
+            Request::post("/v1/submissions")
+                .header("content-type", "application/json")
+                .body(Body::from(body.clone()))
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(s, StatusCode::ACCEPTED, "{v}");
+        let id = v["submission_id"].as_str().unwrap().to_owned();
+        let hk = "11".repeat(32);
+        st.store
+            .apply(
+                &id,
+                &StatePatch {
+                    status: Some(Stage::Failed),
+                    final_score: Some(FinalScore::NoScore(
+                        NoScoreReasonCode::ChallengeInternal as u8,
+                    )),
+                    ..StatePatch::default()
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        gating
+            .set_terminal(
+                "prism",
+                &hk,
+                submission_gating::GatingState::Blocked,
+                Some("install"),
+            )
+            .await
+            .unwrap();
+        assert!(gating.set_updated_at_ms(
+            "prism",
+            &hk,
+            now_ms().saturating_sub(submission_gating::INFRA_RESUBMIT_WINDOW_MS + 1),
+        ));
+        let (s, v) = call(
+            app.clone(),
+            Request::post(format!("/v1/submissions/{id}/retry"))
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(s, StatusCode::ACCEPTED, "{v}");
+        assert_eq!(v["status"], "queued");
+        st.store
+            .apply(
+                &id,
+                &StatePatch {
+                    status: Some(Stage::Failed),
+                    final_score: Some(FinalScore::NoScore(
+                        NoScoreReasonCode::ChallengeInternal as u8,
+                    )),
+                    ..StatePatch::default()
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        gating
+            .set_terminal(
+                "prism",
+                &hk,
+                submission_gating::GatingState::Blocked,
+                Some("install"),
+            )
+            .await
+            .unwrap();
+        assert!(gating.set_updated_at_ms(
+            "prism",
+            &hk,
+            now_ms().saturating_sub(submission_gating::INFRA_RESUBMIT_WINDOW_MS + 1),
+        ));
+        let (s, v) = call(
+            app,
+            Request::post("/v1/submissions")
+                .header("content-type", "application/json")
+                .body(Body::from(body))
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(s, StatusCode::ACCEPTED, "{v}");
+        assert_eq!(v["status"], "queued");
+        assert_eq!(
+            st.store.get(&id).await.unwrap().unwrap().status,
+            Stage::Queued
+        );
+    }
+
     #[tokio::test]
     async fn precheck_detects_copy_without_queuing() {
         let st = state();

crates/prism-orphan/src/lib.rs

@@ -242,6 +242,36 @@ mod tests {
         assert!(matches!(got.final_score, Some(FinalScore::NoScore(_))));
     }
 
+    #[tokio::test]
+    async fn boot_reconcile_requeues_pre_pod_review_without_failing() {
+        let store: Arc<dyn PrismStore> = Arc::new(MemoryPrismStore::default());
+        let r = row("feedfacefeedface", Stage::LlmReview, None);
+        store.insert_queued(&r).await.unwrap();
+        store
+            .apply(
+                &r.id,
+                &prism_store::StatePatch {
+                    status: Some(Stage::LlmReview),
+                    ..Default::default()
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        let active = Arc::new(ActiveJobs::new());
+        let be: Arc<dyn EvalJobBackend> = Arc::new(NoopBackend::dead());
+        let report = reconcile_once(store.as_ref(), &active, None, be, 0, true, None)
+            .await
+            .unwrap();
+        assert_eq!(report.requeued, 1);
+        assert_eq!(report.failed, 0);
+        assert_eq!(report.terminate_attempts, 0);
+        let got = store.get(&r.id).await.unwrap().unwrap();
+        assert_eq!(got.status, Stage::Queued);
+        assert!(got.pod_id.is_none());
+        assert!(got.final_score.is_none());
+    }
+
     #[tokio::test]
     async fn boot_reconcile_resumes_alive_pod_without_terminate() {
         let store: Arc<dyn PrismStore> = Arc::new(MemoryPrismStore::default());

crates/prism-orphan/src/reconcile.rs

@@ -111,7 +111,16 @@ pub async fn reconcile_once(
                 }
             }
         }
-        // Unreattachable mid-pod or pre-measure without pod: fail + best-effort stop.
+        // Pre-pod (screens / claim / not yet rented): never fail-orphan.
+        // A restart here has no Lium instance; fail-closed with `pod (none)`
+        // burned the miner slot and asked them to stop a pod that does not exist.
+        if row.pod_id.is_none() {
+            if requeue_pre_pod(store, &row, reason).await {
+                report.requeued = report.requeued.saturating_add(1);
+            }
+            continue;
+        }
+        // Unreattachable mid-pod: fail + best-effort stop.
         let mut terminated = false;
         let key_present = payer.is_some_and(|p| p.vault.get(&row.id).is_some());
         if let Some(pod) = row.pod_id.as_deref() {
@@ -141,6 +150,32 @@ pub async fn reconcile_once(
     Ok(report)
 }
 
+async fn requeue_pre_pod(store: &dyn PrismStore, row: &SubmissionState, reason: &str) -> bool {
+    let ok = store
+        .apply(
+            &row.id,
+            &StatePatch {
+                status: Some(Stage::Queued),
+                error_detail: None,
+                ..StatePatch::default()
+            },
+            Some(&StageEvent {
+                stage: Stage::Queued,
+                detail: Some(serde_json::json!({
+                    "pre_pod_requeue": true,
+                    "reason": reason,
+                })),
+                at_ms: 0,
+            }),
+        )
+        .await
+        .is_ok();
+    if ok {
+        info!(submission_id = %row.id, reason, "pre-pod mid-flight requeued");
+    }
+    ok
+}
+
 /// Probe pod + vault; on success requeue to `queued` keeping `pod_id`.
 async fn try_resume(
     store: &dyn PrismStore,

docs/PRISM.md

@@ -126,18 +126,21 @@ special: each miner `X-Lium-Api-Key` has its **own** Lium budget (no shared
 process-wide rent serialize queue). The orchestrator **requeues without
 burning** `retry_count` / gating attempts. A background tick re-queues
 failed 429 rows from the last **6 hours**. After an infra `blocked`, the
-miner may **resubmit for up to 30 minutes** (new `POST /v1/submissions` or
-`POST /v1/submissions/{id}/retry` for `ChallengeInternal`); after the window
-the slot stays blocked until the metagraph watcher reopens it (hotkey left /
-replaced).
+miner may **`/retry` or re-POST the same bytes** for `ChallengeInternal`
+without a time cutoff. A *different* ZIP is only accepted inside the
+**30-minute** infra window; after that the slot stays blocked until the
+metagraph watcher reopens it (hotkey left / replaced).
 
 **Training-only entries** gate separately under the composite challenge key
 `prism:train:<arch_id>`: one accepted entry per `(hotkey, arch_id)`, with
 the same auto-retry classes, the same terminal `rejected`/`blocked` states,
 and the same watcher resets (reconciliation is prefix-scoped, so `prism`
 covers every `prism:train:*` row). Idempotency stays the contract-bytes
-`submission_id`: resubmitting identical bytes is an `already-queued` no-op,
-never a gate conflict.
+`submission_id`: resubmitting identical in-flight / successful bytes is an
+`already-queued` no-op (never a gate conflict). A failed `ChallengeInternal`
+row is recovered by that same POST or `/retry`. Pre-pod mid-flight rows
+(`llm_review` / `similarity` / `provisioning` with no `pod_id`) requeue on
+control-plane restart instead of fail-orphan with `pod (none)`.
 
 ## Architecture registry + competition
 

docs/external-miner/prism.md

@@ -136,16 +136,18 @@ versioned descriptor). Trust `/v1/recipe`, not marketing chart labels.
 - Infra failures (Lium pod, review/similarity/LLM infra) **auto-retry up to 3
   times**; harness `EVAL_FAIL` (miner/model code) is terminal for that attempt
   and is **not** auto-retried. Cheat / rejected verdicts are terminal. After an
-  infra failure (`ChallengeInternal`), you may **recover within 30 minutes**
-  via `POST /v1/submissions/{id}/retry` with **`X-Lium-Api-Key`** (required on
-  live when another GPU rent is needed). After 30 minutes the slot stays
-  blocked until your hotkey leaves the metagraph.
+  infra failure (`ChallengeInternal` / `control_plane_restart`), recover with
+  `POST /v1/submissions/{id}/retry` and **`X-Lium-Api-Key`** on live when
+  another GPU rent is needed — **no 30-minute cutoff** on `/retry` or on
+  re-POSTing the **same** pin+patch (that used to return `already-queued`
+  while the row stayed `failed`). A *different* ZIP while the slot is
+  `blocked` is still only accepted inside the 30-minute infra window.
 
 ### Retry vs re-POST
 
 | Action | When | Headers |
 |--------|------|---------|
-| Re-POST the **same** ZIP | Always safe | Same as submit | Returns `200 already-queued` — **no new GPU run**; does not recover a failed row |
+| Re-POST the **same** ZIP | Always safe | Same as submit | In-flight / scored → `200 already-queued`. Failed `ChallengeInternal` → same as `/retry` (`202 queued`) |
 | `POST /v1/submissions/{id}/retry` | Row status is **`failed`** only | **`X-Lium-Api-Key`** on live (infra recovery); admin Bearer for operator non-infra retries | Requeues measure; wrong/missing Lium key → `400 missing_lium_api_key` |
 | `/retry` on non-failed | — | — | `409 not_failed` — hotkey or Bearer alone does not change that |
 

docs/external-miner/troubleshoot.md

@@ -12,7 +12,7 @@
 | `409 schedule` "daily manual run quota exceeded" | Manual anti-spam cap (10/day) — round-loop auto-enqueue does **not** spend it | `GET /v1/quota/{hotkey}` → `manual.remaining`; wait until next UTC day |
 | Active harness but no runs this round | Rare race / restart before auto-enqueue; or eliminated cooldown | Wait for the round tick / ask ops `admin/rounds/current/requeue`; check `eliminated_until_round` |
 | `auto_retry` events, class `install` | Dep won't install (bad name/version, heavy source build) | Design: `GET /v1/runs/{id}/logs`; Prism: `GET /v1/submissions/{id}/logs?since=` |
-| `control_plane_restart` / `harness_detached` | Restart could not reattach (dead pod or unrecoverable BYOK seal) | Stop the Lium pod if still billing; resubmit with `X-Lium-Api-Key`. Healthy pods are resumed automatically — do not kill them on a routine master redeploy. |
+| `control_plane_restart` / `harness_detached` | Restart could not reattach (dead pod or unrecoverable BYOK seal) | If `pod_id` is null, **no Lium pod was rented** — do not hunt a pod. `POST /v1/submissions/{id}/retry` (or re-POST the same ZIP) with `X-Lium-Api-Key`. Healthy pods with a `pod_id` resume automatically. |
 | Run `failed` / Score 0 | Missing pages, timeout, crash | `GET /v1/runs/{id}/events`; ensure three required HTML pages |
 | External call refused (`403`) | Target is internal-blocklisted (metadata IP, loopback, RFC1918/VPC, control plane) | Call public endpoints only; egress is otherwise open |
 | Pages look empty in viewer | Sanitize stripped content | Scripts/`on*` handlers are removed; use static HTML/CSS |
@@ -35,7 +35,7 @@
 | `similar: true` on precheck | Would hit intake copy gate | Change the patch vs prior champions; starting from the operator pin is fine |
 | `429 precheck_quota_exceeded` | 3 prechecks/coldkey/UTC day used | Wait until next UTC day; rotating hotkeys does not reset |
 | `400 missing_lium_api_key` | Live path needs miner-funded Lium | Pass `X-Lium-Api-Key` (your Lium account); see [`prism.md`](prism.md) |
-| `409 not_failed` on `/retry` | Row is not `failed` (queued/running/scored) | `/retry` is only for failed rows. Identical ZIP re-POST → `already-queued` (no-op). After infra failure use `/retry` + `X-Lium-Api-Key` |
+| `409 not_failed` on `/retry` | Row is not `failed` (queued/running/scored) | `/retry` is only for failed rows. In-flight identical ZIP re-POST → `already-queued`. Failed infra: `/retry` or same-ZIP POST recovers the row |
 | `400 missing_lium_api_key` on `/retry` | Failed infra row needs another GPU rent | Send `X-Lium-Api-Key` (hotkey / Bearer alone is not enough) |
 | Stuck `Provisioning` | Lium market / underfunded key / no 1×5090 | Check Lium balance; Prism hard-pins **1× RTX 5090** (non-5090 rejected) |
 | Idempotent replay | Same `submission_id` (pin id + patch bytes) | Expected — returns prior row |