1mod 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;
38pub const MAX_BOT_NAME_BYTES: usize = 128;
40pub 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
56pub(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
195pub(crate) struct RoutinePoll {
197 pub(crate) active: bool,
198 pub(crate) due: Vec<(String, ActiveRoutineRun)>,
199}
200
201pub(crate) enum BeginRun {
203 Started(ActiveRoutineRun),
204 Skipped,
205}
206
207pub(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 let _ = self.0.unlock();
221 }
222}
223
224#[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#[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#[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 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 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 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, ¤t) {
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 #[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 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 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 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 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;