feat(worker): judge versioned sealed runtime programs

backend/tests/test_coordinator.py

@@ -439,6 +439,7 @@ def test_actual_expiry_stopping_quarantine_retry_and_old_attempt(
                     break
                 await asyncio.sleep(0.05)
             assert pending[0]["binding"] == g["binding"]
+            assert pending == [dict(binding=g["binding"], case_count=2)]
             stopped = dict(
                 schema_version=1, kind="judge.stopped", binding=g["binding"], cleanup_complete=False
             )

backend/tests/test_gnu_image_builder.py

@@ -0,0 +1,101 @@
+"""Offline package data only; no compiler, submitted source or image execution."""
+
+import importlib.util
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+SPEC = importlib.util.spec_from_file_location(
+    "gnu_image_builder", Path(__file__).resolve().parents[2] / "scripts/build_gnu_image.py"
+)
+builder = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(builder)
+
+
+def test_absolute_package_link_is_relocated_inside_verified_image(tmp_path):
+    root = tmp_path / "image"
+    target = root / "usr/lib/plugin.so"
+    target.parent.mkdir(parents=True)
+    target.write_bytes(b"synthetic pinned package bytes")
+    link = root / "usr/lib/plugin-link.so"
+    link.symlink_to("/usr/lib/plugin.so")
+    records = {
+        "usr/lib/plugin-link.so": {"kind": "symlink", "target": "/usr/lib/plugin.so"},
+        "usr/lib/plugin.so": {"kind": "file", "sha256": builder.sha(target)},
+    }
+    source, record = builder.resolve_verified(root, records, "usr/lib/plugin-link.so")
+    assert source == target
+    assert source.read_bytes() == b"synthetic pinned package bytes"
+    assert record == records["usr/lib/plugin.so"]
+
+
+@pytest.mark.parametrize("target", ["../../../../outside", "plugin-link.so"])
+def test_package_link_escape_or_cycle_refused(tmp_path, target):
+    link = tmp_path / "usr/lib/plugin-link.so"
+    link.parent.mkdir(parents=True)
+    link.symlink_to(target)
+    records = {"usr/lib/plugin-link.so": {"kind": "symlink", "target": target}}
+    with pytest.raises(ValueError):
+        builder.resolve_verified(tmp_path, records, "usr/lib/plugin-link.so")
+
+
+def test_replaced_or_unverified_ancestor_link_is_refused(tmp_path):
+    library = tmp_path / "usr/lib"
+    library.parent.mkdir()
+    library.symlink_to("/usr/lib")
+    records = {"usr/lib/plugin.so": {"kind": "file", "sha256": "a" * 64}}
+    with pytest.raises(ValueError, match="unverified GNU ancestor"):
+        builder.resolve_verified(tmp_path, records, "usr/lib/plugin.so")
+    records["usr/lib"] = {"kind": "symlink", "target": "/different/path"}
+    with pytest.raises(ValueError, match="differs from verified manifest"):
+        builder.resolve_verified(tmp_path, records, "usr/lib/plugin.so")
+
+
+def test_verified_manifest_does_not_authorize_changed_package_bytes(tmp_path):
+    root = tmp_path / "support"
+    library = root / "usr/lib/ld-linux-x86-64.so.2"
+    library.parent.mkdir(parents=True)
+    library.write_bytes(b"original synthetic library")
+    manifest = {
+        "packages": [
+            {
+                "name": "gcc",
+                "version": "16.2.1+r23+gd564253eb6c8-1",
+                "sha256": "ebac12ceef494c616a36fc6f0f792b365dc59b6ed46ca1a69f138aecc810b6ab",
+            },
+            {
+                "name": "glibc",
+                "version": "2.44+r50+g1848099f063e-3",
+                "sha256": "071d807f4d3bf6b921f86b9a91e83a47ecc026c5821634efa46b09d890e06785",
+            },
+            {
+                "name": "binutils",
+                "version": "2.47-2",
+                "sha256": "62efd213e87590f025cf861b646793d3cacb03e7e26187b73d33b00adaf7a977",
+            },
+        ],
+        "files": [
+            {
+                "path": "usr/lib/ld-linux-x86-64.so.2",
+                "kind": "file",
+                "bytes": library.stat().st_size,
+                "sha256": builder.sha(library),
+            }
+        ],
+    }
+    manifest_path = tmp_path / "manifest.json"
+    manifest_path.write_text(json.dumps(manifest))
+    library.write_bytes(b"changed synthetic library")
+    output = tmp_path / "output"
+    args = SimpleNamespace(
+        support_dir=root,
+        support_manifest=manifest_path,
+        support_manifest_sha256=builder.sha(manifest_path),
+        output=output,
+    )
+    with pytest.raises(ValueError, match="differs from verified public package bytes"):
+        builder.build(args)
+    assert not (output / "manifest.json").exists()
+    assert not list(output.rglob("*.so*"))

backend/tests/test_multi_runtime_coordinator.py

@@ -209,6 +209,13 @@ def test_actual_v2_cancel_requires_matching_stop_protocol_and_fence(
                     headers={"Authorization": "Bearer " + platform},
                 )
             ).status_code == 204
+            pending = await client.get(
+                "/internal/pending-stops", headers=worker_headers(workers[0])
+            )
+            assert pending.status_code == 200
+            assert pending.json() == [
+                dict(schema_version=2, binding=grant["binding"], case_count=2)
+            ]
             stopped = dict(
                 schema_version=2,
                 kind="judge.stopped",

judge/crates/coordinator/src/repository.rs

@@ -460,8 +460,18 @@ impl Repository {
     }
     pub async fn pending_stops(&self, worker: Identifier) -> Result<Value> {
         let s = self.database.session().await?;
-        let rows=s.client.query("SELECT e.id,a.id attempt_id,a.fencing_version,jsonb_array_length(a.manifest->'cases') case_count FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id WHERE e.state='cancelling' AND a.worker_id=$1 ORDER BY e.updated_at,e.id LIMIT 16",&[&worker.as_uuid()]).await?;
+        let rows=s.client.query("SELECT e.id,a.id attempt_id,a.fencing_version,jsonb_array_length(a.manifest->'cases') case_count,(a.manifest->>'schema_version')::int schema_version FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id WHERE e.state='cancelling' AND a.worker_id=$1 ORDER BY e.updated_at,e.id LIMIT 16",&[&worker.as_uuid()]).await?;
-        Ok(Value::Array(rows.iter().map(|r|json!({"binding":LeaseBinding{evaluation_id:identifier(r.get("id")),attempt_id:identifier(r.get("attempt_id")),worker_id:worker,fencing_version:r.get::<_,i64>("fencing_version")as u64},"case_count":r.get::<_,i32>("case_count")})).collect()))
+        let entries = rows.iter().map(|r| {
+            let mut entry=json!({"binding":LeaseBinding{evaluation_id:identifier(r.get("id")),attempt_id:identifier(r.get("attempt_id")),worker_id:worker,fencing_version:r.get::<_,i64>("fencing_version")as u64},"case_count":r.get::<_,i32>("case_count")});
+            match r.get::<_,i32>("schema_version") {
+                // Preserve the old pending-stop response and all v1 report bytes.
+                1 => {},
+                2 => entry["schema_version"] = json!(2),
+                _ => return Err(Error::Invalid),
+            }
+            Ok(entry)
+        }).collect::<Result<Vec<_>>>()?;
+        Ok(Value::Array(entries))
     }
     pub async fn recover(&self) -> Result<u64> {
         let mut s = self.database.session().await?;

judge/crates/engine/src/lib.rs

@@ -229,3 +229,4 @@ mod tests {
 }
 
 pub mod submission;
+pub mod submission_v2;

judge/crates/engine/src/submission.rs

@@ -62,7 +62,7 @@ fn empty_compile(status: CompilationStatus) -> CompilationReport {
         memory_peak_bytes: None,
     }
 }
-fn classify(o: &Observation, expected: &[u8]) -> Verdict {
+pub(crate) fn classify(o: &Observation, expected: &[u8]) -> Verdict {
     match o.kind {
         ExecutionKind::Exited { code: 0 } => {
             if o.stdout == expected {

judge/crates/engine/src/submission_v2.rs

@@ -0,0 +1,478 @@
+//! Explicit v2 preparation and typed program execution; legacy wire stays separate.
+//! Ports own clocks/cancellation/storage/execution; this module is pure domain.
+use super::submission::{
+    ControlPort, ExpectedPort, SkipReason, SkippedCase, StageFailure, classify,
+};
+use crate::{ExecutionKind, Observation};
+use kf_contracts::judging_v2::*;
+use serde::Serialize;
+pub struct Compiled {
+    pub report: CompilationReport,
+    pub cleanup_complete: bool,
+    pub failure: Option<FailureCode>,
+}
+pub trait PreparationPort {
+    fn prepare(
+        &mut self,
+        manifest: &JobManifest,
+        binding: &LeaseBinding,
+        run_id: &str,
+    ) -> Result<Compiled, StageFailure>;
+}
+pub trait RunPort {
+    fn run(
+        &mut self,
+        program: &CompiledProgram,
+        case: &TestCase,
+        limits: &RunLimits,
+        run_id: &str,
+    ) -> Result<Observation, StageFailure>;
+}
+#[derive(Debug, Serialize)]
+pub struct SubmissionResult {
+    pub report: JudgementReport,
+    pub compile_run_id: String,
+    pub case_run_ids: Vec<String>,
+    pub skipped: Vec<SkippedCase>,
+}
+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 failure(code: FailureCode) -> StageFailure {
+    StageFailure {
+        code,
+        cleanup_complete: true,
+    }
+}
+fn gate(p: &impl ControlPort, m: &JobManifest) -> Result<bool, StageFailure> {
+    if p.elapsed_us() >= m.job_wall_time_us {
+        return Err(failure(FailureCode::JobDeadline));
+    }
+    Ok(p.cancelled())
+}
+pub fn judge_submission<P: PreparationPort + RunPort + ExpectedPort + ControlPort>(
+    p: &mut P,
+    m: &JobManifest,
+    binding: LeaseBinding,
+    digest: Digest,
+) -> Result<SubmissionResult, &'static str> {
+    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.prepare(m, &binding, &compile_run_id)),
+    };
+    let mut cleanup_complete;
+    let mut failure_code;
+    let mut compilation = match compilation {
+        Ok(c) => {
+            c.report.validate()?;
+            if c.report.program.as_ref().is_some_and(|program| {
+                program.binding != binding
+                    || program.source != m.source
+                    || program.toolchain != m.toolchain
+                    || program.kind != m.language.contract().program_kind
+                    || (program.kind != ProgramKind::PythonSource
+                        && (program
+                            .files
+                            .iter()
+                            .any(|f| f.artifact.bytes > m.compile_limits.file_bytes)
+                            || program.bytes() > m.compile_limits.workdir_bytes))
+            }) {
+                return Err("preparation program differs from authorized manifest or lease");
+            }
+            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)
+        }
+    };
+    if failure_code.is_none() && p.elapsed_us() >= m.job_wall_time_us {
+        failure_code = Some(FailureCode::JobDeadline);
+        if compilation.status != CompilationStatus::Succeeded {
+            compilation.status = CompilationStatus::SystemError;
+            compilation.program = None;
+        }
+    }
+    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 o.kind == ExecutionKind::InfrastructureFailure {
+                        Verdict::SystemError
+                    } else {
+                        match gate(p, m) {
+                            Err(e) => {
+                                failure_code = Some(e.code);
+                                Verdict::SystemError
+                            }
+                            Ok(true) => Verdict::Cancelled,
+                            Ok(false) if !matches!(o.kind, ExecutionKind::Exited { code: 0 }) => {
+                                classify(&o, &[])
+                            }
+                            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
+                                }
+                            },
+                        }
+                    };
+                    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 {
+        prepared: CompiledProgram,
+        observations: Vec<Observation>,
+        prepare_status: CompilationStatus,
+        prepares: usize,
+        runs: usize,
+        expected_reads: usize,
+        cancel_after: usize,
+        clock: u64,
+        advance_us: u64,
+        fail_cleanup: bool,
+    }
+    impl PreparationPort for Port {
+        fn prepare(
+            &mut self,
+            _: &JobManifest,
+            _: &LeaseBinding,
+            _: &str,
+        ) -> Result<Compiled, StageFailure> {
+            self.prepares += 1;
+            Ok(Compiled {
+                report: CompilationReport {
+                    status: self.prepare_status,
+                    program: (self.prepare_status == CompilationStatus::Succeeded)
+                        .then(|| self.prepared.clone()),
+                    diagnostics: String::new().try_into().unwrap(),
+                    diagnostics_truncated: false,
+                    cpu_time_us: None,
+                    memory_peak_bytes: Some(4),
+                },
+                cleanup_complete: !self.fail_cleanup,
+                failure: None,
+            })
+        }
+    }
+    impl RunPort for Port {
+        fn run(
+            &mut self,
+            program: &CompiledProgram,
+            _: &TestCase,
+            _: &RunLimits,
+            _: &str,
+        ) -> Result<Observation, StageFailure> {
+            assert_eq!(program, &self.prepared);
+            self.runs += 1;
+            Ok(self.observations.remove(0))
+        }
+    }
+    impl ExpectedPort for Port {
+        fn expected(&mut self, case: &TestCase) -> Result<Vec<u8>, StageFailure> {
+            self.expected_reads += 1;
+            Ok(vec![0; case.expected.bytes as usize])
+        }
+    }
+    impl ControlPort for Port {
+        fn elapsed_us(&self) -> u64 {
+            self.clock
+                .saturating_add(self.runs as u64 * self.advance_us)
+        }
+        fn cancelled(&self) -> bool {
+            self.runs >= self.cancel_after
+        }
+        fn stage(&mut self, _: Stage, _: Option<u16>) -> Result<(), StageFailure> {
+            Ok(())
+        }
+    }
+    fn fixture(language: &str) -> (JobManifest, LeaseBinding, Digest, Port) {
+        let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("../../../tests/fixtures/contracts-v2");
+        let m: JobManifest = kf_contracts::judging::decode_wire(
+            &std::fs::read(root.join(format!("judge-manifest.valid-{language}.json"))).unwrap(),
+        )
+        .unwrap();
+        let r: JudgementReport = kf_contracts::judging::decode_wire(
+            &std::fs::read(root.join(format!("judge-result.valid-{language}.json"))).unwrap(),
+        )
+        .unwrap();
+        let p = Port {
+            prepared: r.compilation.program.unwrap(),
+            observations: m
+                .cases
+                .iter()
+                .map(|case| Observation {
+                    kind: ExecutionKind::Exited { code: 0 },
+                    stdout: vec![0; case.expected.bytes as usize],
+                    cpu_time_us: Some(3),
+                    memory_peak_bytes: None,
+                })
+                .collect(),
+            prepare_status: CompilationStatus::Succeeded,
+            prepares: 0,
+            runs: 0,
+            expected_reads: 0,
+            cancel_after: usize::MAX,
+            clock: 0,
+            advance_us: 0,
+            fail_cleanup: false,
+        };
+        (m, r.binding, r.manifest_digest, p)
+    }
+    #[test]
+    fn typed_prepare_once_same_program_every_point_and_unknown_measurements() {
+        for language in ["c17", "cpp20", "python313", "java21"] {
+            let (m, b, d, mut p) = fixture(language);
+            let r = judge_submission(&mut p, &m, b, d).unwrap();
+            assert_eq!(p.prepares, 1);
+            assert_eq!(p.runs, m.cases.len());
+            assert_eq!(r.report.verdict, Verdict::Accepted);
+            assert_eq!(r.report.run_memory_peak_bytes, None);
+            assert_eq!(r.report.compilation.cpu_time_us, None);
+            assert_eq!(r.report.run_cpu_time_us, Some(3 * m.cases.len() as u64));
+            assert!(r.skipped.is_empty());
+        }
+    }
+    #[test]
+    fn bad_program_is_refused_before_any_execution_or_expected_read() {
+        let (m, b, d, mut p) = fixture("java21");
+        p.prepared.source.digest = "f".repeat(64).try_into().unwrap();
+        assert!(judge_submission(&mut p, &m, b, d).is_err());
+        assert_eq!(p.runs, 0);
+        assert_eq!(p.expected_reads, 0);
+        let (m, b, d, mut p) = fixture("python313");
+        p.prepared.binding.fencing_version += 1;
+        assert!(judge_submission(&mut p, &m, b, d).is_err());
+        assert_eq!(p.runs, 0);
+    }
+    #[test]
+    fn preparation_failure_and_cleanup_precedence_never_fake_points() {
+        for status in [
+            CompilationStatus::CompilationError,
+            CompilationStatus::CompilationLimit,
+            CompilationStatus::Cancelled,
+        ] {
+            let (m, b, d, mut p) = fixture("java21");
+            p.prepare_status = status;
+            let r = judge_submission(&mut p, &m, b, d).unwrap();
+            assert_eq!(p.runs, 0);
+            assert_eq!(r.skipped.len(), m.cases.len());
+            assert_eq!(
+                r.report.verdict,
+                if status == CompilationStatus::Cancelled {
+                    Verdict::Cancelled
+                } else {
+                    Verdict::CompilationError
+                }
+            );
+        }
+        let (m, b, d, mut p) = fixture("python313");
+        p.fail_cleanup = true;
+        let r = judge_submission(&mut p, &m, b, d).unwrap();
+        assert_eq!(r.report.failure_code, Some(FailureCode::CleanupFailed));
+        assert!(!r.report.cleanup_complete);
+        assert_eq!(p.runs, 0);
+    }
+    #[test]
+    fn trusted_outcome_precedes_comparison_and_text_cannot_fake_memory_limit() {
+        for (kind, verdict) in [
+            (ExecutionKind::Exited { code: 1 }, Verdict::RuntimeError),
+            (ExecutionKind::MemoryLimit, Verdict::MemoryLimitExceeded),
+            (ExecutionKind::TimeLimit, Verdict::TimeLimitExceeded),
+            (ExecutionKind::OutputLimit, Verdict::OutputLimitExceeded),
+        ] {
+            let (m, b, d, mut p) = fixture("java21");
+            for o in &mut p.observations {
+                o.kind = kind;
+                o.stdout = b"OutOfMemoryError".to_vec();
+            }
+            let r = judge_submission(&mut p, &m, b, d).unwrap();
+            assert_eq!(r.report.verdict, verdict);
+            assert_eq!(p.expected_reads, 0);
+        }
+    }
+    #[test]
+    fn job_deadline_and_cancel_skip_tail_without_fabricated_grades() {
+        let (m, b, d, mut p) = fixture("python313");
+        p.advance_us = 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_eq!(p.runs, 1);
+        assert_eq!(p.expected_reads, 0);
+        let (m, b, d, mut p) = fixture("java21");
+        p.cancel_after = 1;
+        let r = judge_submission(&mut p, &m, b, d).unwrap();
+        assert_eq!(r.report.verdict, Verdict::Cancelled);
+        assert_eq!(r.skipped.len(), m.cases.len() - 1);
+        assert_eq!(p.runs, 1);
+    }
+}

judge/crates/sandbox/src/artifacts.rs

@@ -236,6 +236,7 @@ impl OutputPlan {
         let paths = if self.java_classes {
             let mut paths = Vec::new();
             let mut count = 0;
+            let mut unsupported = false;
             // The entire group is frozen and drained. Discover only bounded,
             // flat compiler outputs through the held private work directory.
             for entry in std::fs::read_dir(format!("/proc/self/fd/{}", work.as_raw_fd()))? {
@@ -251,10 +252,36 @@ impl OutputPlan {
                     continue;
                 }
                 if !kf_contracts::judging_v2::class_name(&name) {
-                    return Err(refused("class export path/type"));
+                    // javac accepts packages and Unicode class names outside
+                    // this protocol's flat ASCII class set. Suppress the whole
+                    // batch so Worker reports unsupported entry as user CE.
+                    // Links/special files remain infrastructure refusals and
+                    // no unrecognized output is ever opened for reading.
+                    let fd = rustix::fs::openat(
+                        work,
+                        name.as_str(),
+                        OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+                        Mode::empty(),
+                    )?;
+                    let st = rustix::fs::fstat(&fd)?;
+                    if st.st_dev != rustix::fs::fstat(work)?.st_dev
+                        || !matches!(
+                            FileType::from_raw_mode(st.st_mode),
+                            FileType::RegularFile | FileType::Directory
+                        )
+                        || (FileType::from_raw_mode(st.st_mode) == FileType::RegularFile
+                            && st.st_nlink != 1)
+                    {
+                        return Err(refused("class export path/type"));
+                    }
+                    unsupported = true;
+                    continue;
                 }
                 paths.push(name);
             }
+            if unsupported {
+                return Ok(Vec::new());
+            }
             paths.sort();
             paths
         } else {
@@ -406,8 +433,33 @@ pub struct StagePublication {
     dev: u64,
     ino: u64,
     files: Vec<StageEntry>,
+    staged_bytes: u64,
     removed: bool,
 }
+const MAX_STAGE_FILES: usize = kf_contracts::judging_v2::MAX_CLASS_FILES + 1;
+const MAX_STAGE_JOURNAL_BYTES: u64 = 65_536;
+fn indexed_program(name: &str) -> bool {
+    name.strip_prefix("program-").is_some_and(|index| {
+        index.len() == 3
+            && index.bytes().all(|b| b.is_ascii_digit())
+            && index
+                .parse::<usize>()
+                .is_ok_and(|i| i < kf_contracts::judging_v2::MAX_CLASS_FILES)
+    })
+}
+fn stage_entries_valid(names: &[&str]) -> bool {
+    let indexed = names.iter().any(|n| indexed_program(n));
+    let mut unique = std::collections::HashSet::new();
+    names.len() <= if indexed { MAX_STAGE_FILES } else { 2 }
+        && names.iter().all(|n| {
+            unique.insert(*n)
+                && if indexed {
+                    *n == "stdin" || indexed_program(n)
+                } else {
+                    matches!(*n, "source" | "program" | "stdin")
+                }
+        })
+}
 #[derive(serde::Serialize, serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 struct StageEntry {
@@ -462,6 +514,7 @@ impl StagePublication {
             dev: st.st_dev,
             ino: st.st_ino,
             files: vec![],
+            staged_bytes: 0,
             removed: false,
         };
         let header = StageOwner {
@@ -537,16 +590,16 @@ impl StagePublication {
                 || j.st_mode & 0o7777 != 0o600
                 || j.st_nlink != 1
                 || j.st_size < 0
-                || j.st_size > 2048
+                || j.st_size as u64 > MAX_STAGE_JOURNAL_BYTES
                 || FileType::from_raw_mode(j.st_mode) != FileType::RegularFile
             {
                 return Err(refused("untrusted stage journal"));
             }
             let mut bytes = String::new();
             readonly_handle(&journal)?
-                .take(2049)
+                .take(MAX_STAGE_JOURNAL_BYTES + 1)
                 .read_to_string(&mut bytes)?;
-            if bytes.len() > 2048 || !bytes.ends_with('\n') {
+            if bytes.len() as u64 > MAX_STAGE_JOURNAL_BYTES || !bytes.ends_with('\n') {
                 return Err(refused("stage journal framing"));
             }
             let mut lines = bytes.lines();
@@ -572,13 +625,7 @@ impl StagePublication {
                         .map_err(|_| refused("stage entry journal"))
                 })
                 .collect::<io::Result<Vec<_>>>()?;
-            let mut unique = std::collections::HashSet::new();
+            if !stage_entries_valid(&entries.iter().map(|e| e.name.as_str()).collect::<Vec<_>>()) {
-            if entries.len() > 2
-                || entries.iter().any(|e| {
-                    !matches!(e.name.as_str(), "source" | "program" | "stdin")
-                        || !unique.insert(e.name.clone())
-                })
-            {
                 return Err(refused("stage entry set"));
             }
             let mut value = Self {
@@ -588,6 +635,7 @@ impl StagePublication {
                 dev: st.st_dev,
                 ino: st.st_ino,
                 files: entries,
+                staged_bytes: 0,
                 removed: false,
             };
             value.cleanup()?;
@@ -599,9 +647,12 @@ impl StagePublication {
         &self.root
     }
     pub fn put(&mut self, name: &str, reader: &mut impl Read, bytes: u64) -> io::Result<()> {
-        if !matches!(name, "source" | "program" | "stdin")
+        let mut names: Vec<_> = self.files.iter().map(|e| e.name.as_str()).collect();
-            || self.files.len() >= 2
+        names.push(name);
+        if !stage_entries_valid(&names)
             || bytes > MAX_SNAPSHOT_BYTES
+            || (indexed_program(name) && bytes == 0)
+            || self.staged_bytes.saturating_add(bytes) > MAX_SNAPSHOT_BYTES
         {
             return Err(refused("fixed staging entry/budget"));
         }
@@ -628,6 +679,7 @@ impl StagePublication {
         file.write_all(b"\n")?;
         file.sync_all()?;
         self.files.push(entry);
+        self.staged_bytes += bytes;
         let mut file = File::from(fd);
         let copied = io::copy(&mut reader.take(bytes + 1), &mut file)?;
         if copied != bytes {
@@ -650,7 +702,7 @@ impl StagePublication {
         use rustix::fd::AsRawFd;
         let actual: Vec<_> =
             std::fs::read_dir(format!("/proc/self/fd/{}", self.root.fd.as_raw_fd()))?
-                .take(5)
+                .take(MAX_STAGE_FILES + 2)
                 .map(|e| Ok(e?.file_name()))
                 .collect::<io::Result<_>>()?;
         if actual.len() != self.files.len() + 1
@@ -710,3 +762,33 @@ impl Drop for StagePublication {
         let _ = self.cleanup();
     }
 }
+
+#[cfg(test)]
+mod stage_policy_tests {
+    use super::*;
+    #[test]
+    fn bounded_indexed_batch_and_legacy_names_are_separate() {
+        assert!(stage_entries_valid(&["source"]));
+        assert!(stage_entries_valid(&["program", "stdin"]));
+        assert!(!stage_entries_valid(&["source", "program", "stdin"]));
+        let mut names: Vec<_> = (0..256).map(|i| format!("program-{i:03}")).collect();
+        names.push("stdin".into());
+        assert!(stage_entries_valid(
+            &names.iter().map(String::as_str).collect::<Vec<_>>()
+        ));
+        for name in [
+            "program-256",
+            "program-00",
+            "program-0000",
+            "program-../a",
+            "Main.class",
+            ".owner",
+            "expected",
+        ] {
+            assert!(!stage_entries_valid(&[name]));
+        }
+        assert!(!stage_entries_valid(&["program-000", "source"]));
+        assert!(!stage_entries_valid(&["program-000", "program"]));
+        assert!(!stage_entries_valid(&["program-000", "program-000"]));
+    }
+}

judge/crates/sandbox/src/image.rs

@@ -46,6 +46,23 @@ fn refused() -> io::Error {
         "image catalog/integrity refused",
     )
 }
+fn catalog_path(kind: ImageKind, name: &str) -> bool {
+    if kind != ImageKind::GnuCxx {
+        return crate::spec::path(name, "image").is_ok();
+    }
+    // GNU's immutable standard-library directories include c++ and libstdc++.
+    // This operator catalog rule never changes caller artifact path validation.
+    !name.is_empty()
+        && name.len() <= 4096
+        && name.split('/').all(|p| {
+            !p.is_empty()
+                && p != "."
+                && p != ".."
+                && p.len() <= 255
+                && p.bytes()
+                    .all(|b| b.is_ascii_alphanumeric() || b"_.-+".contains(&b))
+        })
+}
 pub fn catalog_digest(data: &[u8]) -> Digest {
     format!("{:x}", Sha256::digest(data))
         .try_into()
@@ -81,7 +98,9 @@ impl ImageCatalog {
             .iter()
             .chain(self.files.iter().map(|v| &v.path))
         {
-            crate::spec::path(name, "image").map_err(|_| refused())?;
+            if !catalog_path(self.image_kind, name) {
+                return Err(refused());
+            }
             if name == "manifest.json"
                 || name
                     .split('/')
@@ -290,3 +309,25 @@ fn walk<'a>(
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod path_tests {
+    use super::*;
+    #[test]
+    fn trusted_gnu_library_paths_do_not_widen_request_or_managed_image_paths() {
+        for p in ["usr/include/c++/16.2.1/vector", "usr/lib/libstdc++.so.6"] {
+            assert!(catalog_path(ImageKind::GnuCxx, p));
+            assert!(!catalog_path(ImageKind::CPython313, p));
+            assert!(crate::spec::path(p, "artifact").is_err());
+        }
+        for p in [
+            "/usr/lib",
+            "usr/../lib",
+            "usr//lib",
+            "usr/./lib",
+            "usr/lib\0",
+        ] {
+            assert!(!catalog_path(ImageKind::GnuCxx, p));
+        }
+    }
+}

judge/crates/sandbox/src/launcher/protocol.rs

@@ -244,10 +244,22 @@ impl ControlChannel {
     }
 
     pub(crate) fn execution_deadline(&mut self, wall_us: u64) -> io::Result<()> {
+        self.runtime_deadline(wall_us, Duration::ZERO)
+    }
+
+    /// The ordinary RPC starts before catalog verification and isolation setup.
+    /// Its fixed preparation allowance is separate from the student's measured
+    /// execution wall limit. The factory's job/cancellation watcher stays active.
+    pub(crate) fn request_deadline(&mut self, wall_us: u64) -> io::Result<()> {
+        self.runtime_deadline(wall_us, Duration::from_secs(30))
+    }
+
+    fn runtime_deadline(&mut self, wall_us: u64, preparation: Duration) -> io::Result<()> {
         if wall_us == 0 || wall_us > 120_000_000 || self.poisoned || !self.pending.is_empty() {
             return Err(invalid("execution deadline"));
         }
-        self.deadline = Instant::now() + Duration::from_micros(wall_us) + Duration::from_secs(5);
+        self.deadline =
+            Instant::now() + preparation + Duration::from_micros(wall_us) + Duration::from_secs(5);
         Ok(())
     }
 
@@ -579,6 +591,24 @@ mod tests {
         )
     }
 
+    #[test]
+    fn request_preparation_allowance_is_fixed_and_execution_limit_is_separate() {
+        let (mut request, mut execution) = pair();
+        let start = Instant::now();
+        request.request_deadline(3_000_000).unwrap();
+        execution.execution_deadline(3_000_000).unwrap();
+        assert!(request.deadline.duration_since(start) >= Duration::from_secs(38));
+        assert!(request.deadline.duration_since(start) < Duration::from_secs(39));
+        assert!(execution.deadline.duration_since(start) >= Duration::from_secs(8));
+        assert!(execution.deadline.duration_since(start) < Duration::from_secs(9));
+        request.request_deadline(120_000_000).unwrap();
+        assert!(request.deadline.duration_since(start) < Duration::from_secs(156));
+        assert!(request.request_deadline(0).is_err());
+        assert!(request.request_deadline(120_000_001).is_err());
+        request.poisoned = true;
+        assert!(request.request_deadline(3_000_000).is_err());
+    }
+
     #[test]
     fn descriptor_transport_preserves_identity_and_cloexec() {
         let (mut a, mut b) = pair();

judge/crates/sandbox/src/rpc.rs

@@ -117,7 +117,7 @@ pub(crate) fn execute_channel(
 ) -> io::Result<Executed> {
     c.send(&Request { spec })?;
     c.send_fd(publication)?;
-    c.execution_deadline(validated.spec().limits.wall_time_us)?;
+    c.request_deadline(validated.spec().limits.wall_time_us)?;
     let report = match c.receive::<Reply>()? {
         Reply::Completed { report } => *report,
         Reply::Rejected {
@@ -229,7 +229,7 @@ pub(crate) fn serve_stream(
     }
     let artifacts = ArtifactRoot::from_worker_directory(c.receive_fd()?, ids.worker_uid)?;
     let peer = c.duplicate_stream()?;
-    c.execution_deadline(request.spec.limits.wall_time_us)?;
+    c.request_deadline(request.spec.limits.wall_time_us)?;
     match execution::execute(
         config,
         request.spec,

judge/crates/sandbox/tests/indexed_staging_vm.rs

@@ -0,0 +1,142 @@
+#![cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+//! Indexed Worker publications, actual seals/recovery/foreign preservation in VM only.
+#[allow(dead_code)]
+mod support;
+use kf_sandbox::artifacts::StagePublication;
+use std::{
+    fs,
+    io::Read,
+    os::unix::{
+        fs::{MetadataExt, PermissionsExt, chown},
+        process::CommandExt,
+    },
+    path::PathBuf,
+    process::Command,
+};
+
+fn child(name: &str, path: &std::path::Path) -> std::process::Output {
+    Command::new(std::env::current_exe().unwrap())
+        .env_clear()
+        .env("KF_SANDBOX_INDEPENDENT_VM", "1")
+        .env("KF_INDEXED_STAGE_DIR", path)
+        .uid(1000)
+        .gid(1000)
+        .args([
+            "--ignored",
+            "--exact",
+            name,
+            "--test-threads=1",
+            "--nocapture",
+        ])
+        .output()
+        .unwrap()
+}
+
+#[test]
+#[ignore = "ordinary peer subprocess in capped independent VM"]
+fn leave_indexed_publication_for_recovery() {
+    support::require_vm();
+    assert_eq!(rustix::process::geteuid().as_raw(), 1000);
+    let path = PathBuf::from(std::env::var_os("KF_INDEXED_STAGE_DIR").unwrap());
+    let mut p = StagePublication::create(&path, "dead-owner").unwrap();
+    for i in 0..256 {
+        p.put(&format!("program-{i:03}"), &mut b"class".as_slice(), 5)
+            .unwrap();
+    }
+    p.put("stdin", &mut b"".as_slice(), 0).unwrap();
+    assert!(p.put("program-256", &mut b"class".as_slice(), 5).is_err());
+    assert!(p.put("program", &mut b"class".as_slice(), 5).is_err());
+    assert!(p.put("program-001", &mut b"class".as_slice(), 5).is_err());
+    assert!(fs::metadata(path.join("dead-owner/.owner")).unwrap().len() < 65536);
+    std::mem::forget(p);
+    println!("KF_INDEXED_DEAD_OWNER=1");
+}
+
+#[test]
+#[ignore = "ordinary peer subprocess in capped independent VM"]
+fn ordinary_indexed_staging_client() {
+    support::require_vm();
+    assert_eq!(rustix::process::geteuid().as_raw(), 1000);
+    let path = PathBuf::from(std::env::var_os("KF_INDEXED_STAGE_DIR").unwrap());
+    let mut p = StagePublication::create(&path, "sealed").unwrap();
+    p.put("program-000", &mut b"\0checked\xff".as_slice(), 9)
+        .unwrap();
+    assert!(p.put("program-001", &mut b"".as_slice(), 0).is_err());
+    for name in [
+        "../foreign",
+        "program-1",
+        "program-256",
+        "source",
+        "program",
+    ] {
+        assert!(p.put(name, &mut b"x".as_slice(), 1).is_err());
+    }
+    assert!(
+        p.put("program-001", &mut b"x".as_slice(), 64 * 1024 * 1024 + 1)
+            .is_err()
+    );
+    let snapshot = p.root().snapshot_artifact("program-000", 9).unwrap();
+    fs::write(
+        path.join("sealed/program-000"),
+        b"overwritten backing bytes",
+    )
+    .unwrap();
+    let mut bytes = vec![];
+    snapshot.reader().unwrap().read_to_end(&mut bytes).unwrap();
+    assert_eq!(bytes, b"\0checked\xff");
+    let seals = rustix::fs::fcntl_get_seals(&snapshot).unwrap();
+    assert!(seals.contains(
+        rustix::fs::SealFlags::WRITE
+            | rustix::fs::SealFlags::GROW
+            | rustix::fs::SealFlags::SHRINK
+            | rustix::fs::SealFlags::SEAL
+    ));
+    assert_eq!(StagePublication::recover(&path).unwrap(), 0);
+    p.cleanup().unwrap();
+    let owner = child("leave_indexed_publication_for_recovery", &path);
+    assert!(owner.status.success(), "{owner:?}");
+    assert!(String::from_utf8_lossy(&owner.stdout).contains("KF_INDEXED_DEAD_OWNER=1"));
+    let foreign = path.join("dead-owner/foreign");
+    fs::write(&foreign, b"foreign canary").unwrap();
+    let inode = fs::metadata(&foreign).unwrap().ino();
+    assert!(StagePublication::recover(&path).is_err());
+    assert_eq!(fs::metadata(&foreign).unwrap().ino(), inode);
+    assert_eq!(fs::read(&foreign).unwrap(), b"foreign canary");
+    fs::remove_file(foreign).unwrap();
+    assert_eq!(StagePublication::recover(&path).unwrap(), 1);
+    assert_eq!(fs::read_dir(&path).unwrap().count(), 0);
+    let mut p = StagePublication::create(&path, "replaced").unwrap();
+    p.put("program-000", &mut b"original".as_slice(), 8)
+        .unwrap();
+    let file = path.join("replaced/program-000");
+    fs::rename(&file, path.join("original-retained")).unwrap();
+    fs::write(&file, b"different inode").unwrap();
+    let inode = fs::metadata(&file).unwrap().ino();
+    assert!(p.cleanup().is_err());
+    assert_eq!(fs::metadata(&file).unwrap().ino(), inode);
+    assert_eq!(fs::read(&file).unwrap(), b"different inode");
+    fs::remove_file(&file).unwrap();
+    fs::rename(path.join("original-retained"), &file).unwrap();
+    p.cleanup().unwrap();
+    assert_eq!(fs::read_dir(&path).unwrap().count(), 0);
+    println!(
+        "KF_INDEXED_STAGING_PROOF files=256 stdin=1 bounds=1 seals=1 live_owner=1 dead_owner_recovered=1 foreign_inode_preserved=1 replaced_inode_preserved=1 leftovers=0"
+    );
+}
+
+#[test]
+#[ignore = "actual ordinary indexed staging only inside independent capped no-NIC/no-share VM"]
+fn indexed_staging_recovers_only_owned_inodes_and_preserves_sealed_bytes() {
+    support::require_vm();
+    let dir = PathBuf::from(format!("/kf/indexed-staging-{}", std::process::id()));
+    fs::create_dir(&dir).unwrap();
+    chown(&dir, Some(1000), Some(1000)).unwrap();
+    fs::set_permissions(&dir, fs::Permissions::from_mode(0o700)).unwrap();
+    let output = child("ordinary_indexed_staging_client", &dir);
+    assert!(output.status.success(), "{output:?}");
+    assert!(String::from_utf8_lossy(&output.stdout).contains("KF_INDEXED_STAGING_PROOF"));
+    assert_eq!(fs::read_dir(&dir).unwrap().count(), 0);
+    fs::remove_dir(dir).unwrap();
+    print!("{}", String::from_utf8_lossy(&output.stdout));
+    println!("KF_INDEXED_STAGING_STATUS=0");
+}

judge/crates/sandbox/tests/managed_pipeline_vm.rs

@@ -0,0 +1,509 @@
+#![cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+//! Real ordinary Worker/Engine and default Root, only in the independent VM.
+#[allow(dead_code)]
+mod support;
+use kf_contracts::{judging as v1, judging_v2 as v2};
+use kf_sandbox::{
+    factory::FactoryClient,
+    image::{ImageCatalog, catalog_digest},
+    runtime::{ImageKind, RegisteredImage},
+    spec::{ExecutionProfile, ResourceLimits},
+};
+use sha2::{Digest as _, Sha256};
+use std::{
+    fs,
+    os::unix::{
+        fs::{PermissionsExt, chown},
+        process::CommandExt,
+    },
+    path::{Path, PathBuf},
+    process::{Child, Command, Stdio},
+    time::{Duration, Instant},
+};
+fn store(path: &Path, data: &[u8]) {
+    fs::write(path, data).unwrap();
+    chown(path, Some(1000), Some(1000)).unwrap();
+    fs::set_permissions(path, fs::Permissions::from_mode(0o600)).unwrap();
+}
+fn id(n: u32) -> String {
+    format!("aaaaaaaa-aaaa-4aaa-8aaa-{n:012}")
+}
+fn budget(profile: ExecutionProfile) -> v1::RunLimits {
+    let mut b = ResourceLimits::for_profile(profile);
+    if profile.prepares_program() {
+        b.cpu_time_us = 30_000_000;
+        b.wall_time_us = 60_000_000;
+    } else if matches!(
+        profile,
+        ExecutionProfile::JavaRun | ExecutionProfile::PythonRun
+    ) {
+        b.cpu_time_us = 5_000_000;
+        b.wall_time_us = 10_000_000;
+    }
+    serde_json::from_value(serde_json::to_value(b).unwrap()).unwrap()
+}
+struct Fixture {
+    root: support::Root,
+    dir: PathBuf,
+    process: Child,
+    next: u32,
+}
+impl Fixture {
+    fn new() -> Self {
+        support::require_vm();
+        let mut root = support::Root::new();
+        root.config.max_concurrent_runs = 1;
+        root.config.identities.count = 1;
+        root.config.aggregate.task_cpu_quota_us = 100000;
+        root.config.per_task_ceiling.cpu_time_us = 30_000_000;
+        root.config.per_task_ceiling.wall_time_us = 60_000_000;
+        let mut capabilities = Vec::new();
+        for (name, kind, languages) in [
+            (
+                "python313",
+                ImageKind::LinuxCPython313X86_64V1,
+                vec![v2::Language::Python313],
+            ),
+            (
+                "java21",
+                ImageKind::LinuxOpenJdk21X86_64V1,
+                vec![v2::Language::Java21],
+            ),
+            (
+                "gnu-v2",
+                ImageKind::LinuxGnuCxxX86_64V1,
+                vec![v2::Language::C17, v2::Language::Cpp20],
+            ),
+        ] {
+            let image_root = format!("/var/lib/kf-judge/images/{name}");
+            let raw = fs::read(format!("{image_root}/manifest.json")).unwrap();
+            let c: ImageCatalog = serde_json::from_slice(&raw).unwrap();
+            let toolchain = v2::RuntimeIdentity {
+                toolchain_id: c.toolchain_id,
+                image_kind: c.image_kind,
+                image_revision: c.image_revision,
+                image_digest: catalog_digest(&raw),
+                runtime_version: c.runtime_version,
+            };
+            for language in languages {
+                let (prepare, run) = match language {
+                    v2::Language::C17 => (ExecutionProfile::CompileC, ExecutionProfile::Native),
+                    v2::Language::Cpp20 => (ExecutionProfile::CompileCpp, ExecutionProfile::Native),
+                    v2::Language::Python313 => {
+                        (ExecutionProfile::PythonCheck, ExecutionProfile::PythonRun)
+                    }
+                    v2::Language::Java21 => {
+                        (ExecutionProfile::JavaCompile, ExecutionProfile::JavaRun)
+                    }
+                };
+                capabilities.push(v2::RuntimeCapability {
+                    language,
+                    toolchain: toolchain.clone(),
+                    compile_ceiling: budget(prepare),
+                    run_ceiling: budget(run),
+                });
+            }
+            root.config.runtime_images.push(RegisteredImage {
+                image_root,
+                image_kind: kind,
+                runtime_identity: toolchain,
+                per_task_ceiling: root.config.per_task_ceiling.clone(),
+            });
+        }
+        root.config.clone().validate().unwrap();
+        let dir = PathBuf::from(format!("/kf/managed-pipeline-{}", std::process::id()));
+        fs::create_dir(&dir).unwrap();
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o711)).unwrap();
+        for name in ["artifacts", "staging"] {
+            let p = dir.join(name);
+            fs::create_dir(&p).unwrap();
+            chown(&p, Some(1000), Some(1000)).unwrap();
+            fs::set_permissions(&p, fs::Permissions::from_mode(0o700)).unwrap();
+        }
+        let admission = v2::AdmissionPolicy {
+            schema_version: v2::SchemaVersion,
+            max_concurrent_jobs: 1,
+            prefetch: 1,
+            max_retries: 2,
+            max_job_wall_time_us: 180_000_000,
+            max_package_bytes: v1::MAX_PACKAGE_BYTES,
+            capabilities,
+        };
+        admission.validate().unwrap();
+        store(
+            &dir.join("admission.json"),
+            &serde_json::to_vec(&admission).unwrap(),
+        );
+        fs::write(
+            dir.join("root.json"),
+            serde_json::to_vec(&root.config).unwrap(),
+        )
+        .unwrap();
+        fs::set_permissions(dir.join("root.json"), fs::Permissions::from_mode(0o600)).unwrap();
+        let mut process = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap())
+            .env_clear()
+            .args(["factory", "--config"])
+            .arg(dir.join("root.json"))
+            .args(["--lifetime-seconds", "1200"])
+            .stdout(Stdio::null())
+            .stderr(fs::File::create(dir.join("root-diagnostics.log")).unwrap())
+            .spawn()
+            .unwrap();
+        let deadline = Instant::now() + Duration::from_secs(90);
+        while !fs::metadata("/var/lib/kf-judge/factory/control.sock")
+            .is_ok_and(|m| m.permissions().mode() & 0o777 == 0o660)
+        {
+            assert!(
+                process.try_wait().unwrap().is_none(),
+                "factory startup refused"
+            );
+            assert!(Instant::now() < deadline);
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        Self {
+            root,
+            dir,
+            process,
+            next: 1000,
+        }
+    }
+    fn artifact(&self, data: &[u8]) -> v1::ArtifactReference {
+        let digest: v1::Digest = format!("{:x}", Sha256::digest(data)).try_into().unwrap();
+        store(&self.dir.join("artifacts").join(digest.as_str()), data);
+        v1::ArtifactReference {
+            digest,
+            bytes: data.len() as u64,
+        }
+    }
+    fn execute(
+        &mut self,
+        language: v2::Language,
+        source: &[u8],
+        points: &[(&[u8], &[u8])],
+        verdict: v1::Verdict,
+        legacy: bool,
+    ) -> serde_json::Value {
+        self.next += 10;
+        let mut m: serde_json::Value = serde_json::from_str(include_str!(
+            "../../../../tests/fixtures/contracts-v2/judge-manifest.valid-python313.json"
+        ))
+        .unwrap();
+        let n = self.next;
+        let binding = v1::LeaseBinding {
+            evaluation_id: id(n).try_into().unwrap(),
+            attempt_id: id(n + 1).try_into().unwrap(),
+            worker_id: id(8).try_into().unwrap(),
+            fencing_version: 1,
+        };
+        let admission: v2::AdmissionPolicy =
+            serde_json::from_slice(&fs::read(self.dir.join("admission.json")).unwrap()).unwrap();
+        let cap = admission
+            .capabilities
+            .iter()
+            .find(|c| c.language == language)
+            .unwrap();
+        m["evaluation_id"] = serde_json::json!(id(n));
+        m["submission_id"] = serde_json::json!(id(n + 2));
+        m["source"] = serde_json::to_value(self.artifact(source)).unwrap();
+        m["language"] = serde_json::to_value(language).unwrap();
+        m["toolchain"] = serde_json::to_value(&cap.toolchain).unwrap();
+        m["compile_limits"] = serde_json::to_value(&cap.compile_ceiling).unwrap();
+        m["run_limits"] = serde_json::to_value(&cap.run_ceiling).unwrap();
+        m["job_wall_time_us"] = serde_json::json!(180_000_000);
+        m["cases"] = serde_json::json!(
+            points
+                .iter()
+                .enumerate()
+                .map(|(i, (input, expected))| v1::TestCase {
+                    case_id: id(n + 3 + i as u32).try_into().unwrap(),
+                    input: self.artifact(input),
+                    expected: self.artifact(expected)
+                })
+                .collect::<Vec<_>>()
+        );
+        let canonical = if legacy {
+            m["schema_version"] = serde_json::json!(1);
+            m["toolchain"] = serde_json::json!({"toolchain_id":"gnu-cxx-v1","image_kind":"linux_gnu_cxx_x86_64_v1","image_revision":1});
+            v1::manifest_bytes(&serde_json::from_value(m.clone()).unwrap()).unwrap()
+        } else {
+            v2::manifest_bytes(&serde_json::from_value(m.clone()).unwrap()).unwrap()
+        };
+        let digest = format!("{:x}", Sha256::digest(canonical));
+        let mut job = serde_json::json!({"manifest":m,"binding":binding,"manifest_digest":digest});
+        if !legacy {
+            job["attempt_number"] = serde_json::json!(1);
+        }
+        store(
+            &self.dir.join("job.json"),
+            &serde_json::to_vec(&job).unwrap(),
+        );
+        let mut command = Command::new(std::env::var_os("KF_WORKER_TEST_BIN").unwrap());
+        command
+            .env_clear()
+            .uid(1000)
+            .gid(1000)
+            .arg(if legacy {
+                "judge-submission"
+            } else {
+                "judge-submission-v2"
+            })
+            .args(["--socket", "/var/lib/kf-judge/factory/control.sock"])
+            .arg("--job")
+            .arg(self.dir.join("job.json"))
+            .arg("--artifacts")
+            .arg(self.dir.join("artifacts"))
+            .arg("--staging")
+            .arg(self.dir.join("staging"));
+        if !legacy {
+            command
+                .arg("--admission")
+                .arg(self.dir.join("admission.json"));
+        }
+        let output = command.output().unwrap();
+        if !output.stderr.is_empty() {
+            println!(
+                "KF_PIPELINE_WORKER_DIAGNOSTIC {}",
+                String::from_utf8_lossy(&output.stderr)
+            );
+        }
+        assert!(
+            output.status.success(),
+            "worker failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        let actual: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap();
+        if actual["result"]["report"]["verdict"] != serde_json::to_value(verdict).unwrap() {
+            use std::io::Read;
+            let mut diagnostic = Vec::new();
+            fs::File::open(self.dir.join("root-diagnostics.log"))
+                .unwrap()
+                .take(16_384)
+                .read_to_end(&mut diagnostic)
+                .unwrap();
+            println!(
+                "KF_PIPELINE_ROOT_DIAGNOSTIC {}",
+                String::from_utf8_lossy(&diagnostic)
+            );
+        }
+        assert_eq!(
+            actual["result"]["report"]["verdict"],
+            serde_json::to_value(verdict).unwrap(),
+            "{actual}"
+        );
+        assert_eq!(
+            actual["result"]["report"]["cleanup_complete"], true,
+            "{actual}"
+        );
+        assert_eq!(
+            actual["result"]["report"]["schema_version"],
+            if legacy { 1 } else { 2 }
+        );
+        let stages = actual["stages"].as_array().unwrap();
+        assert!(!stages.is_empty());
+        let mut ids = std::collections::HashSet::new();
+        for s in stages {
+            assert!(s["helper_pid"].as_u64().unwrap() > 1);
+            assert!(ids.insert(s["helper_pid"].as_u64().unwrap()));
+            assert_eq!(s["report"]["errors"], serde_json::json!([]));
+            assert_eq!(s["report"]["cleanup"], "complete");
+        }
+        assert_eq!(
+            stages.len(),
+            if verdict == v1::Verdict::CompilationError {
+                1
+            } else {
+                points.len() + 1
+            }
+        );
+        assert_eq!(fs::read_dir(self.dir.join("staging")).unwrap().count(), 0);
+        let stopped = Command::new(std::env::current_exe().unwrap())
+            .env_clear()
+            .uid(1000)
+            .gid(1000)
+            .args([
+                "--ignored",
+                "--exact",
+                "ordinary_pipeline_stopped_client",
+                "--test-threads=1",
+                "--nocapture",
+            ])
+            .output()
+            .unwrap();
+        assert!(stopped.status.success(), "{stopped:?}");
+        assert!(String::from_utf8_lossy(&stopped.stdout).contains("KF_PIPELINE_STOPPED=1"));
+        for name in ["tasks", "captures", "identities"] {
+            assert_eq!(
+                fs::read_dir(format!("/var/lib/kf-judge/{name}"))
+                    .unwrap()
+                    .count(),
+                0
+            );
+        }
+        println!(
+            "KF_MANAGED_PIPELINE_PROOF {}",
+            serde_json::json!({"worker_uid":1000,"language":language,"legacy":legacy,"job":job,"actual":actual,"staging_empty":true,"global_stopped":true})
+        );
+        actual
+    }
+}
+impl Drop for Fixture {
+    fn drop(&mut self) {
+        let _ = self.process.kill();
+        let _ = self.process.wait();
+        for name in ["requests", "socket.json", "control.sock", "lock"] {
+            let _ = fs::remove_file(Path::new("/var/lib/kf-judge/factory").join(name));
+        }
+        let _ = fs::remove_dir("/var/lib/kf-judge/factory");
+        let _ = fs::remove_dir_all(&self.dir);
+    }
+}
+#[test]
+#[ignore = "ordinary peer subprocess inside the isolated pipeline VM"]
+fn ordinary_pipeline_stopped_client() {
+    assert_eq!(rustix::process::geteuid().as_raw(), 1000);
+    assert_eq!(rustix::process::getegid().as_raw(), 1000);
+    assert!(
+        FactoryClient::new(PathBuf::from("/var/lib/kf-judge/factory/control.sock"))
+            .stopped()
+            .unwrap()
+    );
+    println!("KF_PIPELINE_STOPPED=1");
+}
+#[test]
+#[ignore = "actual ordinary Worker typed Python/Java and old/new GNU only inside capped no-NIC/no-share VM"]
+fn ordinary_worker_prepares_once_and_runs_all_four_runtimes() {
+    let mut f = Fixture::new();
+    let points: &[(&[u8], &[u8])] = &[
+        (b"", b""),
+        (b"\0\xff\n", b"\0\xff\n"),
+        ("你好\n".as_bytes(), "你好\n".as_bytes()),
+    ];
+    let py = br#"import os,sys
+assert not os.path.exists('previous')
+assert not os.path.exists('expected')
+assert not os.path.exists('/kf')
+open('previous','w').close()
+sys.stdout.buffer.write(sys.stdin.buffer.read())
+"#;
+    f.execute(
+        v2::Language::Python313,
+        py,
+        points,
+        v1::Verdict::Accepted,
+        false,
+    );
+    f.execute(
+        v2::Language::Python313,
+        py,
+        &[(b"x", b"y"), (b"", b"")],
+        v1::Verdict::WrongAnswer,
+        false,
+    );
+    f.execute(
+        v2::Language::Python313,
+        b"def bad(:\n",
+        &[(b"", b"")],
+        v1::Verdict::CompilationError,
+        false,
+    );
+    f.execute(
+        v2::Language::Python313,
+        b"raise MemoryError('fake kernel OOM')\n",
+        &[(b"", b"")],
+        v1::Verdict::RuntimeError,
+        false,
+    );
+    let pyloop = f.execute(
+        v2::Language::Python313,
+        b"while True: pass\n",
+        &[(b"", b"")],
+        v1::Verdict::TimeLimitExceeded,
+        false,
+    );
+    assert_eq!(
+        pyloop["result"]["report"]["compilation"]["status"],
+        "succeeded"
+    );
+    let java=br#"import java.io.*; public class Main {static int counter=0;static class Inner {} public static void main(String[] args) throws Exception {new Runnable(){public void run(){}}.run();if(++counter!=1||new File("previous").exists()||new File("expected").exists()||new File("/kf").exists()||Main.class.getResource("Main.java")!=null)throw new AssertionError();new File("previous").createNewFile();Helper.copy();}} class Helper {static void copy() throws Exception {System.in.transferTo(System.out);}}"#;
+    let j = f.execute(
+        v2::Language::Java21,
+        java,
+        points,
+        v1::Verdict::Accepted,
+        false,
+    );
+    assert_eq!(
+        j["result"]["report"]["compilation"]["program"]["files"]
+            .as_array()
+            .unwrap()
+            .len(),
+        4
+    );
+    f.execute(
+        v2::Language::Java21,
+        java,
+        &[(b"x", b"y")],
+        v1::Verdict::WrongAnswer,
+        false,
+    );
+    for source in [
+        &b"public class Main { syntax error }"[..],
+        &b"public class Main {static {if(System.nanoTime()>0){while(true){}}}}"[..],
+        &b"package unwanted; public class Main {public static void main(String[] a){}}"[..],
+        &b"public class Wrong {public static void main(String[] a){}}"[..],
+    ] {
+        f.execute(
+            v2::Language::Java21,
+            source,
+            &[(b"", b"")],
+            v1::Verdict::CompilationError,
+            false,
+        );
+    }
+    f.execute(v2::Language::Java21,b"public class Main {public static void main(String[] a){throw new OutOfMemoryError(\"fake kernel OOM\");}}",&[(b"",b"")],v1::Verdict::RuntimeError,false);
+    gnu_points(&mut f, points);
+    assert!(
+        !fs::read_dir(&f.root.path)
+            .unwrap()
+            .any(|e| e.unwrap().file_type().unwrap().is_dir())
+    );
+    println!("KF_MANAGED_PIPELINE_STATUS=0");
+}
+
+fn gnu_points(f: &mut Fixture, points: &[(&[u8], &[u8])]) {
+    let c=b"#include <stdio.h>\nint main(void){int c;while((c=getchar())!=EOF)putchar(c);return 0;}\n";
+    let cpp = b"#include <iostream>\nint main(){char c;while(std::cin.get(c))std::cout.put(c);}\n";
+    f.execute(v2::Language::C17, c, points, v1::Verdict::Accepted, false);
+    f.execute(
+        v2::Language::Cpp20,
+        cpp,
+        points,
+        v1::Verdict::Accepted,
+        false,
+    );
+    f.execute(v2::Language::C17, c, points, v1::Verdict::Accepted, true);
+    f.execute(
+        v2::Language::Cpp20,
+        cpp,
+        points,
+        v1::Verdict::Accepted,
+        true,
+    );
+}
+#[test]
+#[ignore = "targeted actual ordinary Worker new/legacy GNU after typed staging changes inside capped VM"]
+fn ordinary_worker_runs_versioned_and_legacy_gnu() {
+    let mut f = Fixture::new();
+    let points: &[(&[u8], &[u8])] = &[
+        (b"", b""),
+        (b"\0\xff\n", b"\0\xff\n"),
+        ("你好\n".as_bytes(), "你好\n".as_bytes()),
+    ];
+    gnu_points(&mut f, points);
+    assert!(
+        !fs::read_dir(&f.root.path)
+            .unwrap()
+            .any(|e| e.unwrap().file_type().unwrap().is_dir())
+    );
+    println!("KF_MANAGED_GNU_PIPELINE_STATUS=0");
+}

judge/crates/worker/src/consumer.rs

@@ -1,9 +1,16 @@
 //! Ordinary transport adapter. AMQP is a bounded wake-up; claim is sole authority.
 use crate::pipeline::{ArtifactPort, Attempt};
+use crate::{pipeline_v2::RuntimeAttempt, runtime_specs::AuthorizedJob};
 use futures_lite::{StreamExt, future::block_on};
-use kf_contracts::judging_versions::{JobManifest as VersionedManifest, NodeAdmission};
+use kf_contracts::judging_v2 as v2;
+use kf_contracts::judging_versions::{
+    self as versions, JobManifest as VersionedManifest, NodeAdmission,
+};
 use kf_contracts::{JudgeTaskNotification, SchemaVersion, judging::*};
-use kf_engine::submission::{StageFailure, judge_submission};
+use kf_engine::{
+    submission::{StageFailure, judge_submission},
+    submission_v2,
+};
 use kf_sandbox::{artifacts::StagePublication, factory::FactoryClient};
 use lapin::{Connection, ConnectionProperties, options::*, types::FieldTable};
 use reqwest::blocking::Client;
@@ -20,6 +27,37 @@ use std::{
     time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 const QUEUE: &str = "kf.judge.evaluations.v1";
+/// Transport projection after strict versioned wire decoding. It never changes
+/// the authorized manifest or its canonical bytes.
+#[derive(Clone)]
+struct Grant {
+    version: u16,
+    binding: LeaseBinding,
+    lease_token: LeaseToken,
+    attempt_number: u32,
+    expires_at_unix_ms: u64,
+    manifest_digest: Digest,
+}
+impl From<versions::LeaseGrant> for Grant {
+    fn from(wire: versions::LeaseGrant) -> Self {
+        macro_rules! fields {
+            ($g:ident, $version:literal) => {
+                Self {
+                    version: $version,
+                    binding: $g.binding,
+                    lease_token: $g.lease_token,
+                    attempt_number: $g.attempt_number,
+                    expires_at_unix_ms: $g.expires_at_unix_ms,
+                    manifest_digest: $g.manifest_digest,
+                }
+            };
+        }
+        match wire {
+            versions::LeaseGrant::V1(g) => fields!(g, 1),
+            versions::LeaseGrant::V2(g) => fields!(g, 2),
+        }
+    }
+}
 #[derive(Debug)]
 struct TransportError {
     operation: &'static str,
@@ -139,7 +177,7 @@ impl Remote {
         &self,
         route: &str,
         body: &T,
-        grant: Option<&LeaseGrant>,
+        grant: Option<&Grant>,
     ) -> io::Result<reqwest::blocking::Response> {
         let mut r = self
             .client
@@ -173,8 +211,9 @@ impl Remote {
         let v: Vec<Pending> = serde_json::from_slice(&b)
             .map_err(|_| transport_error("coordinator_pending", "decode"))?;
         if v.len() > 16
-            || v.iter()
+            || v.iter().any(|p| {
-                .any(|p| p.case_count == 0 || p.case_count > MAX_CASES)
+                p.case_count == 0 || p.case_count > MAX_CASES || !matches!(p.schema_version, 1 | 2)
+            })
         {
             return Err(error());
         }
@@ -193,12 +232,7 @@ impl Remote {
             if p.binding.worker_id != c.worker_id {
                 return Err(error());
             }
-            let r = StoppedReport {
+            let r = p.stopped(clean)?;
-                schema_version: SchemaVersion,
-                kind: StoppedKind::Value,
-                binding: p.binding,
-                cleanup_complete: clean,
-            };
             if self.post("/internal/stopped", &r, None)?.status().as_u16() != 204 {
                 return Err(error());
             }
@@ -209,13 +243,37 @@ impl Remote {
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
 struct Pending {
+    #[serde(default = "legacy_version")]
+    schema_version: u16,
     binding: LeaseBinding,
     case_count: usize,
 }
+fn legacy_version() -> u16 {
+    1
+}
+impl Pending {
+    fn stopped(self, clean: bool) -> io::Result<versions::StoppedReport> {
+        match self.schema_version {
+            1 => Ok(versions::StoppedReport::V1(StoppedReport {
+                schema_version: SchemaVersion,
+                kind: StoppedKind::Value,
+                binding: self.binding,
+                cleanup_complete: clean,
+            })),
+            2 => Ok(versions::StoppedReport::V2(v2::StoppedReport {
+                schema_version: v2::SchemaVersion,
+                kind: StoppedKind::Value,
+                binding: self.binding,
+                cleanup_complete: clean,
+            })),
+            _ => Err(error()),
+        }
+    }
+}
 struct RemoteArtifacts {
     client: Client,
     base: String,
-    grant: LeaseGrant,
+    grant: Grant,
 }
 impl RemoteArtifacts {
     fn headers(&self, r: reqwest::blocking::RequestBuilder) -> reqwest::blocking::RequestBuilder {
@@ -234,25 +292,28 @@ impl RemoteArtifacts {
             self.grant.binding.attempt_id.as_uuid()
         )
     }
-    fn manifest(&self) -> io::Result<JobManifest> {
+    fn manifest(&self) -> io::Result<VersionedManifest> {
         let response = self
             .headers(self.client.get(self.url("/manifest")))
             .send()
             .map_err(|_| error())?;
-        let m: JobManifest =
+        authorized_manifest(&Remote::bytes(response, MAX_WIRE_BYTES)?, &self.grant)
-            decode_wire(&Remote::bytes(response, MAX_WIRE_BYTES)?).map_err(|_| error())?;
-        let digest = format!(
-            "{:x}",
-            Sha256::digest(manifest_bytes(&m).map_err(|_| error())?)
-        );
-        if digest != self.grant.manifest_digest.as_str()
-            || m.evaluation_id != self.grant.binding.evaluation_id
-        {
-            return Err(error());
-        }
-        Ok(m)
     }
 }
+fn authorized_manifest(raw: &[u8], grant: &Grant) -> io::Result<VersionedManifest> {
+    let m: VersionedManifest = decode_wire(raw).map_err(|_| error())?;
+    let digest = format!(
+        "{:x}",
+        Sha256::digest(m.canonical_bytes().map_err(|_| error())?)
+    );
+    if digest != grant.manifest_digest.as_str()
+        || m.evaluation_id() != grant.binding.evaluation_id
+        || m.version() != grant.version
+    {
+        return Err(error());
+    }
+    Ok(m)
+}
 impl ArtifactPort for RemoteArtifacts {
     fn read(&mut self, reference: &ArtifactReference) -> io::Result<Vec<u8>> {
         let r = self
@@ -282,7 +343,7 @@ fn lease_deadline(ms: u64) -> io::Result<Instant> {
         .ok_or_else(error)?;
     Ok(Instant::now() + Duration::from_millis(remaining))
 }
-fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Result<()> {
+fn execute(c: &Config, r: &Remote, g: Grant, stop: &AtomicBool) -> io::Result<()> {
     if g.binding.worker_id != c.worker_id {
         return Err(error());
     }
@@ -293,7 +354,7 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu
     };
     let m = artifacts.manifest()?;
     c.admission
-        .admit(&VersionedManifest::V1(m.clone()), g.attempt_number)
+        .admit(&m, g.attempt_number)
         .map_err(|_| error())?;
     let cancel = AtomicBool::new(false);
     let done = AtomicBool::new(false);
@@ -329,23 +390,9 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu
         });
         let result = (|| {
             let mut sequence = 0u64;
-            let mut adapter = Attempt::new(
+            let mut control = |stage, index| {
-                artifacts,
-                c.socket.clone(),
-                c.staging.clone(),
-                m.job_wall_time_us,
-                &cancel,
-            )?
-            .with_control(|stage, index| {
                 sequence += 1;
-                let report = StageReport {
+                let report = stage_report(&m, g.binding.clone(), sequence, stage, index);
-                    schema_version: SchemaVersion,
-                    kind: StageKind::Value,
-                    binding: g.binding.clone(),
-                    sequence,
-                    stage,
-                    case_index: index,
-                };
                 for _ in 0..2 {
                     match r.post("/internal/stage", &report, Some(&g)) {
                         Ok(response) if response.status().as_u16() == 204 => return Ok(()),
@@ -359,22 +406,64 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu
                     code: FailureCode::LeaseLost,
                     cleanup_complete: true,
                 })
-            });
+            };
-            let mut result = judge_submission(
+            let mut report = match &m {
-                &mut adapter,
+                VersionedManifest::V1(manifest) => {
-                &m,
+                    let mut adapter = Attempt::new(
-                g.binding.clone(),
+                        artifacts,
-                g.manifest_digest.clone(),
+                        c.socket.clone(),
-            )
+                        c.staging.clone(),
-            .map_err(|_| error())?;
+                        manifest.job_wall_time_us,
-            transport_cancellation(&mut result.report);
+                        &cancel,
-            result
+                    )?
-                .report
+                    .with_control(&mut control);
+                    versions::JudgementReport::V1(
+                        judge_submission(
+                            &mut adapter,
+                            manifest,
+                            g.binding.clone(),
+                            g.manifest_digest.clone(),
+                        )
+                        .map_err(|_| error())?
+                        .report,
+                    )
+                }
+                VersionedManifest::V2(manifest) => {
+                    let authorized = AuthorizedJob::admit(
+                        manifest.clone(),
+                        g.binding.clone(),
+                        g.manifest_digest.clone(),
+                        g.attempt_number,
+                        &c.admission,
+                    )
+                    .map_err(|_| error())?;
+                    let mut adapter = RuntimeAttempt::new(
+                        authorized,
+                        artifacts,
+                        c.socket.clone(),
+                        c.staging.clone(),
+                        &cancel,
+                    )?
+                    .with_control(&mut control);
+                    versions::JudgementReport::V2(
+                        submission_v2::judge_submission(
+                            &mut adapter,
+                            manifest,
+                            g.binding.clone(),
+                            g.manifest_digest.clone(),
+                        )
+                        .map_err(|_| error())?
+                        .report,
+                    )
+                }
+            };
+            transport_cancellation_versioned(&mut report);
+            report
                 .validate_for_manifest(&m, &g.manifest_digest)
                 .map_err(|_| error())?;
             // Same immutable report only; a transport ambiguity never becomes success.
             for _ in 0..3 {
-                match r.post("/internal/report", &result.report, Some(&g)) {
+                match r.post("/internal/report", &report, Some(&g)) {
                     Ok(response) if response.status().as_u16() == 204 => return Ok(()),
                     Ok(response) if response.status().as_u16() == 409 => break,
                     _ => {}
@@ -388,6 +477,47 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu
         result
     })
 }
+fn stage_report(
+    m: &VersionedManifest,
+    binding: LeaseBinding,
+    sequence: u64,
+    stage: Stage,
+    case_index: Option<u16>,
+) -> versions::StageReport {
+    match m {
+        VersionedManifest::V1(_) => versions::StageReport::V1(StageReport {
+            schema_version: SchemaVersion,
+            kind: StageKind::Value,
+            binding,
+            sequence,
+            stage,
+            case_index,
+        }),
+        VersionedManifest::V2(manifest) => versions::StageReport::V2(v2::StageReport {
+            schema_version: v2::SchemaVersion,
+            kind: StageKind::Value,
+            binding,
+            sequence,
+            stage,
+            case_index,
+            preparation_kind: manifest.language.contract().preparation_kind,
+        }),
+    }
+}
+fn transport_cancellation_versioned(report: &mut versions::JudgementReport) {
+    match report {
+        versions::JudgementReport::V1(r) => transport_cancellation(r),
+        versions::JudgementReport::V2(r) => {
+            if r.verdict == Verdict::Cancelled {
+                r.verdict = Verdict::SystemError;
+                r.failure_code = Some(FailureCode::LeaseLost);
+                if r.compilation.status == CompilationStatus::Cancelled {
+                    r.compilation.status = CompilationStatus::SystemError;
+                }
+            }
+        }
+    }
+}
 // A task stopped by service shutdown/transport loss is an infrastructure failure.
 // User cancellation is authoritative in Coordinator and completes via actual stop proof.
 fn transport_cancellation(report: &mut JudgementReport) {
@@ -486,9 +616,9 @@ async fn consume(c: &Config, r: &Remote, stop: &AtomicBool) -> io::Result<()> {
         let grant = r.post("/internal/claim", &notice, None)?;
         match grant.status().as_u16() {
             200 => {
-                let g: LeaseGrant =
+                let g: versions::LeaseGrant =
                     decode_wire(&Remote::bytes(grant, MAX_WIRE_BYTES)?).map_err(|_| error())?;
-                let result = execute(c, r, g, stop);
+                let result = execute(c, r, g.into(), stop);
                 if result.is_err() && !r.reconcile(c)? {
                     return Err(error());
                 }
@@ -539,6 +669,86 @@ pub fn run(c: Config, stop: Arc<AtomicBool>) -> io::Result<()> {
 mod tests {
     use super::*;
     #[test]
+    fn versioned_manifest_is_bound_to_lease_generation_digest_and_identity() {
+        for (raw, version) in [
+            (
+                &include_bytes!("../../../../tests/fixtures/contracts/judge-manifest.valid.json")[..],
+                1,
+            ),
+            (
+                &include_bytes!(
+                    "../../../../tests/fixtures/contracts-v2/judge-manifest.valid-python313.json"
+                )[..],
+                2,
+            ),
+        ] {
+            let m: VersionedManifest = decode_wire(raw).unwrap();
+            let base: versions::LeaseGrant = decode_wire(include_bytes!(
+                "../../../../tests/fixtures/contracts-v2/judge-lease.valid.json"
+            ))
+            .unwrap();
+            let mut g: Grant = base.into();
+            g.version = version;
+            g.binding.evaluation_id = m.evaluation_id();
+            g.manifest_digest = format!("{:x}", Sha256::digest(m.canonical_bytes().unwrap()))
+                .try_into()
+                .unwrap();
+            assert_eq!(authorized_manifest(raw, &g).unwrap(), m);
+            g.version = 3 - version;
+            assert!(authorized_manifest(raw, &g).is_err());
+            g.version = version;
+            g.manifest_digest = "f".repeat(64).try_into().unwrap();
+            assert!(authorized_manifest(raw, &g).is_err());
+        }
+    }
+    #[test]
+    fn preparation_stage_and_pending_stop_retain_exact_protocol_generation() {
+        let raw = include_bytes!(
+            "../../../../tests/fixtures/contracts-v2/judge-manifest.valid-python313.json"
+        );
+        let m: VersionedManifest = decode_wire(raw).unwrap();
+        let g: versions::LeaseGrant = decode_wire(include_bytes!(
+            "../../../../tests/fixtures/contracts-v2/judge-lease.valid.json"
+        ))
+        .unwrap();
+        let mut g: Grant = g.into();
+        g.binding.evaluation_id = m.evaluation_id();
+        let report = stage_report(&m, g.binding.clone(), 1, Stage::Compiling, None);
+        report.validate_for_manifest(&m).unwrap();
+        let value = serde_json::to_value(report).unwrap();
+        assert_eq!(value["schema_version"], 2);
+        assert_eq!(value["preparation_kind"], "syntax_check");
+        for version in [1, 2, 3] {
+            let mut pending = serde_json::json!({"binding":g.binding,"case_count":2});
+            if version != 1 {
+                pending["schema_version"] = serde_json::json!(version);
+            }
+            let p: Pending = serde_json::from_value(pending).unwrap();
+            if version == 3 {
+                assert!(p.stopped(true).is_err());
+                continue;
+            }
+            let stopped = serde_json::to_value(p.stopped(true).unwrap()).unwrap();
+            assert_eq!(stopped["schema_version"], version);
+            assert_eq!(
+                stopped,
+                serde_json::json!({"schema_version":version,"kind":"judge.stopped","binding":g.binding,"cleanup_complete":true})
+            );
+        }
+        let mut legacy: VersionedManifest = decode_wire(include_bytes!(
+            "../../../../tests/fixtures/contracts/judge-manifest.valid.json"
+        ))
+        .unwrap();
+        if let VersionedManifest::V1(m) = &mut legacy {
+            m.evaluation_id = g.binding.evaluation_id;
+        }
+        let report = stage_report(&legacy, g.binding, 1, Stage::Compiling, None);
+        let value = serde_json::to_value(&report).unwrap();
+        assert_eq!(value["schema_version"], 1);
+        assert!(value.get("preparation_kind").is_none());
+        report.validate_for_manifest(&legacy).unwrap();
+    }
+    #[test]
     fn amqp_numeric_authority_and_default_vhost_are_preserved() {
         for (raw, host, vhost) in [
             ("amqp://fixture:temporary@127.0.0.2:5672/", "127.0.0.2", "/"),

judge/crates/worker/src/java_entry.rs

@@ -0,0 +1,212 @@
+//! Bounded metadata inspection of sealed javac output, never class loading.
+//! Format: https://docs.oracle.com/javase/specs/jvms/se21/html/jvms-4.html
+//! Bytecode verification and execution remain inside the isolated JVM.
+use std::collections::{BTreeMap, HashSet};
+struct Reader<'a>(&'a [u8]);
+impl<'a> Reader<'a> {
+    fn take(&mut self, n: usize) -> Result<&'a [u8], &'static str> {
+        let value = self.0.get(..n).ok_or("truncated class metadata")?;
+        self.0 = &self.0[n..];
+        Ok(value)
+    }
+    fn u8(&mut self) -> Result<u8, &'static str> {
+        Ok(self.take(1)?[0])
+    }
+    fn u16(&mut self) -> Result<u16, &'static str> {
+        Ok(u16::from_be_bytes(
+            self.take(2)?.try_into().expect("two bytes"),
+        ))
+    }
+    fn u32(&mut self) -> Result<u32, &'static str> {
+        Ok(u32::from_be_bytes(
+            self.take(4)?.try_into().expect("four bytes"),
+        ))
+    }
+    fn attributes(&mut self, pool: &[Constant<'a>]) -> Result<(), &'static str> {
+        for _ in 0..self.u16()? {
+            utf8(pool, self.u16()?)?;
+            let length = self.u32()? as usize;
+            self.take(length)?;
+        }
+        Ok(())
+    }
+}
+enum Constant<'a> {
+    Other,
+    Utf8(&'a [u8]),
+    Class(u16),
+}
+fn utf8<'a>(pool: &[Constant<'a>], index: u16) -> Result<&'a [u8], &'static str> {
+    match pool.get(usize::from(index)) {
+        Some(Constant::Utf8(bytes)) if index != 0 => Ok(bytes),
+        _ => Err("invalid class UTF8 reference"),
+    }
+}
+fn class<'a>(pool: &[Constant<'a>], index: u16) -> Result<&'a [u8], &'static str> {
+    match pool.get(usize::from(index)) {
+        Some(Constant::Class(name)) if index != 0 => utf8(pool, *name),
+        _ => Err("invalid class name reference"),
+    }
+}
+struct Info<'a> {
+    parent: &'a [u8],
+    main: bool,
+}
+fn parse<'a>(path: &str, data: &'a [u8]) -> Result<Info<'a>, &'static str> {
+    if data.len() as u64 > kf_contracts::judging::MAX_PACKAGE_BYTES {
+        return Err("class metadata byte budget");
+    }
+    let mut r = Reader(data);
+    if r.u32()? != 0xcafebabe || r.u16()? != 0 || r.u16()? != 65 {
+        return Err("class differs from fixed javac release");
+    }
+    let count = usize::from(r.u16()?);
+    if count < 2 {
+        return Err("empty class constant pool");
+    }
+    let mut pool = vec![Constant::Other];
+    while pool.len() < count {
+        let c = match r.u8()? {
+            1 => {
+                let n = usize::from(r.u16()?);
+                Constant::Utf8(r.take(n)?)
+            }
+            7 => Constant::Class(r.u16()?),
+            3 | 4 | 9 | 10 | 11 | 12 | 17 | 18 => {
+                r.take(4)?;
+                Constant::Other
+            }
+            8 | 16 | 19 | 20 => {
+                r.take(2)?;
+                Constant::Other
+            }
+            15 => {
+                r.take(3)?;
+                Constant::Other
+            }
+            5 | 6 => {
+                r.take(8)?;
+                if pool.len() + 1 >= count {
+                    return Err("wide constant slot budget");
+                }
+                pool.push(Constant::Other);
+                Constant::Other
+            }
+            _ => return Err("unknown class constant"),
+        };
+        pool.push(c);
+    }
+    r.u16()?;
+    if class(&pool, r.u16()?)?
+        != path
+            .strip_suffix(".class")
+            .ok_or("class file path")?
+            .as_bytes()
+    {
+        return Err("class identity differs from sealed file");
+    }
+    let parent = class(&pool, r.u16()?)?;
+    for _ in 0..r.u16()? {
+        class(&pool, r.u16()?)?;
+    }
+    for _ in 0..r.u16()? {
+        r.u16()?;
+        utf8(&pool, r.u16()?)?;
+        utf8(&pool, r.u16()?)?;
+        r.attributes(&pool)?;
+    }
+    let mut main = false;
+    for _ in 0..r.u16()? {
+        let flags = r.u16()?;
+        let name = utf8(&pool, r.u16()?)?;
+        let descriptor = utf8(&pool, r.u16()?)?;
+        main |=
+            flags & 0x0009 == 0x0009 && name == b"main" && descriptor == b"([Ljava/lang/String;)V";
+        r.attributes(&pool)?;
+    }
+    r.attributes(&pool)?;
+    if !r.0.is_empty() {
+        return Err("trailing class metadata");
+    }
+    Ok(Info { parent, main })
+}
+pub(crate) fn has_main(files: &[(String, Vec<u8>)]) -> Result<bool, &'static str> {
+    if files.is_empty()
+        || files.len() > kf_contracts::judging_v2::MAX_CLASS_FILES
+        || files.iter().map(|(_, d)| d.len() as u64).sum::<u64>()
+            > kf_contracts::judging::MAX_PACKAGE_BYTES
+    {
+        return Err("class collection metadata budget");
+    }
+    let mut classes = BTreeMap::new();
+    for (path, bytes) in files {
+        if !kf_contracts::judging_v2::class_name(path) {
+            return Err("unsupported class path");
+        }
+        if classes
+            .insert(
+                path.strip_suffix(".class").expect("class path").as_bytes(),
+                parse(path, bytes)?,
+            )
+            .is_some()
+        {
+            return Err("duplicate class identity");
+        }
+    }
+    let mut name = &b"Main"[..];
+    let mut seen = HashSet::new();
+    while let Some(info) = classes.get(name) {
+        if !seen.insert(name) {
+            return Err("class superclass cycle");
+        }
+        if info.main {
+            return Ok(true);
+        }
+        name = info.parent;
+    }
+    Ok(false)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn minimal(flags: u16, descriptor: &[u8]) -> Vec<u8> {
+        let mut b = vec![0xca, 0xfe, 0xba, 0xbe, 0, 0, 0, 65, 0, 7];
+        for name in [
+            &b"Main"[..],
+            &b"java/lang/Object"[..],
+            &b"main"[..],
+            descriptor,
+        ] {
+            b.push(1);
+            b.extend_from_slice(&(name.len() as u16).to_be_bytes());
+            b.extend_from_slice(name);
+        }
+        b.extend_from_slice(&[7, 0, 1, 7, 0, 2, 0, 0x21, 0, 5, 0, 6, 0, 0, 0, 0, 0, 1]);
+        b.extend_from_slice(&flags.to_be_bytes());
+        b.extend_from_slice(&[0, 3, 0, 4, 0, 0, 0, 0]);
+        b
+    }
+    #[test]
+    fn entry_signature_and_truncation_are_checked_without_loading_classes() {
+        let yes = minimal(9, b"([Ljava/lang/String;)V");
+        assert!(has_main(&[("Main.class".into(), yes.clone())]).unwrap());
+        for flags in [0, 1, 8] {
+            assert!(
+                !has_main(&[(
+                    "Main.class".into(),
+                    minimal(flags, b"([Ljava/lang/String;)V")
+                )])
+                .unwrap()
+            );
+        }
+        assert!(!has_main(&[("Main.class".into(), minimal(9, b"()V"))]).unwrap());
+        for n in 0..yes.len() {
+            assert!(parse("Main.class", &yes[..n]).is_err());
+        }
+        assert!(parse("Other.class", &yes).is_err());
+        let mut extra = yes;
+        extra.push(0);
+        assert!(parse("Main.class", &extra).is_err());
+    }
+}

judge/crates/worker/src/lib.rs

@@ -116,5 +116,14 @@ pub mod native {
 #[cfg(target_os = "linux")]
 pub mod pipeline;
 
+#[cfg(target_os = "linux")]
+pub mod runtime_specs;
+
+#[cfg(target_os = "linux")]
+pub mod pipeline_v2;
+
+#[cfg(target_os = "linux")]
+mod java_entry;
+
 #[cfg(target_os = "linux")]
 pub mod consumer;

judge/crates/worker/src/main.rs

@@ -32,6 +32,18 @@ enum Command {
         #[arg(long)]
         staging: PathBuf,
     },
+    JudgeSubmissionV2 {
+        #[arg(long)]
+        socket: PathBuf,
+        #[arg(long)]
+        job: PathBuf,
+        #[arg(long)]
+        artifacts: PathBuf,
+        #[arg(long)]
+        staging: PathBuf,
+        #[arg(long)]
+        admission: PathBuf,
+    },
     JudgeOnce {
         #[arg(long)]
         socket: PathBuf,
@@ -114,6 +126,36 @@ async fn main() -> ExitCode {
                 ExitCode::from(2)
             }
         }
+        Some(Command::JudgeSubmissionV2 {
+            socket,
+            job,
+            artifacts,
+            staging,
+            admission,
+        }) => {
+            let result = (|| {
+                use std::os::unix::fs::{MetadataExt, PermissionsExt};
+                let m = std::fs::symlink_metadata(&admission)?;
+                if !m.is_file() || m.nlink() != 1 || m.permissions().mode() & 0o7777 != 0o600 {
+                    return Err(std::io::Error::other("operator admission file refused"));
+                }
+                let policy = kf_contracts::judging::decode_wire(
+                    &kf_sandbox::execution::read_bounded_file(&admission, 32768)?,
+                )
+                .map_err(std::io::Error::other)?;
+                kf_worker::pipeline_v2::judge_local(&socket, &job, &artifacts, &staging, &policy)
+            })();
+            match result {
+                Ok(result) => {
+                    println!("{}", serde_json::json!(result));
+                    ExitCode::SUCCESS
+                }
+                Err(_) => {
+                    eprintln!("submission_failed");
+                    ExitCode::from(2)
+                }
+            }
+        }
         Some(Command::JudgeOnce {
             socket,
             job,

judge/crates/worker/src/pipeline.rs

@@ -61,12 +61,12 @@ impl ArtifactPort for LocalArtifacts {
         Ok(data)
     }
 }
-fn digest(data: &[u8]) -> Digest {
+pub(crate) fn digest(data: &[u8]) -> Digest {
     format!("{:x}", Sha256::digest(data))
         .try_into()
         .expect("SHA256 representation")
 }
-fn verify(r: &ArtifactReference, data: &[u8]) -> io::Result<()> {
+pub(crate) fn verify(r: &ArtifactReference, data: &[u8]) -> io::Result<()> {
     if data.len() as u64 != r.bytes || digest(data) != r.digest {
         return Err(io::Error::other("artifact digest mismatch"));
     }
@@ -78,7 +78,7 @@ fn stage_error(code: FailureCode, cleanup_complete: bool) -> StageFailure {
         cleanup_complete,
     }
 }
-fn limits(r: &RunLimits) -> ResourceLimits {
+pub(crate) fn limits(r: &RunLimits) -> ResourceLimits {
     ResourceLimits {
         cpu_time_us: r.cpu_time_us,
         wall_time_us: r.wall_time_us,
@@ -92,13 +92,13 @@ fn limits(r: &RunLimits) -> ResourceLimits {
         workdir_bytes: r.workdir_bytes,
     }
 }
-fn metric(m: &Measurement<u64>) -> Option<u64> {
+pub(crate) fn metric(m: &Measurement<u64>) -> Option<u64> {
     match m {
         Measurement::Measured { value } => Some(*value),
         Measurement::Unavailable { .. } => None,
     }
 }
-fn kind(o: ExecutionOutcome) -> ExecutionKind {
+pub(crate) fn kind(o: ExecutionOutcome) -> ExecutionKind {
     match o {
         ExecutionOutcome::Exited { code } => ExecutionKind::Exited { code },
         ExecutionOutcome::Signaled { .. } => ExecutionKind::Signaled,
@@ -210,7 +210,10 @@ impl<'a, A: ArtifactPort> Attempt<'a, A> {
         self.control = Some(Box::new(control));
         self
     }
-    fn execute(
+    pub(crate) fn staging_path(&self) -> &Path {
+        &self.staging
+    }
+    pub(crate) fn execute(
         &mut self,
         spec: ExecutionSpec,
         publication: &mut StagePublication,

judge/crates/worker/src/pipeline_v2.rs

@@ -0,0 +1,414 @@
+//! Ordinary multi-runtime adapter. User source is only checked/compiled/run by
+//! isolated Root helpers; Worker handles bounded bytes and sealed capabilities.
+use crate::{
+    pipeline::{ArtifactPort, Attempt, LocalArtifacts, StageEvidence, kind, metric, verify},
+    runtime_specs::AuthorizedJob,
+};
+use kf_contracts::{judging as v1, judging_v2 as v2, judging_versions::NodeAdmission};
+use kf_engine::{Observation, submission as old, submission_v2 as engine};
+use kf_sandbox::{
+    artifacts::{SnapshotFile, StagePublication},
+    outcome::ExecutionOutcome,
+};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest as _, Sha256};
+use std::{
+    io::{self, Read},
+    path::{Path, PathBuf},
+    sync::atomic::AtomicBool,
+};
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Job {
+    pub manifest: v2::JobManifest,
+    pub binding: v1::LeaseBinding,
+    pub manifest_digest: v1::Digest,
+    pub attempt_number: u32,
+}
+#[derive(Serialize)]
+pub struct Judged {
+    pub result: engine::SubmissionResult,
+    pub stages: Vec<StageEvidence>,
+}
+struct SealedProgram {
+    metadata: v2::CompiledProgram,
+    files: Vec<SnapshotFile>,
+}
+pub struct RuntimeAttempt<'a, A: ArtifactPort> {
+    authorized: AuthorizedJob,
+    io: Attempt<'a, A>,
+    program: Option<SealedProgram>,
+}
+fn failure(code: v1::FailureCode, cleanup_complete: bool) -> old::StageFailure {
+    old::StageFailure {
+        code,
+        cleanup_complete,
+    }
+}
+fn artifact_error(_: io::Error) -> old::StageFailure {
+    failure(v1::FailureCode::ArtifactUnavailable, true)
+}
+fn snapshot_reference(file: &SnapshotFile) -> Result<v1::ArtifactReference, old::StageFailure> {
+    let mut h = Sha256::new();
+    let mut reader = file.reader().map_err(artifact_error)?;
+    let mut buffer = [0u8; 16384];
+    let mut total = 0u64;
+    loop {
+        let n = reader.read(&mut buffer).map_err(artifact_error)?;
+        if n == 0 {
+            break;
+        }
+        total += n as u64;
+        if total > file.bytes() {
+            return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+        }
+        h.update(&buffer[..n]);
+    }
+    if total != file.bytes() {
+        return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+    }
+    Ok(v1::ArtifactReference {
+        digest: format!("{:x}", h.finalize()).try_into().expect("SHA256"),
+        bytes: total,
+    })
+}
+fn diagnostics(result: &kf_sandbox::rpc::Executed) -> Result<(String, bool), old::StageFailure> {
+    let mut raw = Vec::new();
+    result
+        .stderr
+        .reader()
+        .map_err(artifact_error)?
+        .take(65537)
+        .read_to_end(&mut raw)
+        .map_err(artifact_error)?;
+    let mut text = String::from_utf8_lossy(&raw).into_owned();
+    let mut truncated = result.report.stderr.truncated || text.len() > 65536;
+    if text.len() > 65536 {
+        let mut end = 65536;
+        while !text.is_char_boundary(end) {
+            end -= 1;
+        }
+        text.truncate(end);
+        truncated = true;
+    }
+    Ok((text, truncated))
+}
+impl<'a, A: ArtifactPort> RuntimeAttempt<'a, A> {
+    pub fn new(
+        authorized: AuthorizedJob,
+        artifacts: A,
+        socket: PathBuf,
+        staging: PathBuf,
+        cancellation: &'a AtomicBool,
+    ) -> io::Result<Self> {
+        let io = Attempt::new(
+            artifacts,
+            socket,
+            staging,
+            authorized.manifest().job_wall_time_us,
+            cancellation,
+        )?;
+        Ok(Self {
+            authorized,
+            io,
+            program: None,
+        })
+    }
+    pub fn with_control(
+        mut self,
+        control: impl FnMut(v1::Stage, Option<u16>) -> Result<(), old::StageFailure> + 'a,
+    ) -> Self {
+        self.io = self.io.with_control(control);
+        self
+    }
+    pub fn into_stages(self) -> Vec<StageEvidence> {
+        self.io.stages
+    }
+}
+impl<A: ArtifactPort> engine::PreparationPort for RuntimeAttempt<'_, A> {
+    fn prepare(
+        &mut self,
+        m: &v2::JobManifest,
+        binding: &v1::LeaseBinding,
+        run_id: &str,
+    ) -> Result<engine::Compiled, old::StageFailure> {
+        if m != self.authorized.manifest()
+            || binding != self.authorized.binding()
+            || self.program.is_some()
+        {
+            return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+        }
+        let source = self.io.artifacts.read(&m.source).map_err(artifact_error)?;
+        verify(&m.source, &source).map_err(artifact_error)?;
+        let mut publication =
+            StagePublication::create(self.io.staging_path(), run_id).map_err(artifact_error)?;
+        publication
+            .put("source", &mut source.as_slice(), m.source.bytes)
+            .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, false))?;
+        let checked_source = if m.language == v2::Language::Python313 {
+            let file = publication
+                .root()
+                .snapshot_artifact("source", m.source.bytes)
+                .map_err(artifact_error)?;
+            if snapshot_reference(&file)? != m.source {
+                return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+            }
+            Some(file)
+        } else {
+            None
+        };
+        drop(source);
+        let spec = self
+            .authorized
+            .preparation_spec(run_id.into())
+            .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, true))?;
+        let validated = spec
+            .clone()
+            .validate(&kf_sandbox::spec::SpecPolicy::default())
+            .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, true))?;
+        let result = self.io.execute(spec, &mut publication)?;
+        let mut status = match result
+            .report
+            .classify(&validated)
+            .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, false))?
+        {
+            ExecutionOutcome::Exited { code: 0 } => v1::CompilationStatus::Succeeded,
+            ExecutionOutcome::Exited { .. } | ExecutionOutcome::Signaled { .. } => {
+                v1::CompilationStatus::CompilationError
+            }
+            ExecutionOutcome::LimitExceeded { .. } => v1::CompilationStatus::CompilationLimit,
+            ExecutionOutcome::Cancelled => v1::CompilationStatus::Cancelled,
+            _ => v1::CompilationStatus::SystemError,
+        };
+        let (mut text, truncated) = diagnostics(&result)?;
+        let cpu_time_us = metric(&result.report.usage.cpu_time_us);
+        let memory_peak_bytes = metric(&result.report.usage.memory_peak_bytes);
+        let mut files = Vec::new();
+        let mut sealed = Vec::new();
+        if status == v1::CompilationStatus::Succeeded {
+            if let Some(source) = checked_source {
+                if !result.outputs.is_empty() {
+                    return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+                }
+                files.push(v2::ProgramFile {
+                    path: "main.py".into(),
+                    artifact: m.source.clone(),
+                });
+                sealed.push(source);
+            } else {
+                for output in result.outputs {
+                    files.push(v2::ProgramFile {
+                        path: output.path,
+                        artifact: snapshot_reference(&output.file)?,
+                    });
+                    sealed.push(output.file);
+                }
+            }
+            if m.language == v2::Language::Java21 {
+                // No reflective loading or initialization outside the JVM. A
+                // successfully compiled unsupported entry is a user CE.
+                let mut metadata = Vec::with_capacity(files.len());
+                for (info, file) in files.iter().zip(&sealed) {
+                    let mut bytes = Vec::new();
+                    file.reader()
+                        .map_err(artifact_error)?
+                        .take(info.artifact.bytes + 1)
+                        .read_to_end(&mut bytes)
+                        .map_err(artifact_error)?;
+                    verify(&info.artifact, &bytes).map_err(artifact_error)?;
+                    metadata.push((info.path.clone(), bytes));
+                }
+                let has_main = if metadata.is_empty() {
+                    false
+                } else {
+                    crate::java_entry::has_main(&metadata)
+                        .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, true))?
+                };
+                if !has_main {
+                    status = v1::CompilationStatus::CompilationError;
+                    text="Java submission requires default-package Main with public static void main(String[]).".into();
+                }
+            }
+        }
+        let program = if status == v1::CompilationStatus::Succeeded {
+            let metadata = v2::CompiledProgram {
+                kind: m.language.contract().program_kind,
+                binding: binding.clone(),
+                source: m.source.clone(),
+                toolchain: m.toolchain.clone(),
+                files,
+            };
+            self.authorized
+                .validate_program(&metadata)
+                .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, true))?;
+            self.program = Some(SealedProgram {
+                metadata: metadata.clone(),
+                files: sealed,
+            });
+            Some(metadata)
+        } else {
+            None
+        };
+        Ok(engine::Compiled {
+            report: v2::CompilationReport {
+                status,
+                program,
+                diagnostics: text.try_into().expect("bounded diagnostics"),
+                diagnostics_truncated: truncated,
+                cpu_time_us,
+                memory_peak_bytes,
+            },
+            cleanup_complete: true,
+            failure: (status == v1::CompilationStatus::SystemError)
+                .then_some(v1::FailureCode::SandboxUnavailable),
+        })
+    }
+}
+impl<A: ArtifactPort> engine::RunPort for RuntimeAttempt<'_, A> {
+    fn run(
+        &mut self,
+        program: &v2::CompiledProgram,
+        case: &v1::TestCase,
+        budget: &v1::RunLimits,
+        run_id: &str,
+    ) -> Result<Observation, old::StageFailure> {
+        let index = self
+            .authorized
+            .manifest()
+            .cases
+            .iter()
+            .position(|c| c == case)
+            .ok_or_else(|| failure(v1::FailureCode::ArtifactUnavailable, true))?;
+        if budget != &self.authorized.manifest().run_limits {
+            return Err(failure(v1::FailureCode::SandboxUnavailable, true));
+        }
+        let spec = self
+            .authorized
+            .run_spec(program, index, run_id.into())
+            .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, true))?;
+        let stored = self
+            .program
+            .as_ref()
+            .filter(|s| &s.metadata == program)
+            .ok_or_else(|| failure(v1::FailureCode::ArtifactUnavailable, true))?;
+        if stored.files.len() != program.files.len() {
+            return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+        }
+        let input = self
+            .io
+            .artifacts
+            .read(&case.input)
+            .map_err(artifact_error)?;
+        verify(&case.input, &input).map_err(artifact_error)?;
+        let mut publication =
+            StagePublication::create(self.io.staging_path(), run_id).map_err(artifact_error)?;
+        for ((info, file), artifact) in program.files.iter().zip(&stored.files).zip(&spec.inputs) {
+            if snapshot_reference(file)? != info.artifact {
+                return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+            }
+            publication
+                .put(
+                    &artifact.artifact_id,
+                    &mut file.reader().map_err(artifact_error)?,
+                    info.artifact.bytes,
+                )
+                .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, false))?;
+        }
+        publication
+            .put("stdin", &mut input.as_slice(), case.input.bytes)
+            .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, false))?;
+        drop(input);
+        let validated = spec
+            .clone()
+            .validate(&kf_sandbox::spec::SpecPolicy::default())
+            .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, true))?;
+        let result = self.io.execute(spec, &mut publication)?;
+        let outcome = result
+            .report
+            .classify(&validated)
+            .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, false))?;
+        let mut stdout = Vec::new();
+        result
+            .stdout
+            .reader()
+            .map_err(artifact_error)?
+            .take(budget.stdout_bytes + 1)
+            .read_to_end(&mut stdout)
+            .map_err(artifact_error)?;
+        if stdout.len() as u64 > budget.stdout_bytes {
+            return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+        }
+        Ok(Observation {
+            kind: kind(outcome),
+            stdout,
+            cpu_time_us: metric(&result.report.usage.cpu_time_us),
+            memory_peak_bytes: metric(&result.report.usage.memory_peak_bytes),
+        })
+    }
+}
+impl<A: ArtifactPort> old::ExpectedPort for RuntimeAttempt<'_, A> {
+    fn expected(&mut self, case: &v1::TestCase) -> Result<Vec<u8>, old::StageFailure> {
+        if !self.authorized.manifest().cases.contains(case) {
+            return Err(failure(v1::FailureCode::ArtifactUnavailable, true));
+        }
+        let bytes = self
+            .io
+            .artifacts
+            .read(&case.expected)
+            .map_err(artifact_error)?;
+        verify(&case.expected, &bytes).map_err(artifact_error)?;
+        Ok(bytes)
+    }
+}
+impl<A: ArtifactPort> old::ControlPort for RuntimeAttempt<'_, A> {
+    fn elapsed_us(&self) -> u64 {
+        old::ControlPort::elapsed_us(&self.io)
+    }
+    fn cancelled(&self) -> bool {
+        old::ControlPort::cancelled(&self.io)
+    }
+    fn stage(&mut self, stage: v1::Stage, index: Option<u16>) -> Result<(), old::StageFailure> {
+        old::ControlPort::stage(&mut self.io, stage, index)
+    }
+}
+pub fn judge_local(
+    socket: &Path,
+    job: &Path,
+    artifacts: &Path,
+    staging: &Path,
+    admission: &NodeAdmission,
+) -> io::Result<Judged> {
+    let job: Job = v1::decode_wire(&kf_sandbox::execution::read_bounded_file(
+        job,
+        v1::MAX_WIRE_BYTES as u64,
+    )?)
+    .map_err(io::Error::other)?;
+    let authorized = AuthorizedJob::admit(
+        job.manifest.clone(),
+        job.binding.clone(),
+        job.manifest_digest.clone(),
+        job.attempt_number,
+        admission,
+    )
+    .map_err(io::Error::other)?;
+    let cancellation = AtomicBool::new(false);
+    let mut attempt = RuntimeAttempt::new(
+        authorized,
+        LocalArtifacts::open(artifacts)?,
+        socket.into(),
+        staging.into(),
+        &cancellation,
+    )?;
+    let result = engine::judge_submission(
+        &mut attempt,
+        &job.manifest,
+        job.binding,
+        job.manifest_digest,
+    )
+    .map_err(io::Error::other)?;
+    Ok(Judged {
+        result,
+        stages: attempt.into_stages(),
+    })
+}

judge/crates/worker/src/runtime_specs.rs

@@ -0,0 +1,310 @@
+//! Ordinary v2 admission and closed specifications. This module neither reads
+//! user files nor executes source; every stage still goes through Root gates.
+use crate::pipeline::{compiler_spec, limits};
+use kf_contracts::{judging as v1, judging_v2 as v2, judging_versions as versions};
+use kf_sandbox::spec::{ExecutionProfile as P, ExecutionSpec, InputArtifact, SpecPolicy};
+use sha2::{Digest as _, Sha256};
+
+/// A canonical manifest and lease checked against the operator's capability
+/// registry. Private fields prevent widening budgets after admission.
+pub struct AuthorizedJob {
+    manifest: v2::JobManifest,
+    binding: v1::LeaseBinding,
+    manifest_digest: v1::Digest,
+}
+impl AuthorizedJob {
+    pub fn admit(
+        manifest: v2::JobManifest,
+        binding: v1::LeaseBinding,
+        manifest_digest: v1::Digest,
+        attempt_number: u32,
+        policy: &versions::NodeAdmission,
+    ) -> Result<Self, &'static str> {
+        binding.validate()?;
+        if binding.evaluation_id != manifest.evaluation_id {
+            return Err("lease differs from authorized manifest");
+        }
+        policy.admit(&versions::JobManifest::V2(manifest.clone()), attempt_number)?;
+        let canonical = v2::manifest_bytes(&manifest)?;
+        if format!("{:x}", Sha256::digest(canonical)) != manifest_digest.as_str() {
+            return Err("canonical manifest digest differs from lease");
+        }
+        Ok(Self {
+            manifest,
+            binding,
+            manifest_digest,
+        })
+    }
+    pub fn manifest(&self) -> &v2::JobManifest {
+        &self.manifest
+    }
+    pub fn binding(&self) -> &v1::LeaseBinding {
+        &self.binding
+    }
+    pub fn manifest_digest(&self) -> &v1::Digest {
+        &self.manifest_digest
+    }
+    pub fn preparation_spec(&self, run_id: String) -> Result<ExecutionSpec, &'static str> {
+        let m = &self.manifest;
+        let spec = match m.language {
+            v2::Language::C17 | v2::Language::Cpp20 => {
+                let mut spec = compiler_spec(
+                    if m.language == v2::Language::C17 {
+                        v1::Language::C17
+                    } else {
+                        v1::Language::Cpp20
+                    },
+                    "source".into(),
+                    run_id,
+                    &m.compile_limits,
+                );
+                spec.runtime_identity = Some(m.toolchain.clone());
+                spec
+            }
+            v2::Language::Python313 | v2::Language::Java21 => {
+                let profile = if m.language == v2::Language::Python313 {
+                    P::PythonCheck
+                } else {
+                    P::JavaCompile
+                };
+                let budget = limits(&m.compile_limits);
+                ExecutionSpec {
+                    schema_version: kf_contracts::SchemaVersion,
+                    runtime_identity: Some(m.toolchain.clone()),
+                    run_id,
+                    profile,
+                    argv: kf_sandbox::profiles::argv(profile, &budget)?,
+                    env: Default::default(),
+                    cwd: ".".into(),
+                    inputs: vec![InputArtifact {
+                        artifact_id: "source".into(),
+                        path: m.language.contract().source_name.into(),
+                    }],
+                    stdin_artifact: None,
+                    outputs: vec![],
+                    limits: budget,
+                }
+            }
+        };
+        checked(spec)
+    }
+    /// Metadata is only accepted for this source/runtime/attempt. The adapter
+    /// must separately verify every sealed file's byte count and digest.
+    pub fn validate_program(&self, program: &v2::CompiledProgram) -> v1::Validation {
+        program.validate()?;
+        let m = &self.manifest;
+        if program.binding != self.binding
+            || program.source != m.source
+            || program.toolchain != m.toolchain
+            || program.kind != m.language.contract().program_kind
+        {
+            return Err("program differs from authorized source, runtime or lease");
+        }
+        if program.kind != v2::ProgramKind::PythonSource
+            && (program
+                .files
+                .iter()
+                .any(|f| f.artifact.bytes > m.compile_limits.file_bytes)
+                || program.bytes() > m.compile_limits.workdir_bytes)
+        {
+            return Err("program exceeds preparation output budget");
+        }
+        Ok(())
+    }
+    pub fn run_spec(
+        &self,
+        program: &v2::CompiledProgram,
+        case_index: usize,
+        run_id: String,
+    ) -> Result<ExecutionSpec, &'static str> {
+        self.validate_program(program)?;
+        if case_index >= self.manifest.cases.len() {
+            return Err("test point is outside authorized manifest");
+        }
+        let profile = match self.manifest.language {
+            v2::Language::C17 | v2::Language::Cpp20 => P::Native,
+            v2::Language::Python313 => P::PythonRun,
+            v2::Language::Java21 => P::JavaRun,
+        };
+        let budget = limits(&self.manifest.run_limits);
+        checked(ExecutionSpec {
+            schema_version: kf_contracts::SchemaVersion,
+            runtime_identity: Some(self.manifest.toolchain.clone()),
+            run_id,
+            profile,
+            argv: if profile == P::Native {
+                vec!["/work/main".into()]
+            } else {
+                kf_sandbox::profiles::argv(profile, &budget)?
+            },
+            env: Default::default(),
+            cwd: ".".into(),
+            inputs: program
+                .files
+                .iter()
+                .enumerate()
+                .map(|(i, f)| InputArtifact {
+                    artifact_id: program_artifact_id(i),
+                    path: f.path.clone(),
+                })
+                .collect(),
+            stdin_artifact: Some("stdin".into()),
+            outputs: vec![],
+            limits: budget,
+        })
+    }
+}
+pub(crate) fn program_artifact_id(index: usize) -> String {
+    format!("program-{index:03}")
+}
+fn checked(spec: ExecutionSpec) -> Result<ExecutionSpec, &'static str> {
+    spec.clone()
+        .validate(&SpecPolicy::default())
+        .map_err(|_| "invalid fixed runtime specification")?;
+    Ok(spec)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn fixture<T: serde::de::DeserializeOwned>(name: &str) -> T {
+        let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("../../../tests/fixtures/contracts-v2")
+            .join(format!("{name}.json"));
+        v1::decode_wire(&std::fs::read(path).unwrap()).unwrap()
+    }
+    fn authorized(language: &str) -> (AuthorizedJob, v2::CompiledProgram) {
+        let m: v2::JobManifest = fixture(&format!("judge-manifest.valid-{language}"));
+        let report: v2::JudgementReport = fixture(&format!("judge-result.valid-{language}"));
+        let policy: versions::NodeAdmission = fixture("judge-registry.valid");
+        let digest = format!("{:x}", Sha256::digest(v2::manifest_bytes(&m).unwrap()))
+            .try_into()
+            .unwrap();
+        let job = AuthorizedJob::admit(m, report.binding, digest, 1, &policy).unwrap();
+        (job, report.compilation.program.unwrap())
+    }
+    #[test]
+    fn all_four_languages_use_exact_pins_fixed_commands_and_private_points() {
+        for (language, prepare, run) in [
+            ("c17", P::CompileC, P::Native),
+            ("cpp20", P::CompileCpp, P::Native),
+            ("python313", P::PythonCheck, P::PythonRun),
+            ("java21", P::JavaCompile, P::JavaRun),
+        ] {
+            let (j, program) = authorized(language);
+            let p = j.preparation_spec("prepare".into()).unwrap();
+            assert_eq!(p.profile, prepare);
+            assert_eq!(p.runtime_identity.as_ref(), Some(&j.manifest.toolchain));
+            assert!(p.env.is_empty());
+            assert!(p.stdin_artifact.is_none());
+            assert_eq!(p.inputs[0].path, j.manifest.language.contract().source_name);
+            for index in 0..j.manifest.cases.len() {
+                let r = j
+                    .run_spec(&program, index, format!("point-{index}"))
+                    .unwrap();
+                assert_eq!(r.profile, run);
+                assert_eq!(r.runtime_identity, p.runtime_identity);
+                assert_eq!(r.limits, limits(&j.manifest.run_limits));
+                assert!(r.env.is_empty() && r.outputs.is_empty());
+                assert_eq!(r.stdin_artifact.as_deref(), Some("stdin"));
+                assert_eq!(r.inputs.len(), program.files.len());
+                assert!(
+                    r.inputs
+                        .iter()
+                        .all(|f| f.artifact_id.starts_with("program-"))
+                );
+                // Expected answers, source paths and manifest host paths are not inputs.
+                assert_eq!(
+                    r.inputs.iter().map(|f| &f.path).collect::<Vec<_>>(),
+                    program.files.iter().map(|f| &f.path).collect::<Vec<_>>()
+                );
+            }
+            assert!(
+                j.run_spec(&program, j.manifest.cases.len(), "point".into())
+                    .is_err()
+            );
+        }
+    }
+    #[test]
+    fn lease_manifest_digest_runtime_and_operator_capabilities_are_fenced() {
+        let (j, _) = authorized("python313");
+        let policy: versions::NodeAdmission = fixture("judge-registry.valid");
+        let admit = |m, b, d, n| AuthorizedJob::admit(m, b, d, n, &policy);
+        let mut wrong = j.manifest.clone();
+        wrong.toolchain.image_digest = "f".repeat(64).try_into().unwrap();
+        assert!(admit(wrong, j.binding.clone(), j.manifest_digest.clone(), 1).is_err());
+        let mut b = j.binding.clone();
+        b.evaluation_id = "aaaaaaaa-aaaa-4aaa-8aaa-000000001000"
+            .to_string()
+            .try_into()
+            .unwrap();
+        assert!(admit(j.manifest.clone(), b, j.manifest_digest.clone(), 1).is_err());
+        assert!(
+            admit(
+                j.manifest.clone(),
+                j.binding.clone(),
+                "f".repeat(64).try_into().unwrap(),
+                1
+            )
+            .is_err()
+        );
+        assert!(
+            admit(
+                j.manifest.clone(),
+                j.binding.clone(),
+                j.manifest_digest.clone(),
+                4
+            )
+            .is_err()
+        );
+        let mut wrong = j.manifest.clone();
+        wrong.run_limits.memory_bytes += 1;
+        assert!(admit(wrong, j.binding.clone(), j.manifest_digest.clone(), 1).is_err());
+        let mut legacy: v2::AdmissionRegistry = fixture("judge-registry.valid");
+        legacy.multi_runtime = None;
+        assert!(
+            AuthorizedJob::admit(
+                j.manifest.clone(),
+                j.binding.clone(),
+                j.manifest_digest.clone(),
+                1,
+                &versions::NodeAdmission::Registry(legacy)
+            )
+            .is_err()
+        );
+    }
+    #[test]
+    fn sealed_program_metadata_cannot_change_source_attempt_type_or_budget() {
+        for language in ["c17", "python313", "java21"] {
+            let (j, p) = authorized(language);
+            let mut bad = p.clone();
+            bad.binding.attempt_id = "aaaaaaaa-aaaa-4aaa-8aaa-000000001000"
+                .to_string()
+                .try_into()
+                .unwrap();
+            assert!(j.validate_program(&bad).is_err());
+            bad = p.clone();
+            bad.source.digest = "f".repeat(64).try_into().unwrap();
+            assert!(j.validate_program(&bad).is_err());
+            bad = p.clone();
+            bad.toolchain.image_digest = "f".repeat(64).try_into().unwrap();
+            assert!(j.validate_program(&bad).is_err());
+            bad = p.clone();
+            bad.files[0].path = "../Main.class".into();
+            assert!(j.validate_program(&bad).is_err());
+            bad = p.clone();
+            bad.files[0].artifact.bytes = 0;
+            assert!(j.validate_program(&bad).is_err());
+            if language != "python313" {
+                bad = p;
+                bad.files[0].artifact.bytes = j.manifest.compile_limits.file_bytes + 1;
+                assert!(j.validate_program(&bad).is_err());
+            }
+        }
+        let (j, mut p) = authorized("java21");
+        p.files[0].path = "A$1.class".into();
+        j.validate_program(&p).unwrap();
+        p.files.push(p.files[0].clone());
+        assert!(j.validate_program(&p).is_err());
+    }
+}

scripts/build_gnu_image.py

@@ -0,0 +1,181 @@
+"""Offline GNU catalog from an explicitly pinned, verified public package closure.
+
+No target executable, package hook, submitted source or network installer runs.
+Root publication and real execution acceptance remain separate steps.
+"""
+
+import argparse
+import hashlib
+import json
+import re
+from pathlib import Path
+
+
+def sha(path):
+    with path.open("rb") as source:
+        return hashlib.file_digest(source, "sha256").hexdigest()
+
+
+def canonical(name):
+    return len(name) <= 4096 and all(
+        re.fullmatch(r"[A-Za-z0-9_.+\-]{1,255}", p) and p not in (".", "..")
+        for p in name.split("/")
+    )
+
+
+def resolve_verified(root, records, name):
+    parts = name.split("/")
+    for _ in range(9):
+        changed = False
+        for i in range(len(parts)):
+            prefix = "/".join(parts[: i + 1])
+            record = records.get(prefix)
+            path = root / prefix
+            if record and record["kind"] == "symlink":
+                if not path.is_symlink() or str(path.readlink()) != record["target"]:
+                    raise ValueError("GNU package link differs from verified manifest")
+                target = record["target"]
+                expanded = (
+                    ([] if target.startswith("/") else parts[:i])
+                    + target.split("/")
+                    + parts[i + 1 :]
+                )
+                parts = []
+                for p in expanded:
+                    if p in ("", "."):
+                        continue
+                    if p == "..":
+                        if not parts:
+                            raise ValueError("GNU image-relative link escapes closure")
+                        parts.pop()
+                    else:
+                        parts.append(p)
+                changed = True
+                break
+            if path.is_symlink():
+                raise ValueError("unverified GNU ancestor link")
+        if not changed:
+            relative = "/".join(parts)
+            record = records.get(relative)
+            if record is None or record["kind"] != "file":
+                raise ValueError("GNU link outside verified regular-file closure")
+            source = root / relative
+            source.resolve(strict=True).relative_to(root)
+            return source, record
+    raise ValueError("GNU link cycle or depth budget")
+
+
+def build(args):
+    if sha(args.support_manifest) != args.support_manifest_sha256:
+        raise ValueError("verified GNU support manifest fingerprint differs")
+    manifest = json.loads(args.support_manifest.read_bytes())
+    packages = {p["name"]: p for p in manifest["packages"]}
+    expected = {
+        "gcc": (
+            "16.2.1+r23+gd564253eb6c8-1",
+            "ebac12ceef494c616a36fc6f0f792b365dc59b6ed46ca1a69f138aecc810b6ab",
+        ),
+        "glibc": (
+            "2.44+r50+g1848099f063e-3",
+            "071d807f4d3bf6b921f86b9a91e83a47ecc026c5821634efa46b09d890e06785",
+        ),
+        "binutils": ("2.47-2", "62efd213e87590f025cf861b646793d3cacb03e7e26187b73d33b00adaf7a977"),
+    }
+    for name, (version, digest) in expected.items():
+        if (packages[name]["version"], packages[name]["sha256"]) != (version, digest):
+            raise ValueError("GNU image requires the pinned compiler/libc/linker closure")
+    records = {f["path"]: f for f in manifest["files"]}
+    source_root = args.support_dir.resolve(strict=True)
+    if args.output.exists():
+        raise ValueError("refuse to replace an image")
+    args.output.mkdir(mode=0o700)
+    directories = {"dev", "proc", "tmp", "work", "usr", "lib64"}
+    files = []
+
+    def put(name, source, expected_file):
+        if not canonical(name) or expected_file["kind"] != "file":
+            raise ValueError("noncanonical verified GNU file")
+        if (
+            source.stat().st_size != expected_file["bytes"]
+            or sha(source) != expected_file["sha256"]
+        ):
+            raise ValueError("GNU file differs from verified public package bytes")
+        if len(files) >= 4096 or sum(f["bytes"] for f in files) + source.stat().st_size > 1024**3:
+            raise ValueError("GNU image exceeds finite catalog budget")
+        target = args.output / name
+        target.parent.mkdir(parents=True, exist_ok=True)
+        # Copy to a fresh regular inode; never publish source links/hard links.
+        with source.open("rb") as reader, target.open("xb") as writer:
+            while data := reader.read(1024 * 1024):
+                writer.write(data)
+        executable = bool(source.stat().st_mode & 0o111)
+        target.chmod(0o555 if executable else 0o444)
+        if sha(target) != expected_file["sha256"]:
+            raise ValueError("GNU copy fingerprint differs")
+        files.append(
+            {
+                "path": name,
+                "bytes": target.stat().st_size,
+                "sha256": sha(target),
+                "executable": executable,
+            }
+        )
+        directories.update(p.as_posix() for p in Path(name).parents if p != Path("."))
+
+    for name, record in sorted(records.items()):
+        if not (
+            name.startswith(("usr/include/", "usr/lib/"))
+            or name in ("usr/bin/gcc", "usr/bin/g++", "usr/bin/as", "usr/bin/ld")
+        ):
+            continue
+        if record["kind"] == "directory":
+            continue
+        source, real = resolve_verified(source_root, records, name)
+        put(name, source, real)
+    loader, loader_record = resolve_verified(source_root, records, "usr/lib/ld-linux-x86-64.so.2")
+    put("lib64/ld-linux-x86-64.so.2", loader, loader_record)
+    for name in directories:
+        (args.output / name).mkdir(parents=True, exist_ok=True)
+    catalog = dict(
+        format_version=2,
+        architecture="x86_64_v3",
+        image_kind="linux_gnu_cxx_x86_64_v1",
+        image_revision=1,
+        toolchain_id="gnu-cxx-v1",
+        runtime_version="16.2.1+r23+gd564253eb6c8",
+        directories=sorted(directories),
+        files=sorted(files, key=lambda f: f["path"]),
+    )
+    raw = (json.dumps(catalog, sort_keys=True, separators=(",", ":")) + "\n").encode()
+    if len(raw) > 1024 * 1024:
+        raise ValueError("GNU catalog transport budget")
+    (args.output / "manifest.json").write_bytes(raw)
+    (args.output / "manifest.json").chmod(0o444)
+    for path in sorted((p for p in args.output.rglob("*") if p.is_dir()), reverse=True):
+        path.chmod(0o555)
+    args.output.chmod(0o555)
+    print(
+        json.dumps(
+            dict(
+                image_digest=hashlib.sha256(raw).hexdigest(),
+                files=len(files),
+                bytes=sum(f["bytes"] for f in files),
+                support_manifest_sha256=args.support_manifest_sha256,
+                source_execution=False,
+                root_publication_and_VM_acceptance=False,
+            )
+        )
+    )
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--support-dir", required=True, type=Path)
+    parser.add_argument("--support-manifest", required=True, type=Path)
+    parser.add_argument("--support-manifest-sha256", required=True)
+    parser.add_argument("--output", required=True, type=Path)
+    build(parser.parse_args())
+
+
+if __name__ == "__main__":
+    main()