diff --git a/benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json b/benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json
new file mode 100644
index 0000000000..b790f460a4
--- /dev/null
+++ b/benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json
@@ -0,0 +1,12 @@
+{
+  "slice_class": "non_ranking",
+  "targeted_mechanism": "none",
+  "kind": "harness",
+  "fixtures": [
+    "harness-goldens"
+  ],
+  "slices": [
+    "v3-ledger-busy"
+  ],
+  "note": "Views v3 ledger: busy contention reported as busy/retry, not unreadable or schema drift. No changed path matches RANKING_FENCE_PREFIXES."
+}
diff --git a/crates/aft/src/blob_store/v3.rs b/crates/aft/src/blob_store/v3.rs
index b5f24928dd..0f5f920667 100644
--- a/crates/aft/src/blob_store/v3.rs
+++ b/crates/aft/src/blob_store/v3.rs
@@ -29,6 +29,7 @@ pub const NAMESPACE_BUDGET: u64 = 4 * 1024 * 1024 * 1024;
 /// ages them out by epoch instead of by marking.
 pub const SYMBOL_RETENTION_SWEEPS: i64 = 8;
 const TIMEOUT_MS: i64 = 5_000;
+pub(crate) const BUSY_PREFIX: &str = "v3 store busy: ";
 mod connections;
 mod copy;
 mod initialization;
@@ -156,6 +157,12 @@ pub enum Error {
     Sqlite(rusqlite::Error),
     Incompatible(UnsupportedPersistedFormat),
     Refused(&'static str),
+    Busy(&'static str),
+    Ledger {
+        path: PathBuf,
+        operation: &'static str,
+        source: Box<Error>,
+    },
     InvalidIdentity,
     ReservationRegistryMismatch,
 }
@@ -166,12 +173,77 @@ impl fmt::Display for Error {
             Self::Sqlite(e) => write!(f, "{e}"),
             Self::Incompatible(e) => write!(f, "{e}"),
             Self::Refused(reason) => f.write_str(reason),
+            Self::Busy(reason) => write!(f, "{BUSY_PREFIX}{reason}"),
+            Self::Ledger {
+                path,
+                operation,
+                source,
+            } => write!(
+                f,
+                "{}: {operation} {}: {source}",
+                if self.is_busy() {
+                    "reservation_ledger_busy"
+                } else {
+                    "reservation_ledger_unreadable"
+                },
+                path.display()
+            ),
             Self::InvalidIdentity => f.write_str("invalid v3 family, scope or lock identity"),
             Self::ReservationRegistryMismatch => f.write_str("reservation_registry_mismatch"),
         }
     }
 }
-impl std::error::Error for Error {}
+impl std::error::Error for Error {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Io(error) => Some(error),
+            Self::Sqlite(error) => Some(error),
+            Self::Ledger { source, .. } => Some(source.as_ref()),
+            _ => None,
+        }
+    }
+}
+impl Error {
+    /// Contention is retryable; unreadable or incompatible ledgers are not
+    /// treated as empty accounting or as transient contention.
+    pub fn is_busy(&self) -> bool {
+        match self {
+            Self::Busy(_) => true,
+            Self::Sqlite(rusqlite::Error::SqliteFailure(error, _)) => matches!(
+                error.code,
+                rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
+            ),
+            Self::Io(error) => Self::io_is_busy(error),
+            Self::Ledger { source, .. } => source.is_busy(),
+            _ => false,
+        }
+    }
+
+    fn ledger(path: &Path, operation: &'static str, source: impl Into<Error>) -> Self {
+        Self::Ledger {
+            path: path.to_path_buf(),
+            operation,
+            source: Box::new(source.into()),
+        }
+    }
+
+    fn io_is_busy(error: &std::io::Error) -> bool {
+        // Path context must not hide a Windows sharing violation or a nested
+        // WouldBlock error from the ledger's retry classification.
+        let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(error);
+        while let Some(error) = cause {
+            if let Some(error) = error.downcast_ref::<std::io::Error>() {
+                if error.kind() == std::io::ErrorKind::WouldBlock
+                    || cfg!(windows) && matches!(error.raw_os_error(), Some(32 | 33))
+                {
+                    return true;
+                }
+            }
+            cause = error.source();
+        }
+        false
+    }
+}
 impl From<std::io::Error> for Error {
     fn from(e: std::io::Error) -> Self {
         Self::Io(e)
@@ -541,25 +613,50 @@ fn schema_rows(c: &Connection) -> rusqlite::Result<Vec<SchemaRow>> {
     c.prepare("SELECT type,name,tbl_name,sql FROM sqlite_schema WHERE substr(name,1,7) != 'sqlite_' ORDER BY type,name,tbl_name,sql")?
         .query_map([], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?)))?.collect()
 }
+fn format_read<T>(
+    result: rusqlite::Result<T>,
+    path: &Path,
+    operation: &'static str,
+) -> Result<Option<T>> {
+    match result {
+        Ok(value) => Ok(Some(value)),
+        Err(error) => {
+            let error = Error::Sqlite(error);
+            // A locked format table is not evidence of a missing or corrupt
+            // schema. Other read failures retain the format-drift checks.
+            if error.is_busy() {
+                Err(Error::ledger(path, operation, error))
+            } else {
+                Ok(None)
+            }
+        }
+    }
+}
 fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> {
     #[cfg(test)]
     let _profile = work_profile::span("validate_database");
-    let version = c
-        .query_row(
+    let version = format_read(
+        c.query_row(
             "SELECT format_version FROM format_meta WHERE singleton=1",
             [],
             |r| r.get::<_, u64>(0),
-        )
-        .unwrap_or(0);
+        ),
+        path,
+        "read format version",
+    )?
+    .unwrap_or(0);
     let displayed = fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
     persisted_format::gate(kind.store(), &displayed, path, Some(version))
         .map_err(Error::Incompatible)?;
     if version == 0 {
         return Err(incompatible(kind, path, 0, IncompatibleReason::SchemaDrift));
     }
-    let app = c
-        .query_row("PRAGMA application_id", [], |r| r.get::<_, i64>(0))
-        .unwrap_or(0);
+    let app = format_read(
+        c.query_row("PRAGMA application_id", [], |r| r.get::<_, i64>(0)),
+        path,
+        "read application id",
+    )?
+    .unwrap_or(0);
     if app != kind.app_id() {
         return Err(incompatible(
             kind,
@@ -568,13 +665,16 @@ fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> {
             IncompatibleReason::ApplicationId,
         ));
     }
-    let (schema, layout, text): (u32, u32, String) = c
-        .query_row(
+    let (schema, layout, text): (u32, u32, String) = format_read(
+        c.query_row(
             "SELECT schema_version,layout_tag,schema_text FROM format_meta WHERE singleton=1",
             [],
             |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
-        )
-        .map_err(|_| incompatible(kind, path, version, IncompatibleReason::SchemaDrift))?;
+        ),
+        path,
+        "read format schema",
+    )?
+    .ok_or_else(|| incompatible(kind, path, version, IncompatibleReason::SchemaDrift))?;
     let compiled_schema = match (kind, version, schema) {
         // A registry of the base-reference protocol carries its own exact
         // text; any other version/schema pair falls through to the drift check.
@@ -611,8 +711,9 @@ fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> {
     };
     if !known_version
         || text != format!("{META}{compiled_schema}")
-        || schema_rows(c).ok() != Some(schema_rows(&expected)?)
-        || !initialization::has_seed(c, kind)
+        || format_read(schema_rows(c), path, "read schema objects")?
+            != Some(schema_rows(&expected)?)
+        || !format_read(initialization::has_seed(c, kind), path, "read seed")?.unwrap_or(false)
     {
         return Err(incompatible(
             kind,
@@ -808,14 +909,14 @@ fn for_each_registry(
         let path = entry.path().join("members.sqlite");
         if !path
             .try_exists()
-            .map_err(|e| crate::views::io::at("checking v3 registry", &path, e))?
+            .map_err(|error| Error::ledger(&path, "checking v3 registry", error))?
         {
             continue;
         }
         let name = entry.file_name();
         body(
             name.to_str()
-                .ok_or(Error::Refused("reservation_ledger_unreadable"))?,
+                .ok_or_else(|| Error::ledger(&path, "decode family", Error::InvalidIdentity))?,
             &path,
         )?;
     }
@@ -853,13 +954,13 @@ fn preflight_floor(storage: &Path, request: u64) -> Result<()> {
 }
 fn recover_ledgers(storage: &Path, lock: &ReservationLock) -> Result<()> {
     lock.verify(storage)?;
-    for_each_registry(storage, |family, _| {
+    for_each_registry(storage, |family, path| {
         let recover = LedgerReader::open_existing(storage, family)
             .and_then(|reader| reader.reservation_owners())
             .map_or(true, |(_, abandoned)| abandoned);
         if recover {
             RegistryWriter::open_existing(storage, family, lock)
-                .map_err(|_| Error::Refused("reservation_ledger_unreadable"))?;
+                .map_err(|error| Error::ledger(path, "recover registry", error))?;
         }
         Ok(())
     })
@@ -892,15 +993,19 @@ fn check_floor(storage: &Path, request: u64, own: Option<(&Path, &Connection)>)
                 continue;
             }
             let registry = entry.path().join("members.sqlite");
-            if !registry.try_exists().map_err(|e| {
-                crate::views::io::at("checking v3 reservation registry", &registry, e)
+            if !registry.try_exists().map_err(|error| {
+                Error::ledger(&registry, "checking v3 reservation registry", error)
             })? {
                 continue;
             }
             let name = entry.file_name();
-            let name = name
-                .to_str()
-                .ok_or(Error::Refused("reservation_ledger_unreadable"))?;
+            let name = name.to_str().ok_or_else(|| {
+                Error::ledger(
+                    &registry,
+                    "decode family for admission",
+                    Error::InvalidIdentity,
+                )
+            })?;
             let own_connection = own.filter(|(path, _)| {
                 registry == *path
                     || crate::db::file_identity::identity_of(&registry).is_some_and(|identity| {
@@ -912,7 +1017,9 @@ fn check_floor(storage: &Path, request: u64, own: Option<(&Path, &Connection)>)
                     || LedgerReader::open_existing(storage, name).and_then(|r| r.accounting()),
                     |(_, connection)| accounting(connection),
                 )
-                .map_err(|_| Error::Refused("reservation_ledger_unreadable"))?;
+                .map_err(|error| {
+                    Error::ledger(&registry, "read accounting for admission", error)
+                })?;
             reserved = reserved
                 .checked_add(ledger.reserved_bytes)
                 .ok_or(Error::Refused("device_free_space_floor"))?;
@@ -1179,13 +1286,19 @@ impl<'a> RegistryWriter<'a> {
             // walk did not count, so only a complete recheck that finds the
             // same total confirms it. A recheck the deadline stops is not a
             // confirmation.
+            #[cfg(test)]
+            tests::measurement_recheck_hook()?;
             match allocated_bytes_until(&self.dir, deadline)? {
                 Some(again) if again == bytes => return Ok(Measurement::Recorded(bytes)),
                 Some(_) => {}
                 None => return settle_registry_growth(&mut connection, &self.dir, bytes, counted),
             }
         }
-        Err(Error::Refused("reservation_ledger_unreadable"))
+        Err(Error::ledger(
+            &self.dir,
+            "measure registry",
+            Error::Busy("namespace size did not stabilize after 8 measurements"),
+        ))
     }
     pub fn with_barrier<T>(
         &mut self,
@@ -1228,8 +1341,14 @@ impl<'a> RegistryWriter<'a> {
     pub fn reserve(&mut self, request: u64) -> Result<Reservation> {
         let storage = self.lock.storage.clone();
         recover_ledgers(&storage, self.lock)?;
-        let owner_start = crate::root_cache::process_start_time_ms(std::process::id())
-            .ok_or(Error::Refused("reservation_ledger_unreadable"))?;
+        let owner_start =
+            crate::root_cache::process_start_time_ms(std::process::id()).ok_or_else(|| {
+                Error::ledger(
+                    &self.dir,
+                    "reserve",
+                    std::io::Error::other("current process start time unavailable"),
+                )
+            })?;
         let before = allocated_bytes(&self.dir)?;
         let registry = self.registry_identity()?;
         let path = self.dir.join("members.sqlite");
@@ -1343,7 +1462,11 @@ fn settle_registry_growth(
         checkpoint(connection)?;
         recorded = safe;
     }
-    Err(Error::Refused("reservation_ledger_unreadable"))
+    Err(Error::ledger(
+        dir,
+        "settle registry growth",
+        Error::Busy("registry size did not stabilize after 8 corrections"),
+    ))
 }
 
 #[cfg(test)]
@@ -1360,12 +1483,21 @@ pub(crate) fn grow_and_expire_recheck_for_test() {
 }
 
 fn checkpoint(c: &Connection) -> Result<()> {
-    let (busy, _, _): (i64, i64, i64) =
+    let (busy, log, checkpointed): (i64, i64, i64) =
         c.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| {
             Ok((r.get(0)?, r.get(1)?, r.get(2)?))
         })?;
     if busy != 0 {
-        return Err(Error::Refused("reservation_ledger_unreadable"));
+        return Err(Error::ledger(
+            Path::new(c.path().unwrap_or("<unknown>")),
+            "checkpoint WAL",
+            rusqlite::Error::SqliteFailure(
+                rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
+                Some(format!(
+                    "wal_checkpoint(TRUNCATE): busy={busy}, log={log}, checkpointed={checkpointed}"
+                )),
+            ),
+        ));
     }
     Ok(())
 }
diff --git a/crates/aft/src/blob_store/v3/connections.rs b/crates/aft/src/blob_store/v3/connections.rs
index 110d7b2ce5..81dd344546 100644
--- a/crates/aft/src/blob_store/v3/connections.rs
+++ b/crates/aft/src/blob_store/v3/connections.rs
@@ -12,7 +12,9 @@ enum Slot {
         connection: IdentityConnection,
         writable: bool,
     },
-    Closed,
+    Closed {
+        route: PathBuf,
+    },
 }
 
 pub(super) struct Database {
@@ -27,7 +29,7 @@ impl Deref for Guard<'_> {
     fn deref(&self) -> &Self::Target {
         match &*self.0 {
             Slot::Open { connection, .. } => connection,
-            Slot::Closed => unreachable!("guard is created only for an open slot"),
+            Slot::Closed { .. } => unreachable!("guard is created only for an open slot"),
         }
     }
 }
@@ -35,18 +37,31 @@ impl DerefMut for Guard<'_> {
     fn deref_mut(&mut self) -> &mut Self::Target {
         match &mut *self.0 {
             Slot::Open { connection, .. } => connection,
-            Slot::Closed => unreachable!("guard is created only for an open slot"),
+            Slot::Closed { .. } => unreachable!("guard is created only for an open slot"),
         }
     }
 }
 impl Database {
     pub(super) fn connection(&self) -> Result<Guard<'_>> {
-        let slot = self
+        let mut slot = self
             .slot
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
-        if matches!(*slot, Slot::Closed) {
-            return Err(Error::Refused("reservation_ledger_unreadable"));
+        if let Slot::Closed { route } = &*slot {
+            if matches!((self.identity, identity_of(route)), (Some(expected), Some(actual)) if expected != actual)
+            {
+                return Err(Error::Refused("live_sqlite_file_replaced"));
+            }
+            // If replacing a read-only connection with a writable one fails,
+            // existing readers can reopen read-only on demand. A still-held
+            // SQLite lock returns its busy error; a later read retries after
+            // the lock is released.
+            let connection = open_db(route, self.kind, false, true)
+                .map_err(|source| Error::ledger(route, "reopen shared reader", source))?;
+            *slot = Slot::Open {
+                connection,
+                writable: false,
+            };
         }
         Ok(Guard(slot))
     }
@@ -55,7 +70,7 @@ impl Database {
             .slot
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let reopen = matches!(*slot, Slot::Closed)
+        let reopen = matches!(*slot, Slot::Closed { .. })
             || (!readonly
                 && matches!(
                     *slot,
@@ -69,7 +84,9 @@ impl Database {
             // the family is exclusively held, before opening its replacement.
             // Existing read capabilities still refer to this Database object,
             // so later operations use its replacement connection.
-            *slot = Slot::Closed;
+            *slot = Slot::Closed {
+                route: route.to_path_buf(),
+            };
             let connection = open_db(route, self.kind, create, readonly)?;
             *slot = Slot::Open {
                 connection,
diff --git a/crates/aft/src/blob_store/v3/initialization.rs b/crates/aft/src/blob_store/v3/initialization.rs
index e72ed6a2a7..f8d4f3032b 100644
--- a/crates/aft/src/blob_store/v3/initialization.rs
+++ b/crates/aft/src/blob_store/v3/initialization.rs
@@ -4,32 +4,43 @@
 
 use super::*;
 
-fn has_marker(c: &Connection) -> bool {
-    c.query_row("SELECT 1 FROM format_meta WHERE singleton=1", [], |r| {
-        r.get::<_, i64>(0)
-    })
-    .is_ok()
+fn has_marker(c: &Connection, path: &Path) -> Result<bool> {
+    Ok(format_read(
+        c.query_row("SELECT 1 FROM format_meta WHERE singleton=1", [], |r| {
+            r.get::<_, i64>(0)
+        }),
+        path,
+        "read initialization marker",
+    )?
+    .is_some())
 }
-fn has_any_seed(c: &Connection, kind: Kind) -> bool {
+fn has_any_seed(c: &Connection, path: &Path, kind: Kind) -> Result<bool> {
     let sql = match kind {
         Kind::Registry => "SELECT 1 FROM accounting WHERE singleton=1",
         Kind::Plane(_) => "SELECT 1 FROM store_meta WHERE singleton=1",
     };
-    c.query_row(sql, [], |r| r.get::<_, i64>(0)).is_ok()
+    Ok(format_read(
+        c.query_row(sql, [], |r| r.get::<_, i64>(0)),
+        path,
+        "read initialization seed",
+    )?
+    .is_some())
 }
-pub(super) fn has_seed(c: &Connection, kind: Kind) -> bool {
+pub(super) fn has_seed(c: &Connection, kind: Kind) -> rusqlite::Result<bool> {
     match kind {
-        Kind::Registry => has_any_seed(c, kind),
+        Kind::Registry => c.query_row("SELECT 1 FROM accounting WHERE singleton=1", [], |_| {
+            Ok(true)
+        }),
         Kind::Plane(plane) => c
             .query_row("SELECT plane FROM store_meta WHERE singleton=1", [], |r| {
                 r.get::<_, String>(0)
             })
-            .is_ok_and(|stored| stored == plane.as_str()),
+            .map(|stored| stored == plane.as_str()),
     }
 }
 
-fn resumable(c: &Connection, kind: Kind) -> Result<bool> {
-    let Ok(rows) = schema_rows(c) else {
+fn resumable(c: &Connection, path: &Path, kind: Kind) -> Result<bool> {
+    let Some(rows) = format_read(schema_rows(c), path, "read bootstrap schema")? else {
         return Ok(false);
     };
     let app: i64 = c.pragma_query_value(None, "application_id", |r| r.get(0))?;
@@ -121,10 +132,10 @@ fn resumable(c: &Connection, kind: Kind) -> Result<bool> {
 }
 
 pub(super) fn ensure(c: &mut IdentityConnection, path: &Path, kind: Kind) -> Result<()> {
-    if has_marker(c) && has_any_seed(c, kind) {
+    if has_marker(c, path)? && has_any_seed(c, path, kind)? {
         return Ok(());
     }
-    if !resumable(c, kind)? {
+    if !resumable(c, path, kind)? {
         return Ok(());
     }
     // SQLite must select pointer-map mode before the first write transaction.
@@ -133,10 +144,10 @@ pub(super) fn ensure(c: &mut IdentityConnection, path: &Path, kind: Kind) -> Res
     #[cfg(test)]
     tests::initialization_hook(kind, "auto_vacuum");
     let tx = c.transaction_with_behavior(TransactionBehavior::Immediate)?;
-    if has_marker(&tx) && has_any_seed(&tx, kind) {
+    if has_marker(&tx, path)? && has_any_seed(&tx, path, kind)? {
         return Ok(());
     }
-    if !resumable(&tx, kind)? {
+    if !resumable(&tx, path, kind)? {
         return Ok(());
     }
     // Old interrupted bootstraps can contain a prefix of the schema. Only
diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs
index 8c1dfea3de..573cb43d73 100644
--- a/crates/aft/src/blob_store/v3/tests.rs
+++ b/crates/aft/src/blob_store/v3/tests.rs
@@ -134,6 +134,172 @@ fn db(storage: &Path, family: &str) -> PathBuf {
     family_dir(storage, family).unwrap().join("members.sqlite")
 }
 
+type RecheckHook = Box<dyn FnMut() -> Result<()>>;
+thread_local! {
+    static RECHECK_HOOK: std::cell::RefCell<Option<RecheckHook>> = const { std::cell::RefCell::new(None) };
+}
+pub(super) fn measurement_recheck_hook() -> Result<()> {
+    RECHECK_HOOK.with(|hook| match &mut *hook.borrow_mut() {
+        Some(hook) => hook(),
+        None => Ok(()),
+    })
+}
+
+#[test]
+fn path_context_keeps_io_contention_classifiable() {
+    let path = Path::new("busy-ledger/members.sqlite");
+    let error = crate::views::io::at(
+        "checking v3 registry",
+        path,
+        std::io::Error::from(std::io::ErrorKind::WouldBlock),
+    );
+    let error = Error::Io(std::io::Error::other(error));
+    assert!(error.is_busy(), "{error}");
+    for code in [32, 33] {
+        let error = crate::views::io::at(
+            "resolving v3 database",
+            path,
+            std::io::Error::from_raw_os_error(code),
+        );
+        let error = Error::ledger(path, "open registry", error);
+        assert_eq!(error.is_busy(), cfg!(windows), "{error}");
+        let detail = error.to_string();
+        assert!(detail.contains("resolving v3 database"), "{detail}");
+        assert!(detail.contains("busy-ledger/members.sqlite"), "{detail}");
+    }
+}
+
+#[test]
+fn checkpoint_contention_is_busy_and_preserves_sqlite_cause() {
+    let _guard = serial();
+    let root = tempdir().unwrap();
+    seed(root.path(), A);
+    let path = db(root.path(), A);
+    let lock = ReservationLock::acquire(root.path()).unwrap();
+    let mut writer = RegistryWriter::open_for_barrier(root.path(), A, &lock).unwrap();
+    writer
+        .connection()
+        .unwrap()
+        .busy_timeout(std::time::Duration::ZERO)
+        .unwrap();
+    let reader = IdentityConnection::open_with_flags(
+        &path,
+        OpenFlags::SQLITE_OPEN_READ_ONLY,
+        "v3-checkpoint-blocker",
+    )
+    .unwrap();
+    reader
+        .execute_batch("BEGIN; SELECT * FROM accounting;")
+        .unwrap();
+    writer
+        .with_barrier(|tx| {
+            tx.execute("UPDATE accounting SET watermark=watermark+1", [])?;
+            Ok(())
+        })
+        .unwrap();
+
+    let error = writer.covering_measurement().unwrap_err();
+    assert!(error.is_busy(), "{error}");
+    let detail = error.to_string();
+    assert!(
+        detail.starts_with("reservation_ledger_busy: checkpoint WAL"),
+        "{detail}"
+    );
+    assert!(detail.contains("members.sqlite"), "{detail}");
+    assert!(
+        detail.contains("wal_checkpoint(TRUNCATE): busy=1"),
+        "{detail}"
+    );
+    assert!(matches!(error, Error::Ledger { source, .. }
+        if matches!(*source, Error::Sqlite(rusqlite::Error::SqliteFailure(code, _))
+            if code.code == rusqlite::ErrorCode::DatabaseBusy)));
+
+    reader.execute_batch("ROLLBACK").unwrap();
+    writer.covering_measurement().unwrap();
+    assert_eq!(writer.accounting().unwrap().reserved_bytes, 0);
+}
+
+#[test]
+fn busy_validation_is_not_persisted_as_schema_drift() {
+    let _guard = serial();
+    let root = tempdir().unwrap();
+    seed(root.path(), A);
+    let path = db(root.path(), A);
+    let blocker = IdentityConnection::open_with_flags(
+        &path,
+        OpenFlags::SQLITE_OPEN_READ_WRITE,
+        "v3-validation-blocker",
+    )
+    .unwrap();
+    blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap();
+    let error = match LedgerReader::open_existing(root.path(), A) {
+        Ok(_) => panic!("an exclusively held ledger must not validate through another connection"),
+        Err(error) => error,
+    };
+    assert!(error.is_busy(), "{error}");
+    let detail = error.to_string();
+    assert!(
+        detail.starts_with("reservation_ledger_busy: read format version"),
+        "{detail}"
+    );
+    assert!(detail.contains("members.sqlite"), "{detail}");
+    assert!(detail.contains("database is locked"), "{detail}");
+    blocker.execute_batch("ROLLBACK").unwrap();
+    drop(blocker);
+    assert_eq!(
+        LedgerReader::open_existing(root.path(), A)
+            .unwrap()
+            .accounting()
+            .unwrap()
+            .reserved_bytes,
+        0
+    );
+}
+
+#[test]
+fn changing_derived_database_reports_busy_then_measurement_recovers() {
+    let _guard = serial();
+    let root = tempdir().unwrap();
+    seed(root.path(), A);
+    let derived_dir = view_dir(root.path(), A, "churning-view").unwrap();
+    fs::create_dir_all(&derived_dir).unwrap();
+    let lock = ReservationLock::acquire(root.path()).unwrap();
+    let mut writer = RegistryWriter::open_existing(root.path(), A, &lock).unwrap();
+    let writes = std::rc::Rc::new(std::cell::Cell::new(0));
+    let count = writes.clone();
+    // A derived graph can commit outside the root reservation lock. Create a
+    // new generation between each inventory walk and its confirmation, so
+    // growth does not depend on the filesystem's allocation granularity.
+    RECHECK_HOOK.with(|hook| {
+        *hook.borrow_mut() = Some(Box::new(move || {
+            let derived = IdentityConnection::open(
+                derived_dir.join(format!("derived-{}.sqlite", count.get())),
+                "v3-measurement-churn",
+            )?;
+            derived.execute_batch(
+                "CREATE TABLE growing(payload BLOB); INSERT INTO growing VALUES(zeroblob(16384));",
+            )?;
+            count.set(count.get() + 1);
+            Ok(())
+        }))
+    });
+    let result = writer.covering_measurement();
+    RECHECK_HOOK.with(|hook| *hook.borrow_mut() = None);
+    let error = result.unwrap_err();
+    assert_eq!(writes.get(), 8);
+    assert!(error.is_busy(), "{error}");
+    let detail = error.to_string();
+    assert!(
+        detail.starts_with("reservation_ledger_busy: measure registry"),
+        "{detail}"
+    );
+    assert!(detail.contains("8 measurements"), "{detail}");
+
+    let measured = writer.covering_measurement().unwrap();
+    assert_eq!(measured, allocated_bytes(writer.family_dir()).unwrap());
+    assert_eq!(writer.accounting().unwrap().reserved_bytes, 0);
+}
+
 #[test]
 fn floor_refusal_before_creation_and_budget_recheck_leave_no_plane_or_row() {
     let _guard = serial();
@@ -233,7 +399,14 @@ fn unreadable_peer_ledger_refuses_instead_of_ignoring_reserved_bytes() {
     fs::create_dir_all(&dir).unwrap();
     fs::write(dir.join("members.sqlite"), b"not sqlite").unwrap();
     let error = publish(root.path(), B, 100, None, |_| Ok(())).unwrap_err();
-    assert_eq!(error.to_string(), "reservation_ledger_unreadable");
+    let detail = error.to_string();
+    assert!(!error.is_busy(), "{detail}");
+    assert!(
+        detail.starts_with("reservation_ledger_unreadable: recover registry"),
+        "{detail}"
+    );
+    assert!(detail.contains("members.sqlite"), "{detail}");
+    assert!(detail.contains("schema_drift"), "{detail}");
     assert!(!db(root.path(), B).exists());
 }
 
@@ -1027,6 +1200,86 @@ fn member_kind_constraint_and_v3_namespace_are_not_v2_registration() {
         .starts_with(family_dir(root.path(), A).unwrap()));
 }
 
+fn blocked_read_only_upgrade(root: &Path) -> (LedgerReader, IdentityConnection, ReservationLock) {
+    seed(root, A);
+    let path = db(root, A);
+    let blocker = IdentityConnection::open_with_flags(
+        &path,
+        OpenFlags::SQLITE_OPEN_READ_WRITE,
+        "v3-upgrade-blocker",
+    )
+    .unwrap();
+    // Rollback mode permits an exclusive lock while the idle read-only handle
+    // is still open, making a failed connection handoff deterministic.
+    blocker.execute_batch("PRAGMA journal_mode=DELETE").unwrap();
+    let ledger = LedgerReader::open_existing(root, A).unwrap();
+    blocker.execute_batch("BEGIN EXCLUSIVE").unwrap();
+    let lock = ReservationLock::acquire(root).unwrap();
+    let error = match RegistryWriter::open_existing(root, A, &lock) {
+        Ok(_) => panic!("writer upgrade must be blocked by the exclusive transaction"),
+        Err(error) => error,
+    };
+    assert!(error.is_busy(), "{error}");
+    (ledger, blocker, lock)
+}
+
+#[test]
+fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() {
+    let _guard = serial();
+    let root = tempdir().unwrap();
+    let path = db(root.path(), A);
+    let (ledger, blocker, lock) = blocked_read_only_upgrade(root.path());
+    let error = ledger.accounting().unwrap_err();
+    assert!(
+        error.is_busy(),
+        "existing reader lost the busy cause: {error}"
+    );
+    let detail = error.to_string();
+    assert!(
+        detail.starts_with("reservation_ledger_busy: reopen shared reader"),
+        "{detail}"
+    );
+    assert!(detail.contains("members.sqlite"), "{detail}");
+    assert!(detail.contains("read format version"), "{detail}");
+    assert!(detail.contains("database is locked"), "{detail}");
+    blocker.execute_batch("ROLLBACK").unwrap();
+    drop(blocker);
+    assert_eq!(ledger.accounting().unwrap().reserved_bytes, 0);
+    assert_eq!(
+        ledger
+            .read(|c| Ok(c.pragma_query_value(None, "query_only", |r| r.get::<_, i64>(0))?))
+            .unwrap(),
+        1
+    );
+    assert_eq!(crate::db::file_identity::open_connections(&path), 1);
+    let writer = RegistryWriter::open_existing(root.path(), A, &lock).unwrap();
+    assert!(Arc::ptr_eq(&ledger.database, &writer.database));
+    assert_eq!(crate::db::file_identity::open_connections(&path), 1);
+}
+
+#[test]
+fn failed_upgrade_reader_rejects_a_replaced_database() {
+    let _guard = serial();
+    let root = tempdir().unwrap();
+    let path = db(root.path(), A);
+    let (ledger, blocker, _lock) = blocked_read_only_upgrade(root.path());
+    blocker.execute_batch("ROLLBACK").unwrap();
+    drop(blocker);
+    let original = path.with_extension("original.sqlite");
+    fs::rename(&path, &original).unwrap();
+    // Rebuild the file through SQLite itself: the source fence forbids copying
+    // file-set bytes, and VACUUM INTO gives the same content a new identity.
+    Connection::open(&original)
+        .unwrap()
+        .execute("VACUUM INTO ?1", [path.to_str().unwrap()])
+        .unwrap();
+    assert!(matches!(
+        ledger.accounting(),
+        Err(Error::Refused("live_sqlite_file_replaced"))
+    ));
+    assert_eq!(crate::db::file_identity::open_connections(&path), 0);
+}
+
 #[test]
 fn read_only_first_owner_upgrades_without_a_second_live_connection() {
     let _g = serial();
@@ -1491,7 +1744,14 @@ fn incomplete_marker_with_application_data_is_not_reset() {
     let root = tempdir().unwrap();
     legacy_bootstrap(root.path(), Kind::Registry, 2, true);
     let error = publish(root.path(), B, 1, None, |_| Ok(())).unwrap_err();
-    assert_eq!(error.to_string(), "reservation_ledger_unreadable");
+    let detail = error.to_string();
+    assert!(!error.is_busy(), "{detail}");
+    assert!(
+        detail.starts_with("reservation_ledger_unreadable: recover registry"),
+        "{detail}"
+    );
+    assert!(detail.contains("members.sqlite"), "{detail}");
+    assert!(detail.contains("schema_drift"), "{detail}");
     let path = db(root.path(), A);
     assert_eq!(crate::db::file_identity::open_connections(&path), 0);
     let c = IdentityConnection::open_with_flags(
diff --git a/crates/aft/src/commands/configure.rs b/crates/aft/src/commands/configure.rs
index adb66a7dcc..c6fe7c3e2e 100644
--- a/crates/aft/src/commands/configure.rs
+++ b/crates/aft/src/commands/configure.rs
@@ -8092,6 +8092,8 @@ struct ConfigureMaintenanceContinuation {
     semantic_waits_for_callgraph_start: bool,
     view_load: Option<crossbeam_channel::Receiver<Result<PreparedConfigureView, String>>>,
     replay_retry_at: Option<Instant>,
+    lexical_bind_retry_at: Option<Instant>,
+    lexical_bind_retry_delay: Duration,
     callgraph_upgrade_retry_at: Option<Instant>,
     replay_error_logged: bool,
 }
@@ -8105,13 +8107,23 @@ impl ConfigureMaintenanceContinuation {
             semantic_waits_for_callgraph_start: false,
             view_load: None,
             replay_retry_at: None,
+            lexical_bind_retry_at: None,
+            lexical_bind_retry_delay: Duration::from_secs(1),
             callgraph_upgrade_retry_at: None,
             replay_error_logged: false,
         }
     }
 
     fn retry_at(&self) -> Option<Instant> {
-        self.callgraph_upgrade_retry_at.or(self.replay_retry_at)
+        self.callgraph_upgrade_retry_at
+            .or(self.lexical_bind_retry_at)
+            .or(self.replay_retry_at)
+    }
+
+    fn defer_lexical_bind(&mut self, now: Instant) {
+        self.lexical_bind_retry_at = Some(now + self.lexical_bind_retry_delay);
+        self.lexical_bind_retry_delay =
+            (self.lexical_bind_retry_delay * 2).min(Duration::from_secs(30));
     }
 }
 
@@ -8986,6 +8998,7 @@ fn run_configure_maintenance_unit_inner(
             continuation.stage = ConfigureMaintenanceStage::ViewLoad;
         }
         ConfigureMaintenanceStage::ViewLoad => {
+            let retrying = continuation.lexical_bind_retry_at.take().is_some();
             if crate::views::parent::held_by(ctx, &job.canonical_cache_root) {
                 // A parent folder owns no view; its session reads the children's.
                 ctx.clear_view_runtime();
@@ -9007,6 +9020,18 @@ fn run_configure_maintenance_unit_inner(
                     ctx.clear_view_runtime();
                 }
                 match ctx.lexical_view_snapshot("bind") {
+                    Some(Err(error)) if crate::views::trigram::bind_is_busy(&error) => {
+                        if !retrying {
+                            slog_warn!(
+                                "lexical checkout view busy; configure maintenance will retry: {}",
+                                error
+                            );
+                        }
+                        // A parent only reads child publications; it cannot
+                        // recover a dropped child bind by querying that child.
+                        continuation.defer_lexical_bind(Instant::now());
+                        return ConfigureMaintenanceUnitResult::Continue;
+                    }
                     Some(Err(error)) => {
                         slog_warn!("lexical checkout view unavailable: {}", error);
                     }
@@ -9016,6 +9041,7 @@ fn run_configure_maintenance_unit_inner(
                     }
                     _ => {}
                 }
+                continuation.lexical_bind_retry_delay = Duration::from_secs(1);
             } else {
                 ctx.clear_view_runtime();
             }
@@ -17024,6 +17050,173 @@ mod tests {
         lock
     }
 
+    #[test]
+    fn lexical_bind_busy_markers_do_not_retry_terminal_failures() {
+        for reason in [
+            "family registry I/O error: reservation_ledger_busy: read format: database is locked",
+            "[driver]: v3 store busy: reservation lock timeout",
+            "[driver]: v3 base build busy; retain direct-scan coverage and retry",
+        ] {
+            assert!(crate::views::trigram::bind_is_busy(reason), "{reason}");
+        }
+        for reason in [
+            "reservation_ledger_unreadable: permission denied",
+            "live_sqlite_file_replaced",
+            "a file named busy.rs is unreadable",
+        ] {
+            assert!(!crate::views::trigram::bind_is_busy(reason), "{reason}");
+        }
+    }
+
+    #[test]
+    fn busy_lexical_bind_retries_until_parent_can_read_child_trigram() {
+        let _git_env = crate::test_env::hermetic_git_env_guard();
+        let temp = tempfile::tempdir().unwrap();
+        let base = std::fs::canonicalize(temp.path()).unwrap();
+        let projects = base.join("projects");
+        let root = projects.join("alpha");
+        let storage = base.join("storage");
+        init_git_fixture(&root);
+        let ctx = test_context();
+        let response = super::handle_configure(
+            &configure_request_with_params(json!({
+                "project_root": root,
+                "harness": "opencode",
+                "storage_dir": storage,
+                "config": [user_tier(json!({
+                    "indexes": { "trigram": true, "semantic": false, "callgraph": false },
+                    "views": { "enabled": true },
+                }))],
+            })),
+            &ctx,
+        );
+        assert!(response.success, "{response:?}");
+        assert!(ctx.artifact_identity_provisional_reason().is_none());
+        assert!(ctx.views_route_root());
+        let family = ctx.cached_artifact_cache_key(&root).unwrap();
+        let mut state = super::ConfigureMaintenanceState::default();
+        state.absorb_enqueued(&ctx);
+        while state.jobs.front().unwrap().stage != super::ConfigureMaintenanceStage::ViewLoad {
+            assert!(super::drain_deferred_configure_maintenance_unit(
+                &ctx, &mut state
+            ));
+        }
+
+        let parent = test_context();
+        parent.update_config(|config| {
+            config.project_root = Some(projects.clone());
+            config.views.enabled = true;
+            config.indexes.semantic = false;
+            config.indexes.callgraph = false;
+        });
+        parent.set_canonical_cache_root(projects.clone());
+        let planes = crate::views::parent::RequestedPlanes {
+            trigram: true,
+            semantic: false,
+            callgraph: false,
+        };
+        assert!(crate::views::parent::prepare(&projects, planes));
+        assert!(crate::views::parent::activate(
+            &parent,
+            &projects,
+            &storage,
+            &parent.config().semantic
+        ));
+        let session = crate::views::parent::session_for_root(&projects).unwrap();
+        assert!(session.wait_rounds(1, Duration::from_secs(30)));
+        assert!(!session.children()[0].trigram_ready());
+
+        {
+            let reservation = crate::blob_store::v3::ReservationLock::acquire(&storage).unwrap();
+            crate::blob_store::v3::open_base_refs_registry(&storage, &family, &reservation, None)
+                .unwrap();
+        }
+        let ledger = crate::blob_store::v3::family_dir(&storage, &family)
+            .unwrap()
+            .join("members.sqlite");
+        let blocker = rusqlite::Connection::open(ledger).unwrap();
+        blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap();
+        // Only the configure tail may publish the child; the parent never binds it.
+        while super::drain_deferred_configure_maintenance_unit(&ctx, &mut state) {}
+        assert!(!session.children()[0].trigram_ready());
+        assert_eq!(
+            state.jobs.len(),
+            1,
+            "busy lexical bind was dropped instead of parked"
+        );
+        assert_eq!(
+            state.jobs[0].stage,
+            super::ConfigureMaintenanceStage::ViewLoad
+        );
+        let retry_at = state.jobs[0]
+            .retry_at()
+            .expect("busy bind needs a retry deadline");
+        assert_eq!(
+            state.jobs[0].lexical_bind_retry_delay,
+            Duration::from_secs(2)
+        );
+        state.replay_retry_cutoff = Some(retry_at - Duration::from_nanos(1));
+        assert!(!state.has_ready_job());
+        // Check every backoff step without spending SQLite's busy timeout on
+        // each attempt. The real locked bind above arms the same continuation.
+        for seconds in [2, 4, 8, 16, 30, 30] {
+            let now = Instant::now();
+            state.jobs[0].defer_lexical_bind(now);
+            let retry_at = state.jobs[0].retry_at().unwrap();
+            assert_eq!(retry_at, now + Duration::from_secs(seconds));
+            state.replay_retry_cutoff = Some(retry_at - Duration::from_nanos(1));
+            assert!(!state.has_ready_job());
+            assert_eq!(
+                state.jobs[0].stage,
+                super::ConfigureMaintenanceStage::ViewLoad
+            );
+        }
+        assert!(session.wait_rounds(session.rounds() + 1, Duration::from_secs(30)));
+        let query: RawRequest = serde_json::from_value(json!({
+            "id": "parent-busy-lexical",
+            "command": "semantic_search",
+            "query": "tracked",
+        }))
+        .unwrap();
+        let pending = crate::views::parent::route(&query, &parent).unwrap();
+        assert_eq!(pending.data["complete"], false, "{pending:?}");
+        let text = pending.data["text"].as_str().unwrap();
+        assert!(
+            text.contains("alpha") && text.contains("busy"),
+            "{pending:?}"
+        );
+        blocker.execute_batch("ROLLBACK").unwrap();
+        drop(blocker);
+        assert!(session.wait_rounds(session.rounds() + 1, Duration::from_secs(30)));
+        let pending = crate::views::parent::route(&query, &parent).unwrap();
+        assert_eq!(pending.data["complete"], false, "{pending:?}");
+        let text = pending.data["text"].as_str().unwrap();
+        assert!(
+            text.contains("alpha") && text.contains("busy"),
+            "{pending:?}"
+        );
+        state.replay_retry_cutoff = state.jobs[0].retry_at();
+        assert!(state.has_ready_job());
+        assert!(super::drain_deferred_configure_maintenance_unit(
+            &ctx, &mut state
+        ));
+        assert!(state.jobs[0].lexical_bind_retry_at.is_none());
+        assert_eq!(
+            state.jobs[0].lexical_bind_retry_delay,
+            Duration::from_secs(1)
+        );
+        while super::drain_deferred_configure_maintenance_unit(&ctx, &mut state) {}
+        assert!(state.jobs.is_empty());
+        let deadline = Instant::now() + Duration::from_secs(30);
+        while !session.children()[0].trigram_ready() {
+            assert!(
+                Instant::now() < deadline,
+                "parent never saw the retried lexical publication"
+            );
+            std::thread::sleep(Duration::from_millis(20));
+        }
+    }
+
     fn trigram_bind_request(root: &std::path::Path, storage: &std::path::Path) -> RawRequest {
         configure_request_with_params(json!({
             "project_root": root,
diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs
index 54d3545fdc..fd978f7453 100644
--- a/crates/aft/src/context.rs
+++ b/crates/aft/src/context.rs
@@ -4587,6 +4587,7 @@ pub struct AppContext {
     lexical_view: parking_lot::Mutex<Option<Arc<crate::views::trigram::LexicalView>>>,
     lexical_routes: [AtomicU64; 3],
     callgraph_view: parking_lot::Mutex<Option<Arc<crate::views::callgraph::CallgraphView>>>,
+    callgraph_view_read_diagnostics: parking_lot::Mutex<CallgraphViewReadDiagnostics>,
     callgraph_view_routes: AtomicU64,
     /// Files whose symbols a views-routed root's prewarm reused from, or
     /// stored into, the family's v3 symbol plane.
@@ -4951,6 +4952,47 @@ pub enum CallgraphStoreAccess {
     Error(CallGraphStoreError),
 }
 
+pub(crate) fn checkout_view_callgraph_access(
+    reader: Option<
+        Result<Arc<ReadonlyCallGraphStore>, crate::views::callgraph::CallgraphViewError>,
+    >,
+) -> CallgraphStoreAccess {
+    match reader {
+        Some(Ok(store)) => CallgraphStoreAccess::Ready(store),
+        // Transient contention while opening a view (a locked derived
+        // database, or another driver finishing the shared base) says nothing
+        // about the graph. Report SQLite's busy error, which every caller
+        // already presents as busy/pending with a retry.
+        Some(Err(error)) if error.busy => CallgraphStoreAccess::Error(CallGraphStoreError::Sqlite(
+            rusqlite::Error::SqliteFailure(
+                rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
+                Some(error.reason),
+            ),
+        )),
+        Some(Err(error))
+            if error
+                .reason
+                .contains(crate::views::read::CALLGRAPH_DISABLED) =>
+        {
+            CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(error.reason))
+        }
+        Some(Err(error)) => CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(format!(
+            "callgraph view unavailable: {error}"
+        ))),
+        None => CallgraphStoreAccess::Unavailable,
+    }
+}
+
+/// Acquisition diagnostics for the checkout callgraph view, including the full
+/// cause of retryable SQLite contention and time spent acquiring readers.
+#[derive(Debug, Clone, Default, Serialize)]
+pub struct CallgraphViewReadDiagnostics {
+    pub attempts: u64,
+    pub last_attempt_ms: u64,
+    pub total_attempt_ms: u64,
+    pub last_error: Option<String>,
+}
+
 #[derive(Clone, Copy)]
 enum CallgraphBackgroundWork {
     Ensure,
@@ -5451,6 +5493,9 @@ impl AppContext {
             lexical_view: parking_lot::Mutex::new(None),
             lexical_routes: Default::default(),
             callgraph_view: parking_lot::Mutex::new(None),
+            callgraph_view_read_diagnostics: parking_lot::Mutex::new(
+                CallgraphViewReadDiagnostics::default(),
+            ),
             callgraph_view_routes: AtomicU64::new(0),
             symbol_view_routes: Arc::default(),
             checkout_semantic: Arc::default(),
@@ -11041,7 +11086,13 @@ impl AppContext {
                 crate::views::contracts::WaitOutcome::Installed(snapshot)
                 | crate::views::contracts::WaitOutcome::TimedOut { snapshot, .. } => snapshot,
             };
-            return match runtime.callgraph.reader(&runtime.access, snapshot) {
+            let started = Instant::now();
+            let reader = runtime.callgraph.reader(&runtime.access, snapshot);
+            self.note_callgraph_view_read(
+                started.elapsed(),
+                reader.as_ref().err().map(ToString::to_string),
+            );
+            return match reader {
                 Ok(reader) => CallgraphStoreAccess::Ready(Arc::clone(&reader.store)),
                 Err(error) if error.reason == crate::views::read::CALLGRAPH_DISABLED => {
                     CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(error.reason))
@@ -11055,29 +11106,16 @@ impl AppContext {
             if !demand {
                 return CallgraphStoreAccess::Building;
             }
-            return match self.callgraph_view_reader() {
-                Some(Ok(store)) => CallgraphStoreAccess::Ready(store),
-                // A lock held on the view's database says nothing about the
-                // graph: report SQLite's busy error, which callers present as
-                // building, retry shortly.
-                Some(Err(error)) if error.busy => CallgraphStoreAccess::Error(
-                    CallGraphStoreError::Sqlite(rusqlite::Error::SqliteFailure(
-                        rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
-                        Some(error.reason),
-                    )),
-                ),
-                Some(Err(error))
-                    if error
-                        .reason
-                        .contains(crate::views::read::CALLGRAPH_DISABLED) =>
-                {
-                    CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(error.reason))
-                }
-                Some(Err(error)) => CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(
-                    format!("callgraph view unavailable: {error}"),
-                )),
-                None => CallgraphStoreAccess::Unavailable,
-            };
+            let started = Instant::now();
+            let reader = self.callgraph_view_reader();
+            self.note_callgraph_view_read(
+                started.elapsed(),
+                reader
+                    .as_ref()
+                    .and_then(|reader| reader.as_ref().err())
+                    .map(ToString::to_string),
+            );
+            return checkout_view_callgraph_access(reader);
         }
         if self.config().views.enabled && self.config().indexes.callgraph {
             let on_demand = self.callgraph_build_on_demand();
@@ -12269,6 +12307,23 @@ impl AppContext {
         ));
     }
 
+    /// Passive diagnostic state, including the last acquisition error. A
+    /// successful retry does not erase that error.
+    pub fn callgraph_view_read_diagnostics(&self) -> CallgraphViewReadDiagnostics {
+        self.callgraph_view_read_diagnostics.lock().clone()
+    }
+
+    fn note_callgraph_view_read(&self, elapsed: Duration, error: Option<String>) {
+        let mut diagnostics = self.callgraph_view_read_diagnostics.lock();
+        let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
+        diagnostics.attempts = diagnostics.attempts.saturating_add(1);
+        diagnostics.last_attempt_ms = elapsed_ms;
+        diagnostics.total_attempt_ms = diagnostics.total_attempt_ms.saturating_add(elapsed_ms);
+        if error.is_some() {
+            diagnostics.last_error = error;
+        }
+    }
+
     /// The call-graph reader of this checkout's view, acquired once per
     /// pinned request. `None` means views do not route call-graph reads; an
     /// acquisition error is final and never answered from another store.
@@ -18608,6 +18663,70 @@ mod callgraph_store_for_ops_tests {
         store.sqlite_path().to_path_buf()
     }
 
+    #[test]
+    fn callgraph_view_diagnostics_preserve_busy_cause_and_retry_on_the_same_context() {
+        use crate::blob_store::v3;
+        use crate::db::file_identity::IdentityConnection;
+
+        let _serial = v3::serial_for_test();
+        let (_dirs, ctx) = cold_build_context();
+        let mut config = (*ctx.config()).clone();
+        config.views.enabled = true;
+        ctx.set_config(config);
+        let root = ctx.callgraph_project_root().unwrap();
+        ctx.set_canonical_cache_root(root.clone());
+        assert!(matches!(
+            ctx.callgraph_store_for_ops(),
+            CallgraphStoreAccess::Ready(_)
+        ));
+        ctx.callgraph_view.lock().take();
+
+        let storage = ctx.storage_paths().artifacts;
+        let family = ctx.memoized_artifact_cache_key(&root);
+        let path = v3::family_dir(&storage, &family)
+            .unwrap()
+            .join("members.sqlite");
+        let blocker = IdentityConnection::open_with_flags(
+            &path,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE,
+            "callgraph-view-diagnostics-blocker",
+        )
+        .unwrap();
+        blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap();
+        let CallgraphStoreAccess::Error(error) = ctx.callgraph_store_for_ops() else {
+            panic!("a locked view must retain the SQLite busy verdict");
+        };
+        assert!(
+            matches!(
+                &error,
+                CallGraphStoreError::Sqlite(rusqlite::Error::SqliteFailure(code, _))
+                    if code.extended_code == rusqlite::ffi::SQLITE_BUSY
+            ),
+            "{error}"
+        );
+        assert!(error.is_transient_lock_contention(), "{error}");
+        let busy = ctx.callgraph_view_read_diagnostics();
+        assert_eq!(busy.attempts, 2);
+        let detail = busy
+            .last_error
+            .as_deref()
+            .expect("SQLite busy must retain its acquisition cause");
+        assert!(detail.contains("reservation_ledger_busy"), "{detail}");
+        assert!(detail.contains("read initialization marker"), "{detail}");
+        assert!(detail.contains("members.sqlite"), "{detail}");
+        assert!(detail.contains("database is locked"), "{detail}");
+        blocker.execute_batch("ROLLBACK").unwrap();
+        drop(blocker);
+        assert!(matches!(
+            ctx.callgraph_store_for_ops(),
+            CallgraphStoreAccess::Ready(_)
+        ));
+        let ready = ctx.callgraph_view_read_diagnostics();
+        assert_eq!(ready.attempts, 3);
+        assert_eq!(ready.last_error, busy.last_error);
+        assert!(ready.total_attempt_ms >= ready.last_attempt_ms);
+    }
+
     #[test]
     fn callgraph_busy_readiness_query_reports_building() {
         for read_only in [true, false] {
diff --git a/crates/aft/src/inspect/view_projection_tests.rs b/crates/aft/src/inspect/view_projection_tests.rs
index 14c9881109..8db2f90597 100644
--- a/crates/aft/src/inspect/view_projection_tests.rs
+++ b/crates/aft/src/inspect/view_projection_tests.rs
@@ -770,6 +770,7 @@ fn views_tier2_pending_plane_never_falls_back_to_legacy() {
 #[test]
 fn views_tier2_and_legacy_dead_code_verdicts_match() {
     let (_project, _storage, mut job, request) = view_projection_fixture();
+    let view = crate::views::ViewStore::open(&request.storage, &request.scope).unwrap();
     for source in [
         "export function used() {}\nexport function dead() {}\n",
         "export function used() { dead(); }\nexport function dead() {}\n",
@@ -779,6 +780,9 @@ fn views_tier2_and_legacy_dead_code_verdicts_match() {
     ] {
         write_projection_cache_file(&job.project_root.join("target.ts"), source);
         crate::views::assembly::publish_checkout(&request).unwrap();
+        // Publication makes the immutable generation current before its
+        // detached checkpoint closes the last writer; wait for readability.
+        crate::views::wait_for_derived_checkpoint_for_test(view.view_dir());
         Arc::make_mut(&mut job.config).views.enabled = true;
         let view =
             build_tier2_callgraph_snapshot_with_refresh(&job, true, &job.scope_files).unwrap();
diff --git a/crates/aft/src/migration/v3_import.rs b/crates/aft/src/migration/v3_import.rs
index 43a06d8313..98a6024f41 100644
--- a/crates/aft/src/migration/v3_import.rs
+++ b/crates/aft/src/migration/v3_import.rs
@@ -1360,6 +1360,7 @@ fn storage_gap(error: &v3::Error) -> Option<String> {
         v3::Error::Refused(reason) if STORAGE_GAP_REFUSALS.contains(reason) => {
             Some(format!("storage_gap:{reason}"))
         }
+        v3::Error::Ledger { .. } if !error.is_busy() => Some(format!("storage_gap:{error}")),
         _ => None,
     }
 }
diff --git a/crates/aft/src/migration/v3_import_tests.rs b/crates/aft/src/migration/v3_import_tests.rs
index 3f96c8219a..f982408481 100644
--- a/crates/aft/src/migration/v3_import_tests.rs
+++ b/crates/aft/src/migration/v3_import_tests.rs
@@ -462,3 +462,22 @@ fn a_refused_reservation_is_a_named_storage_gap() {
         .unwrap()
         .exists());
 }
+
+#[test]
+fn ledger_storage_gap_keeps_diagnostics_but_does_not_swallow_busy() {
+    let error = |code| v3::Error::Ledger {
+        path: PathBuf::from("family/members.sqlite"),
+        operation: "recover registry",
+        source: Box::new(v3::Error::Sqlite(rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error::new(code),
+            Some("underlying SQLite cause".into()),
+        ))),
+    };
+    let unreadable = error(rusqlite::ffi::SQLITE_NOTADB);
+    assert_eq!(
+        storage_gap(&unreadable),
+        Some(format!("storage_gap:{unreadable}"))
+    );
+    assert!(unreadable.to_string().contains("underlying SQLite cause"));
+    assert_eq!(storage_gap(&error(rusqlite::ffi::SQLITE_BUSY)), None);
+}
diff --git a/crates/aft/src/views/base_refs.rs b/crates/aft/src/views/base_refs.rs
index db9ff91e66..8e0c3b8fb2 100644
--- a/crates/aft/src/views/base_refs.rs
+++ b/crates/aft/src/views/base_refs.rs
@@ -81,6 +81,7 @@ pub enum BaseRefError {
 impl fmt::Display for BaseRefError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
+            Self::Store(error) if error.is_busy() => write!(f, "{}{error}", v3::BUSY_PREFIX),
             Self::Store(error) => write!(f, "v3 store: {error}"),
             Self::View(error) => write!(f, "{error}"),
             Self::Io(error) => write!(f, "{error}"),
diff --git a/crates/aft/src/views/callgraph.rs b/crates/aft/src/views/callgraph.rs
index 9e890b83d3..8c5a1188cb 100644
--- a/crates/aft/src/views/callgraph.rs
+++ b/crates/aft/src/views/callgraph.rs
@@ -104,6 +104,15 @@ impl ManifestBlobReader for BlobReader {
 
 /// Reads callgraph blobs from the family's v3 callgraph plane store.
 pub struct V3BlobReader(pub crate::blob_store::v3::PlaneReader);
+
+pub(super) fn v3_error(error: crate::blob_store::v3::Error) -> PlaneError {
+    if error.is_busy() {
+        self::error(format!("{V3_STORE_BUSY}: {error}"))
+    } else {
+        self::error(error)
+    }
+}
+
 impl ManifestBlobReader for V3BlobReader {
     fn read_callgraph_blob(&self, key: &str) -> Result<Option<Vec<u8>>, ManifestJoinError> {
         let bytes = parse_hex32(key)
@@ -358,9 +367,9 @@ impl PlaneAdapter for CallgraphPlane {
         }
     }
 }
-/// Why a checkout view could not supply a reader. `busy` marks a transient
-/// SQLite lock on the view's database: the graph exists and the caller should
-/// retry, which is reported as building rather than unavailable.
+/// Why a checkout view could not supply a reader. `busy` marks transient
+/// contention on a view database or its family ledger, so callers can retry
+/// rather than reporting a permanently unavailable graph.
 #[derive(Clone, Debug)]
 pub(crate) struct CallgraphViewError {
     pub reason: String,
@@ -376,6 +385,27 @@ impl From<String> for CallgraphViewError {
     }
 }
 
+impl From<PlaneError> for CallgraphViewError {
+    fn from(error: PlaneError) -> Self {
+        // Base-reference errors enter through the loader's driver stage,
+        // whereas plane materializers return their reason without a stage.
+        let reason = error
+            .reason
+            .strip_prefix("[driver]: ")
+            .unwrap_or(&error.reason);
+        let reason = reason
+            .strip_prefix("family registry I/O error: ")
+            .unwrap_or(reason);
+        Self {
+            busy: reason.starts_with(DERIVED_STORE_BUSY)
+                || reason.starts_with(V3_STORE_BUSY)
+                || reason.starts_with(crate::blob_store::v3::BUSY_PREFIX)
+                || reason.starts_with(V3_BASE_COMPLETION_BUSY),
+            reason: error.to_string(),
+        }
+    }
+}
+
 impl From<&str> for CallgraphViewError {
     fn from(reason: &str) -> Self {
         reason.to_owned().into()
@@ -391,6 +421,8 @@ impl std::fmt::Display for CallgraphViewError {
 /// Prefix of the plane error for a derived database that a SQLite lock kept
 /// from opening.
 pub(crate) const DERIVED_STORE_BUSY: &str = "callgraph derived store is busy";
+const V3_STORE_BUSY: &str = "callgraph v3 store is busy";
+const V3_BASE_COMPLETION_BUSY: &str = "v3 base plane completion is owned elsewhere";
 
 /// Supplies a checkout's verified call graph to one request, without
 /// exposing its storage layout. Implementations never publish v1/v2 data and
@@ -457,7 +489,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource {
         let (mut snapshot, revision) = self
             .driver
             .verified_bind(&self.access, seed)
-            .map_err(|error| error.to_string())?;
+            .map_err(CallgraphViewError::from)?;
         if has_unpublished_changes(&snapshot) {
             // Publish the edits as a new worker generation (a donor clone
             // patched with the diff) before the single install, as a fresh
@@ -471,7 +503,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource {
                     &snapshot,
                     super::first_load::DERIVED_CALLGRAPH_SEEDING,
                 )
-                .map_err(|error| error.to_string())?;
+                .map_err(CallgraphViewError::from)?;
             let cut = snapshot
                 .install_cut()
                 .cloned()
@@ -492,7 +524,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource {
         }
         self.driver
             .install(&self.access, &snapshot, revision)
-            .map_err(|error| error.to_string())?;
+            .map_err(CallgraphViewError::from)?;
         // Read the installed cut directly: `installed_snapshot` would mark a
         // checkout without a steady watcher subscription overflowed at once.
         // That mark only decides how the next bind reconciles; this cut was
@@ -528,10 +560,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource {
                 self.bridge
                     .adapter
                     .open_generation(&self.access, snapshot.generation())
-                    .map_err(|error| CallgraphViewError {
-                        busy: error.reason.starts_with(DERIVED_STORE_BUSY),
-                        reason: error.to_string(),
-                    })?;
+                    .map_err(CallgraphViewError::from)?;
                 self.bridge
                     .adapter
                     .reader(&self.access, &snapshot)
@@ -596,7 +625,7 @@ impl CallgraphView {
         storage: crate::root_cache::StoragePaths,
         family: String,
         policy: crate::blob_store::v2::TrigramPolicy,
-    ) -> Result<Self, String> {
+    ) -> Result<Self, CallgraphViewError> {
         let scope = view_scope(&root);
         let producers = super::manifest_v2::Producers {
             trigram: policy.fingerprint_hex(),
@@ -619,7 +648,7 @@ impl CallgraphView {
                 Arc::new(super::first_load::ConfiguredMembershipWalker),
                 vec![trigram, bridge.clone()],
             )
-            .map_err(|error| error.to_string())?
+            .map_err(CallgraphViewError::from)?
             .with_adapters(vec![adapter]),
         );
         driver.register_write_intent();
@@ -761,6 +790,221 @@ mod tests {
         super::super::donor::events_under(dir)
     }
 
+    #[test]
+    fn v3_checkpoint_contention_reaches_checkout_reader_as_busy() {
+        use crate::blob_store::v3;
+        use crate::db::file_identity::IdentityConnection;
+        let _serial = v3::serial_for_test();
+        let temp = tempfile::tempdir().unwrap();
+        let storage = temp.path().join("storage");
+        let root = checkout(&temp.path().join("checkout"));
+        let graph = view(&root, &storage);
+        assert_eq!(callers(&graph.reader().unwrap(), "a.ts", "target").len(), 1);
+        let plane = v3::plane_path(&storage, "donor-reuse", FamilyPlane::Callgraph).unwrap();
+        let blocker = IdentityConnection::open_with_flags(
+            &plane,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
+            "callgraph-v3-checkpoint-blocker",
+        )
+        .unwrap();
+        // Pin a real WAL snapshot before the next publication appends frames.
+        blocker
+            .execute_batch("BEGIN; SELECT * FROM sqlite_schema;")
+            .unwrap();
+        let added = root.join("c.ts");
+        std::fs::write(
+            &added,
+            "import { caller } from './b';\nexport function top() { return caller(); }\n",
+        )
+        .unwrap();
+        graph.record_change(&added);
+        let error = match graph.reader() {
+            Ok(_) => panic!("a pinned WAL reader should prevent the truncating checkpoint"),
+            Err(error) => error,
+        };
+        assert!(error.busy, "{error}");
+        assert!(error.reason.contains(V3_STORE_BUSY), "{error}");
+        assert!(error.reason.contains("checkpoint WAL"), "{error}");
+        assert!(error.reason.contains("callgraph.sqlite"), "{error}");
+        assert!(
+            error.reason.contains("wal_checkpoint(TRUNCATE): busy=1"),
+            "{error}"
+        );
+        blocker.execute_batch("ROLLBACK").unwrap();
+        drop(blocker);
+        assert_eq!(callers(&graph.reader().unwrap(), "b.ts", "caller").len(), 1);
+    }
+
+    #[test]
+    fn v3_busy_ledger_survives_driver_error_wrapping() {
+        use crate::blob_store::v3;
+        use crate::db::file_identity::IdentityConnection;
+        let _serial = v3::serial_for_test();
+        let temp = tempfile::tempdir().unwrap();
+        let family = "driver-busy";
+        v3::publish(temp.path(), family, 1024, None, |_| Ok(())).unwrap();
+        let path = v3::family_dir(temp.path(), family)
+            .unwrap()
+            .join("members.sqlite");
+        let lock = v3::ReservationLock::acquire(temp.path()).unwrap();
+        let mut writer = v3::RegistryWriter::open_for_barrier(temp.path(), family, &lock).unwrap();
+        let blocker = IdentityConnection::open_with_flags(
+            &path,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
+            "callgraph-v3-ledger-blocker",
+        )
+        .unwrap();
+        blocker
+            .execute_batch("BEGIN; SELECT * FROM accounting;")
+            .unwrap();
+        writer
+            .with_barrier(|tx| {
+                tx.execute("UPDATE accounting SET watermark=watermark+1", [])?;
+                Ok(())
+            })
+            .unwrap();
+        let ledger_error = writer.covering_measurement().unwrap_err();
+        let driver_error = |error| {
+            PlaneError::loader(
+                super::super::contracts::LoaderStage::Driver,
+                super::super::base_refs::BaseRefError::from(error).to_string(),
+            )
+        };
+        let error = CallgraphViewError::from(driver_error(ledger_error));
+        assert!(error.busy, "{error}");
+        assert!(
+            error
+                .reason
+                .starts_with("view loader [driver]: v3 store busy: "),
+            "{error}"
+        );
+        assert!(error.reason.contains("members.sqlite"), "{error}");
+        assert!(
+            error.reason.contains("wal_checkpoint(TRUNCATE): busy=1"),
+            "{error}"
+        );
+        let permanent = CallgraphViewError::from(driver_error(v3::Error::Refused(
+            "reservation_ledger_unreadable",
+        )));
+        assert!(!permanent.busy, "{permanent}");
+        blocker.execute_batch("ROLLBACK").unwrap();
+        writer.covering_measurement().unwrap();
+    }
+
+    #[test]
+    fn v3_busy_ledger_during_view_construction_is_retryable() {
+        use crate::blob_store::v3;
+        use crate::db::file_identity::IdentityConnection;
+        let _serial = v3::serial_for_test();
+        let temp = tempfile::tempdir().unwrap();
+        let storage = temp.path().join("storage");
+        let root = checkout(&temp.path().join("checkout"));
+        v3::publish(&storage, "donor-reuse", 1024, None, |_| Ok(())).unwrap();
+        let path = v3::family_dir(&storage, "donor-reuse")
+            .unwrap()
+            .join("members.sqlite");
+        let blocker = IdentityConnection::open_with_flags(
+            &path,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE,
+            "callgraph-v3-open-blocker",
+        )
+        .unwrap();
+        // SQLite's exclusive locking mode prevents a second connection from
+        // opening the ledger until this transaction and connection are gone.
+        blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap();
+        let construct = || {
+            CallgraphView::new(
+                root.clone(),
+                crate::root_cache::StoragePaths::same(&storage),
+                "donor-reuse".into(),
+                crate::blob_store::v2::TrigramPolicy {
+                    max_file_size: 1 << 20,
+                },
+            )
+        };
+        let error = match construct() {
+            Ok(_) => panic!("view construction must not open an exclusively held ledger"),
+            Err(error) => error,
+        };
+        assert!(error.busy, "{error}");
+        assert!(error.reason.contains("v3 store busy: "), "{error}");
+        blocker.execute_batch("ROLLBACK").unwrap();
+        drop(blocker);
+        assert_eq!(
+            callers(&construct().unwrap().reader().unwrap(), "a.ts", "target").len(),
+            1
+        );
+    }
+
+    #[test]
+    fn v3_base_completion_owned_by_another_driver_answers_building_then_ready() {
+        use super::super::base_builder::{build_base, BaseBuild, BaseBuildInput};
+        use super::super::eviction::DiskBudget;
+
+        let _serial = crate::blob_store::v3::serial_for_test();
+        let temp = tempfile::tempdir().unwrap();
+        let storage = temp.path().join("storage");
+        let root = checkout(&temp.path().join("checkout"));
+        let paths: Vec<_> = FILES
+            .iter()
+            .map(|(name, _)| RelPath::new(name.as_bytes().to_vec()).unwrap())
+            .collect();
+        let BaseBuild::Published { claim, .. } = build_base(
+            BaseBuildInput {
+                storage: &storage,
+                family: "donor-reuse",
+                root: &root,
+                paths: &paths,
+                policy: crate::blob_store::v2::TrigramPolicy {
+                    max_file_size: 1 << 20,
+                },
+                semantic_producer: None,
+                callgraph_producer: PRODUCER.into(),
+                ignore_fingerprint: None,
+                history: None,
+                budget: DiskBudget::default(),
+            },
+            None,
+        )
+        .unwrap() else {
+            panic!("the fixture must own its fresh base")
+        };
+        let outcome =
+            super::super::base_builder::claim_base(&storage, "donor-reuse", claim.scope(), None)
+                .unwrap();
+        let owner_pid = match &outcome {
+            super::super::base_builder::ClaimOutcome::Busy { owner_pid, .. } => *owner_pid,
+            _ => panic!("the live base owner must fence another completion driver"),
+        };
+        assert_eq!(owner_pid, std::process::id());
+        let error = CallgraphViewError::from(PlaneError::loader(
+            super::super::contracts::LoaderStage::Driver,
+            format!("{V3_BASE_COMPLETION_BUSY}: {outcome:?}"),
+        ));
+        assert!(error.busy, "{error}");
+        assert!(error.reason.contains(V3_BASE_COMPLETION_BUSY), "{error}");
+        assert!(
+            error
+                .reason
+                .contains(&format!("owner_pid: {}", std::process::id())),
+            "{error}"
+        );
+        assert!(matches!(
+            crate::context::checkout_view_callgraph_access(Some(Err(error))),
+            crate::context::CallgraphStoreAccess::Error(error)
+                if error.is_transient_lock_contention()
+        ));
+
+        claim.release().unwrap();
+        let graph = view(&root, &storage);
+        let crate::context::CallgraphStoreAccess::Ready(ready) =
+            crate::context::checkout_view_callgraph_access(Some(graph.reader()))
+        else {
+            panic!("the completed base must serve a ready call graph")
+        };
+        assert_eq!(callers(&ready, "a.ts", "target").len(), 1);
+    }
+
     #[test]
     fn sealed_base_donor_serves_checkouts_without_reextraction_or_cold_builds() {
         use super::super::donor::DonorEvent;
diff --git a/crates/aft/src/views/first_load.rs b/crates/aft/src/views/first_load.rs
index 8751060073..07840d7fb1 100644
--- a/crates/aft/src/views/first_load.rs
+++ b/crates/aft/src/views/first_load.rs
@@ -4719,7 +4719,7 @@ impl CompositePlane for CallgraphBridge {
                 }
                 Ok(())
             })
-            .map_err(|e| error(e.to_string()))?;
+            .map_err(super::callgraph::v3_error)?;
         }
         for (_, key) in &keys {
             if !v3_readers.contains(key) {
@@ -5018,7 +5018,7 @@ impl CompositePlane for CallgraphBridge {
                 }
                 Ok(())
             })
-            .map_err(|e| error(e.to_string()))?;
+            .map_err(super::callgraph::v3_error)?;
         }
         for (path, key, _) in &payloads {
             if let Some(super::manifest_v2::EntryV2::Regular {
@@ -5045,7 +5045,7 @@ impl CompositePlane for CallgraphBridge {
                 &route.family,
                 FamilyPlane::Callgraph,
             )
-            .map_err(|e| error(e.to_string()))?,
+            .map_err(super::callgraph::v3_error)?,
         );
         let dir = route.member_dir();
         // Open (and so initialize and recover) the member's view store before
diff --git a/crates/aft/src/views/io.rs b/crates/aft/src/views/io.rs
index badd1b5897..17eaf72dbb 100644
--- a/crates/aft/src/views/io.rs
+++ b/crates/aft/src/views/io.rs
@@ -4,8 +4,8 @@
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Mutex, OnceLock};
 
-/// Keep the error kind available to retry/absence checks while naming the I/O
-/// that failed. OS errors alone do not include the path they operated on.
+/// Keep the error kind and native source available to retry/absence checks
+/// while naming the I/O that failed. OS errors alone do not include its path.
 pub(crate) fn at(
     operation: &str,
     path: &std::path::Path,
@@ -13,10 +13,39 @@ pub(crate) fn at(
 ) -> std::io::Error {
     std::io::Error::new(
         source.kind(),
-        format!("{operation} {}: {source}", path.display()),
+        PathIoError {
+            operation: operation.to_owned(),
+            path: path.to_path_buf(),
+            source,
+        },
     )
 }
 
+#[derive(Debug)]
+struct PathIoError {
+    operation: String,
+    path: std::path::PathBuf,
+    source: std::io::Error,
+}
+
+impl std::fmt::Display for PathIoError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{} {}: {}",
+            self.operation,
+            self.path.display(),
+            self.source
+        )
+    }
+}
+
+impl std::error::Error for PathIoError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(&self.source)
+    }
+}
+
 /// Lifetime totals for measured work sites, shared by every root in this process.
 /// Relaxed loads are intentionally not a transactional snapshot; difference them
 /// only within one daemon lifetime, never sum overlapping per-root intervals.
@@ -341,6 +370,30 @@ mod tests {
         }
     }
 
+    #[test]
+    fn path_context_preserves_native_error_sources() {
+        let error = super::at(
+            "opening",
+            std::path::Path::new("busy-database"),
+            std::io::Error::from_raw_os_error(32),
+        );
+        let context = error
+            .get_ref()
+            .unwrap()
+            .downcast_ref::<super::PathIoError>()
+            .unwrap();
+        assert_eq!(context.source.raw_os_error(), Some(32));
+        assert_eq!(
+            std::error::Error::source(context)
+                .unwrap()
+                .downcast_ref::<std::io::Error>()
+                .unwrap()
+                .raw_os_error(),
+            Some(32)
+        );
+        assert!(error.to_string().starts_with("opening busy-database: "));
+    }
+
     use super::*;
     #[test]
     fn performance_work_counts_real_source_operations() {
diff --git a/crates/aft/src/views/parent/worker.rs b/crates/aft/src/views/parent/worker.rs
index 965ad620f0..bb0d2b0e8c 100644
--- a/crates/aft/src/views/parent/worker.rs
+++ b/crates/aft/src/views/parent/worker.rs
@@ -245,8 +245,13 @@ fn refresh_trigram(session: &ParentSession, child: &Child) {
         Some(&session.storage),
     );
     let Some(artifact) = artifact else {
-        *write(&child.trigram) =
-            Plane::Gap("no trigram index has been built for this repository yet".into());
+        let mut trigram = write(&child.trigram);
+        // Releasing the ledger lock does not itself publish a usable lexical index.
+        // Keep its last busy diagnostic until a usable publication arrives.
+        if matches!(&*trigram, Plane::Gap(reason) if super::super::trigram::bind_is_busy(reason)) {
+            return;
+        }
+        *trigram = Plane::Gap("no trigram index has been built for this repository yet".into());
         return;
     };
     if read(&child.trigram).ready().is_some_and(
diff --git a/crates/aft/src/views/trigram.rs b/crates/aft/src/views/trigram.rs
index c3c42bfd41..932afc395c 100644
--- a/crates/aft/src/views/trigram.rs
+++ b/crates/aft/src/views/trigram.rs
@@ -45,6 +45,15 @@ fn absolute_member_path(
     }
     Ok(root.join(relative))
 }
+pub(crate) fn bind_is_busy(reason: &str) -> bool {
+    // Lexical snapshot errors carry strings. Match only the explicit retry
+    // markers preserved by the v3 registry and base builder, including markers
+    // nested beneath a driver or registry diagnostic.
+    reason.contains(crate::blob_store::v3::BUSY_PREFIX)
+        || reason.contains("reservation_ledger_busy:")
+        || reason.contains("v3 base build busy; retain direct-scan coverage and retry")
+}
+
 /// Checkout-local lexical authority, independent of the semantic worker's
 /// lifetime and model availability. Queries serialize reconciliation so an
 /// installed membership and its payloads always come from the same snapshot.
diff --git a/crates/aft/tests/integration/branch_switch_test.rs b/crates/aft/tests/integration/branch_switch_test.rs
index 6a3afbe73b..645a86f86f 100644
--- a/crates/aft/tests/integration/branch_switch_test.rs
+++ b/crates/aft/tests/integration/branch_switch_test.rs
@@ -1,4 +1,4 @@
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener, TcpStream};
 use std::path::{Path, PathBuf};
@@ -355,28 +355,136 @@ fn configure_context_with_views(
     ctx
 }
 
+#[derive(Debug, Default)]
+struct IndexWaitDiagnostics {
+    status: BTreeMap<&'static str, String>,
+    view_status: BTreeMap<&'static str, BTreeMap<String, usize>>,
+    last_error: BTreeMap<&'static str, String>,
+}
+
+impl IndexWaitDiagnostics {
+    fn observe_views(&mut self, ctx: &AppContext) {
+        use aft::blob_store::v2::FamilyPlane;
+        use aft::views::readiness::PlaneState;
+
+        if let Some(error) = ctx.callgraph_view_read_diagnostics().last_error {
+            self.last_error.insert("callgraph", error);
+        }
+        if let Some(runtime) = ctx.checkout_semantic_runtime() {
+            if let Some(error) = runtime.unavailable_reason() {
+                self.last_error.insert("semantic", error);
+            }
+            let snapshot = runtime.installed();
+            let membership = snapshot.membership();
+            for (name, plane) in [
+                ("callgraph", FamilyPlane::Callgraph),
+                ("semantic", FamilyPlane::Semantic),
+                ("lexical", FamilyPlane::Trigram),
+            ] {
+                let mut counts = BTreeMap::new();
+                for path in membership.keys() {
+                    let state = snapshot.plane_readiness(path, plane);
+                    let status = match state {
+                        Some(PlaneState::Ready { .. }) => "ready".to_owned(),
+                        Some(PlaneState::Pending { reason }) => format!("pending: {reason:?}"),
+                        Some(PlaneState::Failed { .. }) => {
+                            self.last_error
+                                .insert(name, format!("{path:?}: {state:#?}"));
+                            "failed".to_owned()
+                        }
+                        None => "unset".to_owned(),
+                    };
+                    *counts.entry(status).or_insert(0) += 1;
+                }
+                self.view_status.insert(name, counts);
+            }
+        }
+    }
+
+    fn observe_lexical(&mut self, answer: &Value) {
+        self.status
+            .insert("lexical", answer["index_status"].to_string());
+        if answer["success"] == false || !answer["index_read_gap"].is_null() {
+            self.last_error.insert("lexical", format!("{answer:#}"));
+        }
+    }
+
+    fn report(&self, ctx: &AppContext) -> String {
+        // Diagnostics only sample passive state. Running another query here
+        // could wake a retry and hide the state that exhausted the deadline.
+        let health = ctx.try_health_snapshot(&ctx.canonical_cache_root());
+        let planes: BTreeMap<_, _> = ["callgraph", "semantic", "lexical"]
+            .into_iter()
+            .map(|plane| {
+                (
+                    plane,
+                    json!({
+                        "final_status": self.status.get(plane),
+                        "semantic_member_readiness": self.view_status.get(plane),
+                        "last_error": self.last_error.get(plane),
+                    }),
+                )
+            })
+            .collect();
+        format!(
+            "planes={:#}\ncallgraph_view_reads={:#}\nsemantic_slot={}\nlast_callgraph_reconcile={:#?}\nhealth={:#}",
+            json!(planes),
+            json!(ctx.callgraph_view_read_diagnostics()),
+            ctx.checkout_semantic().describe(),
+            ctx.last_callgraph_reconcile(),
+            json!(health),
+        )
+    }
+}
+
 fn wait_until_ready(ctx: &AppContext) -> Arc<aft::callgraph_store::ReadonlyCallGraphStore> {
-    let deadline = Instant::now() + ROW_DEADLINE;
+    wait_until_ready_observed(ctx, &mut IndexWaitDiagnostics::default())
+}
+
+fn wait_until_ready_observed(
+    ctx: &AppContext,
+    diagnostics: &mut IndexWaitDiagnostics,
+) -> Arc<aft::callgraph_store::ReadonlyCallGraphStore> {
+    let started = Instant::now();
+    let deadline = started + ROW_DEADLINE;
     loop {
         aft::runtime_drain::drain_watcher_events(ctx);
         aft::runtime_drain::drain_search_index_events(ctx);
         aft::runtime_drain::drain_callgraph_store_events(ctx);
         aft::runtime_drain::drain_semantic_index_events(ctx);
         aft::runtime_drain::drain_semantic_refresh_events(ctx);
-        let callgraph = match ctx.callgraph_store_for_ops() {
-            CallgraphStoreAccess::Ready(store) => Some(store),
-            CallgraphStoreAccess::Building => None,
+        let mut failure = None;
+        let (callgraph_status, callgraph) = match ctx.callgraph_store_for_ops() {
+            CallgraphStoreAccess::Ready(store) => ("ready", Some(store)),
+            CallgraphStoreAccess::Building => ("building", None),
             CallgraphStoreAccess::Suspended(reason) => {
-                panic!("callgraph suspended while waiting: {reason:?}")
+                failure = Some(format!("callgraph suspended while waiting: {reason:?}"));
+                ("suspended", None)
+            }
+            CallgraphStoreAccess::Unavailable => {
+                failure = Some("callgraph unavailable while waiting".to_owned());
+                ("unavailable", None)
+            }
+            CallgraphStoreAccess::Off => {
+                failure = Some("callgraph index off while waiting".to_owned());
+                ("off", None)
             }
-            CallgraphStoreAccess::Unavailable => panic!("callgraph unavailable while waiting"),
-            CallgraphStoreAccess::Off => panic!("callgraph index off while waiting"),
             // The view reader opens without a busy wait, so a publication
             // checkpointing the derived database can briefly lock it. Commands
-            // report that as a retryable `callgraph_building`; wait it out too.
-            CallgraphStoreAccess::Error(error) if error.is_transient_lock_contention() => None,
-            CallgraphStoreAccess::Error(error) => panic!("callgraph failed while waiting: {error}"),
+            // expose a retryable SQLite busy error; wait it out too.
+            CallgraphStoreAccess::Error(error) => {
+                diagnostics
+                    .last_error
+                    .insert("callgraph", format!("{error}\n{error:#?}"));
+                if !error.is_transient_lock_contention() {
+                    failure = Some(format!("callgraph failed while waiting: {error}"));
+                }
+                ("error", None)
+            }
         };
+        diagnostics
+            .status
+            .insert("callgraph", callgraph_status.to_owned());
         // With views on, grep and glob read the checkout's lexical view on
         // demand and no legacy search index is ever built; their readiness
         // is checked by the grep answers the callers wait for.
@@ -387,17 +495,42 @@ fn wait_until_ready(ctx: &AppContext) -> Arc<aft::callgraph_store::ReadonlyCallG
                 .unwrap_or_else(std::sync::PoisonError::into_inner)
                 .as_ref()
                 .is_some_and(|index| index.ready);
+        diagnostics.status.insert(
+            "lexical",
+            if ctx.config().views.enabled {
+                "views-on; query readiness checked by grep".to_owned()
+            } else {
+                format!("legacy ready={search_ready}")
+            },
+        );
+        let semantic_status = ctx
+            .semantic_index_status()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone();
+        diagnostics
+            .status
+            .insert("semantic", format!("{semantic_status:#?}"));
+        if let SemanticIndexStatus::Failed(error) = &semantic_status {
+            diagnostics.last_error.insert("semantic", error.clone());
+        }
         let semantic_ready = matches!(
-            &*ctx
-                .semantic_index_status()
-                .read()
-                .unwrap_or_else(std::sync::PoisonError::into_inner),
+            &semantic_status,
             SemanticIndexStatus::Ready { refreshing, .. } if refreshing.is_empty()
         );
+        diagnostics.observe_views(ctx);
+        if let Some(failure) = failure {
+            panic!("{failure}\n{}", diagnostics.report(ctx));
+        }
         if let Some(store) = callgraph.filter(|_| search_ready && semantic_ready) {
             return store;
         }
-        assert!(Instant::now() < deadline, "indexes did not become ready");
+        assert!(
+            Instant::now() < deadline,
+            "indexes did not become ready after {:?}\n{}",
+            started.elapsed(),
+            diagnostics.report(ctx)
+        );
         thread::sleep(Duration::from_millis(10));
     }
 }
@@ -526,11 +659,19 @@ fn run_row(label: &str, row: impl FnOnce()) {
 fn wait_for_views_branch(ctx: &AppContext, root: &Path, branch: char) {
     let other = if branch == 'A' { 'B' } else { 'A' };
     let deadline = Instant::now() + ROW_DEADLINE;
+    let mut diagnostics = IndexWaitDiagnostics::default();
     loop {
-        let _ = wait_until_ready(ctx);
+        let _ = wait_until_ready_observed(ctx, &mut diagnostics);
         let present = grep(ctx, &format!("target{branch}"));
         let absent = grep(ctx, &format!("target{other}"));
         let caller_result = callers(ctx, root, branch);
+        diagnostics.observe_lexical(&present);
+        diagnostics.observe_lexical(&absent);
+        if caller_result["success"] == false {
+            diagnostics
+                .last_error
+                .insert("callgraph", format!("{caller_result:#}"));
+        }
         if present["index_status"] == "Ready"
             && present["total_matches"].as_u64().unwrap_or(0) > 0
             && absent["total_matches"] == 0
@@ -541,7 +682,7 @@ fn wait_for_views_branch(ctx: &AppContext, root: &Path, branch: char) {
         }
         assert!(
             Instant::now() < deadline,
-            "views branch {branch} did not converge: present={present:#} absent={absent:#} callers={caller_result:#}"
+            "views branch {branch} did not converge: present={present:#} absent={absent:#} callers={caller_result:#}\n{}", diagnostics.report(ctx)
         );
         thread::sleep(Duration::from_millis(25));
     }
@@ -613,13 +754,20 @@ fn views_round_trip_reuses_semantic_blobs_without_embedding() {
 /// semantic file of the checkout as it is on disk.
 fn wait_for_semantic_view(ctx: &AppContext) {
     let deadline = Instant::now() + ROW_DEADLINE;
+    let mut diagnostics = IndexWaitDiagnostics::default();
     loop {
         aft::runtime_drain::drain_watcher_events(ctx);
         let runtime = ctx.checkout_semantic_runtime();
         let complete = runtime.as_ref().is_some_and(|runtime| {
-            runtime
-                .search(&[1.0, 0.0, 0.0], 1, &|_| true)
-                .is_ok_and(|answer| answer.complete())
+            match runtime.search(&[1.0, 0.0, 0.0], 1, &|_| true) {
+                Ok(answer) => answer.complete(),
+                Err(error) => {
+                    diagnostics
+                        .last_error
+                        .insert("semantic", format!("{error:#?}"));
+                    false
+                }
+            }
         });
         // A switch the watcher has not delivered yet looks complete for the
         // previous branch; the reconciled snapshot must match the disk too.
@@ -640,12 +788,17 @@ fn wait_for_semantic_view(ctx: &AppContext) {
                             .is_some_and(|state| state.is_pending())
                     })
         });
+        diagnostics
+            .status
+            .insert("semantic", format!("complete={complete} current={current}"));
+        diagnostics.observe_views(ctx);
         if complete && current {
             return;
         }
         assert!(
             Instant::now() < deadline,
-            "views-on semantic view never caught up"
+            "views-on semantic view never caught up\n{}",
+            diagnostics.report(ctx)
         );
         thread::sleep(Duration::from_millis(25));
     }
diff --git a/crates/aft/tests/integration/per_checkout_9.rs b/crates/aft/tests/integration/per_checkout_9.rs
index 2b1bd6fd21..410650e588 100644
--- a/crates/aft/tests/integration/per_checkout_9.rs
+++ b/crates/aft/tests/integration/per_checkout_9.rs
@@ -378,6 +378,9 @@ fn configure(root: &Path, storage: &Path, doc: Value) -> Arc<AppContext> {
 }
 
 fn drain(ctx: &AppContext) {
+    // Match the runtime maintenance tick: a busy bind can defer identity
+    // resolution or view publication beyond configure's first drain.
+    aft::runtime_drain::drain_deferred_configure_maintenance(ctx);
     aft::runtime_drain::drain_watcher_events(ctx);
     aft::runtime_drain::drain_search_index_events(ctx);
     aft::runtime_drain::drain_semantic_index_events(ctx);
@@ -390,14 +393,12 @@ fn trigram_artifact(root: &Path, storage: &Path) -> PathBuf {
 }
 
 /// Waits until the child session's trigram index is ready and persisted.
-/// Each round also runs the configure-tail drain both runtimes run on their
-/// maintenance tick: a child whose bind met a busy aft.db settles its identity
-/// there once the database is free, and only then starts its index.
+/// The runtime drain also settles a bind whose identity was deferred by busy
+/// storage, so its index can start once the database is free.
 fn wait_trigram(ctx: &AppContext, root: &Path, storage: &Path) {
     let deadline = Instant::now() + DEADLINE;
     loop {
         drain(ctx);
-        aft::runtime_drain::drain_deferred_configure_maintenance(ctx);
         let ready = ctx
             .search_index()
             .read()
@@ -420,6 +421,59 @@ fn wait_trigram(ctx: &AppContext, root: &Path, storage: &Path) {
     }
 }
 
+#[test]
+fn runtime_drains_retry_child_trigram_after_busy_identity() {
+    let temp = tempfile::tempdir().unwrap();
+    let base = std::fs::canonicalize(temp.path()).unwrap();
+    let root = base.join("alpha");
+    let storage = base.join("storage");
+    init_repo(
+        &root,
+        &[("src/lib.rs", "pub fn retry_marker() {}\n".into())],
+    );
+    std::fs::create_dir_all(&storage).unwrap();
+    let lock = rusqlite::Connection::open(storage.join("aft.db")).unwrap();
+    // Rollback-journal mode blocks identity readers as well as writers.
+    lock.execute_batch("PRAGMA journal_mode=DELETE; BEGIN EXCLUSIVE;")
+        .unwrap();
+    let ctx = configure(&root, &storage, child_trigram_doc());
+    assert!(ctx
+        .degraded_reasons()
+        .iter()
+        .any(|reason| reason.contains("aft.db was busy at bind")));
+    assert!(ctx.cached_artifact_cache_key(&root).is_none());
+    assert!(!ctx.config().indexes.trigram);
+    lock.execute_batch("ROLLBACK").unwrap();
+    drop(lock);
+
+    let deadline = Instant::now() + DEADLINE;
+    loop {
+        drain(&ctx);
+        if ctx
+            .search_index()
+            .read()
+            .unwrap()
+            .as_ref()
+            .is_some_and(|index| index.ready)
+            && trigram_artifact(&root, &storage).is_file()
+        {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "runtime drains never retried the child's busy identity: {:?}",
+            ctx.degraded_reasons()
+        );
+        thread::sleep(Duration::from_millis(20));
+    }
+    assert!(!ctx
+        .degraded_reasons()
+        .iter()
+        .any(|reason| reason.contains("aft.db was busy at bind")));
+    assert!(ctx.config().indexes.trigram);
+    assert!(ctx.cached_artifact_cache_key(&root).is_some());
+}
+
 /// Asks the child session for its call graph until it answers. The first
 /// answer binds the child's v3 call-graph view and publishes it; a later one
 /// publishes the edits made since.
@@ -427,7 +481,6 @@ fn child_callgraph_answer(ctx: &AppContext, root: &Path) -> Value {
     let deadline = Instant::now() + DEADLINE;
     loop {
         drain(ctx);
-        aft::runtime_drain::drain_deferred_configure_maintenance(ctx);
         let answer = aft::commands::callers::handle_callers(
             &request(json!({
                 "id": "child-callers",
diff --git a/scripts/windows-gate.sh b/scripts/windows-gate.sh
index 28788f2b24..770a13fe03 100755
--- a/scripts/windows-gate.sh
+++ b/scripts/windows-gate.sh
@@ -18,6 +18,8 @@ base=origin/main
 full=0
 filter=
 filter_set=0
+test_target=
+features=
 timeout_minutes=60
 show_status=0
 plan_only=0
@@ -30,12 +32,18 @@ while (($#)); do
     --filter)
       if (($# < 2)); then echo "--filter requires a test filter" >&2; exit 2; fi
       filter=$2; filter_set=1; shift 2 ;;
+    --test-target)
+      if (($# < 2)) || [[ -z "$2" ]]; then echo "--test-target requires a target" >&2; exit 2; fi
+      test_target=$2; shift 2 ;;
+    --features)
+      if (($# < 2)) || [[ -z "$2" ]]; then echo "--features requires a feature list" >&2; exit 2; fi
+      features=$2; shift 2 ;;
     --timeout-minutes)
       if (($# < 2)); then echo "--timeout-minutes requires a positive integer" >&2; exit 2; fi
       timeout_minutes=$2; shift 2 ;;
     --status) show_status=1; shift ;;
     --plan) plan_only=1; shift ;;
-    --help|-h) echo 'Run with runon windows; options: --base, --full, --filter, --timeout-minutes, --status, --plan'; exit 0 ;;
+    --help|-h) echo 'Run with runon windows; options: --base, --full, --filter, --test-target, --features, --timeout-minutes, --status, --plan'; exit 0 ;;
     *) echo "Unknown argument: $1" >&2; exit 2 ;;
   esac
 done
@@ -54,6 +62,10 @@ if ((filter_set)) && [[ -z "$filter" ]]; then
   echo "filter must not be empty" >&2
   exit 2
 fi
+if ((full)) && [[ -n "$test_target" ]]; then
+  echo '--full and --test-target cannot be combined' >&2
+  exit 2
+fi
 
 if ((show_status)); then
   printf 'Windows native test runner\nCargo target directory: %s\n' "${CARGO_TARGET_DIR:-$repo_root/target}"
@@ -100,7 +112,9 @@ select_path() {
   esac
 }
 
-if [[ -n "$filter" ]]; then
+if [[ -n "$test_target" ]]; then
+  plan_label="explicit test target $test_target"
+elif [[ -n "$filter" ]]; then
   plan_label='explicit filter'
 else
   base_sha="$(git rev-parse --verify "$base^{commit}")"
@@ -129,6 +143,10 @@ if [[ -n "$filter" ]]; then
   add_job lib '' "$filter"
   if ((full)); then add_job test integration "$filter"; fi
 fi
+if [[ -n "$test_target" ]]; then
+  job_kinds=(); job_targets=(); job_filters=()
+  add_job test "$test_target" "$filter"
+fi
 
 selected_kinds=(); selected_targets=(); selected_filters=()
 for ((i = 0; i < ${#job_kinds[@]}; i++)); do
@@ -159,11 +177,13 @@ print_job() {
   local kind=$1 target=$2 test_filter=$3
   local -a args=(test --locked -j 4 -p agent-file-tools "--$kind")
   [[ -z "$target" ]] || args+=("$target")
+  [[ -z "$features" ]] || args+=(--features "$features")
   [[ -z "$test_filter" ]] || args+=("$test_filter")
   args+=(-- --test-threads 4)
   printf '  cargo'; printf ' %q' "${args[@]}"; printf '\n'
 }
-printf '  storage preflight: cargo test --locked -j 4 -p agent-file-tools --lib gate_hermeticity_tests::gate_resolves_every_user_config_and_state_path_under_the_gate_homes -- --test-threads 4\n'
+printf '  storage preflight:\n'
+print_job lib '' gate_hermeticity_tests::gate_resolves_every_user_config_and_state_path_under_the_gate_homes
 for ((i = 0; i < ${#job_kinds[@]}; i++)); do print_job "${job_kinds[$i]}" "${job_targets[$i]}" "${job_filters[$i]}"; done
 if ((plan_only)); then exit 0; fi
 
@@ -219,6 +239,7 @@ run_job() {
   local kind=$1 target=$2 test_filter=$3 label=$4 remaining output code
   local -a args=(test --locked -j 4 -p agent-file-tools "--$kind")
   [[ -z "$target" ]] || args+=("$target")
+  [[ -z "$features" ]] || args+=(--features "$features")
   [[ -z "$test_filter" ]] || args+=("$test_filter")
   args+=(-- --test-threads 4)
   remaining=$((timeout_minutes * 60 - (SECONDS - started)))