use crate::error::AppError;
use crate::paths::workbenches_dir;
use crate::routine_storage::{append_manual_trigger_log, append_skip_log, write_routine};
use crate::utils::lock::LockRecover;
use crate::utils::time::now_secs;
use crate::routines::agents::load_agent_command;
use crate::routines::cleanup::{
cleanup_expired_workbenches, parse_workbench_name, tmux_session_count,
tmux_session_prefix_alive,
};
use crate::routines::command::{
build_routine_command, inline_prompt_overflow, slugify, tmux_session_prefix, TriggerSource,
TMUX_SESSION_PREFIX,
};
use crate::routines::model::{CleanupResponse, Routine, RoutineStore};
use crate::routines::{max_concurrent_runs, MAX_CONCURRENT_RUNS_ENV};
use super::service_log_tail::{read_log_tail_with_meta, LogWithMeta};
pub fn svc_trigger(store: &RoutineStore, id: &str) -> Result<Routine, AppError> {
if crate::global_lock::is_globally_locked() {
return Err(AppError::Locked("routines are globally locked".into()));
}
let mut lock = store.lock_recover();
let routine = lock.get_mut(id).ok_or(AppError::NotFound)?;
if !routine.enabled {
return Err(AppError::Locked("routine is disabled".into()));
}
if routine.power_saving {
return Err(AppError::Locked("routine is in power-saving mode".into()));
}
let ts = now_secs();
routine.last_manual_trigger_at = Some(ts);
let routine = routine.clone();
drop(lock);
write_routine(&routine).map_err(|_| AppError::Internal)?;
append_manual_trigger_log(&crate::routines::slugify(&routine.title), ts);
spawn_routine_command(&routine, TriggerSource::Manual);
Ok(routine)
}
pub fn svc_trigger_scheduled(store: &RoutineStore, id: &str) -> Result<Routine, AppError> {
if crate::global_lock::is_globally_locked() {
return Err(AppError::Locked("routines are globally locked".into()));
}
let mut lock = store.lock_recover();
let routine = lock.get_mut(id).ok_or(AppError::NotFound)?;
if !routine.enabled {
return Err(AppError::Locked("routine is disabled".into()));
}
if routine.power_saving {
return Err(AppError::Locked("routine is in power-saving mode".into()));
}
if let Some(until) = routine.snoozed_until {
if now_secs() < until {
return Err(AppError::Locked(format!("routine snoozed until {until}")));
}
routine.snoozed_until = None;
let routine = routine.clone();
drop(lock);
write_routine(&routine).map_err(|_| AppError::Internal)?;
spawn_routine_command(&routine, TriggerSource::Scheduled);
return Ok(routine);
}
if let Some(runs) = routine.skip_runs {
if runs > 0 {
routine.skip_runs = (runs > 1).then_some(runs - 1);
let routine = routine.clone();
drop(lock);
write_routine(&routine).map_err(|_| AppError::Internal)?;
return Err(AppError::Locked(format!(
"routine snoozed, skipping this scheduled run ({} more to skip)",
routine.skip_runs.unwrap_or(0)
)));
}
}
let routine = routine.clone();
drop(lock);
spawn_routine_command(&routine, TriggerSource::Scheduled);
Ok(routine)
}
pub(crate) fn sh_bin() -> String {
if let Ok(bin) = std::env::var("MOADIM_SH_BIN") {
return bin;
}
#[cfg(test)]
let fallback = "/nonexistent/moadim-test-sh-guard".to_string();
#[cfg(not(test))]
let fallback = "sh".to_string();
fallback
}
pub fn svc_snooze(
store: &RoutineStore,
id: &str,
snoozed_until: Option<u64>,
skip_runs: Option<u32>,
) -> Result<Routine, AppError> {
if snoozed_until.is_some() && skip_runs.is_some() {
return Err(AppError::BadRequest(
"snoozed_until and skip_runs are mutually exclusive; set only one".into(),
));
}
let mut lock = store.lock_recover();
let routine = lock.get_mut(id).ok_or(AppError::NotFound)?;
routine.snoozed_until = snoozed_until;
routine.skip_runs = skip_runs;
let routine = routine.clone();
drop(lock);
write_routine(&routine).map_err(|_| AppError::Internal)?;
Ok(routine)
}
pub fn svc_set_power_saving(
store: &RoutineStore,
id: &str,
active: bool,
) -> Result<Routine, AppError> {
let mut lock = store.lock_recover();
let routine = lock.get_mut(id).ok_or(AppError::NotFound)?;
routine.power_saving = active;
let routine = routine.clone();
drop(lock);
write_routine(&routine).map_err(|_| AppError::Internal)?;
Ok(routine)
}
fn spawn_routine_command(routine: &Routine, source: TriggerSource) {
match load_agent_command(&routine.agent) {
Ok(agent) => {
if let Some(len) = inline_prompt_overflow(routine, &agent) {
let reason = format!(
"composed prompt is {len} bytes, over the inline-argument limit for agent \
{:?}; skipping launch (would fail silently inside tmux otherwise) — switch \
the agent's args to {{prompt_file}} or shorten the routine's prompt/open \
flags",
routine.agent,
);
log::warn!("trigger: routine {:?} skipped — {reason}", routine.id);
append_skip_log(&slugify(&routine.title), now_secs(), &reason);
return;
}
let session_prefix = tmux_session_prefix(&slugify(&routine.title));
if tmux_session_prefix_alive(&session_prefix) {
let reason = format!(
"a previous run (tmux session prefix {session_prefix:?}) is still active \
(overlap guard)"
);
log::warn!("trigger: routine {:?} skipped — {reason}", routine.id);
append_skip_log(&slugify(&routine.title), now_secs(), &reason);
return;
}
let live = tmux_session_count(TMUX_SESSION_PREFIX);
let cap = max_concurrent_runs();
if cap > 0 && live >= cap {
let reason = format!(
"{live} routine session(s) already running, at or over the global \
concurrency cap of {cap} (set {MAX_CONCURRENT_RUNS_ENV} to raise it); this \
fire will be retried on its next scheduled tick"
);
log::warn!("trigger: routine {:?} skipped — {reason}", routine.id);
append_skip_log(&slugify(&routine.title), now_secs(), &reason);
return;
}
let cmd = build_routine_command(routine, &agent, source);
let mut command = std::process::Command::new(sh_bin());
command.arg("-lc").arg(&cmd);
crate::utils::process::spawn_and_reap(command, "routine command");
}
Err(err) => {
let reason = format!("cannot load agent {:?} ({err})", routine.agent);
log::warn!("trigger: routine {:?} skipped — {reason}", routine.id);
append_skip_log(&slugify(&routine.title), now_secs(), &reason);
}
}
}
pub fn svc_cleanup(store: &RoutineStore) -> CleanupResponse {
let stats = cleanup_expired_workbenches(store);
CleanupResponse {
removed: stats.removed,
freed_bytes: stats.freed_bytes,
}
}
pub(super) fn migrate_workbenches(old_slug: &str, new_slug: &str) {
let Ok(entries) = std::fs::read_dir(workbenches_dir()) else {
return;
};
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
let Some((dir_slug, ts)) = parse_workbench_name(&name) else {
continue;
};
if dir_slug != old_slug {
continue;
}
let from = workbenches_dir().join(&name);
let to = workbenches_dir().join(format!("{new_slug}-{ts}"));
if let Err(err) = std::fs::rename(&from, &to) {
log::warn!("failed to migrate workbench {name} to {new_slug}-{ts}: {err}");
}
}
}
pub fn svc_logs(store: &RoutineStore, id: &str) -> Result<LogWithMeta, AppError> {
let routine = store
.lock_recover()
.get(id)
.cloned()
.ok_or(AppError::NotFound)?;
let slug = slugify(&routine.title);
let mut newest: Option<(u64, String)> = None;
if let Ok(entries) = std::fs::read_dir(workbenches_dir()) {
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
if let Some((dir_slug, ts)) = parse_workbench_name(&name) {
if dir_slug == slug && newest.as_ref().is_none_or(|(newest_ts, _)| ts > *newest_ts)
{
newest = Some((ts, name));
}
}
}
}
let Some((_, dir)) = newest else {
return skip_log_fallback(&slug);
};
let log_path = workbenches_dir().join(dir).join("agent.log");
if !log_path.exists() {
return skip_log_fallback(&slug);
}
read_log_tail_with_meta(&log_path).map_err(|_| AppError::Internal)
}
fn skip_log_fallback(slug: &str) -> Result<LogWithMeta, AppError> {
let skip_log_path = crate::paths::routine_skip_log_path(slug);
if !skip_log_path.exists() {
return Ok(LogWithMeta::empty());
}
read_log_tail_with_meta(&skip_log_path).map_err(|_| AppError::Internal)
}