diff --git a/README.md b/README.md
index 6b36692..408164c 100644
--- a/README.md
+++ b/README.md
@@ -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#单机提交服务的启动与停止)。
+后续接入平台成绩查询与前端，再实现分布式恢复、并行优化和学习功能。
 
 ## 许可证
 
diff --git a/deploy/README.md b/deploy/README.md
index 417025a..8b3f5c6 100644
--- a/deploy/README.md
+++ b/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 只保存逐阶段工程提交。
diff --git a/judge/Cargo.lock b/judge/Cargo.lock
index 45f72a9..82790f3 100644
--- a/judge/Cargo.lock
+++ b/judge/Cargo.lock
@@ -11,6 +11,54 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "amq-protocol"
+version = "7.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "587d313f3a8b4a40f866cc84b6059fe83133bf172165ac3b583129dd211d8e1c"
+dependencies = [
+ "amq-protocol-tcp",
+ "amq-protocol-types",
+ "amq-protocol-uri",
+ "cookie-factory",
+ "nom",
+ "serde",
+]
+
+[[package]]
+name = "amq-protocol-tcp"
+version = "7.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc707ab9aa964a85d9fc25908a3fdc486d2e619406883b3105b48bf304a8d606"
+dependencies = [
+ "amq-protocol-uri",
+ "tcp-stream",
+ "tracing",
+]
+
+[[package]]
+name = "amq-protocol-types"
+version = "7.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf99351d92a161c61ec6ecb213bc7057f5b837dd4e64ba6cb6491358efd770c4"
+dependencies = [
+ "cookie-factory",
+ "nom",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "amq-protocol-uri"
+version = "7.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f89f8273826a676282208e5af38461a07fe939def57396af6ad5997fcf56577d"
+dependencies = [
+ "amq-protocol-types",
+ "percent-encoding",
+ "url",
+]
+
 [[package]]
 name = "anstream"
 version = "1.0.0"
@@ -47,7 +95,7 @@ version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
 dependencies = [
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -58,9 +106,136 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
 dependencies = [
  "anstyle",
  "once_cell_polyfill",
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "async-channel"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2"
+dependencies = [
+ "concurrent-queue",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-executor"
+version = "1.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a"
+dependencies = [
+ "async-task",
+ "concurrent-queue",
+ "fastrand 2.5.0",
+ "futures-lite 2.6.1",
+ "pin-project-lite",
+ "slab",
+]
+
+[[package]]
+name = "async-global-executor"
+version = "3.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13f937e26114b93193065fd44f507aa2e9169ad0cdabbb996920b1fe1ddea7ba"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-io 2.6.0",
+ "async-lock 3.4.2",
+ "blocking",
+ "futures-lite 2.6.1",
+]
+
+[[package]]
+name = "async-global-executor-trait"
+version = "2.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9af57045d58eeb1f7060e7025a1631cbc6399e0a1d10ad6735b3d0ea7f8346ce"
+dependencies = [
+ "async-global-executor",
+ "async-trait",
+ "executor-trait",
+]
+
+[[package]]
+name = "async-io"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0fc5b45d93ef0529756f812ca52e44c221b35341892d3dcc34132ac02f3dd2af"
+dependencies = [
+ "async-lock 2.8.0",
+ "autocfg",
+ "cfg-if",
+ "concurrent-queue",
+ "futures-lite 1.13.0",
+ "log",
+ "parking",
+ "polling 2.8.0",
+ "rustix 0.37.28",
+ "slab",
+ "socket2 0.4.10",
+ "waker-fn",
+]
+
+[[package]]
+name = "async-io"
+version = "2.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc"
+dependencies = [
+ "autocfg",
+ "cfg-if",
+ "concurrent-queue",
+ "futures-io",
+ "futures-lite 2.6.1",
+ "parking",
+ "polling 3.11.0",
+ "rustix 1.1.5",
+ "slab",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "async-lock"
+version = "2.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "287272293e9d8c41773cec55e365490fe034813a2f172f502d6ddcf75b2f582b"
+dependencies = [
+ "event-listener 2.5.3",
+]
+
+[[package]]
+name = "async-lock"
+version = "3.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
+dependencies = [
+ "event-listener 5.4.2",
+ "event-listener-strategy",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-reactor-trait"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a6012d170ad00de56c9ee354aef2e358359deb1ec504254e0e5a3774771de0e"
+dependencies = [
+ "async-io 1.13.0",
+ "async-trait",
+ "futures-core",
+ "reactor-trait",
+]
+
+[[package]]
+name = "async-task"
+version = "4.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de"
+
 [[package]]
 name = "async-trait"
 version = "0.1.92"
@@ -78,6 +253,12 @@ version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
+[[package]]
+name = "autocfg"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
+
 [[package]]
 name = "axum"
 version = "0.8.9"
@@ -136,6 +317,18 @@ version = "0.22.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
 
+[[package]]
+name = "base64"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
+
+[[package]]
+name = "bitflags"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+
 [[package]]
 name = "bitflags"
 version = "2.13.2"
@@ -160,6 +353,19 @@ dependencies = [
  "hybrid-array",
 ]
 
+[[package]]
+name = "blocking"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a70e4329df6cb94385eed412ec92375c3cdd8a6e502493d1229b6414e4036dfa"
+dependencies = [
+ "async-channel",
+ "async-task",
+ "futures-io",
+ "futures-lite 2.6.1",
+ "piper",
+]
+
 [[package]]
 name = "bumpalo"
 version = "3.20.3"
@@ -247,12 +453,27 @@ version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
+[[package]]
+name = "concurrent-queue"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "const-oid"
 version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
 
+[[package]]
+name = "cookie-factory"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9885fa71e26b8ab7855e2ec7cae6e9b380edff76cd052e07c683a0319d51b3a2"
+
 [[package]]
 name = "cpufeatures"
 version = "0.2.17"
@@ -271,6 +492,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6"
+
 [[package]]
 name = "crypto-common"
 version = "0.1.7"
@@ -321,6 +548,23 @@ dependencies = [
  "ctutils",
 ]
 
+[[package]]
+name = "displaydoc"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
+[[package]]
+name = "doc-comment"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "780955b8b195a21ab8e4ac6b60dd1dbdcec1dc6c51c0617964b08c81785e12c9"
+
 [[package]]
 name = "errno"
 version = "0.3.14"
@@ -328,7 +572,42 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "event-listener"
+version = "2.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0"
+
+[[package]]
+name = "event-listener"
+version = "5.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2"
+dependencies = [
+ "parking",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "event-listener-strategy"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
+dependencies = [
+ "event-listener 5.4.2",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "executor-trait"
+version = "2.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13c39dff9342e4e0e16ce96be751eb21a94e94a87bb2f6e63ad1961c2ce109bf"
+dependencies = [
+ "async-trait",
 ]
 
 [[package]]
@@ -337,6 +616,32 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7"
 
+[[package]]
+name = "fastrand"
+version = "1.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be"
+dependencies = [
+ "instant",
+]
+
+[[package]]
+name = "fastrand"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223"
+
+[[package]]
+name = "flume"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "spin",
+]
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.2"
@@ -362,6 +667,40 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
 
+[[package]]
+name = "futures-io"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
+
+[[package]]
+name = "futures-lite"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49a9d51ce47660b1e808d3c990b4709f2f415d928835a17dfd16991515c46bce"
+dependencies = [
+ "fastrand 1.9.0",
+ "futures-core",
+ "futures-io",
+ "memchr",
+ "parking",
+ "pin-project-lite",
+ "waker-fn",
+]
+
+[[package]]
+name = "futures-lite"
+version = "2.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad"
+dependencies = [
+ "fastrand 2.5.0",
+ "futures-core",
+ "futures-io",
+ "parking",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.34"
@@ -381,8 +720,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc"
 dependencies = [
  "futures-core",
+ "futures-io",
  "futures-sink",
  "futures-task",
+ "memchr",
  "pin-project-lite",
  "slab",
 ]
@@ -427,6 +768,18 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
 
+[[package]]
+name = "hermit-abi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024"
+
+[[package]]
+name = "hermit-abi"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284"
+
 [[package]]
 name = "hmac"
 version = "0.13.0"
@@ -508,6 +861,7 @@ dependencies = [
  "pin-project-lite",
  "smallvec",
  "tokio",
+ "want",
 ]
 
 [[package]]
@@ -516,15 +870,154 @@ version = "0.1.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff"
 dependencies = [
+ "base64 0.23.1",
  "bytes",
+ "futures-channel",
+ "futures-util",
  "http",
  "http-body",
+ "httparse",
  "hyper",
+ "ipnet",
+ "libc",
+ "percent-encoding",
  "pin-project-lite",
+ "socket2 0.6.5",
  "tokio",
  "tower-service",
+ "tracing",
 ]
 
+[[package]]
+name = "icu_collections"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513"
+dependencies = [
+ "displaydoc",
+ "potential_utf",
+ "utf8_iter",
+ "yoke",
+ "zerofrom",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_locale_core"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb"
+dependencies = [
+ "displaydoc",
+ "litemap",
+ "tinystr",
+ "writeable",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_normalizer"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f"
+dependencies = [
+ "icu_collections",
+ "icu_normalizer_data",
+ "icu_properties",
+ "icu_provider",
+ "smallvec",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_normalizer_data"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0"
+
+[[package]]
+name = "icu_properties"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148"
+dependencies = [
+ "displaydoc",
+ "icu_collections",
+ "icu_locale_core",
+ "icu_properties_data",
+ "icu_provider",
+ "zerotrie",
+ "zerovec",
+]
+
+[[package]]
+name = "icu_properties_data"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa"
+
+[[package]]
+name = "icu_provider"
+version = "2.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73"
+dependencies = [
+ "displaydoc",
+ "icu_locale_core",
+ "writeable",
+ "yoke",
+ "zerofrom",
+ "zerotrie",
+ "zerovec",
+]
+
+[[package]]
+name = "idna"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de"
+dependencies = [
+ "idna_adapter",
+ "smallvec",
+ "utf8_iter",
+]
+
+[[package]]
+name = "idna_adapter"
+version = "1.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714"
+dependencies = [
+ "icu_normalizer",
+ "icu_properties",
+]
+
+[[package]]
+name = "instant"
+version = "0.1.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
+name = "io-lifetimes"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eae7b9aee968036d54dce06cebaefd919e4472e753296daccd6d344e3e2df0c2"
+dependencies = [
+ "hermit-abi 0.3.9",
+ "libc",
+ "windows-sys 0.48.0",
+]
+
+[[package]]
+name = "ipnet"
+version = "2.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0"
+
 [[package]]
 name = "is_terminal_polyfill"
 version = "1.70.2"
@@ -590,7 +1083,7 @@ dependencies = [
  "clap",
  "kf-contracts",
  "libc",
- "rustix",
+ "rustix 1.1.5",
  "seccompiler",
  "serde",
  "serde_json",
@@ -602,12 +1095,39 @@ name = "kf-worker"
 version = "0.0.1"
 dependencies = [
  "clap",
+ "futures-lite 2.6.1",
  "kf-contracts",
  "kf-engine",
  "kf-sandbox",
+ "lapin",
+ "reqwest",
  "serde",
  "serde_json",
  "sha2 0.10.9",
+ "tokio",
+ "uuid",
+]
+
+[[package]]
+name = "lapin"
+version = "2.5.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "02d2aa4725b9607915fa1a73e940710a3be6af508ce700e56897cbe8847fbb07"
+dependencies = [
+ "amq-protocol",
+ "async-global-executor-trait",
+ "async-reactor-trait",
+ "async-trait",
+ "executor-trait",
+ "flume",
+ "futures-core",
+ "futures-io",
+ "parking_lot",
+ "pinky-swear",
+ "reactor-trait",
+ "serde",
+ "tracing",
+ "waker-fn",
 ]
 
 [[package]]
@@ -628,18 +1148,30 @@ version = "0.1.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d"
 dependencies = [
- "bitflags",
+ "bitflags 2.13.2",
  "libc",
  "plain",
  "redox_syscall 0.9.4",
 ]
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519"
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
 
+[[package]]
+name = "litemap"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae"
+
 [[package]]
 name = "lock_api"
 version = "0.4.14"
@@ -692,6 +1224,12 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
 
+[[package]]
+name = "minimal-lexical"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
+
 [[package]]
 name = "mio"
 version = "1.2.4"
@@ -700,7 +1238,17 @@ checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
 dependencies = [
  "libc",
  "wasi",
- "windows-sys",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "nom"
+version = "7.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
+dependencies = [
+ "memchr",
+ "minimal-lexical",
 ]
 
 [[package]]
@@ -709,7 +1257,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -724,6 +1272,12 @@ version = "1.70.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
 
+[[package]]
+name = "parking"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.5"
@@ -778,19 +1332,72 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
 
+[[package]]
+name = "pinky-swear"
+version = "6.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1ea6e230dd3a64d61bcb8b79e597d3ab6b4c94ec7a234ce687dd718b4f2e657"
+dependencies = [
+ "doc-comment",
+ "flume",
+ "parking_lot",
+ "tracing",
+]
+
+[[package]]
+name = "piper"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1"
+dependencies = [
+ "atomic-waker",
+ "fastrand 2.5.0",
+ "futures-io",
+]
+
 [[package]]
 name = "plain"
 version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
 
+[[package]]
+name = "polling"
+version = "2.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b2d323e8ca7996b3e23126511a523f7e62924d93ecd5ae73b333815b0eb3dce"
+dependencies = [
+ "autocfg",
+ "bitflags 1.3.2",
+ "cfg-if",
+ "concurrent-queue",
+ "libc",
+ "log",
+ "pin-project-lite",
+ "windows-sys 0.48.0",
+]
+
+[[package]]
+name = "polling"
+version = "3.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218"
+dependencies = [
+ "cfg-if",
+ "concurrent-queue",
+ "hermit-abi 0.5.3",
+ "pin-project-lite",
+ "rustix 1.1.5",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "postgres-protocol"
 version = "0.6.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514"
 dependencies = [
- "base64",
+ "base64 0.22.1",
  "byteorder",
  "bytes",
  "fallible-iterator",
@@ -816,6 +1423,15 @@ dependencies = [
  "uuid",
 ]
 
+[[package]]
+name = "potential_utf"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661"
+dependencies = [
+ "zerovec",
+]
+
 [[package]]
 name = "ppv-lite86"
 version = "0.2.21"
@@ -901,13 +1517,24 @@ version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
 
+[[package]]
+name = "reactor-trait"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "438a4293e4d097556730f4711998189416232f009c137389e0f961d2bc0ddc58"
+dependencies = [
+ "async-trait",
+ "futures-core",
+ "futures-io",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
 dependencies = [
- "bitflags",
+ "bitflags 2.13.2",
 ]
 
 [[package]]
@@ -916,7 +1543,7 @@ version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "737970939a87c6fa31e7acad13307bccbb017a073b695b6089a2c484f929e20e"
 dependencies = [
- "bitflags",
+ "bitflags 2.13.2",
 ]
 
 [[package]]
@@ -936,17 +1563,65 @@ version = "0.8.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
 
+[[package]]
+name = "reqwest"
+version = "0.12.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d0946410b9f7b082a427e4ef5c8ff541a88b357bc6c637c40db3a68ac70a36f"
+dependencies = [
+ "base64 0.22.1",
+ "bytes",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "http",
+ "http-body",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "js-sys",
+ "log",
+ "percent-encoding",
+ "pin-project-lite",
+ "serde",
+ "serde_json",
+ "serde_urlencoded",
+ "sync_wrapper",
+ "tokio",
+ "tower",
+ "tower-http",
+ "tower-service",
+ "url",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
+[[package]]
+name = "rustix"
+version = "0.37.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "519165d378b97752ca44bbe15047d5d3409e875f39327546b42ac81d7e18c1b6"
+dependencies = [
+ "bitflags 1.3.2",
+ "errno",
+ "io-lifetimes",
+ "libc",
+ "linux-raw-sys 0.3.8",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "rustix"
 version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d"
 dependencies = [
- "bitflags",
+ "bitflags 2.13.2",
  "errno",
  "libc",
- "linux-raw-sys",
- "windows-sys",
+ "linux-raw-sys 0.12.1",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -1101,6 +1776,16 @@ version = "1.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
 
+[[package]]
+name = "socket2"
+version = "0.4.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f7916fc008ca5542385b89a3d3ce689953c143e9304a9bf8beec1de48994c0d"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "socket2"
 version = "0.6.5"
@@ -1108,9 +1793,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
 dependencies = [
  "libc",
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "spin"
+version = "0.9.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e"
+dependencies = [
+ "lock_api",
+]
+
+[[package]]
+name = "stable_deref_trait"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
+
 [[package]]
 name = "stringprep"
 version = "0.1.5"
@@ -1155,6 +1855,29 @@ name = "sync_wrapper"
 version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263"
+dependencies = [
+ "futures-core",
+]
+
+[[package]]
+name = "synstructure"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
+[[package]]
+name = "tcp-stream"
+version = "0.28.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "495b0abdce3dc1f8fd27240651c9e68890c14e9d9c61527b1ce44d8a5a7bd3d5"
+dependencies = [
+ "cfg-if",
+]
 
 [[package]]
 name = "thread_local"
@@ -1165,6 +1888,16 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "tinystr"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643"
+dependencies = [
+ "displaydoc",
+ "zerovec",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.13.3"
@@ -1182,9 +1915,9 @@ dependencies = [
  "mio",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "socket2 0.6.5",
  "tokio-macros",
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -1218,7 +1951,7 @@ dependencies = [
  "postgres-protocol",
  "postgres-types",
  "rand 0.9.5",
- "socket2",
+ "socket2 0.6.5",
  "tokio",
  "tokio-util",
  "whoami",
@@ -1254,6 +1987,24 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "tower-http"
+version = "0.6.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
+dependencies = [
+ "bitflags 2.13.2",
+ "bytes",
+ "futures-util",
+ "http",
+ "http-body",
+ "pin-project-lite",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "url",
+]
+
 [[package]]
 name = "tower-layer"
 version = "0.3.3"
@@ -1341,6 +2092,12 @@ dependencies = [
  "tracing-serde",
 ]
 
+[[package]]
+name = "try-lock"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
+
 [[package]]
 name = "typenum"
 version = "1.20.1"
@@ -1374,6 +2131,24 @@ version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d"
 
+[[package]]
+name = "url"
+version = "2.5.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed"
+dependencies = [
+ "form_urlencoded",
+ "idna",
+ "percent-encoding",
+ "serde",
+]
+
+[[package]]
+name = "utf8_iter"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
+
 [[package]]
 name = "utf8parse"
 version = "0.2.2"
@@ -1404,6 +2179,21 @@ version = "0.9.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
 
+[[package]]
+name = "waker-fn"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7"
+
+[[package]]
+name = "want"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec4cdd0dd910afe868b7ef477227d8d538b46b3075031afee8a9f2acb0a2ed0b"
+dependencies = [
+ "try-lock",
+]
+
 [[package]]
 name = "wasi"
 version = "0.11.1+wasi-snapshot-preview1"
@@ -1438,6 +2228,17 @@ dependencies = [
  "wasm-bindgen-shared",
 ]
 
+[[package]]
+name = "wasm-bindgen-futures"
+version = "0.4.79"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3cbab34de2d982e9b48e18d216d04c4a6f641066ff19ffb699980f591ee3610e"
+dependencies = [
+ "js-sys",
+ "tokio",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "wasm-bindgen-macro"
 version = "0.2.129"
@@ -1491,12 +2292,43 @@ dependencies = [
  "web-sys",
 ]
 
+[[package]]
+name = "winapi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
+dependencies = [
+ "winapi-i686-pc-windows-gnu",
+ "winapi-x86_64-pc-windows-gnu",
+]
+
+[[package]]
+name = "winapi-i686-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
+
+[[package]]
+name = "winapi-x86_64-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
 
+[[package]]
+name = "windows-sys"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
+dependencies = [
+ "windows-targets",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.61.2"
@@ -1506,12 +2338,98 @@ dependencies = [
  "windows-link",
 ]
 
+[[package]]
+name = "windows-targets"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
+dependencies = [
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc",
+]
+
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
+
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
+
+[[package]]
+name = "windows_i686_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
+
+[[package]]
+name = "windows_i686_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
+
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
+
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
+
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"
+
 [[package]]
 name = "wit-bindgen"
 version = "0.57.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
 
+[[package]]
+name = "writeable"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc"
+
+[[package]]
+name = "yoke"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5"
+dependencies = [
+ "stable_deref_trait",
+ "yoke-derive",
+ "zerofrom",
+]
+
+[[package]]
+name = "yoke-derive"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+ "synstructure",
+]
+
 [[package]]
 name = "zerocopy"
 version = "0.8.62"
@@ -1532,6 +2450,60 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "zerofrom"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272"
+dependencies = [
+ "zerofrom-derive",
+]
+
+[[package]]
+name = "zerofrom-derive"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+ "synstructure",
+]
+
+[[package]]
+name = "zerotrie"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f"
+dependencies = [
+ "displaydoc",
+ "yoke",
+ "zerofrom",
+]
+
+[[package]]
+name = "zerovec"
+version = "0.11.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8"
+dependencies = [
+ "yoke",
+ "zerofrom",
+ "zerovec-derive",
+]
+
+[[package]]
+name = "zerovec-derive"
+version = "0.11.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
 [[package]]
 name = "zmij"
 version = "1.0.23"
diff --git a/judge/README.md b/judge/README.md
index cf7b310..d9eeb69 100644
--- a/judge/README.md
+++ b/judge/README.md
@@ -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、队列或权限依赖。
diff --git a/judge/crates/sandbox/src/factory.rs b/judge/crates/sandbox/src/factory.rs
index 7337643..8898603 100644
--- a/judge/crates/sandbox/src/factory.rs
+++ b/judge/crates/sandbox/src/factory.rs
@@ -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()?;
diff --git a/judge/crates/sandbox/tests/pipeline_vm.rs b/judge/crates/sandbox/tests/pipeline_vm.rs
index cff60ac..42b1589 100644
--- a/judge/crates/sandbox/tests/pipeline_vm.rs
+++ b/judge/crates/sandbox/tests/pipeline_vm.rs
@@ -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!(
diff --git a/judge/crates/worker/Cargo.toml b/judge/crates/worker/Cargo.toml
index 79de4df..70adde9 100644
--- a/judge/crates/worker/Cargo.toml
+++ b/judge/crates/worker/Cargo.toml
@@ -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
diff --git a/judge/crates/worker/src/consumer.rs b/judge/crates/worker/src/consumer.rs
new file mode 100644
index 0000000..c23c29e
--- /dev/null
+++ b/judge/crates/worker/src/consumer.rs
@@ -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();
+    }
+}
diff --git a/judge/crates/worker/src/lib.rs b/judge/crates/worker/src/lib.rs
index 5024723..71603c9 100644
--- a/judge/crates/worker/src/lib.rs
+++ b/judge/crates/worker/src/lib.rs
@@ -114,3 +114,6 @@ pub mod native {
 
 #[cfg(target_os = "linux")]
 pub mod pipeline;
+
+#[cfg(target_os = "linux")]
+pub mod consumer;
diff --git a/judge/crates/worker/src/main.rs b/judge/crates/worker/src/main.rs
index 7828cf6..2951798 100644
--- a/judge/crates/worker/src/main.rs
+++ b/judge/crates/worker/src/main.rs
@@ -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,
diff --git a/judge/crates/worker/src/pipeline.rs b/judge/crates/worker/src/pipeline.rs
index 4121b43..4b7b718 100644
--- a/judge/crates/worker/src/pipeline.rs
+++ b/judge/crates/worker/src/pipeline.rs
@@ -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(