diff --git a/src/autocoder/core/gatekeeper.py b/src/autocoder/core/gatekeeper.py
index b30727c4..4a6985d1 100644
--- a/src/autocoder/core/gatekeeper.py
+++ b/src/autocoder/core/gatekeeper.py
@@ -16,25 +16,28 @@
 - Deterministic logic (no AI needed)
 """
 
-import os
-import subprocess
-import logging
 import contextlib
+import hashlib
 import json
+import logging
+import os
 import shutil
+import subprocess
 import sys
-import hashlib
-from pathlib import Path
-from typing import Dict, Any, Optional
 from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from autocoder.core.engine_settings import load_engine_settings
+from autocoder.reviewers.base import ReviewConfig
+from autocoder.reviewers.factory import get_reviewer
+
+from .git_dirty import get_git_dirty_status
+from .project_config import infer_preset, load_project_config, synthesize_commands_from_preset
 
 # Direct imports (system code = fast!)
 from .test_framework_detector import TestFrameworkDetector
 from .worktree_manager import WorktreeManager
-from .project_config import load_project_config, infer_preset, synthesize_commands_from_preset
-from autocoder.reviewers.base import ReviewConfig
-from autocoder.reviewers.factory import get_reviewer
-from autocoder.core.engine_settings import load_engine_settings
 
 logger = logging.getLogger(__name__)
 
@@ -53,7 +56,9 @@ def _is_yolo_mode() -> bool:
         return raw not in {"", "0", "false", "no", "off"}
 
     @staticmethod
-    def _apply_allow_no_tests(test_results: Dict[str, Any], *, allow_no_tests: bool) -> Dict[str, Any]:
+    def _apply_allow_no_tests(
+        test_results: dict[str, Any], *, allow_no_tests: bool
+    ) -> dict[str, Any]:
         """
         YOLO-only escape hatch when a project has no test command/script.
 
@@ -74,7 +79,7 @@ def _apply_allow_no_tests(test_results: Dict[str, Any], *, allow_no_tests: bool)
         combined = (str(out.get("output", "")) + str(out.get("errors", ""))).lower()
 
         # npm: Missing script: "test"
-        if "npm" in cmd and "missing script" in combined and "\"test\"" in combined:
+        if "npm" in cmd and "missing script" in combined and '"test"' in combined:
             out["passed"] = True
             out["note"] = "No test script detected; allowed by configuration (YOLO mode)"
             return out
@@ -125,16 +130,16 @@ def __init__(self, project_dir: str):
     def verify_and_merge(
         self,
         branch_name: str,
-        worktree_path: Optional[str] = None,
-        agent_id: Optional[str] = None,
+        worktree_path: str | None = None,
+        agent_id: str | None = None,
         *,
         feature_id: int | None = None,
-        main_branch: Optional[str] = None,
+        main_branch: str | None = None,
         fetch_remote: bool = False,
         push_remote: bool = False,
         allow_no_tests: bool = False,
         delete_feature_branch: bool = True,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Verify a feature branch and merge to main if tests pass.
 
@@ -171,9 +176,9 @@ def verify_and_merge(
         temp_worktree_path = None
         verify_branch = None
         approved = False
-        test_results: Dict[str, Any] | None = None
-        verification: Dict[str, Any] = {}
-        review: Dict[str, Any] | None = None
+        test_results: dict[str, Any] | None = None
+        verification: dict[str, Any] = {}
+        review: dict[str, Any] | None = None
 
         def detect_main_branch() -> str:
             if main_branch:
@@ -217,76 +222,19 @@ def origin_exists() -> bool:
             except subprocess.CalledProcessError:
                 return False
 
-        def _git_porcelain() -> list[str]:
-            raw = subprocess.run(
-                ["git", "status", "--porcelain"],
-                cwd=self.project_dir,
-                capture_output=True,
-                text=True,
-            ).stdout
-            lines = [ln for ln in raw.splitlines() if ln.strip()]
-            return lines
-
-        def _split_dirty(lines: list[str]) -> tuple[list[str], list[str]]:
-            # Ignore runtime/build artifacts that the orchestrator/UI creates in the main tree.
-            ignore_substrings = [
-                ".autocoder/",
-                "worktrees/",
-                "agent_system.db",
-                ".eslintrc.json",
-            ]
-            ignore_untracked_filenames = {
-                # Claude Code CLI can leave these behind in the target project root.
-                # They are not part of the user's repo and shouldn't block deterministic merges.
-                ".claude_settings.json",
-                "claude-progress.txt",
-            }
-            ignored: list[str] = []
-            remaining: list[str] = []
-            for ln in lines:
-                target = ln.replace("\\", "/")
-                status = ln[:2]
-                path_part = ln[3:] if len(ln) > 3 else ""
-                # Handle renames like: "R  old -> new"
-                if "->" in path_part:
-                    path_part = path_part.split("->", 1)[-1].strip()
-                filename = path_part.replace("\\", "/").split("/")[-1] if path_part else ""
-
-                # Ignore known runtime artifacts (any status).
-                if any(s in target for s in ignore_substrings):
-                    ignored.append(ln)
-                # Ignore known Claude CLI artifacts only when untracked.
-                elif status == "??" and filename in ignore_untracked_filenames:
-                    ignored.append(ln)
-                # Claude CLI sometimes drops a redundant root-level app_spec.txt even when prompts/app_spec.txt exists.
-                # Treat it as an artifact only in that case, and only when untracked.
-                elif (
-                    status == "??"
-                    and filename == "app_spec.txt"
-                    and (self.project_dir / "prompts" / "app_spec.txt").exists()
-                ):
-                    ignored.append(ln)
-                # AutoCoder prompt scaffolding files are often left untracked in the target project.
-                # They are not part of the feature code and should not block merges.
-                elif status == "??":
-                    rel = path_part.replace("\\", "/")
-                    if rel == "prompts/" or rel == "prompts":
-                        ignored.append(ln)
-                    elif rel.startswith("prompts/"):
-                        rel_name = rel.split("/")[-1] if rel else ""
-                        # Defensive: only ignore the known AutoCoder prompt filenames.
-                        if rel_name == "app_spec.txt" or rel_name.endswith("_prompt.txt"):
-                            ignored.append(ln)
-                        else:
-                            remaining.append(ln)
-                else:
-                    remaining.append(ln)
-            return ignored, remaining
+        # Note: git dirty detection is shared between Gatekeeper and server preflight.
+        # See: autocoder.core.git_dirty
 
-        def _write_artifact(result: Dict[str, Any]) -> None:
+        def _write_artifact(result: dict[str, Any]) -> None:
             try:
                 if feature_id is not None:
-                    out_dir = self.project_dir / ".autocoder" / "features" / str(int(feature_id)) / "gatekeeper"
+                    out_dir = (
+                        self.project_dir
+                        / ".autocoder"
+                        / "features"
+                        / str(int(feature_id))
+                        / "gatekeeper"
+                    )
                 else:
                     out_dir = self.project_dir / ".autocoder" / "gatekeeper"
                 out_dir.mkdir(parents=True, exist_ok=True)
@@ -327,7 +275,7 @@ def _review_config_from_project(cfg: Any) -> ReviewConfig:
                 gemini_model=getattr(r, "gemini_model", None),
             )
 
-        def _run_shell(command: str, *, cwd: Path, timeout_s: int | None = None) -> Dict[str, Any]:
+        def _run_shell(command: str, *, cwd: Path, timeout_s: int | None = None) -> dict[str, Any]:
             try:
                 cmd = _expand_placeholders(command, cwd)
                 result = subprocess.run(
@@ -391,8 +339,8 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
             detected_main = detect_main_branch()
             has_origin = origin_exists()
 
-            porcelain = _git_porcelain()
-            ignored_dirty, remaining_dirty = _split_dirty(porcelain)
+            dirty = get_git_dirty_status(self.project_dir)
+            ignored_dirty, remaining_dirty = dirty.ignored, dirty.remaining
             can_update_ref_without_checkout = bool(ignored_dirty) and not remaining_dirty
             if remaining_dirty:
                 return {
@@ -422,18 +370,32 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                     }
 
             # Step 2: Create temporary worktree for verification
-            temp_worktree_path = self.project_dir / f"verify_temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+            temp_worktree_path = (
+                self.project_dir / f"verify_temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+            )
 
             try:
-                base_ref = f"origin/{detected_main}" if (fetch_remote and has_origin) else detected_main
-                safe_branch_name = branch_name.replace(" ", "-").replace("\\", "-").replace(":", "-")
+                base_ref = (
+                    f"origin/{detected_main}" if (fetch_remote and has_origin) else detected_main
+                )
+                safe_branch_name = (
+                    branch_name.replace(" ", "-").replace("\\", "-").replace(":", "-")
+                )
                 verify_branch = f"verify/{safe_branch_name}"
                 # Create temp worktree on base ref
                 subprocess.run(
-                    ["git", "worktree", "add", "-b", verify_branch, str(temp_worktree_path), base_ref],
+                    [
+                        "git",
+                        "worktree",
+                        "add",
+                        "-b",
+                        verify_branch,
+                        str(temp_worktree_path),
+                        base_ref,
+                    ],
                     cwd=self.project_dir,
                     check=True,
-                    capture_output=True
+                    capture_output=True,
                 )
                 logger.info(f"✓ Created temporary worktree: {temp_worktree_path}")
             except subprocess.CalledProcessError as e:
@@ -441,7 +403,7 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                 return {
                     "approved": False,
                     "reason": "Failed to create temporary worktree for verification",
-                    "error": str(e)
+                    "error": str(e),
                 }
 
             # Step 3: In temp worktree, attempt merge (no commit yet)
@@ -450,7 +412,7 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                     ["git", "merge", "--no-commit", "--no-ff", branch_name],
                     cwd=str(temp_worktree_path),
                     capture_output=True,
-                    text=True
+                    text=True,
                 )
 
                 if merge_result.returncode != 0:
@@ -460,18 +422,14 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                         "approved": False,
                         "reason": "Merge conflict - needs manual resolution",
                         "merge_conflict": True,
-                        "errors": merge_result.stderr
+                        "errors": merge_result.stderr,
                     }
 
                 logger.info("✓ Merged branch in temp worktree (no commit yet)")
 
             except subprocess.CalledProcessError as e:
                 logger.error(f"✗ Merge failed: {e}")
-                return {
-                    "approved": False,
-                    "reason": "Merge command failed",
-                    "error": str(e)
-                }
+                return {"approved": False, "reason": "Merge command failed", "error": str(e)}
 
             # Step 4: Run tests IN TEMP WORKTREE
             logger.info("🧪 Running verification in temporary worktree...")
@@ -494,9 +452,14 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                         selected = Gatekeeper._select_node_install_command(temp_path)
                         if selected:
                             setup_cmd = selected
-                    verification["setup"] = _run_shell(setup_cmd, cwd=temp_path, timeout_s=setup.timeout_s)
+                    verification["setup"] = _run_shell(
+                        setup_cmd, cwd=temp_path, timeout_s=setup.timeout_s
+                    )
                     verification["setup"]["allow_fail"] = bool(getattr(setup, "allow_fail", False))
-                    if not verification["setup"]["passed"] and not verification["setup"]["allow_fail"]:
+                    if (
+                        not verification["setup"]["passed"]
+                        and not verification["setup"]["allow_fail"]
+                    ):
                         result = {
                             "approved": False,
                             "reason": "Setup failed",
@@ -509,7 +472,7 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
 
                 # Run remaining commands; ensure "test" runs early.
                 ordered = ["test", "lint", "typecheck", "format", "build", "acceptance"]
-                seen = set(["setup"])
+                seen = {"setup"}
                 for name in ordered + sorted(k for k in command_specs.keys() if k not in ordered):
                     if name in seen:
                         continue
@@ -517,7 +480,9 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                     if not spec or not getattr(spec, "command", None):
                         continue
                     seen.add(name)
-                    verification[name] = _run_shell(spec.command, cwd=temp_path, timeout_s=spec.timeout_s)
+                    verification[name] = _run_shell(
+                        spec.command, cwd=temp_path, timeout_s=spec.timeout_s
+                    )
                     verification[name]["allow_fail"] = bool(getattr(spec, "allow_fail", False))
                     if name == "test":
                         verification[name] = Gatekeeper._apply_allow_no_tests(
@@ -547,7 +512,9 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                 test_results = verification.get("test")
             else:
                 test_results = self._run_tests_in_directory(str(temp_worktree_path))
-                test_results = Gatekeeper._apply_allow_no_tests(test_results, allow_no_tests=allow_no_tests)
+                test_results = Gatekeeper._apply_allow_no_tests(
+                    test_results, allow_no_tests=allow_no_tests
+                )
                 verification["test"] = test_results
                 if not test_results.get("success", False):
                     result = {
@@ -588,7 +555,8 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                         "skipped": bool(rr.skipped),
                         "reason": rr.reason,
                         "findings": [
-                            {"severity": f.severity, "message": f.message, "file": f.file} for f in (rr.findings or [])
+                            {"severity": f.severity, "message": f.message, "file": f.file}
+                            for f in (rr.findings or [])
                         ],
                         "stdout": rr.stdout,
                         "stderr": rr.stderr,
@@ -607,7 +575,12 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                         "engines": list(review_cfg.engines or []),
                     }
 
-                if review_cfg.mode == "gate" and review and not review.get("approved") and not review.get("skipped"):
+                if (
+                    review_cfg.mode == "gate"
+                    and review
+                    and not review.get("approved")
+                    and not review.get("skipped")
+                ):
                     result = {
                         "approved": False,
                         "reason": "Review failed",
@@ -629,7 +602,7 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                     cwd=str(temp_worktree_path),
                     check=True,
                     capture_output=True,
-                    text=True
+                    text=True,
                 )
                 logger.info(f"✓ Committed merge in temp worktree: {commit_result.stdout.strip()}")
             except subprocess.CalledProcessError as e:
@@ -657,6 +630,7 @@ def _compute_diff_fingerprint(cwd: Path) -> str | None:
                 capture_output=True,
                 text=True,
             ).stdout.strip()
+
             # If the main working tree contains only ignorable runtime artifacts, we want to avoid
             # failing merges due to "untracked file would be overwritten" edge cases (common with
             # Claude CLI artifacts like claude-progress.txt). We still update the working tree when
@@ -749,7 +723,7 @@ def _cleanup_safe_untracked_artifacts(lines: list[str]) -> None:
                 "review": review,
                 "merge_commit": merge_commit_hash,
                 "diff_fingerprint": diff_fingerprint,
-                "timestamp": datetime.now().isoformat()
+                "timestamp": datetime.now().isoformat(),
             }
             _write_artifact(result)
             return result
@@ -763,7 +737,7 @@ def _cleanup_safe_untracked_artifacts(lines: list[str]) -> None:
                         ["git", "worktree", "remove", "-f", str(temp_worktree_path)],
                         cwd=self.project_dir,
                         capture_output=True,
-                        timeout=30
+                        timeout=30,
                     )
                     logger.info(f"✓ Cleaned up temporary worktree: {temp_worktree_path}")
 
@@ -776,7 +750,7 @@ def _cleanup_safe_untracked_artifacts(lines: list[str]) -> None:
                     # Force remove directory as fallback
                     try:
                         shutil.rmtree(str(temp_worktree_path), ignore_errors=True)
-                    except:
+                    except Exception:
                         pass
 
             # Cleanup agent's worktree if provided
@@ -806,7 +780,7 @@ def _cleanup_safe_untracked_artifacts(lines: list[str]) -> None:
                         capture_output=True,
                     )
 
-    def _run_tests(self, timeout: int = 300) -> Dict[str, Any]:
+    def _run_tests(self, timeout: int = 300) -> dict[str, Any]:
         """
         Run the test suite using the detected framework in project directory.
 
@@ -818,7 +792,7 @@ def _run_tests(self, timeout: int = 300) -> Dict[str, Any]:
         """
         return self._run_tests_in_directory(str(self.project_dir), timeout)
 
-    def _run_tests_in_directory(self, directory: str, timeout: int = 300) -> Dict[str, Any]:
+    def _run_tests_in_directory(self, directory: str, timeout: int = 300) -> dict[str, Any]:
         """
         Run the test suite using the detected framework in specified directory.
 
@@ -834,21 +808,13 @@ def _run_tests_in_directory(self, directory: str, timeout: int = 300) -> Dict[st
             cmd = self.test_detector.get_test_command(ci_mode=True)
 
             if not cmd:
-                return {
-                    "success": False,
-                    "error": "No test framework detected"
-                }
+                return {"success": False, "error": "No test framework detected"}
 
             logger.info(f"Running: {cmd}")
 
             # Execute tests in specified directory
             result = subprocess.run(
-                cmd,
-                shell=True,
-                cwd=directory,
-                capture_output=True,
-                text=True,
-                timeout=timeout
+                cmd, shell=True, cwd=directory, capture_output=True, text=True, timeout=timeout
             )
 
             # Parse output
@@ -865,7 +831,7 @@ def _run_tests_in_directory(self, directory: str, timeout: int = 300) -> Dict[st
                 "command": cmd,
                 "output": output,
                 "errors": errors,
-                "summary": self._extract_test_summary(output, errors)
+                "summary": self._extract_test_summary(output, errors),
             }
 
         except subprocess.TimeoutExpired:
@@ -873,16 +839,13 @@ def _run_tests_in_directory(self, directory: str, timeout: int = 300) -> Dict[st
             return {
                 "success": False,
                 "error": f"Tests timed out after {timeout} seconds",
-                "timeout": True
+                "timeout": True,
             }
         except Exception as e:
             logger.error(f"Test execution failed: {e}")
-            return {
-                "success": False,
-                "error": str(e)
-            }
+            return {"success": False, "error": str(e)}
 
-    def _extract_test_summary(self, stdout: str, stderr: str) -> Dict[str, Any]:
+    def _extract_test_summary(self, stdout: str, stderr: str) -> dict[str, Any]:
         """
         Extract test summary from output.
 
@@ -892,13 +855,7 @@ def _extract_test_summary(self, stdout: str, stderr: str) -> Dict[str, Any]:
 
         combined = stdout + stderr
 
-        summary = {
-            "total": None,
-            "passed": None,
-            "failed": None,
-            "skipped": None,
-            "duration": None
-        }
+        summary = {"total": None, "passed": None, "failed": None, "skipped": None, "duration": None}
 
         # Jest/Vitest pattern
         jest_pattern = r"(\d+)\s+passed,\s*(\d+)\s+failed"
@@ -925,12 +882,7 @@ def _extract_test_summary(self, stdout: str, stderr: str) -> Dict[str, Any]:
 
         return summary
 
-    def reject_feature(
-        self,
-        branch_name: str,
-        reason: str,
-        errors: str
-    ) -> Dict[str, Any]:
+    def reject_feature(self, branch_name: str, reason: str, errors: str) -> dict[str, Any]:
         """
         Reject a feature branch and clean up.
 
@@ -947,11 +899,7 @@ def reject_feature(
 
         # Abort any pending merge
         try:
-            subprocess.run(
-                ["git", "merge", "--abort"],
-                cwd=self.project_dir,
-                capture_output=True
-            )
+            subprocess.run(["git", "merge", "--abort"], cwd=self.project_dir, capture_output=True)
             logger.info("✓ Aborted pending merge")
         except subprocess.CalledProcessError:
             pass  # No merge in progress
@@ -962,7 +910,7 @@ def reject_feature(
                 ["git", "reset", "--hard", "origin/main"],
                 cwd=self.project_dir,
                 check=True,
-                capture_output=True
+                capture_output=True,
             )
             logger.info("✓ Reset to origin/main")
         except subprocess.CalledProcessError as e:
@@ -972,7 +920,7 @@ def reject_feature(
             "approved": False,
             "reason": reason,
             "errors": errors,
-            "timestamp": datetime.now().isoformat()
+            "timestamp": datetime.now().isoformat(),
         }
 
     def verify_commands_only(
@@ -982,7 +930,7 @@ def verify_commands_only(
         allow_no_tests: bool = False,
         feature_id: int | None = None,
         agent_id: str | None = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Run deterministic verification commands without merging.
 
@@ -990,10 +938,16 @@ def verify_commands_only(
         before doing a full Gatekeeper temp-worktree merge verification.
         """
 
-        def _write_artifact(result: Dict[str, Any]) -> None:
+        def _write_artifact(result: dict[str, Any]) -> None:
             try:
                 if feature_id is not None:
-                    out_dir = self.project_dir / ".autocoder" / "features" / str(int(feature_id)) / "controller"
+                    out_dir = (
+                        self.project_dir
+                        / ".autocoder"
+                        / "features"
+                        / str(int(feature_id))
+                        / "controller"
+                    )
                 else:
                     out_dir = self.project_dir / ".autocoder" / "controller"
                 out_dir.mkdir(parents=True, exist_ok=True)
@@ -1013,7 +967,7 @@ def _expand_placeholders(command: str, project_dir: Path) -> str:
             venv_py_s = str(venv_py).replace("\\", "/")
             return command.replace("{PY}", py).replace("{VENV_PY}", venv_py_s)
 
-        def _run_shell(command: str, *, cwd: Path, timeout_s: int | None) -> Dict[str, Any]:
+        def _run_shell(command: str, *, cwd: Path, timeout_s: int | None) -> dict[str, Any]:
             cmd = _expand_placeholders(command, cwd)
             try:
                 result = subprocess.run(
@@ -1106,7 +1060,9 @@ def _run_shell(command: str, *, cwd: Path, timeout_s: int | None) -> Dict[str, A
             verification[name] = _run_shell(spec.command, cwd=workdir, timeout_s=spec.timeout_s)
             verification[name]["allow_fail"] = bool(getattr(spec, "allow_fail", False))
             if name == "test":
-                verification[name] = Gatekeeper._apply_allow_no_tests(verification[name], allow_no_tests=allow_no_tests)
+                verification[name] = Gatekeeper._apply_allow_no_tests(
+                    verification[name], allow_no_tests=allow_no_tests
+                )
             if not verification[name]["success"] and not verification[name]["allow_fail"]:
                 result = {
                     "approved": False,
@@ -1143,17 +1099,10 @@ def main():
     """CLI interface for the Gatekeeper."""
     import argparse
 
-    parser = argparse.ArgumentParser(
-        description="Gatekeeper - Verify and merge feature branches"
-    )
-    parser.add_argument(
-        "branch",
-        help="Feature branch name (e.g., feat/user-auth-001)"
-    )
+    parser = argparse.ArgumentParser(description="Gatekeeper - Verify and merge feature branches")
+    parser.add_argument("branch", help="Feature branch name (e.g., feat/user-auth-001)")
     parser.add_argument(
-        "--project-dir",
-        default=".",
-        help="Path to project directory (default: current directory)"
+        "--project-dir", default=".", help="Path to project directory (default: current directory)"
     )
 
     args = parser.parse_args()
@@ -1186,4 +1135,5 @@ def main():
 
 if __name__ == "__main__":
     import sys
+
     sys.exit(main())
diff --git a/src/autocoder/core/git_bootstrap.py b/src/autocoder/core/git_bootstrap.py
index 00a24a41..5e6e4a3d 100644
--- a/src/autocoder/core/git_bootstrap.py
+++ b/src/autocoder/core/git_bootstrap.py
@@ -16,7 +16,6 @@
 import subprocess
 from pathlib import Path
 
-
 _DEFAULT_GITIGNORE_LINES = [
     "",
     "# Common local / build artifacts",
@@ -49,6 +48,8 @@
     "# AutoCoder runtime artifacts",
     ".autocoder/",
     "worktrees/",
+    ".playwright-mcp/",
+    "*.pid",
     ".agent.lock",
     ".progress_cache",
     "agent_system.db",
@@ -140,7 +141,10 @@ def ensure_git_repo_for_parallel(project_dir: Path) -> tuple[bool, str]:
         if proc.returncode != 0:
             proc = _run_git(["git", "init"], cwd=project_dir)
         if proc.returncode != 0:
-            return False, f"git init failed: {(proc.stderr or proc.stdout).strip() or 'unknown error'}"
+            return (
+                False,
+                f"git init failed: {(proc.stderr or proc.stdout).strip() or 'unknown error'}",
+            )
 
     # If HEAD exists already, we are good (don't touch user repo further).
     if _git_has_head(project_dir):
@@ -162,9 +166,14 @@ def ensure_git_repo_for_parallel(project_dir: Path) -> tuple[bool, str]:
         joined = (commit.stderr or commit.stdout or "").strip()
         # If there is nothing to commit, allow an empty initial commit so HEAD exists.
         if "nothing to commit" in joined.lower():
-            commit2 = _run_git(["git", "commit", "--no-gpg-sign", "--allow-empty", "-m", "init"], cwd=project_dir)
+            commit2 = _run_git(
+                ["git", "commit", "--no-gpg-sign", "--allow-empty", "-m", "init"], cwd=project_dir
+            )
             if commit2.returncode != 0:
-                return False, f"git commit failed: {(commit2.stderr or commit2.stdout).strip() or 'unknown error'}"
+                return (
+                    False,
+                    f"git commit failed: {(commit2.stderr or commit2.stdout).strip() or 'unknown error'}",
+                )
         else:
             return False, f"git commit failed: {joined or 'unknown error'}"
 
diff --git a/src/autocoder/core/git_dirty.py b/src/autocoder/core/git_dirty.py
new file mode 100644
index 00000000..f979fa0f
--- /dev/null
+++ b/src/autocoder/core/git_dirty.py
@@ -0,0 +1,126 @@
+"""
+Git Dirty Detection
+===================
+
+Shared helper for determining whether a project git working tree is "dirty" in a way that should
+block deterministic merges (Gatekeeper) or parallel-mode worktree orchestration.
+
+We intentionally ignore known runtime artifacts that AutoCoder / Playwright can create, to avoid
+blocking merges on harmless files.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+
+def _run_git(argv: list[str], *, cwd: Path) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        argv,
+        cwd=str(cwd),
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        check=False,
+    )
+
+
+def git_status_porcelain(project_dir: Path) -> list[str]:
+    raw = _run_git(["git", "status", "--porcelain"], cwd=Path(project_dir)).stdout
+    return [ln for ln in (raw or "").splitlines() if ln.strip()]
+
+
+def split_dirty(lines: list[str], *, project_dir: Path) -> tuple[list[str], list[str]]:
+    """
+    Split `git status --porcelain` lines into:
+    - ignored (runtime/artifacts we don't want to block on)
+    - remaining (real changes that should block deterministic merges)
+    """
+    project_dir = Path(project_dir).resolve()
+
+    ignore_any_status_substrings = [
+        ".autocoder/",
+        "worktrees/",
+        "agent_system.db",
+        "assistant.db",
+        ".progress_cache",
+        ".eslintrc.json",
+    ]
+    ignore_untracked_substrings = [
+        # Playwright MCP verification artifacts / screenshots
+        ".playwright-mcp/",
+    ]
+    ignore_untracked_filenames = {
+        # Claude Code CLI can leave these behind in the target project root.
+        ".claude_settings.json",
+        "claude-progress.txt",
+    }
+    ignore_untracked_globs = [
+        "*.pid",
+    ]
+
+    ignored: list[str] = []
+    remaining: list[str] = []
+    for ln in lines:
+        target = ln.replace("\\", "/")
+        status = ln[:2]
+        path_part = ln[3:] if len(ln) > 3 else ""
+        # Handle renames like: "R  old -> new"
+        if "->" in path_part:
+            path_part = path_part.split("->", 1)[-1].strip()
+        rel = path_part.replace("\\", "/")
+        filename = rel.split("/")[-1] if rel else ""
+
+        if any(s in target for s in ignore_any_status_substrings):
+            ignored.append(ln)
+            continue
+
+        if status == "??":
+            if any(s in rel for s in ignore_untracked_substrings):
+                ignored.append(ln)
+                continue
+            if filename in ignore_untracked_filenames:
+                ignored.append(ln)
+                continue
+            if any(fnmatch.fnmatch(filename, pat) for pat in ignore_untracked_globs):
+                ignored.append(ln)
+                continue
+
+            # Claude CLI sometimes drops a redundant root-level app_spec.txt even when prompts/app_spec.txt exists.
+            if filename == "app_spec.txt" and (project_dir / "prompts" / "app_spec.txt").exists():
+                ignored.append(ln)
+                continue
+
+            # AutoCoder prompt scaffolding files are often left untracked in the target project.
+            if rel == "prompts/" or rel == "prompts":
+                ignored.append(ln)
+                continue
+            if rel.startswith("prompts/"):
+                rel_name = rel.split("/")[-1] if rel else ""
+                if rel_name == "app_spec.txt" or rel_name.endswith("_prompt.txt"):
+                    ignored.append(ln)
+                    continue
+
+        remaining.append(ln)
+
+    return ignored, remaining
+
+
+@dataclass(frozen=True)
+class GitDirtyStatus:
+    ignored: list[str]
+    remaining: list[str]
+
+    @property
+    def is_clean(self) -> bool:
+        return not self.remaining
+
+
+def get_git_dirty_status(project_dir: Path) -> GitDirtyStatus:
+    lines = git_status_porcelain(project_dir)
+    ignored, remaining = split_dirty(lines, project_dir=project_dir)
+    return GitDirtyStatus(ignored=ignored, remaining=remaining)
diff --git a/src/autocoder/core/orchestrator.py b/src/autocoder/core/orchestrator.py
index 36644aef..a4b940c3 100644
--- a/src/autocoder/core/orchestrator.py
+++ b/src/autocoder/core/orchestrator.py
@@ -19,43 +19,47 @@
 """
 
 import asyncio
-import os
-import sys
+import contextlib
+import errno
+import json
 import logging
-import subprocess
+import os
 import shutil
-import json
-import psutil
-import threading
 import socket
-import contextlib
-import errno
-from pathlib import Path
-from typing import Dict, Any, List, Optional, Tuple, Set
+import subprocess
+import sys
+import threading
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
 
-# Direct imports (system code = fast!)
-from .knowledge_base import KnowledgeBase, get_knowledge_base
-from .model_settings import ModelSettings, ModelPreset, get_full_model_id
-from .worktree_manager import WorktreeManager
-from .database import Database, get_database
-from .gatekeeper import Gatekeeper
-from .project_config import load_project_config
-from .engine_settings import load_engine_settings, EngineSettings
-from .logs import prune_worker_logs_from_env, prune_gatekeeper_artifacts_from_env
+import psutil
+
+# Agent imports (for initializer)
+from autocoder.agent import run_autonomous_agent
+from autocoder.agent.prompts import get_app_spec
+from autocoder.generation.feature_backlog import (
+    build_backlog_prompt,
+    infer_feature_count,
+    parse_feature_backlog,
+)
 from autocoder.generation.multi_model import (
     MultiModelGenerateConfig,
     generate_multi_model_artifact,
     generate_multi_model_text,
 )
-from autocoder.generation.feature_backlog import build_backlog_prompt, parse_feature_backlog, infer_feature_count
-
-# Agent imports (for initializer)
-from autocoder.agent import run_autonomous_agent
-from autocoder.agent.prompts import get_app_spec
 
 # MCP server imports (for agents)
-from autocoder.tools import test_mcp, knowledge_mcp, model_settings_mcp, feature_mcp
+from .database import get_database
+from .engine_settings import EngineSettings, load_engine_settings
+from .gatekeeper import Gatekeeper
+
+# Direct imports (system code = fast!)
+from .knowledge_base import get_knowledge_base
+from .logs import prune_gatekeeper_artifacts_from_env, prune_worker_logs_from_env
+from .model_settings import ModelSettings
+from .project_config import load_project_config
+from .worktree_manager import WorktreeManager
 
 logger = logging.getLogger(__name__)
 
@@ -63,6 +67,7 @@
 # Port Allocator - Thread-safe port pool management
 # ============================================================================
 
+
 class PortAllocator:
     """
     Thread-safe port pool allocator for parallel agents.
@@ -82,11 +87,11 @@ class PortAllocator:
     def __init__(
         self,
         *,
-        api_port_range: Optional[Tuple[int, int]] = None,
-        web_port_range: Optional[Tuple[int, int]] = None,
+        api_port_range: tuple[int, int] | None = None,
+        web_port_range: tuple[int, int] | None = None,
         bind_host_ipv4: str = "127.0.0.1",
         bind_host_ipv6: str = "::1",
-        verify_availability: Optional[bool] = None,
+        verify_availability: bool | None = None,
     ):
         """Initialize the port allocator with pools and optional bind checks."""
         self._lock = threading.Lock()
@@ -113,22 +118,22 @@ def __init__(
         self._bind_host_ipv6 = bind_host_ipv6
 
         # Track available and in-use ports
-        self._available_api_ports: Set[int] = set(
+        self._available_api_ports: set[int] = set(
             range(self.api_port_range[0], self.api_port_range[1])
         )
-        self._available_web_ports: Set[int] = set(
+        self._available_web_ports: set[int] = set(
             range(self.web_port_range[0], self.web_port_range[1])
         )
 
-        self._in_use_api_ports: Set[int] = set()
-        self._in_use_web_ports: Set[int] = set()
-        self._blocked_api_ports: Set[int] = set()
-        self._blocked_web_ports: Set[int] = set()
+        self._in_use_api_ports: set[int] = set()
+        self._in_use_web_ports: set[int] = set()
+        self._blocked_api_ports: set[int] = set()
+        self._blocked_web_ports: set[int] = set()
 
         # Track which agent owns which ports
-        self._agent_ports: Dict[str, Tuple[int, int]] = {}
+        self._agent_ports: dict[str, tuple[int, int]] = {}
 
-        logger.info(f"PortAllocator initialized:")
+        logger.info("PortAllocator initialized:")
         logger.info(
             f"  API ports: {self.api_port_range[0]}-{self.api_port_range[1]} "
             f"({len(self._available_api_ports)} available)"
@@ -140,7 +145,7 @@ def __init__(
         logger.info(f"  Port availability verification: {self._verify_availability}")
 
     @staticmethod
-    def _range_from_env(start_env: str, end_env: str, default: Tuple[int, int]) -> Tuple[int, int]:
+    def _range_from_env(start_env: str, end_env: str, default: tuple[int, int]) -> tuple[int, int]:
         start_raw = os.environ.get(start_env)
         end_raw = os.environ.get(end_env)
         if not start_raw and not end_raw:
@@ -213,7 +218,7 @@ def reserve_ports(self, agent_id: str, api_port: int, web_port: int) -> bool:
             self._agent_ports[agent_id] = (api_port, web_port)
             return True
 
-    def allocate_ports(self, agent_id: str) -> Optional[Tuple[int, int]]:
+    def allocate_ports(self, agent_id: str) -> tuple[int, int] | None:
         """
         Allocate a port pair for an agent.
 
@@ -226,15 +231,19 @@ def allocate_ports(self, agent_id: str) -> Optional[Tuple[int, int]]:
         with self._lock:
             # Check if agent already has ports allocated
             if agent_id in self._agent_ports:
-                logger.warning(f"Agent {agent_id} already has ports allocated: {self._agent_ports[agent_id]}")
+                logger.warning(
+                    f"Agent {agent_id} already has ports allocated: {self._agent_ports[agent_id]}"
+                )
                 return self._agent_ports[agent_id]
 
             # Check if we have available ports
             if not self._available_api_ports or not self._available_web_ports:
-                logger.error(f"No ports available! API: {len(self._available_api_ports)}, Web: {len(self._available_web_ports)}")
+                logger.error(
+                    f"No ports available! API: {len(self._available_api_ports)}, Web: {len(self._available_web_ports)}"
+                )
                 return None
 
-            def pick_port(available: Set[int], blocked: Set[int]) -> Optional[int]:
+            def pick_port(available: set[int], blocked: set[int]) -> int | None:
                 for candidate in sorted(available):
                     if self._port_is_available(candidate):
                         return candidate
@@ -263,7 +272,9 @@ def pick_port(available: Set[int], blocked: Set[int]) -> Optional[int]:
             self._agent_ports[agent_id] = (api_port, web_port)
 
             logger.info(f"Allocated ports for {agent_id}: API={api_port}, WEB={web_port}")
-            logger.info(f"  Available ports: API={len(self._available_api_ports)}, Web={len(self._available_web_ports)}")
+            logger.info(
+                f"  Available ports: API={len(self._available_api_ports)}, Web={len(self._available_web_ports)}"
+            )
 
             return (api_port, web_port)
 
@@ -296,11 +307,13 @@ def release_ports(self, agent_id: str) -> bool:
             del self._agent_ports[agent_id]
 
             logger.info(f"Released ports for {agent_id}: API={api_port}, WEB={web_port}")
-            logger.info(f"  Available ports: API={len(self._available_api_ports)}, Web={len(self._available_web_ports)}")
+            logger.info(
+                f"  Available ports: API={len(self._available_api_ports)}, Web={len(self._available_web_ports)}"
+            )
 
             return True
 
-    def get_agent_ports(self, agent_id: str) -> Optional[Tuple[int, int]]:
+    def get_agent_ports(self, agent_id: str) -> tuple[int, int] | None:
         """
         Get the ports allocated to an agent.
 
@@ -313,7 +326,7 @@ def get_agent_ports(self, agent_id: str) -> Optional[Tuple[int, int]]:
         with self._lock:
             return self._agent_ports.get(agent_id)
 
-    def get_status(self) -> Dict[str, Any]:
+    def get_status(self) -> dict[str, Any]:
         """Get current allocator status."""
         with self._lock:
             return {
@@ -322,21 +335,20 @@ def get_status(self) -> Dict[str, Any]:
                     "in_use": len(self._in_use_api_ports),
                     "blocked": len(self._blocked_api_ports),
                     "total": self.api_port_range[1] - self.api_port_range[0],
-                    "range": f"{self.api_port_range[0]}-{self.api_port_range[1]}"
+                    "range": f"{self.api_port_range[0]}-{self.api_port_range[1]}",
                 },
                 "web_ports": {
                     "available": len(self._available_web_ports),
                     "in_use": len(self._in_use_web_ports),
                     "blocked": len(self._blocked_web_ports),
                     "total": self.web_port_range[1] - self.web_port_range[0],
-                    "range": f"{self.web_port_range[0]}-{self.web_port_range[1]}"
+                    "range": f"{self.web_port_range[0]}-{self.web_port_range[1]}",
                 },
                 "active_allocations": len(self._agent_ports),
-                "agents": list(self._agent_ports.keys())
+                "agents": list(self._agent_ports.keys()),
             }
 
 
-
 class Orchestrator:
     """
     Orchestrator manages parallel autonomous coding agents.
@@ -345,12 +357,7 @@ class Orchestrator:
     Includes port pool allocator for managing parallel agent server ports.
     """
 
-    def __init__(
-        self,
-        project_dir: str,
-        max_agents: int = 3,
-        model_preset: str = "balanced"
-    ):
+    def __init__(self, project_dir: str, max_agents: int = 3, model_preset: str = "balanced"):
         """
         Initialize the orchestrator.
 
@@ -382,9 +389,9 @@ def __init__(
         self.port_allocator = PortAllocator()
         self._bootstrap_ports_from_database()
         port_status = self.port_allocator.get_status()
-        self._last_logs_prune_at: Optional[datetime] = None
-        self._last_dependency_check_at: Optional[datetime] = None
-        self._last_regression_spawn_at: Optional[datetime] = None
+        self._last_logs_prune_at: datetime | None = None
+        self._last_dependency_check_at: datetime | None = None
+        self._last_regression_spawn_at: datetime | None = None
         self._planner_breaker_failures: int = 0
         self._planner_breaker_open_until: datetime | None = None
 
@@ -394,17 +401,23 @@ def __init__(
             try:
                 self.model_settings.set_preset(model_preset)
             except ValueError:
-                logger.warning("Unknown model_preset=%s; falling back to persisted settings", model_preset)
+                logger.warning(
+                    "Unknown model_preset=%s; falling back to persisted settings", model_preset
+                )
         self.available_models = self.model_settings.available_models
 
-        logger.info(f"Orchestrator initialized:")
+        logger.info("Orchestrator initialized:")
         logger.info(f"  Project: {self.project_dir}")
         logger.info(f"  Max agents: {max_agents}")
         logger.info(f"  Model preset: {self.model_settings.preset}")
         logger.info(f"  Available models: {self.available_models}")
-        logger.info(f"  Port pools:")
-        logger.info(f"    API: {port_status['api_ports']['range']} ({port_status['api_ports']['available']} available)")
-        logger.info(f"    Web: {port_status['web_ports']['range']} ({port_status['web_ports']['available']} available)")
+        logger.info("  Port pools:")
+        logger.info(
+            f"    API: {port_status['api_ports']['range']} ({port_status['api_ports']['available']} available)"
+        )
+        logger.info(
+            f"    Web: {port_status['web_ports']['range']} ({port_status['web_ports']['available']} available)"
+        )
 
     @staticmethod
     def _parse_db_timestamp(value: object) -> datetime | None:
@@ -422,7 +435,7 @@ def _parse_db_timestamp(value: object) -> datetime | None:
                 return dt.replace(tzinfo=timezone.utc)
         return None
 
-    def _is_expected_worker_process(self, agent_row: Dict[str, Any]) -> bool:
+    def _is_expected_worker_process(self, agent_row: dict[str, Any]) -> bool:
         """
         Guard against PID reuse.
 
@@ -481,7 +494,7 @@ def _is_expected_worker_process(self, agent_row: Dict[str, Any]) -> bool:
 
         return True
 
-    def _salvage_dead_agent(self, agent_row: Dict[str, Any], *, reason: str) -> None:
+    def _salvage_dead_agent(self, agent_row: dict[str, Any], *, reason: str) -> None:
         """
         Handle an agent row whose PID is missing or not a valid worker process.
 
@@ -494,14 +507,20 @@ def _salvage_dead_agent(self, agent_row: Dict[str, Any], *, reason: str) -> None
         feature_id_raw = agent_row.get("feature_id")
         feature_id = int(feature_id_raw) if feature_id_raw is not None else None
 
-        feature: Dict[str, Any] | None = None
+        feature: dict[str, Any] | None = None
         if feature_id is not None:
             with contextlib.suppress(Exception):
                 feature = self.database.get_feature(feature_id)
 
         # If feature is already ready for Gatekeeper, keep it intact and just mark the agent completed.
-        if feature and feature.get("review_status") == "READY_FOR_VERIFICATION" and feature.get("branch_name"):
-            logger.warning("Dead agent %s had a feature ready for verification; salvaging", agent_id)
+        if (
+            feature
+            and feature.get("review_status") == "READY_FOR_VERIFICATION"
+            and feature.get("branch_name")
+        ):
+            logger.warning(
+                "Dead agent %s had a feature ready for verification; salvaging", agent_id
+            )
             with contextlib.suppress(Exception):
                 self.database.mark_agent_completed(agent_id)
             with contextlib.suppress(Exception):
@@ -632,7 +651,9 @@ def _bootstrap_ports_from_database(self) -> None:
                     agent_id,
                     pid,
                 )
-                self._salvage_dead_agent(agent, reason="Agent process missing on orchestrator startup")
+                self._salvage_dead_agent(
+                    agent, reason="Agent process missing on orchestrator startup"
+                )
 
         if reserved:
             logger.info(f"Bootstrapped {reserved} port allocation(s) from database")
@@ -667,7 +688,7 @@ async def _run_initializer(self) -> bool:
                     project_dir=self.project_dir,
                     model=model,
                     max_iterations=1,  # Only run initializer
-                    yolo_mode=False     # Full testing for initializer
+                    yolo_mode=False,  # Full testing for initializer
                 )
 
             # Verify features were created
@@ -681,7 +702,7 @@ async def _run_initializer(self) -> bool:
             # Apply staging if the backlog is large
             self._maybe_stage_initializer_backlog(total_features=int(total_features))
 
-            logger.info(f"   ✅ Initializer completed successfully!")
+            logger.info("   ✅ Initializer completed successfully!")
             logger.info(f"   📊 Created {total_features} features")
             return True
 
@@ -745,13 +766,17 @@ def _maybe_stage_initializer_backlog(self, *, total_features: int) -> None:
         if stage_threshold <= 0 or total_features <= stage_threshold:
             return
 
-        keep = enqueue_count if enqueue_count > 0 else min(stage_threshold, max(1, stage_threshold // 3))
+        keep = (
+            enqueue_count
+            if enqueue_count > 0
+            else min(stage_threshold, max(1, stage_threshold // 3))
+        )
         staged = self.database.stage_features_excluding_top(keep)
         logger.info(
             f"   🧊 Staged {staged} features (threshold={stage_threshold}, kept_enabled={keep})"
         )
 
-    async def run_parallel_agents(self) -> Dict[str, Any]:
+    async def run_parallel_agents(self) -> dict[str, Any]:
         """
         Run multiple agents in parallel until all features are complete.
 
@@ -763,7 +788,6 @@ async def run_parallel_agents(self) -> Dict[str, Any]:
         logger.info("🚀 Starting parallel agent execution...")
 
         start_time = datetime.now()
-        total_completed = 0
         total_failed = 0
         idle_cycles = 0
 
@@ -797,12 +821,14 @@ async def run_parallel_agents(self) -> Dict[str, Any]:
                     logger.info("📝 No features found, running initializer agent...")
                     initializer_success = await self._run_initializer()
                     if not initializer_success:
-                        logger.error("❌ Initializer failed, cannot continue with parallel execution")
+                        logger.error(
+                            "❌ Initializer failed, cannot continue with parallel execution"
+                        )
                         return {
                             "duration_seconds": 0,
                             "features_completed": 0,
                             "features_failed": 0,
-                            "error": "Initializer failed"
+                            "error": "Initializer failed",
                         }
                     # Refresh stats after initializer
                     stats = self.database.get_stats()
@@ -829,7 +855,9 @@ async def run_parallel_agents(self) -> Dict[str, Any]:
                     if self._stop_when_done():
                         logger.info("✅ All features complete!")
                         break
-                    logger.info("✅ Queue empty; waiting for new features (AUTOCODER_STOP_WHEN_DONE=0)")
+                    logger.info(
+                        "✅ Queue empty; waiting for new features (AUTOCODER_STOP_WHEN_DONE=0)"
+                    )
                     await asyncio.sleep(10)
                     continue
 
@@ -861,7 +889,11 @@ async def run_parallel_agents(self) -> Dict[str, Any]:
                         queue_state = self.database.get_pending_queue_state()
 
                     claimable_now = int((queue_state or {}).get("claimable_now") or 0)
-                    pending_total = int((queue_state or {}).get("pending_total") or stats["features"]["pending"] or 0)
+                    pending_total = int(
+                        (queue_state or {}).get("pending_total")
+                        or stats["features"]["pending"]
+                        or 0
+                    )
 
                     if claimable_now > 0:
                         idle_cycles = 0
@@ -903,7 +935,9 @@ async def run_parallel_agents(self) -> Dict[str, Any]:
                             if pending_total > 0 and waiting_backoff > 0:
                                 reason.append(f"waiting on retry window ({waiting_backoff})")
                             msg = ", ".join(reason) if reason else "no claimable features"
-                            logger.info(f"⏳ No claimable pending features; sleeping {sleep_s}s ({msg})")
+                            logger.info(
+                                f"⏳ No claimable pending features; sleeping {sleep_s}s ({msg})"
+                            )
 
                         await asyncio.sleep(sleep_s)
                         continue
@@ -929,10 +963,10 @@ async def run_parallel_agents(self) -> Dict[str, Any]:
             "duration_seconds": duration,
             "features_completed": final_stats["features"]["completed"],
             "features_failed": total_failed,
-            "stats": final_stats
+            "stats": final_stats,
         }
 
-    def _spawn_agents(self, count: int) -> List[str]:
+    def _spawn_agents(self, count: int) -> list[str]:
         """
         Spawn new agents to work on pending features.
 
@@ -973,7 +1007,9 @@ def _spawn_agents(self, count: int) -> List[str]:
             port_pair = self.port_allocator.allocate_ports(agent_id)
             if not port_pair:
                 logger.error(f"   ❌ Failed to allocate ports for {agent_id}")
-                self.database.mark_feature_failed(feature_id=feature_id, reason="No ports available")
+                self.database.mark_feature_failed(
+                    feature_id=feature_id, reason="No ports available"
+                )
                 continue
 
             api_port, web_port = port_pair
@@ -988,7 +1024,9 @@ def _spawn_agents(self, count: int) -> List[str]:
                 )
             except Exception as e:
                 logger.error(f"   ❌ Failed to create worktree for {agent_id}: {e}")
-                self.database.mark_feature_failed(feature_id=feature_id, reason="Worktree creation failed")
+                self.database.mark_feature_failed(
+                    feature_id=feature_id, reason="Worktree creation failed"
+                )
                 # Release ports on failure
                 self.port_allocator.release_ports(agent_id)
                 continue
@@ -1090,7 +1128,7 @@ def _spawn_agents(self, count: int) -> List[str]:
                 env["AUTOCODER_FEATURE_PLAN_PATH"] = str(plan_path)
 
             # Per-agent logs (for debugging and post-mortems)
-            logs_dir = (self.project_dir / ".autocoder" / "logs")
+            logs_dir = self.project_dir / ".autocoder" / "logs"
             logs_dir.mkdir(parents=True, exist_ok=True)
             log_file_path = logs_dir / f"{agent_id}.log"
 
@@ -1100,7 +1138,9 @@ def _spawn_agents(self, count: int) -> List[str]:
             logger.info(f"🚀 Launching {agent_id}:")
             logger.info(f"   Feature: #{feature_id} - {claimed_feature['name']}")
             if use_patch_chain:
-                logger.info(f"   Engines: {', '.join(worker_engines) if worker_engines else 'none'}")
+                logger.info(
+                    f"   Engines: {', '.join(worker_engines) if worker_engines else 'none'}"
+                )
             else:
                 logger.info(f"   Model: {model.upper()}")
             logger.info(f"   Worktree: {worktree_info['worktree_path']}")
@@ -1139,7 +1179,9 @@ def _spawn_agents(self, count: int) -> List[str]:
                     log_file_path=str(log_file_path),
                 )
                 with contextlib.suppress(Exception):
-                    self.database.create_branch(claimed_branch, feature_id=feature_id, agent_id=agent_id)
+                    self.database.create_branch(
+                        claimed_branch, feature_id=feature_id, agent_id=agent_id
+                    )
                 with contextlib.suppress(Exception):
                     self.database.add_activity_event(
                         event_type="agent.spawn",
@@ -1172,7 +1214,9 @@ def _spawn_agents(self, count: int) -> List[str]:
                         data={"error": str(e)},
                     )
                 # Cleanup on failure
-                self.database.mark_feature_failed(feature_id=feature_id, reason="Agent spawn failed")
+                self.database.mark_feature_failed(
+                    feature_id=feature_id, reason="Agent spawn failed"
+                )
                 self.worktree_manager.delete_worktree(agent_id, force=True)
                 self.port_allocator.release_ports(agent_id)
                 continue
@@ -1211,7 +1255,9 @@ def _ensure_feature_plan(self, *, feature: dict, worktree_path: Path) -> str | N
                 plan_error = reason
             else:
                 try:
-                    plan_path = self._generate_feature_plan(feature=feature, worktree_path=worktree_path, cfg=cfg)
+                    plan_path = self._generate_feature_plan(
+                        feature=feature, worktree_path=worktree_path, cfg=cfg
+                    )
                     self._planner_breaker_record_success()
                     return plan_path
                 except Exception as e:
@@ -1219,7 +1265,11 @@ def _ensure_feature_plan(self, *, feature: dict, worktree_path: Path) -> str | N
                     self._planner_breaker_record_failure()
         else:
             with contextlib.suppress(Exception):
-                until = self._planner_breaker_open_until.isoformat() if self._planner_breaker_open_until else ""
+                until = (
+                    self._planner_breaker_open_until.isoformat()
+                    if self._planner_breaker_open_until
+                    else ""
+                )
                 plan_error = f"planner circuit breaker open{(' until ' + until) if until else ''}"
 
         # If a plan file already exists (from a previous attempt), reuse it for required features.
@@ -1252,7 +1302,9 @@ def _ensure_feature_plan(self, *, feature: dict, worktree_path: Path) -> str | N
 
         return None
 
-    def _generate_feature_plan(self, *, feature: dict, worktree_path: Path, cfg: MultiModelGenerateConfig) -> str:
+    def _generate_feature_plan(
+        self, *, feature: dict, worktree_path: Path, cfg: MultiModelGenerateConfig
+    ) -> str:
         """
         Generate a per-feature plan artifact and return its path.
 
@@ -1281,9 +1333,7 @@ def _generate_feature_plan(self, *, feature: dict, worktree_path: Path, cfg: Mul
             f"Feature #{feature_id}: {feature.get('name')}\n"
             f"Category: {feature.get('category')}\n\n"
             f"Description:\n{feature.get('description')}\n\n"
-            f"Steps:\n"
-            + "\n".join([f"- {s}" for s in (feature.get('steps') or [])])
-            + "\n\n"
+            f"Steps:\n" + "\n".join([f"- {s}" for s in (feature.get("steps") or [])]) + "\n\n"
             "Constraints:\n"
             "- Keep it small and actionable.\n"
             "- Include the verification commands to run.\n"
@@ -1321,7 +1371,11 @@ def _planner_plan_text_is_valid(text: str) -> bool:
         lowered = stripped.lower()
         if lowered.startswith("you are synthesizing multiple drafts into one final artifact"):
             return False
-        if "drafts:" in lowered and "user request:" in lowered and "you are synthesizing multiple drafts" in lowered:
+        if (
+            "drafts:" in lowered
+            and "user request:" in lowered
+            and "you are synthesizing multiple drafts" in lowered
+        ):
             return False
 
         return True
@@ -1334,7 +1388,11 @@ def _env_truthy(name: str) -> bool:
     def _engine_settings(self) -> EngineSettings:
         with contextlib.suppress(Exception):
             return load_engine_settings(str(self.project_dir))
-        return self.engine_settings if getattr(self, "engine_settings", None) is not None else EngineSettings.defaults()
+        return (
+            self.engine_settings
+            if getattr(self, "engine_settings", None) is not None
+            else EngineSettings.defaults()
+        )
 
     def _chain_agents(self, stage: str) -> list[str]:
         chain = self._engine_settings().chain_for(stage)  # type: ignore[arg-type]
@@ -1505,7 +1563,9 @@ def _planner_build_plan_cfg(self) -> tuple[MultiModelGenerateConfig, bool, str]:
             synthesizer=effective,  # type: ignore[arg-type]
             timeout_s=self._planner_timeout_s(),
             codex_model=str(os.environ.get("AUTOCODER_CODEX_MODEL", "")).strip(),
-            codex_reasoning_effort=str(os.environ.get("AUTOCODER_CODEX_REASONING_EFFORT", "")).strip(),
+            codex_reasoning_effort=str(
+                os.environ.get("AUTOCODER_CODEX_REASONING_EFFORT", "")
+            ).strip(),
             gemini_model=str(os.environ.get("AUTOCODER_GEMINI_MODEL", "")).strip(),
             claude_model=self._planner_model(),
         )
@@ -1543,10 +1603,14 @@ def _planner_breaker_record_failure(self) -> None:
         self._planner_breaker_failures = int(self._planner_breaker_failures or 0) + 1
         if self._planner_breaker_failures >= threshold:
             cooldown = self._planner_breaker_cooldown_s()
-            self._planner_breaker_open_until = datetime.now(timezone.utc) + timedelta(seconds=cooldown)
+            self._planner_breaker_open_until = datetime.now(timezone.utc) + timedelta(
+                seconds=cooldown
+            )
             self._planner_breaker_failures = 0
 
-    def _write_fallback_feature_plan(self, *, feature: dict, worktree_path: Path, reason: str) -> str | None:
+    def _write_fallback_feature_plan(
+        self, *, feature: dict, worktree_path: Path, reason: str
+    ) -> str | None:
         feature_id = int(feature.get("id") or 0)
         if feature_id <= 0:
             return None
@@ -1574,7 +1638,9 @@ def _write_fallback_feature_plan(self, *, feature: dict, worktree_path: Path, re
                     continue
                 verify_lines.append(f"- {key}: {spec.command}")
         if not verify_lines:
-            verify_lines.append("- Run the project verification commands (tests/lint/typecheck) and ensure Gatekeeper passes.")
+            verify_lines.append(
+                "- Run the project verification commands (tests/lint/typecheck) and ensure Gatekeeper passes."
+            )
 
         text = (
             "# Feature plan (fallback)\n\n"
@@ -1669,9 +1735,15 @@ def _qa_subagent_model(self) -> str:
         if raw in self.available_models:
             return raw
         # QA is short-lived; default to a fast/strong middle tier.
-        return "sonnet" if "sonnet" in self.available_models else (self.available_models[0] if self.available_models else "opus")
+        return (
+            "sonnet"
+            if "sonnet" in self.available_models
+            else (self.available_models[0] if self.available_models else "opus")
+        )
 
-    def _spawn_qa_subagent(self, *, feature_id: int, feature_name: str, branch_name: str) -> str | None:
+    def _spawn_qa_subagent(
+        self, *, feature_id: int, feature_name: str, branch_name: str
+    ) -> str | None:
         """
         Spawn a short-lived QA fixer worker for a rejected feature branch.
 
@@ -1770,7 +1842,7 @@ def _spawn_qa_subagent(self, *, feature_id: int, feature_name: str, branch_name:
         # QA fix prompt injection (used by Claude worker; harmless for CLI fixers).
         env["AUTOCODER_QA_FIX_ENABLED"] = "1"
 
-        logs_dir = (self.project_dir / ".autocoder" / "logs")
+        logs_dir = self.project_dir / ".autocoder" / "logs"
         logs_dir.mkdir(parents=True, exist_ok=True)
         log_file_path = logs_dir / f"{qa_agent_id}.log"
 
@@ -1820,16 +1892,16 @@ def _spawn_qa_subagent(self, *, feature_id: int, feature_name: str, branch_name:
                     message=f"QA sub-agent {qa_agent_id} started for feature #{feature_id}",
                     agent_id=qa_agent_id,
                     feature_id=int(feature_id),
-                        data={
-                            "branch": str(branch_name),
-                            "qa_mode": qa_mode,
-                            "engines": qa_engines,
-                            "model": str(model) if not use_patch_chain else "",
-                            "max_iterations": int(max_iterations),
-                            "api_port": int(api_port),
-                            "web_port": int(web_port),
-                        },
-                    )
+                    data={
+                        "branch": str(branch_name),
+                        "qa_mode": qa_mode,
+                        "engines": qa_engines,
+                        "model": str(model) if not use_patch_chain else "",
+                        "max_iterations": int(max_iterations),
+                        "api_port": int(api_port),
+                        "web_port": int(web_port),
+                    },
+                )
             return qa_agent_id
         except Exception as e:
             logger.error(f"   ❌ Failed to spawn QA sub-agent process: {e}")
@@ -1840,7 +1912,12 @@ def _spawn_qa_subagent(self, *, feature_id: int, feature_name: str, branch_name:
                     message=f"Failed to spawn QA sub-agent for feature #{feature_id}",
                     agent_id=str(qa_agent_id),
                     feature_id=int(feature_id),
-                    data={"error": str(e), "branch": str(branch_name), "qa_mode": qa_mode, "engines": qa_engines},
+                    data={
+                        "error": str(e),
+                        "branch": str(branch_name),
+                        "qa_mode": qa_mode,
+                        "engines": qa_engines,
+                    },
                 )
             with contextlib.suppress(Exception):
                 self.database.requeue_feature(feature_id, preserve_branch=True)
@@ -1872,7 +1949,11 @@ def _regression_pool_model(self) -> str:
         raw = str(os.environ.get("AUTOCODER_REGRESSION_POOL_MODEL", "")).strip().lower()
         if raw in self.available_models:
             return raw
-        return "sonnet" if "sonnet" in self.available_models else (self.available_models[0] if self.available_models else "opus")
+        return (
+            "sonnet"
+            if "sonnet" in self.available_models
+            else (self.available_models[0] if self.available_models else "opus")
+        )
 
     def _regression_pool_max_iterations(self) -> int:
         raw = str(os.environ.get("AUTOCODER_REGRESSION_POOL_MAX_ITERATIONS", "")).strip()
@@ -1889,7 +1970,9 @@ def _count_active_regression_agents(self) -> int:
             return 0
         return sum(1 for a in agents if str(a.get("agent_id") or "").startswith("regression-"))
 
-    def _maybe_spawn_regression_agents(self, *, available_slots: int, completed_count: int) -> List[str]:
+    def _maybe_spawn_regression_agents(
+        self, *, available_slots: int, completed_count: int
+    ) -> list[str]:
         """
         Opportunistically spawn regression testers when there is spare capacity.
 
@@ -1922,7 +2005,7 @@ def _maybe_spawn_regression_agents(self, *, available_slots: int, completed_coun
             self._last_regression_spawn_at = now
         return spawned
 
-    def _spawn_regression_agents(self, count: int) -> List[str]:
+    def _spawn_regression_agents(self, count: int) -> list[str]:
         """
         Spawn regression tester agents (Claude+Playwright) that do NOT participate in Gatekeeper merges.
 
@@ -1992,7 +2075,7 @@ def _spawn_regression_agents(self, count: int) -> List[str]:
             env["VITE_PORT"] = str(web_port)
             env.setdefault("AUTOCODER_AGENT_ID", str(agent_id))
 
-            logs_dir = (self.project_dir / ".autocoder" / "logs")
+            logs_dir = self.project_dir / ".autocoder" / "logs"
             logs_dir.mkdir(parents=True, exist_ok=True)
             log_file_path = logs_dir / f"{agent_id}.log"
 
@@ -2075,11 +2158,8 @@ def _detect_main_branch(self) -> str:
         return "main"
 
     async def run_feature_lifecycle(
-        self,
-        feature_id: int,
-        agent_id: str,
-        model: str
-    ) -> Dict[str, Any]:
+        self, feature_id: int, agent_id: str, model: str
+    ) -> dict[str, Any]:
         """
         Run the complete lifecycle for a single feature.
 
@@ -2105,17 +2185,12 @@ async def run_feature_lifecycle(
         # Get feature details
         feature = self.database.get_feature(feature_id)
         if not feature:
-            return {
-                "success": False,
-                "error": f"Feature {feature_id} not found"
-            }
+            return {"success": False, "error": f"Feature {feature_id} not found"}
 
         # Step 1: Create worktree
         logger.info(f"📁 Creating worktree for {agent_id}...")
         worktree_info = self.worktree_manager.create_worktree(
-            agent_id=agent_id,
-            feature_id=feature_id,
-            feature_name=feature["name"]
+            agent_id=agent_id, feature_id=feature_id, feature_name=feature["name"]
         )
 
         worktree_path = worktree_info["worktree_path"]
@@ -2137,9 +2212,7 @@ async def run_feature_lifecycle(
         try:
             # Simulate agent working
             await self._simulate_agent_work(
-                worktree_path=worktree_path,
-                feature=feature,
-                model=model
+                worktree_path=worktree_path, feature=feature, model=model
             )
 
             # Step 4: Submit to Gatekeeper
@@ -2159,7 +2232,7 @@ async def run_feature_lifecycle(
                     "success": True,
                     "feature_id": feature_id,
                     "agent_id": agent_id,
-                    "verification": verification
+                    "verification": verification,
                 }
             else:
                 logger.warning(f"❌ Feature rejected: {verification['reason']}")
@@ -2173,7 +2246,7 @@ async def run_feature_lifecycle(
                     "success": False,
                     "feature_id": feature_id,
                     "agent_id": agent_id,
-                    "verification": verification
+                    "verification": verification,
                 }
 
         except Exception as e:
@@ -2182,7 +2255,7 @@ async def run_feature_lifecycle(
                 "success": False,
                 "feature_id": feature_id,
                 "agent_id": agent_id,
-                "error": str(e)
+                "error": str(e),
             }
 
         finally:
@@ -2191,7 +2264,7 @@ async def run_feature_lifecycle(
             self.worktree_manager.delete_worktree(agent_id, force=True)
             self.database.unregister_agent(agent_id)
 
-    def _select_model_for_feature(self, feature: Dict[str, Any]) -> str:
+    def _select_model_for_feature(self, feature: dict[str, Any]) -> str:
         """
         Select the best model for a feature using knowledge base.
 
@@ -2231,19 +2304,14 @@ def _select_model_for_feature(self, feature: Dict[str, Any]) -> str:
             "frontend": "opus",
             "testing": "haiku",
             "documentation": "haiku",
-            "infrastructure": "opus"
+            "infrastructure": "opus",
         }
 
         recommended = category_mapping.get(category, "sonnet")
         logger.info(f"   Category mapping recommends: {recommended}")
         return recommended
 
-    async def _simulate_agent_work(
-        self,
-        worktree_path: str,
-        feature: Dict[str, Any],
-        model: str
-    ):
+    async def _simulate_agent_work(self, worktree_path: str, feature: dict[str, Any], model: str):
         """
         Simulate agent working on a feature.
 
@@ -2265,11 +2333,10 @@ async def _simulate_agent_work(
 
         # Simulate checkpoint
         self.worktree_manager.commit_checkpoint(
-            agent_id="simulation",
-            message="Initial implementation"
+            agent_id="simulation", message="Initial implementation"
         )
 
-        logger.info(f"   ✅ Agent completed work (simulated)")
+        logger.info("   ✅ Agent completed work (simulated)")
 
     def _recover_crashed_agents(self):
         """
@@ -2339,7 +2406,9 @@ def _recover_completed_agents(self):
                     )
 
                     if needs_verification:
-                        logger.info(f"🧪 Gatekeeper verifying feature #{feature_id} ({branch_name})...")
+                        logger.info(
+                            f"🧪 Gatekeeper verifying feature #{feature_id} ({branch_name})..."
+                        )
                         with contextlib.suppress(Exception):
                             self.database.add_activity_event(
                                 event_type="gatekeeper.verify",
@@ -2405,7 +2474,9 @@ def _recover_completed_agents(self):
                                         feature_id=int(feature_id),
                                         data={
                                             "reason": str(pre.get("reason") or "").strip(),
-                                            "artifact_path": str(pre.get("artifact_path") or "").strip(),
+                                            "artifact_path": str(
+                                                pre.get("artifact_path") or ""
+                                            ).strip(),
                                             "excerpt": excerpt[:2000],
                                         },
                                     )
@@ -2447,7 +2518,9 @@ def _recover_completed_agents(self):
                                     feature_id=int(feature_id),
                                     data={
                                         "branch": str(branch_name),
-                                        "merge_commit": str(verification.get("merge_commit") or "").strip(),
+                                        "merge_commit": str(
+                                            verification.get("merge_commit") or ""
+                                        ).strip(),
                                     },
                                 )
                             self.database.mark_feature_passing(feature_id)
@@ -2457,7 +2530,9 @@ def _recover_completed_agents(self):
                                     self.database.mark_branch_merged(branch_name, merge_commit)
                         else:
                             reason = verification.get("reason") or "Gatekeeper rejected feature"
-                            logger.warning(f"❌ Gatekeeper rejected feature #{feature_id}: {reason}")
+                            logger.warning(
+                                f"❌ Gatekeeper rejected feature #{feature_id}: {reason}"
+                            )
                             excerpt = self._format_gatekeeper_failure_excerpt(verification)
                             with contextlib.suppress(Exception):
                                 self.database.add_activity_event(
@@ -2468,7 +2543,9 @@ def _recover_completed_agents(self):
                                     feature_id=int(feature_id),
                                     data={
                                         "reason": str(reason).strip(),
-                                        "artifact_path": str(verification.get("artifact_path") or "").strip(),
+                                        "artifact_path": str(
+                                            verification.get("artifact_path") or ""
+                                        ).strip(),
                                         "excerpt": excerpt[:2000],
                                     },
                                 )
@@ -2481,9 +2558,15 @@ def _recover_completed_agents(self):
                                 artifact_path=verification.get("artifact_path"),
                                 diff_fingerprint=verification.get("diff_fingerprint"),
                             )
-                    elif assigned_agent == agent_id and not passes and review_status == "READY_FOR_VERIFICATION":
+                    elif (
+                        assigned_agent == agent_id
+                        and not passes
+                        and review_status == "READY_FOR_VERIFICATION"
+                    ):
                         # READY but missing branch_name is unrecoverable; requeue.
-                        logger.warning(f"Feature #{feature_id} ready for verification but missing branch_name; requeuing")
+                        logger.warning(
+                            f"Feature #{feature_id} ready for verification but missing branch_name; requeuing"
+                        )
                         self.database.mark_feature_failed(
                             feature_id=feature_id,
                             reason="Missing branch_name for verification",
@@ -2532,7 +2615,12 @@ def _prune_worker_logs_if_needed(self) -> None:
                 logger.info(
                     f"Pruned worker logs: deleted_files={result.deleted_files}, deleted_bytes={result.deleted_bytes}"
                 )
-            if str(os.environ.get("AUTOCODER_LOGS_PRUNE_ARTIFACTS", "")).strip().lower() in {"1", "true", "yes", "on"}:
+            if str(os.environ.get("AUTOCODER_LOGS_PRUNE_ARTIFACTS", "")).strip().lower() in {
+                "1",
+                "true",
+                "yes",
+                "on",
+            }:
                 a = prune_gatekeeper_artifacts_from_env(self.project_dir)
                 if a.deleted_files:
                     logger.info(
@@ -2543,7 +2631,10 @@ def _prune_worker_logs_if_needed(self) -> None:
                 keep_rows_raw = str(os.environ.get("AUTOCODER_ACTIVITY_KEEP_ROWS", "")).strip()
                 keep_days = int(keep_days_raw) if keep_days_raw else 14
                 keep_rows = int(keep_rows_raw) if keep_rows_raw else 5000
-                deleted = int(self.database.prune_activity_events(keep_days=keep_days, keep_rows=keep_rows) or 0)
+                deleted = int(
+                    self.database.prune_activity_events(keep_days=keep_days, keep_rows=keep_rows)
+                    or 0
+                )
                 if deleted:
                     logger.info(f"Pruned activity events: deleted={deleted}")
             except Exception:
@@ -2554,18 +2645,23 @@ def _prune_worker_logs_if_needed(self) -> None:
     def _block_unresolvable_dependencies_if_needed(self) -> None:
         # Default: once per minute.
         now = datetime.now()
-        if self._last_dependency_check_at and (now - self._last_dependency_check_at) < timedelta(seconds=60):
+        if self._last_dependency_check_at and (now - self._last_dependency_check_at) < timedelta(
+            seconds=60
+        ):
             return
         self._last_dependency_check_at = now
         n = int(self.database.block_unresolvable_dependencies() or 0)
         if n > 0:
-            logger.warning(f"Dependency health check: blocked {n} feature(s) due to unresolvable dependencies")
+            logger.warning(
+                f"Dependency health check: blocked {n} feature(s) due to unresolvable dependencies"
+            )
 
     @staticmethod
     def _format_gatekeeper_failure_excerpt(verification: dict) -> str:
         """
         Build a concise, actionable error excerpt for retrying agents and UI display.
         """
+
         def _first_text(obj: object, limit: int) -> str:
             s = str(obj or "")
             s = s.replace("\r\n", "\n").strip()
@@ -2575,10 +2671,13 @@ def _first_text(obj: object, limit: int) -> str:
 
         reason = str(verification.get("reason") or "Gatekeeper rejected feature").strip()
         artifact_path = str(verification.get("artifact_path") or "").strip()
+        details = str(verification.get("details") or "").strip()
 
         lines: list[str] = [f"Gatekeeper rejected: {reason}"]
         if artifact_path:
             lines.append(f"Artifact: {artifact_path}")
+        if details:
+            lines.append("Details:\n" + _first_text(details, 1200))
 
         # Prefer deterministic verification command failures.
         ver = verification.get("verification")
@@ -2667,14 +2766,18 @@ def _handle_verification_rejection(
                         feature_id=feature_id,
                         reason=excerpt,
                         artifact_path=str(artifact_path) if artifact_path is not None else None,
-                        diff_fingerprint=str(diff_fingerprint) if diff_fingerprint is not None else None,
+                        diff_fingerprint=(
+                            str(diff_fingerprint) if diff_fingerprint is not None else None
+                        ),
                         preserve_branch=True,
                         next_status="IN_PROGRESS",
                     )
                     failure_recorded = True
                     refreshed = self.database.get_feature(int(feature_id)) or {}
                     if str(refreshed.get("status") or "").upper() == "BLOCKED":
-                        logger.info(f"Feature #{feature_id} is BLOCKED; skipping QA sub-agent spawn")
+                        logger.info(
+                            f"Feature #{feature_id} is BLOCKED; skipping QA sub-agent spawn"
+                        )
                     else:
                         qa_id = self._spawn_qa_subagent(
                             feature_id=int(feature_id),
@@ -2694,7 +2797,7 @@ def _handle_verification_rejection(
                 diff_fingerprint=str(diff_fingerprint) if diff_fingerprint is not None else None,
             )
 
-    def get_status(self) -> Dict[str, Any]:
+    def get_status(self) -> dict[str, Any]:
         """Get current orchestrator status."""
         stats = self.database.get_stats()
         progress = self.database.get_progress()
@@ -2709,7 +2812,7 @@ def get_status(self) -> Dict[str, Any]:
             "progress": progress,
             "active_agents": stats["agents"]["active"],
             "ports": port_status,
-            "timestamp": datetime.now().isoformat()
+            "timestamp": datetime.now().isoformat(),
         }
 
 
@@ -2717,10 +2820,9 @@ def get_status(self) -> Dict[str, Any]:
 # Convenience Functions
 # ============================================================================
 
+
 def create_orchestrator(
-    project_dir: str,
-    max_agents: int = 3,
-    model_preset: str = "balanced"
+    project_dir: str, max_agents: int = 3, model_preset: str = "balanced"
 ) -> Orchestrator:
     """
     Create an orchestrator instance.
@@ -2733,11 +2835,7 @@ def create_orchestrator(
     Returns:
         Orchestrator instance
     """
-    return Orchestrator(
-        project_dir=project_dir,
-        max_agents=max_agents,
-        model_preset=model_preset
-    )
+    return Orchestrator(project_dir=project_dir, max_agents=max_agents, model_preset=model_preset)
 
 
 async def main():
@@ -2748,40 +2846,27 @@ async def main():
         description="Orchestrator - Run parallel autonomous coding agents"
     )
     parser.add_argument(
-        "--project-dir",
-        default=".",
-        help="Path to project directory (default: current directory)"
+        "--project-dir", default=".", help="Path to project directory (default: current directory)"
     )
     parser.add_argument(
-        "--parallel",
-        type=int,
-        default=3,
-        help="Number of parallel agents (default: 3)"
+        "--parallel", type=int, default=3, help="Number of parallel agents (default: 3)"
     )
     parser.add_argument(
         "--preset",
         default="balanced",
         choices=["quality", "balanced", "economy", "cheap", "experimental"],
-        help="Model preset (default: balanced)"
+        help="Model preset (default: balanced)",
     )
+    parser.add_argument("--show-status", action="store_true", help="Show current status and exit")
     parser.add_argument(
-        "--show-status",
-        action="store_true",
-        help="Show current status and exit"
-    )
-    parser.add_argument(
-        "--show-ports",
-        action="store_true",
-        help="Show port allocation status and exit"
+        "--show-ports", action="store_true", help="Show port allocation status and exit"
     )
 
     args = parser.parse_args()
 
     # Create orchestrator
     orchestrator = create_orchestrator(
-        project_dir=args.project_dir,
-        max_agents=args.parallel,
-        model_preset=args.preset
+        project_dir=args.project_dir, max_agents=args.parallel, model_preset=args.preset
     )
 
     # Show port status?
@@ -2791,17 +2876,17 @@ async def main():
         print("PORT ALLOCATION STATUS")
         print("=" * 60)
         ports = status["ports"]
-        print(f"\nAPI Ports:")
+        print("\nAPI Ports:")
         print(f"  Range: {ports['api_ports']['range']}")
         print(f"  Available: {ports['api_ports']['available']}")
         print(f"  In Use: {ports['api_ports']['in_use']}")
-        print(f"\nWeb Ports:")
+        print("\nWeb Ports:")
         print(f"  Range: {ports['web_ports']['range']}")
         print(f"  Available: {ports['web_ports']['available']}")
         print(f"  In Use: {ports['web_ports']['in_use']}")
         print(f"\nActive Allocations: {ports['active_allocations']}")
         if ports["agents"]:
-            print(f"Agents with ports:")
+            print("Agents with ports:")
             for agent in ports["agents"]:
                 p = orchestrator.port_allocator.get_agent_ports(agent)
                 if not p:
@@ -2821,10 +2906,12 @@ async def main():
         print(f"Max Agents: {status['max_agents']}")
         print(f"Model Preset: {status['model_preset']}")
         print(f"Available Models: {', '.join(status['available_models'])}")
-        print(f"\nProgress: {status['progress']['passing']}/{status['progress']['total']} ({status['progress']['percentage']}%)")
+        print(
+            f"\nProgress: {status['progress']['passing']}/{status['progress']['total']} ({status['progress']['percentage']}%)"
+        )
         print(f"Active Agents: {status['active_agents']}")
         ports = status["ports"]
-        print(f"\nPort Usage:")
+        print("\nPort Usage:")
         print(f"  API: {ports['api_ports']['in_use']}/{ports['api_ports']['total']}")
         print(f"  Web: {ports['web_ports']['in_use']}/{ports['web_ports']['total']}")
         print("=" * 60)
diff --git a/src/autocoder/server/main.py b/src/autocoder/server/main.py
index 54d40e22..8cd8daa8 100644
--- a/src/autocoder/server/main.py
+++ b/src/autocoder/server/main.py
@@ -8,60 +8,62 @@
 
 import asyncio
 import mimetypes
-import shutil
 import os
+import shutil
 import sys
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-from fastapi import FastAPI, Request, WebSocket, HTTPException
+from fastapi import FastAPI, HTTPException, Request, WebSocket
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.staticfiles import StaticFiles
 from fastapi.responses import FileResponse, HTMLResponse
+from fastapi.staticfiles import StaticFiles
 
-# Fix MIME types for JavaScript files on Windows
-mimetypes.add_type("application/javascript", ".js")
-mimetypes.add_type("text/css", ".css")
-
-if sys.platform == "win32":
-    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
+from autocoder.core.cli_defaults import get_codex_cli_defaults
+from autocoder.core.port_config import get_ui_allow_remote, get_ui_cors_origins, get_ui_port
 
 from .routers import (
-    projects_router,
-    features_router,
+    activity_router,
     agent_router,
-    spec_creation_router,
+    assistant_chat_router,
+    devserver_router,
+    diagnostics_router,
+    engine_settings_router,
     expand_project_router,
+    features_router,
     filesystem_router,
-    assistant_chat_router,
-    model_settings_router,
+    generate_router,
+    git_router,
     logs_router,
-    activity_router,
+    model_settings_router,
     parallel_router,
-    settings_router,
-    engine_settings_router,
-    project_settings_router,
-    generate_router,
     project_config_router,
-    diagnostics_router,
-    worktrees_router,
-    devserver_router,
+    project_settings_router,
+    projects_router,
+    settings_router,
+    spec_creation_router,
     terminal_router,
     version_router,
+    worktrees_router,
 )
-from .websocket import project_websocket
-from .services.process_manager import cleanup_all_managers
-from .services.assistant_chat_session import cleanup_all_sessions as cleanup_assistant_sessions
-from .services.expand_chat_session import cleanup_all_expand_sessions
-from .services.dev_server_manager import cleanup_all_dev_servers
-from .services.scheduler import restore_schedules, cleanup_schedules
 from .routers.devserver import devserver_websocket
-from .services.terminal_manager import cleanup_all_terminals
 from .routers.terminal import terminal_websocket
 from .schemas import SetupStatus
-from autocoder.core.cli_defaults import get_codex_cli_defaults
-from autocoder.core.port_config import get_ui_port, get_ui_cors_origins, get_ui_allow_remote
+from .services.assistant_chat_session import cleanup_all_sessions as cleanup_assistant_sessions
+from .services.dev_server_manager import cleanup_all_dev_servers
+from .services.expand_chat_session import cleanup_all_expand_sessions
+from .services.process_manager import cleanup_all_managers
+from .services.scheduler import cleanup_schedules, restore_schedules
+from .services.terminal_manager import cleanup_all_terminals
+from .websocket import project_websocket
+
+# Configure default event loop policy early on Windows.
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
 
+# Fix MIME types for JavaScript files on Windows.
+mimetypes.add_type("application/javascript", ".js")
+mimetypes.add_type("text/css", ".css")
 
 # Paths
 ROOT_DIR = Path(__file__).resolve().parent.parent
@@ -127,6 +129,7 @@ async def lifespan(app: FastAPI):
 # Security Middleware
 # ============================================================================
 
+
 @app.middleware("http")
 async def require_localhost(request: Request, call_next):
     """Only allow requests from localhost."""
@@ -166,12 +169,14 @@ async def require_localhost(request: Request, call_next):
 app.include_router(devserver_router)
 app.include_router(terminal_router)
 app.include_router(version_router)
+app.include_router(git_router)
 
 
 # ============================================================================
 # WebSocket Endpoint
 # ============================================================================
 
+
 @app.websocket("/ws/projects/{project_name}")
 async def websocket_endpoint(websocket: WebSocket, project_name: str):
     """WebSocket endpoint for real-time project updates."""
@@ -194,6 +199,7 @@ async def terminal_ws_endpoint(websocket: WebSocket, project_name: str, terminal
 # Setup & Health Endpoints
 # ============================================================================
 
+
 @app.get("/api/health")
 async def health_check():
     """Health check endpoint."""
@@ -204,7 +210,9 @@ async def health_check():
 async def setup_status():
     """Check system setup status."""
     # Check for Claude CLI (allow override; defaults to "claude")
-    cli_command = (os.environ.get("AUTOCODER_CLI_COMMAND") or os.environ.get("CLI_COMMAND") or "claude").strip()
+    cli_command = (
+        os.environ.get("AUTOCODER_CLI_COMMAND") or os.environ.get("CLI_COMMAND") or "claude"
+    ).strip()
     claude_cli = shutil.which(cli_command) is not None
 
     # Claude Code CLI auth storage varies by version/platform:
@@ -282,7 +290,9 @@ async def serve_spa(path: str):
 
         # Fall back to index.html for SPA routing
         return FileResponse(UI_DIST_DIR / "index.html", headers=_ui_index_headers())
+
 else:
+
     @app.get("/")
     async def missing_ui_build():
         """
@@ -328,7 +338,9 @@ async def missing_ui_build():
 
 if __name__ == "__main__":
     import uvicorn
+
     from autocoder.core.port_config import get_ui_host
+
     uvicorn.run(
         "autocoder.server.main:app",
         host=get_ui_host(),
diff --git a/src/autocoder/server/routers/__init__.py b/src/autocoder/server/routers/__init__.py
index 06f37f95..1992cddb 100644
--- a/src/autocoder/server/routers/__init__.py
+++ b/src/autocoder/server/routers/__init__.py
@@ -5,27 +5,28 @@
 FastAPI routers for different API endpoints.
 """
 
-from .projects import router as projects_router
-from .features import router as features_router
+from .activity import router as activity_router
 from .agent import router as agent_router
-from .spec_creation import router as spec_creation_router
-from .filesystem import router as filesystem_router
 from .assistant_chat import router as assistant_chat_router
-from .model_settings import router as model_settings_router
-from .logs import router as logs_router
-from .activity import router as activity_router
-from .parallel import router as parallel_router
-from .settings import router as settings_router
+from .devserver import router as devserver_router
+from .diagnostics import router as diagnostics_router
 from .engine_settings import router as engine_settings_router
-from .project_settings import router as project_settings_router
+from .expand_project import router as expand_project_router
+from .features import router as features_router
+from .filesystem import router as filesystem_router
 from .generate import router as generate_router
+from .git import router as git_router
+from .logs import router as logs_router
+from .model_settings import router as model_settings_router
+from .parallel import router as parallel_router
 from .project_config import router as project_config_router
-from .diagnostics import router as diagnostics_router
-from .worktrees import router as worktrees_router
-from .expand_project import router as expand_project_router
-from .devserver import router as devserver_router
+from .project_settings import router as project_settings_router
+from .projects import router as projects_router
+from .settings import router as settings_router
+from .spec_creation import router as spec_creation_router
 from .terminal import router as terminal_router
 from .version import router as version_router
+from .worktrees import router as worktrees_router
 
 __all__ = [
     "projects_router",
@@ -49,4 +50,5 @@
     "devserver_router",
     "terminal_router",
     "version_router",
+    "git_router",
 ]
diff --git a/src/autocoder/server/routers/agent.py b/src/autocoder/server/routers/agent.py
index 9ace66ed..6e71722f 100644
--- a/src/autocoder/server/routers/agent.py
+++ b/src/autocoder/server/routers/agent.py
@@ -7,17 +7,24 @@
 """
 
 import re
+from datetime import datetime, timedelta
 from pathlib import Path
 
 from fastapi import APIRouter, HTTPException
 
-from datetime import datetime, timedelta
+from autocoder.core.git_bootstrap import ensure_git_repo_for_parallel
+from autocoder.core.git_dirty import get_git_dirty_status
+from autocoder.core.spec_validation import project_setup_status
 
-from ..schemas import AgentStatus, AgentActionResponse, AgentStartRequest, AgentScheduleRequest, AgentScheduleResponse
+from ..schemas import (
+    AgentActionResponse,
+    AgentScheduleRequest,
+    AgentScheduleResponse,
+    AgentStartRequest,
+    AgentStatus,
+)
 from ..services.process_manager import get_manager
-from ..services.scheduler import schedule_run, get_schedule, cancel_schedule
-from autocoder.core.spec_validation import project_setup_status
-from autocoder.core.git_bootstrap import ensure_git_repo_for_parallel
+from ..services.scheduler import cancel_schedule, get_schedule, schedule_run
 
 
 def _get_project_path(project_name: str) -> Path:
@@ -35,11 +42,8 @@ def _get_project_path(project_name: str) -> Path:
 
 def validate_project_name(name: str) -> str:
     """Validate and sanitize project name to prevent path traversal."""
-    if not re.match(r'^[a-zA-Z0-9_-]{1,50}$', name):
-        raise HTTPException(
-            status_code=400,
-            detail="Invalid project name"
-        )
+    if not re.match(r"^[a-zA-Z0-9_-]{1,50}$", name):
+        raise HTTPException(status_code=400, detail="Invalid project name")
     return name
 
 
@@ -49,13 +53,16 @@ def get_project_manager(project_name: str):
     project_dir = _get_project_path(project_name)
 
     if not project_dir:
-        raise HTTPException(status_code=404, detail=f"Project '{project_name}' not found in registry")
+        raise HTTPException(
+            status_code=404, detail=f"Project '{project_name}' not found in registry"
+        )
 
     if not project_dir.exists():
         raise HTTPException(status_code=404, detail=f"Project directory not found: {project_dir}")
 
     return get_manager(project_name, project_dir, ROOT_DIR)
 
+
 @router.get("/status", response_model=AgentStatus)
 async def get_agent_status(project_name: str):
     """Get the current status of the agent for a project."""
@@ -99,10 +106,24 @@ async def start_agent(
                     "Fix manually:\n"
                     "  git init\n"
                     "  git add -A\n"
-                    "  git commit -m \"init\"\n"
+                    '  git commit -m "init"\n'
                     "Or switch to Standard mode."
                 ),
             )
+        dirty = get_git_dirty_status(manager.project_dir)
+        if dirty.remaining:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "error": "git_dirty",
+                    "message": (
+                        "Git working tree is dirty. Gatekeeper cannot merge deterministically while there are "
+                        "uncommitted changes in the project root."
+                    ),
+                    "remaining": dirty.remaining,
+                    "ignored": dirty.ignored,
+                },
+            )
 
     # Parallel and YOLO modes are mutually exclusive
     if request.parallel_mode and request.yolo_mode:
@@ -210,10 +231,24 @@ async def schedule_agent(
                     "Fix manually:\n"
                     "  git init\n"
                     "  git add -A\n"
-                    "  git commit -m \"init\"\n"
+                    '  git commit -m "init"\n'
                     "Or schedule Standard mode."
                 ),
             )
+        dirty = get_git_dirty_status(manager.project_dir)
+        if dirty.remaining:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "error": "git_dirty",
+                    "message": (
+                        "Git working tree is dirty. Gatekeeper cannot merge deterministically while there are "
+                        "uncommitted changes in the project root."
+                    ),
+                    "remaining": dirty.remaining,
+                    "ignored": dirty.ignored,
+                },
+            )
     run_at = request.run_at
     if run_at.tzinfo is not None:
         run_at = run_at.astimezone().replace(tzinfo=None)
diff --git a/src/autocoder/server/routers/git.py b/src/autocoder/server/routers/git.py
new file mode 100644
index 00000000..f9688fd5
--- /dev/null
+++ b/src/autocoder/server/routers/git.py
@@ -0,0 +1,101 @@
+"""
+Git Router
+==========
+
+Endpoints to surface git working-tree state in the UI and provide safe remediation actions.
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from autocoder.core.git_dirty import get_git_dirty_status
+
+
+def _get_project_path(project_name: str) -> Path:
+    from autocoder.agent.registry import get_project_path
+
+    return get_project_path(project_name)
+
+
+router = APIRouter(prefix="/api/projects/{project_name}/git", tags=["git"])
+
+
+def validate_project_name(name: str) -> str:
+    if not re.match(r"^[a-zA-Z0-9_-]{1,50}$", name):
+        raise HTTPException(status_code=400, detail="Invalid project name")
+    return name
+
+
+def _require_project_dir(project_name: str) -> Path:
+    project_name = validate_project_name(project_name)
+    project_dir = _get_project_path(project_name)
+    if not project_dir:
+        raise HTTPException(
+            status_code=404, detail=f"Project '{project_name}' not found in registry"
+        )
+    if not project_dir.exists():
+        raise HTTPException(status_code=404, detail=f"Project directory not found: {project_dir}")
+    return Path(project_dir)
+
+
+class GitStatusResponse(BaseModel):
+    is_clean: bool
+    ignored: list[str] = Field(default_factory=list)
+    remaining: list[str] = Field(default_factory=list)
+
+
+class GitStashRequest(BaseModel):
+    include_untracked: bool = True
+    message: str = "autocoder-ui: stash before parallel run"
+
+
+class GitStashResponse(BaseModel):
+    success: bool
+    message: str
+
+
+@router.get("/status", response_model=GitStatusResponse)
+async def git_status(project_name: str) -> GitStatusResponse:
+    project_dir = _require_project_dir(project_name)
+    st = get_git_dirty_status(project_dir)
+    return GitStatusResponse(is_clean=st.is_clean, ignored=st.ignored, remaining=st.remaining)
+
+
+@router.post("/stash", response_model=GitStashResponse)
+async def git_stash(
+    project_name: str, req: GitStashRequest = GitStashRequest()
+) -> GitStashResponse:
+    project_dir = _require_project_dir(project_name)
+
+    args = ["git", "stash", "push", "-m", str(req.message)]
+    if req.include_untracked:
+        args.append("-u")
+
+    proc = subprocess.run(
+        args,
+        cwd=str(project_dir),
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        check=False,
+    )
+    out = (proc.stdout or "").strip()
+    err = (proc.stderr or "").strip()
+    if proc.returncode != 0:
+        raise HTTPException(
+            status_code=400,
+            detail=f"git stash failed: {(err or out) or 'unknown error'}",
+        )
+
+    # Common outputs:
+    # - "Saved working directory and index state WIP on main: ..."
+    # - "No local changes to save"
+    msg = out or err or "ok"
+    return GitStashResponse(success=True, message=msg)
diff --git a/src/autocoder/server/services/process_manager.py b/src/autocoder/server/services/process_manager.py
index 467ad16f..9ae8f6cf 100644
--- a/src/autocoder/server/services/process_manager.py
+++ b/src/autocoder/server/services/process_manager.py
@@ -29,18 +29,18 @@
 
 # Patterns for sensitive data that should be redacted from output
 SENSITIVE_PATTERNS = [
-    r'sk-[a-zA-Z0-9]{20,}',  # Anthropic API keys
-    r'ANTHROPIC_API_KEY=[^\s]+',
-    r'api[_-]?key[=:][^\s]+',
-    r'token[=:][^\s]+',
-    r'password[=:][^\s]+',
-    r'secret[=:][^\s]+',
-    r'ghp_[a-zA-Z0-9]{36,}',  # GitHub personal access tokens
-    r'gho_[a-zA-Z0-9]{36,}',  # GitHub OAuth tokens
-    r'ghs_[a-zA-Z0-9]{36,}',  # GitHub server tokens
-    r'ghr_[a-zA-Z0-9]{36,}',  # GitHub refresh tokens
-    r'aws[_-]?access[_-]?key[=:][^\s]+',  # AWS keys
-    r'aws[_-]?secret[=:][^\s]+',
+    r"sk-[a-zA-Z0-9]{20,}",  # Anthropic API keys
+    r"ANTHROPIC_API_KEY=[^\s]+",
+    r"api[_-]?key[=:][^\s]+",
+    r"token[=:][^\s]+",
+    r"password[=:][^\s]+",
+    r"secret[=:][^\s]+",
+    r"ghp_[a-zA-Z0-9]{36,}",  # GitHub personal access tokens
+    r"gho_[a-zA-Z0-9]{36,}",  # GitHub OAuth tokens
+    r"ghs_[a-zA-Z0-9]{36,}",  # GitHub server tokens
+    r"ghr_[a-zA-Z0-9]{36,}",  # GitHub refresh tokens
+    r"aws[_-]?access[_-]?key[=:][^\s]+",  # AWS keys
+    r"aws[_-]?secret[=:][^\s]+",
 ]
 
 # Patterns for Claude CLI / API authentication failures.
@@ -66,7 +66,7 @@
 def sanitize_output(line: str) -> str:
     """Remove sensitive information from output lines."""
     for pattern in SENSITIVE_PATTERNS:
-        line = re.sub(pattern, '[REDACTED]', line, flags=re.IGNORECASE)
+        line = re.sub(pattern, "[REDACTED]", line, flags=re.IGNORECASE)
     return line
 
 
@@ -193,7 +193,9 @@ def _get_lock_process(self) -> tuple[int, float | None, psutil.Process] | None:
             return None
 
         try:
-            pid, stored_create_time = self._parse_lock_content(self.lock_file.read_text(encoding="utf-8"))
+            pid, stored_create_time = self._parse_lock_content(
+                self.lock_file.read_text(encoding="utf-8")
+            )
         except (ValueError, OSError):
             self.lock_file.unlink(missing_ok=True)
             return None
@@ -375,7 +377,9 @@ def _create_lock(self) -> bool:
             create_time = None
 
         lock_content = (
-            f"{self.process.pid}:{create_time}" if create_time is not None else str(self.process.pid)
+            f"{self.process.pid}:{create_time}"
+            if create_time is not None
+            else str(self.process.pid)
         )
 
         try:
@@ -448,9 +452,7 @@ async def _stream_output(self) -> None:
             loop = asyncio.get_running_loop()
             while True:
                 # Use run_in_executor for blocking readline
-                line = await loop.run_in_executor(
-                    None, self.process.stdout.readline
-                )
+                line = await loop.run_in_executor(None, self.process.stdout.readline)
                 if not line:
                     break
 
@@ -494,7 +496,7 @@ async def start(
         yolo_mode: bool = False,
         parallel_mode: bool = False,
         parallel_count: int = 3,
-        model_preset: str = "balanced"
+        model_preset: str = "balanced",
     ) -> tuple[bool, str]:
         """
         Start the agent as a subprocess.
@@ -508,6 +510,8 @@ async def start(
         Returns:
             Tuple of (success, message)
         """
+        # Ensure we don't report a stale "running" state after UI/server restarts.
+        await self.healthcheck()
         if self.status in ("running", "paused"):
             return False, f"Agent is already {self.status}"
 
@@ -558,9 +562,7 @@ async def start(
             # Persisted settings override env, but only when explicitly saved.
             env = apply_advanced_settings_env(env)
             # Apply project-scoped runtime overrides (wins over global advanced defaults).
-            env = apply_project_runtime_settings_env(
-                self.project_dir, env, override_existing=True
-            )
+            env = apply_project_runtime_settings_env(self.project_dir, env, override_existing=True)
 
             # Start subprocess with piped stdout/stderr
             # Use project_dir as cwd so Claude SDK sandbox allows access to project files
diff --git a/src/autocoder/server/services/terminal_manager.py b/src/autocoder/server/services/terminal_manager.py
index 2f18b026..dabb9c1c 100644
--- a/src/autocoder/server/services/terminal_manager.py
+++ b/src/autocoder/server/services/terminal_manager.py
@@ -22,10 +22,10 @@
 import sys
 import threading
 import uuid
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Callable, Set
 
 logger = logging.getLogger(__name__)
 
@@ -136,7 +136,7 @@ def __init__(self, project_name: str, project_dir: Path):
         self.project_name = project_name
         self.project_dir = Path(project_dir).resolve()
 
-        self._pty_process: "WinPtyProcess | None" = None
+        self._pty_process: WinPtyProcess | None = None
         self._master_fd: int | None = None
         self._child_pid: int | None = None
 
@@ -144,7 +144,7 @@ def __init__(self, project_name: str, project_dir: Path):
         self._output_task: asyncio.Task | None = None
         self.last_error: str | None = None
 
-        self._output_callbacks: Set[Callable[[bytes], None]] = set()
+        self._output_callbacks: set[Callable[[bytes], None]] = set()
         self._callbacks_lock = threading.Lock()
 
     @property
@@ -203,8 +203,11 @@ async def start(self) -> bool:
                     logger.warning(msg)
                     return False
                 assert WinPtyProcess is not None
+                # WinPTY ultimately relies on CreateProcess; passing an unquoted exe path with spaces
+                # (e.g. "C:\\Program Files\\PowerShell\\7\\pwsh.exe") can fail as "C:\\Program".
+                cmd = subprocess.list2cmdline([shell])
                 self._pty_process = WinPtyProcess.spawn(
-                    [shell],
+                    cmd,
                     cwd=str(self.project_dir),
                     env=os.environ.copy(),
                 )
@@ -281,6 +284,7 @@ async def _read_output_unix(self) -> None:
         loop = asyncio.get_running_loop()
         try:
             while self._is_active and self._master_fd is not None:
+
                 def read_with_select() -> bytes:
                     if self._master_fd is None:
                         return b""
@@ -460,7 +464,9 @@ def delete_terminal(project_name: str, terminal_id: str) -> bool:
     return True
 
 
-def get_terminal_session(project_name: str, project_dir: Path, terminal_id: str | None = None) -> TerminalSession:
+def get_terminal_session(
+    project_name: str, project_dir: Path, terminal_id: str | None = None
+) -> TerminalSession:
     if terminal_id is None:
         terminals = list_terminals(project_name)
         if not terminals:
diff --git a/ui/src/App.tsx b/ui/src/App.tsx
index 597113ac..4ebe3e4f 100644
--- a/ui/src/App.tsx
+++ b/ui/src/App.tsx
@@ -921,6 +921,7 @@ function App() {
             ) : (
               <>
                 <ProgressDashboard
+                  projectName={selectedProject}
                   passing={progress.passing}
                   total={progress.total}
                   percentage={progress.percentage}
diff --git a/ui/src/components/AgentControl.tsx b/ui/src/components/AgentControl.tsx
index e749d19d..817a267c 100644
--- a/ui/src/components/AgentControl.tsx
+++ b/ui/src/components/AgentControl.tsx
@@ -2,6 +2,7 @@ import { Play, Pause, Square, Loader2, Zap, Users, AlertTriangle, X } from 'luci
 import { useMemo, useState } from 'react'
 import { useStartAgent, useStopAgent, usePauseAgent, useResumeAgent } from '../hooks/useProjects'
 import type { AgentStatus } from '../lib/types'
+import { GitDirtyModal } from './GitDirtyModal'
 
 export type RunMode = 'standard' | 'parallel'
 export type ParallelPreset = 'quality' | 'balanced' | 'economy' | 'cheap' | 'experimental' | 'custom'
@@ -48,6 +49,7 @@ export function AgentControl({
   const pauseAgent = usePauseAgent(projectName)
   const resumeAgent = useResumeAgent(projectName)
   const [lastError, setLastError] = useState<string | null>(null)
+  const [showGitDirty, setShowGitDirty] = useState(false)
 
   const isLoading = startAgent.isPending || stopAgent.isPending || pauseAgent.isPending || resumeAgent.isPending
   const startDisabled = Boolean(isLoading || setupRequired)
@@ -61,6 +63,12 @@ export function AgentControl({
     setLastError(null)
 
     const onError = (e: unknown) => {
+      const anyErr: any = e as any
+      const detail = anyErr?.detail
+      if (detail && typeof detail === 'object' && (detail as any).error === 'git_dirty') {
+        setShowGitDirty(true)
+        return
+      }
       const msg = e instanceof Error ? e.message : String(e)
       setLastError(msg)
     }
@@ -115,6 +123,17 @@ export function AgentControl({
     <div className="relative flex items-center gap-2">
       <StatusIndicator status={status} />
 
+      <GitDirtyModal
+        projectName={projectName}
+        isOpen={showGitDirty}
+        onClose={() => setShowGitDirty(false)}
+        onAfterStash={async () => {
+          // After stashing, immediately retry the start with the same settings.
+          setShowGitDirty(false)
+          handleStart()
+        }}
+      />
+
       {(status === 'running' || status === 'paused') && (
         <>
           {yoloMode && (
diff --git a/ui/src/components/GitDirtyModal.tsx b/ui/src/components/GitDirtyModal.tsx
new file mode 100644
index 00000000..93e3ecca
--- /dev/null
+++ b/ui/src/components/GitDirtyModal.tsx
@@ -0,0 +1,140 @@
+import { useEffect, useState } from 'react'
+import { AlertTriangle, RefreshCw, X } from 'lucide-react'
+import { useGitStatus, useGitStash } from '../hooks/useGit'
+
+export function GitDirtyModal({
+  projectName,
+  isOpen,
+  onClose,
+  onAfterStash,
+}: {
+  projectName: string
+  isOpen: boolean
+  onClose: () => void
+  onAfterStash?: () => void
+}) {
+  const statusQuery = useGitStatus(projectName)
+  const stash = useGitStash(projectName)
+  const [localError, setLocalError] = useState<string | null>(null)
+
+  useEffect(() => {
+    if (isOpen) {
+      setLocalError(null)
+    }
+  }, [isOpen])
+
+  if (!isOpen) return null
+
+  const remaining = statusQuery.data?.remaining ?? []
+  const ignored = statusQuery.data?.ignored ?? []
+
+  return (
+    <>
+      <div className="fixed inset-0 bg-black/30 z-50" onClick={onClose} aria-hidden="true" />
+      <div className="fixed inset-0 z-50 flex items-center justify-center p-4">
+        <div
+          className="neo-card w-full max-w-3xl max-h-[90vh] overflow-hidden border-4 border-[var(--color-neo-border)] shadow-[8px_8px_0px_rgba(0,0,0,1)] bg-[var(--color-neo-card)]"
+          onClick={(e) => e.stopPropagation()}
+        >
+          <div className="flex items-center justify-between gap-3 p-4 border-b-3 border-[var(--color-neo-border)]">
+            <div className="flex items-center gap-3">
+              <div className="p-2 bg-[var(--color-neo-danger)] border-3 border-[var(--color-neo-border)] shadow-[2px_2px_0px_rgba(0,0,0,1)]">
+                <AlertTriangle size={18} className="text-white" />
+              </div>
+              <div>
+                <div className="font-display font-bold uppercase tracking-wide">Git working tree dirty</div>
+                <div className="text-xs text-[var(--color-neo-text-secondary)]">
+                  Parallel mode uses Git worktrees; Gatekeeper refuses merges when the main tree has uncommitted changes.
+                </div>
+              </div>
+            </div>
+
+            <button className="neo-btn neo-btn-secondary text-xs" onClick={onClose} title="Close">
+              <X size={16} />
+            </button>
+          </div>
+
+          <div className="p-4 overflow-y-auto max-h-[calc(90vh-140px)] space-y-3">
+            {statusQuery.isLoading ? (
+              <div className="text-sm text-[var(--color-neo-text-secondary)]">Checking git status…</div>
+            ) : null}
+
+            {localError ? (
+              <div className="neo-card p-3 border-3 border-[var(--color-neo-danger)]">
+                <div className="text-sm text-[var(--color-neo-danger)] whitespace-pre-wrap">{localError}</div>
+              </div>
+            ) : null}
+
+            {!statusQuery.isLoading && remaining.length === 0 ? (
+              <div className="neo-card p-3 border-3 border-[var(--color-neo-border)] bg-[var(--color-neo-bg)]">
+                <div className="font-display font-bold uppercase">Looks clean</div>
+                <div className="text-sm text-[var(--color-neo-text-secondary)] mt-1">
+                  No blocking changes detected. You can close this and start the agent again.
+                </div>
+              </div>
+            ) : null}
+
+            {remaining.length > 0 ? (
+              <div className="neo-card p-3">
+                <div className="font-display font-bold uppercase text-sm">Blocking changes</div>
+                <div className="text-xs text-[var(--color-neo-text-secondary)] mt-1">
+                  Commit, restore, or stash these changes to unblock merges.
+                </div>
+                <pre className="mt-2 max-h-[38vh] overflow-auto bg-[var(--color-neo-bg)] border-3 border-[var(--color-neo-border)] p-3 text-xs font-mono whitespace-pre-wrap">
+                  {remaining.join('\n')}
+                </pre>
+              </div>
+            ) : null}
+
+            {ignored.length > 0 ? (
+              <details className="neo-card p-3">
+                <summary className="cursor-pointer font-display font-bold uppercase text-sm">
+                  Ignored artifacts ({ignored.length})
+                </summary>
+                <pre className="mt-2 max-h-[22vh] overflow-auto bg-[var(--color-neo-bg)] border-3 border-[var(--color-neo-border)] p-3 text-xs font-mono whitespace-pre-wrap">
+                  {ignored.join('\n')}
+                </pre>
+              </details>
+            ) : null}
+          </div>
+
+          <div className="p-4 border-t-3 border-[var(--color-neo-border)] flex items-center justify-between gap-2">
+            <button
+              className="neo-btn neo-btn-secondary text-sm"
+              onClick={() => statusQuery.refetch()}
+              disabled={statusQuery.isFetching}
+              title="Refresh git status"
+            >
+              <RefreshCw size={16} />
+              Refresh
+            </button>
+
+            <div className="flex items-center gap-2 justify-end">
+              <button className="neo-btn neo-btn-secondary text-sm" onClick={onClose}>
+                Close
+              </button>
+              <button
+                className="neo-btn neo-btn-primary text-sm"
+                disabled={stash.isPending || remaining.length === 0}
+                onClick={async () => {
+                  setLocalError(null)
+                  try {
+                    await stash.mutateAsync({ include_untracked: true })
+                    await statusQuery.refetch()
+                    onAfterStash?.()
+                  } catch (e: any) {
+                    setLocalError(e instanceof Error ? e.message : String(e))
+                  }
+                }}
+                title={remaining.length === 0 ? 'No blocking changes to stash' : 'Stash changes (reversible)'}
+              >
+                Stash changes
+              </button>
+            </div>
+          </div>
+        </div>
+      </div>
+    </>
+  )
+}
+
diff --git a/ui/src/components/ProgressDashboard.tsx b/ui/src/components/ProgressDashboard.tsx
index bd6cd6c2..46880ad0 100644
--- a/ui/src/components/ProgressDashboard.tsx
+++ b/ui/src/components/ProgressDashboard.tsx
@@ -1,7 +1,11 @@
 import { AlertTriangle, ChevronRight, Wifi, WifiOff } from 'lucide-react'
 import type { AgentStatus } from '../lib/types'
+import { useState } from 'react'
+import { useGitStatus } from '../hooks/useGit'
+import { GitDirtyModal } from './GitDirtyModal'
 
 interface ProgressDashboardProps {
+  projectName?: string
   passing: number
   total: number
   percentage: number
@@ -22,6 +26,7 @@ interface ProgressDashboardProps {
 }
 
 export function ProgressDashboard({
+  projectName,
   passing,
   total,
   percentage,
@@ -35,6 +40,9 @@ export function ProgressDashboard({
   onAgentBadgeClick,
 }: ProgressDashboardProps) {
   const pct = total > 0 ? Math.max(0, Math.min(100, percentage)) : 0
+  const [showGitDirty, setShowGitDirty] = useState(false)
+  const gitStatusQuery = useGitStatus(projectName || '')
+  const gitDirty = Boolean(projectName && gitStatusQuery.data && !gitStatusQuery.data.is_clean)
 
   const isIdle =
     String(agentStatus || '').toLowerCase() === 'running' &&
@@ -62,6 +70,13 @@ export function ProgressDashboard({
 
   return (
     <div className="neo-card px-4 py-3">
+      {projectName ? (
+        <GitDirtyModal
+          projectName={projectName}
+          isOpen={showGitDirty}
+          onClose={() => setShowGitDirty(false)}
+        />
+      ) : null}
       <div className="flex flex-col gap-3 lg:flex-row lg:items-center lg:justify-between">
         <div className="min-w-[260px]">
           <div className="flex flex-wrap items-baseline gap-x-3 gap-y-1">
@@ -137,6 +152,16 @@ export function ProgressDashboard({
                 </span>
               )
             ) : null}
+            {gitDirty ? (
+              <button
+                type="button"
+                className="neo-badge bg-[var(--color-neo-danger)] text-white cursor-pointer hover:opacity-95 focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-[var(--color-neo-danger)] focus-visible:ring-offset-2 focus-visible:ring-offset-[var(--color-neo-bg)]"
+                title="Git working tree has uncommitted changes; Gatekeeper cannot merge"
+                onClick={() => setShowGitDirty(true)}
+              >
+                Git dirty
+              </button>
+            ) : null}
           </div>
         ) : null}
 
diff --git a/ui/src/hooks/useGit.ts b/ui/src/hooks/useGit.ts
new file mode 100644
index 00000000..74becf93
--- /dev/null
+++ b/ui/src/hooks/useGit.ts
@@ -0,0 +1,22 @@
+import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
+import { getGitStatus, gitStash } from '../lib/api'
+
+export function useGitStatus(projectName: string) {
+  return useQuery({
+    queryKey: ['git-status', projectName],
+    queryFn: () => getGitStatus(projectName),
+    staleTime: 5_000,
+    refetchInterval: 10_000,
+    enabled: Boolean(projectName),
+  })
+}
+
+export function useGitStash(projectName: string) {
+  const qc = useQueryClient()
+  return useMutation({
+    mutationFn: (req: Parameters<typeof gitStash>[1]) => gitStash(projectName, req),
+    onSuccess: () => {
+      qc.invalidateQueries({ queryKey: ['git-status', projectName] })
+    },
+  })
+}
diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts
index 054d2df8..41f27d69 100644
--- a/ui/src/lib/api.ts
+++ b/ui/src/lib/api.ts
@@ -42,6 +42,9 @@ import type {
   DevServerActionResponse,
   TerminalInfo,
   BackendVersion,
+  GitStatusResponse,
+  GitStashRequest,
+  GitStashResponse,
 } from './types'
 
 const API_BASE = '/api'
@@ -56,8 +59,19 @@ async function fetchJSON<T>(url: string, options?: RequestInit): Promise<T> {
   })
 
   if (!response.ok) {
-    const error = await response.json().catch(() => ({ detail: 'Unknown error' }))
-    throw new Error(error.detail || `HTTP ${response.status}`)
+    const payload = await response.json().catch(() => ({ detail: 'Unknown error' }))
+    const detail = (payload as any)?.detail ?? payload
+    const message =
+      typeof detail === 'string'
+        ? detail
+        : typeof (payload as any)?.message === 'string'
+          ? (payload as any).message
+          : JSON.stringify(detail, null, 2)
+
+    const err: any = new Error(message || `HTTP ${response.status}`)
+    err.status = response.status
+    err.detail = detail
+    throw err
   }
 
   return response.json()
@@ -327,6 +341,24 @@ export async function cancelAgentSchedule(projectName: string): Promise<AgentSch
   })
 }
 
+// ============================================================================
+// Git API
+// ============================================================================
+
+export async function getGitStatus(projectName: string): Promise<GitStatusResponse> {
+  return fetchJSON(`/projects/${encodeURIComponent(projectName)}/git/status`)
+}
+
+export async function gitStash(projectName: string, req: GitStashRequest = {}): Promise<GitStashResponse> {
+  return fetchJSON(`/projects/${encodeURIComponent(projectName)}/git/stash`, {
+    method: 'POST',
+    body: JSON.stringify({
+      include_untracked: req.include_untracked ?? true,
+      message: req.message ?? 'autocoder-ui: stash before parallel run',
+    }),
+  })
+}
+
 // ============================================================================
 // Worker Logs API
 // ============================================================================
diff --git a/ui/src/lib/types.ts b/ui/src/lib/types.ts
index 6524fe6f..fd2c987a 100644
--- a/ui/src/lib/types.ts
+++ b/ui/src/lib/types.ts
@@ -277,6 +277,22 @@ export interface RetryBlockedResponse {
   group_key?: string | null
 }
 
+export interface GitStatusResponse {
+  is_clean: boolean
+  ignored: string[]
+  remaining: string[]
+}
+
+export interface GitStashRequest {
+  include_untracked?: boolean
+  message?: string
+}
+
+export interface GitStashResponse {
+  success: boolean
+  message: string
+}
+
 export interface PruneWorkerLogsRequest {
   keep_days: number
   keep_files: number