From 4253432be04e9854d1fcd6d8f387aaf260c0b9fb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:02:14 +0300 Subject: [PATCH 01/19] chore(schedule): move schedule types into tinyflows-schedule Relocate the schedule type definitions out of tinyflows-bus and into the dedicated tinyflows-schedule crate, updating the bus module to re-export them so existing callers keep working. This separates the shared data model from the bus implementation. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/{schedule.rs => schedule/mod.rs} | 0 .../src => tinyflows-bus/src/schedule}/types.rs | 0 .../src => tinyflows-bus/src/schedule}/types_tests.rs | 0 3 files changed, 0 insertions(+), 0 deletions(-) rename crates/tinyflows-bus/src/{schedule.rs => schedule/mod.rs} (100%) rename crates/{tinyflows-schedule/src => tinyflows-bus/src/schedule}/types.rs (100%) rename crates/{tinyflows-schedule/src => tinyflows-bus/src/schedule}/types_tests.rs (100%) diff --git a/crates/tinyflows-bus/src/schedule.rs b/crates/tinyflows-bus/src/schedule/mod.rs similarity index 100% rename from crates/tinyflows-bus/src/schedule.rs rename to crates/tinyflows-bus/src/schedule/mod.rs diff --git a/crates/tinyflows-schedule/src/types.rs b/crates/tinyflows-bus/src/schedule/types.rs similarity index 100% rename from crates/tinyflows-schedule/src/types.rs rename to crates/tinyflows-bus/src/schedule/types.rs diff --git a/crates/tinyflows-schedule/src/types_tests.rs b/crates/tinyflows-bus/src/schedule/types_tests.rs similarity index 100% rename from crates/tinyflows-schedule/src/types_tests.rs rename to crates/tinyflows-bus/src/schedule/types_tests.rs From 17c3f076adfa8492e6f1832fd9754050d2111f76 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:02:20 +0300 Subject: [PATCH 02/19] feat(bus): add schedule types for bus scheduling Introduce the schedule types used by the bus to describe when and how scheduled work should run. Auto-committed-on: dragonfly --- crates/tinyflows-bus/Cargo.toml | 1 + crates/tinyflows-bus/src/schedule/types.rs | 12 +----------- 2 files changed, 2 insertions(+), 11 deletions(-) diff --git a/crates/tinyflows-bus/Cargo.toml b/crates/tinyflows-bus/Cargo.toml index 088bd6c4..bba7adb5 100644 --- a/crates/tinyflows-bus/Cargo.toml +++ b/crates/tinyflows-bus/Cargo.toml @@ -9,5 +9,6 @@ publish = false description = "Transport-free TinyBus contract for workflow authoring" [dependencies] +chrono.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/tinyflows-bus/src/schedule/types.rs b/crates/tinyflows-bus/src/schedule/types.rs index debdea59..251ffc81 100644 --- a/crates/tinyflows-bus/src/schedule/types.rs +++ b/crates/tinyflows-bus/src/schedule/types.rs @@ -135,7 +135,7 @@ pub enum Schedule { /// A standard 5-field crontab (`min hour dom mon dow`, weekdays 0–7 /// with 0 and 7 = Sunday) or a crate-native 6/7-field expression /// with a leading seconds column (weekdays 1–7, 1 = Sunday). See - /// [`crate::schedule::normalize_expression`]. + /// `normalize_expression` (served by the module's `NormalizeExpression` member). expr: String, #[serde(default)] tz: Option, @@ -172,11 +172,6 @@ impl<'de> Deserialize<'de> for Schedule { /// This handles callers that send `schedule: "0 9 * * 1"` directly /// instead of the structured form. fn visit_str(self, value: &str) -> Result { - tracing::debug!( - "[cron] Schedule::deserialize: got bare string '{}', \ - coercing to Cron variant", - value - ); Ok(Schedule::Cron { expr: value.to_owned(), tz: None, @@ -185,11 +180,6 @@ impl<'de> Deserialize<'de> for Schedule { } fn visit_string(self, value: String) -> Result { - tracing::debug!( - "[cron] Schedule::deserialize: got bare string '{}', \ - coercing to Cron variant", - value - ); Ok(Schedule::Cron { expr: value, tz: None, From ecd28cb0fc93b1abb00d298a59bdffd87fa1c921 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:02:57 +0300 Subject: [PATCH 03/19] feat(schedule): add frequency, records, requests, and store types Introduce the schedule module's supporting types for frequency configuration, record storage, request handling, and store definitions. These provide the data structures the scheduler needs to persist and query scheduled work. Auto-committed-on: dragonfly --- .../tinyflows-bus/src/schedule/frequency.rs | 82 +++++++ crates/tinyflows-bus/src/schedule/records.rs | 58 +++++ crates/tinyflows-bus/src/schedule/requests.rs | 66 ++++++ .../tinyflows-bus/src/schedule/store_types.rs | 213 ++++++++++++++++++ 4 files changed, 419 insertions(+) create mode 100644 crates/tinyflows-bus/src/schedule/frequency.rs create mode 100644 crates/tinyflows-bus/src/schedule/records.rs create mode 100644 crates/tinyflows-bus/src/schedule/requests.rs create mode 100644 crates/tinyflows-bus/src/schedule/store_types.rs diff --git a/crates/tinyflows-bus/src/schedule/frequency.rs b/crates/tinyflows-bus/src/schedule/frequency.rs new file mode 100644 index 00000000..1c88e70b --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/frequency.rs @@ -0,0 +1,82 @@ +//! Cadence-floor vocabulary shared by schedule validation and its hosts. +use chrono::{DateTime, Duration as ChronoDuration, Utc}; +use serde::{Deserialize, Serialize}; +use std::fmt; + +/// The shortest interval a host should allow between runs of an agent job. +/// Enforced by the module's `ValidateSchedule` (agent mode) member. +pub const MIN_AGENT_JOB_INTERVAL: ChronoDuration = ChronoDuration::minutes(5); + +/// Why a schedule runs more often than a threshold allows. Carries the +/// evidence, not only the verdict, so a log line or an error can quote it. +/// +/// Wire shape: `{"fixed_interval": }` or +/// `{"consecutive_runs": {"first": RFC3339, "second": RFC3339}}`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TooFrequent { + /// A fixed `every_ms` interval shorter than the threshold. + FixedInterval(#[serde(with = "millis")] ChronoDuration), + /// Two consecutive cron occurrences closer together than the threshold. + ConsecutiveRuns { + /// First of the pair. + first: DateTime, + /// Second of the pair. + second: DateTime, + }, +} + +mod millis { + use chrono::Duration; + use serde::{Deserialize, Deserializer, Serializer}; + + pub fn serialize(gap: &Duration, s: S) -> Result { + s.serialize_i64(gap.num_milliseconds()) + } + pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result { + Duration::try_milliseconds(i64::deserialize(d)?) + .ok_or_else(|| serde::de::Error::custom("gap out of range")) + } +} + +impl TooFrequent { + /// The offending gap. + pub fn gap(&self) -> ChronoDuration { + match self { + Self::FixedInterval(gap) => *gap, + Self::ConsecutiveRuns { first, second } => *second - *first, + } + } +} + +impl fmt::Display for TooFrequent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::FixedInterval(gap) => write!(f, "fires every {}", describe_gap(*gap)), + Self::ConsecutiveRuns { first, second } => write!( + f, + "fires at {} and again at {}, {} apart", + first.format("%Y-%m-%d %H:%M:%S UTC"), + second.format("%Y-%m-%d %H:%M:%S UTC"), + describe_gap(*second - *first) + ), + } + } +} + +/// `4 minutes`, `30 seconds`, `1 minute 30 seconds` — whole seconds only. +fn describe_gap(gap: ChronoDuration) -> String { + fn count(n: i64, unit: &str) -> String { + if n == 1 { + format!("{n} {unit}") + } else { + format!("{n} {unit}s") + } + } + let seconds = gap.num_seconds(); + match (seconds / 60, seconds % 60) { + (0, s) => count(s, "second"), + (m, 0) => count(m, "minute"), + (m, s) => format!("{} {}", count(m, "minute"), count(s, "second")), + } +} diff --git a/crates/tinyflows-bus/src/schedule/records.rs b/crates/tinyflows-bus/src/schedule/records.rs new file mode 100644 index 00000000..f13553c6 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/records.rs @@ -0,0 +1,58 @@ +//! Plain-data records the host callbacks carry. A host implementing the +//! `ScheduleHost` callbacks decodes and encodes these instead of hand-rolled +//! JSON; the JSON shapes are the wire contract. +use super::types::DeliveryStatus; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// Host-classified result of one attempt. Permanent failures halt immediately. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Attempt { + /// Whether execution (or final execution plus delivery) succeeded. + pub success: bool, + /// Result text passed unchanged to delivery and run-history persistence. + pub output: String, + /// False for host-classified permanent failures; stops retry immediately. + pub retryable: bool, +} + +/// Persisted outcome after execution and delivery. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunRecord { + /// Stable invocation identity shared by all attempts and delivery. + pub run_id: String, + /// Identity of the job that produced this run. + pub job_id: String, + /// Dispatch-time creation identity. Stores must reject a replacement with + /// the same job ID instead of attaching this outcome to its history. + pub job_created_at: DateTime, + /// Execution start, before the first attempt. + pub started_at: DateTime, + /// Execution finish before delivery; skipped runs use the same start/end. + pub finished_at: DateTime, + /// `ok`, `error`, or `skipped`. + pub status: String, + /// Result text passed unchanged to delivery and run-history persistence. + pub output: String, + /// Delivery outcome, absent on skipped slots. + pub delivery_status: Option, +} + +/// Final mutation computed by the runtime. Stores apply it mechanically. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Completion { + /// Remove a successful auto-delete one-shot, including its stored history, + /// unless a concurrent edit re-enabled or otherwise retained it. + pub remove: bool, + /// Retain a one-shot already disabled at claim time. A store must preserve + /// an explicit re-enable made after that claim. + pub disable: bool, + /// Recurrence computed after delivery; absent for one-shot jobs. + pub next_run: Option>, + /// Execution finish before delivery; skipped runs use the same start/end. + pub finished_at: DateTime, + /// Whether execution (or final execution plus delivery) succeeded. + pub success: bool, + /// Result text passed unchanged to delivery and run-history persistence. + pub output: String, +} diff --git a/crates/tinyflows-bus/src/schedule/requests.rs b/crates/tinyflows-bus/src/schedule/requests.rs new file mode 100644 index 00000000..4f8d1e67 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/requests.rs @@ -0,0 +1,66 @@ +//! Request types for the pure schedule-maths and validation members. The +//! algorithms (cron parsing, time zones, active hours, cadence floor) execute +//! in the module; hosts only send these and read the replies. +use super::types::{CronJob, CronJobPatch, DeliveryConfig, JobOrigin, Schedule, SessionTarget}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// `NextRun`: the first occurrence of `schedule` strictly after `from`. +/// Reply: `DateTime` (RFC3339). Errors on an unparseable schedule. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NextRunRequest { + /// Schedule to evaluate. + pub schedule: Schedule, + /// Instant to search from. + pub from: DateTime, +} + +/// `ValidateSchedule`: reject a schedule that can never fire, or (when +/// `agent` is set) one that runs more often than `MIN_AGENT_JOB_INTERVAL`. +/// Reply: unit; the error message is the author-facing explanation. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ValidateScheduleRequest { + /// Schedule to validate. + pub schedule: Schedule, + /// Reference instant (one-shots must lie after it). + pub now: DateTime, + /// Apply the agent-job cadence floor as well. + #[serde(default)] + pub agent: bool, +} + +/// `RunsCloserThan`: the closest pair of consecutive runs after `from` that +/// is under `min_gap_ms` apart. Reply: `Option`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunsCloserThanRequest { + /// Schedule to inspect. + pub schedule: Schedule, + /// Instant the walk starts from. + pub from: DateTime, + /// Threshold in milliseconds. + pub min_gap_ms: i64, +} + +/// `CheckOriginRequirements`: refuse origin delivery or a `current` session +/// target when the job has no origin conversation. Reply: unit; the error +/// message is the author-facing explanation. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CheckOriginRequest { + /// Requested session target. + pub session_target: SessionTarget, + /// Requested delivery. + pub delivery: DeliveryConfig, + /// The creating conversation, when there is one. + #[serde(default)] + pub origin: Option, +} + +/// `CheckPatch`: refuse a patch no store may apply to `job` (re-targeting a +/// flow-schedule job). Reply: unit. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CheckPatchRequest { + /// The stored job. + pub job: CronJob, + /// The proposed patch. + pub patch: CronJobPatch, +} diff --git a/crates/tinyflows-bus/src/schedule/store_types.rs b/crates/tinyflows-bus/src/schedule/store_types.rs new file mode 100644 index 00000000..7f46508d --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/store_types.rs @@ -0,0 +1,213 @@ +//! Store-wire vocabulary: the agent-job spec, the typed form of a store +//! operation, and the output bound every store applies. +use super::types::{ + CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, Schedule, + SessionTarget, +}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// Largest job output, in bytes, a store keeps in `last_output` or a run; +/// longer output is cut at a char boundary and ends with +/// [`TRUNCATED_OUTPUT_MARKER`]. The cut itself runs in the module +/// (`TruncateCronOutput`, and inside every store write). +pub const MAX_CRON_OUTPUT_BYTES: usize = 16 * 1024; +/// Suffix appended to output cut to [`MAX_CRON_OUTPUT_BYTES`]. +pub const TRUNCATED_OUTPUT_MARKER: &str = "\n...[truncated]"; + +/// Everything an agent job is created with. [`AgentJobSpec::new`] gives an +/// enabled, isolated job with no name, model, definition or origin. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentJobSpec { + /// Display name. + pub name: Option, + /// When the job fires. + pub schedule: Schedule, + /// Prompt the agent runs. + pub prompt: String, + /// Which session the run uses. + pub session_target: SessionTarget, + /// Model override. + pub model: Option, + /// `None` stores [`DeliveryConfig::default`]. + pub delivery: Option, + /// Remove the job after a successful run. + pub delete_after_run: bool, + /// Built-in agent definition to run with. + pub agent_id: Option, + /// Inserted in this state in one write, so an opt-in job is never + /// briefly enabled. + pub enabled: bool, + /// The conversation the job was created from. + pub origin: Option, +} + +impl AgentJobSpec { + /// An enabled, isolated job with no name, model, definition or origin. + pub fn new(schedule: Schedule, prompt: impl Into) -> Self { + Self { + name: None, + schedule, + prompt: prompt.into(), + session_target: SessionTarget::default(), + model: None, + delivery: None, + delete_after_run: false, + agent_id: None, + enabled: true, + origin: None, + } + } +} + +/// What `import_job` did with a job. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ImportedJob { + /// The id the job has in this store. It differs from the imported job's own + /// id for a flow-schedule job; its runs are imported under this id. + pub id: String, + /// Whether the call wrote the job. `false` when the store already held a + /// job under `id`, which is left exactly as it was. + pub written: bool, +} + +/// Typed form of `StoreRequest::operation`. It serializes to exactly the +/// `{"op": ..., ...}` JSON that `STORE_OPERATIONS` documents, so a host may +/// build the untyped value from this or keep sending hand-built JSON. +/// +/// Reply (the `Value` a `ScheduleStore` call returns), per operation: +/// `add`/`add_shell`/`add_agent`/`add_flow`/`get`/`update` a [`CronJob`]; +/// `find_flow` an `Option`; `list`/`due` a `Vec`; +/// `remove`/`set_max_active_jobs` `null`; `clear`/`dedup`/`delete_queued_runs` +/// a count; `list_runs` a `Vec`; `import_job` an [`ImportedJob`]; +/// `import_run` a bool (written); the rest follow the document store's +/// `record_*` results. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "op", rename_all = "snake_case")] +pub enum StoreOperation { + /// A shell job from a bare cron expression. + Add { + /// Cron expression. + expression: String, + /// Shell command. + command: String, + }, + /// A shell job. + AddShell { + /// Display name. + name: Option, + /// When the job fires. + schedule: Schedule, + /// Shell command. + command: String, + }, + /// An agent job from a full spec. + AddAgent { + /// The job to create. + spec: AgentJobSpec, + }, + /// The schedule-trigger job bound to a flow. + AddFlow { + /// Bound flow. + flow_id: String, + /// When it fires. + schedule: Schedule, + }, + /// Look up a flow's schedule-trigger job. + FindFlow { + /// Bound flow. + flow_id: String, + }, + /// Every job. + List, + /// One job by id. + Get { + /// Job id. + job_id: String, + }, + /// Remove a job and its history. + Remove { + /// Job id. + job_id: String, + }, + /// Remove every job. + Clear, + /// Collapse same-named jobs. + Dedup, + /// Enabled jobs due at `now`. + Due { + /// Selection instant. + now: DateTime, + }, + /// Apply a patch. + Update { + /// Job id. + job_id: String, + /// Fields to change. + patch: CronJobPatch, + }, + /// Record last-run bookkeeping. + RecordLastRun { + /// Job id. + job_id: String, + /// Finish instant. + finished_at: DateTime, + /// Outcome. + success: bool, + /// Result text. + output: String, + }, + /// Reschedule after a run. + Reschedule { + /// The job as it ran. + job: CronJob, + /// Outcome. + success: bool, + /// Result text. + output: String, + }, + /// Append a run to a job's history. + RecordRun { + /// Job id. + job_id: String, + /// Start instant. + started_at: DateTime, + /// Finish instant. + finished_at: DateTime, + /// `ok`, `error` or `skipped`. + status: String, + /// Result text. + output: Option, + /// Run duration in milliseconds. + duration_ms: i64, + /// Delivery outcome. + delivery_status: Option, + }, + /// Drop queued (not yet started) runs of a job. + DeleteQueuedRuns { + /// Job id. + job_id: String, + }, + /// A job's run history, newest first. + ListRuns { + /// Job id. + job_id: String, + /// Maximum rows. + limit: usize, + }, + /// Install or clear the scope's enabled-job cap (`null` clears it). + SetMaxActiveJobs { + /// New cap; zero is refused. + limit: Option, + }, + /// Import a job from a retired store, keeping its history fields. + ImportJob { + /// The job to import. + job: CronJob, + }, + /// Import one run of an already imported job. + ImportRun { + /// The run to import. + run: CronRun, + }, +} From 331cdc796a103c353e6ae73c6f48426c2e83f34e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:03:17 +0300 Subject: [PATCH 04/19] feat(schedule): add wire format tests for schedule messages Add round-trip serialization tests covering the schedule message wire format so encoding and decoding stay in sync as the protocol evolves. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/schedule/mod.rs | 102 +++++++++++-- .../tinyflows-bus/src/schedule/wire_tests.rs | 139 ++++++++++++++++++ 2 files changed, 230 insertions(+), 11 deletions(-) create mode 100644 crates/tinyflows-bus/src/schedule/wire_tests.rs diff --git a/crates/tinyflows-bus/src/schedule/mod.rs b/crates/tinyflows-bus/src/schedule/mod.rs index df2df955..eaa2371b 100644 --- a/crates/tinyflows-bus/src/schedule/mod.rs +++ b/crates/tinyflows-bus/src/schedule/mod.rs @@ -2,6 +2,26 @@ //! shape without making hosts link scheduler or storage implementation crates. use serde::{Deserialize, Serialize}; +mod frequency; +mod records; +mod requests; +mod store_types; +pub mod types; + +pub use frequency::{MIN_AGENT_JOB_INTERVAL, TooFrequent}; +pub use records::{Attempt, Completion, RunRecord}; +pub use requests::{ + CheckOriginRequest, CheckPatchRequest, NextRunRequest, RunsCloserThanRequest, + ValidateScheduleRequest, +}; +pub use store_types::{ + AgentJobSpec, ImportedJob, MAX_CRON_OUTPUT_BYTES, StoreOperation, TRUNCATED_OUTPUT_MARKER, +}; +pub use types::{ + ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, + JobPolicy, JobType, Schedule, SessionTarget, delivery_mode, +}; + /// The module has one bus identity and separately served authoring/schedule objects. pub const BUS_NAME: &str = crate::BUS_NAME; /// Separate scheduler interface in the shared module. @@ -11,20 +31,73 @@ pub const OBJECT_PATH: &str = "/ai/tinyhumans/tinyflows/Schedule"; /// Host interface carrying the context-bound runtime ports. pub const CALLBACK_INTERFACE: &str = "ai.tinyhumans.tinyflows.ScheduleHost"; /// Version of the scheduler wire vocabulary. 2 removed the module-owned -/// `sqlite` store backend; storage is the host's document port only. -pub const CONTRACT_VERSION: u32 = 2; +/// `sqlite` store backend; storage is the host's document port only. 3 moved +/// 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; /// Exact advertised scheduler member set. pub const METHODS: &[&str] = &[ - "ScheduleStore", - "StartScheduler", - "StopScheduler", - "TickScheduler", - "DrainScheduler", - "ExecuteJob", - "ParseHumanDelay", - "ValidateDelivery", + members::SCHEDULE_STORE, + members::START_SCHEDULER, + members::STOP_SCHEDULER, + members::TICK_SCHEDULER, + members::DRAIN_SCHEDULER, + members::EXECUTE_JOB, + members::PARSE_HUMAN_DELAY, + members::VALIDATE_DELIVERY, + members::NEXT_RUN, + members::VALIDATE_SCHEDULE, + members::RUNS_CLOSER_THAN, + members::NORMALIZE_EXPRESSION, + members::SCHEDULE_CRON_EXPRESSION, + members::DEFAULT_DELIVERY, + members::CHECK_ORIGIN_REQUIREMENTS, + members::CHECK_PATCH, + members::TRUNCATE_CRON_OUTPUT, ]; +/// Stable scheduler member names, in `METHODS` order. +pub mod members { + /// `(StoreRequest) -> JSON`; see `STORE_OPERATIONS`. + pub const SCHEDULE_STORE: &str = "ScheduleStore"; + /// `(StartScheduler) -> handle: string`. + pub const START_SCHEDULER: &str = "StartScheduler"; + /// `(handle: string) -> ()`. + pub const STOP_SCHEDULER: &str = "StopScheduler"; + /// `(handle: string) -> usize`; manual-tick schedulers only. + pub const TICK_SCHEDULER: &str = "TickScheduler"; + /// `(handle: string) -> ()`; manual-tick schedulers only. + pub const DRAIN_SCHEDULER: &str = "DrainScheduler"; + /// `(scope, CallbackAddress, config, job, run_id) -> Attempt`. + pub const EXECUTE_JOB: &str = "ExecuteJob"; + /// `(delay: string) -> milliseconds: i64`. + pub const PARSE_HUMAN_DELAY: &str = "ParseHumanDelay"; + /// `(DeliveryConfig, Option, Option>) -> ()`. + pub const VALIDATE_DELIVERY: &str = "ValidateDelivery"; + /// `(NextRunRequest) -> DateTime`. + pub const NEXT_RUN: &str = "NextRun"; + /// `(ValidateScheduleRequest) -> ()`. + pub const VALIDATE_SCHEDULE: &str = "ValidateSchedule"; + /// `(RunsCloserThanRequest) -> Option`. + pub const RUNS_CLOSER_THAN: &str = "RunsCloserThan"; + /// `(expression: string) -> string`: POSIX crontab weekdays into the + /// `cron` crate's numbering; errors on a malformed expression. + pub const NORMALIZE_EXPRESSION: &str = "NormalizeExpression"; + /// `(Schedule) -> Option`: the cron expression, if it has one. + pub const SCHEDULE_CRON_EXPRESSION: &str = "ScheduleCronExpression"; + /// `(Option) -> DeliveryConfig`: origin delivery with an + /// origin, proactive without. + pub const DEFAULT_DELIVERY: &str = "DefaultDelivery"; + /// `(CheckOriginRequest) -> ()`. + pub const CHECK_ORIGIN_REQUIREMENTS: &str = "CheckOriginRequirements"; + /// `(CheckPatchRequest) -> ()`. + pub const CHECK_PATCH: &str = "CheckPatch"; + /// `(output: string) -> string`, bounded to `MAX_CRON_OUTPUT_BYTES`. + pub const TRUNCATE_CRON_OUTPUT: &str = "TruncateCronOutput"; +} + /// A host callback object bound to the owning agent/tenant context. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct CallbackAddress { @@ -117,7 +190,7 @@ pub struct StoreRequest { pub operation: serde_json::Value, } -/// Supported `operation.op` values. Other fields use existing job/spec/patch +/// Supported `operation.op` values; `StoreOperation` is their typed form. Other fields use job/spec/patch /// JSON shapes. Run timestamps are RFC3339 and durations are milliseconds. pub const STORE_OPERATIONS: &[&str] = &[ "add", @@ -137,6 +210,9 @@ pub const STORE_OPERATIONS: &[&str] = &[ "record_run", "delete_queued_runs", "list_runs", + "set_max_active_jobs", + "import_job", + "import_run", ]; /// Context-bound document callback interface. The only member is `Document`; @@ -148,3 +224,7 @@ pub const STORE_OPERATIONS: &[&str] = &[ pub const DOCUMENT_INTERFACE: &str = "ai.tinyhumans.tinyflows.ScheduleDocuments"; /// Exact document callback member used by the module. pub const DOCUMENT_METHOD: &str = "Document"; + +#[cfg(test)] +#[path = "mod_tests.rs"] +mod tests; diff --git a/crates/tinyflows-bus/src/schedule/wire_tests.rs b/crates/tinyflows-bus/src/schedule/wire_tests.rs new file mode 100644 index 00000000..1d60bfe8 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/wire_tests.rs @@ -0,0 +1,139 @@ +//! Literal-JSON fixtures pinning the persisted / RPC wire shape of the +//! schedule types. These strings are what a host's job store and RPC clients +//! already hold; if one stops parsing or re-serialising byte-for-byte, a +//! stored job has been orphaned. + +use super::{ + ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, + JobType, Schedule, SessionTarget, +}; + +fn assert_bytes(json: &str, expected: &T) +where + T: serde::Serialize + serde::de::DeserializeOwned + PartialEq + std::fmt::Debug, +{ + let parsed: T = serde_json::from_str(json).expect("fixture parses"); + assert_eq!(&parsed, expected); + assert_eq!(serde_json::to_string(&parsed).unwrap(), json); +} + +#[test] +fn cron_schedule_wire_bytes() { + assert_bytes( + r#"{"kind":"cron","expr":"0 9 * * 1","tz":"Europe/London","active_hours":{"start":"09:00","end":"17:30"}}"#, + &Schedule::Cron { + expr: "0 9 * * 1".into(), + tz: Some("Europe/London".into()), + active_hours: Some(ActiveHours { + start: "09:00".into(), + end: "17:30".into(), + }), + }, + ); + assert_bytes( + r#"{"kind":"cron","expr":"*/5 * * * *","tz":null,"active_hours":null}"#, + &Schedule::Cron { + expr: "*/5 * * * *".into(), + tz: None, + active_hours: None, + }, + ); +} + +#[test] +fn at_and_every_schedule_wire_bytes() { + let at: Schedule = + serde_json::from_str(r#"{"kind":"at","at":"2026-02-16T17:00:00Z"}"#).unwrap(); + assert!(matches!(at, Schedule::At { .. })); + assert_eq!( + serde_json::to_string(&at).unwrap(), + r#"{"kind":"at","at":"2026-02-16T17:00:00Z"}"# + ); + assert_bytes( + r#"{"kind":"every","every_ms":60000}"#, + &Schedule::Every { every_ms: 60_000 }, + ); +} + +#[test] +fn legacy_shapes_still_deserialize() { + // Missing optional keys and the bare-string shorthand both read as Cron. + let no_optionals: Schedule = + serde_json::from_str(r#"{"kind":"cron","expr":"0 9 * * *"}"#).unwrap(); + let bare: Schedule = serde_json::from_str(r#""0 9 * * *""#).unwrap(); + let expected = Schedule::Cron { + expr: "0 9 * * *".into(), + tz: None, + active_hours: None, + }; + assert_eq!(no_optionals, expected); + assert_eq!(bare, expected); +} + +#[test] +fn enum_and_delivery_wire_bytes() { + assert_bytes(r#""shell""#, &JobType::Shell); + assert_bytes(r#""agent""#, &JobType::Agent); + assert_bytes(r#""flow""#, &JobType::Flow); + assert_bytes(r#""isolated""#, &SessionTarget::Isolated); + assert_bytes(r#""main""#, &SessionTarget::Main); + assert_bytes(r#""current""#, &SessionTarget::Current); + assert_bytes(r#""not_requested""#, &DeliveryStatus::NotRequested); + assert_bytes( + r#"{"mode":"none","channel":null,"to":null,"best_effort":true}"#, + &DeliveryConfig::default(), + ); +} + +#[test] +fn job_and_run_wire_bytes() { + let job_json = r#"{"id":"j1","expression":"0 9 * * *","schedule":{"kind":"cron","expr":"0 9 * * *","tz":null,"active_hours":null},"command":"echo hi","prompt":null,"name":"morning","job_type":"shell","session_target":"isolated","model":null,"agent_id":null,"enabled":true,"delivery":{"mode":"none","channel":null,"to":null,"best_effort":true},"delete_after_run":false,"created_at":"2026-02-16T00:00:00Z","next_run":"2026-02-16T09:00:00Z","last_run":null,"last_status":null,"last_output":null}"#; + let job: CronJob = serde_json::from_str(job_json).unwrap(); + assert_eq!(serde_json::to_string(&job).unwrap(), job_json); + + let run_json = r#"{"id":7,"job_id":"j1","started_at":"2026-02-16T09:00:00Z","finished_at":"2026-02-16T09:00:02Z","status":"ok","output":"hi","duration_ms":2000}"#; + let run: CronRun = serde_json::from_str(run_json).unwrap(); + assert_eq!(serde_json::to_string(&run).unwrap(), run_json); +} + +#[test] +fn patch_double_option_wire_semantics() { + let absent: CronJobPatch = serde_json::from_str("{}").unwrap(); + assert_eq!(absent.agent_id, None); + let cleared: CronJobPatch = serde_json::from_str(r#"{"agent_id":null}"#).unwrap(); + assert_eq!(cleared.agent_id, Some(None)); + let set: CronJobPatch = serde_json::from_str(r#"{"agent_id":"welcome"}"#).unwrap(); + assert_eq!(set.agent_id, Some(Some("welcome".into()))); +} + +#[test] +fn origin_wire_bytes() { + assert_bytes( + r#"{"kind":"web","thread_id":"t1","agent_id":"orchestrator"}"#, + &JobOrigin::Web { + thread_id: "t1".into(), + agent_id: Some("orchestrator".into()), + }, + ); + assert_bytes( + r#"{"kind":"channel","channel":"telegram","reply_target":"42","history_key":"telegram:42","sender":"alice","thread_id":"7"}"#, + &JobOrigin::Channel { + channel: "telegram".into(), + reply_target: "42".into(), + history_key: "telegram:42".into(), + sender: Some("alice".into()), + thread_id: Some("7".into()), + }, + ); +} + +#[test] +fn job_with_origin_and_run_with_delivery_status_wire_bytes() { + let job_json = r#"{"id":"j1","expression":"0 9 * * *","schedule":{"kind":"cron","expr":"0 9 * * *","tz":null,"active_hours":null},"command":"","prompt":"brief","name":null,"job_type":"agent","session_target":"current","model":null,"agent_id":null,"enabled":true,"delivery":{"mode":"origin","channel":null,"to":null,"best_effort":true},"delete_after_run":false,"created_at":"2026-02-16T00:00:00Z","next_run":"2026-02-16T09:00:00Z","last_run":null,"last_status":null,"last_output":null,"origin":{"kind":"web","thread_id":"t1"}}"#; + let job: CronJob = serde_json::from_str(job_json).unwrap(); + assert_eq!(serde_json::to_string(&job).unwrap(), job_json); + + let run_json = r#"{"id":7,"job_id":"j1","started_at":"2026-02-16T09:00:00Z","finished_at":"2026-02-16T09:00:02Z","status":"ok","output":"hi","duration_ms":2000,"delivery_status":"delivered"}"#; + let run: CronRun = serde_json::from_str(run_json).unwrap(); + assert_eq!(serde_json::to_string(&run).unwrap(), run_json); +} From 0a257a504723ef03b1aee05b295344dc260afaaa Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:03:51 +0300 Subject: [PATCH 05/19] feat(schedule): add chrono dependency and wire test module Add chrono to the tinyflows-bus dependencies and register a new wire_tests module behind the test configuration so the schedule wire format can be covered by dedicated tests. Auto-committed-on: dragonfly --- Cargo.lock | 1 + crates/tinyflows-bus/src/schedule/mod.rs | 3 + .../tinyflows-bus/src/schedule/mod_tests.rs | 123 ++++++++++++++++++ 3 files changed, 127 insertions(+) create mode 100644 crates/tinyflows-bus/src/schedule/mod_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 7c0cab03..2f2cebb4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2044,6 +2044,7 @@ dependencies = [ name = "tinyflows-bus" version = "0.1.0" dependencies = [ + "chrono", "serde", "serde_json", ] diff --git a/crates/tinyflows-bus/src/schedule/mod.rs b/crates/tinyflows-bus/src/schedule/mod.rs index eaa2371b..d82f6421 100644 --- a/crates/tinyflows-bus/src/schedule/mod.rs +++ b/crates/tinyflows-bus/src/schedule/mod.rs @@ -228,3 +228,6 @@ pub const DOCUMENT_METHOD: &str = "Document"; #[cfg(test)] #[path = "mod_tests.rs"] mod tests; +#[cfg(test)] +#[path = "wire_tests.rs"] +mod wire_tests; diff --git a/crates/tinyflows-bus/src/schedule/mod_tests.rs b/crates/tinyflows-bus/src/schedule/mod_tests.rs new file mode 100644 index 00000000..855d5851 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/mod_tests.rs @@ -0,0 +1,123 @@ +use super::*; +use serde_json::json; + +fn every() -> Schedule { + Schedule::Every { every_ms: 60_000 } +} + +#[test] +fn contract_is_version_three_with_the_exact_member_set() { + assert_eq!(CONTRACT_VERSION, 3); + assert_eq!(METHODS.len(), 17); + let mut unique: Vec<_> = METHODS.to_vec(); + unique.sort_unstable(); + unique.dedup(); + assert_eq!(unique.len(), METHODS.len()); + for name in [ + "ScheduleStore", + "ParseHumanDelay", + "ValidateDelivery", + "NextRun", + "ValidateSchedule", + "RunsCloserThan", + "NormalizeExpression", + "ScheduleCronExpression", + "DefaultDelivery", + "CheckOriginRequirements", + "CheckPatch", + "TruncateCronOutput", + ] { + assert!(METHODS.contains(&name), "{name}"); + } +} + +#[test] +fn every_advertised_store_operation_round_trips_through_the_typed_form() { + let job = json!({"id":"j1","expression":"","schedule":{"kind":"every","every_ms":60000},"command":"c","prompt":null,"name":null,"job_type":"shell","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}); + let run = json!({"id":1,"job_id":"j1","started_at":"2026-01-01T00:00:00Z","finished_at":"2026-01-01T00:00:01Z","status":"ok","output":null,"duration_ms":1000}); + let spec = json!({"name":null,"schedule":{"kind":"every","every_ms":60000},"prompt":"p","session_target":"isolated","model":null,"delivery":null,"delete_after_run":false,"agent_id":null,"enabled":true,"origin":null}); + let operations = [ + json!({"op":"add","expression":"* * * * *","command":"c"}), + json!({"op":"add_shell","name":null,"schedule":{"kind":"every","every_ms":60000},"command":"c"}), + json!({"op":"add_agent","spec":spec}), + json!({"op":"add_flow","flow_id":"f","schedule":{"kind":"every","every_ms":60000}}), + json!({"op":"find_flow","flow_id":"f"}), + json!({"op":"list"}), + json!({"op":"get","job_id":"j"}), + json!({"op":"remove","job_id":"j"}), + json!({"op":"clear"}), + json!({"op":"dedup"}), + json!({"op":"due","now":"2026-01-01T00:00:00Z"}), + json!({"op":"update","job_id":"j","patch":{"enabled":false}}), + json!({"op":"record_last_run","job_id":"j","finished_at":"2026-01-01T00:00:00Z","success":true,"output":"o"}), + json!({"op":"reschedule","job":job,"success":true,"output":"o"}), + json!({"op":"record_run","job_id":"j","started_at":"2026-01-01T00:00:00Z","finished_at":"2026-01-01T00:00:01Z","status":"ok","output":null,"duration_ms":1000,"delivery_status":null}), + json!({"op":"delete_queued_runs","job_id":"j"}), + json!({"op":"list_runs","job_id":"j","limit":5}), + json!({"op":"set_max_active_jobs","limit":3}), + json!({"op":"import_job","job":job}), + json!({"op":"import_run","run":run}), + ]; + assert_eq!(operations.len(), STORE_OPERATIONS.len()); + for (operation, name) in operations.iter().zip(STORE_OPERATIONS) { + assert_eq!(operation["op"], *name); + let typed: StoreOperation = serde_json::from_value(operation.clone()).unwrap(); + assert_eq!(serde_json::to_value(&typed).unwrap()["op"], *name); + } + let cleared: StoreOperation = + serde_json::from_value(json!({"op":"set_max_active_jobs","limit":null})).unwrap(); + assert!(matches!( + cleared, + StoreOperation::SetMaxActiveJobs { limit: None } + )); +} + +#[test] +fn too_frequent_wire_shape_and_display() { + let fixed = TooFrequent::FixedInterval(chrono::Duration::seconds(90)); + assert_eq!(serde_json::to_string(&fixed).unwrap(), r#"{"fixed_interval":90000}"#); + assert_eq!(fixed.to_string(), "fires every 1 minute 30 seconds"); + let pair: TooFrequent = serde_json::from_str( + r#"{"consecutive_runs":{"first":"2026-01-01T00:56:00Z","second":"2026-01-01T01:00:00Z"}}"#, + ) + .unwrap(); + assert_eq!(pair.gap(), chrono::Duration::minutes(4)); + assert!(pair.to_string().ends_with("4 minutes apart")); + assert_eq!(serde_json::from_str::(&serde_json::to_string(&fixed).unwrap()).unwrap(), fixed); + assert_eq!(MIN_AGENT_JOB_INTERVAL, chrono::Duration::minutes(5)); +} + +#[test] +fn callback_records_keep_their_json_shape() { + let attempt: Attempt = + serde_json::from_str(r#"{"success":true,"output":"o","retryable":false}"#).unwrap(); + assert_eq!( + serde_json::to_string(&attempt).unwrap(), + r#"{"success":true,"output":"o","retryable":false}"# + ); + let record_json = r#"{"run_id":"r","job_id":"j","job_created_at":"2026-01-01T00:00:00Z","started_at":"2026-01-01T00:00:01Z","finished_at":"2026-01-01T00:00:02Z","status":"ok","output":"o","delivery_status":"delivered"}"#; + let record: RunRecord = serde_json::from_str(record_json).unwrap(); + assert_eq!(serde_json::to_string(&record).unwrap(), record_json); + let completion_json = r#"{"remove":false,"disable":true,"next_run":null,"finished_at":"2026-01-01T00:00:02Z","success":true,"output":"o"}"#; + let completion: Completion = serde_json::from_str(completion_json).unwrap(); + assert_eq!(serde_json::to_string(&completion).unwrap(), completion_json); +} + +#[test] +fn agent_job_spec_defaults_and_request_types_deserialize() { + let spec = AgentJobSpec::new(every(), "hi"); + assert!(spec.enabled && spec.delivery.is_none() && spec.origin.is_none()); + let back: AgentJobSpec = serde_json::from_value(serde_json::to_value(&spec).unwrap()).unwrap(); + assert_eq!(back.prompt, "hi"); + + let validate: ValidateScheduleRequest = serde_json::from_value( + json!({"schedule":{"kind":"every","every_ms":1000},"now":"2026-01-01T00:00:00Z"}), + ) + .unwrap(); + assert!(!validate.agent); + let origin: CheckOriginRequest = serde_json::from_value( + json!({"session_target":"current","delivery":{"mode":"origin"}}), + ) + .unwrap(); + assert!(origin.origin.is_none()); +} From 3678ac21262748f814549b026a9ba8acab290ac2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:04:26 +0300 Subject: [PATCH 06/19] refactor(schedule): move shared schedule types into tinyflows-bus The schedule contract types (Attempt, RunRecord, Completion, AgentJobSpec, TooFrequent, MIN_AGENT_JOB_INTERVAL and the output truncation constants) now live in tinyflows-bus and are re-exported from their previous locations, so other crates can depend on the contract without pulling in the scheduler. Auto-committed-on: dragonfly --- Cargo.lock | 1 + crates/tinyflows-schedule/Cargo.toml | 1 + .../tinyflows-schedule/src/runtime/ports.rs | 53 +-------------- crates/tinyflows-schedule/src/schedule.rs | 66 ++----------------- crates/tinyflows-schedule/src/store.rs | 47 +------------ crates/tinyflows-schedule/src/store_tests.rs | 1 + crates/tinyflows-schedule/src/types.rs | 3 + 7 files changed, 13 insertions(+), 159 deletions(-) create mode 100644 crates/tinyflows-schedule/src/types.rs diff --git a/Cargo.lock b/Cargo.lock index 2f2cebb4..346c7831 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2153,6 +2153,7 @@ dependencies = [ "cron", "serde", "serde_json", + "tinyflows-bus", "tokio", "tracing", "uuid", diff --git a/crates/tinyflows-schedule/Cargo.toml b/crates/tinyflows-schedule/Cargo.toml index c04ac303..dfd4ddd6 100644 --- a/crates/tinyflows-schedule/Cargo.toml +++ b/crates/tinyflows-schedule/Cargo.toml @@ -11,6 +11,7 @@ homepage.workspace = true readme = "../../README.md" [dependencies] +tinyflows-bus = { path = "../tinyflows-bus" } async-trait.workspace = true tokio.workspace = true uuid.workspace = true diff --git a/crates/tinyflows-schedule/src/runtime/ports.rs b/crates/tinyflows-schedule/src/runtime/ports.rs index 6bff1627..36e7214b 100644 --- a/crates/tinyflows-schedule/src/runtime/ports.rs +++ b/crates/tinyflows-schedule/src/runtime/ports.rs @@ -3,60 +3,9 @@ use crate::{CronJob, DeliveryStatus}; use anyhow::Result; use async_trait::async_trait; use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; use std::time::Duration; -/// Host-classified result of one attempt. Permanent failures halt immediately. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Attempt { - /// Whether execution (or final execution plus delivery) succeeded. - pub success: bool, - /// Result text passed unchanged to delivery and run-history persistence. - pub output: String, - /// False for host-classified permanent failures; stops retry immediately. - pub retryable: bool, -} - -/// Persisted outcome after execution and delivery. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RunRecord { - /// Stable invocation identity shared by all attempts and delivery. - pub run_id: String, - /// Identity of the job that produced this run. - pub job_id: String, - /// Dispatch-time creation identity. Stores must reject a replacement with - /// the same job ID instead of attaching this outcome to its history. - pub job_created_at: DateTime, - /// Execution start, before the first attempt. - pub started_at: DateTime, - /// Execution finish before delivery; skipped runs use the same start/end. - pub finished_at: DateTime, - /// `ok`, `error`, or `skipped`. - pub status: String, - /// Result text passed unchanged to delivery and run-history persistence. - pub output: String, - /// Delivery outcome, absent on skipped slots. - pub delivery_status: Option, -} - -/// Final mutation computed by the runtime. Stores apply it mechanically. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Completion { - /// Remove a successful auto-delete one-shot, including its stored history, - /// unless a concurrent edit re-enabled or otherwise retained it. - pub remove: bool, - /// Retain a one-shot already disabled at claim time. A store must preserve - /// an explicit re-enable made after that claim. - pub disable: bool, - /// Recurrence computed after delivery; absent for one-shot jobs. - pub next_run: Option>, - /// Execution finish before delivery; skipped runs use the same start/end. - pub finished_at: DateTime, - /// Whether execution (or final execution plus delivery) succeeded. - pub success: bool, - /// Result text passed unchanged to delivery and run-history persistence. - pub output: String, -} +pub use tinyflows_bus::schedule::{Attempt, Completion, RunRecord}; /// Storage bound to one agent/tenant scope. #[async_trait] diff --git a/crates/tinyflows-schedule/src/schedule.rs b/crates/tinyflows-schedule/src/schedule.rs index e8ae790d..37426816 100644 --- a/crates/tinyflows-schedule/src/schedule.rs +++ b/crates/tinyflows-schedule/src/schedule.rs @@ -6,13 +6,10 @@ use cron::Schedule as CronExprSchedule; use std::fmt; use std::str::FromStr; -/// The closest together two runs of an *agent* job may be scheduled. Every -/// run is a full inference turn, so a tighter cadence is almost always a -/// misconfiguration that bills accordingly. Shell and flow jobs are not -/// subject to it. Enforced by [`validate_agent_schedule`] when an agent job is -/// created or its schedule changed; the scheduler warns about rows that -/// predate the rule. -pub const MIN_AGENT_JOB_INTERVAL: ChronoDuration = ChronoDuration::minutes(5); +// `MIN_AGENT_JOB_INTERVAL` (the closest together two runs of an agent job may +// be scheduled, enforced by `validate_agent_schedule`) and `TooFrequent` live in +// the bus contract and are re-exported here. +pub use tinyflows_bus::schedule::{MIN_AGENT_JOB_INTERVAL, TooFrequent}; /// Upper bound on cron candidates walked while looking for an occurrence that /// falls inside `active_hours`. `next_run_for_schedule` gets a fresh budget @@ -95,44 +92,6 @@ pub fn validate_agent_schedule(schedule: &Schedule, now: DateTime) -> Resul Ok(()) } -/// Why a schedule runs more often than a threshold allows. Carries the -/// evidence, not only the verdict, so a log line or an error can quote it. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum TooFrequent { - /// A fixed `every_ms` interval shorter than the threshold. - FixedInterval(ChronoDuration), - /// Two consecutive cron occurrences closer together than the threshold. - ConsecutiveRuns { - first: DateTime, - second: DateTime, - }, -} - -impl TooFrequent { - /// The offending gap. - pub fn gap(&self) -> ChronoDuration { - match self { - Self::FixedInterval(gap) => *gap, - Self::ConsecutiveRuns { first, second } => *second - *first, - } - } -} - -impl fmt::Display for TooFrequent { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::FixedInterval(gap) => write!(f, "fires every {}", describe_gap(*gap)), - Self::ConsecutiveRuns { first, second } => write!( - f, - "fires at {} and again at {}, {} apart", - first.format("%Y-%m-%d %H:%M:%S UTC"), - second.format("%Y-%m-%d %H:%M:%S UTC"), - describe_gap(*second - *first) - ), - } - } -} - /// The two consecutive runs of `schedule` after `from` that are closest /// together, if that gap is under `min_gap`. /// @@ -189,23 +148,6 @@ pub fn runs_closer_than( } } -/// `4 minutes`, `30 seconds`, `1 minute 30 seconds` — whole seconds only. -fn describe_gap(gap: ChronoDuration) -> String { - fn count(n: i64, unit: &str) -> String { - if n == 1 { - format!("{n} {unit}") - } else { - format!("{n} {unit}s") - } - } - let seconds = gap.num_seconds(); - match (seconds / 60, seconds % 60) { - (0, s) => count(s, "second"), - (m, 0) => count(m, "minute"), - (m, s) => format!("{} {}", count(m, "minute"), count(s, "second")), - } -} - pub fn schedule_cron_expression(schedule: &Schedule) -> Option { match schedule { Schedule::Cron { expr, .. } => Some(expr.clone()), diff --git a/crates/tinyflows-schedule/src/store.rs b/crates/tinyflows-schedule/src/store.rs index cba54ad8..35b022ec 100644 --- a/crates/tinyflows-schedule/src/store.rs +++ b/crates/tinyflows-schedule/src/store.rs @@ -8,52 +8,9 @@ //! [`CronJob`]: crate::CronJob //! [`CronRun`]: crate::CronRun -use crate::{CronJob, CronJobPatch, DeliveryConfig, JobOrigin, JobType, Schedule, SessionTarget}; +use crate::{CronJob, CronJobPatch, JobType}; -/// Largest job output, in bytes, a store keeps in `last_output` or a run; -/// longer output is cut at a char boundary and ends with -/// [`TRUNCATED_OUTPUT_MARKER`]. -pub const MAX_CRON_OUTPUT_BYTES: usize = 16 * 1024; -/// Suffix appended to output cut to [`MAX_CRON_OUTPUT_BYTES`]. -pub const TRUNCATED_OUTPUT_MARKER: &str = "\n...[truncated]"; - -/// Everything an agent job is created with. [`AgentJobSpec::new`] gives an -/// enabled, isolated job with no name, model, definition or origin. -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct AgentJobSpec { - pub name: Option, - pub schedule: Schedule, - pub prompt: String, - pub session_target: SessionTarget, - pub model: Option, - /// `None` stores [`DeliveryConfig::default`]. - pub delivery: Option, - pub delete_after_run: bool, - /// Built-in agent definition to run with. - pub agent_id: Option, - /// Inserted in this state in one write, so an opt-in job is never - /// briefly enabled. - pub enabled: bool, - /// The conversation the job was created from. - pub origin: Option, -} - -impl AgentJobSpec { - pub fn new(schedule: Schedule, prompt: impl Into) -> Self { - Self { - name: None, - schedule, - prompt: prompt.into(), - session_target: SessionTarget::default(), - model: None, - delivery: None, - delete_after_run: false, - agent_id: None, - enabled: true, - origin: None, - } - } -} +pub use tinyflows_bus::schedule::{AgentJobSpec, MAX_CRON_OUTPUT_BYTES, TRUNCATED_OUTPUT_MARKER}; /// Refuses a patch no store may apply to `job`. /// diff --git a/crates/tinyflows-schedule/src/store_tests.rs b/crates/tinyflows-schedule/src/store_tests.rs index 6b30055e..d855f6b7 100644 --- a/crates/tinyflows-schedule/src/store_tests.rs +++ b/crates/tinyflows-schedule/src/store_tests.rs @@ -1,4 +1,5 @@ use super::*; +use crate::{DeliveryConfig, Schedule, SessionTarget}; #[test] fn short_output_is_kept_and_long_output_is_cut_at_a_char_boundary() { diff --git a/crates/tinyflows-schedule/src/types.rs b/crates/tinyflows-schedule/src/types.rs new file mode 100644 index 00000000..e0a49183 --- /dev/null +++ b/crates/tinyflows-schedule/src/types.rs @@ -0,0 +1,3 @@ +//! The schedule and job/run records. They live in `tinyflows-bus` (the wire +//! contract) and are re-exported here at their original paths. +pub use tinyflows_bus::schedule::types::*; From 31c25860060422f7c9b7a391d40dce0bfa6ce791 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:04:46 +0300 Subject: [PATCH 07/19] refactor(schedule): re-export describe_gap from the bus contract The gap-description helper is now public and re-exported alongside the interval constants, so callers can format durations without duplicating the logic. Auto-committed-on: dragonfly --- crates/tinyflows-bus/src/schedule/frequency.rs | 2 +- crates/tinyflows-bus/src/schedule/mod.rs | 2 +- crates/tinyflows-schedule/src/schedule.rs | 3 +-- 3 files changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/tinyflows-bus/src/schedule/frequency.rs b/crates/tinyflows-bus/src/schedule/frequency.rs index 1c88e70b..65a5374d 100644 --- a/crates/tinyflows-bus/src/schedule/frequency.rs +++ b/crates/tinyflows-bus/src/schedule/frequency.rs @@ -65,7 +65,7 @@ impl fmt::Display for TooFrequent { } /// `4 minutes`, `30 seconds`, `1 minute 30 seconds` — whole seconds only. -fn describe_gap(gap: ChronoDuration) -> String { +pub fn describe_gap(gap: ChronoDuration) -> String { fn count(n: i64, unit: &str) -> String { if n == 1 { format!("{n} {unit}") diff --git a/crates/tinyflows-bus/src/schedule/mod.rs b/crates/tinyflows-bus/src/schedule/mod.rs index d82f6421..ce53e458 100644 --- a/crates/tinyflows-bus/src/schedule/mod.rs +++ b/crates/tinyflows-bus/src/schedule/mod.rs @@ -8,7 +8,7 @@ mod requests; mod store_types; pub mod types; -pub use frequency::{MIN_AGENT_JOB_INTERVAL, TooFrequent}; +pub use frequency::{MIN_AGENT_JOB_INTERVAL, TooFrequent, describe_gap}; pub use records::{Attempt, Completion, RunRecord}; pub use requests::{ CheckOriginRequest, CheckPatchRequest, NextRunRequest, RunsCloserThanRequest, diff --git a/crates/tinyflows-schedule/src/schedule.rs b/crates/tinyflows-schedule/src/schedule.rs index 37426816..04b01384 100644 --- a/crates/tinyflows-schedule/src/schedule.rs +++ b/crates/tinyflows-schedule/src/schedule.rs @@ -3,13 +3,12 @@ use crate::types::{ActiveHours, Schedule}; use anyhow::{Context, Result}; use chrono::{DateTime, Duration as ChronoDuration, NaiveTime, Timelike, Utc}; use cron::Schedule as CronExprSchedule; -use std::fmt; use std::str::FromStr; // `MIN_AGENT_JOB_INTERVAL` (the closest together two runs of an agent job may // be scheduled, enforced by `validate_agent_schedule`) and `TooFrequent` live in // the bus contract and are re-exported here. -pub use tinyflows_bus::schedule::{MIN_AGENT_JOB_INTERVAL, TooFrequent}; +pub use tinyflows_bus::schedule::{MIN_AGENT_JOB_INTERVAL, TooFrequent, describe_gap}; /// Upper bound on cron candidates walked while looking for an occurrence that /// falls inside `active_hours`. `next_run_for_schedule` gets a fresh budget From fb009169f95f42447a2c30e66eb00866ebd2ac91 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:07:15 +0300 Subject: [PATCH 08/19] feat(schedule): add persistent schedule store Schedules are now persisted through a dedicated store so they survive restarts instead of living only in memory. The bus, drivers, and module layers were wired up to read and write through the new store. Auto-committed-on: dragonfly --- .../tinyflows-bus/src/schedule/mod_tests.rs | 17 ++-- .../tinyflows-drivers/src/schedule/import.rs | 12 +-- .../tinyflows-module/src/schedule_service.rs | 56 +++++++++++++- crates/tinyflows-module/src/schedule_store.rs | 77 ++----------------- .../src/schedule_store_tests.rs | 1 + crates/tinyflows-schedule/src/lib.rs | 3 +- crates/tinyflows-schedule/src/store.rs | 4 +- 7 files changed, 80 insertions(+), 90 deletions(-) diff --git a/crates/tinyflows-bus/src/schedule/mod_tests.rs b/crates/tinyflows-bus/src/schedule/mod_tests.rs index 855d5851..67992335 100644 --- a/crates/tinyflows-bus/src/schedule/mod_tests.rs +++ b/crates/tinyflows-bus/src/schedule/mod_tests.rs @@ -75,7 +75,10 @@ fn every_advertised_store_operation_round_trips_through_the_typed_form() { #[test] fn too_frequent_wire_shape_and_display() { let fixed = TooFrequent::FixedInterval(chrono::Duration::seconds(90)); - assert_eq!(serde_json::to_string(&fixed).unwrap(), r#"{"fixed_interval":90000}"#); + assert_eq!( + serde_json::to_string(&fixed).unwrap(), + r#"{"fixed_interval":90000}"# + ); assert_eq!(fixed.to_string(), "fires every 1 minute 30 seconds"); let pair: TooFrequent = serde_json::from_str( r#"{"consecutive_runs":{"first":"2026-01-01T00:56:00Z","second":"2026-01-01T01:00:00Z"}}"#, @@ -83,7 +86,10 @@ fn too_frequent_wire_shape_and_display() { .unwrap(); assert_eq!(pair.gap(), chrono::Duration::minutes(4)); assert!(pair.to_string().ends_with("4 minutes apart")); - assert_eq!(serde_json::from_str::(&serde_json::to_string(&fixed).unwrap()).unwrap(), fixed); + assert_eq!( + serde_json::from_str::(&serde_json::to_string(&fixed).unwrap()).unwrap(), + fixed + ); assert_eq!(MIN_AGENT_JOB_INTERVAL, chrono::Duration::minutes(5)); } @@ -115,9 +121,8 @@ fn agent_job_spec_defaults_and_request_types_deserialize() { ) .unwrap(); assert!(!validate.agent); - let origin: CheckOriginRequest = serde_json::from_value( - json!({"session_target":"current","delivery":{"mode":"origin"}}), - ) - .unwrap(); + let origin: CheckOriginRequest = + serde_json::from_value(json!({"session_target":"current","delivery":{"mode":"origin"}})) + .unwrap(); assert!(origin.origin.is_none()); } diff --git a/crates/tinyflows-drivers/src/schedule/import.rs b/crates/tinyflows-drivers/src/schedule/import.rs index 9fc3ad63..5eeba422 100644 --- a/crates/tinyflows-drivers/src/schedule/import.rs +++ b/crates/tinyflows-drivers/src/schedule/import.rs @@ -32,16 +32,8 @@ use super::jobs::{flow_job_id, new_job_doc, runs_of}; use super::runs::RUN_COUNTER; use super::{CAS_ATTEMPTS, COUNTERS, CronDocuments, JOBS, RUNS, storage_error}; -/// What [`CronDocuments::import_job`] did with a job. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ImportedJob { - /// The id the job has in this store. It differs from the imported job's own - /// id for a flow-schedule job; its runs are imported under this id. - pub id: String, - /// Whether this call wrote the job. `false` when the store already held a - /// job under [`Self::id`], which is left exactly as it was. - pub written: bool, -} +/// What [`CronDocuments::import_job`] did with a job (the bus wire record). +pub use tinyflows_schedule::ImportedJob; impl CronDocuments { /// Imports `job`, keeping its schedule, delivery, origin, policy, claim diff --git a/crates/tinyflows-module/src/schedule_service.rs b/crates/tinyflows-module/src/schedule_service.rs index b00a5b4a..aa31713a 100644 --- a/crates/tinyflows-module/src/schedule_service.rs +++ b/crates/tinyflows-module/src/schedule_service.rs @@ -3,8 +3,13 @@ use super::schedule_callbacks::HostCallbacks; use serde_json::Value; use std::{collections::HashMap, sync::Arc}; use tinybus::{Connection, Error, Result}; -use tinyflows_bus::schedule::{self, CallbackAddress, StartScheduler}; -use tinyflows_schedule::{CronJob, DeliveryConfig, JobOrigin, runtime::*, validation}; +use tinyflows_bus::schedule::{ + self, CallbackAddress, CheckOriginRequest, CheckPatchRequest, NextRunRequest, + RunsCloserThanRequest, StartScheduler, TooFrequent, ValidateScheduleRequest, +}; +use tinyflows_schedule::{ + CronJob, DeliveryConfig, JobOrigin, Schedule, runtime::*, schedule as maths, store, validation, +}; use tokio::{ sync::{Mutex, oneshot}, task::JoinHandle, @@ -174,6 +179,53 @@ impl ScheduleService { validation::validate_delivery(&delivery, origin.as_ref(), allowed_users.as_deref()) .map_err(failure) } + + async fn next_run(&self, request: NextRunRequest) -> Result> { + maths::next_run_for_schedule(&request.schedule, request.from).map_err(failure) + } + async fn validate_schedule(&self, request: ValidateScheduleRequest) -> Result<()> { + if request.agent { + maths::validate_agent_schedule(&request.schedule, request.now) + } else { + maths::validate_schedule(&request.schedule, request.now) + } + .map_err(failure) + } + async fn runs_closer_than( + &self, + request: RunsCloserThanRequest, + ) -> Result> { + let min_gap = chrono::Duration::try_milliseconds(request.min_gap_ms) + .ok_or_else(|| Error::failed("min_gap_ms out of range"))?; + Ok(maths::runs_closer_than( + &request.schedule, + request.from, + min_gap, + )) + } + async fn normalize_expression(&self, expression: String) -> Result { + maths::normalize_expression(&expression).map_err(failure) + } + async fn schedule_cron_expression(&self, schedule: Schedule) -> Result> { + Ok(maths::schedule_cron_expression(&schedule)) + } + async fn default_delivery(&self, origin: Option) -> Result { + Ok(validation::default_delivery(origin.as_ref())) + } + async fn check_origin_requirements(&self, request: CheckOriginRequest) -> Result<()> { + validation::check_origin_requirements( + &request.session_target, + &request.delivery, + request.origin.as_ref(), + ) + .map_err(Error::failed) + } + async fn check_patch(&self, request: CheckPatchRequest) -> Result<()> { + store::check_patch(&request.job, &request.patch).map_err(failure) + } + async fn truncate_cron_output(&self, output: String) -> Result { + Ok(store::truncate_cron_output(&output)) + } } pub(super) async fn setup(connection: &Connection) -> Result<()> { diff --git a/crates/tinyflows-module/src/schedule_store.rs b/crates/tinyflows-module/src/schedule_store.rs index dc963372..41b51a8e 100644 --- a/crates/tinyflows-module/src/schedule_store.rs +++ b/crates/tinyflows-module/src/schedule_store.rs @@ -1,80 +1,14 @@ //! Module-owned persistence dispatch. Hosts pass configuration and JSON only. use anyhow::Result; use chrono::{DateTime, Utc}; -use serde::Deserialize; use serde_json::Value; use std::sync::Arc; use tinybus::Connection; -use tinyflows_bus::schedule::{StoreBackend, StoreConfig, StoreRequest}; +use tinyflows_bus::schedule::{ + StoreBackend, StoreConfig, StoreOperation as Operation, StoreRequest, +}; use tinyflows_drivers::schedule::CronDocuments; -use tinyflows_schedule::{AgentJobSpec, CronJob, CronJobPatch, DeliveryStatus, Schedule}; - -#[derive(Deserialize)] -#[serde(tag = "op", rename_all = "snake_case")] -enum Operation { - Add { - expression: String, - command: String, - }, - AddShell { - name: Option, - schedule: Schedule, - command: String, - }, - AddAgent { - spec: AgentJobSpec, - }, - AddFlow { - flow_id: String, - schedule: Schedule, - }, - FindFlow { - flow_id: String, - }, - List, - Get { - job_id: String, - }, - Remove { - job_id: String, - }, - Clear, - Dedup, - Due { - now: DateTime, - }, - Update { - job_id: String, - patch: CronJobPatch, - }, - RecordLastRun { - job_id: String, - finished_at: DateTime, - success: bool, - output: String, - }, - Reschedule { - job: CronJob, - success: bool, - output: String, - }, - RecordRun { - job_id: String, - started_at: DateTime, - finished_at: DateTime, - status: String, - output: Option, - duration_ms: i64, - delivery_status: Option, - }, - DeleteQueuedRuns { - job_id: String, - }, - ListRuns { - job_id: String, - limit: usize, - }, -} +use tinyflows_schedule::CronJob; /// The module's schedule store: `CronDocuments` over the host's document port. pub(super) struct Store(CronDocuments); @@ -155,6 +89,9 @@ impl Store { ), Operation::DeleteQueuedRuns { job_id } => dispatch!(delete_queued_runs, &job_id), Operation::ListRuns { job_id, limit } => dispatch!(list_runs, &job_id, limit), + Operation::SetMaxActiveJobs { limit } => dispatch!(set_max_active_jobs, limit), + Operation::ImportJob { job } => dispatch!(import_job, &job), + Operation::ImportRun { run } => dispatch!(import_run, &run), }) } } diff --git a/crates/tinyflows-module/src/schedule_store_tests.rs b/crates/tinyflows-module/src/schedule_store_tests.rs index b07000f0..c551ae88 100644 --- a/crates/tinyflows-module/src/schedule_store_tests.rs +++ b/crates/tinyflows-module/src/schedule_store_tests.rs @@ -1,4 +1,5 @@ use super::*; +use tinyflows_schedule::{DeliveryStatus, Schedule}; use async_trait::async_trait; use tinyflows_schedule::runtime::*; use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; diff --git a/crates/tinyflows-schedule/src/lib.rs b/crates/tinyflows-schedule/src/lib.rs index b11d59e7..e40e6c95 100644 --- a/crates/tinyflows-schedule/src/lib.rs +++ b/crates/tinyflows-schedule/src/lib.rs @@ -24,7 +24,8 @@ pub use schedule::{ runs_closer_than, schedule_cron_expression, validate_agent_schedule, validate_schedule, }; pub use store::{ - AgentJobSpec, MAX_CRON_OUTPUT_BYTES, TRUNCATED_OUTPUT_MARKER, check_patch, truncate_cron_output, + AgentJobSpec, ImportedJob, MAX_CRON_OUTPUT_BYTES, TRUNCATED_OUTPUT_MARKER, check_patch, + truncate_cron_output, }; pub use types::{ ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, diff --git a/crates/tinyflows-schedule/src/store.rs b/crates/tinyflows-schedule/src/store.rs index 35b022ec..740753ce 100644 --- a/crates/tinyflows-schedule/src/store.rs +++ b/crates/tinyflows-schedule/src/store.rs @@ -10,7 +10,9 @@ use crate::{CronJob, CronJobPatch, JobType}; -pub use tinyflows_bus::schedule::{AgentJobSpec, MAX_CRON_OUTPUT_BYTES, TRUNCATED_OUTPUT_MARKER}; +pub use tinyflows_bus::schedule::{ + AgentJobSpec, ImportedJob, MAX_CRON_OUTPUT_BYTES, TRUNCATED_OUTPUT_MARKER, +}; /// Refuses a patch no store may apply to `job`. /// From 6df87402e31bdeb27239349a6bfb0d9d7e70aea9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:08:38 +0300 Subject: [PATCH 09/19] test(schedule): add member tests module and share test helpers Wires a new schedule_members_tests module into the service and widens the visibility of the existing test state, connection helper, and document callbacks so the new tests can reuse them. Also reorders imports in the store tests to satisfy formatting. Auto-committed-on: dragonfly --- .../src/schedule_members_tests.rs | 298 ++++++++++++++++++ .../tinyflows-module/src/schedule_service.rs | 3 + .../src/schedule_service_tests.rs | 6 +- .../src/schedule_store_tests.rs | 2 +- 4 files changed, 305 insertions(+), 4 deletions(-) create mode 100644 crates/tinyflows-module/src/schedule_members_tests.rs diff --git a/crates/tinyflows-module/src/schedule_members_tests.rs b/crates/tinyflows-module/src/schedule_members_tests.rs new file mode 100644 index 00000000..7cb6e963 --- /dev/null +++ b/crates/tinyflows-module/src/schedule_members_tests.rs @@ -0,0 +1,298 @@ +//! Every contract-v3 pure member, called through the bus like a host would. +use super::tests::{connections, document_callbacks}; +use super::*; +use serde_json::json; +use tinyflows_bus::schedule::{self, members}; + +async fn proxy() -> (tinybus::Proxy, Connection) { + let (_server, host, _) = connections().await; + let proxy = host + .proxy( + schedule::BUS_NAME, + schedule::OBJECT_PATH, + schedule::INTERFACE, + ) + .unwrap(); + // Keep the server connection alive for the duration of the test. + std::mem::forget(_server); + (proxy, host) +} + +#[tokio::test] +async fn schedule_maths_members_match_the_in_process_functions() { + let (proxy, _host) = proxy().await; + let every = json!({"kind":"every","every_ms":600000}); + let next: chrono::DateTime = proxy + .call( + members::NEXT_RUN, + &(NextRunRequest { + schedule: serde_json::from_value(every.clone()).unwrap(), + from: "2026-01-01T00:00:00Z".parse().unwrap(), + },), + ) + .await + .unwrap(); + assert_eq!(next.to_rfc3339(), "2026-01-01T00:10:00+00:00"); + + let bad: std::result::Result, _> = proxy + .call( + members::NEXT_RUN, + &(json!({"schedule":{"kind":"cron","expr":"not a cron"},"from":"2026-01-01T00:00:00Z"}),), + ) + .await; + assert!(bad.is_err()); + + let normalized: String = proxy + .call(members::NORMALIZE_EXPRESSION, &("0 9 * * 1",)) + .await + .unwrap(); + assert_eq!( + normalized, + maths::normalize_expression("0 9 * * 1").unwrap() + ); + assert!( + proxy + .call::(members::NORMALIZE_EXPRESSION, &("nope",)) + .await + .is_err() + ); + + let expr: Option = proxy + .call( + members::SCHEDULE_CRON_EXPRESSION, + &(json!({"kind":"cron","expr":"0 9 * * *"}),), + ) + .await + .unwrap(); + assert_eq!(expr.as_deref(), Some("0 9 * * *")); + let none: Option = proxy + .call(members::SCHEDULE_CRON_EXPRESSION, &(every,)) + .await + .unwrap(); + assert_eq!(none, None); +} + +#[tokio::test] +async fn validation_members_enforce_the_cadence_floor_only_for_agents() { + let (proxy, _host) = proxy().await; + let request = |agent: bool| json!({"schedule":{"kind":"every","every_ms":60000},"now":"2026-01-01T00:00:00Z","agent":agent}); + proxy + .call::<()>(members::VALIDATE_SCHEDULE, &(request(false),)) + .await + .unwrap(); + let error = proxy + .call::<()>(members::VALIDATE_SCHEDULE, &(request(true),)) + .await + .unwrap_err(); + assert!(error.to_string().contains("minute"), "{error}"); + + let closest: Option = proxy + .call( + members::RUNS_CLOSER_THAN, + &(json!({"schedule":{"kind":"cron","expr":"*/7 * * * *"},"from":"2026-01-01T00:00:00Z","min_gap_ms":300000}),), + ) + .await + .unwrap(); + assert_eq!(closest.unwrap().gap(), chrono::Duration::minutes(4)); + let fixed: Option = proxy + .call( + members::RUNS_CLOSER_THAN, + &(json!({"schedule":{"kind":"every","every_ms":1000},"from":"2026-01-01T00:00:00Z","min_gap_ms":300000}),), + ) + .await + .unwrap(); + assert_eq!( + fixed, + Some(TooFrequent::FixedInterval(chrono::Duration::seconds(1))) + ); + let fine: Option = proxy + .call( + members::RUNS_CLOSER_THAN, + &(json!({"schedule":{"kind":"every","every_ms":3600000},"from":"2026-01-01T00:00:00Z","min_gap_ms":300000}),), + ) + .await + .unwrap(); + assert_eq!(fine, None); +} + +#[tokio::test] +async fn delivery_and_output_members_apply_host_independent_policy() { + let (proxy, _host) = proxy().await; + let origin = json!({"kind":"web","thread_id":"t1"}); + let with_origin: DeliveryConfig = proxy + .call(members::DEFAULT_DELIVERY, &(Some(origin.clone()),)) + .await + .unwrap(); + assert_eq!(with_origin.mode, "origin"); + let without: DeliveryConfig = proxy + .call(members::DEFAULT_DELIVERY, &(None::,)) + .await + .unwrap(); + assert_eq!(without.mode, "proactive"); + + let needs_origin = json!({"session_target":"current","delivery":{"mode":"none"}}); + assert!( + proxy + .call::<()>(members::CHECK_ORIGIN_REQUIREMENTS, &(needs_origin,)) + .await + .is_err() + ); + proxy + .call::<()>( + members::CHECK_ORIGIN_REQUIREMENTS, + &(json!({"session_target":"current","delivery":{"mode":"origin"},"origin":origin}),), + ) + .await + .unwrap(); + + let long = "x".repeat(schedule::MAX_CRON_OUTPUT_BYTES + 10); + let cut: String = proxy + .call(members::TRUNCATE_CRON_OUTPUT, &(long,)) + .await + .unwrap(); + assert_eq!(cut.len(), schedule::MAX_CRON_OUTPUT_BYTES); + assert!(cut.ends_with(schedule::TRUNCATED_OUTPUT_MARKER)); + let short: String = proxy + .call(members::TRUNCATE_CRON_OUTPUT, &("ok",)) + .await + .unwrap(); + assert_eq!(short, "ok"); +} + +#[tokio::test] +async fn check_patch_refuses_retargeting_a_flow_schedule_job() { + let (proxy, _host) = proxy().await; + let job = json!({"id":"flow:f","expression":"","schedule":{"kind":"every","every_ms":60000},"command":"f","prompt":null,"name":null,"job_type":"flow","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}); + proxy + .call::<()>( + members::CHECK_PATCH, + &(json!({"job":job,"patch":{"command":"f","enabled":false}}),), + ) + .await + .unwrap(); + let error = proxy + .call::<()>( + members::CHECK_PATCH, + &(json!({"job":job,"patch":{"command":"other"}}),), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("cannot be changed"), "{error}"); +} + +#[tokio::test] +async fn store_admission_cap_and_import_operations_are_served() { + let (proxy, host) = proxy().await; + host.serve_at( + "/test/Documents".try_into().unwrap(), + document_callbacks::Documents { + docs: { + use tinystoragedrivers_core::StorageBackend; + tinystoragedrivers_core::MemoryStorage::new() + .for_scope(&tinystoragedrivers_core::Scope::local()) + .unwrap() + .documents() + .clone() + }, + conflict_once: Arc::new(std::sync::atomic::AtomicBool::new(false)), + atomic_calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + }, + ) + .await + .unwrap(); + let storage = json!({"backend":{"kind":"documents","callback":{"destination":"test.ScheduleHost","object_path":"/test/Documents"}}}); + let call = |operation: schedule::StoreOperation| json!({"storage":storage,"operation":serde_json::to_value(operation).unwrap()}); + let job: CronJob = serde_json::from_value(json!({"id":"imp-1","expression":"* * * * *","schedule":{"kind":"every","every_ms":3600000},"command":"echo hi","prompt":null,"name":"imported","job_type":"shell","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-01T01:00:00Z","last_run":null,"last_status":null,"last_output":null})).unwrap(); + let imported: schedule::ImportedJob = serde_json::from_value( + proxy + .call::( + members::SCHEDULE_STORE, + &(call(schedule::StoreOperation::ImportJob { + job: job.clone(), + }),), + ) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!( + imported, + schedule::ImportedJob { + id: "imp-1".into(), + written: true + } + ); + let again: schedule::ImportedJob = serde_json::from_value( + proxy + .call::( + members::SCHEDULE_STORE, + &(call(schedule::StoreOperation::ImportJob { job }),), + ) + .await + .unwrap(), + ) + .unwrap(); + assert!(!again.written); + + let run: schedule::CronRun = serde_json::from_value(json!({"id":1,"job_id":"imp-1","started_at":"2026-01-01T00:00:00Z","finished_at":"2026-01-01T00:00:02Z","status":"ok","output":"hi","duration_ms":2000})).unwrap(); + let written: bool = serde_json::from_value( + proxy + .call::( + members::SCHEDULE_STORE, + &(call(schedule::StoreOperation::ImportRun { run }),), + ) + .await + .unwrap(), + ) + .unwrap(); + assert!(written); + + // Admission cap: the memory backend supports transactions, so a cap of + // one is accepted and the existing job counts against it. + let _: Value = proxy + .call( + members::SCHEDULE_STORE, + &(call(schedule::StoreOperation::SetMaxActiveJobs { + limit: Some(1), + }),), + ) + .await + .unwrap(); + let refused = proxy + .call::( + members::SCHEDULE_STORE, + &(call(schedule::StoreOperation::AddShell { + name: None, + schedule: serde_json::from_value(json!({"kind":"every","every_ms":3600000})) + .unwrap(), + command: "echo two".into(), + }),), + ) + .await; + assert!(refused.is_err(), "second enabled job must exceed the cap"); + let _: Value = proxy + .call( + members::SCHEDULE_STORE, + &(call(schedule::StoreOperation::SetMaxActiveJobs { + limit: None, + }),), + ) + .await + .unwrap(); +} + +#[tokio::test] +async fn the_served_surface_is_exactly_the_advertised_member_set() { + let (proxy, _host) = proxy().await; + // Every advertised member must at least be dispatched (not "unknown member"). + for name in schedule::METHODS { + let result = proxy.call::(name, &(json!({}),)).await; + if let Err(error) = result { + let text = error.to_string().to_lowercase(); + assert!( + !text.contains("unknown method") && !text.contains("no such method"), + "{name}: {error}" + ); + } + } +} diff --git a/crates/tinyflows-module/src/schedule_service.rs b/crates/tinyflows-module/src/schedule_service.rs index aa31713a..667ce90b 100644 --- a/crates/tinyflows-module/src/schedule_service.rs +++ b/crates/tinyflows-module/src/schedule_service.rs @@ -238,6 +238,9 @@ pub(super) async fn setup(connection: &Connection) -> Result<()> { Ok(()) } +#[cfg(test)] +#[path = "schedule_members_tests.rs"] +mod members_tests; #[cfg(test)] #[path = "schedule_service_tests.rs"] mod tests; diff --git a/crates/tinyflows-module/src/schedule_service_tests.rs b/crates/tinyflows-module/src/schedule_service_tests.rs index 8551f0f0..2ad165de 100644 --- a/crates/tinyflows-module/src/schedule_service_tests.rs +++ b/crates/tinyflows-module/src/schedule_service_tests.rs @@ -4,7 +4,7 @@ use tinyflows_bus::schedule::CALLBACK_INTERFACE; use tokio::sync::Mutex; #[derive(Default)] -struct State { +pub(super) struct State { records: Mutex>, completions: Mutex>, executions: Mutex>, @@ -43,7 +43,7 @@ impl Callbacks { } } -async fn connections() -> (Connection, Connection, Arc) { +pub(super) async fn connections() -> (Connection, Connection, Arc) { let bus = tinybus::transport::memory::MemoryBus::new(); tinybus::broker::Broker::new().spawn(bus.clone()); let server = Connection::connect(bus.connect().await.unwrap()) @@ -210,7 +210,7 @@ async fn module_job_storage_preserves_policy_updates_and_run_history() { } #[path = "../tests/support/document_callbacks.rs"] -mod document_callbacks; +pub(super) mod document_callbacks; #[tokio::test] async fn module_document_store_retries_policy_migration_conflicts_without_losing_seeded_jobs() { diff --git a/crates/tinyflows-module/src/schedule_store_tests.rs b/crates/tinyflows-module/src/schedule_store_tests.rs index c551ae88..20c8bb44 100644 --- a/crates/tinyflows-module/src/schedule_store_tests.rs +++ b/crates/tinyflows-module/src/schedule_store_tests.rs @@ -1,7 +1,7 @@ use super::*; -use tinyflows_schedule::{DeliveryStatus, Schedule}; use async_trait::async_trait; use tinyflows_schedule::runtime::*; +use tinyflows_schedule::{DeliveryStatus, Schedule}; use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; struct Replacement(CronDocuments); From 6484f253efe5791174957bf24c1e984b0771b9e9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:08:52 +0300 Subject: [PATCH 10/19] test(schedule): return server connection from proxy helper The proxy test helper now hands the server connection back to callers instead of leaking it with mem::forget, so each test keeps it alive for its own duration and the connection is dropped normally at the end. Auto-committed-on: dragonfly --- .../src/schedule_members_tests.rs | 20 +++++++++---------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/crates/tinyflows-module/src/schedule_members_tests.rs b/crates/tinyflows-module/src/schedule_members_tests.rs index 7cb6e963..545634e8 100644 --- a/crates/tinyflows-module/src/schedule_members_tests.rs +++ b/crates/tinyflows-module/src/schedule_members_tests.rs @@ -4,8 +4,8 @@ use super::*; use serde_json::json; use tinyflows_bus::schedule::{self, members}; -async fn proxy() -> (tinybus::Proxy, Connection) { - let (_server, host, _) = connections().await; +async fn proxy() -> (tinybus::Proxy, Connection, Connection) { + let (server, host, _) = connections().await; let proxy = host .proxy( schedule::BUS_NAME, @@ -13,14 +13,12 @@ async fn proxy() -> (tinybus::Proxy, Connection) { schedule::INTERFACE, ) .unwrap(); - // Keep the server connection alive for the duration of the test. - std::mem::forget(_server); - (proxy, host) + (proxy, host, server) } #[tokio::test] async fn schedule_maths_members_match_the_in_process_functions() { - let (proxy, _host) = proxy().await; + let (proxy, _host, _server) = proxy().await; let every = json!({"kind":"every","every_ms":600000}); let next: chrono::DateTime = proxy .call( @@ -74,7 +72,7 @@ async fn schedule_maths_members_match_the_in_process_functions() { #[tokio::test] async fn validation_members_enforce_the_cadence_floor_only_for_agents() { - let (proxy, _host) = proxy().await; + let (proxy, _host, _server) = proxy().await; let request = |agent: bool| json!({"schedule":{"kind":"every","every_ms":60000},"now":"2026-01-01T00:00:00Z","agent":agent}); proxy .call::<()>(members::VALIDATE_SCHEDULE, &(request(false),)) @@ -117,7 +115,7 @@ async fn validation_members_enforce_the_cadence_floor_only_for_agents() { #[tokio::test] async fn delivery_and_output_members_apply_host_independent_policy() { - let (proxy, _host) = proxy().await; + let (proxy, _host, _server) = proxy().await; let origin = json!({"kind":"web","thread_id":"t1"}); let with_origin: DeliveryConfig = proxy .call(members::DEFAULT_DELIVERY, &(Some(origin.clone()),)) @@ -161,7 +159,7 @@ async fn delivery_and_output_members_apply_host_independent_policy() { #[tokio::test] async fn check_patch_refuses_retargeting_a_flow_schedule_job() { - let (proxy, _host) = proxy().await; + let (proxy, _host, _server) = proxy().await; let job = json!({"id":"flow:f","expression":"","schedule":{"kind":"every","every_ms":60000},"command":"f","prompt":null,"name":null,"job_type":"flow","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}); proxy .call::<()>( @@ -182,7 +180,7 @@ async fn check_patch_refuses_retargeting_a_flow_schedule_job() { #[tokio::test] async fn store_admission_cap_and_import_operations_are_served() { - let (proxy, host) = proxy().await; + let (proxy, host, _server) = proxy().await; host.serve_at( "/test/Documents".try_into().unwrap(), document_callbacks::Documents { @@ -283,7 +281,7 @@ async fn store_admission_cap_and_import_operations_are_served() { #[tokio::test] async fn the_served_surface_is_exactly_the_advertised_member_set() { - let (proxy, _host) = proxy().await; + let (proxy, _host, _server) = proxy().await; // Every advertised member must at least be dispatched (not "unknown member"). for name in schedule::METHODS { let result = proxy.call::(name, &(json!({}),)).await; From 51fda3addf49ac8e4ccf140fa0feab74d1dd113b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:08:59 +0300 Subject: [PATCH 11/19] test(schedule): add probe for unknown member Adds a temporary test that calls a bogus member name and panics with the resulting error, to inspect how the proxy reports unknown members. Auto-committed-on: dragonfly --- crates/tinyflows-module/src/schedule_members_tests.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/crates/tinyflows-module/src/schedule_members_tests.rs b/crates/tinyflows-module/src/schedule_members_tests.rs index 545634e8..0616e80d 100644 --- a/crates/tinyflows-module/src/schedule_members_tests.rs +++ b/crates/tinyflows-module/src/schedule_members_tests.rs @@ -294,3 +294,10 @@ async fn the_served_surface_is_exactly_the_advertised_member_set() { } } } + +#[tokio::test] +async fn probe_unknown() { + let (proxy, _h, _s) = proxy().await; + let e = proxy.call::("Bogus", &(json!({}),)).await.unwrap_err(); + panic!("{e}"); +} From 981c8e5fbff93022623863c90bf270f813907e1e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:09:11 +0300 Subject: [PATCH 12/19] test(schedule): tighten unknown member assertion Fold the standalone probe into the served surface test and assert on the UnknownMethod error directly instead of matching lowercased message text. Auto-committed-on: dragonfly --- .../tinyflows-module/src/schedule_members_tests.rs | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/crates/tinyflows-module/src/schedule_members_tests.rs b/crates/tinyflows-module/src/schedule_members_tests.rs index 0616e80d..addcad47 100644 --- a/crates/tinyflows-module/src/schedule_members_tests.rs +++ b/crates/tinyflows-module/src/schedule_members_tests.rs @@ -283,21 +283,15 @@ async fn store_admission_cap_and_import_operations_are_served() { async fn the_served_surface_is_exactly_the_advertised_member_set() { let (proxy, _host, _server) = proxy().await; // Every advertised member must at least be dispatched (not "unknown member"). + let unknown = proxy.call::("Bogus", &(json!({}),)).await; + assert!(unknown.unwrap_err().to_string().contains("UnknownMethod")); for name in schedule::METHODS { let result = proxy.call::(name, &(json!({}),)).await; if let Err(error) = result { - let text = error.to_string().to_lowercase(); assert!( - !text.contains("unknown method") && !text.contains("no such method"), + !error.to_string().contains("UnknownMethod"), "{name}: {error}" ); } } } - -#[tokio::test] -async fn probe_unknown() { - let (proxy, _h, _s) = proxy().await; - let e = proxy.call::("Bogus", &(json!({}),)).await.unwrap_err(); - panic!("{e}"); -} From f90537132124ee9e91b309a63cecefe1ca5ae9ee Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:10:12 +0300 Subject: [PATCH 13/19] docs(schedule): document schedule bus contract v3 The schedule wire records, callback records and agent job spec now live in tinyflows_bus::schedule, so a host can implement the scheduler through tinyflows-bus alone without linking tinyflows-schedule or redeclaring types. The docs describe the new pure members, the added ScheduleStore operations and the CONTRACT_VERSION bump to 3, with tinyflows-schedule re-exporting everything at its old paths and the JSON wire shape unchanged. Auto-committed-on: dragonfly --- CHANGELOG.md | 19 +++++++++++++++++++ crates/tinyflows-module/README.md | 25 +++++++++++++++++++++++++ docs/tinyflows-bus-design.md | 26 +++++++++++++++++++++++++- 3 files changed, 69 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f44f17a9..d8427e13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Schedule bus contract v3: hosts need only `tinyflows-bus` + +- **Moved** the schedule wire records (`Schedule`, `CronJob`, `CronRun`, + `CronJobPatch`, `JobOrigin`, `DeliveryConfig`, `DeliveryStatus`, `JobPolicy`, + `JobType`, `SessionTarget`, `ActiveHours`, `delivery_mode`), `AgentJobSpec`, + and the callback records `Attempt`/`RunRecord`/`Completion` from + `tinyflows-schedule` into `tinyflows_bus::schedule` (`tinyflows-bus` gains + `chrono`). `tinyflows-schedule` re-exports them at the same paths; the JSON + wire shape is unchanged (`wire_tests` pass unchanged and now also pin the bus). + The two `Schedule` deserializer `debug` logs were dropped with the move. +- **Added** the typed `StoreOperation`, `ImportedJob`, `TooFrequent`, + `MIN_AGENT_JOB_INTERVAL` and request types to the contract, and `members::*` + constants for every Schedule member. +- **Added** module members `NextRun`, `ValidateSchedule`, `RunsCloserThan`, + `NormalizeExpression`, `ScheduleCronExpression`, `DefaultDelivery`, + `CheckOriginRequirements`, `CheckPatch`, `TruncateCronOutput`, and + `ScheduleStore` operations `set_max_active_jobs`, `import_job`, `import_run`. +- **Changed** Schedule `CONTRACT_VERSION` 2 -> 3. + ### Script spawner seam (`host-caps`) `caps::host::script` now launches every `code`/`shell` script through a diff --git a/crates/tinyflows-module/README.md b/crates/tinyflows-module/README.md index c5f4ae2c..99d67417 100644 --- a/crates/tinyflows-module/README.md +++ b/crates/tinyflows-module/README.md @@ -53,6 +53,31 @@ existing job/run JSON; mutation-only operations return null. | record_run | job_id, started_at, finished_at (RFC3339), status, output (optional), duration_ms, delivery_status (optional) | null | | delete_queued_runs | job_id | removed count | | list_runs | job_id, limit | runs | +| set_max_active_jobs | limit (number, or null to clear) | null | +| import_job | job (CronJob JSON) | `{id, written}` (an existing id is left untouched) | +| import_run | run (CronRun JSON) | written (bool) | + +`tinyflows_bus::schedule::StoreOperation` is the typed form of `operation`; the +job, run, patch, spec and callback records (`Attempt`, `RunRecord`, `Completion`) +are in `tinyflows_bus::schedule`, so a host implements ScheduleHost and the +document port without redeclaring them. + +### Pure members (contract v3) + +Each takes one JSON argument and returns JSON; failures are bus errors carrying +the author-facing message. + +| Member | Argument | Reply | +| --- | --- | --- | +| NextRun | `{schedule, from}` | RFC3339 instant after `from` | +| ValidateSchedule | `{schedule, now, agent?}`; `agent` adds the 5 minute floor | null | +| RunsCloserThan | `{schedule, from, min_gap_ms}` | `TooFrequent` or null | +| NormalizeExpression | cron expression string | normalised expression | +| ScheduleCronExpression | schedule | expression string or null | +| DefaultDelivery | origin or null | `DeliveryConfig` (`origin` with an origin, else `proactive`) | +| CheckOriginRequirements | `{session_target, delivery, origin?}` | null | +| CheckPatch | `{job, patch}`; refuses re-targeting a flow job | null | +| TruncateCronOutput | output string | output bounded to 16 KiB | A document callback serves `ai.tinyhumans.tinyflows.ScheduleDocuments.Document` with a JSON request and returns `{value: JSON}` or `{error: {kind, message}}`. diff --git a/docs/tinyflows-bus-design.md b/docs/tinyflows-bus-design.md index f31b8c64..43506ccf 100644 --- a/docs/tinyflows-bus-design.md +++ b/docs/tinyflows-bus-design.md @@ -15,7 +15,7 @@ interface per surface, each with its own `CONTRACT_VERSION`: | Surface | Status | Members | | --- | --- | --- | | Authoring | served, v1 | `ValidateGraph`, `BindingFailures`, `AuthoringFailures`, `UnresolvableToolArgs`, `BuilderFailures` (migrate, compat, graph ops gates, preflight) | -| Schedule | served, v2 | scheduler start/stop/tick/drain, `ExecuteJob`, delay and delivery validation | +| Schedule | served, v3 | scheduler start/stop/tick/drain, `ExecuteJob`, `ScheduleStore`, delay and delivery validation, schedule maths and validation (`NextRun`, `ValidateSchedule`, `RunsCloserThan`, `NormalizeExpression`, `ScheduleCronExpression`), `DefaultDelivery`, `CheckOriginRequirements`, `CheckPatch`, `TruncateCronOutput` | | Runs | skeleton, v1 | `StartRun`, `ResumeRun`, `CancelRun`, `PollRun` | | Catalog | skeleton, v1 | list/get/put/delete workflows, `ListRuns`, `NodeKinds`, `ApplyGraphOps`, `ApprovalManifest` | @@ -74,6 +74,30 @@ the host reads `PollRun`. Calls are blocking from the module's executor thread. A callback failure surfaces as the node's error; approval callbacks fail closed (an error is a rejection, never an approval). +## Schedule contract v3: typed records and pure members + +A host reaches the scheduler through `tinyflows-bus` alone: it links neither +`tinyflows-schedule` nor `tinyflows-drivers`, and redeclares no type. v3 makes +that possible: + +- The wire records live in `tinyflows_bus::schedule::types` (`Schedule`, + `CronJob`, `CronRun`, `CronJobPatch`, `JobOrigin`, `DeliveryConfig`, + `DeliveryStatus`, `JobPolicy`, `JobType`, `SessionTarget`, `ActiveHours`, + `delivery_mode`), beside `AgentJobSpec`, `ImportedJob`, the callback records + `Attempt`/`RunRecord`/`Completion`, `StoreOperation` (the typed `op` enum), + `TooFrequent` and the constants `MIN_AGENT_JOB_INTERVAL`, + `MAX_CRON_OUTPUT_BYTES`, `TRUNCATED_OUTPUT_MARKER`. `tinyflows-schedule` + re-exports every one at its old path. JSON shapes are byte-identical, pinned + by `wire_tests` in both crates. Adds `chrono` (no I/O) to the contract. +- New members (algorithms that need the `cron` crate run in the module): + `NextRun(NextRunRequest)`, `ValidateSchedule(ValidateScheduleRequest)` + (`agent: true` adds the cadence floor), `RunsCloserThan(RunsCloserThanRequest)`, + `NormalizeExpression(string)`, `ScheduleCronExpression(Schedule)`, + `DefaultDelivery(Option)`, `CheckOriginRequirements(CheckOriginRequest)`, + `CheckPatch(CheckPatchRequest)`, `TruncateCronOutput(string)`. +- New `ScheduleStore` operations: `set_max_active_jobs`, `import_job`, + `import_run`. + ## Storage-port decision (`tinyflows-sqlite`) Decision: the module does not own a SQLite backend. Schedule contract v2 From 2eecfc6385ee4576ea1ddff498d8504bbdc9c70d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:30:34 +0300 Subject: [PATCH 14/19] chore: files changed crates/tinyflows-module/Cargo.toml,crates/tinyflows-module/src/exports.rs,crate Auto-committed-on: dragonfly --- crates/tinyflows-module/Cargo.toml | 6 ++++++ crates/tinyflows-module/src/exports.rs | 22 +++++++++++++++++++--- crates/tinyflows-module/src/lib.rs | 4 ++++ 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/crates/tinyflows-module/Cargo.toml b/crates/tinyflows-module/Cargo.toml index 99aefd0f..3c5e5abd 100644 --- a/crates/tinyflows-module/Cargo.toml +++ b/crates/tinyflows-module/Cargo.toml @@ -28,6 +28,12 @@ tinybus = { version = ">=0.1.3, <0.2", default-features = false, features = ["ma tinybus-module = ">=0.1.3, <0.2" serde_json.workspace = true +[features] +default = [] +# Swap the C ABI exports for a Rust `linked_module()` so a host can link the +# module into its own executable (several linked modules can then coexist). +static-link = [] + [dev-dependencies] tempfile = "3" tokio.workspace = true diff --git a/crates/tinyflows-module/src/exports.rs b/crates/tinyflows-module/src/exports.rs index b660e8b0..bbedb309 100644 --- a/crates/tinyflows-module/src/exports.rs +++ b/crates/tinyflows-module/src/exports.rs @@ -5,11 +5,11 @@ use std::sync::OnceLock; use tinybus::module::{abi::*, manifest::*}; use tinybus::{BusName, InterfaceName, InterfaceVersion, MemberName, ObjectPath, Version}; -#[unsafe(no_mangle)] +#[cfg_attr(not(feature = "static-link"), unsafe(no_mangle))] pub static TINYBUS_MODULE_ABI_V1: TbAbiDescriptor = TbAbiDescriptor::current(env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION")); -#[unsafe(no_mangle)] +#[cfg_attr(not(feature = "static-link"), unsafe(no_mangle))] pub extern "C" fn tinybus_module_manifest_v1() -> TbSlice { std::panic::catch_unwind(manifest).unwrap_or(TbSlice { ptr: std::ptr::null(), @@ -75,10 +75,26 @@ fn manifest() -> TbSlice { /// /// # Safety /// `host` and `out` must satisfy the TinyBus ABI v1 pointer and lifetime contract. -#[unsafe(no_mangle)] +#[cfg_attr(not(feature = "static-link"), unsafe(no_mangle))] pub unsafe extern "C" fn tinybus_module_init_v1( host: *const TbHostVtable, out: *mut TbModuleVtable, ) -> i32 { unsafe { tinybus_module::start_module(host, out, 2, true, super::service::setup) } } + +/// Entry points and parsed manifest for a host that links this module instead +/// of loading the `cdylib` (local development and host unit tests). +/// +/// # Errors +/// Returns an error if the generated manifest is invalid. +#[cfg(feature = "static-link")] +pub fn linked_module() -> tinybus::Result { + unsafe { + tinybus::module::LinkedModule::from_exports( + &TINYBUS_MODULE_ABI_V1, + tinybus_module_manifest_v1, + tinybus_module_init_v1, + ) + } +} diff --git a/crates/tinyflows-module/src/lib.rs b/crates/tinyflows-module/src/lib.rs index dbefce73..82c6dcda 100644 --- a/crates/tinyflows-module/src/lib.rs +++ b/crates/tinyflows-module/src/lib.rs @@ -11,3 +11,7 @@ mod schedule_service; mod schedule_documents; mod schedule_store; + +#[cfg(feature = "static-link")] +pub use exports::linked_module; +pub use tinyflows_bus::{BUS_NAME, OBJECT_PATH}; From 2442cd76a1509052325cd6aecfe6fb6d6251f71d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:31:05 +0300 Subject: [PATCH 15/19] chore: files changed crates/tinyflows-module/src/schedule_callbacks.rs Auto-committed-on: dragonfly --- .../tinyflows-module/src/schedule_callbacks.rs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/crates/tinyflows-module/src/schedule_callbacks.rs b/crates/tinyflows-module/src/schedule_callbacks.rs index a7911019..c97d4858 100644 --- a/crates/tinyflows-module/src/schedule_callbacks.rs +++ b/crates/tinyflows-module/src/schedule_callbacks.rs @@ -8,6 +8,10 @@ use tinybus::Connection; use tinyflows_bus::schedule::{CALLBACK_INTERFACE, CallbackAddress, callbacks}; use tinyflows_schedule::{CronJob, DeliveryStatus, runtime::*}; +/// How long the host may take to run, report, deliver or trigger one job. +pub(super) const RUN_CALLBACK_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(2 * 60 * 60); + pub(super) struct HostCallbacks { connection: Connection, address: CallbackAddress, @@ -31,11 +35,21 @@ impl HostCallbacks { member: &str, args: &A, ) -> Result { - let proxy = self.connection.proxy( + let mut proxy = self.connection.proxy( &self.address.destination, &self.address.object_path, CALLBACK_INTERFACE, )?; + // Running, failing, delivering and triggering are the host doing the + // job's actual work (an agent turn, a shell command, a flow): the bus + // default of 30 seconds would cut every long run short and make the + // runtime retry work that is still going. + if matches!( + member, + callbacks::EXECUTE | callbacks::FAILED | callbacks::DELIVER | callbacks::SCHEDULE_TRIGGER + ) { + proxy = proxy.with_timeout(RUN_CALLBACK_TIMEOUT); + } Ok(proxy.call(member, args).await?) } } From b145107d7646557354bedb1aa59bb445cc8c5064 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:32:20 +0300 Subject: [PATCH 16/19] feat(import): add schedule-read feature exposing legacy cron reader The legacy `jobs.db` decoding moved out of `import_cron` into a new `legacy_cron` module that returns plain bus records, so a host reaching the scheduler through the bus can replay the file without linking the scheduler implementation. A new `schedule-read` feature gates it, and `ImportReport`'s `wrote` and `invalid` methods are now public so callers can build reports. Auto-committed-on: dragonfly --- Cargo.lock | 1 + crates/tinyflows-import/Cargo.toml | 7 +- crates/tinyflows-import/src/cron.rs | 287 ++------------------ crates/tinyflows-import/src/cron_tests.rs | 1 + crates/tinyflows-import/src/legacy_cron.rs | 291 +++++++++++++++++++++ crates/tinyflows-import/src/lib.rs | 4 + crates/tinyflows-import/src/report.rs | 4 +- 7 files changed, 325 insertions(+), 270 deletions(-) create mode 100644 crates/tinyflows-import/src/legacy_cron.rs diff --git a/Cargo.lock b/Cargo.lock index cb56b5c7..f2629359 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2099,6 +2099,7 @@ dependencies = [ "sha2 0.11.0", "tempfile", "tinyflows", + "tinyflows-bus", "tinyflows-catalog", "tinyflows-drivers", "tinyflows-schedule", diff --git a/crates/tinyflows-import/Cargo.toml b/crates/tinyflows-import/Cargo.toml index b69eb913..0fedf18e 100644 --- a/crates/tinyflows-import/Cargo.toml +++ b/crates/tinyflows-import/Cargo.toml @@ -30,11 +30,16 @@ store = ["engine", "tinyflows/store", "tinyflows-drivers/store"] # `import_flow_catalog`, `import_file_state`. catalog = ["engine", "dep:tinyflows-catalog", "tinyflows-drivers/catalog"] # `import_cron`. -schedule = ["dep:tinyflows-schedule", "tinyflows-drivers/schedule"] +schedule = ["schedule-read", "dep:tinyflows-schedule", "tinyflows-drivers/schedule"] +# `read_legacy_cron`: decodes `jobs.db` into bus records only, for a host that +# replays them through the scheduler module. Links neither the scheduler +# implementation nor the cron store. +schedule-read = ["dep:tinyflows-bus"] [dependencies] tinyflows = { workspace = true, optional = true } tinyflows-catalog = { version = "0.1.0", path = "../tinyflows-catalog", optional = true } +tinyflows-bus = { path = "../tinyflows-bus", optional = true } tinyflows-schedule = { version = "0.1.0", path = "../tinyflows-schedule", optional = true } tinyflows-drivers = { path = "../tinyflows-drivers", default-features = false } tinystoragedrivers-core = { git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.11.0" } diff --git a/crates/tinyflows-import/src/cron.rs b/crates/tinyflows-import/src/cron.rs index 130890f9..d0784853 100644 --- a/crates/tinyflows-import/src/cron.rs +++ b/crates/tinyflows-import/src/cron.rs @@ -1,23 +1,13 @@ -//! `jobs.db`: the cron store before it moved to documents. -//! -//! Two tables, `cron_jobs` and `cron_runs`. `cron_jobs` gained most of its -//! columns over time, so a database from before a column existed reads as if -//! the column held its default. An older `cron_job_policies` side table, which -//! the old store folded into `cron_jobs.retries` / `single_flight` on its next -//! open, is read the same way here. +//! `jobs.db`: the cron store before it moved to documents, copied into a +//! `CronDocuments` store. See `legacy_cron` for the reader. -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::path::Path; -use anyhow::{Context, Result, anyhow}; -use chrono::{DateTime, Utc}; +use anyhow::Result; use tinyflows_drivers::schedule::CronDocuments; -use tinyflows_schedule::{ - CronJob, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, JobPolicy, JobType, Schedule, - SessionTarget, -}; -use crate::legacy_db::{LegacyDb, column_or}; +use crate::legacy_cron::read_legacy_cron; use crate::report::ImportReport; /// The jobs and run history in `jobs_db` (the file the old store was opened @@ -33,33 +23,24 @@ use crate::report::ImportReport; /// example an active-job cap the jobs would exceed). A row that cannot be /// decoded is counted in the report as invalid. pub async fn import_cron(jobs_db: &Path, cron: &CronDocuments) -> Result { - let mut report = ImportReport::default(); - let Some(db) = LegacyDb::open(jobs_db)? else { - return Ok(report); - }; - let tables = db.tables().await?; - if !tables.contains("cron_jobs") { - return Ok(report); - } + let legacy = read_legacy_cron(jobs_db).await?; + let mut report = legacy.report; let mut filed_as = HashMap::new(); - for job in read_jobs(&db, &tables, &mut report).await? { - let source_id = job.id.clone(); - let imported = cron.import_job(&job).await?; + for job in &legacy.jobs { + let imported = cron.import_job(job).await?; report.wrote("jobs", imported.written); - filed_as.insert(source_id, imported.id); + filed_as.insert(job.id.clone(), imported.id); } - if tables.contains("cron_runs") { - for mut run in read_runs(&db, &mut report).await? { - let Some(id) = filed_as.get(&run.job_id) else { - report.invalid( - "runs", - format!("run {}: job '{}' was not imported", run.id, run.job_id), - ); - continue; - }; - run.job_id = id.clone(); - report.wrote("runs", cron.import_run(&run).await?); - } + for mut run in legacy.runs { + let Some(id) = filed_as.get(&run.job_id) else { + report.invalid( + "runs", + format!("run {}: job '{}' was not imported", run.id, run.job_id), + ); + continue; + }; + run.job_id = id.clone(); + report.wrote("runs", cron.import_run(&run).await?); } tracing::info!( target: "tinyflows_import", @@ -70,234 +51,6 @@ pub async fn import_cron(jobs_db: &Path, cron: &CronDocuments) -> Result, - job_type: String, - prompt: Option, - name: Option, - session_target: String, - model: Option, - enabled: i64, - delivery: Option, - delete_after_run: i64, - created_at: String, - next_run: String, - last_run: Option, - last_status: Option, - last_output: Option, - agent_id: Option, - origin: Option, - retries: Option, - single_flight: i64, - claim_generation: i64, -} - -async fn read_jobs( - db: &LegacyDb, - tables: &HashSet, - report: &mut ImportReport, -) -> Result> { - let have = db.columns("cron_jobs").await?; - let select = format!( - "SELECT id, expression, command, {}, {}, {}, {}, {}, {}, {}, {}, {}, created_at, next_run, \ - last_run, last_status, last_output, {}, {}, {}, {}, {} FROM cron_jobs ORDER BY created_at, id", - column_or(&have, "schedule", "NULL"), - column_or(&have, "job_type", "'shell'"), - column_or(&have, "prompt", "NULL"), - column_or(&have, "name", "NULL"), - column_or(&have, "session_target", "'isolated'"), - column_or(&have, "model", "NULL"), - column_or(&have, "enabled", "1"), - column_or(&have, "delivery", "NULL"), - column_or(&have, "delete_after_run", "0"), - column_or(&have, "agent_id", "NULL"), - column_or(&have, "origin", "NULL"), - column_or(&have, "retries", "NULL"), - column_or(&have, "single_flight", "0"), - column_or(&have, "claim_generation", "0"), - ); - let mut rows = db - .read("cron jobs", move |conn| { - let mut statement = conn.prepare(&select)?; - statement - .query_map([], |row| { - Ok(JobRow { - id: row.get(0)?, - expression: row.get(1)?, - command: row.get(2)?, - schedule: row.get(3)?, - job_type: row.get(4)?, - prompt: row.get(5)?, - name: row.get(6)?, - session_target: row.get(7)?, - model: row.get(8)?, - enabled: row.get(9)?, - delivery: row.get(10)?, - delete_after_run: row.get(11)?, - created_at: row.get(12)?, - next_run: row.get(13)?, - last_run: row.get(14)?, - last_status: row.get(15)?, - last_output: row.get(16)?, - agent_id: row.get(17)?, - origin: row.get(18)?, - retries: row.get(19)?, - single_flight: row.get(20)?, - claim_generation: row.get(21)?, - }) - })? - .collect::, _>>() - }) - .await?; - if tables.contains("cron_job_policies") { - let policies = read_policies(db).await?; - for row in &mut rows { - if let Some((retries, single_flight)) = policies.get(&row.id) { - row.retries = *retries; - row.single_flight = *single_flight; - } - } - } - let mut jobs = Vec::new(); - for row in rows { - let id = row.id.clone(); - match decode_job(row) { - Ok(job) => jobs.push(job), - Err(error) => report.invalid("jobs", format!("job '{id}': {error:#}")), - } - } - Ok(jobs) -} - -/// `cron_job_policies`: `job_id -> (retries, single_flight)`. -async fn read_policies(db: &LegacyDb) -> Result, i64)>> { - db.read("cron job policies", |conn| { - let mut statement = - conn.prepare("SELECT job_id, retries, single_flight FROM cron_job_policies")?; - statement - .query_map([], |row| { - Ok((row.get::<_, String>(0)?, (row.get(1)?, row.get(2)?))) - })? - .collect::, _>>() - }) - .await -} - -fn decode_job(row: JobRow) -> Result { - let schedule = match row.schedule.as_deref().map(str::trim) { - Some(raw) if !raw.is_empty() => serde_json::from_str(raw) - .with_context(|| format!("cron schedule JSON does not parse: {raw}"))?, - _ if row.expression.trim().is_empty() => { - return Err(anyhow!("missing schedule and legacy expression")); - } - // A job written before `schedule` existed carried only its expression. - _ => Schedule::Cron { - expr: row.expression.clone(), - tz: None, - active_hours: None, - }, - }; - let delivery = match row.delivery.as_deref().map(str::trim) { - Some(raw) if !raw.is_empty() => serde_json::from_str(raw) - .with_context(|| format!("cron delivery JSON does not parse: {raw}"))?, - _ => DeliveryConfig::default(), - }; - let origin = match row.origin.as_deref().map(str::trim) { - Some(raw) if !raw.is_empty() => Some( - serde_json::from_str::(raw) - .with_context(|| format!("cron origin JSON does not parse: {raw}"))?, - ), - _ => None, - }; - Ok(CronJob { - claim_generation: u64::try_from(row.claim_generation) - .context("claim generation is negative")?, - policy: JobPolicy { - retries: row - .retries - .map(|n| u32::try_from(n.max(0)).unwrap_or(u32::MAX)), - single_flight: row.single_flight != 0, - }, - id: row.id, - expression: row.expression, - schedule, - command: row.command, - job_type: JobType::parse(&row.job_type), - prompt: row.prompt, - name: row.name, - session_target: SessionTarget::parse(&row.session_target), - model: row.model, - agent_id: row.agent_id, - enabled: row.enabled != 0, - delivery, - delete_after_run: row.delete_after_run != 0, - created_at: instant(&row.created_at)?, - next_run: instant(&row.next_run)?, - last_run: row.last_run.as_deref().map(instant).transpose()?, - last_status: row.last_status, - last_output: row.last_output, - origin, - }) -} - -fn instant(raw: &str) -> Result> { - Ok(DateTime::parse_from_rfc3339(raw) - .with_context(|| format!("invalid RFC3339 timestamp: {raw}"))? - .with_timezone(&Utc)) -} - -async fn read_runs(db: &LegacyDb, report: &mut ImportReport) -> Result> { - let have = db.columns("cron_runs").await?; - let select = format!( - "SELECT id, job_id, started_at, finished_at, status, output, duration_ms, {} \ - FROM cron_runs ORDER BY id", - column_or(&have, "delivery_status", "NULL"), - ); - let rows = db - .read("cron runs", move |conn| { - let mut statement = conn.prepare(&select)?; - statement - .query_map([], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - row.get::<_, String>(4)?, - row.get::<_, Option>(5)?, - row.get::<_, Option>(6)?, - row.get::<_, Option>(7)?, - )) - })? - .collect::, _>>() - }) - .await?; - let mut runs = Vec::new(); - for (id, job_id, started_at, finished_at, status, output, duration_ms, delivery_status) in rows - { - match (instant(&started_at), instant(&finished_at)) { - (Ok(started_at), Ok(finished_at)) => runs.push(CronRun { - id, - job_id, - started_at, - finished_at, - status, - output, - duration_ms, - delivery_status: delivery_status.as_deref().and_then(DeliveryStatus::parse), - }), - (Err(error), _) | (_, Err(error)) => { - report.invalid("runs", format!("run {id}: {error:#}")); - } - } - } - Ok(runs) -} - #[cfg(test)] #[path = "cron_tests.rs"] mod tests; diff --git a/crates/tinyflows-import/src/cron_tests.rs b/crates/tinyflows-import/src/cron_tests.rs index 288eafd6..4b21ca2e 100644 --- a/crates/tinyflows-import/src/cron_tests.rs +++ b/crates/tinyflows-import/src/cron_tests.rs @@ -1,4 +1,5 @@ use super::*; +use tinyflows_bus::schedule::types::{CronJob, JobType}; use crate::test_support::{docs, expected, fixtures, json, tree_digest}; use serde_json::Value; use tinystoragedrivers_sqlite::rusqlite::Connection; diff --git a/crates/tinyflows-import/src/legacy_cron.rs b/crates/tinyflows-import/src/legacy_cron.rs new file mode 100644 index 00000000..03f8f749 --- /dev/null +++ b/crates/tinyflows-import/src/legacy_cron.rs @@ -0,0 +1,291 @@ +//! Reads the legacy `jobs.db` (the cron store before it moved to documents) into +//! plain bus records. It names only the transport vocabulary +//! (`tinyflows_bus::schedule::types`), so a host that reaches the scheduler +//! through the bus can replay the file through the module's +//! `import_job` / `import_run` store operations without linking the scheduler +//! implementation. +//! +//! Two tables, `cron_jobs` and `cron_runs`. `cron_jobs` gained most of its +//! columns over time, so a database from before a column existed reads as if +//! the column held its default. An older `cron_job_policies` side table, which +//! the old store folded into `cron_jobs.retries` / `single_flight` on its next +//! open, is read the same way here. + +use std::collections::HashSet; +use std::path::Path; + +use anyhow::{Context, Result, anyhow}; +use chrono::{DateTime, Utc}; +use std::collections::HashMap; +use tinyflows_bus::schedule::types::{ + CronJob, CronRun, DeliveryConfig, DeliveryStatus, JobOrigin, JobPolicy, JobType, Schedule, + SessionTarget, +}; + +use crate::legacy_db::{LegacyDb, column_or}; +use crate::report::ImportReport; + +/// What `jobs.db` held, decoded. +#[derive(Debug, Default)] +pub struct LegacyCron { + /// Every decodable job, oldest first. + pub jobs: Vec, + /// Every decodable run, in row order. Its `job_id` is the job's id in the + /// legacy file; a flow-schedule job is filed under another id by the + /// target store, so a caller remaps it from the id the import returns. + pub runs: Vec, + /// Rows that could not be decoded; also counted in the report as invalid. + pub report: ImportReport, +} + +/// Reads `jobs_db` (the file the old store was opened with). A missing file or +/// one without a `cron_jobs` table reads as empty. +/// +/// # Errors +/// +/// When the file exists but cannot be read. A row that cannot be decoded is +/// counted in the report as invalid. +pub async fn read_legacy_cron(jobs_db: &Path) -> Result { + let mut legacy = LegacyCron::default(); + let Some(db) = LegacyDb::open(jobs_db)? else { + return Ok(legacy); + }; + let tables = db.tables().await?; + if !tables.contains("cron_jobs") { + return Ok(legacy); + } + legacy.jobs = read_jobs(&db, &tables, &mut legacy.report).await?; + if tables.contains("cron_runs") { + legacy.runs = read_runs(&db, &mut legacy.report).await?; + } + Ok(legacy) +} + +/// One `cron_jobs` row with the raw text columns still raw. +struct JobRow { + id: String, + expression: String, + command: String, + schedule: Option, + job_type: String, + prompt: Option, + name: Option, + session_target: String, + model: Option, + enabled: i64, + delivery: Option, + delete_after_run: i64, + created_at: String, + next_run: String, + last_run: Option, + last_status: Option, + last_output: Option, + agent_id: Option, + origin: Option, + retries: Option, + single_flight: i64, + claim_generation: i64, +} + +async fn read_jobs( + db: &LegacyDb, + tables: &HashSet, + report: &mut ImportReport, +) -> Result> { + let have = db.columns("cron_jobs").await?; + let select = format!( + "SELECT id, expression, command, {}, {}, {}, {}, {}, {}, {}, {}, {}, created_at, next_run, \ + last_run, last_status, last_output, {}, {}, {}, {}, {} FROM cron_jobs ORDER BY created_at, id", + column_or(&have, "schedule", "NULL"), + column_or(&have, "job_type", "'shell'"), + column_or(&have, "prompt", "NULL"), + column_or(&have, "name", "NULL"), + column_or(&have, "session_target", "'isolated'"), + column_or(&have, "model", "NULL"), + column_or(&have, "enabled", "1"), + column_or(&have, "delivery", "NULL"), + column_or(&have, "delete_after_run", "0"), + column_or(&have, "agent_id", "NULL"), + column_or(&have, "origin", "NULL"), + column_or(&have, "retries", "NULL"), + column_or(&have, "single_flight", "0"), + column_or(&have, "claim_generation", "0"), + ); + let mut rows = db + .read("cron jobs", move |conn| { + let mut statement = conn.prepare(&select)?; + statement + .query_map([], |row| { + Ok(JobRow { + id: row.get(0)?, + expression: row.get(1)?, + command: row.get(2)?, + schedule: row.get(3)?, + job_type: row.get(4)?, + prompt: row.get(5)?, + name: row.get(6)?, + session_target: row.get(7)?, + model: row.get(8)?, + enabled: row.get(9)?, + delivery: row.get(10)?, + delete_after_run: row.get(11)?, + created_at: row.get(12)?, + next_run: row.get(13)?, + last_run: row.get(14)?, + last_status: row.get(15)?, + last_output: row.get(16)?, + agent_id: row.get(17)?, + origin: row.get(18)?, + retries: row.get(19)?, + single_flight: row.get(20)?, + claim_generation: row.get(21)?, + }) + })? + .collect::, _>>() + }) + .await?; + if tables.contains("cron_job_policies") { + let policies = read_policies(db).await?; + for row in &mut rows { + if let Some((retries, single_flight)) = policies.get(&row.id) { + row.retries = *retries; + row.single_flight = *single_flight; + } + } + } + let mut jobs = Vec::new(); + for row in rows { + let id = row.id.clone(); + match decode_job(row) { + Ok(job) => jobs.push(job), + Err(error) => report.invalid("jobs", format!("job '{id}': {error:#}")), + } + } + Ok(jobs) +} + +/// `cron_job_policies`: `job_id -> (retries, single_flight)`. +async fn read_policies(db: &LegacyDb) -> Result, i64)>> { + db.read("cron job policies", |conn| { + let mut statement = + conn.prepare("SELECT job_id, retries, single_flight FROM cron_job_policies")?; + statement + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, (row.get(1)?, row.get(2)?))) + })? + .collect::, _>>() + }) + .await +} + +fn decode_job(row: JobRow) -> Result { + let schedule = match row.schedule.as_deref().map(str::trim) { + Some(raw) if !raw.is_empty() => serde_json::from_str(raw) + .with_context(|| format!("cron schedule JSON does not parse: {raw}"))?, + _ if row.expression.trim().is_empty() => { + return Err(anyhow!("missing schedule and legacy expression")); + } + // A job written before `schedule` existed carried only its expression. + _ => Schedule::Cron { + expr: row.expression.clone(), + tz: None, + active_hours: None, + }, + }; + let delivery = match row.delivery.as_deref().map(str::trim) { + Some(raw) if !raw.is_empty() => serde_json::from_str(raw) + .with_context(|| format!("cron delivery JSON does not parse: {raw}"))?, + _ => DeliveryConfig::default(), + }; + let origin = match row.origin.as_deref().map(str::trim) { + Some(raw) if !raw.is_empty() => Some( + serde_json::from_str::(raw) + .with_context(|| format!("cron origin JSON does not parse: {raw}"))?, + ), + _ => None, + }; + Ok(CronJob { + claim_generation: u64::try_from(row.claim_generation) + .context("claim generation is negative")?, + policy: JobPolicy { + retries: row + .retries + .map(|n| u32::try_from(n.max(0)).unwrap_or(u32::MAX)), + single_flight: row.single_flight != 0, + }, + id: row.id, + expression: row.expression, + schedule, + command: row.command, + job_type: JobType::parse(&row.job_type), + prompt: row.prompt, + name: row.name, + session_target: SessionTarget::parse(&row.session_target), + model: row.model, + agent_id: row.agent_id, + enabled: row.enabled != 0, + delivery, + delete_after_run: row.delete_after_run != 0, + created_at: instant(&row.created_at)?, + next_run: instant(&row.next_run)?, + last_run: row.last_run.as_deref().map(instant).transpose()?, + last_status: row.last_status, + last_output: row.last_output, + origin, + }) +} + +fn instant(raw: &str) -> Result> { + Ok(DateTime::parse_from_rfc3339(raw) + .with_context(|| format!("invalid RFC3339 timestamp: {raw}"))? + .with_timezone(&Utc)) +} + +async fn read_runs(db: &LegacyDb, report: &mut ImportReport) -> Result> { + let have = db.columns("cron_runs").await?; + let select = format!( + "SELECT id, job_id, started_at, finished_at, status, output, duration_ms, {} \ + FROM cron_runs ORDER BY id", + column_or(&have, "delivery_status", "NULL"), + ); + let rows = db + .read("cron runs", move |conn| { + let mut statement = conn.prepare(&select)?; + statement + .query_map([], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, Option>(5)?, + row.get::<_, Option>(6)?, + row.get::<_, Option>(7)?, + )) + })? + .collect::, _>>() + }) + .await?; + let mut runs = Vec::new(); + for (id, job_id, started_at, finished_at, status, output, duration_ms, delivery_status) in rows + { + match (instant(&started_at), instant(&finished_at)) { + (Ok(started_at), Ok(finished_at)) => runs.push(CronRun { + id, + job_id, + started_at, + finished_at, + status, + output, + duration_ms, + delivery_status: delivery_status.as_deref().and_then(DeliveryStatus::parse), + }), + (Err(error), _) | (_, Err(error)) => { + report.invalid("runs", format!("run {id}: {error:#}")); + } + } + } + Ok(runs) +} + diff --git a/crates/tinyflows-import/src/lib.rs b/crates/tinyflows-import/src/lib.rs index b29c5fd7..40a72a7d 100644 --- a/crates/tinyflows-import/src/lib.rs +++ b/crates/tinyflows-import/src/lib.rs @@ -56,6 +56,8 @@ mod checkpoints; #[cfg(feature = "schedule")] mod cron; mod legacy_db; +#[cfg(feature = "schedule-read")] +mod legacy_cron; mod report; #[cfg(feature = "catalog")] mod state; @@ -71,6 +73,8 @@ pub use catalog::import_flow_catalog; pub use checkpoints::{import_file_checkpoints, import_sqlite_checkpoints}; #[cfg(feature = "schedule")] pub use cron::import_cron; +#[cfg(feature = "schedule-read")] +pub use legacy_cron::{LegacyCron, read_legacy_cron}; pub use report::{ImportReport, Tally}; #[cfg(feature = "catalog")] pub use state::import_file_state; diff --git a/crates/tinyflows-import/src/report.rs b/crates/tinyflows-import/src/report.rs index 2b2f426f..443b37cc 100644 --- a/crates/tinyflows-import/src/report.rs +++ b/crates/tinyflows-import/src/report.rs @@ -22,7 +22,7 @@ pub struct ImportReport { } impl ImportReport { - pub(crate) fn wrote(&mut self, kind: &'static str, written: bool) { + pub fn wrote(&mut self, kind: &'static str, written: bool) { let tally = self.kinds.entry(kind).or_default(); if written { tally.written += 1; @@ -32,7 +32,7 @@ impl ImportReport { } /// Records one record that could not be imported, with why. - pub(crate) fn invalid(&mut self, kind: &'static str, what: impl std::fmt::Display) { + pub fn invalid(&mut self, kind: &'static str, what: impl std::fmt::Display) { self.kinds.entry(kind).or_default().invalid += 1; let message = format!("{kind}: {what}"); tracing::warn!(target: "tinyflows_import", "[import] skipped an unreadable record: {message}"); From 6160d5ad0867ab0218a5662dbb82d3f21183c2c1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 14:32:25 +0300 Subject: [PATCH 17/19] feat(module): static-link entry, long run-callback timeouts, bus-typed legacy cron reader tinyflows-module: add a static-link feature exporting linked_module() so a host can link the module for tests; give the Execute/Failed/Deliver/ScheduleTrigger callbacks a 2h deadline instead of the bus default of 30s. tinyflows-import: add schedule-read / read_legacy_cron, which decodes jobs.db into tinyflows-bus records without linking the scheduler implementation. Co-authored-by: Medulla --- crates/tinyflows-import/src/cron_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows-import/src/cron_tests.rs b/crates/tinyflows-import/src/cron_tests.rs index 4b21ca2e..637635bb 100644 --- a/crates/tinyflows-import/src/cron_tests.rs +++ b/crates/tinyflows-import/src/cron_tests.rs @@ -1,5 +1,5 @@ use super::*; -use tinyflows_bus::schedule::types::{CronJob, JobType}; +use tinyflows_bus::schedule::types::JobType; use crate::test_support::{docs, expected, fixtures, json, tree_digest}; use serde_json::Value; use tinystoragedrivers_sqlite::rusqlite::Connection; From ccb79ef64175dedac91af036fef341f225b39fd6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 15:29:21 +0300 Subject: [PATCH 18/19] fix(schedule): reject elapsed one-shot schedules in next_run The next_run contract promises an occurrence strictly after the given instant, but the in-process maths returned a one-shot's instant unchanged, so an already-elapsed At schedule was handed back as the next run. The service now returns an error for those, with tests covering past, equal and future instants. Auto-committed-on: dragonfly --- crates/tinyflows-import/src/cron_tests.rs | 2 +- crates/tinyflows-import/src/legacy_cron.rs | 1 - crates/tinyflows-import/src/lib.rs | 2 +- .../tinyflows-module/src/schedule_callbacks.rs | 5 ++++- .../src/schedule_members_tests.rs | 18 ++++++++++++++++++ .../tinyflows-module/src/schedule_service.rs | 11 +++++++++++ crates/tinyflows-schedule/Cargo.toml | 2 +- docs/tinyflows-bus-design.md | 3 ++- 8 files changed, 38 insertions(+), 6 deletions(-) diff --git a/crates/tinyflows-import/src/cron_tests.rs b/crates/tinyflows-import/src/cron_tests.rs index 637635bb..a7c7decd 100644 --- a/crates/tinyflows-import/src/cron_tests.rs +++ b/crates/tinyflows-import/src/cron_tests.rs @@ -1,7 +1,7 @@ use super::*; -use tinyflows_bus::schedule::types::JobType; use crate::test_support::{docs, expected, fixtures, json, tree_digest}; use serde_json::Value; +use tinyflows_bus::schedule::types::JobType; use tinystoragedrivers_sqlite::rusqlite::Connection; fn jobs_db() -> std::path::PathBuf { diff --git a/crates/tinyflows-import/src/legacy_cron.rs b/crates/tinyflows-import/src/legacy_cron.rs index 03f8f749..056309d5 100644 --- a/crates/tinyflows-import/src/legacy_cron.rs +++ b/crates/tinyflows-import/src/legacy_cron.rs @@ -288,4 +288,3 @@ async fn read_runs(db: &LegacyDb, report: &mut ImportReport) -> Result>( + members::NEXT_RUN, + &(json!({"schedule":{"kind":"at","at":at},"from":"2026-01-01T00:00:00Z"}),), + ) + .await; + assert!(elapsed.is_err(), "{at} is not after from"); + } + let future: chrono::DateTime = proxy + .call( + members::NEXT_RUN, + &(json!({"schedule":{"kind":"at","at":"2026-01-02T00:00:00Z"},"from":"2026-01-01T00:00:00Z"}),), + ) + .await + .unwrap(); + assert_eq!(future.to_rfc3339(), "2026-01-02T00:00:00+00:00"); + let bad: std::result::Result, _> = proxy .call( members::NEXT_RUN, diff --git a/crates/tinyflows-module/src/schedule_service.rs b/crates/tinyflows-module/src/schedule_service.rs index 667ce90b..526adb6c 100644 --- a/crates/tinyflows-module/src/schedule_service.rs +++ b/crates/tinyflows-module/src/schedule_service.rs @@ -181,6 +181,17 @@ impl ScheduleService { } async fn next_run(&self, request: NextRunRequest) -> Result> { + // The contract promises an occurrence strictly after `from`; the + // in-process function returns a one-shot's instant unchanged, so an + // elapsed `At` is refused here instead of handed back as "next". + if let Schedule::At { at } = &request.schedule { + if *at <= request.from { + return Err(Error::failed(format!( + "one-shot schedule at {at} is not after {}", + request.from + ))); + } + } maths::next_run_for_schedule(&request.schedule, request.from).map_err(failure) } async fn validate_schedule(&self, request: ValidateScheduleRequest) -> Result<()> { diff --git a/crates/tinyflows-schedule/Cargo.toml b/crates/tinyflows-schedule/Cargo.toml index dfd4ddd6..626e3382 100644 --- a/crates/tinyflows-schedule/Cargo.toml +++ b/crates/tinyflows-schedule/Cargo.toml @@ -11,7 +11,7 @@ homepage.workspace = true readme = "../../README.md" [dependencies] -tinyflows-bus = { path = "../tinyflows-bus" } +tinyflows-bus = { version = "0.1.0", path = "../tinyflows-bus" } async-trait.workspace = true tokio.workspace = true uuid.workspace = true diff --git a/docs/tinyflows-bus-design.md b/docs/tinyflows-bus-design.md index 43506ccf..6a0610ba 100644 --- a/docs/tinyflows-bus-design.md +++ b/docs/tinyflows-bus-design.md @@ -83,7 +83,8 @@ that possible: - The wire records live in `tinyflows_bus::schedule::types` (`Schedule`, `CronJob`, `CronRun`, `CronJobPatch`, `JobOrigin`, `DeliveryConfig`, `DeliveryStatus`, `JobPolicy`, `JobType`, `SessionTarget`, `ActiveHours`, - `delivery_mode`), beside `AgentJobSpec`, `ImportedJob`, the callback records + `delivery_mode`). `tinyflows_bus::schedule` exports them together with + `AgentJobSpec`, `ImportedJob`, the callback records `Attempt`/`RunRecord`/`Completion`, `StoreOperation` (the typed `op` enum), `TooFrequent` and the constants `MIN_AGENT_JOB_INTERVAL`, `MAX_CRON_OUTPUT_BYTES`, `TRUNCATED_OUTPUT_MARKER`. `tinyflows-schedule` From 62c3e78c3223f6cf814efafc9df674e6e4c10bb5 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 16 Oct 2026 15:29:56 +0300 Subject: [PATCH 19/19] refactor(schedule_service): collapse nested if into let-chain The one-shot schedule guard now uses a let-chain instead of a nested if, which keeps the same validation behaviour while flattening the control flow. Auto-committed-on: dragonfly --- crates/tinyflows-module/src/schedule_service.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/crates/tinyflows-module/src/schedule_service.rs b/crates/tinyflows-module/src/schedule_service.rs index 526adb6c..ac83973c 100644 --- a/crates/tinyflows-module/src/schedule_service.rs +++ b/crates/tinyflows-module/src/schedule_service.rs @@ -184,13 +184,13 @@ impl ScheduleService { // The contract promises an occurrence strictly after `from`; the // in-process function returns a one-shot's instant unchanged, so an // elapsed `At` is refused here instead of handed back as "next". - if let Schedule::At { at } = &request.schedule { - if *at <= request.from { - return Err(Error::failed(format!( - "one-shot schedule at {at} is not after {}", - request.from - ))); - } + if let Schedule::At { at } = &request.schedule + && *at <= request.from + { + return Err(Error::failed(format!( + "one-shot schedule at {at} is not after {}", + request.from + ))); } maths::next_run_for_schedule(&request.schedule, request.from).map_err(failure) }