feat(sandbox): add bounded standalone launcher control

judge/Cargo.lock

@@ -119,6 +119,12 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "bitflags"
+version = "2.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
+
 [[package]]
 name = "bumpalo"
 version = "3.20.3"
@@ -379,6 +385,7 @@ version = "0.0.1"
 dependencies = [
  "clap",
  "kf-contracts",
+ "rustix",
  "serde",
  "serde_json",
 ]
@@ -406,6 +413,12 @@ version = "0.2.190"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78"
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
+
 [[package]]
 name = "log"
 version = "0.4.34"
@@ -518,6 +531,19 @@ version = "0.8.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
 
+[[package]]
+name = "rustix"
+version = "1.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d"
+dependencies = [
+ "bitflags",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys",
+]
+
 [[package]]
 name = "rustversion"
 version = "1.0.23"

judge/README.md

@@ -91,8 +91,36 @@ FD、栈和其他信号仍按原退出状态记录，缺少专门证据时不猜
 最先停止的原因只用于调度诊断，最终执行分类仍由结果模型依据全部观察确定。
 
 事件仅由可信监督器在核验同一任务的实际操作后产生，不可从执行请求直接反序列化。
-该模块不执行系统调用、不证明隔离、不产生执行许可；有界 IPC、单线程启动器、真实终止与
-清理仍待实现和独立 VM 验证。当前 CLI `run` 保持拒绝执行。
+该状态模块不执行系统调用、不证明隔离、不产生执行许可；真实任务树终止与清理仍待实现和
+独立 VM 验证。当前 CLI `run` 保持拒绝执行。
+
+## 独立启动入口与控制通信
+
+`launcher/entry.rs` 在可信二进制的独立进程中运行，不启动 Tokio、不安装自定义 `pre_exec`
+闭包，也不在 Worker 的 fork 子进程里运行 Rust 初始化逻辑。入口实际检查 `/proc/self/task`
+只有一个线程；未来每次权限操作或 fork 前须再次检查。当前入口以普通身份运行，没有实现
+特权 helper、namespace、降权、seccomp 或用户程序执行，初始化返回准备阶段的能力不可用。
+
+控制端必须先在可信 state 根下创建独占、权限 0700 的每次启动目录和 Unix listener，再通过
+清空环境的 `Command` 启动内部入口；socket 不放在用户工作区，控制通道与用户 stdio 分开。
+入口通过非阻塞、CLOEXEC socket 连接，队列满时直接失败，不在连接阶段无限等待。
+双方用内核 `SO_PEERCRED` 核对预期 PID/UID/GID；父端 PID 来源于仍持有的 `Child`，
+子端父 PID 来源于内核。内部入口不提供用户命令参数，握手消息不能转换成生命周期确认事件。
+
+`launcher/protocol.rs` 使用带版本的 8 字节帧头和最多 1 KiB JSON。先拒绝未知版本、标志及
+非法长度，再读取固定大小缓冲；发送序列化也使用固定缓冲。父子消息类型分开，拒绝多余字段、
+非法 run_id 和未知命令。所有 I/O 共用一个绝对截止时间（入口 5 秒，通道最大 10 秒），
+部分帧不能续期；EOF、解析、超时或写入错误关闭并锁定通道，不能重新同步后继续初始化。
+
+新增 `rustix` 1.1.5 仅用于沙箱的安全 syscall/FD 封装，启用 `net`、`process`，由锁文件固定。
+选择它的 OwnedFd/AsFd 接口完成连接与凭据核对；目前无需 `nix` 或直接 `libc` 调用，
+也未放宽 `unsafe` 限制。参见 [rustix](https://github.com/bytecodealliance/rustix)、
+[socket_peercred](https://docs.rs/rustix/1.1.5/rustix/net/sockopt/fn.socket_peercred.html) 和
+[fork 后的安全约束](https://doc.rust-lang.org/std/os/unix/process/trait.CommandExt.html#tymethod.pre_exec)。
+
+普通 Linux 回归实际启动可信入口，验证单线程、认证握手、准备拒绝、取消和断开后的退出回收；
+协议回归验证边界与失败锁定。这些测试没有运行用户代码或安全攻击，不是 Linux 隔离验收。
+目标进程的阻塞/放行、完整 FD 白名单、进程树监督、阶段回滚及独立 VM 安全验收仍待完成。
 
 ## 可信运行配置与启动前检查
 

judge/crates/sandbox/Cargo.toml

@@ -10,6 +10,7 @@ kf-contracts.workspace = true
 clap.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+rustix = { version = "1.1.5", features = ["net", "process"] }
 
 [lints]
 workspace = true

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

@@ -0,0 +1,79 @@
+//! Runs after a fresh exec of the trusted binary, without a Tokio runtime.
+//! No custom pre_exec closure, fork, user program, or permission change occurs.
+
+use super::protocol::{ChildMessage, ControlChannel, ExpectedPeer, ParentMessage};
+use crate::outcome::{ErrorCode, ExecutionStage};
+use std::io;
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+pub const STARTUP_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Parent must create a listener in a fresh private directory under its trusted
+/// state root before spawning this entry. Socket path is trusted deployment data.
+/// The kernel peer credentials, rather than that path, authenticate the parent.
+pub fn serve(socket_path: &Path) -> io::Result<()> {
+    let deadline = Instant::now() + STARTUP_TIMEOUT;
+    ensure_single_threaded()?;
+    let parent = rustix::process::getppid()
+        .filter(|pid| *pid != rustix::process::Pid::INIT)
+        .ok_or_else(|| {
+            io::Error::new(io::ErrorKind::PermissionDenied, "missing launcher parent")
+        })?;
+    // A full/stalled listener queue must fail, rather than block before the
+    // channel deadline is installed. OwnedFd closes on every failure path.
+    let socket = rustix::net::socket_with(
+        rustix::net::AddressFamily::UNIX,
+        rustix::net::SocketType::STREAM,
+        rustix::net::SocketFlags::CLOEXEC | rustix::net::SocketFlags::NONBLOCK,
+        None,
+    )?;
+    let address = rustix::net::SocketAddrUnix::new(socket_path)?;
+    rustix::net::connect(&socket, &address)?;
+    let stream = UnixStream::from(socket);
+    let mut channel = ControlChannel::new(
+        stream,
+        ExpectedPeer {
+            pid: parent.as_raw_pid() as u32,
+            uid: rustix::process::geteuid().as_raw(),
+            gid: rustix::process::getegid().as_raw(),
+        },
+        deadline,
+    )?;
+    channel.send(&ChildMessage::Hello {
+        pid: std::process::id(),
+        threads: 1,
+    })?;
+    match channel.receive::<ParentMessage>()? {
+        ParentMessage::Initialize { run_id } => {
+            // A transport handshake is never an isolation or exec permit.
+            channel.send(&ChildMessage::InitializationRefused {
+                run_id,
+                stage: ExecutionStage::Preparation,
+                code: ErrorCode::CapabilityUnavailable,
+            })?;
+        }
+        ParentMessage::Cancel {} => channel.send(&ChildMessage::Cancelled {})?,
+    }
+    Ok(())
+}
+
+/// Must be repeated immediately before future privileged/fork operations.
+pub fn ensure_single_threaded() -> io::Result<()> {
+    let mut count = 0;
+    for entry in std::fs::read_dir("/proc/self/task")? {
+        entry?;
+        count += 1;
+        if count > 1 {
+            break;
+        }
+    }
+    if count != 1 {
+        return Err(io::Error::new(
+            io::ErrorKind::PermissionDenied,
+            "launcher requires one thread",
+        ));
+    }
+    Ok(())
+}

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

@@ -1,7 +1,12 @@
 //! Trusted supervisor/launcher coordination, separate from user stdio.
 //!
 //! Lifecycle events describe completed operations, not commands received from
-//! an untrusted request. OS operations and authenticated bounded IPC are pending;
-//! this module grants no execution authority and does not change CLI refusal.
+//! an untrusted request. The standalone entry authenticates its parent and uses
+//! bounded IPC, but refuses initialization until OS isolation is implemented.
 
 pub mod lifecycle;
+
+#[cfg(target_os = "linux")]
+pub mod entry;
+#[cfg(target_os = "linux")]
+pub mod protocol;

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

@@ -0,0 +1,314 @@
+//! Private per-attempt Unix control transport, never user stdin/stdout.
+//! Messages cannot produce Lifecycle events or prove completed OS operations.
+
+use crate::outcome::{ErrorCode, ExecutionStage};
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
+use std::io::{self, Read, Write};
+use std::net::Shutdown;
+use std::os::unix::net::UnixStream;
+use std::time::{Duration, Instant};
+
+pub const MAX_MESSAGE_BYTES: usize = 1024;
+pub const MAX_CONTROL_TIMEOUT: Duration = Duration::from_secs(10);
+const PREFIX: [u8; 4] = [b'K', b'F', 1, 0];
+
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "command", rename_all = "snake_case", deny_unknown_fields)]
+pub enum ParentMessage {
+    Initialize { run_id: String },
+    Cancel {},
+}
+
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)]
+pub enum ChildMessage {
+    Hello {
+        pid: u32,
+        threads: u8,
+    },
+    InitializationRefused {
+        run_id: String,
+        stage: ExecutionStage,
+        code: ErrorCode,
+    },
+    Cancelled {},
+}
+
+mod sealed {
+    pub trait Sealed {}
+}
+
+/// Sealed to keep the control protocol separate from arbitrary execution data.
+pub trait WireMessage: sealed::Sealed + Serialize + DeserializeOwned {
+    fn valid(&self) -> bool;
+}
+
+impl sealed::Sealed for ParentMessage {}
+impl WireMessage for ParentMessage {
+    fn valid(&self) -> bool {
+        match self {
+            Self::Initialize { run_id } => valid_run_id(run_id),
+            Self::Cancel {} => true,
+        }
+    }
+}
+
+impl sealed::Sealed for ChildMessage {}
+impl WireMessage for ChildMessage {
+    fn valid(&self) -> bool {
+        match self {
+            Self::Hello { pid, threads } => *pid > 0 && *threads == 1,
+            Self::InitializationRefused { run_id, .. } => valid_run_id(run_id),
+            Self::Cancelled {} => true,
+        }
+    }
+}
+
+fn valid_run_id(value: &str) -> bool {
+    !value.is_empty()
+        && value.len() <= 64
+        && value
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || b"_-".contains(&byte))
+}
+
+/// Expected credentials come from the spawned Child handle / actual parent,
+/// never the JSON payload. A separate stream is required for every attempt.
+#[derive(Clone, Copy, Debug)]
+pub struct ExpectedPeer {
+    pub pid: u32,
+    pub uid: u32,
+    pub gid: u32,
+}
+
+pub struct ControlChannel {
+    stream: UnixStream,
+    deadline: Instant,
+    poisoned: bool,
+}
+
+impl ControlChannel {
+    pub fn new(stream: UnixStream, peer: ExpectedPeer, deadline: Instant) -> io::Result<Self> {
+        let remaining = deadline
+            .checked_duration_since(Instant::now())
+            .filter(|duration| !duration.is_zero() && *duration <= MAX_CONTROL_TIMEOUT)
+            .ok_or_else(|| {
+                io::Error::new(io::ErrorKind::InvalidInput, "invalid control deadline")
+            })?;
+        let credentials = rustix::net::sockopt::socket_peercred(&stream)?;
+        if credentials.pid.as_raw_pid() as u32 != peer.pid
+            || credentials.uid.as_raw() != peer.uid
+            || credentials.gid.as_raw() != peer.gid
+        {
+            return Err(io::Error::new(
+                io::ErrorKind::PermissionDenied,
+                "unexpected control peer",
+            ));
+        }
+        stream.set_nonblocking(false)?;
+        stream.set_read_timeout(Some(remaining))?;
+        stream.set_write_timeout(Some(remaining))?;
+        Ok(Self {
+            stream,
+            deadline,
+            poisoned: false,
+        })
+    }
+
+    pub fn send<T: WireMessage>(&mut self, message: &T) -> io::Result<()> {
+        let result = self.send_inner(message);
+        self.finish(result)
+    }
+
+    pub fn receive<T: WireMessage>(&mut self) -> io::Result<T> {
+        let result = self.receive_inner();
+        self.finish(result)
+    }
+
+    fn send_inner<T: WireMessage>(&mut self, message: &T) -> io::Result<()> {
+        self.remaining()?;
+        if !message.valid() {
+            return Err(invalid("invalid control message"));
+        }
+        let mut body = FixedBuffer::default();
+        serde_json::to_writer(&mut body, message)
+            .map_err(|_| invalid("control message too large"))?;
+        let mut header = [0u8; 8];
+        header[..4].copy_from_slice(&PREFIX);
+        header[4..].copy_from_slice(&(body.len as u32).to_be_bytes());
+        self.write_bytes(&header)?;
+        self.write_bytes(&body.bytes[..body.len])
+    }
+
+    fn receive_inner<T: WireMessage>(&mut self) -> io::Result<T> {
+        let mut header = [0u8; 8];
+        self.read_bytes(&mut header)?;
+        let length = decode_header(header)?;
+        let mut bytes = [0u8; MAX_MESSAGE_BYTES];
+        self.read_bytes(&mut bytes[..length])?;
+        let message: T = serde_json::from_slice(&bytes[..length])
+            .map_err(|_| invalid("invalid control JSON"))?;
+        if !message.valid() {
+            return Err(invalid("invalid control message"));
+        }
+        Ok(message)
+    }
+
+    fn remaining(&self) -> io::Result<Duration> {
+        if self.poisoned {
+            return Err(io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "control channel failed",
+            ));
+        }
+        self.deadline
+            .checked_duration_since(Instant::now())
+            .filter(|duration| !duration.is_zero())
+            .ok_or_else(|| io::Error::new(io::ErrorKind::TimedOut, "control deadline expired"))
+    }
+
+    fn read_bytes(&mut self, mut bytes: &mut [u8]) -> io::Result<()> {
+        while !bytes.is_empty() {
+            self.stream.set_read_timeout(Some(self.remaining()?))?;
+            match self.stream.read(bytes) {
+                Ok(0) => return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "control EOF")),
+                Ok(count) => bytes = &mut bytes[count..],
+                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
+                Err(error) => return Err(error),
+            }
+        }
+        Ok(())
+    }
+
+    fn write_bytes(&mut self, mut bytes: &[u8]) -> io::Result<()> {
+        while !bytes.is_empty() {
+            self.stream.set_write_timeout(Some(self.remaining()?))?;
+            match self.stream.write(bytes) {
+                Ok(0) => {
+                    return Err(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "control write failed",
+                    ));
+                }
+                Ok(count) => bytes = &bytes[count..],
+                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
+                Err(error) => return Err(error),
+            }
+        }
+        Ok(())
+    }
+
+    fn finish<T>(&mut self, result: io::Result<T>) -> io::Result<T> {
+        if result.is_err() {
+            self.poisoned = true;
+            let _ = self.stream.shutdown(Shutdown::Both);
+        }
+        result
+    }
+}
+
+fn invalid(message: &'static str) -> io::Error {
+    io::Error::new(io::ErrorKind::InvalidData, message)
+}
+
+fn decode_header(header: [u8; 8]) -> io::Result<usize> {
+    if header[..4] != PREFIX {
+        return Err(invalid("unsupported control framing"));
+    }
+    let length =
+        u32::from_be_bytes(header[4..].try_into().expect("four-byte header length")) as usize;
+    if length == 0 || length > MAX_MESSAGE_BYTES {
+        return Err(invalid("invalid control frame length"));
+    }
+    Ok(length)
+}
+
+struct FixedBuffer {
+    bytes: [u8; MAX_MESSAGE_BYTES],
+    len: usize,
+}
+
+impl Default for FixedBuffer {
+    fn default() -> Self {
+        Self {
+            bytes: [0; MAX_MESSAGE_BYTES],
+            len: 0,
+        }
+    }
+}
+
+impl Write for FixedBuffer {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        if bytes.len() > MAX_MESSAGE_BYTES - self.len {
+            return Err(invalid("control message too large"));
+        }
+        self.bytes[self.len..self.len + bytes.len()].copy_from_slice(bytes);
+        self.len += bytes.len();
+        Ok(bytes.len())
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn header_rejects_oversized_length_before_body_read() {
+        for length in [0, 1025, u32::MAX] {
+            let mut header = [0u8; 8];
+            header[..4].copy_from_slice(&PREFIX);
+            header[4..].copy_from_slice(&length.to_be_bytes());
+            assert!(decode_header(header).is_err());
+        }
+    }
+
+    #[test]
+    fn header_rejects_version_and_reserved_flags() {
+        for prefix in [[b'K', b'F', 2, 0], [b'K', b'F', 1, 1], [0; 4]] {
+            let mut header = [0u8; 8];
+            header[..4].copy_from_slice(&prefix);
+            header[4..].copy_from_slice(&1u32.to_be_bytes());
+            assert!(decode_header(header).is_err());
+        }
+    }
+
+    #[test]
+    fn serializer_has_fixed_storage() {
+        let mut buffer = FixedBuffer::default();
+        buffer.write_all(&[0; MAX_MESSAGE_BYTES]).unwrap();
+        assert!(buffer.write_all(b"extra").is_err());
+        assert_eq!(buffer.len, MAX_MESSAGE_BYTES);
+    }
+
+    #[test]
+    fn schemas_reject_unknown_fields_and_direction() {
+        assert!(
+            serde_json::from_str::<ParentMessage>(r#"{"command":"cancel","exec":true}"#).is_err()
+        );
+        assert!(serde_json::from_str::<ChildMessage>(r#"{"command":"cancel"}"#).is_err());
+        assert!(serde_json::from_str::<ParentMessage>(r#"{"command":"release_exec"}"#).is_err());
+    }
+
+    #[test]
+    fn identifiers_and_thread_claims_are_bounded() {
+        for run_id in ["", "../work", "bad id"] {
+            assert!(
+                !ParentMessage::Initialize {
+                    run_id: run_id.into()
+                }
+                .valid()
+            );
+        }
+        assert!(
+            !ParentMessage::Initialize {
+                run_id: "x".repeat(65)
+            }
+            .valid()
+        );
+        assert!(!ChildMessage::Hello { pid: 1, threads: 2 }.valid());
+        assert!(!ChildMessage::Hello { pid: 0, threads: 1 }.valid());
+    }
+}

judge/crates/sandbox/src/main.rs

@@ -28,6 +28,13 @@ enum Command {
     },
     /// Reserved execution entrypoint; refuses execution until isolation is implemented.
     Run,
+    /// Internal trusted launcher control entry; never runs a user program.
+    #[cfg(target_os = "linux")]
+    #[command(hide = true)]
+    Launcher {
+        #[arg(long)]
+        control_socket: PathBuf,
+    },
 }
 
 fn main() -> ExitCode {
@@ -50,6 +57,18 @@ fn main() -> ExitCode {
         }
         Command::Validate { spec } => validate_spec(spec),
         Command::CheckRuntime { config } => check_runtime(config),
+        #[cfg(target_os = "linux")]
+        Command::Launcher { control_socket } => {
+            match kf_sandbox::launcher::entry::serve(&control_socket) {
+                Ok(()) => ExitCode::from(2),
+                Err(error) => spec_error(
+                    "launcher_control_failed",
+                    Some(serde_json::json!({
+                        "io_kind": format!("{:?}", error.kind())
+                    })),
+                ),
+            }
+        }
     }
 }
 

judge/crates/sandbox/tests/launcher_control.rs

@@ -0,0 +1,303 @@
+#![cfg(target_os = "linux")]
+
+use kf_sandbox::launcher::protocol::{ChildMessage, ControlChannel, ExpectedPeer, ParentMessage};
+use kf_sandbox::outcome::{ErrorCode, ExecutionStage};
+use std::fs::{self, DirBuilder};
+use std::io::{self, Write};
+use std::os::unix::fs::DirBuilderExt;
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+static ATTEMPT: AtomicU64 = AtomicU64::new(0);
+
+struct Attempt {
+    directory: PathBuf,
+    child: Option<Child>,
+}
+
+impl Attempt {
+    fn start() -> (Self, ControlChannel, u32) {
+        let directory = std::env::temp_dir().join(format!(
+            "kf-l-{}-{}",
+            std::process::id(),
+            ATTEMPT.fetch_add(1, Ordering::Relaxed)
+        ));
+        DirBuilder::new().mode(0o700).create(&directory).unwrap();
+        let mut attempt = Self {
+            directory,
+            child: None,
+        };
+        let socket = attempt.directory.join("control.sock");
+        let listener = UnixListener::bind(&socket).unwrap();
+        listener.set_nonblocking(true).unwrap();
+        let deadline = Instant::now() + Duration::from_secs(5);
+        let child = Command::new(env!("CARGO_BIN_EXE_kf-sandbox"))
+            .arg("launcher")
+            .arg("--control-socket")
+            .arg(&socket)
+            .env_clear()
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+            .unwrap();
+        let pid = child.id();
+        attempt.child = Some(child);
+        let stream = loop {
+            match listener.accept() {
+                Ok((stream, _)) => break stream,
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                    assert!(
+                        Instant::now() < deadline,
+                        "launcher must connect within deadline"
+                    );
+                    assert!(
+                        attempt
+                            .child
+                            .as_mut()
+                            .unwrap()
+                            .try_wait()
+                            .unwrap()
+                            .is_none()
+                    );
+                    std::thread::sleep(Duration::from_millis(5));
+                }
+                Err(error) => panic!("cannot accept launcher: {error}"),
+            }
+        };
+        let channel = ControlChannel::new(stream, own_peer(pid), deadline).unwrap();
+        assert_eq!(
+            fs::read_dir(format!("/proc/{pid}/task")).unwrap().count(),
+            1
+        );
+        (attempt, channel, pid)
+    }
+
+    fn reap(&mut self) {
+        let deadline = Instant::now() + Duration::from_secs(5);
+        loop {
+            if let Some(status) = self.child.as_mut().unwrap().try_wait().unwrap() {
+                assert_eq!(
+                    status.code(),
+                    Some(2),
+                    "entry never reports execution success"
+                );
+                self.child = None;
+                return;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "launcher must exit and be reaped"
+            );
+            std::thread::sleep(Duration::from_millis(5));
+        }
+    }
+}
+
+impl Drop for Attempt {
+    fn drop(&mut self) {
+        if let Some(child) = &mut self.child {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+        let _ = fs::remove_file(self.directory.join("control.sock"));
+        let _ = fs::remove_dir(&self.directory);
+    }
+}
+
+fn own_peer(pid: u32) -> ExpectedPeer {
+    ExpectedPeer {
+        pid,
+        uid: rustix::process::geteuid().as_raw(),
+        gid: rustix::process::getegid().as_raw(),
+    }
+}
+
+fn pair() -> (ControlChannel, UnixStream) {
+    let (stream, writer) = UnixStream::pair().unwrap();
+    (
+        ControlChannel::new(
+            stream,
+            own_peer(std::process::id()),
+            Instant::now() + Duration::from_secs(2),
+        )
+        .unwrap(),
+        writer,
+    )
+}
+
+#[test]
+fn fresh_single_thread_entry_exchanges_authenticated_messages_then_refuses() {
+    let (mut attempt, mut channel, pid) = Attempt::start();
+    assert_eq!(
+        channel.receive::<ChildMessage>().unwrap(),
+        ChildMessage::Hello { pid, threads: 1 }
+    );
+    channel
+        .send(&ParentMessage::Initialize {
+            run_id: "attempt-001".into(),
+        })
+        .unwrap();
+    assert_eq!(
+        channel.receive::<ChildMessage>().unwrap(),
+        ChildMessage::InitializationRefused {
+            run_id: "attempt-001".into(),
+            stage: ExecutionStage::Preparation,
+            code: ErrorCode::CapabilityUnavailable,
+        }
+    );
+    attempt.reap();
+    assert_eq!(
+        channel.receive::<ChildMessage>().unwrap_err().kind(),
+        io::ErrorKind::UnexpectedEof
+    );
+}
+
+#[test]
+fn parent_can_cancel_before_initialization_and_reap_entry() {
+    let (mut attempt, mut channel, pid) = Attempt::start();
+    assert_eq!(
+        channel.receive::<ChildMessage>().unwrap(),
+        ChildMessage::Hello { pid, threads: 1 }
+    );
+    channel.send(&ParentMessage::Cancel {}).unwrap();
+    assert_eq!(
+        channel.receive::<ChildMessage>().unwrap(),
+        ChildMessage::Cancelled {}
+    );
+    attempt.reap();
+}
+
+#[test]
+fn parent_disconnect_does_not_leave_a_waiting_launcher() {
+    let (mut attempt, mut channel, _) = Attempt::start();
+    assert!(matches!(
+        channel.receive::<ChildMessage>().unwrap(),
+        ChildMessage::Hello { .. }
+    ));
+    drop(channel);
+    attempt.reap();
+}
+
+#[test]
+fn kernel_credentials_reject_an_unexpected_peer() {
+    for peer in [
+        ExpectedPeer {
+            pid: std::process::id() + 1,
+            ..own_peer(std::process::id())
+        },
+        ExpectedPeer {
+            uid: rustix::process::geteuid().as_raw() + 1,
+            ..own_peer(std::process::id())
+        },
+        ExpectedPeer {
+            gid: rustix::process::getegid().as_raw() + 1,
+            ..own_peer(std::process::id())
+        },
+    ] {
+        let (stream, _writer) = UnixStream::pair().unwrap();
+        let result = ControlChannel::new(stream, peer, Instant::now() + Duration::from_secs(1));
+        assert!(matches!(result, Err(error) if error.kind() == io::ErrorKind::PermissionDenied));
+    }
+}
+
+#[test]
+fn truncated_frame_poisoning_prevents_resynchronization() {
+    let (mut channel, mut writer) = pair();
+    writer.write_all(b"KF\x01").unwrap();
+    drop(writer);
+    assert_eq!(
+        channel.receive::<ParentMessage>().unwrap_err().kind(),
+        io::ErrorKind::UnexpectedEof
+    );
+    assert_eq!(
+        channel.send(&ParentMessage::Cancel {}).unwrap_err().kind(),
+        io::ErrorKind::BrokenPipe
+    );
+}
+
+#[test]
+fn two_frames_remain_separate_and_keep_their_direction() {
+    let (mut channel, stream) = pair();
+    let mut writer = ControlChannel::new(
+        stream,
+        own_peer(std::process::id()),
+        Instant::now() + Duration::from_secs(2),
+    )
+    .unwrap();
+    writer
+        .send(&ParentMessage::Initialize {
+            run_id: "run-1".into(),
+        })
+        .unwrap();
+    writer.send(&ParentMessage::Cancel {}).unwrap();
+    assert_eq!(
+        channel.receive::<ParentMessage>().unwrap(),
+        ParentMessage::Initialize {
+            run_id: "run-1".into()
+        }
+    );
+    assert_eq!(
+        channel.receive::<ParentMessage>().unwrap(),
+        ParentMessage::Cancel {}
+    );
+}
+
+#[test]
+fn unknown_fields_poison_channel_before_any_later_command() {
+    let (mut channel, mut writer) = pair();
+    let bytes = br#"{"command":"cancel","extra":true}"#;
+    writer.write_all(b"KF\x01\x00").unwrap();
+    writer
+        .write_all(&(bytes.len() as u32).to_be_bytes())
+        .unwrap();
+    writer.write_all(bytes).unwrap();
+    assert_eq!(
+        channel.receive::<ParentMessage>().unwrap_err().kind(),
+        io::ErrorKind::InvalidData
+    );
+    assert_eq!(
+        channel.receive::<ParentMessage>().unwrap_err().kind(),
+        io::ErrorKind::BrokenPipe
+    );
+}
+
+#[test]
+fn partial_body_cannot_reset_the_absolute_deadline() {
+    let (stream, mut writer) = UnixStream::pair().unwrap();
+    let mut channel = ControlChannel::new(
+        stream,
+        own_peer(std::process::id()),
+        Instant::now() + Duration::from_millis(80),
+    )
+    .unwrap();
+    writer.write_all(b"KF\x01\x00\x00\x00\x00\x10{").unwrap();
+    let started = Instant::now();
+    let error = channel.receive::<ParentMessage>().unwrap_err();
+    assert!(matches!(
+        error.kind(),
+        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
+    ));
+    assert!(started.elapsed() < Duration::from_secs(1));
+    assert_eq!(
+        channel.send(&ParentMessage::Cancel {}).unwrap_err().kind(),
+        io::ErrorKind::BrokenPipe
+    );
+}
+
+#[test]
+fn deadline_must_be_finite_short_and_in_the_future() {
+    for deadline in [
+        Instant::now() - Duration::from_secs(1),
+        Instant::now() + Duration::from_secs(30),
+    ] {
+        let (stream, _writer) = UnixStream::pair().unwrap();
+        assert!(
+            matches!(ControlChannel::new(stream, own_peer(std::process::id()), deadline),
+            Err(error) if error.kind() == io::ErrorKind::InvalidInput)
+        );
+    }
+}