gitea_mcp_server.py

@@ -479,6 +479,7 @@ import issue_lock_worktree  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
+import reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
 
@@ -3429,6 +3430,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_reconcile_already_landed_pr(
     pr_number: int,
@@ -3608,7 +3781,6 @@ def gitea_reconcile_already_landed_pr(
 
     return result
 
-
 @mcp.tool()
 def gitea_close_issue(
     issue_number: int,
@@ -4391,6 +4563,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 "

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,
+    }

review_proofs.py

@@ -18,6 +18,11 @@ import hashlib
 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}$")

role_session_router.py

@@ -76,6 +76,7 @@ AUTHOR_TASKS = frozenset({
     "delete_branch",
     "work_issue",
     "work-issue",
+    "reconcile_landed_pr",
 })
 
 RECONCILER_TASKS = frozenset({
@@ -104,6 +105,7 @@ TASK_REQUIRED_ROLE = {
     "approve_pr": "reviewer",
     "work_issue": "author",
     "work-issue": "author",
+    "reconcile_landed_pr": "author",
     "reconcile_already_landed_pr": "reconciler",
     "reconcile_already_landed": "reconciler",
     "reconcile-landed-pr": "reconciler",

task_capability_map.py

@@ -104,6 +104,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",
+    },
     "reconcile_already_landed_pr": {
         "permission": "gitea.pr.close",
         "role": "reconciler",

tests/test_llm_workflow_split.py

@@ -139,6 +139,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_pr_queue_cleanup_template_references_workflow():
     text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text(
         encoding="utf-8"

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()

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()