refactor: add job completion notifications to Runner

- Job Actor notifies Runner on completion (JobCompleted/JobFailed)
- Runner handlers use fire-and-forget (tokio::spawn) to avoid blocking
- executor::execute_job now returns bool for success/failure

Author: aster-void

src/actor/job/executor.rs

@@ -27,7 +27,7 @@ pub struct WebhookPayload {
     pub attempts: u32,
 }
 
-pub async fn execute_job(job: &Job, sot_path: &PathBuf, runner: &RunnerConfig) {
+pub async fn execute_job(job: &Job, sot_path: &PathBuf, runner: &RunnerConfig) -> bool {
     let job_dir = git::get_job_dir(sot_path, &job.id);
     let work_dir = resolve_work_dir(sot_path, &job.id, &job.working_dir);
     let mut log_file = job
@@ -75,7 +75,7 @@ pub async fn execute_job(job: &Job, sot_path: &PathBuf, runner: &RunnerConfig) {
         let success = handle_result(job, &result, log_file.as_mut());
 
         if success {
-            return;
+            return true;
         }
 
         last_result = Some(result);
@@ -113,6 +113,8 @@ pub async fn execute_job(job: &Job, sot_path: &PathBuf, runner: &RunnerConfig) {
             send_webhook(&webhook.to_url(), &payload).await;
         }
     }
+
+    false
 }
 
 fn resolve_work_dir(sot_path: &PathBuf, job_id: &str, working_dir: &Option<String>) -> PathBuf {

src/actor/job/mod.rs

@@ -1,12 +1,14 @@
 mod executor;
 mod tick;
 
+use crate::actor::runner::{JobCompleted, JobFailed, RunnerActor};
 use crate::config::{Concurrency, Job, RunnerConfig};
 use crate::git;
 use std::path::PathBuf;
 use tokio::task::JoinHandle;
 use tracing::{error, info};
 use xtra::prelude::*;
+use xtra::refcount::Weak;
 
 use executor::execute_job;
 use tick::is_job_due;
@@ -16,18 +18,25 @@ pub struct JobActor {
     job: Job,
     sot_path: PathBuf,
     runner: RunnerConfig,
+    runner_addr: Option<Address<RunnerActor, Weak>>,
     pending_sync: bool,
     handles: Vec<JoinHandle<()>>,
     tick_handle: Option<JoinHandle<()>>,
     stopping: bool,
 }
 
 impl JobActor {
-    pub fn new(job: Job, sot_path: PathBuf, runner: RunnerConfig) -> Self {
+    pub fn new(
+        job: Job,
+        sot_path: PathBuf,
+        runner: RunnerConfig,
+        runner_addr: Option<Address<RunnerActor, Weak>>,
+    ) -> Self {
         Self {
             job,
             sot_path,
             runner,
+            runner_addr,
             pending_sync: true, // Initial sync needed
             handles: Vec::new(),
             tick_handle: None,
@@ -222,30 +231,61 @@ impl JobActor {
     }
 
     fn spawn_job(&mut self) {
+        let job_id = self.job.id.clone();
         let job = self.job.clone();
         let sot_path = self.sot_path.clone();
         let runner = self.runner.clone();
+        let runner_addr = self.runner_addr.clone();
 
         let handle = tokio::spawn(async move {
-            execute_job(&job, &sot_path, &runner).await;
+            let success = execute_job(&job, &sot_path, &runner).await;
+            if let Some(addr) = runner_addr {
+                let msg = if success {
+                    Either::Left(JobCompleted { job_id })
+                } else {
+                    Either::Right(JobFailed { job_id })
+                };
+                match msg {
+                    Either::Left(m) => { let _ = addr.send(m).await; }
+                    Either::Right(m) => { let _ = addr.send(m).await; }
+                }
+            }
         });
 
         self.handles.push(handle);
     }
 
     fn spawn_waiting_job(&mut self) {
+        let job_id = self.job.id.clone();
         let job = self.job.clone();
         let sot_path = self.sot_path.clone();
         let runner = self.runner.clone();
+        let runner_addr = self.runner_addr.clone();
         let previous_handles = std::mem::take(&mut self.handles);
 
         let handle = tokio::spawn(async move {
             for prev_handle in previous_handles {
                 let _ = prev_handle.await;
             }
-            execute_job(&job, &sot_path, &runner).await;
+            let success = execute_job(&job, &sot_path, &runner).await;
+            if let Some(addr) = runner_addr {
+                let msg = if success {
+                    Either::Left(JobCompleted { job_id })
+                } else {
+                    Either::Right(JobFailed { job_id })
+                };
+                match msg {
+                    Either::Left(m) => { let _ = addr.send(m).await; }
+                    Either::Right(m) => { let _ = addr.send(m).await; }
+                }
+            }
         });
 
         self.handles.push(handle);
     }
 }
+
+enum Either<L, R> {
+    Left(L),
+    Right(R),
+}

src/actor/runner/mod.rs

@@ -10,6 +10,7 @@ use std::time::Duration;
 use tokio::task::JoinHandle;
 use tracing::{error, info, warn};
 use xtra::prelude::*;
+use xtra::refcount::Weak;
 
 const CONFIG_FILE: &str = "rollcron.yaml";
 
@@ -22,6 +23,7 @@ pub struct RunnerActor {
     job_actors: HashMap<String, Address<JobActor>>,
     poll_handle: Option<JoinHandle<()>>,
     supervisor_handle: Option<JoinHandle<()>>,
+    self_addr: Option<Address<Self, Weak>>,
 }
 
 impl RunnerActor {
@@ -39,12 +41,19 @@ impl RunnerActor {
             job_actors: HashMap::new(),
             poll_handle: None,
             supervisor_handle: None,
+            self_addr: None,
         }
     }
 
     fn spawn_job_actor(&mut self, job: Job) {
         let job_id = job.id.clone();
-        let actor = JobActor::new(job, self.sot_path.clone(), self.runner_config.clone());
+        let runner_addr = self.self_addr.clone();
+        let actor = JobActor::new(
+            job,
+            self.sot_path.clone(),
+            self.runner_config.clone(),
+            runner_addr,
+        );
         let addr = xtra::spawn_tokio(actor, Mailbox::unbounded());
         self.job_actors.insert(job_id, addr);
     }
@@ -55,6 +64,7 @@ impl Actor for RunnerActor {
 
     async fn started(&mut self, mailbox: &Mailbox<Self>) -> Result<(), Self::Stop> {
         let addr = mailbox.address();
+        self.self_addr = Some(addr.clone());
 
         // Start git poll loop
         let source = self.source.clone();
@@ -82,9 +92,11 @@ impl Actor for RunnerActor {
             handle.abort();
         }
 
-        // Shutdown all job actors
+        // Shutdown all job actors (fire-and-forget)
         for (_, addr) in self.job_actors.drain() {
-            let _ = addr.send(Shutdown).await;
+            tokio::spawn(async move {
+                let _ = addr.send(Shutdown).await;
+            });
         }
 
         info!(target: "rollcron::runner", "Runner actor stopped");
@@ -142,24 +154,26 @@ impl Handler<ConfigUpdate> for RunnerActor {
             .cloned()
             .collect();
 
-        // Remove deleted jobs
+        // Remove deleted jobs (fire-and-forget)
         for job_id in to_remove {
             if let Some(addr) = self.job_actors.remove(&job_id) {
                 info!(target: "rollcron::runner", job_id = %job_id, "Removing job actor");
-                let _ = addr.send(Shutdown).await;
+                tokio::spawn(async move {
+                    let _ = addr.send(Shutdown).await;
+                });
             }
         }
 
         // Update or create jobs
         for (job_id, job) in new_job_ids {
             if let Some(addr) = self.job_actors.get(&job_id) {
-                // Update existing job
-                let _ = addr
-                    .send(SyncNeeded {
-                        sot_path: msg.sot_path.clone(),
-                    })
-                    .await;
-                let _ = addr.send(Update(job)).await;
+                // Update existing job (fire-and-forget)
+                let addr = addr.clone();
+                let sot_path = msg.sot_path.clone();
+                tokio::spawn(async move {
+                    let _ = addr.send(SyncNeeded { sot_path }).await;
+                    let _ = addr.send(Update(job)).await;
+                });
             } else {
                 // Create new job
                 let job_dir = git::get_job_dir(&self.sot_path, &job_id);
@@ -238,3 +252,29 @@ impl Handler<GracefulShutdown> for RunnerActor {
         ctx.stop_self();
     }
 }
+
+/// Job execution completed successfully
+pub struct JobCompleted {
+    pub job_id: String,
+}
+
+impl Handler<JobCompleted> for RunnerActor {
+    type Return = ();
+
+    async fn handle(&mut self, msg: JobCompleted, _ctx: &mut Context<Self>) {
+        info!(target: "rollcron::runner", job_id = %msg.job_id, "Job completed");
+    }
+}
+
+/// Job execution failed after all retries
+pub struct JobFailed {
+    pub job_id: String,
+}
+
+impl Handler<JobFailed> for RunnerActor {
+    type Return = ();
+
+    async fn handle(&mut self, msg: JobFailed, _ctx: &mut Context<Self>) {
+        warn!(target: "rollcron::runner", job_id = %msg.job_id, "Job failed");
+    }
+}