feat: use bare repo + worktree instead of full clone and add slurm job logs resolution logic

Author: lkstrp

app/backends/base.py

@@ -6,10 +6,12 @@
 from pathlib import Path, PurePosixPath
 from typing import TYPE_CHECKING
 
+from app.utils import bare_repo_dir, parse_default_branch
+
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, Awaitable, Callable
 
-    from app.models import WorkflowFileInfo
+    from app.models import SnkmtJobResponse, WorkflowFileInfo
 
 logger = logging.getLogger(__name__)
 
@@ -60,19 +62,103 @@ def work_dir(self, job_id: str) -> str:
         ...
 
     @abstractmethod
+    async def _run_git_cmd(self, *args: str) -> str:
+        """Run a git command and return its stdout as a string.
+
+        Implementations must raise on non-zero exit codes.
+        """
+        ...
+
+    @abstractmethod
+    async def _copy_local_workflow(self, src: str, dst: str) -> None:
+        """Copy a local workflow directory to the work directory."""
+        ...
+
+    @abstractmethod
+    def _scratch_dir(self) -> str:
+        """Return the scratch directory path for this backend."""
+        ...
+
     async def prepare(
         self,
         job_id: str,
         workflow: str,
         git_ref: str | None = None,
     ) -> tuple[str, str | None, str | None]:
-        """
-        Prepare workflow in a new working directory.
+        """Prepare workflow in a new working directory.
 
-        If workflow is a URL, clone the repo (checking out git_ref if provided).
-        If it is a local path, upload the directory to the compute host.
+        If workflow is a URL, set up a bare repo and create a worktree.
+        If it is a local path, copy/upload the directory.
         """
-        ...
+        work_dir = self.work_dir(job_id)
+        git_sha: str | None = None
+
+        if workflow.startswith(("http://", "https://")):
+            repo_dir = bare_repo_dir(self._scratch_dir(), workflow)
+            resolved_ref = git_ref
+
+            # 1. Ensure bare repo exists (idempotent)
+            await self._run_git_cmd("git", "init", "--bare", repo_dir)
+
+            # 2. Resolve default branch if git_ref not provided
+            if resolved_ref is None:
+                out = await self._run_git_cmd(
+                    "git",
+                    "ls-remote",
+                    "--symref",
+                    workflow,
+                    "HEAD",
+                )
+                resolved_ref = parse_default_branch(out)
+
+            # 3. Fetch to a job-scoped ref
+            job_ref = f"refs/snakedispatch/{job_id}"
+            await self._run_git_cmd(
+                "git",
+                "-C",
+                repo_dir,
+                "fetch",
+                workflow,
+                f"{resolved_ref}:{job_ref}",
+            )
+
+            # 4. Create worktree (detached HEAD)
+            await self._run_git_cmd(
+                "git",
+                "-C",
+                repo_dir,
+                "worktree",
+                "add",
+                "--detach",
+                work_dir,
+                job_ref,
+            )
+
+            # 5. Extract SHA
+            out = await self._run_git_cmd(
+                "git",
+                "-C",
+                work_dir,
+                "rev-parse",
+                "HEAD",
+            )
+            git_sha = out.strip()
+
+            # 6. Clean up temp ref
+            await self._run_git_cmd(
+                "git",
+                "-C",
+                repo_dir,
+                "update-ref",
+                "-d",
+                job_ref,
+            )
+
+            git_ref = resolved_ref
+        else:
+            await self._copy_local_workflow(workflow, work_dir)
+
+        return work_dir, git_ref, git_sha
 
     @abstractmethod
     async def setup(
@@ -184,6 +270,14 @@ async def check_connectivity(self) -> bool:
         """Check whether the backend is reachable. Returns True if healthy."""
         ...
 
+    def resolve_job_logs(
+        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(
         self,

app/backends/local.py

@@ -35,51 +35,39 @@ def __init__(self, config: LocalConfig) -> None:
     def work_dir(self, job_id: str) -> str:
         return f"{self._config.scratch_dir}/jobs/{job_id}"
 
+    def _scratch_dir(self) -> str:
+        return self._config.scratch_dir
+
+    async def _run_git_cmd(self, *args: str) -> str:
+        """Run a git subprocess locally, returning stdout."""
+        if "init" in args:
+            Path(args[-1]).parent.mkdir(parents=True, exist_ok=True)
+        proc = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            cmd_str = " ".join(args)
+            msg = (
+                f"git command failed (exit {proc.returncode}): {cmd_str}\n"
+                f"stderr: {(stderr or b'').decode()}"
+            )
+            raise RuntimeError(msg)
+        return (stdout or b"").decode()
+
+    async def _copy_local_workflow(self, src: str, dst: str) -> None:
+        await asyncio.to_thread(shutil.copytree, src, dst, dirs_exist_ok=True)
+
     async def prepare(
         self,
         job_id: str,
         workflow: str,
         git_ref: str | None = None,
     ) -> tuple[str, str | None, str | None]:
         Path(self._config.scratch_dir).mkdir(parents=True, exist_ok=True)
-        work_dir = self.work_dir(job_id)
-        git_sha: str | None = None
-
-        if workflow.startswith(("http://", "https://")):
-            cmd = ["git", "clone", "--depth=1"]
-            if git_ref:
-                cmd += ["--branch", git_ref]
-            cmd += [workflow, work_dir]
-            proc = await asyncio.create_subprocess_exec(
-                *cmd,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-            _, stderr = await proc.communicate()
-            if proc.returncode != 0:
-                msg = f"git clone failed (exit {proc.returncode}): {stderr.decode()}"
-                raise RuntimeError(msg)
-
-            proc = await asyncio.create_subprocess_exec(
-                "git",
-                "-C",
-                work_dir,
-                "rev-parse",
-                "HEAD",
-                "--abbrev-ref",
-                "HEAD",
-                stdout=asyncio.subprocess.PIPE,
-            )
-            out, _ = await proc.communicate()
-            lines = out.decode().strip().splitlines()
-            git_sha = lines[0] if lines else None
-            git_ref = lines[1] if len(lines) > 1 else None
-        else:
-            src = Path(workflow)
-            dst = Path(work_dir)
-            await asyncio.to_thread(shutil.copytree, src, dst, dirs_exist_ok=True)
-
-        return work_dir, git_ref, git_sha
+        return await super().prepare(job_id, workflow, git_ref)
 
     async def setup(
         self,
@@ -94,7 +82,6 @@ async def setup(
                 await asyncio.to_thread(full_path.write_text, content)
                 logger.debug("Wrote setup file: %s", full_path)
 
-
     async def launch(
         self,
         job_id: str,
@@ -202,11 +189,12 @@ async def check_job_status(self, job_id: str, work_dir: str) -> int | None:
             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
+            else:
+                return None  # still running
         return None
 
     async def list_workflow_files(

app/backends/slurm_ssh.py

@@ -17,7 +17,7 @@
 
 from app.backends.base import CHUNK_SIZE, SNKMT_DB_FILENAME, ComputeBackend
 from app.config import SlurmSSHConfig
-from app.models import JobStatus, WorkflowFileInfo
+from app.models import JobStatus, SnkmtJobResponse, WorkflowFileInfo
 from app.utils import (
     build_wrapper_script,
     enforce_error_limit,
@@ -218,34 +218,21 @@ async def _upload_dir(self, local_dir: str, remote_dir: str) -> None:
                 msg = f"local tar failed with exit {local_proc.returncode}"
                 raise RuntimeError(msg)
 
-    async def prepare(
-        self,
-        job_id: str,
-        workflow: str,
-        git_ref: str | None = None,
-    ) -> tuple[str, str | None, str | None]:
-        """Clone/upload workflow and return (work_dir, git_ref, git_sha)."""
-        work_dir = self.work_dir(job_id)
-        git_sha: str | None = None
-
-        if workflow.startswith(("http://", "https://")):
-            cmd = "git clone --depth=1 "
-            if git_ref:
-                cmd += f"--branch {shlex.quote(git_ref)} "
-            cmd += f"{shlex.quote(workflow)} {shlex.quote(work_dir)}"
-            await self._run_ssh(cmd)
+    def _scratch_dir(self) -> str:
+        return self._config.scratch_dir
 
-            result = await self._run_ssh(
-                f"git -C {shlex.quote(work_dir)} rev-parse HEAD && "
-                f"git -C {shlex.quote(work_dir)} rev-parse --abbrev-ref HEAD"
-            )
-            lines = result.stdout.strip().splitlines()
-            git_sha = lines[0] if lines else None
-            git_ref = lines[1] if len(lines) > 1 else None
-        else:
-            await self._upload_dir(workflow, work_dir)
+    async def _run_git_cmd(self, *args: str) -> str:
+        """Run a git command over SSH, returning stdout."""
+        quoted = [shlex.quote(a) for a in args]
+        # For init, ensure parent dirs exist remotely
+        if "init" in args:
+            repo_path = shlex.quote(args[-1])
+            await self._run_ssh(f"mkdir -p $(dirname {repo_path})")
+        result = await self._run_ssh(" ".join(quoted))
+        return result.stdout or ""
 
-        return work_dir, git_ref, git_sha
+    async def _copy_local_workflow(self, src: str, dst: str) -> None:
+        await self._upload_dir(src, dst)
 
     async def setup(
         self,
@@ -268,7 +255,6 @@ async def setup(
                         await f.write(content)
                     logger.debug("Wrote setup file: %s", full_path)
 
-
     async def launch(
         self,
         job_id: str,
@@ -520,6 +506,45 @@ async def sync_snkmt_db(self, job_id: str, work_dir: str, local_path: Path) -> N
         finally:
             await self._run_ssh(f"rm -f {shlex.quote(remote_bak)}", check=False)
 
+    def resolve_job_logs(
+        self,
+        jobs: list[SnkmtJobResponse],
+        workflow_files: list[WorkflowFileInfo] | None,
+    ) -> None:
+        """Match slurm log files to jobs via structured path convention.
+
+        Expects logs at: logs/slurm/{rule}/{wildcards_string}/{slurm_id}.out
+        If multiple logs exist (retries), picks the latest (highest slurm ID).
+        """
+        # Index slurm log files by directory prefix
+        slurm_logs: dict[str, list[str]] = {}
+        if workflow_files:
+            for wf in workflow_files:
+                p = wf["path"]
+                if p.startswith("logs/slurm/") and p.endswith(".out"):
+                    dir_prefix = p.rsplit("/", 1)[0]
+                    slurm_logs.setdefault(dir_prefix, []).append(p)
+
+        def _slurm_id(path: str) -> int:
+            fname = path.rsplit("/", 1)[-1].removesuffix(".out")
+            try:
+                return int(fname)
+            except ValueError:
+                return -1
+
+        for job in jobs:
+            if job.wildcards:
+                wildcards_str = ",".join(
+                    f"{k}={v}" for k, v in sorted(job.wildcards.items())
+                )
+                slurm_dir = f"logs/slurm/{job.rule}/{wildcards_str}"
+            else:
+                slurm_dir = f"logs/slurm/{job.rule}"
+
+            candidates = slurm_logs.get(slurm_dir, [])
+            if candidates:
+                job.log = max(candidates, key=_slurm_id)
+
     async def cleanup(
         self,
         job_id: str,
@@ -528,15 +553,13 @@ async def cleanup(
         # 1. Kill the wrapper process if a PID file exists
         pid_file = f"{shlex.quote(work_dir)}/.pid"
         await self._run_ssh(
-            f"test -f {pid_file} && "
-            f"kill $(cat {pid_file}) 2>/dev/null || true",
+            f"test -f {pid_file} && kill $(cat {pid_file}) 2>/dev/null || true",
             check=False,
         )
         # Wait before sending SIGKILL in case SIGTERM wasn't enough
         await asyncio.sleep(5)
         await self._run_ssh(
-            f"test -f {pid_file} && "
-            f"kill -9 $(cat {pid_file}) 2>/dev/null || true",
+            f"test -f {pid_file} && kill -9 $(cat {pid_file}) 2>/dev/null || true",
             check=False,
         )
         # 2. Cancel only SLURM jobs launched from this work_dir

app/models.py

@@ -177,6 +177,7 @@ class SnkmtJobResponse(BaseModel):
     started_at: datetime | None = None
     completed_at: datetime | None = None
     files: list[SnkmtFileResponse] = []
+    log: str | None = None
 
     @field_validator("started_at", "completed_at", mode="before")
     @classmethod