fix(prism): WTA emission, pre-pod screens, 8 concurrent evals

bins/prism-challenge/src/main.rs

@@ -70,7 +70,7 @@ struct Cli {
     #[arg(
         long,
         env = "PRISM_MAX_CONCURRENT_EVALS",
-        default_value_t = 1,
+        default_value_t = 8,
         global = true
     )]
     max_concurrent_evals: u32,

crates/challenge-agentic/src/lib.rs

@@ -18,6 +18,7 @@ mod agent;
 mod llm;
 mod prompts;
 mod sim;
+mod static_checks;
 mod tools;
 mod types;
 
@@ -26,6 +27,7 @@ pub use challenge_ast::{copy_gate, CopyGateHit, GateCorpusEntry};
 pub use llm::{load_api_key_file, DEFAULT_MODEL};
 pub use prompts::{AGENTIC_PROMPT_VERSION, DESIGN_DOMAIN_RULES, PRISM_DOMAIN_RULES};
 pub use sim::{SimAgent, SIM_CHEAT_BPS, SIM_SUSPICIOUS_BPS};
+pub use static_checks::{static_source_cheat, training_has_telemetry_hooks, StaticCheatHit};
 pub use types::{
     AgenticBackend, AgenticError, AgenticVerdict, CheatCode, ContainerReviewRequest, CorpusEntry,
     ReviewRequest, VerdictKind, OPENROUTER_API_BASE,

crates/challenge-agentic/src/sim.rs

@@ -79,6 +79,10 @@ impl AgenticBackend for SimAgent {
             return Ok(v);
         }
 
+        // Source-only screens are also available via [`crate::static_source_cheat`]
+        // for the pre-pod orchestrator path; sim keeps the in-review copies so
+        // metrics-relative checks and corpus AST still share one backend.
+
         if let Some(v) = pages_scrape_cheat_verdict(req)? {
             return Ok(v);
         }
@@ -187,12 +191,7 @@ fn telemetry_hooks_verdict(
         return None;
     }
     let (path, src) = primaries.iter().find(|(p, _)| p.ends_with("training.py"))?;
-    let imports_shim = src.contains("prism_telemetry")
-        || src.contains("ctx[\"telemetry\"]")
-        || src.contains("ctx['telemetry']");
-    let calls_report = src.contains(".report(");
-    let calls_finish = src.contains("finish_evaluation(");
-    if imports_shim && calls_report && calls_finish {
+    if crate::training_has_telemetry_hooks(src) {
         return None;
     }
     Some(AgenticVerdict {

crates/challenge-agentic/src/static_checks.rs

@@ -0,0 +1,94 @@
+//! Cheap source-only cheat screens (no GPU, no private eval assets).
+//!
+//! Run these **before** renting a Lium pod so a bad submission fails fast
+//! instead of burning hours of GPU. Metrics/receipt consistency checks stay
+//! post-eval (they need harness output).
+
+use crate::types::CheatCode;
+
+/// One static source finding.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StaticCheatHit {
+    /// Cheat taxonomy code.
+    pub code: CheatCode,
+    /// Human-readable reason (safe to surface in error_detail).
+    pub rationale: String,
+}
+
+/// Scan miner sources for cheap, deterministic cheat patterns.
+///
+/// Order: hardcoded `METRICS_JSON=` short-circuit first, then missing Prism
+/// telemetry hooks in `training.py`. Returns the first hit.
+#[must_use]
+pub fn static_source_cheat(
+    architecture_py: &str,
+    training_py: &str,
+) -> Option<StaticCheatHit> {
+    for (path, src) in [
+        ("architecture.py", architecture_py),
+        ("training.py", training_py),
+    ] {
+        if src.contains("METRICS_JSON=") {
+            return Some(StaticCheatHit {
+                code: CheatCode::EvalShortCircuit,
+                rationale: format!("static: hardcoded METRICS_JSON in {path}"),
+            });
+        }
+    }
+    if !training_has_telemetry_hooks(training_py) {
+        return Some(StaticCheatHit {
+            code: CheatCode::MissingTelemetryHooks,
+            rationale: "static: training.py missing prism_telemetry report/finish_evaluation hooks"
+                .into(),
+        });
+    }
+    None
+}
+
+/// Prism telemetry-hook contract (recipe ≥ 1.1.0).
+#[must_use]
+pub fn training_has_telemetry_hooks(training_py: &str) -> bool {
+    let imports_shim = training_py.contains("prism_telemetry")
+        || training_py.contains("ctx[\"telemetry\"]")
+        || training_py.contains("ctx['telemetry']");
+    let calls_report = training_py.contains(".report(");
+    let calls_finish = training_py.contains("finish_evaluation(");
+    imports_shim && calls_report && calls_finish
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn metrics_json_short_circuit() {
+        let hit = static_source_cheat(
+            "def build_model(ctx):\n    pass\n",
+            "def train(m, ctx):\n    print('METRICS_JSON={}')\n",
+        )
+        .expect("hit");
+        assert_eq!(hit.code, CheatCode::EvalShortCircuit);
+    }
+
+    #[test]
+    fn missing_hooks() {
+        let hit = static_source_cheat(
+            "def build_model(ctx):\n    pass\n",
+            "def train(m, ctx):\n    return {}\n",
+        )
+        .expect("hit");
+        assert_eq!(hit.code, CheatCode::MissingTelemetryHooks);
+    }
+
+    #[test]
+    fn clean_hooks() {
+        let train = concat!(
+            "import prism_telemetry\n",
+            "def train(m, ctx):\n",
+            "    prism_telemetry.report(loss=1.0, step=1)\n",
+            "    prism_telemetry.finish_evaluation()\n",
+            "    return {}\n",
+        );
+        assert!(static_source_cheat("def build_model(ctx):\n    pass\n", train).is_none());
+    }
+}

crates/prism-challenge/src/orchestrator.rs

@@ -1,23 +1,27 @@
 //! Lium job orchestrator: DB-backed state machine, recovery, epoch emitter.
 //!
-//! Workers claim `queued` rows, rent + run the recipe, run master-side LLM
-//! review + cheap similarity + agentic anti-cheat, and compute the
+//! Workers claim `queued` rows, run cheap source screens (copy gate, static
+//! cheat patterns, AST similarity) **before** renting a Lium pod, then run
+//! the recipe + master-side LLM review + agentic anti-cheat, and compute the
 //! chain-facing score. Leaf emission is decoupled from finalizes: the
 //! epoch-close emitter ([`prism_emit::EpochEmitter`], driven by
 //! [`Orchestrator::run_emitter`]) assigns every newly-finalized row to the
 //! next chain-epoch boundary's D24 set via the emission outbox
 //! (`emitted_epoch` watermark + emit cursor), so independent same-epoch
 //! scorers all land and each scoring run is assigned exactly once. Positive
-//! scores then carry into later epochs' competition sets until superseded.
-//! All state lives in the store, so the API is a pure projection and restarts
-//! sweep orphans.
+//! scores then carry into later epochs' competition sets until superseded;
+//! leaf emission applies WTA so only the single best hotkey gets Prism's
+//! share. All state lives in the store, so the API is a pure projection and
+//! restarts sweep orphans.
 
 use std::sync::Arc;
 use std::time::Duration;
 
 use bundle::NoScoreReasonCode;
 use chain::ChainClient;
-use challenge_agentic::{copy_gate, AgenticBackend, AgenticVerdict, VerdictKind};
+use challenge_agentic::{
+    copy_gate, static_source_cheat, AgenticBackend, AgenticVerdict, VerdictKind,
+};
 use challenge_common::{expected_set_at_chain, PinnedBlockHash};
 use crypto::KEY_LEN;
 use prism_emit::EpochEmitter;
@@ -330,12 +334,34 @@ impl<C: ChainClient + Send> Orchestrator<C> {
         let id = row.id.clone();
         info!(submission_id = %id, miner = %row.miner_hotkey, "prism eval start");
 
-        // Phase 0: pre-LLM copy gate on architecture.py (created_at ordered).
-        // A byte/AST copy of a strictly-earlier architecture is terminal
-        // `rejected` with Score(0) — no pod time, no LLM spend.
+        // Phase 0: pre-pod cheap screens (no GPU, no private eval assets).
+        // Copy gate → static cheat patterns → AST similarity. Fail-fast with
+        // Score(0) so a bad submission never rents a Lium pod (~6h waste).
         if self.copy_gate_step(&row).await {
             return Ok(());
         }
+        if self.static_source_step(&row).await {
+            return Ok(());
+        }
+        let similarity = match self.similarity_step(&id, &row).await {
+            Ok(v) => v,
+            Err(e) => {
+                if self.maybe_auto_retry(&row, "ast_infra", &e).await {
+                    return Ok(());
+                }
+                self.fail_terminal(&row, "ast_infra", &e).await;
+                return Ok(());
+            }
+        };
+        if matches!(
+            similarity.kind,
+            prism_review::SimilarityKind::Copied | prism_review::SimilarityKind::Suspicious
+        ) {
+            let detail = format!("pre-pod similarity: {:?}", similarity.kind);
+            self.reject_pre_pod(&row, Some(similarity), None, detail)
+                .await;
+            return Ok(());
+        }
 
         // Phase 1: provision + recipe exec + terminate (always verified).
         // Lium/infra failures auto-retry (install class); budget exhaustion is
@@ -359,19 +385,8 @@ impl<C: ChainClient + Send> Orchestrator<C> {
             return Ok(());
         };
 
-        // Phase 3: cheap similarity (AST infra → auto-retry, then terminal).
-        let similarity = match self.similarity_step(&id, &row).await {
-            Ok(v) => v,
-            Err(e) => {
-                if self.maybe_auto_retry(&row, "ast_infra", &e).await {
-                    return Ok(());
-                }
-                self.fail_terminal(&row, "ast_infra", &e).await;
-                return Ok(());
-            }
-        };
-
-        // Phase 4: agentic anti-cheat (LLM infra → auto-retry, then terminal).
+        // Phase 3: agentic anti-cheat (needs metrics/receipt; post-pod).
+        // Source-only screens already ran pre-pod; this catches metrics forge.
         let Some(agentic) = self
             .agentic_step(&id, &row, metrics.as_ref(), receipt.as_ref())
             .await
@@ -474,28 +489,73 @@ impl<C: ChainClient + Send> Orchestrator<C> {
             }],
             prompt_version: prism_review::SIMILARITY_PROMPT_VERSION,
         };
+        self.reject_pre_pod(
+            row,
+            Some(similarity),
+            Some(serde_json::json!({
+                "gate": "copy_created_at",
+                "nearest_id": hit.nearest_id,
+                "similarity_bps": hit.similarity_bps,
+                "byte_identical": hit.byte_identical,
+            })),
+            format!(
+                "copy gate: architecture clones {} (bps={})",
+                hit.nearest_id, hit.similarity_bps
+            ),
+        )
+        .await;
+        true
+    }
+
+    /// Static source cheat screen (METRICS_JSON / telemetry hooks). Pre-pod.
+    /// Returns `true` when the row was finalized terminal `rejected`.
+    async fn static_source_step(&self, row: &SubmissionState) -> bool {
+        let Some(hit) = static_source_cheat(&row.architecture_py, &row.training_py) else {
+            return false;
+        };
+        warn!(
+            submission_id = %row.id,
+            code = ?hit.code,
+            rationale = %hit.rationale,
+            "static source cheat rejected (pod skipped)"
+        );
+        self.reject_pre_pod(
+            row,
+            None,
+            Some(serde_json::json!({
+                "gate": "static_source",
+                "cheat_code": format!("{:?}", hit.code),
+                "rationale": hit.rationale,
+            })),
+            hit.rationale.clone(),
+        )
+        .await;
+        true
+    }
+
+    /// Terminal Score(0) reject before any Lium rent. Shared by copy gate,
+    /// static screens, and pre-pod similarity.
+    async fn reject_pre_pod(
+        &self,
+        row: &SubmissionState,
+        similarity: Option<SimilarityVerdict>,
+        detail: Option<serde_json::Value>,
+        error_detail: String,
+    ) {
         let _ = self
             .store
             .apply(
                 &row.id,
                 &StatePatch {
                     status: Some(Stage::Rejected),
                     final_score: Some(FinalScore::Score(0)),
-                    similarity: Some(similarity),
-                    error_detail: Some(format!(
-                        "copy gate: architecture clones {} (bps={})",
-                        hit.nearest_id, hit.similarity_bps
-                    )),
+                    similarity,
+                    error_detail: Some(error_detail),
                     ..StatePatch::default()
                 },
                 Some(&StageEvent {
                     stage: Stage::Rejected,
-                    detail: Some(serde_json::json!({
-                        "gate": "copy_created_at",
-                        "nearest_id": hit.nearest_id,
-                        "similarity_bps": hit.similarity_bps,
-                        "byte_identical": hit.byte_identical,
-                    })),
+                    detail,
                     at_ms: 0,
                 }),
             )
@@ -510,9 +570,6 @@ impl<C: ChainClient + Send> Orchestrator<C> {
                 )
                 .await;
         }
-        // The Score(0) enters the emission outbox; the epoch-close emitter
-        // lands it in the next boundary's D24 set.
-        true
     }
 
     /// Pod phase. Returns `(bpb, receipt)` on full success.

crates/prism-challenge/tests/cheat_arch_copy.rs

@@ -136,11 +136,13 @@ async fn baseline_arch_train_copy_scores_zero() {
 
     assert!(orch.cycle_once().await.unwrap());
     let row = store.get(&id).await.unwrap().expect("row");
-    assert!(
-        matches!(row.status, Stage::Terminated | Stage::Failed),
-        "status={:?}",
+    assert_eq!(
+        row.status,
+        Stage::Rejected,
+        "baseline arch copy must fail pre-pod similarity, got {:?}",
         row.status
     );
+    assert!(row.pod_id.is_none(), "arch copy must not rent a pod");
     assert_eq!(
         row.final_score,
         Some(FinalScore::Score(0)),

crates/prism-challenge/tests/cheat_metrics.rs

@@ -141,11 +141,16 @@ def train(model, ctx):
 
     assert!(orch.cycle_once().await.unwrap());
     let row = store.get(&id).await.unwrap().expect("row");
-    assert!(
-        matches!(row.status, Stage::Terminated | Stage::Failed),
-        "status={:?}",
+    assert_eq!(
+        row.status,
+        Stage::Rejected,
+        "static METRICS_JSON screen must reject pre-pod, got {:?}",
         row.status
     );
+    assert!(
+        row.pod_id.is_none(),
+        "hardcoded METRICS_JSON must not rent a pod"
+    );
     assert_eq!(
         row.final_score,
         Some(FinalScore::Score(0)),

crates/prism-challenge/tests/copy_gate.rs

@@ -239,9 +239,10 @@ async fn ast_copy_with_renames_is_rejected() {
 }
 
 #[tokio::test]
-async fn same_arch_same_timestamp_passes_the_gate() {
+async fn same_arch_same_timestamp_passes_copy_gate_then_similarity() {
     // created_at ties cannot be ordered → the copy gate must NOT reject;
-    // the row proceeds to the normal pipeline (sim: terminates with a score).
+    // pre-pod cheap similarity still catches the identical architecture
+    // (Score(0), no pod) before any Lium rent.
     let store = Arc::new(MemoryPrismStore::new());
     let chain = Arc::new(LockedFake(Mutex::new(fake_chain())));
     let orch = Arc::new(mk_orchestrator(&store, &chain));
@@ -275,10 +276,11 @@ async fn same_arch_same_timestamp_passes_the_gate() {
         .await
         .unwrap()
         .expect("row b");
-    assert_ne!(b.status, Stage::Rejected, "tie must not hard-reject");
-    // The LLM similarity path (SimReviewer, arch-only) still judges the copy.
+    assert_eq!(b.status, Stage::Rejected, "status={:?}", b.status);
+    assert!(b.pod_id.is_none(), "tie copy must not rent a pod");
     assert_eq!(b.final_score, Some(FinalScore::Score(0)));
-    assert!(matches!(b.status, Stage::Terminated | Stage::Failed));
+    let sim = b.similarity.expect("similarity recorded");
+    assert!(matches!(sim.kind, prism_review::SimilarityKind::Copied));
 }
 
 #[tokio::test]