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]>
This commit is contained in:
2026-07-05 17:41:50 -04:00
committed by sysadmin
co-authored by Claude Fable 5
parent 5b9fcaefbf
commit 8ec69cdd35
6 changed files with 316 additions and 9 deletions
+119
View File
@@ -13,6 +13,7 @@ Configuration (mcp_config.json):
"env": {}
}
"""
import json
import os
import re
import sys
@@ -174,6 +175,11 @@ import issue_duplicate_gate # noqa: E402
import role_session_router # 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
@@ -505,6 +511,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,
@@ -532,6 +616,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}