fix(automation): refuse incomplete stored job documents

crates/tracedecay-automation-runtime/src/automation/jobs.rs

@@ -107,10 +107,11 @@ struct AutomationJobsFile {
     jobs: Vec<AutomationJob>,
 }
 
-#[derive(Debug, Deserialize, Default)]
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
 struct AutomationJobsRawFile {
-    #[serde(default)]
-    jobs: Vec<Value>,
+    schema_version: Option<u32>,
+    jobs: Option<Vec<Value>>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -195,22 +196,54 @@ pub async fn load_jobs(dashboard_root: &Path) -> Result<Vec<AutomationJob>> {
     let read_path = path.clone();
     match crate::automation::blocking_fs(move || std::fs::read(read_path)).await {
         Ok(bytes) => {
-            let file = serde_json::from_slice::<AutomationJobsRawFile>(&bytes).map_err(|e| {
+            let file = serde_json::from_slice::<AutomationJobsRawFile>(&bytes).map_err(|_| {
                 TraceDecayError::Config {
-                    message: format!("failed to parse automation jobs '{}': {e}", path.display()),
+                    message: format!("invalid automation jobs envelope '{}'", path.display()),
                 }
             })?;
-            let mut loaded = Vec::with_capacity(file.jobs.len());
-            for (index, entry) in file.jobs.into_iter().enumerate() {
-                match serde_json::from_value::<AutomationJob>(entry) {
-                    Ok(job) => loaded.push(job),
-                    Err(error) => tracing::warn!(
-                        index,
-                        path = %path.display(),
-                        %error,
-                        "skipped corrupt automation job entry"
+            let schema_version = file.schema_version.ok_or_else(|| TraceDecayError::Config {
+                message: format!(
+                    "automation jobs envelope '{}' is missing schema_version",
+                    path.display()
+                ),
+            })?;
+            let jobs = file.jobs.ok_or_else(|| TraceDecayError::Config {
+                message: format!(
+                    "automation jobs envelope '{}' is missing jobs",
+                    path.display()
+                ),
+            })?;
+            if schema_version != JOBS_SCHEMA_VERSION {
+                return Err(TraceDecayError::Config {
+                    message: format!(
+                        "automation jobs envelope '{}' uses unsupported schema_version {}",
+                        path.display(),
+                        schema_version
                     ),
-                }
+                });
+            }
+            let mut loaded = Vec::with_capacity(jobs.len());
+            for (index, entry) in jobs.into_iter().enumerate() {
+                let delivery = entry.get("delivery").cloned();
+                let job = match serde_json::from_value::<AutomationJob>(entry) {
+                    Ok(job) => job,
+                    Err(_) => {
+                        let issue = if delivery.is_some_and(|delivery| {
+                            serde_json::from_value::<JobDelivery>(delivery).is_err()
+                        }) {
+                            "invalid delivery"
+                        } else {
+                            "invalid job"
+                        };
+                        return Err(TraceDecayError::Config {
+                            message: format!(
+                                "automation jobs envelope '{}' contains {issue} at index {index}",
+                                path.display()
+                            ),
+                        });
+                    }
+                };
+                loaded.push(job);
             }
             Ok(loaded)
         }
@@ -1056,6 +1089,10 @@ async fn run_pre_run_command(
     ))
 }
 
+#[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used)]
+#[path = "jobs/persistence_tests.rs"]
+mod persistence_tests;
 #[cfg(all(test, unix))]
 #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 #[path = "jobs/pre_run_command_tests.rs"]

crates/tracedecay-automation-runtime/src/automation/jobs/persistence_tests.rs

@@ -0,0 +1,147 @@
+use std::fs;
+
+use serde_json::{Value, json};
+use tracedecay_domain::errors::TraceDecayError;
+
+use super::{jobs_configured_for_scheduler, jobs_path, load_jobs, save_jobs};
+
+async fn write_document(root: &std::path::Path, value: Value) -> Vec<u8> {
+    let bytes = serde_json::to_vec(&value).expect("fixture document serializes");
+    fs::write(jobs_path(root), &bytes).expect("fixture document writes");
+    bytes
+}
+
+async fn assert_load_refused(root: &std::path::Path, document: Value, context: &str) {
+    let original = write_document(root, document).await;
+    let error = load_jobs(root)
+        .await
+        .expect_err("incomplete stored jobs must not become an editable list");
+    let TraceDecayError::Config { message } = error else {
+        panic!("stored jobs failure must remain a typed configuration error");
+    };
+    assert!(message.contains(context), "{message}");
+    assert!(!message.contains("private-job-payload"), "{message}");
+    assert_eq!(
+        fs::read(jobs_path(root)).expect("document remains readable"),
+        original
+    );
+}
+
+#[tokio::test]
+async fn mixed_rows_refuse_complete_load_and_scheduler_admission() {
+    let root = tempfile::tempdir().expect("temporary dashboard root");
+    let document = json!({
+        "schema_version": 1,
+        "jobs": [
+            {"id":"job-a", "name":"A", "prompt":"Do A"},
+            {"id":"job-b", "name":"B", "private":"private-job-payload"}
+        ]
+    });
+    let original = write_document(root.path(), document).await;
+
+    let error = load_jobs(root.path())
+        .await
+        .expect_err("one bad row must not be filtered from a rewrite source");
+    let TraceDecayError::Config { message } = error else {
+        panic!("mixed stored rows must fail as configuration");
+    };
+    assert!(message.contains("index 1"), "{message}");
+    assert!(!message.contains("private-job-payload"), "{message}");
+    assert_eq!(
+        fs::read(jobs_path(root.path())).expect("document remains readable"),
+        original
+    );
+    assert!(jobs_configured_for_scheduler(root.path()).await.is_err());
+}
+
+#[tokio::test]
+async fn unsupported_and_incomplete_envelopes_are_not_current_empty_files() {
+    let root = tempfile::tempdir().expect("temporary dashboard root");
+    for (document, context) in [
+        (json!({"schema_version": 9, "jobs": []}), "schema_version"),
+        (json!({"schema_version": 1}), "jobs"),
+        (json!({"jobs": []}), "schema_version"),
+        (
+            json!({"schema_version": "private-job-payload", "jobs": []}),
+            "envelope",
+        ),
+        (json!({"schema_version": 1, "jobs": {}}), "envelope"),
+        (
+            json!({
+                "schema_version": 1,
+                "jobs": [],
+                "private-job-payload": "private-job-payload"
+            }),
+            "envelope",
+        ),
+    ] {
+        assert_load_refused(root.path(), document, context).await;
+    }
+}
+
+#[tokio::test]
+async fn all_invalid_rows_are_not_reported_as_an_empty_configuration() {
+    let root = tempfile::tempdir().expect("temporary dashboard root");
+    assert_load_refused(
+        root.path(),
+        json!({
+            "schema_version": 1,
+            "jobs": [{"name":"missing id and prompt"}]
+        }),
+        "index 0",
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn malformed_delivery_refuses_the_whole_document() {
+    let root = tempfile::tempdir().expect("temporary dashboard root");
+    assert_load_refused(
+        root.path(),
+        json!({
+            "schema_version": 1,
+            "jobs": [{
+                "id":"job-a",
+                "name":"A",
+                "prompt":"Do A",
+                "delivery":{"mode":"private-job-payload", "url":"private-job-payload"}
+            }]
+        }),
+        "invalid delivery at index 0",
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn valid_job_extension_fields_survive_load_and_save() {
+    let root = tempfile::tempdir().expect("temporary dashboard root");
+    write_document(
+        root.path(),
+        json!({
+            "schema_version": 1,
+            "jobs": [{
+                "id":"job-a",
+                "name":"A",
+                "prompt":"Do A",
+                "future_field":{"mode":"preserve", "rank":3}
+            }]
+        }),
+    )
+    .await;
+
+    let jobs = load_jobs(root.path())
+        .await
+        .expect("valid current jobs load");
+    assert_eq!(jobs.len(), 1);
+    save_jobs(root.path(), &jobs)
+        .await
+        .expect("valid current jobs save");
+    let saved: Value = serde_json::from_slice(
+        &fs::read(jobs_path(root.path())).expect("saved document is readable"),
+    )
+    .expect("saved document remains JSON");
+    assert_eq!(
+        saved["jobs"][0]["future_field"],
+        json!({"mode":"preserve", "rank":3})
+    );
+}

crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_config_tests.rs

@@ -1,5 +1,6 @@
 use serde_json::json;
 use tracedecay_contracts::retained_surfaces::AutomationSkipReasonV1;
+use tracedecay_domain::errors::TraceDecayError;
 
 use super::{
     AutomationJob, JOBS_SCHEMA_VERSION, JobDelivery, job_schedule_decision,
@@ -16,14 +17,20 @@ use crate::automation::run_ledger::{
 async fn corrupt_jobs_file_surfaces_error_instead_of_no_work() {
     let temp = tempfile::TempDir::new().unwrap();
     let root = temp.path();
-    std::fs::write(jobs_path(root), b"{ this is not valid json").unwrap();
+    let path = jobs_path(root);
+    let original = b"{ this is not valid json";
+    std::fs::write(&path, original).unwrap();
     let err = jobs_configured_for_scheduler(root)
         .await
         .expect_err("corrupt jobs file must surface an error");
-    assert!(
-        err.to_string().contains("failed to parse automation jobs"),
-        "error should carry the parse cause: {err}"
+    let TraceDecayError::Config { message } = err else {
+        panic!("corrupt jobs must fail as a typed configuration error");
+    };
+    assert_eq!(
+        message,
+        format!("invalid automation jobs envelope '{}'", path.display())
     );
+    assert_eq!(std::fs::read(path).unwrap(), original);
 }
 
 #[tokio::test]

crates/tracedecay-dashboard-api/src/automation_jobs_api.rs

@@ -363,12 +363,70 @@ fn not_found(job_id: &str) -> JsonError {
 #[cfg(test)]
 #[allow(clippy::expect_used)]
 mod tests {
+    use std::fs;
     use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::mpsc;
     use std::time::Duration;
 
     use super::*;
     use crate::{DashboardAutomationAuthorityErrorV1, DashboardAutomationAuthorityV1};
+    use tracedecay_automation_runtime::automation::jobs::jobs_path;
+
+    #[tokio::test]
+    async fn malformed_sibling_refuses_every_crud_write_without_reconcile() {
+        for operation in ["create", "update", "delete"] {
+            let (_project, mut state) =
+                crate::events_api::dashboard_state_fixture("project.dashboard-jobs-lossless-crud")
+                    .await;
+            let reconciles = Arc::new(AtomicUsize::new(0));
+            let reconcile_count = Arc::clone(&reconciles);
+            state.automation_scheduler_reconciler = Some(Arc::new(move || {
+                reconcile_count.fetch_add(1, Ordering::SeqCst);
+                Box::pin(async {
+                    tracedecay_contracts::retrieval::AutomationSchedulerReconcileOutcome::Started
+                })
+            }));
+            let raw = br#"{"schema_version":1,"jobs":[{"id":"job-a","name":"A","prompt":"Do A"},{"id":"job-b","name":"B","private":"private-job-payload"}]}"#;
+            let path = jobs_path(&state.dashboard_root);
+            fs::write(&path, raw).expect("malformed fixture document writes");
+
+            let result = match operation {
+                "create" => create(
+                    State(state.clone()),
+                    Json(json!({"id":"job-c", "name":"C", "prompt":"Do C"})),
+                )
+                .await
+                .map(|_| ()),
+                "update" => update(
+                    State(state.clone()),
+                    AxumPath("job-a".to_owned()),
+                    Json(json!({"name":"Renamed"})),
+                )
+                .await
+                .map(|_| ()),
+                "delete" => delete(State(state), AxumPath("job-a".to_owned()))
+                    .await
+                    .map(|_| ()),
+                _ => unreachable!(),
+            };
+
+            let (status, body) = result.expect_err("CRUD must refuse a partial stored list");
+            assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{operation}");
+            assert!(
+                !body["detail"]
+                    .as_str()
+                    .unwrap_or_default()
+                    .contains("private-job-payload"),
+                "{operation} leaked stored data: {body:?}"
+            );
+            assert_eq!(
+                fs::read(&path).expect("stored document remains readable"),
+                raw
+            );
+            assert_eq!(reconciles.load(Ordering::SeqCst), 0, "{operation}");
+        }
+    }
 
     #[test]
     fn accepted_user_job_runs_without_a_tokio_runtime() {

crates/tracedecay/tests/automation_runner_test/jobs.rs

@@ -21,6 +21,7 @@ use tracedecay_automation_runtime::automation::scheduler::{
     AutomationSchedule, cron_is_due, parse_schedule,
 };
 use tracedecay_contracts::retained_surfaces::AutomationSkipReasonV1;
+use tracedecay_domain::errors::TraceDecayError;
 
 fn sample_job(id: &str) -> AutomationJob {
     AutomationJob {
@@ -127,7 +128,7 @@ async fn job_persistence_round_trips() {
 }
 
 #[tokio::test]
-async fn load_jobs_skips_corrupt_entries_without_losing_valid_jobs() {
+async fn load_jobs_refuses_corrupt_entries_without_rewriting_the_document() {
     let temp = tempdir().unwrap();
     let root = temp.path();
     let valid_one = sample_job("valid-one");
@@ -145,15 +146,15 @@ async fn load_jobs_skips_corrupt_entries_without_losing_valid_jobs() {
             valid_two
         ]
     });
-    fs::write(
-        root.join("automation_jobs.json"),
-        serde_json::to_vec_pretty(&document).unwrap(),
-    )
-    .unwrap();
+    let bytes = serde_json::to_vec_pretty(&document).unwrap();
+    let path = root.join("automation_jobs.json");
+    fs::write(&path, &bytes).unwrap();
 
-    let loaded = load_jobs(root).await.unwrap();
-    let ids: Vec<&str> = loaded.iter().map(|job| job.id.as_str()).collect();
-    assert_eq!(ids, vec!["valid-one", "valid-two"]);
+    let error = load_jobs(root).await.unwrap_err();
+    assert!(
+        matches!(error, TraceDecayError::Config { message } if message.contains("invalid job at index 1"))
+    );
+    assert_eq!(fs::read(path).unwrap(), bytes);
 }
 
 #[tokio::test]