feat(recoverable-job): wire API (cancel, view run, ...)

src/common/di.rs

@@ -2145,7 +2145,12 @@ impl AppServiceFactory {
         // periodic-triggered recoverable job's first tick sees a clean
         // slate. See `docs/plan/job-registry.md` Part 2.
         use crate::infrastructure::scheduler::JobStoreProvider as _;
-        match app_state.core.job_store_provider.boot_recovery_sweep().await {
+        match app_state
+            .core
+            .job_store_provider
+            .boot_recovery_sweep()
+            .await
+        {
             Ok(0) => tracing::debug!(
                 target: "oxicloud::scheduler",
                 event = "recoverable.boot_recovery",
@@ -2212,8 +2217,7 @@ pub struct CoreServices {
     /// `svc.register_recoverable_job(&registry, &job_store_provider).await`.
     /// Boot-time crash-recovery sweep is run in `build_app_state` right
     /// after this provider is created.
-    pub job_store_provider:
+    pub job_store_provider: Arc<crate::infrastructure::scheduler::PgJobStoreProvider>,
-        Arc<crate::infrastructure::scheduler::PgJobStoreProvider>,
 }
 
 /// Container for repository services

src/infrastructure/scheduler/mod.rs

@@ -34,7 +34,7 @@ pub use handler::JobHandler;
 pub use pg_job_store::{PgJobStore, PgJobStoreProvider};
 pub use recoverable::{
     JobStore, JobStoreProvider, OpenedRun, RecoverableAdapter, RecoverableJobHandler, RunOutcome,
-    RunStatus, run_or_resume,
+    RunStatus, RunSummary, run_or_resume,
 };
 pub use registry::{JobEntry, JobRegistry, JobSummary, RegisterError};
 pub use types::{ErrCause, JobOutcome, JobRunArgs};

src/infrastructure/scheduler/pg_job_store.rs

@@ -16,7 +16,7 @@ use uuid::Uuid;
 
 use crate::common::errors::DomainError;
 
-use super::recoverable::{JobStore, JobStoreProvider, OpenedRun, RunStatus};
+use super::recoverable::{JobStore, JobStoreProvider, OpenedRun, RunStatus, RunSummary};
 
 // ─── PgJobStore — bound to one run ──────────────────────────────────────────
 
@@ -239,8 +239,132 @@ impl JobStoreProvider for PgJobStoreProvider {
         .map_err(|e| map_sqlx_err("boot_recovery_sweep", e))?;
         Ok(result.rows_affected())
     }
+
+    async fn list_runs(&self, job_name: &str, limit: u32) -> Result<Vec<RunSummary>, DomainError> {
+        // Cap the limit at 100 defensively — the API layer should
+        // also clamp, but a broken caller shouldn't tank the DB.
+        let capped = limit.min(100) as i64;
+        let rows: Vec<RunSummaryRow> = sqlx::query_as(RUN_SUMMARY_SELECT_LIST)
+            .bind(job_name)
+            .bind(capped)
+            .fetch_all(self.pool.as_ref())
+            .await
+            .map_err(|e| map_sqlx_err("list_runs", e))?;
+        rows.into_iter().map(row_to_summary).collect()
     }
 
+    async fn get_run_by_id(&self, run_id: Uuid) -> Result<Option<RunSummary>, DomainError> {
+        let row: Option<RunSummaryRow> = sqlx::query_as(RUN_SUMMARY_SELECT_BY_ID)
+            .bind(run_id)
+            .fetch_optional(self.pool.as_ref())
+            .await
+            .map_err(|e| map_sqlx_err("get_run_by_id", e))?;
+        row.map(row_to_summary).transpose()
+    }
+
+    async fn request_cancel(&self, job_name: &str) -> Result<Option<Uuid>, DomainError> {
+        // Only Running → CancelRequested flips. `Paused` can be
+        // cancelled by not resuming — no need for a state change.
+        // `CancelRequested` already is what it is.
+        // Multiple Running rows shouldn't exist (partial unique index),
+        // but LIMIT 1 is defensive.
+        let flipped: Option<(Uuid,)> = sqlx::query_as(
+            r#"
+            UPDATE jobs.recoverable_runs
+               SET status           = 'CancelRequested',
+                   last_progress_at = NOW()
+             WHERE id = (
+                    SELECT id FROM jobs.recoverable_runs
+                     WHERE job_name = $1
+                       AND status = 'Running'
+                     ORDER BY started_at DESC
+                     LIMIT 1
+                   )
+            RETURNING id
+            "#,
+        )
+        .bind(job_name)
+        .fetch_optional(self.pool.as_ref())
+        .await
+        .map_err(|e| map_sqlx_err("request_cancel", e))?;
+        Ok(flipped.map(|(id,)| id))
+    }
+}
+
+// ─── Shared row → RunSummary decoder ────────────────────────────────────────
+
+/// Row shape returned by the run-summary SELECTs. Kept as a distinct
+/// type so both `list_runs` and `get_run_by_id` share the projection
+/// (SQL column list + decoder). Order matches the SELECT below.
+type RunSummaryRow = (
+    Uuid,                  // id
+    String,                // job_name
+    String,                // status
+    DateTime<Utc>,         // started_at
+    DateTime<Utc>,         // last_progress_at
+    Option<DateTime<Utc>>, // completed_at
+    Option<Vec<u8>>,       // cursor
+    serde_json::Value,     // stats
+    serde_json::Value,     // params
+    Option<String>,        // error_message
+);
+
+const RUN_SUMMARY_COLUMNS: &str = "id, job_name, status, started_at, last_progress_at, completed_at, cursor, stats, params, error_message";
+
+// `format!` isn't const, but `concat!` gives us a &'static str at compile
+// time — worth it so the SELECT strings show up in tracing / SQL logs
+// as one contiguous line instead of a runtime string build.
+const RUN_SUMMARY_SELECT_LIST: &str = concat!(
+    "SELECT id, job_name, status, started_at, last_progress_at, completed_at, cursor, stats, params, error_message ",
+    "FROM jobs.recoverable_runs ",
+    "WHERE job_name = $1 ",
+    "ORDER BY started_at DESC ",
+    "LIMIT $2"
+);
+
+const RUN_SUMMARY_SELECT_BY_ID: &str = concat!(
+    "SELECT id, job_name, status, started_at, last_progress_at, completed_at, cursor, stats, params, error_message ",
+    "FROM jobs.recoverable_runs ",
+    "WHERE id = $1"
+);
+
+fn row_to_summary(row: RunSummaryRow) -> Result<RunSummary, DomainError> {
+    let (
+        id,
+        job_name,
+        status_str,
+        started_at,
+        last_progress_at,
+        completed_at,
+        cursor,
+        stats,
+        params,
+        error_message,
+    ) = row;
+    let status = RunStatus::parse(&status_str).ok_or_else(|| {
+        DomainError::internal_error("JobStore", format!("unknown status: {status_str}"))
+    })?;
+    Ok(RunSummary {
+        id,
+        job_name,
+        status,
+        started_at,
+        last_progress_at,
+        completed_at,
+        stats,
+        params,
+        cursor_hex: cursor.map(hex::encode),
+        error_message,
+    })
+}
+
+// Suppress the dead-code lint on the column list — kept as a
+// human-readable constant even though the actual SELECTs currently
+// inline it. Future rewrites of the SELECTs (e.g. adding stats
+// projection) will use it.
+#[allow(dead_code)]
+const _RUN_SUMMARY_COLUMNS_UNUSED: &str = RUN_SUMMARY_COLUMNS;
+
 /// Row shape returned by `open_or_start`'s SELECT — factored out
 /// so clippy's `type_complexity` lint doesn't yell at the query.
 type ExistingRun = (Uuid, String, DateTime<Utc>, Option<Vec<u8>>);
@@ -283,10 +407,7 @@ impl PgJobStoreProvider {
                 })?;
                 match status {
                     RunStatus::Running | RunStatus::CancelRequested => {
-                        Ok(OpenedRun::AlreadyActive {
+                        Ok(OpenedRun::AlreadyActive { run_id: id, status })
-                            run_id: id,
-                            status,
-                        })
                     }
                     RunStatus::Paused => {
                         // Flip to Running and hand back the cursor.
@@ -313,20 +434,15 @@ impl PgJobStoreProvider {
                         .bind(id)
                         .fetch_optional(self.pool.as_ref())
                         .await
-                        .map_err(|e| {
+                        .map_err(|e| OpenErr::Fatal(map_sqlx_err("open_or_start.resume", e)))?;
-                            OpenErr::Fatal(map_sqlx_err("open_or_start.resume", e))
-                        })?;
                         let (started_at, cursor_bytes) = row.ok_or_else(|| {
                             OpenErr::Fatal(DomainError::internal_error(
                                 "JobStore",
                                 format!("run vanished during resume: {id}"),
                             ))
                         })?;
-                        let store: Arc<dyn JobStore> = Arc::new(PgJobStore::new(
+                        let store: Arc<dyn JobStore> =
-                            self.pool.clone(),
+                            Arc::new(PgJobStore::new(self.pool.clone(), id, started_at));
-                            id,
-                            started_at,
-                        ));
                         Ok(OpenedRun::Resumed {
                             store,
                             cursor: cursor_bytes.unwrap_or_default(),

src/infrastructure/scheduler/recoverable.rs

@@ -256,6 +256,54 @@ pub trait JobStoreProvider: Send + Sync {
     /// via `POST /api/admin/jobs/{name}/trigger`, which calls
     /// `open_or_start` and picks up the Paused cursor.
     async fn boot_recovery_sweep(&self) -> Result<u64, DomainError>;
+
+    /// Latest N runs for `job_name`, newest first, terminal + non-terminal
+    /// both included. Powers `GET /api/admin/jobs/{name}/runs`. `limit`
+    /// caps the return size; the API layer clamps it too.
+    async fn list_runs(&self, job_name: &str, limit: u32) -> Result<Vec<RunSummary>, DomainError>;
+
+    /// Fetch one run by id. Powers `GET /api/admin/jobs/{name}/runs/{id}`.
+    /// Returns `None` when the id doesn't exist (unknown or pruned).
+    async fn get_run_by_id(&self, run_id: Uuid) -> Result<Option<RunSummary>, DomainError>;
+
+    /// Request cancellation of the CURRENT active run for `job_name`
+    /// by flipping its status from `Running` → `CancelRequested`.
+    /// Returns the run's id when a Running row was flipped, `None`
+    /// when there was no Running row to cancel (nothing in flight,
+    /// or the latest non-terminal row is already `Paused` /
+    /// `CancelRequested`).
+    ///
+    /// Cooperative — the handler still needs to poll `store.status()`
+    /// and return `RunOutcome::Paused` at the next safe boundary. If
+    /// the handler doesn't poll, cancel is a no-op until the run
+    /// completes naturally.
+    async fn request_cancel(&self, job_name: &str) -> Result<Option<Uuid>, DomainError>;
+}
+
+/// Serialisable snapshot of one `jobs.recoverable_runs` row, returned
+/// by the admin listing + get-run endpoints.
+#[derive(Debug, Clone, Serialize)]
+pub struct RunSummary {
+    pub id: Uuid,
+    pub job_name: String,
+    pub status: RunStatus,
+    pub started_at: DateTime<Utc>,
+    pub last_progress_at: DateTime<Utc>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub completed_at: Option<DateTime<Utc>>,
+    /// `stats` JSONB dump — job-specific counters (scanned_count,
+    /// migrated_blobs, findings_this_run, …).
+    pub stats: serde_json::Value,
+    /// `params` JSONB dump — per-run params captured at start
+    /// (grace_window_secs, source_backend, …).
+    pub params: serde_json::Value,
+    /// Cursor as hex — omitted when null. Operators occasionally want
+    /// to inspect this for "where did the scan get to" diagnostics;
+    /// the raw bytes are opaque per-job so we render as hex.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor_hex: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error_message: Option<String>,
 }
 
 /// Result of [`JobStoreProvider::open_or_start`].
@@ -324,11 +372,7 @@ pub async fn run_or_resume(
         }
         RunOutcome::Paused { cursor } => {
             let cursor_hex = hex::encode(&cursor);
-            log_terminal_write_err(
+            log_terminal_write_err("mark_paused", run_id, store.mark_paused(Some(cursor)).await);
-                "mark_paused",
-                run_id,
-                store.mark_paused(Some(cursor)).await,
-            );
             JobOutcome::ok_with(
                 0,
                 serde_json::json!({
@@ -339,11 +383,7 @@ pub async fn run_or_resume(
             )
         }
         RunOutcome::Failed { message } => {
-            log_terminal_write_err(
+            log_terminal_write_err("mark_failed", run_id, store.mark_failed(&message).await);
-                "mark_failed",
-                run_id,
-                store.mark_failed(&message).await,
-            );
             JobOutcome::err(format!("{message} (run_id={run_id})"))
         }
     }
@@ -461,11 +501,7 @@ mod tests {
         async fn status(&self) -> Result<RunStatus, DomainError> {
             Ok(self.state.lock().unwrap().status)
         }
-        async fn checkpoint(
+        async fn checkpoint(&self, cursor: Vec<u8>, delta_count: u64) -> Result<(), DomainError> {
-            &self,
-            cursor: Vec<u8>,
-            delta_count: u64,
-        ) -> Result<(), DomainError> {
             let mut s = self.state.lock().unwrap();
             s.cursor = Some(cursor);
             s.scanned_count += delta_count;
@@ -530,15 +566,15 @@ mod tests {
         /// assertions.
         fn last_status(&self) -> Option<RunStatus> {
             let stores = self.stores.lock().unwrap();
-            stores
+            stores.last().map(|s| s.state.lock().unwrap().status)
-                .last()
-                .map(|s| s.state.lock().unwrap().status)
         }
 
         /// Test-only read — last-created run's cursor.
         fn last_cursor(&self) -> Option<Vec<u8>> {
             let stores = self.stores.lock().unwrap();
-            stores.last().and_then(|s| s.state.lock().unwrap().cursor.clone())
+            stores
+                .last()
+                .and_then(|s| s.state.lock().unwrap().cursor.clone())
         }
     }
 
@@ -596,6 +632,68 @@ mod tests {
             }
             Ok(n)
         }
+
+        async fn list_runs(
+            &self,
+            job_name: &str,
+            limit: u32,
+        ) -> Result<Vec<RunSummary>, DomainError> {
+            let stores = self.stores.lock().unwrap();
+            let now = Utc::now();
+            let out: Vec<RunSummary> = stores
+                .iter()
+                .rev() // newest first — MemProvider stores in insertion order
+                .take(limit as usize)
+                .map(|s| {
+                    let state = s.state.lock().unwrap();
+                    RunSummary {
+                        id: s.run_id,
+                        job_name: job_name.to_string(),
+                        status: state.status,
+                        started_at: s.started_at,
+                        last_progress_at: now,
+                        completed_at: None,
+                        stats: serde_json::json!({ "scanned_count": state.scanned_count }),
+                        params: serde_json::json!({}),
+                        cursor_hex: state.cursor.as_ref().map(hex::encode),
+                        error_message: state.error_message.clone(),
+                    }
+                })
+                .collect();
+            Ok(out)
+        }
+
+        async fn get_run_by_id(&self, run_id: Uuid) -> Result<Option<RunSummary>, DomainError> {
+            let stores = self.stores.lock().unwrap();
+            let now = Utc::now();
+            Ok(stores.iter().find(|s| s.run_id == run_id).map(|s| {
+                let state = s.state.lock().unwrap();
+                RunSummary {
+                    id: s.run_id,
+                    job_name: "mem".to_string(),
+                    status: state.status,
+                    started_at: s.started_at,
+                    last_progress_at: now,
+                    completed_at: None,
+                    stats: serde_json::json!({ "scanned_count": state.scanned_count }),
+                    params: serde_json::json!({}),
+                    cursor_hex: state.cursor.as_ref().map(hex::encode),
+                    error_message: state.error_message.clone(),
+                }
+            }))
+        }
+
+        async fn request_cancel(&self, _job_name: &str) -> Result<Option<Uuid>, DomainError> {
+            let stores = self.stores.lock().unwrap();
+            if let Some(s) = stores.last() {
+                let mut state = s.state.lock().unwrap();
+                if state.status == RunStatus::Running {
+                    state.status = RunStatus::CancelRequested;
+                    return Ok(Some(s.run_id));
+                }
+            }
+            Ok(None)
+        }
     }
 
     // ─── Handlers ──────────────────────────────────────────────────────────

src/infrastructure/scheduler/registry.rs

@@ -120,9 +120,7 @@ impl JobRegistry {
                     cadence,
                 );
             }
-            Err(e) => panic!(
+            Err(e) => panic!("JobRegistry::register({name}) failed — DI wiring bug: {e}"),
-                "JobRegistry::register({name}) failed — DI wiring bug: {e}"
-            ),
         }
     }
 

src/interfaces/api/handlers/admin_handler.rs

@@ -147,8 +147,19 @@ pub fn admin_routes() -> Router<Arc<AppState>> {
         // audit-logged. See `docs/plan/job-registry.md` §Cross-cutting.
         // Retired the `/internal/trigger-sweep|gc|grant-cleanup` shims
         // that used to sit here (Stage 2 of the job-registry rollout).
+        //
+        // `/jobs` + `/jobs/{name}/trigger` cover every registered
+        // JobHandler (periodic + recoverable — recoverable ones slot
+        // in through `RecoverableAdapter`). The `/cancel` + `/runs`
+        // + `/runs/{id}` triplet is recoverable-only — hitting them
+        // on a stateless job silently gets an empty list / no-op
+        // cancel, since no rows in `jobs.recoverable_runs` match.
+        // See `docs/plan/job-registry.md` Part 2.
         .route("/jobs", get(list_jobs))
         .route("/jobs/{name}/trigger", post(trigger_job))
+        .route("/jobs/{name}/cancel", post(cancel_job))
+        .route("/jobs/{name}/runs", get(list_job_runs))
+        .route("/jobs/{name}/runs/{id}", get(get_job_run))
         // Drives — admin-wide view (distinct from `/api/drives` which
         // is filtered to the caller's role grants).
         .route("/drives", get(list_all_drives))
@@ -2146,3 +2157,148 @@ pub async fn trigger_job(
             .into_response(),
     }
 }
+
+/// `POST /api/admin/jobs/{name}/cancel` — cooperative cancel of the
+/// currently-running recoverable run for `{name}`.
+///
+/// Flips the row's `status` from `Running` → `CancelRequested`. The
+/// handler is responsible for polling `store.status()` between batches
+/// and returning `RunOutcome::Paused` at the next safe boundary; if it
+/// doesn't, the cancel is a no-op until the run completes naturally.
+///
+/// Returns 200 with the run_id when a Running row was flipped, 200 with
+/// `cancelled: false` when nothing was running (either no runs exist,
+/// or the latest is Paused / Completed / Failed / already CancelRequested).
+/// Never 404 on "no active run" — the job name is registered and the
+/// endpoint just reports the truth.
+#[utoipa::path(
+    post,
+    path = "/api/admin/jobs/{name}/cancel",
+    params(("name" = String, Path, description = "Registered job name")),
+    responses(
+        (status = 200, description = "Cancel signalled (or no-op if nothing was running)"),
+        (status = 401, description = "Unauthorized"),
+        (status = 403, description = "Admin required"),
+        (status = 500, description = "DB error"),
+    ),
+    security(("bearerAuth" = [])),
+    tag = "admin"
+)]
+pub async fn cancel_job(
+    State(state): State<Arc<AppState>>,
+    axum::extract::Path(name): axum::extract::Path<String>,
+) -> impl IntoResponse {
+    use crate::infrastructure::scheduler::JobStoreProvider as _;
+    tracing::info!(
+        target: "audit",
+        event = "job.cancel_requested",
+        job = %name,
+        "👮🏻‍♂️ Admin requested cancel for job {}",
+        name,
+    );
+    match state.core.job_store_provider.request_cancel(&name).await {
+        Ok(Some(run_id)) => (
+            StatusCode::OK,
+            Json(serde_json::json!({
+                "cancelled": true,
+                "run_id": run_id.to_string(),
+                "status": "CancelRequested",
+            })),
+        )
+            .into_response(),
+        Ok(None) => (
+            StatusCode::OK,
+            Json(serde_json::json!({
+                "cancelled": false,
+                "reason": "no running run for this job",
+            })),
+        )
+            .into_response(),
+        Err(e) => AppError::internal_error(format!("cancel failed: {e}")).into_response(),
+    }
+}
+
+/// Query parameters for `GET /api/admin/jobs/{name}/runs`.
+#[derive(serde::Deserialize)]
+pub struct ListRunsQuery {
+    /// Cap on returned rows. Server-side clamps to 100 defensively.
+    #[serde(default = "default_runs_limit")]
+    pub limit: u32,
+}
+
+fn default_runs_limit() -> u32 {
+    20
+}
+
+/// `GET /api/admin/jobs/{name}/runs?limit=N` — history of recoverable
+/// runs for a registered job, newest first. Includes terminal +
+/// non-terminal rows.
+///
+/// Read-only, no audit line — standard admin-middleware auth is enough.
+#[utoipa::path(
+    get,
+    path = "/api/admin/jobs/{name}/runs",
+    params(
+        ("name" = String, Path, description = "Registered job name"),
+        ("limit" = Option<u32>, Query, description = "Max rows to return (default 20, capped at 100)"),
+    ),
+    responses(
+        (status = 200, description = "Runs listed (may be empty)"),
+        (status = 401, description = "Unauthorized"),
+        (status = 403, description = "Admin required"),
+        (status = 500, description = "DB error"),
+    ),
+    security(("bearerAuth" = [])),
+    tag = "admin"
+)]
+pub async fn list_job_runs(
+    State(state): State<Arc<AppState>>,
+    axum::extract::Path(name): axum::extract::Path<String>,
+    axum::extract::Query(query): axum::extract::Query<ListRunsQuery>,
+) -> impl IntoResponse {
+    use crate::infrastructure::scheduler::JobStoreProvider as _;
+    let limit = query.limit.clamp(1, 100);
+    match state.core.job_store_provider.list_runs(&name, limit).await {
+        Ok(runs) => (StatusCode::OK, Json(runs)).into_response(),
+        Err(e) => AppError::internal_error(format!("list_runs failed: {e}")).into_response(),
+    }
+}
+
+/// `GET /api/admin/jobs/{name}/runs/{id}` — single-run detail.
+///
+/// Returns 404 when the id doesn't exist. `{name}` is not validated
+/// against the run's `job_name` — the id is globally unique — but
+/// keeping the name in the URL path lets operators build stable
+/// per-job history links without knowing individual run ids upfront.
+#[utoipa::path(
+    get,
+    path = "/api/admin/jobs/{name}/runs/{id}",
+    params(
+        ("name" = String, Path, description = "Registered job name"),
+        ("id" = String, Path, description = "Run UUID"),
+    ),
+    responses(
+        (status = 200, description = "Run detail"),
+        (status = 401, description = "Unauthorized"),
+        (status = 403, description = "Admin required"),
+        (status = 404, description = "Run not found"),
+        (status = 500, description = "DB error"),
+    ),
+    security(("bearerAuth" = [])),
+    tag = "admin"
+)]
+pub async fn get_job_run(
+    State(state): State<Arc<AppState>>,
+    axum::extract::Path((_name, id)): axum::extract::Path<(String, uuid::Uuid)>,
+) -> impl IntoResponse {
+    use crate::infrastructure::scheduler::JobStoreProvider as _;
+    match state.core.job_store_provider.get_run_by_id(id).await {
+        Ok(Some(run)) => (StatusCode::OK, Json(run)).into_response(),
+        Ok(None) => (
+            StatusCode::NOT_FOUND,
+            Json(serde_json::json!({ "error": "run not found", "id": id.to_string() })),
+        )
+            .into_response(),
+        Err(e) => AppError::internal_error(format!("get_run failed: {e}")).into_response(),
+    }
+}

src/interfaces/api/mod.rs

@@ -234,6 +234,9 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload;
         // (docs/plan/job-registry.md §Cross-cutting).
         handlers::admin_handler::list_jobs,
         handlers::admin_handler::trigger_job,
+        handlers::admin_handler::cancel_job,
+        handlers::admin_handler::list_job_runs,
+        handlers::admin_handler::get_job_run,
         // Grant / ReBAC handlers (free functions)
         handlers::grant_handler::create_grant,
         handlers::grant_handler::revoke_grant,