use crate::channels::{
Channel, DiscordChannel, MattermostChannel, SendMessage, SlackChannel, TelegramChannel,
};
use crate::config::Config;
use crate::cron::{
due_jobs, next_run_for_schedule, record_last_run, record_run, remove_job, reschedule_after_run,
update_job, CronJob, CronJobPatch, DeliveryConfig, JobType, Schedule, SessionTarget,
};
use crate::security::SecurityPolicy;
use anyhow::Result;
use chrono::{DateTime, Utc};
use futures_util::{stream, StreamExt};
use std::process::Stdio;
use std::sync::Arc;
use tokio::process::Command;
use tokio::time::{self, Duration};
const MIN_POLL_SECONDS: u64 = 5;
const SHELL_JOB_TIMEOUT_SECS: u64 = 120;
const SCHEDULER_COMPONENT: &str = "scheduler";
pub async fn run(config: Config) -> Result<()> {
let poll_secs = config.reliability.scheduler_poll_secs.max(MIN_POLL_SECONDS);
let mut interval = time::interval(Duration::from_secs(poll_secs));
interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
let security = Arc::new(SecurityPolicy::from_config(
&config.autonomy,
&config.workspace_dir,
));
crate::health::mark_component_ok(SCHEDULER_COMPONENT);
loop {
interval.tick().await;
crate::health::mark_component_ok(SCHEDULER_COMPONENT);
let jobs = match due_jobs(&config, Utc::now()) {
Ok(jobs) => jobs,
Err(e) => {
crate::health::mark_component_error(SCHEDULER_COMPONENT, e.to_string());
tracing::warn!("Scheduler query failed: {e}");
continue;
}
};
process_due_jobs(&config, &security, jobs, SCHEDULER_COMPONENT).await;
}
}
pub async fn execute_job_now(config: &Config, job: &CronJob) -> (bool, String) {
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
execute_job_with_retry(config, &security, job).await
}
async fn execute_job_with_retry(
config: &Config,
security: &SecurityPolicy,
job: &CronJob,
) -> (bool, String) {
let mut last_output = String::new();
let retries = config.reliability.scheduler_retries;
let mut backoff_ms = config.reliability.provider_backoff_ms.max(200);
for attempt in 0..=retries {
let (success, output) = match job.job_type {
JobType::Shell => run_job_command(config, security, job).await,
JobType::Agent => run_agent_job(config, security, job).await,
};
last_output = output;
if success {
return (true, last_output);
}
if last_output.starts_with("blocked by security policy:") {
return (false, last_output);
}
if attempt < retries {
let jitter_ms = u64::from(Utc::now().timestamp_subsec_millis() % 250);
time::sleep(Duration::from_millis(backoff_ms + jitter_ms)).await;
backoff_ms = (backoff_ms.saturating_mul(2)).min(30_000);
}
}
(false, last_output)
}
async fn process_due_jobs(
config: &Config,
security: &Arc<SecurityPolicy>,
jobs: Vec<CronJob>,
component: &str,
) {
crate::health::mark_component_ok(component);
let max_concurrent = config.scheduler.max_concurrent.max(1);
let mut in_flight =
stream::iter(
jobs.into_iter().map(|job| {
let config = config.clone();
let security = Arc::clone(security);
let component = component.to_owned();
async move {
execute_and_persist_job(&config, security.as_ref(), &job, &component).await
}
}),
)
.buffer_unordered(max_concurrent);
while let Some((job_id, success, output)) = in_flight.next().await {
if !success {
tracing::warn!("Scheduler job '{job_id}' failed: {output}");
}
}
}
async fn execute_and_persist_job(
config: &Config,
security: &SecurityPolicy,
job: &CronJob,
component: &str,
) -> (String, bool, String) {
crate::health::mark_component_ok(component);
warn_if_high_frequency_agent_job(job);
let started_at = Utc::now();
let (success, output) = execute_job_with_retry(config, security, job).await;
let finished_at = Utc::now();
let success = persist_job_result(config, job, success, &output, started_at, finished_at).await;
(job.id.clone(), success, output)
}
async fn run_agent_job(
config: &Config,
security: &SecurityPolicy,
job: &CronJob,
) -> (bool, String) {
if !security.can_act() {
return (
false,
"blocked by security policy: autonomy is read-only".to_string(),
);
}
if security.is_rate_limited() {
return (
false,
"blocked by security policy: rate limit exceeded".to_string(),
);
}
if !security.record_action() {
return (
false,
"blocked by security policy: action budget exhausted".to_string(),
);
}
let name = job.name.clone().unwrap_or_else(|| "cron-job".to_string());
let prompt = job.prompt.clone().unwrap_or_default();
let prefixed_prompt = format!("[cron:{} {name}] {prompt}", job.id);
let model_override = job.model.clone();
let run_result = match job.session_target {
SessionTarget::Main | SessionTarget::Isolated => {
crate::agent::run(
config.clone(),
Some(prefixed_prompt),
None,
model_override,
config.default_temperature,
vec![],
false,
)
.await
}
};
match run_result {
Ok(response) => (
true,
if response.trim().is_empty() {
"agent job executed".to_string()
} else {
response
},
),
Err(e) => (false, format!("agent job failed: {e}")),
}
}
async fn persist_job_result(
config: &Config,
job: &CronJob,
mut success: bool,
output: &str,
started_at: DateTime<Utc>,
finished_at: DateTime<Utc>,
) -> bool {
let duration_ms = (finished_at - started_at).num_milliseconds();
if let Err(e) = deliver_if_configured(config, job, output).await {
if job.delivery.best_effort {
tracing::warn!("Cron delivery failed (best_effort): {e}");
} else {
success = false;
tracing::warn!("Cron delivery failed: {e}");
}
}
let _ = record_run(
config,
&job.id,
started_at,
finished_at,
if success { "ok" } else { "error" },
Some(output),
duration_ms,
);
if is_one_shot_auto_delete(job) {
if success {
if let Err(e) = remove_job(config, &job.id) {
tracing::warn!("Failed to remove one-shot cron job after success: {e}");
}
} else {
let _ = record_last_run(config, &job.id, finished_at, false, output);
if let Err(e) = update_job(
config,
&job.id,
CronJobPatch {
enabled: Some(false),
..CronJobPatch::default()
},
) {
tracing::warn!("Failed to disable failed one-shot cron job: {e}");
}
}
return success;
}
if let Err(e) = reschedule_after_run(config, job, success, output) {
tracing::warn!("Failed to persist scheduler run result: {e}");
}
success
}
fn is_one_shot_auto_delete(job: &CronJob) -> bool {
job.delete_after_run && matches!(job.schedule, Schedule::At { .. })
}
fn warn_if_high_frequency_agent_job(job: &CronJob) {
if !matches!(job.job_type, JobType::Agent) {
return;
}
let too_frequent = match &job.schedule {
Schedule::Every { every_ms } => *every_ms < 5 * 60 * 1000,
Schedule::Cron { .. } => {
let now = Utc::now();
match (
next_run_for_schedule(&job.schedule, now),
next_run_for_schedule(&job.schedule, now + chrono::Duration::seconds(1)),
) {
(Ok(a), Ok(b)) => (b - a).num_minutes() < 5,
_ => false,
}
}
Schedule::At { .. } => false,
};
if too_frequent {
tracing::warn!(
"Cron agent job '{}' is scheduled more frequently than every 5 minutes",
job.id
);
}
}
async fn deliver_if_configured(config: &Config, job: &CronJob, output: &str) -> Result<()> {
let delivery: &DeliveryConfig = &job.delivery;
if !delivery.mode.eq_ignore_ascii_case("announce") {
return Ok(());
}
let channel = delivery
.channel
.as_deref()
.ok_or_else(|| anyhow::anyhow!("delivery.channel is required for announce mode"))?;
let target = delivery
.to
.as_deref()
.ok_or_else(|| anyhow::anyhow!("delivery.to is required for announce mode"))?;
deliver_announcement(config, channel, target, output).await
}
pub(crate) async fn deliver_announcement(
config: &Config,
channel: &str,
target: &str,
output: &str,
) -> Result<()> {
match channel.to_ascii_lowercase().as_str() {
"telegram" => {
let tg = config
.channels_config
.telegram
.as_ref()
.ok_or_else(|| anyhow::anyhow!("telegram channel not configured"))?;
let channel = TelegramChannel::new(
tg.bot_token.clone(),
tg.allowed_users.clone(),
tg.mention_only,
);
channel.send(&SendMessage::new(output, target)).await?;
}
"discord" => {
let dc = config
.channels_config
.discord
.as_ref()
.ok_or_else(|| anyhow::anyhow!("discord channel not configured"))?;
let channel = DiscordChannel::new(
dc.bot_token.clone(),
dc.guild_id.clone(),
dc.allowed_users.clone(),
dc.listen_to_bots,
dc.mention_only,
);
channel.send(&SendMessage::new(output, target)).await?;
}
"slack" => {
let sl = config
.channels_config
.slack
.as_ref()
.ok_or_else(|| anyhow::anyhow!("slack channel not configured"))?;
let channel = SlackChannel::new(
sl.bot_token.clone(),
sl.channel_id.clone(),
sl.allowed_users.clone(),
);
channel.send(&SendMessage::new(output, target)).await?;
}
"mattermost" => {
let mm = config
.channels_config
.mattermost
.as_ref()
.ok_or_else(|| anyhow::anyhow!("mattermost channel not configured"))?;
let channel = MattermostChannel::new(
mm.url.clone(),
mm.bot_token.clone(),
mm.channel_id.clone(),
mm.allowed_users.clone(),
mm.thread_replies.unwrap_or(true),
mm.mention_only.unwrap_or(false),
);
channel.send(&SendMessage::new(output, target)).await?;
}
other => anyhow::bail!("unsupported delivery channel: {other}"),
}
Ok(())
}
async fn run_job_command(
config: &Config,
security: &SecurityPolicy,
job: &CronJob,
) -> (bool, String) {
run_job_command_with_timeout(
config,
security,
job,
Duration::from_secs(SHELL_JOB_TIMEOUT_SECS),
)
.await
}
async fn run_job_command_with_timeout(
config: &Config,
security: &SecurityPolicy,
job: &CronJob,
timeout: Duration,
) -> (bool, String) {
if !security.can_act() {
return (
false,
"blocked by security policy: autonomy is read-only".to_string(),
);
}
if security.is_rate_limited() {
return (
false,
"blocked by security policy: rate limit exceeded".to_string(),
);
}
if !security.is_command_allowed(&job.command) {
return (
false,
format!(
"blocked by security policy: command not allowed: {}",
job.command
),
);
}
if let Some(path) = security.forbidden_path_argument(&job.command) {
return (
false,
format!("blocked by security policy: forbidden path argument: {path}"),
);
}
if !security.record_action() {
return (
false,
"blocked by security policy: action budget exhausted".to_string(),
);
}
let child = match Command::new("sh")
.arg("-lc")
.arg(&job.command)
.current_dir(&config.workspace_dir)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.spawn()
{
Ok(child) => child,
Err(e) => return (false, format!("spawn error: {e}")),
};
match time::timeout(timeout, child.wait_with_output()).await {
Ok(Ok(output)) => {
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
let combined = format!(
"status={}\nstdout:\n{}\nstderr:\n{}",
output.status,
stdout.trim(),
stderr.trim()
);
(output.status.success(), combined)
}
Ok(Err(e)) => (false, format!("spawn error: {e}")),
Err(_) => (
false,
format!("job timed out after {}s", timeout.as_secs_f64()),
),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
use crate::cron::{self, DeliveryConfig};
use crate::security::SecurityPolicy;
use chrono::{Duration as ChronoDuration, Utc};
use tempfile::TempDir;
async fn test_config(tmp: &TempDir) -> Config {
let config = Config {
workspace_dir: tmp.path().join("workspace"),
config_path: tmp.path().join("config.toml"),
..Config::default()
};
tokio::fs::create_dir_all(&config.workspace_dir)
.await
.unwrap();
config
}
fn test_job(command: &str) -> CronJob {
CronJob {
id: "test-job".into(),
expression: "* * * * *".into(),
schedule: crate::cron::Schedule::Cron {
expr: "* * * * *".into(),
tz: None,
},
command: command.into(),
prompt: None,
name: None,
job_type: JobType::Shell,
session_target: SessionTarget::Isolated,
model: None,
enabled: true,
delivery: DeliveryConfig::default(),
delete_after_run: false,
created_at: Utc::now(),
next_run: Utc::now(),
last_run: None,
last_status: None,
last_output: None,
}
}
fn unique_component(prefix: &str) -> String {
format!("{prefix}-{}", uuid::Uuid::new_v4())
}
#[tokio::test]
async fn run_job_command_success() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let job = test_job("echo scheduler-ok");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(success);
assert!(output.contains("scheduler-ok"));
assert!(output.contains("status=exit status: 0"));
}
#[tokio::test]
async fn run_job_command_failure() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let job = test_job("ls definitely_missing_file_for_scheduler_test");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("definitely_missing_file_for_scheduler_test"));
assert!(output.contains("status=exit status:"));
}
#[tokio::test]
async fn run_job_command_times_out() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["sleep".into()];
let job = test_job("sleep 1");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) =
run_job_command_with_timeout(&config, &security, &job, Duration::from_millis(50)).await;
assert!(!success);
assert!(output.contains("job timed out after"));
}
#[tokio::test]
async fn run_job_command_blocks_disallowed_command() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["echo".into()];
let job = test_job("curl https://evil.example");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("command not allowed"));
}
#[tokio::test]
async fn run_job_command_blocks_forbidden_path_argument() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["cat".into()];
let job = test_job("cat /etc/passwd");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("forbidden path argument"));
assert!(output.contains("/etc/passwd"));
}
#[tokio::test]
async fn run_job_command_blocks_forbidden_option_assignment_path_argument() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["grep".into()];
let job = test_job("grep --file=/etc/passwd root ./src");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("forbidden path argument"));
assert!(output.contains("/etc/passwd"));
}
#[tokio::test]
async fn run_job_command_blocks_forbidden_short_option_attached_path_argument() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["grep".into()];
let job = test_job("grep -f/etc/passwd root ./src");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("forbidden path argument"));
assert!(output.contains("/etc/passwd"));
}
#[tokio::test]
async fn run_job_command_blocks_tilde_user_path_argument() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["cat".into()];
let job = test_job("cat ~root/.ssh/id_rsa");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("forbidden path argument"));
assert!(output.contains("~root/.ssh/id_rsa"));
}
#[tokio::test]
async fn run_job_command_blocks_input_redirection_path_bypass() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.allowed_commands = vec!["cat".into()];
let job = test_job("cat </etc/passwd");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("command not allowed"));
}
#[tokio::test]
async fn run_job_command_blocks_readonly_mode() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.level = crate::security::AutonomyLevel::ReadOnly;
let job = test_job("echo should-not-run");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("read-only"));
}
#[tokio::test]
async fn run_job_command_blocks_rate_limited() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.max_actions_per_hour = 0;
let job = test_job("echo should-not-run");
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_job_command(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("rate limit exceeded"));
}
#[tokio::test]
async fn execute_job_with_retry_recovers_after_first_failure() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.reliability.scheduler_retries = 1;
config.reliability.provider_backoff_ms = 1;
config.autonomy.allowed_commands = vec!["sh".into()];
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
tokio::fs::write(
config.workspace_dir.join("retry-once.sh"),
"#!/bin/sh\nif [ -f retry-ok.flag ]; then\n echo recovered\n exit 0\nfi\ntouch retry-ok.flag\nexit 1\n",
)
.await
.unwrap();
let job = test_job("sh ./retry-once.sh");
let (success, output) = execute_job_with_retry(&config, &security, &job).await;
assert!(success);
assert!(output.contains("recovered"));
}
#[tokio::test]
async fn execute_job_with_retry_exhausts_attempts() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.reliability.scheduler_retries = 1;
config.reliability.provider_backoff_ms = 1;
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let job = test_job("ls always_missing_for_retry_test");
let (success, output) = execute_job_with_retry(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("always_missing_for_retry_test"));
}
#[tokio::test]
async fn run_agent_job_returns_error_without_provider_key() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let mut job = test_job("");
job.job_type = JobType::Agent;
job.prompt = Some("Say hello".into());
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_agent_job(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("agent job failed:"));
}
#[tokio::test]
async fn run_agent_job_blocks_readonly_mode() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.level = crate::security::AutonomyLevel::ReadOnly;
let mut job = test_job("");
job.job_type = JobType::Agent;
job.prompt = Some("Say hello".into());
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_agent_job(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("read-only"));
}
#[tokio::test]
async fn run_agent_job_blocks_rate_limited() {
let tmp = TempDir::new().unwrap();
let mut config = test_config(&tmp).await;
config.autonomy.max_actions_per_hour = 0;
let mut job = test_job("");
job.job_type = JobType::Agent;
job.prompt = Some("Say hello".into());
let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
let (success, output) = run_agent_job(&config, &security, &job).await;
assert!(!success);
assert!(output.contains("blocked by security policy"));
assert!(output.contains("rate limit exceeded"));
}
#[tokio::test]
async fn process_due_jobs_marks_component_ok_even_when_idle() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let security = Arc::new(SecurityPolicy::from_config(
&config.autonomy,
&config.workspace_dir,
));
let component = unique_component("scheduler-idle");
crate::health::mark_component_error(&component, "pre-existing error");
process_due_jobs(&config, &security, Vec::new(), &component).await;
let snapshot = crate::health::snapshot_json();
let entry = &snapshot["components"][component.as_str()];
assert_eq!(entry["status"], "ok");
assert!(entry["last_ok"].as_str().is_some());
assert!(entry["last_error"].is_null());
}
#[tokio::test]
async fn process_due_jobs_failure_does_not_mark_component_unhealthy() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let job = test_job("ls definitely_missing_file_for_scheduler_component_health_test");
let security = Arc::new(SecurityPolicy::from_config(
&config.autonomy,
&config.workspace_dir,
));
let component = unique_component("scheduler-fail");
crate::health::mark_component_ok(&component);
process_due_jobs(&config, &security, vec![job], &component).await;
let snapshot = crate::health::snapshot_json();
let entry = &snapshot["components"][component.as_str()];
assert_eq!(entry["status"], "ok");
}
#[tokio::test]
async fn persist_job_result_records_run_and_reschedules_shell_job() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let job = cron::add_job(&config, "*/5 * * * *", "echo ok").unwrap();
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, true, "ok", started, finished).await;
assert!(success);
let runs = cron::list_runs(&config, &job.id, 10).unwrap();
assert_eq!(runs.len(), 1);
let updated = cron::get_job(&config, &job.id).unwrap();
assert_eq!(updated.last_status.as_deref(), Some("ok"));
}
#[tokio::test]
async fn persist_job_result_success_deletes_one_shot() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let at = Utc::now() + ChronoDuration::minutes(10);
let job = cron::add_agent_job(
&config,
Some("one-shot".into()),
crate::cron::Schedule::At { at },
"Hello",
SessionTarget::Isolated,
None,
None,
true,
)
.unwrap();
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, true, "ok", started, finished).await;
assert!(success);
let lookup = cron::get_job(&config, &job.id);
assert!(lookup.is_err());
}
#[tokio::test]
async fn persist_job_result_failure_disables_one_shot() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let at = Utc::now() + ChronoDuration::minutes(10);
let job = cron::add_agent_job(
&config,
Some("one-shot".into()),
crate::cron::Schedule::At { at },
"Hello",
SessionTarget::Isolated,
None,
None,
true,
)
.unwrap();
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, false, "boom", started, finished).await;
assert!(!success);
let updated = cron::get_job(&config, &job.id).unwrap();
assert!(!updated.enabled);
assert_eq!(updated.last_status.as_deref(), Some("error"));
}
#[tokio::test]
async fn persist_job_result_success_deletes_one_shot_shell_job() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let at = Utc::now() + ChronoDuration::minutes(10);
let job = cron::add_once_at(&config, at, "echo one-shot-shell").unwrap();
assert!(job.delete_after_run);
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, true, "ok", started, finished).await;
assert!(success);
let lookup = cron::get_job(&config, &job.id);
assert!(lookup.is_err());
}
#[tokio::test]
async fn persist_job_result_failure_disables_one_shot_shell_job() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let at = Utc::now() + ChronoDuration::minutes(10);
let job = cron::add_once_at(&config, at, "echo one-shot-shell").unwrap();
assert!(job.delete_after_run);
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, false, "boom", started, finished).await;
assert!(!success);
let updated = cron::get_job(&config, &job.id).unwrap();
assert!(!updated.enabled);
assert_eq!(updated.last_status.as_deref(), Some("error"));
}
#[tokio::test]
async fn persist_job_result_delivery_failure_non_best_effort_marks_error() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let job = cron::add_agent_job(
&config,
Some("announce-job".into()),
crate::cron::Schedule::Cron {
expr: "*/5 * * * *".into(),
tz: None,
},
"deliver this",
SessionTarget::Isolated,
None,
Some(DeliveryConfig {
mode: "announce".into(),
channel: Some("telegram".into()),
to: Some("123456".into()),
best_effort: false,
}),
false,
)
.unwrap();
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, true, "ok", started, finished).await;
assert!(!success);
let updated = cron::get_job(&config, &job.id).unwrap();
assert!(updated.enabled);
assert_eq!(updated.last_status.as_deref(), Some("error"));
let runs = cron::list_runs(&config, &job.id, 10).unwrap();
assert_eq!(runs.len(), 1);
assert_eq!(runs[0].status, "error");
}
#[tokio::test]
async fn persist_job_result_delivery_failure_best_effort_keeps_success() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let job = cron::add_agent_job(
&config,
Some("announce-job-best-effort".into()),
crate::cron::Schedule::Cron {
expr: "*/5 * * * *".into(),
tz: None,
},
"deliver this",
SessionTarget::Isolated,
None,
Some(DeliveryConfig {
mode: "announce".into(),
channel: Some("telegram".into()),
to: Some("123456".into()),
best_effort: true,
}),
false,
)
.unwrap();
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, true, "ok", started, finished).await;
assert!(success);
let updated = cron::get_job(&config, &job.id).unwrap();
assert!(updated.enabled);
assert_eq!(updated.last_status.as_deref(), Some("ok"));
let runs = cron::list_runs(&config, &job.id, 10).unwrap();
assert_eq!(runs.len(), 1);
assert_eq!(runs[0].status, "ok");
}
#[tokio::test]
async fn persist_job_result_at_schedule_without_delete_after_run_is_not_deleted() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let at = Utc::now() + ChronoDuration::minutes(10);
let job = cron::add_agent_job(
&config,
Some("at-no-autodelete".into()),
crate::cron::Schedule::At { at },
"Hello",
SessionTarget::Isolated,
None,
None,
false,
)
.unwrap();
assert!(!job.delete_after_run);
let started = Utc::now();
let finished = started + ChronoDuration::milliseconds(10);
let success = persist_job_result(&config, &job, true, "ok", started, finished).await;
assert!(success);
let updated = cron::get_job(&config, &job.id).unwrap();
assert!(updated.enabled);
assert_eq!(updated.last_status.as_deref(), Some("ok"));
}
#[tokio::test]
async fn deliver_if_configured_handles_none_and_invalid_channel() {
let tmp = TempDir::new().unwrap();
let config = test_config(&tmp).await;
let mut job = test_job("echo ok");
assert!(deliver_if_configured(&config, &job, "x").await.is_ok());
job.delivery = DeliveryConfig {
mode: "announce".into(),
channel: Some("invalid".into()),
to: Some("target".into()),
best_effort: true,
};
let err = deliver_if_configured(&config, &job, "x").await.unwrap_err();
assert!(err.to_string().contains("unsupported delivery channel"));
}
}