diff --git a/crates/harness-runtime/src/human.rs b/crates/harness-runtime/src/human.rs
index 0f0a733..ea708ec 100644
--- a/crates/harness-runtime/src/human.rs
+++ b/crates/harness-runtime/src/human.rs
@@ -614,21 +614,13 @@ fn rollback(harness: &Harness, prefix: &Path, to: Option<&str>) -> Result<()> {
         }));
     };
 
-    if present.exposed.as_deref() == Some(version) {
-        println!(
-            "{} already runs {version}; nothing to do.",
-            declared.command
-        );
-        return Ok(());
-    }
-
     let rolled = setup_core::software::rollback(&declared, prefix, version)?;
     println!(
         "{} now runs {}.",
         rolled.executable.display(),
         rolled.version
     );
-    if let Some(previous) = present.exposed {
+    if let Some(previous) = present.exposed.filter(|previous| previous != version) {
         println!("  it ran {previous} before this, and that tree is still here");
     }
     Ok(())
diff --git a/crates/harness-runtime/src/software.rs b/crates/harness-runtime/src/software.rs
index c450450..f08f5c7 100644
--- a/crates/harness-runtime/src/software.rs
+++ b/crates/harness-runtime/src/software.rs
@@ -29,9 +29,9 @@
 //! evict every configuration backup the target had.
 //!
 //! Software operations hold a prefix lock and revalidate the plan's bounded
-//! content observation before recovery or effects. Install stages the version
-//! before exposing its entry point. A per-command journal owns staging recovery;
-//! installed-version ownership and exact-plan replay remain separate boundaries.
+//! content observation before effects. Install stages the version before exposing
+//! its entry point. A pending per-command journal refuses new operations; implicit
+//! recovery would perform effects that the new plan did not authorize.
 
 use std::fs;
 use std::path::{Path, PathBuf};
@@ -244,7 +244,7 @@ fn acquire_prefix(
     root: &Path,
     operation: Operation,
     expected_software_digest: &str,
-) -> Result<setup_core::lock::TargetLock> {
+) -> Result<(setup_core::lock::TargetLock, software::Writer)> {
     // A plan binds the bytes below this prefix, not just its absolute name.
     // Check before creating bookkeeping, then again under the writer lock.
     let before = setup_core::software_prefix::observe(root, harness.control_directory)?;
@@ -278,6 +278,10 @@ fn acquire_prefix(
         true
     };
 
+    // Contention must refuse before this provider creates any bookkeeping that
+    // another command's prefix observation would include as an unrelated change.
+    let writer = software::Writer::acquire(root)?;
+
     // The lock lives in this provider's own dotted directory inside the prefix,
     // not at its root. `acquire` takes a control directory, and a `target.lock`
     // sitting beside `bin/` in a program directory would be both misnamed and
@@ -299,7 +303,7 @@ fn acquire_prefix(
     }
     guard.annotate(&format!("{} {operation}", harness.provider_id))?;
 
-    Ok(guard)
+    Ok((guard, writer))
 }
 
 fn installation_input(
@@ -395,12 +399,13 @@ pub(crate) fn apply(
                 "software_remove downloads nothing, so it takes no --software-artifact",
             ));
         }
-        let _guard = acquire_prefix(harness, &root, operation, expected_software_digest)?;
-        let recovered = setup_core::software::recover(&declared, &root)?;
-        let removed = software::remove(&declared, &root)?;
+        software::require_idle(&declared, &root)?;
+        let (_guard, writer) = acquire_prefix(harness, &root, operation, expected_software_digest)?;
+        software::require_idle(&declared, &root)?;
+        let removed = writer.remove(&declared)?;
         return Ok(serde_json::json!({
             "state": "verified",
-            "recovered": recovered,
+            "recovered": [],
             "operation": operation.as_str(),
             "command": declared.command,
             "version": declared.version,
@@ -411,11 +416,11 @@ pub(crate) fn apply(
     let (artifact, input) =
         installation_input(&declared, operation, planned_artifacts, downloaded)?;
 
-    // Reject incomplete or incorrect inputs before creating prefix bookkeeping
-    // or recovering any previously observed staging. The kernel consumes this
-    // same held input and verifies the extraction stream before promotion.
-    // Recovery only touches physical directories named by its staging journal.
-    let _guard = acquire_prefix(harness, &root, operation, expected_software_digest)?;
+    // Validate the held input before any bookkeeping. A pending transaction
+    // refuses before the lock and again under it; a new plan owns no recovery.
+    software::require_idle(&declared, &root)?;
+    let (_guard, writer) = acquire_prefix(harness, &root, operation, expected_software_digest)?;
+    software::require_idle(&declared, &root)?;
 
     // Re-checked here, not trusted from the plan: applying happens later, and
     // the prefix could have been emptied in between. The plan's digest binds
@@ -436,12 +441,11 @@ pub(crate) fn apply(
         ));
     }
 
-    let recovered = setup_core::software::recover(&declared, &root)?;
-    let installed = software::install_verified(&declared, artifact, input, &root)?;
+    let installed = writer.install_verified(&declared, artifact, input)?;
 
     Ok(serde_json::json!({
         "state": "verified",
-        "recovered": recovered,
+        "recovered": [],
         "operation": operation.as_str(),
         "command": declared.command,
         "version": installed.version,
diff --git a/crates/harness-runtime/src/wire.rs b/crates/harness-runtime/src/wire.rs
index 916157b..8a10471 100644
--- a/crates/harness-runtime/src/wire.rs
+++ b/crates/harness-runtime/src/wire.rs
@@ -9119,7 +9119,19 @@ mod tests {
         // took of the thing this provider actually owns.
         let target = seeded("software-slots");
         let file = downloaded(&target, TEST_PAYLOAD);
-        plan_then_install(&target, "software_install", Some(&file));
+        let prefix = ready_prefix(&target);
+        let writer = setup_core::software::Writer::acquire(Path::new(&prefix)).unwrap();
+        let planned = software_plan(&target, "software_install");
+        let extra = apply_args(&target, &prefix, "software_install", &planned, Some(&file));
+        let borrowed: Vec<&str> = extra.iter().map(String::as_str).collect();
+        let error = refuse(args("apply-operation", &target, &borrowed));
+        assert_eq!(error.reason(), Some(WireReason::ProviderUnavailable));
+        assert!(!Path::new(&prefix).join(TEST.control_directory).exists());
+        drop(writer);
+        assert_eq!(
+            apply_planned(&target, &prefix, "software_install", &planned, Some(&file))["state"],
+            "verified"
+        );
 
         let slots = target.join(TEST.control_directory).join("backups");
         let taken = fs::read_dir(&slots).map_or(0, Iterator::count);
@@ -9127,6 +9139,21 @@ mod tests {
             taken, 0,
             "a software install captured a configuration backup"
         );
+
+        // A matching version marker must not hide a foreign human rollback launcher.
+        let exposed = Path::new(&prefix).join("bin/test-harness");
+        fs::remove_file(&exposed).unwrap();
+        fs::write(&exposed, b"foreign launcher").unwrap();
+        let error = crate::human::run(
+            &TEST,
+            crate::human::Command::Rollback {
+                prefix: PathBuf::from(&prefix),
+                to: Some(TEST.software.unwrap().version.to_owned()),
+            },
+        )
+        .unwrap_err();
+        assert_eq!(error.reason(), Some(WireReason::RecoveryRequired));
+        assert_eq!(fs::read(exposed).unwrap(), b"foreign launcher");
     }
 
     #[test]
diff --git a/crates/setup-core/src/lock.rs b/crates/setup-core/src/lock.rs
index b6f5e56..c1db24d 100644
--- a/crates/setup-core/src/lock.rs
+++ b/crates/setup-core/src/lock.rs
@@ -68,6 +68,7 @@ fn release_in_process(path: &Path) {
 pub struct TargetLock {
     file: File,
     path: PathBuf,
+    identity: (u64, u64),
 }
 
 impl TargetLock {
@@ -87,8 +88,8 @@ impl TargetLock {
             ));
         }
 
-        let acquire_os_lock = || -> Result<File> {
-            let open_lock = || -> std::io::Result<File> {
+        let acquire_os_lock = || -> Result<(File, (u64, u64))> {
+            let open_lock = || -> std::io::Result<(File, (u64, u64))> {
                 let parent = control_directory
                     .parent()
                     .ok_or_else(|| std::io::Error::other("control directory has no parent"))?;
@@ -117,9 +118,9 @@ impl TargetLock {
                 if !metadata.is_file() || metadata.nlink() != 1 {
                     return Err(std::io::Error::other("lock is not a regular file"));
                 }
-                Ok(file.into_std())
+                Ok((file.into_std(), (metadata.dev(), metadata.ino())))
             };
-            let file = open_lock().map_err(|source| {
+            let (file, identity) = open_lock().map_err(|source| {
                 Error::new(
                     ReasonCode::StateUnavailable,
                     format!("cannot open lock file {}", path.display()),
@@ -133,11 +134,15 @@ impl TargetLock {
                 )
                 .with_source(source)
             })?;
-            Ok(file)
+            Ok((file, identity))
         };
 
         match acquire_os_lock() {
-            Ok(file) => Ok(Self { file, path }),
+            Ok((file, identity)) => Ok(Self {
+                file,
+                path,
+                identity,
+            }),
             Err(error) => {
                 // The in-process claim must not outlive a failed acquisition, or
                 // the next attempt would be refused by this process forever.
@@ -153,6 +158,43 @@ impl TargetLock {
         &self.path
     }
 
+    /// Verify that the named lock still refers to this held regular file.
+    ///
+    /// # Errors
+    /// Refuses an inaccessible, replaced, linked or aliased lock entry.
+    pub fn revalidate(&self) -> Result<()> {
+        let check = || -> std::io::Result<()> {
+            let parent = self
+                .path
+                .parent()
+                .ok_or_else(|| std::io::Error::other("lock has no parent"))?;
+            let directory = Dir::open_ambient_dir(parent, cap_std::ambient_authority())?;
+            let named = directory.symlink_metadata(LOCK_FILE_NAME)?;
+            if !named.is_file()
+                || named.is_symlink()
+                || named.nlink() != 1
+                || (named.dev(), named.ino()) != self.identity
+            {
+                return Err(std::io::Error::other("the held lock entry changed"));
+            }
+            #[cfg(windows)]
+            {
+                use cap_std::fs::MetadataExt as _;
+                if named.file_attributes() & 0x400 != 0 {
+                    return Err(std::io::Error::other("the lock is a reparse point"));
+                }
+            }
+            Ok(())
+        };
+        check().map_err(|source| {
+            Error::new(
+                ReasonCode::StateUnavailable,
+                "the prefix writer lock was replaced",
+            )
+            .with_source(source)
+        })
+    }
+
     /// Record a non-secret owner note inside the lock file.
     ///
     /// The note is diagnostic only. Nothing reads it to make a decision, because
diff --git a/crates/setup-core/src/software.rs b/crates/setup-core/src/software.rs
index 3a073af..e64f869 100644
--- a/crates/setup-core/src/software.rs
+++ b/crates/setup-core/src/software.rs
@@ -16,14 +16,21 @@
 //! program can serve several targets, which a program living inside one of them
 //! could not.
 
+mod cleanup;
 mod exposure;
 mod input;
+mod launch;
 mod ownership;
 mod records;
+mod removal;
 mod staging;
+mod switch;
+pub(crate) mod writer;
 
 pub use input::VerifiedArtifact;
+pub use writer::Writer;
 
+#[cfg(test)]
 use exposure::expose;
 
 use std::fs;
@@ -519,6 +526,19 @@ pub fn install_verified(
     mut source: VerifiedArtifact,
     root: &Path,
 ) -> Result<Installed> {
+    require_idle(software, root)?;
+    source.begin(artifact)?;
+    records::io(fs::create_dir_all(root))?;
+    Writer::acquire(root)?.install_verified(software, artifact, source)
+}
+
+fn install_locked(
+    software: &Software,
+    artifact: &Artifact,
+    mut source: VerifiedArtifact,
+    root: &Path,
+) -> Result<Installed> {
+    require_idle(software, root)?;
     source.begin(artifact)?;
     let version_root = root.join(software.version);
 
@@ -537,11 +557,10 @@ pub fn install_verified(
     let staging = transaction.stage_path();
     let output = transaction.destination()?;
 
-    let (executable, files) = match artifact.shape {
+    let files = match artifact.shape {
         Shape::Raw => {
-            let placed = staging.join(software.command);
             archive::place_executable_into(source.reader(), Path::new(software.command), &output)?;
-            (placed, 1)
+            1
         }
         Shape::GzipTar | Shape::Zip => {
             let entries = if artifact.shape == Shape::Zip {
@@ -568,7 +587,7 @@ pub fn install_verified(
                     ),
                 ));
             }
-            (staging.join(artifact.member), entries.len())
+            entries.len()
         }
     };
 
@@ -578,16 +597,10 @@ pub fn install_verified(
 
     transaction.promote()?;
 
-    let executable = version_root.join(
-        executable
-            .strip_prefix(&staging)
-            .unwrap_or_else(|_| Path::new(software.command)),
-    );
-
     let exposed = root
         .join("bin")
         .join(exposed_name(software.command, artifact.member));
-    expose(&executable, &exposed, software.version, software.command)?;
+    transaction.expose()?;
     transaction.complete()?;
 
     Ok(Installed {
@@ -608,17 +621,44 @@ pub fn install_verified(
 /// # Errors
 /// Refuses malformed journals, changed identities and unavailable filesystem operations.
 pub fn recover(software: &Software, root: &Path) -> Result<Vec<String>> {
-    let Some(transaction) = staging::Staging::load(root, software.command)? else {
+    if !records::leaf(software.command) {
+        return Err(records::refuse());
+    }
+    if require_idle(software, root).is_ok() {
+        return Ok(Vec::new());
+    }
+    Writer::acquire(root)?.recover(software)
+}
+
+fn recover_locked(software: &Software, root: &Path) -> Result<Vec<String>> {
+    let installation = staging::Staging::load(root, software.command)?;
+    let removal = removal::Removal::load(root, software.command)?;
+    let switch = switch::Switch::load(root, software.command)?;
+    if usize::from(installation.is_some())
+        + usize::from(removal.is_some())
+        + usize::from(switch.is_some())
+        > 1
+    {
+        return Err(records::refuse());
+    }
+    if let Some(switch) = switch {
+        let version = switch.version().to_owned();
+        switch.complete()?;
+        return Ok(vec![format!(
+            "completed the recorded launch switch to {version}"
+        )]);
+    }
+    if let Some(removal) = removal {
+        let version = removal.version().to_owned();
+        removal.complete()?;
+        return Ok(vec![format!("completed the recorded removal of {version}")]);
+    }
+    let Some(transaction) = installation else {
         return Ok(Vec::new());
     };
     let version = transaction.version().to_owned();
     if transaction.recover()? {
-        let executable = transaction.executable();
-        let exposed = root.join("bin").join(exposed_name(
-            software.command,
-            &executable.to_string_lossy(),
-        ));
-        expose(&executable, &exposed, &version, software.command)?;
+        transaction.expose()?;
         transaction.complete()?;
         Ok(vec![format!(
             "completed the recorded installation of {version}"
@@ -630,121 +670,58 @@ pub fn recover(software: &Software, root: &Path) -> Result<Vec<String>> {
     }
 }
 
+/// Refuse a new operation while an install, removal or launch switch is pending.
+///
+/// A different plan cannot authorize recovery of an earlier installation. This
+/// check reads only the journal entry; malformed records also remain untouched.
+/// The caller must repeat it under the prefix lock before making any effect.
+///
+/// # Errors
+/// Returns `RecoveryRequired` for an existing or inaccessible journal.
+pub fn require_idle(software: &Software, root: &Path) -> Result<()> {
+    if !records::leaf(software.command) {
+        return Err(records::refuse());
+    }
+    for journal in [
+        format!(".nddev-software-{}.transaction.json", software.command),
+        removal::journal(software.command),
+        switch::journal(software.command),
+    ] {
+        match fs::symlink_metadata(root.join(journal)) {
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            _ => {
+                return Err(Error::new(
+                    ReasonCode::RecoveryRequired,
+                    "a software operation is pending; a new plan cannot recover or replace its recorded operation",
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
 /// Remove one installed version, and the exposed command if it pointed at it.
 ///
 /// # Errors
 ///
 /// Refuses an unrecorded or modified tree, or a filesystem failure during removal.
 pub fn remove(software: &Software, root: &Path) -> Result<bool> {
-    let version_root = root.join(software.version);
-    if !version_root.exists() {
+    require_idle(software, root)?;
+    if fs::symlink_metadata(root).is_err_and(|error| error.kind() == std::io::ErrorKind::NotFound) {
         return Ok(false);
     }
+    Writer::acquire(root)?.remove(software)
+}
 
-    // Asked **before** the tree goes, and that ordering is the whole fix.
-    // `exposed` resolves the command into the version directory it points at,
-    // and a directory that has just been removed resolves to nothing -- so
-    // reading afterwards cannot tell "it named this one" from "it named another
-    // one" and the answer would always be the same.
-    //
-    // The sentence above this function has said *"and the exposed command if it
-    // pointed at it"* since it was written. The code took the command whenever
-    // the tree existed. So the ordinary sequence after a bad release -- install,
-    // update, roll back, take the bad one off -- deleted the command that was
-    // running the good one and left a complete version tree nothing could start.
-    let exposed_version =
-        Present::under_named(root, software.command, software.member_here()).exposed;
-
-    if !ownership::remove(root, software.command, software.version)? {
+fn remove_locked(software: &Software, root: &Path) -> Result<bool> {
+    require_idle(software, root)?;
+    let Some(removal) = removal::Removal::begin(root, software)? else {
         return Ok(false);
-    }
-
-    // Some(this one)  -- the command named what was just taken; it goes too.
-    // Some(another)   -- the command names a version still installed; it stays.
-    // None            -- nothing usable was exposed, so anything left behind is
-    //                    a leftover rather than somebody's entry point.
-    let ours = exposed_version.as_deref() == Some(software.version);
-    if !ours && exposed_version.is_some() {
-        return Ok(true);
-    }
-
-    let exposed = root
-        .join("bin")
-        .join(exposed_name(software.command, software.member_here()));
-    if exposed.symlink_metadata().is_ok() {
-        crate::lock::remove_file(&exposed).map_err(|error| {
-            Error::new(
-                ReasonCode::StateUnavailable,
-                format!("{} could not be removed: {error}", exposed.display()),
-            )
-            .with_source(error)
-        })?;
-    }
-    // The record goes with the command it described. A marker outliving it
-    // would name a version nothing runs.
-    let remove_record = |path: &Path| -> Result<()> {
-        match crate::lock::remove_file(path) {
-            Ok(()) => Ok(()),
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
-            Err(error) => Err(Error::new(
-                ReasonCode::StateUnavailable,
-                "software launch record could not be removed",
-            )
-            .with_source(error)),
-        }
     };
-    remove_record(&Present::marker(root, software.command))?;
-    if Manifest::read(root, software.command)
-        .is_some_and(|record| record.version == software.version)
-    {
-        remove_record(&Manifest::path(root, software.command))?;
-    }
+    removal.complete()?;
     Ok(true)
 }
 
-/// Where a version tree can hold its executable, most specific first.
-///
-/// Looked for rather than assumed. This build pins one version and knows where
-/// *its* executable sits inside the archive; an older tree was written by an
-/// older build, whose artifact table this one does not carry. Every shape it
-/// could have used is tried, and none is guessed at: if the file is not there,
-/// the refusal says where it looked.
-///
-/// **The platform is a parameter and not a `cfg!`**, for the same reason
-/// [`exposed_name_on`] takes one. That is not hypothetical here.
-/// [`Software::member_hint`] answers with the *first* artifact's member
-/// whatever host is asking, the tables in this repository list Linux first, and
-/// so a Windows rollback looked for `package/bin/opencode` while the file on
-/// disk was `package/bin/opencode.exe`. Ubuntu and macOS passed; only the
-/// evidence run on Windows failed, and a `cfg!` here would have left the fix
-/// unprovable from the machine that wrote it.
-fn executable_candidates(
-    software: &Software,
-    version_root: &Path,
-    os: &str,
-    arch: &str,
-) -> Vec<PathBuf> {
-    let mut candidates = Vec::new();
-    let mut push = |relative: &str| {
-        let path = version_root.join(relative);
-        if !candidates.contains(&path) {
-            candidates.push(path);
-        }
-    };
-    // This host's own member first: it is the only one that is right by
-    // construction rather than by the table happening to be ordered well.
-    if let Ok(artifact) = software.artifact_for(os, arch) {
-        push(artifact.member);
-    }
-    // Then the shapes an older build could have written.
-    push(software.member_hint());
-    push(software.command);
-    if os == "windows" {
-        push(&format!("{}.exe", software.command));
-    }
-    candidates
-}
-
 /// Point the exposed command back at a version that is already on disk.
 ///
 /// Installing 1.0.6 leaves 1.0.5 in its own directory and moves only the
@@ -769,6 +746,11 @@ fn executable_candidates(
 /// Refuses a version that is not installed, naming the ones that are, and a
 /// version tree that holds no executable this build can find.
 pub fn rollback(software: &Software, root: &Path, to: &str) -> Result<Installed> {
+    require_idle(software, root)?;
+    Writer::acquire(root)?.rollback(software, to)
+}
+
+fn rollback_locked(software: &Software, root: &Path, to: &str) -> Result<Installed> {
     let present = Present::under_named(root, software.command, software.member_here());
     if !present.versions.iter().any(|found| found == to) {
         return Err(Error::new(
@@ -789,31 +771,12 @@ pub fn rollback(software: &Software, root: &Path, to: &str) -> Result<Installed>
         ));
     }
 
-    let version_root = root.join(to);
-    let (os, arch) = crate::platform_of_this_host();
-    let candidates = executable_candidates(software, &version_root, os, arch);
-    let Some(executable) = candidates.iter().find(|path| path.is_file()) else {
-        return Err(Error::new(
-            ReasonCode::StateUnavailable,
-            format!(
-                "the {to} tree holds no {} executable; looked at {}",
-                software.command,
-                candidates
-                    .iter()
-                    .map(|path| path.display().to_string())
-                    .collect::<Vec<_>>()
-                    .join(" and ")
-            ),
-        ));
-    };
-
-    let exposed = root
-        .join("bin")
-        .join(exposed_name(software.command, software.member_here()));
-    expose(executable, &exposed, to, software.command)?;
+    let switch = switch::Switch::begin(root, software.command, to)?;
+    let exposed = switch.executable();
+    switch.complete()?;
     Ok(Installed {
         version: to.to_owned(),
-        root: version_root,
+        root: root.join(to),
         executable: exposed,
         files: 0,
     })
@@ -937,34 +900,31 @@ impl Manifest {
         root.join("bin").join(format!(".{command}.manifest.json"))
     }
 
-    /// The record, or `None` where there is not one to read.
-    ///
-    /// Absent is a real state and not a failure: a prefix written before this
-    /// existed has no record, and calling that tampering would refuse every
-    /// installation made by an earlier release of this provider. Unparseable is
-    /// also `None` -- a record that cannot be read cannot accuse anything.
+    /// Distinguish absence from an unreadable, malformed or unsupported record.
     #[must_use]
-    pub fn read(root: &Path, command: &str) -> Option<Self> {
-        match Self::inspect(root, command) {
-            ManifestState::Present(found) => Some(found),
-            ManifestState::Missing
-            | ManifestState::Unreadable
-            | ManifestState::Malformed
-            | ManifestState::Unsupported { .. } => None,
-        }
-    }
-
-    /// Distinguish a missing legacy record from a corrupt or unsupported one.
-    ///
-    /// [`Manifest::read`] collapsed every failure into `None`, so a truncated
-    /// JSON file disabled verification the same way an older prefix with no
-    /// record did. Launch has to treat those as different states.
-    #[must_use]
-    pub fn inspect(root: &Path, command: &str) -> ManifestState {
-        match fs::read_to_string(Self::path(root, command)) {
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => ManifestState::Missing,
+    pub fn inspect(path: &Path, command: &str) -> ManifestState {
+        let read = || -> Result<Option<Vec<u8>>> {
+            if !records::leaf(command) {
+                return Err(records::refuse());
+            }
+            let root = match records::open_root(path) {
+                Ok(root) => root,
+                Err(error) => match fs::symlink_metadata(path) {
+                    Err(missing) if missing.kind() == std::io::ErrorKind::NotFound => {
+                        return Ok(None);
+                    }
+                    _ => return Err(error),
+                },
+            };
+            let Some(bin) = records::present(&root, "bin")? else {
+                return Ok(None);
+            };
+            records::read_bytes(&bin, &format!(".{command}.manifest.json"), 16 * 1024)
+        };
+        match read() {
+            Ok(None) => ManifestState::Missing,
             Err(_) => ManifestState::Unreadable,
-            Ok(raw) => match serde_json::from_str::<Self>(&raw) {
+            Ok(Some(raw)) => match serde_json::from_slice::<Self>(&raw) {
                 Err(_) => ManifestState::Malformed,
                 Ok(found) if found.schema_version != 1 => ManifestState::Unsupported {
                     schema_version: found.schema_version,
@@ -1212,14 +1172,27 @@ mod tests {
     fn rollback_points_the_command_at_a_version_already_on_disk() {
         let (at, artifact) = staged("rollback", b"#!/bin/sh\necho new\n", CODEX_MEMBER);
         let root = at.join("prefix");
+        let previous = Software {
+            version: "1.2.2",
+            ..software()
+        };
+        install(&previous, &artifact, &at.join("artifact.tgz"), &root).unwrap();
         let installed = install(&software(), &artifact, &at.join("artifact.tgz"), &root).unwrap();
         assert_eq!(installed.version, "1.2.3");
 
-        // What an update leaves behind: the previous tree, untouched.
-        let older = root.join("1.2.2");
-        fs::create_dir_all(older.join("package/vendor/x86_64-unknown-linux-musl/bin")).unwrap();
-        fs::write(older.join(CODEX_MEMBER), b"#!/bin/sh\necho old\n").unwrap();
-
+        let entry = root.join("bin/codex");
+        let retained = at.join("retained-launcher");
+        fs::rename(&entry, &retained).unwrap();
+        fs::write(&entry, b"foreign launcher").unwrap();
+        assert!(rollback(&software(), &root, "1.2.2").is_err());
+        assert_eq!(fs::read(&entry).unwrap(), b"foreign launcher");
+        fs::remove_file(&entry).unwrap();
+        fs::rename(retained, entry).unwrap();
+        let older = root.join("1.2.2").join(CODEX_MEMBER);
+        fs::write(&older, b"changed retained version").unwrap();
+        assert!(rollback(&software(), &root, "1.2.2").is_err());
+        assert_eq!(fs::read(&older).unwrap(), b"changed retained version");
+        fs::write(older, b"#!/bin/sh\necho new\n").unwrap();
         let rolled = rollback(&software(), &root, "1.2.2").unwrap();
         assert_eq!(rolled.version, "1.2.2");
 
@@ -1239,6 +1212,43 @@ mod tests {
                 .as_deref(),
             Some("1.2.3")
         );
+        // A real durable switch record survives exit before its launch renames.
+        drop(switch::Switch::begin(&root, "codex", "1.2.2").unwrap());
+        assert!(rollback(&software(), &root, "1.2.3").is_err());
+        assert!(remove(&software(), &root).is_err());
+        assert!(install(&software(), &artifact, &at.join("artifact.tgz"), &root).is_err());
+        let receipt = root.join(".nddev-software-codex-1.2.2.installed.json");
+        let original = fs::read(&receipt).unwrap();
+        let mut changed: serde_json::Value = serde_json::from_slice(&original).unwrap();
+        changed["artifact_sha256"] = format!("sha256:{}", "0".repeat(64)).into();
+        fs::write(&receipt, serde_json::to_vec(&changed).unwrap()).unwrap();
+        assert!(recover(&software(), &root).is_err());
+        fs::write(receipt, original).unwrap();
+        let marker = root.join("bin/.codex.version");
+        fs::write(&marker, b"foreign marker").unwrap();
+        let before = crate::software_prefix::observe(&root, ".codex-setup-system")
+            .unwrap()
+            .digest;
+        assert!(recover(&software(), &root).is_err());
+        assert_eq!(
+            crate::software_prefix::observe(&root, ".codex-setup-system")
+                .unwrap()
+                .digest,
+            before
+        );
+        // The new marker landed while the command and manifest still name the predecessor.
+        fs::write(marker, b"1.2.2").unwrap();
+        assert_eq!(recover(&software(), &root).unwrap().len(), 1);
+        assert!(recover(&software(), &root).unwrap().is_empty());
+        assert_eq!(
+            Present::under_named(&root, "codex", CODEX_MEMBER)
+                .exposed
+                .as_deref(),
+            Some("1.2.2")
+        );
+        assert!(root.join("1.2.3").join(CODEX_MEMBER).is_file());
+        assert!(!root.join(switch::journal("codex")).exists());
+        fs::remove_dir_all(at).unwrap();
     }
 
     /// Removing a version nobody is running leaves the running one runnable.
@@ -1258,12 +1268,13 @@ mod tests {
     fn removing_an_inactive_version_leaves_the_active_one_exposed() {
         let (at, artifact) = staged("remove-inactive", b"#!/bin/sh\necho new\n", CODEX_MEMBER);
         let root = at.join("prefix");
+        let previous = Software {
+            version: "1.2.2",
+            ..software()
+        };
+        install(&previous, &artifact, &at.join("artifact.tgz"), &root).unwrap();
         install(&software(), &artifact, &at.join("artifact.tgz"), &root).unwrap();
 
-        // The version an update left behind, and the one a rollback selects.
-        let older = root.join("1.2.2");
-        fs::create_dir_all(older.join("package/vendor/x86_64-unknown-linux-musl/bin")).unwrap();
-        fs::write(older.join(CODEX_MEMBER), b"#!/bin/sh\necho old\n").unwrap();
         rollback(&software(), &root, "1.2.2").unwrap();
         assert_eq!(
             Present::under_named(&root, "codex", CODEX_MEMBER)
@@ -1465,6 +1476,7 @@ mod tests {
         fs::write(&marker, "1.2.3").unwrap();
         let staging = marker.with_extension("version.incoming");
         fs::write(&staging, b"unrelated temporary file").unwrap();
+        let predecessor = launch::Snapshot::prepare(&root, "codex", CODEX_MEMBER).unwrap();
         let manifest = Manifest::path(&root, "codex");
         fs::remove_file(&manifest).unwrap();
         fs::create_dir(&manifest).unwrap(); // the destination is not a metadata file
@@ -1473,6 +1485,7 @@ mod tests {
             &root.join("bin").join("codex"),
             "1.2",
             "codex",
+            &predecessor,
         );
         assert!(
             refused.is_err(),
@@ -1706,19 +1719,17 @@ mod tests {
         );
     }
 
-    /// A tree an older build wrote may not put the executable where this
-    /// build's artifacts do. Both shapes are tried and neither is guessed at:
-    /// if the file is not there, the refusal says where it looked.
+    /// A directory name is not evidence that a provider installed the tree.
     #[test]
-    fn a_version_tree_with_no_executable_is_refused_naming_where_it_looked() {
+    fn rollback_refuses_an_unrecorded_version_without_changing_the_launcher() {
         let (at, artifact) = staged("rollback-empty", b"x", CODEX_MEMBER);
         let root = at.join("prefix");
         install(&software(), &artifact, &at.join("artifact.tgz"), &root).unwrap();
         fs::create_dir_all(root.join("1.2.2")).unwrap();
 
         let error = rollback(&software(), &root, "1.2.2").unwrap_err();
-        assert!(error.detail().contains("1.2.2"), "{}", error.detail());
-        assert!(error.detail().contains("looked at"), "{}", error.detail());
+        assert_eq!(error.reason(), ReasonCode::RecoveryRequired);
+        assert!(root.join("1.2.2").is_dir());
         assert_eq!(
             Present::under_named(&root, "codex", CODEX_MEMBER)
                 .exposed
@@ -1727,95 +1738,6 @@ mod tests {
         );
     }
 
-    /// Two platforms whose executables are named differently inside the
-    /// archive, which is the ordinary case and not an exotic one: every
-    /// `windows/*` artifact in this repository ends in `.exe` and no other
-    /// does.
-    const CROSS_ARTIFACTS: &[Artifact] = &[
-        Artifact {
-            platform: "linux/x86_64",
-            url: "https://example.invalid/linux.tgz",
-            bytes: 0,
-            sha256: "sha256:0",
-            shape: Shape::GzipTar,
-            member: "package/bin/tool",
-        },
-        Artifact {
-            platform: "windows/x86_64",
-            url: "https://example.invalid/windows.tgz",
-            bytes: 0,
-            sha256: "sha256:0",
-            shape: Shape::GzipTar,
-            member: "package/bin/tool.exe",
-        },
-    ];
-
-    fn cross() -> Software {
-        Software {
-            version: "1.2.3",
-            command: "tool",
-            delivery: Delivery::Artifacts(CROSS_ARTIFACTS),
-            unsupported: &[],
-            previous: None,
-        }
-    }
-
-    /// A rollback on Windows looked for the Linux member and refused a tree
-    /// that held the executable all along.
-    ///
-    /// `member_hint` answers with the *first* artifact's member whatever host
-    /// asks, and every table here lists Linux first. The evidence run caught it
-    /// on `windows-latest` while Ubuntu and macOS passed, which is exactly the
-    /// shape a `cfg!` would have made unprovable from the machine that fixes
-    /// it -- so the platform is a parameter and this runs everywhere.
-    #[test]
-    fn the_candidate_list_is_this_hosts_member_and_not_the_tables_first() {
-        let root = Path::new("/prefix/1.2.2");
-
-        let windows = executable_candidates(&cross(), root, "windows", "x86_64");
-        assert_eq!(
-            windows.first(),
-            Some(&root.join("package/bin/tool.exe")),
-            "this host's own member comes first: {windows:?}"
-        );
-        assert!(
-            windows.contains(&root.join("tool.exe")),
-            "the bare command with its extension is a shape an older tree may \
-             have used: {windows:?}"
-        );
-
-        let linux = executable_candidates(&cross(), root, "linux", "x86_64");
-        assert_eq!(
-            linux.first(),
-            Some(&root.join("package/bin/tool")),
-            "{linux:?}"
-        );
-        assert!(
-            !linux.contains(&root.join("tool.exe")),
-            "the Windows shape is not offered to a platform that cannot run it: {linux:?}"
-        );
-
-        // The defect, stated as the thing that must not come back: the hint is
-        // the Linux member on both hosts, so a list built from it alone sends a
-        // Windows rollback to a file that is not there.
-        assert_eq!(cross().member_hint(), "package/bin/tool");
-        assert_ne!(
-            windows.first(),
-            Some(&root.join(cross().member_hint())),
-            "a Windows candidate list must not start at the hint"
-        );
-    }
-
-    /// An unsupported platform still gets a usable list rather than an empty
-    /// one: the tree may have been written by a build that supported it.
-    #[test]
-    fn a_platform_with_no_artifact_still_offers_the_older_shapes() {
-        let root = Path::new("/prefix/1.2.2");
-        let found = executable_candidates(&software(), root, "windows", "x86_64");
-        assert!(!found.is_empty(), "{found:?}");
-        assert!(found.contains(&root.join("codex.exe")), "{found:?}");
-    }
-
     fn scratch(name: &str) -> PathBuf {
         let path =
             std::env::temp_dir().join(format!("setup-core-software-{name}-{}", std::process::id()));
@@ -1972,6 +1894,8 @@ mod tests {
             65_536
         );
         assert_eq!(recover(&software(), &root).unwrap().len(), 1);
+        assert!(remove(&software(), &root).unwrap());
+        assert!(!root.join("1.2.3").exists());
         fs::remove_dir_all(&at).unwrap();
     }
 
@@ -1979,6 +1903,14 @@ mod tests {
     fn reinstallation_preserves_modified_trees_and_replaces_unchanged_ones() {
         let (at, artifact) = staged("twice", b"first", CODEX_MEMBER);
         let root = at.join("software");
+        fs::create_dir_all(root.join("bin")).unwrap();
+        let entry = root.join("bin/codex");
+        fs::write(&entry, b"foreign launcher").unwrap();
+        assert!(install(&software(), &artifact, &at.join("artifact.tgz"), &root).is_err());
+        assert_eq!(fs::read(&entry).unwrap(), b"foreign launcher");
+        assert!(!root.join("1.2.3").exists());
+        assert!(!root.join(".nddev-software-codex.transaction.json").exists());
+        fs::remove_file(&entry).unwrap();
         install(&software(), &artifact, &at.join("artifact.tgz"), &root).unwrap();
         let stray = root.join("1.2.3/package/left-over");
         fs::write(&stray, b"unrecorded content").unwrap();
@@ -2050,6 +1982,37 @@ mod tests {
         let installed = install(&software(), &artifact, &at.join("artifact.tgz"), &root).unwrap();
         assert!(installed.executable.symlink_metadata().is_ok());
 
+        let retained_entry = at.join("retained-entry");
+        fs::rename(&installed.executable, &retained_entry).unwrap();
+        fs::write(&installed.executable, b"foreign launcher").unwrap();
+        assert!(remove(&software(), &root).is_err());
+        assert_eq!(
+            fs::read(&installed.executable).unwrap(),
+            b"foreign launcher"
+        );
+        assert!(root.join("1.2.3").is_dir());
+        fs::remove_file(&installed.executable).unwrap();
+        fs::rename(&retained_entry, &installed.executable).unwrap();
+
+        let marker = Present::marker(&root, "codex");
+        fs::write(&marker, b"another-version").unwrap();
+        assert!(remove(&software(), &root).is_err());
+        assert!(root.join("1.2.3").is_dir());
+        fs::write(&marker, b"1.2.3").unwrap();
+        let manifest = Manifest::path(&root, "codex");
+        let manifest_bytes = fs::read(&manifest).unwrap();
+        fs::write(&manifest, b"{").unwrap();
+        assert!(remove(&software(), &root).is_err());
+        assert!(root.join("1.2.3").is_dir());
+        fs::write(&manifest, &manifest_bytes).unwrap();
+
+        let linked_manifest = at.join("manifest-link");
+        fs::hard_link(&manifest, &linked_manifest).unwrap();
+        assert_eq!(Manifest::inspect(&root, "codex"), ManifestState::Unreadable);
+        assert!(remove(&software(), &root).is_err());
+        assert!(root.join("1.2.3").is_dir());
+        fs::remove_file(&linked_manifest).unwrap();
+
         // A receipt does not authorize deleting files added after installation.
         let foreign = root.join("1.2.3/personal-file");
         fs::write(&foreign, b"preserve").unwrap();
diff --git a/crates/setup-core/src/software/cleanup.rs b/crates/setup-core/src/software/cleanup.rs
new file mode 100644
index 0000000..7e1f0ef
--- /dev/null
+++ b/crates/setup-core/src/software/cleanup.rs
@@ -0,0 +1,116 @@
+//! Delete only unchanged remaining members of a recorded installation.
+
+use std::{
+    collections::BTreeMap,
+    path::Path,
+    time::{Duration, Instant},
+};
+
+use cap_fs_ext::{DirExt, MetadataExt};
+use cap_std::fs::{Dir, Metadata};
+
+use super::{
+    exposure,
+    records::{Identity, io, refuse, sync},
+};
+use crate::{
+    Result,
+    software_prefix::{self, Entry, EntryKind},
+};
+
+fn mode(metadata: &Metadata) -> u32 {
+    #[cfg(unix)]
+    {
+        use cap_std::fs::PermissionsExt;
+        metadata.permissions().mode() & 0o7777
+    }
+    #[cfg(not(unix))]
+    {
+        u32::from(metadata.permissions().readonly())
+    }
+}
+
+fn remove_member(root: &Dir, entry: &Entry) -> Result<()> {
+    let path = Path::new(&entry.path);
+    let mut parent = io(root.try_clone())?;
+    for component in path.parent().ok_or_else(refuse)?.components() {
+        parent = io(parent.open_dir_nofollow(component.as_os_str()))?;
+        Identity::of(&parent)?;
+    }
+    let name = path.file_name().ok_or_else(refuse)?;
+    let before = io(parent.symlink_metadata(name))?;
+    if mode(&before) != entry.mode {
+        return Err(refuse());
+    }
+    match entry.kind {
+        EntryKind::Directory => {
+            if !before.is_dir() || before.is_symlink() {
+                return Err(refuse());
+            }
+            let directory = io(parent.open_dir_nofollow(name))?;
+            Identity::of(&directory)?;
+            let held = io(directory.dir_metadata())?;
+            if before.dev() != held.dev() || before.ino() != held.ino() {
+                return Err(refuse());
+            }
+            // Empty-only deletion cannot recursively consume a newly added file.
+            io(directory.remove_open_dir())?;
+        }
+        EntryKind::File | EntryKind::Link => {
+            let observed = match entry.kind {
+                EntryKind::File if before.is_file() && !before.is_symlink() => {
+                    io(exposure::payload(&parent, Path::new(name)))?.2
+                }
+                EntryKind::Link if before.is_symlink() => io(parent.read_link_contents(name))?
+                    .to_str()
+                    .ok_or_else(refuse)?
+                    .to_owned(),
+                _ => return Err(refuse()),
+            };
+            let expected = if entry.kind == EntryKind::File {
+                format!("sha256:{}", entry.payload)
+            } else {
+                entry.payload.clone()
+            };
+            let after = io(parent.symlink_metadata(name))?;
+            if observed != expected
+                || before.dev() != after.dev()
+                || before.ino() != after.ino()
+                || before.len() != after.len()
+                || before.modified().ok() != after.modified().ok()
+                || mode(&after) != entry.mode
+            {
+                return Err(refuse());
+            }
+            io(parent.remove_file_or_symlink(name))?;
+        }
+    }
+    sync(&parent)
+}
+
+pub(super) fn inspect_remaining(directory: &Dir, owned: &[Entry]) -> Result<Vec<Entry>> {
+    software_prefix::inventory_digest(owned).map_err(|_| refuse())?;
+    let (_, remaining) = software_prefix::inventory_directory(directory)?;
+    let expected: BTreeMap<_, _> = owned.iter().map(|entry| (&entry.path, entry)).collect();
+    if remaining
+        .iter()
+        .any(|entry| expected.get(&entry.path).copied() != Some(entry))
+    {
+        return Err(refuse());
+    }
+    Ok(remaining)
+}
+
+pub(super) fn remaining(directory: &Dir, owned: &[Entry]) -> Result<()> {
+    // Observe the complete remaining subset before making any deletion. The
+    // reverse preorder visits children before their directories.
+    let remaining = inspect_remaining(directory, owned)?;
+    let started = Instant::now();
+    for entry in remaining.iter().skip(1).rev() {
+        if started.elapsed() > Duration::from_secs(20) {
+            return Err(refuse());
+        }
+        remove_member(directory, entry)?;
+    }
+    Ok(())
+}
diff --git a/crates/setup-core/src/software/exposure.rs b/crates/setup-core/src/software/exposure.rs
index b661546..06be8f8 100644
--- a/crates/setup-core/src/software/exposure.rs
+++ b/crates/setup-core/src/software/exposure.rs
@@ -11,7 +11,7 @@ use cap_fs_ext::{DirExt, FollowSymlinks, MetadataExt, OpenOptionsFollowExt, Open
 use cap_std::fs::{Dir, File, OpenOptions};
 use sha2::{Digest, Sha256};
 
-use super::Manifest;
+use super::{Manifest, launch};
 use crate::{Error, ReasonCode, Result, digest};
 
 fn invalid() -> io::Error {
@@ -103,7 +103,7 @@ impl Drop for Prepared<'_> {
     }
 }
 
-fn payload(root: &Dir, relative: &Path) -> io::Result<(Dir, File, String)> {
+pub(super) fn payload(root: &Dir, relative: &Path) -> io::Result<(Dir, File, String)> {
     let mut parent = root.try_clone()?;
     for part in relative.parent().ok_or_else(invalid)?.components() {
         let Component::Normal(name) = part else {
@@ -164,6 +164,20 @@ fn payload(root: &Dir, relative: &Path) -> io::Result<(Dir, File, String)> {
     ))
 }
 
+#[cfg(not(unix))]
+pub(super) fn wrapper(executable: &Path) -> io::Result<Option<Vec<u8>>> {
+    use super::{MemberKind, member_kind};
+    let path = executable.to_str().ok_or_else(invalid)?;
+    let invocation = match member_kind(path, true) {
+        MemberKind::JavaScript => "node",
+        MemberKind::CommandScript => "call",
+        MemberKind::Native => return Ok(None),
+    };
+    Ok(Some(
+        format!("@{invocation} \"{path}\" %*\r\n").into_bytes(),
+    ))
+}
+
 fn metadata_destination(directory: &Dir, name: &str) -> io::Result<()> {
     match directory.symlink_metadata(name) {
         Ok(metadata) if metadata.is_file() && !metadata.is_symlink() => {
@@ -186,6 +200,7 @@ pub(super) fn expose(
     exposed: &Path,
     version: &str,
     command: &str,
+    predecessor: &launch::Snapshot,
 ) -> Result<()> {
     let perform = || -> io::Result<()> {
         let root_path = exposed
@@ -199,6 +214,14 @@ pub(super) fn expose(
         let root = parent.open_dir_nofollow(root_path.file_name().ok_or_else(invalid)?)?;
         plain(&root)?;
         let relative = executable.strip_prefix(root_path).map_err(|_| invalid())?;
+        let member = relative
+            .strip_prefix(version)
+            .map_err(|_| invalid())?
+            .to_str()
+            .ok_or_else(invalid)?;
+        predecessor
+            .validate(root_path, command, member)
+            .map_err(|_| invalid())?;
         let (source_parent, source, sha256) = payload(&root, relative)?;
         let manifest = Manifest {
             schema_version: 1,
@@ -208,11 +231,6 @@ pub(super) fn expose(
         };
         let body = serde_json::to_vec(&manifest).map_err(io::Error::other)?;
 
-        match root.create_dir("bin") {
-            Ok(()) => sync(&root)?,
-            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
-            Err(error) => return Err(error),
-        }
         let bin = root.open_dir_nofollow("bin")?;
         plain(&bin)?;
         let name = exposed
@@ -244,11 +262,33 @@ pub(super) fn expose(
             Ok(())
         };
         check_identity()?;
-        // Readers may observe a mixed generation until all three renames land.
-        // The staging journal stays present and can finish these writes again.
-        entry.commit(name)?;
-        marker.commit(&marker_name)?;
-        manifest.commit(&manifest_name)?;
+        let prepared = [
+            (entry, name),
+            (marker, marker_name.as_str()),
+            (manifest, manifest_name.as_str()),
+        ];
+        let replacements = prepared
+            .iter()
+            .map(|(entry, _)| {
+                launch::stamp(&bin, &entry.name, 8 * 1024 * 1024 * 1024)
+                    .map_err(|_| invalid())?
+                    .ok_or_else(invalid)
+            })
+            .collect::<io::Result<Vec<_>>>()?;
+        // Refuse every foreign entry before replacing any member. During a
+        // recorded recovery, a member may already contain this exact result.
+        for (index, replacement) in replacements.iter().enumerate() {
+            predecessor
+                .check_replacement(&bin, index, replacement)
+                .map_err(|_| invalid())?;
+        }
+        for (index, (entry, name)) in prepared.into_iter().enumerate() {
+            check_identity()?;
+            predecessor
+                .check_replacement(&bin, index, &replacements[index])
+                .map_err(|_| invalid())?;
+            entry.commit(name)?;
+        }
         check_identity()
     };
     perform().map_err(|error| {
@@ -278,17 +318,7 @@ fn prepare_entry<'a>(
         use super::{MemberKind, member_kind};
         match member_kind(&executable.to_string_lossy(), true) {
             MemberKind::JavaScript | MemberKind::CommandScript => {
-                let invocation =
-                    if member_kind(&executable.to_string_lossy(), true) == MemberKind::JavaScript {
-                        "node"
-                    } else {
-                        "call"
-                    };
-                Prepared::bytes(
-                    bin,
-                    name,
-                    format!("@{invocation} \"{}\" %*\r\n", executable.display()).as_bytes(),
-                )
+                Prepared::bytes(bin, name, &wrapper(executable)?.ok_or_else(invalid)?)
             }
             MemberKind::Native => {
                 let source_name = executable.file_name().ok_or_else(invalid)?;
diff --git a/crates/setup-core/src/software/launch.rs b/crates/setup-core/src/software/launch.rs
new file mode 100644
index 0000000..6271ce3
--- /dev/null
+++ b/crates/setup-core/src/software/launch.rs
@@ -0,0 +1,398 @@
+//! Observe coherent launch ownership and retain exact entry preconditions.
+
+use std::path::{Path, PathBuf};
+
+use cap_fs_ext::MetadataExt;
+use cap_std::fs::Dir;
+use serde::{Deserialize, Serialize};
+
+use super::{Manifest, Software, exposure, records};
+use crate::Result;
+use records::{Identity, io, open_root, present, refuse, sync};
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct Stamp {
+    device: u64,
+    inode: u64,
+    length: u64,
+    mode: u32,
+    link: bool,
+    content: String,
+}
+
+pub(super) fn stamp(bin: &Dir, name: &str, limit: u64) -> Result<Option<Stamp>> {
+    let before = match bin.symlink_metadata(name) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(refuse().with_source(error)),
+    };
+    if before.len() > limit {
+        return Err(refuse());
+    }
+    let link = before.is_symlink();
+    let content = if link {
+        io(bin.read_link_contents(name))?
+            .to_str()
+            .ok_or_else(refuse)?
+            .to_owned()
+    } else {
+        io(exposure::payload(bin, Path::new(name)))?.2
+    };
+    let after = io(bin.symlink_metadata(name))?;
+    if before.dev() != after.dev()
+        || before.ino() != after.ino()
+        || before.len() != after.len()
+        || before.modified().ok() != after.modified().ok()
+        || before.permissions() != after.permissions()
+        || link != after.is_symlink()
+    {
+        return Err(refuse());
+    }
+    #[cfg(unix)]
+    let mode = {
+        use cap_std::fs::PermissionsExt;
+        before.permissions().mode() & 0o7777
+    };
+    #[cfg(not(unix))]
+    let mode = u32::from(before.permissions().readonly());
+    Ok(Some(Stamp {
+        device: before.dev(),
+        inode: before.ino(),
+        length: before.len(),
+        mode,
+        link,
+        content,
+    }))
+}
+
+fn names(command: &str, member: &str) -> [String; 3] {
+    [
+        super::exposed_name(command, member),
+        format!(".{command}.version"),
+        format!(".{command}.manifest.json"),
+    ]
+}
+
+struct Observed {
+    entries: Vec<(String, Stamp)>,
+    manifest: Manifest,
+}
+
+fn inspect(
+    path: &Path,
+    root: &Dir,
+    bin: &Dir,
+    command: &str,
+    member: &str,
+) -> Result<Option<Observed>> {
+    let names = names(command, member);
+    let observed = names
+        .iter()
+        .zip([8 * 1024 * 1024 * 1024, 1024, 16 * 1024])
+        .map(|(name, limit)| stamp(bin, name, limit))
+        .collect::<Result<Vec<_>>>()?;
+    if observed.iter().all(Option::is_none) {
+        return Ok(None);
+    }
+    let entries = names
+        .into_iter()
+        .zip(observed)
+        .map(|(name, held)| Ok((name, held.ok_or_else(refuse)?)))
+        .collect::<Result<Vec<_>>>()?;
+    let marker = records::read_bytes(bin, &entries[1].0, 1024)?.ok_or_else(refuse)?;
+    let manifest: Manifest = records::read(bin, &entries[2].0, 16 * 1024)?.ok_or_else(refuse)?;
+    if manifest.schema_version != 1
+        || !records::leaf(&manifest.version)
+        || marker != manifest.version.as_bytes()
+        || !super::ownership::digest_valid(&manifest.executable_sha256)
+        || !records::member_valid(&manifest.executable)
+        || !manifest
+            .executable
+            .starts_with(&format!("{}/", manifest.version))
+    {
+        return Err(refuse());
+    }
+    let executable = Path::new(&manifest.executable);
+    let (_, _, digest) = io(exposure::payload(root, executable))?;
+    if digest != manifest.executable_sha256 {
+        return Err(refuse());
+    }
+    let command_entry = &entries[0].1;
+    #[cfg(unix)]
+    {
+        let valid = if command_entry.link {
+            Path::new(&command_entry.content) == path.join(executable)
+        } else {
+            command_entry.content == digest
+        };
+        if !valid {
+            return Err(refuse());
+        }
+    }
+    #[cfg(not(unix))]
+    {
+        let expected = io(exposure::wrapper(&path.join(executable)))?;
+        let valid = if let Some(bytes) = expected {
+            records::read_bytes(bin, &entries[0].0, 16 * 1024)? == Some(bytes)
+        } else {
+            !command_entry.link && command_entry.content == digest
+        };
+        if !valid {
+            return Err(refuse());
+        }
+    }
+    Ok(Some(Observed { entries, manifest }))
+}
+
+/// A captured predecessor, retained before any version-tree promotion.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct Snapshot {
+    root_identity: Identity,
+    bin_identity: Identity,
+    entries: Vec<(String, Option<Stamp>)>,
+}
+
+impl Snapshot {
+    pub(super) fn prepare(path: &Path, command: &str, member: &str) -> Result<Self> {
+        if !records::leaf(command) || !records::member_valid(member) {
+            return Err(refuse());
+        }
+        let root = open_root(path)?;
+        let bin = if let Some(bin) = present(&root, "bin")? {
+            bin
+        } else {
+            io(root.create_dir("bin"))?;
+            sync(&root)?;
+            present(&root, "bin")?.ok_or_else(refuse)?
+        };
+        let entries = match inspect(path, &root, &bin, command, member)? {
+            Some(Observed { entries, .. }) => entries
+                .into_iter()
+                .map(|(name, stamp)| (name, Some(stamp)))
+                .collect(),
+            None => names(command, member)
+                .into_iter()
+                .map(|name| (name, None))
+                .collect(),
+        };
+        let snapshot = Self {
+            root_identity: Identity::of(&root)?,
+            bin_identity: Identity::of(&bin)?,
+            entries,
+        };
+        snapshot.revalidate(path, command, member)?;
+        Ok(snapshot)
+    }
+
+    pub(super) fn validate(&self, path: &Path, command: &str, member: &str) -> Result<()> {
+        let root = open_root(path)?;
+        let bin = present(&root, "bin")?.ok_or_else(refuse)?;
+        if Identity::of(&root)? != self.root_identity
+            || Identity::of(&bin)? != self.bin_identity
+            || self.entries.len() != 3
+        {
+            return Err(refuse());
+        }
+        let populated = self
+            .entries
+            .iter()
+            .filter(|(_, entry)| entry.is_some())
+            .count();
+        if populated != 0 && populated != 3 {
+            return Err(refuse());
+        }
+        for (index, ((name, original), expected)) in
+            self.entries.iter().zip(names(command, member)).enumerate()
+        {
+            let limit = [8 * 1024 * 1024 * 1024, 1024, 16 * 1024][index];
+            if *name != expected
+                || original.as_ref().is_some_and(|entry| {
+                    entry.mode > 0o7777
+                        || entry.length > limit
+                        || (entry.link && index != 0)
+                        || (!entry.link && !super::ownership::digest_valid(&entry.content))
+                })
+            {
+                return Err(refuse());
+            }
+        }
+        Ok(())
+    }
+
+    pub(super) fn revalidate(&self, path: &Path, command: &str, member: &str) -> Result<()> {
+        self.validate(path, command, member)?;
+        let bin = present(&open_root(path)?, "bin")?.ok_or_else(refuse)?;
+        for ((name, original), limit) in
+            self.entries
+                .iter()
+                .zip([8 * 1024 * 1024 * 1024, 1024, 16 * 1024])
+        {
+            if stamp(&bin, name, limit)? != *original {
+                return Err(refuse());
+            }
+        }
+        Ok(())
+    }
+
+    pub(super) fn check_replacement(
+        &self,
+        bin: &Dir,
+        index: usize,
+        replacement: &Stamp,
+    ) -> Result<()> {
+        let (name, original) = self.entries.get(index).ok_or_else(refuse)?;
+        let limit = [8 * 1024 * 1024 * 1024, 1024, 16 * 1024]
+            .get(index)
+            .copied()
+            .ok_or_else(refuse)?;
+        let current = stamp(bin, name, limit)?;
+        if current == *original {
+            return Ok(());
+        }
+        // A recorded promotion may have completed any prefix of the three
+        // renames. Accept only the exact intended payload on a later attempt.
+        if current.as_ref().is_some_and(|found| {
+            found.length == replacement.length
+                && found.mode == replacement.mode
+                && found.link == replacement.link
+                && found.content == replacement.content
+        }) {
+            return Ok(());
+        }
+        Err(refuse())
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct Record {
+    root_identity: Identity,
+    bin_identity: Identity,
+    entries: Vec<(String, Stamp)>,
+}
+
+pub(super) struct Retraction {
+    path: PathBuf,
+    bin: Dir,
+    record: Record,
+}
+
+impl Retraction {
+    /// All three records must agree. Absence or ambiguity grants no ownership.
+    pub(super) fn prepare(path: &Path, software: &Software) -> Result<Option<Self>> {
+        let root = open_root(path)?;
+        let Some(bin) = present(&root, "bin")? else {
+            return Ok(None);
+        };
+        let Some(Observed { entries, manifest }) =
+            inspect(path, &root, &bin, software.command, software.member_here())?
+        else {
+            return Ok(None);
+        };
+        // Another coherent exposure is preserved in its entirety.
+        if manifest.version != software.version {
+            return Ok(None);
+        }
+        let receipt = super::ownership::Receipt::read(&root, software.command, software.version)?
+            .ok_or_else(refuse)?;
+        if manifest.executable != format!("{}/{}", software.version, receipt.member()) {
+            return Err(refuse());
+        }
+        let retraction = Self {
+            path: path.to_owned(),
+            record: Record {
+                root_identity: Identity::of(&root)?,
+                bin_identity: Identity::of(&bin)?,
+                entries,
+            },
+            bin,
+        };
+        retraction.revalidate(false)?;
+        Ok(Some(retraction))
+    }
+
+    pub(super) fn into_record(self) -> Record {
+        self.record
+    }
+
+    pub(super) fn resume(
+        path: &Path,
+        command: &str,
+        version: &str,
+        member: &str,
+        record: Record,
+    ) -> Result<Self> {
+        let names = [
+            super::exposed_name(command, member),
+            format!(".{command}.version"),
+            format!(".{command}.manifest.json"),
+        ];
+        if record.entries.len() != 3 {
+            return Err(refuse());
+        }
+        for ((name, stamp), (expected, limit)) in
+            record.entries.iter().zip(names.into_iter().zip([
+                8 * 1024 * 1024 * 1024,
+                1024,
+                16 * 1024,
+            ]))
+        {
+            if *name != expected || stamp.length > limit || stamp.mode > 0o7777 {
+                return Err(refuse());
+            }
+            if stamp.link {
+                if *name != super::exposed_name(command, member)
+                    || Path::new(&stamp.content) != path.join(version).join(member)
+                {
+                    return Err(refuse());
+                }
+            } else if !super::ownership::digest_valid(&stamp.content) {
+                return Err(refuse());
+            }
+        }
+        let root = open_root(path)?;
+        let bin = present(&root, "bin")?.ok_or_else(refuse)?;
+        let retraction = Self {
+            path: path.to_owned(),
+            bin,
+            record,
+        };
+        retraction.revalidate(true)?;
+        Ok(retraction)
+    }
+
+    fn revalidate(&self, allow_missing: bool) -> Result<()> {
+        let root = open_root(&self.path)?;
+        if Identity::of(&root)? != self.record.root_identity
+            || Identity::of(&present(&root, "bin")?.ok_or_else(refuse)?)?
+                != self.record.bin_identity
+        {
+            return Err(refuse());
+        }
+        for (name, expected) in &self.record.entries {
+            match stamp(&self.bin, name, expected.length)? {
+                Some(found) if found == *expected => {}
+                None if allow_missing => {}
+                _ => return Err(refuse()),
+            }
+        }
+        Ok(())
+    }
+
+    pub(super) fn complete(self) -> Result<()> {
+        self.revalidate(true)?;
+        for (name, expected) in self.record.entries {
+            match stamp(&self.bin, &name, expected.length)? {
+                Some(found) if found == expected => {
+                    io(self.bin.remove_file(name))?;
+                    sync(&self.bin)?;
+                }
+                None => {}
+                _ => return Err(refuse()),
+            }
+        }
+        Ok(())
+    }
+}
diff --git a/crates/setup-core/src/software/ownership.rs b/crates/setup-core/src/software/ownership.rs
index c6e61da..23a5f93 100644
--- a/crates/setup-core/src/software/ownership.rs
+++ b/crates/setup-core/src/software/ownership.rs
@@ -1,11 +1,9 @@
 //! Installation receipts bind a version tree, never just its directory name.
 
-use std::path::Path;
-
 use cap_std::fs::Dir;
 use serde::{Deserialize, Serialize};
 
-use super::records::{self, Identity, io, leaf, member_valid, open_root, present, refuse, sync};
+use super::records::{self, Identity, io, leaf, member_valid, present, refuse, sync};
 use crate::{
     Result,
     software_prefix::{self, Entry, INVENTORY_LIMIT},
@@ -44,25 +42,60 @@ fn name(command: &str, version: &str) -> Result<String> {
 }
 
 impl Receipt {
+    pub(super) fn digest(&self) -> Result<String> {
+        crate::digest::of_domain_canonical_json(
+            "nddev-software-receipt/1",
+            &serde_json::to_value(self).map_err(|_| refuse())?,
+        )
+    }
+
+    pub(super) fn member(&self) -> &str {
+        &self.member
+    }
+
+    pub(super) fn tree_identity(&self) -> Identity {
+        self.tree_identity
+    }
+
+    pub(super) fn entries(&self) -> &[Entry] {
+        &self.entries
+    }
+
+    pub(super) fn remove_record(&self, root: &Dir) -> Result<()> {
+        match Self::read(root, &self.command, &self.version)? {
+            Some(ref current) if current == self => {
+                io(root.remove_file(name(&self.command, &self.version)?))?;
+                sync(root)
+            }
+            None => Ok(()),
+            _ => Err(refuse()),
+        }
+    }
+
     pub(super) fn read(root: &Dir, command: &str, version: &str) -> Result<Option<Self>> {
         let Some(record): Option<Self> =
             records::read(root, &name(command, version)?, INVENTORY_LIMIT)?
         else {
             return Ok(None);
         };
-        if record.schema_version != 1
-            || record.command != command
-            || record.version != version
-            || record.root_identity != Identity::of(root)?
-            || !digest_valid(&record.artifact_sha256)
-            || !digest_valid(&record.tree_digest)
-            || !member_valid(&record.member)
-            || software_prefix::inventory_digest(&record.entries).map_err(|_| refuse())?
-                != record.tree_digest
+        record.validate(root, command, version)?;
+        Ok(Some(record))
+    }
+
+    pub(super) fn validate(&self, root: &Dir, command: &str, version: &str) -> Result<()> {
+        if self.schema_version != 1
+            || self.command != command
+            || self.version != version
+            || self.root_identity != Identity::of(root)?
+            || !digest_valid(&self.artifact_sha256)
+            || !digest_valid(&self.tree_digest)
+            || !member_valid(&self.member)
+            || software_prefix::inventory_digest(&self.entries).map_err(|_| refuse())?
+                != self.tree_digest
         {
             return Err(refuse());
         }
-        Ok(Some(record))
+        Ok(())
     }
 
     pub(super) fn verify(&self, directory: &Dir) -> Result<&str> {
@@ -105,29 +138,3 @@ pub(super) fn record_installation(
     };
     records::write(root, &name(command, version)?, &record, INVENTORY_LIMIT)
 }
-
-/// Refuse unknown or modified trees before deleting any version content.
-pub(super) fn remove(root_path: &Path, command: &str, version: &str) -> Result<bool> {
-    let root = open_root(root_path)?;
-    let Some(directory) = present(&root, version)? else {
-        return Ok(false);
-    };
-    let record = Receipt::read(&root, command, version)?.ok_or_else(|| {
-        crate::Error::new(
-            crate::ReasonCode::RecoveryRequired,
-            "software version has no installation receipt; reinstall its exact artifact to adopt unchanged legacy bytes",
-        )
-    })?;
-    record.verify(&directory)?;
-    if Identity::of(&open_root(root_path)?)? != record.root_identity {
-        return Err(refuse());
-    }
-    // Consume the held directory, so a changed name cannot select another tree.
-    io(directory.remove_open_dir_all())?;
-    if Receipt::read(&root, command, version)?.as_ref() != Some(&record) {
-        return Err(refuse());
-    }
-    io(root.remove_file(name(command, version)?))?;
-    sync(&root)?;
-    Ok(true)
-}
diff --git a/crates/setup-core/src/software/records.rs b/crates/setup-core/src/software/records.rs
index de9c5f6..204214d 100644
--- a/crates/setup-core/src/software/records.rs
+++ b/crates/setup-core/src/software/records.rs
@@ -101,6 +101,16 @@ pub(super) fn open_root(path: &Path) -> Result<Dir> {
     io(parent.open_dir_nofollow(name))
 }
 
+pub(super) fn optional_root(path: &Path) -> Result<Option<Dir>> {
+    match open_root(path) {
+        Ok(root) => Ok(Some(root)),
+        Err(error) => match std::fs::symlink_metadata(path) {
+            Err(missing) if missing.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            _ => Err(error),
+        },
+    }
+}
+
 pub(super) fn present(parent: &Dir, name: &str) -> Result<Option<Dir>> {
     match parent.open_dir_nofollow(name) {
         Ok(directory) => {
@@ -113,6 +123,12 @@ pub(super) fn present(parent: &Dir, name: &str) -> Result<Option<Dir>> {
 }
 
 pub(super) fn read<T: DeserializeOwned>(root: &Dir, name: &str, limit: usize) -> Result<Option<T>> {
+    read_bytes(root, name, limit)?
+        .map(|bytes| serde_json::from_slice(&bytes).map_err(|_| refuse()))
+        .transpose()
+}
+
+pub(super) fn read_bytes(root: &Dir, name: &str, limit: usize) -> Result<Option<Vec<u8>>> {
     let mut options = OpenOptions::new();
     options.read(true).follow(FollowSymlinks::No).nonblock(true);
     let mut file = match root.open_with(name, &options) {
@@ -147,9 +163,7 @@ pub(super) fn read<T: DeserializeOwned>(root: &Dir, name: &str, limit: usize) ->
     {
         return Err(refuse());
     }
-    serde_json::from_slice(&bytes)
-        .map(Some)
-        .map_err(|_| refuse())
+    Ok(Some(bytes))
 }
 
 pub(super) fn write<T: Serialize>(root: &Dir, name: &str, record: &T, limit: usize) -> Result<()> {
diff --git a/crates/setup-core/src/software/removal.rs b/crates/setup-core/src/software/removal.rs
new file mode 100644
index 0000000..9dc8307
--- /dev/null
+++ b/crates/setup-core/src/software/removal.rs
@@ -0,0 +1,322 @@
+//! Recorded version removal with resumable, inventory-bound cleanup.
+
+use std::path::{Path, PathBuf};
+
+use cap_std::fs::Dir;
+use serde::{Deserialize, Serialize};
+
+use super::{Software, cleanup, launch, ownership::Receipt, records};
+use crate::{Result, software_prefix::INVENTORY_LIMIT};
+use records::{Identity, io, leaf, open_root, present, refuse, sync, unique};
+
+const RECORD_LIMIT: usize = INVENTORY_LIMIT + 64 * 1024;
+
+#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+enum Phase {
+    Prepared,
+    Quarantined,
+    Cleaned,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+struct Record {
+    schema_version: u32,
+    command: String,
+    version: String,
+    quarantine: String,
+    root_identity: Identity,
+    receipt: Receipt,
+    exposure: Option<launch::Record>,
+    phase: Phase,
+}
+
+pub(super) struct Removal {
+    root: Dir,
+    path: PathBuf,
+    journal: String,
+    record: Record,
+}
+
+pub(super) fn journal(command: &str) -> String {
+    format!(".nddev-software-{command}.removal.json")
+}
+
+impl Removal {
+    pub(super) fn begin(path: &Path, software: &Software) -> Result<Option<Self>> {
+        super::require_idle(software, path)?;
+        let root = open_root(path)?;
+        let Some(directory) = present(&root, software.version)? else {
+            return Ok(None);
+        };
+        let receipt = Receipt::read(&root, software.command, software.version)?.ok_or_else(|| {
+            crate::Error::new(crate::ReasonCode::RecoveryRequired,
+                "software version has no installation receipt; reinstall its exact artifact to adopt unchanged legacy bytes")
+        })?;
+        receipt.verify(&directory)?;
+        drop(directory);
+        let exposure =
+            launch::Retraction::prepare(path, software)?.map(launch::Retraction::into_record);
+        let removal = Self {
+            record: Record {
+                schema_version: 1,
+                command: software.command.to_owned(),
+                version: software.version.to_owned(),
+                quarantine: format!(".removing-{}-{}", software.command, unique()?),
+                root_identity: Identity::of(&root)?,
+                receipt,
+                exposure,
+                phase: Phase::Prepared,
+            },
+            root,
+            path: path.to_owned(),
+            journal: journal(software.command),
+        };
+        removal.save()?;
+        Ok(Some(removal))
+    }
+
+    pub(super) fn load(path: &Path, command: &str) -> Result<Option<Self>> {
+        if !leaf(command) {
+            return Err(refuse());
+        }
+        let Some(root) = records::optional_root(path)? else {
+            return Ok(None);
+        };
+        let Some(record): Option<Record> = records::read(&root, &journal(command), RECORD_LIMIT)?
+        else {
+            return Ok(None);
+        };
+        if record.schema_version != 1
+            || record.command != command
+            || !leaf(&record.version)
+            || !leaf(&record.quarantine)
+            || !record
+                .quarantine
+                .starts_with(&format!(".removing-{command}-"))
+            || record.root_identity != Identity::of(&root)?
+        {
+            return Err(refuse());
+        }
+        record.receipt.validate(&root, command, &record.version)?;
+        Ok(Some(Self {
+            root,
+            path: path.to_owned(),
+            journal: journal(command),
+            record,
+        }))
+    }
+
+    pub(super) fn version(&self) -> &str {
+        &self.record.version
+    }
+
+    fn check_root(&self) -> Result<()> {
+        if Identity::of(&open_root(&self.path)?)? != self.record.root_identity {
+            return Err(refuse());
+        }
+        Ok(())
+    }
+
+    fn save(&self) -> Result<()> {
+        self.check_root()?;
+        records::write(&self.root, &self.journal, &self.record, RECORD_LIMIT)
+    }
+
+    fn quarantine(&mut self) -> Result<()> {
+        let current = present(&self.root, &self.record.version)?;
+        let quarantined = present(&self.root, &self.record.quarantine)?;
+        match (current, quarantined) {
+            (Some(directory), None) => {
+                self.record.receipt.verify(&directory)?;
+                drop(directory);
+                io(self
+                    .root
+                    .rename(&self.record.version, &self.root, &self.record.quarantine))?;
+                let directory = present(&self.root, &self.record.quarantine)?.ok_or_else(refuse)?;
+                self.record.receipt.verify(&directory)?;
+                sync(&self.root)?;
+            }
+            (None, Some(directory)) => {
+                self.record.receipt.verify(&directory)?;
+            }
+            _ => return Err(refuse()),
+        }
+        self.record.phase = Phase::Quarantined;
+        self.save()
+    }
+
+    pub(super) fn complete(mut self) -> Result<()> {
+        self.check_root()?;
+        let receipt = Receipt::read(&self.root, &self.record.command, &self.record.version)?;
+        if receipt.as_ref() != Some(&self.record.receipt)
+            && !(receipt.is_none() && self.record.phase == Phase::Cleaned)
+        {
+            return Err(refuse());
+        }
+        if self.record.phase == Phase::Prepared {
+            self.quarantine()?;
+        }
+        if present(&self.root, &self.record.version)?.is_some() {
+            return Err(refuse());
+        }
+        let remaining = present(&self.root, &self.record.quarantine)?;
+        if remaining
+            .as_ref()
+            .map(Identity::of)
+            .transpose()?
+            .is_some_and(|identity| identity != self.record.receipt.tree_identity())
+        {
+            return Err(refuse());
+        }
+        if self.record.phase == Phase::Cleaned {
+            if remaining.is_some() {
+                return Err(refuse());
+            }
+        } else {
+            let exposure = self
+                .record
+                .exposure
+                .as_ref()
+                .map(|exposure| {
+                    launch::Retraction::resume(
+                        &self.path,
+                        &self.record.command,
+                        &self.record.version,
+                        self.record.receipt.member(),
+                        exposure.clone(),
+                    )
+                })
+                .transpose()?;
+            if let Some(directory) = remaining {
+                cleanup::remaining(&directory, self.record.receipt.entries())?;
+                self.check_root()?;
+                io(directory.remove_open_dir())?;
+                sync(&self.root)?;
+            }
+            if let Some(exposure) = exposure {
+                exposure.complete()?;
+            }
+            self.record.phase = Phase::Cleaned;
+            self.save()?;
+        }
+        self.record.receipt.remove_record(&self.root)?;
+        self.check_root()?;
+        io(self.root.remove_file(&self.journal))?;
+        sync(&self.root)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::panic)]
+
+    use super::*;
+    use crate::{
+        archive::build::{Dialect, Item, gzip_tar},
+        digest,
+        software::{self, Artifact, Delivery, Shape},
+    };
+    use std::fs;
+
+    #[test]
+    fn interrupted_removal_preserves_foreign_entries_and_resumes_owned_subsets() {
+        let at = std::env::temp_dir().join(format!("software-removal-{}", unique().unwrap()));
+        fs::create_dir(&at).unwrap();
+        let root = at.join("prefix");
+        let bytes = gzip_tar(
+            &[
+                Item::file("codex", b"payload", 0o755),
+                Item::file("docs/a", b"first", 0o644),
+                Item::file("docs/b", b"second", 0o644),
+            ],
+            Dialect::Gnu,
+        );
+        let archive = at.join("artifact.tgz");
+        fs::write(&archive, &bytes).unwrap();
+        let artifact = Artifact {
+            platform: "linux/x86_64",
+            url: "https://example.invalid/tool.tgz",
+            bytes: bytes.len() as u64,
+            sha256: Box::leak(digest::of_bytes(&bytes).into_boxed_str()),
+            shape: Shape::GzipTar,
+            member: "codex",
+        };
+        let declared = Software {
+            version: "1.2.3",
+            command: "codex",
+            delivery: Delivery::Artifacts(&[]),
+            unsupported: &[],
+            previous: None,
+        };
+        for interruption in 0..4 {
+            software::install(&declared, &artifact, &archive, &root).unwrap();
+            let mut removal = Removal::begin(&root, &declared).unwrap().unwrap();
+            let quarantine = root.join(&removal.record.quarantine);
+            assert!(software::remove(&declared, &root).is_err());
+            assert!(software::install(&declared, &artifact, &archive, &root).is_err());
+            if interruption > 0 {
+                removal.quarantine().unwrap();
+            }
+            if interruption == 1 {
+                // Interruption halfway through version and launch-entry cleanup.
+                fs::remove_file(quarantine.join("docs/a")).unwrap();
+                fs::remove_file(root.join("bin/.codex.version")).unwrap();
+                fs::write(quarantine.join("personal"), b"preserve").unwrap();
+                assert!(software::recover(&declared, &root).is_err());
+                assert_eq!(fs::read(quarantine.join("docs/b")).unwrap(), b"second");
+                assert_eq!(fs::read(quarantine.join("personal")).unwrap(), b"preserve");
+                fs::remove_file(quarantine.join("personal")).unwrap();
+                fs::write(quarantine.join("docs/b"), b"changed").unwrap();
+                assert!(software::recover(&declared, &root).is_err());
+                assert_eq!(fs::read(quarantine.join("docs/b")).unwrap(), b"changed");
+                fs::write(quarantine.join("docs/b"), b"second").unwrap();
+            } else if interruption == 2 {
+                let moved = root.join("retained-tree");
+                fs::rename(&quarantine, &moved).unwrap();
+                fs::create_dir(&quarantine).unwrap();
+                fs::write(quarantine.join("personal"), b"foreign directory").unwrap();
+                assert!(software::recover(&declared, &root).is_err());
+                assert_eq!(
+                    fs::read(quarantine.join("personal")).unwrap(),
+                    b"foreign directory"
+                );
+                fs::remove_file(quarantine.join("personal")).unwrap();
+                fs::remove_dir(&quarantine).unwrap();
+                fs::rename(moved, &quarantine).unwrap();
+            } else if interruption == 3 {
+                let directory = present(&removal.root, &removal.record.quarantine)
+                    .unwrap()
+                    .unwrap();
+                cleanup::remaining(&directory, removal.record.receipt.entries()).unwrap();
+                directory.remove_open_dir().unwrap();
+                launch::Retraction::resume(
+                    &root,
+                    "codex",
+                    "1.2.3",
+                    "codex",
+                    removal.record.exposure.clone().unwrap(),
+                )
+                .unwrap()
+                .complete()
+                .unwrap();
+                removal.record.phase = Phase::Cleaned;
+                removal.save().unwrap();
+                removal.record.receipt.remove_record(&removal.root).unwrap();
+            }
+            drop(removal);
+            assert_eq!(software::recover(&declared, &root).unwrap().len(), 1);
+            assert!(software::recover(&declared, &root).unwrap().is_empty());
+            software::require_idle(&declared, &root).unwrap();
+            assert!(!root.join("1.2.3").exists() && !quarantine.exists());
+            assert!(
+                !root
+                    .join(".nddev-software-codex-1.2.3.installed.json")
+                    .exists()
+            );
+            assert_eq!(fs::read_dir(root.join("bin")).unwrap().count(), 0);
+        }
+        fs::remove_dir_all(at).unwrap();
+    }
+}
diff --git a/crates/setup-core/src/software/staging.rs b/crates/setup-core/src/software/staging.rs
index d521790..7a57a28 100644
--- a/crates/setup-core/src/software/staging.rs
+++ b/crates/setup-core/src/software/staging.rs
@@ -6,11 +6,17 @@ use cap_fs_ext::DirExt;
 use cap_std::fs::Dir;
 use serde::{Deserialize, Serialize};
 
-use super::ownership;
 use super::records::{
     self, Identity, io, leaf, member_valid, open_root, present, refuse, sync, unique,
 };
-use crate::{Result, archive::Destination};
+use super::{cleanup, launch, ownership};
+use crate::{
+    Result,
+    archive::Destination,
+    software_prefix::{self, Entry, INVENTORY_LIMIT},
+};
+
+const RECORD_LIMIT: usize = INVENTORY_LIMIT + 64 * 1024;
 
 #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]
@@ -33,10 +39,12 @@ struct Record {
     stage_identity: Identity,
     previous_identity: Option<Identity>,
     previous_digest: Option<String>,
+    previous_entries: Option<Vec<Entry>>,
     legacy: bool,
     artifact_sha256: String,
     sealed_digest: Option<String>,
     phase: Phase,
+    launch: launch::Snapshot,
 }
 
 pub(super) struct Staging {
@@ -70,21 +78,27 @@ impl Staging {
         }
         let previous = present(&directory, version)?;
         let receipt = ownership::Receipt::read(&directory, command, version)?;
-        let previous_digest = if let Some(receipt) = &receipt {
-            Some(
+        let previous_inventory = if let Some(receipt) = &receipt {
+            Some((
                 receipt
                     .verify(previous.as_ref().ok_or_else(refuse)?)?
                     .to_owned(),
-            )
+                receipt.entries().to_vec(),
+            ))
         } else {
             previous
                 .as_ref()
-                .map(crate::software_prefix::digest_directory)
+                .map(software_prefix::inventory_directory)
                 .transpose()?
         };
+        let (previous_digest, previous_entries) = match previous_inventory {
+            Some((digest, entries)) => (Some(digest), Some(entries)),
+            None => (None, None),
+        };
         let legacy = previous.is_some() && receipt.is_none();
         let previous_identity = previous.as_ref().map(Identity::of).transpose()?;
         drop(previous);
+        let launch = launch::Snapshot::prepare(root, command, member)?;
         let nonce = unique()?;
         let stage = format!(".incoming-{command}-{nonce}");
         let quarantine = format!(".replaced-{command}-{nonce}");
@@ -92,7 +106,7 @@ impl Staging {
         let stage_identity = Identity::of(&io(directory.open_dir_nofollow(&stage))?)?;
         let transaction = Self {
             record: Record {
-                schema_version: 1,
+                schema_version: 3,
                 command: command.to_owned(),
                 version: version.to_owned(),
                 member: member.to_owned(),
@@ -102,10 +116,12 @@ impl Staging {
                 stage_identity,
                 previous_identity,
                 previous_digest,
+                previous_entries,
                 legacy,
                 artifact_sha256: artifact_sha256.to_owned(),
                 sealed_digest: None,
                 phase: Phase::Extracting,
+                launch,
             },
             root: directory,
             path: root.to_owned(),
@@ -119,21 +135,18 @@ impl Staging {
         if !leaf(command) {
             return Err(refuse());
         }
-        let root = match open_root(path) {
-            Ok(root) => root,
-            Err(error) => match std::fs::symlink_metadata(path) {
-                Err(missing) if missing.kind() == std::io::ErrorKind::NotFound => return Ok(None),
-                _ => return Err(error),
-            },
+        let Some(root) = records::optional_root(path)? else {
+            return Ok(None);
         };
         let journal = format!(".nddev-software-{command}.transaction.json");
-        let Some(record): Option<Record> = records::read(&root, &journal, 16 * 1024)? else {
+        let Some(record): Option<Record> = records::read(&root, &journal, RECORD_LIMIT)? else {
             return Ok(None);
         };
-        if record.schema_version != 1
+        if record.schema_version != 3
             || record.command != command
             || !ownership::digest_valid(&record.artifact_sha256)
             || record.previous_identity.is_some() != record.previous_digest.is_some()
+            || record.previous_identity.is_some() != record.previous_entries.is_some()
             || record
                 .previous_digest
                 .as_deref()
@@ -151,6 +164,16 @@ impl Staging {
         {
             return Err(refuse());
         }
+        if record
+            .previous_entries
+            .as_deref()
+            .map(software_prefix::inventory_digest)
+            .transpose()?
+            != record.previous_digest
+        {
+            return Err(refuse());
+        }
+        record.launch.validate(path, command, &record.member)?;
         Ok(Some(Self {
             root,
             path: path.to_owned(),
@@ -161,7 +184,7 @@ impl Staging {
 
     fn save(&self) -> Result<()> {
         self.check_root()?;
-        records::write(&self.root, &self.journal, &self.record, 16 * 1024)
+        records::write(&self.root, &self.journal, &self.record, RECORD_LIMIT)
     }
 
     fn check_root(&self) -> Result<()> {
@@ -200,8 +223,26 @@ impl Staging {
             .join(&self.record.member)
     }
 
+    pub(super) fn expose(&self) -> Result<()> {
+        let executable = self.executable();
+        let exposed = self.path.join("bin").join(super::exposed_name(
+            &self.record.command,
+            &self.record.member,
+        ));
+        super::exposure::expose(
+            &executable,
+            &exposed,
+            &self.record.version,
+            &self.record.command,
+            &self.record.launch,
+        )
+    }
+
     pub(super) fn promote(&mut self) -> Result<()> {
         self.check_root()?;
+        self.record
+            .launch
+            .revalidate(&self.path, &self.record.command, &self.record.member)?;
         let stage = self.checked(&self.record.stage, self.record.stage_identity)?;
         if present(&self.root, &self.record.quarantine)?.is_some() {
             return Err(refuse());
@@ -274,6 +315,7 @@ impl Staging {
             if self.record.sealed_digest.as_deref() != Some(digest.as_str()) {
                 return Err(refuse());
             }
+            self.check_predecessor()?;
             return Ok(true);
         }
         if self.record.phase == Phase::Promoted || stage.is_none() {
@@ -306,9 +348,24 @@ impl Staging {
         Ok(false)
     }
 
+    fn check_predecessor(&self) -> Result<Option<Dir>> {
+        let remaining = present(&self.root, &self.record.quarantine)?;
+        if let Some(directory) = &remaining {
+            if Some(Identity::of(directory)?) != self.record.previous_identity {
+                return Err(refuse());
+            }
+            cleanup::inspect_remaining(
+                directory,
+                self.record.previous_entries.as_deref().ok_or_else(refuse)?,
+            )?;
+        }
+        Ok(remaining)
+    }
+
     pub(super) fn complete(self) -> Result<()> {
         self.check_root()?;
         self.checked(&self.record.version, self.record.stage_identity)?;
+        let predecessor = self.check_predecessor()?;
         ownership::record_installation(
             &self.root,
             &self.record.command,
@@ -318,14 +375,14 @@ impl Staging {
             self.record.sealed_digest.as_deref().ok_or_else(refuse)?,
             self.record.stage_identity,
         )?;
-        if let Some(quarantine) = present(&self.root, &self.record.quarantine)? {
-            if Some(Identity::of(&quarantine)?) != self.record.previous_identity
-                || Some(crate::software_prefix::digest_directory(&quarantine)?)
-                    != self.record.previous_digest
-            {
-                return Err(refuse());
-            }
-            io(quarantine.remove_open_dir_all())?;
+        if let Some(quarantine) = predecessor {
+            cleanup::remaining(
+                &quarantine,
+                self.record.previous_entries.as_deref().ok_or_else(refuse)?,
+            )?;
+            self.check_root()?;
+            io(quarantine.remove_open_dir())?;
+            sync(&self.root)?;
         }
         io(self.root.remove_file(&self.journal))?;
         sync(&self.root)
@@ -343,6 +400,10 @@ mod tests {
         "sha256:0000000000000000000000000000000000000000000000000000000000000000";
 
     #[test]
+    #[allow(
+        clippy::too_many_lines,
+        reason = "One persisted lifecycle drives the interruption and foreign-entry controls."
+    )]
     fn recovery_changes_only_recorded_directories_across_interruption_windows() {
         let root = std::env::temp_dir().join(format!("software-staging-{}", unique().unwrap()));
         fs::create_dir(&root).unwrap();
@@ -359,12 +420,22 @@ mod tests {
         fs::create_dir(root.join(".replaced-unrelated")).unwrap();
         fs::write(root.join("unrelated.incoming"), b"unrelated").unwrap();
         assert!(software::recover(&declared, &root).unwrap().is_empty());
+        software::require_idle(&declared, &root).unwrap();
 
         let transaction = Staging::begin(&root, "codex", "1.2.3", "codex", ARTIFACT).unwrap();
         let partial = transaction.stage_path();
         fs::write(partial.join("codex"), b"partial").unwrap();
         drop(transaction);
+        assert_eq!(
+            software::require_idle(&declared, &root)
+                .unwrap_err()
+                .reason(),
+            crate::ReasonCode::RecoveryRequired
+        );
+        assert!(software::remove(&declared, &root).is_err());
+        assert_eq!(fs::read(partial.join("codex")).unwrap(), b"partial");
         assert_eq!(software::recover(&declared, &root).unwrap().len(), 1);
+        software::require_idle(&declared, &root).unwrap();
         assert!(!partial.exists());
         assert!(!root.join("1.2.3").exists());
 
@@ -404,6 +475,12 @@ mod tests {
         // Promotion landed, but its phase update and exposure did not.
         let mut transaction = Staging::begin(&root, "codex", "1.2.3", "codex", ARTIFACT).unwrap();
         fs::write(transaction.stage_path().join("codex"), b"new").unwrap();
+        let entry = root.join("bin/codex");
+        fs::write(&entry, b"foreign during extraction").unwrap();
+        assert!(transaction.promote().is_err());
+        assert_eq!(fs::read(root.join("1.2.3/codex")).unwrap(), b"previous");
+        assert_eq!(fs::read(&entry).unwrap(), b"foreign during extraction");
+        fs::remove_file(&entry).unwrap();
         transaction.promote().unwrap();
         transaction.record.phase = Phase::Promoting;
         transaction.save().unwrap();
@@ -412,10 +489,55 @@ mod tests {
         assert!(software::recover(&declared, &root).is_err());
         assert!(!root.join("bin/codex").exists());
         fs::write(root.join("1.2.3/codex"), b"new").unwrap();
+        fs::write(&entry, b"foreign after promotion").unwrap();
+        assert!(software::recover(&declared, &root).is_err());
+        assert_eq!(fs::read(&entry).unwrap(), b"foreign after promotion");
+        assert!(!root.join("bin/.codex.version").exists());
+        fs::remove_file(&entry).unwrap();
+        // One of the three recorded activation renames completed before exit.
+        #[cfg(unix)]
+        std::os::unix::fs::symlink(root.join("1.2.3/codex"), &entry).unwrap();
+        #[cfg(not(unix))]
+        fs::copy(root.join("1.2.3/codex"), &entry).unwrap();
         software::recover(&declared, &root).unwrap();
         assert_eq!(fs::read(root.join("bin/codex")).unwrap(), b"new");
         assert!(software::recover(&declared, &root).unwrap().is_empty());
 
+        // Predecessor cleanup was interrupted after deleting a recorded member.
+        let mut transaction = Staging::begin(&root, "codex", "1.2.3", "codex", ARTIFACT).unwrap();
+        fs::write(transaction.stage_path().join("codex"), b"newer").unwrap();
+        transaction.promote().unwrap();
+        transaction.expose().unwrap();
+        let quarantine = root.join(&transaction.record.quarantine);
+        fs::remove_file(quarantine.join("codex")).unwrap();
+        let foreign = quarantine.join("foreign");
+        fs::write(&foreign, b"preserve").unwrap();
+        drop(transaction);
+        let before = crate::software_prefix::observe(&root, ".unused")
+            .unwrap()
+            .digest;
+        let launch_before =
+            serde_json::to_value(launch::Snapshot::prepare(&root, "codex", "codex").unwrap())
+                .unwrap();
+        assert!(software::recover(&declared, &root).is_err());
+        assert_eq!(
+            before,
+            crate::software_prefix::observe(&root, ".unused")
+                .unwrap()
+                .digest
+        );
+        assert_eq!(
+            launch_before,
+            serde_json::to_value(launch::Snapshot::prepare(&root, "codex", "codex").unwrap(),)
+                .unwrap()
+        );
+        assert_eq!(fs::read(&foreign).unwrap(), b"preserve");
+        fs::remove_file(foreign).unwrap();
+        assert_eq!(software::recover(&declared, &root).unwrap().len(), 1);
+        assert!(!quarantine.exists());
+        assert_eq!(fs::read(root.join("bin/codex")).unwrap(), b"newer");
+        assert!(software::recover(&declared, &root).unwrap().is_empty());
+
         // Matching names are insufficient when the actual directory moved.
         let mut transaction = Staging::begin(&root, "codex", "1.2.3", "codex", ARTIFACT).unwrap();
         let original = transaction.stage_path();
@@ -429,7 +551,7 @@ mod tests {
         assert!(software::recover(&declared, &root).is_err());
         assert_eq!(fs::read(original.join("foreign")).unwrap(), b"foreign");
         assert_eq!(fs::read(moved.join("codex")).unwrap(), b"recorded");
-        assert_eq!(fs::read(root.join("bin/codex")).unwrap(), b"new");
+        assert_eq!(fs::read(root.join("bin/codex")).unwrap(), b"newer");
         assert_eq!(fs::read(&unrelated).unwrap(), b"unrelated");
         assert!(root.join(".replaced-unrelated").is_dir());
         assert_eq!(
diff --git a/crates/setup-core/src/software/switch.rs b/crates/setup-core/src/software/switch.rs
new file mode 100644
index 0000000..a938bab
--- /dev/null
+++ b/crates/setup-core/src/software/switch.rs
@@ -0,0 +1,137 @@
+//! A durable launch switch to an unchanged, already installed version.
+
+use std::path::{Path, PathBuf};
+
+use cap_std::fs::Dir;
+use serde::{Deserialize, Serialize};
+
+use super::{exposure, launch, ownership::Receipt, records};
+use crate::Result;
+use records::{Identity, io, leaf, open_root, present, refuse, sync};
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+struct Record {
+    schema_version: u32,
+    command: String,
+    version: String,
+    member: String,
+    receipt_digest: String,
+    predecessor: launch::Snapshot,
+}
+
+pub(super) struct Switch {
+    root: Dir,
+    path: PathBuf,
+    record: Record,
+}
+
+pub(super) fn journal(command: &str) -> String {
+    format!(".nddev-software-{command}.switch.json")
+}
+
+impl Switch {
+    pub(super) fn begin(path: &Path, command: &str, version: &str) -> Result<Self> {
+        if !leaf(command) || !leaf(version) {
+            return Err(refuse());
+        }
+        let root = open_root(path)?;
+        match root.symlink_metadata(journal(command)) {
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            _ => return Err(refuse()),
+        }
+        let receipt = Receipt::read(&root, command, version)?.ok_or_else(refuse)?;
+        let predecessor = launch::Snapshot::prepare(path, command, receipt.member())?;
+        let switch = Self {
+            record: Record {
+                schema_version: 1,
+                command: command.to_owned(),
+                version: version.to_owned(),
+                member: receipt.member().to_owned(),
+                receipt_digest: receipt.digest()?,
+                predecessor,
+            },
+            root,
+            path: path.to_owned(),
+        };
+        switch.verify()?;
+        records::write(&switch.root, &journal(command), &switch.record, 16 * 1024)?;
+        Ok(switch)
+    }
+
+    pub(super) fn load(path: &Path, command: &str) -> Result<Option<Self>> {
+        if !leaf(command) {
+            return Err(refuse());
+        }
+        let Some(root) = records::optional_root(path)? else {
+            return Ok(None);
+        };
+        let Some(record): Option<Record> = records::read(&root, &journal(command), 16 * 1024)?
+        else {
+            return Ok(None);
+        };
+        if record.schema_version != 1
+            || record.command != command
+            || !leaf(&record.version)
+            || !records::member_valid(&record.member)
+            || !super::ownership::digest_valid(&record.receipt_digest)
+        {
+            return Err(refuse());
+        }
+        let switch = Self {
+            root,
+            path: path.to_owned(),
+            record,
+        };
+        switch
+            .record
+            .predecessor
+            .validate(path, command, &switch.record.member)?;
+        Ok(Some(switch))
+    }
+
+    pub(super) fn version(&self) -> &str {
+        &self.record.version
+    }
+
+    pub(super) fn executable(&self) -> PathBuf {
+        self.path.join("bin").join(super::exposed_name(
+            &self.record.command,
+            &self.record.member,
+        ))
+    }
+
+    fn verify(&self) -> Result<()> {
+        if Identity::of(&self.root)? != Identity::of(&open_root(&self.path)?)? {
+            return Err(refuse());
+        }
+        self.record
+            .predecessor
+            .validate(&self.path, &self.record.command, &self.record.member)?;
+        let receipt = Receipt::read(&self.root, &self.record.command, &self.record.version)?
+            .ok_or_else(refuse)?;
+        if receipt.digest()? != self.record.receipt_digest || receipt.member() != self.record.member
+        {
+            return Err(refuse());
+        }
+        receipt.verify(&present(&self.root, &self.record.version)?.ok_or_else(refuse)?)?;
+        Ok(())
+    }
+
+    pub(super) fn complete(self) -> Result<()> {
+        self.verify()?;
+        exposure::expose(
+            &self
+                .path
+                .join(&self.record.version)
+                .join(&self.record.member),
+            &self.executable(),
+            &self.record.version,
+            &self.record.command,
+            &self.record.predecessor,
+        )?;
+        self.verify()?;
+        io(self.root.remove_file(journal(&self.record.command)))?;
+        sync(&self.root)
+    }
+}
diff --git a/crates/setup-core/src/software/writer.rs b/crates/setup-core/src/software/writer.rs
new file mode 100644
index 0000000..2d2da78
--- /dev/null
+++ b/crates/setup-core/src/software/writer.rs
@@ -0,0 +1,235 @@
+//! One held writer for every software command sharing a prefix.
+
+use std::path::{Path, PathBuf};
+
+use cap_fs_ext::{DirExt, MetadataExt};
+use cap_std::fs::Dir;
+
+use super::{Artifact, Installed, Software, VerifiedArtifact, records};
+use crate::{Result, lock::TargetLock};
+use records::{Identity, io, open_root, refuse, sync};
+
+pub(crate) const CONTROL: &str = ".nddev-software";
+
+pub(crate) fn validate_control(root: &Dir) -> Result<()> {
+    let Some(control) = records::present(root, CONTROL)? else {
+        return Ok(());
+    };
+    for entry in io(control.entries())? {
+        let name = io(entry)?.file_name();
+        if name != crate::lock::LOCK_FILE_NAME {
+            return Err(refuse());
+        }
+        let metadata = io(control.symlink_metadata(name))?;
+        if !metadata.is_file() || metadata.is_symlink() || metadata.nlink() != 1 {
+            return Err(refuse());
+        }
+        #[cfg(windows)]
+        {
+            use cap_std::fs::MetadataExt as _;
+            if metadata.file_attributes() & 0x400 != 0 {
+                return Err(refuse());
+            }
+        }
+    }
+    Ok(())
+}
+
+/// A prefix writer held across observation and all software effects.
+///
+/// Every harness uses the same lock. A caller with a plan must revalidate its
+/// prefix precondition after acquiring this writer and before invoking a method.
+pub struct Writer {
+    root: PathBuf,
+    identity: Identity,
+    control_identity: Identity,
+    lock: TargetLock,
+}
+
+impl Writer {
+    /// Lock an existing plain prefix without waiting for another writer.
+    ///
+    /// # Errors
+    /// Refuses contention, aliases, unrecognized control data or changed roots.
+    pub fn acquire(root: &Path) -> Result<Self> {
+        let directory = open_root(root)?;
+        let identity = Identity::of(&directory)?;
+        let created = match directory.create_dir(CONTROL) {
+            Ok(()) => true,
+            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => false,
+            Err(error) => return Err(refuse().with_source(error)),
+        };
+        let control = io(directory.open_dir_nofollow(CONTROL))?;
+        let control_identity = Identity::of(&control)?;
+        // Only the reserved lock entry is allowed. An empty directory left by
+        // interrupted initialization is valid and needs no ownership guesses.
+        if !created {
+            validate_control(&directory)?;
+        }
+        let lock = TargetLock::acquire(&root.join(CONTROL))?;
+        if Identity::of(&open_root(root)?)? != identity
+            || Identity::of(&io(open_root(root)?.open_dir_nofollow(CONTROL))?)? != control_identity
+        {
+            return Err(refuse());
+        }
+        sync(&control)?;
+        if created {
+            sync(&directory)?;
+        }
+        validate_control(&directory)?;
+        Ok(Self {
+            root: root.to_owned(),
+            identity,
+            control_identity,
+            lock,
+        })
+    }
+
+    fn revalidate(&self) -> Result<()> {
+        let root = open_root(&self.root)?;
+        if Identity::of(&root)? != self.identity
+            || Identity::of(&io(root.open_dir_nofollow(CONTROL))?)? != self.control_identity
+        {
+            return Err(refuse());
+        }
+        validate_control(&root)?;
+        self.lock.revalidate()
+    }
+
+    /// Install a held verified artifact while retaining this writer.
+    ///
+    /// # Errors
+    /// Refuses pending transactions, changed input, ownership conflicts or I/O failures.
+    pub fn install_verified(
+        &self,
+        software: &Software,
+        artifact: &Artifact,
+        source: VerifiedArtifact,
+    ) -> Result<Installed> {
+        self.revalidate()?;
+        super::install_locked(software, artifact, source, &self.root)
+    }
+
+    /// Remove an unchanged recorded version while retaining this writer.
+    ///
+    /// # Errors
+    /// Refuses pending transactions, changed owned entries or I/O failures.
+    pub fn remove(&self, software: &Software) -> Result<bool> {
+        self.revalidate()?;
+        super::remove_locked(software, &self.root)
+    }
+
+    /// Complete only the recorded kernel transaction while retaining this writer.
+    ///
+    /// # Errors
+    /// Refuses conflicting records, changed owned entries or I/O failures.
+    pub fn recover(&self, software: &Software) -> Result<Vec<String>> {
+        self.revalidate()?;
+        super::recover_locked(software, &self.root)
+    }
+
+    /// Select an installed version while retaining this writer.
+    ///
+    /// # Errors
+    /// Refuses pending transactions, unavailable versions or I/O failures.
+    pub fn rollback(&self, software: &Software, to: &str) -> Result<Installed> {
+        self.revalidate()?;
+        super::require_idle(software, &self.root)?;
+        super::rollback_locked(software, &self.root, to)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::panic)]
+
+    use super::*;
+    use crate::{
+        ReasonCode,
+        software::{self, Delivery},
+        software_prefix,
+    };
+    use std::fs;
+
+    #[test]
+    fn one_prefix_writer_excludes_other_commands_and_refuses_substituted_metadata() {
+        let root =
+            std::env::temp_dir().join(format!("software-writer-{}", records::unique().unwrap()));
+        fs::create_dir(&root).unwrap();
+        let before = software_prefix::observe(&root, ".first-provider")
+            .unwrap()
+            .digest;
+        let other = Software {
+            version: "1.2.3",
+            command: "other",
+            delivery: Delivery::Artifacts(&[]),
+            unsupported: &[],
+            previous: None,
+        };
+        let writer = Writer::acquire(&root).unwrap();
+        assert_eq!(
+            before,
+            software_prefix::observe(&root, ".first-provider")
+                .unwrap()
+                .digest
+        );
+        assert!(Writer::acquire(&root).is_err());
+        assert_eq!(
+            software::remove(&other, &root).unwrap_err().reason(),
+            ReasonCode::LockUnavailable
+        );
+        assert_eq!(
+            software::rollback(&other, &root, "1.2.3")
+                .unwrap_err()
+                .reason(),
+            ReasonCode::LockUnavailable
+        );
+        let pending = root.join(".nddev-software-other.transaction.json");
+        fs::write(&pending, b"generated incomplete transaction").unwrap();
+        assert_eq!(
+            software::recover(&other, &root).unwrap_err().reason(),
+            ReasonCode::LockUnavailable
+        );
+        assert_eq!(
+            fs::read(&pending).unwrap(),
+            b"generated incomplete transaction"
+        );
+        fs::remove_file(pending).unwrap();
+        assert!(!writer.remove(&other).unwrap());
+        #[cfg(unix)]
+        {
+            let lock = root.join(CONTROL).join(crate::lock::LOCK_FILE_NAME);
+            let previous = root.join("held-lock");
+            fs::rename(&lock, &previous).unwrap();
+            fs::write(&lock, b"replacement lock").unwrap();
+            assert!(writer.remove(&other).is_err());
+            assert_eq!(fs::read(&lock).unwrap(), b"replacement lock");
+            fs::remove_file(&lock).unwrap();
+            fs::rename(previous, lock).unwrap();
+        }
+        drop(writer);
+        drop(Writer::acquire(&root).unwrap());
+        // Windows byte-range locks may reject content reads. The seal test
+        // concerns reserved names in an artifact, not a concurrently held lock.
+        let held = open_root(&root).unwrap();
+        let (_, inventory) = software_prefix::inventory_directory(&held).unwrap();
+        assert!(
+            inventory
+                .iter()
+                .any(|entry| entry.path == ".nddev-software/target.lock")
+        );
+        drop(held);
+        let unknown = root.join(CONTROL).join("personal");
+        fs::write(&unknown, b"preserve").unwrap();
+        assert!(Writer::acquire(&root).is_err());
+        assert!(software_prefix::observe(&root, ".first-provider").is_err());
+        assert_eq!(fs::read(&unknown).unwrap(), b"preserve");
+        fs::remove_file(unknown).unwrap();
+        // Empty interrupted initialization resumes without a format receipt.
+        let lock = root.join(CONTROL).join(crate::lock::LOCK_FILE_NAME);
+        fs::remove_file(&lock).unwrap();
+        drop(Writer::acquire(&root).unwrap());
+        assert!(lock.is_file());
+        fs::remove_dir_all(root).unwrap();
+    }
+}
diff --git a/crates/setup-core/src/software_prefix.rs b/crates/setup-core/src/software_prefix.rs
index d514b0d..c6f1d14 100644
--- a/crates/setup-core/src/software_prefix.rs
+++ b/crates/setup-core/src/software_prefix.rs
@@ -1,6 +1,6 @@
 //! Bounded, lossless software-prefix observations for plan preconditions.
 //!
-//! Only the provider's top-level control directory is excluded. Links contribute
+//! Only validated shared writer metadata and the provider control directory are excluded. Links contribute
 //! their literal destinations; no link or special file is opened for content.
 //! This observation establishes neither ownership nor permission to remove files.
 
@@ -262,7 +262,10 @@ impl Reading {
         for entry in io(dir.entries())? {
             self.check()?;
             let name = io(entry)?.file_name();
-            if depth == 0 && name == control {
+            if depth == 0
+                && (name == control
+                    || (!control.is_empty() && name == crate::software::writer::CONTROL))
+            {
                 continue;
             }
             self.entries += 1;
@@ -395,6 +398,7 @@ pub fn observe(root: &Path, control: &str) -> Result<Observation> {
         ))?;
         let name = root.file_name().ok_or_else(invalid)?;
         let dir = io(parent.open_dir_nofollow(name))?;
+        crate::software::writer::validate_control(&dir)?;
         reading.walk(&dir, "", 0, control)?;
         let reopened = io(parent.open_dir_nofollow(name))?;
         if !same(&io(dir.dir_metadata())?, &io(reopened.dir_metadata())?) {
@@ -536,9 +540,25 @@ mod tests {
                 assert!(observe(&root, ".control").is_err());
                 fs::remove_file(invalid).unwrap();
             }
-            let socket = UnixListener::bind(root.join("socket")).unwrap();
-            assert!(observe(&root, ".control").is_err());
+            // macOS's user temporary directory can exceed sockaddr_un's
+            // pathname budget. Keep this negative control in an exclusively
+            // created short directory; never change the process-wide cwd.
+            let socket_root = Path::new("/tmp").join(format!(
+                "nddev-socket-{}-{}",
+                std::process::id(),
+                std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .unwrap()
+                    .as_nanos()
+            ));
+            fs::create_dir(&socket_root).unwrap();
+            assert!(observe(&socket_root, ".control").is_ok());
+            let socket = UnixListener::bind(socket_root.join("socket")).unwrap();
+            assert!(observe(&socket_root, ".control").is_err());
             drop(socket);
+            fs::remove_file(socket_root.join("socket")).unwrap();
+            assert!(observe(&socket_root, ".control").is_ok());
+            fs::remove_dir(socket_root).unwrap();
         }
         fs::remove_dir_all(parent).unwrap();
     }