fix: resolve conflicts for PR #388

Merge prgs/master; unify reconciler profile docs (#304) with master's
already-landed close tool guidance (#310).

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

author_mutation_worktree.py

@@ -0,0 +1,105 @@
+"""Branches-only author mutation worktree guard (#274).
+
+Author/coder mutations must run from a session-owned worktree under the
+project's ``branches/`` directory, never from the stable control checkout.
+"""
+
+from __future__ import annotations
+
+import os
+
+BASE_BRANCHES = frozenset({"master", "main", "dev"})
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").rstrip("/")
+
+
+def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
+    """True when *path* resolves inside ``<project_root>/branches/``."""
+    normalized = _normalize_path(path)
+    if not normalized:
+        return False
+    if "/branches/" in f"{normalized}/":
+        return True
+    if normalized.endswith("/branches"):
+        return True
+    if project_root:
+        root = _normalize_path(os.path.realpath(project_root))
+        real = _normalize_path(os.path.realpath(path))
+        if real.startswith(f"{root}/"):
+            rel = real[len(root) + 1 :]
+            return rel == "branches" or rel.startswith("branches/")
+    return False
+
+
+def resolve_mutation_workspace(
+    worktree_path: str | None,
+    project_root: str,
+    *,
+    active_worktree_env: str | None = None,
+    author_worktree_env: str | None = None,
+) -> str:
+    """Resolve the workspace path inspected before author mutations."""
+    for candidate in (worktree_path, active_worktree_env, author_worktree_env):
+        text = (candidate or "").strip()
+        if text:
+            return os.path.realpath(os.path.abspath(text))
+    return os.path.realpath(project_root)
+
+
+def assess_author_mutation_worktree(
+    *,
+    workspace_path: str,
+    project_root: str,
+    current_branch: str | None = None,
+    base_branches: frozenset[str] | None = None,
+) -> dict:
+    """Fail closed when author mutations are not rooted under ``branches/``."""
+    bases = base_branches or BASE_BRANCHES
+    reasons: list[str] = []
+    root = os.path.realpath(project_root)
+    workspace = os.path.realpath(workspace_path)
+    branch = (current_branch or "").strip()
+
+    under_branches = is_path_under_branches(workspace, root)
+    if not under_branches:
+        if workspace == root:
+            reasons.append(
+                "author mutation blocked: workspace is the stable control checkout; "
+                "create or switch to a session-owned worktree under branches/"
+            )
+        else:
+            reasons.append(
+                f"author mutation blocked: workspace '{workspace}' is not under "
+                f"'{root}/branches/'; create a branches/<task> worktree first"
+            )
+
+    if not under_branches and workspace == root and branch and branch not in bases:
+        reasons.append(
+            f"control checkout drift: branch '{branch}' is not a stable base "
+            f"branch ({'/'.join(sorted(bases))})"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "project_root": root,
+        "workspace_path": workspace,
+        "under_branches": is_path_under_branches(workspace, root),
+        "current_branch": branch or None,
+    }
+
+
+def format_author_mutation_worktree_error(assessment: dict) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    root = assessment.get("project_root") or "(unknown)"
+    reasons = "; ".join(assessment.get("reasons") or ["unknown branches-only violation"])
+    return (
+        f"Branches-only mutation guard (#274): {reasons}. "
+        f"project root: {root}; workspace: {workspace}. "
+        "Create a session-owned worktree under branches/ before mutating."
+    )

docs/gitea-dual-namespace-deployment.md

@@ -23,7 +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 (#304) |
+| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#304 profile; #310 close tool) |
 
 Properties:
 

docs/gitea-execution-profiles.md

@@ -226,11 +226,12 @@ 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 (#304)
+## Reconciler profile for already-landed open PRs (#304 / #310)
 
 Normal author and reviewer profiles must not gain broad `gitea.pr.close`
-authority. Already-landed open PRs need a dedicated reconciler profile such
-as `prgs-reconciler` with a narrow operation set:
+authority. Already-landed open PRs (head SHA is an ancestor of the target
+branch) need a dedicated reconciler profile such as `prgs-reconciler` with a
+narrow operation set:
 
 - `gitea.read`
 - `gitea.pr.comment`
@@ -244,8 +245,11 @@ Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
 
 Launch a static `gitea-reconciler` MCP namespace with
 `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
-`reconciler_profile.assess_reconciler_profile`; the gated close MCP tool and
-ancestor proofs ship with the #310 reconciler close stack.
+`reconciler_profile.assess_reconciler_profile` (#304). Use the
+`gitea_reconcile_already_landed_pr` tool (#310). 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
 

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

@@ -397,6 +397,28 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
             _preflight_resolved_role = resolved_role
 
 
+def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
+    """#274: author mutations must run from a branches/ session worktree."""
+    if _preflight_resolved_role == "reviewer":
+        return
+    workspace = author_mutation_worktree.resolve_mutation_workspace(
+        worktree_path,
+        PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
+    )
+    git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+    assessment = author_mutation_worktree.assess_author_mutation_worktree(
+        workspace_path=workspace,
+        project_root=PROJECT_ROOT,
+        current_branch=git_state.get("current_branch"),
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            author_mutation_worktree.format_author_mutation_worktree_error(assessment)
+        )
+
+
 def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
@@ -426,32 +448,33 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 "file edits before mutation (fail closed). "
                 f"{_format_preflight_workspace_details(details)}"
             )
-        return
-
-    if _preflight_whoami_violation:
-        raise RuntimeError(
-            "Pre-flight order violation: Workspace file edits occurred before "
-            f"gitea_whoami verification (fail closed). Offending files: "
-            f"{_format_preflight_files(_preflight_whoami_violation_files)}"
-        )
-    if _preflight_capability_violation:
-        raise RuntimeError(
-            "Pre-flight order violation: Workspace file edits occurred before "
-            f"gitea_resolve_task_capability verification (fail closed). Offending files: "
-            f"{_format_preflight_files(_preflight_capability_violation_files)}"
-        )
-
-    if _preflight_resolved_role == "reviewer":
-        current = _get_workspace_porcelain()
-        baseline = _preflight_capability_baseline_porcelain or ""
-        reviewer_delta = _new_tracked_changes_since(baseline, current)
-        _preflight_reviewer_violation_files = reviewer_delta
-        if reviewer_delta:
+    else:
+        if _preflight_whoami_violation:
             raise RuntimeError(
-                "Reviewer role violation: Reviewer profile is forbidden from modifying "
-                "tracked workspace files (fail closed). Offending files: "
-                f"{_format_preflight_files(reviewer_delta)}"
+                "Pre-flight order violation: Workspace file edits occurred before "
+                f"gitea_whoami verification (fail closed). Offending files: "
+                f"{_format_preflight_files(_preflight_whoami_violation_files)}"
             )
+        if _preflight_capability_violation:
+            raise RuntimeError(
+                "Pre-flight order violation: Workspace file edits occurred before "
+                f"gitea_resolve_task_capability verification (fail closed). Offending files: "
+                f"{_format_preflight_files(_preflight_capability_violation_files)}"
+            )
+
+        if _preflight_resolved_role == "reviewer":
+            current = _get_workspace_porcelain()
+            baseline = _preflight_capability_baseline_porcelain or ""
+            reviewer_delta = _new_tracked_changes_since(baseline, current)
+            _preflight_reviewer_violation_files = reviewer_delta
+            if reviewer_delta:
+                raise RuntimeError(
+                    "Reviewer role violation: Reviewer profile is forbidden from modifying "
+                    "tracked workspace files (fail closed). Offending files: "
+                    f"{_format_preflight_files(reviewer_delta)}"
+                )
+
+    _enforce_branches_only_author_mutation(worktree_path)
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -476,6 +499,8 @@ 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 author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
@@ -3429,6 +3454,186 @@ def gitea_reconcile_merged_cleanups(
     return {"success": True, "performed": True, **report}
 
 
+@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,
@@ -3940,8 +4145,10 @@ def _role_kind(allowed, forbidden) -> str:
 
     'reconciler' can close already-landed PRs without review/author powers;
     '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.
     """
     if reconciler_profile.is_reconciler_profile(allowed, forbidden):
         return "reconciler"
@@ -3949,10 +4156,17 @@ def _role_kind(allowed, forbidden) -> str:
         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"
@@ -4732,6 +4946,7 @@ _RUNTIME_CAPABILITY_TASKS = (
     "review_pr",
     "merge_pr",
     "close_issue",
+    "reconcile_already_landed_pr",
 )
 
 
@@ -4783,6 +4998,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)

review_proofs.py

@@ -1710,6 +1710,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)
@@ -1886,6 +1891,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)"
@@ -1999,6 +2009,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 []
@@ -4901,6 +4915,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,
@@ -5334,6 +5354,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,6 +74,14 @@ AUTHOR_TASKS = frozenset({
     "comment_pr",
     "address_pr_change_requests",
     "delete_branch",
+    "work_issue",
+    "work-issue",
+})
+
+RECONCILER_TASKS = frozenset({
+    "reconcile_already_landed_pr",
+    "reconcile_already_landed",
+    "reconcile-landed-pr",
 })
 
 TASK_REQUIRED_ROLE = {
@@ -94,6 +102,11 @@ TASK_REQUIRED_ROLE = {
     "pr-queue-cleanup": "reviewer",
     "request_changes_pr": "reviewer",
     "approve_pr": "reviewer",
+    "work_issue": "author",
+    "work-issue": "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

@@ -104,6 +104,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_author_mutation_worktree.py

@@ -0,0 +1,109 @@
+"""Tests for branches-only author mutation worktree guard (#274)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import author_mutation_worktree as amw  # noqa: E402
+
+
+class TestPathUnderBranches(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_branches_subdirectory_passes(self):
+        self.assertTrue(
+            amw.is_path_under_branches(f"{self.ROOT}/branches/issue-274", self.ROOT)
+        )
+
+    def test_control_checkout_fails(self):
+        self.assertFalse(amw.is_path_under_branches(self.ROOT, self.ROOT))
+
+    def test_sibling_directory_fails(self):
+        self.assertFalse(
+            amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
+        )
+
+
+class TestAssessAuthorMutationWorktree(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_branches_worktree_passes(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=f"{self.ROOT}/branches/issue-274",
+            project_root=self.ROOT,
+            current_branch="feat/issue-274-branches-only-worktrees",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_control_checkout_blocks(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=self.ROOT,
+            project_root=self.ROOT,
+            current_branch="master",
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertIn("control checkout", result["reasons"][0])
+
+    def test_drifted_control_branch_blocks(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=self.ROOT,
+            project_root=self.ROOT,
+            current_branch="feat/some-task",
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("drift" in r for r in result["reasons"]))
+
+    def test_branches_worktree_as_project_root_passes(self):
+        """MCP launched from a branches/ worktree uses that path as PROJECT_ROOT."""
+        worktree_root = f"{self.ROOT}/branches/issue-274"
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=worktree_root,
+            project_root=worktree_root,
+            current_branch="feat/issue-274-branches-only-worktrees",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+
+class TestPreflightIntegration(unittest.TestCase):
+    def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self):
+        import mcp_server
+
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "author"
+        control_root = "/repo/Gitea-Tools"
+        with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
+            with mock.patch.dict(
+                "os.environ",
+                {"GITEA_TEST_PORCELAIN": ""},
+                clear=False,
+            ):
+                with self.assertRaises(RuntimeError) as ctx:
+                    mcp_server.verify_preflight_purity()
+        self.assertIn("Branches-only mutation guard", str(ctx.exception))
+
+    def test_verify_preflight_allows_branches_worktree(self):
+        import mcp_server
+
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "author"
+        worktree = "/repo/Gitea-Tools/branches/issue-274"
+        with mock.patch.dict(
+            "os.environ",
+            {"GITEA_TEST_PORCELAIN": ""},
+            clear=False,
+        ):
+            mcp_server.verify_preflight_purity(worktree_path=worktree)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_commit_payloads.py

@@ -55,7 +55,15 @@ class TestCommitPayloads(unittest.TestCase):
         with open(self.config_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(CONFIG))
 
-        self.locked_worktree_dir = tempfile.TemporaryDirectory()
+        repo_root = os.path.realpath(
+            os.path.join(os.path.dirname(__file__), os.pardir)
+        )
+        branches_root = os.path.join(repo_root, "branches")
+        os.makedirs(branches_root, exist_ok=True)
+        self.locked_worktree_dir = tempfile.TemporaryDirectory(
+            dir=branches_root,
+            prefix="test-commit-payloads-",
+        )
         self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
 
         self.lock_file_path = "/tmp/gitea_issue_lock.json"
@@ -94,6 +102,7 @@ class TestCommitPayloads(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TEST_PORCELAIN": "",
+            "GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
         }
 
     @patch("mcp_server.api_request")

tests/test_llm_workflow_split.py

@@ -201,6 +201,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(