diff --git a/deployment/aliyun/research/README.md b/deployment/aliyun/research/README.md index df097d870..df2dc202d 100644 --- a/deployment/aliyun/research/README.md +++ b/deployment/aliyun/research/README.md @@ -275,6 +275,12 @@ Submit with `alpha-harness prediction dispatch submit --submission FILE --contex CONTEXT --namespace NS`. The query-free result URL is the duplicate guard; each Job has isolated storage. Treat rendered Secret `stringData` as sensitive. +Read Job and Pod milestones without mutation using `alpha-harness prediction +dispatch status --context CONTEXT --namespace NS --job-name JOB`. Snapshot-ready +and evaluator-started remain `null` unless `--evidence execution-evidence.json` +is supplied. Evidence is accepted only when its mission ID, mission SHA, and +snapshot SHA match the immutable Job annotations; a mismatch fails closed. + The URL Secret must always contain `resume-url` and `resume-sha256`; set both to empty strings for the first attempt. A paused or failed runner still uploads its results and append-only state to the attempt's immutable result URL before the diff --git a/rust_hft/alpha-harness/app/src/cli.rs b/rust_hft/alpha-harness/app/src/cli.rs index 64d256e50..682dbfe14 100644 --- a/rust_hft/alpha-harness/app/src/cli.rs +++ b/rust_hft/alpha-harness/app/src/cli.rs @@ -102,6 +102,7 @@ enum PredictionCommand { #[derive(Debug, Subcommand)] enum PredictionDispatchCommand { Render(PredictionDispatchRenderArgs), + Status(PredictionDispatchStatusArgs), Submit(PredictionDispatchSubmitArgs), } @@ -123,6 +124,18 @@ pub struct PredictionDispatchSubmitArgs { pub namespace: String, } +#[derive(Debug, Clone, Args)] +pub struct PredictionDispatchStatusArgs { + #[arg(long)] + pub context: String, + #[arg(long)] + pub namespace: String, + #[arg(long)] + pub job_name: String, + #[arg(long)] + pub evidence: Option, +} + #[derive(Debug, Clone, Args)] pub struct PredictionExecuteArgs { #[arg(long)] @@ -649,6 +662,7 @@ pub async fn run(cli: Cli) -> anyhow::Result<()> { } PredictionCommand::Dispatch { command } => match command { PredictionDispatchCommand::Render(args) => prediction_dispatch::render(args), + PredictionDispatchCommand::Status(args) => prediction_dispatch::status(args), PredictionDispatchCommand::Submit(args) => prediction_dispatch::submit(args), }, }, @@ -809,6 +823,12 @@ mod tests { assert!(Cli::try_parse_from(args.split_whitespace()).is_ok()); } + #[test] + fn parses_prediction_dispatch_status_with_explicit_cluster_identity() { + let args = "alpha-harness prediction dispatch status --context ack --namespace monday-research --job-name prediction-job"; + assert!(Cli::try_parse_from(args.split_whitespace()).is_ok()); + } + #[cfg(unix)] #[tokio::test(flavor = "current_thread")] async fn prediction_snapshot_runs_blocking_pipeline_outside_async_runtime() { diff --git a/rust_hft/alpha-harness/app/src/prediction_dispatch.rs b/rust_hft/alpha-harness/app/src/prediction_dispatch.rs index 8de0b2815..7981b060a 100644 --- a/rust_hft/alpha-harness/app/src/prediction_dispatch.rs +++ b/rust_hft/alpha-harness/app/src/prediction_dispatch.rs @@ -1,12 +1,16 @@ use crate::{ - cli::{print_json, PredictionDispatchRenderArgs, PredictionDispatchSubmitArgs}, + cli::{ + print_json, PredictionDispatchRenderArgs, PredictionDispatchStatusArgs, + PredictionDispatchSubmitArgs, + }, mission_runner::normalized_sha256, }; use anyhow::{bail, Context}; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; use std::{ + collections::BTreeMap, ffi::OsString, fs::File, io::{Read, Write}, @@ -61,6 +65,106 @@ struct RenderedSubmission { result_identity_sha256: String, } +#[derive(Deserialize)] +struct StatusJob { + metadata: StatusMetadata, + #[serde(default)] + status: StatusJobState, +} + +#[derive(Deserialize)] +struct StatusMetadata { + name: String, + namespace: String, + uid: String, + #[serde(default)] + annotations: BTreeMap, +} + +#[derive(Default, Deserialize)] +struct StatusJobState { + #[serde(default)] + conditions: Vec, +} + +#[derive(Deserialize)] +struct StatusPodList { + items: Vec, +} + +#[derive(Deserialize)] +struct StatusPod { + metadata: StatusPodMetadata, + #[serde(default)] + status: StatusPodState, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct StatusPodMetadata { + #[serde(default)] + owner_references: Vec, +} + +#[derive(Deserialize)] +struct StatusOwnerReference { + uid: String, + name: String, + kind: String, + #[serde(default)] + controller: bool, +} + +#[derive(Default, Deserialize)] +#[serde(rename_all = "camelCase")] +struct StatusPodState { + #[serde(default)] + conditions: Vec, + #[serde(default)] + container_statuses: Vec, +} + +#[derive(Deserialize)] +struct StatusCondition { + #[serde(rename = "type")] + kind: String, + status: String, +} + +#[derive(Deserialize)] +struct StatusContainer { + state: StatusContainerState, +} + +#[derive(Deserialize)] +struct StatusContainerState { + running: Option, + terminated: Option, +} + +#[derive(Deserialize)] +struct StatusEvidence { + lane: String, + mission_id: String, + mission_sha256: String, + snapshot_archive_sha256: String, +} + +#[derive(Serialize)] +struct PredictionStatus { + job_name: String, + namespace: String, + mission_id: String, + mission_sha256: String, + snapshot_sha256: String, + submitted: bool, + scheduled: bool, + image_ready: bool, + snapshot_ready: Option, + evaluator_started: Option, + completed: bool, +} + pub fn render(args: PredictionDispatchRenderArgs) -> anyhow::Result<()> { let submission = load_submission(&args.submission)?; let rendered = render_submission(submission, &args.namespace)?; @@ -135,6 +239,136 @@ pub fn submit(args: PredictionDispatchSubmitArgs) -> anyhow::Result<()> { })) } +pub fn status(args: PredictionDispatchStatusArgs) -> anyhow::Result<()> { + validate_cluster_target(&args.context, &args.namespace)?; + validate_dns_label("prediction Job name", &args.job_name)?; + let job = kubectl_json( + &args.context, + &args.namespace, + ["get", "job", &args.job_name, "-o", "json"], + "read prediction Job status", + )?; + let job_uid = status_job_uid(&job)?; + let selector = format!("batch.kubernetes.io/controller-uid={job_uid}"); + let pods = kubectl_json( + &args.context, + &args.namespace, + ["get", "pods", "-l", &selector, "-o", "json"], + "read prediction Pod status", + )?; + let evidence = args + .evidence + .as_deref() + .map(load_status_evidence) + .transpose()?; + let derived = derive_status(&job, &pods, evidence.as_ref())?; + if derived.job_name != args.job_name || derived.namespace != args.namespace { + bail!("Kubernetes Job readback does not match the requested immutable identity"); + } + print_json(&json!({"context": args.context, "status": derived})) +} + +fn status_job_uid(job: &Value) -> anyhow::Result { + let job: StatusJob = + serde_json::from_value(job.clone()).context("parse prediction Job readback")?; + validate_identifier("prediction Job UID", &job.metadata.uid)?; + Ok(job.metadata.uid) +} + +fn load_status_evidence(path: &Path) -> anyhow::Result { + let file = File::open(path) + .with_context(|| format!("open prediction execution evidence {}", path.display()))?; + if file.metadata()?.len() > MAX_SUBMISSION_BYTES { + bail!("prediction execution evidence exceeds {MAX_SUBMISSION_BYTES} bytes"); + } + let mut bytes = Vec::new(); + file.take(MAX_SUBMISSION_BYTES + 1) + .read_to_end(&mut bytes)?; + if bytes.len() as u64 > MAX_SUBMISSION_BYTES { + bail!("prediction execution evidence exceeds {MAX_SUBMISSION_BYTES} bytes"); + } + serde_json::from_slice(&bytes) + .with_context(|| format!("parse prediction execution evidence {}", path.display())) +} + +fn derive_status( + job: &Value, + pods: &Value, + evidence: Option<&Value>, +) -> anyhow::Result { + let job: StatusJob = + serde_json::from_value(job.clone()).context("parse prediction Job readback")?; + let pods: StatusPodList = + serde_json::from_value(pods.clone()).context("parse prediction Pod readback")?; + let annotation = |key: &str| { + job.metadata + .annotations + .get(key) + .map(String::as_str) + .with_context(|| format!("prediction Job is missing {key} annotation")) + }; + if annotation("research.monday/lane")? != "prediction_market" { + bail!("Kubernetes Job is not a prediction research Job"); + } + let mission_id = annotation("research.monday/mission-id")?.to_owned(); + validate_identifier("Job mission id", &mission_id)?; + let mission_sha256 = + normalized_sha256("Job mission", annotation("research.monday/mission-sha256")?)?; + let snapshot_sha256 = normalized_sha256( + "Job snapshot", + annotation("research.monday/snapshot-sha256")?, + )?; + let research_ready = if let Some(evidence) = evidence { + let evidence: StatusEvidence = serde_json::from_value(evidence.clone()) + .context("parse prediction execution evidence")?; + if evidence.lane != "prediction_market" + || evidence.mission_id != mission_id + || normalized_sha256("evidence mission", &evidence.mission_sha256)? != mission_sha256 + || normalized_sha256("evidence snapshot", &evidence.snapshot_archive_sha256)? + != snapshot_sha256 + { + bail!("prediction execution evidence does not match the immutable Job identity"); + } + Some(true) + } else { + None + }; + let condition = |conditions: &[StatusCondition], kind: &str| { + conditions + .iter() + .any(|condition| condition.kind == kind && condition.status == "True") + }; + let owned_pods = pods.items.iter().filter(|pod| { + pod.metadata.owner_references.iter().any(|owner| { + owner.controller + && owner.kind == "Job" + && owner.name == job.metadata.name + && owner.uid == job.metadata.uid + }) + }); + let scheduled = owned_pods + .clone() + .any(|pod| condition(&pod.status.conditions, "PodScheduled")); + let image_ready = owned_pods.clone().any(|pod| { + pod.status.container_statuses.iter().any(|container| { + container.state.running.is_some() || container.state.terminated.is_some() + }) + }); + Ok(PredictionStatus { + job_name: job.metadata.name, + namespace: job.metadata.namespace, + mission_id, + mission_sha256, + snapshot_sha256, + submitted: true, + scheduled, + image_ready, + snapshot_ready: research_ready, + evaluator_started: research_ready, + completed: condition(&job.status.conditions, "Complete"), + }) +} + fn load_submission(path: &Path) -> anyhow::Result { let file = File::open(path) .with_context(|| format!("open prediction submission {}", path.display()))?; @@ -677,6 +911,74 @@ mod tests { } } + #[test] + fn status_derives_only_kubernetes_milestones_without_evidence() { + let status = derive_status(&status_job(), &status_pods(), None).unwrap(); + assert!(status.submitted && status.scheduled && status.image_ready && status.completed); + assert_eq!(status.snapshot_ready, None); + assert_eq!(status.evaluator_started, None); + } + + #[test] + fn matching_execution_evidence_asserts_research_milestones() { + let evidence = status_evidence(); + let status = derive_status(&status_job(), &status_pods(), Some(&evidence)).unwrap(); + assert_eq!(status.snapshot_ready, Some(true)); + assert_eq!(status.evaluator_started, Some(true)); + } + + #[test] + fn status_rejects_evidence_from_another_mission_or_snapshot() { + for field in ["mission_id", "mission_sha256", "snapshot_archive_sha256"] { + let mut evidence = status_evidence(); + evidence[field] = if field == "mission_id" { + json!("mission-2") + } else { + json!("e".repeat(64)) + }; + assert!(derive_status(&status_job(), &status_pods(), Some(&evidence)).is_err()); + } + } + + #[test] + fn pod_from_another_job_cannot_advance_milestones() { + let mut pods = status_pods(); + pods["items"][0]["metadata"]["ownerReferences"][0]["uid"] = json!("other-uid"); + let status = derive_status(&status_job(), &pods, None).unwrap(); + assert!(!status.scheduled && !status.image_ready); + } + + fn status_job() -> Value { + json!({ + "metadata": {"name": "prediction-job", "namespace": "monday-research", "uid": "job-uid", "annotations": { + "research.monday/lane": "prediction_market", + "research.monday/mission-id": "mission-1", + "research.monday/mission-sha256": "c".repeat(64), + "research.monday/snapshot-sha256": "d".repeat(64) + }}, + "status": {"conditions": [{"type": "Complete", "status": "True"}]} + }) + } + + fn status_pods() -> Value { + json!({"items": [{"metadata": {"ownerReferences": [{ + "uid": "job-uid", "name": "prediction-job", "kind": "Job", "controller": true + }]}, "status": { + "conditions": [{"type": "PodScheduled", "status": "True"}], + "containerStatuses": [{"state": {"running": {}}}] + }}]}) + } + + fn status_evidence() -> Value { + json!({ + "lane": "prediction_market", + "mission_id": "mission-1", + "mission_sha256": "c".repeat(64), + "snapshot_archive_sha256": "d".repeat(64), + "runner_exit_code": 0 + }) + } + fn valid_submission() -> PredictionSubmission { PredictionSubmission { attempt_id: "btc-5m-attempt-001".to_owned(),