everruns_core/
session_schedule.rs

1// Session schedule domain types
2//
3// Represents scheduled tasks bound to a session.
4// When a schedule fires, a message is injected into the session to trigger a turn.
5
6use chrono::{DateTime, Utc};
7use serde::{Deserialize, Serialize};
8use uuid::Uuid;
9
10use crate::principal::PrincipalSummary;
11use crate::typed_id::{PrincipalId, ScheduleId, SessionId};
12
13#[cfg(feature = "openapi")]
14use utoipa::ToSchema;
15
16/// Maximum number of active schedules per session.
17pub const MAX_ACTIVE_SCHEDULES_PER_SESSION: u32 = 5;
18
19/// Default minimum seconds between consecutive recurring schedule fires.
20pub const DEFAULT_MIN_INTERVAL_SECONDS: i64 = 300;
21
22/// Default maximum number of active (enabled) schedules per org.
23pub const DEFAULT_MAX_SCHEDULES_PER_ORG: i64 = 100;
24
25/// Minimum seconds between consecutive recurring session-schedule fires.
26///
27/// Configurable via `SESSION_SCHEDULE_MIN_INTERVAL_SECONDS`; default 300 (5 min).
28/// Each fire dispatches a real worker turn, so on open-signup deployments a
29/// `* * * * *` (every-minute) cron is a sustained amplifier of operator compute.
30/// This is the session-schedule sibling of the app `schedule` channel's
31/// `SCHEDULE_CHANNEL_MIN_INTERVAL_SECONDS` (see `specs/app-invocation-channels.md`).
32pub fn min_interval_seconds() -> i64 {
33    std::env::var("SESSION_SCHEDULE_MIN_INTERVAL_SECONDS")
34        .ok()
35        .and_then(|v| v.parse::<i64>().ok())
36        .filter(|&v| v > 0)
37        .unwrap_or(DEFAULT_MIN_INTERVAL_SECONDS)
38}
39
40/// Maximum number of active (enabled) schedules per org.
41///
42/// Configurable via `RESOURCE_LIMIT_MAX_SESSION_SCHEDULES_PER_ORG`; default 100.
43/// `MAX_ACTIVE_SCHEDULES_PER_SESSION` only bounds a single session, so without an
44/// org-wide cap unlimited sessions imply unlimited active schedules per org. Uses
45/// the `RESOURCE_LIMIT_*` env family so the SaaS wrapper sets it per plan via
46/// `PlanResourceLimits::apply_to_env`.
47pub fn max_active_schedules_per_org() -> i64 {
48    std::env::var("RESOURCE_LIMIT_MAX_SESSION_SCHEDULES_PER_ORG")
49        .ok()
50        .and_then(|v| v.parse::<i64>().ok())
51        .filter(|&v| v > 0)
52        .unwrap_or(DEFAULT_MAX_SCHEDULES_PER_ORG)
53}
54
55/// Returns the minimum interval (seconds) between the next few consecutive
56/// triggers of `cron_expression`, or `None` when the expression cannot be parsed
57/// or fires fewer than twice.
58///
59/// Accepts 5-field (`min hour dom mon dow`) and 6/7-field cron forms; 5-field is
60/// normalized to the seconds-aware form the `cron` crate expects (sec=0, year=*).
61/// This is more permissive than the app schedule channel's
62/// `normalize_cron_expression` (which accepts only 5 or 7 fields): here we also
63/// accept the 6-field seconds form the agent may already pass through to the store.
64pub fn cron_min_interval_seconds(cron_expression: &str) -> Option<i64> {
65    use std::str::FromStr;
66    let fields: Vec<&str> = cron_expression.split_whitespace().collect();
67    let normalized = match fields.len() {
68        5 => format!("0 {} *", fields.join(" ")),
69        6 | 7 => cron_expression.to_string(),
70        _ => return None,
71    };
72    let schedule = cron::Schedule::from_str(&normalized).ok()?;
73    let upcoming: Vec<_> = schedule.upcoming(chrono::Utc).take(3).collect();
74    if upcoming.len() < 2 {
75        return None;
76    }
77    upcoming
78        .windows(2)
79        .map(|w| (w[1] - w[0]).num_seconds())
80        .min()
81}
82
83/// Validate that a recurring cron does not fire more often than the configured
84/// minimum interval. Returns a user-facing error string when it fires too often.
85///
86/// Unparseable expressions pass here (return `Ok`) and are rejected later at
87/// next-trigger computation, so this gate never false-rejects a valid cron form
88/// it does not recognize.
89pub fn validate_cron_min_interval(cron_expression: &str) -> Result<(), String> {
90    let min_limit = min_interval_seconds();
91    if let Some(interval) = cron_min_interval_seconds(cron_expression)
92        && interval < min_limit
93    {
94        return Err(format!(
95            "Schedule cron must fire no more than once every {min_limit} seconds (≥ {} min); expression fires every {interval} seconds",
96            min_limit / 60
97        ));
98    }
99    Ok(())
100}
101
102/// Outcome of a failed session-schedule limit check.
103///
104/// Distinguishes a store/count failure (surface as an internal error) from a
105/// limit rejection (surface as a user-facing tool error) so callers preserve the
106/// same behavior they had with the inline checks.
107pub enum ScheduleLimitError {
108    /// Counting active schedules failed.
109    Store(crate::error::AgentLoopError),
110    /// A limit was exceeded; carries the user-facing message.
111    Rejected(String),
112}
113
114/// Enforce the create-time session-schedule limits shared by every agent entry
115/// point (`create_schedule` and `spawn_background` with a `schedule` arg):
116/// per-session cap, per-org cap, and minimum recurring cron interval. Pass the
117/// recurring `cron_expression` (None for one-shot schedules, which skip the
118/// interval gate). Each fire dispatches a real worker turn, so these bound
119/// operator compute on open-signup deployments (see `specs/threat-model.md`
120/// TM-SCHED-001).
121pub async fn validate_schedule_create_limits<T: crate::traits::SessionScheduleStore + ?Sized>(
122    store: &T,
123    session_id: SessionId,
124    cron_expression: Option<&str>,
125) -> std::result::Result<(), ScheduleLimitError> {
126    let per_session = store
127        .count_active_schedules(session_id)
128        .await
129        .map_err(ScheduleLimitError::Store)?;
130    if per_session >= MAX_ACTIVE_SCHEDULES_PER_SESSION {
131        return Err(ScheduleLimitError::Rejected(format!(
132            "Maximum {MAX_ACTIVE_SCHEDULES_PER_SESSION} active schedules per session. Cancel an existing schedule first."
133        )));
134    }
135
136    let max_per_org = max_active_schedules_per_org();
137    let per_org = store
138        .count_active_org_schedules()
139        .await
140        .map_err(ScheduleLimitError::Store)?;
141    if i64::from(per_org) >= max_per_org {
142        return Err(ScheduleLimitError::Rejected(format!(
143            "Maximum {max_per_org} active schedules per org reached. Cancel an existing schedule first."
144        )));
145    }
146
147    if let Some(cron) = cron_expression {
148        validate_cron_min_interval(cron).map_err(ScheduleLimitError::Rejected)?;
149    }
150
151    Ok(())
152}
153
154/// Type of schedule: one-shot or recurring.
155#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
156#[cfg_attr(feature = "openapi", derive(ToSchema))]
157#[serde(rename_all = "lowercase")]
158pub enum ScheduleType {
159    /// Fires once at `scheduled_at` then auto-disables.
160    OneShot,
161    /// Fires on a cron schedule.
162    Recurring,
163}
164
165/// A session-scoped schedule.
166#[derive(Debug, Clone, Serialize, Deserialize)]
167#[cfg_attr(feature = "openapi", derive(ToSchema))]
168pub struct SessionSchedule {
169    /// Unique identifier (format: sched_{32-hex}).
170    #[cfg_attr(feature = "openapi", schema(value_type = String, example = "sched_01933b5a00007000800000000000001"))]
171    pub id: ScheduleId,
172    /// Session this schedule belongs to.
173    #[cfg_attr(feature = "openapi", schema(value_type = String, example = "session_01933b5a00007000800000000000001"))]
174    pub session_id: SessionId,
175    /// Owning principal for this schedule.
176    #[cfg_attr(feature = "openapi", schema(value_type = String, example = "principal_01933b5a000070008000000000000001"))]
177    pub owner_principal_id: PrincipalId,
178    /// Denormalized effective human owner of the owning principal lineage.
179    #[serde(skip_serializing_if = "Option::is_none")]
180    pub resolved_owner_user_id: Option<Uuid>,
181    /// Owning principal summary.
182    #[serde(skip_serializing_if = "Option::is_none")]
183    pub owner: Option<PrincipalSummary>,
184    /// Effective human owner summary.
185    #[serde(skip_serializing_if = "Option::is_none")]
186    pub effective_owner: Option<PrincipalSummary>,
187    /// What the agent should do when the schedule fires.
188    pub description: String,
189    /// Cron expression for recurring schedules (None for one-shot).
190    #[serde(skip_serializing_if = "Option::is_none")]
191    pub cron_expression: Option<String>,
192    /// One-shot trigger time (None for recurring).
193    #[serde(skip_serializing_if = "Option::is_none")]
194    pub scheduled_at: Option<DateTime<Utc>>,
195    /// IANA timezone for cron interpretation.
196    pub timezone: String,
197    /// Whether the schedule is active.
198    pub enabled: bool,
199    /// Computed type based on cron_expression vs scheduled_at.
200    pub schedule_type: ScheduleType,
201    /// Next computed trigger time.
202    #[serde(skip_serializing_if = "Option::is_none")]
203    pub next_trigger_at: Option<DateTime<Utc>>,
204    /// Last time this schedule fired.
205    #[serde(skip_serializing_if = "Option::is_none")]
206    pub last_triggered_at: Option<DateTime<Utc>>,
207    /// Total number of times this schedule has fired.
208    pub trigger_count: u32,
209    pub created_at: DateTime<Utc>,
210    pub updated_at: DateTime<Utc>,
211}
212
213impl SessionSchedule {
214    /// Derive schedule type from fields.
215    pub fn derive_type(cron_expression: &Option<String>) -> ScheduleType {
216        if cron_expression.is_some() {
217            ScheduleType::Recurring
218        } else {
219            ScheduleType::OneShot
220        }
221    }
222}
223
224/// Test-only RAII guard that saves a process-global env var, applies a change,
225/// and restores the original value (or absence) on drop. Env is shared across the
226/// parallel test binary, so saving/restoring keeps env-reading tests independent
227/// of each other and of the environment the suite was launched with.
228#[cfg(test)]
229pub(crate) struct EnvVarGuard {
230    key: &'static str,
231    prev: Option<String>,
232}
233
234#[cfg(test)]
235impl EnvVarGuard {
236    /// Save the current value and unset the var.
237    pub(crate) fn unset(key: &'static str) -> Self {
238        let prev = std::env::var(key).ok();
239        unsafe { std::env::remove_var(key) };
240        Self { key, prev }
241    }
242}
243
244#[cfg(test)]
245impl Drop for EnvVarGuard {
246    fn drop(&mut self) {
247        match &self.prev {
248            Some(v) => unsafe { std::env::set_var(self.key, v) },
249            None => unsafe { std::env::remove_var(self.key) },
250        }
251    }
252}
253
254#[cfg(test)]
255mod limit_tests {
256    use super::*;
257
258    #[test]
259    fn min_interval_default_is_300() {
260        let _g = EnvVarGuard::unset("SESSION_SCHEDULE_MIN_INTERVAL_SECONDS");
261        assert_eq!(min_interval_seconds(), DEFAULT_MIN_INTERVAL_SECONDS);
262    }
263
264    #[test]
265    fn cron_interval_every_minute_is_60() {
266        // 5-field every-minute cron.
267        assert_eq!(cron_min_interval_seconds("* * * * *"), Some(60));
268    }
269
270    #[test]
271    fn cron_interval_every_5_min_is_300() {
272        assert_eq!(cron_min_interval_seconds("*/5 * * * *"), Some(300));
273    }
274
275    #[test]
276    fn cron_interval_six_field_every_30s() {
277        // 6-field (seconds) form, every 30 seconds.
278        assert_eq!(cron_min_interval_seconds("*/30 * * * * *"), Some(30));
279    }
280
281    #[test]
282    fn cron_interval_unparseable_is_none() {
283        assert_eq!(cron_min_interval_seconds("not a cron"), None);
284        assert_eq!(cron_min_interval_seconds("* * *"), None);
285    }
286
287    #[test]
288    fn validate_rejects_every_minute_at_default() {
289        let _g = EnvVarGuard::unset("SESSION_SCHEDULE_MIN_INTERVAL_SECONDS");
290        assert!(validate_cron_min_interval("* * * * *").is_err());
291    }
292
293    #[test]
294    fn validate_accepts_daily() {
295        let _g = EnvVarGuard::unset("SESSION_SCHEDULE_MIN_INTERVAL_SECONDS");
296        assert!(validate_cron_min_interval("0 3 * * *").is_ok());
297    }
298
299    #[test]
300    fn validate_accepts_unparseable() {
301        // Unrecognized forms pass the gate; the create path rejects them later.
302        assert!(validate_cron_min_interval("garbage").is_ok());
303    }
304}
everruns_core/session_schedule.rs

everruns_core/
session_schedule.rs
