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 = 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
54pub(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
152pub(crate) enum BeginRun {
154 Started(ActiveRoutineRun),
155 Skipped,
156}
157
158pub(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 let _ = self.0.unlock();
172 }
173}
174
175#[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#[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#[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 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 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 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, ¤t) {
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 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 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 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 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;