diff --git a/crates/tracedecay-automation-runtime/src/automation/jobs.rs b/crates/tracedecay-automation-runtime/src/automation/jobs.rs index b5d9cf53ae..f93d40e115 100644 --- a/crates/tracedecay-automation-runtime/src/automation/jobs.rs +++ b/crates/tracedecay-automation-runtime/src/automation/jobs.rs @@ -107,10 +107,11 @@ struct AutomationJobsFile { jobs: Vec, } -#[derive(Debug, Deserialize, Default)] +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] struct AutomationJobsRawFile { - #[serde(default)] - jobs: Vec, + schema_version: Option, + jobs: Option>, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -195,22 +196,54 @@ pub async fn load_jobs(dashboard_root: &Path) -> Result> { 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::(&bytes).map_err(|e| { + let file = serde_json::from_slice::(&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::(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::(entry) { + Ok(job) => job, + Err(_) => { + let issue = if delivery.is_some_and(|delivery| { + serde_json::from_value::(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"] diff --git a/crates/tracedecay-automation-runtime/src/automation/jobs/persistence_tests.rs b/crates/tracedecay-automation-runtime/src/automation/jobs/persistence_tests.rs new file mode 100644 index 0000000000..189584ba72 --- /dev/null +++ b/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 { + 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}) + ); +} diff --git a/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_config_tests.rs b/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_config_tests.rs index 9f48988244..bca8c2149f 100644 --- a/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_config_tests.rs +++ b/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] diff --git a/crates/tracedecay-dashboard-api/src/automation_jobs_api.rs b/crates/tracedecay-dashboard-api/src/automation_jobs_api.rs index c225d4c064..40f6718d9f 100644 --- a/crates/tracedecay-dashboard-api/src/automation_jobs_api.rs +++ b/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() { diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 4195884b7e..bad308cc63 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/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]