Carry verified upload evidence through backup finalization

crates/libs/common/core/src/types/object_store.rs

@@ -605,5 +605,60 @@ impl FromStr for ExactStorageUri {
     }
 }
 
+/// Canonical stable snapshot metadata for an operator recovery approval.
+/// Generated transport timestamps/sizes and node-local send-source handles are
+/// deliberately excluded. Every other field, including future fields, is bound.
+#[cfg(feature = "serde")]
+pub fn snapshot_metadata_binding(bytes: &[u8]) -> Result<Vec<u8>, SnapshotMetadataBindingError> {
+    let mut value: serde_json::Value = serde_json::from_slice(bytes)?;
+    let map = value
+        .as_object_mut()
+        .ok_or(SnapshotMetadataBindingError::Shape)?;
+    if map.get("version").and_then(serde_json::Value::as_u64) != Some(2)
+        || [
+            "sandbox_id",
+            "local_artifact_id",
+            "artifact_kind",
+            "artifact_origin",
+            "execution",
+            "source",
+        ]
+        .iter()
+        .any(|key| map.get(*key).is_none_or(serde_json::Value::is_null))
+    {
+        return Err(SnapshotMetadataBindingError::Shape);
+    }
+    for key in ["created_at", "size_bytes", "rootfs_send_source_id"] {
+        map.remove(key);
+    }
+    fn sort(value: &mut serde_json::Value) {
+        match value {
+            serde_json::Value::Object(map) => {
+                for child in map.values_mut() {
+                    sort(child);
+                }
+                map.sort_keys();
+            }
+            serde_json::Value::Array(values) => {
+                for child in values {
+                    sort(child);
+                }
+            }
+            _ => {}
+        }
+    }
+    sort(&mut value);
+    Ok(serde_json::to_vec(&value)?)
+}
+
+#[cfg(feature = "serde")]
+#[derive(Debug, thiserror::Error)]
+pub enum SnapshotMetadataBindingError {
+    #[error("recovery requires version 2 artifact identity and execution metadata")]
+    Shape,
+    #[error("invalid snapshot metadata: {0}")]
+    Json(#[from] serde_json::Error),
+}
+
 #[cfg(test)]
 mod tests;

crates/libs/common/jobs/src/db.rs

@@ -503,11 +503,26 @@ pub fn complete_leased_job(
     job_id: &JobId,
     worker_id: &str,
     bytes: i64,
+) -> rusqlite::Result<bool> {
+    complete_leased_job_with_result(conn, job_id, worker_id, None, bytes, None)
+}
+
+pub fn complete_leased_job_with_result(
+    conn: &Connection,
+    job_id: &JobId,
+    worker_id: &str,
+    expected_attempt: Option<i32>,
+    bytes: i64,
+    payload: Option<&str>,
 ) -> rusqlite::Result<bool> {
     let rows = conn.execute(
         "UPDATE jobs
          SET status = 'completed',
              bytes_processed = ?3,
+             payload = CASE WHEN json_valid(payload) AND json_type(payload) = 'object' THEN
+                 CASE WHEN ?4 IS NULL THEN json_remove(payload, '$.completion_result')
+                 ELSE json_set(payload, '$.completion_result', json_object('version', 1, 'attempt', attempts, 'value', json(?4))) END
+                 ELSE payload END,
              error = NULL,
              error_kind = NULL,
              failure_reason = NULL,
@@ -517,8 +532,9 @@ pub fn complete_leased_job(
              heartbeat_at = NULL
          WHERE job_id = ?1
            AND lease_owner = ?2
-           AND status = 'running'",
-        params![job_id, worker_id, bytes],
+           AND status = 'running'
+           AND (?5 IS NULL OR attempts = ?5)",
+        params![job_id, worker_id, bytes, payload, expected_attempt],
     )?;
     Ok(rows > 0)
 }
@@ -625,14 +641,27 @@ pub fn complete_job_attempt(
     job_id: &JobId,
     expected_attempts: i32,
     bytes: i64,
+) -> rusqlite::Result<bool> {
+    complete_job_attempt_with_result(conn, job_id, expected_attempts, bytes, None)
+}
+
+pub fn complete_job_attempt_with_result(
+    conn: &Connection,
+    job_id: &JobId,
+    expected_attempts: i32,
+    bytes: i64,
+    payload: Option<&str>,
 ) -> rusqlite::Result<bool> {
     let changed = conn.execute(
         "UPDATE jobs
-         SET status = 'completed', bytes_processed = ?2,
+         SET status = 'completed', bytes_processed = ?2, payload = CASE WHEN json_valid(payload) AND json_type(payload) = 'object' THEN
+                 CASE WHEN ?4 IS NULL THEN json_remove(payload, '$.completion_result')
+                 ELSE json_set(payload, '$.completion_result', json_object('version', 1, 'attempt', attempts, 'value', json(?4))) END
+                 ELSE payload END,
              error = NULL, error_kind = NULL, failure_reason = NULL,
              completed_at = datetime('now')
          WHERE job_id = ?1 AND status = 'running' AND attempts = ?3",
-        params![job_id, bytes, expected_attempts],
+        params![job_id, bytes, expected_attempts, payload],
     )?;
     Ok(changed == 1)
 }

crates/libs/common/jobs/src/db/tests.rs

@@ -534,3 +534,73 @@ fn count_failed_jobs_by_dedup_key_counts_one_failure_kind_of_one_key() {
         0
     );
 }
+#[test]
+fn completion_evidence_is_atomic_and_fenced() {
+    for leased in [false, true] {
+        let conn = mem_conn();
+        let id = fixture_job_id("content-proof");
+        insert_job(
+            &conn,
+            InsertJob {
+                job_id: &id,
+                kind: "push",
+                payload: "{\"source\":\"owned\"}",
+                subject_id: None,
+                dedup_key: None,
+                max_attempts: 3,
+                traceparent: None,
+                priority: JobPriority::default(),
+            },
+        )
+        .unwrap();
+        conn.execute(
+            "UPDATE jobs SET status='running',attempts=2,lease_owner='current' WHERE job_id=?1",
+            [id],
+        )
+        .unwrap();
+        let complete = |owner, attempt, payload| {
+            if leased {
+                complete_leased_job_with_result(&conn, &id, owner, Some(attempt), 14, Some(payload))
+            } else {
+                complete_job_attempt_with_result(&conn, &id, attempt, 14, Some(payload))
+            }
+        };
+        assert!(!complete("previous", 1, "wrong-proof").unwrap());
+        assert!(!complete("current", 1, "stale-attempt-proof").unwrap());
+        let unchanged = get_job(&conn, &id).unwrap().unwrap();
+        assert_eq!(unchanged.status, "running");
+        assert_eq!(unchanged.payload, "{\"source\":\"owned\"}");
+        conn.execute(
+            "UPDATE jobs SET payload=json_set(payload,'$.updated_during_upload',1) WHERE job_id=?1",
+            [id],
+        )
+        .unwrap();
+        assert!(
+            complete(
+                "current",
+                2,
+                "{\"source\":\"owned\",\"proof\":\"verified\"}"
+            )
+            .unwrap()
+        );
+        let completed = get_job(&conn, &id).unwrap().unwrap();
+        assert_eq!(completed.status, "completed");
+        assert_eq!(completed.bytes_processed, 14);
+        assert!(completed.payload.contains("verified"));
+        let payload: serde_json::Value = serde_json::from_str(&completed.payload).unwrap();
+        assert_eq!(payload["updated_during_upload"], 1);
+        assert_eq!(payload["source"], "owned");
+        assert_eq!(
+            crate::completion_result::<serde_json::Value>(&completed).unwrap()["proof"],
+            "verified"
+        );
+        let mut old_retry = completed.clone();
+        old_retry.attempts += 1;
+        assert!(crate::completion_result::<serde_json::Value>(&old_retry).is_none());
+        assert!(!complete("current", 2, "replacement-proof").unwrap());
+        assert_eq!(
+            get_job(&conn, &id).unwrap().unwrap().payload,
+            completed.payload
+        );
+    }
+}

crates/libs/common/jobs/src/lib.rs

@@ -368,6 +368,31 @@ impl JobError {
     }
 }
 
+/// A successful execution and an optional service-owned result containing its
+/// completion evidence. The worker publishes both in the same fenced SQLite
+/// transition; partial or superseded executions never publish their evidence.
+#[derive(Debug)]
+pub struct JobCompletion {
+    pub bytes_processed: i64,
+    pub result: Option<serde_json::Value>,
+}
+
+/// Read optional completion evidence only when its schema and producing attempt
+/// match. Old workers, malformed results and rollback schemas remain readable.
+pub fn completion_result<T: serde::de::DeserializeOwned>(job: &Job) -> Option<T> {
+    if job.status != "completed" {
+        return None;
+    }
+    let payload: serde_json::Value = serde_json::from_str(&job.payload).ok()?;
+    let envelope = payload.get("completion_result")?;
+    if envelope.get("version")?.as_u64()? != 1
+        || envelope.get("attempt")?.as_i64()? != i64::from(job.attempts)
+    {
+        return None;
+    }
+    serde_json::from_value(envelope.get("value")?.clone()).ok()
+}
+
 /// Defines the execution logic for a job kind.
 ///
 /// Implementations must be `Clone` + `Send` + `Sync` + `'static` so they
@@ -381,6 +406,21 @@ pub trait JobExecutor: Clone + Send + Sync + 'static {
         cancellation: tokio_util::sync::CancellationToken,
     ) -> Result<i64, JobError>;
 
+    /// Executors may return evidence merged into the current intent at completion.
+    /// Existing executors preserve their intent payload unchanged.
+    async fn execute_with_completion(
+        &self,
+        job: &Job,
+        cancellation: tokio_util::sync::CancellationToken,
+    ) -> Result<JobCompletion, JobError> {
+        self.execute(job, cancellation)
+            .await
+            .map(|bytes_processed| JobCompletion {
+                bytes_processed,
+                result: None,
+            })
+    }
+
     /// Remove executor-owned partial state after cancellation. This runs only
     /// after the execute future has stopped, so cleanup cannot race a writer.
     async fn cleanup_cancelled(&self, _job: &Job) -> Result<(), JobError> {

crates/libs/common/jobs/src/worker.rs

@@ -1,4 +1,4 @@
 use pplx_space_common_core::types::{JobId, SubjectId};
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
@@ -1583,9 +1583,11 @@
                 self.finish_cancelled_job(job).await;
                 return;
             }
-            let result =
-                tokio::time::timeout(self.job_timeout, self.executor.execute(&job, cancellation))
-                    .await;
+            let result = tokio::time::timeout(
+                self.job_timeout,
+                self.executor.execute_with_completion(&job, cancellation),
+            )
+            .await;
             self.cancellation_tokens
                 .lock()
                 .expect("common jobs cancellation mutex poisoned")
@@ -1605,8 +1607,12 @@
             }
 
             match result {
-                Ok(Ok(bytes)) => {
-                    if self.complete_job(&job, lease_owner.as_deref(), bytes).await {
+                Ok(Ok(completion)) => {
+                    let bytes = completion.bytes_processed;
+                    if self
+                        .complete_job(&job, lease_owner.as_deref(), &completion)
+                        .await
+                    {
                         span.record("otel.status_code", "OK");
                         tracing::info!(
                             job_id = %job.job_id,
@@ -1724,14 +1730,33 @@
     /// remote work; abandoning this write strands nonleased running rows.
     /// A persistent database outage has no completion deadline. Bounded drain
     /// must report non-quiescence rather than dropping this ownership.
-    async fn complete_job(&self, job: &Job, lease_owner: Option<&str>, bytes: i64) -> bool {
+    async fn complete_job(
+        &self,
+        job: &Job,
+        lease_owner: Option<&str>,
+        completion: &crate::JobCompletion,
+    ) -> bool {
+        let result_json = completion.result.as_ref().map(serde_json::Value::to_string);
         let mut retry_delay = Duration::from_secs(1);
         loop {
             let result = {
                 let conn = self.lock_db_for_background();
                 match lease_owner {
-                    Some(owner) => db::complete_leased_job(&conn, &job.job_id, owner, bytes),
-                    None => db::complete_job_attempt(&conn, &job.job_id, job.attempts, bytes),
+                    Some(owner) => db::complete_leased_job_with_result(
+                        &conn,
+                        &job.job_id,
+                        owner,
+                        Some(job.attempts),
+                        completion.bytes_processed,
+                        result_json.as_deref(),
+                    ),
+                    None => db::complete_job_attempt_with_result(
+                        &conn,
+                        &job.job_id,
+                        job.attempts,
+                        completion.bytes_processed,
+                        result_json.as_deref(),
+                    ),
                 }
             };
             match result {