From 93fb41c6089b7e1e596aa377681e5e9fc7fd962e Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 9 Oct 2026 03:34:04 +0800 Subject: [PATCH] feat(engine): judge immutable submissions through bounded execution ports --- judge/Cargo.lock | 1 + judge/crates/engine/Cargo.toml | 3 + judge/crates/engine/src/lib.rs | 2 + judge/crates/engine/src/submission.rs | 530 ++++++++++++++++++++++++++ 4 files changed, 536 insertions(+) create mode 100644 judge/crates/engine/src/submission.rs diff --git a/judge/Cargo.lock b/judge/Cargo.lock index 1587721..cf8799e 100644 --- a/judge/Cargo.lock +++ b/judge/Cargo.lock @@ -378,6 +378,7 @@ version = "0.0.1" dependencies = [ "kf-contracts", "serde", + "serde_json", ] [[package]] diff --git a/judge/crates/engine/Cargo.toml b/judge/crates/engine/Cargo.toml index 7df5187..b9c5e94 100644 --- a/judge/crates/engine/Cargo.toml +++ b/judge/crates/engine/Cargo.toml @@ -11,3 +11,6 @@ serde.workspace = true [lints] workspace = true + +[dev-dependencies] +serde_json.workspace = true diff --git a/judge/crates/engine/src/lib.rs b/judge/crates/engine/src/lib.rs index 8f23a1e..b6c2570 100644 --- a/judge/crates/engine/src/lib.rs +++ b/judge/crates/engine/src/lib.rs @@ -227,3 +227,5 @@ mod tests { assert_eq!(p.calls, 0); } } + +pub mod submission; diff --git a/judge/crates/engine/src/submission.rs b/judge/crates/engine/src/submission.rs new file mode 100644 index 0000000..a3e277b --- /dev/null +++ b/judge/crates/engine/src/submission.rs @@ -0,0 +1,530 @@ +//! One immutable submission, one compilation, bounded sequential test points. +//! Ports own clocks/cancellation/storage/execution; this module is pure domain. +use crate::{ExecutionKind, Observation}; +use kf_contracts::{SchemaVersion, judging::*}; +use serde::Serialize; +#[derive(Debug)] +pub struct StageFailure { + pub code: FailureCode, + pub cleanup_complete: bool, +} +pub struct Compiled { + pub report: CompilationReport, + pub cleanup_complete: bool, + pub failure: Option, +} +pub trait CompilationPort { + fn compile(&mut self, manifest: &JobManifest, run_id: &str) -> Result; +} +pub trait RunPort { + fn run( + &mut self, + program: &ArtifactReference, + case: &TestCase, + limits: &RunLimits, + run_id: &str, + ) -> Result; +} +pub trait ExpectedPort { + fn expected(&mut self, case: &TestCase) -> Result, StageFailure>; +} +pub trait ControlPort { + fn elapsed_us(&self) -> u64; + fn cancelled(&self) -> bool; + fn stage(&mut self, stage: Stage, case_index: Option) -> Result<(), StageFailure>; +} +#[derive(Debug, Serialize)] +pub struct SkippedCase { + pub case_id: Identifier, + pub reason: SkipReason, +} +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum SkipReason { + CompilationFailed, + Cancelled, + InfrastructureFailure, +} +#[derive(Debug, Serialize)] +pub struct SubmissionResult { + pub report: JudgementReport, + pub compile_run_id: String, + pub case_run_ids: Vec, + pub skipped: Vec, +} +fn empty_compile(status: CompilationStatus) -> CompilationReport { + CompilationReport { + status, + program: None, + diagnostics: String::new().try_into().expect("empty diagnostics"), + diagnostics_truncated: false, + cpu_time_us: None, + memory_peak_bytes: None, + } +} +fn classify(o: &Observation, expected: &[u8]) -> Verdict { + match o.kind { + ExecutionKind::Exited { code: 0 } => { + if o.stdout == expected { + Verdict::Accepted + } else { + Verdict::WrongAnswer + } + } + ExecutionKind::Exited { .. } | ExecutionKind::Signaled | ExecutionKind::OtherLimit => { + Verdict::RuntimeError + } + ExecutionKind::TimeLimit => Verdict::TimeLimitExceeded, + ExecutionKind::MemoryLimit => Verdict::MemoryLimitExceeded, + ExecutionKind::OutputLimit => Verdict::OutputLimitExceeded, + ExecutionKind::Cancelled => Verdict::Cancelled, + ExecutionKind::InfrastructureFailure => Verdict::SystemError, + } +} +fn failure(code: FailureCode) -> StageFailure { + StageFailure { + code, + cleanup_complete: true, + } +} +fn gate(p: &impl ControlPort, m: &JobManifest) -> Result { + if p.elapsed_us() >= m.job_wall_time_us { + return Err(failure(FailureCode::JobDeadline)); + } + Ok(p.cancelled()) +} +pub fn judge_submission( + p: &mut P, + m: &JobManifest, + binding: LeaseBinding, + digest: Digest, +) -> Result { + m.validate()?; + binding.validate()?; + if binding.evaluation_id != m.evaluation_id { + return Err("submission binding differs"); + } + let compile_run_id = format!("c-{}", binding.attempt_id.as_uuid()); + let compilation = match gate(p, m) { + Ok(true) => Ok(Compiled { + report: empty_compile(CompilationStatus::Cancelled), + cleanup_complete: true, + failure: None, + }), + Err(e) => Err(e), + Ok(false) => p + .stage(Stage::Compiling, None) + .and_then(|()| p.compile(m, &compile_run_id)), + }; + let mut cleanup_complete; + let mut failure_code; + let compilation = match compilation { + Ok(c) => { + c.report.validate()?; + cleanup_complete = c.cleanup_complete; + failure_code = c.failure; + if !cleanup_complete { + failure_code = Some(FailureCode::CleanupFailed); + } + if failure_code.is_some() { + empty_compile(CompilationStatus::SystemError) + } else { + c.report + } + } + Err(e) => { + cleanup_complete = e.cleanup_complete; + failure_code = Some(if cleanup_complete { + e.code + } else { + FailureCode::CleanupFailed + }); + empty_compile(CompilationStatus::SystemError) + } + }; + let mut cases = Vec::with_capacity(m.cases.len()); + let mut run_ids = Vec::with_capacity(m.cases.len()); + if let Some(program) = &compilation.program { + for (index, c) in m.cases.iter().enumerate() { + match gate(p, m) { + Ok(true) => break, + Err(e) => { + failure_code = Some(e.code); + break; + } + Ok(false) => {} + } + let run_id = format!("r-{}-{index}", binding.attempt_id.as_uuid()); + let observation = p + .stage(Stage::Running, Some(index as u16)) + .and_then(|()| p.run(program, c, &m.run_limits, &run_id)); + run_ids.push(run_id); + let result = match observation { + Err(e) => { + cleanup_complete = e.cleanup_complete; + failure_code = Some(if cleanup_complete { + e.code + } else { + FailureCode::CleanupFailed + }); + CaseReport { + case_id: c.case_id, + verdict: Verdict::SystemError, + cpu_time_us: None, + memory_peak_bytes: None, + } + } + Ok(o) => { + let verdict = if matches!(o.kind, ExecutionKind::Exited { code: 0 }) { + match gate(p, m) { + Err(e) => { + failure_code = Some(e.code); + Verdict::SystemError + } + Ok(true) => Verdict::Cancelled, + Ok(false) => match p.expected(c) { + Ok(bytes) + if bytes.len() as u64 == c.expected.bytes + && bytes.len() as u64 <= m.run_limits.stdout_bytes => + { + classify(&o, &bytes) + } + Ok(_) => { + failure_code = Some(FailureCode::ArtifactUnavailable); + Verdict::SystemError + } + Err(e) => { + failure_code = Some(e.code); + Verdict::SystemError + } + }, + } + } else { + classify(&o, &[]) + }; + if verdict == Verdict::SystemError && failure_code.is_none() { + failure_code = Some(FailureCode::SandboxUnavailable); + } + CaseReport { + case_id: c.case_id, + verdict, + cpu_time_us: o.cpu_time_us, + memory_peak_bytes: o.memory_peak_bytes, + } + } + }; + let stop = matches!(result.verdict, Verdict::SystemError | Verdict::Cancelled); + cases.push(result); + if stop { + break; + } + } + } + let verdict = if failure_code.is_some() { + Verdict::SystemError + } else { + match compilation.status { + CompilationStatus::CompilationError | CompilationStatus::CompilationLimit => { + Verdict::CompilationError + } + CompilationStatus::Cancelled => Verdict::Cancelled, + CompilationStatus::SystemError => { + failure_code = Some(FailureCode::SandboxUnavailable); + Verdict::SystemError + } + CompilationStatus::Succeeded => { + if cases.len() != m.cases.len() + || cases.iter().any(|c| c.verdict == Verdict::Cancelled) + { + Verdict::Cancelled + } else { + cases + .iter() + .find(|c| c.verdict != Verdict::Accepted) + .map_or(Verdict::Accepted, |c| c.verdict) + } + } + } + }; + let cpu = if !cases.is_empty() && cases.iter().all(|c| c.cpu_time_us.is_some()) { + Some( + cases + .iter() + .try_fold(0_u64, |n, c| n.checked_add(c.cpu_time_us.unwrap_or(0))) + .ok_or("CPU sum overflow")?, + ) + } else { + None + }; + let memory = if !cases.is_empty() && cases.iter().all(|c| c.memory_peak_bytes.is_some()) { + cases.iter().filter_map(|c| c.memory_peak_bytes).max() + } else { + None + }; + let reason = if verdict == Verdict::Cancelled { + SkipReason::Cancelled + } else if verdict == Verdict::CompilationError { + SkipReason::CompilationFailed + } else { + SkipReason::InfrastructureFailure + }; + let skipped = m.cases[cases.len()..] + .iter() + .map(|c| SkippedCase { + case_id: c.case_id, + reason, + }) + .collect(); + let report = JudgementReport { + schema_version: SchemaVersion, + kind: ResultKind::Value, + binding, + manifest_digest: digest, + compilation, + cases, + verdict, + cleanup_complete, + failure_code, + run_cpu_time_us: cpu, + run_memory_peak_bytes: memory, + }; + report.validate_for_manifest(m, &report.manifest_digest)?; + Ok(SubmissionResult { + report, + compile_run_id, + case_run_ids: run_ids, + skipped, + }) +} +#[cfg(test)] +mod tests { + use super::*; + struct Port { + observed: Vec, + calls: usize, + compile_calls: usize, + expected_calls: usize, + cancel_after: usize, + clock: u64, + fail_compile: bool, + } + impl CompilationPort for Port { + fn compile(&mut self, _: &JobManifest, _: &str) -> Result { + self.compile_calls += 1; + Ok(Compiled { + report: CompilationReport { + status: if self.fail_compile { + CompilationStatus::CompilationError + } else { + CompilationStatus::Succeeded + }, + program: (!self.fail_compile).then(|| ArtifactReference { + digest: "a".repeat(64).try_into().unwrap(), + bytes: 1, + }), + diagnostics: String::new().try_into().unwrap(), + diagnostics_truncated: false, + cpu_time_us: Some(5), + memory_peak_bytes: Some(10), + }, + cleanup_complete: true, + failure: None, + }) + } + } + impl RunPort for Port { + fn run( + &mut self, + _: &ArtifactReference, + _: &TestCase, + _: &RunLimits, + _: &str, + ) -> Result { + self.calls += 1; + Ok(self.observed.remove(0)) + } + } + impl ExpectedPort for Port { + fn expected(&mut self, c: &TestCase) -> Result, StageFailure> { + self.expected_calls += 1; + Ok(vec![0; c.expected.bytes as usize]) + } + } + impl ControlPort for Port { + fn elapsed_us(&self) -> u64 { + self.clock + } + fn cancelled(&self) -> bool { + self.calls >= self.cancel_after + } + fn stage(&mut self, _: Stage, _: Option) -> Result<(), StageFailure> { + Ok(()) + } + } + fn fixture() -> (JobManifest, LeaseBinding, Digest) { + let m: JobManifest = serde_json::from_str(include_str!( + "../../../../tests/fixtures/contracts/judge-manifest.valid.json" + )) + .unwrap(); + let mut value: serde_json::Value = serde_json::from_str(include_str!( + "../../../../tests/fixtures/contracts/judge-result.valid.json" + )) + .unwrap(); + let b: LeaseBinding = serde_json::from_value(value["binding"].take()).unwrap(); + (m, b, "b".repeat(64).try_into().unwrap()) + } + fn port(kinds: Vec) -> Port { + Port { + observed: kinds + .into_iter() + .map(|kind| Observation { + kind, + stdout: vec![], + cpu_time_us: Some(3), + memory_peak_bytes: None, + }) + .collect(), + calls: 0, + compile_calls: 0, + expected_calls: 0, + cancel_after: usize::MAX, + clock: 0, + fail_compile: false, + } + } + #[test] + fn compile_once_order_exact_bytes_and_measurements() { + let (mut m, b, d) = fixture(); + for c in &mut m.cases { + c.expected.bytes = 0; + } + let mut p = port(vec![ExecutionKind::Exited { code: 0 }; m.cases.len()]); + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(p.compile_calls, 1); + assert_eq!(p.calls, m.cases.len()); + assert_eq!(r.report.verdict, Verdict::Accepted); + assert_eq!(r.report.run_cpu_time_us, Some(3 * m.cases.len() as u64)); + assert_eq!(r.report.run_memory_peak_bytes, None); + assert!(r.skipped.is_empty()); + assert_eq!(r.report.compilation.cpu_time_us, Some(5)); + } + #[test] + fn cancellation_and_ce_skip_without_faking_cases() { + let (m, b, d) = fixture(); + let mut p = port(vec![]); + p.fail_compile = true; + let r = judge_submission(&mut p, &m, b.clone(), d.clone()).unwrap(); + assert_eq!(r.report.verdict, Verdict::CompilationError); + assert_eq!(r.skipped.len(), m.cases.len()); + assert_eq!(p.calls, 0); + p.fail_compile = false; + p.cancel_after = 0; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.verdict, Verdict::Cancelled); + assert_eq!(p.compile_calls, 1); + } + #[test] + fn nonzero_does_not_read_or_compare_expected() { + let (m, b, d) = fixture(); + let mut p = port(vec![ExecutionKind::Exited { code: 7 }; m.cases.len()]); + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.verdict, Verdict::RuntimeError); + assert_eq!(p.expected_calls, 0); + } + #[test] + fn infrastructure_and_deadline_override_prior_wrong_answer() { + let (m, b, d) = fixture(); + let mut p = port(vec![ExecutionKind::InfrastructureFailure]); + let r = judge_submission(&mut p, &m, b.clone(), d.clone()).unwrap(); + assert_eq!(p.calls, 1); + assert_eq!(r.report.failure_code, Some(FailureCode::SandboxUnavailable)); + p.clock = m.job_wall_time_us; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.failure_code, Some(FailureCode::JobDeadline)); + assert!(r.report.cases.is_empty()); + } + #[test] + fn byte_boundaries_and_nonzero_precedence() { + for (out, expected, v) in [ + (vec![], vec![], Verdict::Accepted), + (vec![0, 255, 10], vec![0, 255, 10], Verdict::Accepted), + (vec![0, 255, 10], vec![0, 255], Verdict::WrongAnswer), + (vec![10], vec![], Verdict::WrongAnswer), + ] { + let o = Observation { + kind: ExecutionKind::Exited { code: 0 }, + stdout: out, + cpu_time_us: None, + memory_peak_bytes: None, + }; + assert_eq!(classify(&o, &expected), v); + } + } + #[test] + fn fatal_precedence_missing_metrics_and_midpoint_cancel() { + let (mut m, b, d) = fixture(); + for c in &mut m.cases { + c.expected.bytes = 0; + } + let mut p = port(vec![ + ExecutionKind::Exited { code: 0 }, + ExecutionKind::InfrastructureFailure, + ]); + p.observed[0].stdout = vec![1]; + let r = judge_submission(&mut p, &m, b.clone(), d.clone()).unwrap(); + assert_eq!(r.report.cases[0].verdict, Verdict::WrongAnswer); + assert_eq!(r.report.verdict, Verdict::SystemError); + assert_eq!(r.skipped.len(), 1); + let mut p = port(vec![ExecutionKind::Exited { code: 0 }; 3]); + p.cancel_after = 1; + let r = judge_submission(&mut p, &m, b.clone(), d.clone()).unwrap(); + assert_eq!(p.calls, 1); + assert_eq!(r.report.verdict, Verdict::Cancelled); + assert_eq!(r.skipped.len(), 2); + let mut p = port(vec![ExecutionKind::Exited { code: 0 }; 3]); + p.observed[1].cpu_time_us = None; + p.observed[0].memory_peak_bytes = Some(100); + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.verdict, Verdict::Accepted); + assert_eq!(r.report.run_cpu_time_us, None); + assert_eq!(r.report.run_memory_peak_bytes, None); + } + #[test] + fn maximum_point_count_is_bounded_and_run_ids_are_unique() { + let (mut m, b, d) = fixture(); + m.cases = (0..MAX_CASES) + .map(|i| TestCase { + case_id: format!("aaaaaaaa-aaaa-4aaa-8aaa-{:012}", 1000 + i) + .try_into() + .unwrap(), + input: m.cases[0].input.clone(), + expected: ArtifactReference { + digest: m.cases[0].expected.digest.clone(), + bytes: 0, + }, + }) + .collect(); + m.job_wall_time_us = 900_000_000; + let mut p = port(vec![ExecutionKind::Exited { code: 0 }; MAX_CASES]); + let r = judge_submission(&mut p, &m, b.clone(), d.clone()).unwrap(); + assert_eq!(p.compile_calls, 1); + assert_eq!(p.calls, MAX_CASES); + assert_eq!( + r.case_run_ids + .iter() + .collect::>() + .len(), + MAX_CASES + ); + m.cases.push(TestCase { + case_id: "aaaaaaaa-aaaa-4aaa-8aaa-000000002000" + .to_string() + .try_into() + .unwrap(), + input: m.cases[0].input.clone(), + expected: m.cases[0].expected.clone(), + }); + assert!(judge_submission(&mut p, &m, b, d).is_err()); + assert_eq!(p.compile_calls, 1); + } +}