From c39737d5d016f606bb754391310b8ba3f32e406e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 12 Oct 2026 16:45:50 +0300 Subject: [PATCH 1/4] test(schedule): cover malformed legacy policy rejection Add tests asserting that malformed legacy retry budgets and single-flight values cause scheduling to fail closed in both the in-memory and SQLite stores. The seeded job and the offending policy row are checked to remain untouched so an invalid policy stays available for repair. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/schedule/mod_tests.rs | 59 +++++++++++++++++++ .../src/schedule/policy_tests.rs | 49 +++++++++++++++ 2 files changed, 108 insertions(+) diff --git a/crates/tinyflows-drivers/src/schedule/mod_tests.rs b/crates/tinyflows-drivers/src/schedule/mod_tests.rs index 7cfb2a37..5478a3a5 100644 --- a/crates/tinyflows-drivers/src/schedule/mod_tests.rs +++ b/crates/tinyflows-drivers/src/schedule/mod_tests.rs @@ -5,6 +5,65 @@ use chrono::Utc; use serde_json::json; use tinystoragedrivers_core::{MemoryStorage, Precondition}; +async fn refuses_malformed_legacy_policy(policy: serde_json::Value) { + use tinystoragedrivers_core::CollectionSpec; + let store = store(); + let job = store + .add_shell_job(Some("system".into()), daily(), "system:seeded") + .await + .unwrap(); + let before = store.docs.get(JOBS, &job.id).await.unwrap().unwrap(); + store + .docs + .ensure_collection(&CollectionSpec::new("cron_job_policies")) + .await + .unwrap(); + store + .docs + .put( + "cron_job_policies", + &job.id, + policy.clone(), + Precondition::Absent, + ) + .await + .unwrap(); + + assert!( + store.get_job(&job.id).await.is_err(), + "malformed legacy policy must prevent scheduling: {policy}" + ); + assert_eq!( + store.docs.get(JOBS, &job.id).await.unwrap().unwrap().doc, + before.doc, + "refusing a policy must preserve the seeded job" + ); + assert_eq!( + store + .docs + .get("cron_job_policies", &job.id) + .await + .unwrap() + .unwrap() + .doc, + policy, + "an invalid policy must remain available for repair" + ); +} + +#[tokio::test] +async fn malformed_legacy_retry_budgets_fail_closed_without_mutating_seeded_jobs() { + for retries in [json!("2"), json!(-1), json!(u64::from(u32::MAX) + 1)] { + refuses_malformed_legacy_policy(json!({"retries": retries, "single_flight": true})).await; + } +} + +#[tokio::test] +async fn malformed_legacy_single_flight_fails_closed_without_deleting_the_policy() { + refuses_malformed_legacy_policy(json!({"retries": 0, "single_flight": "true"})).await; + refuses_malformed_legacy_policy(json!({"retries": 0})).await; +} + #[test] fn limits_default_like_the_sqlite_store_and_clamp_to_one() { let store = store(); diff --git a/crates/tinyflows-sqlite/src/schedule/policy_tests.rs b/crates/tinyflows-sqlite/src/schedule/policy_tests.rs index bde5c777..cf3e9fb2 100644 --- a/crates/tinyflows-sqlite/src/schedule/policy_tests.rs +++ b/crates/tinyflows-sqlite/src/schedule/policy_tests.rs @@ -1,6 +1,55 @@ use super::*; use rusqlite::Connection; +#[test] +fn malformed_legacy_retry_budgets_fail_closed_without_mutating_seeded_jobs() { + for retries in [-1_i64, i64::from(u32::MAX) + 1] { + let dir = tempfile::tempdir().unwrap(); + let opts = CronStoreOptions::new(dir.path().join("jobs.db")); + let job = add_agent_job_with_definition( + &opts, + Some("system".into()), + tinyflows_schedule::Schedule::Every { every_ms: 300_000 }, + "seeded prompt", + tinyflows_schedule::SessionTarget::Isolated, + None, + None, + false, + Some("welcome".into()), + true, + ) + .unwrap(); + let conn = Connection::open(&opts.db_path).unwrap(); + conn.execute_batch("CREATE TABLE cron_job_policies(job_id TEXT PRIMARY KEY, retries INTEGER, single_flight INTEGER NOT NULL DEFAULT 0);").unwrap(); + conn.execute( + "INSERT INTO cron_job_policies VALUES (?1, ?2, 1)", + rusqlite::params![&job.id, retries], + ) + .unwrap(); + + assert!( + get_job(&opts, &job.id).is_err(), + "invalid retry budget {retries} must prevent scheduling" + ); + let preserved: i64 = conn + .query_row( + "SELECT retries FROM cron_job_policies WHERE job_id=?1", + [&job.id], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(preserved, retries); + let unchanged: (Option, i64, String) = conn + .query_row( + "SELECT retries, single_flight, prompt FROM cron_jobs WHERE id=?1", + [&job.id], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .unwrap(); + assert_eq!(unchanged, (None, 0, "seeded prompt".into())); + } +} + #[test] fn legacy_policy_rows_migrate_without_changing_seeded_jobs() { let dir = tempfile::tempdir().unwrap(); From 84381aaa52f5be55ce0925df182670dcea01d5e1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 12 Oct 2026 16:46:30 +0300 Subject: [PATCH 2/4] fix(schedule): reject malformed legacy cron policies instead of coercing Legacy cron policy values were silently clamped or defaulted during migration, so out-of-range retries and non-boolean single_flight flags were accepted without warning. Both the SQLite migration and the document policy reader now validate these fields and fail with an error, leaving the policy table unchanged when the data is malformed. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/schedule/policy.rs | 16 ++++++++++------ crates/tinyflows-sqlite/src/schedule/schema.rs | 13 ++++++++++++- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/crates/tinyflows-drivers/src/schedule/policy.rs b/crates/tinyflows-drivers/src/schedule/policy.rs index 88fdbca8..d90d2ddd 100644 --- a/crates/tinyflows-drivers/src/schedule/policy.rs +++ b/crates/tinyflows-drivers/src/schedule/policy.rs @@ -23,16 +23,20 @@ impl CronDocuments { return Ok(None); } let mut doc = doc.clone(); - let retries = policy - .doc - .get("retries") - .and_then(Value::as_u64) - .map(|n| u32::try_from(n).unwrap_or(u32::MAX)); + let retries = match policy.doc.get("retries") { + None | Some(Value::Null) => None, + Some(value) => Some( + value + .as_u64() + .and_then(|n| u32::try_from(n).ok()) + .ok_or_else(|| anyhow!("Invalid legacy cron policy retries"))?, + ), + }; let single_flight = policy .doc .get("single_flight") .and_then(Value::as_bool) - .unwrap_or(false); + .ok_or_else(|| anyhow!("Invalid legacy cron policy single_flight"))?; doc["policy"] = serde_json::to_value(tinyflows_schedule::JobPolicy { retries, single_flight, diff --git a/crates/tinyflows-sqlite/src/schedule/schema.rs b/crates/tinyflows-sqlite/src/schedule/schema.rs index ccb3c927..8cf20de0 100644 --- a/crates/tinyflows-sqlite/src/schedule/schema.rs +++ b/crates/tinyflows-sqlite/src/schedule/schema.rs @@ -260,8 +260,19 @@ fn migrate_policies(conn: &Connection) -> Result<()> { let exists: bool = tx.query_row( "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='cron_job_policies')", [], |r| r.get(0))?; if exists { + let malformed: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM cron_job_policies + WHERE (retries IS NOT NULL AND + (typeof(retries) != 'integer' OR retries < 0 OR retries > 4294967295)) + OR typeof(single_flight) != 'integer' OR single_flight NOT IN (0, 1))", + [], + |row| row.get(0), + )?; + if malformed { + anyhow::bail!("Invalid legacy cron policy; migration left the policy table unchanged"); + } tx.execute_batch("UPDATE cron_jobs SET - retries = (SELECT CASE WHEN retries IS NULL THEN NULL ELSE MIN(4294967295, MAX(0, retries)) END FROM cron_job_policies p WHERE p.job_id=cron_jobs.id), + retries = (SELECT retries FROM cron_job_policies p WHERE p.job_id=cron_jobs.id), single_flight = (SELECT single_flight != 0 FROM cron_job_policies p WHERE p.job_id=cron_jobs.id) WHERE EXISTS(SELECT 1 FROM cron_job_policies p WHERE p.job_id=cron_jobs.id); DROP TABLE cron_job_policies;")?; From d5e43d5dfc74f1b0ecc93c3a77f36f06a8cf56c9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 24 Oct 2026 07:00:36 +0000 Subject: [PATCH 3/4] test(schedule): update import to tinystoragedrivers_api The test helper now imports CollectionSpec from tinystoragedrivers_api instead of the removed tinystoragedrivers_core crate, keeping the malformed legacy policy test compiling against the current dependency layout. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyflows-drivers/src/schedule/mod_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows-drivers/src/schedule/mod_tests.rs b/crates/tinyflows-drivers/src/schedule/mod_tests.rs index d50cfb7f..7b11cf52 100644 --- a/crates/tinyflows-drivers/src/schedule/mod_tests.rs +++ b/crates/tinyflows-drivers/src/schedule/mod_tests.rs @@ -7,7 +7,7 @@ use serde_json::json; use tinyflows_storage::Precondition; async fn refuses_malformed_legacy_policy(policy: serde_json::Value) { - use tinystoragedrivers_core::CollectionSpec; + use tinystoragedrivers_api::CollectionSpec; let store = store(); let job = store .add_shell_job(Some("system".into()), daily(), "system:seeded") From 2cc31cbc91256c485cdf411a14af339db5ed3cc0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 24 Oct 2026 07:00:50 +0000 Subject: [PATCH 4/4] test(schedule): update import to tinystoragedrivers_types The malformed legacy policy test now imports CollectionSpec from tinystoragedrivers_types instead of tinystoragedrivers_api, matching the crate that actually exports it. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- crates/tinyflows-drivers/src/schedule/mod_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows-drivers/src/schedule/mod_tests.rs b/crates/tinyflows-drivers/src/schedule/mod_tests.rs index 7b11cf52..a9b7734a 100644 --- a/crates/tinyflows-drivers/src/schedule/mod_tests.rs +++ b/crates/tinyflows-drivers/src/schedule/mod_tests.rs @@ -7,7 +7,7 @@ use serde_json::json; use tinyflows_storage::Precondition; async fn refuses_malformed_legacy_policy(policy: serde_json::Value) { - use tinystoragedrivers_api::CollectionSpec; + use tinystoragedrivers_types::CollectionSpec; let store = store(); let job = store .add_shell_job(Some("system".into()), daily(), "system:seeded")