feat(author-workflow): enforce exact issue lock before branch/commit/push/PR (Issue #204)

Recreation of the #204 work from closed PR #205 (invalid provenance), rebuilt
cleanly on master under the prgs author identity with no PR #203 content:

- Add gitea_lock_issue MCP tool: locks exactly one issue to its branch name,
  fails closed on branch/issue-number mismatch and on issues already tied to
  an open PR (by head branch or Closes/Fixes reference).
- gitea_create_pr now requires the issue lock: head must match the locked
  branch, title/body must contain Closes/Fixes #<locked issue> exactly, and
  ambiguous references (equivalent / related / same as) are rejected.
- scripts/worktree-start refuses to create an issue-linked worktree unless the
  lock file exists and matches the requested branch.
- assess_controller_handoff rejects handoffs whose selected issue / opened PR
  fields carry multiple numbers or fuzzy equivalence wording.
- Tests: TestIssueLocking (lock + create_pr gates), handoff exact-reference
  tests, worktree-start lock coverage.

Closes #204

Co-Authored-By: Claude Fable 5 <[email protected]>

mcp_server.py

@@ -13,6 +13,7 @@ Configuration (mcp_config.json):
         "env": {}
     }
 """
+import json
 import os
 import re
 import sys
@@ -47,6 +48,11 @@ import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
 
 
+# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
+# consumed by gitea_create_pr and scripts/worktree-start.
+ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+
+
 def _reveal_endpoints() -> bool:
     """Admin/debug opt-in (#120): include endpoint URLs and token source
     names in tool output. Off by default so normal LLM-facing responses
@@ -340,6 +346,84 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
+@mcp.tool()
+def gitea_lock_issue(
+    issue_number: int,
+    branch_name: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Lock exactly one Gitea issue and its branch name to ensure durable tracking.
+
+    Args:
+        issue_number: The tracking issue number.
+        branch_name: The branch name (must match (fix|feat|docs|chore)/issue-<issue_number>-<desc>).
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override Gitea host.
+        org: Override Org.
+        repo: Override Repo.
+    """
+    # 1. Enforce branch name includes issue number
+    expected_pattern = f"issue-{issue_number}"
+    if expected_pattern not in branch_name:
+        raise ValueError(
+            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
+        )
+
+    # 2. Check if the issue already has an open PR (reuse protection)
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    try:
+        prs = api_get_all(url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+
+    for pr in prs:
+        pr_head = pr.get("head", {}).get("ref", "")
+        pr_title = pr.get("title", "")
+        pr_body = pr.get("body", "")
+
+        if expected_pattern in pr_head:
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
+            )
+
+        patterns = [
+            f"closes #{issue_number}",
+            f"fixes #{issue_number}",
+        ]
+        text_to_check = f"{pr_title} {pr_body}".lower()
+        if any(p in text_to_check for p in patterns):
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
+            )
+
+    data = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "remote": remote,
+        "org": o,
+        "repo": r,
+    }
+
+    try:
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(data, f)
+    except Exception as e:
+        raise RuntimeError(f"Could not write issue lock file: {e}")
+
+    return {
+        "success": True,
+        "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+    }
+
+
 @mcp.tool()
 def gitea_create_pr(
     title: str,
@@ -367,6 +451,41 @@ def gitea_create_pr(
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
     """
     h, o, r = _resolve(remote, host, org, repo)
+
+    # ── Issue Lock Validation (Issue #194 / #196) ──
+    if not os.path.exists(ISSUE_LOCK_FILE):
+        raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
+
+    try:
+        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
+            lock_data = json.load(f)
+    except Exception as e:
+        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
+
+    locked_issue = lock_data.get("issue_number")
+    locked_branch = lock_data.get("branch_name")
+
+    if head != locked_branch:
+        raise ValueError(
+            f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
+        )
+
+    # Check for forbidden terms anywhere in title/body
+    forbidden_terms = ["equivalent", "related", "same as"]
+    text_to_check = f"{title} {body}".lower()
+    for term in forbidden_terms:
+        if term in text_to_check:
+            raise ValueError(
+                f"PR title or body contains forbidden term '{term}' (fail closed)"
+            )
+
+    # Ensure Closes #<locked_issue> or Fixes #<locked_issue> is present exactly
+    closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE)
+    if not closes_pattern.search(text_to_check):
+        raise ValueError(
+            f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
+        )
+
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}

review_proofs.py

@@ -627,6 +627,35 @@ def assess_controller_handoff(report_text, role=None):
             "reasons": [f"handoff missing required field: {m}"
                         for m in missing],
         }
+
+    # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
+    fields_dict = {}
+    for line in section:
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            k, v = stripped.split(":", 1)
+            fields_dict[k.strip().lower()] = v.strip()
+
+    for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
+        val = fields_dict.get(alias)
+        if val:
+            numbers = re.findall(r"\d+", val)
+            has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
+            if len(numbers) != 1 or has_forbidden:
+                field_name = "Selected issue/PR"
+                for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())):
+                    if alias in aliases:
+                        field_name = name
+                        break
+                return {
+                    "verdict": "incomplete",
+                    "downgraded": True,
+                    "missing_fields": [field_name],
+                    "reasons": [
+                        f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
+                    ],
+                }
+
     return {
         "verdict": "complete",
         "downgraded": False,

scripts/worktree-start

@@ -40,6 +40,16 @@ start_ref="${2:-prgs/master}"
 
 # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
 if [[ "$allow_unlinked" -eq 0 ]]; then
+  if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
+    echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
+    exit 2
+  fi
+  locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
+  if [[ "$branch" != "$locked_branch" ]]; then
+    echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
+    exit 2
+  fi
+
   if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
      || [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
     :

tests/test_mcp_server.py

@@ -33,6 +33,7 @@ from mcp_server import (  # noqa: E402
     gitea_submit_pr_review,
     gitea_list_issue_comments,
     gitea_create_issue_comment,
+    gitea_lock_issue,
 )
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
@@ -85,10 +86,13 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_pr(self, _auth, mock_api):
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {}, clear=True):
-            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
         payload = mock_api.call_args[0][3]
@@ -97,10 +101,13 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
-            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
 
@@ -2296,3 +2303,83 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
             "gitea.issue.comment", reviewer["allowed_operations"],
             reviewer.get("forbidden_operations", []))
         self.assertTrue(ok)
+
+
+class TestIssueLocking(unittest.TestCase):
+    """Test issue locking and PR gating constraints."""
+
+    def tearDown(self):
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_success(self, _auth, mock_api):
+        mock_api.return_value = [] # no open PRs
+        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
+
+    def test_lock_issue_mismatch_branch_fails(self):
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
+        self.assertIn("must contain locked issue pattern", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
+        mock_api.return_value = [{
+            "number": 200,
+            "head": {"ref": "feat/issue-196-boundary"},
+            "title": "Some PR",
+            "body": "No closes ref"
+        }]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
+        mock_api.return_value = [{
+            "number": 200,
+            "head": {"ref": "feat/other-branch"},
+            "title": "Some PR",
+            "body": "fixes #196"
+        }]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_missing_lock_fails(self, _auth):
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("Issue lock is missing", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_branch_mismatch_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with self.assertRaises(ValueError) as ctx:
+            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
+        self.assertIn("does not match locked branch", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_forbidden_terms_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        for term in ("equivalent to #196", "related to #196", "same as #196"):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
+            self.assertIn("contains forbidden term", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_missing_closes_ref_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with self.assertRaises(ValueError) as ctx:
+            gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))

tests/test_review_proofs.py

@@ -775,6 +775,40 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("No review/merge confirmation", result["missing_fields"])
 
+    def test_author_role_rejects_equivalent_or_multiple_issues(self):
+        # 1. equivalent reference blocked
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: Issue #194 / #196 equivalent",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_eq, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Selected issue", res["missing_fields"])
+
+        # 2. multiple issues blocked
+        incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #194, #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_multi, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Selected issue", res["missing_fields"])
+
+    def test_author_role_rejects_fuzzy_pr_number(self):
+        incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: PR #203 / #204 equivalent",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_pr, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("PR number opened", res["missing_fields"])
+
     def test_inventory_role_requires_inventory_fields(self):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",

tests/test_worktrees.py

@@ -14,11 +14,39 @@ BRANCHES = REPO / "branches"
 
 
 def run(script, *args):
-    proc = subprocess.run(
+    branch = None
-        ["bash", str(SCRIPTS / script), *args],
+    for arg in args:
-        capture_output=True, text=True, cwd=str(REPO),
+        if not arg.startswith("-"):
-    )
+            branch = arg
-    return proc.returncode, proc.stdout, proc.stderr
+            break
+
+    lock_file = Path("/tmp/gitea_issue_lock.json")
+    created_lock = False
+    if script == "worktree-start" and branch:
+        import re
+        import json
+        m = re.search(r"issue-(\d+)", branch)
+        if not m:
+            m = re.search(r"pr-(\d+)", branch)
+        issue_num = int(m.group(1)) if m else 999
+        lock_file.write_text(json.dumps({
+            "issue_number": issue_num,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools"
+        }), encoding="utf-8")
+        created_lock = True
+
+    try:
+        proc = subprocess.run(
+            ["bash", str(SCRIPTS / script), *args],
+            capture_output=True, text=True, cwd=str(REPO),
+        )
+        return proc.returncode, proc.stdout, proc.stderr
+    finally:
+        if created_lock and lock_file.exists():
+            lock_file.unlink()
 
 
 class TestWorktreeStart(unittest.TestCase):