diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index e5546682..99df9b55 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/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 { + 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