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/Cargo.lock b/Cargo.lock index 97422a3d..f2629359 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2032,6 +2032,7 @@ dependencies = [ name = "tinyflows-bus" version = "0.1.0" dependencies = [ + "chrono", "serde", "serde_json", ] @@ -2098,6 +2099,7 @@ dependencies = [ "sha2 0.11.0", "tempfile", "tinyflows", + "tinyflows-bus", "tinyflows-catalog", "tinyflows-drivers", "tinyflows-schedule", @@ -2140,6 +2142,7 @@ dependencies = [ "cron", "serde", "serde_json", + "tinyflows-bus", "tokio", "tracing", "uuid", 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/frequency.rs b/crates/tinyflows-bus/src/schedule/frequency.rs new file mode 100644 index 00000000..65a5374d --- /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. +pub 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.rs b/crates/tinyflows-bus/src/schedule/mod.rs similarity index 60% rename from crates/tinyflows-bus/src/schedule.rs rename to crates/tinyflows-bus/src/schedule/mod.rs index df2df955..ce53e458 100644 --- a/crates/tinyflows-bus/src/schedule.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, describe_gap}; +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,10 @@ 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; +#[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..67992335 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/mod_tests.rs @@ -0,0 +1,128 @@ +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()); +} 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, + }, +} diff --git a/crates/tinyflows-bus/src/schedule/types.rs b/crates/tinyflows-bus/src/schedule/types.rs new file mode 100644 index 00000000..251ffc81 --- /dev/null +++ b/crates/tinyflows-bus/src/schedule/types.rs @@ -0,0 +1,459 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Deserializer, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)] +#[serde(rename_all = "lowercase")] +pub enum JobType { + #[default] + Shell, + Agent, + /// A flow schedule trigger binding. The job's + /// `command` column carries the bound flow id. The runtime dispatches + /// it directly through `runtime::TriggerSink`, without a host event bus. + Flow, +} + +impl JobType { + pub fn as_str(&self) -> &'static str { + match self { + Self::Shell => "shell", + Self::Agent => "agent", + Self::Flow => "flow", + } + } + + pub fn parse(raw: &str) -> Self { + if raw.eq_ignore_ascii_case("agent") { + Self::Agent + } else if raw.eq_ignore_ascii_case("flow") { + Self::Flow + } else { + Self::Shell + } + } +} + +/// Which session an agent job runs in. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)] +#[serde(rename_all = "lowercase")] +pub enum SessionTarget { + /// A fresh, detached session with no conversation bound to it. + #[default] + Isolated, + /// The host's main session. + Main, + /// A detached run whose context and result delivery are bound to the + /// conversation that created the job ([`CronJob::origin`]). + Current, +} + +impl SessionTarget { + pub fn as_str(&self) -> &'static str { + match self { + Self::Isolated => "isolated", + Self::Main => "main", + Self::Current => "current", + } + } + + /// Case-insensitive; anything unrecognised reads as [`Self::Isolated`]. + pub fn parse(raw: &str) -> Self { + if raw.eq_ignore_ascii_case("main") { + Self::Main + } else if raw.eq_ignore_ascii_case("current") { + Self::Current + } else { + Self::Isolated + } + } +} + +/// The conversation a job was created from, captured at creation time so a +/// run can bind its context to it and route its result back there. +/// +/// Serializes internally tagged by `kind` (`"web"` or `"channel"`); absent +/// optional fields are omitted. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum JobOrigin { + /// Desktop/web chat thread. + Web { + /// The chat thread the job was created in. + thread_id: String, + /// The agent that owned that thread, when it was not the default one. + #[serde(default, skip_serializing_if = "Option::is_none")] + agent_id: Option, + }, + /// External messaging channel (Telegram, Discord, Slack, ...). + Channel { + /// Channel backend name, e.g. `"telegram"`. + channel: String, + /// Where a reply is sent on that channel (chat, room or user id). + reply_target: String, + /// The host's conversation-history key for that chat. + history_key: String, + /// The sender whose message created the job. + #[serde(default, skip_serializing_if = "Option::is_none")] + sender: Option, + /// Channel-side thread or topic within the chat. + #[serde(default, skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, +} + +impl JobOrigin { + /// The serialized `kind` tag: `"web"` or `"channel"`. + pub fn kind_str(&self) -> &'static str { + match self { + Self::Web { .. } => "web", + Self::Channel { .. } => "channel", + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ActiveHours { + pub start: String, + pub end: String, +} + +/// A cron-job schedule. +/// +/// Serializes as an internally-tagged object (`{"kind": "cron", ...}`). +/// Deserializes from **either** that object form **or** a bare cron-expression +/// string like `"0 9 * * 1"` — the bare-string form is treated as +/// `Schedule::Cron { expr, tz: None, active_hours: None }`. +/// +/// The bare-string shorthand exists because agents and some older frontend +/// callers pass `schedule: "0 9 * * 1"` directly instead of the structured +/// object. Accepting it here prevents Sentry issue CORE-RUST-FY +/// ("invalid type: string, expected internally tagged enum Schedule"). +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "lowercase")] +pub enum Schedule { + Cron { + /// 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 + /// `normalize_expression` (served by the module's `NormalizeExpression` member). + expr: String, + #[serde(default)] + tz: Option, + #[serde(default)] + active_hours: Option, + }, + At { + at: DateTime, + }, + Every { + every_ms: u64, + }, +} + +impl<'de> Deserialize<'de> for Schedule { + fn deserialize>(deserializer: D) -> Result { + use serde::de::{self, MapAccess, Visitor}; + use std::fmt; + + struct ScheduleVisitor; + + impl<'de> Visitor<'de> for ScheduleVisitor { + type Value = Schedule; + + fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "a cron-schedule object ({{\"kind\":\"cron\",\"expr\":\"...\"}}) \ + or a bare cron-expression string" + ) + } + + /// Accept a bare string as `Schedule::Cron { expr, .. }`. + /// This handles callers that send `schedule: "0 9 * * 1"` directly + /// instead of the structured form. + fn visit_str(self, value: &str) -> Result { + Ok(Schedule::Cron { + expr: value.to_owned(), + tz: None, + active_hours: None, + }) + } + + fn visit_string(self, value: String) -> Result { + Ok(Schedule::Cron { + expr: value, + tz: None, + active_hours: None, + }) + } + + /// Accept the standard internally-tagged object form. + fn visit_map>(self, map: A) -> Result { + // Delegate to the serde-derived tagged-enum logic by + // deserializing from a collected map value. + #[derive(Deserialize)] + #[serde(tag = "kind", rename_all = "lowercase")] + enum ScheduleTagged { + Cron { + expr: String, + #[serde(default)] + tz: Option, + #[serde(default)] + active_hours: Option, + }, + At { + at: DateTime, + }, + Every { + every_ms: u64, + }, + } + + let tagged = + ScheduleTagged::deserialize(de::value::MapAccessDeserializer::new(map))?; + Ok(match tagged { + ScheduleTagged::Cron { + expr, + tz, + active_hours, + } => Schedule::Cron { + expr, + tz, + active_hours, + }, + ScheduleTagged::At { at } => Schedule::At { at }, + ScheduleTagged::Every { every_ms } => Schedule::Every { every_ms }, + }) + } + } + + deserializer.deserialize_any(ScheduleVisitor) + } +} + +/// The [`DeliveryConfig::mode`] values a host understands. +pub mod delivery_mode { + /// Keep the result in the run history only. + pub const NONE: &str = "none"; + /// Announce the result on [`DeliveryConfig::channel`] / [`DeliveryConfig::to`]. + pub const ANNOUNCE: &str = "announce"; + /// Surface the result as a proactive message. + pub const PROACTIVE: &str = "proactive"; + /// Commit the result into the job's origin conversation + /// ([`CronJob::origin`](super::CronJob::origin)) and send it once there. + pub const ORIGIN: &str = "origin"; +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct DeliveryConfig { + /// One of the [`delivery_mode`] constants. `"origin"` routes the result + /// to the conversation recorded in [`CronJob::origin`]. + #[serde(default)] + pub mode: String, + #[serde(default)] + pub channel: Option, + #[serde(default)] + pub to: Option, + #[serde(default = "default_true")] + pub best_effort: bool, +} + +impl Default for DeliveryConfig { + fn default() -> Self { + Self { + mode: delivery_mode::NONE.to_string(), + channel: None, + to: None, + best_effort: true, + } + } +} + +fn default_true() -> bool { + true +} + +/// Per-job retry override and single-flight missed-slot recording. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct JobPolicy { + /// None uses scheduler defaults; Some(0) runs exactly once. + pub retries: Option, + /// Record recurring slots skipped while the same job is already running. + pub single_flight: bool, +} +impl JobPolicy { + /// Whether serialization can omit this historically compatible policy. + pub fn is_default(&self) -> bool { + *self == Self::default() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CronJob { + /// Durable occurrence claim generation; legacy jobs start at zero. Stores + /// increment this atomically when claiming and fence stale completions. + #[serde(default, skip_serializing_if = "claim_generation_is_zero")] + pub claim_generation: u64, + #[serde(default, skip_serializing_if = "JobPolicy::is_default")] + /// Retry override and single-flight semantics, defaulting for older jobs. + pub policy: JobPolicy, + pub id: String, + pub expression: String, + pub schedule: Schedule, + pub command: String, + pub prompt: Option, + pub name: Option, + pub job_type: JobType, + pub session_target: SessionTarget, + pub model: Option, + /// Optional built-in agent definition ID (e.g. `"welcome"`, + /// `"morning_briefing"`). When set, the cron scheduler + /// resolves the agent definition from the registry and runs with the + /// definition's prompt, tool allowlist, iteration cap, and model hint + /// instead of the generic `OpenHumanSessionHost::from_config` path. + pub agent_id: Option, + pub enabled: bool, + pub delivery: DeliveryConfig, + pub delete_after_run: bool, + pub created_at: DateTime, + pub next_run: DateTime, + pub last_run: Option>, + pub last_status: Option, + pub last_output: Option, + /// The conversation that created the job, when it was created from one. + /// Absent on jobs that predate origin capture. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub origin: Option, +} + +fn claim_generation_is_zero(generation: &u64) -> bool { + *generation == 0 +} + +impl CronJob { + /// Compare the inputs that determine execution and delivery, excluding + /// display names and last-run bookkeeping that may change independently. + pub fn same_execution_definition(&self, other: &Self) -> bool { + self.command == other.command + && self.prompt == other.prompt + && self.job_type == other.job_type + && self.session_target == other.session_target + && self.model == other.model + && self.agent_id == other.agent_id + && self.delivery == other.delivery + && self.origin == other.origin + && self.policy == other.policy + && self.delete_after_run == other.delete_after_run + } +} + +/// What happened to a run's result after the run finished. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum DeliveryStatus { + /// The result was delivered to its destination. + Delivered, + /// Delivery was requested but deliberately skipped (e.g. empty output). + Suppressed, + /// Delivery was attempted and failed. + Failed, + /// The job's delivery mode asked for no delivery. + NotRequested, +} + +impl DeliveryStatus { + pub fn as_str(&self) -> &'static str { + match self { + Self::Delivered => "delivered", + Self::Suppressed => "suppressed", + Self::Failed => "failed", + Self::NotRequested => "not_requested", + } + } + + /// Case-insensitive; `None` for an unknown value. + pub fn parse(raw: &str) -> Option { + [ + Self::Delivered, + Self::Suppressed, + Self::Failed, + Self::NotRequested, + ] + .into_iter() + .find(|status| raw.eq_ignore_ascii_case(status.as_str())) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CronRun { + pub id: i64, + pub job_id: String, + pub started_at: DateTime, + pub finished_at: DateTime, + pub status: String, + pub output: Option, + pub duration_ms: Option, + /// Outcome of delivering this run's result; absent on runs recorded + /// without one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub delivery_status: Option, +} + +/// Deserialize a nullable patch field with true double-option semantics: +/// +/// | wire | result | meaning | +/// | --------------- | ------------- | ------------- | +/// | key absent | `None` | no change | +/// | key present `null` | `Some(None)` | clear the value | +/// | key present value | `Some(Some(v))` | set the value | +/// +/// A plain `#[derive(Deserialize)]` on `Option>` collapses the absent +/// and the `null` cases *both* to the outer `None`, so "clear over the wire" +/// (for example, `{"agent_id": null}`) silently deserializes as "no change" — a no-op. Used +/// with `#[serde(default, deserialize_with = "deserialize_double_option")]`, +/// this helper restores the distinction: serde only invokes it when the key is +/// *present*, so a present `null` becomes `Some(None)` and a present value +/// becomes `Some(Some(v))`, while an absent key falls back to the `default` +/// (`None`). +fn deserialize_double_option<'de, D, T>(deserializer: D) -> Result>, D::Error> +where + D: Deserializer<'de>, + T: Deserialize<'de>, +{ + Ok(Some(Option::::deserialize(deserializer)?)) +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct CronJobPatch { + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Replace the complete job policy; None leaves it unchanged. + pub policy: Option, + pub schedule: Option, + pub command: Option, + pub prompt: Option, + pub name: Option, + pub enabled: Option, + pub delivery: Option, + pub model: Option, + pub session_target: Option, + pub delete_after_run: Option, + /// `Option>` distinguishes "no change" (`None`) from + /// "clear the agent definition" (`Some(None)`). See + /// [`deserialize_double_option`] for why the custom deserializer is required + /// to honor a wire `null` as a clear rather than a silent no-op. + #[serde(default, deserialize_with = "deserialize_double_option")] + pub agent_id: Option>, + /// Set (`Some(Some(_))`) or clear (`Some(None)`) the job's origin + /// conversation; `None` leaves it unchanged (and is omitted when + /// serialized, so a round trip never turns it into a clear). + #[serde( + default, + deserialize_with = "deserialize_double_option", + skip_serializing_if = "Option::is_none" + )] + pub origin: Option>, +} + +#[cfg(test)] +#[path = "types_tests.rs"] +mod tests; 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 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); +} 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-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..a7c7decd 100644 --- a/crates/tinyflows-import/src/cron_tests.rs +++ b/crates/tinyflows-import/src/cron_tests.rs @@ -1,6 +1,7 @@ use super::*; 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 new file mode 100644 index 00000000..056309d5 --- /dev/null +++ b/crates/tinyflows-import/src/legacy_cron.rs @@ -0,0 +1,290 @@ +//! 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..9752dd99 100644 --- a/crates/tinyflows-import/src/lib.rs +++ b/crates/tinyflows-import/src/lib.rs @@ -55,6 +55,8 @@ mod catalog; mod checkpoints; #[cfg(feature = "schedule")] mod cron; +#[cfg(feature = "schedule-read")] +mod legacy_cron; mod legacy_db; mod report; #[cfg(feature = "catalog")] @@ -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}"); 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/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/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}; diff --git a/crates/tinyflows-module/src/schedule_callbacks.rs b/crates/tinyflows-module/src/schedule_callbacks.rs index a7911019..752b7fab 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,24 @@ 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?) } } 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..8c96fa5b --- /dev/null +++ b/crates/tinyflows-module/src/schedule_members_tests.rs @@ -0,0 +1,315 @@ +//! 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, Connection) { + let (server, host, _) = connections().await; + let proxy = host + .proxy( + schedule::BUS_NAME, + schedule::OBJECT_PATH, + schedule::INTERFACE, + ) + .unwrap(); + (proxy, host, server) +} + +#[tokio::test] +async fn schedule_maths_members_match_the_in_process_functions() { + let (proxy, _host, _server) = 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"); + + for at in ["2026-01-01T00:00:00Z", "2025-12-31T00:00:00Z"] { + let elapsed = proxy + .call::>( + 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, + &(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, _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),)) + .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, _server) = 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, _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::<()>( + 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, _server) = 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, _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 { + assert!( + !error.to_string().contains("UnknownMethod"), + "{name}: {error}" + ); + } + } +} diff --git a/crates/tinyflows-module/src/schedule_service.rs b/crates/tinyflows-module/src/schedule_service.rs index b00a5b4a..ac83973c 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,64 @@ impl ScheduleService { validation::validate_delivery(&delivery, origin.as_ref(), allowed_users.as_deref()) .map_err(failure) } + + 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 + && *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<()> { + 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<()> { @@ -186,6 +249,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.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..20c8bb44 100644 --- a/crates/tinyflows-module/src/schedule_store_tests.rs +++ b/crates/tinyflows-module/src/schedule_store_tests.rs @@ -1,6 +1,7 @@ use super::*; use async_trait::async_trait; use tinyflows_schedule::runtime::*; +use tinyflows_schedule::{DeliveryStatus, Schedule}; use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; struct Replacement(CronDocuments); diff --git a/crates/tinyflows-schedule/Cargo.toml b/crates/tinyflows-schedule/Cargo.toml index c04ac303..626e3382 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 = { version = "0.1.0", path = "../tinyflows-bus" } async-trait.workspace = true tokio.workspace = true uuid.workspace = true 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/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..04b01384 100644 --- a/crates/tinyflows-schedule/src/schedule.rs +++ b/crates/tinyflows-schedule/src/schedule.rs @@ -3,16 +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; -/// 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, 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 @@ -95,44 +91,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 +147,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..740753ce 100644 --- a/crates/tinyflows-schedule/src/store.rs +++ b/crates/tinyflows-schedule/src/store.rs @@ -8,52 +8,11 @@ //! [`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, ImportedJob, 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 index debdea59..e0a49183 100644 --- a/crates/tinyflows-schedule/src/types.rs +++ b/crates/tinyflows-schedule/src/types.rs @@ -1,469 +1,3 @@ -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Deserializer, Serialize}; - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)] -#[serde(rename_all = "lowercase")] -pub enum JobType { - #[default] - Shell, - Agent, - /// A flow schedule trigger binding. The job's - /// `command` column carries the bound flow id. The runtime dispatches - /// it directly through `runtime::TriggerSink`, without a host event bus. - Flow, -} - -impl JobType { - pub fn as_str(&self) -> &'static str { - match self { - Self::Shell => "shell", - Self::Agent => "agent", - Self::Flow => "flow", - } - } - - pub fn parse(raw: &str) -> Self { - if raw.eq_ignore_ascii_case("agent") { - Self::Agent - } else if raw.eq_ignore_ascii_case("flow") { - Self::Flow - } else { - Self::Shell - } - } -} - -/// Which session an agent job runs in. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)] -#[serde(rename_all = "lowercase")] -pub enum SessionTarget { - /// A fresh, detached session with no conversation bound to it. - #[default] - Isolated, - /// The host's main session. - Main, - /// A detached run whose context and result delivery are bound to the - /// conversation that created the job ([`CronJob::origin`]). - Current, -} - -impl SessionTarget { - pub fn as_str(&self) -> &'static str { - match self { - Self::Isolated => "isolated", - Self::Main => "main", - Self::Current => "current", - } - } - - /// Case-insensitive; anything unrecognised reads as [`Self::Isolated`]. - pub fn parse(raw: &str) -> Self { - if raw.eq_ignore_ascii_case("main") { - Self::Main - } else if raw.eq_ignore_ascii_case("current") { - Self::Current - } else { - Self::Isolated - } - } -} - -/// The conversation a job was created from, captured at creation time so a -/// run can bind its context to it and route its result back there. -/// -/// Serializes internally tagged by `kind` (`"web"` or `"channel"`); absent -/// optional fields are omitted. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum JobOrigin { - /// Desktop/web chat thread. - Web { - /// The chat thread the job was created in. - thread_id: String, - /// The agent that owned that thread, when it was not the default one. - #[serde(default, skip_serializing_if = "Option::is_none")] - agent_id: Option, - }, - /// External messaging channel (Telegram, Discord, Slack, ...). - Channel { - /// Channel backend name, e.g. `"telegram"`. - channel: String, - /// Where a reply is sent on that channel (chat, room or user id). - reply_target: String, - /// The host's conversation-history key for that chat. - history_key: String, - /// The sender whose message created the job. - #[serde(default, skip_serializing_if = "Option::is_none")] - sender: Option, - /// Channel-side thread or topic within the chat. - #[serde(default, skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, -} - -impl JobOrigin { - /// The serialized `kind` tag: `"web"` or `"channel"`. - pub fn kind_str(&self) -> &'static str { - match self { - Self::Web { .. } => "web", - Self::Channel { .. } => "channel", - } - } -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct ActiveHours { - pub start: String, - pub end: String, -} - -/// A cron-job schedule. -/// -/// Serializes as an internally-tagged object (`{"kind": "cron", ...}`). -/// Deserializes from **either** that object form **or** a bare cron-expression -/// string like `"0 9 * * 1"` — the bare-string form is treated as -/// `Schedule::Cron { expr, tz: None, active_hours: None }`. -/// -/// The bare-string shorthand exists because agents and some older frontend -/// callers pass `schedule: "0 9 * * 1"` directly instead of the structured -/// object. Accepting it here prevents Sentry issue CORE-RUST-FY -/// ("invalid type: string, expected internally tagged enum Schedule"). -#[derive(Debug, Clone, Serialize, PartialEq, Eq)] -#[serde(tag = "kind", rename_all = "lowercase")] -pub enum Schedule { - Cron { - /// 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`]. - expr: String, - #[serde(default)] - tz: Option, - #[serde(default)] - active_hours: Option, - }, - At { - at: DateTime, - }, - Every { - every_ms: u64, - }, -} - -impl<'de> Deserialize<'de> for Schedule { - fn deserialize>(deserializer: D) -> Result { - use serde::de::{self, MapAccess, Visitor}; - use std::fmt; - - struct ScheduleVisitor; - - impl<'de> Visitor<'de> for ScheduleVisitor { - type Value = Schedule; - - fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - f, - "a cron-schedule object ({{\"kind\":\"cron\",\"expr\":\"...\"}}) \ - or a bare cron-expression string" - ) - } - - /// Accept a bare string as `Schedule::Cron { expr, .. }`. - /// 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, - active_hours: None, - }) - } - - 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, - active_hours: None, - }) - } - - /// Accept the standard internally-tagged object form. - fn visit_map>(self, map: A) -> Result { - // Delegate to the serde-derived tagged-enum logic by - // deserializing from a collected map value. - #[derive(Deserialize)] - #[serde(tag = "kind", rename_all = "lowercase")] - enum ScheduleTagged { - Cron { - expr: String, - #[serde(default)] - tz: Option, - #[serde(default)] - active_hours: Option, - }, - At { - at: DateTime, - }, - Every { - every_ms: u64, - }, - } - - let tagged = - ScheduleTagged::deserialize(de::value::MapAccessDeserializer::new(map))?; - Ok(match tagged { - ScheduleTagged::Cron { - expr, - tz, - active_hours, - } => Schedule::Cron { - expr, - tz, - active_hours, - }, - ScheduleTagged::At { at } => Schedule::At { at }, - ScheduleTagged::Every { every_ms } => Schedule::Every { every_ms }, - }) - } - } - - deserializer.deserialize_any(ScheduleVisitor) - } -} - -/// The [`DeliveryConfig::mode`] values a host understands. -pub mod delivery_mode { - /// Keep the result in the run history only. - pub const NONE: &str = "none"; - /// Announce the result on [`DeliveryConfig::channel`] / [`DeliveryConfig::to`]. - pub const ANNOUNCE: &str = "announce"; - /// Surface the result as a proactive message. - pub const PROACTIVE: &str = "proactive"; - /// Commit the result into the job's origin conversation - /// ([`CronJob::origin`](super::CronJob::origin)) and send it once there. - pub const ORIGIN: &str = "origin"; -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct DeliveryConfig { - /// One of the [`delivery_mode`] constants. `"origin"` routes the result - /// to the conversation recorded in [`CronJob::origin`]. - #[serde(default)] - pub mode: String, - #[serde(default)] - pub channel: Option, - #[serde(default)] - pub to: Option, - #[serde(default = "default_true")] - pub best_effort: bool, -} - -impl Default for DeliveryConfig { - fn default() -> Self { - Self { - mode: delivery_mode::NONE.to_string(), - channel: None, - to: None, - best_effort: true, - } - } -} - -fn default_true() -> bool { - true -} - -/// Per-job retry override and single-flight missed-slot recording. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] -pub struct JobPolicy { - /// None uses scheduler defaults; Some(0) runs exactly once. - pub retries: Option, - /// Record recurring slots skipped while the same job is already running. - pub single_flight: bool, -} -impl JobPolicy { - /// Whether serialization can omit this historically compatible policy. - pub fn is_default(&self) -> bool { - *self == Self::default() - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CronJob { - /// Durable occurrence claim generation; legacy jobs start at zero. Stores - /// increment this atomically when claiming and fence stale completions. - #[serde(default, skip_serializing_if = "claim_generation_is_zero")] - pub claim_generation: u64, - #[serde(default, skip_serializing_if = "JobPolicy::is_default")] - /// Retry override and single-flight semantics, defaulting for older jobs. - pub policy: JobPolicy, - pub id: String, - pub expression: String, - pub schedule: Schedule, - pub command: String, - pub prompt: Option, - pub name: Option, - pub job_type: JobType, - pub session_target: SessionTarget, - pub model: Option, - /// Optional built-in agent definition ID (e.g. `"welcome"`, - /// `"morning_briefing"`). When set, the cron scheduler - /// resolves the agent definition from the registry and runs with the - /// definition's prompt, tool allowlist, iteration cap, and model hint - /// instead of the generic `OpenHumanSessionHost::from_config` path. - pub agent_id: Option, - pub enabled: bool, - pub delivery: DeliveryConfig, - pub delete_after_run: bool, - pub created_at: DateTime, - pub next_run: DateTime, - pub last_run: Option>, - pub last_status: Option, - pub last_output: Option, - /// The conversation that created the job, when it was created from one. - /// Absent on jobs that predate origin capture. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub origin: Option, -} - -fn claim_generation_is_zero(generation: &u64) -> bool { - *generation == 0 -} - -impl CronJob { - /// Compare the inputs that determine execution and delivery, excluding - /// display names and last-run bookkeeping that may change independently. - pub fn same_execution_definition(&self, other: &Self) -> bool { - self.command == other.command - && self.prompt == other.prompt - && self.job_type == other.job_type - && self.session_target == other.session_target - && self.model == other.model - && self.agent_id == other.agent_id - && self.delivery == other.delivery - && self.origin == other.origin - && self.policy == other.policy - && self.delete_after_run == other.delete_after_run - } -} - -/// What happened to a run's result after the run finished. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum DeliveryStatus { - /// The result was delivered to its destination. - Delivered, - /// Delivery was requested but deliberately skipped (e.g. empty output). - Suppressed, - /// Delivery was attempted and failed. - Failed, - /// The job's delivery mode asked for no delivery. - NotRequested, -} - -impl DeliveryStatus { - pub fn as_str(&self) -> &'static str { - match self { - Self::Delivered => "delivered", - Self::Suppressed => "suppressed", - Self::Failed => "failed", - Self::NotRequested => "not_requested", - } - } - - /// Case-insensitive; `None` for an unknown value. - pub fn parse(raw: &str) -> Option { - [ - Self::Delivered, - Self::Suppressed, - Self::Failed, - Self::NotRequested, - ] - .into_iter() - .find(|status| raw.eq_ignore_ascii_case(status.as_str())) - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CronRun { - pub id: i64, - pub job_id: String, - pub started_at: DateTime, - pub finished_at: DateTime, - pub status: String, - pub output: Option, - pub duration_ms: Option, - /// Outcome of delivering this run's result; absent on runs recorded - /// without one. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub delivery_status: Option, -} - -/// Deserialize a nullable patch field with true double-option semantics: -/// -/// | wire | result | meaning | -/// | --------------- | ------------- | ------------- | -/// | key absent | `None` | no change | -/// | key present `null` | `Some(None)` | clear the value | -/// | key present value | `Some(Some(v))` | set the value | -/// -/// A plain `#[derive(Deserialize)]` on `Option>` collapses the absent -/// and the `null` cases *both* to the outer `None`, so "clear over the wire" -/// (for example, `{"agent_id": null}`) silently deserializes as "no change" — a no-op. Used -/// with `#[serde(default, deserialize_with = "deserialize_double_option")]`, -/// this helper restores the distinction: serde only invokes it when the key is -/// *present*, so a present `null` becomes `Some(None)` and a present value -/// becomes `Some(Some(v))`, while an absent key falls back to the `default` -/// (`None`). -fn deserialize_double_option<'de, D, T>(deserializer: D) -> Result>, D::Error> -where - D: Deserializer<'de>, - T: Deserialize<'de>, -{ - Ok(Some(Option::::deserialize(deserializer)?)) -} - -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct CronJobPatch { - #[serde(default, skip_serializing_if = "Option::is_none")] - /// Replace the complete job policy; None leaves it unchanged. - pub policy: Option, - pub schedule: Option, - pub command: Option, - pub prompt: Option, - pub name: Option, - pub enabled: Option, - pub delivery: Option, - pub model: Option, - pub session_target: Option, - pub delete_after_run: Option, - /// `Option>` distinguishes "no change" (`None`) from - /// "clear the agent definition" (`Some(None)`). See - /// [`deserialize_double_option`] for why the custom deserializer is required - /// to honor a wire `null` as a clear rather than a silent no-op. - #[serde(default, deserialize_with = "deserialize_double_option")] - pub agent_id: Option>, - /// Set (`Some(Some(_))`) or clear (`Some(None)`) the job's origin - /// conversation; `None` leaves it unchanged (and is omitted when - /// serialized, so a round trip never turns it into a clear). - #[serde( - default, - deserialize_with = "deserialize_double_option", - skip_serializing_if = "Option::is_none" - )] - pub origin: Option>, -} - -#[cfg(test)] -#[path = "types_tests.rs"] -mod tests; +//! 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::*; diff --git a/docs/tinyflows-bus-design.md b/docs/tinyflows-bus-design.md index f31b8c64..6a0610ba 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,31 @@ 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`). `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` + 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