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