Allow fixed calendar time for manual scheduler passes

crates/tinyflows-bus/src/schedule/mod.rs

@@ -35,8 +35,9 @@ pub const CALLBACK_INTERFACE: &str = "ai.tinyhumans.tinyflows.ScheduleHost";
 /// the job/run/policy records and the callback records into this crate
 /// (`types`, `Attempt`/`RunRecord`/`Completion`, `AgentJobSpec`,
 /// `StoreOperation`) and added the schedule-maths and validation members and
-/// the `set_max_active_jobs`/`import_job`/`import_run` store operations.
-pub const CONTRACT_VERSION: u32 = 3;
+/// the `set_max_active_jobs`/`import_job`/`import_run` store operations. 4 adds
+/// optional `config.now` for a fixed calendar instant on manual schedulers.
+pub const CONTRACT_VERSION: u32 = 4;
 /// Exact advertised scheduler member set.
 pub const METHODS: &[&str] = &[
     members::SCHEDULE_STORE,
@@ -115,6 +116,9 @@ pub struct StartScheduler {
     /// Context-bound I/O callbacks used for every scheduled run.
     pub callback: CallbackAddress,
     /// `poll_secs`, `max_concurrent`, `retries`, `backoff_ms`; omitted keys use defaults.
+    /// Manual schedulers additionally accept `now` (RFC3339), fixing calendar
+    /// time for due selection, run records and rescheduling without changing
+    /// the machine clock. Omit it to use system time. Automatic loops refuse it.
     pub config: serde_json::Value,
     #[serde(default)]
     /// True to drive ticks/drain through calls; false starts the module-owned loop.

crates/tinyflows-bus/src/schedule/mod_tests.rs

@@ -6,8 +6,8 @@ fn every() -> Schedule {
 }
 
 #[test]
-fn contract_is_version_three_with_the_exact_member_set() {
-    assert_eq!(CONTRACT_VERSION, 3);
+fn contract_is_version_four_with_the_exact_member_set() {
+    assert_eq!(CONTRACT_VERSION, 4);
     assert_eq!(METHODS.len(), 17);
     let mut unique: Vec<_> = METHODS.to_vec();
     unique.sort_unstable();

crates/tinyflows-module/Cargo.toml

@@ -36,4 +36,4 @@ static-link = []
 
 [dev-dependencies]
 tempfile = "3"
-tokio.workspace = true
+tokio = { workspace = true, features = ["test-util"] }

crates/tinyflows-module/README.md

@@ -15,6 +15,14 @@ limit when a running job defers its occurrence. Stores without this operation
 fail that poll rather than silently starve later due jobs. The runtime retains
 the original batch size and leaves deferred occurrences unchanged.
 
+A host-driven pass can set `StartScheduler.config.now` to an RFC3339 instant
+with `manual_ticks: true`. Due selection, run timestamps, and the next recurring
+occurrence all use that fixed instant, so a test can advance calendar time
+without waiting or changing the machine clock. Start a new manual scheduler
+for another instant. Retry sleeps still use Tokio timers. Omitting `now` keeps
+the system clock; a fixed instant is refused for an automatic poll loop.
+This option requires Schedule contract version 4.
+
 RecordRun callbacks receive job_created_at, captured from the dispatched job.
 Match that creation before appending history; an ID alone can refer to a
 replacement flow job. The module-owned backends perform this check. Completion

crates/tinyflows-module/src/schedule_clock.rs

@@ -0,0 +1,20 @@
+//! The fixed calendar instant of a host-driven scheduler pass.
+
+use chrono::{DateTime, Utc};
+use tinyflows_schedule::runtime::Clock;
+
+/// Keeps due selection, run records and recurring schedules on the same time.
+pub(super) struct FixedClock(pub(super) DateTime<Utc>);
+
+#[async_trait::async_trait]
+impl Clock for FixedClock {
+    fn now(&self) -> DateTime<Utc> {
+        self.0
+    }
+
+    async fn sleep(&self, duration: std::time::Duration) {
+        // Calendar time is fixed for the pass; retry backoff still yields to
+        // Tokio's timer, which a test can pause and advance independently.
+        tokio::time::sleep(duration).await;
+    }
+}

crates/tinyflows-module/src/schedule_clock_tests.rs

@@ -0,0 +1,123 @@
+//! Manual passes use one supplied calendar instant across dispatch and records.
+
+use super::*;
+use chrono::{Duration, Utc};
+use serde_json::json;
+
+#[tokio::test(start_paused = true)]
+async fn retry_sleep_uses_tokio_time_without_changing_the_calendar_instant() {
+    let now = Utc::now();
+    let clock = clock::FixedClock(now);
+    let began = tokio::time::Instant::now();
+    clock.sleep(std::time::Duration::from_secs(10)).await;
+    assert_eq!(began.elapsed(), std::time::Duration::from_secs(10));
+    assert_eq!(clock.now(), now);
+}
+
+#[tokio::test]
+async fn a_manual_pass_uses_the_supplied_time_for_due_jobs_and_run_records() {
+    let (_server, host, state) = tests::connections().await;
+    let proxy = host
+        .proxy(
+            schedule::BUS_NAME,
+            schedule::OBJECT_PATH,
+            schedule::INTERFACE,
+        )
+        .unwrap();
+    let now = Utc::now() + Duration::days(1);
+    let mut job = tests::job("shell");
+    job["schedule"] = json!({"kind":"at", "at": now});
+    job["next_run"] = json!(now);
+    state.jobs.lock().await.push(job);
+    let mut repeating = tests::job("shell");
+    repeating["id"] = json!("repeat");
+    repeating["next_run"] = json!(now);
+    state.jobs.lock().await.push(repeating);
+
+    // The exact same future job is not due before the supplied instant.
+    for (at, expected) in [(now - Duration::seconds(1), 0), (now, 2)] {
+        let request = json!({
+            "scope": "fixed-clock", "callback": tests::callback(),
+            "manual_ticks": true, "config": {"now": at},
+        });
+        let handle: String = proxy
+            .call(schedule::members::START_SCHEDULER, &(request,))
+            .await
+            .unwrap();
+        let count: usize = proxy
+            .call(schedule::members::TICK_SCHEDULER, &(&handle,))
+            .await
+            .unwrap();
+        assert_eq!(count, expected);
+        proxy
+            .call::<()>(schedule::members::DRAIN_SCHEDULER, &(&handle,))
+            .await
+            .unwrap();
+        proxy
+            .call::<()>(schedule::members::STOP_SCHEDULER, &(&handle,))
+            .await
+            .unwrap();
+    }
+    let times = state.observed_times.lock().await;
+    assert!(times.contains(&(now - Duration::seconds(1))));
+    assert!(times.contains(&now));
+    let records = state.records.lock().await;
+    assert_eq!(records.len(), 2);
+    for record in records.iter() {
+        assert_eq!(record["started_at"], json!(now));
+        assert_eq!(record["finished_at"], json!(now));
+    }
+    assert!(
+        state
+            .completions
+            .lock()
+            .await
+            .iter()
+            .any(|completion| { completion["next_run"] == json!(now + Duration::seconds(60)) })
+    );
+}
+
+#[tokio::test]
+async fn a_fixed_calendar_time_is_refused_for_an_automatic_loop() {
+    let (_server, host, _) = tests::connections().await;
+    let proxy = host
+        .proxy(
+            schedule::BUS_NAME,
+            schedule::OBJECT_PATH,
+            schedule::INTERFACE,
+        )
+        .unwrap();
+    let request = json!({
+        "scope": "fixed-loop", "callback": tests::callback(),
+        "config": {"now": Utc::now()},
+    });
+    let error = proxy
+        .call::<String>(schedule::members::START_SCHEDULER, &(request,))
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("manual ticks"), "{error}");
+}
+
+#[tokio::test]
+async fn an_invalid_calendar_time_is_refused() {
+    let (_server, host, _) = tests::connections().await;
+    let proxy = host
+        .proxy(
+            schedule::BUS_NAME,
+            schedule::OBJECT_PATH,
+            schedule::INTERFACE,
+        )
+        .unwrap();
+    for now in [json!("not a date"), json!(42), Value::Null] {
+        let request = json!({
+            "scope": "invalid-clock", "callback": tests::callback(),
+            "manual_ticks": true, "config": {"now": now},
+        });
+        assert!(
+            proxy
+                .call::<String>(schedule::members::START_SCHEDULER, &(request,))
+                .await
+                .is_err()
+        );
+    }
+}

crates/tinyflows-module/src/schedule_service.rs

@@ -1,4 +1,6 @@
 //! Separate Schedule object in the same loadable TinyFlows module.
+#[path = "schedule_clock.rs"]
+mod clock;
 use super::schedule_callbacks::HostCallbacks;
 use serde_json::Value;
 use std::{collections::HashMap, sync::Arc};
@@ -67,8 +69,20 @@ impl ScheduleService {
     }
 
     async fn start_scheduler(&self, request: StartScheduler) -> Result<String> {
+        let now: Option<chrono::DateTime<chrono::Utc>> = request
+            .config
+            .get("now")
+            .map(|now| serde_json::from_value(now.clone()))
+            .transpose()
+            .map_err(failure)?;
+        if now.is_some() && !request.manual_ticks {
+            return Err(Error::failed("a fixed calendar time requires manual ticks"));
+        }
         let config: SchedulerConfig = serde_json::from_value(request.config).map_err(failure)?;
         let mut ports = HostCallbacks::ports(self.connection.clone(), request.callback);
+        if let Some(now) = now {
+            ports.clock = Arc::new(clock::FixedClock(now));
+        }
         if let Some(storage) = request.storage {
             ports.store = Arc::new(
                 super::schedule_store::Store::new(self.connection.clone(), storage)
@@ -255,3 +269,7 @@ mod members_tests;
 #[cfg(test)]
 #[path = "schedule_service_tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "schedule_clock_tests.rs"]
+mod clock_tests;

crates/tinyflows-module/src/schedule_service_tests.rs

@@ -5,16 +5,46 @@ use tokio::sync::Mutex;
 
 #[derive(Default)]
 pub(super) struct State {
-    records: Mutex<Vec<Value>>,
-    completions: Mutex<Vec<Value>>,
+    pub(super) records: Mutex<Vec<Value>>,
+    pub(super) jobs: Mutex<Vec<Value>>,
+    pub(super) observed_times: Mutex<Vec<chrono::DateTime<chrono::Utc>>>,
+    pub(super) completions: Mutex<Vec<Value>>,
     executions: Mutex<Vec<String>>,
     triggers: Mutex<Vec<String>>,
 }
 struct Callbacks(Arc<State>);
 #[tinybus::interface(name = "ai.tinyhumans.tinyflows.ScheduleHost")]
 impl Callbacks {
-    async fn due_jobs(&self, _now: String) -> Result<Vec<Value>> {
-        Ok(vec![])
+    async fn due_jobs(&self, now: chrono::DateTime<chrono::Utc>) -> Result<Vec<Value>> {
+        self.0.observed_times.lock().await.push(now);
+        Ok(self
+            .0
+            .jobs
+            .lock()
+            .await
+            .iter()
+            .filter(|job| {
+                job["enabled"] == true
+                    && serde_json::from_value::<chrono::DateTime<chrono::Utc>>(
+                        job["next_run"].clone(),
+                    )
+                    .unwrap()
+                        <= now
+            })
+            .cloned()
+            .collect())
+    }
+    async fn due_jobs_excluding(
+        &self,
+        now: chrono::DateTime<chrono::Utc>,
+        excluded: Vec<String>,
+    ) -> Result<Vec<Value>> {
+        Ok(self
+            .due_jobs(now)
+            .await?
+            .into_iter()
+            .filter(|job| !excluded.iter().any(|id| job["id"] == *id))
+            .collect())
     }
     async fn claim_slot(&self, _job: Value, _next_run: String) -> Result<()> {
         Ok(())
@@ -63,13 +93,13 @@ pub(super) async fn connections() -> (Connection, Connection, Arc<State>) {
     host.request_name("test.ScheduleHost").await.unwrap();
     (server, host, state)
 }
-fn callback() -> CallbackAddress {
+pub(super) fn callback() -> CallbackAddress {
     CallbackAddress {
         destination: "test.ScheduleHost".into(),
         object_path: "/test/ScheduleHost".into(),
     }
 }
-fn job(kind: &str) -> Value {
+pub(super) fn job(kind: &str) -> Value {
     json!({"id":"one","expression":"","schedule":{"kind":"every","every_ms":60000},"command":"flow-a","prompt":null,"name":null,"job_type":kind,"session_target":"isolated","model":null,"agent_id":null,"enabled":true,"delivery":{"mode":"none"},"delete_after_run":false,"created_at":"2026-01-01T00:00:00Z","next_run":"2026-01-01T00:00:00Z","last_run":null,"last_status":null,"last_output":null})
 }
 

crates/tinyflows-module/tests/module_artifact_tests.rs

@@ -110,7 +110,7 @@ async fn compiled_module_serves_validation_without_linking_engine_into_client()
             destination: "test.ScheduleCallbacks".into(),
             object_path: "/test/ScheduleCallbacks".into(),
         },
-        config: json!({}),
+        config: json!({"now": "2027-01-01T00:00:00Z"}),
         manual_ticks: true,
         storage: None,
     };
@@ -133,6 +133,12 @@ async fn compiled_module_serves_validation_without_linking_engine_into_client()
         .unwrap();
     assert_eq!(*state.triggers.lock().unwrap(), ["artifact-flow"]);
     assert_eq!(state.records.lock().unwrap()[0]["status"], "ok");
+    for field in ["started_at", "finished_at"] {
+        assert_eq!(
+            state.records.lock().unwrap()[0][field],
+            "2027-01-01T00:00:00Z"
+        );
+    }
     assert_eq!(state.completions.lock().unwrap()[0]["disable"], true);
 
     assert_eq!(