From 06383ff79e5e2a8d250ea3f1ce82cb4b82c67314 Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 14:03:20 +0800 Subject: [PATCH 1/2] feat(harness): gate prediction dispatch on admission --- .../app/src/prediction_dispatch.rs | 580 +++++++++++++++++- 1 file changed, 553 insertions(+), 27 deletions(-) diff --git a/rust_hft/alpha-harness/app/src/prediction_dispatch.rs b/rust_hft/alpha-harness/app/src/prediction_dispatch.rs index 7981b060a..3ac633fa1 100644 --- a/rust_hft/alpha-harness/app/src/prediction_dispatch.rs +++ b/rust_hft/alpha-harness/app/src/prediction_dispatch.rs @@ -3,7 +3,7 @@ use crate::{ print_json, PredictionDispatchRenderArgs, PredictionDispatchStatusArgs, PredictionDispatchSubmitArgs, }, - mission_runner::normalized_sha256, + mission_runner::{configured_sibling_binary, normalized_sha256}, }; use anyhow::{bail, Context}; use serde::{Deserialize, Serialize}; @@ -11,16 +11,21 @@ use serde_json::{json, Value}; use sha2::{Digest, Sha256}; use std::{ collections::BTreeMap, - ffi::OsString, fs::File, io::{Read, Write}, - path::Path, + path::{Path, PathBuf}, process::{Command, Output, Stdio}, + sync::mpsc, + thread, + time::Duration, }; const MAX_SUBMISSION_BYTES: u64 = 1024 * 1024; +const MAX_ADMISSION_RESPONSE_BYTES: u64 = 16 * 1024; +const SNAPSHOT_ADMISSION_TIMEOUT: Duration = Duration::from_secs(30); const RESOURCE_PROFILE: &str = "standard-v1"; const ACTIVE_DEADLINE_SECONDS: u64 = 1800; +const SNAPSHOT_ADMISSION_SCHEMA_VERSION: &str = "monday.prediction.snapshot_admission.v1"; #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] @@ -40,6 +45,55 @@ struct PredictionSubmission { resume_url: Option, #[serde(default)] resume_sha256: Option, + catalog_partition_artifact: CatalogPartitionArtifactRef, + compiler_source_identity: String, + build_input_identity: String, + task_capability: String, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct CatalogPartitionArtifactRef { + path: String, + artifact_sha256: String, + payload_sha256: String, +} + +#[derive(Serialize)] +struct SnapshotAdmissionRequest<'a> { + schema_version: &'static str, + catalog_partition_artifact: &'a CatalogPartitionArtifactRef, + compiler_source_identity: &'a str, + compiler_image_identity: String, + build_input_identity: &'a str, + task_capability: &'a str, +} + +#[derive(Deserialize)] +#[serde(tag = "status", rename_all = "lowercase", deny_unknown_fields)] +enum SnapshotAdmissionResponse { + Admitted { + schema_version: String, + snapshot_contract_id: String, + snapshot_digest: String, + partition_digest: String, + policy_identity: String, + task_capability: String, + immutable_image_identity: String, + }, + Rejected { + schema_version: String, + rejection: String, + }, +} + +struct SnapshotAdmission { + snapshot_contract_id: String, + snapshot_digest: String, + partition_digest: String, + policy_identity: String, + task_capability: String, + immutable_image_identity: String, } struct ValidatedSubmission { @@ -57,6 +111,21 @@ struct ValidatedSubmission { secret_name: String, } +struct AdmittedSubmission { + validated: ValidatedSubmission, + admission: SnapshotAdmission, +} + +enum AdmissionDecision { + Admitted(Box), + Rejected(String), +} + +enum SnapshotAdmissionDecision { + Admitted(SnapshotAdmission), + Rejected(String), +} + #[derive(Debug)] struct RenderedSubmission { manifest: Value, @@ -167,7 +236,16 @@ struct PredictionStatus { pub fn render(args: PredictionDispatchRenderArgs) -> anyhow::Result<()> { let submission = load_submission(&args.submission)?; - let rendered = render_submission(submission, &args.namespace)?; + let validated = validate_submission(submission)?; + let sibling = configured_sibling_binary( + "MONDAY_PREDICTION_SNAPSHOT_BIN", + "monday-prediction-snapshot", + )?; + let admitted = match admit_submission(validated, &sibling)? { + AdmissionDecision::Admitted(admitted) => *admitted, + AdmissionDecision::Rejected(rejection) => return report_admission_rejection(&rejection), + }; + let rendered = render_admitted_submission(admitted, &args.namespace)?; print_json(&json!({ "job_name": rendered.job_name, "secret_name": rendered.secret_name, @@ -180,15 +258,46 @@ pub fn submit(args: PredictionDispatchSubmitArgs) -> anyhow::Result<()> { validate_cluster_target(&args.context, &args.namespace)?; let submission = load_submission(&args.submission)?; let validated = validate_submission(submission)?; + let sibling = configured_sibling_binary( + "MONDAY_PREDICTION_SNAPSHOT_BIN", + "monday-prediction-snapshot", + )?; + submit_validated_submission(args, validated, &sibling, &kubectl_binary()) +} + +#[cfg(test)] +fn submit_with_binaries( + args: PredictionDispatchSubmitArgs, + sibling: &Path, + kubectl: &Path, +) -> anyhow::Result<()> { + validate_cluster_target(&args.context, &args.namespace)?; + let submission = load_submission(&args.submission)?; + let validated = validate_submission(submission)?; + submit_validated_submission(args, validated, sibling, kubectl) +} + +fn submit_validated_submission( + args: PredictionDispatchSubmitArgs, + validated: ValidatedSubmission, + sibling: &Path, + kubectl: &Path, +) -> anyhow::Result<()> { + let admitted = match admit_submission(validated, sibling)? { + AdmissionDecision::Admitted(admitted) => *admitted, + AdmissionDecision::Rejected(rejection) => return report_admission_rejection(&rejection), + }; let existing = existing_result_jobs( + kubectl, &args.context, &args.namespace, - &validated.result_identity_label, + &admitted.validated.result_identity_label, )?; ensure_result_available(&existing)?; - let rendered = render_validated_submission(validated, &args.namespace)?; + let rendered = render_admitted_submission(admitted, &args.namespace)?; let secret_body = serde_json::to_vec(&rendered.manifest["items"][0])?; let output = kubectl_with_input( + kubectl, &args.context, &args.namespace, ["create", "-f", "-"], @@ -197,6 +306,7 @@ pub fn submit(args: PredictionDispatchSubmitArgs) -> anyhow::Result<()> { ensure_kubectl_success(output, "create immutable prediction input Secret")?; let job_body = serde_json::to_vec(&rendered.manifest["items"][1])?; let output = kubectl_with_input( + kubectl, &args.context, &args.namespace, ["create", "-f", "-"], @@ -206,6 +316,7 @@ pub fn submit(args: PredictionDispatchSubmitArgs) -> anyhow::Result<()> { ensure_kubectl_success(output, "create immutable prediction research Job") { let readback = existing_result_jobs( + kubectl, &args.context, &args.namespace, &rendered.result_identity_sha256[..32], @@ -217,7 +328,7 @@ pub fn submit(args: PredictionDispatchSubmitArgs) -> anyhow::Result<()> { ) { Some(true) => true, Some(false) => { - delete_input_secret(&args.context, &args.namespace, &rendered.secret_name)?; + delete_input_secret(kubectl, &args.context, &args.namespace, &rendered.secret_name)?; return Err(error); } None => return Err(error.context(match readback { @@ -240,9 +351,11 @@ pub fn submit(args: PredictionDispatchSubmitArgs) -> anyhow::Result<()> { } pub fn status(args: PredictionDispatchStatusArgs) -> anyhow::Result<()> { + let kubectl = kubectl_binary(); validate_cluster_target(&args.context, &args.namespace)?; validate_dns_label("prediction Job name", &args.job_name)?; let job = kubectl_json( + &kubectl, &args.context, &args.namespace, ["get", "job", &args.job_name, "-o", "json"], @@ -251,6 +364,7 @@ pub fn status(args: PredictionDispatchStatusArgs) -> anyhow::Result<()> { let job_uid = status_job_uid(&job)?; let selector = format!("batch.kubernetes.io/controller-uid={job_uid}"); let pods = kubectl_json( + &kubectl, &args.context, &args.namespace, ["get", "pods", "-l", &selector, "-o", "json"], @@ -439,18 +553,241 @@ fn validate_submission(submission: PredictionSubmission) -> anyhow::Result anyhow::Result { + admit_submission_with_timeout(validated, sibling, SNAPSHOT_ADMISSION_TIMEOUT) +} + +fn admit_submission_with_timeout( + validated: ValidatedSubmission, + sibling: &Path, + timeout: Duration, +) -> anyhow::Result { + let compiler_source_identity = immutable_sha256_identity( + "compiler source identity", + &validated.submission.compiler_source_identity, + )?; + let build_input_identity = immutable_sha256_identity( + "build input identity", + &validated.submission.build_input_identity, + )?; + let artifact_sha256 = immutable_sha256_identity( + "catalog partition artifact", + &validated + .submission + .catalog_partition_artifact + .artifact_sha256, + )?; + let payload_sha256 = immutable_sha256_identity( + "catalog partition payload", + &validated + .submission + .catalog_partition_artifact + .payload_sha256, + )?; + if validated + .submission + .catalog_partition_artifact + .path + .is_empty() + || validated + .submission + .catalog_partition_artifact + .path + .chars() + .any(char::is_control) + { + bail!("catalog partition artifact path is invalid"); + } + let request = SnapshotAdmissionRequest { + schema_version: SNAPSHOT_ADMISSION_SCHEMA_VERSION, + catalog_partition_artifact: &CatalogPartitionArtifactRef { + path: validated.submission.catalog_partition_artifact.path.clone(), + artifact_sha256, + payload_sha256, + }, + compiler_source_identity: &compiler_source_identity, + compiler_image_identity: format!("sha256:{}", validated.image_digest), + build_input_identity: &build_input_identity, + task_capability: &validated.submission.task_capability, + }; + let request = serde_json::to_vec(&request).context("serialize snapshot admission request")?; + let mut child = Command::new(sibling) + .arg("--admit-authenticated-snapshot") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .with_context(|| format!("start snapshot admission sibling {}", sibling.display()))?; + let mut stdin = child + .stdin + .take() + .context("snapshot admission sibling stdin is unavailable")?; + if let Err(error) = stdin.write_all(&request) { + let _ = child.kill(); + let _ = child.wait(); + return Err(error).context("write snapshot admission request"); + } + drop(stdin); + let stdout = child + .stdout + .take() + .context("snapshot admission sibling stdout is unavailable")?; + let (sender, receiver) = mpsc::sync_channel(1); + let reader = thread::spawn(move || { + let _ = sender.send(read_bounded_admission_response(stdout)); + }); + let output = match receiver.recv_timeout(timeout) { + Ok(output) => output, + Err(mpsc::RecvTimeoutError::Timeout) => { + let _ = child.kill(); + let _ = child.wait(); + let _ = reader.join(); + bail!( + "snapshot admission sibling timed out after {} seconds", + timeout.as_secs() + ); + } + Err(mpsc::RecvTimeoutError::Disconnected) => { + let _ = child.kill(); + let _ = child.wait(); + bail!("snapshot admission sibling response reader failed"); + } + }; + reader + .join() + .map_err(|_| anyhow::anyhow!("snapshot admission sibling response reader panicked"))?; + let status = child + .wait() + .context("wait for snapshot admission sibling")?; + let output = output?; + if !status.success() { + bail!("snapshot admission sibling exited unsuccessfully"); + } + match parse_snapshot_admission_response(&output, &validated)? { + SnapshotAdmissionDecision::Admitted(admission) => { + Ok(AdmissionDecision::Admitted(Box::new(AdmittedSubmission { + validated, + admission, + }))) + } + SnapshotAdmissionDecision::Rejected(rejection) => { + Ok(AdmissionDecision::Rejected(rejection)) + } + } +} + +fn read_bounded_admission_response(mut reader: impl Read) -> anyhow::Result> { + let mut output = Vec::new(); + let mut buffer = [0_u8; 4096]; + let mut exceeded = false; + loop { + let read = reader.read(&mut buffer)?; + if read == 0 { + break; + } + let remaining = MAX_ADMISSION_RESPONSE_BYTES.saturating_add(1) as usize - output.len(); + let copied = read.min(remaining); + output.extend_from_slice(&buffer[..copied]); + exceeded |= copied < read || output.len() as u64 > MAX_ADMISSION_RESPONSE_BYTES; + } + if exceeded { + bail!("snapshot admission response exceeds {MAX_ADMISSION_RESPONSE_BYTES} bytes"); + } + Ok(output) +} + +fn parse_snapshot_admission_response( + output: &[u8], + validated: &ValidatedSubmission, +) -> anyhow::Result { + let output = std::str::from_utf8(output).context("snapshot admission response is not UTF-8")?; + let response = output + .strip_suffix('\n') + .filter(|line| !line.is_empty() && !line.contains('\n')) + .context("snapshot admission response must be exactly one non-empty JSON line")?; + let response: SnapshotAdmissionResponse = + serde_json::from_str(response).context("snapshot admission response is invalid")?; + match response { + SnapshotAdmissionResponse::Rejected { + schema_version, + rejection, + } => { + if schema_version != SNAPSHOT_ADMISSION_SCHEMA_VERSION || rejection.is_empty() { + bail!("snapshot admission response is invalid"); + } + Ok(SnapshotAdmissionDecision::Rejected(rejection)) + } + SnapshotAdmissionResponse::Admitted { + schema_version, + snapshot_contract_id, + snapshot_digest, + partition_digest, + policy_identity, + task_capability, + immutable_image_identity, + } => { + if schema_version != SNAPSHOT_ADMISSION_SCHEMA_VERSION + || task_capability != validated.submission.task_capability + || immutable_image_identity != format!("sha256:{}", validated.image_digest) + || immutable_sha256_identity("admitted snapshot digest", &snapshot_digest)? + != format!("sha256:{}", validated.snapshot_sha256) + { + bail!("snapshot admission response does not bind the submitted immutable identity"); + } + Ok(SnapshotAdmissionDecision::Admitted(SnapshotAdmission { + snapshot_contract_id: immutable_sha256_identity( + "admitted snapshot contract", + &snapshot_contract_id, + )?, + snapshot_digest, + partition_digest: immutable_sha256_identity( + "admitted partition", + &partition_digest, + )?, + policy_identity: immutable_sha256_identity("admitted policy", &policy_identity)?, + task_capability, + immutable_image_identity, + })) + } + } +} + +fn report_admission_rejection(rejection: &str) -> anyhow::Result<()> { + print_json(&json!({ + "schema_version": SNAPSHOT_ADMISSION_SCHEMA_VERSION, + "status": "rejected", + "rejection": rejection, + }))?; + bail!("snapshot admission rejected: {rejection}") +} + +fn immutable_sha256_identity(label: &str, value: &str) -> anyhow::Result { + let digest = value + .strip_prefix("sha256:") + .with_context(|| format!("{label} must use sha256:<64 lowercase hex>"))?; + if value != format!("sha256:{digest}") || digest != normalized_sha256(label, digest)? { + bail!("{label} must use sha256:<64 lowercase hex>"); + } + Ok(value.to_owned()) +} + +fn render_admitted_submission( + admitted: AdmittedSubmission, namespace: &str, ) -> anyhow::Result { validate_dns_label("namespace", namespace)?; - render_validated_submission(validate_submission(submission)?, namespace) + render_validated_submission(admitted, namespace) } fn render_validated_submission( - validated: ValidatedSubmission, + admitted: AdmittedSubmission, namespace: &str, ) -> anyhow::Result { + let admission = admitted.admission; + let validated = admitted.validated; let resume_url = validated.submission.resume_url.as_deref().unwrap_or(""); let resume_sha256 = validated.resume_sha256.as_deref().unwrap_or(""); let labels = json!({ @@ -465,6 +802,12 @@ fn render_validated_submission( "research.monday/mission-object": validated.mission_object, "research.monday/snapshot-sha256": validated.snapshot_sha256, "research.monday/snapshot-object": validated.snapshot_object, + "research.monday/snapshot-contract-id": admission.snapshot_contract_id, + "research.monday/admitted-snapshot-digest": admission.snapshot_digest, + "research.monday/partition-digest": admission.partition_digest, + "research.monday/policy-identity": admission.policy_identity, + "research.monday/task-capability": admission.task_capability, + "research.monday/admitted-image-identity": admission.immutable_image_identity, "research.monday/result-object": validated.result_object, "research.monday/result-identity-sha256": validated.result_identity_sha256, "research.monday/image-digest": validated.image_digest, @@ -668,17 +1011,21 @@ fn sha256_text(value: &str) -> String { format!("{:x}", Sha256::digest(value.as_bytes())) } -fn kubectl_binary() -> OsString { - std::env::var_os("MONDAY_KUBECTL_BIN").unwrap_or_else(|| OsString::from("kubectl")) +fn kubectl_binary() -> PathBuf { + std::env::var_os("MONDAY_KUBECTL_BIN") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("kubectl")) } fn existing_result_jobs( + kubectl: &Path, context: &str, namespace: &str, result_identity_label: &str, ) -> anyhow::Result> { let selector = format!("research.monday/result-id={result_identity_label}"); let jobs = kubectl_json( + kubectl, context, namespace, ["get", "jobs", "-l", selector.as_str(), "-o", "json"], @@ -728,12 +1075,13 @@ fn create_failure_recovered( } fn kubectl_json( + kubectl: &Path, context: &str, namespace: &str, args: [&str; N], action: &str, ) -> anyhow::Result { - let output = Command::new(kubectl_binary()) + let output = Command::new(kubectl) .arg("--context") .arg(context) .arg("--namespace") @@ -746,12 +1094,13 @@ fn kubectl_json( } fn kubectl_with_input( + kubectl: &Path, context: &str, namespace: &str, args: [&str; N], input: &[u8], ) -> anyhow::Result { - let mut child = Command::new(kubectl_binary()) + let mut child = Command::new(kubectl) .arg("--context") .arg(context) .arg("--namespace") @@ -770,8 +1119,13 @@ fn kubectl_with_input( child.wait_with_output().context("wait for kubectl") } -fn delete_input_secret(context: &str, namespace: &str, secret_name: &str) -> anyhow::Result<()> { - let output = Command::new(kubectl_binary()) +fn delete_input_secret( + kubectl: &Path, + context: &str, + namespace: &str, + secret_name: &str, +) -> anyhow::Result<()> { + let output = Command::new(kubectl) .arg("--context") .arg(context) .arg("--namespace") @@ -799,8 +1153,11 @@ mod tests { #[test] fn renders_the_blessed_immutable_prediction_job() { - let rendered = render_submission(valid_submission(), "monday-research") - .expect("valid immutable submission must render"); + let rendered = render_admitted_submission( + admitted_submission_for_test(valid_submission()), + "monday-research", + ) + .expect("valid immutable submission must render"); let secret = &rendered.manifest["items"][0]; let job = &rendered.manifest["items"][1]; @@ -825,6 +1182,10 @@ mod tests { job["metadata"]["annotations"]["research.monday/result-identity-sha256"], rendered.result_identity_sha256 ); + assert_eq!( + job["metadata"]["annotations"]["research.monday/partition-digest"], + format!("sha256:{}", "e".repeat(64)) + ); assert!(rendered.job_name.starts_with("prediction-")); } @@ -836,7 +1197,7 @@ mod tests { ] { let mut submission = valid_submission(); submission.image = image; - assert!(render_submission(submission, "monday-research").is_err()); + assert!(validate_submission(submission).is_err()); } } @@ -845,8 +1206,9 @@ mod tests { let mut submission = valid_submission(); submission.snapshot_sha256.clear(); - let error = render_submission(submission, "monday-research") - .expect_err("snapshot without an authenticated digest must fail"); + let error = validate_submission(submission) + .err() + .expect("snapshot without an authenticated digest must fail"); assert!(error.to_string().contains("snapshot SHA256 is invalid")); } @@ -865,8 +1227,9 @@ mod tests { submission.result_put_url = "https://oss-internal/results/latest/results.zip?signature=x".to_owned(); - let error = render_submission(submission, "monday-research") - .expect_err("mutable output identity must fail before Job creation"); + let error = validate_submission(submission) + .err() + .expect("mutable output identity must fail before Job creation"); assert!(error.to_string().contains("exact attempt id")); } @@ -879,7 +1242,7 @@ mod tests { ] { let mut submission = valid_submission(); submission.mission_url = url.to_owned(); - assert!(render_submission(submission, "monday-research").is_err()); + assert!(validate_submission(submission).is_err()); } } @@ -889,8 +1252,9 @@ mod tests { submission.resume_url = Some("https://oss-internal/results/prior.zip?signature=x".to_owned()); - let error = render_submission(submission, "monday-research") - .expect_err("incomplete resume pair must fail"); + let error = validate_submission(submission) + .err() + .expect("incomplete resume pair must fail"); assert!(error.to_string().contains("must be supplied together")); } @@ -911,6 +1275,146 @@ mod tests { } } + #[cfg(unix)] + #[test] + fn rejected_sibling_admission_never_reaches_kubectl() { + use std::os::unix::fs::PermissionsExt; + + let root = tempfile::tempdir().expect("create dispatch test root"); + let submission = root.path().join("submission.json"); + std::fs::write( + &submission, + serde_json::to_vec(&json!({ + "attempt_id": "btc-5m-attempt-001", + "mission_id": "btc-5m-mission-001", + "image": format!("registry/research-runner@sha256:{}", "a".repeat(64)), + "evaluator_version": format!("sha256:{}", "b".repeat(64)), + "resource_profile": RESOURCE_PROFILE, + "mission_url": "https://oss-internal/missions/mission.json?signature=x", + "mission_sha256": "c".repeat(64), + "snapshot_url": "https://oss-internal/snapshots/snapshot.zip?signature=x", + "snapshot_sha256": "d".repeat(64), + "result_put_url": "https://oss-internal/results/btc-5m-attempt-001/results.zip?signature=x", + "llm_secret_name": "monday-prediction-llm", + "catalog_partition_artifact": { + "path": "catalog/catalog-partition-deadbeef.json", + "artifact_sha256": format!("sha256:{}", "e".repeat(64)), + "payload_sha256": format!("sha256:{}", "f".repeat(64)), + }, + "compiler_source_identity": format!("sha256:{}", "1".repeat(64)), + "build_input_identity": format!("sha256:{}", "2".repeat(64)), + "task_capability": "btc_5m_backtest", + })) + .expect("serialize submission"), + ) + .expect("write submission"); + let sibling = root.path().join("monday-prediction-snapshot"); + std::fs::write( + &sibling, + "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' '{\"schema_version\":\"monday.prediction.snapshot_admission.v1\",\"status\":\"rejected\",\"rejection\":\"unsupported_task\"}'\n", + ) + .expect("write sibling"); + std::fs::set_permissions(&sibling, std::fs::Permissions::from_mode(0o700)) + .expect("make sibling executable"); + let kubectl = root.path().join("kubectl"); + let kubectl_log = root.path().join("kubectl-called"); + std::fs::write( + &kubectl, + format!("#!/bin/sh\ntouch '{}'\nexit 1\n", kubectl_log.display()), + ) + .expect("write kubectl sentinel"); + std::fs::set_permissions(&kubectl, std::fs::Permissions::from_mode(0o700)) + .expect("make kubectl sentinel executable"); + + let error = submit_with_binaries( + PredictionDispatchSubmitArgs { + submission, + context: "ack".to_owned(), + namespace: "monday-research".to_owned(), + }, + &sibling, + &kubectl, + ) + .expect_err("rejected admission must fail dispatch after reporting a typed result"); + + assert!(error.to_string().contains("unsupported_task")); + assert!( + !kubectl_log.exists(), + "rejection must occur before any Kubernetes read or write" + ); + } + + #[test] + fn admission_response_must_be_one_line_and_bind_submission_identities() { + let validated = validate_submission(valid_submission()).expect("valid submission"); + let admitted = serde_json::to_vec(&json!({ + "schema_version": SNAPSHOT_ADMISSION_SCHEMA_VERSION, + "status": "admitted", + "snapshot_contract_id": format!("sha256:{}", "1".repeat(64)), + "snapshot_digest": format!("sha256:{}", "d".repeat(64)), + "partition_digest": format!("sha256:{}", "e".repeat(64)), + "policy_identity": format!("sha256:{}", "f".repeat(64)), + "task_capability": "btc_5m_backtest", + "immutable_image_identity": format!("sha256:{}", "a".repeat(64)), + })) + .expect("serialize admitted response"); + let mut one_line = admitted.clone(); + one_line.push(b'\n'); + assert!(parse_snapshot_admission_response(&one_line, &validated).is_ok()); + + let rejected = serde_json::to_vec(&json!({ + "schema_version": SNAPSHOT_ADMISSION_SCHEMA_VERSION, + "status": "rejected", + "rejection": "unsupported_task", + })) + .expect("serialize rejected response"); + let mut rejected = rejected; + rejected.push(b'\n'); + assert!(matches!( + parse_snapshot_admission_response(&rejected, &validated), + Ok(SnapshotAdmissionDecision::Rejected(reason)) if reason == "unsupported_task" + )); + + let mut extra_output = serde_json::to_vec(&json!({ + "schema_version": SNAPSHOT_ADMISSION_SCHEMA_VERSION, + "status": "rejected", + "rejection": "unsupported_task", + })) + .expect("serialize rejected response"); + extra_output.extend_from_slice(b"\nextra"); + let mut mismatched = + serde_json::from_slice::(&admitted).expect("parse admitted response"); + mismatched["snapshot_digest"] = json!(format!("sha256:{}", "e".repeat(64))); + let mut mismatched = serde_json::to_vec(&mismatched).expect("serialize mismatch"); + mismatched.push(b'\n'); + for output in [admitted, extra_output, mismatched] { + assert!(parse_snapshot_admission_response(&output, &validated).is_err()); + } + } + + #[cfg(unix)] + #[test] + fn admission_timeout_kills_a_stalled_sibling() { + use std::os::unix::fs::PermissionsExt; + + let root = tempfile::tempdir().expect("create admission timeout root"); + let sibling = root.path().join("monday-prediction-snapshot"); + std::fs::write(&sibling, "#!/bin/sh\ncat >/dev/null\nsleep 1\n") + .expect("write stalled sibling"); + std::fs::set_permissions(&sibling, std::fs::Permissions::from_mode(0o700)) + .expect("make stalled sibling executable"); + + let error = admit_submission_with_timeout( + validate_submission(valid_submission()).expect("valid submission"), + &sibling, + Duration::from_millis(10), + ) + .err() + .expect("stalled sibling must time out"); + + assert!(error.to_string().contains("timed out")); + } + #[test] fn status_derives_only_kubernetes_milestones_without_evidence() { let status = derive_status(&status_job(), &status_pods(), None).unwrap(); @@ -995,6 +1499,28 @@ mod tests { llm_secret_name: "monday-prediction-llm".to_owned(), resume_url: None, resume_sha256: None, + catalog_partition_artifact: CatalogPartitionArtifactRef { + path: "catalog/catalog-partition-deadbeef.json".to_owned(), + artifact_sha256: format!("sha256:{}", "e".repeat(64)), + payload_sha256: format!("sha256:{}", "f".repeat(64)), + }, + compiler_source_identity: format!("sha256:{}", "1".repeat(64)), + build_input_identity: format!("sha256:{}", "2".repeat(64)), + task_capability: "btc_5m_backtest".to_owned(), + } + } + + fn admitted_submission_for_test(submission: PredictionSubmission) -> AdmittedSubmission { + AdmittedSubmission { + validated: validate_submission(submission).expect("valid test submission"), + admission: SnapshotAdmission { + snapshot_contract_id: format!("sha256:{}", "3".repeat(64)), + snapshot_digest: format!("sha256:{}", "d".repeat(64)), + partition_digest: format!("sha256:{}", "e".repeat(64)), + policy_identity: format!("sha256:{}", "f".repeat(64)), + task_capability: "btc_5m_backtest".to_owned(), + immutable_image_identity: format!("sha256:{}", "a".repeat(64)), + }, } } } From 800287d0a421e98e997a1e06f0e85bced8d42351 Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 14:38:06 +0800 Subject: [PATCH 2/2] fix(harness): bind admission through prediction runner --- rust_hft/alpha-harness/app/src/cli.rs | 12 + .../app/src/prediction_dispatch.rs | 245 +++++++++++++++--- .../app/src/prediction_runner.rs | 96 ++++++- 3 files changed, 312 insertions(+), 41 deletions(-) diff --git a/rust_hft/alpha-harness/app/src/cli.rs b/rust_hft/alpha-harness/app/src/cli.rs index 682dbfe14..0549ce65b 100644 --- a/rust_hft/alpha-harness/app/src/cli.rs +++ b/rust_hft/alpha-harness/app/src/cli.rs @@ -148,6 +148,10 @@ pub struct PredictionExecuteArgs { pub snapshot_url: String, #[arg(long)] pub snapshot_sha256: String, + #[arg(long)] + pub snapshot_contract_id: String, + #[arg(long)] + pub snapshot_digest: String, /// Read-only cache directory containing `.zip` archives. #[arg(long)] pub snapshot_cache_dir: Option, @@ -801,6 +805,10 @@ mod tests { root.path().join("missing-snapshot.zip").into_os_string(), OsString::from("--snapshot-sha256"), OsString::from("b".repeat(64)), + OsString::from("--snapshot-contract-id"), + OsString::from(format!("sha256:{}", "c".repeat(64))), + OsString::from("--snapshot-digest"), + OsString::from("0123456789abcdef"), OsString::from("--result-put-url"), root.path().join("results.zip").into_os_string(), ]) @@ -981,6 +989,10 @@ printf '%s\n' '{{"schema_version":"research_snapshot_v2","snapshot_hash":"012345 "snapshot.zip", "--snapshot-sha256", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "--snapshot-contract-id", + "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd", + "--snapshot-digest", + "0123456789abcdef", "--resume-url", "previous-results.zip", "--resume-sha256", diff --git a/rust_hft/alpha-harness/app/src/prediction_dispatch.rs b/rust_hft/alpha-harness/app/src/prediction_dispatch.rs index 3ac633fa1..ab4ca0007 100644 --- a/rust_hft/alpha-harness/app/src/prediction_dispatch.rs +++ b/rust_hft/alpha-harness/app/src/prediction_dispatch.rs @@ -13,11 +13,11 @@ use std::{ collections::BTreeMap, fs::File, io::{Read, Write}, - path::{Path, PathBuf}, - process::{Command, Output, Stdio}, + path::{Component, Path, PathBuf}, + process::{Child, Command, Output, Stdio}, sync::mpsc, thread, - time::Duration, + time::{Duration, Instant}, }; const MAX_SUBMISSION_BYTES: u64 = 1024 * 1024; @@ -39,6 +39,7 @@ struct PredictionSubmission { mission_sha256: String, snapshot_url: String, snapshot_sha256: String, + snapshot_contract_id: String, result_put_url: String, llm_secret_name: String, #[serde(default)] @@ -503,12 +504,14 @@ fn validate_submission(submission: PredictionSubmission) -> anyhow::Result output, Err(mpsc::RecvTimeoutError::Timeout) => { - let _ = child.kill(); - let _ = child.wait(); - let _ = reader.join(); + terminate_admission_child(&mut child); bail!( "snapshot admission sibling timed out after {} seconds", timeout.as_secs() ); } Err(mpsc::RecvTimeoutError::Disconnected) => { - let _ = child.kill(); - let _ = child.wait(); + terminate_admission_child(&mut child); bail!("snapshot admission sibling response reader failed"); } }; reader .join() .map_err(|_| anyhow::anyhow!("snapshot admission sibling response reader panicked"))?; - let status = child - .wait() - .context("wait for snapshot admission sibling")?; + let status = wait_for_admission_child_exit(&mut child, timeout)?; let output = output?; if !status.success() { bail!("snapshot admission sibling exited unsuccessfully"); @@ -679,6 +666,35 @@ fn admit_submission_with_timeout( } } +fn terminate_admission_child(child: &mut Child) { + let _ = child.kill(); + let _ = child.wait(); +} + +fn wait_for_admission_child_exit( + child: &mut Child, + timeout: Duration, +) -> anyhow::Result { + let deadline = Instant::now() + timeout; + loop { + match child.try_wait() { + Ok(Some(status)) => return Ok(status), + Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(10)), + Ok(None) => { + terminate_admission_child(child); + bail!( + "snapshot admission sibling did not exit within {} seconds after responding", + timeout.as_secs() + ); + } + Err(error) => { + terminate_admission_child(child); + return Err(error).context("poll snapshot admission sibling exit"); + } + } + } +} + fn read_bounded_admission_response(mut reader: impl Read) -> anyhow::Result> { let mut output = Vec::new(); let mut buffer = [0_u8; 4096]; @@ -731,9 +747,9 @@ fn parse_snapshot_admission_response( } => { if schema_version != SNAPSHOT_ADMISSION_SCHEMA_VERSION || task_capability != validated.submission.task_capability + || snapshot_contract_id != validated.submission.snapshot_contract_id || immutable_image_identity != format!("sha256:{}", validated.image_digest) - || immutable_sha256_identity("admitted snapshot digest", &snapshot_digest)? - != format!("sha256:{}", validated.snapshot_sha256) + || validate_snapshot_digest(&snapshot_digest).is_err() { bail!("snapshot admission response does not bind the submitted immutable identity"); } @@ -774,6 +790,17 @@ fn immutable_sha256_identity(label: &str, value: &str) -> anyhow::Result Ok(value.to_owned()) } +fn validate_snapshot_digest(value: &str) -> anyhow::Result<()> { + if value.len() != 16 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')) + { + bail!("admitted snapshot digest must be exactly 16 lowercase ASCII hex characters"); + } + Ok(()) +} + fn render_admitted_submission( admitted: AdmittedSubmission, namespace: &str, @@ -879,11 +906,13 @@ fn render_validated_submission( "--mission-sha256", "$(MISSION_SHA256)", "--snapshot-url", "$(SNAPSHOT_URL)", "--snapshot-sha256", "$(SNAPSHOT_SHA256)", + "--snapshot-contract-id", "$(SNAPSHOT_CONTRACT_ID)", + "--snapshot-digest", "$(SNAPSHOT_DIGEST)", "--resume-url", "$(RESUME_URL)", "--resume-sha256", "$(RESUME_SHA256)", "--result-put-url", "$(RESULT_PUT_URL)" ], - "env": prediction_environment(&validated), + "env": prediction_environment(&validated, &admission), "resources": { "requests": { "cpu": "3500m", "memory": "8Gi" }, "limits": { "cpu": "3500m", "memory": "12Gi" }, @@ -916,7 +945,7 @@ fn render_validated_submission( }) } -fn prediction_environment(validated: &ValidatedSubmission) -> Value { +fn prediction_environment(validated: &ValidatedSubmission, admission: &SnapshotAdmission) -> Value { let secret_ref = |key: &str| { json!({ "name": validated.secret_name, @@ -931,6 +960,8 @@ fn prediction_environment(validated: &ValidatedSubmission) -> Value { { "name": "RESUME_SHA256", "valueFrom": { "secretKeyRef": secret_ref("resume-sha256") } }, { "name": "MISSION_SHA256", "value": validated.mission_sha256 }, { "name": "SNAPSHOT_SHA256", "value": validated.snapshot_sha256 }, + { "name": "SNAPSHOT_CONTRACT_ID", "value": admission.snapshot_contract_id }, + { "name": "SNAPSHOT_DIGEST", "value": admission.snapshot_digest }, { "name": "MONDAY_PREDICTION_LLM_BASE_URL", "valueFrom": { "secretKeyRef": { "name": validated.submission.llm_secret_name, "key": "base-url" } } }, { "name": "MONDAY_PREDICTION_LLM_MODEL", "valueFrom": { "secretKeyRef": { "name": validated.submission.llm_secret_name, "key": "model" } } }, { "name": "MONDAY_PREDICTION_LLM_API_KEY", "valueFrom": { "secretKeyRef": { "name": validated.submission.llm_secret_name, "key": "api-key", "optional": true } } }, @@ -987,6 +1018,54 @@ fn validate_identifier(label: &str, value: &str) -> anyhow::Result<()> { Ok(()) } +fn validate_task_capability(value: &str) -> anyhow::Result<()> { + let bytes = value.as_bytes(); + if bytes.is_empty() + || bytes.len() > 63 + || !bytes[0].is_ascii_lowercase() + || !bytes[bytes.len() - 1].is_ascii_alphanumeric() + || !bytes.iter().all(|byte| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-') + }) + { + bail!("task capability must be a lowercase safe identifier"); + } + Ok(()) +} + +fn validate_catalog_partition_artifact_path(value: &str) -> anyhow::Result<()> { + let path = Path::new(value); + let has_windows_prefix = value.as_bytes().get(1) == Some(&b':') + && value + .as_bytes() + .first() + .is_some_and(u8::is_ascii_alphabetic); + if value.is_empty() + || value.chars().any(char::is_control) + || path.is_absolute() + || has_windows_prefix + || path.components().any(|component| { + matches!( + component, + Component::ParentDir | Component::RootDir | Component::Prefix(_) + ) + }) + { + bail!("catalog partition artifact path is invalid"); + } + let name = path + .file_name() + .and_then(|name| name.to_str()) + .context("catalog partition artifact path must end in a UTF-8 filename")?; + let suffix = name + .strip_prefix("catalog-partition-") + .and_then(|name| name.strip_suffix(".json")); + if suffix.is_none_or(str::is_empty) { + bail!("catalog partition artifact path must name catalog-partition-*.json"); + } + Ok(()) +} + fn validate_dns_label(label: &str, value: &str) -> anyhow::Result<()> { let bytes = value.as_bytes(); if bytes.is_empty() @@ -1186,6 +1265,16 @@ mod tests { job["metadata"]["annotations"]["research.monday/partition-digest"], format!("sha256:{}", "e".repeat(64)) ); + let container = &job["spec"]["template"]["spec"]["containers"][0]; + assert!(container["args"] + .as_array() + .is_some_and(|args| args.contains(&json!("--snapshot-contract-id")))); + assert!(container["env"] + .as_array() + .is_some_and(|values| values.iter().any(|value| { + value["name"] == "SNAPSHOT_CONTRACT_ID" + && value["value"] == format!("sha256:{}", "1".repeat(64)) + }))); assert!(rendered.job_name.starts_with("prediction-")); } @@ -1294,6 +1383,7 @@ mod tests { "mission_sha256": "c".repeat(64), "snapshot_url": "https://oss-internal/snapshots/snapshot.zip?signature=x", "snapshot_sha256": "d".repeat(64), + "snapshot_contract_id": format!("sha256:{}", "1".repeat(64)), "result_put_url": "https://oss-internal/results/btc-5m-attempt-001/results.zip?signature=x", "llm_secret_name": "monday-prediction-llm", "catalog_partition_artifact": { @@ -1351,7 +1441,7 @@ mod tests { "schema_version": SNAPSHOT_ADMISSION_SCHEMA_VERSION, "status": "admitted", "snapshot_contract_id": format!("sha256:{}", "1".repeat(64)), - "snapshot_digest": format!("sha256:{}", "d".repeat(64)), + "snapshot_digest": "0123456789abcdef", "partition_digest": format!("sha256:{}", "e".repeat(64)), "policy_identity": format!("sha256:{}", "f".repeat(64)), "task_capability": "btc_5m_backtest", @@ -1382,12 +1472,18 @@ mod tests { })) .expect("serialize rejected response"); extra_output.extend_from_slice(b"\nextra"); - let mut mismatched = + let mut invalid_digest = serde_json::from_slice::(&admitted).expect("parse admitted response"); - mismatched["snapshot_digest"] = json!(format!("sha256:{}", "e".repeat(64))); - let mut mismatched = serde_json::to_vec(&mismatched).expect("serialize mismatch"); - mismatched.push(b'\n'); - for output in [admitted, extra_output, mismatched] { + invalid_digest["snapshot_digest"] = json!("0123456789abcdeg"); + let mut invalid_digest = serde_json::to_vec(&invalid_digest).expect("serialize mismatch"); + invalid_digest.push(b'\n'); + let mut contract_mismatch = + serde_json::from_slice::(&admitted).expect("parse admitted response"); + contract_mismatch["snapshot_contract_id"] = json!(format!("sha256:{}", "2".repeat(64))); + let mut contract_mismatch = + serde_json::to_vec(&contract_mismatch).expect("serialize contract mismatch"); + contract_mismatch.push(b'\n'); + for output in [admitted, extra_output, invalid_digest, contract_mismatch] { assert!(parse_snapshot_admission_response(&output, &validated).is_err()); } } @@ -1407,7 +1503,7 @@ mod tests { let error = admit_submission_with_timeout( validate_submission(valid_submission()).expect("valid submission"), &sibling, - Duration::from_millis(10), + Duration::from_millis(100), ) .err() .expect("stalled sibling must time out"); @@ -1415,6 +1511,78 @@ mod tests { assert!(error.to_string().contains("timed out")); } + #[cfg(unix)] + #[test] + fn admission_timeout_does_not_join_a_reader_held_by_a_descendant() { + use std::os::unix::fs::PermissionsExt; + + let root = tempfile::tempdir().expect("create inherited-stdout timeout root"); + let sibling = root.path().join("monday-prediction-snapshot"); + std::fs::write(&sibling, "#!/bin/sh\ncat >/dev/null\nsleep 1 &\nexit 0\n") + .expect("write descendant-stalling sibling"); + std::fs::set_permissions(&sibling, std::fs::Permissions::from_mode(0o700)) + .expect("make descendant-stalling sibling executable"); + + let started = Instant::now(); + let error = admit_submission_with_timeout( + validate_submission(valid_submission()).expect("valid submission"), + &sibling, + Duration::from_millis(10), + ) + .err() + .expect("inherited stdout must time out without joining its reader"); + + assert!(error.to_string().contains("timed out")); + assert!(started.elapsed() < Duration::from_millis(500)); + } + + #[cfg(unix)] + #[test] + fn admission_exit_wait_kills_a_sibling_that_does_not_exit_after_response() { + let mut child = Command::new("/bin/sh") + .args(["-c", "sleep 1"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start response-stalling sibling"); + let error = wait_for_admission_child_exit(&mut child, Duration::from_millis(10)) + .expect_err("response-stalling sibling must be killed and reaped"); + + assert!(error.to_string().contains("did not exit"), "{error:#}"); + } + + #[test] + fn validates_catalog_partition_artifact_path_and_task_capability() { + for path in [ + "catalog/catalog-partition-deadbeef.json", + "nested/catalog/catalog-partition-deadbeef.json", + ] { + validate_catalog_partition_artifact_path(path).expect("valid relative catalog path"); + } + for path in [ + "", + "/catalog/catalog-partition-deadbeef.json", + "../catalog-partition-deadbeef.json", + "C:\\catalog\\catalog-partition-deadbeef.json", + "catalog/other.json", + "catalog/catalog-partition-.json", + "catalog/catalog-partition-deadbeef.json\n", + ] { + assert!( + validate_catalog_partition_artifact_path(path).is_err(), + "{path}" + ); + } + validate_task_capability("btc_5m_backtest").expect("supported syntax"); + for capability in ["", "BTC_5m_backtest", "btc 5m", "btc/5m", "_btc_5m"] { + assert!( + validate_task_capability(capability).is_err(), + "{capability}" + ); + } + } + #[test] fn status_derives_only_kubernetes_milestones_without_evidence() { let status = derive_status(&status_job(), &status_pods(), None).unwrap(); @@ -1494,6 +1662,7 @@ mod tests { mission_sha256: "c".repeat(64), snapshot_url: "https://oss-internal/snapshots/snapshot.zip?signature=x".to_owned(), snapshot_sha256: "d".repeat(64), + snapshot_contract_id: format!("sha256:{}", "1".repeat(64)), result_put_url: "https://oss-internal/results/btc-5m-attempt-001/results.zip?signature=x".to_owned(), llm_secret_name: "monday-prediction-llm".to_owned(), @@ -1514,8 +1683,8 @@ mod tests { AdmittedSubmission { validated: validate_submission(submission).expect("valid test submission"), admission: SnapshotAdmission { - snapshot_contract_id: format!("sha256:{}", "3".repeat(64)), - snapshot_digest: format!("sha256:{}", "d".repeat(64)), + snapshot_contract_id: format!("sha256:{}", "1".repeat(64)), + snapshot_digest: "0123456789abcdef".to_owned(), partition_digest: format!("sha256:{}", "e".repeat(64)), policy_identity: format!("sha256:{}", "f".repeat(64)), task_capability: "btc_5m_backtest".to_owned(), diff --git a/rust_hft/alpha-harness/app/src/prediction_runner.rs b/rust_hft/alpha-harness/app/src/prediction_runner.rs index d157cd6a0..2a7733ce5 100644 --- a/rust_hft/alpha-harness/app/src/prediction_runner.rs +++ b/rust_hft/alpha-harness/app/src/prediction_runner.rs @@ -32,6 +32,13 @@ struct PredictionMissionIdentity { search_policy_snapshot_id: String, } +#[derive(Debug, Deserialize)] +struct PredictionSnapshotIdentity { + schema_version: String, + snapshot_hash: String, + snapshot_contract_hash: String, +} + #[derive(Debug, Serialize)] struct PredictionExecutionEvidence<'a> { lane: &'static str, @@ -104,6 +111,7 @@ fn execute_with_runner(args: PredictionExecuteArgs, runner: &Path) -> anyhow::Re ) })?; extract_archive(&snapshot_archive, snapshot_dir.path(), None)?; + verify_admitted_snapshot_identity(&args, &mission, snapshot_dir.path())?; let resume_bundle_sha256 = if let Some((resume_url, resume_sha256)) = resume_source(&args)? { let resume_archive = input_dir.join("resume.zip"); @@ -176,6 +184,8 @@ fn validate_execute_args(args: &PredictionExecuteArgs) -> anyhow::Result<()> { args.mission_sha256.as_str(), args.snapshot_url.as_str(), args.snapshot_sha256.as_str(), + args.snapshot_contract_id.as_str(), + args.snapshot_digest.as_str(), args.result_put_url.as_str(), ] .iter() @@ -183,6 +193,8 @@ fn validate_execute_args(args: &PredictionExecuteArgs) -> anyhow::Result<()> { { bail!("prediction execution paths, URLs, and hashes are required"); } + immutable_sha256_identity("prediction snapshot contract", &args.snapshot_contract_id)?; + validate_snapshot_digest(&args.snapshot_digest)?; resume_source(args)?; Ok(()) } @@ -254,6 +266,47 @@ fn validate_mission_identity(mission: &PredictionMissionIdentity) -> anyhow::Res Ok(()) } +fn verify_admitted_snapshot_identity( + args: &PredictionExecuteArgs, + mission: &PredictionMissionIdentity, + snapshot_dir: &Path, +) -> anyhow::Result<()> { + let snapshot: PredictionSnapshotIdentity = serde_json::from_slice( + &std::fs::read(snapshot_dir.join("manifest.json")) + .context("read extracted prediction snapshot manifest")?, + ) + .context("prediction snapshot manifest identity is invalid JSON")?; + if snapshot.schema_version != "research_snapshot_v2" + || mission.data_snapshot_id != args.snapshot_contract_id + || snapshot.snapshot_contract_hash != args.snapshot_contract_id + || snapshot.snapshot_hash != args.snapshot_digest + { + bail!("prediction mission, admitted snapshot contract, and snapshot manifest do not match"); + } + Ok(()) +} + +fn immutable_sha256_identity(label: &str, value: &str) -> anyhow::Result<()> { + let digest = value + .strip_prefix("sha256:") + .with_context(|| format!("{label} must use sha256:<64 lowercase hex>"))?; + if value != format!("sha256:{digest}") || digest != normalized_sha256(label, digest)? { + bail!("{label} must use sha256:<64 lowercase hex>"); + } + Ok(()) +} + +fn validate_snapshot_digest(value: &str) -> anyhow::Result<()> { + if value.len() != 16 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')) + { + bail!("prediction snapshot digest must be exactly 16 lowercase ASCII hex characters"); + } + Ok(()) +} + fn ensure_empty_results_directory(results_dir: &Path) -> anyhow::Result<()> { data_mission::ensure_real_directory(results_dir, "prediction results")?; if std::fs::read_dir(results_dir)? @@ -365,6 +418,29 @@ mod tests { std::fs::remove_dir_all(fixture.root).unwrap(); } + #[cfg(unix)] + #[test] + fn execute_rejects_admitted_contract_mismatch_before_starting_runner() { + use std::os::unix::fs::PermissionsExt; + + let fixture = execute_fixture("admitted-contract-mismatch"); + let marker = fixture.root.join("runner-started"); + let runner = fixture.root.join("must-not-start"); + std::fs::write(&runner, format!("#!/bin/sh\ntouch {}\n", marker.display())).unwrap(); + std::fs::set_permissions(&runner, std::fs::Permissions::from_mode(0o700)).unwrap(); + let mut args = fixture.args; + args.snapshot_contract_id = format!("sha256:{}", "2".repeat(64)); + + let error = execute_with_runner(args, &runner) + .expect_err("mismatched admitted contract must prevent runner execution"); + + assert!(error.to_string().contains( + "prediction mission, admitted snapshot contract, and snapshot manifest do not match" + )); + assert!(!marker.exists()); + std::fs::remove_dir_all(fixture.root).unwrap(); + } + #[cfg(unix)] #[test] fn execute_reuses_verified_snapshot_cache_across_isolated_attempts() { @@ -558,7 +634,7 @@ mod tests { let resumed_runner = fixture.root.join("resumed-runner"); std::fs::write( &resumed_runner, - "#!/bin/sh\ntest \"$(cat \"$2/manifest.json\")\" = '{}' || exit 4\nprintf '{\"status\":\"budget_exhausted\"}\\n' > \"$3/summary.json\"\n", + "#!/bin/sh\ntest -f \"$2/manifest.json\" || exit 4\nprintf '{\"status\":\"budget_exhausted\"}\\n' > \"$3/summary.json\"\n", ) .unwrap(); std::fs::set_permissions(&resumed_runner, std::fs::Permissions::from_mode(0o700)).unwrap(); @@ -691,10 +767,12 @@ mod tests { fn execute_fixture(name: &str) -> ExecuteFixture { let root = temporary_root(name); let mission_path = root.join("mission.json"); + let snapshot_contract_id = format!("sha256:{}", "1".repeat(64)); + let snapshot_digest = "0123456789abcdef"; let mission = serde_json::json!({ "mission_id": "prediction-test", "lane": "prediction_market", - "data_snapshot_id": "sha256:snapshot-contract", + "data_snapshot_id": snapshot_contract_id, "search_policy_snapshot_id": "sha256:evaluator-version" }); let mission_bytes = serde_json::to_vec(&mission).unwrap(); @@ -704,7 +782,17 @@ mod tests { archive .start_file("manifest.json", SimpleFileOptions::default()) .unwrap(); - archive.write_all(b"{}\n").unwrap(); + archive + .write_all( + serde_json::to_string(&serde_json::json!({ + "schema_version": "research_snapshot_v2", + "snapshot_hash": snapshot_digest, + "snapshot_contract_hash": snapshot_contract_id, + })) + .unwrap() + .as_bytes(), + ) + .unwrap(); archive.finish().unwrap(); let result_path = root.join("published.zip"); let args = PredictionExecuteArgs { @@ -713,6 +801,8 @@ mod tests { mission_sha256: format!("{:x}", Sha256::digest(&mission_bytes)), snapshot_url: snapshot_path.to_string_lossy().into_owned(), snapshot_sha256: sha256_file(&snapshot_path).unwrap(), + snapshot_contract_id, + snapshot_digest: snapshot_digest.to_owned(), snapshot_cache_dir: None, resume_url: None, resume_sha256: None,