feat(mcp): bind issue lock to author scratch worktree (#249)

gitea_lock_issue and gitea_create_pr accept worktree_path so lock
preconditions validate the caller's scratch clone instead of the shared
MCP server CWD. Lock records store the validated path; PR creation fails
closed when the declared path does not match.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

issue_lock_worktree.py

@@ -0,0 +1,156 @@
+"""Issue-lock worktree validation (#249).
+
+Author issue locks must validate the caller's own scratch clone (or declared
+worktree path), not the shared MCP server working directory. A clean scratch at
+``master``/``main`` must remain lockable while an unrelated session leaves the
+shared dev worktree dirty or on a feature branch.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+from reviewer_worktree import parse_dirty_tracked_files
+
+AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
+BASE_BRANCHES = frozenset({"master", "main"})
+
+
+def resolve_author_worktree_path(
+    explicit: str | None,
+    project_root: str,
+) -> str:
+    """Resolve the author worktree path for lock/PR gates."""
+    path = (explicit or "").strip()
+    if not path:
+        path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
+    if not path:
+        path = project_root
+    return os.path.realpath(os.path.abspath(path))
+
+
+def read_worktree_git_state(worktree_path: str) -> dict:
+    """Read branch name and porcelain status from a git worktree."""
+    path = (worktree_path or "").strip()
+    if not path:
+        return {"current_branch": None, "porcelain_status": ""}
+
+    branch_res = subprocess.run(
+        ["git", "-C", path, "branch", "--show-current"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    current_branch = (branch_res.stdout or "").strip() or None
+
+    status_res = subprocess.run(
+        ["git", "-C", path, "status", "--porcelain"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return {
+        "current_branch": current_branch,
+        "porcelain_status": status_res.stdout or "",
+    }
+
+
+def assess_issue_lock_worktree(
+    *,
+    worktree_path: str,
+    current_branch: str | None,
+    porcelain_status: str,
+    base_branches: frozenset[str] | None = None,
+) -> dict:
+    """Fail closed when lock preconditions are not met on the declared worktree."""
+    bases = base_branches or BASE_BRANCHES
+    reasons: list[str] = []
+    path = (worktree_path or "").strip()
+    if not path:
+        reasons.append("worktree path not declared for issue lock; fail closed")
+        return _assessment(False, reasons, path, None, [])
+
+    branch = (current_branch or "").strip()
+    dirty_files = parse_dirty_tracked_files(porcelain_status)
+
+    if dirty_files:
+        reasons.append(
+            "tracked file edits exist before issue lock; "
+            "lock must precede implementation work"
+        )
+    if not branch:
+        reasons.append(
+            "current branch unknown (detached HEAD?); issue lock must be taken "
+            f"from base branch ({_base_list(bases)})"
+        )
+    elif branch not in bases:
+        reasons.append(
+            f"issue lock must be taken from base branch ({_base_list(bases)}), "
+            f"not '{branch}'"
+        )
+
+    proven = not reasons
+    return _assessment(proven, reasons, path, branch or None, dirty_files)
+
+
+def format_issue_lock_worktree_error(assessment: dict) -> str:
+    """Format a single fail-closed error for ``gitea_lock_issue``."""
+    reasons = list(assessment.get("reasons") or [])
+    if not reasons:
+        reasons = ["issue lock worktree validation failed"]
+    return "; ".join(reasons) + " (fail closed)"
+
+
+def verify_pr_worktree_matches_lock(
+    locked_worktree_path: str | None,
+    declared_worktree_path: str | None,
+    project_root: str,
+) -> dict:
+    """PR creation must use the same worktree the lock was validated against."""
+    locked = (locked_worktree_path or "").strip()
+    if not locked:
+        return {"proven": True, "block": False, "reasons": []}
+
+    declared = resolve_author_worktree_path(declared_worktree_path, project_root)
+    locked_real = os.path.realpath(locked)
+    declared_real = os.path.realpath(declared)
+    if locked_real != declared_real:
+        return {
+            "proven": False,
+            "block": True,
+            "reasons": [
+                f"PR worktree '{declared_real}' does not match locked worktree "
+                f"'{locked_real}' (fail closed)"
+            ],
+            "locked_worktree_path": locked_real,
+            "declared_worktree_path": declared_real,
+        }
+    return {
+        "proven": True,
+        "block": False,
+        "reasons": [],
+        "locked_worktree_path": locked_real,
+        "declared_worktree_path": declared_real,
+    }
+
+
+def _base_list(bases: frozenset[str]) -> str:
+    return "/".join(sorted(bases))
+
+
+def _assessment(
+    proven: bool,
+    reasons: list[str],
+    worktree_path: str,
+    current_branch: str | None,
+    dirty_files: list[str],
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "worktree_path": worktree_path or None,
+        "current_branch": current_branch,
+        "dirty_files": dirty_files,
+    }