feat(core): remote-existence audit heals files whose Drive objects vanished (#171)

## Why

Today a user deleted their backup's Drive folders and emptied the trash.
That left **4,526 `file_state` rows pointing at hard-deleted
`drive_file_id`s.**

#168 fixed half of it: a file that *later changes* plans an UPDATE, the
update 404s, and the stale id self-heals. But a file that **never
changes never plans an update** - so it would have stayed silently
un-backed-up **forever**: no error shown, restore impossible. This PR
adds the proactive pass that closes that gap, and verifies the adjacent
folder-deletion recovery path.

## Part 1: the audit

**Enumeration, not per-file GETs.** New required
`RemoteStore::list_source_object_ids`. Every object the executor creates
carries `appProperties driven.source_id`, so one paged `files.list`
returns a source's whole live footprint in ~N/1000 requests. Folders
carry only `driven.folder_marker`, so they are excluded for free. A new
`ID_ONLY_FIELDS` projection (`fields=nextPageToken,files(id)`) avoids
pulling nine fields per row and discarding eight on a large source.

`SOURCE_ID_KEY` now has **one** definition in `driven-drive`,
re-exported by the executor. Two copies that drifted would make the
audit match nothing, judge every id dead, and re-upload the whole
source.

**The heal.** Dead FILE: clear `drive_file_id` + `drive_md5` and stamp
the `REQUEUE_FORCE_RESCAN_MTIME_NS` sentinel (the exact remediation
applied by hand during the incident), so the next scan re-emits the path
in *either* scan mode and the executor re-creates it. Dead BUNDLE
object: `heal_dead_bundle` re-queues the members and drops the `bundles`
row in **one transaction** - members must be read before the row is
deleted, because the `bundle_id` FK cascades the membership rows away.
Members then re-upload individually, which is the same standalone
promotion a bundled member already gets whenever it changes.

**The safety property.** The audit infers "gone" from **absence**, so an
enumeration that cannot be completed writes **nothing** and returns
`Err`. A partial listing would name live objects as dead and churn the
entire source; retrying next cycle is free. This is also why the trait
method has **no default body** - "no live objects" is not a safe
degradation, it is that same failure arriving silently, so
`BreakerReportingStore` and every test double must delegate or error
explicitly.

**Scheduling.** Once per source per process, **plus** every deep-verify
cycle. Startup-once matters because the damage appeared overnight and
`deep_verify_interval_secs` defaults to a week. It runs **before** the
scan, so a heal completes in one cycle instead of two. A source is
marked done only on success, so a transient failure retries rather than
being skipped until restart; an audit error never fails the cycle.

**Reporting.** Capped at 20 per-file `drive.remote_file_missing` WARN
rows - the incident would have written 4,526 - plus one
`remote_audit_done` Info row whose `file_count` carries the **full**
healed count, so the cap hides nothing. A clean audit writes nothing at
all.

## Part 2: ancestor-deletion recovery - verified, and it found a real
wedge

The answer is yes, it recovers, but **not** for the reason the brief
guessed, and there was a genuine bug.

Not the mechanism: nothing in the orchestrator halts a source on
`dest_folder_missing` (grep returns no matches), so that was never the
failure mode.

The real one: an intermediate folder cached in `parent_dirs` can be dead
while a **deeper component is not cached** - a *new subdirectory* under
a deleted-but-cached parent. `ensure_folder` then 404s **inside**
`resolve_remote_target`, which maps to `UploadError::Fatal` and
**aborted the entire `execute()` for the source**. And because
`invalidate_parent_dirs` only ran *after* the target resolved, the
poisoned cache entry **survived for the process lifetime** - every later
file under that chain failed identically. A permanent wedge until
restart.

Fix: `ensure_parents` now drops the path's cached chain and re-ensures
**once** from the source root, gated on the invalidation having actually
removed something. That gate is what keeps a deleted destination
**root** on its unchanged fail-fast mass-delete guard - the root is
never in this cache, so it removes nothing, never retries, and is never
silently re-created. Only the subfolder chain self-heals.

Three cases are pinned: cold cache, warm cache with the whole chain
cached (recovers in two cycles via the pre-existing invalidate path - a
per-op failure, never an aborted cycle), and the new-subdirectory case
that was broken.

## Test-infrastructure notes

- The fake gains `delete_folder_tree` - real Drive **cascades** a folder
deletion, while `trash`/`delete_permanent` touch one object, and that
cascade is the incident.
- New `with_source_listing_broken` fault, scoped to the enumeration, so
a test can let its setup uploads land and break *only* the audit.
- One e2e fixture hand-created an orphan **without** the `source_id`
stamp - unfaithful to the executor, which has stamped it since v0.1.0
(verified present in `v1.0.0`, so there is no back-compat hazard for
existing installs). It is now stamped. A duplicate re-upload in that
test is what caught it.

## Gates

`cargo fmt`; `clippy -p driven-core -p driven-drive --all-targets -D
warnings` clean; `cargo test -p driven-core -p driven-drive` all green
(533 tests); `cargo check --workspace --all-targets` clean; vitest 515
passed; prettier clean.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

https://claude.ai/code/session_01JLB3E2Jm7knNJd37fVpH8X

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: pmaxhogan

crates/driven-core/src/executor.rs

@@ -1152,6 +1152,120 @@ pub trait StateRepo: Send + Sync {
         path: &RelativePath,
     ) -> Result<()>;
 
+    /// Every `(relative_path, drive_file_id)` pair this source records - i.e.
+    /// every file whose bytes are supposed to exist as a STANDALONE Drive
+    /// object right now.
+    ///
+    /// The recorded half of the remote-existence audit: the executor diffs
+    /// these ids against the live ids Drive reports
+    /// ([`driven_drive::remote_store::RemoteStore::list_source_object_ids`])
+    /// and heals whatever is recorded but no longer live. Rows with a NULL
+    /// `drive_file_id` are EXCLUDED - a bundled member (whose bytes live inside
+    /// a `.tar.gz`) and a never-uploaded row both have no standalone object to
+    /// audit, and treating their absence from Drive as damage would re-upload
+    /// them on every pass.
+    ///
+    /// Leaner than [`Self::load_source_file_state`] on purpose: the audit needs
+    /// two columns, and a source can hold hundreds of thousands of rows.
+    ///
+    /// The default returns an empty list, which reads as "this source records
+    /// no standalone objects" and makes the audit a no-op - the SAFE
+    /// degradation for a fake that models no `file_state` (mirrors
+    /// [`Self::list_empty_bundles`]).
+    async fn list_file_state_drive_ids(
+        &self,
+        source: SourceId,
+    ) -> Result<Vec<(RelativePath, String)>> {
+        let _ = source;
+        Ok(Vec::new())
+    }
+
+    /// Every `(bundle_id, drive_file_id)` this source records - the bundle half
+    /// of the remote-existence audit's recorded set.
+    ///
+    /// Distinct from [`Self::list_empty_bundles`], which returns only the
+    /// MEMBERLESS bundles that GC should trash. The audit needs ALL of them,
+    /// because a bundle whose Drive object was deleted out-of-band is still
+    /// full of members - members whose bytes are now nowhere. Default empty
+    /// (audit no-op), like the other bundle accessors.
+    async fn list_bundles_for_source(&self, source: SourceId) -> Result<Vec<(String, String)>> {
+        let _ = source;
+        Ok(Vec::new())
+    }
+
+    /// Re-queue one `file_state` row for a fresh upload after its recorded
+    /// Drive object was proven GONE.
+    ///
+    /// Three writes in one targeted UPDATE:
+    /// - `drive_file_id = NULL`, so the executor's next execution of this path
+    ///   takes the CREATE branch (it decides create-vs-update from this column,
+    ///   not from the plan) instead of a doomed UPDATE against a dead id;
+    /// - `drive_md5 = NULL`, since the md5 described bytes that no longer
+    ///   exist and would otherwise linger as a claim about a live object;
+    /// - `mtime_ns = force_rescan_mtime_ns`, the caller's sentinel. Clearing
+    ///   the id ALONE is not enough: the scanner treats a file as unchanged
+    ///   when its `(size, mtime_ns)` equals the stored row's, so an untouched
+    ///   file would never be re-emitted and the cleared id would sit there
+    ///   forever. A sentinel no real filesystem can produce guarantees the
+    ///   very next scan re-emits the path in EITHER scan mode.
+    ///
+    /// The row itself survives - only its remote pointer was wrong - so the
+    /// file's recorded history and its `bundle_members` linkage (if any) are
+    /// untouched. A missing row is an idempotent no-op.
+    ///
+    /// The default errors rather than silently succeeding: a repo that cannot
+    /// perform the heal must not report a file as healed when its row still
+    /// points at a dead object. Unreachable under the default listings above,
+    /// which return nothing to heal.
+    async fn requeue_file_state_for_reupload(
+        &self,
+        source: SourceId,
+        path: &RelativePath,
+        force_rescan_mtime_ns: i64,
+    ) -> Result<()> {
+        let _ = (source, path, force_rescan_mtime_ns);
+        Err(anyhow::anyhow!(
+            "requeue_file_state_for_reupload is not implemented by this StateRepo"
+        ))
+    }
+
+    /// Heal one bundle whose `.tar.gz` Drive object is GONE, returning the
+    /// member paths that were re-queued.
+    ///
+    /// In ONE transaction: read the bundle's members, re-queue each member's
+    /// `file_state` row exactly as
+    /// [`Self::requeue_file_state_for_reupload`] does, then delete the
+    /// `bundles` row (whose `bundle_id` FK cascades the now-meaningless
+    /// `bundle_members` rows away).
+    ///
+    /// The ordering is load-bearing and the reason this is one method rather
+    /// than a loop at the call site: deleting the bundle row cascades the
+    /// membership rows, so once it is gone the member list is UNRECOVERABLE.
+    /// Members must be read - and re-queued - first. Doing it in a transaction
+    /// additionally means a crash mid-heal cannot leave members orphaned from a
+    /// deleted bundle with no re-upload pending.
+    ///
+    /// Each member keeps `drive_file_id = NULL` (it never had a standalone
+    /// object), so the sentinel mtime is what does the work: the next scan
+    /// re-emits the path and the planner - seeing an existing `file_state` row,
+    /// which makes the file "changed" rather than "genuinely new" - schedules it
+    /// as an INDIVIDUAL upload. That is the same standalone promotion a bundled
+    /// member gets whenever it changes, so the recovery path is one the
+    /// executor already exercises rather than a bespoke re-bundle.
+    ///
+    /// Default errors, for the same reason as
+    /// [`Self::requeue_file_state_for_reupload`].
+    async fn heal_dead_bundle(
+        &self,
+        bundle_id: &str,
+        force_rescan_mtime_ns: i64,
+    ) -> Result<Vec<RelativePath>> {
+        let _ = (bundle_id, force_rescan_mtime_ns);
+        Err(anyhow::anyhow!(
+            "heal_dead_bundle is not implemented by this StateRepo"
+        ))
+    }
+
     /// R2-P1-3 (DESIGN s5.4): increment the CONSECUTIVE checksum-mismatch
     /// counter for `(source, path)` by one and return the NEW count. After the
     /// 3rd consecutive mismatch the executor marks the file

crates/driven-core/src/orchestrator.rs

@@ -1576,6 +1576,106 @@ impl StateRepo for SqliteStateRepo {
         Ok(())
     }
 
+    async fn list_file_state_drive_ids(
+        &self,
+        source: SourceId,
+    ) -> Result<Vec<(RelativePath, String)>> {
+        let source_str = source.to_string();
+        // Runtime `sqlx::query` (NOT the compile-checked `query!` macro) so this
+        // additive method needs NO `.sqlx` cache regeneration - the same reason
+        // `clear_file_state_drive_file_id` uses it.
+        let rows: Vec<(String, String)> = sqlx::query_as(
+            "SELECT relative_path, drive_file_id FROM file_state \
+             WHERE source_id = ?1 AND drive_file_id IS NOT NULL",
+        )
+        .bind(source_str.as_str())
+        .fetch_all(&self.pool)
+        .await?;
+        let mut out = Vec::with_capacity(rows.len());
+        for (path, drive_file_id) in rows {
+            out.push((relative_path_from_string(path)?, drive_file_id));
+        }
+        Ok(out)
+    }
+
+    async fn list_bundles_for_source(&self, source: SourceId) -> Result<Vec<(String, String)>> {
+        let source_str = source.to_string();
+        let rows: Vec<(String, String)> =
+            sqlx::query_as("SELECT id, drive_file_id FROM bundles WHERE source_id = ?1")
+                .bind(source_str.as_str())
+                .fetch_all(&self.pool)
+                .await?;
+        Ok(rows)
+    }
+
+    async fn requeue_file_state_for_reupload(
+        &self,
+        source: SourceId,
+        path: &RelativePath,
+        force_rescan_mtime_ns: i64,
+    ) -> Result<()> {
+        let source_str = source.to_string();
+        let path_str = path.as_str().to_string();
+        // A targeted three-column UPDATE: it cannot clobber a column a
+        // concurrent edit may have changed, and a missing row updates 0 rows
+        // (idempotent no-op).
+        sqlx::query(
+            "UPDATE file_state SET drive_file_id = NULL, drive_md5 = NULL, mtime_ns = ?3 \
+             WHERE source_id = ?1 AND relative_path = ?2",
+        )
+        .bind(source_str.as_str())
+        .bind(path_str.as_str())
+        .bind(force_rescan_mtime_ns)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    async fn heal_dead_bundle(
+        &self,
+        bundle_id: &str,
+        force_rescan_mtime_ns: i64,
+    ) -> Result<Vec<RelativePath>> {
+        let mut tx = self.pool.begin().await?;
+
+        // Read the members FIRST: the `DELETE FROM bundles` below cascades
+        // `bundle_members` away via its `bundle_id` FK, after which this list
+        // cannot be recovered.
+        let members: Vec<(String, String)> = sqlx::query_as(
+            "SELECT source_id, relative_path FROM bundle_members WHERE bundle_id = ?1",
+        )
+        .bind(bundle_id)
+        .fetch_all(&mut *tx)
+        .await?;
+
+        let mut paths = Vec::with_capacity(members.len());
+        for (source_id, relative_path) in &members {
+            // Members carry `drive_file_id = NULL` already (their bytes lived
+            // in the bundle), so the sentinel mtime is what forces the rescan.
+            // The NULLs are still written so the statement is identical to
+            // `requeue_file_state_for_reupload` and stays correct if a member
+            // ever does hold an id.
+            sqlx::query(
+                "UPDATE file_state SET drive_file_id = NULL, drive_md5 = NULL, mtime_ns = ?3 \
+                 WHERE source_id = ?1 AND relative_path = ?2",
+            )
+            .bind(source_id.as_str())
+            .bind(relative_path.as_str())
+            .bind(force_rescan_mtime_ns)
+            .execute(&mut *tx)
+            .await?;
+            paths.push(relative_path_from_string(relative_path.clone())?);
+        }
+
+        sqlx::query("DELETE FROM bundles WHERE id = ?1")
+            .bind(bundle_id)
+            .execute(&mut *tx)
+            .await?;
+
+        tx.commit().await?;
+        Ok(paths)
+    }
+
     async fn bump_checksum_mismatch_count(
         &self,
         source: SourceId,

crates/driven-core/src/state/mod.rs

@@ -44,7 +44,7 @@ use driven_core::types::{
 use driven_crypto::key::SourceKey;
 use driven_crypto::{ContentDecryptor, DrivenCryptoSuite, SourceCryptoSuite, HEADER_LEN};
 
-use driven_drive::fake::{InMemoryRemoteStore, CLIENT_OP_UUID_KEY};
+use driven_drive::fake::{InMemoryRemoteStore, CLIENT_OP_UUID_KEY, SOURCE_ID_KEY};
 use driven_drive::remote_store::{DriveContext, RemoteStore, UploadBody};
 
 use driven_power::{PowerSource, PowerState};
@@ -1039,10 +1039,22 @@ async fn reconcile_requeue_reuploads_changed_bytes_on_next_cycle() {
     let old_bytes = b"OLD uploaded bytes".to_vec();
     let new_bytes = b"NEW locally-edited bytes - different length".to_vec();
 
+    // The source is built FIRST so the orphan below can be stamped with its id,
+    // exactly as the executor would have stamped it.
+    let src = source_in(account, src_dir.path(), &folder);
+    state.upsert_source(&src).await.unwrap();
+
     // The orphan landed on Drive with the OLD bytes + its client_op_uuid.
     let op_uuid = uuid::Uuid::new_v4().to_string();
     let mut app = std::collections::HashMap::new();
     app.insert(CLIENT_OP_UUID_KEY.to_string(), op_uuid.clone());
+    // The orphan stands in for an object the EXECUTOR created, and the executor
+    // stamps the owning source on everything it creates (it has done so since
+    // v0.1.0). Omitting it made this fixture unfaithful in a way that now
+    // matters: the remote-existence audit enumerates a source's live objects by
+    // exactly this key, so an UNSTAMPED object reads as deleted - and the
+    // adopted orphan would be re-uploaded as a duplicate.
+    app.insert(SOURCE_ID_KEY.to_string(), src.id.to_string());
     let created = remote
         .create(
             &folder,
@@ -1057,8 +1069,6 @@ async fn reconcile_requeue_reuploads_changed_bytes_on_next_cycle() {
     // But locally the file now holds the NEW bytes (edited after the upload,
     // before the lost commit).
     write_file(src_dir.path(), "drift.bin", &new_bytes);
-    let src = source_in(account, src_dir.path(), &folder);
-    state.upsert_source(&src).await.unwrap();
     let rel = RelativePath::try_from("drift.bin".to_string()).unwrap();
 
     let clock = Arc::new(FakeClock::new());

crates/driven-core/src/state/sqlite.rs

@@ -193,6 +193,29 @@ impl InMemoryRemoteStore {
         self
     }
 
+    /// Latches "the source-object enumeration cannot be completed": every
+    /// [`crate::remote_store::RemoteStore::list_source_object_ids`] call fails.
+    /// Every OTHER call - uploads, reads, trashes - keeps working.
+    ///
+    /// Targets that ONE call on purpose. It models the failure the
+    /// remote-existence audit must survive without writing anything: the
+    /// enumeration is the only input that tells the audit which objects are
+    /// still alive, so a test needs the setup uploads to land normally and then
+    /// exactly that request to fail. A blanket read fault would break the
+    /// setup too and prove nothing.
+    ///
+    /// The audit infers "gone" from ABSENCE, so this is the fault that would
+    /// otherwise re-upload an entire source - the reason
+    /// `list_source_object_ids` returns `Err` rather than a partial set.
+    ///
+    /// Latches for the lifetime of the store.
+    pub fn with_source_listing_broken(self) -> Self {
+        self.faults
+            .source_listing_broken
+            .store(true, Ordering::Release);
+        self
+    }
+
     /// Latches the destination-folder-readonly state. Every subsequent
     /// write-target request returns `drive.dest_folder_permission_denied`
     /// (SPEC s24). Read-only calls keep working - mirrors the user