Skip to main content

mobius_gateway/
bots.rs

1//! Gateway-owned Bot profiles, routines, run history, and schedule matching.
2
3mod storage;
4
5use std::collections::BTreeSet;
6use std::fs::{File, OpenOptions, TryLockError};
7use std::io::Read as _;
8#[cfg(unix)]
9use std::os::unix::fs::{MetadataExt as _, OpenOptionsExt as _, PermissionsExt as _};
10use std::path::{Component, Path, PathBuf};
11use std::str::FromStr as _;
12
13use chrono::{TimeZone as _, Timelike as _, Utc};
14use chrono_tz::Tz;
15use croner::Cron;
16use mobius::protocol::MAX_MESSAGE_BYTES;
17use serde::{Deserialize, Serialize};
18use uuid::Uuid;
19
20use self::storage::BotStorage;
21use crate::config::validate_agent_composition;
22use crate::wire::{
23    AgentComposition, BotRecord, ProviderTint, Routine, RoutineRun, RoutineRunStatus,
24    RoutineSchedule, RoutineScheduleKind, VersionedAgentConfig,
25};
26use crate::{Error, Result};
27
28const STATE_VERSION: u32 = 6;
29const STATE_FILE: &str = storage::STATE_FILE;
30const STATE_LOCK_FILE: &str = "bots-state.lock";
31const ROUTINES_DIR: &str = "routines";
32const ROUTINE_SUBMISSION_PREFIX: &str =
33    "# Routine\n\nThe instructions below relate to a routine task.";
34const MAX_ROUTINE_INSTRUCTIONS_BYTES: usize =
35    MAX_MESSAGE_BYTES - ROUTINE_SUBMISSION_PREFIX.len() - 2;
36const MAX_STATE_BYTES: u64 = 1024 * 1024;
37const MAX_HANDLE_BYTES: usize = 64;
38/// Maximum UTF-8 byte length of a Bot display name.
39pub const MAX_BOT_NAME_BYTES: usize = 128;
40/// Maximum UTF-8 byte length of a Bot description.
41pub const MAX_BOT_DESCRIPTION_BYTES: usize = 2 * 1024;
42pub(crate) const MOBIUS_HANDLE: &str = "mobius";
43const USER_HANDLE: &str = "user";
44const MOBIUS_NAME: &str = "Mobius";
45pub(crate) const MOBIUS_DESCRIPTION: &str = "You are möbius, a concise coding agent. Inspect the real code path before editing, make the smallest focused change, and preserve unrelated work.";
46const BOT_TINTS: [ProviderTint; 7] = [
47    ProviderTint::Blue,
48    ProviderTint::Teal,
49    ProviderTint::Green,
50    ProviderTint::Yellow,
51    ProviderTint::Orange,
52    ProviderTint::Red,
53    ProviderTint::Purple,
54];
55
56/// Gateway-wide persistent Bot profiles, routines, and run history.
57pub(crate) struct BotStore {
58    state_dir: PathBuf,
59    routines_dir: PathBuf,
60    storage: BotStorage,
61    pub(crate) prepared: tokio::sync::Mutex<
62        std::collections::BTreeMap<String, std::sync::Arc<crate::assembly::PreparedBot>>,
63    >,
64    pub(crate) preparation_generation: std::sync::atomic::AtomicU64,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
68#[serde(deny_unknown_fields)]
69pub(crate) struct StoredRoutine {
70    pub(crate) id: String,
71    pub(crate) bot_id: String,
72    pub(crate) workspace: PathBuf,
73    pub(crate) instructions: PathBuf,
74    pub(crate) schedule: RoutineSchedule,
75    pub(crate) ends_at: Option<i64>,
76    pub(crate) enabled: bool,
77    pub(crate) next_run_at: Option<i64>,
78    pub(crate) last_matched_minute: Option<i64>,
79}
80
81impl StoredRoutine {
82    fn reset_next_run(&mut self, now: i64) -> Result<()> {
83        self.last_matched_minute = None;
84        self.next_run_at = match self.schedule.kind {
85            RoutineScheduleKind::Once => self.schedule.at,
86            RoutineScheduleKind::Interval => Some(
87                now.checked_add(
88                    i64::try_from(self.schedule.every_seconds.ok_or_else(|| {
89                        Error::Config("interval schedule is missing its interval".into())
90                    })?)
91                    .map_err(|_| Error::Config("interval schedule is too large".into()))?,
92                )
93                .ok_or_else(|| Error::Config("interval schedule overflows its timestamp".into()))?,
94            ),
95            RoutineScheduleKind::Cron => Some(next_cron_occurrence(&self.schedule, now, true)?),
96        };
97        Ok(())
98    }
99
100    fn advance_interval(&mut self, now: i64) -> Result<()> {
101        let every =
102            i64::try_from(self.schedule.every_seconds.ok_or_else(|| {
103                Error::Config("interval schedule is missing its interval".into())
104            })?)
105            .map_err(|_| Error::Config("interval schedule is too large".into()))?;
106        let next = self
107            .next_run_at
108            .ok_or_else(|| Error::Config("interval schedule has no next run".into()))?;
109        let missed = (now.saturating_sub(next) / every).saturating_add(1);
110        self.next_run_at = Some(
111            next.checked_add(every.saturating_mul(missed))
112                .ok_or_else(|| Error::Config("interval schedule overflows its timestamp".into()))?,
113        );
114        Ok(())
115    }
116
117    fn is_finished(&self, now: i64) -> bool {
118        match self.schedule.kind {
119            RoutineScheduleKind::Once => {
120                self.next_run_at.is_none()
121                    || self
122                        .ends_at
123                        .is_some_and(|ends_at| self.next_run_at.is_some_and(|next| next > ends_at))
124            }
125            RoutineScheduleKind::Interval => self
126                .ends_at
127                .is_some_and(|ends_at| self.next_run_at.is_none_or(|next| next > ends_at)),
128            RoutineScheduleKind::Cron => self
129                .ends_at
130                .is_some_and(|ends_at| ends_at.div_euclid(60) < now.div_euclid(60)),
131        }
132    }
133
134    fn next_run_at(&self, now: i64) -> Option<i64> {
135        if self.is_finished(now) || !self.enabled {
136            return None;
137        }
138        if self.schedule.kind != RoutineScheduleKind::Cron {
139            return self.next_run_at;
140        }
141        let next = self
142            .next_run_at
143            .or_else(|| next_cron_occurrence(&self.schedule, now, false).ok())?;
144        self.ends_at
145            .map_or(Some(next), |ends_at| (next <= ends_at).then_some(next))
146    }
147}
148
149fn next_cron_occurrence(
150    schedule: &RoutineSchedule,
151    now: i64,
152    inclusive_current_minute: bool,
153) -> Result<i64> {
154    let expression = schedule
155        .expression
156        .as_deref()
157        .ok_or_else(|| Error::Config("cron schedule is missing its expression".into()))?;
158    let cron = Cron::from_str(expression)
159        .map_err(|error| Error::Config(format!("invalid persisted cron schedule: {error}")))?;
160    let time_zone = schedule
161        .time_zone
162        .as_deref()
163        .ok_or_else(|| Error::Config("cron schedule is missing its time zone".into()))?
164        .parse::<Tz>()
165        .map_err(|error| Error::Config(format!("invalid persisted cron time zone: {error}")))?;
166    let utc_minute = Utc
167        .timestamp_opt(now, 0)
168        .single()
169        .and_then(|time| time.with_second(0))
170        .ok_or_else(|| Error::Config("cron timestamp is outside the supported range".into()))?;
171    let local_time = utc_minute.with_timezone(&time_zone);
172    let minimum = local_time.timestamp();
173    let mut cursor = local_time;
174    let mut inclusive = inclusive_current_minute;
175    let mut previous = None;
176    loop {
177        let next = cron
178            .find_next_occurrence(&cursor, inclusive)
179            .map_err(|error| Error::Config(format!("invalid persisted cron schedule: {error}")))?;
180        let timestamp = next.timestamp();
181        if timestamp > minimum || (inclusive_current_minute && timestamp == minimum) {
182            return Ok(timestamp);
183        }
184        if previous.is_some_and(|previous| timestamp <= previous) {
185            return Err(Error::Config(
186                "persisted cron schedule did not advance its timestamp".into(),
187            ));
188        }
189        previous = Some(timestamp);
190        cursor = next;
191        inclusive = false;
192    }
193}
194
195/// One scheduler tick derived from a single locked catalog snapshot.
196pub(crate) struct RoutinePoll {
197    pub(crate) active: bool,
198    pub(crate) due: Vec<(String, ActiveRoutineRun)>,
199}
200
201/// Result of reserving one task invocation.
202pub(crate) enum BeginRun {
203    Started(ActiveRoutineRun),
204    Skipped,
205}
206
207/// A durable running invocation whose file lock is held until completion.
208pub(crate) struct ActiveRoutineRun {
209    run_id: String,
210    session_id: String,
211    _lock: RoutineLock,
212}
213
214#[derive(Debug)]
215struct RoutineLock(File);
216
217impl Drop for RoutineLock {
218    fn drop(&mut self) {
219        // Closing alone leaves the lock held while a fork inherits the file descriptor.
220        let _ = self.0.unlock();
221    }
222}
223
224/// Validated Bot deletion whose routine locks stay held through gateway cleanup.
225#[derive(Debug)]
226pub(crate) struct BotDeletion {
227    bot_id: String,
228    expected_revision: u64,
229    routine_ids: BTreeSet<String>,
230    instructions: BTreeSet<PathBuf>,
231    state_lock: Option<File>,
232    _routine_locks: Vec<RoutineLock>,
233}
234
235impl BotDeletion {
236    pub(crate) fn release_state_lock(&mut self) {
237        drop(self.state_lock.take());
238    }
239}
240
241/// Validated routine deletion whose lock stays held through gateway cleanup.
242#[derive(Debug)]
243pub(crate) struct RoutineDeletion {
244    routine_id: String,
245    session_ids: BTreeSet<String>,
246    instructions: PathBuf,
247    _state_lock: File,
248    _lock: RoutineLock,
249}
250
251/// Durable forward-recovery record for a cross-owner Bot cascade.
252#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
253#[serde(deny_unknown_fields)]
254pub(crate) struct PendingBotDeletion {
255    pub(crate) bot_id: String,
256    pub(crate) expected_revision: u64,
257    pub(crate) session_roots: Vec<String>,
258    pub(crate) session_ids: Vec<String>,
259    instruction_paths: Vec<PathBuf>,
260}
261
262impl RoutineDeletion {
263    pub(crate) fn session_ids(&self) -> &BTreeSet<String> {
264        &self.session_ids
265    }
266}
267
268impl ActiveRoutineRun {
269    pub(crate) fn session_id(&self) -> &str {
270        &self.session_id
271    }
272}
273
274#[derive(Debug, Clone, Serialize, Deserialize)]
275#[serde(deny_unknown_fields)]
276struct BotState {
277    version: u32,
278    bots: Vec<StoredBot>,
279    routines: Vec<StoredRoutine>,
280    pending_bot_deletion: Option<PendingBotDeletion>,
281}
282
283#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
284#[serde(deny_unknown_fields)]
285struct StoredBot {
286    id: String,
287    handle: String,
288    name: String,
289    description: String,
290    tint: ProviderTint,
291    config: VersionedAgentConfig,
292}
293
294impl StoredBot {
295    fn record(&self) -> Result<BotRecord> {
296        let (accepts_file_attachments, routine_interaction_policy) =
297            crate::assembly::bot_semantics(&self.config.config)?;
298        Ok(BotRecord {
299            id: self.id.clone(),
300            handle: self.handle.clone(),
301            name: self.name.clone(),
302            description: self.description.clone(),
303            tint: self.tint,
304            config: self.config.clone(),
305            accepts_file_attachments,
306            routine_interaction_policy,
307        })
308    }
309}
310
311#[cfg(test)]
312impl From<&BotRecord> for StoredBot {
313    fn from(bot: &BotRecord) -> Self {
314        Self {
315            id: bot.id.clone(),
316            handle: bot.handle.clone(),
317            name: bot.name.clone(),
318            description: bot.description.clone(),
319            tint: bot.tint,
320            config: bot.config.clone(),
321        }
322    }
323}
324
325impl Default for BotState {
326    fn default() -> Self {
327        Self {
328            version: STATE_VERSION,
329            bots: Vec::new(),
330            routines: Vec::new(),
331            pending_bot_deletion: None,
332        }
333    }
334}
335
336impl BotStore {
337    /// Opens or creates owner-only Bot state.
338    pub(crate) fn open(state_dir: &Path) -> Result<Self> {
339        let state_dir = std::fs::canonicalize(state_dir)?;
340        let routines_dir = private_routines_dir(&state_dir)?;
341        let path = state_dir.join(STATE_FILE);
342        let (storage, persisted) = BotStorage::open(&path)?;
343        let store = Self {
344            state_dir,
345            routines_dir,
346            storage,
347            prepared: tokio::sync::Mutex::default(),
348            preparation_generation: std::sync::atomic::AtomicU64::default(),
349        };
350        let state = store.fresh_state()?;
351        if persisted && !state.bots.iter().any(|bot| bot.handle == MOBIUS_HANDLE) {
352            return Err(Error::Config(
353                "persisted Bot state has no built-in @mobius Bot".into(),
354            ));
355        }
356        let bot_ids = state
357            .bots
358            .iter()
359            .map(|bot| bot.id.clone())
360            .collect::<BTreeSet<_>>();
361        store.storage.validate_run_owners(&bot_ids)?;
362        store.storage.recover_interrupted_runs()?;
363        Ok(store)
364    }
365
366    /// Creates the ordinary built-in Bot only before Bot state has ever existed.
367    pub(crate) fn seed_default(
368        &self,
369        defaults: &VersionedAgentConfig,
370    ) -> Result<Option<BotRecord>> {
371        let _file_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
372        _file_lock.lock()?;
373        if self.storage.load_catalog()?.is_some() {
374            return Ok(None);
375        }
376        let config = defaults.config.clone();
377        validate_agent_composition(&config)?;
378        let mut state = BotState::default();
379        let bot = StoredBot {
380            id: Uuid::new_v4().to_string(),
381            handle: MOBIUS_HANDLE.into(),
382            name: MOBIUS_NAME.into(),
383            description: MOBIUS_DESCRIPTION.into(),
384            tint: ProviderTint::default(),
385            config: VersionedAgentConfig {
386                revision: 1,
387                config,
388            },
389        };
390        let record = bot.record()?;
391        state.bots.push(bot);
392        validate_state(&state, &self.routines_dir)?;
393        self.save(&state)?;
394        Ok(Some(record))
395    }
396
397    pub(crate) fn create_bot(
398        &self,
399        name: &str,
400        description: &str,
401        config: AgentComposition,
402    ) -> Result<BotRecord> {
403        let name = validate_name(name)?;
404        let description = validate_description(description)?;
405        validate_agent_composition(&config)?;
406        self.update(|state| {
407            let id = Uuid::new_v4().to_string();
408            let handle = next_handle(state, &name, &id);
409            let tint = next_tint(state);
410            let bot = StoredBot {
411                id,
412                handle,
413                name,
414                description,
415                tint,
416                config: VersionedAgentConfig {
417                    revision: 1,
418                    config,
419                },
420            };
421            let record = bot.record()?;
422            state.bots.push(bot);
423            Ok(record)
424        })
425    }
426
427    pub(crate) fn update_bot(
428        &self,
429        id: &str,
430        expected_revision: u64,
431        name: &str,
432        description: &str,
433        tint: ProviderTint,
434        config: AgentComposition,
435    ) -> Result<BotRecord> {
436        let name = validate_name(name)?;
437        let description = validate_description(description)?;
438        validate_agent_composition(&config)?;
439        self.update(|state| {
440            let handle = next_handle(state, &name, id);
441            let bot = find_bot_mut(state, id)?;
442            if bot.config.revision != expected_revision {
443                return Err(Error::Config(format!(
444                    "Bot configuration revision changed from {expected_revision} to {}",
445                    bot.config.revision
446                )));
447            }
448            if bot.name != name && bot.handle != MOBIUS_HANDLE {
449                bot.handle = handle;
450            }
451            bot.name = name;
452            bot.description = description;
453            bot.tint = tint;
454            let config = VersionedAgentConfig {
455                revision: expected_revision
456                    .checked_add(1)
457                    .ok_or_else(|| Error::Config("Bot revision overflow".into()))?,
458                config,
459            };
460            bot.config = config;
461            bot.record()
462        })
463    }
464
465    pub(crate) fn prepare_bot_deletion(
466        &self,
467        id: &str,
468        expected_revision: u64,
469    ) -> Result<BotDeletion> {
470        let state_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
471        state_lock.lock()?;
472        let state = self.fresh_state()?;
473        let bot = state
474            .bots
475            .iter()
476            .find(|bot| bot.id == id)
477            .ok_or_else(|| Error::Config(format!("unknown Bot `{id}`")))?;
478        if bot.handle == MOBIUS_HANDLE {
479            return Err(Error::Config(
480                "the built-in @mobius Bot cannot be deleted".into(),
481            ));
482        }
483        if bot.config.revision != expected_revision {
484            return Err(Error::Config(format!(
485                "Bot configuration revision changed from {expected_revision} to {}",
486                bot.config.revision
487            )));
488        }
489        let routines = state
490            .routines
491            .iter()
492            .filter(|routine| routine.bot_id == id)
493            .cloned()
494            .collect::<Vec<_>>();
495        let routine_ids = routines
496            .iter()
497            .map(|routine| routine.id.clone())
498            .collect::<BTreeSet<_>>();
499        let instructions = routines
500            .iter()
501            .map(|routine| routine.instructions.clone())
502            .collect::<BTreeSet<_>>();
503        drop(state);
504        let mut routine_locks = Vec::with_capacity(routine_ids.len());
505        for routine_id in &routine_ids {
506            let Some(lock) = self.try_routine_lock(routine_id)? else {
507                return Err(Error::Config(format!(
508                    "routine {routine_id} is currently running"
509                )));
510            };
511            routine_locks.push(lock);
512        }
513        for routine in &routines {
514            self.read_routine_instructions(routine)?;
515        }
516        Ok(BotDeletion {
517            bot_id: id.into(),
518            expected_revision,
519            routine_ids,
520            instructions,
521            state_lock: Some(state_lock),
522            _routine_locks: routine_locks,
523        })
524    }
525
526    pub(crate) fn record_bot_deletion(
527        &self,
528        deletion: &mut BotDeletion,
529        session_roots: &[String],
530        session_ids: &[String],
531    ) -> Result<PendingBotDeletion> {
532        let intent = PendingBotDeletion {
533            bot_id: deletion.bot_id.clone(),
534            expected_revision: deletion.expected_revision,
535            session_roots: session_roots.to_vec(),
536            session_ids: session_ids.to_vec(),
537            instruction_paths: deletion.instructions.iter().cloned().collect(),
538        };
539        let intent = self.update_locked(|state| {
540            let bot = find_bot_mut(state, &intent.bot_id)?;
541            if bot.config.revision != intent.expected_revision {
542                return Err(Error::Config(format!(
543                    "Bot configuration revision changed from {} to {}",
544                    intent.expected_revision, bot.config.revision
545                )));
546            }
547            if let Some(pending) = &state.pending_bot_deletion
548                && pending != &intent
549            {
550                return Err(Error::Config(
551                    "another Bot deletion is awaiting recovery".into(),
552                ));
553            }
554            state.pending_bot_deletion = Some(intent.clone());
555            Ok(intent.clone())
556        })?;
557        deletion.release_state_lock();
558        Ok(intent)
559    }
560
561    pub(crate) fn pending_bot_deletion(&self) -> Result<Option<PendingBotDeletion>> {
562        Ok(self.fresh_state()?.pending_bot_deletion)
563    }
564
565    pub(crate) fn clear_bot_deletion(&self, bot_id: &str) -> Result<()> {
566        let state_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
567        state_lock.lock()?;
568        self.update_locked(|state| {
569            let pending = state
570                .pending_bot_deletion
571                .as_ref()
572                .ok_or_else(|| Error::Config("Bot deletion recovery is not pending".into()))?;
573            if pending.bot_id != bot_id {
574                return Err(Error::Config(
575                    "a different Bot deletion is awaiting recovery".into(),
576                ));
577            }
578            state.pending_bot_deletion = None;
579            Ok(())
580        })
581    }
582
583    pub(crate) fn cleanup_bot_deletion_files(&self, intent: &PendingBotDeletion) -> Result<()> {
584        for path in &intent.instruction_paths {
585            if path.parent() != Some(self.routines_dir.as_path()) {
586                return Err(Error::Config(
587                    "pending Bot deletion instructions left the private routine directory".into(),
588                ));
589            }
590            remove_if_present(path)?;
591        }
592        Ok(())
593    }
594
595    pub(crate) fn delete_bot(&self, deletion: BotDeletion) -> Result<BotRecord> {
596        let BotDeletion {
597            bot_id,
598            expected_revision,
599            routine_ids,
600            instructions,
601            state_lock,
602            _routine_locks,
603        } = deletion;
604        let state_lock = match state_lock {
605            Some(state_lock) => state_lock,
606            None => {
607                let state_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
608                state_lock.lock()?;
609                state_lock
610            }
611        };
612        let mut state = self.fresh_state()?;
613        let index = {
614            if let Some(pending) = &state.pending_bot_deletion
615                && (pending.bot_id != bot_id || pending.expected_revision != expected_revision)
616            {
617                return Err(Error::Config(
618                    "a different Bot deletion is awaiting recovery".into(),
619                ));
620            }
621            let index = state
622                .bots
623                .iter()
624                .position(|bot| bot.id == bot_id)
625                .ok_or_else(|| Error::Config(format!("unknown Bot `{bot_id}`")))?;
626            let bot = &state.bots[index];
627            if bot.handle == MOBIUS_HANDLE {
628                return Err(Error::Config(
629                    "the built-in @mobius Bot cannot be deleted".into(),
630                ));
631            }
632            if bot.config.revision != expected_revision {
633                return Err(Error::Config(format!(
634                    "Bot configuration revision changed from {expected_revision} to {}",
635                    bot.config.revision
636                )));
637            }
638            let current_routine_ids = state
639                .routines
640                .iter()
641                .filter(|routine| routine.bot_id == bot_id)
642                .map(|routine| routine.id.clone())
643                .collect::<BTreeSet<_>>();
644            if current_routine_ids != routine_ids {
645                return Err(Error::Config(
646                    "Bot routine state changed during deletion".into(),
647                ));
648            }
649            let current_instructions = state
650                .routines
651                .iter()
652                .filter(|routine| routine.bot_id == bot_id)
653                .map(|routine| routine.instructions.clone())
654                .collect::<BTreeSet<_>>();
655            if current_instructions != instructions {
656                return Err(Error::Config(
657                    "Bot routine instructions changed during deletion".into(),
658                ));
659            }
660            index
661        };
662        let bot = state.bots.remove(index).record()?;
663        state.routines.retain(|routine| routine.bot_id != bot_id);
664        validate_state(&state, &self.routines_dir)?;
665        let catalog = catalog_json(&state)?;
666        self.storage
667            .delete_runs_and_save_catalog(&catalog, None, Some(&bot_id))?;
668        drop(_routine_locks);
669        drop(state_lock);
670        for path in &instructions {
671            let _ = remove_if_present(path);
672        }
673        Ok(bot)
674    }
675
676    pub(crate) fn bots(&self) -> Result<Vec<BotRecord>> {
677        self.fresh_state()?
678            .bots
679            .iter()
680            .map(StoredBot::record)
681            .collect()
682    }
683
684    pub(crate) fn bot(&self, id: &str) -> Result<BotRecord> {
685        self.fresh_state()?
686            .bots
687            .iter()
688            .find(|bot| bot.id == id)
689            .ok_or_else(|| Error::Config(format!("unknown Bot `{id}`")))?
690            .record()
691    }
692
693    #[cfg(test)]
694    pub(crate) fn mobius(&self) -> Result<BotRecord> {
695        self.fresh_state()?
696            .bots
697            .iter()
698            .find(|bot| bot.handle == MOBIUS_HANDLE)
699            .ok_or_else(|| Error::Config("the built-in @mobius Bot is missing".into()))?
700            .record()
701    }
702
703    /// Writes and registers one Bot-owned routine.
704    pub(crate) fn create_routine(
705        &self,
706        bot_id: &str,
707        workspace: &Path,
708        instructions: &str,
709        schedule: RoutineSchedule,
710        ends_at: Option<i64>,
711    ) -> Result<StoredRoutine> {
712        let workspace = validate_workspace(workspace)?;
713        validate_instructions(instructions)?;
714        validate_schedule(&schedule, ends_at)?;
715        let instructions = instructions.trim();
716        let path = self.new_instruction_path();
717        crate::publication::publish(&path, instructions.as_bytes(), true)?;
718        let result = self.update(|state| {
719            find_bot_mut(state, bot_id)?;
720            let now = Utc::now().timestamp();
721            let mut routine = StoredRoutine {
722                id: Uuid::new_v4().to_string(),
723                bot_id: bot_id.into(),
724                workspace,
725                instructions: path.clone(),
726                schedule,
727                ends_at,
728                enabled: true,
729                next_run_at: Some(now),
730                last_matched_minute: None,
731            };
732            routine.reset_next_run(now)?;
733            state.routines.push(routine.clone());
734            Ok(routine)
735        });
736        match result {
737            Ok(routine) => Ok(routine),
738            Err(error) => match std::fs::remove_file(&path) {
739                Ok(()) => Err(error),
740                Err(rollback) => Err(Error::Config(format!(
741                    "{error}; removing the unregistered routine failed: {rollback}"
742                ))),
743            },
744        }
745    }
746
747    pub(crate) fn routine_records(&self, bot_id: Option<&str>, now: i64) -> Result<Vec<Routine>> {
748        let state = self.fresh_state()?;
749        state
750            .routines
751            .iter()
752            .filter(|stored| bot_id.is_none_or(|bot_id| stored.bot_id == bot_id))
753            .map(|stored| self.routine_record_from(stored, now))
754            .collect()
755    }
756
757    pub(crate) fn routine_record(&self, id: &str, now: i64) -> Result<Routine> {
758        let state = self.fresh_state()?;
759        let stored = state
760            .routines
761            .iter()
762            .find(|routine| routine.id == id)
763            .ok_or_else(|| Error::Config(format!("unknown routine `{id}`")))?;
764        self.routine_record_from(stored, now)
765    }
766
767    pub(crate) fn has_active_routines(&self, now: i64) -> Result<bool> {
768        let state = self.fresh_state()?;
769        Ok(state
770            .routines
771            .iter()
772            .any(|routine| routine.enabled && !routine.is_finished(now))
773            || self.storage.has_running_routines()?)
774    }
775
776    #[expect(
777        clippy::too_many_arguments,
778        reason = "one routine replacement keeps its validated fields explicit"
779    )]
780    pub(crate) fn update_routine(
781        &self,
782        id: &str,
783        bot_id: &str,
784        workspace: &Path,
785        instructions: &str,
786        schedule: RoutineSchedule,
787        ends_at: Option<i64>,
788        enabled: bool,
789    ) -> Result<StoredRoutine> {
790        self.bot(bot_id)?;
791        let workspace = validate_workspace(workspace)?;
792        validate_instructions(instructions)?;
793        validate_schedule(&schedule, ends_at)?;
794        let existing = self.routine(id)?;
795        let Some(_lock) = self.try_routine_lock(&existing.id)? else {
796            return Err(Error::Config(format!(
797                "routine {} is currently running",
798                existing.id
799            )));
800        };
801        let path = self.new_instruction_path();
802        crate::publication::publish(&path, instructions.trim().as_bytes(), true)?;
803        let result = self.update(|state| {
804            find_bot_mut(state, bot_id)?;
805            let index = resolve_routine(&state.routines, &existing.id)?;
806            let stored = &mut state.routines[index];
807            stored.bot_id = bot_id.into();
808            stored.workspace = workspace;
809            stored.instructions.clone_from(&path);
810            stored.schedule = schedule;
811            stored.ends_at = ends_at;
812            stored.enabled = enabled;
813            stored.reset_next_run(Utc::now().timestamp())?;
814            Ok(state.routines[index].clone())
815        });
816        match result {
817            Ok(routine) => {
818                let _ = remove_if_present(&existing.instructions);
819                Ok(routine)
820            }
821            Err(error) => match remove_if_present(&path) {
822                Ok(()) => Err(error),
823                Err(cleanup) => Err(Error::Config(format!(
824                    "{error}; removing the unregistered routine instructions failed: {cleanup}"
825                ))),
826            },
827        }
828    }
829
830    pub(crate) fn prepare_routine_deletion(&self, id: &str) -> Result<RoutineDeletion> {
831        let state_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
832        state_lock.lock()?;
833        let routine = self.routine(id)?;
834        let Some(lock) = self.try_routine_lock(&routine.id)? else {
835            return Err(Error::Config(format!(
836                "routine {} is currently running",
837                routine.id
838            )));
839        };
840        self.read_routine_instructions(&routine)?;
841        let state = self.fresh_state()?;
842        let index = resolve_routine(&state.routines, &routine.id)?;
843        let routine = &state.routines[index];
844        let session_ids = self
845            .storage
846            .session_ids_for_routine(&routine.id)?
847            .into_iter()
848            .collect();
849        Ok(RoutineDeletion {
850            routine_id: routine.id.clone(),
851            session_ids,
852            instructions: routine.instructions.clone(),
853            _state_lock: state_lock,
854            _lock: lock,
855        })
856    }
857
858    pub(crate) fn delete_routine(&self, deletion: RoutineDeletion) -> Result<StoredRoutine> {
859        let RoutineDeletion {
860            routine_id,
861            session_ids,
862            instructions,
863            _state_lock,
864            _lock,
865        } = deletion;
866        let mut state = self.fresh_state()?;
867        let index = {
868            let index = resolve_routine(&state.routines, &routine_id)?;
869            if state.routines[index].instructions != instructions {
870                return Err(Error::Config(
871                    "routine instructions changed during deletion".into(),
872                ));
873            }
874            let current_session_ids = self
875                .storage
876                .session_ids_for_routine(&routine_id)?
877                .into_iter()
878                .collect::<BTreeSet<_>>();
879            if current_session_ids != session_ids {
880                return Err(Error::Config(
881                    "routine run state changed during deletion".into(),
882                ));
883            }
884            index
885        };
886        let deleted = state.routines.remove(index);
887        validate_state(&state, &self.routines_dir)?;
888        let catalog = catalog_json(&state)?;
889        self.storage
890            .delete_runs_and_save_catalog(&catalog, Some(&routine_id), None)?;
891        drop(_lock);
892        drop(_state_lock);
893        let _ = remove_if_present(&instructions);
894        Ok(deleted)
895    }
896
897    pub(crate) fn routine(&self, id: &str) -> Result<StoredRoutine> {
898        let state = self.fresh_state()?;
899        Ok(state.routines[resolve_routine(&state.routines, id)?].clone())
900    }
901
902    pub(crate) fn routine_input(&self, id: &str) -> Result<(StoredRoutine, String)> {
903        let state = self.fresh_state()?;
904        let routine = state
905            .routines
906            .iter()
907            .find(|routine| routine.id == id)
908            .ok_or_else(|| Error::Config(format!("unknown routine `{id}`")))?;
909        let instructions = self.read_routine_instructions(routine)?;
910        let input = format!("{ROUTINE_SUBMISSION_PREFIX}\n\n{instructions}");
911        Ok((routine.clone(), input))
912    }
913
914    fn routine_record_from(&self, stored: &StoredRoutine, now: i64) -> Result<Routine> {
915        Ok(Routine {
916            id: stored.id.clone(),
917            bot_id: stored.bot_id.clone(),
918            workspace: stored.workspace.clone(),
919            instructions: self.read_routine_instructions(stored)?,
920            schedule: stored.schedule.clone(),
921            ends_at: stored.ends_at,
922            enabled: stored.enabled,
923            finished: stored.is_finished(now),
924            next_run_at: stored.next_run_at(now),
925        })
926    }
927
928    fn read_routine_instructions(&self, routine: &StoredRoutine) -> Result<String> {
929        let path = std::fs::canonicalize(&routine.instructions)?;
930        if !path.is_file() || path.parent() != Some(self.routines_dir.as_path()) {
931            return Err(Error::Config(
932                "routine instructions must remain inside the private gateway routine directory"
933                    .into(),
934            ));
935        }
936        let mut file = File::open(&path)?;
937        let opened = file.metadata()?;
938        let verified = std::fs::canonicalize(&routine.instructions)?;
939        let current = std::fs::metadata(&verified)?;
940        if verified != path || !same_file(&opened, &current) {
941            return Err(Error::Config(
942                "routine instructions changed while they were being opened".into(),
943            ));
944        }
945        let limit = u64::try_from(MAX_ROUTINE_INSTRUCTIONS_BYTES).unwrap_or(u64::MAX);
946        let mut bytes = Vec::new();
947        std::io::Read::by_ref(&mut file)
948            .take(limit + 1)
949            .read_to_end(&mut bytes)?;
950        if bytes.len() > MAX_ROUTINE_INSTRUCTIONS_BYTES {
951            return Err(Error::Config(format!(
952                "routine instructions exceed the {MAX_ROUTINE_INSTRUCTIONS_BYTES}-byte input limit"
953            )));
954        }
955        let input = String::from_utf8(bytes)
956            .map_err(|_| Error::Config("routine instructions are not valid UTF-8".into()))?;
957        validate_instructions(&input)?;
958        Ok(input)
959    }
960
961    /// Reserves due routines and records their invocations atomically.
962    #[cfg(test)]
963    pub(crate) fn take_due(&self, now: i64) -> Result<Vec<(String, ActiveRoutineRun)>> {
964        Ok(self.poll_due(now)?.due)
965    }
966
967    /// Observes routine activity and reserves due work from one state load.
968    pub(crate) fn poll_due(&self, now: i64) -> Result<RoutinePoll> {
969        let state_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
970        state_lock.lock()?;
971        let mut state = self.fresh_state()?;
972        let active = state
973            .routines
974            .iter()
975            .any(|routine| routine.enabled && !routine.is_finished(now))
976            || self.storage.has_running_routines()?;
977        if state.pending_bot_deletion.is_some() {
978            return Ok(RoutinePoll {
979                active,
980                due: Vec::new(),
981            });
982        }
983        let minute = now.div_euclid(60);
984        let mut runs = Vec::new();
985        let mut due = Vec::new();
986        let mut schedule_changed = false;
987        for routine in &mut state.routines {
988            if routine.enabled
989                && !routine.is_finished(now)
990                && routine.schedule.kind == RoutineScheduleKind::Cron
991            {
992                let stale = routine
993                    .next_run_at
994                    .is_some_and(|next| next.div_euclid(60) < minute);
995                if routine.next_run_at.is_none() || stale {
996                    routine.next_run_at = Some(next_cron_occurrence(&routine.schedule, now, true)?);
997                    schedule_changed = true;
998                }
999            }
1000        }
1001        for index in 0..state.routines.len() {
1002            let routine = &state.routines[index];
1003            if !routine.enabled || routine.is_finished(now) {
1004                continue;
1005            }
1006            let should_run = routine.last_matched_minute != Some(minute)
1007                && routine.next_run_at.is_some_and(|next| next <= now);
1008            if !should_run {
1009                continue;
1010            }
1011            let routine = state.routines[index].clone();
1012            {
1013                let stored = &mut state.routines[index];
1014                stored.last_matched_minute = Some(minute);
1015                match stored.schedule.kind {
1016                    RoutineScheduleKind::Once => stored.next_run_at = None,
1017                    RoutineScheduleKind::Interval => stored.advance_interval(now)?,
1018                    RoutineScheduleKind::Cron => {
1019                        stored.next_run_at =
1020                            Some(next_cron_occurrence(&stored.schedule, now, false)?);
1021                    }
1022                }
1023                schedule_changed = true;
1024            }
1025            let run = match self.try_routine_lock(&routine.id)? {
1026                Some(lock) => {
1027                    let run = new_run(&routine, RoutineRunStatus::Running, None);
1028                    due.push((
1029                        routine.id,
1030                        ActiveRoutineRun {
1031                            run_id: run.id.clone(),
1032                            session_id: run
1033                                .session_id
1034                                .clone()
1035                                .expect("a running routine reserves its session ID"),
1036                            _lock: lock,
1037                        },
1038                    ));
1039                    run
1040                }
1041                None => new_run(
1042                    &routine,
1043                    RoutineRunStatus::Skipped,
1044                    Some("the previous invocation is still running".into()),
1045                ),
1046            };
1047            runs.push(run);
1048        }
1049        if !runs.is_empty() {
1050            validate_state(&state, &self.routines_dir)?;
1051            let catalog = catalog_json(&state)?;
1052            self.storage.save_catalog_and_runs(&catalog, &runs)?;
1053        } else if schedule_changed {
1054            validate_state(&state, &self.routines_dir)?;
1055            self.save(&state)?;
1056        }
1057        Ok(RoutinePoll { active, due })
1058    }
1059
1060    /// Starts an overlap-locked invocation or records an overlap skip.
1061    pub(crate) fn begin_run(&self, id: &str) -> Result<BeginRun> {
1062        self.begin_run_inner(id, || {})
1063    }
1064
1065    fn begin_run_inner(&self, id: &str, after_resolve: impl FnOnce()) -> Result<BeginRun> {
1066        let routine = self.stored_routine(id)?;
1067        after_resolve();
1068        let Some(lock) = self.try_routine_lock(&routine.id)? else {
1069            let _file_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
1070            _file_lock.lock()?;
1071            let state = self.fresh_state()?;
1072            if state.pending_bot_deletion.is_some() {
1073                return Err(Error::Config(
1074                    "Bot deletion recovery must finish before changing Bot state".into(),
1075                ));
1076            }
1077            let routine = state
1078                .routines
1079                .iter()
1080                .find(|stored| stored.id == routine.id)
1081                .cloned()
1082                .ok_or_else(|| Error::Config(format!("unknown routine `{}`", routine.id)))?;
1083            if !self.storage.has_running(&routine.id)? {
1084                return Err(Error::Config(format!(
1085                    "routine {} is currently being modified",
1086                    routine.id
1087                )));
1088            }
1089            self.storage.insert_run(&new_run(
1090                &routine,
1091                RoutineRunStatus::Skipped,
1092                Some("the previous invocation is still running".into()),
1093            ))?;
1094            return Ok(BeginRun::Skipped);
1095        };
1096        let routine = self.stored_routine(&routine.id)?;
1097        let run = new_run(&routine, RoutineRunStatus::Running, None);
1098        self.insert_run(&run)?;
1099        Ok(BeginRun::Started(ActiveRoutineRun {
1100            run_id: run.id,
1101            session_id: run
1102                .session_id
1103                .expect("a running routine reserves its session ID"),
1104            _lock: lock,
1105        }))
1106    }
1107
1108    /// Completes a running invocation and releases its overlap lock.
1109    pub(crate) fn finish_run(
1110        &self,
1111        run: ActiveRoutineRun,
1112        status: RoutineRunStatus,
1113        message: Option<String>,
1114    ) -> Result<RoutineRun> {
1115        if status == RoutineRunStatus::Running {
1116            return Err(Error::Config(
1117                "a completed routine run cannot remain running".into(),
1118            ));
1119        }
1120        let state_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
1121        state_lock.lock()?;
1122        self.storage
1123            .finish_run(&run.run_id, status, Utc::now().timestamp(), message)
1124    }
1125
1126    /// Returns newest-first run history for one routine.
1127    pub(crate) fn history(&self, id: Option<&str>) -> Result<Vec<RoutineRun>> {
1128        let state = self.fresh_state()?;
1129        let routine_id = id
1130            .map(|id| self.resolve_history_routine(&state, id))
1131            .transpose()?;
1132        self.storage.history(routine_id.as_deref())
1133    }
1134
1135    fn resolve_history_routine(&self, state: &BotState, id: &str) -> Result<String> {
1136        validate_routine_id_prefix(id)?;
1137        let history_ids = self.storage.history_routine_candidates(id)?;
1138        let mut ids = state
1139            .routines
1140            .iter()
1141            .map(|routine| routine.id.as_str())
1142            .chain(history_ids.iter().map(String::as_str))
1143            .filter(|routine_id| routine_id.starts_with(id))
1144            .collect::<BTreeSet<_>>();
1145        if ids.contains(id) {
1146            return Ok(id.into());
1147        }
1148        let resolved = ids
1149            .pop_first()
1150            .ok_or_else(|| Error::Config(format!("unknown routine `{id}`")))?;
1151        if !ids.is_empty() {
1152            return Err(Error::Config(format!(
1153                "routine ID prefix `{id}` is ambiguous"
1154            )));
1155        }
1156        Ok(resolved.into())
1157    }
1158
1159    pub(crate) fn run(&self, id: &str) -> Result<RoutineRun> {
1160        let _state = self.fresh_state()?;
1161        self.storage.run(id)
1162    }
1163
1164    pub(crate) fn delete_run(&self, id: &str) -> Result<RoutineRun> {
1165        let _file_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
1166        _file_lock.lock()?;
1167        let state = self.fresh_state()?;
1168        if state.pending_bot_deletion.is_some() {
1169            return Err(Error::Config(
1170                "Bot deletion recovery must finish before changing Bot state".into(),
1171            ));
1172        }
1173        self.storage.delete_run(id)
1174    }
1175
1176    fn stored_routine(&self, id: &str) -> Result<StoredRoutine> {
1177        self.fresh_state()?
1178            .routines
1179            .into_iter()
1180            .find(|routine| routine.id == id)
1181            .ok_or_else(|| Error::Config(format!("unknown routine `{id}`")))
1182    }
1183
1184    fn try_routine_lock(&self, id: &str) -> Result<Option<RoutineLock>> {
1185        let file = open_private_lock(self.state_dir.join(format!("routine-{id}.lock")))?;
1186        match file.try_lock() {
1187            Ok(()) => Ok(Some(RoutineLock(file))),
1188            Err(TryLockError::WouldBlock) => Ok(None),
1189            Err(TryLockError::Error(error)) => Err(error.into()),
1190        }
1191    }
1192
1193    fn new_instruction_path(&self) -> PathBuf {
1194        self.routines_dir
1195            .join(format!("{}.md", Uuid::new_v4().as_hyphenated()))
1196    }
1197
1198    fn update<T>(&self, mutate: impl FnOnce(&mut BotState) -> Result<T>) -> Result<T> {
1199        let _file_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
1200        _file_lock.lock()?;
1201        self.update_locked(|state| {
1202            if state.pending_bot_deletion.is_some() {
1203                return Err(Error::Config(
1204                    "Bot deletion recovery must finish before changing Bot state".into(),
1205                ));
1206            }
1207            mutate(state)
1208        })
1209    }
1210
1211    fn update_locked<T>(&self, mutate: impl FnOnce(&mut BotState) -> Result<T>) -> Result<T> {
1212        let mut state = self.fresh_state()?;
1213        let result = mutate(&mut state)?;
1214        validate_state(&state, &self.routines_dir)?;
1215        self.save(&state)?;
1216        Ok(result)
1217    }
1218
1219    fn save(&self, state: &BotState) -> Result<()> {
1220        self.storage.save_catalog(&catalog_json(state)?)
1221    }
1222
1223    fn insert_run(&self, run: &RoutineRun) -> Result<()> {
1224        let _file_lock = open_private_lock(self.state_dir.join(STATE_LOCK_FILE))?;
1225        _file_lock.lock()?;
1226        let state = self.fresh_state()?;
1227        if state.pending_bot_deletion.is_some() {
1228            return Err(Error::Config(
1229                "Bot deletion recovery must finish before changing Bot state".into(),
1230            ));
1231        }
1232        self.storage.insert_run(run)
1233    }
1234
1235    fn fresh_state(&self) -> Result<BotState> {
1236        let state = self
1237            .storage
1238            .load_catalog()?
1239            .map(|contents| serde_json::from_str(&contents))
1240            .transpose()?
1241            .unwrap_or_default();
1242        validate_state(&state, &self.routines_dir)?;
1243        Ok(state)
1244    }
1245}
1246
1247fn validate_session_id(session_id: &str) -> Result<()> {
1248    if session_id.trim().is_empty() {
1249        return Err(Error::Config("routine session ID cannot be empty".into()));
1250    }
1251    Ok(())
1252}
1253
1254fn catalog_json(state: &BotState) -> Result<String> {
1255    let contents = serde_json::to_string_pretty(state)?;
1256    if u64::try_from(contents.len()).unwrap_or(u64::MAX) > MAX_STATE_BYTES {
1257        return Err(Error::Config("Bot state is too large".into()));
1258    }
1259    Ok(contents)
1260}
1261
1262fn next_handle(state: &BotState, name: &str, id: &str) -> String {
1263    let mut base = String::new();
1264    let mut separator = false;
1265    for character in name.chars() {
1266        if character.is_ascii_alphanumeric() {
1267            if separator && !base.is_empty() && base.len() < MAX_HANDLE_BYTES {
1268                base.push('-');
1269            }
1270            separator = false;
1271            if base.len() < MAX_HANDLE_BYTES {
1272                base.push(character.to_ascii_lowercase());
1273            }
1274        } else {
1275            separator = true;
1276        }
1277    }
1278    while base.ends_with('-') {
1279        base.pop();
1280    }
1281    if base.is_empty() {
1282        base.push_str("bot");
1283    }
1284    if base != USER_HANDLE
1285        && !state
1286            .bots
1287            .iter()
1288            .any(|bot| bot.id != id && bot.handle == base)
1289    {
1290        return base;
1291    }
1292    for index in 2_u64.. {
1293        let suffix = format!("-{index}");
1294        let prefix_len = MAX_HANDLE_BYTES.saturating_sub(suffix.len());
1295        let prefix = base[..base.len().min(prefix_len)].trim_end_matches('-');
1296        let candidate = format!("{prefix}{suffix}");
1297        if candidate != USER_HANDLE
1298            && !state
1299                .bots
1300                .iter()
1301                .any(|bot| bot.id != id && bot.handle == candidate)
1302        {
1303            return candidate;
1304        }
1305    }
1306    unreachable!("the Bot handle suffix space is unbounded")
1307}
1308
1309fn next_tint(state: &BotState) -> ProviderTint {
1310    BOT_TINTS
1311        .iter()
1312        .copied()
1313        .find(|tint| state.bots.iter().all(|bot| bot.tint != *tint))
1314        .unwrap_or(BOT_TINTS[state.bots.len() % BOT_TINTS.len()])
1315}
1316
1317fn validate_handle(handle: &str) -> Result<String> {
1318    let handle = handle.trim();
1319    if handle.is_empty()
1320        || handle.len() > MAX_HANDLE_BYTES
1321        || !handle.bytes().all(|byte| {
1322            byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_')
1323        })
1324    {
1325        return Err(Error::Config(format!(
1326            "Bot handle must be 1–{MAX_HANDLE_BYTES} lowercase ASCII letters, digits, dashes, or underscores"
1327        )));
1328    }
1329    if handle == USER_HANDLE {
1330        return Err(Error::Config("Bot handle `user` is reserved".into()));
1331    }
1332    Ok(handle.into())
1333}
1334
1335fn validate_name(name: &str) -> Result<String> {
1336    let name = name.trim();
1337    if name.is_empty() || name.len() > MAX_BOT_NAME_BYTES {
1338        return Err(Error::Config(format!(
1339            "Bot name must be 1–{MAX_BOT_NAME_BYTES} bytes"
1340        )));
1341    }
1342    Ok(name.into())
1343}
1344
1345fn validate_description(description: &str) -> Result<String> {
1346    let description = description.trim();
1347    if description.is_empty() || description.len() > MAX_BOT_DESCRIPTION_BYTES {
1348        return Err(Error::Config(format!(
1349            "Bot description must be 1–{MAX_BOT_DESCRIPTION_BYTES} bytes"
1350        )));
1351    }
1352    Ok(description.into())
1353}
1354
1355fn validate_workspace(workspace: &Path) -> Result<PathBuf> {
1356    let workspace = std::fs::canonicalize(workspace)?;
1357    if !workspace.is_dir() {
1358        return Err(Error::Config(
1359            "routine workspace must be a directory".into(),
1360        ));
1361    }
1362    Ok(workspace)
1363}
1364
1365fn validate_stored_workspace(workspace: &Path) -> Result<()> {
1366    if !workspace.is_absolute()
1367        || workspace
1368            .components()
1369            .any(|component| matches!(component, Component::CurDir | Component::ParentDir))
1370    {
1371        return Err(Error::Config(
1372            "persisted routine workspace must be an absolute normalized path".into(),
1373        ));
1374    }
1375    Ok(())
1376}
1377
1378fn validate_routine_id_prefix(id: &str) -> Result<()> {
1379    if id.is_empty() || id.chars().any(char::is_whitespace) {
1380        return Err(Error::Config("routine ID cannot be empty".into()));
1381    }
1382    Ok(())
1383}
1384
1385fn validate_instructions(instructions: &str) -> Result<()> {
1386    let instructions = instructions.trim();
1387    if instructions.is_empty() {
1388        return Err(Error::Config("routine instructions cannot be empty".into()));
1389    }
1390    if instructions.len() > MAX_ROUTINE_INSTRUCTIONS_BYTES {
1391        return Err(Error::Config(format!(
1392            "routine instructions exceed the {MAX_ROUTINE_INSTRUCTIONS_BYTES}-byte input limit"
1393        )));
1394    }
1395    Ok(())
1396}
1397
1398fn validate_schedule(schedule: &RoutineSchedule, ends_at: Option<i64>) -> Result<()> {
1399    let populated = [
1400        schedule.at.is_some(),
1401        schedule.every_seconds.is_some(),
1402        schedule.expression.is_some(),
1403    ]
1404    .into_iter()
1405    .filter(|populated| *populated)
1406    .count();
1407    match schedule.kind {
1408        RoutineScheduleKind::Once
1409            if populated == 1 && schedule.at.is_some() && schedule.time_zone.is_none() =>
1410        {
1411            if ends_at.is_some_and(|ends_at| schedule.at.is_some_and(|at| at > ends_at)) {
1412                return Err(Error::Config(
1413                    "a once schedule cannot end before it runs".into(),
1414                ));
1415            }
1416        }
1417        RoutineScheduleKind::Interval
1418            if populated == 1
1419                && schedule.every_seconds.is_some()
1420                && schedule.time_zone.is_none() =>
1421        {
1422            if schedule.every_seconds.unwrap_or_default() < 60 {
1423                return Err(Error::Config("interval must be at least 60 seconds".into()));
1424            }
1425        }
1426        RoutineScheduleKind::Cron
1427            if populated == 1 && schedule.expression.is_some() && schedule.time_zone.is_some() =>
1428        {
1429            let time_zone = schedule.time_zone.as_deref().unwrap_or_default();
1430            time_zone
1431                .parse::<Tz>()
1432                .map_err(|error| Error::Config(format!("invalid cron time zone: {error}")))?;
1433            let expression = schedule.expression.as_deref().unwrap_or_default();
1434            let fields = expression.split_ascii_whitespace().collect::<Vec<_>>();
1435            if fields.len() != 5
1436                || fields.iter().any(|field| {
1437                    field.is_empty()
1438                        || !field.chars().all(|character| {
1439                            character.is_ascii_alphanumeric()
1440                                || matches!(character, '*' | '/' | ',' | '-')
1441                        })
1442                })
1443            {
1444                return Err(Error::Config(
1445                    "schedule must be a five-field cron expression".into(),
1446                ));
1447            }
1448            Cron::from_str(expression)
1449                .map_err(|error| Error::Config(format!("invalid cron schedule: {error}")))?;
1450        }
1451        _ => {
1452            return Err(Error::Config(
1453                "schedule fields do not match the selected schedule kind".into(),
1454            ));
1455        }
1456    }
1457    if ends_at.is_some_and(|ends_at| ends_at <= 0) {
1458        return Err(Error::Config("schedule end time must be positive".into()));
1459    }
1460    Ok(())
1461}
1462
1463fn validate_state(state: &BotState, routines_dir: &Path) -> Result<()> {
1464    if state.version != STATE_VERSION {
1465        return Err(Error::Config(format!(
1466            "unsupported Bot state version {}",
1467            state.version
1468        )));
1469    }
1470    let mut bot_ids = BTreeSet::new();
1471    let mut handles = BTreeSet::new();
1472    for bot in &state.bots {
1473        let parsed = Uuid::parse_str(&bot.id)
1474            .map_err(|_| Error::Config("invalid persisted Bot ID".into()))?;
1475        if parsed.to_string() != bot.id || !bot_ids.insert(bot.id.as_str()) {
1476            return Err(Error::Config("duplicate persisted Bot ID".into()));
1477        }
1478        if !handles.insert(bot.handle.as_str()) {
1479            return Err(Error::Config("duplicate persisted Bot handle".into()));
1480        }
1481        if validate_handle(&bot.handle)? != bot.handle
1482            || validate_name(&bot.name)? != bot.name
1483            || validate_description(&bot.description)? != bot.description
1484        {
1485            return Err(Error::Config(
1486                "persisted Bot identity is not normalized".into(),
1487            ));
1488        }
1489        if bot.config.revision == 0 {
1490            return Err(Error::Config(
1491                "persisted Bot revision must be positive".into(),
1492            ));
1493        }
1494        validate_agent_composition(&bot.config.config)?;
1495    }
1496    let mut ids = BTreeSet::new();
1497    let mut paths = BTreeSet::new();
1498    for routine in &state.routines {
1499        let parsed = Uuid::parse_str(&routine.id)
1500            .map_err(|_| Error::Config("invalid persisted routine ID".into()))?;
1501        if parsed.to_string() != routine.id || !ids.insert(routine.id.as_str()) {
1502            return Err(Error::Config("duplicate persisted routine ID".into()));
1503        }
1504        if !bot_ids.contains(routine.bot_id.as_str()) {
1505            return Err(Error::Config("persisted routine has no Bot".into()));
1506        }
1507        validate_stored_workspace(&routine.workspace)?;
1508        if !routine.instructions.is_absolute()
1509            || routine.instructions.parent() != Some(routines_dir)
1510            || !paths.insert(routine.instructions.as_path())
1511        {
1512            return Err(Error::Config(
1513                "persisted routine path is outside the private gateway routine directory".into(),
1514            ));
1515        }
1516        validate_schedule(&routine.schedule, routine.ends_at)?;
1517        if routine.next_run_at.is_some_and(|next| next <= 0) {
1518            return Err(Error::Config("invalid persisted routine next run".into()));
1519        }
1520    }
1521    if let Some(pending) = &state.pending_bot_deletion {
1522        let parsed = Uuid::parse_str(&pending.bot_id)
1523            .map_err(|_| Error::Config("invalid pending Bot deletion ID".into()))?;
1524        if parsed.to_string() != pending.bot_id || pending.expected_revision == 0 {
1525            return Err(Error::Config("invalid pending Bot deletion".into()));
1526        }
1527        if let Some(bot) = state.bots.iter().find(|bot| bot.id == pending.bot_id)
1528            && bot.config.revision != pending.expected_revision
1529        {
1530            return Err(Error::Config(
1531                "pending Bot deletion revision changed".into(),
1532            ));
1533        }
1534        for session_id in pending.session_roots.iter().chain(&pending.session_ids) {
1535            validate_session_id(session_id)?;
1536        }
1537        if pending
1538            .session_roots
1539            .iter()
1540            .any(|root| !pending.session_ids.contains(root))
1541        {
1542            return Err(Error::Config(
1543                "pending Bot deletion root is outside its session set".into(),
1544            ));
1545        }
1546        if pending
1547            .instruction_paths
1548            .iter()
1549            .any(|path| !path.is_absolute() || path.parent() != Some(routines_dir))
1550        {
1551            return Err(Error::Config(
1552                "pending Bot deletion instructions are outside the private routine directory"
1553                    .into(),
1554            ));
1555        }
1556    }
1557    Ok(())
1558}
1559
1560fn resolve_routine(routines: &[StoredRoutine], id: &str) -> Result<usize> {
1561    validate_routine_id_prefix(id)?;
1562    if let Some(index) = routines.iter().position(|routine| routine.id == id) {
1563        return Ok(index);
1564    }
1565    let mut matches = routines
1566        .iter()
1567        .enumerate()
1568        .filter(|(_, routine)| routine.id.starts_with(id));
1569    let (index, _) = matches
1570        .next()
1571        .ok_or_else(|| Error::Config(format!("unknown routine `{id}`")))?;
1572    if matches.next().is_some() {
1573        return Err(Error::Config(format!(
1574            "routine ID prefix `{id}` is ambiguous"
1575        )));
1576    }
1577    Ok(index)
1578}
1579
1580fn new_run(
1581    routine: &StoredRoutine,
1582    status: RoutineRunStatus,
1583    message: Option<String>,
1584) -> RoutineRun {
1585    let now = Utc::now().timestamp();
1586    RoutineRun {
1587        id: Uuid::new_v4().to_string(),
1588        routine_id: routine.id.clone(),
1589        bot_id: routine.bot_id.clone(),
1590        started_at: now,
1591        finished_at: (status != RoutineRunStatus::Running).then_some(now),
1592        status,
1593        session_id: (status == RoutineRunStatus::Running).then(|| Uuid::new_v4().to_string()),
1594        message,
1595    }
1596}
1597
1598fn find_bot_mut<'a>(state: &'a mut BotState, id: &str) -> Result<&'a mut StoredBot> {
1599    state
1600        .bots
1601        .iter_mut()
1602        .find(|bot| bot.id == id)
1603        .ok_or_else(|| Error::Config(format!("unknown Bot `{id}`")))
1604}
1605
1606fn open_private_lock(path: PathBuf) -> Result<File> {
1607    let mut options = OpenOptions::new();
1608    options.read(true).write(true).create(true).truncate(false);
1609    #[cfg(unix)]
1610    options.mode(0o600);
1611    let file = options.open(path)?;
1612    #[cfg(unix)]
1613    file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
1614    Ok(file)
1615}
1616
1617fn private_routines_dir(state_dir: &Path) -> Result<PathBuf> {
1618    let path = state_dir.join(ROUTINES_DIR);
1619    std::fs::create_dir_all(&path)?;
1620    let path = std::fs::canonicalize(path)?;
1621    if path.parent() != Some(state_dir) || !path.is_dir() {
1622        return Err(Error::Config(
1623            "gateway routine directory must be a real directory inside gateway state".into(),
1624        ));
1625    }
1626    #[cfg(unix)]
1627    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
1628    Ok(path)
1629}
1630
1631fn remove_if_present(path: &Path) -> Result<()> {
1632    match std::fs::remove_file(path) {
1633        Ok(()) => Ok(()),
1634        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
1635        Err(error) => Err(error.into()),
1636    }
1637}
1638
1639#[cfg(unix)]
1640fn same_file(left: &std::fs::Metadata, right: &std::fs::Metadata) -> bool {
1641    left.dev() == right.dev() && left.ino() == right.ino()
1642}
1643
1644#[cfg(not(unix))]
1645fn same_file(_left: &std::fs::Metadata, _right: &std::fs::Metadata) -> bool {
1646    true
1647}
1648
1649#[cfg(test)]
1650mod tests;