feat(jobs): the engine binds a run to the flags it started with

`run_or_resume` now records `JobRunArgs` in `params` on a Fresh open and
restores them on Resume, passing the restored args to the handler rather
than whatever the resuming caller supplied.

Two problems, and the engine is the only place both are guaranteed.

**A resumed run must not change mode.** Handlers read `args` on every
call, so a paused `?repair=true` import resumed by a plain trigger
silently continued as import-only: the deletion half never finished and
nothing reported it. `?deep=true` had the same hole — a paused bit-rot
scan resumed shallow while still presenting as the run that started
deep. `blobs_consistency` and `manifests_consistency` had hand-rolled
this for their own two flags; the three import jobs had not, and fixing
it per-handler means every future job remembering.

**The run row should say what it did.** For a destructive job, "did this
run delete anything?" is answerable only from `params`, which is what an
operator reads afterwards. Before this, a repair run and a discovery run
were indistinguishable in the history.

Deliberately not overridable on resume: adding `?repair=true` to a
resume would apply it to the remaining entries only, producing a run
that half-deleted. Cancel and start fresh is the honest way to change
your mind.

A missing key reads as false/None, so a run paused before this existed
resumes under-acting rather than deleting under a flag nobody gave it.
Failure to record the flags fails the run instead of guessing — acting
under unrecorded flags is the one thing worth refusing for jobs that
delete.

The flag list is hardcoded here. Letting each job declare its own
parameters — name, type, default — is the better shape and is its own
change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/infrastructure/scheduler/recoverable.rs

@@ -199,6 +199,63 @@ impl RunOutcome {
     }
 }
 
+/// Write `JobRunArgs` to `params` on a Fresh run, or read them back on a
+/// Resumed one.
+///
+/// Returns the args the handler should actually use. On resume that is
+/// whatever the original run recorded, NOT what the resuming caller
+/// passed — see the call site in [`run_or_resume`] for why changing mode
+/// mid-run is refused.
+///
+/// Every flag is stored as a string, matching the `params` convention the
+/// progress fields already use, and each is read back independently: a run
+/// paused before this existed simply has no keys, and each missing one
+/// falls back to `false` / `None`. That is the safe direction — a resumed
+/// legacy run under-acts rather than deleting under a flag nobody gave it.
+async fn persist_or_restore_args(
+    store: &dyn JobStore,
+    args: &JobRunArgs,
+    is_fresh: bool,
+) -> Result<JobRunArgs, String> {
+    const FLAGS: [&str; 3] = ["force", "deep", "repair"];
+
+    if is_fresh {
+        for (key, value) in FLAGS.iter().zip([args.force, args.deep, args.repair]) {
+            let v = if value { "true" } else { "false" };
+            store
+                .set_string_param(key, v)
+                .await
+                .map_err(|e| format!("persist `{key}` to params: {e}"))?;
+        }
+        // `storage` is absent rather than empty when unset, so a run that
+        // did not scope itself does not grow a key claiming it did.
+        if let Some(name) = &args.storage {
+            store
+                .set_string_param("storage", name)
+                .await
+                .map_err(|e| format!("persist `storage` to params: {e}"))?;
+        }
+        return Ok(args.clone());
+    }
+
+    let mut restored = JobRunArgs::default();
+    for (key, slot) in FLAGS.iter().zip([
+        &mut restored.force,
+        &mut restored.deep,
+        &mut restored.repair,
+    ]) {
+        *slot = match store.get_string_param(key).await {
+            Ok(v) => v.as_deref() == Some("true"),
+            Err(e) => return Err(format!("read `{key}` from params: {e}")),
+        };
+    }
+    restored.storage = store
+        .get_string_param("storage")
+        .await
+        .map_err(|e| format!("read `storage` from params: {e}"))?;
+    Ok(restored)
+}
+
 // ─── Traits — implementor + port ────────────────────────────────────────────
 
 /// The implementor-facing contract for a long-running, restart-tolerant
@@ -797,10 +854,44 @@ pub async fn run_or_resume(
         }
     }
 
+    // Bind the run to the flags it started with.
+    //
+    // A Fresh run records its `JobRunArgs` in `params`; a Resumed run reads
+    // them back and runs with THOSE, ignoring whatever the resuming caller
+    // passed. Two reasons, and the engine is the only place both are
+    // guaranteed:
+    //
+    // **A resumed run must not change mode.** Handlers read `args` on every
+    // call, so a paused `?repair=true` import resumed by a plain trigger
+    // silently continued as import-only — the deletion half never finished
+    // and nothing said so. The same held for `?deep=true`: a paused bit-rot
+    // scan resumed shallow while still reporting as the run that started
+    // deep. Fixing it per-handler meant every job remembering, and three of
+    // them did not.
+    //
+    // **The run row should say what it did.** For a destructive job, "did
+    // this run delete anything?" is answerable only from `params`, and that
+    // is what an operator reads afterwards.
+    //
+    // Deliberately NOT overridable on resume. Adding `?repair=true` to a
+    // resume would apply it to the remaining entries only, producing a run
+    // that half-deleted — the honest way to change your mind is to cancel
+    // and start fresh.
+    let args = match persist_or_restore_args(&*store, args, is_fresh).await {
+        Ok(effective) => effective,
+        Err(e) => {
+            // Fail the run rather than guess. Proceeding would mean acting
+            // under flags nothing recorded, which for the jobs that delete
+            // is the one thing worth refusing.
+            log_terminal_write_err("mark_failed", run_id, store.mark_failed(&e).await);
+            return JobOutcome::err(e);
+        }
+    };
+
     // Dispatch. Terminal writes to `jobs.recoverable_runs` happen
     // here (NOT in the handler) so the row always ends in a state
     // that matches what the handler returned.
-    let outcome = job.run_resumable(&*store, args, resume_cursor).await;
+    let outcome = job.run_resumable(&*store, &args, resume_cursor).await;
 
     // Fetch the terminal run summary so we can surface aggregate
     // stats (finding_count, scanned_count) on the outer JobOutcome