Merge pull request #2 from QJHWC/fix/ci-cross-platform

fix: stabilize cross-platform CI and SSH timeout cleanup

Author: QJHWC

.gitattributes

@@ -0,0 +1,8 @@
+* text=auto eol=lf
+
+*.eps binary
+*.jpeg binary
+*.jpg binary
+*.pdf binary
+*.png binary
+*.zip binary

.github/workflows/ci.yml

@@ -16,8 +16,8 @@ jobs:
         python: ["3.10", "3.11", "3.12"]
     runs-on: ${{ matrix.os }}
     steps:
-      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
-      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd
+      - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405
         with:
           python-version: ${{ matrix.python }}
           cache: pip
@@ -29,7 +29,7 @@ jobs:
       - run: python -m pip install --upgrade pip
       - run: python -m pip install -r requirements-dev.txt
       - run: python -m pip check
-      - run: python -m pytest -q
+      - run: python -m pytest -q --tb=short
       - run: python -m ruff check .
       - run: python -m mypy paperforge
       - if: matrix.python == '3.11'
@@ -42,8 +42,8 @@ jobs:
   real-compute-integration:
     runs-on: ubuntu-24.04
     steps:
-      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
-      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd
+      - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405
         with:
           python-version: "3.11"
           cache: pip
@@ -58,7 +58,7 @@ jobs:
           PAPERFORGE_KUBECTL: /usr/local/bin/kubectl
           PAPERFORGE_KUBERNETES_CONTEXT: kind-paperforge-v3
           PAPERFORGE_KUBERNETES_NAMESPACE: paperforge-v3
-        run: python -m pytest -q tests/integration/test_real_compute_backends.py
+        run: python -m pytest -q --tb=short tests/integration/test_real_compute_backends.py
       - if: always()
         run: |
           kind delete cluster --name paperforge-v3 || true

engine/perform_review.py

@@ -301,7 +301,7 @@ def load_paper(pdf_path, num_pages=None, min_size=100):
 
 
 def load_review(path):
-    with open(path) as json_file:
+    with open(path, encoding="utf-8") as json_file:
         loaded = json.load(json_file)
     return loaded["review"]
 
@@ -332,7 +332,7 @@ def get_review_fewshot_examples(num_fs_examples=1):
     ):
         txt_path = paper.replace(".pdf", ".txt")
         if os.path.exists(txt_path):
-            with open(txt_path) as f:
+            with open(txt_path, encoding="utf-8") as f:
                 paper_text = f.read()
         else:
             paper_text = load_paper(paper)

engine/run_lock.py

@@ -2,8 +2,7 @@
 
 Prevents concurrent processes from writing to the same workspace
 (e.g. two training sweeps or a sync + writeup running in parallel).
-Uses POSIX fcntl advisory locking; on platforms without fcntl support,
-locking is unavailable and callers will receive a RuntimeError.
+Uses POSIX ``flock`` or the Windows CRT byte-range lock.
 """
 
 from __future__ import annotations
@@ -13,25 +12,59 @@
 from collections.abc import Iterator
 from contextlib import contextmanager
 from pathlib import Path
-from typing import TextIO
+from typing import TextIO, TypedDict
 
 try:
     import fcntl
-except Exception as exc:
+except ImportError:
     fcntl = None  # type: ignore[assignment]
-    _FCNTL_IMPORT_ERROR = exc
-else:
-    _FCNTL_IMPORT_ERROR = None
+
+try:
+    import msvcrt
+except ImportError:
+    msvcrt = None  # type: ignore[assignment]
 
 
 _REENTRANT_GUARD = threading.RLock()
-_REENTRANT_LOCKS: dict[str, dict[str, object]] = {}
+
+
+class _HeldLock(TypedDict):
+    fp: TextIO
+    count: int
+
+
+_REENTRANT_LOCKS: dict[str, _HeldLock] = {}
 
 
 class RunLockTimeoutError(TimeoutError):
     """Raised when the workspace lock cannot be acquired within timeout."""
 
 
+def _lock_nonblocking(lock_fp: TextIO) -> None:
+    lock_fp.seek(0)
+    if fcntl is not None:
+        fcntl.flock(lock_fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return
+    if msvcrt is not None:
+        msvcrt.locking(lock_fp.fileno(), msvcrt.LK_NBLCK, 1)
+        return
+    raise RuntimeError("no supported process lock implementation is available")
+
+
+def _unlock(lock_fp: TextIO) -> None:
+    lock_fp.seek(0)
+    if fcntl is not None:
+        fcntl.flock(lock_fp.fileno(), fcntl.LOCK_UN)
+    elif msvcrt is not None:
+        msvcrt.locking(lock_fp.fileno(), msvcrt.LK_UNLCK, 1)
+
+
+def _is_lock_contention(exc: OSError) -> bool:
+    if isinstance(exc, BlockingIOError):
+        return True
+    return msvcrt is not None and exc.errno in {13, 33, 36}
+
+
 def acquire_run_lock(
     run_dir: Path,
     timeout: int = 30,
@@ -43,8 +76,8 @@ def acquire_run_lock(
     The caller must keep the returned file handle alive until the
     protected operation completes; closing the handle releases the lock.
     """
-    if fcntl is None:
-        raise RuntimeError(f"fcntl unavailable on this platform: {_FCNTL_IMPORT_ERROR}")
+    if fcntl is None and msvcrt is None:
+        raise RuntimeError("no supported process lock implementation is available")
 
     run_dir = run_dir.resolve()
     run_dir.mkdir(parents=True, exist_ok=True)
@@ -54,23 +87,29 @@ def acquire_run_lock(
         existing = _REENTRANT_LOCKS.get(str(lock_path))
         if existing is not None:
             existing["count"] = int(existing["count"]) + 1
-            return existing["fp"]  # type: ignore[return-value]
+            return existing["fp"]
 
     lock_path.touch(exist_ok=True)
-    fp = lock_path.open("w", encoding="utf-8")
+    fp = lock_path.open("a+", encoding="utf-8")
+    if lock_path.stat().st_size == 0:
+        fp.write("\0")
+        fp.flush()
 
     start = time.monotonic()
     waiting_printed = False
     while True:
         try:
-            fcntl.flock(fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            _lock_nonblocking(fp)
             with _REENTRANT_GUARD:
                 _REENTRANT_LOCKS[str(lock_path)] = {
                     "fp": fp,
                     "count": 1,
                 }
             return fp
-        except BlockingIOError:
+        except OSError as exc:
+            if not _is_lock_contention(exc):
+                fp.close()
+                raise
             elapsed = time.monotonic() - start
             if timeout >= 0 and elapsed >= float(timeout):
                 fp.close()
@@ -84,8 +123,6 @@ def acquire_run_lock(
 
 
 def release_run_lock(lock_fp: TextIO) -> None:
-    if fcntl is None:
-        return
     lock_path = Path(getattr(lock_fp, "name", "")).resolve()
     with _REENTRANT_GUARD:
         existing = _REENTRANT_LOCKS.get(str(lock_path))
@@ -96,7 +133,7 @@ def release_run_lock(lock_fp: TextIO) -> None:
                 return
             _REENTRANT_LOCKS.pop(str(lock_path), None)
     try:
-        fcntl.flock(lock_fp.fileno(), fcntl.LOCK_UN)
+        _unlock(lock_fp)
     finally:
         lock_fp.close()
 

paperforge/compute/_local_supervisor.py

@@ -42,6 +42,13 @@ def _atomic_json(path: Path, payload: dict[str, Any]) -> None:
             Path(temporary).unlink(missing_ok=True)
 
 
+def _kill_process_group(pid: int, sig: int) -> None:
+    killpg = getattr(os, "killpg", None)
+    if killpg is None:
+        raise AttributeError("process-group signaling is unavailable")
+    killpg(pid, sig)
+
+
 def _terminate(process: subprocess.Popen[bytes]) -> None:
     if process.poll() is not None:
         return
@@ -55,7 +62,7 @@ def _terminate(process: subprocess.Popen[bytes]) -> None:
             process.terminate()
     else:
         try:
-            os.killpg(process.pid, signal.SIGTERM)
+            _kill_process_group(process.pid, signal.SIGTERM)
         except (AttributeError, ProcessLookupError):
             process.terminate()
     try:
@@ -65,7 +72,10 @@ def _terminate(process: subprocess.Popen[bytes]) -> None:
             process.kill()
         else:
             try:
-                os.killpg(process.pid, signal.SIGKILL)
+                _kill_process_group(
+                    process.pid,
+                    getattr(signal, "SIGKILL", signal.SIGTERM),
+                )
             except (AttributeError, ProcessLookupError):
                 process.kill()
         process.wait(timeout=5)

paperforge/compute/docker.py

@@ -10,7 +10,7 @@
 import time
 from collections.abc import Sequence
 from dataclasses import dataclass
-from pathlib import Path, PurePosixPath
+from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import Any
 
 from paperforge.path_safety import (
@@ -48,6 +48,14 @@
 )
 
 
+def _host_container_user() -> str:
+    getuid = vars(os).get("getuid")
+    getgid = vars(os).get("getgid")
+    if not callable(getuid) or not callable(getgid):
+        return "65532:65532"
+    return f"{getuid()}:{getgid()}"
+
+
 @dataclass(frozen=True)
 class DockerConfig:
     image: str
@@ -61,7 +69,15 @@ class DockerConfig:
     def __post_init__(self) -> None:
         if not self.image or "\x00" in self.image:
             raise ValueError("docker image must be non-empty")
-        if not _RUNTIME_PATTERN.fullmatch(self.runtime):
+        windows_runtime = PureWindowsPath(self.runtime)
+        safe_windows_runtime = (
+            os.name == "nt"
+            and windows_runtime.is_absolute()
+            and ".." not in windows_runtime.parts
+            and not self.runtime.startswith(("\\\\", "//"))
+            and not any(ord(character) < 32 for character in self.runtime)
+        )
+        if not _RUNTIME_PATTERN.fullmatch(self.runtime) and not safe_windows_runtime:
             raise ValueError("docker runtime contains unsafe characters")
         workdir = PurePosixPath(self.container_workdir)
         if not workdir.is_absolute() or ".." in workdir.parts:
@@ -219,12 +235,7 @@ def _submit_argv(
             "--security-opt",
             "no-new-privileges",
             "--user",
-            self.config.container_user
-            or (
-                f"{os.getuid()}:{os.getgid()}"
-                if hasattr(os, "getuid")
-                else "65532:65532"
-            ),
+            self.config.container_user or _host_container_user(),
             "--pids-limit",
             "512",
             "--tmpfs",

paperforge/compute/local.py

@@ -745,29 +745,34 @@ def _supervisor_alive(self, job_id: str, metadata: dict[str, Any]) -> bool:
             powershell = shutil.which("powershell.exe") or shutil.which("pwsh.exe")
             if powershell is None:
                 return False
-            inspected = subprocess.run(
-                [
-                    powershell,
-                    "-NoLogo",
-                    "-NoProfile",
-                    "-NonInteractive",
-                    "-Command",
-                    (
-                        "$p = Get-CimInstance Win32_Process -Filter "
-                        f"\"ProcessId = {raw_pid}\"; "
-                        "if ($null -ne $p) { [Console]::Out.Write($p.CommandLine) }"
-                    ),
-                ],
-                check=False,
-                capture_output=True,
-                text=True,
-            )
-            command = inspected.stdout.strip()
-            return (
-                inspected.returncode == 0
-                and "paperforge.compute._local_supervisor" in command
-                and str(metadata["launch_path"]) in command
-            )
+            for attempt in range(3):
+                inspected = subprocess.run(
+                    [
+                        powershell,
+                        "-NoLogo",
+                        "-NoProfile",
+                        "-NonInteractive",
+                        "-Command",
+                        (
+                            "$p = Get-CimInstance Win32_Process -Filter "
+                            f"\"ProcessId = {raw_pid}\"; "
+                            "if ($null -ne $p) { "
+                            "[Console]::Out.Write($p.CommandLine) }"
+                        ),
+                    ],
+                    check=False,
+                    capture_output=True,
+                    text=True,
+                )
+                command = inspected.stdout.strip()
+                if inspected.returncode == 0 and command:
+                    return (
+                        "paperforge.compute._local_supervisor" in command
+                        and str(metadata["launch_path"]) in command
+                    )
+                if attempt < 2:
+                    time.sleep(0.05)
+            return False
         try:
             os.kill(raw_pid, 0)
         except (OSError, ProcessLookupError):
@@ -788,7 +793,8 @@ def _supervisor_alive(self, job_id: str, metadata: dict[str, Any]) -> bool:
     @staticmethod
     def _signal_supervisor(pid: int, *, force: bool) -> None:
         if os.name != "nt":
-            os.kill(pid, signal.SIGKILL if force else signal.SIGTERM)
+            force_signal = getattr(signal, "SIGKILL", signal.SIGTERM)
+            os.kill(pid, force_signal if force else signal.SIGTERM)
             return
         if force:
             subprocess.run(
@@ -930,6 +936,11 @@ def status(self, job_id: str, *, execute: bool = False) -> JobResult:
             return completion
         if self._supervisor_alive(job_id, dict(previous.metadata)):
             return previous
+        time.sleep(0.05)
+        completion = self._completion_result(job_id, previous)
+        if completion is not None:
+            self._remember(job_id, result=completion)
+            return completion
         process = self._processes.get(job_id)
         if process is not None and process.poll() is not None:
             time.sleep(0.05)