fix(jobs): retain domain recovery and establish lifecycle durability once

Keep the original standalone consolidation storage recovery owner beyond fast settlement, with typed temporary causes, capped backoff and current binding/attempt/input/grant/cleanup checks. Permanent constraints and authority loss retain diagnostics without blind execution retries.

Separate successful lifecycle durability establishment from ordinary canonical lock acquisition. Pin and recheck ancestors and scoped lock identity before/after every OS lock wait; never cache DB identity or authorization. Retry all sync steps after failure, reestablish for each new manager, and retain full persistence for configured Host symlinks. Add bounded manager-local diagnostics and actual replacement/failure controls.

Validation: Feature-storage ordinary46/test-support50 and docs; standalone ordinary147+11+9(1ignored)+9; Subject196; Runtime628+20+20; exact-current connected/cleanup15; root check/fmt/names/diff. Server uses package-only opt0/debug0/codegen256, not normal-profile performance proof. Keep historical Journal phase-unknown failure separate; initial fixture failure and later admission-order correction are retained in scratch evidence. Fresh whole-source/all-seven-AC formal review remains pending; no integration, operating DB migration or case closure.

crates/feature-storage/src/identity_tests.rs

@@ -315,6 +315,9 @@ fn reserved_adoption_reconciles_at_each_real_identity_commit_boundary() {
         let lifecycle = owner::Lifecycle {
             root: temp.path().into(),
             scope: SCOPE.into(),
+            established: Arc::default(),
+            #[cfg(feature = "test-support")]
+            observation: Arc::default(),
         };
         let expected = {
             let _lease = lifecycle.lock(true).unwrap();

crates/feature-storage/src/lib.rs

@@ -151,6 +151,11 @@ impl FeatureRegistration {
     }
 }
 
+#[cfg(feature = "test-support")]
+mod lifecycle_observation;
+#[cfg(feature = "test-support")]
+pub use lifecycle_observation::LifecycleObservation;
+
 #[cfg(feature = "test-support")]
 #[path = "open_observation.rs"]
 mod open_observation;
@@ -194,6 +199,9 @@ struct FeatureStorageInner {
     id: Uuid,
     root: PathBuf,
     state: Mutex<FeatureStorageState>,
+    lifecycle_established: Arc<owner::EstablishedLocks>,
+    #[cfg(feature = "test-support")]
+    lifecycle_observation: Arc<lifecycle_observation::Counters>,
 }
 
 struct FeatureStorageState {
@@ -251,10 +259,20 @@ impl FeatureStorage {
                     maintenance: HashSet::new(),
                     deletion_fences: HashSet::new(),
                 }),
+                lifecycle_established: Arc::default(),
+                #[cfg(feature = "test-support")]
+                lifecycle_observation: Arc::default(),
             }),
         }
     }
 
+    /// Read-only cumulative lifecycle diagnostics shared by this manager's Host
+    /// and Feature handles, independent of per-registration open observations.
+    #[cfg(feature = "test-support")]
+    pub fn lifecycle_observation(&self) -> LifecycleObservation {
+        self.inner.lifecycle_observation.snapshot()
+    }
+
     pub fn for_server_database(database_path: impl AsRef<Path>) -> Result<Self> {
         let database_path = database_path.as_ref();
         let parent = database_path.parent().ok_or_else(|| {
@@ -359,6 +377,9 @@ impl FeatureStorage {
                 accepting_operations: AtomicBool::new(true),
                 connection: Mutex::new(Some(connection)),
                 root: self.inner.root.clone(),
+                lifecycle_established: self.inner.lifecycle_established.clone(),
+                #[cfg(feature = "test-support")]
+                lifecycle_observation: self.inner.lifecycle_observation.clone(),
                 path: path.into(),
                 binding,
             }),
@@ -471,6 +492,9 @@ impl FeatureStorage {
         owner::Lifecycle {
             root: self.inner.root.clone(),
             scope: scope_id.into(),
+            established: self.inner.lifecycle_established.clone(),
+            #[cfg(feature = "test-support")]
+            observation: self.inner.lifecycle_observation.clone(),
         }
     }
 
@@ -574,6 +598,9 @@ impl FeatureStorage {
             accepting_operations: AtomicBool::new(true),
             connection: Mutex::new(Some(connection)),
             root: self.inner.root.clone(),
+            lifecycle_established: self.inner.lifecycle_established.clone(),
+            #[cfg(feature = "test-support")]
+            lifecycle_observation: self.inner.lifecycle_observation.clone(),
             path,
             binding,
         });
@@ -897,6 +924,13 @@ pub struct ScopedFeatureStorage {
 pub type WorkspaceFeatureStorage = ScopedFeatureStorage;
 
 impl ScopedFeatureStorage {
+    /// Manager-wide lifecycle totals, including other scopes and retained handles.
+    /// This scope handle neither filters the observation nor changes storage I/O.
+    #[cfg(feature = "test-support")]
+    pub fn lifecycle_observation(&self) -> LifecycleObservation {
+        self.manager.lifecycle_observation()
+    }
+
     pub fn scope_id(&self) -> &str {
         &self.scope_id
     }
@@ -1142,6 +1176,9 @@ struct FeatureDatabaseInner {
     accepting_operations: AtomicBool,
     connection: Mutex<Option<Connection>>,
     root: PathBuf,
+    lifecycle_established: Arc<owner::EstablishedLocks>,
+    #[cfg(feature = "test-support")]
+    lifecycle_observation: Arc<lifecycle_observation::Counters>,
     path: PathBuf,
     binding: owner::Binding,
 }
@@ -1175,6 +1212,9 @@ impl FeatureDatabase {
         let lease = owner::Lifecycle {
             root: self.inner.root.clone(),
             scope: self.inner.scope_id.clone(),
+            established: self.inner.lifecycle_established.clone(),
+            #[cfg(feature = "test-support")]
+            observation: self.inner.lifecycle_observation.clone(),
         }
         .lock(false)?;
         self.inner.ensure_accepting()?;
@@ -1232,6 +1272,9 @@ impl FeatureDatabase {
         let lifecycle = owner::Lifecycle {
             root: self.inner.root.clone(),
             scope: self.inner.scope_id.clone(),
+            established: self.inner.lifecycle_established.clone(),
+            #[cfg(feature = "test-support")]
+            observation: self.inner.lifecycle_observation.clone(),
         };
         let _lease = lifecycle.lock(false).map_err(E::from)?;
         self.inner.ensure_accepting().map_err(E::from)?;
@@ -1308,6 +1351,9 @@ impl FeatureDatabaseInner {
         owner::Lifecycle {
             root: self.root.clone(),
             scope: self.scope_id.clone(),
+            established: self.lifecycle_established.clone(),
+            #[cfg(feature = "test-support")]
+            observation: self.lifecycle_observation.clone(),
         }
         .verify(
             &self.binding,
@@ -1753,6 +1799,149 @@ mod tests {
         FeatureRegistration::new(feature_id, MIGRATIONS)
     }
 
+    #[cfg(feature = "test-support")]
+    #[test]
+    fn lifecycle_observation_is_manager_local_and_counts_retained_feature_operations() {
+        use std::os::unix::fs::MetadataExt;
+
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("storage");
+        let manager = FeatureStorage::new(&root);
+        let observer = manager.clone();
+        // Even managers sharing a physical root must not share observation state.
+        let other = FeatureStorage::new(&root);
+        let feature = manager.register(registration("test-feature")).unwrap();
+        let database = manager.scope("workspace").unwrap().open(&feature).unwrap();
+        assert_eq!(
+            other.lifecycle_observation(),
+            LifecycleObservation::default()
+        );
+        let other_feature = other.register(registration("test-feature")).unwrap();
+        let other_database = other
+            .scope("workspace")
+            .unwrap()
+            .open(&other_feature)
+            .unwrap();
+        assert_eq!(database.identity(), other_database.identity());
+        let identity = database.identity().clone();
+        let lock_path = root.join(".lifecycle/workspace.lock");
+        let lock_inode = fs::metadata(&lock_path).unwrap().ino();
+        let other_before = other.lifecycle_observation();
+        let opens_before = feature.open_observation();
+        let before = observer.lifecycle_observation();
+
+        database
+            .with_connection(|connection| {
+                let synchronous: u32 =
+                    connection.query_row("PRAGMA synchronous", [], |row| row.get(0))?;
+                assert_eq!(synchronous, 2, "durability policy remains FULL");
+                Ok(())
+            })
+            .unwrap();
+        let after_read = observer.lifecycle_observation();
+        let read = after_read.since(before);
+        assert_eq!(read.acquisitions, 1);
+        assert_eq!(
+            read.sync_attempts, 0,
+            "same durable lifecycle needs no re-sync"
+        );
+        assert_eq!(read.lock_attempts, 1);
+        assert_eq!(read.sync_nanoseconds, 0);
+        assert!(read.lock_nanoseconds > 0);
+        assert_eq!((read.failed_syncs, read.failed_locks), (0, 0));
+
+        insert(&database.clone(), "observed commit").unwrap();
+        let after_write = observer.lifecycle_observation();
+        let write = after_write.since(after_read);
+        assert_eq!(write.acquisitions, 1);
+        assert_eq!(write.sync_attempts, 0);
+        assert_eq!(write.lock_attempts, 1);
+        assert_eq!(write.sync_nanoseconds, 0);
+        assert!(write.lock_nanoseconds > 0);
+        assert_eq!((write.failed_syncs, write.failed_locks), (0, 0));
+        assert_eq!(after_write, manager.lifecycle_observation());
+        assert_eq!(other_before, other.lifecycle_observation());
+        assert_eq!(opens_before, feature.open_observation());
+        assert_eq!(database.identity(), &identity);
+        assert_eq!(fs::metadata(&lock_path).unwrap().ino(), lock_inode);
+        // Re-read committed data through the independent manager's verified DB.
+        assert_eq!(values(&other_database).unwrap(), ["observed commit"]);
+    }
+
+    #[cfg(feature = "test-support")]
+    #[test]
+    fn lifecycle_observation_includes_retained_host_catalog_operations() {
+        let temp = tempfile::tempdir().unwrap();
+        let manager = FeatureStorage::new(temp.path().join("storage"));
+        let database = manager
+            .open_host_catalog::<FeatureStorageError>(
+                &temp.path().join("host.sqlite"),
+                "host-provision",
+                "isolated test Host",
+                |connection| {
+                    connection.execute_batch(
+                        "CREATE TABLE items (id INTEGER PRIMARY KEY, value TEXT NOT NULL UNIQUE);",
+                    )?;
+                    Ok(())
+                },
+            )
+            .unwrap();
+        let before = manager.lifecycle_observation();
+        assert_eq!(before.acquisitions, 1);
+        assert_eq!(before.sync_attempts, 1);
+        assert_eq!(before.lock_attempts, 1);
+        let identity = database.identity().clone();
+        insert(&database, "host commit").unwrap();
+        assert_eq!(values(&database).unwrap(), ["host commit"]);
+        let delta = manager.lifecycle_observation().since(before);
+        assert_eq!(delta.acquisitions, 2);
+        assert_eq!(delta.sync_attempts, 0);
+        assert_eq!(delta.lock_attempts, 2);
+        assert_eq!(delta.sync_nanoseconds, 0);
+        assert!(delta.lock_nanoseconds > 0);
+        assert_eq!((delta.failed_syncs, delta.failed_locks), (0, 0));
+        assert_eq!(database.identity(), &identity);
+    }
+
+    #[cfg(feature = "test-support")]
+    #[test]
+    fn lifecycle_observation_counts_failed_directory_attempts_without_suppressing_retry() {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("storage");
+        fs::create_dir(&root).unwrap();
+        let obstacle = root.join(".lifecycle");
+        fs::write(&obstacle, b"not a directory").unwrap();
+        let manager = FeatureStorage::new(&root);
+        let feature = manager.register(registration("test-feature")).unwrap();
+        let scoped = manager.scope("workspace").unwrap();
+        for _ in 0..2 {
+            assert!(matches!(
+                scoped.open(&feature),
+                Err(FeatureStorageError::Io(_))
+            ));
+        }
+        let failed = manager.lifecycle_observation();
+        assert_eq!(failed.acquisitions, 2);
+        assert_eq!(failed.sync_attempts, 2);
+        assert_eq!(failed.failed_syncs, 2);
+        assert!(failed.sync_nanoseconds > 0);
+        assert_eq!(failed.lock_attempts, 0);
+        assert_eq!(failed.lock_nanoseconds, 0);
+        assert_eq!(failed.failed_locks, 0);
+        assert_eq!(feature.open_observation(), OpenObservation::default());
+        assert!(!root.join("workspace/features/test-feature.sqlite").exists());
+
+        fs::remove_file(obstacle).unwrap();
+        let database = scoped.open(&feature).unwrap();
+        insert(&database, "retry committed").unwrap();
+        assert_eq!(values(&database).unwrap(), ["retry committed"]);
+        let retried = manager.lifecycle_observation().since(failed);
+        assert!(retried.acquisitions > 0);
+        assert_eq!(retried.sync_attempts, 1);
+        assert_eq!(retried.lock_attempts, retried.acquisitions);
+        assert_eq!((retried.failed_syncs, retried.failed_locks), (0, 0));
+    }
+
     #[test]
     fn metadata_ancestor_links_are_synced_before_sqlite_initialization() {
         let temp = tempfile::tempdir().unwrap();

crates/feature-storage/src/lifecycle_lock.rs

@@ -0,0 +1,219 @@
+//! Durability establishment is local to a manager, not inferred from existence.
+//! Retained descriptors pin the established ancestor chain and scoped lock inode;
+//! each acquisition rechecks the configured/canonical paths before and after the
+//! OS lock. A replacement is rejected for the lifetime of that proof. New
+//! managers (including new processes) repeat all persistence steps. Failed
+//! establishment publishes no proof. Configured Host symlinks remain supported,
+//! but their mutable links require full establishment on every acquisition.
+//!
+//! This is only durability evidence. The caller must still verify the current
+//! owner catalog, DB identity, fences and authorization under the returned lease.
+//! As with the owner, advisory checks require a Host-protected filesystem; they
+//! do not sandbox native writers racing after the final path check.
+use super::*;
+
+/// Instance-local durability evidence, shared only by a manager and its handles.
+/// It never substitutes for the owner catalog, DB identity, fences or authority.
+#[derive(Default)]
+pub(crate) struct EstablishedLocks {
+    scopes: Mutex<HashMap<String, Arc<Proof>>>,
+}
+
+struct Directory {
+    path: PathBuf,
+    file: File,
+    symlink: bool,
+}
+
+struct Proof {
+    root: PathBuf,
+    canonical_root: PathBuf,
+    directories: Vec<Directory>,
+    lock_path: PathBuf,
+    lock: File,
+    reusable: bool,
+}
+
+fn conflict() -> FeatureStorageError {
+    FeatureStorageError::Identity("canonical lifecycle lock path was replaced or redirected".into())
+}
+
+fn same_file(left: &fs::Metadata, right: &fs::Metadata) -> bool {
+    left.dev() == right.dev() && left.ino() == right.ino()
+}
+
+fn regular_lock(path: &Path) -> Result<fs::Metadata> {
+    let metadata = fs::symlink_metadata(path)?;
+    if !metadata.is_file() {
+        return Err(conflict());
+    }
+    Ok(metadata)
+}
+
+impl Proof {
+    fn capture(root: &Path, scope: &str) -> Result<Self> {
+        let root = if root.is_absolute() {
+            root.to_path_buf()
+        } else {
+            std::env::current_dir()?.join(root)
+        };
+        let directory = root.join(".lifecycle");
+        // A configured Host root may contain symlinks. The lock namespace itself
+        // must not: following a replacement there could split the advisory lock.
+        if !fs::symlink_metadata(&directory)?.is_dir() {
+            return Err(conflict());
+        }
+        let canonical_root = fs::canonicalize(&root)?;
+        let resolved = fs::canonicalize(&directory)?;
+        if resolved != canonical_root.join(".lifecycle") {
+            return Err(conflict());
+        }
+        let mut directories = Vec::new();
+        let mut seen = HashSet::new();
+        for path in directory.ancestors().chain(resolved.ancestors()) {
+            if seen.insert(path.to_path_buf()) {
+                let symlink = fs::symlink_metadata(path)?.is_symlink();
+                let file = File::open(path)?;
+                if !file.metadata()?.is_dir() {
+                    return Err(conflict());
+                }
+                directories.push(Directory {
+                    path: path.into(),
+                    file,
+                    symlink,
+                });
+            }
+        }
+        let lock_path = directory.join(format!("{scope}.lock"));
+        match fs::symlink_metadata(&lock_path) {
+            Ok(_) => {
+                regular_lock(&lock_path)?;
+            }
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error.into()),
+        }
+        let lock = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .open(&lock_path)?;
+        let proof = Self {
+            root,
+            canonical_root,
+            reusable: directories.iter().all(|entry| !entry.symlink),
+            directories,
+            lock_path,
+            lock,
+        };
+        proof.check()?;
+        Ok(proof)
+    }
+
+    fn check(&self) -> Result<()> {
+        if fs::canonicalize(&self.root)? != self.canonical_root {
+            return Err(conflict());
+        }
+        for directory in &self.directories {
+            let link = fs::symlink_metadata(&directory.path)?;
+            if link.is_symlink() != directory.symlink
+                || !same_file(&fs::metadata(&directory.path)?, &directory.file.metadata()?)
+            {
+                return Err(conflict());
+            }
+        }
+        if !same_file(&regular_lock(&self.lock_path)?, &self.lock.metadata()?)
+            || fs::canonicalize(&self.lock_path)?
+                != self
+                    .canonical_root
+                    .join(".lifecycle")
+                    .join(self.lock_path.file_name().ok_or_else(conflict)?)
+        {
+            return Err(conflict());
+        }
+        Ok(())
+    }
+
+    fn open(&self) -> Result<File> {
+        self.check()?;
+        // A fresh open description is required for each flock lease. Cloning
+        // the pinned File would share one OS lock across independent operations.
+        let lock = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(&self.lock_path)?;
+        if !same_file(&lock.metadata()?, &self.lock.metadata()?) {
+            return Err(conflict());
+        }
+        Ok(lock)
+    }
+}
+
+impl Lifecycle {
+    pub(super) fn lock_with_sync(
+        &self,
+        exclusive: bool,
+        sync: &mut impl FnMut(&Path) -> std::io::Result<()>,
+    ) -> Result<File> {
+        #[cfg(feature = "test-support")]
+        self.observation.acquisition();
+        let proof = {
+            let mut scopes = self.established.scopes.lock().map_err(|_| {
+                FeatureStorageError::Operation("lifecycle establishment mutex poisoned".into())
+            })?;
+            let previous = scopes.get(&self.scope);
+            if let Some(proof) = previous {
+                // Replacements are rejected, not silently accepted as a new
+                // lock domain while an old manager/handle might still hold it.
+                proof.check()?;
+            }
+            if let Some(proof) = previous.filter(|proof| proof.reusable) {
+                proof.clone()
+            } else {
+                #[cfg(feature = "test-support")]
+                let started = self.observation.sync_start();
+                let result = (|| {
+                    let directory = self.root.join(".lifecycle");
+                    fs::create_dir_all(&directory)?;
+                    let proof = Proof::capture(&self.root, &self.scope)?;
+                    // Re-sync even existing ancestors: their presence is not
+                    // evidence that an earlier attempt persisted their links.
+                    durable_create_directory(&directory, sync)?;
+                    proof.lock.sync_all()?;
+                    sync(&directory)?;
+                    proof.check()?;
+                    Ok::<_, FeatureStorageError>(Arc::new(proof))
+                })();
+                #[cfg(feature = "test-support")]
+                self.observation.sync_finished(started, result.is_err());
+                let proof = result?;
+                // Publish only after every durability step and path recheck.
+                // Pinned descriptors prevent inode reuse from reviving proof.
+                scopes.insert(self.scope.clone(), proof.clone());
+                proof
+            }
+        };
+        let lock = proof.open()?;
+        #[cfg(feature = "test-support")]
+        let started = self.observation.lock_start();
+        let result = if exclusive {
+            lock.lock()
+        } else {
+            lock.lock_shared()
+        };
+        #[cfg(feature = "test-support")]
+        self.observation.lock_finished(started, result.is_err());
+        result?;
+        // Do not hold the instance mutex while waiting for another process.
+        // Check again after the OS wait, before the caller can touch its DB.
+        proof.check()?;
+        if !same_file(&lock.metadata()?, &proof.lock.metadata()?) {
+            return Err(conflict());
+        }
+        Ok(lock)
+    }
+}
+
+#[cfg(test)]
+#[path = "lifecycle_lock_tests.rs"]
+mod tests;

crates/feature-storage/src/lifecycle_lock_tests.rs

@@ -0,0 +1,255 @@
+//! Component tests use the production manager, lifecycle locks and filesystem.
+//! Only directory sync failure is injected; DB identity checks remain production.
+use super::*;
+use std::os::unix::fs::symlink;
+
+fn sync_paths(lifecycle: &Lifecycle) -> Vec<PathBuf> {
+    let mut paths = Vec::new();
+    drop(
+        lifecycle
+            .lock_with_sync(false, &mut |path| {
+                paths.push(path.into());
+                sync_directory(path)
+            })
+            .unwrap(),
+    );
+    paths
+}
+
+#[test]
+fn failed_establishment_retries_every_link_even_when_directories_and_lock_exist() {
+    let reference = tempfile::tempdir().unwrap();
+    let reference_manager = FeatureStorage::new(reference.path().join("nested/storage"));
+    let count = sync_paths(&reference_manager.lifecycle("scope")).len();
+    assert!(count > 2, "ancestor chain and final lock-directory sync");
+    for fail_at in 0..count {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("nested/storage");
+        let manager = FeatureStorage::new(&root);
+        let lifecycle = manager.lifecycle("scope");
+        let mut calls = 0;
+        let result = lifecycle.lock_with_sync(false, &mut |path| {
+            let index = calls;
+            calls += 1;
+            if index == fail_at {
+                Err(std::io::Error::other("injected directory sync failure"))
+            } else {
+                sync_directory(path)
+            }
+        });
+        assert!(
+            matches!(result, Err(FeatureStorageError::Io(_))),
+            "{fail_at}"
+        );
+        assert_eq!(calls, fail_at + 1);
+        assert!(root.join(".lifecycle/scope.lock").is_file());
+        assert!(lifecycle.established.scopes.lock().unwrap().is_empty());
+        let retried = sync_paths(&lifecycle);
+        assert_eq!(
+            retried.len(),
+            count,
+            "all durability steps retry at {fail_at}"
+        );
+        assert_eq!(
+            retried.first(),
+            retried.last(),
+            "lock directory persisted last"
+        );
+        drop(
+            lifecycle
+                .lock_with_sync(false, &mut |_| panic!("already established"))
+                .unwrap(),
+        );
+    }
+}
+
+#[test]
+fn establishment_is_shared_with_handles_but_not_new_managers_or_scopes() {
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("storage");
+    let manager = FeatureStorage::new(&root);
+    let paths = sync_paths(&manager.lifecycle("scope"));
+    assert!(!paths.is_empty());
+    assert!(sync_paths(&manager.clone().lifecycle("scope")).is_empty());
+    assert_eq!(sync_paths(&manager.lifecycle("other")), paths);
+    let other = FeatureStorage::new(&root);
+    assert_eq!(
+        sync_paths(&other.lifecycle("scope")),
+        paths,
+        "existence is not proof"
+    );
+    assert_eq!(
+        sync_paths(&FeatureStorage::new(&root).lifecycle("scope")),
+        paths
+    );
+}
+
+fn open_database(root: &Path) -> (FeatureStorage, RegisteredFeature, FeatureDatabase) {
+    let manager = FeatureStorage::new(root);
+    let registration = manager
+        .register(FeatureRegistration::new("feature", &[]))
+        .unwrap();
+    let database = manager.scope("scope").unwrap().open(&registration).unwrap();
+    database
+        .transaction(|tx| {
+            tx.execute_batch(
+                "CREATE TABLE marker(value TEXT); INSERT INTO marker VALUES('persisted');",
+            )?;
+            Ok(())
+        })
+        .unwrap();
+    (manager, registration, database)
+}
+
+fn rejected(
+    manager: &FeatureStorage,
+    registration: &RegisteredFeature,
+    database: &FeatureDatabase,
+) {
+    assert!(
+        database
+            .with_connection(|_| -> Result<()> { panic!("replaced lock cannot enter DB closure") })
+            .is_err()
+    );
+    assert!(database.guard(database.identity()).is_err());
+    assert!(manager.scope("scope").unwrap().open(registration).is_err());
+}
+
+#[test]
+fn physical_root_directory_lock_and_ancestor_replacement_reject_retained_access() {
+    for replacement in ["root", "directory", "lock", "ancestor"] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("parent/storage");
+        let (manager, registration, database) = open_database(&root);
+        let path = match replacement {
+            "root" => root.clone(),
+            "directory" => root.join(".lifecycle"),
+            "lock" => root.join(".lifecycle/scope.lock"),
+            "ancestor" => root.parent().unwrap().to_path_buf(),
+            _ => unreachable!(),
+        };
+        let moved = temp.path().join("moved");
+        fs::rename(&path, &moved).unwrap();
+        if replacement == "lock" {
+            fs::write(&path, b"").unwrap();
+        } else {
+            fs::create_dir_all(&path).unwrap();
+        }
+        rejected(&manager, &registration, &database);
+        // Rejecting must not erase the old proof and accept the replacement on retry.
+        rejected(&manager, &registration, &database);
+        fs::remove_file(&path)
+            .or_else(|_| fs::remove_dir_all(&path))
+            .unwrap();
+        fs::rename(&moved, &path).unwrap();
+        let value: String = database
+            .with_connection(|connection| {
+                Ok(connection.query_row("SELECT value FROM marker", [], |r| r.get(0))?)
+            })
+            .unwrap();
+        assert_eq!(value, "persisted");
+    }
+}
+
+#[test]
+fn symlink_replacement_cannot_redirect_or_split_an_established_lock() {
+    for replacement in ["root", "directory", "lock"] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("storage");
+        let (manager, registration, database) = open_database(&root);
+        let path = match replacement {
+            "root" => root.clone(),
+            "directory" => root.join(".lifecycle"),
+            "lock" => root.join(".lifecycle/scope.lock"),
+            _ => unreachable!(),
+        };
+        let moved = temp.path().join("moved");
+        fs::rename(&path, &moved).unwrap();
+        symlink(&moved, &path).unwrap();
+        // The target inode is still the original; following metadata alone is insufficient.
+        rejected(&manager, &registration, &database);
+        fs::remove_file(&path).unwrap();
+        fs::rename(&moved, &path).unwrap();
+        drop(database.guard(database.identity()).unwrap());
+    }
+}
+
+#[test]
+fn initial_lock_namespace_symlinks_are_rejected_without_establishment_proof() {
+    for directory_link in [true, false] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("storage");
+        fs::create_dir(&root).unwrap();
+        let target = temp.path().join("target");
+        if directory_link {
+            fs::create_dir(&target).unwrap();
+            symlink(&target, root.join(".lifecycle")).unwrap();
+        } else {
+            fs::create_dir(root.join(".lifecycle")).unwrap();
+            fs::write(&target, b"").unwrap();
+            symlink(&target, root.join(".lifecycle/scope.lock")).unwrap();
+        }
+        let manager = FeatureStorage::new(&root);
+        let lifecycle = manager.lifecycle("scope");
+        assert!(matches!(
+            lifecycle.lock(false),
+            Err(FeatureStorageError::Identity(_))
+        ));
+        assert!(lifecycle.established.scopes.lock().unwrap().is_empty());
+    }
+}
+
+#[test]
+fn configured_host_symlinks_reestablish_and_reject_changed_canonical_root() {
+    let temp = tempfile::tempdir().unwrap();
+    let target = temp.path().join("target");
+    fs::create_dir(&target).unwrap();
+    let root = temp.path().join("alias");
+    symlink(&target, &root).unwrap();
+    let manager = FeatureStorage::new(&root);
+    let lifecycle = manager.lifecycle("scope");
+    let paths = sync_paths(&lifecycle);
+    assert!(paths.contains(&target));
+    // Mutable symlink links are not pinned by ordinary directory descriptors;
+    // therefore no successful proof can suppress their durability steps.
+    assert_eq!(sync_paths(&lifecycle), paths);
+    let next = temp.path().join("next");
+    fs::create_dir(&next).unwrap();
+    fs::remove_file(&root).unwrap();
+    symlink(&next, &root).unwrap();
+    assert!(matches!(
+        lifecycle.lock(false),
+        Err(FeatureStorageError::Identity(_))
+    ));
+}
+
+#[cfg(feature = "test-support")]
+#[test]
+fn replacement_during_os_lock_wait_is_rejected_before_database_access() {
+    use std::time::{Duration, Instant};
+
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().join("storage");
+    let (manager, registration, database) = open_database(&root);
+    let guard = database.guard(database.identity()).unwrap();
+    let attempts = manager.lifecycle_observation().lock_attempts;
+    let lifecycle = manager.lifecycle("scope");
+    let waiter = std::thread::spawn(move || lifecycle.lock(true));
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while manager.lifecycle_observation().lock_attempts == attempts {
+        assert!(
+            Instant::now() < deadline,
+            "waiter did not reach OS lock call"
+        );
+        std::thread::yield_now();
+    }
+    let lock_path = root.join(".lifecycle/scope.lock");
+    fs::rename(&lock_path, root.join(".lifecycle/held.lock")).unwrap();
+    fs::write(&lock_path, b"").unwrap();
+    drop(guard);
+    assert!(matches!(
+        waiter.join().unwrap(),
+        Err(FeatureStorageError::Identity(_))
+    ));
+    rejected(&manager, &registration, &database);
+}

crates/feature-storage/src/lifecycle_observation.rs

@@ -0,0 +1,114 @@
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Instant;
+
+/// Cumulative lifecycle observations for one storage manager and its clones/handles.
+/// Counts attempts before starting I/O, including attempts that return errors.
+/// `acquisitions` counts entries to lifecycle acquisition; `sync_attempts` counts
+/// lifecycle durability establishment attempts (the full ancestor sync plus
+/// lockfile and lock-directory persistence), and `lock_attempts` counts OS
+/// shared/exclusive lock calls, excluding lockfile open failures. Once this
+/// manager has established and rechecked the same physical paths, ordinary
+/// acquisitions need no establishment attempt. A fresh manager establishes again.
+///
+/// Durations sum monotonic elapsed nanoseconds around those calls, including
+/// errors and OS lock waits, but not lease holding time or database operations.
+/// Overlapping threads contribute independently: aggregates are not wall time.
+/// Fields saturate at u64::MAX and are independently sampled with relaxed loads;
+/// a concurrent snapshot need not describe one atomic point in time. In-flight
+/// calls contribute attempts before their duration or failure is recorded.
+/// These diagnostics retain no database or authority and are not termination,
+/// durability, or successful-work evidence; they never control storage I/O.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct LifecycleObservation {
+    pub acquisitions: u64,
+    pub sync_attempts: u64,
+    pub sync_nanoseconds: u64,
+    pub failed_syncs: u64,
+    pub lock_attempts: u64,
+    pub lock_nanoseconds: u64,
+    pub failed_locks: u64,
+}
+
+impl LifecycleObservation {
+    /// Saturating differences between snapshots from the same manager.
+    pub fn since(self, before: Self) -> Self {
+        Self {
+            acquisitions: self.acquisitions.saturating_sub(before.acquisitions),
+            sync_attempts: self.sync_attempts.saturating_sub(before.sync_attempts),
+            sync_nanoseconds: self
+                .sync_nanoseconds
+                .saturating_sub(before.sync_nanoseconds),
+            failed_syncs: self.failed_syncs.saturating_sub(before.failed_syncs),
+            lock_attempts: self.lock_attempts.saturating_sub(before.lock_attempts),
+            lock_nanoseconds: self
+                .lock_nanoseconds
+                .saturating_sub(before.lock_nanoseconds),
+            failed_locks: self.failed_locks.saturating_sub(before.failed_locks),
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+pub(super) struct Counters {
+    acquisitions: AtomicU64,
+    sync_attempts: AtomicU64,
+    sync_nanoseconds: AtomicU64,
+    failed_syncs: AtomicU64,
+    lock_attempts: AtomicU64,
+    lock_nanoseconds: AtomicU64,
+    failed_locks: AtomicU64,
+}
+
+impl Counters {
+    pub(super) fn snapshot(&self) -> LifecycleObservation {
+        LifecycleObservation {
+            acquisitions: self.acquisitions.load(Ordering::Relaxed),
+            sync_attempts: self.sync_attempts.load(Ordering::Relaxed),
+            sync_nanoseconds: self.sync_nanoseconds.load(Ordering::Relaxed),
+            failed_syncs: self.failed_syncs.load(Ordering::Relaxed),
+            lock_attempts: self.lock_attempts.load(Ordering::Relaxed),
+            lock_nanoseconds: self.lock_nanoseconds.load(Ordering::Relaxed),
+            failed_locks: self.failed_locks.load(Ordering::Relaxed),
+        }
+    }
+
+    pub(super) fn acquisition(&self) {
+        add(&self.acquisitions, 1);
+    }
+
+    pub(super) fn sync_start(&self) -> Instant {
+        add(&self.sync_attempts, 1);
+        Instant::now()
+    }
+
+    pub(super) fn sync_finished(&self, started: Instant, failed: bool) {
+        let elapsed = started.elapsed();
+        add(&self.sync_nanoseconds, nanoseconds(elapsed));
+        if failed {
+            add(&self.failed_syncs, 1);
+        }
+    }
+
+    pub(super) fn lock_start(&self) -> Instant {
+        add(&self.lock_attempts, 1);
+        Instant::now()
+    }
+
+    pub(super) fn lock_finished(&self, started: Instant, failed: bool) {
+        let elapsed = started.elapsed();
+        add(&self.lock_nanoseconds, nanoseconds(elapsed));
+        if failed {
+            add(&self.failed_locks, 1);
+        }
+    }
+}
+
+fn nanoseconds(duration: std::time::Duration) -> u64 {
+    duration.as_nanos().min(u128::from(u64::MAX)) as u64
+}
+
+fn add(counter: &AtomicU64, amount: u64) {
+    let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
+        Some(value.saturating_add(amount))
+    });
+}

crates/feature-storage/src/owner.rs

@@ -16,6 +16,10 @@ use std::os::unix::fs::MetadataExt;
 
 const IDENTITY_TABLE: &str = "__yoi_storage_identity";
 
+#[path = "lifecycle_lock.rs"]
+mod lifecycle_lock;
+pub(super) use lifecycle_lock::EstablishedLocks;
+
 /// Expected storage target, retained with authorization and rechecked on resume.
 /// This is a physical identity, not an authorization credential or a Job lease.
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -36,27 +40,14 @@ pub(super) struct Binding {
 pub(super) struct Lifecycle {
     pub root: PathBuf,
     pub scope: String,
+    pub established: Arc<EstablishedLocks>,
+    #[cfg(feature = "test-support")]
+    pub observation: Arc<lifecycle_observation::Counters>,
 }
 
 impl Lifecycle {
     pub fn lock(&self, exclusive: bool) -> Result<File> {
-        durable_create_directory(&self.root.join(".lifecycle"), &mut sync_directory)?;
-        let lock = OpenOptions::new()
-            .read(true)
-            .write(true)
-            .create(true)
-            .truncate(false)
-            .open(
-                self.root
-                    .join(".lifecycle")
-                    .join(format!("{}.lock", self.scope)),
-            )?;
-        if exclusive {
-            lock.lock()?;
-        } else {
-            lock.lock_shared()?;
-        }
-        Ok(lock)
+        self.lock_with_sync(exclusive, &mut sync_directory)
     }
 
     fn catalog(&self, create: bool) -> Result<Connection> {

crates/standalone/src/subjektiv.rs

@@ -381,30 +381,80 @@ enum ConsolidationRecoveryError {
     Service(#[from] crate::jobs::JobServiceError),
     #[error(transparent)]
     Domain(#[from] subjektiv::api::OperationError),
+    #[error(transparent)]
+    Writer(#[from] subjektiv::journal::JournalError),
 }
 
 impl ConsolidationRecoveryError {
     fn retryable_read_or_storage(&self) -> bool {
         use crate::jobs::{JobServiceError, JobStoreError};
-        matches!(
-            self,
-            Self::Service(JobServiceError::Store(
-                JobStoreError::Sqlite(_) | JobStoreError::Storage(_) | JobStoreError::Io(_)
-            )) | Self::Domain(subjektiv::api::OperationError::Storage(_))
-            // The Host's writer/Subject liveness reads wrap storage errors in
-            // Unavailable. Recheck them within this same finite read-only budget;
-            // their failure NEVER authorizes a receipt, release, or execution.
-            | Self::Host(SubjektivHostError::Unavailable(_))
-        )
+        fn sqlite(error: &rusqlite::Error) -> bool {
+            matches!(error, rusqlite::Error::SqliteFailure(cause, _)
+                if matches!(cause.code,
+                    rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
+                    | rusqlite::ErrorCode::SystemIoFailure | rusqlite::ErrorCode::DiskFull
+                    | rusqlite::ErrorCode::CannotOpen))
+        }
+        fn io(error: &std::io::Error) -> bool {
+            matches!(
+                error.kind(),
+                std::io::ErrorKind::Interrupted
+                    | std::io::ErrorKind::WouldBlock
+                    | std::io::ErrorKind::TimedOut
+            )
+        }
+        fn storage(error: &feature_storage::FeatureStorageError) -> bool {
+            match error {
+                feature_storage::FeatureStorageError::Sqlite(error) => sqlite(error),
+                feature_storage::FeatureStorageError::Io(error) => io(error),
+                _ => false,
+            }
+        }
+        match self {
+            Self::Service(JobServiceError::Store(JobStoreError::Sqlite(error))) => sqlite(error),
+            Self::Service(JobServiceError::Store(JobStoreError::Io(error))) => io(error),
+            Self::Service(JobServiceError::Store(JobStoreError::Storage(error)))
+            | Self::Domain(subjektiv::api::OperationError::Storage(error))
+            | Self::Writer(subjektiv::journal::JournalError::Storage(error))
+            | Self::Writer(subjektiv::journal::JournalError::Subject(
+                subjektiv::SubjektivError::Storage(error),
+            )) => storage(error),
+            // Authority loss, permanent SQL constraints, malformed data and
+            // untyped Host unavailability never acquire a storage retry owner.
+            _ => false,
+        }
     }
 }
 
+#[derive(Debug, PartialEq, Eq)]
+enum ConsolidationRecoveryProgress {
+    FastRetry,
+    RetainedStorage,
+    Blocked,
+}
+
 struct ConsolidationRecoveryOwner {
-    running: bool,
+    progress: ConsolidationRecoveryProgress,
     passes: usize,
     error: ConsolidationRecoveryError,
 }
 
+impl ConsolidationRecoveryOwner {
+    fn next_action(&self) -> &'static str {
+        match self.progress {
+            ConsolidationRecoveryProgress::FastRetry => {
+                "reconcile the same attempt after short storage backoff"
+            }
+            ConsolidationRecoveryProgress::RetainedStorage => {
+                "recheck the original binding/attempt/input/grant and same domain operation with capped storage backoff"
+            }
+            ConsolidationRecoveryProgress::Blocked => {
+                "resolve the recorded permanent storage or authority cause; no automatic execution or cleanup"
+            }
+        }
+    }
+}
+
 pub(crate) struct LocalSubjectHost {
     writer: subjektiv::journal::JournalWriterBinding,
     self_weak: std::sync::Weak<Self>,
@@ -813,6 +863,23 @@ impl LocalSubjectHost {
         self.recover_journals();
     }
 
+    /// Recheck the actual writer store without erasing its typed storage cause.
+    /// A released writer/closed Host is authority loss, not a transient outage.
+    fn ensure_consolidation_live(&self) -> Result<(), ConsolidationRecoveryError> {
+        if !self.live.load(Ordering::Acquire) {
+            return Err(
+                SubjektivHostError::Unavailable("local Subject Host is closed".into()).into(),
+            );
+        }
+        if !self.catalog.store.journal_writer_active(&self.writer)? {
+            return Err(SubjektivHostError::Unavailable(
+                "Subject writer binding was released".into(),
+            )
+            .into());
+        }
+        Ok(())
+    }
+
     /// Prefer a complete saved result. Otherwise close only this failed attempt's
     /// domain reservation, retaining unresolved work and the original failure.
     /// Neither path reruns the finalizer or proves prior execution shutdown.
@@ -821,7 +888,7 @@ impl LocalSubjectHost {
         jobs: &crate::jobs::StandaloneJobs,
         observed: &crate::jobs::JobSnapshot,
     ) -> Result<bool, ConsolidationRecoveryError> {
-        self.ensure_live()?;
+        self.ensure_consolidation_live()?;
         let current = jobs.get(&observed.request.job_id)?;
         if current.attempt.attempt_id != observed.attempt.attempt_id
             || current.input_digest != observed.input_digest
@@ -852,7 +919,7 @@ impl LocalSubjectHost {
             &current.request.job_id,
             &current.attempt.attempt_id,
         )? {
-            self.ensure_live()?;
+            self.ensure_consolidation_live()?;
             let settled = jobs.accept_reconciled(&job::JobResultSubmission {
                 job_id: current.request.job_id,
                 attempt_id: current.attempt.attempt_id,
@@ -891,7 +958,7 @@ impl LocalSubjectHost {
             &current.input_digest,
         ))
         .map_err(|error| SubjektivHostError::InvalidArgument(error.to_string()))?;
-        self.ensure_live()?;
+        self.ensure_consolidation_live()?;
         // Recheck the exact Host/grant before persistence. The service/store also
         // fence the current attempt and independently require ALL executions released.
         let checked = jobs.get(&current.request.job_id)?;
@@ -910,23 +977,16 @@ impl LocalSubjectHost {
         Ok(true)
     }
 
-    /// Keep a finite recovery owner for read/storage failure of this exact
-    /// snapshot. No startup-only repair, disconnect polling, new attempt or model
-    /// call. Exhaustion retains the typed cause and original durable obligation;
-    /// a fresh Host can reconcile it, never infer success from budget exhaustion.
+    /// Retain only typed temporary storage failures of this exact operation.
+    /// After eight fast passes the same owner uses capped backoff, not another
+    /// execution budget or a healthy-other-store probe. Each pass rereads the
+    /// current binding/attempt/input/grant/retirement and actual domain receipts.
     fn own_consolidation_recovery(
         &self,
         jobs: crate::jobs::StandaloneJobs,
         observed: crate::jobs::JobSnapshot,
         error: ConsolidationRecoveryError,
     ) {
-        eprintln!(
-            "standalone consolidation Job {} reconciliation blocked: {error}",
-            observed.request.job_id
-        );
-        if !error.retryable_read_or_storage() {
-            return;
-        }
         let key = (
             observed.request.job_id.clone(),
             observed.attempt.attempt_id.clone(),
@@ -936,14 +996,8 @@ impl LocalSubjectHost {
             .consolidation_recovery
             .lock()
             .expect("consolidation recovery poisoned");
-        if let Some(owner) = owners.get(&key) {
-            if !owner.running {
-                eprintln!(
-                    "standalone consolidation Job {} retains exhausted reconciliation after {} passes: {}; next action: resolve cause and restore the same Host binding",
-                    key.0, owner.passes, owner.error
-                );
-            }
-            return; // Repeated disconnect cannot refill a running or exhausted budget.
+        if owners.contains_key(&key) {
+            return; // Duplicate notifications cannot refill a budget or revive a blocked owner.
         }
         let Some(host) = self.self_weak.upgrade() else {
             return;
@@ -951,18 +1005,42 @@ impl LocalSubjectHost {
         let Ok(runtime) = tokio::runtime::Handle::try_current() else {
             return;
         };
-        owners.insert(
-            key.clone(),
-            ConsolidationRecoveryOwner {
-                running: true,
-                passes: 1,
-                error,
+        let retry = error.retryable_read_or_storage();
+        let owner = ConsolidationRecoveryOwner {
+            progress: if retry {
+                ConsolidationRecoveryProgress::FastRetry
+            } else {
+                ConsolidationRecoveryProgress::Blocked
             },
+            passes: 1,
+            error,
+        };
+        eprintln!(
+            "standalone consolidation Job {} reconciliation {:?} after {} passes: {}; next action: {}",
+            key.0,
+            owner.progress,
+            owner.passes,
+            owner.error,
+            owner.next_action()
         );
+        owners.insert(key.clone(), owner);
+        if !retry {
+            return;
+        }
         runtime.spawn(async move {
-            const PASSES: usize = 8;
-            for pass in 2..=PASSES {
-                tokio::time::sleep(std::time::Duration::from_millis(100 << (pass - 2).min(5))).await;
+            const FAST_PASSES: usize = 8;
+            let mut pass = 1usize;
+            let mut retained_delay = std::time::Duration::from_secs(1);
+            loop {
+                let delay = if pass < FAST_PASSES {
+                    std::time::Duration::from_millis(100 << (pass - 1).min(5))
+                } else {
+                    let delay = retained_delay;
+                    retained_delay = (retained_delay * 2).min(std::time::Duration::from_secs(30));
+                    delay
+                };
+                tokio::time::sleep(delay).await;
+                pass = pass.saturating_add(1);
                 let result = host.recover_consolidation_job(&jobs, &observed);
                 let mut owners = host.consolidation_recovery.lock().expect("consolidation recovery poisoned");
                 match result {
@@ -973,15 +1051,18 @@ impl LocalSubjectHost {
                         return;
                     }
                     Err(error) => {
-                        let retry = error.retryable_read_or_storage() && pass < PASSES;
+                        let retry = error.retryable_read_or_storage();
                         let owner = owners.get_mut(&key).expect("retained consolidation owner");
                         owner.passes = pass;
-                        owner.running = retry;
                         owner.error = error;
-                        if !retry {
-                            eprintln!("standalone consolidation Job {} reconciliation stopped after {} passes: {}; original Failed/input/receipts retained; next action: resolve storage/authority failure and restore the same Host binding", key.0, owner.passes, owner.error);
-                            return;
+                        owner.progress = if !retry { ConsolidationRecoveryProgress::Blocked }
+                            else if pass < FAST_PASSES { ConsolidationRecoveryProgress::FastRetry }
+                            else { ConsolidationRecoveryProgress::RetainedStorage };
+                        if !retry || pass == FAST_PASSES {
+                            eprintln!("standalone consolidation Job {} reconciliation {:?} after {} passes: {}; original input/receipts retained; next action: {}",
+                                key.0, owner.progress, owner.passes, owner.error, owner.next_action());
                         }
+                        if !retry { return; }
                     }
                 }
             }
@@ -991,7 +1072,7 @@ impl LocalSubjectHost {
     fn consolidation_grant(
         &self,
         snapshot: &crate::jobs::JobSnapshot,
-    ) -> Result<LocalConsolidationGrant, SubjektivHostError> {
+    ) -> Result<LocalConsolidationGrant, ConsolidationRecoveryError> {
         let grant: LocalConsolidationGrant =
             serde_json::from_value(snapshot.domain_grant.clone().ok_or_else(|| {
                 SubjektivHostError::InvalidArgument("missing explicit Subject delegation".into())
@@ -1004,16 +1085,17 @@ impl LocalSubjectHost {
         {
             return Err(SubjektivHostError::InvalidArgument(
                 "persisted Job grant/input does not match local Subject capability".into(),
-            ));
+            )
+            .into());
         }
         let subject = self
             .catalog
             .store
             .subject(&self.binding.subject_id)
-            .map_err(domain_error)?
+            .map_err(subjektiv::api::subjektiv_store_error)?
             .ok_or_else(|| SubjektivHostError::Unavailable("Subject missing".into()))?;
         if subject.state != subjektiv::SubjectState::Active {
-            return Err(SubjektivHostError::Unavailable("Subject is retired".into()));
+            return Err(SubjektivHostError::Unavailable("Subject is retired".into()).into());
         }
         Ok(grant)
     }

crates/standalone/src/subjektiv/tests.rs

@@ -99,6 +99,16 @@ impl ConsolidationRecoveryFixture {
         }
     }
 
+    fn domain_path(&self) -> PathBuf {
+        self.body
+            .host
+            .catalog
+            .root
+            .join("features")
+            .join(&self.body.host.catalog.scope_id)
+            .join("features/subjektiv.sqlite")
+    }
+
     fn domain_closure(&self) -> Option<String> {
         use rusqlite::OptionalExtension;
         rusqlite::Connection::open(&self.body.jobs_path)
@@ -356,8 +366,226 @@ async fn failed_consolidation_retains_unreleased_stale_inactive_migrated_and_era
     }
 }
 
+fn assert_real_busy(error: &ConsolidationRecoveryError, domain: bool) {
+    let cause = match error {
+        ConsolidationRecoveryError::Domain(subjektiv::api::OperationError::Storage(
+            feature_storage::FeatureStorageError::Sqlite(rusqlite::Error::SqliteFailure(cause, _)),
+        )) if domain => cause,
+        ConsolidationRecoveryError::Service(crate::jobs::JobServiceError::Store(
+            crate::jobs::JobStoreError::Sqlite(rusqlite::Error::SqliteFailure(cause, _)),
+        )) if !domain => cause,
+        other => panic!("expected typed Busy from original domain={domain}: {other:?}"),
+    };
+    assert_eq!(cause.code, rusqlite::ErrorCode::DatabaseBusy);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_consolidation_real_sqlite_busy_beyond_fast_budget_recovers_without_restore() {
+    real_sqlite_busy_consolidation_recovery(false, None).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_consolidation_real_domain_busy_beyond_fast_budget_recovers_without_restore() {
+    real_sqlite_busy_consolidation_recovery(true, None).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_consolidation_real_busy_retained_owner_rejects_stale_domain() {
+    real_sqlite_busy_consolidation_recovery(true, Some("stale")).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_consolidation_real_busy_retained_owner_rejects_retired_input() {
+    real_sqlite_busy_consolidation_recovery(true, Some("retired")).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_consolidation_real_busy_retained_owner_requires_actual_execution_cleanup() {
+    real_sqlite_busy_consolidation_recovery(true, Some("cleanup")).await;
+}
+
+async fn real_sqlite_busy_consolidation_recovery(domain: bool, control: Option<&str>) {
+    let mut fixture = ConsolidationRecoveryFixture::new();
+    if control != Some("cleanup") {
+        fixture.confirm_execution_joined();
+    }
+    if control == Some("stale") {
+        fixture.generation(&fixture.saved.attempt.attempt_id);
+        fixture
+            .body
+            .host
+            .catalog
+            .store
+            .create_memory(
+                &fixture.body.binding.subject_id,
+                subjektiv::MemoryDraft::active(
+                    memory::extract::CandidateKind::Lesson,
+                    "changed",
+                    "body",
+                    "reason",
+                    "change",
+                ),
+            )
+            .unwrap();
+    }
+    let before = fixture.saved.clone();
+    let path = if domain {
+        fixture.domain_path()
+    } else {
+        fixture.body.jobs_path.clone()
+    };
+    let connection = rusqlite::Connection::open(path).unwrap();
+    connection.execute_batch("BEGIN IMMEDIATE;").unwrap();
+    let error = fixture
+        .body
+        .host
+        .recover_consolidation_job(&fixture.jobs, &before)
+        .unwrap_err();
+    assert_real_busy(&error, domain);
+    eprintln!(
+        "S1 real SQLite Busy domain={domain}; unchanged exact attempt; holding through pass8"
+    );
+    fixture
+        .body
+        .host
+        .own_consolidation_recovery(fixture.jobs.clone(), before.clone(), error);
+    tokio::time::timeout(std::time::Duration::from_secs(60), async {
+        loop {
+            let reached = fixture
+                .body
+                .host
+                .consolidation_recovery
+                .lock()
+                .unwrap()
+                .values()
+                .any(|owner| owner.passes >= 8);
+            if reached {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("real SQLite fault must remain through all eight passes");
+    {
+        let owners = fixture.body.host.consolidation_recovery.lock().unwrap();
+        assert_eq!(owners.len(), 1);
+        let owner = owners.values().next().unwrap();
+        assert_eq!(
+            owner.progress,
+            ConsolidationRecoveryProgress::RetainedStorage
+        );
+        assert_real_busy(&owner.error, domain);
+        assert!(owner.next_action().contains("same domain operation"));
+    }
+    assert!(fixture.domain_closure().is_none());
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    if control == Some("retired") {
+        // While the original Feature DB is STILL unavailable, explicitly erase
+        // the Job's original input. A healthy Job DB is not Feature recovery.
+        fixture
+            .jobs
+            .reclaim_input(
+                &before.request.job_id,
+                &before.input_digest,
+                crate::jobs::JobInputRetirementReason::ExplicitErasure,
+            )
+            .unwrap();
+    }
+    let expected = fixture.jobs.get(&before.request.job_id).unwrap();
+    eprintln!("S1 pass8 observed while real lock still held; releasing only SQLite transaction");
+    connection.execute_batch("ROLLBACK;").unwrap();
+    // No restore, recover_jobs, notification, repeated own call, new Job, model or finalizer.
+    if let Some(control) = control {
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            loop {
+                let owners = fixture.body.host.consolidation_recovery.lock().unwrap();
+                let stopped = if control == "cleanup" {
+                    owners.is_empty()
+                } else {
+                    owners
+                        .values()
+                        .any(|owner| owner.progress == ConsolidationRecoveryProgress::Blocked)
+                };
+                drop(owners);
+                if stopped {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("retained owner must honor current stale/retirement/cleanup refusal");
+        assert!(fixture.domain_closure().is_none(), "{control}");
+        assert_eq!(
+            fixture.jobs.get(&before.request.job_id).unwrap(),
+            expected,
+            "{control}"
+        );
+        assert_eq!(
+            expected.executions, before.executions,
+            "no invented cleanup proof"
+        );
+        if control == "cleanup" {
+            assert!(!expected.executions_released());
+            assert!(
+                expected
+                    .executions
+                    .iter()
+                    .all(|execution| !execution.writes_revoked)
+            );
+            assert!(!fixture.jobs.unreleased_granted().unwrap().is_empty());
+        } else {
+            let owners = fixture.body.host.consolidation_recovery.lock().unwrap();
+            let owner = owners.values().next().unwrap();
+            assert!(owner.passes > 8);
+            assert!(!owner.error.retryable_read_or_storage());
+            assert!(
+                owner
+                    .next_action()
+                    .contains("no automatic execution or cleanup")
+            );
+        }
+        return;
+    }
+    tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        while fixture.domain_closure().is_none() {
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("same original domain obligation must recover after only fault release");
+    assert!(fixture.jobs.unreleased_granted().unwrap().is_empty());
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    assert!(
+        fixture
+            .body
+            .host
+            .consolidation_recovery
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    let retained = subjektiv::job::inspect_abandoned_attempt(
+        &fixture.body.host.catalog.store,
+        &fixture.body.binding.subject_id,
+        &["candidate".into()],
+        &before.request.job_id,
+        &before.attempt.attempt_id,
+    )
+    .unwrap();
+    assert!(
+        retained.prepared_generation_id.is_none(),
+        "no finalizer replay"
+    );
+    assert!(retained.disposition_ids.is_empty(), "no model effects");
+    assert_eq!(retained.unresolved_candidate_ids, vec!["candidate"]);
+    let reopened = crate::job_store::JobStore::open(&fixture.body.jobs_path).unwrap();
+    assert_eq!(reopened.get(&before.request.job_id).unwrap(), before);
+}
+
 #[tokio::test]
-async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt_recovery() {
+async fn failed_consolidation_permanent_receipt_constraint_retains_cause_without_retry() {
     let mut fixture = ConsolidationRecoveryFixture::new();
     fixture.confirm_execution_joined();
     let before = fixture.saved.clone();
@@ -375,7 +603,7 @@ async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt
         .host
         .recover_consolidation_job(&fixture.jobs, &before)
         .unwrap_err();
-    assert!(error.retryable_read_or_storage());
+    assert!(!error.retryable_read_or_storage());
     assert!(fixture.domain_closure().is_none());
     fixture
         .body
@@ -384,24 +612,23 @@ async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt
     connection
         .execute_batch("DROP TRIGGER fail_domain_closure;")
         .unwrap();
-    // No disconnect, recover_jobs call, restart, new Job, model or finalizer.
-    tokio::time::timeout(std::time::Duration::from_secs(5), async {
-        while fixture.domain_closure().is_none() {
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .unwrap();
-    assert!(fixture.jobs.unreleased_granted().unwrap().is_empty());
+    // Removing a permanent constraint does not reclassify it as a transient
+    // outage or refill the same owner's budget. Diagnose it, never blind retry.
+    tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+    assert!(fixture.domain_closure().is_none());
     assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    let owners = fixture.body.host.consolidation_recovery.lock().unwrap();
+    let owner = owners.values().next().unwrap();
+    assert_eq!(owner.progress, ConsolidationRecoveryProgress::Blocked);
+    assert_eq!(owner.passes, 1);
+    assert!(matches!(&owner.error,
+        ConsolidationRecoveryError::Service(crate::jobs::JobServiceError::Store(
+            crate::jobs::JobStoreError::Sqlite(rusqlite::Error::SqliteFailure(cause, _))))
+        if cause.code == rusqlite::ErrorCode::ConstraintViolation));
     assert!(
-        fixture
-            .body
-            .host
-            .consolidation_recovery
-            .lock()
-            .unwrap()
-            .is_empty()
+        owner
+            .next_action()
+            .contains("permanent storage or authority")
     );
 }
 

crates/workspace-server/src/server/tests/backend_job_tests.rs

@@ -1161,6 +1161,21 @@ async fn backend_job_cleanup_commit_failure_retries_old_attempt_without_touching
         api.retry_backend_job(&request.job_id).is_err(),
         "retry must not overlap an incompletely cleaned attempt"
     );
+    // This is a permanent SQL constraint, not a transient-storage recovery test.
+    // Wait for the failed owner's admission lease to leave before explicitly
+    // sweeping; otherwise that request can coalesce with the still-ending owner.
+    // Map absence here is only admission ordering, never execution/cleanup proof.
+    tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        while api
+            .backend_job_recovery
+            .cleanup_owner_ids()
+            .contains(&first.attempt.attempt_id)
+        {
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("permanent-failure cleanup owner must return before explicit sweep");
     api.config_store
         .with_conn(|conn| {
             conn.execute_batch("DROP TRIGGER fail_job_cleanup_catalog_commit;")?;

crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs

@@ -56,9 +56,27 @@ struct ConnectedJournalModel {
     dialogs: Arc<Mutex<Vec<Request>>>,
     journal_inputs: Arc<Mutex<Vec<String>>>,
     journal_observations: Arc<Mutex<Vec<Value>>>,
+    lifecycle_storage: Arc<Mutex<Option<crate::WorkspaceFeatureStorage>>>,
+    clock: Arc<Mutex<Option<std::time::Instant>>>,
+    http_observations: Arc<Mutex<Vec<Value>>>,
+    poll_observations: Arc<Mutex<Vec<Value>>>,
 }
 
 impl ConnectedJournalModel {
+    fn elapsed_micros(&self) -> u64 {
+        self.clock.lock().unwrap().as_ref().map_or(0, |clock| {
+            clock.elapsed().as_micros().min(u64::MAX as u128) as u64
+        })
+    }
+
+    fn observe_bounded(observations: &Mutex<Vec<Value>>, value: Value) {
+        let mut observations = observations.lock().unwrap();
+        if observations.len() == 64 {
+            observations.remove(0);
+        }
+        observations.push(value);
+    }
+
     fn journal_response(request: &Request) -> Vec<Event> {
         // Consume the actual Controller-rendered input, not a fixture generation or
         // a Source tool response. Decode also checks complete bodies/ranges/digests.
@@ -264,6 +282,14 @@ impl LlmClient for ConnectedJournalModel {
             }
             observations.push(json!({
                 "observed_at": Utc::now().to_rfc3339(),
+                "elapsed_micros": self.elapsed_micros(),
+                "lifecycle_manager": self.lifecycle_storage.lock().unwrap().as_ref()
+                    .map(|storage| { let timing = storage.lifecycle_observation(); json!({
+                        "acquisitions":timing.acquisitions,"sync_attempts":timing.sync_attempts,
+                        "sync_nanoseconds":timing.sync_nanoseconds,"lock_attempts":timing.lock_attempts,
+                        "lock_nanoseconds":timing.lock_nanoseconds,"failed_syncs":timing.failed_syncs,
+                        "failed_locks":timing.failed_locks,
+                    }) }),
                 "tool_results": request.items.iter().rev().filter_map(|item| match item {
                     Item::ToolResult { call_id, is_error, summary, .. } =>
                         Some(json!({"call_id":call_id,"is_error":is_error,"summary":summary.chars().take(256).collect::<String>()})),
@@ -352,15 +378,17 @@ async fn connected_wait_for_recording(
 ) {
     let host = JournalHostContext::validated_session(store, attr.clone()).unwrap();
     tokio::time::timeout(std::time::Duration::from_secs(30), async {
-        let mut tick = tokio::time::interval(std::time::Duration::from_millis(25));
         loop {
-            tick.tick().await;
+            let poll_started = model.elapsed_micros();
             let jobs = api
                 .store
                 .list_backend_journal_jobs(TEST_WORKSPACE_ID, &attr.session_id, None, 100)
                 .unwrap();
-            let generation_closed =
-                store
+            let ready = !jobs.is_empty()
+                && jobs
+                    .iter()
+                    .all(|job| job.state == BackendJobState::Completed)
+                && store
                     .journal_for_session(&host)
                     .unwrap()
                     .is_some_and(|journal| {
@@ -368,22 +396,21 @@ async fn connected_wait_for_recording(
                             .active_journal_generation(&host, &journal)
                             .unwrap()
                             .is_none()
-                    });
-            if !jobs.is_empty()
-                && generation_closed
-                && jobs
-                    .iter()
-                    .all(|job| job.state == BackendJobState::Completed)
+                    })
                 && !store.journal_retention_required(&host).unwrap()
                 && api
                     .runtime
                     .worker(&RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id))
                     .unwrap()
                     .worker_state
-                    .is_some_and(|state| state.catalog_status() == protocol::WorkerStatus::Idle)
-            {
+                    .is_some_and(|state| state.catalog_status() == protocol::WorkerStatus::Idle);
+            ConnectedJournalModel::observe_bounded(&model.poll_observations,
+                json!({"phase":phase,"start_micros":poll_started,
+                    "end_micros":model.elapsed_micros()}));
+            if ready {
                 break;
             }
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
         }
     })
     .await
@@ -399,12 +426,20 @@ async fn connected_wait_for_recording(
             (job.request.job_id.clone(), history, workers)
         }).collect::<Vec<_>>();
         panic!(
-            "real Controller recording did not complete: phase={phase}; jobs={jobs:?}; attempts={attempts:?}; recovery_owners={:?}; cleanup_owners={:?}; model_observations={:?}",
+            "real Controller recording did not complete: phase={phase}; jobs={jobs:?}; attempts={attempts:?}; recovery_owners={:?}; cleanup_owners={:?}; model_observations={:?}; http_observations={:?}; poll_observations={:?}",
             api.backend_job_recovery.recovery_owner_ids(),
             api.backend_job_recovery.cleanup_owner_ids(),
             model.journal_observations.lock().unwrap(),
+            model.http_observations.lock().unwrap(),
+            model.poll_observations.lock().unwrap(),
         )
     });
+    eprintln!(
+        "recording progress timing: phase={phase}; model={:?}; http={:?}; polls={:?}",
+        model.journal_observations.lock().unwrap(),
+        model.http_observations.lock().unwrap(),
+        model.poll_observations.lock().unwrap()
+    );
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]
@@ -457,14 +492,25 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup()
     .await
     .unwrap();
     seed_test_delivery_binding(&api);
+    *model.clock.lock().unwrap() = Some(std::time::Instant::now());
+    *model.lifecycle_storage.lock().unwrap() = Some(api.feature_storage.clone());
     let requests = Arc::new(Mutex::new(Vec::new()));
     let observed = requests.clone();
+    let timing_model = model.clone();
     let app = build_router(api.clone()).layer(axum::middleware::from_fn(
         move |request: axum::extract::Request, next: axum::middleware::Next| {
             let observed = observed.clone();
+            let timing_model = timing_model.clone();
             async move {
                 let path = request.uri().path().to_string();
+                let started = timing_model.elapsed_micros();
+                ConnectedJournalModel::observe_bounded(&timing_model.http_observations,
+                    json!({"event":"ingress","path":path.chars().take(256).collect::<String>(),"start_micros":started}));
                 let response = next.run(request).await;
+                ConnectedJournalModel::observe_bounded(&timing_model.http_observations,
+                    json!({"event":"return","path":path.chars().take(256).collect::<String>(),
+                        "start_micros":started,"end_micros":timing_model.elapsed_micros(),
+                        "status":response.status().as_u16()}));
                 observed.lock().unwrap().push((path, response.status()));
                 response
             }