treewide: fix multiple safety and correctness issues

- config: rewrite normalize_cron() to parse DOW values numerically,
  fixing false replacements like "10" -> "17" or "0-6" -> "7-6"
- scheduler: fix Concurrency::Wait blocking by spawning waiting jobs
  asynchronously instead of awaiting in the actor handler
- git: improve sync_to_job_dir atomicity with backup/rename/remove
  pattern to minimize window where job_dir doesn't exist
- executor: add path traversal validation for env_file paths
- scheduler: extend is_job_due tolerance to ±1s for clock drift
- git: add LC_ALL=C to git commands for consistent English output
- git: add TempDirGuard RAII for reliable temp directory cleanup

Author: aster-void

src/config.rs

@@ -168,22 +168,63 @@ fn normalize_cron(cron: &str) -> String {
     }
 
     // Day-of-week is the 5th field (index 4)
-    // Replace standalone "0" with "7" for Sunday
-    let dow = fields[4]
-        .replace(",0,", ",7,")
-        .replace(",0", ",7")
-        .replace("0,", "7,")
-        .replace("0-", "7-");
-
-    // Handle standalone "0"
-    let dow = if dow == "0" { "7".to_string() } else { dow };
+    // Parse each element properly to avoid false replacements (e.g., "10" -> "17")
+    let dow = normalize_dow_field(fields[4]);
 
     format!(
         "{} {} {} {} {}",
         fields[0], fields[1], fields[2], fields[3], dow
     )
 }
 
+/// Normalize a single day-of-week field, converting 0 (POSIX Sunday) to 7 (cron crate Sunday).
+fn normalize_dow_field(field: &str) -> String {
+    // Handle special cases
+    if field == "*" || field == "?" {
+        return field.to_string();
+    }
+
+    // Split by comma for lists (e.g., "0,3,5")
+    field
+        .split(',')
+        .map(|part| normalize_dow_part(part))
+        .collect::<Vec<_>>()
+        .join(",")
+}
+
+/// Normalize a single part of a day-of-week field (handles ranges like "0-6" and steps like "0/2").
+fn normalize_dow_part(part: &str) -> String {
+    // Handle step values (e.g., "0/2" or "0-6/2")
+    if let Some((range_part, step)) = part.split_once('/') {
+        let normalized_range = normalize_dow_range(range_part);
+        return format!("{}/{}", normalized_range, step);
+    }
+
+    normalize_dow_range(part)
+}
+
+/// Normalize a range or single value (e.g., "0", "0-6").
+fn normalize_dow_range(part: &str) -> String {
+    // Handle ranges (e.g., "0-6")
+    if let Some((start, end)) = part.split_once('-') {
+        let start_normalized = normalize_dow_value(start);
+        let end_normalized = normalize_dow_value(end);
+        return format!("{}-{}", start_normalized, end_normalized);
+    }
+
+    // Single value
+    normalize_dow_value(part)
+}
+
+/// Normalize a single day-of-week value: "0" -> "7", others unchanged.
+fn normalize_dow_value(val: &str) -> String {
+    if val == "0" {
+        "7".to_string()
+    } else {
+        val.to_string()
+    }
+}
+
 pub fn parse_config(content: &str) -> Result<(RunnerConfig, Vec<Job>)> {
     let config: Config = serde_yaml::from_str(content)
         .map_err(|e| anyhow!("Failed to parse YAML: {}", e))?;
@@ -948,4 +989,26 @@ jobs:
         let (_, jobs) = parse_config(yaml).unwrap();
         assert_eq!(jobs[0].webhook.len(), 3);
     }
+
+    #[test]
+    fn test_normalize_dow_field() {
+        // Standalone 0 -> 7
+        assert_eq!(normalize_dow_field("0"), "7");
+        // Other values unchanged
+        assert_eq!(normalize_dow_field("1"), "1");
+        assert_eq!(normalize_dow_field("*"), "*");
+        assert_eq!(normalize_dow_field("?"), "?");
+        // List with 0
+        assert_eq!(normalize_dow_field("0,3,5"), "7,3,5");
+        assert_eq!(normalize_dow_field("1,0,3"), "1,7,3");
+        // Range starting with 0
+        assert_eq!(normalize_dow_field("0-6"), "7-6");
+        // Step value
+        assert_eq!(normalize_dow_field("0/2"), "7/2");
+        assert_eq!(normalize_dow_field("0-6/2"), "7-6/2");
+        // Values that should NOT be affected (regression test for "10" -> "17" bug)
+        assert_eq!(normalize_dow_field("10"), "10"); // Invalid but should not be mangled
+        assert_eq!(normalize_dow_field("1-5"), "1-5");
+        assert_eq!(normalize_dow_field("20"), "20"); // Invalid but should not be mangled
+    }
 }

src/git.rs

@@ -2,6 +2,30 @@ use anyhow::{Context, Result};
 use std::path::{Path, PathBuf};
 use std::process::Command;
 
+/// RAII guard that removes a directory on drop unless disarmed.
+struct TempDirGuard<'a> {
+    path: &'a Path,
+    keep: bool,
+}
+
+impl<'a> TempDirGuard<'a> {
+    fn new(path: &'a Path) -> Self {
+        Self { path, keep: false }
+    }
+
+    fn disarm(&mut self) {
+        self.keep = true;
+    }
+}
+
+impl Drop for TempDirGuard<'_> {
+    fn drop(&mut self) {
+        if !self.keep && self.path.exists() {
+            let _ = std::fs::remove_dir_all(self.path);
+        }
+    }
+}
+
 /// Ensures repo is cloned/synced to cache. Returns (cache path, commit_range if updated).
 pub fn ensure_repo(source: &str) -> Result<(PathBuf, Option<String>)> {
     let cache_dir = get_cache_dir(source)?;
@@ -41,6 +65,7 @@ fn sync_repo(dest: &Path) -> Result<Option<String>> {
     let has_upstream = Command::new("git")
         .args(["rev-parse", "--abbrev-ref", "@{upstream}"])
         .current_dir(dest)
+        .env("LC_ALL", "C") // Ensure consistent English output
         .output()
         .map(|o| o.status.success())
         .unwrap_or(false);
@@ -49,6 +74,7 @@ fn sync_repo(dest: &Path) -> Result<Option<String>> {
         let output = Command::new("git")
             .args(["pull", "--ff-only"])
             .current_dir(dest)
+            .env("LC_ALL", "C") // Ensure consistent English output
             .output()?;
 
         if !output.status.success() {
@@ -129,14 +155,17 @@ pub fn sync_to_job_dir(sot_path: &Path, job_dir: &Path) -> Result<()> {
     }
     std::fs::create_dir_all(&temp_dir)?;
 
+    // RAII guard ensures temp_dir is cleaned up on any error path
+    let mut temp_guard = TempDirGuard::new(&temp_dir);
+
     // Use git archive for all repos (both local and remote use git clone now)
     let archive = Command::new("git")
         .args(["archive", "HEAD"])
         .current_dir(sot_path)
         .output()?;
 
     if !archive.status.success() {
-        std::fs::remove_dir_all(&temp_dir)?;
+        // temp_guard will clean up on drop
         let stderr = String::from_utf8_lossy(&archive.stderr);
         anyhow::bail!("git archive failed: {}", stderr);
     }
@@ -159,18 +188,39 @@ pub fn sync_to_job_dir(sot_path: &Path, job_dir: &Path) -> Result<()> {
 
     let status = extract.wait()?;
     if !status.success() {
-        std::fs::remove_dir_all(&temp_dir)?;
+        // temp_guard will clean up on drop
         anyhow::bail!("tar extraction failed with exit code: {:?}", status.code());
     }
 
-    // Atomic swap: remove old, rename temp to target
+    // Disarm the guard before rename - we'll handle cleanup manually from here
+    temp_guard.disarm();
+
+    // Safe swap: rename old to backup, rename temp to target, then remove backup.
+    // This minimizes the window where job_dir doesn't exist.
+    let backup_dir = job_dir.with_extension("old");
+
+    // Clean up any leftover backup directory
+    if backup_dir.exists() {
+        let _ = std::fs::remove_dir_all(&backup_dir);
+    }
+
     if job_dir.exists() {
-        std::fs::remove_dir_all(job_dir)?;
+        // Move existing to backup first
+        std::fs::rename(job_dir, &backup_dir).with_context(|| {
+            format!("Failed to rename {} to backup", job_dir_str)
+        })?;
     }
+
+    // Move new directory into place
     std::fs::rename(&temp_dir, job_dir).with_context(|| {
         format!("Failed to rename {} to {}", temp_dir_str, job_dir_str)
     })?;
 
+    // Remove backup (ignore errors as it's cleanup)
+    if backup_dir.exists() {
+        let _ = std::fs::remove_dir_all(&backup_dir);
+    }
+
     Ok(())
 }
 

src/scheduler/executor.rs

@@ -165,6 +165,43 @@ async fn run_command(job: &Job, work_dir: &PathBuf, sot_path: &PathBuf, runner:
     }
 }
 
+/// Validate that a path stays within the base directory (prevents path traversal).
+fn validate_env_file_path(
+    env_file_path: &str,
+    base_dir: &PathBuf,
+    context: &str,
+) -> anyhow::Result<PathBuf> {
+    let expanded = env::expand_string(env_file_path);
+    let full_path = base_dir.join(&expanded);
+
+    // Canonicalize to resolve .. and symlinks
+    match (full_path.canonicalize(), base_dir.canonicalize()) {
+        (Ok(resolved), Ok(base)) if resolved.starts_with(&base) => Ok(resolved),
+        (Ok(_), Ok(_)) => {
+            anyhow::bail!(
+                "{} env_file '{}': path traversal detected (must be within base directory)",
+                context,
+                env_file_path
+            )
+        }
+        (Err(e), _) => {
+            anyhow::bail!(
+                "{} env_file '{}': file not found or inaccessible: {}",
+                context,
+                env_file_path,
+                e
+            )
+        }
+        (_, Err(e)) => {
+            anyhow::bail!(
+                "{} base directory not accessible: {}",
+                context,
+                e
+            )
+        }
+    }
+}
+
 fn merge_env_vars(
     job: &Job,
     work_dir: &PathBuf,
@@ -175,9 +212,8 @@ fn merge_env_vars(
 
     // 1. Start with runner.env_file (loaded from sot_path)
     if let Some(env_file_path) = &runner.env_file {
-        let expanded = env::expand_string(env_file_path);
-        let full_path = sot_path.join(&expanded);
-        let vars = env::load_env_from_path(&full_path)?;
+        let validated_path = validate_env_file_path(env_file_path, sot_path, "runner")?;
+        let vars = env::load_env_from_path(&validated_path)?;
         env_vars.extend(vars);
     }
 
@@ -190,9 +226,8 @@ fn merge_env_vars(
 
     // 3. Merge job.env_file (loaded from work_dir)
     if let Some(env_file_path) = &job.env_file {
-        let expanded = env::expand_string(env_file_path);
-        let full_path = work_dir.join(&expanded);
-        let vars = env::load_env_from_path(&full_path)?;
+        let validated_path = validate_env_file_path(env_file_path, work_dir, &format!("job:{}", job.id))?;
+        let vars = env::load_env_from_path(&validated_path)?;
         env_vars.extend(vars);
     }
 

src/scheduler/mod.rs

@@ -192,14 +192,11 @@ impl Scheduler {
                         "{} Waiting for {} previous run(s) to complete",
                         tag, running_count
                     );
-                    if let Some(handles) = self.job_handles.remove(&job.id) {
-                        for handle in handles {
-                            let _ = handle.await;
-                        }
-                    }
-                    self.cleanup_finished_handles(&job.id);
+                    // Spawn waiting job asynchronously to avoid blocking the actor
+                    self.spawn_waiting_job(job, addr);
+                } else {
+                    self.spawn_job(job, addr);
                 }
-                self.spawn_job(job, addr);
             }
             Concurrency::Skip => {
                 if running_count > 0 {
@@ -238,6 +235,34 @@ impl Scheduler {
 
         self.job_handles.entry(job_id).or_default().push(handle);
     }
+
+    /// Spawn a job that waits for previous runs to complete before executing.
+    /// This avoids blocking the actor by spawning a separate task that handles the waiting.
+    fn spawn_waiting_job(&mut self, job: Job, addr: Address<Scheduler, Weak>) {
+        let job_id = job.id.clone();
+        let job_id_for_log = job.id.clone();
+        let work_dir = resolve_work_dir(&self.sot_path, &job.id, &job.working_dir);
+        let sot_path = self.sot_path.clone();
+        let runner = self.runner.clone();
+
+        // Take ownership of existing handles to wait on them
+        let previous_handles = self.job_handles.remove(&job.id).unwrap_or_default();
+
+        let handle = tokio::spawn(async move {
+            // Wait for all previous runs to complete
+            for prev_handle in previous_handles {
+                let _ = prev_handle.await;
+            }
+
+            // Now execute the new job
+            execute_job(&job, &work_dir, &sot_path, &runner).await;
+            if let Err(e) = addr.send(JobCompleted).await {
+                eprintln!("[job:{}] Failed to notify completion: {}", job_id_for_log, e);
+            }
+        });
+
+        self.job_handles.entry(job_id).or_default().push(handle);
+    }
 }
 
 // === Helper Functions ===
@@ -252,7 +277,9 @@ where
     let now = Utc::now().with_timezone(&tz);
     if let Some(next) = schedule.upcoming(tz).next() {
         let until_next = (next - now).num_milliseconds();
-        until_next <= SCHEDULE_TOLERANCE_MS && until_next >= 0
+        // Allow jobs to be triggered slightly after their scheduled time
+        // (e.g., due to minor clock drift or processing delay)
+        until_next <= SCHEDULE_TOLERANCE_MS && until_next >= -SCHEDULE_TOLERANCE_MS
     } else {
         false
     }