already_landed_reconcile.py

@@ -0,0 +1,142 @@
+"""Already-landed open PR reconciliation gates (#310).
+
+Reconciler workflows may close an open PR only when the PR head SHA is proven
+an ancestor of a freshly fetched target branch. Arbitrary PR closure is denied.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from typing import Any
+
+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"
+
+
+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 assess_already_landed_reconciliation(
+    *,
+    pr: dict[str, Any],
+    project_root: str,
+    remote: str,
+    target_branch: str,
+    target_fetch: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Return eligibility and proof for reconciling an open PR."""
+    pr_number = int(pr.get("number") or 0)
+    pr_state = (pr.get("state") or "").strip().lower()
+    head_sha = (pr.get("head") or {}).get("sha") if isinstance(pr.get("head"), dict) else None
+    if not head_sha and isinstance(pr.get("head"), str):
+        head_sha = None
+    head_ref = (pr.get("head") or {}).get("ref") if isinstance(pr.get("head"), dict) else pr.get("head")
+    base_ref = (pr.get("base") or {}).get("ref") if isinstance(pr.get("base"), dict) else pr.get("base")
+    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": [],
+    }
+    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["close_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["close_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["close_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["close_allowed"] = True
+        return result
+
+    result["eligibility_class"] = ELIGIBILITY_NOT_LANDED
+    result["close_allowed"] = False
+    result["reasons"].append(
+        f"PR head {head_sha} is not an ancestor of {target_ref}"
+    )
+    return result

docs/gitea-dual-namespace-deployment.md

@@ -23,6 +23,7 @@ launched with exactly one static execution profile:
 |-----------------------------|----------------|-------------|
 | `gitea-author` | an author profile | implement issues, push branches, open PRs, comment |
 | `gitea-reviewer` | a reviewer profile | review, approve/request changes, merge |
+| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#310) |
 
 Properties:
 

docs/gitea-execution-profiles.md

@@ -226,6 +226,24 @@ explicit operator-directed closure of a contaminated PR). If `close_pr` ever
 resolves as unknown, agents must fail closed rather than fall back to the
 edit path.
 
+## Reconciler profile for already-landed open PRs (#310)
+
+Normal author and reviewer profiles must not gain broad PR-close authority.
+Already-landed open PRs (head SHA is an ancestor of the target branch) need a
+dedicated reconciler profile such as `prgs-reconciler` with:
+
+- `gitea.read`
+- `gitea.pr.comment`
+- `gitea.issue.comment`
+- `gitea.issue.close`
+- `gitea.pr.close`
+
+Use the `gitea-reconciler` MCP namespace (static profile launch) and the
+`gitea_reconcile_already_landed_pr` tool. The resolver task is
+`reconcile_already_landed_pr`. PR close is allowed only after live PR fetch,
+fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
+heads are not already landed cannot be closed through this path.
+
 ## Identity and fail-closed rules
 
 Before **any** mutating action, a workflow must know both:

docs/llm-workflow-runbooks.md

@@ -169,6 +169,25 @@ To avoid the bottleneck of relaunching/restarting the MCP server to switch betwe
   * `mcp__gitea-reviewer__*` (for reviewing PRs, approving, requesting changes, merging)
 * **Trust Model:** Separate tokens remain separate in the keychain/environment. Each instance operates under its own `GITEA_MCP_PROFILE` and enforces its own `allowed_operations`. A runtime `whoami` identity check is still performed independently, and self-review/self-merge checks remain strictly mandatory. The dual-server pattern is a operational convenience and never a security bypass.
 * **Reviewer-Identity PR Creation Deadlock:** Reviewer/merge identities must not create PRs or push branches. Doing so makes the reviewer identity the PR author in Gitea, blocking subsequent independent review and causing a review deadlock. Normally, PRs must be created by the author/work identity (`gitea-author`), leaving the reviewer identity (`gitea-reviewer`) clean and available for independent review and merge.
+* **Reconciler namespace (#310):** Register a third static instance for
+  already-landed PR cleanup when review queues block on open PRs whose heads
+  already landed on `master`:
+
+```json
+"gitea-reconciler": {
+  "command": "/path/to/Gitea-Tools/venv/bin/python3",
+  "args": ["/path/to/Gitea-Tools/mcp_server.py"],
+  "env": {
+    "GITEA_MCP_CONFIG": "/path/to/.config/gitea-tools/profiles.json",
+    "GITEA_MCP_PROFILE": "prgs-reconciler"
+  }
+}
+```
+
+  The reconciler profile grants `gitea.pr.close` only for
+  `gitea_reconcile_already_landed_pr` after ancestry proof — not for normal
+  review or author workflows.
+
 * **Fallback:** If the dual-profile MCP launcher pattern is not supported or configured in the client, the LLM must relaunch or restart the client/MCP with the correct profile environment variable before claiming or working on any tasks.
 
 ## Setup runbook — interactive menu

gitea_mcp_server.py

@@ -476,6 +476,7 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
 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
@@ -3601,6 +3602,185 @@ def gitea_scan_already_landed_open_prs(
     }
 
 
+@mcp.tool()
+def gitea_reconcile_already_landed_pr(
+    pr_number: int,
+    close_pr: bool = False,
+    close_linked_issue: bool = False,
+    post_comment: bool = False,
+    comment_body: str = "",
+    target_branch: str = "master",
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Reconcile an open PR whose head is already on the target branch (#310).
+
+    Read-only assessment always runs. PR close requires ``gitea.pr.close`` and
+    proof that the pinned PR head SHA is an ancestor of a freshly fetched
+    target branch. Arbitrary PR closure is denied.
+
+    Args:
+        pr_number: Open PR to reconcile.
+        close_pr: When True, close the PR after already-landed proof passes.
+        close_linked_issue: When True, close a linked issue after live fetch
+            when the issue is open and ``gitea.issue.close`` is allowed.
+        post_comment: When True, post ``comment_body`` on the PR thread.
+        comment_body: PR comment text when ``post_comment`` is True.
+        target_branch: Target 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, optional mutations, and recovery guidance.
+    """
+    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"),
+        }
+
+    if close_pr:
+        close_block = _profile_operation_gate("gitea.pr.close")
+        if close_block:
+            return {
+                "success": False,
+                "performed": False,
+                "pr_number": pr_number,
+                "required_permission": "gitea.pr.close",
+                "reasons": close_block,
+                "permission_report": _permission_block_report("gitea.pr.close"),
+                "safe_next_action": (
+                    "Launch the prgs-reconciler MCP namespace or configure a "
+                    "profile with gitea.pr.close for already-landed cleanup."
+                ),
+            }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    base = repo_api_url(h, o, r)
+    pr_url = f"{base}/pulls/{pr_number}"
+    pr = api_request("GET", pr_url, auth)
+
+    assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+        pr=pr,
+        project_root=PROJECT_ROOT,
+        remote=remote,
+        target_branch=target_branch,
+    )
+
+    result: dict = {
+        "success": True,
+        "performed": False,
+        "pr_number": pr_number,
+        "pr_state": assessment.get("pr_state"),
+        "candidate_head_sha": assessment.get("candidate_head_sha"),
+        "target_branch": assessment.get("target_branch"),
+        "target_branch_sha": assessment.get("target_branch_sha"),
+        "ancestor_proof": assessment.get("ancestor_proof"),
+        "eligibility_class": assessment.get("eligibility_class"),
+        "linked_issue": assessment.get("linked_issue"),
+        "close_allowed": assessment.get("close_allowed"),
+        "git_ref_mutations": assessment.get("git_ref_mutations") or [],
+        "reasons": list(assessment.get("reasons") or []),
+        "pr_closed": False,
+        "issue_closed": False,
+        "pr_comment_posted": False,
+    }
+
+    if not assessment.get("close_allowed"):
+        result["success"] = False
+        result["safe_next_action"] = (
+            "Do not close this PR via the reconciler path until "
+            "already-landed proof passes."
+        )
+        return result
+
+    verify_preflight_purity(remote)
+
+    if post_comment and comment_body.strip():
+        comment_block = _profile_operation_gate("gitea.pr.comment")
+        if comment_block:
+            result["success"] = False
+            result["reasons"].extend(comment_block)
+            result["permission_report"] = _permission_block_report(
+                "gitea.pr.comment"
+            )
+            return result
+        comment_url = f"{base}/issues/{pr_number}/comments"
+        with _audited(
+            "comment_pr",
+            host=h,
+            remote=remote,
+            org=o,
+            repo=r,
+            pr_number=pr_number,
+            request_metadata={"source": "reconcile_already_landed_pr"},
+        ):
+            api_request("POST", comment_url, auth, {"body": comment_body.strip()})
+        result["pr_comment_posted"] = True
+        result["performed"] = True
+
+    if close_pr:
+        patch_url = f"{base}/pulls/{pr_number}"
+        request_metadata = {
+            "source": "reconcile_already_landed_pr",
+            "required_permission": "gitea.pr.close",
+            "candidate_head_sha": assessment.get("candidate_head_sha"),
+            "target_branch_sha": assessment.get("target_branch_sha"),
+            "ancestor_proof": assessment.get("ancestor_proof"),
+        }
+        with _audited(
+            "close_pr",
+            host=h,
+            remote=remote,
+            org=o,
+            repo=r,
+            pr_number=pr_number,
+            request_metadata=request_metadata,
+        ):
+            closed = api_request("PATCH", patch_url, auth, {"state": "closed"})
+        result["pr_closed"] = True
+        result["performed"] = True
+        result["pr_state"] = closed.get("state", "closed")
+
+    linked_issue = assessment.get("linked_issue")
+    if close_linked_issue and linked_issue:
+        issue_block = _profile_operation_gate("gitea.issue.close")
+        if issue_block:
+            result["reasons"].extend(issue_block)
+            result["issue_close_blocked"] = True
+            return result
+        issue_url = f"{base}/issues/{linked_issue}"
+        issue = api_request("GET", issue_url, auth)
+        result["linked_issue_live_status"] = issue.get("state")
+        if (issue.get("state") or "").lower() == "open":
+            with _audited(
+                "close_issue",
+                host=h,
+                remote=remote,
+                org=o,
+                repo=r,
+                issue_number=linked_issue,
+                request_metadata={"source": "reconcile_already_landed_pr"},
+            ):
+                api_request(
+                    "PATCH",
+                    issue_url,
+                    auth,
+                    {"state": "closed"},
+                )
+            result["issue_closed"] = True
+            result["performed"] = True
+
+    return result
+
 @mcp.tool()
 def gitea_close_issue(
     issue_number: int,
@@ -4111,17 +4291,26 @@ def _role_kind(allowed, forbidden) -> str:
     """Classify the active profile from its normalized operations.
 
     'author' can create PRs / push branches; 'reviewer' can approve/merge;
-    'mixed' can do both (a config smell — the reviewer-deadlock invariant
-    forbids it in v2 configs); 'limited' can do neither.
+    'reconciler' can close already-landed PRs via ``gitea.pr.close`` without
+    review/author powers; 'mixed' can do both (a config smell — the
+    reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do
+    neither.
     """
     def can(op):
         return gitea_config.check_operation(op, allowed, forbidden)[0]
     review = can("gitea.pr.approve") or can("gitea.pr.merge")
     author = can("gitea.pr.create") or can("gitea.branch.push")
+    reconciler = (
+        can("gitea.pr.close")
+        and not review
+        and not author
+    )
     if review and author:
         return "mixed"
     if review:
         return "reviewer"
+    if reconciler:
+        return "reconciler"
     if author:
         return "author"
     return "limited"
@@ -4923,6 +5112,7 @@ _RUNTIME_CAPABILITY_TASKS = (
     "review_pr",
     "merge_pr",
     "close_issue",
+    "reconcile_already_landed_pr",
 )
 
 
@@ -4974,6 +5164,7 @@ def _build_runtime_task_capabilities(
         "review_pr": "can_review_prs",
         "merge_pr": "can_merge_prs",
         "close_issue": "can_close_issues",
+        "reconcile_already_landed_pr": "can_reconcile_already_landed_prs",
     }
     for task in _RUNTIME_CAPABILITY_TASKS:
         permission = task_capability_map.required_permission(task)

migrate_profiles.py

@@ -31,6 +31,8 @@ REVIEWER_DEFAULT_FORBIDDEN = ["branch", "commit", "push", "open_pr"]
 def infer_role(name, execution_profile):
     """Return the unambiguous role for a legacy profile name, or None."""
     haystack = f"{name} {execution_profile or ''}".lower()
+    if "reconciler" in haystack:
+        return "reconciler"
     has_author = "author" in haystack
     has_reviewer = "reviewer" in haystack
     if has_author == has_reviewer:

review_proofs.py

@@ -1715,6 +1715,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         if report_text
         else {"proven": True, "block": False, "reasons": [], "violations": []}
     )
+    reconcile_inventory = (
+        assess_reconcile_inventory_report(report_text)
+        if report_text and _RECONCILE_INVENTORY_HINT.search(report_text)
+        else {"proven": True, "block": False, "reasons": []}
+    )
     reconcile_linked_issue = (
         assess_reconcile_linked_issue_report(report_text)
         if report_text and _RECONCILE_LINKED_ISSUE_HINT.search(report_text)
@@ -1891,6 +1896,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "queue-status report violates loaded workflow proof gates (#339)"
         )
         downgrade_reasons.extend(queue_status_report.get("reasons", []))
+    if not reconcile_inventory.get("proven"):
+        downgrade_reasons.append(
+            "reconciliation PR inventory lacks pagination proof (#308)"
+        )
+        downgrade_reasons.extend(reconcile_inventory.get("reasons", []))
     if not reconcile_linked_issue.get("proven"):
         downgrade_reasons.append(
             "reconciliation linked issue status lacks live fetch proof (#300)"
@@ -2004,6 +2014,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "queue_status_violations": list(
             queue_status_report.get("violations") or []
         ),
+        "reconcile_inventory_proven": bool(reconcile_inventory.get("proven")),
+        "reconcile_inventory_violations": list(
+            reconcile_inventory.get("reasons") or []
+        ),
         "reconcile_linked_issue_proven": bool(reconcile_linked_issue.get("proven")),
         "reconcile_linked_issue_violations": list(
             reconcile_linked_issue.get("reasons") or []
@@ -4906,6 +4920,12 @@ _QUEUE_STATUS_REPORT_HINT = re.compile(
     re.I,
 )
 
+_RECONCILE_INVENTORY_HINT = re.compile(
+    r"already[- ]landed reconciliation|reconcile[- ]landed|"
+    r"open pr inventory|inventory pagination proof",
+    re.I,
+)
+
 _RECONCILE_LINKED_ISSUE_HINT = re.compile(
     r"already[- ]landed|reconcil|linked issue(?:\s+live)?\s+status",
     re.I,
@@ -5339,6 +5359,13 @@ def assess_validation_worktree_edit_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_reconcile_inventory_report(report_text, **kwargs):
+    """#308: prove PR inventory pagination in reconciliation reports."""
+    from reviewer_reconcile_inventory import assess_reconcile_inventory_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_reconcile_linked_issue_report(report_text, **kwargs):
     """#300: prove linked issue status was fetched live in reconciliation handoffs."""
     from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report as _assess

reviewer_reconcile_inventory.py

@@ -0,0 +1,162 @@
+"""Reconciliation PR inventory pagination proof verifier (#308)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_COMPLETE_SCAN_CLAIM_RE = re.compile(
+    r"\b(?:all already[- ]landed(?:\s+prs?)?(?:\s+were)?\s+found|"
+    r"complete queue scan|all open prs? (?:were )?(?:found|checked|inventoried|listed)|"
+    r"inventory (?:is )?complete|exhaustive (?:pr )?inventory|"
+    r"no additional already[- ]landed|scanned (?:the )?(?:full )?open pr queue)\b",
+    re.I,
+)
+
+_FINALITY_RE = re.compile(
+    r"is_final_page\s*:\s*true|has_more\s*:\s*false|no next page|final[- ]page|"
+    r"pagination_complete\s*:\s*true|inventory pagination proof\s*:\s*(?!assumed|none\b).+",
+    re.I,
+)
+
+_REQUESTED_PAGE_SIZE_RE = re.compile(
+    r"requested page size\s*:\s*(\d+)|page[- ]?size\s*(?:=|:)\s*(\d+)",
+    re.I,
+)
+_PAGES_FETCHED_RE = re.compile(
+    r"pages? fetched\s*:\s*(\d+)|page count\s*:\s*(\d+)",
+    re.I,
+)
+_RETURNED_PER_PAGE_RE = re.compile(
+    r"returned pr count(?: per page)?\s*:|open pr count per page|"
+    r"returned \d+ open prs? per page|per[- ]page counts?\s*:",
+    re.I,
+)
+_EXACT_LIMIT_RETURN_RE = re.compile(
+    r"returned\s+(\d+)\s+open prs?|listed\s+(\d+)\s+open prs?",
+    re.I,
+)
+
+_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
+    r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
+    r"default (?:gitea )?page[- ]?size|assumed complete",
+    re.I,
+)
+
+
+def _int_from_groups(match: re.Match[str] | None) -> int | None:
+    if not match:
+        return None
+    for group in match.groups():
+        if group:
+            return int(group)
+    return None
+
+
+def _pagination_proven(text: str, session: dict) -> bool:
+    if session.get("pagination_complete") or session.get("inventory_complete"):
+        return True
+    if _FINALITY_RE.search(text):
+        return True
+    pages = session.get("pages_fetched")
+    if isinstance(pages, int) and pages >= 1 and session.get("is_final_page"):
+        return True
+    return False
+
+
+def _metadata_present(text: str, session: dict) -> list[str]:
+    missing: list[str] = []
+    has_page_size = bool(
+        _REQUESTED_PAGE_SIZE_RE.search(text)
+        or session.get("requested_page_size")
+    )
+    has_pages_fetched = bool(
+        _PAGES_FETCHED_RE.search(text) or session.get("pages_fetched")
+    )
+    has_returned_counts = bool(
+        _RETURNED_PER_PAGE_RE.search(text) or session.get("returned_per_page")
+    )
+    if not has_page_size:
+        missing.append("requested page size")
+    if not has_pages_fetched:
+        missing.append("page count fetched")
+    if not has_returned_counts:
+        missing.append("returned PR count per page")
+    if not _pagination_proven(text, session):
+        missing.append("final-page/no-next-page proof")
+    return missing
+
+
+def assess_reconcile_inventory_report(
+    report_text: str,
+    *,
+    inventory_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate PR inventory pagination proof in reconciliation reports (#308)."""
+    text = report_text or ""
+    session = dict(inventory_session or {})
+    reasons: list[str] = []
+
+    claims_scan = bool(_COMPLETE_SCAN_CLAIM_RE.search(text))
+    lists_open_prs = bool(
+        re.search(r"open prs? listed|listed open prs?|pr inventory", text, re.I)
+    )
+
+    if not claims_scan and not lists_open_prs and not session.get("inventory_required"):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "inventory_claimed": False,
+            "safe_next_action": "proceed",
+        }
+
+    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text) and not _pagination_proven(text, session):
+        reasons.append(
+            "reconciliation inventory assumed complete from page-size guess "
+            "without final-page proof (#308)"
+        )
+
+    if claims_scan and not _pagination_proven(text, session):
+        reasons.append(
+            "complete queue scan or all already-landed claim requires "
+            "pagination finality proof (#308)"
+        )
+
+    missing = _metadata_present(text, session)
+    if (claims_scan or lists_open_prs) and missing:
+        reasons.append(
+            "reconciliation inventory missing pagination metadata: "
+            + ", ".join(missing)
+        )
+
+    requested = session.get("requested_page_size")
+    returned = session.get("returned_page_size")
+    if isinstance(requested, int) and isinstance(returned, int):
+        if returned >= requested > 0 and not _pagination_proven(text, session):
+            reasons.append(
+                "exactly-full first reconciliation page without final-page proof (#308)"
+            )
+    else:
+        for match in _EXACT_LIMIT_RETURN_RE.finditer(text):
+            count = _int_from_groups(match)
+            if count in {10, 20, 50} and not _pagination_proven(text, session):
+                reasons.append(
+                    "exactly-full first reconciliation page without final-page proof (#308)"
+                )
+                break
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": list(dict.fromkeys(reasons)),
+        "inventory_claimed": claims_scan or lists_open_prs,
+        "pagination_proven": _pagination_proven(text, session),
+        "safe_next_action": (
+            "include requested page size, pages fetched, per-page counts, and "
+            "final-page/no-next-page proof before claiming complete scan"
+            if reasons
+            else "proceed"
+        ),
+    }

role_session_router.py

@@ -74,7 +74,14 @@ AUTHOR_TASKS = frozenset({
     "comment_pr",
     "address_pr_change_requests",
     "delete_branch",
+    "work_issue",
+    "work-issue",
     "reconcile_landed_pr",
+})
+
+RECONCILER_TASKS = frozenset({
+    "reconcile_already_landed_pr",
+    "reconcile_already_landed",
     "reconcile-landed-pr",
 })
 
@@ -96,8 +103,12 @@ TASK_REQUIRED_ROLE = {
     "pr-queue-cleanup": "reviewer",
     "request_changes_pr": "reviewer",
     "approve_pr": "reviewer",
+    "work_issue": "author",
+    "work-issue": "author",
     "reconcile_landed_pr": "author",
-    "reconcile-landed-pr": "author",
+    "reconcile_already_landed_pr": "reconciler",
+    "reconcile_already_landed": "reconciler",
+    "reconcile-landed-pr": "reconciler",
     # #309: reconciler tasks close already-landed PRs/issues only.
     "reconcile_close_landed_pr": "reconciler",
     "reconcile_close_landed_issue": "reconciler",

task_capability_map.py

@@ -112,6 +112,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    "reconcile_already_landed_pr": {
+        "permission": "gitea.pr.close",
+        "role": "reconciler",
+    },
     # #309: dedicated reconciler path for already-landed open PRs. Exact
     # close capabilities only — never review/approve/request_changes/merge.
     "reconcile_close_landed_pr": {

tests/test_already_landed_reconcile.py

@@ -0,0 +1,131 @@
+"""Tests for already-landed PR reconciliation gates (#310)."""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import already_landed_reconcile
+
+
+class TestAssessAlreadyLandedReconciliation(unittest.TestCase):
+    def test_open_pr_already_landed_allows_close(self):
+        with patch(
+            "already_landed_reconcile.is_head_ancestor_of_ref",
+            return_value=True,
+        ):
+            assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+                pr={
+                    "number": 278,
+                    "state": "open",
+                    "title": "Fix thing (Closes #263)",
+                    "body": "",
+                    "head": {"ref": "feat/x", "sha": "abc123"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_branch": "master",
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                    "git_fetch_command": "git fetch prgs master",
+                },
+            )
+        self.assertTrue(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_ALREADY_LANDED,
+        )
+        self.assertEqual(assessment["linked_issue"], 263)
+        self.assertTrue(assessment["ancestor_proof"])
+
+    def test_not_landed_pr_denies_close(self):
+        with patch(
+            "already_landed_reconcile.is_head_ancestor_of_ref",
+            return_value=False,
+        ):
+            assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+                pr={
+                    "number": 99,
+                    "state": "open",
+                    "title": "WIP",
+                    "body": "",
+                    "head": {"ref": "feat/y", "sha": "fff111"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_branch": "master",
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                },
+            )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_NOT_LANDED,
+        )
+
+    def test_stale_target_branch_denies_close(self):
+        assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+            pr={
+                "number": 99,
+                "state": "open",
+                "title": "WIP",
+                "body": "",
+                "head": {"ref": "feat/y", "sha": "fff111"},
+                "base": {"ref": "master"},
+            },
+            project_root="/tmp/repo",
+            remote="prgs",
+            target_branch="master",
+            target_fetch={
+                "success": False,
+                "target_branch": "master",
+                "target_ref": "prgs/master",
+                "target_branch_sha": None,
+                "reasons": ["git fetch failed"],
+            },
+        )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_STALE_TARGET,
+        )
+
+    def test_closed_pr_denies_close(self):
+        assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+            pr={
+                "number": 50,
+                "state": "closed",
+                "title": "Done",
+                "body": "",
+                "head": {"ref": "feat/z", "sha": "aaa"},
+                "base": {"ref": "master"},
+            },
+            project_root="/tmp/repo",
+            remote="prgs",
+            target_branch="master",
+            target_fetch={
+                "success": True,
+                "target_branch": "master",
+                "target_ref": "prgs/master",
+                "target_branch_sha": "deadbeef",
+            },
+        )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_PR_NOT_OPEN,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -211,6 +211,12 @@ def test_validation_worktree_edit_verifier_exported():
     assert callable(assess_validation_worktree_edit_report)
 
 
+def test_reconcile_inventory_verifier_exported():
+    from review_proofs import assess_reconcile_inventory_report
+
+    assert callable(assess_reconcile_inventory_report)
+
+
 def test_reconcile_linked_issue_verifier_exported():
     from review_proofs import assess_reconcile_linked_issue_report
 

tests/test_reconcile_already_landed_pr_tool.py

@@ -0,0 +1,149 @@
+"""Tests for gitea_reconcile_already_landed_pr MCP tool (#310)."""
+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_reconcile_already_landed_pr
+
+RECONCILER_PROFILE = {
+    "profile_name": "prgs-reconciler",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.comment",
+        "gitea.issue.comment",
+        "gitea.issue.close",
+        "gitea.pr.close",
+    ],
+    "forbidden_operations": [
+        "gitea.pr.approve",
+        "gitea.pr.merge",
+        "gitea.pr.create",
+        "gitea.branch.push",
+    ],
+    "audit_label": "prgs-reconciler",
+}
+
+AUTHOR_PROFILE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.create",
+        "gitea.issue.comment",
+    ],
+    "forbidden_operations": ["gitea.pr.close", "gitea.pr.approve", "gitea.pr.merge"],
+    "audit_label": "prgs-author",
+}
+
+OPEN_LANDED_PR = {
+    "number": 278,
+    "title": "Already landed (Closes #263)",
+    "body": "",
+    "state": "open",
+    "head": {"ref": "feat/issue-263", "sha": "2a544b78d1371925761d16f1c451bb3b2984470e"},
+    "base": {"ref": "master"},
+}
+
+
+class TestReconcileAlreadyLandedPrTool(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.verify_preflight_purity").start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def _set_profile(self, profile):
+        patch("mcp_server.get_profile", return_value=profile).start()
+
+    def test_close_blocked_without_pr_close_capability(self):
+        self._set_profile(AUTHOR_PROFILE)
+        res = gitea_reconcile_already_landed_pr(
+            pr_number=278,
+            close_pr=True,
+            remote="prgs",
+        )
+        self.assertFalse(res["success"])
+        self.assertFalse(res["performed"])
+        self.assertEqual(res["required_permission"], "gitea.pr.close")
+        self.mock_api.assert_not_called()
+
+    def test_not_landed_pr_close_denied_after_live_fetch(self):
+        self._set_profile(RECONCILER_PROFILE)
+        self.mock_api.return_value = OPEN_LANDED_PR
+        with patch(
+            "mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
+            return_value={
+                "pr_state": "open",
+                "candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
+                "target_branch": "master",
+                "target_branch_sha": "deadbeef",
+                "ancestor_proof": False,
+                "eligibility_class": "NOT_ALREADY_LANDED",
+                "linked_issue": 263,
+                "close_allowed": False,
+                "git_ref_mutations": ["git fetch prgs master"],
+                "reasons": ["not ancestor"],
+            },
+        ):
+            res = gitea_reconcile_already_landed_pr(
+                pr_number=278,
+                close_pr=True,
+                remote="prgs",
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res.get("pr_closed"))
+        patch_calls = [
+            c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
+        ]
+        self.assertEqual(patch_calls, [])
+
+    def test_already_landed_close_allowed_with_capability(self):
+        self._set_profile(RECONCILER_PROFILE)
+
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "GET" and "/pulls/278" in url:
+                return dict(OPEN_LANDED_PR)
+            if method == "PATCH" and "/pulls/278" in url:
+                closed = dict(OPEN_LANDED_PR)
+                closed["state"] = "closed"
+                return closed
+            return {}
+
+        self.mock_api.side_effect = api_side_effect
+        with patch(
+            "mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
+            return_value={
+                "pr_state": "open",
+                "candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
+                "target_branch": "master",
+                "target_branch_sha": "e017d80256217f17b0f421cd051cfa5cbcf5ae0f",
+                "ancestor_proof": True,
+                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+                "linked_issue": 263,
+                "close_allowed": True,
+                "git_ref_mutations": ["git fetch prgs master"],
+                "reasons": [],
+            },
+        ):
+            res = gitea_reconcile_already_landed_pr(
+                pr_number=278,
+                close_pr=True,
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+        self.assertTrue(res["performed"])
+        self.assertTrue(res["pr_closed"])
+        self.assertEqual(res["eligibility_class"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -275,6 +275,18 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertEqual(res["required_role_kind"], "author")
             self.assertIn("author", res["exact_safe_next_action"].lower())
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_reconcile_already_landed_pr_requires_close(self, _auth, _api):
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="reconcile_already_landed_pr", remote="prgs"
+            )
+            self.assertEqual(res["requested_task"], "reconcile_already_landed_pr")
+            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
+            self.assertEqual(res["required_role_kind"], "reconciler")
+            self.assertFalse(res["allowed_in_current_session"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api):

tests/test_reviewer_reconcile_inventory.py

@@ -0,0 +1,102 @@
+"""Tests for reconciliation inventory pagination verifier (#308)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_reconcile_inventory import assess_reconcile_inventory_report  # noqa: E402
+
+
+def _good_report() -> str:
+    return "\n".join([
+        "## Already-landed reconciliation",
+        "Open PR inventory:",
+        "- Requested page size: 50",
+        "- Pages fetched: 2",
+        "- Returned PR count per page: page1=50, page2=6",
+        "- Inventory pagination proof: is_final_page: true, has_more: false",
+        "All already-landed PRs in the scanned open queue were found.",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+    ])
+
+
+class TestReconcileInventoryPagination(unittest.TestCase):
+    def test_complete_metadata_passes(self):
+        result = assess_reconcile_inventory_report(_good_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["pagination_proven"])
+
+    def test_no_inventory_claim_skips_checks(self):
+        report = "## Controller Handoff\n- Selected PR: #278"
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["inventory_claimed"])
+
+    def test_first_page_below_limit_with_finality_passes(self):
+        report = "\n".join([
+            "Open PRs listed for reconciliation.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=6",
+            "- Inventory pagination proof: is_final_page: true",
+        ])
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_exactly_full_first_page_without_finality_blocks(self):
+        report = "\n".join([
+            "Complete queue scan of open PRs.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=50",
+            "Listed 50 open PRs on the first page.",
+        ])
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={"requested_page_size": 50, "returned_page_size": 50},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("full first" in r.lower() for r in result["reasons"]))
+
+    def test_missing_pagination_metadata_blocks(self):
+        report = "All open PRs were inventoried. Inventory is complete."
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("metadata" in r.lower() for r in result["reasons"]))
+
+    def test_default_page_size_assumption_blocks(self):
+        report = (
+            "Exhaustive PR inventory: fewer than the default Gitea page size returned."
+        )
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("page-size" in r.lower() or "page size" in r.lower()
+                            for r in result["reasons"]))
+
+    def test_session_pagination_complete_passes(self):
+        report = "All already-landed PRs were found after scanning open PRs."
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={
+                "pagination_complete": True,
+                "requested_page_size": 50,
+                "pages_fetched": 2,
+                "returned_per_page": [50, 3],
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reconcile_inventory_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_session_router.py

@@ -52,6 +52,22 @@ CONFIG = {
             ],
             "execution_profile": "prgs-reviewer",
         },
+        "prgs-reconciler": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reconciler",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_RECONCILER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.comment", "gitea.issue.comment",
+                "gitea.issue.close", "gitea.pr.close",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "execution_profile": "prgs-reconciler",
+        },
     },
     "rules": {"allow_runtime_switching": False},
 }
@@ -88,6 +104,7 @@ class TestRoleSessionRouter(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            "GITEA_TOKEN_RECONCILER": "reconciler-pass",
         }
 
     def test_reviewer_task_under_author_profile_wrong_role_stop(self):
@@ -197,6 +214,24 @@ class TestRoleSessionRouter(unittest.TestCase):
         )
         self.assertFalse(route["downstream_allowed"])
 
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reconciler-pass")
+    def test_reconciler_task_under_reconciler_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reconciler")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="reconcile_already_landed_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
+    def test_reconciler_task_under_author_profile_wrong_role_stop(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="reconcile_already_landed_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+
     def test_activate_profile_blocked_in_static_mode(self):
         with patch.dict(os.environ, self._env("prgs-author")):
             result = mcp_server.gitea_activate_profile(