From 2ac5e7df3ac2d32fb6e465980a31845a9168786d Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Thu, 13 Aug 2026 12:11:13 +0000 Subject: [PATCH] feat(prism): resume mid-flight Lium jobs across control-plane restart Detach harness under setsid so GPU work survives master bounce; boot reconcile reattaches when the pod is alive and the BYOK seal is present, failing closed only when unreattachable. --- Cargo.lock | 2 + crates/prism-challenge/src/orchestrator.rs | 428 +++++++----------- crates/prism-challenge/tests/e2e_v3_wiring.rs | 5 +- crates/prism-lium-harness/Cargo.toml | 1 + crates/prism-lium-harness/src/detached.rs | 229 ++++++++++ crates/prism-lium-harness/src/lib.rs | 9 + crates/prism-lium/src/client.rs | 320 ++++++------- crates/prism-lium/src/lib.rs | 20 + crates/prism-lium/src/sim.rs | 19 + crates/prism-lium/src/ssh.rs | 53 --- crates/prism-orphan/Cargo.toml | 1 + crates/prism-orphan/src/lib.rs | 77 +++- crates/prism-orphan/src/reconcile.rs | 76 +++- crates/prism-orphan/src/terminal.rs | 150 ++++++ crates/prism-pipeline/src/lib.rs | 4 +- crates/prism-pipeline/src/pipeline.rs | 6 + docs/COMPLETENESS.md | 2 +- docs/PRISM.md | 23 +- docs/external-miner/prism.md | 23 +- docs/external-miner/troubleshoot.md | 2 +- .../prism-enable-lium-and-emission.md | 7 + 21 files changed, 904 insertions(+), 553 deletions(-) create mode 100644 crates/prism-lium-harness/src/detached.rs create mode 100644 crates/prism-orphan/src/terminal.rs diff --git a/Cargo.lock b/Cargo.lock index 875fedc54..1ae2f4136 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3984,6 +3984,7 @@ dependencies = [ "prism-lium-types", "prism-recipe", "prism-tree", + "serde_json", ] [[package]] @@ -4019,6 +4020,7 @@ dependencies = [ "prism-lium", "prism-lium-payer", "prism-pipeline", + "prism-review", "prism-store", "serde_json", "submission-gating", diff --git a/crates/prism-challenge/src/orchestrator.rs b/crates/prism-challenge/src/orchestrator.rs index 4f79efb4f..b47f77f6e 100644 --- a/crates/prism-challenge/src/orchestrator.rs +++ b/crates/prism-challenge/src/orchestrator.rs @@ -15,15 +15,19 @@ use challenge_agentic::{ use challenge_common::{expected_set_at_chain, GatewayClient, PinnedBlockHash}; use crypto::KEY_LEN; use prism_emit::EpochEmitter; +use prism_lium::HARNESS_ABSENT; use prism_lium::{EvalJobBackend, InstanceSpec, RemoteExecResult}; use prism_lium_payer::PayerBackendFactory; use prism_orphan::{ - spawn_log_watch, ActiveJobs, LogBuffer, DEFAULT_LOG_POLL_SECS, DEFAULT_ORPHAN_GRACE_SECS, + fail_terminal, finish_measure, reject_gating, reject_pre_pod, spawn_log_watch, ActiveJobs, + LogBuffer, DEFAULT_LOG_POLL_SECS, DEFAULT_ORPHAN_GRACE_SECS, +}; +use prism_pipeline::{ + gating_key, measurement_patch, mid_pod_resume, resume_measurement, ScoringMode, }; -use prism_pipeline::{gating_key, measurement_patch, resume_measurement, ScoringMode}; use prism_recipe::{BASELINE_ARCHITECTURE_PY, BASELINE_TRAINING_PY}; use prism_review::{ReviewBackend, SimilarityVerdict, SourceSnippet}; -use submission_gating::{GatingState, GatingStore}; +use submission_gating::GatingStore; use tokio::time::sleep; use tracing::{info, warn}; @@ -36,31 +40,18 @@ use prism_store::{FinalScore, PrismStore, Stage, StageEvent, StatePatch, Submiss /// Worker + emitter settings. #[derive(Debug, Clone)] pub struct OrchestratorConfig { - /// Netuid. pub netuid: u16, - /// Pod price cap. pub max_price_per_hour: f64, - /// Pod lifetime hours cap (train cap + margin). pub max_lifetime_hours: f64, - /// SSH public keys for rent. pub ssh_public_keys: Vec, - /// Optional image digest pin. pub image_digest: Option, - /// Queue polling cadence. pub claim_poll: Duration, - /// Emitter tick cadence (chain-epoch boundary detection lag). pub emit_poll: Duration, - /// Rent attempt budget before `failed`. pub max_attempts: u32, - /// Similarity / agentic corpus size (champions + baseline). pub similarity_corpus_limit: u32, - /// Stuck sweep grace (seconds). Default 10h (> wait-RUNNING + 6h train + SSH margin). pub stuck_grace_secs: u64, - /// Local/e2e only: pause after each published stage. Zero in production. pub stage_delay: Duration, - /// Auto-retry budget for infra-class failures (default 3). pub auto_retry_max: u32, - /// Scoring mode (`PRISM_SCORING_MODE`; default shadow). pub scoring_mode: ScoringMode, pub orphan_grace_secs: u64, } @@ -91,7 +82,6 @@ pub struct Orchestrator { cfg: OrchestratorConfig, store: Arc, backend: Arc, - /// When set, each measure builds a miner-billed [`EvalJobBackend`] from the vault. payer: Option, reviewer: Arc, agentic: Arc, @@ -105,7 +95,6 @@ pub struct Orchestrator { } impl Orchestrator { - /// Construct. #[must_use] #[allow(clippy::too_many_arguments)] pub fn new( @@ -143,30 +132,24 @@ impl Orchestrator { self } - /// Miner-funded Lium: resolve a per-submission client from the vault. #[must_use] pub fn with_payer(mut self, payer: PayerBackendFactory) -> Self { self.payer = Some(payer); self } - /// Attach the submission gating store (terminal states + retry attempts). #[must_use] pub fn with_gating(mut self, gating: Arc) -> Self { self.gating = Some(gating); self } - /// Attach the v3 eval store (composite runs + Zone B ingest). Default - /// `None` skips the composite path entirely — legacy behavior is - /// bit-identical (no battery parse, no eval rows, `composite: None`). #[must_use] pub fn with_eval_store(mut self, eval_store: Option>) -> Self { self.eval_store = eval_store; self } - /// Attach the top-model GitHub publisher (absent = publish step no-ops). #[must_use] pub fn with_topmodel( mut self, @@ -176,7 +159,6 @@ impl Orchestrator { self } - /// Backend that bills `submission_id` (miner vault or operator/sim). fn backend_for(&self, submission_id: &str) -> Result, String> { match &self.payer { Some(p) => p.resolve(submission_id, Arc::clone(&self.backend)), @@ -184,19 +166,16 @@ impl Orchestrator { } } - /// Config getter (API views). #[must_use] pub const fn cfg(&self) -> &OrchestratorConfig { &self.cfg } - /// Emitter accessor (tests / diagnostics). #[must_use] pub const fn emitter(&self) -> &EpochEmitter { &self.emitter } - /// Claim loop (spawn one per concurrency permit). pub async fn run_worker(self: Arc) where C: Sync, @@ -213,7 +192,6 @@ impl Orchestrator { } } - /// Stuck-row sweeper loop (skips rows held by a live worker). pub async fn run_sweeper(self: Arc) where C: Sync, @@ -229,7 +207,6 @@ impl Orchestrator { } } - /// One orphan reconcile pass (`boot` ⇒ grace 0). pub async fn reconcile_orphans( &self, boot: bool, @@ -283,15 +260,19 @@ impl Orchestrator { if self.maybe_auto_retry(&row, "install", &msg).await { continue; } - self.fail_terminal(&row, "install", &msg).await; + fail_terminal( + self.store.as_ref(), + self.gating.as_ref(), + &row, + "install", + &msg, + ) + .await; } Ok(()) } /// One claim→finalize cycle; `Ok(true)` when one row was worked. - /// - /// # Errors - /// Claim fault only; per-row business errors become `failed` rows. pub async fn cycle_once(&self) -> Result { let row = self.store.claim_next().await.map_err(|e| e.to_string())?; let Some(row) = row else { return Ok(false) }; @@ -299,8 +280,6 @@ impl Orchestrator { Ok(true) } - /// Requeue on infra failures while auto-retry budget lasts (`false` → - /// terminal). Lium **429** requeues without burning `retry_count` / gating. 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 { @@ -348,61 +327,6 @@ impl Orchestrator { true } - /// Gating `rejected` for cheat-class terminals (composite key for - /// training-only rows, `prism` otherwise). - async fn reject_gating(&self, row: &SubmissionState) { - if let Some(g) = &self.gating { - let _ = g - .set_terminal( - &gating_key(row.arch_id.as_deref()), - &row.miner_hotkey, - GatingState::Rejected, - None, - ) - .await; - } - } - - /// Terminal failure: `failed` row + gating `blocked`. The - /// `NoScore(ChallengeInternal)` enters the emission outbox and lands in - /// the next epoch-boundary leaf set (`run_emitter`). - async fn fail_terminal(&self, row: &SubmissionState, class: &str, msg: &str) { - let _ = self - .store - .apply( - &row.id, - &StatePatch { - status: Some(Stage::Failed), - error_detail: Some(msg.to_owned()), - final_score: Some(FinalScore::NoScore( - NoScoreReasonCode::ChallengeInternal as u8, - )), - ..StatePatch::default() - }, - Some(&StageEvent { - stage: Stage::Failed, - detail: Some(serde_json::json!({"class": class, "error": msg})), - at_ms: 0, - }), - ) - .await; - if let Some(g) = &self.gating { - let _ = g - .set_terminal( - &gating_key(row.arch_id.as_deref()), - &row.miner_hotkey, - GatingState::Blocked, - Some(class), - ) - .await; - } - } - - /// Returns `true` (row finalized terminal — caller stops) on a - /// harness-flagged parameter-cap breach: the row goes terminal - /// `rejected` Score(0) and the miner is gating-rejected. The breach is - /// miner-attributable and machine-verified at build time, so there is - /// no measured score and no review/similarity/agentic spend. async fn cap_terminal(&self, row: &SubmissionState, m: Option<&RemoteExecResult>) -> bool { let Some(m) = m.filter(|m| cap_flag(m)) else { return false; @@ -422,13 +346,10 @@ impl Orchestrator { at_ms: 0, }; let _ = self.store.apply(&row.id, &patch, Some(&event)).await; - self.reject_gating(row).await; + reject_gating(self.gating.as_ref(), row).await; true } - /// Terminal stage event for a finalized row: agentic audit blob plus the - /// scoring mode + version the final score was computed under (v2 shadow, - /// v3 composite). fn terminal_event(&self, status: Stage, agentic: &AgenticVerdict) -> StageEvent { StageEvent { stage: status, @@ -441,16 +362,14 @@ impl Orchestrator { } } - /// Process one submission end-to-end. pub async fn run_row(&self, row: SubmissionState) -> Result<(), String> { let id = row.id.clone(); let _active = self.active.enter(&id); info!(submission_id = %id, miner = %row.miner_hotkey, "prism eval start"); - let Some(pre) = self.pre_pod_screens(&id, &row).await else { + let Some((similarity, review)) = self.screens_or_resume(&id, &row).await else { return Ok(()); }; - let (similarity, review, _pre_agentic) = pre; let (measured, fresh) = match resume_measurement(&row) { Some(mr) => (Ok(mr), false), @@ -463,7 +382,14 @@ impl Orchestrator { if self.maybe_auto_retry(&row, "install", &msg).await { return Ok(()); } - self.fail_terminal(&row, "install", &msg).await; + fail_terminal( + self.store.as_ref(), + self.gating.as_ref(), + &row, + "install", + &msg, + ) + .await; return Ok(()); } }; @@ -489,7 +415,7 @@ impl Orchestrator { agentic.verdict, VerdictKind::Cheat | VerdictKind::Suspicious ) { - self.reject_gating(&row).await; + reject_gating(self.gating.as_ref(), &row).await; } let blob = metrics @@ -546,10 +472,6 @@ impl Orchestrator { Ok(()) } - /// Pre-pod screens: copy → static → similarity → LLM quality → agentic. - /// Returns gates when the row may proceed to Lium rent; `None` when - /// already finalized (rejected / failed / retrying). OpenRouter / agentic - /// infra errors fail closed here — they must never rent a pod. async fn pre_pod_screens( &self, id: &str, @@ -571,13 +493,17 @@ impl Orchestrator { if self.maybe_auto_retry(row, "ast_infra", &e).await { return None; } - self.fail_terminal(row, "ast_infra", &e).await; + fail_terminal( + self.store.as_ref(), + self.gating.as_ref(), + row, + "ast_infra", + &e, + ) + .await; return None; } }; - // Hard-reject LLM `Copied`, and high-confidence `Suspicious` - // (score ≥ 0.9 with non-trope evidence). Below-threshold / trope-only - // Suspicious is not a wipe — parser coercion + combine_final agree. if prism_review::cheap_similarity_hard_zeros( similarity.kind, similarity.score, @@ -587,8 +513,15 @@ impl Orchestrator { "pre-pod similarity: {:?} score={:.2}", similarity.kind, similarity.score ); - self.reject_pre_pod(row, Some(similarity), None, detail) - .await; + reject_pre_pod( + self.store.as_ref(), + self.gating.as_ref(), + row, + Some(similarity), + None, + detail, + ) + .await; return None; } let review = self.review_step(id, row).await?; @@ -598,7 +531,9 @@ impl Orchestrator { VerdictKind::Cheat | VerdictKind::Suspicious ) { let detail = format!("pre-pod agentic: {:?}", agentic.verdict); - self.reject_pre_pod( + reject_pre_pod( + self.store.as_ref(), + self.gating.as_ref(), row, Some(similarity), Some(serde_json::json!({ @@ -613,14 +548,6 @@ impl Orchestrator { Some((similarity, review, agentic)) } - /// Pre-LLM copy gate on `architecture.py`. Returns `true` when the row was - /// finalized terminal `rejected` (caller must stop processing). - /// - /// The corpus is **champions** (Score>0 current top + historical ex-tops), - /// ordered by store `created_at`; the published baseline is exempt by id - /// prefix inside [`copy_gate`]. Ties / unknown timestamps fall through to - /// the LLM similarity review. Training-only rows (`arch_id` set) skip the - /// gate entirely: their architecture is registry-identical by design. async fn copy_gate_step(&self, row: &SubmissionState) -> bool { if row.arch_id.is_some() { return false; @@ -652,7 +579,9 @@ impl Orchestrator { }], prompt_version: prism_review::SIMILARITY_PROMPT_VERSION, }; - self.reject_pre_pod( + reject_pre_pod( + self.store.as_ref(), + self.gating.as_ref(), row, Some(similarity), Some(serde_json::json!({ @@ -670,9 +599,6 @@ impl Orchestrator { true } - /// Static source cheat screen (`METRICS_JSON` / non-causal mix / telemetry - /// hooks; recipe 2.0: delta telemetry/network/eval-leak). Pre-pod. - /// Returns `true` when the row was finalized terminal `rejected`. async fn static_source_step(&self, row: &SubmissionState) -> bool { let patch = row .tree_blob @@ -689,7 +615,9 @@ impl Orchestrator { rationale = %hit.rationale, "static source cheat rejected (pod skipped)" ); - self.reject_pre_pod( + reject_pre_pod( + self.store.as_ref(), + self.gating.as_ref(), row, None, Some(serde_json::json!({ @@ -703,87 +631,83 @@ impl Orchestrator { true } - /// Terminal Score(0) reject before any Lium rent. Shared by copy gate, - /// static screens, and pre-pod similarity. - async fn reject_pre_pod( + async fn screens_or_resume( &self, + id: &str, row: &SubmissionState, - similarity: Option, - detail: Option, - error_detail: String, - ) { - let _ = self - .store - .apply( - &row.id, - &StatePatch { - status: Some(Stage::Rejected), - final_score: Some(FinalScore::Score(0)), - similarity, - error_detail: Some(error_detail), - ..StatePatch::default() - }, - Some(&StageEvent { - stage: Stage::Rejected, - detail, - at_ms: 0, - }), - ) - .await; - if let Some(g) = &self.gating { - let _ = g - .set_terminal( - &gating_key(row.arch_id.as_deref()), - &row.miner_hotkey, - GatingState::Rejected, - None, + ) -> Option<(SimilarityVerdict, prism_review::ReviewVerdict)> { + if mid_pod_resume(row) { + if let (Some(s), Some(r)) = (row.similarity.clone(), row.review.clone()) { + return Some((s, r)); + } + } + let (s, r, _) = self.pre_pod_screens(id, row).await?; + if !mid_pod_resume(row) { + let _ = self + .store + .apply( + id, + &StatePatch { + review: Some(r.clone()), + similarity: Some(s.clone()), + ..StatePatch::default() + }, + Some(&StageEvent { + stage: Stage::Provisioning, + detail: Some(serde_json::json!({"checkpoint": "pre_measure"})), + at_ms: 0, + }), ) .await; } + Some((s, r)) } - /// Pod phase. Returns `(bpb, receipt)` on full success. async fn measure( &self, id: &str, row: &SubmissionState, ) -> Result<(prism_lium::RemoteExecResult, prism_lium::EvalReceipt), String> { self.to_stage(id, Stage::Provisioning).await?; - let backend = self.backend_for(id)?; - - let spec = InstanceSpec { - name: format!("prism-{}", &id[..12]), - max_lifetime_hours: self.cfg.max_lifetime_hours, - max_price_per_hour: self.cfg.max_price_per_hour, - gpu_count: 1, - image_digest: self.cfg.image_digest.clone(), - ssh_public_keys: self.cfg.ssh_public_keys.clone(), - ssh_key_name: Some("prism-mission-worker".into()), - preferred_offer_id: None, - template_id: None, - template_name: None, // default recipe template (prism-recipe-v2 w/ sshd) + let resume = mid_pod_resume(row); + let (pod_id, provider) = if let Some(pid) = row.pod_id.clone() { + ( + pid, + row.pod_provider.clone().unwrap_or_else(|| "lium".into()), + ) + } else { + let spec = InstanceSpec { + name: format!("prism-{}", &id[..12.min(id.len())]), + max_lifetime_hours: self.cfg.max_lifetime_hours, + max_price_per_hour: self.cfg.max_price_per_hour, + gpu_count: 1, + image_digest: self.cfg.image_digest.clone(), + ssh_public_keys: self.cfg.ssh_public_keys.clone(), + ssh_key_name: Some("prism-mission-worker".into()), + preferred_offer_id: None, + template_id: None, + template_name: None, + }; + let inst = backend + .provision(&spec) + .await + .map_err(|e| format!("provision: {e}"))?; + (inst.id, inst.provider) }; self.to_stage(id, Stage::Running).await?; - - let inst = backend - .provision(&spec) - .await - .map_err(|e| format!("provision: {e}"))?; - let pod_id = inst.id.clone(); let _ = self .store .apply( id, &StatePatch { pod_id: Some(pod_id.clone()), - pod_provider: Some(inst.provider.clone()), + pod_provider: Some(provider.clone()), ..StatePatch::default() }, None, ) .await; - let stop_tx = spawn_log_watch( Arc::clone(&self.logs), Arc::clone(&self.store), @@ -793,67 +717,34 @@ impl Orchestrator { pod_id.clone(), DEFAULT_LOG_POLL_SECS, ); - - #[rustfmt::skip] - let metrics = backend - .exec_eval(&pod_id, &row.architecture_py, &row.training_py, row.tree_blob.as_deref()) - .await; - let _ = stop_tx.send(true); - - if let Ok(ref m) = metrics { - let dest = prism_lium::artifact_dir_for(id); - match backend - .harvest_artifacts(&pod_id, &dest, id.as_bytes(), m.n_params) - .await - { - Ok(path) => { - info!( - submission_id = %id, - path = %path.display(), - n_params = ?m.n_params, - "checkpoint secure-received" - ); - } - Err(e) => { - warn!(submission_id = %id, error = %e, "checkpoint secure receive failed"); + let (arch, train, tree) = ( + row.architecture_py.as_str(), + row.training_py.as_str(), + row.tree_blob.as_deref(), + ); + let metrics = if resume { + match backend.resume_eval(&pod_id).await { + Ok(m) => Ok(m), + Err(e) if e.to_string().contains(HARNESS_ABSENT) => { + backend.exec_eval(&pod_id, arch, train, tree).await } + Err(e) => Err(e), } - } - - // Always terminate + verify (billing guard, receipt gate). - if let Err(e) = backend.terminate(&pod_id).await { - warn!(error = %e, %pod_id, "terminate failed"); - } - let mut termination_verified = backend.verify_terminated(&pod_id).await.unwrap_or(false); - if !termination_verified { - tokio::time::sleep(Duration::from_secs(5)).await; - termination_verified = backend.verify_terminated(&pod_id).await.unwrap_or(false); - } - if let Some(p) = &self.payer { - p.vault.remove(id); - } - - let receipt = prism_lium::EvalReceipt { - provider: inst.provider.clone(), - pod_id: pod_id.clone(), - image_digest: self.cfg.image_digest.clone().unwrap_or_default(), - submission_hash: prism_lium::EvalReceipt::hash_submission( - &row.architecture_py, - &row.training_py, - ), - metrics_hash: metrics.as_ref().ok().map_or_else( - || "none".into(), - |m| { - prism_lium::EvalReceipt::hash_metrics_bytes( - &serde_json::to_vec(m).unwrap_or_default(), - ) - }, - ), - termination_verified, + } else { + backend.exec_eval(&pod_id, arch, train, tree).await }; - - let metrics = metrics.map_err(|e| format!("exec: {e}"))?; - Ok((metrics, receipt)) + let _ = stop_tx.send(true); + finish_measure( + &backend, + self.payer.as_ref(), + id, + row, + &pod_id, + provider, + self.cfg.image_digest.clone().unwrap_or_default(), + metrics, + ) + .await } async fn review_step( @@ -873,7 +764,14 @@ impl Orchestrator { if self.maybe_auto_retry(row, "llm_infra", &msg).await { return None; } - self.fail_terminal(row, "llm_infra", &msg).await; + fail_terminal( + self.store.as_ref(), + self.gating.as_ref(), + row, + "llm_infra", + &msg, + ) + .await; None } } @@ -885,9 +783,6 @@ impl Orchestrator { row: &SubmissionState, ) -> Result { self.to_stage(id, Stage::Similarity).await?; - // Training-only rows train a registry architecture: similarity is - // exempt by definition (the arch copy judgment happened when the - // owner's architecture submission was reviewed). if let Some(a) = &row.arch_id { return Ok(SimilarityVerdict { kind: prism_review::SimilarityKind::Original, @@ -906,7 +801,6 @@ impl Orchestrator { .map_err(|e| format!("similarity: {e}")) } - /// Agentic anti-cheat on sources + metrics/receipt. Fail-closed on error. async fn agentic_step( &self, id: &str, @@ -952,7 +846,14 @@ impl Orchestrator { if self.maybe_auto_retry(row, "llm_infra", &msg).await { return None; } - self.fail_terminal(row, "llm_infra", &msg).await; + fail_terminal( + self.store.as_ref(), + self.gating.as_ref(), + row, + "llm_infra", + &msg, + ) + .await; None } } @@ -991,21 +892,16 @@ impl Orchestrator { architecture_py: BASELINE_ARCHITECTURE_PY.into(), training_py: BASELINE_TRAINING_PY.into(), }]; - for r in recent { - if r.id == candidate.id || same_miner(candidate, &r) { - continue; - } - let label = if r.id.len() >= 8 { - format!("subm:{}", &r.id[..8]) - } else { - format!("subm:{}", r.id) - }; - v.push(SourceSnippet { - label, - architecture_py: r.architecture_py.clone(), - training_py: r.training_py.clone(), - }); - } + v.extend( + recent + .into_iter() + .filter(|r| r.id != candidate.id && !same_miner(candidate, r)) + .map(|r| SourceSnippet { + label: format!("subm:{}", &r.id[..r.id.len().min(8)]), + architecture_py: r.architecture_py, + training_py: r.training_py, + }), + ); v } @@ -1032,8 +928,6 @@ impl Orchestrator { Ok(()) } - /// Emitter loop: one D24-complete leaf set per chain epoch (epoch-close - /// batching, exactly-once outbox — see `prism-emit` docs). pub async fn run_emitter(self: Arc) where C: Sync, @@ -1046,21 +940,10 @@ impl Orchestrator { } } - /// One emitter tick: read the live chain epoch + expected set, then let - /// the outbox recover/emit. `Ok(None)` = this epoch already emitted. - /// - /// # Errors - /// Chain / store / sign / submit failures (retried next tick). pub async fn emitter_tick(&self) -> Result, String> { let state = chain::gather_schedule_state(self.chain.as_ref(), self.cfg.netuid) .map_err(|e| format!("schedule: {e}"))?; - // Label with the *current* chain epoch, not the pre-coinbase +1: the - // expected set below is pinned at `last_epoch_block` (the current - // epoch's start block), so a pre-run +1 label attaches the *previous* - // boundary's metagraph to the new epoch number. The other >0-bps - // challenge pins the same way, and D24 requires both same-label sets - // to cover the seal block's metagraph exactly — the +1 skew made - // every boundary churn a permanent 409 (`IncompleteParticipantSet`). + // Pin expected set at `last_epoch_block` (current epoch), not +1. let epoch = state.subnet_epoch_index; let block_hash = self .chain @@ -1089,9 +972,6 @@ impl Orchestrator { } } -/// Harness terminal payload flag: a miner-attributable parameter-cap breach -/// (the harness refused the model at build and emitted a minimal -/// METRICS_JSON instead of measuring; recipe ≥1.3.0). fn cap_flag(m: &RemoteExecResult) -> bool { m.extra .get("cap_exceeded") diff --git a/crates/prism-challenge/tests/e2e_v3_wiring.rs b/crates/prism-challenge/tests/e2e_v3_wiring.rs index 904c4bc4a..a392591c7 100644 --- a/crates/prism-challenge/tests/e2e_v3_wiring.rs +++ b/crates/prism-challenge/tests/e2e_v3_wiring.rs @@ -215,10 +215,9 @@ async fn cap_exceeded_is_terminal_score_zero() { let row = store.get(&id).await.unwrap().expect("row"); assert_eq!(row.status, Stage::Rejected, "cap breach rejects terminally"); assert_eq!(row.final_score, Some(FinalScore::Score(0))); - // Never a measured score and no LLM/similarity spend. + // Cap is measured at build; no bpb and no post-measure agentic re-spend. + // Pre-pod screens may already be checkpointed for resume durability. assert!(row.bpb.is_none(), "no measured bpb on a cap breach"); - assert!(row.review.is_none(), "no review on a cap breach"); - assert!(row.similarity.is_none(), "no similarity on a cap breach"); assert_eq!(row.retry_count, 0, "miner-attributable: no auto-retry"); let detail = row.error_detail.unwrap_or_default(); assert!( diff --git a/crates/prism-lium-harness/Cargo.toml b/crates/prism-lium-harness/Cargo.toml index 8c44ddbc4..f06245e69 100644 --- a/crates/prism-lium-harness/Cargo.toml +++ b/crates/prism-lium-harness/Cargo.toml @@ -14,6 +14,7 @@ prism-automodel = { path = "../prism-automodel" } prism-lium-types = { path = "../prism-lium-types" } prism-recipe = { path = "../prism-recipe" } prism-tree = { path = "../prism-tree" } +serde_json = "1" [lints] workspace = true diff --git a/crates/prism-lium-harness/src/detached.rs b/crates/prism-lium-harness/src/detached.rs new file mode 100644 index 000000000..130612d1d --- /dev/null +++ b/crates/prism-lium-harness/src/detached.rs @@ -0,0 +1,229 @@ +//! Detached pod harness: survive control-plane SSH drops and allow reattach. + +use prism_lium_types::{LiumError, RemoteExecResult}; + +/// Parent-emitted train-done marker (exact raw line; miner stdout is prefixed). +pub const TRAIN_DONE_MARKER: &str = "PHASE_TRAIN_DONE"; +/// Error token when the pod has no harness to reattach to. +pub const HARNESS_ABSENT: &str = "harness_absent"; + +/// Classify a harvested harness log for the poll / resume loop. +#[derive(Debug, Clone, PartialEq)] +#[allow(clippy::large_enum_variant)] +pub enum HarnessProgress { + /// Still training / evaluating. + Running, + /// Train finished; master must stage eval assets (private tier). + NeedsAssets, + /// Terminal success / cap breach payload available. + Done(Box), + /// Process ended without a parseable terminal marker. + Failed(String), +} + +/// Build the remote command that starts `main.py` detached (idempotent). +#[must_use] +pub fn detach_launch_cmd(env_exports: &str, timeout_secs: u64) -> String { + // `setsid` + stdin closed: SSH session death must not kill the harness. + format!( + r#"set -e +cd /tmp/prism_eval +if [ -f harness.pid ] && kill -0 "$(cat harness.pid)" 2>/dev/null; then + echo DETACH_ALREADY + exit 0 +fi +if grep -qE '^(EVAL_OK|CAP_EXCEEDED)$' harness.log 2>/dev/null; then + echo DETACH_DONE + exit 0 +fi +rm -f harness.exit +: > harness.log +{env_exports}nohup setsid bash -c 'timeout --kill-after=60 {timeout_secs} python3 main.py; echo $? > harness.exit' \ + >> harness.log 2>&1 < /dev/null & +echo $! > harness.pid +echo DETACH_STARTED +"# + ) +} + +/// Probe command: pid alive / terminal markers / assets ready. +pub const HARNESS_PROBE_CMD: &str = r#"set +e +cd /tmp/prism_eval 2>/dev/null || { echo 'STATE=absent'; exit 0; } +alive=0 +if [ -f harness.pid ] && kill -0 "$(cat harness.pid)" 2>/dev/null; then alive=1; fi +done=0 +grep -qE '^(EVAL_OK|CAP_EXCEEDED)$' harness.log 2>/dev/null && done=1 +train=0 +grep -qxF 'PHASE_TRAIN_DONE' harness.log 2>/dev/null && train=1 +ready=0 +[ -f eval-assets/.ready ] && ready=1 +log=0 +[ -f harness.log ] && log=1 +echo "STATE=ok alive=$alive done=$done train=$train ready=$ready log=$log" +"#; + +/// Parsed probe from [`HARNESS_PROBE_CMD`]. +#[derive(Debug, Clone, Copy, Default)] +#[allow(clippy::struct_excessive_bools)] +pub struct HarnessProbe { + pub present: bool, + pub pid_alive: bool, + pub terminal: bool, + pub train_done: bool, + pub assets_ready: bool, + pub has_log: bool, +} + +impl HarnessProbe { + /// True when a resume poll can continue (or harvest a finished log). + #[must_use] + pub const fn attachable(self) -> bool { + self.present && (self.pid_alive || self.terminal || self.has_log || self.train_done) + } +} + +/// Parse [`HARNESS_PROBE_CMD`] stdout. +#[must_use] +pub fn parse_harness_probe(stdout: &str) -> HarnessProbe { + let line = stdout + .lines() + .rev() + .find(|l| l.starts_with("STATE=")) + .unwrap_or(""); + if line.contains("STATE=absent") || line.is_empty() { + return HarnessProbe::default(); + } + let flag = |k: &str| line.contains(&format!("{k}=1")); + HarnessProbe { + present: true, + pid_alive: flag("alive"), + terminal: flag("done"), + train_done: flag("train"), + assets_ready: flag("ready"), + has_log: flag("log"), + } +} + +/// Progress from log text + whether assets still need staging. +#[must_use] +pub fn classify_log(log: &str, assets_configured: bool, assets_ready: bool) -> HarnessProgress { + if let Ok(m) = parse_metrics_output(log, 0, log) { + return HarnessProgress::Done(Box::new(m)); + } + let train_done = log.lines().any(|l| l.trim_end() == TRAIN_DONE_MARKER); + if assets_configured && train_done && !assets_ready { + return HarnessProgress::NeedsAssets; + } + if log + .lines() + .any(|l| l.trim_end() == "EVAL_OK" || l.trim_end() == "CAP_EXCEEDED") + { + return HarnessProgress::Failed(truncate(log, 4000)); + } + HarnessProgress::Running +} + +/// Parse harness output: `EVAL_OK` / `CAP_EXCEEDED` + `METRICS_JSON=`. +pub fn parse_metrics_output( + stdout: &str, + returncode: i32, + stderr_tail: &str, +) -> Result { + if !stdout.contains("EVAL_OK") && !stdout.contains("CAP_EXCEEDED") { + return Err(LiumError::Exec(format!( + "harness failed (code {}): {}", + returncode, + truncate(stderr_tail, 4000) + ))); + } + let line = stdout + .lines() + .find(|l| l.starts_with("METRICS_JSON=")) + .ok_or_else(|| LiumError::Exec("harness EVAL_OK without METRICS_JSON".into()))?; + let v: RemoteExecResult = serde_json::from_str(&line["METRICS_JSON=".len()..]) + .map_err(|e| LiumError::Exec(format!("metrics json: {e}")))?; + if v.extra + .get("cap_exceeded") + .and_then(serde_json::Value::as_bool) + == Some(true) + { + return Ok(v); + } + if !v.bpb.is_finite() || v.bpb <= 0.0 { + return Err(LiumError::Exec(format!( + "harness bpb not finite: {}", + v.bpb + ))); + } + Ok(v) +} + +fn truncate(s: &str, max: usize) -> String { + if s.len() <= max { + return s.to_owned(); + } + s[s.len().saturating_sub(max)..].to_owned() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn probe_parses_flags() { + let p = parse_harness_probe("noise\nSTATE=ok alive=1 done=0 train=1 ready=0 log=1\n"); + assert!(p.present && p.pid_alive && p.train_done && p.has_log && !p.terminal); + assert!(p.attachable()); + assert!(!parse_harness_probe("STATE=absent\n").present); + } + + #[test] + fn classify_needs_assets_and_done() { + let log = "PHASE_TRAIN_DONE\n"; + assert_eq!(classify_log(log, true, false), HarnessProgress::NeedsAssets); + let good = "noise\nMETRICS_JSON={\"bpb\":2.5,\"tokens_seen\":1,\"wall_clock_seconds\":1.0,\"gpu_type\":null,\"notes\":\"n\",\"eval_tier\":\"public\"}\nEVAL_OK\n"; + assert!(matches!( + classify_log(good, false, false), + HarnessProgress::Done(_) + )); + } + + #[test] + fn parse_metrics_output_gates_eval_ok_and_bpb() { + let good = "noise\nMETRICS_JSON={\"bpb\":2.5,\"tokens_seen\":1,\"wall_clock_seconds\":1.0,\"gpu_type\":null,\"notes\":\"n\",\"eval_tier\":\"public\"}\nEVAL_OK\n"; + let v = parse_metrics_output(good, 0, "").unwrap(); + assert!((v.bpb - 2.5).abs() < 1e-9); + assert!(parse_metrics_output("no ok line", 1, "boom") + .unwrap_err() + .to_string() + .contains("harness failed")); + let bad = "METRICS_JSON={\"bpb\":-1.0,\"tokens_seen\":1,\"wall_clock_seconds\":1.0,\"gpu_type\":null,\"notes\":\"n\"}\nEVAL_OK\n"; + assert!(parse_metrics_output(bad, 0, "") + .unwrap_err() + .to_string() + .contains("bpb")); + } + + #[test] + fn parse_metrics_output_accepts_cap_exceeded_terminal() { + let out = "noise\nMETRICS_JSON={\"bpb\":0.0,\"tokens_seen\":0,\"wall_clock_seconds\":3.0,\"gpu_type\":null,\"notes\":\"parameter cap exceeded\",\"n_params\":999000000,\"cap_exceeded\":true}\nCAP_EXCEEDED\n"; + let v = parse_metrics_output(out, 3, "").unwrap(); + assert_eq!( + v.extra.get("cap_exceeded").and_then(|x| x.as_bool()), + Some(true) + ); + let missing_flag = "METRICS_JSON={\"bpb\":0.0,\"tokens_seen\":0,\"wall_clock_seconds\":3.0,\"gpu_type\":null,\"notes\":\"n\"}\nCAP_EXCEEDED\n"; + assert!(parse_metrics_output(missing_flag, 3, "") + .unwrap_err() + .to_string() + .contains("not finite")); + } + + #[test] + fn detach_launch_uses_setsid() { + let cmd = detach_launch_cmd("export FOO='bar'\n", 100); + assert!(cmd.contains("setsid")); + assert!(cmd.contains("harness.pid")); + assert!(cmd.contains("DETACH_ALREADY")); + } +} diff --git a/crates/prism-lium-harness/src/lib.rs b/crates/prism-lium-harness/src/lib.rs index 0318eb030..5bd0802d4 100644 --- a/crates/prism-lium-harness/src/lib.rs +++ b/crates/prism-lium-harness/src/lib.rs @@ -1,11 +1,20 @@ //! Pod harness packaging and allowlisted environment helpers for Lium runs. #![forbid(unsafe_code)] +#![allow(clippy::missing_errors_doc)] +#![allow(clippy::redundant_closure_for_method_calls)] + +mod detached; use std::path::PathBuf; use prism_lium_types::LiumError; +pub use detached::{ + classify_log, detach_launch_cmd, parse_harness_probe, parse_metrics_output, HarnessProbe, + HarnessProgress, HARNESS_ABSENT, HARNESS_PROBE_CMD, TRAIN_DONE_MARKER, +}; + /// Pod-side staging directory for eval assets. pub const EVAL_ASSETS_POD_DIR: &str = "/tmp/prism_eval/eval-assets"; diff --git a/crates/prism-lium/src/client.rs b/crates/prism-lium/src/client.rs index f726e4bdd..df6546b92 100644 --- a/crates/prism-lium/src/client.rs +++ b/crates/prism-lium/src/client.rs @@ -18,25 +18,20 @@ fn rent_pool() -> &'static RentPool { use crate::ssh::{ parse_ssh_target, resolve_private_key, ssh_exec, ssh_exec_allow_fail, ssh_exec_stdin, - ssh_exec_streaming, truncate_tail, SshTarget, + truncate_tail, SshTarget, }; use crate::{EvalJobBackend, HARNESS_LOG_RETAIN_BYTES, LIUM_API_BASE_URL, MIN_LIFETIME_HOURS}; use prism_lium_harness::{ - eval_assets_dir, harness_env_pairs, harness_upload_tar, random_seed_hex, EVAL_ASSETS_POD_DIR, - HARNESS_BOOTSTRAP, HARNESS_EXTRACT_CMD, + classify_log, detach_launch_cmd, eval_assets_dir, harness_env_pairs, harness_upload_tar, + parse_harness_probe, parse_metrics_output, random_seed_hex, HarnessProgress, + EVAL_ASSETS_POD_DIR, HARNESS_ABSENT, HARNESS_BOOTSTRAP, HARNESS_EXTRACT_CMD, HARNESS_PROBE_CMD, + TRAIN_DONE_MARKER, }; use prism_lium_types::{CostGuardrailError, LiumError}; use prism_lium_types::{ GpuPreference, Instance, InstanceSpec, LiumSshConfig, Offer, RemoteExecResult, }; -/// Parent-emitted marker line: the train-phase process group is dead and -/// the parent gate is (about to be) waiting for eval assets. Matched as an -/// exact raw line — miner stdout is relayed with an `[harness] v3| ` prefix -/// by the runner, so a miner cannot forge it. The harness side emits this -/// via a bare `print` at the gate (see module docs / E5 report). -const DEFAULT_TRAIN_DONE_MARKER: &str = "PHASE_TRAIN_DONE"; - // cu13.0.2-dinD: sshd from image init (empty startup). Other marketplace // images lack a stable sshd under Lium's metachar-free startup rules. const RECIPES_TEMPLATE_IMAGE: &str = "daturaai/pytorch"; @@ -406,29 +401,11 @@ impl LiumClient { }) } - /// Live recipe eval: wait RUNNING → SSH nvidia-smi → stage the harness - /// package ([`prism_recipe::HARNESS_FILES`]) + miner sources as a tar - /// over ssh stdin → run `main.py` → parse `METRICS_JSON` (v1 and v2 - /// payloads both accepted). - /// - /// The harness trains the miner's code on the pinned fineweb-edu shard and - /// scores the frozen val cut; no metric comes from hashing sources. + /// Live recipe eval: wait RUNNING → stage harness → **detach** `main.py` + /// → poll `harness.log` (survives control-plane restart / SSH drop). /// - /// **Two-phase (private tier)**: when the master-side - /// `PRISM_EVAL_ASSETS_DIR` env points at the operator's private - /// eval-assets directory, the run switches to the streaming flow in - /// [`Self::exec_eval_two_phase`]: harness output is consumed line-by-line - /// and on the parent-emitted train-done marker - /// ([`DEFAULT_TRAIN_DONE_MARKER`], exact raw line — unforgeable by the - /// miner) the client (a) streams the assets tar into - /// [`EVAL_ASSETS_POD_DIR`] over a second ssh channel, (b) writes - /// `SECRET_SEED` (fresh 128-bit hex, file-only — env delivery would leak - /// it to the train-phase child via inherited env), and (c) touches - /// `.ready`. The harness env names the pod assets dir - /// (`PRISM_EVAL_ASSETS_DIR`) so its post-train gate holds for `.ready` - /// instead of falling back. A run that completes without staging while - /// assets were configured is refused (fail-closed against a silent - /// public-tier downgrade). + /// Private-tier assets: when `PRISM_EVAL_ASSETS_DIR` is set, the poller + /// stages the pack on [`TRAIN_DONE_MARKER`] (exact raw line). async fn exec_eval_live( &self, instance_id: &str, @@ -444,8 +421,6 @@ impl LiumClient { let train_cap_secs = (self.ssh.train_hours_cap * 3600.0) as u64; let timeout_secs = train_cap_secs.saturating_add(3600); - // Stage the harness tar over ssh stdin before the run: argv-embedded - // base64 payloads exceeded MAX_ARG_STRLEN at local spawn (BUG-5). let tar = harness_upload_tar(architecture_py, training_py, tree_blob)?; let (att, rty) = (self.ssh.ssh_attempts, self.ssh.ssh_retry_secs); ssh_exec_stdin(&target, &key, HARNESS_EXTRACT_CMD, &tar, att, rty, 300).await?; @@ -454,107 +429,150 @@ impl LiumClient { #[allow(clippy::format_collect)] let env: String = pairs .iter() - .map(|(k, v)| format!("{k}='{v}' \\\n")) + .map(|(k, v)| format!("export {k}='{v}'\n")) .collect(); - let remote = - format!("{HARNESS_BOOTSTRAP}cd /tmp/prism_eval\n{env}timeout --kill-after=60 {timeout_secs} python3 main.py"); + let launch = format!( + "{HARNESS_BOOTSTRAP}{}", + detach_launch_cmd(&env, timeout_secs) + ); + let out = ssh_exec_allow_fail(&target, &key, &launch, att, rty, 120).await?; + if !out.stdout.contains("DETACH_STARTED") + && !out.stdout.contains("DETACH_ALREADY") + && !out.stdout.contains("DETACH_DONE") + { + return Err(LiumError::Exec(format!( + "detach launch failed: {}", + truncate_tail(&out.stderr, 400) + ))); + } + self.poll_detached_eval( + instance_id, + &target, + &key, + assets.as_deref(), + train_cap_secs.saturating_add(3900), + ) + .await + } - self.exec_eval_stream( + /// Reattach: probe harness, refuse if absent, else poll to terminal. + async fn resume_eval_live(&self, instance_id: &str) -> Result { + let _running = self.wait_until_running(instance_id).await?; + let target = self.resolve_ssh_target(instance_id).await?; + let key = resolve_private_key(self.ssh.private_key_path.as_deref())?; + let (att, rty) = (self.ssh.ssh_attempts, self.ssh.ssh_retry_secs); + let probe_out = + ssh_exec_allow_fail(&target, &key, HARNESS_PROBE_CMD, att.max(1), rty, 60).await?; + let probe = parse_harness_probe(&probe_out.stdout); + if !probe.attachable() { + return Err(LiumError::Exec(format!( + "{HARNESS_ABSENT}: no harness on pod {instance_id}" + ))); + } + let assets = eval_assets_dir(); + let train_cap_secs = (self.ssh.train_hours_cap * 3600.0) as u64; + self.poll_detached_eval( instance_id, &target, &key, - &remote, assets.as_deref(), train_cap_secs.saturating_add(3900), ) .await } - /// Stream the harness run line-by-line (stderr merged remote-side with - /// `2>&1`). With `assets` (private tier), the parent-emitted train-done - /// marker triggers [`Self::stage_eval_assets`] over a second ssh channel - /// while the train-phase process group is dead. Fail-closed: a run that - /// completes without staging while assets were configured, or that then - /// reports a non-private tier, is rejected. - async fn exec_eval_stream( + /// Poll `harness.log` until metrics, staging assets on train-done. + async fn poll_detached_eval( &self, instance_id: &str, target: &SshTarget, key: &Path, - remote: &str, assets: Option<&Path>, timeout_secs: u64, ) -> Result { + let start = Instant::now(); + let period = Duration::from_secs(20); + let mut staged = false; let marker = std::env::var("PRISM_TRAIN_DONE_MARKER") - .unwrap_or_else(|_| DEFAULT_TRAIN_DONE_MARKER.to_owned()); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>(); - // Miner stdout is relayed with an `[harness] v3| ` prefix, so only an - // exact raw line (parent-emitted) trips staging — never miner output. - let mut on_line = move |line: &str| { - if assets.is_some() && line.trim_end() == marker { - let _ = tx.send(()); + .unwrap_or_else(|_| TRAIN_DONE_MARKER.to_owned()); + loop { + if start.elapsed() >= Duration::from_secs(timeout_secs) { + let h = self + .harvest_logs_inner(instance_id) + .await + .unwrap_or_default(); + return Err(LiumError::Exec(format!( + "harness poll timed out after {timeout_secs}s; harvested: {}", + truncate_tail(&h, 4000) + ))); } - }; - let remote = format!("{remote} 2>&1"); - let run = ssh_exec_streaming(target, key, &remote, timeout_secs, &mut on_line); - tokio::pin!(run); - let mut staged = false; - let out = loop { - tokio::select! { - _ = rx.recv(), if !staged => { + let rty = self.ssh.ssh_retry_secs; + let probe_out = ssh_exec_allow_fail(target, key, HARNESS_PROBE_CMD, 1, rty, 45).await?; + let probe = parse_harness_probe(&probe_out.stdout); + if probe.assets_ready { + staged = true; + } + let log = self + .harvest_logs_inner(instance_id) + .await + .unwrap_or_default(); + // Prefer exact configured marker when present in full log harvest. + let train_line = log.lines().any(|l| l.trim_end() == marker); + match classify_log(&log, assets.is_some(), staged || probe.assets_ready) { + HarnessProgress::Done(res) => { + if assets.is_some() && !staged && !probe.assets_ready { + return Err(LiumError::Exec( + "eval assets configured but harness never emitted the train-done marker — refusing public-tier result under PRISM_EVAL_ASSETS_DIR".into(), + )); + } + let tier_ok = matches!(res.eval_tier.as_deref(), Some("public" | "private")); + if (staged || probe.assets_ready) && !tier_ok { + return Err(LiumError::Exec(format!( + "eval assets staged but harness reported eval_tier={:?} (want \"public\"|\"private\")", + res.eval_tier + ))); + } + return Ok(*res); + } + HarnessProgress::NeedsAssets | HarnessProgress::Running + if assets.is_some() + && !staged + && !probe.assets_ready + && (train_line || probe.train_done) => + { if let Some(a) = assets { self.stage_eval_assets(target, key, a).await?; staged = true; - info!("eval assets staged post-train (seed written, .ready touched)"); + info!("eval assets staged post-train (detached poll)"); } } - r = &mut run => match r { - Ok(o) => break o, - Err(e) => { - // Session timed out / dropped — a second ssh pulls the - // on-pod log so the fatal tail survives the dead pipe. - let h = self.harvest_logs_inner(instance_id).await.unwrap_or_default(); - return Err(LiumError::Exec(if h.trim().is_empty() { - format!("harness transport: {e}") - } else { - format!("harness transport: {e}; harvested: {h}") - })); + HarnessProgress::Failed(msg) => { + return Err(LiumError::Exec(msg)); + } + HarnessProgress::Running | HarnessProgress::NeedsAssets => { + if !probe.pid_alive && probe.has_log && !probe.terminal { + // Process died without terminal markers. + let _ = parse_metrics_output(&log, 1, &log)?; + return Err(LiumError::Exec(format!( + "harness exited without EVAL_OK; harvested: {}", + truncate_tail(&log, 4000) + ))); } - }, + } } - }; - let tail: String = out - .stdout - .lines() - .rev() - .take(40) - .collect::>() - .join("\n"); - // Nothing came back over the channel: fall back to the on-pod log so the - // failure detail is not an empty string. - let tail = if tail.trim().is_empty() { - self.harvest_logs_inner(instance_id) - .await - .unwrap_or_default() - } else { - tail - }; - let res = parse_metrics_output(&out.stdout, out.returncode, &tail)?; - if assets.is_some() && !staged { - return Err(LiumError::Exec( - "eval assets configured but harness never emitted the train-done marker — refusing public-tier result under PRISM_EVAL_ASSETS_DIR".into(), - )); + sleep(period).await; } - // Staged packs default to `public` (HF held-out). `private` remains - // valid for optional contamination / secret-seed mirrors. - let tier_ok = matches!(res.eval_tier.as_deref(), Some("public" | "private")); - if staged && !tier_ok { - return Err(LiumError::Exec(format!( - "eval assets staged but harness reported eval_tier={:?} (want \"public\"|\"private\")", - res.eval_tier - ))); + } + + async fn instance_running_live(&self, instance_id: &str) -> Result { + match self.status(instance_id).await { + Ok(inst) => { + let st = inst.status.to_ascii_uppercase(); + Ok(RUNNING_STATUSES.iter().any(|s| st == *s)) + } + Err(LiumError::Api(_)) => Ok(false), + Err(e) => Err(e), } - Ok(res) } /// Stage the operator assets into the pod workdir post-train in a single @@ -655,43 +673,6 @@ impl LiumClient { } } -/// Parse the harness run output: `EVAL_OK` gate, `METRICS_JSON=` line, bpb -/// sanity. `stderr_tail` feeds the failure message (legacy path: ssh -/// stderr; detached path: the run-log tail). -fn parse_metrics_output( - stdout: &str, - returncode: i32, - stderr_tail: &str, -) -> Result { - // Miner-attributable param-cap breach: the harness emits the exact - // `CAP_EXCEEDED` line + a minimal METRICS_JSON (bpb sentinel 0) instead - // of measuring. Not EVAL_OK — but a valid terminal parse. - if !stdout.contains("EVAL_OK") && !stdout.contains("CAP_EXCEEDED") { - return Err(LiumError::Exec(format!( - "harness failed (code {}): {}", - returncode, - truncate(stderr_tail, 4000) - ))); - } - let line = stdout - .lines() - .find(|l| l.starts_with("METRICS_JSON=")) - .ok_or_else(|| LiumError::Exec("harness EVAL_OK without METRICS_JSON".into()))?; - let v: RemoteExecResult = serde_json::from_str(&line["METRICS_JSON=".len()..]) - .map_err(|e| LiumError::Exec(format!("metrics json: {e}")))?; - // Terminal cap payload: skip the bpb gate (sentinel 0 by design). - if v.extra.get("cap_exceeded").and_then(Value::as_bool) == Some(true) { - return Ok(v); - } - if !v.bpb.is_finite() || v.bpb <= 0.0 { - return Err(LiumError::Exec(format!( - "harness bpb not finite: {}", - v.bpb - ))); - } - Ok(v) -} - /// First string value found at any of `keys` (top-level object lookups). fn get_str<'a>(v: &'a Value, keys: &[&str]) -> Option<&'a str> { keys.iter().find_map(|k| v.get(k).and_then(|x| x.as_str())) @@ -972,6 +953,14 @@ impl EvalJobBackend for LiumClient { .await } + async fn instance_running(&self, instance_id: &str) -> Result { + self.instance_running_live(instance_id).await + } + + async fn resume_eval(&self, instance_id: &str) -> Result { + self.resume_eval_live(instance_id).await + } + async fn harvest_logs(&self, instance_id: &str) -> Result { self.harvest_logs_inner(instance_id).await } @@ -1340,7 +1329,7 @@ mod tests { fn train_done_marker_match_is_exact_line_only() { // Miner stdout is relayed with the `[harness] v3| ` prefix, so a // forged marker inside miner output must NOT match. - let m = DEFAULT_TRAIN_DONE_MARKER; + let m = TRAIN_DONE_MARKER; assert!("[harness] v3| PHASE_TRAIN_DONE".trim_end() != m); assert!(" PHASE_TRAIN_DONE".trim_end() != m); assert_eq!("PHASE_TRAIN_DONE".trim_end(), m); @@ -1362,42 +1351,6 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } - #[test] - fn parse_metrics_output_gates_eval_ok_and_bpb() { - let good = "noise\nMETRICS_JSON={\"bpb\":2.5,\"tokens_seen\":1,\"wall_clock_seconds\":1.0,\"gpu_type\":null,\"notes\":\"n\",\"eval_tier\":\"public\"}\nEVAL_OK\n"; - let v = parse_metrics_output(good, 0, "").unwrap(); - assert_eq!(v.eval_tier.as_deref(), Some("public")); - assert!(parse_metrics_output("no ok line", 1, "boom") - .unwrap_err() - .to_string() - .contains("harness failed")); - let bad = "METRICS_JSON={\"bpb\":-1.0,\"tokens_seen\":1,\"wall_clock_seconds\":1.0,\"gpu_type\":null,\"notes\":\"n\"}\nEVAL_OK\n"; - assert!(parse_metrics_output(bad, 0, "") - .unwrap_err() - .to_string() - .contains("not finite")); - } - - #[test] - fn parse_metrics_output_accepts_cap_exceeded_terminal() { - // No EVAL_OK; the exact CAP_EXCEEDED line + JSON flag skip the bpb - // gate (the cap payload carries a bpb sentinel 0 by design). - let out = "noise\nMETRICS_JSON={\"bpb\":0.0,\"tokens_seen\":0,\"wall_clock_seconds\":3.0,\"gpu_type\":null,\"notes\":\"parameter cap exceeded\",\"n_params\":999000000,\"cap_exceeded\":true}\nCAP_EXCEEDED\n"; - let v = parse_metrics_output(out, 3, "").unwrap(); - assert_eq!( - v.extra.get("cap_exceeded").and_then(Value::as_bool), - Some(true) - ); - assert_eq!(v.n_params, Some(999_000_000)); - // The bare line without the JSON flag is not a terminal parse: it - // falls through to the bpb gate and is rejected (forge/incomplete). - let missing_flag = "METRICS_JSON={\"bpb\":0.0,\"tokens_seen\":0,\"wall_clock_seconds\":3.0,\"gpu_type\":null,\"notes\":\"n\"}\nCAP_EXCEEDED\n"; - assert!(parse_metrics_output(missing_flag, 3, "") - .unwrap_err() - .to_string() - .contains("not finite")); - } - #[test] fn debug_redacts_key() { let c = LiumClient::with_base_url("super-secret-key-xyz", "http://example").unwrap(); @@ -1537,16 +1490,15 @@ mod tests { assert!(HARNESS_EXTRACT_CMD.len() < 1024); let mut env = String::new(); for (k, v) in harness_env_pairs(6.0, "NVIDIA GeForce RTX 5090", true) { - let _ = writeln!(env, "{k}='{v}' \\"); + let _ = writeln!(env, "export {k}='{v}'"); } - let remote = format!( - "{HARNESS_BOOTSTRAP}cd /tmp/prism_eval\n{env}timeout --kill-after=60 25200 python3 main.py" - ); + let remote = format!("{HARNESS_BOOTSTRAP}{}", detach_launch_cmd(&env, 25200)); assert!( remote.len() < 8 * 1024, "run argv is {} bytes", remote.len() ); + assert!(remote.contains("setsid")); assert!(!remote.contains("base64"), "no payload embedding in argv"); // The stdin payload scales with miner size; argv never does. let big = harness_upload_tar(&"a".repeat(200_000), &"t".repeat(200_000), None).unwrap(); diff --git a/crates/prism-lium/src/lib.rs b/crates/prism-lium/src/lib.rs index 5d5be8813..d0dbac783 100644 --- a/crates/prism-lium/src/lib.rs +++ b/crates/prism-lium/src/lib.rs @@ -42,6 +42,10 @@ pub use prism_artifacts::{ artifact_dir_for, artifact_root, checkpoint_path_for, ensure_artifact_root, write_sim_checkpoint, MAX_CHECKPOINT_BYTES, POD_WORKDIR, }; +pub use prism_lium_harness::{ + classify_log, parse_harness_probe, parse_metrics_output, HarnessProbe, HarnessProgress, + HARNESS_ABSENT, TRAIN_DONE_MARKER, +}; pub use sim::SimLiumBackend; pub use ssh::{parse_ssh_target, resolve_private_key, truncate_tail, SshTarget}; // The data contract lives in `prism-lium-types` (per-crate LOC cap); it is @@ -93,6 +97,22 @@ pub trait EvalJobBackend: Send + Sync { "artifact harvest not supported on this backend".into(), )) } + + /// True when the provider still lists the instance as rentable/running. + async fn instance_running(&self, _instance_id: &str) -> Result { + Ok(false) + } + + /// Reattach to a detached harness already on `instance_id` (no re-upload). + /// + /// Returns [`LiumError::Exec`] containing [`HARNESS_ABSENT`] when the pod + /// is up but no harness log/pid exists — caller may start a fresh + /// [`Self::exec_eval`] on the same pod. + async fn resume_eval(&self, instance_id: &str) -> Result { + Err(LiumError::Exec(format!( + "{HARNESS_ABSENT}: resume unsupported on this backend ({instance_id})" + ))) + } } /// Tail bytes retained for harness stderr / harvested logs in error_detail. diff --git a/crates/prism-lium/src/sim.rs b/crates/prism-lium/src/sim.rs index f1bb2d584..b93020a13 100644 --- a/crates/prism-lium/src/sim.rs +++ b/crates/prism-lium/src/sim.rs @@ -175,6 +175,25 @@ impl EvalJobBackend for SimLiumBackend { ) -> Result { prism_artifacts::write_sim_checkpoint(dest_dir, seed) } + + async fn instance_running(&self, instance_id: &str) -> Result { + let map = self + .pods + .lock() + .map_err(|_| LiumError::Api("sim lock poisoned".into()))?; + Ok(map.get(instance_id).copied().unwrap_or(false)) + } + + async fn resume_eval(&self, instance_id: &str) -> Result { + if !self.instance_running(instance_id).await? { + return Err(LiumError::Exec(format!( + "{}: sim pod gone", + crate::HARNESS_ABSENT + ))); + } + // Sim has no detached log; treat resume as a fresh deterministic eval. + self.exec_eval(instance_id, "resume", "resume", None).await + } } #[cfg(test)] diff --git a/crates/prism-lium/src/ssh.rs b/crates/prism-lium/src/ssh.rs index b437cbec7..fa2c7b7fb 100644 --- a/crates/prism-lium/src/ssh.rs +++ b/crates/prism-lium/src/ssh.rs @@ -331,59 +331,6 @@ pub async fn ssh_exec_stdin( .await } -/// Run a remote command invoking `on_line` for each stdout line as it -/// arrives (merge stderr into stdout remote-side with `2>&1`). Single -/// attempt; the full output is returned and a non-zero exit is NOT an -/// error — the caller interprets the payload (harness run pattern). -pub async fn ssh_exec_streaming( - target: &SshTarget, - private_key: &Path, - remote_cmd: &str, - timeout_secs: u64, - on_line: &mut (dyn FnMut(&str) + Send), -) -> Result { - use tokio::io::{AsyncBufReadExt as _, AsyncReadExt as _}; - let run = async { - let mut child = ssh_command(target, private_key, remote_cmd) - .spawn() - .map_err(|e| format!("ssh spawn: {e}"))?; - let stdout = child.stdout.take().ok_or("ssh stdout pipe")?; - // Drain local ssh-client stderr (banners / keepalive noise). Leaving - // the pipe unread can fill the kernel buffer and stall the session - // mid-harness even when the remote merges 2>&1 into stdout. - let mut stderr = child.stderr.take().ok_or("ssh stderr pipe")?; - let stderr_task = tokio::spawn(async move { - let mut buf = Vec::new(); - let _ = stderr.read_to_end(&mut buf).await; - buf - }); - let mut lines = tokio::io::BufReader::new(stdout).lines(); - let mut acc = String::new(); - while let Some(line) = lines - .next_line() - .await - .map_err(|e| format!("ssh read: {e}"))? - { - on_line(&line); - acc.push_str(&line); - acc.push('\n'); - } - let status = child.wait().await.map_err(|e| format!("ssh wait: {e}"))?; - let stderr_buf = stderr_task.await.unwrap_or_default(); - let stderr = String::from_utf8_lossy(&stderr_buf).into_owned(); - Ok((acc, stderr, status.code().unwrap_or(-1))) - }; - match tokio::time::timeout(Duration::from_secs(timeout_secs), run).await { - Ok(Ok((stdout, stderr, returncode))) => Ok(SshExecOutput { - returncode, - stdout, - stderr, - }), - Ok(Err(e)) => Err(LiumError::Exec(e)), - Err(_) => Err(LiumError::Exec("ssh timed out".into())), - } -} - /// Successful SSH run output. #[derive(Debug, Clone)] pub struct SshExecOutput { diff --git a/crates/prism-orphan/Cargo.toml b/crates/prism-orphan/Cargo.toml index 780248916..cc9827bc1 100644 --- a/crates/prism-orphan/Cargo.toml +++ b/crates/prism-orphan/Cargo.toml @@ -14,6 +14,7 @@ bundle = { path = "../bundle" } prism-lium = { path = "../prism-lium" } prism-lium-payer = { path = "../prism-lium-payer" } prism-pipeline = { path = "../prism-pipeline" } +prism-review = { path = "../prism-review" } prism-store = { path = "../prism-store" } serde_json = "1" submission-gating = { path = "../submission-gating" } diff --git a/crates/prism-orphan/src/lib.rs b/crates/prism-orphan/src/lib.rs index 711f03017..4e53f7ec3 100644 --- a/crates/prism-orphan/src/lib.rs +++ b/crates/prism-orphan/src/lib.rs @@ -1,9 +1,10 @@ -//! Mid-flight orphan detection for Prism after control-plane restarts. +//! Mid-flight orphan / resume for Prism after control-plane restarts. //! //! Workers register in [`ActiveJobs`] for the whole `run_row` hold. On boot -//! (and periodically) [`reconcile_once`] fails provisioning/running rows whose -//! worker is gone, best-effort terminates the pod when the BYOK vault still -//! has a key, and requeues post-measure review stages so they can resume. +//! (and periodically) [`reconcile_once`] **resume-first**: mid-pod rows whose +//! Lium instance is still alive are requeued with `pod_id` kept (no terminate). +//! Only unreattachable rows fail-closed (`control_plane_restart` / +//! `harness_detached`). Post-measure review stages requeue as before. #![forbid(unsafe_code)] #![allow(clippy::missing_errors_doc)] @@ -13,6 +14,7 @@ mod active; mod logs; mod reconcile; +mod terminal; pub use active::{ActiveGuard, ActiveJobs}; pub use logs::{logs_json, LogBuffer, LogLine}; @@ -20,6 +22,7 @@ pub use reconcile::{ midflight_rows, reconcile_once, ReconcileReport, CLASS_ORPHAN, REASON_CONTROL_PLANE_RESTART, REASON_HARNESS_DETACHED, }; +pub use terminal::{fail_terminal, finish_measure, reject_gating, reject_pre_pod}; /// Default detach grace for periodic reconcile (seconds). Boot uses zero. pub const DEFAULT_ORPHAN_GRACE_SECS: u64 = 90; @@ -110,7 +113,24 @@ mod tests { }; use prism_store::{FinalScore, MemoryPrismStore, PrismStore, Stage, SubmissionState}; - struct NoopBackend; + struct NoopBackend { + running: bool, + terminates: std::sync::Mutex, + } + impl NoopBackend { + fn dead() -> Self { + Self { + running: false, + terminates: std::sync::Mutex::new(0), + } + } + fn alive() -> Self { + Self { + running: true, + terminates: std::sync::Mutex::new(0), + } + } + } #[async_trait] impl EvalJobBackend for NoopBackend { async fn list_offers( @@ -123,6 +143,7 @@ mod tests { Err(LiumError::Exec("noop".into())) } async fn terminate(&self, _instance_id: &str) -> Result<(), LiumError> { + *self.terminates.lock().unwrap() += 1; Ok(()) } async fn verify_terminated(&self, _instance_id: &str) -> Result { @@ -137,6 +158,9 @@ mod tests { ) -> Result { Err(LiumError::Exec("noop".into())) } + async fn instance_running(&self, _instance_id: &str) -> Result { + Ok(self.running) + } } fn row(id: &str, status: Stage, pod: Option<&str>) -> SubmissionState { @@ -168,7 +192,7 @@ mod tests { } #[tokio::test] - async fn boot_reconcile_marks_running_orphan_failed() { + async fn boot_reconcile_marks_dead_pod_orphan_failed() { let store: Arc = Arc::new(MemoryPrismStore::default()); let r = row("deadbeefdeadbeef", Stage::Running, Some("pod-1")); store.insert_queued(&r).await.unwrap(); @@ -187,11 +211,12 @@ mod tests { .unwrap(); let active = Arc::new(ActiveJobs::new()); - let be: Arc = Arc::new(NoopBackend); + let be: Arc = Arc::new(NoopBackend::dead()); let report = reconcile_once(store.as_ref(), &active, None, be, 0, true, None) .await .unwrap(); assert_eq!(report.failed, 1); + assert_eq!(report.resumed, 0); let got = store.get(&r.id).await.unwrap().unwrap(); assert_eq!(got.status, Stage::Failed); let err = got.error_detail.unwrap(); @@ -203,6 +228,39 @@ mod tests { assert!(matches!(got.final_score, Some(FinalScore::NoScore(_)))); } + #[tokio::test] + async fn boot_reconcile_resumes_alive_pod_without_terminate() { + let store: Arc = Arc::new(MemoryPrismStore::default()); + let r = row("cafebabecafebabe", Stage::Running, Some("pod-live")); + store.insert_queued(&r).await.unwrap(); + store + .apply( + &r.id, + &prism_store::StatePatch { + status: Some(Stage::Running), + pod_id: Some("pod-live".into()), + pod_provider: Some("lium".into()), + ..Default::default() + }, + None, + ) + .await + .unwrap(); + let active = Arc::new(ActiveJobs::new()); + let be = Arc::new(NoopBackend::alive()); + let be_trait: Arc = be.clone(); + let report = reconcile_once(store.as_ref(), &active, None, be_trait, 0, true, None) + .await + .unwrap(); + assert_eq!(report.resumed, 1); + assert_eq!(report.failed, 0); + assert_eq!(report.terminate_attempts, 0); + assert_eq!(*be.terminates.lock().unwrap(), 0); + let got = store.get(&r.id).await.unwrap().unwrap(); + assert_eq!(got.status, Stage::Queued); + assert_eq!(got.pod_id.as_deref(), Some("pod-live")); + } + #[tokio::test] async fn active_worker_not_orphaned() { let store: Arc = Arc::new(MemoryPrismStore::default()); @@ -222,11 +280,12 @@ mod tests { .unwrap(); let active = Arc::new(ActiveJobs::new()); let _g = active.enter(&r.id); - let be: Arc = Arc::new(NoopBackend); + let be: Arc = Arc::new(NoopBackend::alive()); let report = reconcile_once(store.as_ref(), &active, None, be, 90, false, None) .await .unwrap(); assert_eq!(report.failed, 0); + assert_eq!(report.resumed, 0); assert_eq!( store.get(&r.id).await.unwrap().unwrap().status, Stage::Running @@ -283,7 +342,7 @@ mod tests { .await .unwrap(); let active = Arc::new(ActiveJobs::new()); - let be: Arc = Arc::new(NoopBackend); + let be: Arc = Arc::new(NoopBackend::dead()); let report = reconcile_once(store.as_ref(), &active, None, be, 0, true, None) .await .unwrap(); diff --git a/crates/prism-orphan/src/reconcile.rs b/crates/prism-orphan/src/reconcile.rs index 43704c776..eb28a24c2 100644 --- a/crates/prism-orphan/src/reconcile.rs +++ b/crates/prism-orphan/src/reconcile.rs @@ -1,11 +1,11 @@ -//! Fail / requeue mid-flight rows whose worker is not alive in this process. +//! Resume-first mid-flight reconcile; fail only when unreattachable. use std::sync::Arc; use bundle::NoScoreReasonCode; use prism_lium::EvalJobBackend; use prism_lium_payer::PayerBackendFactory; -use prism_pipeline::resume_measurement; +use prism_pipeline::{mid_pod_resume, resume_measurement}; use prism_store::{FinalScore, PrismStore, Stage, StageEvent, StatePatch, SubmissionState}; use submission_gating::{GatingState, GatingStore}; use tracing::{info, warn}; @@ -24,6 +24,8 @@ pub struct ReconcileReport { pub failed: u32, /// Post-measure review rows requeued. pub requeued: u32, + /// Mid-pod rows requeued for resume (pod left running). + pub resumed: u32, /// Pods where terminate was attempted. pub terminate_attempts: u32, } @@ -49,8 +51,9 @@ pub async fn midflight_rows(store: &dyn PrismStore) -> Result { + report.resumed = report.resumed.saturating_add(1); + continue; + } + Ok(false) => {} + Err(e) => { + warn!(submission_id = %row.id, error = %e, "pod resume probe failed"); + } + } + } + // Unreattachable mid-pod or pre-measure without 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() { @@ -126,6 +141,55 @@ pub async fn reconcile_once( Ok(report) } +/// Probe pod + vault; on success requeue to `queued` keeping `pod_id`. +async fn try_resume( + store: &dyn PrismStore, + payer: Option<&PayerBackendFactory>, + operator: Arc, + row: &SubmissionState, + reason: &str, +) -> Result { + let Some(pod) = row.pod_id.as_deref() else { + return Ok(false); + }; + if let Some(p) = payer { + if p.vault.get(&row.id).is_none() && !p.allow_operator_fallback { + return Ok(false); + } + } + let be = match payer { + Some(p) => p.resolve(&row.id, Arc::clone(&operator))?, + None => operator, + }; + let alive = be.instance_running(pod).await.map_err(|e| e.to_string())?; + if !alive { + return Ok(false); + } + // Keep pod_id / screens; workers claim Queued and call resume_eval. + store + .apply( + &row.id, + &StatePatch { + status: Some(Stage::Queued), + error_detail: None, + ..StatePatch::default() + }, + Some(&StageEvent { + stage: Stage::Queued, + detail: Some(serde_json::json!({ + "pod_resume": true, + "reason": reason, + "pod_id": pod, + })), + at_ms: 0, + }), + ) + .await + .map_err(|e| e.to_string())?; + info!(submission_id = %row.id, %pod, reason, "mid-pod resume queued (pod left running)"); + Ok(true) +} + fn should_skip( row: &SubmissionState, active: &crate::ActiveJobs, @@ -137,7 +201,7 @@ fn should_skip( const WEDGE_SECS: u64 = 30 * 60; return active.age_secs(&row.id).is_none_or(|age| age < WEDGE_SECS); } - // Process restart: every inactive mid-flight row is an orphan immediately. + // Process restart: every inactive mid-flight row is considered immediately. if boot { return false; } diff --git a/crates/prism-orphan/src/terminal.rs b/crates/prism-orphan/src/terminal.rs new file mode 100644 index 000000000..6653b1bfa --- /dev/null +++ b/crates/prism-orphan/src/terminal.rs @@ -0,0 +1,150 @@ +//! Shared terminal / measure teardown helpers for orchestrator + reconcile. + +use std::sync::Arc; +use std::time::Duration; + +use bundle::NoScoreReasonCode; +use prism_lium::{EvalJobBackend, EvalReceipt, RemoteExecResult}; +use prism_lium_payer::PayerBackendFactory; +use prism_pipeline::gating_key; +use prism_review::SimilarityVerdict; +use prism_store::{FinalScore, PrismStore, Stage, StageEvent, StatePatch, SubmissionState}; +use submission_gating::{GatingState, GatingStore}; +use tracing::{info, warn}; + +/// Terminal `failed` + gating `blocked`. +pub async fn fail_terminal( + store: &dyn PrismStore, + gating: Option<&Arc>, + row: &SubmissionState, + class: &str, + msg: &str, +) { + let _ = store + .apply( + &row.id, + &StatePatch { + status: Some(Stage::Failed), + error_detail: Some(msg.to_owned()), + final_score: Some(FinalScore::NoScore( + NoScoreReasonCode::ChallengeInternal as u8, + )), + ..StatePatch::default() + }, + Some(&StageEvent { + stage: Stage::Failed, + detail: Some(serde_json::json!({"class": class, "error": msg})), + at_ms: 0, + }), + ) + .await; + if let Some(g) = gating { + let _ = g + .set_terminal( + &gating_key(row.arch_id.as_deref()), + &row.miner_hotkey, + GatingState::Blocked, + Some(class), + ) + .await; + } +} + +/// Gating `rejected` for cheat-class terminals. +pub async fn reject_gating(gating: Option<&Arc>, row: &SubmissionState) { + if let Some(g) = gating { + let _ = g + .set_terminal( + &gating_key(row.arch_id.as_deref()), + &row.miner_hotkey, + GatingState::Rejected, + None, + ) + .await; + } +} + +/// Terminal Score(0) reject before any Lium rent. +pub async fn reject_pre_pod( + store: &dyn PrismStore, + gating: Option<&Arc>, + row: &SubmissionState, + similarity: Option, + detail: Option, + error_detail: String, +) { + let _ = store + .apply( + &row.id, + &StatePatch { + status: Some(Stage::Rejected), + final_score: Some(FinalScore::Score(0)), + similarity, + error_detail: Some(error_detail), + ..StatePatch::default() + }, + Some(&StageEvent { + stage: Stage::Rejected, + detail, + at_ms: 0, + }), + ) + .await; + reject_gating(gating, row).await; +} + +/// Harvest artifacts, terminate pod, drop vault key, build receipt. +pub async fn finish_measure( + backend: &Arc, + payer: Option<&PayerBackendFactory>, + id: &str, + row: &SubmissionState, + pod_id: &str, + provider: String, + image_digest: String, + metrics: Result, +) -> Result<(RemoteExecResult, EvalReceipt), String> { + if let Ok(ref m) = metrics { + let dest = prism_lium::artifact_dir_for(id); + match backend + .harvest_artifacts(pod_id, &dest, id.as_bytes(), m.n_params) + .await + { + Ok(path) => { + info!( + submission_id = %id, + path = %path.display(), + n_params = ?m.n_params, + "checkpoint secure-received" + ); + } + Err(e) => { + warn!(submission_id = %id, error = %e, "checkpoint secure receive failed"); + } + } + } + if let Err(e) = backend.terminate(pod_id).await { + warn!(error = %e, %pod_id, "terminate failed"); + } + let mut termination_verified = backend.verify_terminated(pod_id).await.unwrap_or(false); + if !termination_verified { + tokio::time::sleep(Duration::from_secs(5)).await; + termination_verified = backend.verify_terminated(pod_id).await.unwrap_or(false); + } + if let Some(p) = payer { + p.vault.remove(id); + } + let receipt = EvalReceipt { + provider, + pod_id: pod_id.to_owned(), + image_digest, + submission_hash: EvalReceipt::hash_submission(&row.architecture_py, &row.training_py), + metrics_hash: metrics.as_ref().ok().map_or_else( + || "none".into(), + |m| EvalReceipt::hash_metrics_bytes(&serde_json::to_vec(m).unwrap_or_default()), + ), + termination_verified, + }; + let metrics = metrics.map_err(|e| format!("exec: {e}"))?; + Ok((metrics, receipt)) +} diff --git a/crates/prism-pipeline/src/lib.rs b/crates/prism-pipeline/src/lib.rs index 65b2ce0d3..f892df681 100644 --- a/crates/prism-pipeline/src/lib.rs +++ b/crates/prism-pipeline/src/lib.rs @@ -23,8 +23,8 @@ pub use composite::{ }; pub use config::PrismConfig; pub use pipeline::{ - measurement_patch, resume_measurement, run_eval_pipeline, run_sim_pipeline, PipelineError, - PipelineInput, PipelineResult, + measurement_patch, mid_pod_resume, resume_measurement, run_eval_pipeline, run_sim_pipeline, + PipelineError, PipelineInput, PipelineResult, }; pub use precheck::{ corpus_from_rows, ephemeral_candidate, evaluate_copy_precheck, gate_corpus_from_rows, diff --git a/crates/prism-pipeline/src/pipeline.rs b/crates/prism-pipeline/src/pipeline.rs index d5c4a1a20..7d07414f8 100644 --- a/crates/prism-pipeline/src/pipeline.rs +++ b/crates/prism-pipeline/src/pipeline.rs @@ -167,6 +167,12 @@ pub fn resume_measurement( Some((metrics, receipt)) } +/// True when a mid-pod row should reattach (has `pod_id`, no completed measure). +#[must_use] +pub fn mid_pod_resume(row: &prism_store::SubmissionState) -> bool { + row.pod_id.is_some() && resume_measurement(row).is_none() +} + /// Store patch persisting a completed measurement (receipt + metrics + bpb); /// the metrics blob also lands the per-step telemetry series master-side. #[must_use] diff --git a/docs/COMPLETENESS.md b/docs/COMPLETENESS.md index 6bdf36337..358c510ed 100644 --- a/docs/COMPLETENESS.md +++ b/docs/COMPLETENESS.md @@ -104,7 +104,7 @@ Agent/operator contracts: root [`AGENTS.md`](../AGENTS.md), [`deploy/AGENTS.md`] | design rating / elimination | done | Integer Elo (K=32), bottom 20% / 4-round cooldown, exact-E leaves. | | design API | done | Harness/quota/runs/viewer/annotate/ops on `:8093`. | | prism Lium backend | done | `PRISM_FORCE_SIM=false` in staging; the binary logs `eval_backend=lium`. API key is mounted from a file so it never appears in `docker inspect`. | -| prism orchestration | done | DB-backed claim/execute/review/similarity/score state machine (`prism_submission` + append-only `prism_stage_event`), pre-pod screens (copy gate + static cheat + AST similarity) before Lium rent, sweeper (10h grace + pre-reclaim log harvest; skips live workers), **boot + periodic orphan reconcile** (`control_plane_restart` / `harness_detached`, sealed BYOK TTL vault for pod cleanup, `GET /v1/submissions/{id}/logs`), epoch-close batched D24 leaf emission with **WTA** (`prism-emit` outbox: `emitted_epoch` watermark + `prism_emit_cursor` + positive-score carry + `apply_wta`, migration 0012). `PRISM_MAX_CONCURRENT_EVALS` default/prod = 8. | +| prism orchestration | done | DB-backed claim/execute/review/similarity/score state machine (`prism_submission` + append-only `prism_stage_event`), pre-pod screens (copy gate + static cheat + AST similarity) before Lium rent, sweeper (10h grace + pre-reclaim log harvest; skips live workers), **detached harness + resume-first boot/periodic reconcile** (reattach live pods via sealed BYOK; fail-closed only when unreattachable — `control_plane_restart` / `harness_detached`; `GET /v1/submissions/{id}/logs`), epoch-close batched D24 leaf emission with **WTA** (`prism-emit` outbox: `emitted_epoch` watermark + `prism_emit_cursor` + positive-score carry + `apply_wta`, migration 0012). `PRISM_MAX_CONCURRENT_EVALS` default/prod = 8. | | prism recipe v1 | done | `prism-recipe` contract, fineweb-edu pinned shard (URL + SHA-256, harness re-verifies), 6h train / 7h pod caps, baseline sources, recipe pin hex on the API. | | prism v3 harness | done (branch `prism-better`) | Multi-file harness package (`main.py` + `prismlib/`, miner code in `unshare --net` subprocess), seeded train stream with authoritative token counter, G6 probes, `prismlib/cheatguard.py` AST audit, METRICS_JSON v2, miner-chosen tokenizer, G5 RULER/BABILong/natural (pretrain-only), `RECIPE_VERSION 1.4.0`. | | prism v3 eval battery | done (branch `prism-better`) | G1–G8 under `harness/eval/` (intrinsic, downstream, recall, reasoning, long-context, curve, inference, stability) + `eval/public_dev/` anchors family + `tests/smoke_battery.py`. | diff --git a/docs/PRISM.md b/docs/PRISM.md index 7bf275981..579dd29c5 100644 --- a/docs/PRISM.md +++ b/docs/PRISM.md @@ -73,15 +73,20 @@ stateDiagram-v2 ``` All transitions are append-only events in `prism_stage_event`; the row state -lives in `prism_submission`. On boot (and every ~30s) `recover_on_boot` / -orphan reconcile fails mid-flight `provisioning`/`running` rows whose worker -is not alive in this process (`control_plane_restart` / `harness_detached`), -best-effort terminates the pod when the BYOK vault still has a key (memory + -optional short-TTL sealed file under `PRISM_PAYER_VAULT_DIR`), and requeues -post-measure review stages. The stuck sweeper remains a **10h** backstop -(aligned above wait-RUNNING + 6h train + SSH margin) and skips live workers. -`GET /v1/submissions/{id}/logs?since=` exposes harvested harness tails + -heartbeats while a pod is measuring. +lives in `prism_submission`. Live measure runs the harness **detached** on the +pod (`setsid` + `harness.log` / `harness.pid`) so a control-plane restart does +not SIGHUP GPU work. On boot (and every ~30s) orphan reconcile is +**resume-first**: mid-flight `provisioning`/`running` rows whose Lium pod is +still alive and whose BYOK key can be restored from the short-TTL sealed vault +(`PRISM_PAYER_VAULT_DIR`) are requeued with `pod_id` kept — the orchestrator +reattaches (log/event poll → wait terminal → harvest → score) without +terminating the pod. Only unreattachable rows fail-closed +(`control_plane_restart` / `harness_detached`) with best-effort terminate. +Post-measure review stages still requeue. Residual gap: expired seal + no +operator fallback ⇒ cannot call Lium API ⇒ fail-orphan (miner must stop the +pod and resubmit). The stuck sweeper remains a **10h** backstop and skips live +workers. `GET /v1/submissions/{id}/logs?since=` exposes harvested harness tails ++ heartbeats while a pod is measuring. Evaluation (Lium / Sim, review, agentic, leaf emit) is **master-only**. Validators never run `prism-challenge` — they fetch sealed weights only. diff --git a/docs/external-miner/prism.md b/docs/external-miner/prism.md index 8bf9c337d..e170c054e 100644 --- a/docs/external-miner/prism.md +++ b/docs/external-miner/prism.md @@ -67,17 +67,18 @@ X-Lium-Api-Key: The key is held in master memory for that submission and may also land in a **short-TTL encrypted seal file** on the master host (never in Postgres, never -logged) so a control-plane restart can still stop your pod. Missing key on -live → `400 missing_lium_api_key`. Cost guardrails (`max_price_per_hour`, -lifetime) still apply so a bad key cannot rent unbounded SKUs through the -orchestrator. - -If the challenge process restarts mid-run, your submission is marked failed -promptly with `control_plane_restart` / `harness_detached` (not left `running` -for hours). When the seal is missing, stop the pod yourself on Lium, then -resubmit with `X-Lium-Api-Key`. Poll `GET /v1/submissions/{id}/events` and -`GET /v1/submissions/{id}/logs?since=` for live stage heartbeats and harness -tails while the run is healthy. +logged) so a control-plane restart can **resume** your pod (or stop it if +reattach is impossible). Missing key on live → `400 missing_lium_api_key`. +Cost guardrails (`max_price_per_hour`, lifetime) still apply so a bad key +cannot rent unbounded SKUs through the orchestrator. + +If the challenge process restarts mid-run while your Lium pod is still +training/evaling, master **reattaches** quietly (same submission id; pod is +not killed). You only see `control_plane_restart` / `harness_detached` when +the pod is already dead or the sealed key expired and master cannot talk to +Lium — then stop the pod yourself and resubmit with `X-Lium-Api-Key`. Poll +`GET /v1/submissions/{id}/events` and `GET /v1/submissions/{id}/logs?since=` +for live stage heartbeats and harness tails while the run is healthy. ## Submit diff --git a/docs/external-miner/troubleshoot.md b/docs/external-miner/troubleshoot.md index f2232a99e..6ca016600 100644 --- a/docs/external-miner/troubleshoot.md +++ b/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` | Challenge process restarted mid-pod | Stop the Lium pod if still billing; resubmit with `X-Lium-Api-Key` | +| `control_plane_restart` / `harness_detached` | Restart could not reattach (dead pod or expired 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. | | 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 | diff --git a/docs/runbooks/prism-enable-lium-and-emission.md b/docs/runbooks/prism-enable-lium-and-emission.md index dff4c6c9f..86fdbd473 100644 --- a/docs/runbooks/prism-enable-lium-and-emission.md +++ b/docs/runbooks/prism-enable-lium-and-emission.md @@ -8,6 +8,13 @@ 4. Run inventory probe → single rent smoke → terminate → `verify_terminated`. 5. Prod default is `PRISM_MAX_CONCURRENT_EVALS=8` (orchestrator worker count / semaphore). Dial down only if the Lium lease pool cannot absorb the load. +6. **Control-plane restart / redeploy (GPU-safe):** keep + `PRISM_PAYER_VAULT_DIR` + `PRISM_PAYER_VAULT_KEY_FILE` on a durable volume. + Healthy mid-flight pods are resumed (not terminated). Do not manually kill + Lium pods after a routine `prism-challenge` bounce — only stop pods that + surface `control_plane_restart` / `harness_detached` (dead pod or expired + seal). Prefer rolling the challenge image when no pods are in + `provisioning`/`running`, or accept resume after boot. ## Emission ceremony (shared with design)