refactor: some cleanup, tests and deduplication

Author: lkstrp

.gitignore

@@ -12,3 +12,4 @@ __pycache__/
 *.pyc
 .pytest_cache/
 .ruff_cache/
+/data

app/backends/base.py

@@ -270,13 +270,12 @@ async def check_connectivity(self) -> bool:
         """Check whether the backend is reachable. Returns True if healthy."""
         ...
 
-    def resolve_job_logs(
+    def resolve_job_logs(  # noqa: B027
         self,
         jobs: list[SnkmtJobResponse],
         workflow_files: list[WorkflowFileInfo] | None,
     ) -> None:
         """Set job.log for each job based on backend-specific log paths."""
-        return
 
     @abstractmethod
     async def cleanup(

app/backends/slurm_ssh.py

@@ -48,6 +48,29 @@
 MONITOR_DEAD_SENTINEL = "DEAD"
 
 
+def _build_status_check_cmd(work_dir_quoted: str) -> str:
+    """Build shell command to check job exit status via .exitcode / PID probe."""
+    return (
+        f"test -f {work_dir_quoted}/.exitcode && cat {work_dir_quoted}/.exitcode "
+        f"|| (kill -0 $(cat {work_dir_quoted}/.pid 2>/dev/null) 2>/dev/null "
+        f"&& echo {MONITOR_RUNNING_SENTINEL} "
+        f"|| echo {MONITOR_DEAD_SENTINEL})"
+    )
+
+
+def _parse_job_status(status: str) -> int | None:
+    """Parse status check output → exit code, -1 for dead, or None for running."""
+    if status == MONITOR_RUNNING_SENTINEL:
+        return None
+    if status == MONITOR_DEAD_SENTINEL:
+        return -1
+    try:
+        return int(status)
+    except ValueError:
+        logger.warning("Unparsable job status value: %r, treating as running", status)
+        return None
+
+
 def _build_rsync_filter(cache_dirs: list[str]) -> str:
     """Build rsync --include/--exclude args from cache_dirs patterns.
 
@@ -84,8 +107,8 @@ def _add(arg: str) -> None:
 class SlurmSSHBackend(ComputeBackend):
     """
     Compute backend that connects to an HPC head node via SSH,
-    clones a workflow, runs Snakemake via pixi in a detached process,
-    and monitors via polling.
+    fetches a workflow into a bare repo and creates a worktree,
+    runs Snakemake via pixi in a detached process, and monitors via polling.
     """
 
     def __init__(self, config: SlurmSSHConfig) -> None:
@@ -314,16 +337,11 @@ async def monitor(
 
         while True:
             try:
-                # Read new log bytes, print a marker, then print the exit
-                # code (or "RUNNING" if still alive)
                 wd = shlex.quote(work_dir)
                 cmd = (
                     f"tail -c +{offset + 1} {wd}/.stdout.log 2>/dev/null; "
                     f"echo '{MONITOR_LOG_MARKER}'; "
-                    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})"
+                    f"{_build_status_check_cmd(wd)}"
                 )
                 result = await self._run_ssh(cmd, check=False)
                 stdout = result.stdout or ""
@@ -336,32 +354,20 @@ async def monitor(
                 consecutive_errors = 0
 
                 if new_log_data:
-                    # Advance offset so next poll only reads new bytes
                     offset += len(new_log_data.encode("utf-8", errors="replace"))
                     for line in new_log_data.splitlines():
                         log_callback(line)
 
-                # "RUNNING" = still going
-                # "DEAD" = process killed without writing .exitcode
-                # number = exit code
-                if status_part == MONITOR_DEAD_SENTINEL:
+                exit_code = _parse_job_status(status_part)
+                if exit_code == -1:
                     logger.warning(
                         "Job %s: wrapper process died without writing "
                         ".exitcode (likely OOM or SIGKILL)",
                         job_id,
                     )
                     return -1
-                if status_part != MONITOR_RUNNING_SENTINEL:
-                    try:
-                        return int(status_part)
-                    except ValueError:
-                        logger.warning(
-                            "Unexpected status value: %r, treating as running",
-                            status_part,
-                        )
-                        # Sleep for poll_interval before retrying
-                        await asyncio.sleep(self._config.poll_interval)
-                        continue
+                if exit_code is not None:
+                    return exit_code
 
             except (TimeoutError, OSError, asyncssh.Error) as exc:
                 consecutive_errors += 1
@@ -374,22 +380,8 @@ async def monitor(
     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
+        result = await self._run_ssh(_build_status_check_cmd(wd), check=False)
+        return _parse_job_status((result.stdout or "").strip())
 
     async def check_connectivity(self) -> bool:
         """Check SSH connectivity and scratch filesystem health."""

app/routes/jobs.py

@@ -76,12 +76,22 @@ def _build_outputs_response(
     return JobOutputsResponse(files=files)
 
 
-_RESPONSE_FIELDS = tuple(JobResponse.model_fields.keys())
-
-
 def _build_job_response(record: JobRecord) -> JobResponse:
-    """Build a JobResponse from a JobRecord using _RESPONSE_FIELDS."""
-    return JobResponse(**{f: getattr(record, f) for f in _RESPONSE_FIELDS})
+    """Build a JobResponse from a JobRecord with explicit field mapping."""
+    return JobResponse(
+        job_id=record.job_id,
+        status=record.status,
+        workflow=record.workflow,
+        configfile=record.configfile,
+        git_ref=record.git_ref,
+        git_sha=record.git_sha,
+        exit_code=record.exit_code,
+        created_at=record.created_at,
+        started_at=record.started_at,
+        completed_at=record.completed_at,
+        total_job_count=record.total_job_count,
+        jobs_finished=record.jobs_finished,
+    )
 
 
 @router.post("/jobs", response_model=JobResponse, status_code=201)

app/routes/snkmt.py

@@ -11,7 +11,7 @@
 import sqlite3
 from collections.abc import Callable
 from pathlib import Path as FilePath
-from typing import Annotated, Any, TypeVar
+from typing import Annotated, Any
 
 from fastapi import APIRouter, Depends, HTTPException, Path
 
@@ -33,7 +33,6 @@
 
 router = APIRouter()
 
-_T = TypeVar("_T")
 
 # Job statuses where snkmt data is not yet available (before workflow execution starts).
 _PRE_EXECUTION_STATUSES = frozenset({JobStatus.PENDING, JobStatus.SETUP})
@@ -69,7 +68,7 @@ def _require_snkmt(
     return snkmt_db, record.work_dir, record.workflow_files
 
 
-async def _run_snkmt_query[T](job_id: str, query: Callable[[], _T]) -> _T:
+async def _run_snkmt_query[T](job_id: str, query: Callable[[], T]) -> T:
     """Run query in a thread and map sqlite3.DatabaseError to HTTP 502."""
     try:
         return await asyncio.to_thread(query)

app/snkmt.py

@@ -85,11 +85,14 @@ class ErrorRow(TypedDict):
     rule_name: str | None
 
 
-def _rename_end_time_column(row: dict[str, Any]) -> dict[str, Any]:
-    """Rename DB column end_time → completed_at to match our naming."""
-    row = dict(row)
-    row["completed_at"] = row.pop("end_time", None)
-    return row
+def _rename_end_time_column(row: sqlite3.Row) -> dict[str, Any]:
+    """Rename DB column ``end_time`` → ``completed_at`` to match our TypedDicts.
+
+    Callers cast the result to the appropriate TypedDict (WorkflowRow or JobRow).
+    """
+    out = dict(row)
+    out["completed_at"] = out.pop("end_time", None)
+    return out
 
 
 def safe_json_loads(value: str | None) -> dict[str, Any] | None:
@@ -135,7 +138,7 @@ def get_workflow(conn: sqlite3.Connection, workflow_id: str) -> WorkflowRow | No
     ).fetchone()
     if row is None:
         return None
-    return cast("WorkflowRow", _rename_end_time_column(dict(row)))
+    return cast("WorkflowRow", _rename_end_time_column(row))
 
 
 def get_rules(conn: sqlite3.Connection, workflow_id: str) -> list[RuleRow]:
@@ -154,7 +157,7 @@ def get_jobs(conn: sqlite3.Connection, workflow_id: str) -> list[JobRow]:
         "WHERE j.workflow_id = ?",
         (workflow_id,),
     ).fetchall()
-    return [cast("JobRow", _rename_end_time_column(dict(r))) for r in rows]
+    return [cast("JobRow", _rename_end_time_column(r)) for r in rows]
 
 
 def get_job_files_by_snakemake_id(
@@ -254,7 +257,7 @@ def get_jobs_by_rule(
         "WHERE j.workflow_id = ? AND r.name = ?",  # noqa: S608 — parameterized query
         (workflow_id, rule_name),
     ).fetchall()
-    return [cast("JobRow", _rename_end_time_column(dict(r))) for r in rows]
+    return [cast("JobRow", _rename_end_time_column(r)) for r in rows]
 
 
 def get_rule_by_name(

app/store.py

@@ -142,24 +142,8 @@ def restore_from_disk(self) -> None:
                 kwargs = {k: data[k] for k in _PERSIST_FIELDS}
                 record = JobRecord(**kwargs)
                 self._jobs[record.job_id] = record
-            except json.JSONDecodeError as exc:
-                logger.warning(
-                    "Skipping corrupt job file (malformed JSON): %s: %s", json_path, exc
-                )
-            except KeyError as exc:
-                logger.warning(
-                    "Skipping corrupt job file (missing field %s): %s", exc, json_path
-                )
-            except ValueError as exc:
-                logger.warning(
-                    "Skipping corrupt job file (invalid field value): %s: %s",
-                    json_path,
-                    exc,
-                )
-            except TypeError as exc:
-                logger.warning(
-                    "Skipping corrupt job file (type error): %s: %s", json_path, exc
-                )
+            except (json.JSONDecodeError, KeyError, ValueError, TypeError) as exc:
+                logger.warning("Skipping corrupt job file: %s: %s", json_path, exc)
 
     def mark_setup(
         self, job_id: str, work_dir: str, git_ref: str | None, git_sha: str | None