From 749e480baf26f9e80ca57ed3abcf35041e80eab2 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 09:25:55 -0400 Subject: [PATCH] feat: add reconciliation workflow MCP tools for already-landed PRs (Closes #301) Expose read-only assess/scan tools and capability planning so already-landed open PRs can be reconciled without review or merge. Register the gitea-reconcile-landed-pr skill and workflow-source verification. --- gitea_mcp_server.py | 195 +++++++++++++++++ reconciliation_workflow.py | 292 +++++++++++++++++++++++++ review_proofs.py | 5 + role_session_router.py | 4 + task_capability_map.py | 8 + tests/test_llm_workflow_split.py | 10 + tests/test_reconciliation_mcp_tools.py | 106 +++++++++ tests/test_reconciliation_workflow.py | 149 +++++++++++++ 8 files changed, 769 insertions(+) create mode 100644 reconciliation_workflow.py create mode 100644 tests/test_reconciliation_mcp_tools.py create mode 100644 tests/test_reconciliation_workflow.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b5415eb..abd5ab7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -477,6 +477,7 @@ import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import merged_cleanup_reconcile # noqa: E402 +import reconciliation_workflow # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -3425,6 +3426,178 @@ def gitea_reconcile_merged_cleanups( return {"success": True, "performed": True, **report} +@mcp.tool() +def gitea_assess_already_landed_reconciliation( + pr_number: int, + target_branch: str = "master", + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Read-only: assess already-landed reconciliation for one open PR (#301). + + Fetches the PR live, refreshes the target branch, checks ancestor proof, + and returns a capability plan for the active profile. Does not review, + approve, request changes, merge, or mutate Gitea state. + + Args: + pr_number: Open PR to assess. + target_branch: Branch used for ancestry proof (default master). + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with eligibility proof, capability plan, and safe next action. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "performed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) + + assessment = reconciliation_workflow.assess_open_pr_reconciliation( + pr=pr, + project_root=PROJECT_ROOT, + remote=remote, + target_branch=target_branch, + ) + + profile = get_profile() + capabilities = reconciliation_workflow.profile_reconciliation_capabilities( + profile.get("allowed_operations"), + profile.get("forbidden_operations"), + ) + plan = reconciliation_workflow.resolve_reconciliation_plan( + assessment=assessment, + capabilities=capabilities, + ) + + return { + "success": True, + "performed": False, + "pr_number": pr_number, + "assessment": assessment, + "capabilities": capabilities, + "plan": plan, + "workflow_source": "skills/llm-project-workflow/workflows/reconcile-landed-pr.md", + "task_mode": "reconcile-landed-pr", + "review_merge_allowed": False, + } + + +@mcp.tool() +def gitea_scan_already_landed_open_prs( + target_branch: str = "master", + limit: int = 50, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Read-only: list open PRs already landed on the target branch (#301). + + Traverses open PR inventory with pagination proof, assesses each candidate, + and returns PRs classified as ``ALREADY_LANDED_RECONCILE_REQUIRED``. Does + not review, approve, merge, or mutate Gitea state. + + Args: + target_branch: Branch used for ancestry proof (default master). + limit: Max open PRs to inspect across pages. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with candidates, pagination metadata, and target branch SHA proof. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "performed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + url = f"{repo_api_url(h, o, r)}/pulls?state=open" + + open_prs: list[dict] = [] + current_page = 1 + pages_fetched = 0 + last_meta: dict = {} + per_page = min(50, max(1, int(limit))) + while pages_fetched < 100 and len(open_prs) < limit: + raw_page, last_meta = api_fetch_page( + url, auth, page=current_page, limit=per_page + ) + pages_fetched += 1 + remaining = limit - len(open_prs) + open_prs.extend(raw_page[:remaining]) + if last_meta["is_final_page"] or len(open_prs) >= limit: + break + current_page += 1 + + pagination = { + "page": 1, + "per_page": per_page, + "returned_count": len(open_prs), + "has_more": False, + "next_page": None, + "is_final_page": bool(last_meta.get("is_final_page")), + "pages_fetched": pages_fetched, + "inventory_complete": bool(last_meta.get("is_final_page")), + "total_count": len(open_prs) if last_meta.get("is_final_page") else None, + } + + target_fetch = reconciliation_workflow.fetch_target_branch( + PROJECT_ROOT, remote, target_branch + ) + + candidates: list[dict] = [] + for pr in open_prs: + assessment = reconciliation_workflow.assess_open_pr_reconciliation( + pr=pr, + project_root=PROJECT_ROOT, + remote=remote, + target_branch=target_branch, + target_fetch=target_fetch, + ) + if assessment.get("eligibility_class") == ( + reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED + ): + candidates.append(assessment) + + return { + "success": True, + "performed": False, + "target_branch": target_branch, + "target_branch_sha": target_fetch.get("target_branch_sha"), + "git_ref_mutations": ( + [target_fetch["git_fetch_command"]] + if target_fetch.get("git_fetch_command") + else [] + ), + "candidates": candidates, + "candidate_count": len(candidates), + "pagination": pagination, + "task_mode": "reconcile-landed-pr", + "review_merge_allowed": False, + } + + @mcp.tool() def gitea_close_issue( issue_number: int, @@ -4182,6 +4355,28 @@ _PROJECT_SKILLS = { "Do not merge unless the operator explicitly authorizes it.", ], }, + "gitea-reconcile-landed-pr": { + "description": "Reconcile open PRs whose heads are already on the " + "target branch without review or merge.", + "when_to_use": "An open PR blocks the review queue but its head SHA " + "is already an ancestor of master; use the dedicated " + "reconciliation path instead of review/merge.", + "required_operations": ["gitea.read"], + "status": "available", + "notes": "Load skills/llm-project-workflow/workflows/reconcile-landed-pr.md " + "first. PR close requires exact gitea.pr.close; review/merge " + "capabilities must not be used on this path.", + "steps": [ + "Resolve task: gitea_resolve_task_capability(task='reconcile_landed_pr').", + "Load canonical workflow via mcp_get_skill_guide or " + "skills/llm-project-workflow/workflows/reconcile-landed-pr.md.", + "Scan queue: gitea_scan_already_landed_open_prs (pagination proof).", + "Assess one PR: gitea_assess_already_landed_reconciliation.", + "Follow the plan outcome: full close only with gitea.pr.close, " + "comment-then-stop when close is missing, recovery handoff otherwise.", + "Never approve, request changes, or merge on this path.", + ], + }, "gitea-pr-merge": { "description": "Gated merge of an approved pull request.", "when_to_use": "Reviewer/merger profile with explicit operator " diff --git a/reconciliation_workflow.py b/reconciliation_workflow.py new file mode 100644 index 0000000..48867b7 --- /dev/null +++ b/reconciliation_workflow.py @@ -0,0 +1,292 @@ +"""Already-landed PR reconciliation workflow helpers (#301). + +Read-only assessment and capability planning for reconciling open PRs whose +heads are already ancestors of the target branch. Does not invoke review or +merge paths. +""" + +from __future__ import annotations + +import subprocess +from typing import Any + +import gitea_config +from merged_cleanup_reconcile import extract_linked_issue, is_head_ancestor_of_ref + +ELIGIBILITY_ALREADY_LANDED = "ALREADY_LANDED_RECONCILE_REQUIRED" +ELIGIBILITY_NOT_LANDED = "NOT_ALREADY_LANDED" +ELIGIBILITY_STALE_TARGET = "TARGET_BRANCH_UNVERIFIED" +ELIGIBILITY_PR_NOT_OPEN = "PR_NOT_OPEN" + +OUTCOME_FULL_RECONCILE = "FULL_RECONCILE_CLOSE_ALLOWED" +OUTCOME_PARTIAL_COMMENT = "PARTIAL_RECONCILE_COMMENT_THEN_STOP" +OUTCOME_RECOVERY_HANDOFF = "RECOVERY_HANDOFF_ONLY" +OUTCOME_NOT_LANDED = "NOT_LANDED_NO_ACTION" +OUTCOME_GATE_NOT_PROVEN = "GATE_NOT_PROVEN" + +RECONCILE_WORKFLOW_MARKERS = ( + "workflows/reconcile-landed-pr.md", + "reconcile-landed-pr.md", +) + +RECONCILE_TASK_MARKERS = ( + "reconcile-landed-pr", + "reconcile already-landed", + "reconcile_already_landed", +) + + +def fetch_target_branch(project_root: str, remote: str, branch: str) -> dict[str, Any]: + """Fetch *branch* from *remote* and return the resolved SHA.""" + ref = f"{remote}/{branch}" + fetch = subprocess.run( + ["git", "-C", project_root, "fetch", remote, branch], + capture_output=True, + text=True, + check=False, + ) + if fetch.returncode != 0: + return { + "success": False, + "target_branch": branch, + "target_ref": ref, + "target_branch_sha": None, + "reasons": [ + f"git fetch {remote} {branch} failed: " + f"{(fetch.stderr or fetch.stdout or '').strip()}" + ], + } + + rev = subprocess.run( + ["git", "-C", project_root, "rev-parse", ref], + capture_output=True, + text=True, + check=False, + ) + if rev.returncode != 0: + return { + "success": False, + "target_branch": branch, + "target_ref": ref, + "target_branch_sha": None, + "reasons": [ + f"git rev-parse {ref} failed: " + f"{(rev.stderr or rev.stdout or '').strip()}" + ], + } + + return { + "success": True, + "target_branch": branch, + "target_ref": ref, + "target_branch_sha": (rev.stdout or "").strip(), + "reasons": [], + "git_fetch_command": f"git fetch {remote} {branch}", + } + + +def profile_reconciliation_capabilities( + allowed: list[str] | None, + forbidden: list[str] | None, +) -> dict[str, bool]: + """Map active profile operations to reconciliation capabilities.""" + allowed = allowed or [] + forbidden = forbidden or [] + + def can(op: str) -> bool: + return gitea_config.check_operation(op, allowed, forbidden)[0] + + return { + "read": can("gitea.read"), + "comment_pr": can("gitea.pr.comment"), + "comment_issue": can("gitea.issue.comment"), + "close_pr": can("gitea.pr.close"), + "close_issue": can("gitea.issue.close"), + "review_pr": can("gitea.pr.review") or can("gitea.pr.approve"), + "merge_pr": can("gitea.pr.merge"), + } + + +def assess_open_pr_reconciliation( + *, + pr: dict[str, Any], + project_root: str, + remote: str, + target_branch: str, + target_fetch: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Assess whether an open PR is eligible for already-landed reconciliation.""" + pr_number = int(pr.get("number") or 0) + pr_state = (pr.get("state") or "").strip().lower() + head = pr.get("head") or {} + head_sha = head.get("sha") if isinstance(head, dict) else None + head_ref = head.get("ref") if isinstance(head, dict) else None + base = pr.get("base") or {} + base_ref = base.get("ref") if isinstance(base, dict) else None + title = pr.get("title") or "" + body = pr.get("body") or "" + + fetch_result = target_fetch or fetch_target_branch( + project_root, remote, target_branch + ) + linked_issue = extract_linked_issue(title, body) + + result: dict[str, Any] = { + "pr_number": pr_number, + "pr_state": pr_state, + "candidate_head_sha": head_sha, + "head_ref": head_ref, + "base_ref": base_ref or target_branch, + "target_branch": target_branch, + "target_branch_sha": fetch_result.get("target_branch_sha"), + "linked_issue": linked_issue, + "git_ref_mutations": [], + "reasons": [], + "review_merge_allowed": False, + } + if fetch_result.get("git_fetch_command"): + result["git_ref_mutations"].append(fetch_result["git_fetch_command"]) + + if pr_state != "open": + result["eligibility_class"] = ELIGIBILITY_PR_NOT_OPEN + result["ancestor_proof"] = None + result["reconciliation_allowed"] = False + result["reasons"].append(f"PR #{pr_number} state is {pr_state!r}, not open") + return result + + if not fetch_result.get("success"): + result["eligibility_class"] = ELIGIBILITY_STALE_TARGET + result["ancestor_proof"] = None + result["reconciliation_allowed"] = False + result["reasons"].extend(fetch_result.get("reasons") or []) + return result + + target_ref = fetch_result.get("target_ref") or f"{remote}/{target_branch}" + ancestor = is_head_ancestor_of_ref(project_root, head_sha, target_ref) + result["ancestor_proof"] = ancestor + + if ancestor is None: + result["eligibility_class"] = ELIGIBILITY_STALE_TARGET + result["reconciliation_allowed"] = False + result["reasons"].append( + f"ancestor check failed for head {head_sha!r} against {target_ref}" + ) + return result + + if ancestor: + result["eligibility_class"] = ELIGIBILITY_ALREADY_LANDED + result["reconciliation_allowed"] = True + return result + + result["eligibility_class"] = ELIGIBILITY_NOT_LANDED + result["reconciliation_allowed"] = False + result["reasons"].append( + f"PR head {head_sha} is not an ancestor of {target_ref}" + ) + return result + + +def resolve_reconciliation_plan( + *, + assessment: dict[str, Any], + capabilities: dict[str, bool] | None, +) -> dict[str, Any]: + """Map eligibility + profile capabilities to a reconciliation outcome.""" + caps = capabilities or {} + reasons: list[str] = [] + + if not assessment.get("reconciliation_allowed"): + outcome = OUTCOME_NOT_LANDED + if assessment.get("eligibility_class") in { + ELIGIBILITY_STALE_TARGET, + ELIGIBILITY_PR_NOT_OPEN, + }: + outcome = OUTCOME_GATE_NOT_PROVEN + reasons.extend(assessment.get("reasons") or []) + return { + "outcome": outcome, + "close_pr_allowed": False, + "comment_pr_allowed": False, + "close_issue_allowed": False, + "comment_issue_allowed": False, + "review_merge_allowed": False, + "missing_capabilities": _missing_reconciliation_capabilities(caps), + "safe_next_action": ( + "Do not reconcile via review/merge; repair missing proof first." + ), + "reasons": reasons, + } + + close_pr = bool(caps.get("close_pr")) + comment_pr = bool(caps.get("comment_pr")) + close_issue = bool(caps.get("close_issue")) + comment_issue = bool(caps.get("comment_issue")) + + if caps.get("review_pr") or caps.get("merge_pr"): + reasons.append( + "reconciliation path must not use review/merge capabilities" + ) + + if close_pr: + outcome = OUTCOME_FULL_RECONCILE + safe_next_action = ( + "Run reconciliation close via exact gitea.pr.close after proof; " + "do not approve or merge." + ) + elif comment_pr: + outcome = OUTCOME_PARTIAL_COMMENT + safe_next_action = ( + "Post one reconciliation comment with ancestor proof, then stop " + "for an authorized close profile." + ) + else: + outcome = OUTCOME_RECOVERY_HANDOFF + safe_next_action = ( + "Produce a recovery handoff naming missing gitea.pr.close and/or " + "gitea.pr.comment; do not loop through review/merge." + ) + reasons.append("gitea.pr.close is not available in the active profile") + + return { + "outcome": outcome, + "close_pr_allowed": close_pr, + "comment_pr_allowed": comment_pr, + "close_issue_allowed": close_issue, + "comment_issue_allowed": comment_issue, + "review_merge_allowed": False, + "missing_capabilities": _missing_reconciliation_capabilities(caps), + "safe_next_action": safe_next_action, + "reasons": reasons, + } + + +def _missing_reconciliation_capabilities(caps: dict[str, bool]) -> list[str]: + missing = [] + if not caps.get("read"): + missing.append("gitea.read") + if not caps.get("close_pr"): + missing.append("gitea.pr.close") + if not caps.get("comment_pr"): + missing.append("gitea.pr.comment") + return missing + + +def assess_reconcile_workflow_source(report_text: str) -> dict[str, Any]: + """Reconciliation reports must cite the canonical workflow file.""" + lower = (report_text or "").lower() + reasons = [] + if not any(marker in lower for marker in RECONCILE_WORKFLOW_MARKERS): + reasons.append( + "reconciliation report missing workflow source " + "(workflows/reconcile-landed-pr.md)" + ) + if not any(marker in lower for marker in RECONCILE_TASK_MARKERS): + reasons.append( + "reconciliation report missing task mode declaration " + "(reconcile-landed-pr)" + ) + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index 89e3701..e6a4518 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -17,6 +17,11 @@ here weakens or replaces them. import re import issue_duplicate_gate +from reconciliation_workflow import ( + RECONCILE_TASK_MARKERS, + RECONCILE_WORKFLOW_MARKERS, + assess_reconcile_workflow_source, +) from reviewer_worktree import assess_reviewer_worktree_proof _FULL_SHA = re.compile(r"^[0-9a-f]{40}$") diff --git a/role_session_router.py b/role_session_router.py index a7db807..d14750e 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -72,6 +72,8 @@ AUTHOR_TASKS = frozenset({ "comment_pr", "address_pr_change_requests", "delete_branch", + "reconcile_landed_pr", + "reconcile-landed-pr", }) TASK_REQUIRED_ROLE = { @@ -90,6 +92,8 @@ TASK_REQUIRED_ROLE = { "blind_pr_queue_review": "reviewer", "request_changes_pr": "reviewer", "approve_pr": "reviewer", + "reconcile_landed_pr": "author", + "reconcile-landed-pr": "author", } WRONG_ROLE_REVIEWER_MSG = ( diff --git a/task_capability_map.py b/task_capability_map.py index 9d162c1..16c1221 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -96,6 +96,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + "reconcile_landed_pr": { + "permission": "gitea.read", + "role": "author", + }, + "reconcile-landed-pr": { + "permission": "gitea.read", + "role": "author", + }, } # Issue-mutating MCP tools and their resolver task keys. diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 550c007..48d2f5e 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -118,6 +118,16 @@ def test_start_issue_template_references_work_issue_workflow(): assert "workflows/work-issue.md" in text +def test_reconcile_skill_registered_in_mcp_server(): + from mcp_server import _PROJECT_SKILLS + + assert "gitea-reconcile-landed-pr" in _PROJECT_SKILLS + steps = _PROJECT_SKILLS["gitea-reconcile-landed-pr"]["steps"] + joined = " ".join(steps).lower() + assert "gitea_scan_already_landed_open_prs" in joined + assert "gitea_assess_already_landed_reconciliation" in joined + + def test_reviewer_fallback_verifier_exported(): from review_proofs import assess_reviewer_fallback_report diff --git a/tests/test_reconciliation_mcp_tools.py b/tests/test_reconciliation_mcp_tools.py new file mode 100644 index 0000000..974df92 --- /dev/null +++ b/tests/test_reconciliation_mcp_tools.py @@ -0,0 +1,106 @@ +"""Tests for reconciliation MCP assessment tools (#301).""" +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import mcp_server +from mcp_server import ( + gitea_assess_already_landed_reconciliation, + gitea_scan_already_landed_open_prs, +) + +AUTHOR_PROFILE = { + "profile_name": "prgs-author", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.issue.comment", + "gitea.issue.close", + ], + "forbidden_operations": ["gitea.pr.close"], + "audit_label": "prgs-author", +} + +OPEN_PR = { + "number": 278, + "title": "Landed (Closes #263)", + "body": "", + "state": "open", + "head": {"ref": "feat/x", "sha": "a" * 40}, + "base": {"ref": "master"}, +} + + +class TestReconciliationMcpTools(unittest.TestCase): + def setUp(self): + self.mock_api = patch("mcp_server.api_request").start() + self.mock_auth = patch( + "mcp_server.get_auth_header", return_value="token test" + ).start() + patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE).start() + mcp_server._IDENTITY_CACHE.clear() + + def tearDown(self): + patch.stopall() + mcp_server._IDENTITY_CACHE.clear() + + def test_assess_returns_plan_without_mutations(self): + self.mock_api.return_value = OPEN_PR + with patch( + "mcp_server.reconciliation_workflow.assess_open_pr_reconciliation", + return_value={ + "reconciliation_allowed": True, + "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED", + "candidate_head_sha": OPEN_PR["head"]["sha"], + "target_branch_sha": "deadbeef", + "linked_issue": 263, + "review_merge_allowed": False, + }, + ): + res = gitea_assess_already_landed_reconciliation( + pr_number=278, remote="prgs" + ) + self.assertTrue(res["success"]) + self.assertFalse(res["performed"]) + self.assertEqual(res["plan"]["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP") + self.assertFalse(res["review_merge_allowed"]) + patch_calls = [ + c for c in self.mock_api.call_args_list if c.args[0] == "PATCH" + ] + self.assertEqual(patch_calls, []) + + def test_scan_returns_candidates_with_pagination(self): + with patch( + "mcp_server.api_fetch_page", + return_value=([OPEN_PR], { + "page": 1, + "per_page": 50, + "is_final_page": True, + "has_more": False, + "next_page": None, + }), + ), patch( + "mcp_server.reconciliation_workflow.fetch_target_branch", + return_value={ + "success": True, + "target_branch_sha": "deadbeef", + "git_fetch_command": "git fetch prgs master", + }, + ), patch( + "mcp_server.reconciliation_workflow.assess_open_pr_reconciliation", + return_value={ + "pr_number": 278, + "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED", + "reconciliation_allowed": True, + }, + ): + res = gitea_scan_already_landed_open_prs(remote="prgs", limit=50) + self.assertTrue(res["success"]) + self.assertEqual(res["candidate_count"], 1) + self.assertTrue(res["pagination"]["inventory_complete"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reconciliation_workflow.py b/tests/test_reconciliation_workflow.py new file mode 100644 index 0000000..e3235c4 --- /dev/null +++ b/tests/test_reconciliation_workflow.py @@ -0,0 +1,149 @@ +"""Tests for already-landed reconciliation workflow helpers (#301).""" +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import reconciliation_workflow +from review_proofs import assess_reconcile_workflow_source + + +class TestReconciliationAssessment(unittest.TestCase): + def test_already_landed_open_pr(self): + with patch( + "reconciliation_workflow.is_head_ancestor_of_ref", + return_value=True, + ): + assessment = reconciliation_workflow.assess_open_pr_reconciliation( + pr={ + "number": 278, + "state": "open", + "title": "Fix (Closes #263)", + "body": "", + "head": {"ref": "feat/x", "sha": "abc" * 13 + "a"}, + "base": {"ref": "master"}, + }, + project_root="/tmp/repo", + remote="prgs", + target_branch="master", + target_fetch={ + "success": True, + "target_ref": "prgs/master", + "target_branch_sha": "deadbeef", + "git_fetch_command": "git fetch prgs master", + }, + ) + self.assertTrue(assessment["reconciliation_allowed"]) + self.assertEqual( + assessment["eligibility_class"], + reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED, + ) + + def test_not_landed_pr(self): + with patch( + "reconciliation_workflow.is_head_ancestor_of_ref", + return_value=False, + ): + assessment = reconciliation_workflow.assess_open_pr_reconciliation( + pr={ + "number": 1, + "state": "open", + "title": "WIP", + "body": "", + "head": {"ref": "feat/y", "sha": "fff" * 13 + "f"}, + "base": {"ref": "master"}, + }, + project_root="/tmp/repo", + remote="prgs", + target_branch="master", + target_fetch={ + "success": True, + "target_ref": "prgs/master", + "target_branch_sha": "deadbeef", + }, + ) + self.assertFalse(assessment["reconciliation_allowed"]) + + +class TestReconciliationPlan(unittest.TestCase): + def test_full_close_when_close_pr_allowed(self): + plan = reconciliation_workflow.resolve_reconciliation_plan( + assessment={ + "reconciliation_allowed": True, + "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED, + }, + capabilities={ + "close_pr": True, + "comment_pr": True, + "close_issue": True, + "comment_issue": True, + }, + ) + self.assertEqual( + plan["outcome"], + reconciliation_workflow.OUTCOME_FULL_RECONCILE, + ) + self.assertTrue(plan["close_pr_allowed"]) + self.assertFalse(plan["review_merge_allowed"]) + + def test_comment_then_stop_without_close(self): + plan = reconciliation_workflow.resolve_reconciliation_plan( + assessment={ + "reconciliation_allowed": True, + "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED, + }, + capabilities={ + "close_pr": False, + "comment_pr": True, + }, + ) + self.assertEqual( + plan["outcome"], + reconciliation_workflow.OUTCOME_PARTIAL_COMMENT, + ) + + def test_recovery_handoff_without_comment_or_close(self): + plan = reconciliation_workflow.resolve_reconciliation_plan( + assessment={ + "reconciliation_allowed": True, + "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED, + }, + capabilities={"close_pr": False, "comment_pr": False}, + ) + self.assertEqual( + plan["outcome"], + reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF, + ) + + def test_not_landed_no_action(self): + plan = reconciliation_workflow.resolve_reconciliation_plan( + assessment={ + "reconciliation_allowed": False, + "eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED, + "reasons": ["not ancestor"], + }, + capabilities={"close_pr": True}, + ) + self.assertEqual( + plan["outcome"], + reconciliation_workflow.OUTCOME_NOT_LANDED, + ) + + +class TestReconcileWorkflowSource(unittest.TestCase): + def test_valid_report_passes(self): + report = ( + "Task mode: reconcile-landed-pr\n" + "Workflow source: workflows/reconcile-landed-pr.md\n" + ) + result = assess_reconcile_workflow_source(report) + self.assertTrue(result["complete"]) + + def test_missing_workflow_fails(self): + result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr") + self.assertFalse(result["complete"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file