feat(worker): inject engine port for authenticated sandbox execution

judge/Cargo.lock

@@ -377,6 +377,7 @@ name = "kf-engine"
 version = "0.0.1"
 dependencies = [
  "kf-contracts",
+ "serde",
 ]
 
 [[package]]
@@ -400,6 +401,7 @@ dependencies = [
  "kf-contracts",
  "kf-engine",
  "kf-sandbox",
+ "serde",
  "serde_json",
 ]
 

judge/README.md

@@ -33,6 +33,31 @@ kf-sandbox run --config /etc/kf-judge/runtime.json --spec request.json   --artif
 
 完成任务树终止、cgroup 空读回、owned 子进程回收、最终计量和导出后，才移除资源并释放 UID。stdout/stderr 和输出文件为不可变快照。普通文件入口按 run ID 排他发布 `stdout`、`stderr`、`artifact-N`，先写自身临时目录再原子安装，不覆盖已有目录；清理未完成时不发布。
 
+## Worker 和 Engine 的最小链路
+
+服务管理器先在有限监督器 cgroup 内启动一个新的 `kf-sandbox serve-one --config /etc/kf-judge/runtime.json --socket /run/kf-judge/task.sock`。socket 的祖先须由 root 持有、不可被 Worker 写入并允许其搜索；socket 排他创建为 root:worker 的 0660，既有路径不覆盖。一次新进程只接收一次请求，等待连接和准备帧有绝对截止时间，不提供常驻消费循环。
+
+普通用户运行：
+
+```sh
+kf-worker judge-once --socket /run/kf-judge/task.sock \
+  --job native-job.json --artifacts /srv/kf-worker/publication
+```
+
+本地单用例作业示例（不是平台任务通知协议）：
+
+```json
+{"request":{"run_id":"case-001","program_artifact":"program-001","stdin_artifact":"input-001"},"expected_artifact":"expected-001"}
+```
+
+Worker 选取业务授权的自身发布目录，准备有界封印的期望输出，并向 Engine 注入 `ExecutionPort`。适配器构造 native 规格，仅绑定程序与 stdin，不将期望输出放入用户运行根。helper 根据 `SO_PEERCRED` 核验配置中的 Worker UID/GID；客户端核验 helper 的 root 凭据。有界私有帧与 `SCM_RIGHTS` 只传持有的目录能力和只读封印输出，收到的 FD 重新检查类型、长度、访问模式、封印和 CLOEXEC，不传任意宿主挂载路径。
+
+helper 不接管 root 或其他身份的输入工件；每任务保留相同的隔离和预算门控。执行中客户端断开或出现额外控制数据会取消全树，清理后退出。socket 清理核对自身 inode，不删除外来替换。业务发布目录的选择与业务授权仍属于可信 Worker；文件所有权不是业务授权证明。
+
+Engine 仅在真实、无基础设施错误的 exit 0 后逐字节比较输出。非零退出/信号为 RuntimeError，CPU/墙钟、内存、输出限制分别映射 TLE/MLE/OLE，取消与 SystemError 单独表达；不可用计量保持 null。汇总在 Engine 内按输入顺序确定，基础设施错误或取消停止后续执行；空用例和重复 run ID 在执行前拒绝。`judge-once` 返回结构化 verdict、用例与实际计量。新增 `serde` 直接依赖用于本地作业/领域结果类型，Engine 不引入 sandbox 或 OS 库。
+
+首轮提供 native 单用例和编译/运行沙箱能力；平台队列消费、完整提交编排、交互题、SPJ 和部署仍未实现。`kf-worker --check` 只检查入口，不消费或执行任务。权限 helper 和镜像必须由部署方配置，Worker 不调用 sudo、不修改账号或权限。
+
 ## 监督和恢复
 
 CPU 总时间与墙钟使用可信监督器和单调时钟；CPU 带宽仅用于节流。内存 OOM 和 pids 超限依据所属 cgroup 的真实事件及计量，不能凭 SIGKILL 推断 MLE。并发排空二进制 stdout/stderr，独立限额、截断和提前 stdin 关闭均保留实际证据。监督器最终报告保留计量不可用状态、全部限制观察、取消、退出状态和清理错误；比较与成绩属于 Engine。

judge/crates/engine/Cargo.toml

@@ -7,6 +7,7 @@ license.workspace = true
 
 [dependencies]
 kf-contracts.workspace = true
+serde.workspace = true
 
 [lints]
 workspace = true

judge/crates/engine/src/lib.rs

@@ -1,6 +1,229 @@
-//! Reserved for deterministic judging orchestration (JS-05/JS-06).
-//!
-//! Owns language adaptation, case order, output comparison and verdict aggregation.
-//! Execution is supplied through a port; this crate must not access SQL, AMQP,
-//! HTTP, privilege setup or hidden host paths directly.
-//! No fake verdict or user-program execution is provided by this scaffold.
+//! Deterministic comparison and aggregation. The injected port owns execution;
+//! this crate has no OS, sandbox, privilege or storage dependency.
+use serde::{Deserialize, Serialize};
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct CaseRequest {
+    pub run_id: String,
+    pub program_artifact: String,
+    pub stdin_artifact: Option<String>,
+}
+pub struct Case {
+    pub request: CaseRequest,
+    pub expected: Vec<u8>,
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum ExecutionKind {
+    Exited { code: u8 },
+    Signaled,
+    TimeLimit,
+    MemoryLimit,
+    OutputLimit,
+    OtherLimit,
+    Cancelled,
+    InfrastructureFailure,
+}
+pub struct Observation {
+    pub kind: ExecutionKind,
+    pub stdout: Vec<u8>,
+    pub cpu_time_us: Option<u64>,
+    pub memory_peak_bytes: Option<u64>,
+}
+pub trait ExecutionPort {
+    fn execute(&mut self, request: &CaseRequest) -> Result<Observation, PortError>;
+}
+#[derive(Debug)]
+pub struct PortError;
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Verdict {
+    Accepted,
+    WrongAnswer,
+    RuntimeError,
+    TimeLimitExceeded,
+    MemoryLimitExceeded,
+    OutputLimitExceeded,
+    Cancelled,
+    SystemError,
+}
+#[derive(Debug, Serialize, Deserialize)]
+pub struct CaseResult {
+    pub run_id: String,
+    pub verdict: Verdict,
+    pub cpu_time_us: Option<u64>,
+    pub memory_peak_bytes: Option<u64>,
+}
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Judgement {
+    pub verdict: Verdict,
+    pub cases: Vec<CaseResult>,
+}
+pub fn judge(port: &mut impl ExecutionPort, cases: &[Case]) -> Result<Judgement, PortError> {
+    if cases.is_empty() || cases.len() > 256 {
+        return Err(PortError);
+    }
+    let mut seen = std::collections::HashSet::new();
+    if cases.iter().any(|c| !seen.insert(&c.request.run_id)) {
+        return Err(PortError);
+    }
+    let mut results = Vec::with_capacity(cases.len());
+    for c in cases {
+        let result = match port.execute(&c.request) {
+            Err(_) => CaseResult {
+                run_id: c.request.run_id.clone(),
+                verdict: Verdict::SystemError,
+                cpu_time_us: None,
+                memory_peak_bytes: None,
+            },
+            Ok(o) => {
+                let verdict = match o.kind {
+                    ExecutionKind::Exited { code: 0 } => {
+                        if o.stdout == c.expected {
+                            Verdict::Accepted
+                        } else {
+                            Verdict::WrongAnswer
+                        }
+                    }
+                    ExecutionKind::Exited { .. }
+                    | ExecutionKind::Signaled
+                    | ExecutionKind::OtherLimit => Verdict::RuntimeError,
+                    ExecutionKind::TimeLimit => Verdict::TimeLimitExceeded,
+                    ExecutionKind::MemoryLimit => Verdict::MemoryLimitExceeded,
+                    ExecutionKind::OutputLimit => Verdict::OutputLimitExceeded,
+                    ExecutionKind::Cancelled => Verdict::Cancelled,
+                    ExecutionKind::InfrastructureFailure => Verdict::SystemError,
+                };
+                CaseResult {
+                    run_id: c.request.run_id.clone(),
+                    verdict,
+                    cpu_time_us: o.cpu_time_us,
+                    memory_peak_bytes: o.memory_peak_bytes,
+                }
+            }
+        };
+        let stop = matches!(result.verdict, Verdict::SystemError | Verdict::Cancelled);
+        results.push(result);
+        if stop {
+            break;
+        }
+    }
+    let verdict = results
+        .iter()
+        .find(|r| matches!(r.verdict, Verdict::SystemError | Verdict::Cancelled))
+        .or_else(|| results.iter().find(|r| r.verdict != Verdict::Accepted))
+        .map_or(Verdict::Accepted, |r| r.verdict);
+    Ok(Judgement {
+        verdict,
+        cases: results,
+    })
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    struct Port {
+        kinds: std::collections::VecDeque<Result<ExecutionKind, PortError>>,
+        calls: usize,
+    }
+    impl ExecutionPort for Port {
+        fn execute(&mut self, _: &CaseRequest) -> Result<Observation, PortError> {
+            self.calls += 1;
+            Ok(Observation {
+                kind: self.kinds.pop_front().unwrap()?,
+                stdout: vec![0, 255, 10],
+                cpu_time_us: Some(42),
+                memory_peak_bytes: None,
+            })
+        }
+    }
+    fn case(id: &str, expected: Vec<u8>) -> Case {
+        Case {
+            request: CaseRequest {
+                run_id: id.into(),
+                program_artifact: "program".into(),
+                stdin_artifact: None,
+            },
+            expected,
+        }
+    }
+    #[test]
+    fn exact_binary_comparison_and_nonzero_precedence() {
+        let mut p = Port {
+            kinds: [
+                Ok(ExecutionKind::Exited { code: 0 }),
+                Ok(ExecutionKind::Exited { code: 0 }),
+                Ok(ExecutionKind::Exited { code: 7 }),
+            ]
+            .into(),
+            calls: 0,
+        };
+        let j = judge(
+            &mut p,
+            &[
+                case("a", vec![0, 255, 10]),
+                case("b", vec![0, 255]),
+                case("c", vec![0, 255, 10]),
+            ],
+        )
+        .unwrap();
+        assert_eq!(j.verdict, Verdict::WrongAnswer);
+        assert_eq!(
+            j.cases.iter().map(|c| c.verdict).collect::<Vec<_>>(),
+            vec![
+                Verdict::Accepted,
+                Verdict::WrongAnswer,
+                Verdict::RuntimeError
+            ]
+        );
+        assert_eq!(j.cases[0].cpu_time_us, Some(42));
+        assert_eq!(j.cases[0].memory_peak_bytes, None);
+    }
+    #[test]
+    fn every_execution_failure_has_a_verdict() {
+        for (k, v) in [
+            (ExecutionKind::TimeLimit, Verdict::TimeLimitExceeded),
+            (ExecutionKind::MemoryLimit, Verdict::MemoryLimitExceeded),
+            (ExecutionKind::OutputLimit, Verdict::OutputLimitExceeded),
+            (ExecutionKind::OtherLimit, Verdict::RuntimeError),
+            (ExecutionKind::Signaled, Verdict::RuntimeError),
+            (ExecutionKind::Cancelled, Verdict::Cancelled),
+            (ExecutionKind::InfrastructureFailure, Verdict::SystemError),
+        ] {
+            let mut p = Port {
+                kinds: [Ok(k)].into(),
+                calls: 0,
+            };
+            assert_eq!(judge(&mut p, &[case("a", vec![])]).unwrap().verdict, v);
+        }
+    }
+    #[test]
+    fn infrastructure_failure_stops_and_overrides_prior_answer() {
+        let mut p = Port {
+            kinds: [
+                Ok(ExecutionKind::Exited { code: 0 }),
+                Err(PortError),
+                Ok(ExecutionKind::Exited { code: 0 }),
+            ]
+            .into(),
+            calls: 0,
+        };
+        let j = judge(
+            &mut p,
+            &[case("a", vec![]), case("b", vec![]), case("c", vec![])],
+        )
+        .unwrap();
+        assert_eq!(j.verdict, Verdict::SystemError);
+        assert_eq!(p.calls, 2);
+        assert_eq!(j.cases[1].cpu_time_us, None);
+    }
+    #[test]
+    fn empty_or_reused_run_ids_do_not_execute() {
+        let mut p = Port {
+            kinds: [].into(),
+            calls: 0,
+        };
+        assert!(judge(&mut p, &[]).is_err());
+        assert!(judge(&mut p, &[case("a", vec![]), case("a", vec![])]).is_err());
+        assert_eq!(p.calls, 0);
+    }
+}

judge/crates/sandbox/src/artifacts.rs

@@ -35,6 +35,23 @@ pub struct SnapshotFile {
 }
 
 impl SnapshotFile {
+    pub(crate) fn from_received(fd: OwnedFd, bytes: u64) -> io::Result<Self> {
+        let st = rustix::fs::fstat(&fd)?;
+        if FileType::from_raw_mode(st.st_mode) != FileType::RegularFile
+            || st.st_size < 0
+            || st.st_size as u64 != bytes
+            || bytes > MAX_SNAPSHOT_BYTES
+            || rustix::fs::fcntl_getfl(&fd)?.intersection(OFlags::ACCMODE) != OFlags::RDONLY
+            || !rustix::fs::fcntl_get_seals(&fd)?.contains(SEALS)
+            || !rustix::io::fcntl_getfd(&fd)?.contains(rustix::io::FdFlags::CLOEXEC)
+        {
+            return Err(refused("invalid received snapshot"));
+        }
+        Ok(Self {
+            file: File::from(fd),
+            bytes,
+        })
+    }
     pub fn bytes(&self) -> u64 {
         self.bytes
     }
@@ -106,10 +123,40 @@ pub struct ArtifactRoot {
     owner: u32,
 }
 
+impl AsFd for ArtifactRoot {
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.fd.as_fd()
+    }
+}
 impl ArtifactRoot {
+    pub fn open(path: &std::path::Path) -> io::Result<Self> {
+        Self::from_directory(rustix::fs::open(
+            path,
+            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        )?)
+    }
+    pub fn snapshot_artifact(&self, id: &str, limit: u64) -> io::Result<SnapshotFile> {
+        if limit == 0 || limit > MAX_SNAPSHOT_BYTES {
+            return Err(refused("snapshot ceiling"));
+        }
+        let mut budget = SnapshotBudget {
+            file_bytes: limit,
+            remaining_bytes: limit,
+        };
+        self.snapshot_id(id, &mut budget, false)
+    }
+    pub(crate) fn from_worker_directory(fd: OwnedFd, owner: u32) -> io::Result<Self> {
+        if owner < 1000 {
+            return Err(refused("ordinary publication owner"));
+        }
+        Self::with_owner(fd, owner)
+    }
     pub fn from_directory(fd: OwnedFd) -> io::Result<Self> {
+        Self::with_owner(fd, rustix::process::geteuid().as_raw())
+    }
+    fn with_owner(fd: OwnedFd, owner: u32) -> io::Result<Self> {
         let stat = rustix::fs::fstat(&fd)?;
-        let owner = rustix::process::geteuid().as_raw();
         if FileType::from_raw_mode(stat.st_mode) != FileType::Directory
             || stat.st_uid != owner
             || stat.st_mode & 0o022 != 0

judge/crates/sandbox/src/execution.rs

@@ -76,13 +76,19 @@ fn read_pinned(fd: OwnedFd, limit: u64) -> io::Result<Vec<u8>> {
     }
     Ok(bytes)
 }
-pub fn read_spec(path: &Path) -> io::Result<ExecutionSpec> {
+pub fn read_bounded_file(path: &Path, limit: u64) -> io::Result<Vec<u8>> {
+    if limit == 0 || limit > crate::spec::MAX_SPEC_BYTES {
+        return Err(invalid("bounded JSON ceiling"));
+    }
     let fd = rustix::fs::open(
         path,
         OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
         Mode::empty(),
     )?;
-    serde_json::from_slice(&read_pinned(fd, crate::spec::MAX_SPEC_BYTES)?)
+    read_pinned(fd, limit)
+}
+pub fn read_spec(path: &Path) -> io::Result<ExecutionSpec> {
+    serde_json::from_slice(&read_bounded_file(path, crate::spec::MAX_SPEC_BYTES)?)
         .map_err(|_| invalid("invalid execution JSON"))
 }
 pub fn load_config(path: &Path) -> io::Result<ValidatedRuntimeConfig> {
@@ -207,7 +213,7 @@ pub(crate) fn execute(
     config: &ValidatedRuntimeConfig,
     request: ExecutionSpec,
     artifacts: &ArtifactRoot,
-    mode: run::MonitorMode,
+    mode: run::MonitorMode<'_>,
 ) -> Result<(run::Completed, ValidatedSpec), RunFailure> {
     authorize(config).map_err(|e| refused("runtime_not_ready", &e))?;
     let spec = request

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

@@ -1,9 +1,5 @@
-//! Trusted supervisor/launcher coordination, separate from user stdio.
-//!
-//! Lifecycle events describe completed operations, not commands received from
-//! an untrusted request. The standalone entry authenticates its parent and uses
-//! bounded IPC, and refuses user execution. The separate namespace bootstrap initializes
-//! only fixed OS isolation operations and remains gated at the missing root stage.
+//! Trusted launcher coordination with per-stage kernel acknowledgement, bounded
+//! private control and one final exec gate; user stdio never carries commands.
 
 pub mod lifecycle;
 

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

@@ -43,7 +43,7 @@ pub enum ChildMessage {
 }
 
 /// A distinct fixed initialization protocol. There is no argv, host path,
-/// readiness claim. VM acceptance builds add a separately verified final gate.
+/// readiness claim. The final execution gate requires actual kernel acknowledgements.
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "command", rename_all = "snake_case", deny_unknown_fields)]
 pub enum NamespaceCommand {
@@ -105,7 +105,7 @@ pub enum NamespaceEvent {
     Cancelled {},
 }
 
-mod sealed {
+pub(crate) mod sealed {
     pub trait Sealed {}
 }
 

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

@@ -39,7 +39,8 @@ pub(crate) struct RunProof {
     pub order: Vec<&'static str>,
 }
 #[derive(Clone, Copy)]
-pub(crate) enum MonitorMode {
+pub(crate) enum MonitorMode<'a> {
+    Peer(&'a rustix::fd::OwnedFd),
     Normal,
     #[cfg(feature = "vm-acceptance")]
     CancelAfter(u64),
@@ -78,15 +79,28 @@ fn consume(event: BudgetEvent) -> (ResourceUsage, Vec<LimitObservation>) {
     }
 }
 
+fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result<bool> {
+    use rustix::event::{PollFd, PollFlags, poll};
+    let mut f = [PollFd::new(
+        fd,
+        PollFlags::IN | PollFlags::HUP | PollFlags::ERR,
+    )];
+    poll(
+        &mut f,
+        Some(&rustix::event::Timespec {
+            tv_sec: 0,
+            tv_nsec: 0,
+        }),
+    )?;
+    Ok(!f[0].revents().is_empty())
+}
 pub(crate) fn execute(
     config: &ValidatedRuntimeConfig,
     spec: &ValidatedSpec,
     binary: &Path,
     artifacts: &ArtifactRoot,
-    mode: MonitorMode,
+    mode: MonitorMode<'_>,
 ) -> Result<Completed, InitializationFailure> {
-    #[cfg(not(feature = "vm-acceptance"))]
-    let _ = mode;
     let mut owned = OwnedChildren::begin().map_err(|e| InitializationFailure {
         failure: error(ExecutionStage::Preparation, ErrorCode::OperationFailed, &e),
         cleanup_failure: None,
@@ -114,10 +128,7 @@ pub(crate) fn execute(
     let bootstrap = target.root().bootstrap_pid();
     let mut errors = Vec::new();
     let mut output_limits = Vec::new();
-    #[cfg(feature = "vm-acceptance")]
     let mut cancelled = None;
-    #[cfg(not(feature = "vm-acceptance"))]
-    let cancelled: Option<u64> = None;
     let mut main = None;
     let mut reaped_descendants = 0;
     let mut order = vec!["verified_isolation", "exec_gate"];
@@ -146,6 +157,14 @@ pub(crate) fn execute(
             });
         }
         let released = (|| {
+            if let MonitorMode::Peer(fd) = mode
+                && peer_closed(fd)?
+            {
+                return Err(io::Error::new(
+                    io::ErrorKind::Interrupted,
+                    "client disconnected before exec",
+                ));
+            }
             ns.control.send(&NamespaceCommand::ReleaseExecution {})?;
             budget.release_prepared_gate(&[bootstrap, init], &ns.init_pidfd)?;
             if ns.control.receive::<NamespaceEvent>()? != (NamespaceEvent::ExecutionSpawned {}) {
@@ -256,8 +275,10 @@ pub(crate) fn execute(
                         monitoring_failed = true;
                     }
                 }
-                #[cfg(feature = "vm-acceptance")]
                 match mode {
+                    MonitorMode::Peer(fd) if peer_closed(fd).unwrap_or(true) => {
+                        cancelled = Some(wall(&usage));
+                    }
                     #[cfg(feature = "vm-acceptance")]
                     MonitorMode::CancelAfter(t) if wall(&usage) >= t => {
                         cancelled = Some(wall(&usage))

judge/crates/sandbox/src/lib.rs

@@ -1,7 +1,5 @@
-//! Linux execution boundary: policy/contract checks and read-only observations.
-//!
-//! Execution contracts and lifecycle ordering precede OS launcher/isolation work.
-//! Until that work is verified, the CLI cannot execute any user program.
+//! Linux execution boundary: policy, kernel gates, bounded supervision and recovery.
+//! Ordinary operator/Worker entries require trusted authority and active isolation.
 
 pub mod launcher;
 pub mod outcome;
@@ -32,3 +30,6 @@ mod isolation;
 
 #[cfg(target_os = "linux")]
 pub mod execution;
+
+#[cfg(target_os = "linux")]
+pub mod rpc;

judge/crates/sandbox/src/main.rs

@@ -40,6 +40,13 @@ enum Command {
         #[arg(long)]
         output: Option<PathBuf>,
     },
+    /// One authenticated Worker request, in a dedicated finite root helper scope.
+    ServeOne {
+        #[arg(long)]
+        config: PathBuf,
+        #[arg(long)]
+        socket: PathBuf,
+    },
     /// Internal trusted launcher control entry; never runs a user program.
     #[cfg(target_os = "linux")]
     #[command(hide = true)]
@@ -120,6 +127,21 @@ fn main() -> ExitCode {
                 spec_error("unsupported_platform", None)
             }
         }
+        Command::ServeOne { config, socket } => {
+            #[cfg(target_os = "linux")]
+            match kf_sandbox::rpc::serve_one(&config, &socket) {
+                Ok(result) => {
+                    println!("{result}");
+                    ExitCode::SUCCESS
+                }
+                Err(_) => spec_error("worker_helper_refused", None),
+            }
+            #[cfg(not(target_os = "linux"))]
+            {
+                let _ = (config, socket);
+                spec_error("unsupported_platform", None)
+            }
+        }
         Command::Validate { spec } => validate_spec(spec),
         Command::CheckRuntime { config } => check_runtime(config),
         #[cfg(target_os = "linux")]

judge/crates/sandbox/src/outcome.rs

@@ -247,7 +247,7 @@ impl std::error::Error for ReportError {}
 
 impl ExecutionReport {
     /// Consistency checks on trusted observations. This does not verify OS
-    /// isolation and never changes the CLI's refusal to execute.
+    /// isolation or supply an execution permit.
     pub fn classify(&self, spec: &ValidatedSpec) -> Result<ExecutionOutcome, ReportError> {
         let limits = &spec.spec().limits;
         if self.run_id != spec.spec().run_id || self.errors.len() > 16 || self.limits.len() > 8 {

judge/crates/sandbox/src/rpc.rs

@@ -0,0 +1,245 @@
+//! One request per fresh privileged process. Kernel peer credentials authenticate
+//! both endpoints; only the configured Worker may provide its own publication FD.
+use crate::{
+    artifacts::{ArtifactRoot, OutputSnapshot, SnapshotFile},
+    execution,
+    launcher::{
+        protocol::{ControlChannel, ExpectedPeer, WireMessage},
+        run,
+    },
+    outcome::{ExecutionError, ExecutionReport},
+    spec::{ExecutionSpec, SpecPolicy},
+};
+use rustix::{
+    fd::OwnedFd,
+    fs::{Mode, OFlags},
+};
+use serde::{Deserialize, Serialize};
+use std::{
+    io,
+    os::unix::{
+        fs::{MetadataExt, PermissionsExt},
+        net::{UnixListener, UnixStream},
+    },
+    path::Path,
+    time::{Duration, Instant},
+};
+fn denied() -> io::Error {
+    io::Error::new(io::ErrorKind::PermissionDenied, "sandbox RPC refused")
+}
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct Request {
+    spec: ExecutionSpec,
+}
+#[derive(Serialize, Deserialize)]
+#[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)]
+pub(crate) enum Reply {
+    Completed {
+        report: Box<ExecutionReport>,
+    },
+    Rejected {
+        failure: ExecutionError,
+        cleanup_failure: Option<ExecutionError>,
+    },
+}
+impl crate::launcher::protocol::sealed::Sealed for Request {}
+impl crate::launcher::protocol::sealed::Sealed for Reply {}
+impl WireMessage for Request {
+    const MAX_BYTES: usize = crate::spec::MAX_SPEC_BYTES as usize + 128;
+    fn valid(&self) -> bool {
+        self.spec.run_id.len() <= 64
+    }
+}
+impl WireMessage for Reply {
+    const MAX_BYTES: usize = crate::spec::MAX_SPEC_BYTES as usize + 128;
+    fn valid(&self) -> bool {
+        match self {
+            Self::Completed { report } => {
+                report.artifacts.len() <= 16
+                    && report.errors.len() <= 16
+                    && report.limits.len() <= 8
+            }
+            Self::Rejected { .. } => true,
+        }
+    }
+}
+fn channel(stream: UnixStream, uid: u32, gid: u32) -> io::Result<ControlChannel> {
+    let c = rustix::net::sockopt::socket_peercred(&stream)?;
+    ControlChannel::new(
+        stream,
+        ExpectedPeer {
+            pid: c.pid.as_raw_pid() as u32,
+            uid,
+            gid,
+        },
+        Instant::now() + Duration::from_secs(5),
+    )
+}
+#[derive(Debug)]
+pub struct Rejected {
+    pub failure: ExecutionError,
+    pub cleanup_failure: Option<ExecutionError>,
+}
+impl std::fmt::Display for Rejected {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("sandbox request rejected")
+    }
+}
+impl std::error::Error for Rejected {}
+pub struct Executed {
+    pub report: ExecutionReport,
+    pub stdout: SnapshotFile,
+    pub stderr: SnapshotFile,
+    pub outputs: Vec<OutputSnapshot>,
+}
+/// No privileges are acquired: this endpoint requires an ordinary uid/gid.
+pub fn execute(
+    socket: &Path,
+    spec: ExecutionSpec,
+    publication: &ArtifactRoot,
+) -> io::Result<Executed> {
+    if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000 {
+        return Err(denied());
+    }
+    let validated = spec
+        .clone()
+        .validate(&SpecPolicy::default())
+        .map_err(|_| denied())?;
+    let mut c = channel(UnixStream::connect(socket)?, 0, 0)?;
+    c.send(&Request { spec })?;
+    c.send_fd(publication)?;
+    c.execution_deadline(validated.spec().limits.wall_time_us)?;
+    let report = match c.receive::<Reply>()? {
+        Reply::Completed { report } => *report,
+        Reply::Rejected {
+            failure,
+            cleanup_failure,
+        } => {
+            return Err(io::Error::new(
+                io::ErrorKind::PermissionDenied,
+                Rejected {
+                    failure,
+                    cleanup_failure,
+                },
+            ));
+        }
+    };
+    report.classify(&validated).map_err(|_| denied())?;
+    let stdout = SnapshotFile::from_received(c.receive_fd()?, report.stdout.retained_bytes)?;
+    let stderr = SnapshotFile::from_received(c.receive_fd()?, report.stderr.retained_bytes)?;
+    let mut outputs = Vec::new();
+    for a in &report.artifacts {
+        outputs.push(OutputSnapshot {
+            path: a.path.clone(),
+            file: SnapshotFile::from_received(c.receive_fd()?, a.bytes)?,
+        });
+    }
+    Ok(Executed {
+        report,
+        stdout,
+        stderr,
+        outputs,
+    })
+}
+struct SocketLease<'a> {
+    parent: OwnedFd,
+    name: &'a std::ffi::OsStr,
+    dev: u64,
+    ino: u64,
+}
+impl Drop for SocketLease<'_> {
+    fn drop(&mut self) {
+        if let Ok(s) = rustix::fs::statat(
+            &self.parent,
+            self.name,
+            rustix::fs::AtFlags::SYMLINK_NOFOLLOW,
+        ) && s.st_dev == self.dev
+            && s.st_ino == self.ino
+        {
+            let _ = rustix::fs::unlinkat(&self.parent, self.name, rustix::fs::AtFlags::empty());
+        }
+    }
+}
+/// Operator supplies protected config/socket paths; a service manager provisions
+/// the finite helper scope. No internal sudo, account changes or deployment.
+pub fn serve_one(config_path: &Path, socket: &Path) -> io::Result<serde_json::Value> {
+    let config = execution::load_config(config_path)?;
+    execution::authorize(&config)?;
+    let parent = crate::isolation::root::open_directory(socket.parent().ok_or_else(denied)?)?;
+    let name = socket.file_name().ok_or_else(denied)?;
+    // Refuse replacement, including an existing stale/foreign socket.
+    if rustix::fs::statat(&parent, name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW).is_ok() {
+        return Err(denied());
+    }
+    let listener = UnixListener::bind(socket)?;
+    let m = std::fs::symlink_metadata(socket)?;
+    let lease = SocketLease {
+        parent,
+        name,
+        dev: m.dev(),
+        ino: m.ino(),
+    };
+    let socket_fd = rustix::fs::openat(
+        &lease.parent,
+        name,
+        OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+        Mode::empty(),
+    )?;
+    let stat = rustix::fs::fstat(&socket_fd)?;
+    if stat.st_dev != lease.dev || stat.st_ino != lease.ino || stat.st_uid != 0 {
+        return Err(denied());
+    }
+    // Ancestors are non-writable to Worker. Operations cannot follow a substituted path.
+    std::os::unix::fs::chown(socket, Some(0), Some(config.config().identities.worker_gid))?;
+    std::fs::set_permissions(socket, std::fs::Permissions::from_mode(0o660))?;
+    listener.set_nonblocking(true)?;
+    let deadline = Instant::now() + Duration::from_secs(5);
+    let stream = loop {
+        match listener.accept() {
+            Ok((s, _)) => break s,
+            Err(e) if e.kind() == io::ErrorKind::WouldBlock && Instant::now() < deadline => {
+                std::thread::sleep(Duration::from_millis(5))
+            }
+            Err(e) => return Err(e),
+        }
+    };
+    drop(listener);
+    let ids = &config.config().identities;
+    let mut c = channel(stream, ids.worker_uid, ids.worker_gid)?;
+    let request = c.receive::<Request>()?;
+    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)?;
+    match execution::execute(
+        &config,
+        request.spec,
+        &artifacts,
+        run::MonitorMode::Peer(&peer),
+    ) {
+        Ok((result, _)) => {
+            let report = result.report;
+            // Cleanup failures remain in the report; caller may never claim a successful verdict.
+            if report.cancelled_at_wall_time_us.is_some() {
+                return Ok(serde_json::json!({"report":report,"proof":result.proof}));
+            }
+            c.send(&Reply::Completed {
+                report: Box::new(report.clone()),
+            })?;
+            c.send_fd(&result.stdout)?;
+            c.send_fd(&result.stderr)?;
+            for o in result.outputs {
+                c.send_fd(&o.file)?;
+            }
+            Ok(serde_json::json!({"report":report,"proof":result.proof}))
+        }
+        Err(e) => {
+            let failure = e.failure.ok_or_else(denied)?;
+            c.send(&Reply::Rejected {
+                failure,
+                cleanup_failure: e.cleanup_failure,
+            })?;
+            Err(denied())
+        }
+    }
+}

judge/crates/sandbox/tests/fixtures/worker_disconnect.c

@@ -0,0 +1,25 @@
+/* Fixed independent-VM controller only. Not installed in the ordinary CLI. */
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+#include <stdio.h>
+#include <stdint.h>
+#include <fcntl.h>
+#include <string.h>
+int main(int argc,char **argv) {
+    if(argc!=4 || getuid()!=0 || setgid(1000) || setuid(1000))return 100;
+    int d=open(argv[2],O_RDONLY|O_DIRECTORY|O_CLOEXEC);
+    int f=open(argv[3],O_RDONLY|O_CLOEXEC);
+    char body[66000];int n=(int)read(f,body,sizeof body);if(d<0||f<0||n<=0||n>=66000)return 101;close(f);
+    int s=socket(AF_UNIX,SOCK_STREAM|SOCK_CLOEXEC,0);struct sockaddr_un a={0};a.sun_family=AF_UNIX;
+    if(strlen(argv[1])>=sizeof a.sun_path){return 102;}
+    strcpy(a.sun_path,argv[1]);
+    if(connect(s,(struct sockaddr*)&a,sizeof a))return 103;
+    unsigned char header[8]={'K','F',1,0,(unsigned char)((unsigned)n>>24),(unsigned char)((unsigned)n>>16),(unsigned char)((unsigned)n>>8),(unsigned char)n};
+    if(write(s,header,8)!=8||write(s,body,(size_t)n)!=n)return 104;
+    char marker=(char)0xa5;struct iovec i={&marker,1};union {struct cmsghdr align;char bytes[CMSG_SPACE(sizeof(int))];}control={0};
+    struct msghdr m={0};m.msg_iov=&i;m.msg_iovlen=1;m.msg_control=control.bytes;m.msg_controllen=sizeof control.bytes;
+    struct cmsghdr *c=CMSG_FIRSTHDR(&m);c->cmsg_level=SOL_SOCKET;c->cmsg_type=SCM_RIGHTS;c->cmsg_len=CMSG_LEN(sizeof(int));memcpy(CMSG_DATA(c),&d,sizeof d);
+    if(sendmsg(s,&m,MSG_NOSIGNAL)!=1)return 105;
+    close(d);sleep(2);close(s);return 0;
+}

judge/crates/sandbox/tests/fixtures/worker_native.c

@@ -0,0 +1,10 @@
+#include <unistd.h>
+int main(void) {
+    unsigned char b[4096];
+    ssize_t n=read(0,b,sizeof b);
+    if(n<0)return 8;
+    if(n>0 && b[0]=='#')return 7;
+    if(n>0 && b[0]=='!'){if(write(1,"live",4)!=4)return 9;sleep(10);return 0;}
+    do {if(n>0 && write(1,b,(size_t)n)!=n)return 10;}while((n=read(0,b,sizeof b))>0);
+    return n<0 ? 11 : 0;
+}

judge/crates/sandbox/tests/worker_vm.rs

@@ -0,0 +1,240 @@
+#![cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+#[allow(dead_code)]
+mod support;
+use std::{
+    fs,
+    os::unix::{
+        fs::{PermissionsExt, chown},
+        net::UnixStream,
+        process::CommandExt,
+    },
+    path::{Path, PathBuf},
+    process::{Child, Command, Stdio},
+    time::{Duration, Instant},
+};
+struct Fixture {
+    root: support::Root,
+    dir: PathBuf,
+    config: PathBuf,
+    socket: PathBuf,
+}
+impl Fixture {
+    fn new() -> Self {
+        support::require_vm();
+        let mut root = support::Root::new();
+        root.config.aggregate.task_cpu_quota_us = 100_000;
+        let dir = PathBuf::from(format!("/kf/worker-{}", std::process::id()));
+        fs::create_dir(&dir).unwrap();
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o711)).unwrap();
+        let config = dir.join("runtime.json");
+        fs::write(&config, serde_json::to_vec(&root.config).unwrap()).unwrap();
+        fs::set_permissions(&config, fs::Permissions::from_mode(0o600)).unwrap();
+        let p = dir.join("inputs");
+        fs::create_dir(&p).unwrap();
+        chown(&p, Some(1000), Some(1000)).unwrap();
+        fs::set_permissions(&p, fs::Permissions::from_mode(0o700)).unwrap();
+        fs::copy("/kf/bin/worker-native", p.join("program")).unwrap();
+        Self::worker_file(&p.join("program"));
+        let socket = dir.join("socket");
+        Self {
+            root,
+            dir,
+            config,
+            socket,
+        }
+    }
+    fn worker_file(p: &Path) {
+        chown(p, Some(1000), Some(1000)).unwrap();
+        fs::set_permissions(p, fs::Permissions::from_mode(0o600)).unwrap();
+    }
+    fn job(&self, id: &str, input: &[u8], expected: &[u8]) {
+        for (name, data) in [("stdin", input), ("expected", expected)] {
+            let p = self.dir.join(format!("inputs/{name}"));
+            fs::write(&p, data).unwrap();
+            Self::worker_file(&p);
+        }
+        let j = serde_json::json!({"request":{"run_id":id,"program_artifact":"program","stdin_artifact":"stdin"},"expected_artifact":"expected"});
+        fs::write(self.dir.join("job.json"), serde_json::to_vec(&j).unwrap()).unwrap();
+        Self::worker_file(&self.dir.join("job.json"));
+    }
+    fn helper(&self) -> Child {
+        let mut c = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap())
+            .env_clear()
+            .args(["serve-one", "--config"])
+            .arg(&self.config)
+            .arg("--socket")
+            .arg(&self.socket)
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()
+            .unwrap();
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !self.socket.exists() {
+            assert!(
+                c.try_wait().unwrap().is_none(),
+                "helper exited before socket"
+            );
+            assert!(Instant::now() < deadline);
+            std::thread::sleep(Duration::from_millis(5));
+        }
+        // Socket creation precedes mode installation; wait for actual published permissions.
+        while fs::metadata(&self.socket).unwrap().permissions().mode() & 0o777 != 0o660 {
+            assert!(Instant::now() < deadline);
+            std::thread::sleep(Duration::from_millis(2));
+        }
+        c
+    }
+    fn empty(&self) {
+        for p in ["tasks", "identities", "captures"] {
+            let p = PathBuf::from(format!("/var/lib/kf-judge/{p}"));
+            if p.exists() {
+                assert_eq!(fs::read_dir(p).unwrap().count(), 0);
+            }
+        }
+        assert!(
+            !fs::read_dir(&self.root.path).unwrap().any(|e| e
+                .unwrap()
+                .file_type()
+                .unwrap()
+                .is_dir())
+        );
+        assert!(!self.socket.exists());
+    }
+    fn execute(&self, id: &str, input: &[u8], expected: &[u8], verdict: &str) {
+        self.job(id, input, expected);
+        let h = self.helper();
+        let w = Command::new(std::env::var_os("KF_WORKER_TEST_BIN").unwrap())
+            .env_clear()
+            .args(["judge-once", "--socket"])
+            .arg(&self.socket)
+            .arg("--job")
+            .arg(self.dir.join("job.json"))
+            .arg("--artifacts")
+            .arg(self.dir.join("inputs"))
+            .uid(1000)
+            .gid(1000)
+            .output()
+            .unwrap();
+        assert!(w.status.success(), "{}", String::from_utf8_lossy(&w.stderr));
+        let j: serde_json::Value = serde_json::from_slice(&w.stdout).unwrap();
+        assert_eq!(j["verdict"], verdict, "{j}");
+        assert_eq!(j["cases"][0]["run_id"], id);
+        assert!(j["cases"][0]["cpu_time_us"].as_u64().unwrap() > 0);
+        let o = h.wait_with_output().unwrap();
+        assert!(o.status.success(), "{}", String::from_utf8_lossy(&o.stderr));
+        let v: serde_json::Value = serde_json::from_slice(&o.stdout).unwrap();
+        assert_eq!(v["report"]["cleanup"], "complete");
+        assert_eq!(v["report"]["errors"], serde_json::json!([]));
+        self.empty();
+        println!(
+            "KF_WORKER_CHAIN_PROOF {}",
+            serde_json::json!({"worker_uid":1000,"worker_gid":1000,"kernel_peer_authentication":true,"judgement":j,"sandbox":v,"expected_artifact_not_mounted":true})
+        );
+    }
+}
+impl Drop for Fixture {
+    fn drop(&mut self) {
+        let _ = fs::remove_dir_all(&self.dir);
+    }
+}
+#[test]
+#[ignore = "independent capped VM: actual ordinary Worker -> Engine -> default-build root helper"]
+fn native_binary_comparison_and_execution_failures() {
+    let f = Fixture::new();
+    f.execute("worker-ac", &[0, 255, 10], &[0, 255, 10], "accepted");
+    f.execute("worker-wa", &[0, 255, 10], &[0, 255], "wrong_answer");
+    f.execute("worker-re", b"#", b"", "runtime_error");
+    f.execute("worker-tle", b"!", b"live", "time_limit_exceeded");
+}
+#[test]
+#[ignore = "independent capped VM: real kernel peer credential denial and foreign socket preservation"]
+fn mismatched_peer_and_existing_socket_are_refused() {
+    let f = Fixture::new();
+    let h = f.helper();
+    let c = UnixStream::connect(&f.socket).unwrap();
+    drop(c);
+    let o = h.wait_with_output().unwrap();
+    assert_eq!(o.status.code(), Some(2));
+    assert!(o.stdout.is_empty());
+    f.empty();
+    f.job("worker-owner", b"!", b"");
+    let publication = f.dir.join("inputs");
+    chown(&publication, Some(0), Some(0)).unwrap();
+    fs::set_permissions(&publication, fs::Permissions::from_mode(0o555)).unwrap();
+    let h = f.helper();
+    let mut spec: kf_sandbox::spec::ExecutionSpec =
+        serde_json::from_str(include_str!("fixtures/spec.valid.json")).unwrap();
+    spec.run_id = "worker-owner".into();
+    spec.inputs[0].artifact_id = "program".into();
+    spec.stdin_artifact = Some("stdin".into());
+    let path = f.dir.join("wire.json");
+    fs::write(
+        &path,
+        serde_json::to_vec(&serde_json::json!({"spec":spec})).unwrap(),
+    )
+    .unwrap();
+    Fixture::worker_file(&path);
+    let c = Command::new("/kf/bin/worker-disconnect")
+        .arg(&f.socket)
+        .arg(&publication)
+        .arg(&path)
+        .output()
+        .unwrap();
+    assert!(c.status.success());
+    let o = h.wait_with_output().unwrap();
+    assert_eq!(o.status.code(), Some(2));
+    assert!(o.stdout.is_empty());
+    f.empty();
+    chown(&publication, Some(1000), Some(1000)).unwrap();
+    fs::set_permissions(&publication, fs::Permissions::from_mode(0o700)).unwrap();
+    fs::write(&f.socket, b"foreign socket name").unwrap();
+    let o = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap())
+        .args(["serve-one", "--config"])
+        .arg(&f.config)
+        .arg("--socket")
+        .arg(&f.socket)
+        .output()
+        .unwrap();
+    assert_eq!(o.status.code(), Some(2));
+    assert_eq!(fs::read(&f.socket).unwrap(), b"foreign socket name");
+    fs::remove_file(&f.socket).unwrap();
+    f.empty();
+    println!(
+        "KF_WORKER_AUTH_PROOF wrong_root_peer_refused=1 root_owned_publication_refused=1 foreign_socket_preserved=1 zero_resources=1"
+    );
+}
+#[test]
+#[ignore = "independent capped VM: authenticated ordinary client disconnect while actual task is live"]
+fn client_disconnect_cancels_entire_task_and_cleans() {
+    let f = Fixture::new();
+    f.job("worker-cancel", b"!", b"");
+    let h = f.helper();
+    let mut spec: kf_sandbox::spec::ExecutionSpec =
+        serde_json::from_str(include_str!("fixtures/spec.valid.json")).unwrap();
+    spec.run_id = "worker-cancel".into();
+    spec.inputs[0].artifact_id = "program".into();
+    spec.stdin_artifact = Some("stdin".into());
+    let path = f.dir.join("wire.json");
+    fs::write(
+        &path,
+        serde_json::to_vec(&serde_json::json!({"spec":spec})).unwrap(),
+    )
+    .unwrap();
+    Fixture::worker_file(&path);
+    let c = Command::new("/kf/bin/worker-disconnect")
+        .arg(&f.socket)
+        .arg(f.dir.join("inputs"))
+        .arg(&path)
+        .output()
+        .unwrap();
+    assert!(c.status.success(), "{}", String::from_utf8_lossy(&c.stderr));
+    let o = h.wait_with_output().unwrap();
+    assert!(o.status.success(), "{}", String::from_utf8_lossy(&o.stderr));
+    let v: serde_json::Value = serde_json::from_slice(&o.stdout).unwrap();
+    assert!(v["report"]["cancelled_at_wall_time_us"].as_u64().unwrap() > 0);
+    assert_eq!(v["report"]["stdout"]["retained_bytes"], 4);
+    assert_eq!(v["report"]["cleanup"], "complete");
+    assert_eq!(v["report"]["errors"], serde_json::json!([]));
+    f.empty();
+    println!("KF_WORKER_DISCONNECT_PROOF {v}");
+}

judge/crates/worker/Cargo.toml

@@ -7,6 +7,7 @@ license.workspace = true
 
 [dependencies]
 kf-contracts.workspace = true
+serde.workspace = true
 kf-engine.workspace = true
 kf-sandbox.workspace = true
 clap.workspace = true

judge/crates/worker/src/lib.rs

@@ -0,0 +1,113 @@
+//! Ordinary Worker adapter. Privileged execution and artifact capabilities stay
+//! in sandbox; comparison and the final summary stay in Engine.
+#[cfg(target_os = "linux")]
+pub mod native {
+    use kf_engine::{Case, CaseRequest, ExecutionKind, ExecutionPort, Observation, PortError};
+    use kf_sandbox::{
+        artifacts::ArtifactRoot,
+        outcome::{ExecutionOutcome, LimitKind, Measurement},
+        spec::{ExecutionProfile, ExecutionSpec, InputArtifact, ResourceLimits},
+    };
+    use serde::Deserialize;
+    use std::{
+        io::{self, Read},
+        path::{Path, PathBuf},
+    };
+    #[derive(Deserialize)]
+    #[serde(deny_unknown_fields)]
+    pub struct Job {
+        pub request: CaseRequest,
+        pub expected_artifact: String,
+    }
+    pub struct SandboxPort {
+        socket: PathBuf,
+        artifacts: ArtifactRoot,
+    }
+    impl ExecutionPort for SandboxPort {
+        fn execute(&mut self, request: &CaseRequest) -> Result<Observation, PortError> {
+            let spec = ExecutionSpec {
+                schema_version: kf_contracts::SchemaVersion,
+                run_id: request.run_id.clone(),
+                profile: ExecutionProfile::Native,
+                argv: vec!["/work/main".into()],
+                env: std::collections::BTreeMap::new(),
+                cwd: ".".into(),
+                inputs: vec![InputArtifact {
+                    artifact_id: request.program_artifact.clone(),
+                    path: "main".into(),
+                }],
+                stdin_artifact: request.stdin_artifact.clone(),
+                outputs: vec![],
+                limits: ResourceLimits::for_profile(ExecutionProfile::Native),
+            };
+            let validated = spec
+                .clone()
+                .validate(&kf_sandbox::spec::SpecPolicy::default())
+                .map_err(|_| PortError)?;
+            let result = kf_sandbox::rpc::execute(&self.socket, spec, &self.artifacts)
+                .map_err(|_| PortError)?;
+            let outcome = result.report.classify(&validated).map_err(|_| PortError)?;
+            let kind = match outcome {
+                ExecutionOutcome::Exited { code } => ExecutionKind::Exited { code },
+                ExecutionOutcome::Signaled { .. } => ExecutionKind::Signaled,
+                ExecutionOutcome::LimitExceeded {
+                    primary: LimitKind::Memory,
+                } => ExecutionKind::MemoryLimit,
+                ExecutionOutcome::LimitExceeded {
+                    primary: LimitKind::CpuTime | LimitKind::WallTime,
+                } => ExecutionKind::TimeLimit,
+                ExecutionOutcome::LimitExceeded {
+                    primary: LimitKind::Stdout | LimitKind::Stderr,
+                } => ExecutionKind::OutputLimit,
+                ExecutionOutcome::LimitExceeded { .. } => ExecutionKind::OtherLimit,
+                ExecutionOutcome::Cancelled => ExecutionKind::Cancelled,
+                ExecutionOutcome::InfrastructureFailure | ExecutionOutcome::NotStarted => {
+                    ExecutionKind::InfrastructureFailure
+                }
+            };
+            let mut stdout = Vec::new();
+            result
+                .stdout
+                .reader()
+                .map_err(|_| PortError)?
+                .read_to_end(&mut stdout)
+                .map_err(|_| PortError)?;
+            let value = |m: Measurement<u64>| match m {
+                Measurement::Measured { value } => Some(value),
+                Measurement::Unavailable { .. } => None,
+            };
+            Ok(Observation {
+                kind,
+                stdout,
+                cpu_time_us: value(result.report.usage.cpu_time_us),
+                memory_peak_bytes: value(result.report.usage.memory_peak_bytes),
+            })
+        }
+    }
+    pub fn judge_once(
+        socket: &Path,
+        job_path: &Path,
+        artifact_path: &Path,
+    ) -> io::Result<kf_engine::Judgement> {
+        let data =
+            kf_sandbox::execution::read_bounded_file(job_path, kf_sandbox::spec::MAX_SPEC_BYTES)?;
+        let job: Job =
+            serde_json::from_slice(&data).map_err(|_| io::Error::other("invalid native job"))?;
+        let artifacts = ArtifactRoot::open(artifact_path)?;
+        let expected = artifacts.snapshot_artifact(
+            &job.expected_artifact,
+            ResourceLimits::for_profile(ExecutionProfile::Native).stdout_bytes,
+        )?;
+        let mut bytes = Vec::new();
+        expected.reader()?.read_to_end(&mut bytes)?;
+        let case = Case {
+            request: job.request,
+            expected: bytes,
+        };
+        let mut port = SandboxPort {
+            socket: socket.to_owned(),
+            artifacts,
+        };
+        kf_engine::judge(&mut port, &[case]).map_err(|_| io::Error::other("invalid judge request"))
+    }
+}

judge/crates/worker/src/main.rs

@@ -1,28 +1,64 @@
-use clap::Parser;
-use std::process::ExitCode;
-
+use clap::{Parser, Subcommand};
+use std::{path::PathBuf, process::ExitCode};
 #[derive(Parser)]
-#[command(version, about = "KF judge worker entrypoint")]
+#[command(
+    version,
+    about = "KF ordinary judge worker",
+    args_conflicts_with_subcommands = true
+)]
 struct Cli {
-    /// Check the process entrypoint without consuming or executing tasks.
     #[arg(long)]
     check: bool,
+    #[command(subcommand)]
+    command: Option<Command>,
+}
+#[derive(Subcommand)]
+enum Command {
+    JudgeOnce {
+        #[arg(long)]
+        socket: PathBuf,
+        #[arg(long)]
+        job: PathBuf,
+        #[arg(long)]
+        artifacts: PathBuf,
+    },
 }
-
 fn main() -> ExitCode {
-    if !Cli::parse().check {
-        eprintln!("judge worker task loop is not implemented; use --check");
-        return ExitCode::from(2);
+    let cli = Cli::parse();
+    if cli.check {
+        println!(
+            "{}",
+            serde_json::json!({"service":"judge-worker","version":env!("CARGO_PKG_VERSION"),"status":"entry_ready","contract_version":kf_contracts::SCHEMA_VERSION,"consuming_tasks":false})
+        );
+        return ExitCode::SUCCESS;
+    }
+    match cli.command {
+        Some(Command::JudgeOnce {
+            socket,
+            job,
+            artifacts,
+        }) => {
+            #[cfg(target_os = "linux")]
+            match kf_worker::native::judge_once(&socket, &job, &artifacts) {
+                Ok(result) => {
+                    println!("{}", serde_json::json!(result));
+                    ExitCode::SUCCESS
+                }
+                Err(_) => {
+                    eprintln!("{}", serde_json::json!({"error":"judge_request_failed"}));
+                    ExitCode::from(2)
+                }
+            }
+            #[cfg(not(target_os = "linux"))]
+            {
+                let _ = (socket, job, artifacts);
+                eprintln!("unsupported worker platform");
+                ExitCode::from(2)
+            }
+        }
+        None => {
+            eprintln!("worker consumer loop unavailable; use --check or judge-once");
+            ExitCode::from(2)
+        }
     }
-    println!(
-        "{}",
-        serde_json::json!({
-            "service": "judge-worker",
-            "version": env!("CARGO_PKG_VERSION"),
-            "status": "scaffold",
-            "contract_version": kf_contracts::SCHEMA_VERSION,
-            "consuming_tasks": false
-        })
-    );
-    ExitCode::SUCCESS
 }