fix: recover orphaned running jobs

Author: lkstrp

app/backends/base.py

@@ -111,6 +111,15 @@ async def monitor(
         """
         ...
 
+    @abstractmethod
+    async def check_job_status(self, job_id: str, work_dir: str) -> int | None:
+        """Check if a job process has finished without blocking.
+
+        Returns:
+            Exit code if the process has exited, None if still running.
+        """
+        ...
+
     @abstractmethod
     async def list_workflow_files(
         self,

app/backends/local.py

@@ -198,6 +198,27 @@ async def _drain_log(off: int) -> int:
 
             await asyncio.sleep(self._config.poll_interval)
 
+    async def check_job_status(self, job_id: str, work_dir: str) -> int | None:
+        """Check if a job process has finished without blocking."""
+        wd = Path(work_dir)
+        exitcode_path = wd / ".exitcode"
+        if exitcode_path.exists():
+            try:
+                return int(exitcode_path.read_text(encoding="utf-8").strip())
+            except (ValueError, OSError):
+                return None
+        pid_path = wd / ".pid"
+        if pid_path.exists():
+            try:
+                pid = int(pid_path.read_text(encoding="utf-8").strip())
+                os.kill(pid, 0)
+                return None  # still running
+            except (ProcessLookupError, PermissionError):
+                return -1  # dead without exitcode
+            except (ValueError, OSError):
+                return None
+        return None
+
     async def list_workflow_files(
         self,
         job_id: str,

app/backends/slurm_ssh.py

@@ -393,6 +393,26 @@ async def monitor(
 
             await asyncio.sleep(self._config.poll_interval)
 
+    async def check_job_status(self, job_id: str, work_dir: str) -> int | None:
+        """Check if a job process has finished without blocking."""
+        wd = shlex.quote(work_dir)
+        cmd = (
+            f"test -f {wd}/.exitcode && cat {wd}/.exitcode "
+            f"|| (kill -0 $(cat {wd}/.pid 2>/dev/null) 2>/dev/null "
+            f"&& echo {MONITOR_RUNNING_SENTINEL} "
+            f"|| echo {MONITOR_DEAD_SENTINEL})"
+        )
+        result = await self._run_ssh(cmd, check=False)
+        status = (result.stdout or "").strip()
+        if status == MONITOR_RUNNING_SENTINEL:
+            return None
+        if status == MONITOR_DEAD_SENTINEL:
+            return -1
+        try:
+            return int(status)
+        except ValueError:
+            return None
+
     async def check_connectivity(self) -> bool:
         """Check SSH connectivity and scratch filesystem health."""
         try:

app/tasks.py

@@ -223,6 +223,29 @@ async def sync_job_data_loop(
                 continue
             if not record.work_dir:
                 continue
+
+            # Recover orphaned RUNNING jobs (no active monitor task)
+            if record.task is None or record.task.done():
+                try:
+                    exit_code = await backend.check_job_status(
+                        job_id, record.work_dir
+                    )
+                    if exit_code is not None:
+                        logger.warning(
+                            "Recovering orphaned job %s (exit code %d)",
+                            job_id,
+                            exit_code,
+                        )
+                        store.mark_finished(job_id, exit_code)
+                        store.persist(record)
+                        continue
+                except Exception:
+                    logger.warning(
+                        "Failed to check orphaned job %s",
+                        job_id,
+                        exc_info=True,
+                    )
+
             _flush_logs(store, job_id)
             snkmt_db_path = store.get_snkmt_db_path(job_id)
             if snkmt_db_path:

tests/test_base.py

@@ -48,6 +48,9 @@ async def save_cache(self, job_id, work_dir, cache_key, cache_dirs):
     async def sync_snkmt_db(self, job_id, work_dir, local_path):
         pass
 
+    async def check_job_status(self, job_id, work_dir):
+        return None
+
     async def check_connectivity(self):
         return True