asched_core/routine/
mod.rs

1//! Machine-local scheduled routines.
2//!
3//! Definitions are stored per canonical working directory. The daemon is the
4//! only writer; application clients should use [`RoutineClient`]. The
5//! versioned Unix-socket protocol remains available in [`ipc`] for diagnostics
6//! and protocol-level integrations.
7
8mod client;
9mod cron;
10pub mod daemon;
11pub mod execution;
12pub mod ipc;
13pub mod store;
14
15pub use client::{RoutineClient, STARTUP_FD_ENV};
16pub use cron::{CronSchedule, LocalTime};
17
18use serde::{Deserialize, Serialize};
19use std::path::PathBuf;
20
21pub const PROJECT_CONFIG_VERSION: u32 = 2;
22pub const RUNTIME_STATE_VERSION: u32 = 2;
23pub const TRANSACTION_VERSION: u32 = 1;
24/// Current project-config version. Retained as a source-compatibility alias.
25pub const SCHEMA_VERSION: u32 = PROJECT_CONFIG_VERSION;
26pub const PROTOCOL_VERSION: u32 = 3;
27pub const MAX_RUNS: usize = 20;
28pub const MAX_EVENT_PAYLOAD_BYTES: usize = 64 * 1024;
29pub const MAX_EVENT_RECEIPTS: usize = 4_096;
30
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(rename_all = "snake_case")]
33pub enum Trigger {
34    Cron(String),
35    Event { kind: String },
36}
37
38impl Trigger {
39    pub fn validated(mut self) -> Result<Self, RoutineError> {
40        match &mut self {
41            Self::Cron(expression) => {
42                *expression = expression.trim().to_string();
43                CronSchedule::parse(expression)?;
44            }
45            Self::Event { kind } => {
46                *kind = kind.trim().to_string();
47                if kind.is_empty() {
48                    return Err(RoutineError::Validation(
49                        "event kind must not be empty".into(),
50                    ));
51                }
52                if kind.chars().any(char::is_whitespace)
53                    || kind.chars().any(char::is_control)
54                    || !kind.contains('.')
55                    || kind.split('.').any(str::is_empty)
56                {
57                    return Err(RoutineError::Validation(
58                        "event kind must be a namespaced string without whitespace or control characters"
59                            .into(),
60                    ));
61                }
62            }
63        }
64        Ok(self)
65    }
66}
67
68#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
69// ^ [[Routine Scheduling and Run Lifecycle]]
70pub struct Routine {
71    pub name: String,
72    pub trigger: Trigger,
73    pub command: Vec<String>,
74    pub prompt: String,
75    #[serde(default = "default_enabled")]
76    pub enabled: bool,
77}
78
79const fn default_enabled() -> bool {
80    true
81}
82
83impl<'de> Deserialize<'de> for Routine {
84    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
85    where
86        D: serde::Deserializer<'de>,
87    {
88        #[derive(Deserialize)]
89        struct StoredRoutine {
90            name: String,
91            #[serde(default)]
92            trigger: Option<Trigger>,
93            #[serde(default)]
94            cron: Option<String>,
95            command: Vec<String>,
96            prompt: String,
97            #[serde(default = "default_enabled")]
98            enabled: bool,
99        }
100        let stored = StoredRoutine::deserialize(deserializer)?;
101        let trigger = match (stored.trigger, stored.cron) {
102            (Some(trigger), None) => trigger,
103            (None, Some(cron)) => Trigger::Cron(cron),
104            (Some(_), Some(_)) => {
105                return Err(serde::de::Error::custom(
106                    "routine must contain exactly one trigger representation",
107                ))
108            }
109            (None, None) => return Err(serde::de::Error::missing_field("trigger")),
110        };
111        Ok(Self {
112            name: stored.name,
113            trigger,
114            command: stored.command,
115            prompt: stored.prompt,
116            enabled: stored.enabled,
117        })
118    }
119}
120
121impl Routine {
122    pub fn validated(mut self) -> Result<Self, RoutineError> {
123        self.name = self.name.trim().to_string();
124        if self.name.is_empty() {
125            return Err(RoutineError::Validation("name must not be empty".into()));
126        }
127        if self.name.contains('/') || self.name.chars().any(char::is_control) {
128            return Err(RoutineError::Validation(
129                "name must not contain '/' or control characters".into(),
130            ));
131        }
132        self.trigger = self.trigger.validated()?;
133        if self.command.is_empty() || self.command.iter().any(|arg| arg.is_empty()) {
134            return Err(RoutineError::Validation(
135                "command must contain nonempty argv items".into(),
136            ));
137        }
138        if self
139            .command
140            .iter()
141            .filter(|arg| arg.as_str() == "{prompt}")
142            .count()
143            > 1
144        {
145            return Err(RoutineError::Validation(
146                "command may contain at most one exact {prompt} argument".into(),
147            ));
148        }
149        Ok(self)
150    }
151}
152
153#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(rename_all = "snake_case")]
155pub enum RunStatus {
156    Running,
157    Succeeded,
158    Failed,
159    SpawnFailed,
160    Cancelled,
161    Interrupted,
162}
163
164#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
165#[serde(rename_all = "snake_case")]
166pub enum RunCause {
167    #[default]
168    Manual,
169    Cron {
170        scheduled_epoch_minute: i64,
171    },
172    Event {
173        kind: String,
174        event_id: String,
175    },
176}
177
178#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct RunRecord {
180    pub id: String,
181    pub routine: String,
182    pub started_epoch: i64,
183    pub finished_epoch: Option<i64>,
184    #[serde(default)]
185    pub cause: RunCause,
186    /// Retained in memory for source compatibility; v2 readers should use `cause`.
187    #[serde(default, skip_serializing_if = "Option::is_none")]
188    pub scheduled_epoch_minute: Option<i64>,
189    pub status: RunStatus,
190    pub exit_code: Option<i32>,
191    #[serde(default)]
192    pub pid: Option<i32>,
193    /// OS process-start token used to reject a recycled PID/process group.
194    #[serde(default)]
195    pub process_start: Option<String>,
196    pub final_output: String,
197    pub stdout_path: PathBuf,
198    pub stderr_path: PathBuf,
199}
200
201#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
202#[serde(rename_all = "snake_case")]
203pub enum FireOutcome {
204    Handled { routines: Vec<RoutineFire> },
205    Deduplicated,
206    NoMatch,
207}
208
209#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
210#[serde(rename_all = "snake_case")]
211pub enum RoutineFire {
212    Started { name: String },
213    AlreadyRunning { name: String },
214}
215
216#[derive(Debug, Clone, Serialize, Deserialize)]
217pub struct Capabilities {
218    pub can_edit: bool,
219    pub can_delete: bool,
220    pub can_run: bool,
221    pub can_cancel: bool,
222    pub can_rename: bool,
223    pub can_toggle_enabled: bool,
224}
225
226impl Capabilities {
227    pub fn for_running(running: bool) -> Self {
228        Self {
229            can_edit: true,
230            can_delete: true,
231            can_run: !running,
232            can_cancel: running,
233            can_rename: !running,
234            can_toggle_enabled: true,
235        }
236    }
237}
238
239#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
240#[serde(rename_all = "snake_case")]
241pub enum RoutineErrorKind {
242    Validation,
243    Duplicate,
244    NotFound,
245    Conflict,
246    ProjectCollision,
247    AlreadyRunning,
248    ProtocolMismatch,
249    Unavailable,
250    Io,
251    Corrupt,
252}
253
254impl std::fmt::Display for RoutineErrorKind {
255    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
256        let value = serde_json::to_value(self).map_err(|_| std::fmt::Error)?;
257        formatter.write_str(value.as_str().ok_or(std::fmt::Error)?)
258    }
259}
260
261#[derive(Debug, thiserror::Error)]
262pub enum RoutineError {
263    #[error("invalid routine: {0}")]
264    Validation(String),
265    #[error("routine '{0}' already exists")]
266    Duplicate(String),
267    #[error("routine '{0}' not found")]
268    NotFound(String),
269    #[error("stale config revision: expected {expected}, actual {actual}")]
270    Conflict { expected: u64, actual: u64 },
271    #[error("project identity collision: expected {expected}, stored {stored}")]
272    ProjectCollision { expected: PathBuf, stored: PathBuf },
273    #[error("routine '{0}' is already running")]
274    AlreadyRunning(String),
275    #[error("protocol mismatch: client {client}, daemon {daemon}")]
276    ProtocolMismatch { client: u32, daemon: u32 },
277    #[error("daemon unavailable: {0}")]
278    Unavailable(String),
279    #[error("I/O error: {0}")]
280    Io(String),
281    #[error("invalid stored data: {0}")]
282    Corrupt(String),
283    #[error("routine daemon {kind}: {message}")]
284    RemoteDaemon {
285        kind: RoutineErrorKind,
286        message: String,
287    },
288}
289
290impl RoutineError {
291    pub fn kind(&self) -> RoutineErrorKind {
292        match self {
293            Self::Validation(_) => RoutineErrorKind::Validation,
294            Self::Duplicate(_) => RoutineErrorKind::Duplicate,
295            Self::NotFound(_) => RoutineErrorKind::NotFound,
296            Self::Conflict { .. } => RoutineErrorKind::Conflict,
297            Self::ProjectCollision { .. } => RoutineErrorKind::ProjectCollision,
298            Self::AlreadyRunning(_) => RoutineErrorKind::AlreadyRunning,
299            Self::ProtocolMismatch { .. } => RoutineErrorKind::ProtocolMismatch,
300            Self::Unavailable(_) => RoutineErrorKind::Unavailable,
301            Self::Io(_) => RoutineErrorKind::Io,
302            Self::Corrupt(_) => RoutineErrorKind::Corrupt,
303            Self::RemoteDaemon { kind, .. } => *kind,
304        }
305    }
306}
307
308impl From<std::io::Error> for RoutineError {
309    fn from(value: std::io::Error) -> Self {
310        Self::Io(value.to_string())
311    }
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317
318    #[test]
319    fn legacy_cron_and_typed_event_routines_share_one_serialization_boundary() {
320        let legacy = r#"
321name = "legacy"
322cron = "0 9 * * *"
323command = ["/bin/true"]
324prompt = ""
325enabled = true
326"#;
327        let cron: Routine = toml::from_str(legacy).unwrap();
328        assert_eq!(cron.trigger, Trigger::Cron("0 9 * * *".into()));
329
330        let event = Routine {
331            name: "event".into(),
332            trigger: Trigger::Event {
333                kind: "filesystem.changed".into(),
334            },
335            command: vec!["/bin/true".into()],
336            prompt: String::new(),
337            enabled: true,
338        };
339        let encoded = toml::to_string(&event).unwrap();
340        let decoded: Routine = toml::from_str(&encoded).unwrap();
341        assert_eq!(decoded, event);
342        assert!(!encoded.contains("cron"));
343    }
344
345    #[test]
346    fn routine_validation_rejects_ambiguous_boundaries() {
347        for routine in [
348            Routine {
349                name: " ".into(),
350                trigger: Trigger::Cron("* * * * *".into()),
351                command: vec!["x".into()],
352                prompt: String::new(),
353                enabled: true,
354            },
355            Routine {
356                name: "x".into(),
357                trigger: Trigger::Cron("60 * * * *".into()),
358                command: vec!["x".into()],
359                prompt: String::new(),
360                enabled: true,
361            },
362            Routine {
363                name: "x".into(),
364                trigger: Trigger::Cron("* * * * *".into()),
365                command: vec![],
366                prompt: String::new(),
367                enabled: true,
368            },
369            Routine {
370                name: "x".into(),
371                trigger: Trigger::Cron("* * * * *".into()),
372                command: vec!["{prompt}".into(), "{prompt}".into()],
373                prompt: String::new(),
374                enabled: true,
375            },
376        ] {
377            assert!(routine.validated().is_err());
378        }
379    }
380
381    #[test]
382    fn event_trigger_validation_rejects_empty_namespace_segments() {
383        for kind in [".", ".changed", "source.", "source..changed"] {
384            assert!(matches!(
385                (Trigger::Event { kind: kind.into() }).validated(),
386                Err(RoutineError::Validation(_))
387            ));
388        }
389    }
390
391    #[test]
392    fn given_ansi_and_control_characters_in_name_when_validated_then_each_is_rejected() {
393        let rejected = [
394            "\u{1b}[31mdanger",
395            "line\nbreak",
396            "tab\tname",
397            "delete\u{7f}",
398        ]
399        .into_iter()
400        .all(|name| {
401            matches!(
402                (Routine {
403                    name: name.into(),
404                    trigger: Trigger::Cron("* * * * *".into()),
405                    command: vec!["/bin/true".into()],
406                    prompt: String::new(),
407                    enabled: true,
408                })
409                .validated(),
410                Err(RoutineError::Validation(_))
411            )
412        });
413
414        assert!(rejected);
415    }
416}
asched_core/routine/mod.rs

asched_core/routine/
mod.rs
