diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0bc82e4..5fdb218 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -503,6 +503,7 @@ import issue_lock_worktree # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 +import issue_lock_adoption # noqa: E402 import merged_cleanup_reconcile # noqa: E402 import reconciler_profile # noqa: E402 import reconciliation_workflow # noqa: E402 @@ -641,6 +642,19 @@ def _branch_entry_name(branch: dict | str) -> str: return str(branch.get("name") or branch.get("ref") or "") +def _branch_entry_commit_sha(branch: dict | str) -> str | None: + """Best-effort head SHA for a Gitea branch entry (None when absent).""" + if not isinstance(branch, dict): + return None + commit = branch.get("commit") + if isinstance(commit, dict): + sha = commit.get("id") or commit.get("sha") + if sha: + return str(sha) + sha = branch.get("commit_sha") + return str(sha) if sha else None + + 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 @@ -1146,13 +1160,28 @@ def gitea_lock_issue( branches = api_get_all(branch_url, auth) except Exception as e: raise RuntimeError(f"Could not list branches to verify issue lock: {e}") - for branch in branches: - name = _branch_entry_name(branch) - if expected_pattern in name: - raise ValueError( - f"Issue #{issue_number} already has matching branch '{name}' " - "(fail closed)" - ) + # #442: distinguish the issue's own already-pushed branch (adoptable for + # lock recovery) from a different same-issue branch (competing work, still + # fail-closed per #400). Open-PR and active-lease checks already ran above, + # so reaching here means no open PR and no competing live lock. + existing_branch_entries = [ + { + "name": _branch_entry_name(branch), + "commit_sha": _branch_entry_commit_sha(branch), + } + for branch in branches + ] + adoption = issue_lock_adoption.assess_own_branch_adoption( + issue_number=issue_number, + requested_branch=branch_name, + existing_branches=existing_branch_entries, + ) + if adoption["block"]: + competing = ", ".join(adoption["competing_branches"]) + raise ValueError( + f"Issue #{issue_number} already has matching branch '{competing}' " + "that is not the requested branch (fail closed)" + ) work_lease = _build_author_issue_work_lease( issue_number=issue_number, @@ -1190,6 +1219,21 @@ def gitea_lock_issue( "worktree_path": resolved_worktree, "work_lease": work_lease, } + if adoption["adopt"]: + # #442: recovery lock over the issue's own already-pushed branch. + result["adoption"] = issue_lock_adoption.build_adoption_proof( + issue_number=issue_number, + branch_name=branch_name, + assessment=adoption, + open_pr_checked=True, + competing_lock_checked=True, + lock_file_path=ISSUE_LOCK_FILE, + lock_file_status="written", + ) + result["message"] = ( + f"Adopted existing branch '{branch_name}' and locked issue " + f"#{issue_number} for recovery (fail-closed check complete)." + ) if agent_artifacts: result["warnings"] = [ "Agent temp artifacts at repo root (delete before implementation): " diff --git a/issue_lock_adoption.py b/issue_lock_adoption.py new file mode 100644 index 0000000..86c1ef9 --- /dev/null +++ b/issue_lock_adoption.py @@ -0,0 +1,142 @@ +"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442). + +When an issue's own already-pushed branch exists, lock reacquisition must be +allowed (adoption) instead of being treated as #400 duplicate competing work. +This module isolates the pure decision so it can be unit-tested apart from the +MCP server's live Gitea calls. + +Adoption is granted only for the issue's *exact* requested branch. Any other +branch that merely contains the same ``issue-`` marker is competing work and +stays fail-closed. Open-PR, competing-live-lock, capability, and worktree +safety checks are enforced by the caller before this decision is consulted; +this module additionally records whether they passed for proof purposes. +""" + +from __future__ import annotations + +ADOPT = "adopt_existing_branch" +BLOCK_COMPETING = "block_competing_branch" +NO_MATCH = "no_matching_branch" + + +def _branch_name(entry) -> str: + if isinstance(entry, dict): + return str(entry.get("name") or "") + return str(entry or "") + + +def _branch_sha(entry) -> str | None: + if isinstance(entry, dict): + sha = entry.get("commit_sha") + if sha: + return str(sha) + return None + + +def assess_own_branch_adoption( + *, + issue_number: int, + requested_branch: str, + existing_branches, +) -> dict: + """Decide whether an existing matching branch is adoptable. + + Args: + issue_number: The tracking issue number being locked. + requested_branch: The exact branch the caller wants to lock. + existing_branches: Iterable of remote branch entries — either names or + dicts with ``name`` and optional ``commit_sha``. + + Returns: + dict with: + * ``outcome`` — one of ADOPT / BLOCK_COMPETING / NO_MATCH + * ``adopt`` (bool), ``block`` (bool) + * ``reason`` (str) + * ``matched_branch`` (str | None), ``matched_head_sha`` (str | None) + * ``competing_branches`` (list[str]) + + ADOPT: the issue's exact branch exists and no other same-issue branch does. + BLOCK_COMPETING: at least one same-issue branch is not the requested branch. + NO_MATCH: no branch carries the issue marker — normal lock path applies. + """ + marker = f"issue-{issue_number}" + requested = (requested_branch or "").strip() + + matches: list[tuple[str, str | None]] = [] + for entry in existing_branches or []: + name = _branch_name(entry).strip() + if marker in name: + matches.append((name, _branch_sha(entry))) + + competing = sorted({name for name, _ in matches if name != requested}) + exact = [(name, sha) for name, sha in matches if name == requested] + + # Fail closed whenever any non-requested same-issue branch exists, even if + # the requested branch is also present: ownership is then ambiguous. + if competing: + return { + "outcome": BLOCK_COMPETING, + "adopt": False, + "block": True, + "reason": ( + f"issue #{issue_number} already has matching branch(es) " + f"{competing} that are not the requested branch " + f"'{requested}' (fail closed)" + ), + "matched_branch": None, + "matched_head_sha": None, + "competing_branches": competing, + } + + if exact: + name, sha = exact[0] + return { + "outcome": ADOPT, + "adopt": True, + "block": False, + "reason": ( + f"existing branch '{name}' is the exact requested branch for " + f"issue #{issue_number}; adopting it for lock recovery" + ), + "matched_branch": name, + "matched_head_sha": sha, + "competing_branches": [], + } + + return { + "outcome": NO_MATCH, + "adopt": False, + "block": False, + "reason": f"no existing branch matches issue #{issue_number}", + "matched_branch": None, + "matched_head_sha": None, + "competing_branches": [], + } + + +def build_adoption_proof( + *, + issue_number: int, + branch_name: str, + assessment: dict, + open_pr_checked: bool, + competing_lock_checked: bool, + lock_file_path: str, + lock_file_status: str, +) -> dict: + """Assemble the proof block returned by ``gitea_lock_issue`` on adoption. + + Requirement #4: adoption results must carry issue number, branch name, + branch head commit, adoption reason, no-existing-PR proof, no-competing- + live-lock proof, and lock file path/status. + """ + return { + "issue_number": issue_number, + "branch_name": branch_name, + "branch_head_commit": assessment.get("matched_head_sha"), + "adoption_reason": assessment.get("reason"), + "no_existing_pr_proof": bool(open_pr_checked), + "no_competing_live_lock_proof": bool(competing_lock_checked), + "lock_file_path": lock_file_path, + "lock_file_status": lock_file_status, + } diff --git a/tests/test_issue_lock_adoption.py b/tests/test_issue_lock_adoption.py new file mode 100644 index 0000000..8ae3747 --- /dev/null +++ b/tests/test_issue_lock_adoption.py @@ -0,0 +1,124 @@ +"""Unit tests for own-branch lock adoption decision (#442).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from issue_lock_adoption import ( # noqa: E402 + ADOPT, + BLOCK_COMPETING, + NO_MATCH, + assess_own_branch_adoption, + build_adoption_proof, +) + +REQ = "feat/issue-420-server-code-parity" + + +class TestAssessOwnBranchAdoption(unittest.TestCase): + def test_exact_own_branch_is_adopted(self): + result = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": REQ, "commit_sha": "934688a"}], + ) + self.assertEqual(result["outcome"], ADOPT) + self.assertTrue(result["adopt"]) + self.assertFalse(result["block"]) + self.assertEqual(result["matched_branch"], REQ) + self.assertEqual(result["matched_head_sha"], "934688a") + + def test_exact_own_branch_adopted_when_sha_missing(self): + result = assess_own_branch_adoption( + issue_number=420, requested_branch=REQ, existing_branches=[REQ] + ) + self.assertEqual(result["outcome"], ADOPT) + self.assertIsNone(result["matched_head_sha"]) + + def test_different_branch_same_issue_blocks(self): + result = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": "feat/issue-420-other-work"}], + ) + self.assertEqual(result["outcome"], BLOCK_COMPETING) + self.assertTrue(result["block"]) + self.assertFalse(result["adopt"]) + self.assertIn("feat/issue-420-other-work", result["competing_branches"]) + self.assertIn("fail closed", result["reason"]) + + def test_own_branch_plus_competing_branch_blocks(self): + # Ambiguous ownership: fail closed even though the exact branch exists. + result = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": REQ}, {"name": "feat/issue-420-rogue"}], + ) + self.assertEqual(result["outcome"], BLOCK_COMPETING) + self.assertTrue(result["block"]) + self.assertEqual(result["competing_branches"], ["feat/issue-420-rogue"]) + + def test_no_matching_branch_is_normal_path(self): + result = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": "feat/issue-999-unrelated"}], + ) + self.assertEqual(result["outcome"], NO_MATCH) + self.assertFalse(result["block"]) + self.assertFalse(result["adopt"]) + + def test_empty_branch_list_is_normal_path(self): + result = assess_own_branch_adoption( + issue_number=420, requested_branch=REQ, existing_branches=[] + ) + self.assertEqual(result["outcome"], NO_MATCH) + + def test_substring_issue_number_not_falsely_matched_as_exact(self): + # A different issue's branch must not be adopted as ours. + result = assess_own_branch_adoption( + issue_number=42, + requested_branch="feat/issue-42-thing", + existing_branches=[{"name": "feat/issue-420-server-code-parity"}], + ) + # 'issue-42' is a substring of 'issue-420', so marker matches but the + # name differs -> competing block, never adoption of the wrong branch. + self.assertEqual(result["outcome"], BLOCK_COMPETING) + self.assertNotEqual(result["matched_branch"], "feat/issue-420-server-code-parity") + + +class TestBuildAdoptionProof(unittest.TestCase): + def test_proof_has_all_required_fields(self): + assessment = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": REQ, "commit_sha": "934688a"}], + ) + proof = build_adoption_proof( + issue_number=420, + branch_name=REQ, + assessment=assessment, + open_pr_checked=True, + competing_lock_checked=True, + lock_file_path="/tmp/gitea_issue_lock.json", + lock_file_status="written", + ) + for key in ( + "issue_number", + "branch_name", + "branch_head_commit", + "adoption_reason", + "no_existing_pr_proof", + "no_competing_live_lock_proof", + "lock_file_path", + "lock_file_status", + ): + self.assertIn(key, proof) + self.assertEqual(proof["branch_head_commit"], "934688a") + self.assertTrue(proof["no_existing_pr_proof"]) + self.assertTrue(proof["no_competing_live_lock_proof"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 4002b5e..eb24802 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3127,6 +3127,105 @@ class TestIssueLocking(unittest.TestCase): gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertIn("already has matching branch", str(ctx.exception)) + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=_clean_master_git_state_for_lock(), + ) + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_adopts_own_existing_branch(self, _auth, mock_api, _git_state): + # #442: the issue's own already-pushed branch is adoptable for recovery. + sha = "934688a71dd19d9b46369e73fb3ac356ee0cc211" + + def _api(url, *a, **k): + if "/pulls" in url: + return [] + if "/branches" in url: + return [{"name": "feat/issue-196-mutations", "commit": {"id": sha}}] + return [] + + mock_api.side_effect = _api + res = gitea_lock_issue( + issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs" + ) + self.assertTrue(res["success"]) + self.assertIn("adoption", res) + proof = res["adoption"] + self.assertEqual(proof["issue_number"], 196) + self.assertEqual(proof["branch_name"], "feat/issue-196-mutations") + self.assertEqual(proof["branch_head_commit"], sha) + self.assertTrue(proof["no_existing_pr_proof"]) + self.assertTrue(proof["no_competing_live_lock_proof"]) + self.assertEqual(proof["lock_file_path"], ISSUE_LOCK_FILE) + self.assertIn("adopt", proof["adoption_reason"].lower()) + self.assertTrue(os.path.exists(ISSUE_LOCK_FILE)) + + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=_clean_master_git_state_for_lock(), + ) + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_open_pr_blocks_adoption(self, _auth, mock_api, _git_state): + # Even with the exact own branch present, an open PR blocks adoption. + def _api(url, *a, **k): + if "/pulls" in url: + return [{ + "number": 200, + "head": {"ref": "feat/issue-196-mutations"}, + "title": "WIP", + "body": "", + }] + if "/branches" in url: + return [{"name": "feat/issue-196-mutations", "commit": {"id": "abc"}}] + return [] + + mock_api.side_effect = _api + 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)) + self.assertFalse(os.path.exists(ISSUE_LOCK_FILE)) + + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=_clean_master_git_state_for_lock(), + ) + @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_proceeds_after_adopted_lock( + self, _auth, mock_api, mock_req, _git_state, _role + ): + # #442 recovery goal: after adopting the own branch, create_pr proceeds. + sha = "934688a71dd19d9b46369e73fb3ac356ee0cc211" + + def _api(url, *a, **k): + if "/pulls" in url: + return [] + if "/branches" in url: + return [{"name": "feat/issue-196-mutations", "commit": {"id": sha}}] + return [] + + mock_api.side_effect = _api + lock_res = gitea_lock_issue( + issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs" + ) + self.assertIn("adoption", lock_res) + + mock_req.return_value = {"number": 7, "html_url": "https://gitea.prgs.cc/pulls/7"} + with patch.dict(os.environ, CREATE_PR_ENV, clear=True): + pr = gitea_create_pr( + title="feat: parity gate Closes #196", + head="feat/issue-196-mutations", + base="master", + remote="prgs", + ) + self.assertEqual(pr["number"], 7) + def test_lock_issue_blocks_active_same_operation_lease(self): with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: json.dump({