feat: Retry zombie jobs and emit events

Pass EventBus into zombie_handler_task and use fail_job_or_retry to
determine the outcome. On permanent failure publish
SchedulerEvent::JobCompleted
(with final state and resources); on retry publish
SchedulerEvent::JobUpdated.
Add a NoopExecutor test stub and
zombie_handler_retries_jobs_with_remaining_retry_budget
test. Update the spawn call to pass a cloned EventBus instance.

Author: AndPuQing

src/multicall/gflowd/scheduler_runtime/event_loop.rs

@@ -37,8 +37,12 @@ pub async fn run_event_driven(
         ),
         // Zombie handler - reacts to zombie events
         tokio::spawn(
-            super::monitors::zombie_handler_task(event_bus.subscribe(), Arc::clone(&shared_state))
-                .instrument(tracing::info_span!("zombie_handler_task")),
+            super::monitors::zombie_handler_task(
+                event_bus.subscribe(),
+                Arc::clone(&shared_state),
+                Arc::clone(&event_bus),
+            )
+            .instrument(tracing::info_span!("zombie_handler_task")),
         ),
         // Timeout monitor - checks time limits every 10s
         tokio::spawn(

src/multicall/gflowd/scheduler_runtime/monitors.rs

@@ -112,6 +112,7 @@ pub(super) async fn zombie_monitor_task(state: SharedState, event_bus: Arc<Event
 pub(super) async fn zombie_handler_task(
     mut events: tokio::sync::broadcast::Receiver<EventEnvelope>,
     state: SharedState,
+    event_bus: Arc<EventBus>,
 ) {
     loop {
         match events.recv().await {
@@ -121,27 +122,38 @@ pub(super) async fn zombie_handler_task(
                 let SchedulerEvent::ZombieJobDetected { job_id } = event.event else {
                     continue;
                 };
-                // Get run_name before acquiring write lock
-                let run_name = {
+                let (_success, gpu_ids, memory_mb) = {
                     let state_guard = state.read().await;
-                    state_guard
-                        .scheduler
-                        .get_job_spec(job_id)
-                        .and_then(|spec| spec.run_name.clone())
+                    if let Some(job) = state_guard.get_job(job_id) {
+                        (true, job.gpu_ids.clone(), job.memory_limit_mb)
+                    } else {
+                        (false, None, None)
+                    }
                 };
 
-                // Update job state (write lock)
                 let mut state_guard = state.write().await;
-                // Use fail_job to properly update group_running_count index
-                state_guard.scheduler.fail_job(job_id);
-                state_guard.mark_dirty();
-                tracing::info!(job_id, "Marked zombie job as failed");
-                drop(state_guard); // Release lock before disabling PipePane
-
-                // Disable PipePane if session still exists (no lock held)
-                // This handles the case where the session was manually killed but PipePane might still be active
-                if let Some(rn) = run_name {
-                    disable_pipe_pane_for_job(job_id, &rn, true);
+                let outcome = state_guard.fail_job_or_retry(job_id).await;
+                drop(state_guard);
+
+                match outcome {
+                    FailJobOutcome::Failed => {
+                        tracing::info!(job_id, "Marked zombie job as failed");
+                        event_bus.publish(SchedulerEvent::JobCompleted {
+                            job_id,
+                            final_state: JobState::Failed,
+                            gpu_ids,
+                            memory_mb,
+                        });
+                    }
+                    FailJobOutcome::Retried { retry_attempt } => {
+                        tracing::info!(
+                            job_id,
+                            retry_attempt,
+                            "Marked zombie job for automatic retry"
+                        );
+                        event_bus.publish(SchedulerEvent::JobUpdated { job_id });
+                    }
+                    FailJobOutcome::NotFound => {}
                 }
             }
             Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
@@ -381,9 +393,20 @@ fn calculate_next_reservation_transition(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::multicall::gflowd::scheduler_runtime::SchedulerRuntime;
+    use gflow::core::executor::Executor;
+    use gflow::core::job::Job;
     use gflow::core::reservation::{GpuReservation, GpuSpec, ReservationStatus};
     use std::time::{Duration, SystemTime};
 
+    struct NoopExecutor;
+
+    impl Executor for NoopExecutor {
+        fn execute(&self, _job: &Job) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
     #[test]
     fn zombie_check_allows_legacy_jobs_without_start_time() {
         let now = SystemTime::now();
@@ -411,6 +434,69 @@ mod tests {
         assert!(should_check_missing_session_as_zombie(started_at, now));
     }
 
+    #[tokio::test]
+    async fn zombie_handler_retries_jobs_with_remaining_retry_budget() {
+        let dir = tempfile::tempdir().unwrap();
+        let state = Arc::new(tokio::sync::RwLock::new(
+            SchedulerRuntime::with_state_path(
+                Box::new(NoopExecutor),
+                dir.path().to_path_buf(),
+                None,
+                gflow::core::gpu_allocation::GpuAllocationStrategy::Sequential,
+                gflow::config::ProjectsConfig::default(),
+            )
+            .unwrap(),
+        ));
+        let event_bus = Arc::new(EventBus::new(16));
+        let mut observed_events = event_bus.subscribe();
+
+        let job_id = {
+            let mut state_guard = state.write().await;
+            let job = Job::builder()
+                .command("false")
+                .submitted_by("alice")
+                .max_retry(Some(1))
+                .build();
+            let (job_id, _run_name, _job) = state_guard.submit_job(job).await.unwrap();
+            let prepared = state_guard.scheduler.prepare_jobs_for_execution();
+            assert_eq!(prepared.len(), 1);
+            assert_eq!(prepared[0].id, job_id);
+            job_id
+        };
+
+        let handler = tokio::spawn(zombie_handler_task(
+            event_bus.subscribe(),
+            Arc::clone(&state),
+            Arc::clone(&event_bus),
+        ));
+
+        event_bus.publish(SchedulerEvent::ZombieJobDetected { job_id });
+
+        let event = tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let envelope = observed_events.recv().await.unwrap();
+                if let SchedulerEvent::JobUpdated {
+                    job_id: updated_job_id,
+                } = envelope.event
+                {
+                    break updated_job_id;
+                }
+            }
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(event, job_id);
+
+        let state_guard = state.read().await;
+        let job = state_guard.get_job(job_id).unwrap();
+        assert_eq!(job.state, JobState::Queued);
+        assert_eq!(job.retry_attempt, 1);
+        drop(state_guard);
+
+        handler.abort();
+    }
+
     #[test]
     fn test_calculate_next_transition_no_reservations() {
         let reservations = vec![];