feat(worker): consume durable jobs with fenced execution and recovery

2026-10-09 08:38:20 +08:00
parent 76b565cce8
commit 19a071573b
11 changed files with 1869 additions and 34 deletions
@@ -4,8 +4,9 @@
 前端使用 Vue 3、TypeScript、Element Plus 和 Vite。

 当前版本为 **0.0.1**：沙箱已有受控隔离、C/C++ 编译及 native 执行；平台已实现真实身份、
-不可变题目、工件及有效租约读取基础。受控 helper 工厂与本地 C/C++ 编译一次、多测试点评测
-已经接通；完整提交投递、协调器调度与前端业务接入仍在开发，不能将前端预览成绩当作实际结果。
+不可变题目、工件与幂等提交创建。事务 outbox、持久化 Coordinator、普通 Worker 和受控
+helper 工厂已接通 C/C++ 编译一次、多测试点评测；前端业务接入仍在开发，不能将前端预览
+成绩当作实际结果。

 ## 工程划分

@@ -68,7 +69,9 @@ make sandbox-probe  # 只读观察 Linux 环境，不创建隔离资源
 模型服务与第三方插件尚未接入。

 开发中的规划文档仅在本地维护，不包含在远端仓库中。
-后续逐步接通单机提交判题，再实现分布式恢复、并行优化和学习功能。
+单机提交服务的配置、最小权限、停止恢复与独立 VM 验收要求见
+[判题说明](judge/README.md#单机持久化协调与常驻-worker)和[部署说明](deploy/README.md#单机提交服务的启动与停止)。
+后续接入平台成绩查询与前端，再实现分布式恢复、并行优化和学习功能。

 ## 许可证

@@ -36,9 +36,30 @@ API 存活检查和前端预览不依赖这些容器，可以先独立启动。
 回收后安排新记录周期，不自动删除记录。清理失败保留身份租约并停止新分配；
 重启只根据已记录归属和 inode 回收，未知/外来内容保留并拒绝启动。

-`kf-worker judge-submission` 是受控本地适配入口，仍不消费队列。
+`kf-worker judge-submission` 是受控本地适配入口。常驻 `kf-worker serve --config`
+使用有界 AMQP 通知和经过服务身份认证的 Coordinator claim；配置与依赖见
+[判题工程说明](../judge/README.md#单机持久化协调与常驻-worker)。
 它验证不可变清单摘要，C17/C++20 仅映射到固定 GNU 命令，编译一次；
 成功产物经完整清理后封印并计算 SHA-256，每点分别申请 helper、stdin 与工作区。
 期望值逐点读取，留在比较端，不传给工厂/用户文件系统。
 新增锁定 `sha2 0.10.9` 仅供 Worker 验证输入及编译产物摘要；
 Sandbox 同版本仅为独立 VM 测试数据生成的开发依赖，Engine 不依赖加密或 OS 库。
+
+### 单机提交服务的启动与停止
+
+1. Backend Alembic 迁移；创建分别仅继承平台/协调器 NOLOGIN 角色的专用登录账号，
+   平台角色只可插入初始评测，执行状态仅由 Coordinator 写入。
+2. 普通平台服务持有私有工件目录；使用应用服务导入不可变题目和创建测试账号。
+3. 验证固定 GNU 镜像、Root 配置及工厂服务/任务 cgroup 的真实上限，再启动工厂。
+4. 独立启动 API、Coordinator、outbox publisher 和普通 Worker；单机 HTTP/AMQP
+   仅数值 loopback，PostgreSQL loopback/Unix socket。任何入口都不自动创建表。
+5. Coordinator readiness 验证数据库和权限；仍需独立 VM 验证真实执行门控。
+   Root 工厂不可持有平台/MQ/数据库口令，用户运行镜像不含服务文件。
+6. 先向 Worker 发送 SIGTERM，使当前任务取消/清理，再停止 publisher/API/Coordinator，
+   最后停止工厂和本地依赖。强制死亡后的旧 attempt 必须撤销并取得实际停止证明；
+   `cleanup_failed` 不等于普通终态，先核对保留资源再由管理员恢复工厂/Worker。
+
+这些入口和顺序不表示生产部署已完成。隔离/超限与用户程序仅在无 NIC、无宿主共享
+挂载、无业务凭据且整体上限已读回的独立测试 VM 内验收。开发 Compose 仍只提供可信
+数据库和 broker，不在宿主或业务服务器运行用户代码。workflow 如需启用由指定
+GitHub 私有仓库承载，并设置有限并发；Gitea 只保存逐阶段工程提交。
@@ -5,9 +5,10 @@
 | `kf-contracts` | 平台与判题服务的通知契约和共享样例 |
 | `kf-engine` | 执行编排端口、输出比较和判题汇总；不进行权限操作 |
 | `kf-sandbox` | Linux namespace、运行根、身份、seccomp、cgroup、监督和清理 |
+| `kf-coordinator` | PostgreSQL 事务、领取/租约/阶段/结果、取消和恢复；不操作账号或题库 |
 | `kf-worker` | 普通用户工件准备与执行适配器注入；不操作评测数据库 |

-依赖保持 `worker → engine / sandbox / contracts`、`engine → contracts`、`sandbox → contracts`。原始系统调用和权限操作集中在 sandbox 内。
+依赖保持 `worker → engine / sandbox / contracts`、`coordinator → contracts`、`engine → contracts`、`sandbox → contracts`。原始系统调用和权限操作集中在 sandbox 内。

 ## 普通执行入口

@@ -56,7 +57,8 @@ helper 不接管 root 或其他身份的输入工件；每任务保留相同的

 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、不修改账号或权限。
+本地已提供 native 单用例、编译一次多测试点与平台消费/持久化协调入口；
+用户成绩展示、交互题、SPJ 和生产部署仍未实现。`kf-worker --check` 只检查入口，不消费或执行任务。权限 helper 和镜像必须由部署方配置，Worker 不调用 sudo、不修改账号或权限。

 ## 编译一次、多测试点

@@ -101,3 +103,52 @@ make check
 宿主普通检查不执行权限、恶意或超限测试。`vm-acceptance` 仅增加独立 VM 的固定验收入口与故障注入，不是生产执行授权。ignored 的 Linux 测试必须在无宿主共享挂载、无业务凭据、无 NIC 且整体资源上限已实际读回的独立 VM 内执行；`KF_SANDBOX_INDEPENDENT_VM=1` 只是测试防误用标记，不能代替环境证据。未执行或跳过不能计为 Linux 安全验收。

 覆盖真实 GNU C/C++ 编译后运行、危险调用、宿主和跨任务文件、权限提升、网络、CPU/内存/pids/输出/文件限制，以及取消、初始化回滚、真实 owner SIGKILL、服务恢复和重复清理。验收记录保存在本地忽略的计划目录；不包含远端开发文档或业务凭据。
+
+## 单机持久化协调与常驻 Worker
+
+先通过 Backend Alembic 迁移，分别创建仅继承 `kf_oj_platform`、`kf_oj_coordinator`
+的普通数据库登录角色。Coordinator 不迁移、不读账号/写题库；Worker 不接收数据库
+连接、不写评测状态、不调用 sudo。令牌是内部服务身份，与浏览器会话及作业短期租约分离。
+
+```sh
+kf-coordinator --config /etc/kf-judge/coordinator.json
+kf-worker serve --config /etc/kf-judge/worker.json
+kf-worker check-stopped --socket /var/lib/kf-judge/factory/control.sock
+```
+
+两个配置都须为 0600、有界、单链接普通文件。Coordinator 配置包含 `listen`、
+`database_url`、`workers`（`id`/64位十六进制 `token`）、不同的 `platform_token`、
+`admission`、`lease_seconds`（5–300）及 `recovery_seconds`（1–60）。
+Worker 配置包含 `worker_id`、`service_token`、`coordinator_url`、`platform_url`、
+`amqp_url`、固定 `socket`、自身 0700 `staging` 与同一 `admission`。
+本轮只允许单机数值 loopback HTTP/AMQP 与 loopback/绝对 Unix socket PostgreSQL；
+未提供跨机器 TLS 模式。Worker 固定并发/prefetch 为1，缺少有效领取权不能执行。
+
+`/health/live` 表示进程存活；配置后 `/health/ready` 连接数据库并核对最小权限。
+内部 HTTP 先认证再解析有界请求，整体10秒，数据库连接槽位4，handler槽位16。
+claim 使用数据库时间、行锁和有限全局容量锁，提交后才返回新 attempt/递增 fence/
+短期令牌与清单摘要。阶段严格递增，续租和结果核对当前绑定与令牌；用例、attempt
+及终态同事务提交后才确认。完整同一报告可重复确认，修改或旧 attempt 被拒绝。
+
+取消与过期撤销执行权，先进入 `cancelling`；不能把租约到期直接当作任务已停止。
+Worker 续租失败在有界期限内取消 helper，通过 Root 工厂的停止查询核对 owned journal、
+任务目录、身份和 cgroup 全部排空，才发送服务身份认证的停止报告。
+未知资源保留并使工厂停止分配；Coordinator 保留 `cleanup_failed`，占用容量，
+由管理员处理明确的外来资源后重启工厂/Worker。停止查询不能选取宿主路径、命令或任务。
+只有确认实际清理的基础设施失败可按 admission 有限重试，普通成绩不重试。
+
+恢复扫描每批最多16项；已投递而30秒后仍待领取的事件重新通知，broker/Worker 断线
+采用最多30秒退避。AMQP 操作各5秒，HTTP 连接1秒/请求2秒；结果最多重试3次相同报告。
+Worker SIGTERM 取消当前执行后退出，SIGKILL 由 Root EOF 清理、租约过期和重启协调收敛。
+消息手工 ack 只表示通知已处理，持久化状态与补偿负责恢复，不把 ack 当成最终成绩。
+
+输出预算必须同时满足每流不超过单文件上限、stdout/stderr 合计不超过捕获存储
+上限；Python/Rust 清单在准入时拒绝不一致限额，实际沙箱仍独立核对。连续测试点
+遇到完成 helper 尚待回收的暂忙槽位时，仅对不消耗 run ID 的 Busy 分配响应做
+最多2秒/64次有界等待，仍受作业截止时间与取消约束；拒绝、重放和执行响应不明
+不会重新执行同一请求。
+
+新增外部依赖职责：Coordinator 的 `tokio-postgres 0.7.15` 只负责专用角色数据库事务；
+Worker 的 `lapin 2.5.5` / `futures-lite 2.6.1` 负责 AMQP 通知，`reqwest 0.12.24`
+负责有界 loopback 请求并禁用代理与重定向；`uuid 1.27.0` 生成身份，`sha2 0.10.9`
+验证清单/工件/报告摘要。Engine 不引入数据库、HTTP、队列或权限依赖。
@@ -29,12 +29,19 @@ fn denied() -> io::Error {
 #[serde(deny_unknown_fields)]
 struct Acquire {
    run_id: String,
+    #[serde(default, skip_serializing_if = "is_false")]
+    check_stopped: bool,
+}
+fn is_false(v: &bool) -> bool {
+    !v
 }
 #[derive(Serialize, Deserialize)]
 #[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)]
 enum Allocation {
    Ready { run_id: String, helper_pid: u32 },
    Refused {},
+    Busy {},
+    Stopped { cleanup_complete: bool },
 }
 impl crate::launcher::protocol::sealed::Sealed for Acquire {}
 impl crate::launcher::protocol::sealed::Sealed for Allocation {}
@@ -49,10 +56,19 @@ impl WireMessage for Allocation {
            Self::Ready { run_id, helper_pid } => {
                *helper_pid > 1 && crate::spec::identifier(run_id, "run_id").is_ok()
            }
-            Self::Refused {} => true,
+            Self::Refused {} | Self::Busy {} | Self::Stopped { .. } => true,
        }
    }
 }
+#[derive(Debug)]
+struct FactoryBusy;
+impl std::fmt::Display for FactoryBusy {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("helper factory busy")
+    }
+}
+impl std::error::Error for FactoryBusy {}
+
 pub struct FactoryClient {
    socket: PathBuf,
 }
@@ -60,6 +76,23 @@ impl FactoryClient {
    pub fn new(socket: PathBuf) -> Self {
        Self { socket }
    }
+    /// Root-authenticated global drain proof. No task/path/command is selected.
+    /// False/unknown never releases execution authority at Coordinator.
+    pub fn stopped(&self) -> io::Result<bool> {
+        if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000
+        {
+            return Err(denied());
+        }
+        let mut channel = rpc::channel(UnixStream::connect(&self.socket)?, 0, 0)?;
+        channel.send(&Acquire {
+            run_id: "stop-check".into(),
+            check_stopped: true,
+        })?;
+        match channel.receive::<Allocation>()? {
+            Allocation::Stopped { cleanup_complete } => Ok(cleanup_complete),
+            _ => Err(denied()),
+        }
+    }
    /// The caller holds only ordinary artifact and Unix socket capabilities.
    pub fn execute(
        &self,
@@ -74,6 +107,45 @@ impl FactoryClient {
        publication: &ArtifactRoot,
        deadline: Option<Instant>,
        cancellation: Option<&std::sync::atomic::AtomicBool>,
+    ) -> io::Result<(rpc::Executed, u32)> {
+        // A completed helper may still be awaiting factory reap when the next
+        // point arrives. Busy does not consume the run ID. Retry only this typed
+        // allocation response, never a rejected request or an ambiguous execution.
+        let allocation_deadline = Instant::now() + Duration::from_secs(2);
+        for _ in 0..64 {
+            if deadline.is_some_and(|d| Instant::now() >= d) {
+                return Err(io::Error::new(
+                    io::ErrorKind::TimedOut,
+                    "helper allocation deadline",
+                ));
+            }
+            if cancellation.is_some_and(|c| c.load(std::sync::atomic::Ordering::Acquire)) {
+                return Err(io::Error::new(
+                    io::ErrorKind::Interrupted,
+                    "helper allocation cancelled",
+                ));
+            }
+            let result = self.execute_once(spec.clone(), publication, deadline, cancellation);
+            if !result
+                .as_ref()
+                .err()
+                .is_some_and(|e| e.get_ref().is_some_and(|v| v.is::<FactoryBusy>()))
+            {
+                return result;
+            }
+            if Instant::now() >= allocation_deadline {
+                return result;
+            }
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        Err(io::Error::new(io::ErrorKind::WouldBlock, FactoryBusy))
+    }
+    fn execute_once(
+        &self,
+        spec: ExecutionSpec,
+        publication: &ArtifactRoot,
+        deadline: Option<Instant>,
+        cancellation: Option<&std::sync::atomic::AtomicBool>,
    ) -> io::Result<(rpc::Executed, u32)> {
        if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000
        {
@@ -105,9 +177,17 @@ impl FactoryClient {
                let mut channel = rpc::channel(stream, 0, 0)?;
                channel.send(&Acquire {
                    run_id: spec.run_id.clone(),
+                    check_stopped: false,
                })?;
                let pid = match channel.receive::<Allocation>()? {
                    Allocation::Ready { run_id, helper_pid } if run_id == spec.run_id => helper_pid,
+                    Allocation::Busy {} => {
+                        return Err(io::Error::new(io::ErrorKind::WouldBlock, FactoryBusy));
+                    }
+                    Allocation::Refused {} => {
+                        eprintln!("sandbox factory allocation refused");
+                        return Err(denied());
+                    }
                    _ => return Err(denied()),
                };
                Ok((
@@ -197,6 +277,31 @@ fn idle_recovery(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<
        std::thread::sleep(Duration::from_millis(10));
    }
 }
+fn drained(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<bool> {
+    let p = config.config();
+    for root in [
+        PathBuf::from(&p.tasks_root),
+        PathBuf::from(&p.state_root).join("identities"),
+        PathBuf::from(&p.state_root).join("captures"),
+    ] {
+        match std::fs::read_dir(root) {
+            Ok(mut entries) => {
+                if entries.next().transpose()?.is_some() {
+                    return Ok(false);
+                }
+            }
+            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
+            Err(e) => return Err(e),
+        }
+    }
+    for entry in std::fs::read_dir(&p.delegated_cgroup_root)? {
+        let entry = entry?;
+        if entry.file_type()?.is_dir() || entry.file_type()?.is_symlink() {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
 /// Fixed executable/config come solely from the trusted service invocation.
 /// Runtime is finite even when the service manager is misconfigured; cgroup
 /// reservations are read back by authorize, and tasks retain their own gates.
@@ -339,13 +444,33 @@ pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> {
            let Ok(request) = channel.receive::<Acquire>() else {
                continue;
            };
-            if active.len() >= policy.max_concurrent_runs as usize
-                || ids.len() >= MAX_IDS
-                || ids.contains(&request.run_id)
-            {
+            if request.check_stopped {
+                if !active.is_empty() {
+                    let _ = channel.send(&Allocation::Stopped {
+                        cleanup_complete: false,
+                    });
+                    continue;
+                }
+                let recovered = idle_recovery(&config);
+                // Unknown unjournaled task directories also prevent a drain proof.
+                let empty = recovered.is_ok() && drained(&config)?;
+                channel.send(&Allocation::Stopped {
+                    cleanup_complete: empty,
+                })?;
+                if !empty {
+                    return Err(denied());
+                }
+                continue;
+            }
+            // Replay/ledger exhaustion is terminal even while all slots are busy.
+            if ids.len() >= MAX_IDS || ids.contains(&request.run_id) {
                let _ = channel.send(&Allocation::Refused {});
                continue;
            }
+            if active.len() >= policy.max_concurrent_runs as usize {
+                let _ = channel.send(&Allocation::Busy {});
+                continue;
+            }
            // Persist before allocating; IDs remain consumed across restarts.
            writeln!(ledger, "{}", request.run_id)?;
            ledger.sync_all()?;
@@ -404,7 +404,7 @@ fn ordinary_control_probe() {
            assert_eq!(ready["event"], "ready");
            let mut second = UnixStream::connect(&socket).unwrap();
            raw_send(&mut second, &serde_json::json!({"run_id":"capacity-extra"}));
-            assert_eq!(raw_receive(&mut second)["event"], "refused");
+            assert_eq!(raw_receive(&mut second)["event"], "busy");
            drop(first);
            drop(second);
            println!(
@@ -13,6 +13,11 @@ kf-sandbox.workspace = true
 clap.workspace = true
 serde_json.workspace = true
 sha2 = "=0.10.9"
+lapin = { version = "=2.5.5", default-features = false }
+reqwest = { version = "=0.12.24", default-features = false, features = ["blocking", "json"] }
+futures-lite = "=2.6.1"
+tokio.workspace = true
+uuid = { version = "=1.27.0", features = ["v4", "serde"] }

 [lints]
 workspace = true
@@ -0,0 +1,586 @@
+//! Ordinary transport adapter. AMQP is a bounded wake-up; claim is sole authority.
+use crate::pipeline::{ArtifactPort, Attempt};
+use futures_lite::{StreamExt, future::block_on};
+use kf_contracts::{JudgeTaskNotification, SchemaVersion, judging::*};
+use kf_engine::submission::{StageFailure, judge_submission};
+use kf_sandbox::{artifacts::StagePublication, factory::FactoryClient};
+use lapin::{Connection, ConnectionProperties, options::*, types::FieldTable};
+use reqwest::blocking::Client;
+use serde::Deserialize;
+use sha2::{Digest as _, Sha256};
+use std::{
+    fs,
+    io::{self, Read},
+    path::{Path, PathBuf},
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+const QUEUE: &str = "kf.judge.evaluations.v1";
+#[derive(Debug)]
+struct TransportError {
+    operation: &'static str,
+    category: &'static str,
+}
+impl std::fmt::Display for TransportError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}: {}", self.operation, self.category)
+    }
+}
+impl std::error::Error for TransportError {}
+fn transport_error(operation: &'static str, category: &'static str) -> io::Error {
+    io::Error::other(TransportError {
+        operation,
+        category,
+    })
+}
+fn error() -> io::Error {
+    io::Error::other("worker operation refused")
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Config {
+    pub worker_id: Identifier,
+    pub service_token: LeaseToken,
+    pub coordinator_url: String,
+    pub platform_url: String,
+    pub amqp_url: String,
+    pub socket: PathBuf,
+    pub staging: PathBuf,
+    pub admission: AdmissionPolicy,
+}
+impl Config {
+    pub fn load(path: &Path) -> io::Result<Self> {
+        use std::os::unix::fs::{MetadataExt, PermissionsExt};
+        let m = fs::symlink_metadata(path)?;
+        if !m.is_file()
+            || m.nlink() != 1
+            || m.permissions().mode() & 0o777 != 0o600
+            || m.len() > 32768
+        {
+            return Err(error());
+        }
+        let c: Self = serde_json::from_slice(&fs::read(path)?).map_err(|_| error())?;
+        for (raw, scheme) in [
+            (&c.coordinator_url, "http"),
+            (&c.platform_url, "http"),
+            (&c.amqp_url, "amqp"),
+        ] {
+            let u = reqwest::Url::parse(raw).map_err(|_| error())?;
+            if u.scheme() != scheme
+                || !u.host_str().is_some_and(|s| {
+                    s.trim_matches(['[', ']'])
+                        .parse::<std::net::IpAddr>()
+                        .is_ok_and(|i| i.is_loopback())
+                })
+                || u.query().is_some()
+                || u.fragment().is_some()
+                || (scheme == "http"
+                    && (!u.username().is_empty() || u.password().is_some() || u.path() != "/"))
+            {
+                return Err(error());
+            }
+        }
+        if !c.socket.is_absolute()
+            || !c.staging.is_absolute()
+            || c.admission.max_concurrent_jobs != 1
+            || c.admission.prefetch != 1
+        {
+            return Err(error());
+        }
+        c.admission.validate().map_err(|_| error())?;
+        amqp_uri(&c.amqp_url)?;
+        Ok(c)
+    }
+}
+fn amqp_uri(raw: &str) -> io::Result<lapin::uri::AMQPUri> {
+    let url = reqwest::Url::parse(raw).map_err(|_| error())?;
+    let host = url.host_str().ok_or_else(error)?.trim_matches(['[', ']']);
+    if url.scheme() != "amqp"
+        || !host
+            .parse::<std::net::IpAddr>()
+            .is_ok_and(|ip| ip.is_loopback())
+    {
+        return Err(error());
+    }
+    let mut uri: lapin::uri::AMQPUri = raw.parse().map_err(|_| error())?;
+    // Locked URI parser uses domain() and otherwise substitutes localhost for IPs.
+    uri.authority.host = host.into();
+    // Python publisher treats an empty/root path as default vhost; normalize identically.
+    if url.path().is_empty() || url.path() == "/" {
+        uri.vhost = "/".into();
+    }
+    Ok(uri)
+}
+#[derive(Clone)]
+struct Remote {
+    client: Client,
+    base: String,
+    service: String,
+}
+impl Remote {
+    fn new(c: &Config) -> io::Result<Self> {
+        Ok(Self {
+            client: Client::builder()
+                .no_proxy()
+                .redirect(reqwest::redirect::Policy::none())
+                .connect_timeout(Duration::from_secs(1))
+                .timeout(Duration::from_secs(2))
+                .build()
+                .map_err(|_| error())?,
+            base: c.coordinator_url.trim_end_matches('/').into(),
+            service: c.service_token.expose_secret().into(),
+        })
+    }
+    fn post<T: serde::Serialize>(
+        &self,
+        route: &str,
+        body: &T,
+        grant: Option<&LeaseGrant>,
+    ) -> io::Result<reqwest::blocking::Response> {
+        let mut r = self
+            .client
+            .post(format!("{}{route}", self.base))
+            .bearer_auth(&self.service)
+            .json(body);
+        if let Some(g) = grant {
+            r = r.header("x-lease-token", g.lease_token.expose_secret());
+        }
+        r.send().map_err(|_| error())
+    }
+    fn bytes(mut r: reqwest::blocking::Response, max: usize) -> io::Result<Vec<u8>> {
+        if !r.status().is_success() || r.content_length().is_some_and(|n| n > max as u64) {
+            return Err(error());
+        }
+        let mut b = Vec::new();
+        (&mut r).take(max as u64 + 1).read_to_end(&mut b)?;
+        if b.len() > max {
+            return Err(error());
+        }
+        Ok(b)
+    }
+    fn pending(&self) -> io::Result<Vec<Pending>> {
+        let r = self
+            .client
+            .get(format!("{}/internal/pending-stops", self.base))
+            .bearer_auth(&self.service)
+            .send()
+            .map_err(|_| transport_error("coordinator_pending", "send"))?;
+        let b = Self::bytes(r, 16384)?;
+        let v: Vec<Pending> = serde_json::from_slice(&b)
+            .map_err(|_| transport_error("coordinator_pending", "decode"))?;
+        if v.len() > 16
+            || v.iter()
+                .any(|p| p.case_count == 0 || p.case_count > MAX_CASES)
+        {
+            return Err(error());
+        }
+        Ok(v)
+    }
+    fn reconcile(&self, c: &Config) -> io::Result<bool> {
+        let pending = self.pending()?;
+        if pending.is_empty() {
+            return Ok(true);
+        }
+        StagePublication::recover(&c.staging)?;
+        let clean = FactoryClient::new(c.socket.clone())
+            .stopped()
+            .unwrap_or(false);
+        for p in pending {
+            if p.binding.worker_id != c.worker_id {
+                return Err(error());
+            }
+            let r = StoppedReport {
+                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());
+            }
+        }
+        Ok(clean)
+    }
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Pending {
+    binding: LeaseBinding,
+    case_count: usize,
+}
+struct RemoteArtifacts {
+    client: Client,
+    base: String,
+    grant: LeaseGrant,
+}
+impl RemoteArtifacts {
+    fn headers(&self, r: reqwest::blocking::RequestBuilder) -> reqwest::blocking::RequestBuilder {
+        r.bearer_auth(self.grant.lease_token.expose_secret())
+            .header(
+                "x-worker-id",
+                self.grant.binding.worker_id.as_uuid().to_string(),
+            )
+            .header("x-fencing-version", self.grant.binding.fencing_version)
+    }
+    fn url(&self, suffix: &str) -> String {
+        format!(
+            "{}/api/v1/internal/evaluations/{}/attempts/{}{suffix}",
+            self.base,
+            self.grant.binding.evaluation_id.as_uuid(),
+            self.grant.binding.attempt_id.as_uuid()
+        )
+    }
+    fn manifest(&self) -> io::Result<JobManifest> {
+        let response = self
+            .headers(self.client.get(self.url("/manifest")))
+            .send()
+            .map_err(|_| error())?;
+        let m: JobManifest =
+            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)
+    }
+}
+impl ArtifactPort for RemoteArtifacts {
+    fn read(&mut self, reference: &ArtifactReference) -> io::Result<Vec<u8>> {
+        let r = self
+            .headers(
+                self.client
+                    .get(self.url(&format!("/artifacts/{}", reference.digest.as_str()))),
+            )
+            .send()
+            .map_err(|_| error())?;
+        let bytes = Remote::bytes(r, reference.bytes as usize)?;
+        if bytes.len() as u64 != reference.bytes
+            || format!("{:x}", Sha256::digest(&bytes)) != reference.digest.as_str()
+        {
+            return Err(error());
+        }
+        Ok(bytes)
+    }
+}
+fn lease_deadline(ms: u64) -> io::Result<Instant> {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(|_| error())?
+        .as_millis() as u64;
+    let remaining = ms
+        .checked_sub(now)
+        .filter(|v| *v >= 500 && *v <= 300000)
+        .ok_or_else(error)?;
+    Ok(Instant::now() + Duration::from_millis(remaining))
+}
+fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Result<()> {
+    if g.binding.worker_id != c.worker_id {
+        return Err(error());
+    }
+    let artifacts = RemoteArtifacts {
+        client: r.client.clone(),
+        base: c.platform_url.trim_end_matches('/').into(),
+        grant: g.clone(),
+    };
+    let m = artifacts.manifest()?;
+    c.admission
+        .admit(&m, g.attempt_number)
+        .map_err(|_| error())?;
+    let cancel = AtomicBool::new(false);
+    let done = AtomicBool::new(false);
+    let initial = lease_deadline(g.expires_at_unix_ms)?;
+    std::thread::scope(|scope| {
+        let renew = scope.spawn(|| {
+            let mut deadline = initial;
+            let mut next = Instant::now() + Duration::from_secs(1);
+            while !done.load(Ordering::Acquire) {
+                if stop.load(Ordering::Acquire) || Instant::now() >= deadline {
+                    cancel.store(true, Ordering::Release);
+                    break;
+                }
+                if Instant::now() >= next {
+                    let result = (|| {
+                        let response = r.post("/internal/renew", &g.binding, Some(&g))?;
+                        let b = Remote::bytes(response, 1024)?;
+                        let v: serde_json::Value =
+                            serde_json::from_slice(&b).map_err(|_| error())?;
+                        lease_deadline(v["expires_at_unix_ms"].as_u64().ok_or_else(error)?)
+                    })();
+                    match result {
+                        Ok(d) => deadline = d,
+                        Err(_) => {
+                            cancel.store(true, Ordering::Release);
+                            break;
+                        }
+                    }
+                    next = Instant::now() + Duration::from_secs(1);
+                }
+                std::thread::sleep(Duration::from_millis(20));
+            }
+        });
+        let result = (|| {
+            let mut sequence = 0u64;
+            let mut adapter = Attempt::new(
+                artifacts,
+                c.socket.clone(),
+                c.staging.clone(),
+                m.job_wall_time_us,
+                &cancel,
+            )?
+            .with_control(|stage, index| {
+                sequence += 1;
+                let report = StageReport {
+                    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(()),
+                        Ok(response) if response.status().as_u16() == 409 => break,
+                        _ => {}
+                    }
+                    std::thread::sleep(Duration::from_millis(100));
+                }
+                cancel.store(true, Ordering::Release);
+                Err(StageFailure {
+                    code: FailureCode::LeaseLost,
+                    cleanup_complete: true,
+                })
+            });
+            let mut result = judge_submission(
+                &mut adapter,
+                &m,
+                g.binding.clone(),
+                g.manifest_digest.clone(),
+            )
+            .map_err(|_| error())?;
+            transport_cancellation(&mut result.report);
+            result
+                .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)) {
+                    Ok(response) if response.status().as_u16() == 204 => return Ok(()),
+                    Ok(response) if response.status().as_u16() == 409 => break,
+                    _ => {}
+                }
+                std::thread::sleep(Duration::from_millis(200));
+            }
+            Err(error())
+        })();
+        done.store(true, Ordering::Release);
+        renew.join().map_err(|_| error())?;
+        result
+    })
+}
+// 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) {
+    if report.verdict == Verdict::Cancelled {
+        report.verdict = Verdict::SystemError;
+        report.failure_code = Some(FailureCode::LeaseLost);
+        if report.compilation.status == CompilationStatus::Cancelled {
+            report.compilation.status = CompilationStatus::SystemError;
+        }
+    }
+}
+async fn bounded<T>(
+    operation: &'static str,
+    future: impl std::future::Future<Output = lapin::Result<T>>,
+) -> io::Result<T> {
+    tokio::time::timeout(Duration::from_secs(5), future)
+        .await
+        .map_err(|_| transport_error(operation, "deadline"))?
+        .map_err(|failure| {
+            transport_error(
+                operation,
+                match failure {
+                    lapin::Error::IOError(_) => "io",
+                    lapin::Error::ProtocolError(_) => "protocol",
+                    lapin::Error::MissingHeartbeatError => "heartbeat",
+                    _ => "refused",
+                },
+            )
+        })
+}
+async fn consume(c: &Config, r: &Remote, stop: &AtomicBool) -> io::Result<()> {
+    if !r.reconcile(c)? {
+        return Err(error());
+    }
+    let connection = bounded(
+        "amqp_connect",
+        Connection::connect_uri(amqp_uri(&c.amqp_url)?, ConnectionProperties::default()),
+    )
+    .await?;
+    let channel = bounded("amqp_channel", connection.create_channel()).await?;
+    bounded("amqp_qos", channel.basic_qos(1, BasicQosOptions::default())).await?;
+    let mut arguments = FieldTable::default();
+    arguments.insert("x-max-length".into(), 64i32.into());
+    arguments.insert("x-max-length-bytes".into(), 32768i32.into());
+    arguments.insert(
+        "x-overflow".into(),
+        lapin::types::AMQPValue::LongString("reject-publish".into()),
+    );
+    arguments.insert("x-message-ttl".into(), 600000i32.into());
+    bounded(
+        "amqp_declare",
+        channel.queue_declare(
+            QUEUE,
+            QueueDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            arguments,
+        ),
+    )
+    .await?;
+    let mut consumer = bounded(
+        "amqp_consume",
+        channel.basic_consume(
+            QUEUE,
+            "kf-worker",
+            BasicConsumeOptions::default(),
+            FieldTable::default(),
+        ),
+    )
+    .await?;
+    while !stop.load(Ordering::Acquire) {
+        // A bounded read wait also reconciles revoked/expired attempts with no message.
+        let delivery = tokio::time::timeout(Duration::from_secs(1), consumer.next()).await;
+        let delivery = match delivery {
+            Err(_) => {
+                if !r.reconcile(c)? {
+                    return Err(error());
+                }
+                continue;
+            }
+            Ok(None) => break,
+            Ok(Some(v)) => v.map_err(|_| transport_error("amqp_read", "refused"))?,
+        };
+        let notice: JudgeTaskNotification = match decode_wire(&delivery.data) {
+            Ok(v) => v,
+            Err(_) => {
+                bounded(
+                    "amqp_reject",
+                    delivery.reject(BasicRejectOptions { requeue: false }),
+                )
+                .await?;
+                continue;
+            }
+        };
+        let grant = r.post("/internal/claim", &notice, None)?;
+        match grant.status().as_u16() {
+            200 => {
+                let g: LeaseGrant =
+                    decode_wire(&Remote::bytes(grant, MAX_WIRE_BYTES)?).map_err(|_| error())?;
+                let result = execute(c, r, g, stop);
+                if result.is_err() && !r.reconcile(c)? {
+                    return Err(error());
+                }
+                // If report was unconfirmed, DB expiry/recovery owns the next notification.
+                bounded("amqp_ack", delivery.ack(BasicAckOptions::default())).await?;
+            }
+            204 | 429 => bounded("amqp_ack", delivery.ack(BasicAckOptions::default())).await?,
+            _ => return Err(error()),
+        }
+    }
+    bounded("amqp_close", connection.close(200, "worker stopped")).await?;
+    Ok(())
+}
+pub fn run(c: Config, stop: Arc<AtomicBool>) -> io::Result<()> {
+    StagePublication::recover(&c.staging)?;
+    let remote = Remote::new(&c)?;
+    let mut failures = 0u32;
+    while !stop.load(Ordering::Acquire) {
+        let result = block_on(consume(&c, &remote, &stop));
+        if stop.load(Ordering::Acquire) {
+            break;
+        }
+        if result.is_ok() {
+            failures = 0;
+        } else {
+            failures = failures.saturating_add(1).min(5);
+            let code = result
+                .err()
+                .and_then(|e| e.into_inner())
+                .and_then(|e| e.downcast::<TransportError>().ok());
+            if let Some(code) = code {
+                eprintln!("worker transport unavailable: {code}");
+            } else {
+                eprintln!("worker transport unavailable: refused");
+            }
+        }
+        let deadline = Instant::now() + Duration::from_secs(2u64.pow(failures).min(30));
+        while Instant::now() < deadline && !stop.load(Ordering::Acquire) {
+            std::thread::sleep(Duration::from_millis(50));
+        }
+    }
+    // A still-live lease may take a bounded expiry interval to become pending.
+    let _ = remote.reconcile(&c);
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[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", "/"),
+            (
+                "amqp://fixture:temporary@127.0.0.1:5672/%2f",
+                "127.0.0.1",
+                "/",
+            ),
+            ("amqp://fixture:temporary@[::1]:5672/named", "::1", "named"),
+        ] {
+            let uri = amqp_uri(raw).unwrap();
+            assert_eq!(uri.authority.host, host);
+            assert_eq!(uri.vhost, vhost);
+        }
+        assert!(amqp_uri("amqp://fixture:temporary@example.invalid/").is_err());
+        assert!(amqp_uri("amqp://fixture:temporary@192.0.2.1/").is_err());
+    }
+    #[test]
+    fn stopped_transport_cannot_claim_user_cancellation() {
+        let mut report: JudgementReport = decode_wire(include_bytes!(
+            "../../../../tests/fixtures/contracts/judge-result.valid.json"
+        ))
+        .unwrap();
+        transport_cancellation(&mut report);
+        assert_eq!(report.verdict, Verdict::Accepted);
+        assert_eq!(report.failure_code, None);
+        report.cases.last_mut().unwrap().verdict = Verdict::Cancelled;
+        report.verdict = Verdict::Cancelled;
+        report.validate().unwrap();
+        transport_cancellation(&mut report);
+        assert_eq!(report.failure_code, Some(FailureCode::LeaseLost));
+        assert_eq!(report.verdict, Verdict::SystemError);
+        report.validate().unwrap();
+        report.cases.clear();
+        report.run_cpu_time_us = None;
+        report.run_memory_peak_bytes = None;
+        report.compilation.program = None;
+        report.compilation.status = CompilationStatus::Cancelled;
+        report.verdict = Verdict::Cancelled;
+        report.failure_code = None;
+        report.validate().unwrap();
+        transport_cancellation(&mut report);
+        assert_eq!(report.compilation.status, CompilationStatus::SystemError);
+        report.validate().unwrap();
+    }
+}
@@ -114,3 +114,6 @@ pub mod native {

 #[cfg(target_os = "linux")]
 pub mod pipeline;
+
+#[cfg(target_os = "linux")]
+pub mod consumer;
@@ -14,6 +14,14 @@ struct Cli {
 }
 #[derive(Subcommand)]
 enum Command {
+    CheckStopped {
+        #[arg(long)]
+        socket: PathBuf,
+    },
+    Serve {
+        #[arg(long)]
+        config: PathBuf,
+    },
    JudgeSubmission {
        #[arg(long)]
        socket: PathBuf,
@@ -33,7 +41,8 @@ enum Command {
        artifacts: PathBuf,
    },
 }
-fn main() -> ExitCode {
+#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
+async fn main() -> ExitCode {
    let cli = Cli::parse();
    if cli.check {
        println!(
@@ -43,6 +52,43 @@ fn main() -> ExitCode {
        return ExitCode::SUCCESS;
    }
    match cli.command {
+        Some(Command::CheckStopped { socket }) => {
+            match kf_sandbox::factory::FactoryClient::new(socket).stopped() {
+                Ok(clean) => {
+                    println!("{}", serde_json::json!({"cleanup_complete":clean}));
+                    ExitCode::SUCCESS
+                }
+                Err(_) => {
+                    eprintln!("stop proof unavailable");
+                    ExitCode::from(2)
+                }
+            }
+        }
+        Some(Command::Serve { config }) => {
+            let c = match kf_worker::consumer::Config::load(&config) {
+                Ok(c) => c,
+                Err(_) => {
+                    eprintln!("worker configuration refused");
+                    return ExitCode::from(2);
+                }
+            };
+            let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+            let watch = stop.clone();
+            tokio::spawn(async move {
+                let mut term =
+                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+                        .expect("signal registration");
+                tokio::select! {_ = term.recv()=>{},_ = tokio::signal::ctrl_c()=>{}}
+                watch.store(true, std::sync::atomic::Ordering::Release);
+            });
+            match tokio::task::spawn_blocking(move || kf_worker::consumer::run(c, stop)).await {
+                Ok(Ok(())) => ExitCode::SUCCESS,
+                _ => {
+                    eprintln!("worker unavailable");
+                    ExitCode::from(2)
+                }
+            }
+        }
        Some(Command::JudgeSubmission {
            socket,
            job,
@@ -162,6 +162,7 @@ pub fn compiler_spec(
        limits: limits(budget),
    }
 }
+type StageControl<'a> = Box<dyn FnMut(Stage, Option<u16>) -> Result<(), StageFailure> + 'a>;
 pub struct Attempt<'a, A: ArtifactPort> {
    pub artifacts: A,
    pub stages: Vec<StageEvidence>,
@@ -171,6 +172,7 @@ pub struct Attempt<'a, A: ArtifactPort> {
    started: Instant,
    deadline: Instant,
    program: Option<(ArtifactReference, SnapshotFile)>,
+    control: Option<StageControl<'a>>,
 }
 impl<'a, A: ArtifactPort> Attempt<'a, A> {
    pub fn new(
@@ -197,8 +199,16 @@ impl<'a, A: ArtifactPort> Attempt<'a, A> {
            started,
            deadline,
            program: None,
+            control: None,
        })
    }
+    pub fn with_control(
+        mut self,
+        control: impl FnMut(Stage, Option<u16>) -> Result<(), StageFailure> + 'a,
+    ) -> Self {
+        self.control = Some(Box::new(control));
+        self
+    }
    fn execute(
        &mut self,
        spec: ExecutionSpec,
@@ -216,8 +226,17 @@ impl<'a, A: ArtifactPort> Attempt<'a, A> {
            Some(self.cancellation),
        );
        let cleanup = publication.cleanup().is_ok();
-        let (result, pid) =
-            result.map_err(|_| stage_error(FailureCode::SandboxUnavailable, false))?;
+        let (result, pid) = result.map_err(|error| {
+            if let Some(rejected) = error.get_ref().and_then(|e| e.downcast_ref::<kf_sandbox::rpc::Rejected>()) {
+                eprintln!("{}", serde_json::json!({"operation":"sandbox_rejected", "failure":rejected.failure, "cleanup_failure":rejected.cleanup_failure}));
+            } else {
+                eprintln!("{}", serde_json::json!({"operation":"sandbox_rpc", "kind":format!("{:?}",error.kind())}));
+            }
+            stage_error(FailureCode::SandboxUnavailable, false)
+        })?;
+        if !cleanup {
+            eprintln!("sandbox stage publication cleanup refused");
+        }
        if !cleanup {
            return Err(stage_error(FailureCode::CleanupFailed, false));
        }
@@ -438,8 +457,12 @@ impl<A: ArtifactPort> ControlPort for Attempt<'_, A> {
    fn cancelled(&self) -> bool {
        self.cancellation.load(Ordering::Acquire)
    }
-    fn stage(&mut self, _: Stage, _: Option<u16>) -> Result<(), StageFailure> {
-        Ok(())
+    fn stage(&mut self, stage: Stage, index: Option<u16>) -> Result<(), StageFailure> {
+        if let Some(control) = self.control.as_mut() {
+            control(stage, index)
+        } else {
+            Ok(())
+        }
    }
 }
 pub fn judge_local(