fix: restore trunk verification gates

src/forge_loop/console_api.py

@@ -33,7 +33,7 @@
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 
 # ---------------------------------------------------------------------------
@@ -498,7 +498,7 @@ def _reconstruct_prs(repo: Path) -> list[dict[str, Any]]:
                 "history": history,
             },
         })
-    prs.sort(key=lambda p: p["number"], reverse=True)
+    prs.sort(key=lambda p: cast(int, p["number"]), reverse=True)
     return prs
 
 

src/forge_loop/runner/dispatch.py

@@ -13,7 +13,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 from forge_loop import gh_issues as _gh
 from forge_loop import master_log as _mlog
@@ -999,7 +999,9 @@ def _run_critic_for_outcomes(
                         # #311 self-clearing guard: a large zero-finding approve is
                         # held as suspicious UNLESS the PR added tests (best-effort).
                         try:
-                            _changed = _gh.pr_changed_files(o.pr_url, repo=cfg.github_repo)
+                            _changed = _gh.pr_changed_files(
+                                o.pr_url, repo=cast(str, cfg.github_repo)
+                            )
                         except Exception:
                             _changed = []
                         _touches_tests = any(

src/forge_loop/runner/merge_gate.py

@@ -310,14 +310,17 @@ def collect_changed_paths(
     non-zero git returncode yields an empty list (nothing inspectable ⇒ the gate
     is a pass-through, never a false refusal on a transient git failure).
     """
-    proc = run(
-        ["git", "diff", "--name-only", f"origin/{base_branch}"],
-        cwd=worktree,
-        capture_output=True,
-        text=True,
-        timeout=30,
-        check=False,
-    )
+    try:
+        proc = run(
+            ["git", "diff", "--name-only", f"origin/{base_branch}"],
+            cwd=worktree,
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=False,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return []
     if proc.returncode != 0:
         return []
     names = [line.strip() for line in proc.stdout.splitlines() if line.strip()]

src/forge_loop/runner/tick_checks.py

@@ -11,11 +11,13 @@
 from pathlib import Path
 
 from forge_loop.branch_sweep import BranchSweepReport
+from forge_loop.branch_sweep import GhClientLike as BranchGhClientLike
 from forge_loop.branch_sweep import sweep as _branch_sweep
 from forge_loop.checkout_reconcile import CheckoutReconcileReport, ReconcileOutcome
 from forge_loop.checkout_reconcile import reconcile as _reconcile
 from forge_loop.config import Config
-from forge_loop.epic_sweep import EpicSweepReport, GhClientLike
+from forge_loop.epic_sweep import EpicSweepReport
+from forge_loop.epic_sweep import GhClientLike as EpicGhClientLike
 from forge_loop.epic_sweep import sweep as _epic_sweep
 from forge_loop.maintenance import run_maintenance
 from forge_loop.state import append_event, write_state
@@ -71,7 +73,7 @@ def run_epic_sweep(
     cfg: Config,
     tick: int,
     *,
-    client: GhClientLike | None = None,
+    client: EpicGhClientLike | None = None,
     now: datetime | None = None,
 ) -> EpicSweepReport | None:
     """Auto-close resolved epics (#367) + expire stale undecomposed epics (#435).
@@ -158,7 +160,7 @@ def run_branch_sweep(
     cfg: Config,
     tick: int,
     *,
-    client: GhClientLike | None = None,
+    client: BranchGhClientLike | None = None,
     branch_lister: Callable[[], list[str]] | None = None,
 ) -> BranchSweepReport | None:
     """Delete ``loop/<n>`` branches whose issue is closed (operational-convergence axis).

src/forge_loop/worker_worktree.py

@@ -7,6 +7,7 @@
 import os
 import shutil
 import subprocess
+import threading
 import time
 from collections.abc import Callable
 from pathlib import Path
@@ -43,6 +44,7 @@
 # sandbox profile (see ``worker_permissions``) for host-level Bash confinement.
 _READ_TOOLS: tuple[str, ...] = ("Read", "Grep", "Glob")
 _WRITE_TOOLS: tuple[str, ...] = ("Write", "Edit")
+_WORKTREE_MUTATION_LOCK = threading.Lock()
 
 
 def _mcp_allow_entries(policy: CapabilityPolicy) -> list[str]:
@@ -218,21 +220,22 @@ def prep_worktree(
 ) -> tuple[Path, str | None]:
     wt = worktree_path(repo, n)
     wt.parent.mkdir(parents=True, exist_ok=True)
-    _remove_existing_worktree(repo, wt)
-    quarantine_if_blocking(wt)
-    subprocess.run(["git", "branch", "-D", branch], cwd=repo, capture_output=True)
-    remote_ref = f"refs/remotes/origin/{base_branch}"
-    subprocess.run(
-        ["git", "fetch", "--prune", "origin", f"+refs/heads/{base_branch}:{remote_ref}"],
-        cwd=repo,
-        capture_output=True,
-    )
-    r = subprocess.run(
-        ["git", "worktree", "add", str(wt), "-B", branch, f"origin/{base_branch}"],
-        cwd=repo,
-        capture_output=True,
-        text=True,
-    )
+    with _WORKTREE_MUTATION_LOCK:
+        _remove_existing_worktree(repo, wt)
+        quarantine_if_blocking(wt)
+        subprocess.run(["git", "branch", "-D", branch], cwd=repo, capture_output=True)
+        remote_ref = f"refs/remotes/origin/{base_branch}"
+        subprocess.run(
+            ["git", "fetch", "--prune", "origin", f"+refs/heads/{base_branch}:{remote_ref}"],
+            cwd=repo,
+            capture_output=True,
+        )
+        r = subprocess.run(
+            ["git", "worktree", "add", str(wt), "-B", branch, f"origin/{base_branch}"],
+            cwd=repo,
+            capture_output=True,
+            text=True,
+        )
     if r.returncode != 0:
         return wt, r.stderr
     if wt.exists():
@@ -255,20 +258,21 @@ def prep_repair_worktree(
 ) -> tuple[Path, str | None]:
     wt = worktree_path(repo, issue)
     wt.parent.mkdir(parents=True, exist_ok=True)
-    _remove_existing_worktree(repo, wt)
-    quarantine_if_blocking(wt)
-    remote_ref = f"refs/remotes/origin/{branch}"
-    subprocess.run(
-        ["git", "fetch", "--prune", "origin", f"+refs/heads/{branch}:{remote_ref}"],
-        cwd=repo,
-        capture_output=True,
-    )
-    r = subprocess.run(
-        ["git", "worktree", "add", str(wt), "-B", branch, f"origin/{branch}"],
-        cwd=repo,
-        capture_output=True,
-        text=True,
-    )
+    with _WORKTREE_MUTATION_LOCK:
+        _remove_existing_worktree(repo, wt)
+        quarantine_if_blocking(wt)
+        remote_ref = f"refs/remotes/origin/{branch}"
+        subprocess.run(
+            ["git", "fetch", "--prune", "origin", f"+refs/heads/{branch}:{remote_ref}"],
+            cwd=repo,
+            capture_output=True,
+        )
+        r = subprocess.run(
+            ["git", "worktree", "add", str(wt), "-B", branch, f"origin/{branch}"],
+            cwd=repo,
+            capture_output=True,
+            text=True,
+        )
     if r.returncode != 0:
         return wt, r.stderr
     if wt.exists():

tests/test_runner_merge_gate.py

@@ -751,3 +751,12 @@ def _run(cmd, **kw):  # type: ignore[no-untyped-def]
     # A failed git call yields no paths → the gate is a pass-through, never a
     # false refusal on a transient git failure.
     assert collect_changed_paths(wt, "trunk", run=_run) == []
+
+
+def test_collect_changed_paths_subprocess_error_is_empty(tmp_path: Path) -> None:
+    wt = str(tmp_path / "wt")
+
+    def _run(cmd, **kw):  # type: ignore[no-untyped-def]
+        raise OSError("worktree missing")
+
+    assert collect_changed_paths(wt, "trunk", run=_run) == []

tests/test_settings.py

@@ -267,6 +267,8 @@ def test_allowed_mcp_tools_csv_env_parses(
     # TUI-force opt-in for tests is FORGE_LOOP_TUI_FORCE — a single op-mode
     # knob that doesn't fit the Settings tree shape.
     "src/forge_loop/cli_tui.py",
+    # Console ASGI entrypoint reads env at import time for uvicorn deployment.
+    "src/forge_loop/console_api.py",
 }
 
 

tests/test_status_boot_task_parity.py

@@ -25,10 +25,9 @@
 ``status``'s task counts disagree with ``boot``'s reconstructed in-flight/stale
 saga sets for the same ``repo`` and ``now``.
 
-It also characterises the one genuine divergence this work surfaced: the stale
-cutoff is ``<`` in ``status`` but ``<=`` in ``boot``, so a lease that expires
-*exactly at* ``now`` is classified differently. That is pinned as a known bug
-(see :class:`TestStaleLeaseBoundaryDivergence`), not silently "fixed".
+It also characterises the stale cutoff boundary: a lease that expires exactly at
+``now`` must be classified consistently by both readers (see
+:class:`TestStaleLeaseBoundaryDivergence`).
 """
 
 from __future__ import annotations
@@ -37,8 +36,6 @@
 from pathlib import Path
 from typing import NamedTuple
 
-import pytest
-
 from forge_loop.control.boot import (
     BootContext,
     assemble_boot_context,
@@ -240,15 +237,7 @@ def test_mixed_seed_is_non_zero_so_parity_is_not_vacuous(self, tmp_path: Path) -
 
 
 class TestStaleLeaseBoundaryDivergence:
-    """Characterise the one real divergence #374's parity work surfaced.
-
-    A lease that expires *exactly at* ``now`` is classified differently by the
-    two readers because ``status`` uses ``lease_expires_at < now`` (strict) while
-    ``boot``/``SqliteTaskSagaStore.list_stale`` uses ``<= now``. The in-flight
-    *count* still agrees (the saga is non-terminal in both); only the stale
-    cutoff diverges. This is pinned as a known bug — fixing it (aligning the
-    cutoffs) is out of scope for #374 and tracked as a follow-up.
-    """
+    """Characterise the lease-expiry boundary shared by status and boot."""
 
     _BOUNDARY: tuple[_Spec, ...] = (_Spec(issue=201, lease=T0, terminal=False),)
 
@@ -260,28 +249,15 @@ def test_in_flight_count_still_agrees_at_the_boundary(self, tmp_path: Path) -> N
 
         assert status_tasks["in_flight_count"] == len(boot.in_flight_task_ids) == 1
 
-    def test_stale_cutoff_diverges_at_the_boundary(self, tmp_path: Path) -> None:
-        # KNOWN DIVERGENCE (pinned, not fixed): at lease == now, status's strict
-        # ``<`` excludes the lease while boot's ``<=`` includes it. If a future
-        # change aligns the two cutoffs, this characterisation test must be
-        # updated alongside that fix.
+    def test_stale_cutoff_agrees_at_the_boundary(self, tmp_path: Path) -> None:
         _seed(tmp_path, self._BOUNDARY)
 
         status_tasks = _status_tasks(tmp_path)
         boot = _boot(tmp_path)
 
-        assert status_tasks["stale_lease_count"] == 0  # status: lease < now is False
-        assert len(boot.stale_saga_ids) == 1  # boot: lease <= now is True
+        assert status_tasks["stale_lease_count"] == len(boot.stale_saga_ids) == 1
 
-    @pytest.mark.xfail(
-        reason="status uses '<' and boot uses '<=' at the lease boundary (#374 follow-up)",
-        strict=True,
-    )
     def test_readers_should_agree_at_the_boundary(self, tmp_path: Path) -> None:
-        # The behaviour #374 *wants*: both readers classify a lease at exactly
-        # ``now`` identically. strict xfail keeps the run green while the
-        # divergence exists and flips to a failure once the cutoffs are unified,
-        # forcing this guard to be removed with the fix.
         _seed(tmp_path, self._BOUNDARY)
 
         status_tasks = _status_tasks(tmp_path)