feat(sandbox): recover service-owned tasks after supervisor restart

judge/crates/sandbox/src/acceptance.rs

@@ -10,6 +10,7 @@ use std::{fs, io, io::Read, os::unix::fs::PermissionsExt, path::Path};
 pub(crate) mod isolation;
 mod normal;
 mod resources;
+pub(crate) mod restart;
 pub(crate) fn verify_guest(require_env: bool) -> io::Result<()> {
     if (require_env && std::env::var("KF_SANDBOX_INDEPENDENT_VM").as_deref() != Ok("1"))
         || fs::read_to_string("/kf/INDEPENDENT_VM")?.trim() != "isolated-test-guest"
@@ -116,6 +117,9 @@ fn prefix(file: &crate::artifacts::SnapshotFile) -> io::Result<String> {
 }
 pub fn run_case(case: &str) -> io::Result<serde_json::Value> {
     verify_guest(true)?;
+    if case.starts_with("restart-") {
+        return restart::run_case(case);
+    }
     if case.starts_with("normal-") {
         return normal::run_case(case);
     }

judge/crates/sandbox/src/acceptance/restart.rs

@@ -0,0 +1,147 @@
+//! Fixed real owner-death/restart cases in the independent capped VM.
+use super::{Fixture, describe};
+use crate::{
+    artifacts::ArtifactRoot,
+    launcher::{
+        capture::CaptureLease,
+        reap::OwnedChildren,
+        run::{self, MonitorMode},
+        seccomp::FilteredTarget,
+    },
+    runtime::{RuntimeConfig, ValidatedRuntimeConfig},
+    spec::{ExecutionSpec, SpecPolicy, ValidatedSpec},
+};
+use rustix::fs::{Mode, OFlags};
+use std::{
+    fs,
+    io::{self, Read},
+    os::unix::fs::PermissionsExt,
+    process::{Child, Command, Stdio},
+    time::{Duration, Instant},
+};
+pub(crate) fn ready(
+    config: &ValidatedRuntimeConfig,
+    spec: &ValidatedSpec,
+    stage: &str,
+) -> io::Result<Child> {
+    super::verify_guest(true)?;
+    let mut child = Command::new(std::env::current_exe()?)
+        .args(["vm-case", "restart-residual"])
+        .env_clear()
+        .env("KF_SANDBOX_INDEPENDENT_VM", "1")
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()?;
+    // This PID belongs to our unreaped direct Child, not a request/frame.
+    let setup = (|| {
+        fs::write(
+            format!(
+                "{}/kf-{}/cgroup.procs",
+                config.config().delegated_cgroup_root,
+                spec.spec().run_id
+            ),
+            child.id().to_string(),
+        )?;
+        let value = serde_json::json!({"stage":stage,"owner_pid":std::process::id(),"residual_pid":child.id(),"run_id":spec.spec().run_id});
+        fs::write("/kf/restart-ready.next", serde_json::to_vec(&value)?)?;
+        fs::rename("/kf/restart-ready.next", "/kf/restart-ready")?;
+        Ok::<_, io::Error>(())
+    })();
+    if let Err(e) = setup {
+        let _ = child.kill();
+        let _ = child.wait();
+        return Err(e);
+    }
+    Ok(child)
+}
+fn config() -> io::Result<ValidatedRuntimeConfig> {
+    let fd = rustix::fs::open(
+        "/kf/restart-runtime.json",
+        OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+        Mode::empty(),
+    )?;
+    let st = rustix::fs::fstat(&fd)?;
+    if st.st_uid != 0
+        || st.st_gid != 0
+        || st.st_mode & 0o7777 != 0o600
+        || st.st_nlink != 1
+        || rustix::fs::FileType::from_raw_mode(st.st_mode) != rustix::fs::FileType::RegularFile
+    {
+        return Err(io::Error::other("trusted restart fixture config ownership"));
+    }
+    let mut bytes = Vec::new();
+    std::fs::File::from(fd)
+        .take(crate::runtime::MAX_CONFIG_BYTES + 1)
+        .read_to_end(&mut bytes)?;
+    if bytes.len() as u64 > crate::runtime::MAX_CONFIG_BYTES {
+        return Err(io::Error::other("restart config bound"));
+    }
+    let c: RuntimeConfig = serde_json::from_slice(&bytes)?;
+    if !c
+        .delegated_cgroup_root
+        .starts_with("/sys/fs/cgroup/kf-acceptance-")
+    {
+        return Err(io::Error::other("fixed acceptance cgroup only"));
+    }
+    c.validate().map_err(io::Error::other)
+}
+pub(super) fn run_case(case: &str) -> io::Result<serde_json::Value> {
+    if case == "restart-recover" {
+        return serde_json::to_value(crate::cleanup::recover(&config()?)?)
+            .map_err(io::Error::other);
+    }
+    if case == "restart-residual" {
+        // Real partial-preparation residual: no parent-death guarantee. It is
+        // bounded by the owning task and the independently capped VM, and its
+        // own 30s deadline prevents an unbounded acceptance helper.
+        std::thread::sleep(Duration::from_secs(30));
+        return Ok(serde_json::json!({"residual_deadline":true}));
+    }
+    if !matches!(case, "restart-prepared" | "restart-live") {
+        return Err(io::Error::other("unknown fixed restart case"));
+    }
+    let fixture = Fixture::new("/kf/bin/isolation-probe")?;
+    let config = fixture.config()?;
+    let root: ArtifactRoot = fixture.root()?;
+    let mut request: ExecutionSpec =
+        serde_json::from_str(include_str!("../../tests/fixtures/spec.valid.json"))?;
+    request.run_id = case.into();
+    request.argv.push("peer".into());
+    request.stdin_artifact = None;
+    request.limits.cpu_time_us = 2_000_000;
+    request.limits.wall_time_us = 30_000_000;
+    let spec = request
+        .validate(&SpecPolicy::default())
+        .map_err(io::Error::other)?;
+    if case == "restart-live" {
+        let completed = run::execute(
+            &config,
+            &spec,
+            &std::env::current_exe()?,
+            &root,
+            MonitorMode::RestartReady,
+        )
+        .map_err(|e| io::Error::other(format!("restart live {:?}", e.failure)))?;
+        return describe(case, &spec, &completed, false);
+    }
+    let mut owned = OwnedChildren::begin()?;
+    crate::launcher::capture::private_supervisor()?;
+    let mut target = FilteredTarget::prepare(&config, &spec, &std::env::current_exe()?, &root)
+        .map_err(|e| io::Error::other(format!("restart prepared {:?}", e.failure)))?;
+    let (mut lease, dir) = CaptureLease::prepare(&config, &spec)?;
+    let (pump, stdio) = crate::io::PipePump::prepare(target.take_stdin(), &spec, dir)?;
+    target.root_mut().target.prepare_execution(&spec, stdio)?;
+    let _child = ready(&config, &spec, "prepared")?;
+    let deadline = Instant::now() + Duration::from_secs(15);
+    while Instant::now() < deadline {
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    drop(pump);
+    lease.cleanup()?;
+    let result = target.refuse_at(crate::outcome::ExecutionStage::Exec, None);
+    owned.finish()?;
+    fs::remove_file("/kf/restart-ready")?;
+    fs::set_permissions(&fixture.publications, fs::Permissions::from_mode(0o700))?;
+    Ok(serde_json::json!({"unexpected_probe_deadline":true,"cleanup_error":result.cleanup_failure}))
+}

judge/crates/sandbox/src/cgroup.rs

@@ -55,6 +55,7 @@ pub struct DelegatedCgroup {
     policy: SpecPolicy,
     quota: u64,
     period: u64,
+    recovery_config: ValidatedRuntimeConfig,
 }
 
 impl DelegatedCgroup {
@@ -117,6 +118,7 @@ impl DelegatedCgroup {
             policy: config.spec_policy(),
             quota: budget.task_cpu_quota_us,
             period: budget.cpu_period_us,
+            recovery_config: config.clone(),
         })
     }
 
@@ -192,6 +194,12 @@ impl DelegatedCgroup {
             // bootstrap. Charge all initialization and descendant CPU from
             // enrollment; never estimate/subtract setup or reset peak later.
             task.baseline = Some(task.accounting_snapshot()?);
+            crate::cleanup::register(
+                &self.recovery_config,
+                spec,
+                crate::cleanup::Resource::Cgroup,
+                &task.fd,
+            )?;
             Ok::<_, io::Error>(())
         })();
         if let Err(source) = setup {
@@ -442,7 +450,7 @@ fn open_directory(path: &Path) -> io::Result<OwnedFd> {
     Ok(fd)
 }
 
-fn read(fd: &OwnedFd, name: &str) -> io::Result<String> {
+pub(crate) fn read(fd: &OwnedFd, name: &str) -> io::Result<String> {
     let file = rustix::fs::openat(
         fd,
         name,
@@ -459,7 +467,7 @@ fn read(fd: &OwnedFd, name: &str) -> io::Result<String> {
     Ok(text)
 }
 
-fn write(fd: &OwnedFd, name: &str, value: &str) -> io::Result<()> {
+pub(crate) fn write(fd: &OwnedFd, name: &str, value: &str) -> io::Result<()> {
     let file = rustix::fs::openat(
         fd,
         name,

judge/crates/sandbox/src/cleanup.rs

@@ -0,0 +1,714 @@
+//! Crash recovery of durable, root-owned task journals, never name-only adoption.
+//! Only fixed descendants of operator-validated roots are eligible. Unknown,
+//! replaced or incomplete resources remain reserved and produce a separate error.
+use crate::{
+    outcome::{ErrorCode, ExecutionError, ExecutionStage},
+    runtime::ValidatedRuntimeConfig,
+    spec::ValidatedSpec,
+};
+use rustix::{
+    fd::{AsRawFd, OwnedFd},
+    fs::{AtFlags, FlockOperation, Mode, OFlags},
+};
+use serde::{Deserialize, Serialize};
+use std::{
+    fs::File,
+    io::{self, Read, Write},
+    path::Path,
+    time::{Duration, Instant},
+};
+const SERVICE: &str = "kf-sandbox-task-v2";
+const MAX_RECORD: u64 = 8192;
+const DIR: OFlags = OFlags::RDONLY
+    .union(OFlags::DIRECTORY)
+    .union(OFlags::NOFOLLOW)
+    .union(OFlags::CLOEXEC);
+fn denied(message: &'static str) -> io::Error {
+    io::Error::new(io::ErrorKind::PermissionDenied, message)
+}
+fn private(fd: &OwnedFd, mode: u32) -> io::Result<()> {
+    let st = rustix::fs::fstat(fd)?;
+    if st.st_uid != 0 || st.st_gid != 0 || st.st_mode & 0o7777 != mode {
+        return Err(denied("unsafe recovery ownership/mode"));
+    }
+    Ok(())
+}
+#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+struct Inode {
+    dev: u64,
+    ino: u64,
+}
+impl Inode {
+    fn of(fd: &OwnedFd) -> io::Result<Self> {
+        let s = rustix::fs::fstat(fd)?;
+        Ok(Self {
+            dev: s.st_dev,
+            ino: s.st_ino,
+        })
+    }
+    fn check(&self, fd: &OwnedFd) -> io::Result<()> {
+        if *self != Self::of(fd)? {
+            return Err(denied("recorded resource inode changed"));
+        }
+        Ok(())
+    }
+}
+#[derive(Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+struct Owner {
+    boot_id: String,
+    pid: u32,
+    start_ticks: u64,
+}
+fn start_ticks(text: &str) -> io::Result<u64> {
+    // comm may contain spaces and ')'; fields after its final ')' start at 3.
+    text.rsplit_once(") ")
+        .and_then(|(_, tail)| tail.split_whitespace().nth(19))
+        .and_then(|n| n.parse().ok())
+        .filter(|n| *n > 0)
+        .ok_or_else(|| denied("invalid owner process start time"))
+}
+fn boot_id() -> io::Result<String> {
+    let id = std::fs::read_to_string("/proc/sys/kernel/random/boot_id")?
+        .trim()
+        .to_owned();
+    if id.len() != 36 || !id.bytes().all(|b| b.is_ascii_hexdigit() || b == b'-') {
+        return Err(denied("invalid boot identity"));
+    }
+    Ok(id)
+}
+impl Owner {
+    fn current() -> io::Result<Self> {
+        Ok(Self {
+            boot_id: boot_id()?,
+            pid: std::process::id(),
+            start_ticks: start_ticks(&std::fs::read_to_string("/proc/self/stat")?)?,
+        })
+    }
+    fn alive(&self) -> io::Result<bool> {
+        // A different kernel boot invalidates cgroup inode identity. Fail closed,
+        // rather than treating a persistent record as authority over new groups.
+        if self.boot_id != boot_id()? {
+            return Err(denied("recovery record belongs to another kernel boot"));
+        }
+        match std::fs::read_to_string(format!("/proc/{}/stat", self.pid)) {
+            Ok(s) => Ok(start_ticks(&s)? == self.start_ticks),
+            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(e),
+        }
+    }
+}
+#[derive(Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+struct Anchors {
+    state: Inode,
+    tasks: Inode,
+    captures: Inode,
+    cgroup: Inode,
+}
+struct Roots {
+    state: OwnedFd,
+    tasks: OwnedFd,
+    captures: OwnedFd,
+    cgroup: OwnedFd,
+}
+impl Roots {
+    fn open(config: &ValidatedRuntimeConfig) -> io::Result<Self> {
+        let c = config.config();
+        let open = crate::isolation::root::open_directory;
+        let state = open(Path::new(&c.state_root))?;
+        let tasks = open(Path::new(&c.tasks_root))?;
+        let captures = rustix::fs::openat(&state, "captures", DIR, Mode::empty())?;
+        private(&captures, 0o700)?;
+        let cgroup = open(Path::new(&c.delegated_cgroup_root))?;
+        if rustix::fs::fstatfs(&cgroup)?.f_type as u64 != 0x6367_7270 {
+            return Err(denied("recovery anchor is not cgroup v2"));
+        }
+        Ok(Self {
+            state,
+            tasks,
+            captures,
+            cgroup,
+        })
+    }
+    fn anchors(&self) -> io::Result<Anchors> {
+        Ok(Anchors {
+            state: Inode::of(&self.state)?,
+            tasks: Inode::of(&self.tasks)?,
+            captures: Inode::of(&self.captures)?,
+            cgroup: Inode::of(&self.cgroup)?,
+        })
+    }
+}
+#[derive(Clone, Copy)]
+pub(crate) enum Resource {
+    Task,
+    Cgroup,
+    #[cfg(feature = "vm-acceptance")]
+    Capture,
+}
+#[derive(Debug, Default, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+struct Resources {
+    task: Option<Inode>,
+    cgroup: Option<Inode>,
+    capture: Option<Inode>,
+}
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+struct Record {
+    service: String,
+    version: u32,
+    run_id: String,
+    uid: u32,
+    gid: u32,
+    owner: Owner,
+    anchors: Anchors,
+    resources: Resources,
+}
+fn read_record(slot: &OwnedFd) -> io::Result<Record> {
+    let fd = rustix::fs::openat(
+        slot,
+        "owner.json",
+        OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+        Mode::empty(),
+    )?;
+    private(&fd, 0o600)?;
+    let s = rustix::fs::fstat(&fd)?;
+    if rustix::fs::FileType::from_raw_mode(s.st_mode) != rustix::fs::FileType::RegularFile
+        || s.st_nlink != 1
+        || s.st_size < 0
+        || s.st_size as u64 > MAX_RECORD
+    {
+        return Err(denied("invalid journal file type/link/size"));
+    }
+    let mut bytes = Vec::new();
+    // Inspect a path-only capability before reading: a foreign FIFO/device
+    // named owner.json must never block or perform device I/O at startup.
+    let reader = rustix::fs::open(
+        format!("/proc/self/fd/{}", fd.as_raw_fd()),
+        OFlags::RDONLY | OFlags::CLOEXEC,
+        Mode::empty(),
+    )?;
+    File::from(reader)
+        .take(MAX_RECORD + 1)
+        .read_to_end(&mut bytes)?;
+    if bytes.len() as u64 > MAX_RECORD {
+        return Err(denied("journal read bound"));
+    }
+    let record: Record =
+        serde_json::from_slice(&bytes).map_err(|_| denied("unrecognized recovery record"))?;
+    crate::spec::identifier(&record.run_id, "run_id")
+        .map_err(|_| denied("invalid recovery task identifier"))?;
+    if record.service != SERVICE
+        || record.version != 2
+        || record.owner.pid <= 1
+        || record.owner.start_ticks == 0
+    {
+        return Err(denied("unrecognized recovery owner"));
+    }
+    Ok(record)
+}
+fn store(slot: &OwnedFd, record: &Record) -> io::Result<()> {
+    let mut bytes = serde_json::to_vec(record)?;
+    bytes.push(b'\n');
+    if bytes.len() as u64 > MAX_RECORD {
+        return Err(denied("journal write bound"));
+    }
+    let fd = rustix::fs::openat(
+        slot,
+        ".owner.next",
+        OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+        Mode::from_raw_mode(0o600),
+    )?;
+    let mut file = File::from(fd);
+    file.write_all(&bytes)?;
+    file.sync_all()?;
+    rustix::fs::renameat(slot, ".owner.next", slot, "owner.json")?;
+    rustix::fs::fsync(slot)?;
+    Ok(())
+}
+pub(crate) fn initialize_slot(
+    config: &ValidatedRuntimeConfig,
+    spec: &ValidatedSpec,
+    slot: &OwnedFd,
+    uid: u32,
+    gid: u32,
+) -> io::Result<()> {
+    rustix::fs::flock(slot, FlockOperation::NonBlockingLockExclusive)?;
+    private(slot, 0o700)?;
+    let state = crate::isolation::root::open_directory(Path::new(&config.config().state_root))?;
+    match rustix::fs::mkdirat(&state, "captures", Mode::from_raw_mode(0o700)) {
+        Ok(()) | Err(rustix::io::Errno::EXIST) => {}
+        Err(e) => return Err(e.into()),
+    }
+    let record = Record {
+        service: SERVICE.into(),
+        version: 2,
+        run_id: spec.spec().run_id.clone(),
+        uid,
+        gid,
+        owner: Owner::current()?,
+        anchors: Roots::open(config)?.anchors()?,
+        resources: Resources::default(),
+    };
+    store(slot, &record)
+}
+/// Primitive-only tests have no identity lease. A full task records each inode
+/// before that resource can enroll a user process or mount a capture filesystem.
+pub(crate) fn register(
+    config: &ValidatedRuntimeConfig,
+    spec: &ValidatedSpec,
+    resource: Resource,
+    fd: &OwnedFd,
+) -> io::Result<()> {
+    let state = crate::isolation::root::open_directory(Path::new(&config.config().state_root))?;
+    let parent = match rustix::fs::openat(&state, "identities", DIR, Mode::empty()) {
+        Ok(fd) => fd,
+        Err(rustix::io::Errno::NOENT) => return Ok(()),
+        Err(e) => return Err(e.into()),
+    };
+    private(&parent, 0o700)?;
+    let current = Owner::current()?;
+    for index in 0..config.config().max_concurrent_runs {
+        let (uid, gid) = config
+            .task_identity(index)
+            .ok_or_else(|| denied("identity slot"))?;
+        let slot = match rustix::fs::openat(&parent, format!("uid-{uid}"), DIR, Mode::empty()) {
+            Ok(fd) => fd,
+            Err(rustix::io::Errno::NOENT) => continue,
+            Err(e) => return Err(e.into()),
+        };
+        private(&slot, 0o700)?;
+        let Ok(mut record) = read_record(&slot) else {
+            continue;
+        };
+        if record.owner != current || record.run_id != spec.spec().run_id {
+            continue;
+        }
+        if record.uid != uid
+            || record.gid != gid
+            || record.anchors != Roots::open(config)?.anchors()?
+        {
+            return Err(denied("journal binding mismatch"));
+        }
+        let value = match resource {
+            Resource::Task => &mut record.resources.task,
+            Resource::Cgroup => &mut record.resources.cgroup,
+            #[cfg(feature = "vm-acceptance")]
+            Resource::Capture => &mut record.resources.capture,
+        };
+        if value.is_some() {
+            return Err(denied("resource already registered"));
+        }
+        *value = Some(Inode::of(fd)?);
+        return store(&slot, &record);
+    }
+    Ok(())
+}
+pub(crate) fn verify_owned(
+    config: &ValidatedRuntimeConfig,
+    spec: &ValidatedSpec,
+    capture: bool,
+) -> io::Result<()> {
+    let roots = Roots::open(config)?;
+    let parent = rustix::fs::openat(&roots.state, "identities", DIR, Mode::empty())?;
+    private(&parent, 0o700)?;
+    let current = Owner::current()?;
+    for index in 0..config.config().max_concurrent_runs {
+        let (uid, gid) = config
+            .task_identity(index)
+            .ok_or_else(|| denied("identity slot"))?;
+        let slot = match rustix::fs::openat(&parent, format!("uid-{uid}"), DIR, Mode::empty()) {
+            Ok(fd) => fd,
+            Err(rustix::io::Errno::NOENT) => continue,
+            Err(e) => return Err(e.into()),
+        };
+        private(&slot, 0o700)?;
+        let Ok(record) = read_record(&slot) else {
+            continue;
+        };
+        if record.owner == current
+            && record.run_id == spec.spec().run_id
+            && record.uid == uid
+            && record.gid == gid
+            && record.anchors == roots.anchors()?
+            && record.resources.task.is_some()
+            && record.resources.cgroup.is_some()
+            && (!capture || record.resources.capture.is_some())
+        {
+            return Ok(());
+        }
+    }
+    Err(denied("task has no complete durable recovery binding"))
+}
+fn names(fd: &OwnedFd) -> io::Result<Vec<String>> {
+    let entries: Vec<_> = std::fs::read_dir(format!("/proc/self/fd/{}", fd.as_raw_fd()))?
+        .take(257)
+        .map(|e| {
+            e?.file_name()
+                .into_string()
+                .map_err(|_| denied("non-UTF8 recovery entry"))
+        })
+        .collect::<io::Result<_>>()?;
+    if entries.len() > 256 {
+        return Err(denied("recovery directory bound"));
+    }
+    Ok(entries)
+}
+pub(crate) fn verify_release(slot: &OwnedFd) -> io::Result<()> {
+    if names(slot)?
+        .iter()
+        .any(|n| n != "owner.json" && n != ".owner.next")
+    {
+        return Err(denied("unknown identity directory content"));
+    }
+    for name in names(slot)? {
+        let fd = rustix::fs::openat(
+            slot,
+            name,
+            OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        )?;
+        private(&fd, 0o600)?;
+        let s = rustix::fs::fstat(&fd)?;
+        if rustix::fs::FileType::from_raw_mode(s.st_mode) != rustix::fs::FileType::RegularFile
+            || s.st_nlink != 1
+            || s.st_size < 0
+            || s.st_size as u64 > MAX_RECORD
+        {
+            return Err(denied("unknown journal replacement"));
+        }
+    }
+    Ok(())
+}
+pub(crate) fn verify_released_resources(
+    config: &ValidatedRuntimeConfig,
+    slot: &OwnedFd,
+    uid: u32,
+    gid: u32,
+) -> io::Result<()> {
+    let record = read_record(slot)?;
+    if record.owner != Owner::current()? || record.uid != uid || record.gid != gid {
+        return Err(denied("identity owner record changed"));
+    }
+    let roots = Roots::open(config)?;
+    if record.anchors != roots.anchors()? {
+        return Err(denied("identity release anchors changed"));
+    }
+    // Reserve the UID until *all* fixed resource paths are gone. A failed
+    // capture unmount/delete must not release an identity just because the
+    // process group and task root have already been removed. O_PATH also
+    // rejects a replacement symlink/special file without opening its target.
+    let name = format!("kf-{}", record.run_id);
+    for parent in [&roots.tasks, &roots.cgroup, &roots.captures] {
+        match rustix::fs::openat(
+            parent,
+            name.as_str(),
+            OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        ) {
+            Err(rustix::io::Errno::NOENT) => {}
+            Err(e) => return Err(e.into()),
+            Ok(_) => return Err(denied("identity still owns a residual resource")),
+        }
+    }
+    Ok(())
+}
+fn mount_id(fd: &OwnedFd) -> io::Result<u64> {
+    std::fs::read_to_string(format!("/proc/self/fdinfo/{}", fd.as_raw_fd()))?
+        .lines()
+        .find_map(|s| s.strip_prefix("mnt_id:"))
+        .and_then(|s| s.trim().parse().ok())
+        .ok_or_else(|| denied("mount identity missing"))
+}
+fn same_mount(parent: &OwnedFd, fd: &OwnedFd) -> io::Result<()> {
+    if mount_id(parent)? != mount_id(fd)? {
+        return Err(denied("unexpected recovery submount"));
+    }
+    Ok(())
+}
+fn resource(parent: &OwnedFd, name: &str, inode: Option<Inode>) -> io::Result<Option<OwnedFd>> {
+    let fd = match rustix::fs::openat(parent, name, DIR, Mode::empty()) {
+        Ok(fd) => fd,
+        Err(rustix::io::Errno::NOENT) => return Ok(None),
+        Err(e) => return Err(e.into()),
+    };
+    inode
+        .ok_or_else(|| denied("resource has no durable ownership binding"))?
+        .check(&fd)?;
+    private(&fd, 0o700)?;
+    same_mount(parent, &fd)?;
+    Ok(Some(fd))
+}
+fn remove_directory(parent: &OwnedFd, name: &str, held: &OwnedFd) -> io::Result<()> {
+    let named = rustix::fs::openat(parent, name, DIR, Mode::empty())?;
+    Inode::of(held)?.check(&named)?;
+    same_mount(parent, &named)?;
+    rustix::fs::unlinkat(parent, name, AtFlags::REMOVEDIR)?;
+    Ok(())
+}
+fn wait_event(group: &OwnedFd, key: &str, value: &str, deadline: Instant) -> io::Result<()> {
+    loop {
+        if crate::cgroup::read(group, "cgroup.events")?
+            .lines()
+            .any(|s| s.split_once(' ') == Some((key, value)))
+        {
+            return Ok(());
+        }
+        if Instant::now() >= deadline {
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "recovery group deadline",
+            ));
+        }
+        std::thread::sleep(Duration::from_millis(2));
+    }
+}
+fn stop(group: &OwnedFd) -> io::Result<u32> {
+    if rustix::fs::fstatfs(group)?.f_type as u64 != 0x6367_7270
+        || crate::cgroup::read(group, "cgroup.type")?.trim() != "domain"
+    {
+        return Err(denied("invalid recovery task cgroup"));
+    }
+    for name in names(group)? {
+        if rustix::fs::openat(group, name, DIR, Mode::empty()).is_ok() {
+            return Err(denied("unexpected nested task cgroup"));
+        }
+    }
+    let deadline = Instant::now() + Duration::from_secs(2);
+    crate::cgroup::write(group, "cgroup.freeze", "1")?;
+    wait_event(group, "frozen", "1", deadline)?;
+    let text = crate::cgroup::read(group, "cgroup.procs")?;
+    let mut fds = Vec::new();
+    for pid in text.split_whitespace() {
+        if fds.len() >= 128 {
+            return Err(denied("recovery process bound"));
+        }
+        let pid = pid
+            .parse::<i32>()
+            .ok()
+            .and_then(rustix::process::Pid::from_raw)
+            .ok_or_else(|| denied("invalid group member"))?;
+        match rustix::process::pidfd_open(pid, rustix::process::PidfdFlags::empty()) {
+            Ok(fd) => fds.push(fd),
+            Err(rustix::io::Errno::SRCH) => {}
+            Err(e) => return Err(e.into()),
+        }
+    }
+    crate::cgroup::write(group, "cgroup.kill", "1")?;
+    wait_event(group, "populated", "0", deadline)?;
+    if !crate::cgroup::read(group, "cgroup.procs")?
+        .trim()
+        .is_empty()
+    {
+        return Err(denied("recovery group still populated"));
+    }
+    // Startup is not the parent of the old supervisor's descendants: do not
+    // wait(-1) or signal numeric PIDs. Held pidfds prove exit; their actual
+    // parent/system init owns zombie reaping, including across service restart.
+    loop {
+        let mut poll: Vec<_> = fds
+            .iter()
+            .map(|fd| rustix::event::PollFd::new(fd, rustix::event::PollFlags::IN))
+            .collect();
+        rustix::event::poll(
+            &mut poll,
+            Some(&rustix::event::Timespec {
+                tv_sec: 0,
+                tv_nsec: 0,
+            }),
+        )?;
+        if poll.iter().all(|p| {
+            p.revents()
+                .intersects(rustix::event::PollFlags::IN | rustix::event::PollFlags::HUP)
+        }) {
+            return Ok(fds.len() as u32);
+        }
+        if Instant::now() >= deadline {
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "recovery pidfd exit deadline",
+            ));
+        }
+        std::thread::sleep(Duration::from_millis(2));
+    }
+}
+#[derive(Debug, Serialize)]
+pub struct RecoveryItem {
+    pub uid: u32,
+    pub run_id: Option<String>,
+    pub status: &'static str,
+    pub stopped_processes: u32,
+    pub cleanup_error: Option<ExecutionError>,
+}
+#[derive(Debug, Default, Serialize)]
+pub struct RecoveryReport {
+    pub items: Vec<RecoveryItem>,
+}
+fn reclaim(
+    roots: &Roots,
+    identities: &OwnedFd,
+    slot: &OwnedFd,
+    name: &str,
+    record: &Record,
+    stopped: &mut u32,
+) -> io::Result<()> {
+    if record.anchors != roots.anchors()? {
+        return Err(denied("recovery anchor binding changed"));
+    }
+    let task_name = format!("kf-{}", record.run_id);
+    let group = resource(&roots.cgroup, &task_name, record.resources.cgroup)?;
+    let task = resource(&roots.tasks, &task_name, record.resources.task)?;
+    let capture = resource(&roots.captures, &task_name, record.resources.capture)?;
+    *stopped = if let Some(group) = &group {
+        stop(group)?
+    } else {
+        0
+    };
+    if let Some(group) = &group {
+        remove_directory(&roots.cgroup, &task_name, group)?;
+    }
+    if let Some(task) = &task {
+        let entries = names(task)?;
+        if entries.iter().any(|n| n != "root") {
+            return Err(denied("unknown task directory content"));
+        }
+        if entries.iter().any(|n| n == "root") {
+            let root = rustix::fs::openat(task, "root", DIR, Mode::empty())?;
+            private(&root, 0o700)?;
+            same_mount(task, &root)?;
+            if !names(&root)?.is_empty() {
+                return Err(denied("unexpected task root contents"));
+            }
+            remove_directory(task, "root", &root)?;
+        }
+        remove_directory(&roots.tasks, &task_name, task)?;
+    }
+    if let Some(capture) = &capture {
+        if !names(capture)?.is_empty() {
+            return Err(denied("unexpected capture contents"));
+        }
+        remove_directory(&roots.captures, &task_name, capture)?;
+    }
+    verify_release(slot)?;
+    for file in [".owner.next", "owner.json"] {
+        match rustix::fs::unlinkat(slot, file, AtFlags::empty()) {
+            Ok(()) | Err(rustix::io::Errno::NOENT) => {}
+            Err(e) => return Err(e.into()),
+        }
+    }
+    remove_directory(identities, name, slot)?;
+    rustix::fs::fsync(identities)?;
+    Ok(())
+}
+/// Call in the fresh privileged service-start helper, before accepting runs.
+/// A live owner or held directory flock is skipped; failures retain the UID.
+pub fn recover(config: &ValidatedRuntimeConfig) -> io::Result<RecoveryReport> {
+    crate::launcher::entry::ensure_single_threaded()?;
+    if !rustix::process::geteuid().is_root() {
+        return Err(denied("privileged recovery helper required"));
+    }
+    let state = crate::isolation::root::open_directory(Path::new(&config.config().state_root))?;
+    let identities = match rustix::fs::openat(&state, "identities", DIR, Mode::empty()) {
+        Ok(fd) => fd,
+        Err(rustix::io::Errno::NOENT) => return Ok(RecoveryReport::default()),
+        Err(e) => return Err(e.into()),
+    };
+    private(&identities, 0o700)?;
+    let roots = Roots::open(config)?;
+    let mut report = RecoveryReport::default();
+    for index in 0..config.config().max_concurrent_runs {
+        let (uid, gid) = config
+            .task_identity(index)
+            .ok_or_else(|| denied("recovery identity mapping"))?;
+        let name = format!("uid-{uid}");
+        let slot = match rustix::fs::openat(&identities, name.as_str(), DIR, Mode::empty()) {
+            Ok(fd) => fd,
+            Err(rustix::io::Errno::NOENT) => continue,
+            Err(e) => {
+                report.items.push(RecoveryItem {
+                    uid,
+                    run_id: None,
+                    status: "unrecognized",
+                    stopped_processes: 0,
+                    cleanup_error: Some(crate::launcher::blocked::failure(
+                        ExecutionStage::Cleanup,
+                        ErrorCode::CleanupIncomplete,
+                        Some(&e.into()),
+                    )),
+                });
+                continue;
+            }
+        };
+        let mut item = RecoveryItem {
+            uid,
+            run_id: None,
+            status: "unrecognized",
+            stopped_processes: 0,
+            cleanup_error: None,
+        };
+        let result = (|| {
+            private(&slot, 0o700)?;
+            match rustix::fs::flock(&slot, FlockOperation::NonBlockingLockExclusive) {
+                Ok(()) => {}
+                Err(rustix::io::Errno::WOULDBLOCK) => {
+                    item.status = "active";
+                    return Ok(());
+                }
+                Err(e) => return Err(e.into()),
+            }
+            let Ok(record) = read_record(&slot) else {
+                return Ok(());
+            };
+            item.run_id = Some(record.run_id.clone());
+            if record.uid != uid || record.gid != gid {
+                return Err(denied("recorded identity mapping changed"));
+            }
+            if record.owner.alive()? {
+                item.status = "active";
+                return Ok(());
+            }
+            reclaim(
+                &roots,
+                &identities,
+                &slot,
+                &name,
+                &record,
+                &mut item.stopped_processes,
+            )?;
+            item.status = "reclaimed";
+            Ok::<_, io::Error>(())
+        })();
+        if let Err(e) = result {
+            item.status = "cleanup_failed";
+            item.cleanup_error = Some(crate::launcher::blocked::failure(
+                ExecutionStage::Cleanup,
+                ErrorCode::CleanupIncomplete,
+                Some(&e),
+            ));
+        }
+        report.items.push(item);
+    }
+    Ok(report)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn owner_start_time_parses_parenthesized_comm_and_rejects_truncation() {
+        let tail = (0..20)
+            .map(|n| if n == 19 { "12345".into() } else { "1".into() })
+            .collect::<Vec<String>>()
+            .join(" ");
+        assert_eq!(
+            start_ticks(&format!("7 (name ) has spaces) {tail}")).unwrap(),
+            12345
+        );
+        for bad in ["", "7 (name) S 1", "7 name S 1"] {
+            assert!(start_ticks(bad).is_err());
+        }
+    }
+}

judge/crates/sandbox/src/isolation/root.rs

@@ -180,6 +180,7 @@ impl RootLease {
         };
         if let Err(source) = (|| {
             safe_directory(&lease.task)?;
+            crate::cleanup::register(config, spec, crate::cleanup::Resource::Task, &lease.task)?;
             rustix::fs::mkdirat(&lease.task, "root", Mode::from_raw_mode(0o700))?;
             lease.root_created = true;
             Ok::<_, io::Error>(())

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

@@ -75,6 +75,9 @@ impl CaptureLease {
             removed: false,
         };
         let mount = (|| {
+            let bare = rustix::fs::openat(&lease.parent, lease.name.as_str(), DIR, Mode::empty())?;
+            crate::cleanup::register(config, spec, crate::cleanup::Resource::Capture, &bare)?;
+            drop(bare);
             let data = std::ffi::CString::new(format!("size={size},nr_inodes=16,mode=0700"))?;
             rustix::mount::mount(
                 "tmpfs",

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

@@ -8,7 +8,7 @@ use crate::runtime::ValidatedRuntimeConfig;
 use crate::spec::ValidatedSpec;
 use rustix::fd::OwnedFd;
 use rustix::fs::{AtFlags, Mode, OFlags};
-use std::{io, io::Write, path::Path};
+use std::{io, path::Path};
 
 const DIR: OFlags = OFlags::RDONLY
     .union(OFlags::DIRECTORY)
@@ -26,12 +26,14 @@ fn private(fd: &OwnedFd) -> io::Result<()> {
 }
 
 struct IdentityLease {
+    config: ValidatedRuntimeConfig,
     parent: OwnedFd,
     slot: OwnedFd,
     name: String,
     uid: u32,
     gid: u32,
     removed: bool,
+    initialized: bool,
 }
 impl IdentityLease {
     fn acquire(config: &ValidatedRuntimeConfig, spec: &ValidatedSpec) -> Result<Self, RootError> {
@@ -71,33 +73,20 @@ impl IdentityLease {
                 }
             };
             let mut lease = Self {
+                config: config.clone(),
                 parent,
                 slot,
                 name,
                 uid,
                 gid,
                 removed: false,
+                initialized: false,
             };
             let setup = (|| {
                 rustix::fs::fchmod(&lease.slot, Mode::from_raw_mode(0o700))?;
                 private(&lease.slot)?;
-                let marker = rustix::fs::openat(
-                    &lease.slot,
-                    "owner.json",
-                    OFlags::WRONLY
-                        | OFlags::CREATE
-                        | OFlags::EXCL
-                        | OFlags::CLOEXEC
-                        | OFlags::NOFOLLOW,
-                    Mode::from_raw_mode(0o600),
-                )?;
-                let mut marker = std::fs::File::from(marker);
-                serde_json::to_writer(
-                    &mut marker,
-                    &serde_json::json!({"version":1,"run_id":spec.spec().run_id,"uid":uid,"gid":gid}),
-                )?;
-                marker.write_all(b"\n")?;
-                marker.sync_all()?;
+                crate::cleanup::initialize_slot(config, spec, &lease.slot, uid, gid)?;
+                lease.initialized = true;
                 rustix::fs::fsync(&lease.slot)?;
                 rustix::fs::fsync(&lease.parent)?;
                 Ok::<_, io::Error>(())
@@ -124,6 +113,19 @@ impl IdentityLease {
         if (a.st_dev, a.st_ino) != (b.st_dev, b.st_ino) {
             return Err(denied("identity marker changed"));
         }
+        crate::cleanup::verify_release(&self.slot)?;
+        if self.initialized {
+            crate::cleanup::verify_released_resources(
+                &self.config,
+                &self.slot,
+                self.uid,
+                self.gid,
+            )?;
+        }
+        match rustix::fs::unlinkat(&self.slot, ".owner.next", AtFlags::empty()) {
+            Ok(()) | Err(rustix::io::Errno::NOENT) => {}
+            Err(e) => return Err(e.into()),
+        }
         match rustix::fs::unlinkat(&self.slot, "owner.json", AtFlags::empty()) {
             Ok(()) | Err(rustix::io::Errno::NOENT) => {}
             Err(e) => return Err(e.into()),
@@ -179,7 +181,9 @@ impl IdentityReducedTarget {
                 return Err(e);
             }
         };
-        if let Err(e) = target.target.root.initialize_rlimits() {
+        if let Err(e) = crate::cleanup::verify_owned(config, spec, false)
+            .and_then(|()| target.target.root.initialize_rlimits())
+        {
             let mut e = target
                 .target
                 .root

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

@@ -57,13 +57,13 @@ pub(crate) fn cancel_stage(
             } else {
                 drop(stdio);
             }
+            drop(pump);
+            lease.cleanup()?;
+            lease.cleanup()?;
             let result = t.refuse_at(crate::outcome::ExecutionStage::Exec, None);
             if result.cleanup_failure.is_some() {
                 return Err(fail(result));
             }
-            drop(pump);
-            lease.cleanup()?;
-            lease.cleanup()?;
         }
         _ => return Err(io::Error::other("fixed preparation stage")),
     }

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

@@ -44,6 +44,7 @@ pub(crate) enum MonitorMode {
     Normal,
     CancelAfter(u64),
     FailAfter(u64),
+    RestartReady,
 }
 fn capture(status: CaptureStatus, id: &str) -> OutputCapture {
     OutputCapture {
@@ -103,6 +104,7 @@ pub(crate) fn execute(
         let work = target.root().pin_work()?;
         let (lease, dir) = CaptureLease::prepare(config, spec)?;
         let (pump, stdio) = PipePump::prepare(target.take_stdin(), spec, dir)?;
+        crate::cleanup::verify_owned(config, spec, true)?;
         target.root_mut().target.prepare_execution(spec, stdio)?;
         Ok::<_, io::Error>((work, lease, pump))
     })();
@@ -159,6 +161,19 @@ pub(crate) fn execute(
                 None
             }
         };
+        // Fixed VM crash probe only: retain a real supervisor-owned residual
+        // child in the same marked cgroup to exercise post-crash startup kill.
+        let _restart_child = if matches!(mode, MonitorMode::RestartReady) && main_fd.is_some() {
+            match crate::acceptance::restart::ready(config, spec, "live") {
+                Ok(child) => Some(child),
+                Err(e) => {
+                    errors.push(error(ExecutionStage::Exec, ErrorCode::OperationFailed, &e));
+                    None
+                }
+            }
+        } else {
+            None
+        };
         let mut monitoring_failed = !errors.is_empty();
         if !monitoring_failed {
             loop {

judge/crates/sandbox/src/lib.rs

@@ -21,6 +21,8 @@ pub mod io;
 #[cfg(target_os = "linux")]
 pub mod accounting;
 #[cfg(target_os = "linux")]
+pub mod cleanup;
+#[cfg(target_os = "linux")]
 pub mod supervisor;
 
 #[cfg(all(target_os = "linux", feature = "vm-acceptance"))]

judge/crates/sandbox/src/runtime.rs

@@ -77,7 +77,7 @@ fn invalid(field: &'static str, code: &'static str) -> ConfigError {
 }
 
 /// Immutable structural policy; deliberately not an execution permit.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct ValidatedRuntimeConfig(RuntimeConfig);
 
 impl ValidatedRuntimeConfig {
@@ -154,6 +154,12 @@ impl RuntimeConfig {
         }) {
             return Err(invalid("roots", "identity_lease_overlap"));
         }
+        let capture_root = format!("{}/captures", self.state_root);
+        if [&self.tasks_root, &self.image_root].iter().any(|root| {
+            *root == &capture_root || inside(root, &capture_root) || inside(&capture_root, root)
+        }) {
+            return Err(invalid("roots", "capture_lease_overlap"));
+        }
         if !inside(&self.delegated_cgroup_root, "/sys/fs/cgroup") {
             return Err(invalid("delegated_cgroup_root", "not_delegated_subtree"));
         }

judge/crates/sandbox/src/spec.rs

@@ -208,7 +208,7 @@ fn error(field: &'static str, code: ValidationCode) -> SpecError {
     SpecError { field, code }
 }
 
-fn identifier(value: &str, field: &'static str) -> Result<(), SpecError> {
+pub(crate) fn identifier(value: &str, field: &'static str) -> Result<(), SpecError> {
     if value.is_empty()
         || value.len() > 64
         || !value

judge/crates/sandbox/tests/restart_vm.rs

@@ -0,0 +1,387 @@
+#![cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+#[allow(dead_code)]
+mod support;
+use rustix::{
+    fd::{AsFd, OwnedFd},
+    process::{WaitId, WaitIdOptions},
+};
+use std::{
+    fs,
+    os::unix::{
+        fs::{PermissionsExt, symlink},
+        process::ExitStatusExt,
+    },
+    process::{Child, Command, Stdio},
+    time::{Duration, Instant},
+};
+const LEASE: &str = "/var/lib/kf-judge/identities/uid-100000/owner.json";
+fn command(case: &str) -> Command {
+    let mut c = Command::new(support::binary());
+    c.args(["vm-case", case])
+        .env_clear()
+        .env("KF_SANDBOX_INDEPENDENT_VM", "1");
+    c
+}
+fn recover() -> serde_json::Value {
+    let o = command("restart-recover").output().unwrap();
+    assert!(
+        o.status.success(),
+        "{} {}",
+        String::from_utf8_lossy(&o.stdout),
+        String::from_utf8_lossy(&o.stderr)
+    );
+    let v: serde_json::Value = serde_json::from_slice(&o.stdout).unwrap();
+    println!("KF_RESTART_RECOVERY_PROOF {v}");
+    v
+}
+fn record() -> serde_json::Value {
+    serde_json::from_slice(&fs::read(LEASE).unwrap()).unwrap()
+}
+fn store(v: &serde_json::Value) {
+    fs::write(LEASE, serde_json::to_vec(v).unwrap()).unwrap();
+}
+fn empty() {
+    for dir in ["tasks", "identities", "captures"] {
+        assert_eq!(
+            fs::read_dir(format!("/var/lib/kf-judge/{dir}"))
+                .unwrap()
+                .count(),
+            0
+        );
+    }
+}
+struct Owner {
+    child: Option<Child>,
+    cg: String,
+    publications: String,
+    run_id: String,
+    members: Vec<OwnedFd>,
+    previous: Option<rustix::process::Pid>,
+    residual: u32,
+}
+impl Owner {
+    fn start(stage: &str) -> Self {
+        support::require_vm();
+        assert!(support::children().is_empty());
+        let previous = rustix::process::child_subreaper().unwrap();
+        rustix::process::set_child_subreaper(rustix::process::Pid::from_raw(1)).unwrap();
+        let _ = fs::remove_file("/kf/restart-ready");
+        let mut child = command(&format!("restart-{stage}"))
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()
+            .unwrap();
+        let owner = child.id();
+        let cg = format!("/sys/fs/cgroup/kf-acceptance-{owner}");
+        let deadline = Instant::now() + Duration::from_secs(8);
+        let ready: serde_json::Value = loop {
+            assert!(
+                child.try_wait().unwrap().is_none(),
+                "supervisor exited before ready"
+            );
+            assert!(
+                Instant::now() < deadline,
+                "crash fixture readiness deadline"
+            );
+            if let Ok(bytes) = fs::read("/kf/restart-ready") {
+                break serde_json::from_slice(&bytes).unwrap();
+            }
+            std::thread::sleep(Duration::from_millis(5));
+        };
+        assert_eq!(ready["owner_pid"], owner);
+        assert_eq!(ready["stage"], stage);
+        let run_id = ready["run_id"].as_str().unwrap().to_owned();
+        if stage == "live" {
+            loop {
+                let pids = fs::read_to_string(format!("{cg}/kf-{run_id}/cgroup.procs")).unwrap();
+                if pids.split_whitespace().any(|p| {
+                    fs::read(format!("/proc/{p}/root/work/peer-ready")).is_ok_and(|s| s == b"ready")
+                }) {
+                    break;
+                }
+                assert!(Instant::now() < deadline);
+                std::thread::sleep(Duration::from_millis(5));
+            }
+        }
+        let members = fs::read_to_string(format!("{cg}/kf-{run_id}/cgroup.procs"))
+            .unwrap()
+            .split_whitespace()
+            .map(|n| {
+                rustix::process::pidfd_open(
+                    rustix::process::Pid::from_raw(n.parse().unwrap()).unwrap(),
+                    rustix::process::PidfdFlags::empty(),
+                )
+                .unwrap()
+            })
+            .collect();
+        let r = record();
+        assert_eq!(r["owner"]["pid"], owner);
+        assert_eq!(r["version"], 2);
+        assert_eq!(r["service"], "kf-sandbox-task-v2");
+        for key in ["task", "cgroup", "capture"] {
+            assert!(r["resources"][key]["ino"].as_u64().unwrap() > 0);
+        }
+        let mut cfg: kf_sandbox::runtime::RuntimeConfig = serde_json::from_str(include_str!(
+            "../../../../deploy/sandbox.runtime.example.json"
+        ))
+        .unwrap();
+        cfg.delegated_cgroup_root = cg.clone();
+        fs::write(
+            "/kf/restart-runtime.json",
+            serde_json::to_vec(&cfg).unwrap(),
+        )
+        .unwrap();
+        fs::set_permissions(
+            "/kf/restart-runtime.json",
+            fs::Permissions::from_mode(0o600),
+        )
+        .unwrap();
+        Self {
+            child: Some(child),
+            cg,
+            publications: format!("/kf/execution-publications-{owner}"),
+            run_id,
+            members,
+            previous,
+            residual: ready["residual_pid"].as_u64().unwrap() as u32,
+        }
+    }
+    fn crash(&mut self) {
+        self.child.as_mut().unwrap().kill().unwrap();
+        assert_eq!(self.child.take().unwrap().wait().unwrap().signal(), Some(9));
+        assert!(
+            fs::read_to_string(format!("/proc/{}/stat", self.residual)).is_ok(),
+            "real partial-preparation residual must survive owner SIGKILL"
+        );
+    }
+    fn group(&self) -> String {
+        format!("{}/kf-{}", self.cg, self.run_id)
+    }
+    fn task(&self) -> String {
+        format!("/var/lib/kf-judge/tasks/kf-{}", self.run_id)
+    }
+    fn reap(&self) {
+        let deadline = Instant::now() + Duration::from_secs(3);
+        for fd in &self.members {
+            loop {
+                match rustix::process::waitid(
+                    WaitId::PidFd(fd.as_fd()),
+                    WaitIdOptions::EXITED | WaitIdOptions::NOHANG,
+                ) {
+                    Ok(Some(_)) | Err(rustix::io::Errno::CHILD) => break,
+                    Ok(None) => {
+                        assert!(Instant::now() < deadline);
+                        std::thread::sleep(Duration::from_millis(5));
+                    }
+                    Err(e) => panic!("owned pidfd wait: {e}"),
+                }
+            }
+            let mut p = [rustix::event::PollFd::new(fd, rustix::event::PollFlags::IN)];
+            rustix::event::poll(
+                &mut p,
+                Some(&rustix::event::Timespec {
+                    tv_sec: 0,
+                    tv_nsec: 0,
+                }),
+            )
+            .unwrap();
+            assert!(
+                p[0].revents()
+                    .intersects(rustix::event::PollFlags::IN | rustix::event::PollFlags::HUP)
+            );
+        }
+        assert!(
+            support::children().is_empty(),
+            "all owned adopted children reaped"
+        );
+    }
+    fn finish(&self) {
+        self.reap();
+        assert!(!std::path::Path::new(&self.group()).exists());
+        empty();
+        assert_eq!(recover()["items"], serde_json::json!([]));
+        println!(
+            "KF_RESTART_KERNEL_PROOF {}",
+            serde_json::json!({"run_id":self.run_id,"residual_pid":self.residual,"owner_killed_sigkill":true,"kernel_pidfd_exits_verified":self.members.len(),"only_owned_adopted_children_reaped":true,"resources_removed":true,"repeat_cleanup_empty":true})
+        );
+    }
+}
+impl Drop for Owner {
+    fn drop(&mut self) {
+        if let Some(child) = self.child.as_mut() {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+        let _ = fs::write(format!("{}/cgroup.kill", self.cg), "1");
+        let deadline = Instant::now() + Duration::from_secs(3);
+        while Instant::now() < deadline {
+            match rustix::process::wait(rustix::process::WaitOptions::NOHANG) {
+                Ok(Some(_)) => {}
+                Err(rustix::io::Errno::CHILD) => break,
+                _ => std::thread::sleep(Duration::from_millis(5)),
+            }
+        }
+        let _ = fs::remove_dir(self.group());
+        let _ = fs::remove_dir(&self.cg);
+        // Fixture input publications belong to the trusted caller, not to
+        // sandbox recovery. Remove only this known test-owned publication.
+        let _ = fs::remove_dir_all(&self.publications);
+        let _ = fs::remove_file("/kf/restart-ready");
+        let _ = fs::remove_file("/kf/restart-runtime.json");
+        let _ = rustix::process::set_child_subreaper(self.previous);
+    }
+}
+#[test]
+#[ignore = "independent capped VM: real SIGKILL and service restart before/after user exec with surviving residual"]
+fn actual_owner_death_reclaims_prepared_and_live_tasks() {
+    for stage in ["prepared", "live"] {
+        let mut owner = Owner::start(stage);
+        assert_eq!(recover()["items"][0]["status"], "active");
+        assert!(owner.child.as_mut().unwrap().try_wait().unwrap().is_none());
+        owner.crash();
+        let v = recover();
+        assert_eq!(v["items"][0]["status"], "reclaimed", "{v}");
+        assert!(v["items"][0]["stopped_processes"].as_u64().unwrap() > 0);
+        owner.finish();
+        drop(owner);
+        empty();
+    }
+}
+#[test]
+#[ignore = "independent capped VM: failed resource deletion must preserve separate error, UID and foreign bytes"]
+fn cleanup_failure_is_independent_and_reserved_until_explicit_repair() {
+    let mut owner = Owner::start("live");
+    fs::write(
+        format!("{}/foreign", owner.task()),
+        b"preserve this foreign content",
+    )
+    .unwrap();
+    owner.crash();
+    let v = recover();
+    assert_eq!(v["items"][0]["status"], "cleanup_failed");
+    assert_eq!(v["items"][0]["cleanup_error"]["stage"], "cleanup");
+    assert_eq!(v["items"][0]["cleanup_error"]["code"], "cleanup_incomplete");
+    assert_eq!(
+        fs::read(format!("{}/foreign", owner.task())).unwrap(),
+        b"preserve this foreign content"
+    );
+    assert!(std::path::Path::new(LEASE).exists());
+    assert!(!std::path::Path::new(&owner.group()).exists());
+    owner.reap();
+    assert_eq!(recover()["items"][0]["status"], "cleanup_failed");
+    fs::remove_file(format!("{}/foreign", owner.task())).unwrap();
+    assert_eq!(recover()["items"][0]["status"], "reclaimed");
+    owner.finish();
+}
+#[test]
+#[ignore = "independent capped VM: replaced cgroup inode, symlink, boot/PID identity and unknown markers stay protected"]
+fn replaced_resources_and_unrecognized_markers_are_not_adopted() {
+    let mut owner = Owner::start("prepared");
+    owner.crash();
+    let saved = record();
+    // A reused owner PID/start pair is live even without a surviving flock.
+    let stat = fs::read_to_string("/proc/self/stat").unwrap();
+    let ticks = stat
+        .rsplit_once(") ")
+        .unwrap()
+        .1
+        .split_whitespace()
+        .nth(19)
+        .unwrap()
+        .parse::<u64>()
+        .unwrap();
+    let mut v = saved.clone();
+    v["owner"]["pid"] = std::process::id().into();
+    v["owner"]["start_ticks"] = ticks.into();
+    store(&v);
+    assert_eq!(recover()["items"][0]["status"], "active");
+    store(&saved);
+    v = saved.clone();
+    v["owner"]["boot_id"] = "00000000-0000-0000-0000-000000000000".into();
+    store(&v);
+    assert_eq!(recover()["items"][0]["status"], "cleanup_failed");
+    store(&saved);
+    // Startup must inspect a foreign FIFO without reading or blocking on it.
+    let held_record = format!("{LEASE}.held");
+    fs::rename(LEASE, &held_record).unwrap();
+    rustix::fs::mknodat(
+        rustix::fs::CWD,
+        LEASE,
+        rustix::fs::FileType::Fifo,
+        rustix::fs::Mode::from_raw_mode(0o600),
+        0,
+    )
+    .unwrap();
+    let before = Instant::now();
+    assert_eq!(recover()["items"][0]["status"], "unrecognized");
+    assert!(before.elapsed() < Duration::from_secs(2));
+    fs::remove_file(LEASE).unwrap();
+    fs::rename(held_record, LEASE).unwrap();
+    v = saved.clone();
+    v["resources"]["cgroup"] = serde_json::Value::Null;
+    store(&v);
+    assert_eq!(recover()["items"][0]["status"], "cleanup_failed");
+    store(&saved);
+    // cgroup v2 deliberately forbids rename. The VM operator first stops and
+    // removes its known original fixture, then creates a genuinely different
+    // cgroup at the same name while retaining the old recorded inode.
+    fs::write(format!("{}/cgroup.kill", owner.group()), "1").unwrap();
+    owner.reap();
+    fs::remove_dir(owner.group()).unwrap();
+    fs::create_dir(owner.group()).unwrap();
+    fs::set_permissions(owner.group(), fs::Permissions::from_mode(0o700)).unwrap();
+    let mut foreign = command("restart-residual")
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .unwrap();
+    fs::write(
+        format!("{}/cgroup.procs", owner.group()),
+        foreign.id().to_string(),
+    )
+    .unwrap();
+    assert_eq!(recover()["items"][0]["status"], "cleanup_failed");
+    assert!(foreign.try_wait().unwrap().is_none());
+    foreign.kill().unwrap();
+    foreign.wait().unwrap();
+    fs::remove_dir(owner.group()).unwrap();
+    let old_task = format!("{}-held", owner.task());
+    fs::rename(owner.task(), &old_task).unwrap();
+    fs::create_dir("/kf/protected-restart-directory").unwrap();
+    fs::write(
+        "/kf/protected-restart-directory/foreign",
+        b"do not traverse",
+    )
+    .unwrap();
+    symlink("/kf/protected-restart-directory", owner.task()).unwrap();
+    assert_eq!(recover()["items"][0]["status"], "cleanup_failed");
+    assert_eq!(
+        fs::read("/kf/protected-restart-directory/foreign").unwrap(),
+        b"do not traverse"
+    );
+    fs::remove_file(owner.task()).unwrap();
+    fs::rename(old_task, owner.task()).unwrap();
+    fs::remove_file("/kf/protected-restart-directory/foreign").unwrap();
+    fs::remove_dir("/kf/protected-restart-directory").unwrap();
+    let unknown = "/var/lib/kf-judge/identities/uid-100001";
+    fs::create_dir(unknown).unwrap();
+    fs::set_permissions(unknown, fs::Permissions::from_mode(0o700)).unwrap();
+    let legacy = b"{\"version\":1,\"run_id\":\"foreign\",\"uid\":100001,\"gid\":100001}";
+    fs::write(format!("{unknown}/owner.json"), legacy).unwrap();
+    fs::set_permissions(
+        format!("{unknown}/owner.json"),
+        fs::Permissions::from_mode(0o600),
+    )
+    .unwrap();
+    let v = recover();
+    assert_eq!(v["items"][0]["status"], "reclaimed");
+    assert_eq!(v["items"][1]["status"], "unrecognized");
+    assert_eq!(fs::read(format!("{unknown}/owner.json")).unwrap(), legacy);
+    fs::remove_file(format!("{unknown}/owner.json")).unwrap();
+    fs::remove_dir(unknown).unwrap();
+    owner.finish();
+    println!(
+        "KF_RESTART_OWNERSHIP_PROOF live_owner_pid_start=1 boot_mismatch_refused=1 unrecorded_resource_refused=1 fifo_not_opened=1 replaced_cgroup_and_process_untouched=1 symlink_not_followed=1 legacy_marker_untouched=1"
+    );
+}

judge/crates/sandbox/tests/runtime_policy.rs

@@ -318,3 +318,21 @@ fn configuration_cannot_enable_execution_or_bypass_required_fields() {
     invalid.as_object_mut().unwrap().remove("aggregate");
     assert!(serde_json::from_value::<RuntimeConfig>(invalid).is_err());
 }
+
+#[test]
+fn task_and_image_roots_cannot_overlap_private_capture_storage() {
+    for field in ["tasks", "image"] {
+        for path in [
+            "/var/lib/kf-judge/captures",
+            "/var/lib/kf-judge/captures/nested",
+        ] {
+            let mut c = config();
+            if field == "tasks" {
+                c.tasks_root = path.into();
+            } else {
+                c.image_root = path.into();
+            }
+            assert_eq!(c.validate().unwrap_err().code, "capture_lease_overlap");
+        }
+    }
+}