feat(prism): operator retry route for failed submissions

- `POST /v1/submissions/{id}/retry`: requeue a failed row by clearing pod,
  receipt, bpb, review, similarity, score and error columns (new
  `reset_prism_submission_for_retry` SQL + `PrismStore::reset_for_retry`).
- Guarded by `AppState.retry_max` (= orchestrator max_attempts, 2).
- Journal entry `{"op":"retry"}` appended to the stage log.

Author: echobt

bins/prism-challenge/src/main.rs

@@ -29,6 +29,8 @@ use prism_challenge::{
 };
 use prism_lium::{EvalJobBackend, LiumClient, LiumSshConfig, SimLiumBackend};
 use prism_review::{OpenRouterClient, ReviewBackend, SimReviewer};
+const MAX_ATTEMPTS: u32 = 2;
+
 use tokio::net::TcpListener;
 use tokio::sync::Semaphore;
 use trustroot::encode_hex;
@@ -266,6 +268,7 @@ async fn cmd_serve(cli: Cli) -> Result<(), String> {
         backend_mode: Box::leak(
             format!("{backend_mode}/{reviewer_mode}/{store_mode}").into_boxed_str(),
         ),
+        retry_max: MAX_ATTEMPTS,
     });
     let app = submission_router(Arc::clone(&state));
 
@@ -287,7 +290,7 @@ async fn cmd_serve(cli: Cli) -> Result<(), String> {
         ssh_public_keys: ssh_pks,
         image_digest: None,
         claim_poll: Duration::from_millis(750),
-        max_attempts: 2,
+        max_attempts: MAX_ATTEMPTS,
         similarity_corpus_limit: 6,
         llm_weight: cli.llm_weight,
         stuck_grace_secs: 7 * 3600,

crates/db/src/prism_store.rs

@@ -215,6 +215,25 @@ pub async fn update_prism_submission(
     Ok(row)
 }
 
+/// Reset a failed row for a retry: clears all execution/score fields and
+/// re-queues it. `retry_count` is bumped (policy enforced by the caller).
+///
+/// # Errors
+/// SQL error / 0 rows for id.
+pub async fn reset_prism_submission_for_retry(
+    pool: &PgPool,
+    id: &str,
+) -> Result<PrismSubmissionRow, DbError> {
+    let q = format!(
+        "UPDATE prism_submission SET            status = 'queued', pod_id = NULL, pod_provider = NULL,            receipt_json = NULL, metrics_json = NULL, bpb = NULL,            review_json = NULL, similarity_json = NULL,            kind = NULL, score = NULL, absence_reason = NULL,            error_detail = NULL, retry_count = retry_count + 1, updated_at = now()          WHERE id = $1 RETURNING {COLS}"
+    );
+    let row = sqlx::query_as::<_, PrismSubmissionRow>(&q)
+        .bind(id)
+        .fetch_one(pool)
+        .await?;
+    Ok(row)
+}
+
 /// Append a stage event.
 ///
 /// # Errors

crates/prism-challenge/src/api.rs

@@ -43,6 +43,8 @@ pub struct AppState {
     pub netuid: u16,
     /// Eval backend label (`lium` / `sim`) for the status view.
     pub backend_mode: &'static str,
+    /// Max orchestrator attempts per submission (retry guard).
+    pub retry_max: u32,
 }
 
 /// Router over the full API surface.
@@ -53,6 +55,7 @@ pub fn submission_router(state: Arc<AppState>) -> Router {
         .route("/v1/submissions", get(list_submissions))
         .route("/v1/submissions/{id}", get(get_submission))
         .route("/v1/submissions/{id}/events", get(get_events))
+        .route("/v1/submissions/{id}/retry", post(post_retry))
         .route("/v1/status", get(get_status))
         .route("/v1/jobs", get(get_jobs))
         .route("/v1/recipe", get(get_recipe))
@@ -165,6 +168,37 @@ async fn get_events(State(st): State<Arc<AppState>>, Path(id): Path<String>) ->
     }
 }
 
+/// `POST /v1/submissions/{id}/retry` — requeue a failed row (guard: max attempts).
+async fn post_retry(State(st): State<Arc<AppState>>, Path(id): Path<String>) -> Response {
+    let row = match st.store.get(&id).await {
+        Ok(Some(r)) => r,
+        Ok(None) => return json_err(StatusCode::NOT_FOUND, "unknown_submission", &id),
+        Err(e) => return json_err(StatusCode::INTERNAL_SERVER_ERROR, "store", &e.to_string()),
+    };
+    if row.status != Stage::Failed {
+        return json_err(
+            StatusCode::CONFLICT,
+            "not_failed",
+            &format!("status={}", row.status.as_str()),
+        );
+    }
+    if row.retry_count >= st.retry_max {
+        return json_err(
+            StatusCode::CONFLICT,
+            "retry_exhausted",
+            &format!("retry_count={} max={}", row.retry_count, st.retry_max),
+        );
+    }
+    match st.store.reset_for_retry(&id).await {
+        Ok(_row) => (
+            StatusCode::ACCEPTED,
+            Json(json!({"submission_id": id, "status": "queued"})),
+        )
+            .into_response(),
+        Err(e) => json_err(StatusCode::INTERNAL_SERVER_ERROR, "store", &e.to_string()),
+    }
+}
+
 async fn get_status(State(st): State<Arc<AppState>>) -> Response {
     let queued = st
         .store
@@ -331,7 +365,7 @@ mod tests {
     use axum::body::Body;
     use axum::http::Request;
     use http_body_util::BodyExt;
-    use prism_store::MemoryPrismStore;
+    use prism_store::{MemoryPrismStore, StatePatch};
     use tower::ServiceExt;
 
     fn state() -> Arc<AppState> {
@@ -340,9 +374,87 @@ mod tests {
             epoch: std::sync::atomic::AtomicU64::new(7),
             netuid: 541,
             backend_mode: "sim",
+            retry_max: 2,
         })
     }
 
+    #[tokio::test]
+    async fn retry_requeues_failed_then_guard_blocks() {
+        let st = state();
+        let app = submission_router(Arc::clone(&st));
+        let id = crate::submission::submission_id(&crate::example_valid_request());
+        // Seed via POST.
+        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))
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(v["submission_id"], id);
+        // Force failed.
+        st.store
+            .apply(
+                &id,
+                &StatePatch {
+                    status: Some(Stage::Failed),
+                    ..StatePatch::default()
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        let (s, v) = call(
+            app.clone(),
+            Request::post(format!("/v1/submissions/{id}/retry"))
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(s, StatusCode::ACCEPTED, "{v}");
+        // Re-fail and retry again → retry_max=2 blocks the third.
+        st.store
+            .apply(
+                &id,
+                &StatePatch {
+                    status: Some(Stage::Failed),
+                    ..StatePatch::default()
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        let (s, _v) = call(
+            app.clone(),
+            Request::post(format!("/v1/submissions/{id}/retry"))
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(s, StatusCode::ACCEPTED);
+        st.store
+            .apply(
+                &id,
+                &StatePatch {
+                    status: Some(Stage::Failed),
+                    ..StatePatch::default()
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        let (s, _v) = call(
+            app,
+            Request::post(format!("/v1/submissions/{id}/retry"))
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await;
+        assert_eq!(s, StatusCode::CONFLICT);
+    }
+
     async fn call(app: Router, req: Request<Body>) -> (StatusCode, Value) {
         let res = app.oneshot(req).await.unwrap();
         let status = res.status();

crates/prism-store/src/dbprism.rs

@@ -237,6 +237,23 @@ impl PrismStore for DbPrismStore {
         Ok(row_to_state(row))
     }
 
+    async fn reset_for_retry(&self, id: &str) -> Result<SubmissionState, StoreError> {
+        let row = dbs::reset_prism_submission_for_retry(&self.pool, id)
+            .await
+            .map_err(|e| StoreError::Backend(e.to_string()))?;
+        dbs::insert_prism_stage_event(
+            &self.pool,
+            &dbs::NewPrismStageEvent {
+                submission_id: id,
+                stage: "queued",
+                detail: Some(serde_json::json!({"op": "retry"})),
+            },
+        )
+        .await
+        .map_err(|e| StoreError::Backend(e.to_string()))?;
+        Ok(row_to_state(row))
+    }
+
     async fn list(
         &self,
         status: Option<&str>,

crates/prism-store/src/store.rs

@@ -196,6 +196,11 @@ pub trait PrismStore: Send + Sync + std::fmt::Debug {
         event: Option<&StageEvent>,
     ) -> Result<SubmissionState, StoreError>;
 
+    /// Retry reset: clears exec/score fields and re-queues a failed row.
+    /// Implementations MUST actually null the pod/receipt/score columns
+    /// (SQL) or reset the in-memory row mirror-equivalently.
+    async fn reset_for_retry(&self, id: &str) -> Result<SubmissionState, StoreError>;
+
     /// Newsfeed listing for the API.
     async fn list(
         &self,
@@ -325,6 +330,29 @@ impl PrismStore for MemoryPrismStore {
         Ok(out)
     }
 
+    async fn reset_for_retry(&self, id: &str) -> Result<SubmissionState, StoreError> {
+        let mut rows = self
+            .rows
+            .lock()
+            .map_err(|_| StoreError::Backend("poison".into()))?;
+        let row = rows
+            .iter_mut()
+            .find(|r| r.id == id)
+            .ok_or(StoreError::NotFound)?;
+        row.status = Stage::Queued;
+        row.pod_id = None;
+        row.pod_provider = None;
+        row.receipt = None;
+        row.bpb = None;
+        row.review = None;
+        row.similarity = None;
+        row.final_score = None;
+        row.error_detail = None;
+        row.retry_count = row.retry_count.saturating_add(1);
+        row.updated_at_ms = now_ms();
+        Ok(row.clone())
+    }
+
     async fn list(
         &self,
         status: Option<&str>,