feat(storage key rot): add admin panel

src/infrastructure/scheduler/handler.rs

@@ -76,4 +76,20 @@ pub trait JobHandler: Send + Sync {
     ///
     /// See trait-level docs for guidance on when to return Ok vs Err.
     async fn run(&self, args: &JobRunArgs) -> JobOutcome;
+
+    /// `true` iff this handler persists per-run rows to
+    /// `jobs.recoverable_runs` (cursor + findings + resume). Surfaced
+    /// on [`crate::infrastructure::scheduler::registry::JobSummary`]
+    /// so the admin UI can decide whether the row is expandable to
+    /// show a run history + findings drawer, without hardcoding a
+    /// name-based allowlist.
+    ///
+    /// Default is `false` — Part 1 periodic handlers (`TrashCleanup`,
+    /// `StorageReconcile`, `GrantCleanup`, `DedupGc`) don't have runs
+    /// or findings. `RecoverableAdapter` overrides to `true` so every
+    /// tenant registered via `register_recoverable_job` flips the flag
+    /// automatically at registration time.
+    fn is_recoverable(&self) -> bool {
+        false
+    }
 }

src/infrastructure/scheduler/recoverable.rs

@@ -802,6 +802,14 @@ impl JobHandler for RecoverableAdapter {
     async fn run(&self, args: &JobRunArgs) -> JobOutcome {
         run_or_resume(self.inner.clone(), self.provider.clone(), args).await
     }
+    fn is_recoverable(&self) -> bool {
+        // Every tenant registered through `register_recoverable_job` is
+        // wrapped by this adapter, so this flag flips true for exactly
+        // the set of jobs whose runs + findings the admin UI should
+        // let operators drill into. No name-based allowlists needed
+        // downstream.
+        true
+    }
 }
 
 // ─── Ergonomics: JobRegistry extension for recoverable jobs ─────────────────

src/infrastructure/scheduler/registry.rs

@@ -225,6 +225,7 @@ impl JobRegistry {
                     last_run_at,
                     last_outcome,
                     running: state.current_run_start.is_some(),
+                    recoverable: entry.handler.is_recoverable(),
                 }
             })
             .collect()
@@ -288,6 +289,10 @@ pub enum RegisterError {
 /// - `running` — true iff the in-flight permit is currently held
 ///   (either the supervisor tick is in progress or an admin trigger
 ///   raced in).
+/// - `recoverable` — true iff the job persists runs + findings to
+///   `jobs.recoverable_runs`. Consumed by the admin UI to decide
+///   whether the row is expandable (drawer with run history +
+///   findings) and to gate the retention/purge action.
 #[derive(Debug, Clone, Serialize)]
 pub struct JobSummary {
     pub name: String,
@@ -300,6 +305,7 @@ pub struct JobSummary {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub last_outcome: Option<JobOutcome>,
     pub running: bool,
+    pub recoverable: bool,
 }
 
 #[cfg(test)]

src/interfaces/api/handlers/admin_handler.rs

@@ -685,6 +685,27 @@ pub async fn trigger_storage_rotate(
         )));
     }
 
+    // Refuse on non-active entry. `storage.blobs` describes what's on
+    // the ACTIVE backend; walking it against a stale target produces a
+    // `rotation_failed` finding per missing blob (pure noise) and can't
+    // actually normalise anything the app reads. The right recipe for
+    // "normalise a different backend" is: migrate to it (blobs land in
+    // the head-pair's format on arrival — no rotation needed).
+    let active = state
+        .core
+        .active_backend_name
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .clone();
+    if name != active {
+        return Err(AppError::bad_request(format!(
+            "storage_rotate refuses non-active entry `{name}` — the DB blob registry \
+             describes the active entry (`{active}`), so walking it against a stale \
+             target produces spurious `rotation_failed` findings. Activate `{name}` \
+             first via `Migrate & activate`, then rotate."
+        )));
+    }
+
     // Concurrency guard per plan: at most one encryption-touching
     // recoverable run at a time across the whole app. Rotation
     // rewrites blobs in place; migration copies + swaps; running