diff --git a/api/crates/layerfs-api/src/machine/stop.rs b/api/crates/layerfs-api/src/machine/stop.rs
index 90a11099df..0c0dedc807 100644
--- a/api/crates/layerfs-api/src/machine/stop.rs
+++ b/api/crates/layerfs-api/src/machine/stop.rs
@@ -1,6 +1,6 @@
 //! `machine.stop`: stop a machine's worker and VM; its disks stay.
-use crate::calls::table::Kind::Name;
-use crate::calls::table::{a, call, Call};
+use crate::calls::table::Kind::{Bool, Name};
+use crate::calls::table::{a, call, opt, Call};
 use crate::calls::Context;
 use layerfs_bridge::control::SdkValue;
 #[cfg(feature = "machine")]
@@ -12,7 +12,7 @@ pub(crate) const CALL: Call = call(
     "machine",
     "stop",
     &[a("name", Name)],
-    &[],
+    &[opt("force", Bool)],
     "stop a machine's worker and VM; its disks stay",
     run,
 );
@@ -21,7 +21,11 @@ pub(crate) const CALL: Call = call(
 #[cfg(feature = "machine")]
 fn run(context: &mut Context<'_>, args: &SdkValue) -> Outcome {
     let name = MachineName::parse(args.get_text("name").unwrap_or(""))?;
-    layerfs_control::machine::stop(&context.home, &name)
+    if args.get_bool("force") {
+        layerfs_control::machine::stop_force(&context.home, &name)
+    } else {
+        layerfs_control::machine::stop(&context.home, &name)
+    }
 }
 
 /// Without the `machine` feature (the guest worker's build) it is
diff --git a/api/crates/layerfs-api/tests/machine_force.rs b/api/crates/layerfs-api/tests/machine_force.rs
new file mode 100644
index 0000000000..c20e077afb
--- /dev/null
+++ b/api/crates/layerfs-api/tests/machine_force.rs
@@ -0,0 +1,265 @@
+//! Provider-direct force stop preserves storage and avoids a dead worker entirely.
+#![cfg(feature = "machine")]
+use layerfs_api::home::{self, paths, read_record, write_record};
+use layerfs_bridge::control::SdkValue;
+use std::{net::TcpListener, os::unix::fs::PermissionsExt, path::Path};
+fn t(value: &str) -> SdkValue {
+    SdkValue::text(value)
+}
+fn call(
+    home: &Path,
+    name: &str,
+    args: &[(&str, SdkValue)],
+) -> Result<SdkValue, layerfs_bridge::control::SdkRefusal> {
+    layerfs_api::call(
+        Some(home),
+        name,
+        SdkValue::Map(
+            args.iter()
+                .map(|(name, value)| ((*name).to_owned(), value.clone()))
+                .collect(),
+        ),
+        &mut |_| {},
+    )
+}
+#[test]
+fn force_stop_bypasses_in_use_and_dead_worker_preserving_disk_store_and_records() {
+    let root = std::env::temp_dir().join(format!("lfs-force-{}", std::process::id()));
+    std::fs::create_dir_all(&root).unwrap();
+    let home = root.join("home");
+    std::fs::create_dir_all(&home).unwrap();
+    std::fs::set_permissions(&home, std::fs::Permissions::from_mode(0o700)).unwrap();
+    home::open(&home).unwrap();
+    let tool = root.join("limactl");
+    std::fs::write(
+        &tool,
+        r#"#!/bin/sh
+d=${0%/*}
+echo "$*" >> "$d/calls"
+case "$1" in
+ list) printf 'Running\t/nonexistent\n';;
+ stop) test "$2" = --force || exit 19; test ! -f "$d/refuse" || exit 18;;
+ *) exit 17;;
+esac
+"#,
+    )
+    .unwrap();
+    std::fs::set_permissions(&tool, std::fs::Permissions::from_mode(0o700)).unwrap();
+    std::env::set_var("PATH", &root);
+    write_record(
+        &paths::machine_record(&home, "vm"),
+        &SdkValue::map([
+            ("name", t("vm")),
+            ("platform", t("lima")),
+            ("state", t("running")),
+        ]),
+    )
+    .unwrap();
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    listener.set_nonblocking(true).unwrap();
+    let worker = SdkValue::map([
+        ("address", t(&listener.local_addr().unwrap().to_string())),
+        ("secret", t("aa")),
+    ]);
+    write_record(&paths::worker_record(&home, "vm"), &worker).unwrap();
+    for (sandbox, machine) in [("one", "vm"), ("two", "vm")] {
+        write_record(
+            &paths::sandbox_record(&home, sandbox),
+            &SdkValue::map([
+                ("name", t(sandbox)),
+                ("id", t(sandbox)),
+                ("machine", t(machine)),
+                ("project", t("shop/app")),
+            ]),
+        )
+        .unwrap();
+        write_record(
+            &paths::workspace_dir(&home, sandbox).join("work.json"),
+            &SdkValue::map([("name", t("work")), ("path", t("/workspaces/one"))]),
+        )
+        .unwrap();
+    }
+    let (store, disk) = (paths::store_file(&home, "shop"), root.join("disk.img"));
+    write_record(
+        &paths::store_record(&home, "shop"),
+        &SdkValue::map([
+            ("name", t("shop")),
+            ("kind", t("sqlite")),
+            ("machine", t("vm")),
+            ("profile", t("disposable")),
+        ]),
+    )
+    .unwrap();
+    let store_record = std::fs::read(paths::store_record(&home, "shop")).unwrap();
+    std::fs::write(&store, b"saved-store").unwrap();
+    std::fs::write(&disk, b"saved-disk").unwrap();
+    let denied = call(&home, "machine.stop", &[("name", t("vm"))]).unwrap_err();
+    assert_eq!(denied.code, "InUse");
+    let stopped = call(
+        &home,
+        "machine.stop",
+        &[("name", t("vm")), ("force", SdkValue::Bool(true))],
+    )
+    .unwrap();
+    assert_eq!(stopped.get_text("state"), Some("stopped"));
+    assert!(stopped.get_bool("forced"));
+    assert_eq!(
+        stopped.get("stopped_sandboxes"),
+        Some(&SdkValue::List(vec![t("one"), t("two")]))
+    );
+    assert_eq!(
+        stopped.get("unknown"),
+        Some(&SdkValue::List(vec![t("one:work"), t("two:work")]))
+    );
+    assert!(
+        matches!(listener.accept(),Err(error) if error.kind()==std::io::ErrorKind::WouldBlock),
+        "worker must never be contacted"
+    );
+    assert!(!paths::worker_record(&home, "vm").exists());
+    assert!(paths::machine_record(&home, "vm").exists());
+    let sandbox = call(&home, "sandbox.get", &[("name", t("one"))]).unwrap();
+    assert_eq!(sandbox.get_text("state"), Some("gone"));
+    let logs = std::fs::read_to_string(root.join("calls")).unwrap();
+    assert_eq!(
+        logs.lines()
+            .filter(|line| line.starts_with("stop --force"))
+            .count(),
+        1
+    );
+    assert!(!logs.contains("start") && !logs.contains("delete"));
+    assert_eq!(std::fs::read(&store).unwrap(), b"saved-store");
+    assert_eq!(
+        std::fs::read(paths::store_record(&home, "shop")).unwrap(),
+        store_record
+    );
+    assert_eq!(std::fs::read(&disk).unwrap(), b"saved-disk");
+    assert!(read_record(&paths::machine_record(&home, "vm"))
+        .unwrap()
+        .unwrap()
+        .get("op")
+        .is_none());
+    let again = call(
+        &home,
+        "machine.stop",
+        &[("name", t("vm")), ("force", SdkValue::Bool(true))],
+    )
+    .unwrap();
+    assert_eq!(again.get_text("state"), Some("stopped"));
+    // A failed later provider operation retains worker/state and explicit unresolved custody.
+    write_record(&paths::worker_record(&home, "vm"), &worker).unwrap();
+    std::fs::write(root.join("refuse"), b"fail once").unwrap();
+    let before = std::fs::read(paths::worker_record(&home, "vm")).unwrap();
+    let failed = call(
+        &home,
+        "machine.stop",
+        &[("name", t("vm")), ("force", SdkValue::Bool(true))],
+    )
+    .unwrap_err();
+    assert_eq!(failed.code, "Failed");
+    assert_eq!(
+        std::fs::read(paths::worker_record(&home, "vm")).unwrap(),
+        before
+    );
+    let uncertain = call(&home, "sandbox.list", &[]).unwrap();
+    assert!(format!("{uncertain:?}").contains("unresolved"));
+    let machine = read_record(&paths::machine_record(&home, "vm"))
+        .unwrap()
+        .unwrap();
+    assert!(machine.get("op").unwrap().get_bool("uncertain"));
+    // Only a fresh explicit request reconciles the uncertain stop; no internal retry occurred.
+    std::fs::remove_file(root.join("refuse")).unwrap();
+    let recovered = call(
+        &home,
+        "machine.stop",
+        &[("name", t("vm")), ("force", SdkValue::Bool(true))],
+    )
+    .unwrap();
+    assert_eq!(recovered.get_text("state"), Some("stopped"));
+    assert_eq!(
+        call(&home, "sandbox.get", &[("name", t("one"))])
+            .unwrap()
+            .get_text("state"),
+        Some("gone")
+    );
+    let native = call(
+        &home,
+        "machine.stop",
+        &[("name", t("native")), ("force", SdkValue::Bool(true))],
+    )
+    .unwrap();
+    assert_eq!(native.get_text("state"), Some("ready"));
+    std::fs::remove_dir_all(root).unwrap();
+}
+
+#[test]
+fn native_force_stop_refused_before_any_signal_leaves_no_uncertain_op() {
+    let root = std::env::temp_dir().join(format!("lfs-force-pre-{}", std::process::id()));
+    std::fs::create_dir_all(&root).unwrap();
+    let home = root.join("home");
+    std::fs::create_dir_all(&home).unwrap();
+    std::fs::set_permissions(&home, std::fs::Permissions::from_mode(0o700)).unwrap();
+    home::open(&home).unwrap();
+    // a-lost: started, but its runtime record is missing, so its stop is
+    // refused before anything is signalled and nothing about it is uncertain.
+    // b-done: already stopped with its machine; the stop goes on to it.
+    // c-new: never started, so nothing is stopped or recorded for it.
+    let sandboxes = [
+        ("a-lost", Some("a-lost"), false),
+        ("b-done", Some("b-done"), true),
+        ("c-new", None, false),
+    ];
+    for (name, id, stopped) in sandboxes {
+        let mut fields = vec![
+            ("name", t(name)),
+            ("machine", t("native")),
+            ("project", t("shop/app")),
+        ];
+        if let Some(id) = id {
+            fields.push(("id", t(id)));
+        }
+        if stopped {
+            fields.push(("machine_stopped", SdkValue::Bool(true)));
+        }
+        let record = SdkValue::Map(
+            fields
+                .into_iter()
+                .map(|(field, value)| (field.to_owned(), value))
+                .collect(),
+        );
+        write_record(&paths::sandbox_record(&home, name), &record).unwrap();
+    }
+    let refused = call(
+        &home,
+        "machine.stop",
+        &[("name", t("native")), ("force", SdkValue::Bool(true))],
+    )
+    .unwrap_err();
+    assert_eq!(refused.code, "Unavailable", "{refused:?}");
+    assert_eq!(
+        refused.detail.get("refused_sandboxes"),
+        Some(&SdkValue::List(vec![t("a-lost")]))
+    );
+    assert_eq!(
+        refused.detail.get("stopped_sandboxes"),
+        Some(&SdkValue::List(vec![t("b-done")]))
+    );
+    let op = read_record(&paths::machine_record(&home, "native"))
+        .unwrap()
+        .and_then(|record| record.get("op").cloned());
+    assert!(
+        op.is_none(),
+        "a refusal before any signal left an op on native: {op:?}"
+    );
+    for name in ["a-lost", "c-new"] {
+        let sandbox = read_record(&paths::sandbox_record(&home, name))
+            .unwrap()
+            .unwrap();
+        for field in ["machine_stopped", "machine_stop_unknown"] {
+            assert!(
+                sandbox.get(field).is_none(),
+                "an unsignalled Sandbox was marked {field}: {sandbox:?}"
+            );
+        }
+    }
+    std::fs::remove_dir_all(root).unwrap();
+}
diff --git a/api/crates/layerfs-api/tests/machine_force_native.rs b/api/crates/layerfs-api/tests/machine_force_native.rs
new file mode 100644
index 0000000000..2bdf3702ea
--- /dev/null
+++ b/api/crates/layerfs-api/tests/machine_force_native.rs
@@ -0,0 +1,257 @@
+//! Live native machine force stop and Sandbox force close with mounted, executing Workspaces.
+#![cfg(all(target_os = "macos", feature = "machine"))]
+use layerfs_api::home::{self, paths, read_record, write_record};
+use layerfs_bridge::control::SdkValue;
+use layerfs_bridge::process::LocalProcess;
+use std::{
+    path::Path,
+    process::{Child, Command},
+    time::{Duration, Instant},
+};
+fn t(value: &str) -> SdkValue {
+    SdkValue::text(value)
+}
+fn call(home: &Path, name: &str, args: &[(&str, SdkValue)]) -> SdkValue {
+    layerfs_api::call(
+        Some(home),
+        name,
+        SdkValue::Map(
+            args.iter()
+                .map(|(name, value)| ((*name).to_owned(), value.clone()))
+                .collect(),
+        ),
+        &mut |_| {},
+    )
+    .unwrap_or_else(|error| panic!("{name}: {error:?}"))
+}
+struct Helper(Child);
+impl Drop for Helper {
+    fn drop(&mut self) {
+        if self.0.try_wait().ok().flatten().is_none() {
+            let _ = self.0.kill();
+            let _ = self.0.wait();
+        }
+    }
+}
+#[test]
+fn exec_helper() {
+    let Ok(home) = std::env::var("LAYERFS_FORCE_NATIVE_HELPER") else {
+        return;
+    };
+    let home = Path::new(&home);
+    let name = std::env::var("LAYERFS_FORCE_NATIVE_NAME").unwrap();
+    let _ = layerfs_api::call(
+        Some(home),
+        "workspace.exec",
+        SdkValue::map([
+            ("ws", t(&format!("{name}:work"))),
+            (
+                "command",
+                t(&format!(
+                    "printf dirty > pending.txt; echo $$ > '{}'; sleep 60",
+                    home.join(format!("{name}.pid")).display()
+                )),
+            ),
+        ]),
+        &mut |_| {},
+    );
+}
+fn live_case(home: &Path, name: &str, machine: bool) {
+    call(
+        home,
+        "sandbox.create",
+        &[
+            ("project", t("shop/app")),
+            ("name", t(name)),
+            ("machine", t("native")),
+        ],
+    );
+    let mounted = call(
+        home,
+        "workspace.mount",
+        &[
+            ("sandbox", t(name)),
+            ("branch", t("work")),
+            ("name", t("work")),
+        ],
+    );
+    let mount = mounted.get_text("path").unwrap().to_owned();
+    let record = read_record(&paths::sandbox_record(home, name))
+        .unwrap()
+        .unwrap();
+    let daemon = record.get("daemon").unwrap();
+    let process = LocalProcess::observe(daemon.get_int("pid").unwrap() as u32).unwrap();
+    let mut helper = Helper(
+        Command::new(std::env::current_exe().unwrap())
+            .args(["exec_helper", "--exact", "--nocapture"])
+            .env("LAYERFS_FORCE_NATIVE_HELPER", home)
+            .env("LAYERFS_FORCE_NATIVE_NAME", name)
+            .spawn()
+            .unwrap(),
+    );
+    let marker = home.join(format!("{name}.pid"));
+    let until = Instant::now() + Duration::from_secs(10);
+    while !marker.exists() {
+        assert!(Instant::now() < until, "exec never became ready");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    let command = LocalProcess::observe(
+        std::fs::read_to_string(marker)
+            .unwrap()
+            .trim()
+            .parse()
+            .unwrap(),
+    )
+    .unwrap();
+    assert!(process.alive() && command.alive());
+    if machine {
+        let stopped = call(
+            home,
+            "machine.stop",
+            &[("name", t("native")), ("force", SdkValue::Bool(true))],
+        );
+        assert_eq!(stopped.get_text("state"), Some("ready"));
+        assert_eq!(
+            call(home, "sandbox.get", &[("name", t(name))]).get_text("state"),
+            Some("gone")
+        );
+        // An explicit second stop of retained terminal records is an idempotent no-op.
+        let again = call(
+            home,
+            "machine.stop",
+            &[("name", t("native")), ("force", SdkValue::Bool(true))],
+        );
+        assert_eq!(again.get_text("state"), Some("ready"));
+    } else {
+        let closed = call(
+            home,
+            "sandbox.close",
+            &[("name", t(name)), ("force", SdkValue::Bool(true))],
+        );
+        assert!(!paths::sandbox_record(home, name).exists());
+        assert!(
+            closed.get("unknown").is_some(),
+            "active Workspace frontier is explicitly unknown"
+        );
+    }
+    assert!(
+        !process.alive() && !command.alive(),
+        "exact daemon and command processes ended"
+    );
+    let until = Instant::now() + Duration::from_secs(5);
+    while helper.0.try_wait().unwrap().is_none() {
+        assert!(Instant::now() < until, "exec helper remained live");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    let mounts = Command::new("/sbin/mount").output().unwrap();
+    assert!(
+        !String::from_utf8_lossy(&mounts.stdout).contains(&format!(" on {mount} ")),
+        "native mount remains attached"
+    );
+    assert!(paths::store_file(home, "shop").exists());
+    assert!(paths::machine_record(home, "native").exists());
+    let saved = call(
+        home,
+        "file.read",
+        &[
+            ("project", t("shop/app")),
+            ("branch", t("work")),
+            ("path", t("/README.md")),
+        ],
+    );
+    assert_eq!(saved.get_text("text"), Some("saved\n"));
+}
+/// A recorded pid that now names an unrelated live process: the forced stop
+/// refuses that Sandbox before any signal, leaves no op on `native`, and
+/// signals neither process.
+fn identity_case(home: &Path) {
+    call(
+        home,
+        "sandbox.create",
+        &[
+            ("project", t("shop/app")),
+            ("name", t("moved")),
+            ("machine", t("native")),
+        ],
+    );
+    let path = paths::sandbox_record(home, "moved");
+    let original = read_record(&path).unwrap().unwrap();
+    let daemon = original.get("daemon").unwrap().clone();
+    let real = LocalProcess::observe(daemon.get_int("pid").unwrap() as u32).unwrap();
+    let mut other = Helper(Command::new("/bin/sleep").arg("30").spawn().unwrap());
+    let moved = home::with_field(daemon, "pid", SdkValue::Int(i64::from(other.0.id())));
+    write_record(&path, &home::with_field(original.clone(), "daemon", moved)).unwrap();
+    let refused = layerfs_api::call(
+        Some(home),
+        "machine.stop",
+        SdkValue::map([("name", t("native")), ("force", SdkValue::Bool(true))]),
+        &mut |_| {},
+    )
+    .unwrap_err();
+    assert_eq!(
+        refused.detail.get("refused_sandboxes"),
+        Some(&SdkValue::List(vec![t("moved")])),
+        "{refused:?}"
+    );
+    assert!(
+        other.0.try_wait().unwrap().is_none(),
+        "the unrelated process was signalled"
+    );
+    assert!(real.alive(), "the real daemon was signalled");
+    let op = read_record(&paths::machine_record(home, "native"))
+        .unwrap()
+        .and_then(|record| record.get("op").cloned());
+    assert!(
+        op.is_none(),
+        "a refusal before any signal left {op:?} on native"
+    );
+    write_record(&path, &original).unwrap();
+    call(
+        home,
+        "sandbox.close",
+        &[("name", t("moved")), ("force", SdkValue::Bool(true))],
+    );
+    assert!(!real.alive(), "the restored record's daemon ended");
+}
+#[test]
+fn native_force_stops_live_daemon_and_command_and_detaches_mount_without_opening_store() {
+    let Ok(daemon) = std::env::var("LAYERFS_DAEMON") else {
+        eprintln!("NOT_RUN: explicit LAYERFS_DAEMON required for live native force proof");
+        return;
+    };
+    let home = std::env::temp_dir().join(format!("lfs-native-force-{}", std::process::id()));
+    home::open(&home).unwrap();
+    let source = home.join("source");
+    std::fs::create_dir(&source).unwrap();
+    std::fs::write(source.join("README.md"), b"saved\n").unwrap();
+    call(&home, "store.create", &[("name", t("shop"))]);
+    assert_eq!(
+        read_record(&paths::store_record(&home, "shop"))
+            .unwrap()
+            .unwrap()
+            .get_text("profile"),
+        Some("disposable")
+    );
+    call(
+        &home,
+        "project.init",
+        &[
+            ("project", t("shop/app")),
+            ("folder", t(&source.display().to_string())),
+        ],
+    );
+    call(
+        &home,
+        "branch.fork",
+        &[
+            ("project", t("shop/app")),
+            ("name", t("work")),
+            ("from_branch", t("main")),
+        ],
+    );
+    home::import_env(&home, &[("LAYERFS_DAEMON".to_owned(), daemon)]).unwrap();
+    live_case(&home, "live", true);
+    live_case(&home, "close", false);
+    identity_case(&home);
+    std::fs::remove_dir_all(&home).unwrap();
+}
diff --git a/api/crates/layerfs-cli/tests/capabilities.rs b/api/crates/layerfs-cli/tests/capabilities.rs
index 40e91bc4b6..b1a598cce8 100644
--- a/api/crates/layerfs-cli/tests/capabilities.rs
+++ b/api/crates/layerfs-cli/tests/capabilities.rs
@@ -98,3 +98,36 @@ fn invalid_discovery_options_are_usage_errors() {
     }
     assert!(!home.exists());
 }
+
+#[test]
+fn explicit_force_options_are_discoverable_alongside_existing_integration_calls() {
+    let home = tempdir("force-capabilities").join("missing");
+    let out = run(&mut layerfs(&home, &["help", "--json"]));
+    assert!(out.status.success(), "{}", text(&out.stderr));
+    let value = json::parse(&text(&out.stdout)).unwrap();
+    let Some(SdkValue::List(calls)) = value.get("calls") else {
+        panic!("calls list");
+    };
+    for (family, name) in [("machine", "stop"), ("sandbox", "close")] {
+        let call = calls
+            .iter()
+            .find(|call| {
+                call.get_text("family") == Some(family) && call.get_text("call") == Some(name)
+            })
+            .unwrap();
+        let Some(SdkValue::List(options)) = call.get("options") else {
+            panic!("call options");
+        };
+        let force = options
+            .iter()
+            .find(|option| option.get_text("name") == Some("force"))
+            .unwrap();
+        assert_eq!(force.get_text("kind"), Some("bool"));
+        assert_eq!(force.get("optional"), Some(&SdkValue::Bool(true)));
+    }
+    assert!(calls
+        .iter()
+        .any(|call| call.get_text("family") == Some("project")
+            && call.get_text("call") == Some("export")));
+    assert!(!home.exists());
+}
diff --git a/api/crates/layerfs-cli/tests/common/stream_worker.rs b/api/crates/layerfs-cli/tests/common/stream_worker.rs
index 4dd7614360..6552d3ad9e 100644
--- a/api/crates/layerfs-cli/tests/common/stream_worker.rs
+++ b/api/crates/layerfs-cli/tests/common/stream_worker.rs
@@ -73,7 +73,7 @@ fn create(more: &SdkValue) -> SdkValue {
         SdkValue::map([
             ("sandbox", more.get("sandbox").cloned().unwrap_or_default()),
             ("name", t("lfs-x-t1")),
-            ("image", t("local")),
+            ("image", t(&format!("sha256:{}", "c".repeat(64)))),
             ("platform", t("docker")),
             ("runtime", t("namespaces")),
             ("route", t("fuse")),
diff --git a/api/crates/layerfs-cli/tests/stream_framed.rs b/api/crates/layerfs-cli/tests/stream_framed.rs
index e83396886a..fcc92851fb 100644
--- a/api/crates/layerfs-cli/tests/stream_framed.rs
+++ b/api/crates/layerfs-cli/tests/stream_framed.rs
@@ -24,7 +24,13 @@ fn result_is_final_even_for_command_exit_2_125_or_cancellation() {
             &home,
             &["workspace", "exec", "d1:one", command, "--events"],
         ));
-        assert_eq!(out.status.code(), Some(0), "{}", text(&out.stderr));
+        assert_eq!(
+            out.status.code(),
+            Some(0),
+            "stdout={} stderr={}",
+            text(&out.stdout),
+            text(&out.stderr)
+        );
         assert!(out.stderr.is_empty());
         let frames = frames(&out.stdout);
         assert!(frames[0].get("progress").is_some());
@@ -52,7 +58,13 @@ fn output_over_one_mib_and_binary_bytes_are_preserved_in_order() {
         &home,
         &["workspace", "exec", "d1:one", "framed-big", "--events"],
     ));
-    assert_eq!(out.status.code(), Some(0), "{}", text(&out.stderr));
+    assert_eq!(
+        out.status.code(),
+        Some(0),
+        "stdout={} stderr={}",
+        text(&out.stdout),
+        text(&out.stderr)
+    );
     let frames = frames(&out.stdout);
     let bytes: Vec<_> = frames.iter().filter_map(|f| f.get("bytes")).collect();
     assert_eq!(bytes.len(), 3);
diff --git a/control/crates/layerfs-control/src/lib.rs b/control/crates/layerfs-control/src/lib.rs
index 0e2377af70..a2cf1d05d7 100644
--- a/control/crates/layerfs-control/src/lib.rs
+++ b/control/crates/layerfs-control/src/lib.rs
@@ -12,6 +12,8 @@ pub mod doctor;
 pub mod inventory;
 #[cfg(feature = "machine")]
 pub mod machine;
+#[cfg(feature = "machine")]
+mod machine_force;
 pub mod migrate;
 pub mod project;
 pub mod store;
diff --git a/control/crates/layerfs-control/src/machine.rs b/control/crates/layerfs-control/src/machine.rs
index 450b615f60..f651620685 100644
--- a/control/crates/layerfs-control/src/machine.rs
+++ b/control/crates/layerfs-control/src/machine.rs
@@ -63,3 +63,8 @@ pub fn remove(home: &Path, name: &MachineName) -> Outcome {
 pub fn update(home: &Path, request: &MachineUpdate) -> Outcome {
     layerfs_machine::update(home, request, RELATED)
 }
+
+/// Explicit crash stop of a machine and this home's Sandboxes, preserving data.
+pub fn stop_force(home: &Path, name: &MachineName) -> Outcome {
+    crate::machine_force::stop(home, name)
+}
diff --git a/control/crates/layerfs-control/src/machine_force.rs b/control/crates/layerfs-control/src/machine_force.rs
new file mode 100644
index 0000000000..05b6cc3449
--- /dev/null
+++ b/control/crates/layerfs-control/src/machine_force.rs
@@ -0,0 +1,310 @@
+//! Explicit provider crash stop without a worker request, VM start or data deletion.
+use crate::inventory;
+use layerfs_bridge::control::{SdkRefusal, SdkValue};
+use layerfs_home::home::{self, lock, paths};
+use layerfs_home::names::MachineName;
+use layerfs_home::op::Op;
+use layerfs_home::outcome::{failed, refused, Outcome};
+use layerfs_machine::platform::{self as adapters, Platform};
+use layerfs_machine::record::MachineRecord;
+use layerfs_machine::{machine_refusal as refusal, machine_spec};
+use layerfs_sandbox::record::SandboxRecord;
+use layerfs_sandbox::{direct, files, reopen};
+use std::path::Path;
+type Result<T> = std::result::Result<T, SdkRefusal>;
+
+pub(crate) fn stop(home: &Path, name: &MachineName) -> Outcome {
+    let home = home.to_path_buf();
+    let name = name.as_str();
+    force_record(&home, name)?;
+    let _start = layerfs_machine::force_stop_lock(&home, name)?;
+    let intent = Op::begin(
+        "machine.stop",
+        "force",
+        SdkValue::map([("force", SdkValue::Bool(true))]),
+    )?;
+    let (spec, hosted, sandboxes) = {
+        let _home = lock::take(&home)?;
+        let machine = force_record(&home, name)?;
+        let hosted = inventory::hosted_on(&home, name)?;
+        let sandboxes = inventory::sandboxes(&home)?
+            .into_iter()
+            .filter(|sandbox| {
+                sandbox.machine() == name || hosted.iter().any(|child| sandbox.machine() == child)
+            })
+            .collect::<Vec<_>>();
+        for sandbox in &sandboxes {
+            if sandbox
+                .op()
+                .is_some_and(|op| op.name() == "sandbox.create" && op.owner_alive())
+            {
+                return Err(refused(
+                    "Busy",
+                    format!(
+                        "Sandbox {} is still being created; no machine stop was attempted",
+                        sandbox.name()
+                    ),
+                ));
+            }
+        }
+        let spec = machine_spec(&home, &home_id(&home)?, &machine)?;
+        for child in &hosted {
+            force_record(&home, child)?;
+        }
+        for machine in std::iter::once(name).chain(hosted.iter().map(String::as_str)) {
+            update(&home, machine, |record| record.with_op(&intent))?;
+        }
+        (spec, hosted, sandboxes)
+    };
+    let mut unknown = Vec::new();
+    let mut stopped = Vec::new();
+    let mut skipped = Vec::new();
+    let mut native_attempt = None;
+    let attempted = (|| -> Result<()> {
+        if spec.kind == Platform::Native {
+            let mut first_refusal = None;
+            for sandbox in &sandboxes {
+                // A refusal before the signal leaves this runtime as it was,
+                // so the stop goes on to the next Sandbox: only a sent stop
+                // makes an outcome uncertain.
+                let runtime = match native_runtime(&home, sandbox) {
+                    Ok(runtime) => runtime,
+                    Err(refusal) => {
+                        skipped.push(sandbox.name().to_owned());
+                        first_refusal.get_or_insert(refusal);
+                        continue;
+                    }
+                };
+                match runtime {
+                    // Nothing ever ran: nothing is stopped or recorded.
+                    Native::NeverStarted => continue,
+                    Native::Stopped => {}
+                    Native::Stop(mut runtime) => {
+                        native_attempt = Some(sandbox);
+                        runtime.stop(0).map_err(failed)?;
+                    }
+                }
+                mark(&home, sandbox, true)?;
+                native_attempt = None;
+                stopped.push(sandbox.name().to_owned());
+                unknown.extend(workspaces(&home, sandbox)?);
+            }
+            if let Some(refusal) = first_refusal {
+                return Err(refusal);
+            }
+        } else {
+            adapters::stop_force(&spec).map_err(refusal)?;
+            for sandbox in &sandboxes {
+                mark(&home, sandbox, true)?;
+                stopped.push(sandbox.name().to_owned());
+                let observed = workspaces(&home, sandbox)?;
+                if observed.is_empty() {
+                    unknown.push(format!("{}:*", sandbox.name()));
+                } else {
+                    unknown.extend(observed);
+                }
+            }
+        }
+        Ok(())
+    })();
+    if let Err(mut error) = attempted {
+        // Native Sandboxes are recorded one by one as they stop, so only the
+        // one being signalled is uncertain; a provider stop covers them all.
+        let uncertain = match spec.kind {
+            Platform::Native => native_attempt.is_some(),
+            _ => error.code != "Unsupported" || !stopped.is_empty(),
+        };
+        // Recording the outcome must not replace the stop's own refusal: a
+        // failure here is reported beside it.
+        let mut custody = Vec::new();
+        let mut keep = |result: Result<()>| {
+            if let Err(failure) = result {
+                custody.push(SdkValue::text(format!(
+                    "{}: {}",
+                    failure.code, failure.message
+                )));
+            }
+        };
+        keep((|| {
+            let _home = lock::take(&home)?;
+            for machine in std::iter::once(name).chain(hosted.iter().map(String::as_str)) {
+                update(&home, machine, |record| {
+                    if uncertain {
+                        record.with_op(&intent.clone().as_uncertain())
+                    } else {
+                        record.without_op()
+                    }
+                })?;
+            }
+            Ok(())
+        })());
+        // A multi-container provider stop may have completed only part of its work.
+        // Keep that uncertainty distinct from both a running or a confirmed gone runtime.
+        if spec.kind != Platform::Native && error.code != "Unsupported" {
+            for sandbox in &sandboxes {
+                if !stopped.iter().any(|name| name == sandbox.name()) {
+                    keep(mark(&home, sandbox, false));
+                }
+            }
+        } else if let Some(sandbox) = native_attempt {
+            if !stopped.iter().any(|name| name == sandbox.name()) {
+                keep(mark(&home, sandbox, false));
+            }
+        }
+        for (key, list) in [
+            ("stopped_sandboxes", stopped),
+            ("refused_sandboxes", skipped),
+        ] {
+            error.detail = home::with_field(
+                error.detail,
+                key,
+                SdkValue::List(list.into_iter().map(SdkValue::Text).collect()),
+            );
+        }
+        error.detail = home::with_field(
+            error.detail,
+            "unknown",
+            SdkValue::List(unknown.into_iter().map(SdkValue::Text).collect()),
+        );
+        if !custody.is_empty() {
+            error.detail =
+                home::with_field(error.detail, "custody_errors", SdkValue::List(custody));
+        }
+        return Err(error);
+    }
+    {
+        let _home = lock::take(&home)?;
+        for machine in std::iter::once(name).chain(hosted.iter().map(String::as_str)) {
+            match std::fs::remove_file(paths::worker_record(&home, machine)) {
+                Ok(()) => {}
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+                Err(error) => return Err(failed(error)),
+            }
+            update(&home, machine, |record| {
+                record.without_op().with(
+                    "state",
+                    SdkValue::text(if machine == "native" {
+                        "ready"
+                    } else {
+                        "stopped"
+                    }),
+                )
+            })?;
+        }
+    }
+    // Do not call get(): a live probe could contact the worker we just stopped.
+    let mut out = layerfs_machine::stopped_summary(&home, name, crate::machine::RELATED)?;
+    for (key, value) in [
+        ("forced", SdkValue::Bool(true)),
+        (
+            "stopped_sandboxes",
+            SdkValue::List(stopped.into_iter().map(SdkValue::Text).collect()),
+        ),
+        (
+            "unknown",
+            SdkValue::List(unknown.into_iter().map(SdkValue::Text).collect()),
+        ),
+    ] {
+        out = home::with_field(out, key, value);
+    }
+    Ok(out)
+}
+
+/// A new explicit force call may reconcile an unknown stop that this
+/// process left, or one whose owner has ended (`Unresolved`); one a live
+/// process still holds stays `Busy`. It never silently retries, nor clears
+/// another kind of unresolved operation.
+fn force_record(home: &Path, name: &str) -> Result<MachineRecord> {
+    let record = MachineRecord::of(home, name)?;
+    if record.op().is_some_and(|op| {
+        op.name() == "machine.stop" && op.uncertain() && (op.mine() || !op.owner_alive())
+    }) {
+        return Ok(record);
+    }
+    record.not_held()?;
+    Ok(record)
+}
+
+/// Record an observed terminal fact, retaining runtime files and exact identity.
+fn mark(home: &Path, sandbox: &SandboxRecord, stopped: bool) -> Result<()> {
+    let _home = lock::take(home)?;
+    let _sandbox = layerfs_sandbox::record::lock(home, sandbox.name())?;
+    if let Some(now) = inventory::sandbox(home, sandbox.name())? {
+        if now.to_record().get("runtime_id") != sandbox.to_record().get("runtime_id") {
+            return Err(refused(
+                "Unknown",
+                "the Sandbox runtime identity changed during the stop; its new record was retained",
+            ));
+        }
+        now.with_machine_stop(stopped).write(home)?;
+    }
+    if stopped {
+        direct::release(home, sandbox);
+    }
+    Ok(())
+}
+
+/// What a forced stop finds for one native Sandbox. Native is the host
+/// itself: only this home's recorded runtime processes stop.
+enum Native {
+    /// Its recorded runtime, reopened for the stop; nothing sent yet.
+    Stop(Box<layerfs_sandbox_backend::Sandbox>),
+    /// Its runtime is already stopped.
+    Stopped,
+    /// It never started (no id and no runtime record).
+    NeverStarted,
+}
+
+/// The Sandbox's recorded native runtime, reopened for a stop without sending
+/// anything, or why there is nothing to stop.
+fn native_runtime(home: &Path, sandbox: &SandboxRecord) -> Result<Native> {
+    if sandbox.machine_stopped() {
+        return Ok(Native::Stopped);
+    }
+    let host = layerfs_sandbox::force::native_host(home, sandbox.name())?;
+    let Some(daemon) = files::started(&sandbox.dir(home))? else {
+        if sandbox.id().is_some() {
+            return Err(refused(
+                "Unavailable",
+                format!(
+                    "Sandbox {} has no recorded native runtime; its stop cannot be confirmed",
+                    sandbox.name()
+                ),
+            ));
+        }
+        sandbox.free()?;
+        return Ok(Native::NeverStarted);
+    };
+    Ok(
+        match reopen::recorded(&host, sandbox.name(), &daemon)?.for_stop()? {
+            Some(runtime) => Native::Stop(runtime),
+            None => Native::Stopped,
+        },
+    )
+}
+
+/// These are lost/unknown uncommitted frontiers, never manufactured commits.
+fn workspaces(home: &Path, sandbox: &SandboxRecord) -> Result<Vec<String>> {
+    let mut workspaces = Vec::new();
+    for (path, record) in files::workspaces(&sandbox.dir(home))? {
+        let fallback = path
+            .file_name()
+            .and_then(|name| name.to_str())
+            .unwrap_or("unknown");
+        let value = record.name().or_else(|| record.path()).unwrap_or(fallback);
+        workspaces.push(format!("{}:{value}", sandbox.name()));
+    }
+    Ok(workspaces)
+}
+
+fn home_id(home: &Path) -> Result<String> {
+    Ok(home::opened(home)?.id)
+}
+
+fn update(
+    home: &Path,
+    name: &str,
+    change: impl FnOnce(MachineRecord) -> MachineRecord,
+) -> Result<()> {
+    change(MachineRecord::of(home, name)?).write(home)
+}
diff --git a/control/crates/layerfs-machine/src/endpoint.rs b/control/crates/layerfs-machine/src/endpoint.rs
index ed71b648ad..7f30823955 100644
--- a/control/crates/layerfs-machine/src/endpoint.rs
+++ b/control/crates/layerfs-machine/src/endpoint.rs
@@ -99,6 +99,7 @@ fn start(
 /// A wait for it, at most [`START_WAIT`], is not a retry.
 pub type StartLock = layerfs_home::home::entity_lock::EntityLock;
 
+/// Take the machine lifecycle lock before a start, create, remove or forced stop.
 pub fn start_lock(home: &Path, name: &str) -> Result<StartLock, SdkRefusal> {
     let path = layerfs_home::home::paths::machine_dir(home, name).join("start.lock");
     let held = format!("machine {name} was being started, made or removed by another process");
diff --git a/control/crates/layerfs-machine/src/lib.rs b/control/crates/layerfs-machine/src/lib.rs
index b01dd35222..9ce32cdf03 100644
--- a/control/crates/layerfs-machine/src/lib.rs
+++ b/control/crates/layerfs-machine/src/lib.rs
@@ -76,3 +76,10 @@ pub use spec::{
 pub use status::status;
 #[cfg(feature = "machine")]
 pub use update::{update, MachineUpdate};
+
+/// The machine lifecycle lock, shared by the spanning forced-stop call.
+#[cfg(feature = "machine")]
+pub use endpoint::start_lock as force_stop_lock;
+/// A confirmed stopped machine summary without a worker or provider probe.
+#[cfg(feature = "machine")]
+pub use power::stopped_summary;
diff --git a/control/crates/layerfs-machine/src/power.rs b/control/crates/layerfs-machine/src/power.rs
index abc612099d..088d6b614c 100644
--- a/control/crates/layerfs-machine/src/power.rs
+++ b/control/crates/layerfs-machine/src/power.rs
@@ -68,3 +68,17 @@ pub fn stop(home: &Path, name: &MachineName, related: Related) -> Outcome {
     }
     super::get::get(&home, name, related)
 }
+
+/// A provider-confirmed terminal state, without contacting its stopped worker.
+pub fn stopped_summary(home: &Path, name: &str, related: Related) -> Outcome {
+    let machine = super::record::MachineRecord::of(home, name)?;
+    super::list::summary(
+        home,
+        &machine,
+        Some((
+            if name == "native" { "ready" } else { "stopped" }.to_owned(),
+            None,
+        )),
+        related,
+    )
+}
diff --git a/control/crates/layerfs-platform/src/adapter.rs b/control/crates/layerfs-platform/src/adapter.rs
index bb24fabf36..5ea1a8104c 100644
--- a/control/crates/layerfs-platform/src/adapter.rs
+++ b/control/crates/layerfs-platform/src/adapter.rs
@@ -24,7 +24,7 @@ pub(crate) trait Adapter: Sync {
     /// Starts a stopped machine or its worker; `None`: nothing to start.
     fn start(&self, spec: &Spec, bundle: &Bundle) -> Result<Option<Running>, MachineError>;
     /// Stops the worker and, where there is one, the VM.
-    fn stop(&self, spec: &Spec) -> Result<(), MachineError>;
+    fn stop(&self, spec: &Spec, force: bool) -> Result<(), MachineError>;
     /// Provider runtime state only, without worker contact or guest commands.
     fn provider_state(&self, spec: &Spec) -> Result<State, MachineError>;
     /// The machine's state now; starts nothing.
diff --git a/control/crates/layerfs-platform/src/lib.rs b/control/crates/layerfs-platform/src/lib.rs
index 1053cceaaa..ee51b4a1c9 100644
--- a/control/crates/layerfs-platform/src/lib.rs
+++ b/control/crates/layerfs-platform/src/lib.rs
@@ -31,7 +31,7 @@ pub use platform::{ExecutorKind, HostOs, How, KernelRoute, Platform, Profile, Ro
 pub use probe::{native_missing, probe, Probe};
 pub use setup::{
     cached_image, copy_in, copy_out, create, forward, forward_close, load_image, provider_within,
-    reach, remove, start, status, status_within, stop, State, Status,
+    reach, remove, start, status, status_within, stop, stop_force, State, Status,
 };
 pub use spec::{
     parse_size, Bundle, Endpoint, Forward, Lasting, Running, Spec, Tools, BUNDLE, DATA, IMAGE,
diff --git a/control/crates/layerfs-platform/src/platforms/apple_container/adapter.rs b/control/crates/layerfs-platform/src/platforms/apple_container/adapter.rs
index 8bcc1bb770..f39baa590b 100644
--- a/control/crates/layerfs-platform/src/platforms/apple_container/adapter.rs
+++ b/control/crates/layerfs-platform/src/platforms/apple_container/adapter.rs
@@ -39,8 +39,8 @@ impl Adapter for AppleContainer {
     fn start(&self, spec: &Spec, bundle: &Bundle) -> Result<Option<Running>, MachineError> {
         start(spec, bundle).map(Some)
     }
-    fn stop(&self, spec: &Spec) -> Result<(), MachineError> {
-        stop(spec)
+    fn stop(&self, spec: &Spec, force: bool) -> Result<(), MachineError> {
+        stop(spec, force)
     }
     fn provider_state(&self, spec: &Spec) -> Result<crate::setup::State, MachineError> {
         super::observe::observe(spec)
diff --git a/control/crates/layerfs-platform/src/platforms/apple_container/power.rs b/control/crates/layerfs-platform/src/platforms/apple_container/power.rs
index 8406632f03..546f587080 100644
--- a/control/crates/layerfs-platform/src/platforms/apple_container/power.rs
+++ b/control/crates/layerfs-platform/src/platforms/apple_container/power.rs
@@ -18,8 +18,9 @@ pub(crate) fn start(spec: &Spec, bundle: &Bundle) -> Result<Running, MachineErro
     bring_up(spec, bundle, port)
 }
 
-pub(crate) fn stop(spec: &Spec) -> Result<(), MachineError> {
-    if observe(spec)? == State::Running {
+pub(crate) fn stop(spec: &Spec, force: bool) -> Result<(), MachineError> {
+    let state = observe(spec)?;
+    if state == State::Running || (force && state == State::Failed) {
         checked(spec, CLI, &["stop", &spec.vm_name], CREATE)?;
     }
     Ok(())
diff --git a/control/crates/layerfs-platform/src/platforms/docker/adapter.rs b/control/crates/layerfs-platform/src/platforms/docker/adapter.rs
index 99c49124ad..25145485ca 100644
--- a/control/crates/layerfs-platform/src/platforms/docker/adapter.rs
+++ b/control/crates/layerfs-platform/src/platforms/docker/adapter.rs
@@ -55,8 +55,8 @@ impl Adapter for Docker {
     fn start(&self, spec: &Spec, bundle: &Bundle) -> Result<Option<Running>, MachineError> {
         start(spec, bundle).map(Some)
     }
-    fn stop(&self, spec: &Spec) -> Result<(), MachineError> {
-        stop(spec)
+    fn stop(&self, spec: &Spec, force: bool) -> Result<(), MachineError> {
+        stop(spec, force)
     }
     fn provider_state(&self, spec: &Spec) -> Result<crate::setup::State, MachineError> {
         super::provider_state(spec)
diff --git a/control/crates/layerfs-platform/src/platforms/docker/power.rs b/control/crates/layerfs-platform/src/platforms/docker/power.rs
index 493cc56c41..b834d39cf0 100644
--- a/control/crates/layerfs-platform/src/platforms/docker/power.rs
+++ b/control/crates/layerfs-platform/src/platforms/docker/power.rs
@@ -123,13 +123,22 @@ pub(crate) fn start(spec: &Spec, bundle: &Bundle) -> Result<Running, MachineErro
 }
 
 /// Stops the worker container; Docker itself and the Store volume stay.
-pub(crate) fn stop(spec: &Spec) -> Result<(), MachineError> {
-    if inspect(spec)?.is_none() {
+pub(crate) fn stop(spec: &Spec, force: bool) -> Result<(), MachineError> {
+    let Some((state, role)) = inspect(spec)? else {
+        return Ok(());
+    };
+    if role != "machine-worker" {
+        return Err(MachineError::new(
+            Code::InvalidInput,
+            "the recorded container is not a LayerFS worker; it is left alone",
+        ));
+    }
+    if !matches!(state.as_str(), "running" | "restarting" | "paused") {
         return Ok(());
     }
     run(
         spec,
-        &["stop", "-t", "5", &spec.vm_name],
+        &["stop", "-t", if force { "0" } else { "5" }, &spec.vm_name],
         "stopping the worker container",
     )?
     .answer("stopping the worker container")?;
diff --git a/control/crates/layerfs-platform/src/platforms/docker/windows.rs b/control/crates/layerfs-platform/src/platforms/docker/windows.rs
index df206e7d25..eb7bf93767 100644
--- a/control/crates/layerfs-platform/src/platforms/docker/windows.rs
+++ b/control/crates/layerfs-platform/src/platforms/docker/windows.rs
@@ -23,7 +23,7 @@ pub(crate) fn guest(spec: &Spec) -> Result<Guest, MachineError> {
     Err(unsupported(spec))
 }
 
-pub(crate) fn stop(spec: &Spec) -> Result<(), MachineError> {
+pub(crate) fn stop(spec: &Spec, _force: bool) -> Result<(), MachineError> {
     Err(unsupported(spec))
 }
 
diff --git a/control/crates/layerfs-platform/src/platforms/lima/adapter.rs b/control/crates/layerfs-platform/src/platforms/lima/adapter.rs
index 0efdc5b13c..8bc121835e 100644
--- a/control/crates/layerfs-platform/src/platforms/lima/adapter.rs
+++ b/control/crates/layerfs-platform/src/platforms/lima/adapter.rs
@@ -36,8 +36,8 @@ impl Adapter for Lima {
     fn start(&self, spec: &Spec, bundle: &Bundle) -> Result<Option<Running>, MachineError> {
         start(spec, bundle).map(Some)
     }
-    fn stop(&self, spec: &Spec) -> Result<(), MachineError> {
-        stop(spec)
+    fn stop(&self, spec: &Spec, force: bool) -> Result<(), MachineError> {
+        stop(spec, force)
     }
     fn provider_state(&self, spec: &Spec) -> Result<crate::setup::State, MachineError> {
         super::observe::observe(spec).map(|(state, _)| state)
diff --git a/control/crates/layerfs-platform/src/platforms/lima/power.rs b/control/crates/layerfs-platform/src/platforms/lima/power.rs
index 9e08085326..a5668f3cba 100644
--- a/control/crates/layerfs-platform/src/platforms/lima/power.rs
+++ b/control/crates/layerfs-platform/src/platforms/lima/power.rs
@@ -20,9 +20,15 @@ pub(crate) fn start(spec: &Spec, bundle: &Bundle) -> Result<Running, MachineErro
 }
 
 /// `limactl stop` keeps the instance and its disk; the worker ends with it.
-pub(crate) fn stop(spec: &Spec) -> Result<(), MachineError> {
-    if observe(spec)?.0 == State::Running {
-        checked(spec, CLI, &["stop", &spec.vm_name], BOOT)?;
+pub(crate) fn stop(spec: &Spec, force: bool) -> Result<(), MachineError> {
+    let state = observe(spec)?.0;
+    if state == State::Running || (force && state == State::Failed) {
+        let mut words = vec!["stop"];
+        if force {
+            words.push("--force");
+        }
+        words.push(&spec.vm_name);
+        checked(spec, CLI, &words, BOOT)?;
     }
     Ok(())
 }
diff --git a/control/crates/layerfs-platform/src/platforms/native/adapter.rs b/control/crates/layerfs-platform/src/platforms/native/adapter.rs
index fd6660b0b5..70bbfa9c80 100644
--- a/control/crates/layerfs-platform/src/platforms/native/adapter.rs
+++ b/control/crates/layerfs-platform/src/platforms/native/adapter.rs
@@ -30,7 +30,7 @@ impl Adapter for Native {
     fn start(&self, spec: &Spec, _bundle: &Bundle) -> Result<Option<Running>, MachineError> {
         start(spec)
     }
-    fn stop(&self, _spec: &Spec) -> Result<(), MachineError> {
+    fn stop(&self, _spec: &Spec, _force: bool) -> Result<(), MachineError> {
         Ok(())
     }
     fn provider_state(&self, _spec: &Spec) -> Result<crate::setup::State, MachineError> {
diff --git a/control/crates/layerfs-platform/src/setup/calls.rs b/control/crates/layerfs-platform/src/setup/calls.rs
index 8fb2e21e1d..e8896768a2 100644
--- a/control/crates/layerfs-platform/src/setup/calls.rs
+++ b/control/crates/layerfs-platform/src/setup/calls.rs
@@ -70,7 +70,14 @@ pub fn start(spec: &Spec, bundle: &Bundle) -> Result<Option<Running>, MachineErr
 /// `docker` stops only its worker; `native` has nothing to stop.
 pub fn stop(spec: &Spec) -> Result<(), MachineError> {
     spec.validate()?;
-    adapter(spec.kind).stop(spec)
+    adapter(spec.kind).stop(spec, false)
+}
+
+/// Stops directly through the provider, without contacting or starting its worker.
+/// The caller owns Sandbox/Workspace reconciliation; registered machines and disks stay.
+pub fn stop_force(spec: &Spec) -> Result<(), MachineError> {
+    spec.validate()?;
+    adapter(spec.kind).stop(spec, true)
 }
 
 /// `machine remove` deletes a VM and its data disk. `InUse` is the
diff --git a/control/crates/layerfs-platform/src/setup/mod.rs b/control/crates/layerfs-platform/src/setup/mod.rs
index 9f23307f64..4b3b0d3735 100644
--- a/control/crates/layerfs-platform/src/setup/mod.rs
+++ b/control/crates/layerfs-platform/src/setup/mod.rs
@@ -10,5 +10,5 @@ pub(crate) mod worker;
 pub(crate) use calls::with_worker;
 pub use calls::{
     cached_image, copy_in, copy_out, create, forward, forward_close, load_image, provider_within,
-    reach, remove, start, status, status_within, stop, State, Status,
+    reach, remove, start, status, status_within, stop, stop_force, State, Status,
 };
diff --git a/control/crates/layerfs-platform/tests/machine_force.rs b/control/crates/layerfs-platform/tests/machine_force.rs
new file mode 100644
index 0000000000..82933c6454
--- /dev/null
+++ b/control/crates/layerfs-platform/tests/machine_force.rs
@@ -0,0 +1,126 @@
+//! Force stop uses only provider power operations, retaining machines and disks.
+#![cfg(unix)]
+#[path = "support/standin.rs"]
+mod standin;
+use layerfs_platform::{self as machine, Code, Platform};
+use standin::Scratch;
+
+#[test]
+fn native_force_stop_never_stops_the_host() {
+    let scratch = Scratch::new("force-native");
+    machine::stop_force(&scratch.spec(Platform::Native)).unwrap();
+    assert!(scratch.log().is_empty());
+}
+
+#[test]
+fn docker_force_stops_only_the_owned_worker_with_zero_grace() {
+    for state in ["running", "paused", "restarting"] {
+        let scratch = Scratch::new("force-docker");
+        scratch.tool("docker", &format!(
+            "case \"$3\" in inspect) printf '{state}\\tmachine-worker\\n';; stop) exit 0;; *) exit 99;; esac"
+        ));
+        machine::stop_force(&scratch.spec(Platform::Docker)).unwrap();
+        let log = scratch.log();
+        let calls: Vec<_> = log.lines().collect();
+        assert_eq!(calls.len(), 2, "{log}");
+        assert!(calls[0].contains(" inspect -f "), "{log}");
+        assert_eq!(
+            calls[1],
+            "docker -H unix:///var/run/docker.sock stop -t 0 lfs-dev"
+        );
+    }
+}
+
+#[test]
+fn docker_force_preserves_unowned_containers_and_reports_inspect_failure() {
+    for (answer, code) in [
+        ("printf 'running\\tother-role\\n'", Code::InvalidInput),
+        ("echo 'engine unavailable' >&2; exit 1", Code::Failed),
+    ] {
+        let scratch = Scratch::new("force-docker-refuse");
+        scratch.tool("docker", answer);
+        let error = machine::stop_force(&scratch.spec(Platform::Docker)).unwrap_err();
+        assert_eq!(error.code, code, "{error}");
+        assert_eq!(scratch.log().lines().count(), 1);
+    }
+}
+
+#[test]
+fn docker_missing_or_stopped_worker_does_not_boot_or_remove_anything() {
+    for answer in [
+        "echo 'No such object: lfs-dev' >&2; exit 1",
+        "printf 'exited\\tmachine-worker\\n'",
+    ] {
+        let scratch = Scratch::new("force-docker-idle");
+        scratch.tool("docker", answer);
+        machine::stop_force(&scratch.spec(Platform::Docker)).unwrap();
+        assert_eq!(scratch.log().lines().count(), 1);
+    }
+}
+
+#[test]
+fn ordinary_docker_stop_retains_grace_period() {
+    let scratch = Scratch::new("normal-docker-stop");
+    scratch.tool("docker", "case \"$3\" in inspect) printf 'running\\tmachine-worker\\n';; stop) exit 0;; *) exit 99;; esac");
+    machine::stop(&scratch.spec(Platform::Docker)).unwrap();
+    assert!(scratch.log().ends_with(" stop -t 5 lfs-dev\n"));
+}
+
+#[test]
+fn lima_force_stops_running_and_broken_instances_without_guest_calls() {
+    for state in ["Running", "Broken"] {
+        let scratch = Scratch::new("force-lima");
+        scratch.tool(
+            "limactl",
+            &format!(
+            "case \"$1\" in list) printf '{state}\\t/unused\\n';; stop) exit 0;; *) exit 99;; esac"
+        ),
+        );
+        machine::stop_force(&scratch.spec(Platform::Lima)).unwrap();
+        let log = scratch.log();
+        assert_eq!(log.lines().count(), 2, "{log}");
+        assert!(log.ends_with("limactl stop --force lfs-dev\n"), "{log}");
+    }
+}
+
+#[test]
+fn lima_idle_force_and_normal_stop_do_not_change_disks() {
+    for state in ["Stopped", ""] {
+        let scratch = Scratch::new("force-lima-idle");
+        scratch.tool("limactl", &format!("printf '{state}\\n'"));
+        machine::stop_force(&scratch.spec(Platform::Lima)).unwrap();
+        assert_eq!(scratch.log().lines().count(), 1);
+    }
+    let scratch = Scratch::new("normal-lima-stop");
+    scratch.tool(
+        "limactl",
+        "case \"$1\" in list) printf 'Running\\t/unused\\n';; stop) exit 0;; *) exit 99;; esac",
+    );
+    machine::stop(&scratch.spec(Platform::Lima)).unwrap();
+    assert!(scratch.log().ends_with("limactl stop lfs-dev\n"));
+}
+
+#[test]
+fn apple_force_stops_running_or_failed_container_without_guest_calls() {
+    for state in ["running", "failed"] {
+        let scratch = Scratch::new("force-apple");
+        scratch.tool("container", &format!(
+            "case \"$1\" in inspect) echo '[{{\"status\":{{\"state\":\"{state}\"}}}}]';; stop) exit 0;; *) exit 99;; esac"
+        ));
+        machine::stop_force(&scratch.spec(Platform::AppleContainer)).unwrap();
+        assert_eq!(
+            scratch.log(),
+            "container inspect lfs-dev\ncontainer stop lfs-dev\n"
+        );
+    }
+}
+
+#[test]
+fn apple_missing_or_stopped_container_does_not_start_or_delete() {
+    for answer in ["[]", "[{\"status\":{\"state\":\"stopped\"}}]"] {
+        let scratch = Scratch::new("force-apple-idle");
+        scratch.tool("container", &format!("echo '{answer}'"));
+        machine::stop_force(&scratch.spec(Platform::AppleContainer)).unwrap();
+        assert_eq!(scratch.log(), "container inspect lfs-dev\n");
+    }
+}
diff --git a/control/crates/layerfs-sandbox-backend/src/backend/host_process/sandbox.rs b/control/crates/layerfs-sandbox-backend/src/backend/host_process/sandbox.rs
index db5723fcdc..49ed80ea75 100644
--- a/control/crates/layerfs-sandbox-backend/src/backend/host_process/sandbox.rs
+++ b/control/crates/layerfs-sandbox-backend/src/backend/host_process/sandbox.rs
@@ -84,6 +84,31 @@ impl Sandbox {
         owner.endpoint()?;
         Ok(owner)
     }
+    /// Reopen only for an explicit crash stop, including a daemon already gone.
+    /// No endpoint or daemon RPC is needed. Stop still matches pid, start and boot.
+    pub fn reopen_for_stop(
+        directory: PathBuf,
+        id: ContainerId,
+        process: LocalProcess,
+        port: u16,
+        labels: Vec<(String, String)>,
+    ) -> Result<Self, Box<WireFailure>> {
+        let mut owner = Self::unstarted(id, directory, labels);
+        owner.process = Some(process);
+        owner.port = port;
+        if !owner.directory.is_absolute() || port == 0 {
+            return Err(owner.invalid("saved local Sandbox"));
+        }
+        if process.boot.is_none() {
+            return Err(owner.invalid("saved local daemon has no boot identity; mounts retained"));
+        }
+        if let Ok(current) = LocalProcess::observe(process.pid) {
+            if current.alive() && !process.alive() {
+                return Err(owner.invalid("saved local daemon identity changed; mounts retained"));
+            }
+        }
+        Ok(owner)
+    }
     /// A recorded Sandbox that [`Self::reopen`] found stopped
     /// ([`WireFailure::container_stopped`]): only its one Delete is admitted.
     pub fn stopped(
diff --git a/control/crates/layerfs-sandbox-backend/tests/local_stop_mount.rs b/control/crates/layerfs-sandbox-backend/tests/local_stop_mount.rs
index 26a5a6b6e9..f086a5fa1c 100644
--- a/control/crates/layerfs-sandbox-backend/tests/local_stop_mount.rs
+++ b/control/crates/layerfs-sandbox-backend/tests/local_stop_mount.rs
@@ -6,6 +6,32 @@ use layerfs_bridge::process::LocalProcess;
 use layerfs_sandbox_backend::{host_process::Sandbox, ContainerId};
 use std::{path::PathBuf, process::Command};
 
+#[test]
+fn force_reopen_refuses_unconfirmed_process_identity_before_stop() {
+    let process = LocalProcess::observe(std::process::id()).unwrap();
+    for unconfirmed in [
+        LocalProcess {
+            boot: None,
+            ..process
+        },
+        LocalProcess {
+            start: process.start + 1,
+            ..process
+        },
+    ] {
+        let result = Sandbox::reopen_for_stop(
+            std::env::temp_dir().join("layerfs-force-unconfirmed-owner"),
+            ContainerId::parse(&"1".repeat(64)).unwrap(),
+            unconfirmed,
+            12345,
+            Vec::new(),
+        );
+        let error = result.unwrap_err();
+        assert!(format!("{error:?}").contains("mounts retained"));
+        assert!(process.alive());
+    }
+}
+
 #[test]
 #[ignore = "requires an explicitly supplied disposable mounted Sandbox"]
 fn stop_detaches_mounts_before_returning_without_delete() {
diff --git a/control/crates/layerfs-sandbox/src/close.rs b/control/crates/layerfs-sandbox/src/close.rs
index 188c10f7f3..e9e1321373 100644
--- a/control/crates/layerfs-sandbox/src/close.rs
+++ b/control/crates/layerfs-sandbox/src/close.rs
@@ -104,13 +104,22 @@ pub fn close(home: &Path, events: &mut Events<'_>, request: &CloseRequest) -> Re
             }
         }
     }
-    let closed = on_machine(
-        &home,
-        events,
-        &record,
-        Call::Close { force, held },
-        crate::dispatch::call,
-    );
+    let closed = if force && record.machine() == "native" {
+        // a forced close on `native` needs no Store or daemon discovery,
+        // even when active mounts or a broken Store would block the
+        // ordinary route
+        crate::force::native_free(&home)
+            .and_then(|()| crate::force::native_host(&home, name))
+            .and_then(|host| crate::force::close(&host, held.as_deref().unwrap_or(&[])))
+    } else {
+        on_machine(
+            &home,
+            events,
+            &record,
+            Call::Close { force, held },
+            crate::dispatch::call,
+        )
+    };
     let _lock = home::lock::take(&home)?;
     // on `native` the Sandbox's own side removed the folder already
     let dir = record.dir(&home);
@@ -140,10 +149,16 @@ pub fn close(home: &Path, events: &mut Events<'_>, request: &CloseRequest) -> Re
             op.log_uncertain(&home, "sandbox", name, &refusal.message);
             Err(refusal)
         }
-        // a certain failure leaves the Sandbox as it was
+        // a certain failure leaves the Sandbox as it was, except a forced
+        // close whose runtime already stopped: the record says so
         Err(refusal) => {
             if let Some(now) = SandboxRecord::read(&home, name)? {
-                now.without_op().write(&home)?;
+                let stopped = refusal.detail.get_bool(crate::force::RUNTIME_STOPPED);
+                match stopped {
+                    true => now.without_op().with_machine_stop(true),
+                    false => now.without_op(),
+                }
+                .write(&home)?;
             }
             Err(refusal)
         }
diff --git a/control/crates/layerfs-sandbox/src/direct.rs b/control/crates/layerfs-sandbox/src/direct.rs
index 9cc6c22d01..8846cf6368 100644
--- a/control/crates/layerfs-sandbox/src/direct.rs
+++ b/control/crates/layerfs-sandbox/src/direct.rs
@@ -324,6 +324,9 @@ pub fn capacity(home: &Path, record: &SandboxRecord) -> Outcome {
 /// Store's lease (the Sandbox half of `workspace::direct::call`).
 fn served(home: &Path, record: &SandboxRecord, call: Call) -> Outcome {
     let home = home.to_path_buf();
+    if let Some(machine) = layerfs_machine::record::MachineRecord::read(&home, record.machine())? {
+        machine.not_held()?;
+    }
     let target = resolve::target(&home, record.project().unwrap_or(""))?;
     let _lease = lease::hold(&home, &target.store)?;
     dispatch::call(host(&home, record, &target, Message::Absent), call)
diff --git a/control/crates/layerfs-sandbox/src/force.rs b/control/crates/layerfs-sandbox/src/force.rs
new file mode 100644
index 0000000000..3562a9090e
--- /dev/null
+++ b/control/crates/layerfs-sandbox/src/force.rs
@@ -0,0 +1,100 @@
+//! The forced close of a `native` Sandbox, which bypasses Store and daemon
+//! discovery and live unmount admission: only its runtime stops; the
+//! machine, its Stores and saved history stay. A forced `machine.stop` on
+//! `native` stops each Sandbox's runtime the same way ([`native_host`],
+//! `Recorded::stop_now`).
+use crate::host::{Deploy, Host, Message};
+use layerfs_bridge::control::{SdkRefusal, SdkValue};
+use layerfs_home::home::{self, paths};
+use layerfs_home::outcome::{failed, refused, Outcome};
+use std::path::Path;
+
+/// The host operation's view of `native` Sandbox `sandbox` from its own
+/// records alone: no Store, project or daemon is resolved.
+pub fn native_host(home: &Path, sandbox: &str) -> Result<Host, SdkRefusal> {
+    Ok(Host {
+        home: home.to_path_buf(),
+        store: String::new(),
+        project: String::new(),
+        sandbox: Some(sandbox.to_owned()),
+        store_record: None,
+        store_path: None,
+        deploy: Some(Deploy {
+            local_dir: Some(paths::run_root(home).display().to_string()),
+            container_prefix: Some(format!("lfs-{}-", &home::opened(home)?.id[..8])),
+            ..Deploy::default()
+        }),
+        postgresql: None,
+        message: Message::Absent,
+        started: None,
+    })
+}
+
+/// The refusal detail field a forced close sets once its runtime stopped.
+pub const RUNTIME_STOPPED: &str = "runtime_stopped";
+
+/// Refuses while a live process holds `native` (a forced `machine.stop` in
+/// progress, for one). An unresolved intent does not stop a forced close,
+/// whether an ended process left it or this process did.
+pub fn native_free(home: &Path) -> Result<(), SdkRefusal> {
+    let Some(machine) = layerfs_machine::record::MachineRecord::read(home, "native")? else {
+        return Ok(());
+    };
+    match machine.op() {
+        Some(op) if op.mine() && op.uncertain() => Ok(()),
+        _ => machine.not_running(),
+    }
+}
+
+/// Forced `sandbox.close` of the Sandbox `host` names, on this process's
+/// kernel: no daemon control call, Store open, active-Workspace refusal or
+/// retry. The backend's Stop checks the runtime identity and ends its
+/// mounts and processes; its Delete independently checks the physical
+/// detach before the runtime goes. Answers each recorded or `held`
+/// Workspace as `unknown`: its uncommitted changes end with the Sandbox.
+pub fn close(host: &Host, held: &[String]) -> Outcome {
+    let id = host.sandbox()?.to_owned();
+    let directory = crate::files::sandbox_dir(host)?;
+    let _guard = layerfs_store::project::discovery::guard(&directory)?;
+    let value = crate::files::started(&directory)?
+        .ok_or_else(|| refused("NoSuchSandbox", format!("no Sandbox {id} is recorded")))?;
+    let mut unknown = crate::files::workspaces(&directory)?
+        .into_iter()
+        .map(|(path, record)| {
+            record
+                .path()
+                .map(str::to_owned)
+                .unwrap_or_else(|| path.display().to_string())
+        })
+        .collect::<Vec<_>>();
+    for path in held {
+        if !unknown.contains(path) {
+            unknown.push(path.to_owned());
+        }
+    }
+    crate::reopen::recorded(host, &id, &value)?.stop_now()?;
+    // The runtime is confirmed stopped from here on: a later failure says so,
+    // so the record shows it stopped instead of as it was.
+    let after_stop = |mut refusal: SdkRefusal| {
+        refusal.detail = home::with_field(refusal.detail, RUNTIME_STOPPED, SdkValue::Bool(true));
+        refusal
+    };
+    crate::reopen::recorded(host, &id, &value)
+        .map_err(after_stop)?
+        .stopped()
+        .delete()
+        .map_err(|error| after_stop(failed(error)))?;
+    crate::registry::SANDBOXES
+        .lock()
+        .map_err(|_| after_stop(refused("Failed", "the Sandbox table's lock was poisoned")))?
+        .remove(&directory);
+    std::fs::remove_dir_all(&directory).map_err(|error| after_stop(failed(error)))?;
+    Ok(SdkValue::map([
+        ("kind", SdkValue::text("Closed")),
+        ("sandbox", SdkValue::text(id)),
+        (
+            "unknown",
+            SdkValue::List(unknown.into_iter().map(SdkValue::Text).collect()),
+        ),
+    ]))
+}
diff --git a/control/crates/layerfs-sandbox/src/get.rs b/control/crates/layerfs-sandbox/src/get.rs
index a85959aaf9..cff92c7f34 100644
--- a/control/crates/layerfs-sandbox/src/get.rs
+++ b/control/crates/layerfs-sandbox/src/get.rs
@@ -112,6 +112,7 @@ pub fn get(home: &Path, events: &mut Events<'_>, request: &GetRequest) -> Result
     // (P4): a machine that is down answers the record with no Workspaces
     // and why
     let listed = match record.id() {
+        _ if record.machine_stopped() => Ok(Ok(SdkValue::Null)),
         // one reached directly lists its daemon's Workspaces from here (#535 G4)
         Some(_) if direct::forwarded(&record) => direct::list(&home, &record),
         Some(_) => reach(
diff --git a/control/crates/layerfs-sandbox/src/lib.rs b/control/crates/layerfs-sandbox/src/lib.rs
index d376e97536..c52fb96ba3 100644
--- a/control/crates/layerfs-sandbox/src/lib.rs
+++ b/control/crates/layerfs-sandbox/src/lib.rs
@@ -32,6 +32,8 @@ pub mod exec;
 pub mod family;
 #[doc(hidden)]
 pub mod files;
+/// The forced close of a `native` Sandbox, from its own records alone.
+pub mod force;
 mod get;
 #[doc(hidden)]
 pub mod host;
diff --git a/control/crates/layerfs-sandbox/src/record/sandbox.rs b/control/crates/layerfs-sandbox/src/record/sandbox.rs
index 1da883f82c..49377d3396 100644
--- a/control/crates/layerfs-sandbox/src/record/sandbox.rs
+++ b/control/crates/layerfs-sandbox/src/record/sandbox.rs
@@ -62,6 +62,8 @@ enum Field {
     RuntimeId(String),
     Daemon(DaemonRecord),
     Left(String),
+    MachineStopped(bool),
+    MachineStopUnknown(bool),
 }
 
 impl fields::Field for Field {
@@ -83,6 +85,14 @@ impl fields::Field for Field {
             "runtime_id" => Self::RuntimeId(text()?),
             "daemon" => Self::Daemon(DaemonRecord::decode(value)?),
             "left" => Self::Left(text()?),
+            "machine_stopped" => Self::MachineStopped(match value {
+                SdkValue::Bool(v) => *v,
+                _ => return None,
+            }),
+            "machine_stop_unknown" => Self::MachineStopUnknown(match value {
+                SdkValue::Bool(v) => *v,
+                _ => return None,
+            }),
             _ => return None,
         })
     }
@@ -104,6 +114,8 @@ impl fields::Field for Field {
             Self::RuntimeId(_) => "runtime_id",
             Self::Daemon(_) => "daemon",
             Self::Left(_) => "left",
+            Self::MachineStopped(_) => "machine_stopped",
+            Self::MachineStopUnknown(_) => "machine_stop_unknown",
         }
     }
 
@@ -112,6 +124,7 @@ impl fields::Field for Field {
             Self::V(n) | Self::Created(n) => SdkValue::Int(*n),
             Self::Op(op) => op.to_record().clone(),
             Self::Daemon(daemon) => daemon.to_record(),
+            Self::MachineStopped(v) | Self::MachineStopUnknown(v) => SdkValue::Bool(*v),
             Self::Name(text)
             | Self::Project(text)
             | Self::Machine(text)
@@ -382,6 +395,23 @@ impl SandboxRecord {
         }))
     }
 
+    /// Whether a provider-confirmed crash stop ended this exact runtime.
+    pub fn machine_stopped(&self) -> bool {
+        self.fields
+            .get(|f| match f {
+                Field::MachineStopped(v) => Some(*v),
+                _ => None,
+            })
+            .unwrap_or(false)
+    }
+
+    /// Record a confirmed stop, or an uncertain attempted stop, symmetrically.
+    pub fn with_machine_stop(mut self, stopped: bool) -> Self {
+        self.fields.set(Field::MachineStopped(stopped));
+        self.fields.set(Field::MachineStopUnknown(!stopped));
+        self
+    }
+
     /// The Sandbox's state (#536 §7.2), from its record, its `op` and,
     /// where the call asked, the daemon's answer (`gone`: its machine
     /// answered that no such Sandbox runs): `creating` or `closing` while
@@ -392,6 +422,19 @@ impl SandboxRecord {
     /// (compute-plane 2.0 P-1: it ended behind control's back; its
     /// `sandbox.close` clears it), else `running`.
     pub fn state(&self, gone: bool) -> &'static str {
+        if self.machine_stopped() {
+            return "gone";
+        }
+        if self
+            .fields
+            .get(|f| match f {
+                Field::MachineStopUnknown(v) => Some(*v),
+                _ => None,
+            })
+            .unwrap_or(false)
+        {
+            return "unresolved";
+        }
         let op = self.op();
         match op.as_ref().map(|op| (op.standing(), op.name())) {
             Some((Standing::Running, "sandbox.close")) => "closing",
diff --git a/control/crates/layerfs-sandbox/src/reopen.rs b/control/crates/layerfs-sandbox/src/reopen.rs
index 6add67b3ee..2ddd38c07c 100644
--- a/control/crates/layerfs-sandbox/src/reopen.rs
+++ b/control/crates/layerfs-sandbox/src/reopen.rs
@@ -283,6 +283,37 @@ fn local_record(_: &Host, id: &str, _: &DaemonRecord) -> Result<Recorded> {
     ))
 }
 impl Recorded {
+    /// Stop an exact recorded runtime without endpoint admission or daemon RPC.
+    pub fn stop_now(self) -> Result<()> {
+        match self.for_stop()? {
+            Some(mut owner) => owner.stop(0).map_err(failed),
+            None => Ok(()),
+        }
+    }
+    /// The exact recorded runtime, reopened for a stop that has not started:
+    /// nothing is signalled here, so a refusal leaves the runtime as it was.
+    /// `None`: the runtime is already stopped.
+    pub fn for_stop(self) -> Result<Option<Box<Sandbox>>> {
+        Ok(Some(match self {
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
+            Self::Local {
+                directory,
+                id,
+                process,
+                port,
+                labels,
+            } => Box::new(Sandbox::HostProcess(
+                host_process::Sandbox::reopen_for_stop(directory, id, process, port, labels)
+                    .map_err(failed)?,
+            )),
+            #[cfg(target_os = "linux")]
+            Self::Namespaces(runtime, id, record) => match runtime.reopen(id, &record) {
+                Ok(owner) => Box::new(owner),
+                Err(error) if error.container_stopped() => return Ok(None),
+                Err(error) => return Err(failed(error)),
+            },
+        }))
+    }
     /// The runtime, observed running once (it starts nothing).
     pub fn reopen(self) -> std::result::Result<Box<Sandbox>, Box<WireFailure>> {
         match self {
diff --git a/control/crates/layerfs-sandbox/src/route.rs b/control/crates/layerfs-sandbox/src/route.rs
index 9db887870d..5a988c6979 100644
--- a/control/crates/layerfs-sandbox/src/route.rs
+++ b/control/crates/layerfs-sandbox/src/route.rs
@@ -250,6 +250,9 @@ fn host_op(
 ) -> Outcome {
     let home = home.to_path_buf();
     let machine = record.machine();
+    if let Some(machine) = layerfs_machine::record::MachineRecord::read(&home, machine)? {
+        machine.not_held()?;
+    }
     let project = record.project().unwrap_or("");
     let target = resolve::target(&home, project)?;
     let Asked {
diff --git a/control/crates/layerfs-sandbox/src/workspace/direct.rs b/control/crates/layerfs-sandbox/src/workspace/direct.rs
index 705b631ecf..5ad11881f7 100644
--- a/control/crates/layerfs-sandbox/src/workspace/direct.rs
+++ b/control/crates/layerfs-sandbox/src/workspace/direct.rs
@@ -31,6 +31,9 @@ pub fn call(
     message: Message,
 ) -> Outcome {
     let home = home.to_path_buf();
+    if let Some(machine) = layerfs_machine::record::MachineRecord::read(&home, record.machine())? {
+        machine.not_held()?;
+    }
     let target = resolve::target(&home, record.project().unwrap_or(""))?;
     let _lease = lease::hold(&home, &target.store)?;
     let host = host(&home, record, &target, message);
diff --git a/control/crates/layerfs-store/src/place.rs b/control/crates/layerfs-store/src/place.rs
index a15c954d5c..b1e9c9994d 100644
--- a/control/crates/layerfs-store/src/place.rs
+++ b/control/crates/layerfs-store/src/place.rs
@@ -47,6 +47,9 @@ pub fn plan(home: &Path, store: &StoreRef) -> Result<Executor> {
 /// Where a call on machine `machine` runs: its Sandboxes' host operations,
 /// and the Store calls of the sqlite Stores on it.
 pub fn on_machine(home: &Path, machine: &str) -> Result<Executor> {
+    if let Some(record) = layerfs_machine::record::MachineRecord::read(home, machine)? {
+        record.not_held()?;
+    }
     if layerfs_machine::runs_here(home, machine)? {
         return Ok(Executor::Local);
     }
diff --git a/control/crates/layerfs-store/src/sqlite/transfer/steps.rs b/control/crates/layerfs-store/src/sqlite/transfer/steps.rs
index 4ecc4981df..ef250836ef 100644
--- a/control/crates/layerfs-store/src/sqlite/transfer/steps.rs
+++ b/control/crates/layerfs-store/src/sqlite/transfer/steps.rs
@@ -61,7 +61,8 @@ fn digest(file: &Path, progress: Option<&mut dyn FnMut(u64, u64)>) -> Outcome {
     ]))
 }
 
-/// Checkpoints and closes a crashed opener's leftover `-wal` beside `file`,
+/// Checkpoints and closes a leftover `-wal` beside `file` (a crashed opener's,
+/// or one whose close could not wait for the writer lane),
 /// through the same `seal` a graceful close takes (#86): a Store whose last
 /// opener was killed keeps its committed frames in the log, which blocks a
 /// snapshot until one opener folds them in. Nothing runs while no live
diff --git a/control/crates/layerfs-store/tests/lease_race.rs b/control/crates/layerfs-store/tests/lease_race.rs
new file mode 100644
index 0000000000..7a3a38bc07
--- /dev/null
+++ b/control/crates/layerfs-store/tests/lease_race.rs
@@ -0,0 +1,56 @@
+//! A holder can release its lease after stale inspection and before removal.
+#![cfg(unix)]
+use layerfs_home::home::{self, paths};
+use layerfs_store::{lease, record};
+use nix::sys::stat::Mode;
+use std::fs::OpenOptions;
+use std::io::Write;
+use std::os::unix::fs::OpenOptionsExt;
+use std::sync::mpsc;
+use std::time::{Duration, Instant};
+
+#[test]
+fn stale_lease_released_after_inspection_is_already_gone() {
+    let home = std::env::temp_dir().join(format!("layerfs-lease-race-{}", std::process::id()));
+    let directory = paths::lease_dir(&home, "shop");
+    home::private_dir(&directory).unwrap();
+    let path = directory.join("4294967295");
+    nix::unistd::mkfifo(&path, Mode::S_IRUSR | Mode::S_IWUSR).unwrap();
+    let (send, recv) = mpsc::channel();
+    let root = home.clone();
+    let reader = std::thread::spawn(move || {
+        let _lock = record::lock(&root, "shop").unwrap();
+        send.send(lease::holders(&root, "shop")).unwrap();
+    });
+    // Opening the FIFO for writing succeeds only after holders opened its read
+    // descriptor. Unlink before supplying EOF: its snapshot then names a lease
+    // already released by its holder, without timing guesses or product hooks.
+    let until = Instant::now() + Duration::from_secs(5);
+    let mut writer = loop {
+        match OpenOptions::new()
+            .write(true)
+            .custom_flags(libc::O_NONBLOCK)
+            .open(&path)
+        {
+            Ok(file) => break file,
+            Err(error) if error.raw_os_error() == Some(libc::ENXIO) => {
+                assert!(Instant::now() < until, "lease inspection never opened FIFO");
+                std::thread::sleep(Duration::from_millis(5));
+            }
+            Err(error) => panic!("open lease FIFO: {error}"),
+        }
+    };
+    std::fs::remove_file(&path).unwrap();
+    writer
+        .write_all(b"0 0 00000000000000000000000000000000\n")
+        .unwrap();
+    drop(writer);
+    let result = recv.recv_timeout(Duration::from_secs(5)).unwrap();
+    reader.join().unwrap();
+    assert!(result.unwrap().is_empty());
+    assert!(
+        !home.join("log/leases.jsonl").exists(),
+        "no removal was performed to log"
+    );
+    std::fs::remove_dir_all(home).unwrap();
+}
diff --git a/docs/architecture/api/sdk/machine.md b/docs/architecture/api/sdk/machine.md
index b66f8a5e7e..a73b553941 100644
--- a/docs/architecture/api/sdk/machine.md
+++ b/docs/architecture/api/sdk/machine.md
@@ -33,7 +33,7 @@ The `machine` calls: list, inspect, create, start, stop, update and remove the m
 | [`machine.get`](#machineget) | summary |
 | [`machine.status`](#machinestatus) | `{ name, platform, state, free_disk, tool_version, down_reason, stores, op? }` |
 | [`machine.start`](#machinestart) | summary, once the worker answers |
-| [`machine.stop`](#machinestop) | summary, `state: "stopped"` |
+| [`machine.stop`](#machinestop) | summary, `state: "stopped"`; with `force`, also `forced`, `stopped_sandboxes`, `unknown` |
 | [`machine.create`](#machinecreate) | summary, `state: "running"` |
 | [`machine.remove`](#machineremove) | `{ name, removed: true, note? }` |
 | [`machine.update`](#machineupdate) | summary |
@@ -213,18 +213,46 @@ Stops a machine's worker and, where it has one, its VM; disks and Stores
 stay.
 
 ```text
-JS      await layerfs.machine.stop(name)
-Python  layerfs.machine.stop(name)
-CLI     layerfs machine stop <name>
+JS      await layerfs.machine.stop(name, { force })
+Python  layerfs.machine.stop(name, *, force=None)
+CLI     layerfs machine stop <name> [--force]
 ```
 
 `native` stops nothing; `docker` ends its worker container only; `lima` and
 `apple-container` end the worker and VM. `worker.json` is deleted and the
 state recorded `stopped`. Nothing stops a machine because it is unused.
-Errors: `InUse` while Sandboxes are recorded on it (`detail.related`,
-`can_stop_machine`); `Unavailable` (`start.lock` held 600 s, or the provider did
-not answer: records stay as they were); `Unknown` (a provider command timed out:
-the machine may be stopped while still recorded with its worker).
+Errors: `InUse` while Sandboxes are recorded on it and `force` is not set
+(`detail.related`, `can_stop_machine`); `InvalidInput` on `docker` when the
+container named for the machine is not a LayerFS worker (its role label), which
+is left alone; `Unavailable` (`start.lock` held 600 s,
+or the provider did not answer: records stay as they were); `Unknown` (a
+provider command timed out: the machine may be stopped while still recorded
+with its worker).
+
+`force` (Bool, default false) stops the machine while Sandboxes, mounted
+Workspaces or commands are active; their uncommitted state may be lost. It goes
+straight to the provider: it does not start the machine, call its worker,
+commit, unmount Workspaces one by one, or delete any machine, Sandbox, Store or
+disk record. On `native` it ends only this home's recorded Sandbox runtimes and
+commands, without opening the Store or calling the daemon, and the machine
+stays `ready`. On `docker`, ending the worker container ends its Sandboxes; the
+Store volume stays. `lima` uses `limactl stop --force`, also when the host
+agent reports a broken VM. It returns the recorded summary without a worker
+probe, plus `forced: true`, `stopped_sandboxes: [name]` and
+`unknown: [workspace]` (`sandbox:*` when a worker-side Workspace frontier could
+not be listed; these are not saved Commits). Stopped Sandboxes are recorded
+`gone` for explicit reclaim, or `unresolved` when the stop could not be
+confirmed. A Sandbox that never started is neither stopped nor listed. On `native` a Sandbox is uncertain only once its stop has been
+sent: a refusal before that (no recorded runtime, or a recorded process
+identity that no longer matches) leaves that Sandbox as it was, and the stop
+goes on to the next one. The call then answers the first such refusal, with
+`detail.refused_sandboxes`, `stopped_sandboxes` and `unknown`, and leaves no op
+on `native`. If recording the outcome fails, the stop's own refusal is still
+answered, with `detail.custody_errors`. A later forced stop may settle an
+uncertain stop this process left, or one whose process has ended; one a live
+process still holds is `Busy`. A provider failure or uncertain outcome is
+reported, never replayed or turned into success
+([`machine_force.rs`](../../../../control/crates/layerfs-control/src/machine_force.rs)).
 
 ### machine.create
 
diff --git a/docs/architecture/api/sdk/sandbox.md b/docs/architecture/api/sdk/sandbox.md
index 20fd9788d7..166f653851 100644
--- a/docs/architecture/api/sdk/sandbox.md
+++ b/docs/architecture/api/sdk/sandbox.md
@@ -148,6 +148,16 @@ CLI     layerfs sandbox close <name> [--force]
 running command is cancelled first (5000 ms grace), and uncommitted changes
 and processes end with the Sandbox
 ([`exec/cancel.rs`](../../../../control/crates/layerfs-sandbox/src/exec/cancel.rs) `settle`).
+On `native`, a forced close works from the Sandbox's own records instead: it
+opens no Store and calls no daemon, checks the runtime's identity, ends its
+mounts and processes, and answers every recorded or held Workspace as
+`unknown` ([`force.rs`](../../../../control/crates/layerfs-sandbox/src/force.rs)).
+It does not settle commands one by one or check mounts first: the runtime's
+stop ends them. It is `Busy` while a live process holds `native` (a forced
+`machine.stop`, for one), but not for an unresolved intent, whether an ended
+process or this one left it. If the
+runtime stopped and a later step (delete, folder removal) failed, the refusal
+carries `detail.runtime_stopped: true` and the record is marked stopped.
 
 | Case | Answer |
 | --- | --- |
diff --git a/docs/architecture/control/sandboxes/README.md b/docs/architecture/control/sandboxes/README.md
index 6d420a489e..bc5bd7ce38 100644
--- a/docs/architecture/control/sandboxes/README.md
+++ b/docs/architecture/control/sandboxes/README.md
@@ -64,6 +64,7 @@ runtime owns command launch, streams, exit status and cancel
 | Direct route | A Sandbox whose daemon port its machine forwards to this host (`lima`'s `ssh -L`) keeps its keys in the home's `control.json` (0600); its Workspace calls dial the daemon from the home, while exec still goes through the worker. | [`direct.rs`](../../../../control/crates/layerfs-sandbox/src/direct.rs) |
 | `sandbox.get` capacity | Reads the daemon's `backend.daemon.connection_failures` on a lent connection, bounded to 2 s and fenced at that deadline. | [`daemon/capacity.rs`](../../../../control/crates/layerfs-sandbox/src/daemon/capacity.rs), [`get.rs`](../../../../control/crates/layerfs-sandbox/src/get.rs) |
 | Retired docker records | A record whose `backend` or `runtime` is `docker`, or a started docker-machine record with no `runtime`: every call but `sandbox.close` is `Unsupported`, naming `docker rm -f <container>`; `sandbox.close` forgets it. | [`record/sandbox.rs`](../../../../control/crates/layerfs-sandbox/src/record/sandbox.rs) `retired_container`, [`reopen.rs`](../../../../control/crates/layerfs-sandbox/src/reopen.rs) `retired`, [`close.rs`](../../../../control/crates/layerfs-sandbox/src/close.rs) `forget` |
+| Machine-stop fields (record format) | A forced `machine.stop` adds two Bool fields to the Sandbox's `record.json`: `machine_stopped` (its runtime was stopped with the machine; shown `gone`) and `machine_stop_unknown` (the stop was not confirmed; shown `unresolved`). An absent field reads `false`, so older records need no change and `"v"` stays `1`. A build older than these fields keeps them as read but shows such a Sandbox as it was before the stop. | [`record/sandbox.rs`](../../../../control/crates/layerfs-sandbox/src/record/sandbox.rs) `machine_stopped`, `with_machine_stop`; [`machine_force.rs`](../../../../control/crates/layerfs-control/src/machine_force.rs) `mark` |
 
 ## Invariants
 
diff --git a/docs/architecture/engine/store/sqlite.md b/docs/architecture/engine/store/sqlite.md
index f2a04f2a8e..855eba95c0 100644
--- a/docs/architecture/engine/store/sqlite.md
+++ b/docs/architecture/engine/store/sqlite.md
@@ -56,6 +56,7 @@ UPDATE and DELETE on `pack` and `pack_unit`. Open checks `application_id`,
 | Guard | `busy_timeout=0`; `SQLITE_DBCONFIG_DEFENSIVE` on | same |
 | Settings | `profile::apply`, then read back into `ConnectionProfile`; a mismatch is `Integrity`; an existing Store not in `wal` mode is refused | [`profile.rs`](../../../../engine/crates/layerfs-persistence/src/backend/sqlite/session/profile.rs) |
 | Checkpoint thread | writable Disposable session only: a second connection on its own thread; failure to start it refuses the open | [`background.rs`](../../../../engine/crates/layerfs-persistence/src/backend/sqlite/session/background.rs) |
+| Last close | A writable session's close waits for `Writer::acquire(path)` like any write (or reuses the lane an uncertain transaction already holds), so its close, which may be SQLite's last, checkpoints and removes `-wal`/`-shm` under the lane; a pull's verify close and a restore's in-use check rely on that. If the wait itself fails (interrupted, or the directory cannot be opened), the close sets `SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE` first, so SQLite takes no exclusive lock, runs no checkpoint and removes nothing; there is no second attempt. If even that setting fails, the connection is kept until process exit. A read-only or diagnostic session cannot checkpoint or remove those files and takes no lane, so it never waits behind a writer kept after an uncertain write. The checkpoint thread's connection closes first and never takes it. One exception: an open whose checkpoint thread fails to start drops its new connection after the open's lane is released | [`lifecycle.rs`](../../../../engine/crates/layerfs-persistence/src/backend/sqlite/session/lifecycle.rs) |
 
 | Setting | Disposable | Source |
 | --- | --- | --- |
diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/lifecycle/seal.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/lifecycle/seal.rs
index e7259fb616..f9a02f37c0 100644
--- a/engine/crates/layerfs-persistence/src/backend/sqlite/lifecycle/seal.rs
+++ b/engine/crates/layerfs-persistence/src/backend/sqlite/lifecycle/seal.rs
@@ -12,7 +12,7 @@ impl Session {
         }
         // The checkpoint thread ends and closes its connection first.
         drop(self.background.take());
-        let state = self
+        let mut state = self
             .state
             .into_inner()
             .map_err(|_| PersistenceError::Uncertain)?;
@@ -22,13 +22,10 @@ impl Session {
         // The same advisory lane as a write transaction: a seal is the sole
         // owner's offline handoff, so no other opener may write while the
         // checkpoint, close and sidecar removal below run.
-        #[cfg(any(target_os = "linux", target_os = "macos"))]
-        let state = {
-            let mut state = state;
+        if state.writer_hold.is_none() {
             state.writer_hold =
                 Some(crate::backend::sqlite::lane::writer_admission::Writer::acquire(&self.path)?);
-            state
-        };
+        }
         #[cfg(target_os = "macos")]
         let identity = super::seal_allocation::Identity::capture(&self.path)?;
         #[cfg(target_os = "macos")]
@@ -43,10 +40,11 @@ impl Session {
         if row.get::<i64>(0)? != 0 {
             return Err(PersistenceError::Busy);
         }
-        state
-            .connection
-            .close()
-            .map_err(|(_, error)| rows::error(error))?;
+        let connection = state.connection.take().ok_or(PersistenceError::Malformed)?;
+        if let Err((connection, error)) = connection.close() {
+            state.connection.restore(connection);
+            return Err(rows::error(error).into());
+        }
         // SQLite has checked-closed before advisory ownership is released.
         // Earlier error exits drop State.connection before its retained owner.
         // The sidecar check below runs while the lane is still held: an opener
@@ -70,7 +68,7 @@ impl Session {
                 }
             }
         }
-        if let Some(writer) = state.writer_hold {
+        if let Some(writer) = state.writer_hold.take() {
             writer.release().map_err(|_| PersistenceError::Uncertain)?;
         }
         #[cfg(target_os = "macos")]
diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/session/background.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/session/background.rs
index e4981e609f..6efb4375cb 100644
--- a/engine/crates/layerfs-persistence/src/backend/sqlite/session/background.rs
+++ b/engine/crates/layerfs-persistence/src/backend/sqlite/session/background.rs
@@ -103,6 +103,9 @@ impl Checkpointer {
 impl Drop for Checkpointer {
     /// Ends the thread after its current pass and closes its connection,
     /// before the session's own connection closes.
+    /// This connection is therefore never SQLite's last one: State still owns
+    /// the main connection. Do not reacquire its directory lane here; a
+    /// quarantined State may already retain it while this join completes.
     fn drop(&mut self) {
         self.wake.take();
         if let Some(thread) = self.thread.take() {
diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/session/connection.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/session/connection.rs
index 3149ee5380..c080697dae 100644
--- a/engine/crates/layerfs-persistence/src/backend/sqlite/session/connection.rs
+++ b/engine/crates/layerfs-persistence/src/backend/sqlite/session/connection.rs
@@ -145,7 +145,11 @@ pub struct SqlWork {
     pub connects: u64,
 }
 pub(crate) struct State {
-    pub(crate) connection: Connection,
+    pub(crate) connection: super::lifecycle::ConnectionSlot,
+    /// The Store path whose lane a writable session takes around its last
+    /// close; `None` for read-only and diagnostic sessions, whose close cannot
+    /// checkpoint or remove `-wal`/`-shm` and so takes no lane.
+    pub(crate) close_lane: Option<PathBuf>,
     pub(crate) quarantined: bool,
     pub(crate) work: RefCell<SqlWork>,
     pub(crate) writer_hold: Option<crate::backend::sqlite::lane::writer_admission::Writer>,
@@ -333,7 +337,8 @@ impl Session {
             admission: crate::backend::admission::Admission::default(),
             path: path.to_path_buf(),
             state: Mutex::new(State {
-                connection,
+                connection: super::lifecycle::ConnectionSlot::new(connection),
+                close_lane: writable.then(|| path.to_path_buf()),
                 quarantined: false,
                 work,
                 writer_hold: None,
diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/session/lifecycle.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/session/lifecycle.rs
new file mode 100644
index 0000000000..20a22f5e84
--- /dev/null
+++ b/engine/crates/layerfs-persistence/src/backend/sqlite/session/lifecycle.rs
@@ -0,0 +1,78 @@
+//! Connection custody at the last Session close, including quarantined owners.
+use super::connection::State;
+use crate::backend::sqlite::lane::writer_admission::Writer;
+use rusqlite::config::DbConfig;
+use rusqlite::Connection;
+use std::ops::Deref;
+
+/// Allows checked seal and last-owner Drop to close before releasing admission.
+pub(crate) struct ConnectionSlot(Option<Connection>);
+
+impl ConnectionSlot {
+    pub(crate) fn new(connection: Connection) -> Self {
+        Self(Some(connection))
+    }
+
+    pub(crate) fn take(&mut self) -> Option<Connection> {
+        self.0.take()
+    }
+
+    pub(crate) fn restore(&mut self, connection: Connection) {
+        self.0 = Some(connection);
+    }
+
+    fn is_closed(&self) -> bool {
+        self.0.is_none()
+    }
+}
+
+impl Deref for ConnectionSlot {
+    type Target = Connection;
+
+    fn deref(&self) -> &Connection {
+        self.0.as_ref().expect("live Session connection")
+    }
+}
+
+impl Drop for State {
+    fn drop(&mut self) {
+        if self.connection.is_closed() {
+            return;
+        }
+        let writer = match (self.writer_hold.take(), &self.close_lane) {
+            // An uncertain transaction already owns the lane. Taking it again
+            // through an independent directory descriptor would deadlock.
+            (Some(writer), _) => Some(writer),
+            // A read-only or diagnostic connection cannot checkpoint or remove
+            // the WAL and SHM files, so its close takes no lane.
+            (None, None) => None,
+            // A writable close waits for the lane like any write: its close
+            // may be SQLite's last, which checkpoints and removes the WAL and
+            // SHM files that a pull's verify and a restore's in-use check
+            // rely on. If the wait itself fails (interrupted, or the Store's
+            // directory cannot be opened), the close turns SQLite's
+            // checkpoint-on-close off first, so it takes no exclusive lock and
+            // removes nothing; there is no second attempt at the lane.
+            (None, Some(path)) => match Writer::acquire(path) {
+                Ok(writer) => Some(writer),
+                Err(_) => {
+                    let quiet = self
+                        .connection
+                        .set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, true);
+                    if quiet.is_err() {
+                        // Drop cannot report it: retain the connection until
+                        // process exit rather than tear the WAL down unadmitted.
+                        std::mem::forget(self.connection.take());
+                        return;
+                    }
+                    None
+                }
+            },
+        };
+        // The background connection has already ended; a writable close that
+        // holds the lane may be SQLite's last and checkpoint/delete the WAL
+        // and SHM files.
+        drop(self.connection.take());
+        drop(writer);
+    }
+}
diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/session/mod.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/session/mod.rs
index 2c5b88c914..4127ec2f0b 100644
--- a/engine/crates/layerfs-persistence/src/backend/sqlite/session/mod.rs
+++ b/engine/crates/layerfs-persistence/src/backend/sqlite/session/mod.rs
@@ -2,6 +2,7 @@
 pub(crate) mod background;
 pub(crate) mod checkpoint;
 pub(crate) mod connection;
+pub(crate) mod lifecycle;
 pub(crate) mod prepared;
 pub(crate) mod profile;
 pub(crate) mod query;
diff --git a/engine/crates/layerfs-persistence/tests/sqlite_lock_custody.rs b/engine/crates/layerfs-persistence/tests/sqlite_lock_custody.rs
index e1d919c135..2c9c480002 100644
--- a/engine/crates/layerfs-persistence/tests/sqlite_lock_custody.rs
+++ b/engine/crates/layerfs-persistence/tests/sqlite_lock_custody.rs
@@ -1,6 +1,6 @@
 //! Cooperative readiness must preserve SQLite's own POSIX lock custody.
 //! Uses public Stores and an independently bounded raw SQLite writer process.
-#![cfg(target_os = "linux")]
+#![cfg(any(target_os = "linux", target_os = "macos"))]
 mod support;
 use layerfs_content::{FinalizedObject, ObjectRole};
 use layerfs_history::HistoryCatalogConfig;
diff --git a/engine/crates/layerfs-persistence/tests/writer_admission.rs b/engine/crates/layerfs-persistence/tests/writer_admission.rs
index 77a90b227d..dcb888af97 100644
--- a/engine/crates/layerfs-persistence/tests/writer_admission.rs
+++ b/engine/crates/layerfs-persistence/tests/writer_admission.rs
@@ -168,6 +168,8 @@ impl Held {
                 }
             }
         };
+        // Darwin inherits the listener's nonblocking flag on accepted sockets.
+        socket.set_nonblocking(false).unwrap();
         let mut held = Self { child, socket };
         held.socket.set_nonblocking(false).unwrap();
         held.socket
@@ -350,6 +352,71 @@ fn another_process_admission_precedes_the_single_sqlite_begin() {
     drop(original);
 }
 
+#[test]
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn writable_last_close_waits_for_the_lane_then_removes_the_wal() {
+    // A pull's verify and a restore's in-use check rely on the last close
+    // folding the WAL in: a writable close waits for a held lane instead of
+    // leaving the WAL behind.
+    let temp = support::Temp::new("wa-close");
+    let path = temp.join("store.sqlite");
+    let wal = temp.join("store.sqlite-wal");
+    let handles = create(&path);
+    assert!(wal.exists(), "the test needs a WAL file before the close");
+    let held = Held::start(&path, &temp.join("s"));
+    let (result_tx, result_rx) = mpsc::channel();
+    let closer = std::thread::spawn(move || {
+        drop(handles);
+        result_tx.send(()).unwrap();
+    });
+    let premature = result_rx.recv_timeout(Duration::from_millis(200)).ok();
+    let wal_kept = wal.exists();
+    held.release();
+    if premature.is_none() {
+        result_rx.recv_timeout(Duration::from_secs(3)).unwrap();
+    }
+    closer.join().unwrap();
+    assert!(
+        premature.is_none(),
+        "a writable close bypassed the held lane"
+    );
+    assert!(wal_kept, "the WAL went before the close was admitted");
+    assert!(!wal.exists(), "the admitted last close left the WAL");
+}
+
+#[test]
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn read_only_close_does_not_wait_for_a_held_writer_lane() {
+    // A writer kept after an uncertain write holds the directory lane until
+    // its session goes; a reader's close must still return meanwhile.
+    let temp = support::Temp::new("wa-ro-close");
+    let path = temp.join("store.sqlite");
+    let original = create(&path);
+    let reader = SqliteHandles::open_read_only(
+        config(&path),
+        &authority().binding_key,
+        authority().cursor_key,
+    )
+    .unwrap();
+    let held = Held::start(&path, &temp.join("s"));
+    let (result_tx, result_rx) = mpsc::channel();
+    let closer = std::thread::spawn(move || {
+        drop(reader);
+        result_tx.send(()).unwrap();
+    });
+    let closed = result_rx.recv_timeout(Duration::from_secs(2)).ok();
+    held.release();
+    if closed.is_none() {
+        result_rx.recv_timeout(Duration::from_secs(3)).unwrap();
+    }
+    closer.join().unwrap();
+    assert!(
+        closed.is_some(),
+        "a read-only close waited for the writer lane"
+    );
+    drop(original);
+}
+
 #[test]
 #[cfg(any(target_os = "linux", target_os = "macos"))]
 fn two_opened_handles_admit_eight_distinct_reservations_without_conflict() {