feat: require validation cwd and HEAD proof in reviewer reports (Closes #398)

Add assess_validation_cwd_proof_report so reviewer final reports must prove
pwd, git rev-parse HEAD, git status, and expected PR head SHA before
claiming validation ran in a branches/ worktree. Wire the verifier into
final_report_validator and gitea_validate_review_final_report.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

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/llm-workflow-runbooks.md

@@ -273,7 +273,20 @@ Never create a parallel branch or PR for the same issue unless the old branch
 is proven abandoned and the takeover is recorded.
 
 Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
-mutations), `status:in-progress`, and claim comments. Full portable wording:
+mutations), `status:in-progress`, and claim comments. `gitea_lock_issue`
+records an `author_issue_work` lease in the issue-lock payload with issue
+number, optional PR number, branch, worktree path, claimant identity/profile,
+created timestamp, expiry timestamp, and last heartbeat timestamp. An active
+same-issue/same-operation lease blocks duplicate work. An expired lease still
+blocks takeover until a recovery review records why the prior work is abandoned,
+completed, or unsafe to continue.
+
+Remote branches matching the issue number are also treated as active work unless
+the recovery review proves the branch is abandoned or superseded. Never delete
+or clean up a branch when it has an active lease, dirty worktree, open PR, or is
+the only copy of unmerged work.
+
+Full portable wording:
 [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 
 ## Global LLM Worktree Rule

docs/wiki/Workflow.md

@@ -17,6 +17,10 @@
 0. Work Selection Rule — verify a work lease before any mutations (open PRs,
    issue-linked PRs, branches, worktrees, dirty worktrees, active leases/
    handoffs, merged-PR completion). Stop if another session owns the lease.
+   `gitea_lock_issue` records an operation-scoped `author_issue_work` lease
+   with issue, branch, worktree, claimant, created, expiry, and heartbeat
+   fields; active or expired same-operation leases require recovery review
+   before takeover.
 0b. Global LLM Worktree Rule — main checkout on `master`/`main`/`dev` only;
    mutate only from a `branches/` worktree after proving root, cwd, branch,
    stable main-checkout branch, and session worktree path (no exceptions).
@@ -31,4 +35,4 @@
 Wiki publication, credential provisioning, and cross-repo mirror operations are
 operator-confirmed actions — see [Runbooks](Runbooks.md).
 
-Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.
+Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.

final_report_validator.py

@@ -92,7 +92,8 @@ _ALREADY_LANDED_GATE_FIRED_RE = re.compile(
     re.IGNORECASE,
 )
 _ELIGIBLE_REVIEW_RE = re.compile(
-    r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
+    r"eligible for (?:review|merge)|ready for (?:review|merge)|"
+    r"(?:next|oldest)\s+eligible\s+pr",
     re.IGNORECASE,
 )
 _APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
@@ -121,11 +122,37 @@ _RECONCILE_STALE_FIELDS = (
     "scratch worktree used",
     "review decision",
     "merge result",
+    "issue lock proof",
+    "claim/comment status",
+    "workspace mutations",
 )
+# Issue #307: reconciliation handoffs must carry the full canonical schema.
 _RECONCILE_REQUIRED_FIELDS = (
+    "Task",
+    "Repo",
+    "Role/profile",
+    "Identity",
     "Selected PR",
-    "Eligibility class",
+    "PR live state",
+    "Candidate head SHA",
+    "Target branch",
+    "Target branch SHA",
+    "Ancestor proof",
+    "Linked issue",
     "Linked issue live status",
+    "Eligibility class",
+    "Capabilities proven",
+    "Missing capabilities",
+    "PR comments posted",
+    "Issue comments posted",
+    "PRs closed",
+    "Issues closed",
+    "Git ref mutations",
+    "Worktree mutations",
+    "MCP/Gitea mutations",
+    "External-state mutations",
+    "Read-only diagnostics",
+    "Blocker",
     "Safe next action",
     "No review/merge confirmation",
 )
@@ -414,19 +441,57 @@ def _rule_reviewer_git_fetch_readonly(
                 "classify git fetch under Git ref mutations, not read-only diagnostics",
             )
         ]
-    if not git_ref_reported and _GIT_FETCH_RE.search(text):
+    if not git_ref_reported:
+        # Issue #307: an observed git fetch (report text or action log)
+        # must be listed under Git ref mutations.
         return [
             validator_finding(
                 "reviewer.git_fetch_readonly",
                 "downgrade",
                 "Git ref mutations",
-                "git fetch mentioned without Git ref mutation classification",
+                "git fetch occurred without Git ref mutation classification",
                 "record git fetch under Git ref mutations",
             )
         ]
     return []
 
 
+def _rule_reviewer_validation_cwd_proof(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+) -> list[dict[str, str]]:
+    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report
+
+    session = validation_session or {}
+    claims = (
+        session.get("validation_ran")
+        or session.get("command")
+        or session.get("baseline_validation_ran")
+    )
+    if not claims and "validation command:" not in (report_text or "").lower():
+        return []
+
+    result = assess_validation_cwd_proof_report(
+        report_text,
+        validation_session=session,
+    )
+    if result.get("proven") or not result.get("claims_validation"):
+        return []
+    severity = "block" if result.get("violations") else "downgrade"
+    return [
+        validator_finding(
+            "reviewer.validation_cwd_proof",
+            severity,
+            "Validation cwd/HEAD proof",
+            reason,
+            result.get("safe_next_action")
+            or "document pwd, HEAD SHA, and explicit cwd before validation",
+        )
+        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
+    ]
+
+
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -668,10 +733,18 @@ def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]
     return []
 
 
-def _rule_reconcile_stale_author_fields(report_text: str) -> list[dict[str, str]]:
+def _rule_reconcile_stale_author_fields(
+    report_text: str,
+    *,
+    session_pr_opened: bool = False,
+) -> list[dict[str, str]]:
     text = (report_text or "").lower()
     findings = []
     for field in _RECONCILE_STALE_FIELDS:
+        if field == "pr number opened" and session_pr_opened:
+            # Issue #307: allowed only when a PR was actually opened
+            # in this session.
+            continue
         if f"{field}:" in text:
             findings.append(
                 validator_finding(
@@ -827,6 +900,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_mutation_categories,
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_validation_command,
+        _rule_reviewer_validation_cwd_proof,
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
@@ -932,6 +1006,8 @@ def assess_final_report_validator(
     mutations_observed: bool = False,
     local_edits: bool = False,
     issue_filing_lock: dict | None = None,
+    session_pr_opened: bool = False,
+    validation_session: dict | None = None,
 ) -> dict[str, Any]:
     """Validate final-report text against task-specific proof rules (#327).
 
@@ -986,6 +1062,8 @@ def assess_final_report_validator(
         "action_log": action_log,
         "mutations_observed": mutations_observed,
         "local_edits": local_edits,
+        "session_pr_opened": session_pr_opened,
+        "validation_session": validation_session,
     }
 
     for rule in _RULES_BY_TASK.get(normalized_kind, ()):

gitea_mcp_server.py

@@ -21,6 +21,7 @@ import json
 import functools
 import contextlib
 import subprocess
+from datetime import datetime, timedelta, timezone
 
 
 # Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
@@ -397,6 +398,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 +449,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
 
@@ -477,8 +501,10 @@ 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 reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
 
@@ -486,6 +512,132 @@ import native_mcp_preference  # noqa: E402
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
 # consumed by gitea_create_pr and scripts/worktree-start.
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+WORK_LEASE_TTL_HOURS = 4
+AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
+VALID_WORK_LEASE_OPERATIONS = frozenset({
+    AUTHOR_ISSUE_WORK_LEASE,
+    "review_pr_work",
+    "fix_pr_changes",
+    "cleanup_branch_work",
+    "issue_filing_work",
+    "recovery_work",
+})
+
+
+def _work_lease_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _work_lease_timestamp(value: datetime) -> str:
+    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    try:
+        return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except ValueError:
+        return None
+
+
+def _load_existing_issue_lock() -> dict | None:
+    if not os.path.exists(ISSUE_LOCK_FILE):
+        return None
+    try:
+        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
+            data = json.load(f)
+        return data if isinstance(data, dict) else None
+    except Exception:
+        return None
+
+
+def _work_lease_claimant(host: str | None) -> dict:
+    profile = get_profile()
+    username = _IDENTITY_CACHE.get(host) if host else None
+    if username is None and host and not _preflight_in_test_mode():
+        username = _authenticated_username(host)
+    return {
+        "username": username,
+        "profile": profile.get("profile_name"),
+    }
+
+
+def _build_author_issue_work_lease(
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    host: str | None,
+) -> dict:
+    created = _work_lease_now()
+    expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
+    return {
+        "operation_type": AUTHOR_ISSUE_WORK_LEASE,
+        "issue_number": issue_number,
+        "pr_number": None,
+        "branch": branch_name,
+        "worktree_path": worktree_path,
+        "claimant": _work_lease_claimant(host),
+        "created_at": _work_lease_timestamp(created),
+        "expires_at": _work_lease_timestamp(expires),
+        "last_heartbeat_at": _work_lease_timestamp(created),
+    }
+
+
+def _active_work_lease_block(
+    existing_lock: dict | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    operation_type: str,
+) -> str | None:
+    if not existing_lock:
+        return None
+    existing_issue = existing_lock.get("issue_number")
+    existing_branch = existing_lock.get("branch_name")
+    existing_worktree = existing_lock.get("worktree_path")
+    lease = existing_lock.get("work_lease")
+    existing_operation = (
+        lease.get("operation_type")
+        if isinstance(lease, dict)
+        else AUTHOR_ISSUE_WORK_LEASE
+    )
+    if existing_issue != issue_number or existing_operation != operation_type:
+        return None
+
+    same_owner = (
+        existing_branch == branch_name
+        and os.path.realpath(str(existing_worktree or "")) == os.path.realpath(worktree_path)
+    )
+    expires_at = (
+        _parse_work_lease_timestamp(lease.get("expires_at"))
+        if isinstance(lease, dict)
+        else None
+    )
+    if expires_at and expires_at <= _work_lease_now():
+        return (
+            f"Issue #{issue_number} has an expired {operation_type} lease on "
+            f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
+            "Recovery review is required before takeover (fail closed)"
+        )
+    if same_owner:
+        return None
+    return (
+        f"Issue #{issue_number} already has an active {operation_type} lease on "
+        f"branch '{existing_branch}' from worktree '{existing_worktree}' "
+        "(fail closed)"
+    )
+
+
+def _branch_entry_name(branch: dict | str) -> str:
+    if isinstance(branch, str):
+        return branch
+    if not isinstance(branch, dict):
+        return ""
+    return str(branch.get("name") or branch.get("ref") or "")
 
 
 def _reveal_endpoints() -> bool:
@@ -933,6 +1085,16 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, PROJECT_ROOT
     )
+    active_lease_block = _active_work_lease_block(
+        _load_existing_issue_lock(),
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+    )
+    if active_lease_block:
+        raise RuntimeError(active_lease_block)
+
     git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
     verify_preflight_purity(remote, worktree_path=resolved_worktree)
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
@@ -978,6 +1140,25 @@ def gitea_lock_issue(
                 f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
             )
 
+    branch_url = f"{repo_api_url(h, o, r)}/branches"
+    try:
+        branches = api_get_all(branch_url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
+    for branch in branches:
+        name = _branch_entry_name(branch)
+        if expected_pattern in name:
+            raise ValueError(
+                f"Issue #{issue_number} already has matching branch '{name}' "
+                "(fail closed)"
+            )
+
+    work_lease = _build_author_issue_work_lease(
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        host=h,
+    )
     data = {
         "issue_number": issue_number,
         "branch_name": branch_name,
@@ -985,6 +1166,7 @@ def gitea_lock_issue(
         "org": o,
         "repo": r,
         "worktree_path": resolved_worktree,
+        "work_lease": work_lease,
     }
 
     try:
@@ -1005,6 +1187,7 @@ def gitea_lock_issue(
         "issue_number": issue_number,
         "branch_name": branch_name,
         "worktree_path": resolved_worktree,
+        "work_lease": work_lease,
     }
     if agent_artifacts:
         result["warnings"] = [
@@ -3429,6 +3612,178 @@ def gitea_reconcile_merged_cleanups(
     return {"success": True, "performed": True, **report}
 
 
+@mcp.tool()
+def gitea_assess_already_landed_reconciliation(
+    pr_number: int,
+    target_branch: str = "master",
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: assess already-landed reconciliation for one open PR (#301).
+
+    Fetches the PR live, refreshes the target branch, checks ancestor proof,
+    and returns a capability plan for the active profile. Does not review,
+    approve, request changes, merge, or mutate Gitea state.
+
+    Args:
+        pr_number: Open PR to assess.
+        target_branch: Branch used for ancestry proof (default master).
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with eligibility proof, capability plan, and safe next action.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
+
+    assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+        pr=pr,
+        project_root=PROJECT_ROOT,
+        remote=remote,
+        target_branch=target_branch,
+    )
+
+    profile = get_profile()
+    capabilities = reconciliation_workflow.profile_reconciliation_capabilities(
+        profile.get("allowed_operations"),
+        profile.get("forbidden_operations"),
+    )
+    plan = reconciliation_workflow.resolve_reconciliation_plan(
+        assessment=assessment,
+        capabilities=capabilities,
+    )
+
+    return {
+        "success": True,
+        "performed": False,
+        "pr_number": pr_number,
+        "assessment": assessment,
+        "capabilities": capabilities,
+        "plan": plan,
+        "workflow_source": "skills/llm-project-workflow/workflows/reconcile-landed-pr.md",
+        "task_mode": "reconcile-landed-pr",
+        "review_merge_allowed": False,
+    }
+
+
+@mcp.tool()
+def gitea_scan_already_landed_open_prs(
+    target_branch: str = "master",
+    limit: int = 50,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: list open PRs already landed on the target branch (#301).
+
+    Traverses open PR inventory with pagination proof, assesses each candidate,
+    and returns PRs classified as ``ALREADY_LANDED_RECONCILE_REQUIRED``. Does
+    not review, approve, merge, or mutate Gitea state.
+
+    Args:
+        target_branch: Branch used for ancestry proof (default master).
+        limit: Max open PRs to inspect across pages.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with candidates, pagination metadata, and target branch SHA proof.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    open_prs: list[dict] = []
+    current_page = 1
+    pages_fetched = 0
+    last_meta: dict = {}
+    per_page = min(50, max(1, int(limit)))
+    while pages_fetched < 100 and len(open_prs) < limit:
+        raw_page, last_meta = api_fetch_page(
+            url, auth, page=current_page, limit=per_page
+        )
+        pages_fetched += 1
+        remaining = limit - len(open_prs)
+        open_prs.extend(raw_page[:remaining])
+        if last_meta["is_final_page"] or len(open_prs) >= limit:
+            break
+        current_page += 1
+
+    pagination = {
+        "page": 1,
+        "per_page": per_page,
+        "returned_count": len(open_prs),
+        "has_more": False,
+        "next_page": None,
+        "is_final_page": bool(last_meta.get("is_final_page")),
+        "pages_fetched": pages_fetched,
+        "inventory_complete": bool(last_meta.get("is_final_page")),
+        "total_count": len(open_prs) if last_meta.get("is_final_page") else None,
+    }
+
+    target_fetch = reconciliation_workflow.fetch_target_branch(
+        PROJECT_ROOT, remote, target_branch
+    )
+
+    candidates: list[dict] = []
+    for pr in open_prs:
+        assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+            pr=pr,
+            project_root=PROJECT_ROOT,
+            remote=remote,
+            target_branch=target_branch,
+            target_fetch=target_fetch,
+        )
+        if assessment.get("eligibility_class") == (
+            reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED
+        ):
+            candidates.append(assessment)
+
+    return {
+        "success": True,
+        "performed": False,
+        "target_branch": target_branch,
+        "target_branch_sha": target_fetch.get("target_branch_sha"),
+        "git_ref_mutations": (
+            [target_fetch["git_fetch_command"]]
+            if target_fetch.get("git_fetch_command")
+            else []
+        ),
+        "candidates": candidates,
+        "candidate_count": len(candidates),
+        "pagination": pagination,
+        "task_mode": "reconcile-landed-pr",
+        "review_merge_allowed": False,
+    }
+
+
 @mcp.tool()
 def gitea_reconcile_already_landed_pr(
     pr_number: int,
@@ -3608,7 +3963,6 @@ def gitea_reconcile_already_landed_pr(
 
     return result
 
-
 @mcp.tool()
 def gitea_close_issue(
     issue_number: int,
@@ -4391,6 +4745,28 @@ _PROJECT_SKILLS = {
             "Do not merge unless the operator explicitly authorizes it.",
         ],
     },
+    "gitea-reconcile-landed-pr": {
+        "description": "Reconcile open PRs whose heads are already on the "
+                       "target branch without review or merge.",
+        "when_to_use": "An open PR blocks the review queue but its head SHA "
+                       "is already an ancestor of master; use the dedicated "
+                       "reconciliation path instead of review/merge.",
+        "required_operations": ["gitea.read"],
+        "status": "available",
+        "notes": "Load skills/llm-project-workflow/workflows/reconcile-landed-pr.md "
+                 "first. PR close requires exact gitea.pr.close; review/merge "
+                 "capabilities must not be used on this path.",
+        "steps": [
+            "Resolve task: gitea_resolve_task_capability(task='reconcile_landed_pr').",
+            "Load canonical workflow via mcp_get_skill_guide or "
+            "skills/llm-project-workflow/workflows/reconcile-landed-pr.md.",
+            "Scan queue: gitea_scan_already_landed_open_prs (pagination proof).",
+            "Assess one PR: gitea_assess_already_landed_reconciliation.",
+            "Follow the plan outcome: full close only with gitea.pr.close, "
+            "comment-then-stop when close is missing, recovery handoff otherwise.",
+            "Never approve, request changes, or merge on this path.",
+        ],
+    },
     "gitea-pr-merge": {
         "description": "Gated merge of an approved pull request.",
         "when_to_use": "Reviewer/merger profile with explicit operator "
@@ -4915,6 +5291,7 @@ _RUNTIME_CAPABILITY_TASKS = (
     "create_issue",
     "comment_issue",
     "create_pr",
+    "work_issue",
     "review_pr",
     "merge_pr",
     "close_issue",
@@ -4967,6 +5344,7 @@ def _build_runtime_task_capabilities(
         "create_issue": "can_create_issues",
         "comment_issue": "can_comment_on_issues",
         "create_pr": "can_author_prs",
+        "work_issue": "can_work_issues",
         "review_pr": "can_review_prs",
         "merge_pr": "can_merge_prs",
         "close_issue": "can_close_issues",
@@ -5119,6 +5497,37 @@ def gitea_get_shell_health() -> dict:
     return native_mcp_preference.shell_health_status()
 
 
+@mcp.tool()
+def gitea_validate_review_final_report(
+    report_text: str,
+    review_decision_lock: dict | None = None,
+    linked_issue_lock: dict | None = None,
+    validation_report: dict | None = None,
+    action_log: list[dict] | None = None,
+    mutations_observed: bool = False,
+    local_edits: bool = False,
+    validation_session: dict | None = None,
+) -> dict:
+    """Read-only: machine-validate reviewer final report schema before output (#391).
+
+    Composes the composable final-report validator with review-specific schema
+    gates (legacy fields, mutation categories, merge/issue claims, blocked
+    handoff replay, and narrative/handoff consistency).
+    """
+    from review_final_report_schema import assess_review_final_report_schema
+
+    return assess_review_final_report_schema(
+        report_text,
+        review_decision_lock=review_decision_lock,
+        linked_issue_lock=linked_issue_lock,
+        validation_report=validation_report,
+        action_log=action_log,
+        mutations_observed=mutations_observed,
+        local_edits=local_edits,
+        validation_session=validation_session,
+    )
+
+
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",

reconciliation_workflow.py

@@ -0,0 +1,292 @@
+"""Already-landed PR reconciliation workflow helpers (#301).
+
+Read-only assessment and capability planning for reconciling open PRs whose
+heads are already ancestors of the target branch. Does not invoke review or
+merge paths.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from typing import Any
+
+import gitea_config
+from merged_cleanup_reconcile import extract_linked_issue, is_head_ancestor_of_ref
+
+ELIGIBILITY_ALREADY_LANDED = "ALREADY_LANDED_RECONCILE_REQUIRED"
+ELIGIBILITY_NOT_LANDED = "NOT_ALREADY_LANDED"
+ELIGIBILITY_STALE_TARGET = "TARGET_BRANCH_UNVERIFIED"
+ELIGIBILITY_PR_NOT_OPEN = "PR_NOT_OPEN"
+
+OUTCOME_FULL_RECONCILE = "FULL_RECONCILE_CLOSE_ALLOWED"
+OUTCOME_PARTIAL_COMMENT = "PARTIAL_RECONCILE_COMMENT_THEN_STOP"
+OUTCOME_RECOVERY_HANDOFF = "RECOVERY_HANDOFF_ONLY"
+OUTCOME_NOT_LANDED = "NOT_LANDED_NO_ACTION"
+OUTCOME_GATE_NOT_PROVEN = "GATE_NOT_PROVEN"
+
+RECONCILE_WORKFLOW_MARKERS = (
+    "workflows/reconcile-landed-pr.md",
+    "reconcile-landed-pr.md",
+)
+
+RECONCILE_TASK_MARKERS = (
+    "reconcile-landed-pr",
+    "reconcile already-landed",
+    "reconcile_already_landed",
+)
+
+
+def fetch_target_branch(project_root: str, remote: str, branch: str) -> dict[str, Any]:
+    """Fetch *branch* from *remote* and return the resolved SHA."""
+    ref = f"{remote}/{branch}"
+    fetch = subprocess.run(
+        ["git", "-C", project_root, "fetch", remote, branch],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if fetch.returncode != 0:
+        return {
+            "success": False,
+            "target_branch": branch,
+            "target_ref": ref,
+            "target_branch_sha": None,
+            "reasons": [
+                f"git fetch {remote} {branch} failed: "
+                f"{(fetch.stderr or fetch.stdout or '').strip()}"
+            ],
+        }
+
+    rev = subprocess.run(
+        ["git", "-C", project_root, "rev-parse", ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if rev.returncode != 0:
+        return {
+            "success": False,
+            "target_branch": branch,
+            "target_ref": ref,
+            "target_branch_sha": None,
+            "reasons": [
+                f"git rev-parse {ref} failed: "
+                f"{(rev.stderr or rev.stdout or '').strip()}"
+            ],
+        }
+
+    return {
+        "success": True,
+        "target_branch": branch,
+        "target_ref": ref,
+        "target_branch_sha": (rev.stdout or "").strip(),
+        "reasons": [],
+        "git_fetch_command": f"git fetch {remote} {branch}",
+    }
+
+
+def profile_reconciliation_capabilities(
+    allowed: list[str] | None,
+    forbidden: list[str] | None,
+) -> dict[str, bool]:
+    """Map active profile operations to reconciliation capabilities."""
+    allowed = allowed or []
+    forbidden = forbidden or []
+
+    def can(op: str) -> bool:
+        return gitea_config.check_operation(op, allowed, forbidden)[0]
+
+    return {
+        "read": can("gitea.read"),
+        "comment_pr": can("gitea.pr.comment"),
+        "comment_issue": can("gitea.issue.comment"),
+        "close_pr": can("gitea.pr.close"),
+        "close_issue": can("gitea.issue.close"),
+        "review_pr": can("gitea.pr.review") or can("gitea.pr.approve"),
+        "merge_pr": can("gitea.pr.merge"),
+    }
+
+
+def assess_open_pr_reconciliation(
+    *,
+    pr: dict[str, Any],
+    project_root: str,
+    remote: str,
+    target_branch: str,
+    target_fetch: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Assess whether an open PR is eligible for already-landed reconciliation."""
+    pr_number = int(pr.get("number") or 0)
+    pr_state = (pr.get("state") or "").strip().lower()
+    head = pr.get("head") or {}
+    head_sha = head.get("sha") if isinstance(head, dict) else None
+    head_ref = head.get("ref") if isinstance(head, dict) else None
+    base = pr.get("base") or {}
+    base_ref = base.get("ref") if isinstance(base, dict) else None
+    title = pr.get("title") or ""
+    body = pr.get("body") or ""
+
+    fetch_result = target_fetch or fetch_target_branch(
+        project_root, remote, target_branch
+    )
+    linked_issue = extract_linked_issue(title, body)
+
+    result: dict[str, Any] = {
+        "pr_number": pr_number,
+        "pr_state": pr_state,
+        "candidate_head_sha": head_sha,
+        "head_ref": head_ref,
+        "base_ref": base_ref or target_branch,
+        "target_branch": target_branch,
+        "target_branch_sha": fetch_result.get("target_branch_sha"),
+        "linked_issue": linked_issue,
+        "git_ref_mutations": [],
+        "reasons": [],
+        "review_merge_allowed": False,
+    }
+    if fetch_result.get("git_fetch_command"):
+        result["git_ref_mutations"].append(fetch_result["git_fetch_command"])
+
+    if pr_state != "open":
+        result["eligibility_class"] = ELIGIBILITY_PR_NOT_OPEN
+        result["ancestor_proof"] = None
+        result["reconciliation_allowed"] = False
+        result["reasons"].append(f"PR #{pr_number} state is {pr_state!r}, not open")
+        return result
+
+    if not fetch_result.get("success"):
+        result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
+        result["ancestor_proof"] = None
+        result["reconciliation_allowed"] = False
+        result["reasons"].extend(fetch_result.get("reasons") or [])
+        return result
+
+    target_ref = fetch_result.get("target_ref") or f"{remote}/{target_branch}"
+    ancestor = is_head_ancestor_of_ref(project_root, head_sha, target_ref)
+    result["ancestor_proof"] = ancestor
+
+    if ancestor is None:
+        result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
+        result["reconciliation_allowed"] = False
+        result["reasons"].append(
+            f"ancestor check failed for head {head_sha!r} against {target_ref}"
+        )
+        return result
+
+    if ancestor:
+        result["eligibility_class"] = ELIGIBILITY_ALREADY_LANDED
+        result["reconciliation_allowed"] = True
+        return result
+
+    result["eligibility_class"] = ELIGIBILITY_NOT_LANDED
+    result["reconciliation_allowed"] = False
+    result["reasons"].append(
+        f"PR head {head_sha} is not an ancestor of {target_ref}"
+    )
+    return result
+
+
+def resolve_reconciliation_plan(
+    *,
+    assessment: dict[str, Any],
+    capabilities: dict[str, bool] | None,
+) -> dict[str, Any]:
+    """Map eligibility + profile capabilities to a reconciliation outcome."""
+    caps = capabilities or {}
+    reasons: list[str] = []
+
+    if not assessment.get("reconciliation_allowed"):
+        outcome = OUTCOME_NOT_LANDED
+        if assessment.get("eligibility_class") in {
+            ELIGIBILITY_STALE_TARGET,
+            ELIGIBILITY_PR_NOT_OPEN,
+        }:
+            outcome = OUTCOME_GATE_NOT_PROVEN
+        reasons.extend(assessment.get("reasons") or [])
+        return {
+            "outcome": outcome,
+            "close_pr_allowed": False,
+            "comment_pr_allowed": False,
+            "close_issue_allowed": False,
+            "comment_issue_allowed": False,
+            "review_merge_allowed": False,
+            "missing_capabilities": _missing_reconciliation_capabilities(caps),
+            "safe_next_action": (
+                "Do not reconcile via review/merge; repair missing proof first."
+            ),
+            "reasons": reasons,
+        }
+
+    close_pr = bool(caps.get("close_pr"))
+    comment_pr = bool(caps.get("comment_pr"))
+    close_issue = bool(caps.get("close_issue"))
+    comment_issue = bool(caps.get("comment_issue"))
+
+    if caps.get("review_pr") or caps.get("merge_pr"):
+        reasons.append(
+            "reconciliation path must not use review/merge capabilities"
+        )
+
+    if close_pr:
+        outcome = OUTCOME_FULL_RECONCILE
+        safe_next_action = (
+            "Run reconciliation close via exact gitea.pr.close after proof; "
+            "do not approve or merge."
+        )
+    elif comment_pr:
+        outcome = OUTCOME_PARTIAL_COMMENT
+        safe_next_action = (
+            "Post one reconciliation comment with ancestor proof, then stop "
+            "for an authorized close profile."
+        )
+    else:
+        outcome = OUTCOME_RECOVERY_HANDOFF
+        safe_next_action = (
+            "Produce a recovery handoff naming missing gitea.pr.close and/or "
+            "gitea.pr.comment; do not loop through review/merge."
+        )
+        reasons.append("gitea.pr.close is not available in the active profile")
+
+    return {
+        "outcome": outcome,
+        "close_pr_allowed": close_pr,
+        "comment_pr_allowed": comment_pr,
+        "close_issue_allowed": close_issue,
+        "comment_issue_allowed": comment_issue,
+        "review_merge_allowed": False,
+        "missing_capabilities": _missing_reconciliation_capabilities(caps),
+        "safe_next_action": safe_next_action,
+        "reasons": reasons,
+    }
+
+
+def _missing_reconciliation_capabilities(caps: dict[str, bool]) -> list[str]:
+    missing = []
+    if not caps.get("read"):
+        missing.append("gitea.read")
+    if not caps.get("close_pr"):
+        missing.append("gitea.pr.close")
+    if not caps.get("comment_pr"):
+        missing.append("gitea.pr.comment")
+    return missing
+
+
+def assess_reconcile_workflow_source(report_text: str) -> dict[str, Any]:
+    """Reconciliation reports must cite the canonical workflow file."""
+    lower = (report_text or "").lower()
+    reasons = []
+    if not any(marker in lower for marker in RECONCILE_WORKFLOW_MARKERS):
+        reasons.append(
+            "reconciliation report missing workflow source "
+            "(workflows/reconcile-landed-pr.md)"
+        )
+    if not any(marker in lower for marker in RECONCILE_TASK_MARKERS):
+        reasons.append(
+            "reconciliation report missing task mode declaration "
+            "(reconcile-landed-pr)"
+        )
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }

review_final_report_schema.py

@@ -0,0 +1,321 @@
+"""Reviewer final-report schema verification before session output (#391).
+
+Composes the composable ``assess_final_report_validator`` (#327) with
+review-specific schema gates required before a reviewer session completes.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from final_report_validator import (
+    assess_final_report_validator,
+    validator_finding,
+    _handoff_fields,
+)
+
+_LEGACY_STALE_FIELDS = (
+    "pinned reviewed head",
+    "scratch worktree used",
+    "workspace mutations",
+)
+
+_REVIEWED_HEAD_RE = re.compile(
+    r"(?:pinned reviewed head|reviewed head sha)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_VALIDATION_PASS_RE = re.compile(
+    r"validation\s*:\s*(?:pass|passed|strong|ok|green)",
+    re.IGNORECASE,
+)
+_MERGED_CLAIM_RE = re.compile(
+    r"\b(?:merged|merge result\s*:\s*merged|pr\s+#\d+\s+merged)\b",
+    re.IGNORECASE,
+)
+_MERGE_RESULT_RE = re.compile(r"merge result\s*:", re.IGNORECASE)
+_ANCESTRY_PROOF_RE = re.compile(
+    r"(?:ancestor proof|target branch sha|merge commit)",
+    re.IGNORECASE,
+)
+_ISSUE_CLOSED_RE = re.compile(
+    r"(?:linked issue(?:\s+live)?\s+status\s*:\s*closed|issue\s+#\d+\s+closed)",
+    re.IGNORECASE,
+)
+_LIVE_ISSUE_PROOF_RE = re.compile(
+    r"gitea_view_issue|(?:issue|linked issue).{0,40}fetched live|live fetch proof",
+    re.IGNORECASE,
+)
+_FULL_SUITE_PASS_RE = re.compile(
+    r"(?:full suite passed|full test suite passed|\d+\s+passed,\s*0\s+failed)",
+    re.IGNORECASE,
+)
+_TESTS_IGNORED_RE = re.compile(
+    r"(?:ignored tests|tests?\s+ignored|skipped tests|not run|tests?\s+skipped)",
+    re.IGNORECASE,
+)
+_BLOCKED_HANDOFF_RE = re.compile(
+    r"(?:blocker\s*:|recovery handoff|infra_stop|capability stop|mutation blocked)",
+    re.IGNORECASE,
+)
+_REPLAY_COMMAND_RE = re.compile(
+    r"(?:gitea_submit_pr_review|gitea_merge_pr|gitea_review_pr|"
+    r"submitted\s+['\"]approve['\"]|replay approve|replay merge)",
+    re.IGNORECASE,
+)
+_PENDING_RE = re.compile(r"\bPENDING\b", re.IGNORECASE)
+_APPROVED_RE = re.compile(
+    r"(?:review decision\s*:\s*approve|submitted\s+approve|official review submitted)",
+    re.IGNORECASE,
+)
+_DRY_RUN_RE = re.compile(r"dry[- ]run", re.IGNORECASE)
+_FINDING_READY_RE = re.compile(
+    r"(?:finding ready|ready to submit|not yet submitted)",
+    re.IGNORECASE,
+)
+_NARRATIVE_APPROVE_RE = re.compile(
+    r"(?:^|\n)\s*(?:##\s+)?(?:summary|verdict|recommendation)\b[^\n]*\bapprove\b",
+    re.IGNORECASE | re.MULTILINE,
+)
+_HANDOFF_DECISION_RE = re.compile(
+    r"review decision\s*:\s*(\w+)",
+    re.IGNORECASE,
+)
+
+
+def _rule_legacy_stale_fields(report_text: str) -> list[dict[str, str]]:
+    fields = _handoff_fields(report_text)
+    findings: list[dict[str, str]] = []
+    for stale in _LEGACY_STALE_FIELDS:
+        if stale in fields and fields[stale].lower() not in {"", "none", "n/a"}:
+            findings.append(
+                validator_finding(
+                    "reviewer.legacy_stale_field",
+                    "block",
+                    stale.title(),
+                    f"legacy or stale handoff field '{stale}' must not appear in "
+                    "canonical reviewer final reports",
+                    "remove stale fields; use Candidate head SHA and precise "
+                    "mutation categories instead",
+                )
+            )
+    return findings
+
+
+def _rule_reviewed_head_without_validation(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _REVIEWED_HEAD_RE.search(text):
+        return []
+    if _VALIDATION_PASS_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "reviewer.reviewed_head_without_validation",
+            "block",
+            "Pinned reviewed head",
+            "reviewed/pinned head SHA reported without validation pass proof",
+            "document validation command, cwd, and pass result before pinning head SHA",
+        )
+    ]
+
+
+def _rule_merged_without_proof(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _MERGED_CLAIM_RE.search(text):
+        return []
+    if _MERGE_RESULT_RE.search(text) and _ANCESTRY_PROOF_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "reviewer.merged_without_proof",
+            "block",
+            "Merge result",
+            "merged outcome claimed without merge result and target ancestry proof",
+            "include Merge result, target branch SHA, and ancestor proof before claiming merged",
+        )
+    ]
+
+
+def _rule_issue_closed_without_live_proof(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _ISSUE_CLOSED_RE.search(text):
+        return []
+    fields = _handoff_fields(text)
+    status_value = fields.get("linked issue live status", "")
+    readonly_value = fields.get("read-only diagnostics", "")
+    proof_blob = f"{status_value} {readonly_value}".lower()
+    if _LIVE_ISSUE_PROOF_RE.search(proof_blob):
+        return []
+    return [
+        validator_finding(
+            "reviewer.issue_closed_without_live_proof",
+            "block",
+            "Linked issue",
+            "issue closed claimed without live linked-issue verification proof",
+            "fetch linked issue live (gitea_view_issue) and record Linked issue live status",
+        )
+    ]
+
+
+def _rule_full_suite_pass_ignored_tests(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not (_FULL_SUITE_PASS_RE.search(text) and _TESTS_IGNORED_RE.search(text)):
+        return []
+    if re.search(r"explicitly\s+(?:ignored|skipped)", text, re.IGNORECASE):
+        return []
+    return [
+        validator_finding(
+            "reviewer.full_suite_pass_ignored_tests",
+            "block",
+            "Validation",
+            "full suite pass claimed while tests were ignored or skipped without "
+            "explicit disclosure",
+            "state which tests were ignored/skipped or downgrade validation verdict",
+        )
+    ]
+
+
+def _rule_blocked_handoff_replay(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _BLOCKED_HANDOFF_RE.search(text):
+        return []
+    if not _REPLAY_COMMAND_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "reviewer.blocked_handoff_replay",
+            "block",
+            "Safe next action",
+            "blocked recovery handoff includes direct approve or merge replay commands",
+            "restart the full workflow after the blocker clears; do not replay mutations",
+        )
+    ]
+
+
+def _rule_narrative_handoff_drift(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    narrative_approve = bool(_NARRATIVE_APPROVE_RE.search(text))
+    decision_match = _HANDOFF_DECISION_RE.search(text)
+    if not narrative_approve or not decision_match:
+        return []
+    handoff_decision = decision_match.group(1).strip().lower()
+    if handoff_decision in {"approve", "approved"}:
+        return []
+    return [
+        validator_finding(
+            "reviewer.narrative_handoff_drift",
+            "block",
+            "Review decision",
+            f"narrative approves PR but controller handoff says '{handoff_decision}'",
+            "align narrative summary with controller handoff Review decision field",
+        )
+    ]
+
+
+def _rule_review_state_ambiguous(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    findings: list[dict[str, str]] = []
+    if _PENDING_RE.search(text) and _APPROVED_RE.search(text):
+        findings.append(
+            validator_finding(
+                "reviewer.pending_vs_approved",
+                "block",
+                "Review decision",
+                "report conflates PENDING review state with APPROVED outcome",
+                "distinguish official submitted review from pending or ready-to-submit state",
+            )
+        )
+    if _DRY_RUN_RE.search(text) and _APPROVED_RE.search(text):
+        if "dry-run only" not in text.lower():
+            findings.append(
+                validator_finding(
+                    "reviewer.dry_run_vs_submitted",
+                    "block",
+                    "Review mutations",
+                    "dry-run language mixed with official approve/submitted claims",
+                    "mark dry-run explicitly or document the live review mutation proof",
+                )
+            )
+    if _FINDING_READY_RE.search(text) and _APPROVED_RE.search(text):
+        findings.append(
+            validator_finding(
+                "reviewer.finding_ready_vs_submitted",
+                "downgrade",
+                "Review decision",
+                "finding-ready wording combined with submitted-approve claims",
+                "state whether review was submitted live or only prepared for submission",
+            )
+        )
+    return findings
+
+
+_SCHEMA_RULES = (
+    _rule_legacy_stale_fields,
+    _rule_reviewed_head_without_validation,
+    _rule_merged_without_proof,
+    _rule_issue_closed_without_live_proof,
+    _rule_full_suite_pass_ignored_tests,
+    _rule_blocked_handoff_replay,
+    _rule_narrative_handoff_drift,
+    _rule_review_state_ambiguous,
+)
+
+
+def _merge_validator_results(
+    base: dict[str, Any],
+    extra_findings: list[dict[str, str]],
+) -> dict[str, Any]:
+    findings = list(base.get("findings") or []) + extra_findings
+    blocked = base.get("blocked") or any(f["severity"] == "block" for f in extra_findings)
+    downgraded = base.get("downgraded") or any(
+        f["severity"] == "downgrade" for f in extra_findings
+    )
+    grade = base.get("grade", "A")
+    if blocked:
+        grade = "blocked"
+    elif downgraded and grade == "A":
+        grade = "downgraded"
+    reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
+    safe_next = base.get("safe_next_action") or "none"
+    if extra_findings:
+        safe_next = extra_findings[0].get("safe_next_action", safe_next)
+    return {
+        **base,
+        "grade": grade,
+        "blocked": blocked,
+        "downgraded": downgraded,
+        "findings": findings,
+        "reasons": reasons,
+        "safe_next_action": safe_next,
+        "complete": grade == "A",
+        "schema_rules_applied": len(_SCHEMA_RULES),
+    }
+
+
+def assess_review_final_report_schema(
+    report_text: str,
+    *,
+    review_decision_lock: dict | None = None,
+    linked_issue_lock: dict | None = None,
+    validation_report: dict | None = None,
+    action_log: list[dict] | None = None,
+    mutations_observed: bool = False,
+    local_edits: bool = False,
+    validation_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate reviewer final report text before session completion (#391)."""
+    base = assess_final_report_validator(
+        report_text,
+        "review_pr",
+        review_decision_lock=review_decision_lock,
+        linked_issue_lock=linked_issue_lock,
+        validation_report=validation_report,
+        action_log=action_log,
+        mutations_observed=mutations_observed,
+        local_edits=local_edits,
+        validation_session=validation_session,
+    )
+    extra: list[dict[str, str]] = []
+    for rule in _SCHEMA_RULES:
+        extra.extend(rule(report_text))
+    return _merge_validator_results(base, extra)

review_proofs.py

@@ -18,6 +18,11 @@ import hashlib
 import re
 
 import issue_duplicate_gate
+from reconciliation_workflow import (
+    RECONCILE_TASK_MARKERS,
+    RECONCILE_WORKFLOW_MARKERS,
+    assess_reconcile_workflow_source,
+)
 from reviewer_worktree import assess_reviewer_worktree_proof
 
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
@@ -1742,6 +1747,17 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reasons": [],
             "violations": [],
         }
+    if report_text:
+        already_landed_classification = assess_already_landed_classification_report(
+            report_text
+        )
+    else:
+        already_landed_classification = {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "violations": [],
+        }
 
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
@@ -1921,6 +1937,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reviewer baseline validation proof missing or failed (#325)"
         )
         downgrade_reasons.extend(baseline_validation.get("reasons", []))
+    if not already_landed_classification.get("proven"):
+        downgrade_reasons.append(
+            "already-landed classification wording missing or invalid (#295)"
+        )
+        downgrade_reasons.extend(already_landed_classification.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -2035,6 +2056,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "baseline_validation_violations": list(
             baseline_validation.get("violations") or []
         ),
+        "already_landed_classification_proven": bool(
+            already_landed_classification.get("proven")
+        ),
+        "already_landed_classification_violations": list(
+            already_landed_classification.get("reasons") or []
+        ),
     }
 
 
@@ -2301,6 +2328,30 @@ CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
     ("Terminal review mutation", ("terminal review mutation",)),
 )
 
+WORK_ISSUE_WORKFLOW_MARKERS = (
+    "workflows/work-issue.md",
+    "work-issue.md",
+)
+
+WORK_ISSUE_TASK_MARKERS = (
+    "work-issue",
+    "work issue",
+)
+
+WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
+    ("Selected PR", ("selected pr",)),
+    ("Review decision", ("review decision",)),
+    ("Merge result", ("merge result",)),
+    ("Merge preflight", ("merge preflight",)),
+    ("Already-landed gate", ("already-landed gate",)),
+    ("Pinned reviewed head", ("pinned reviewed head",)),
+    ("Terminal review mutation", ("terminal review mutation",)),
+    ("Duplicate search terms", ("duplicate search terms",)),
+    ("Issues created", ("issues created",)),
+    ("Issues skipped as duplicates", ("issues skipped as duplicates",)),
+    ("Requested issue task", ("requested issue task",)),
+)
+
 
 def _handoff_section_lines(report_text):
     """Return the lines of the Controller Handoff section, or None."""
@@ -3493,6 +3544,104 @@ def assess_create_issue_workflow_source(report_text: str) -> dict:
     }
 
 
+def assess_work_issue_workflow_source(report_text: str) -> dict:
+    """#139: work-issue reports must cite the canonical workflow file."""
+    lower = (report_text or "").lower()
+    reasons = []
+    if not any(marker in lower for marker in WORK_ISSUE_WORKFLOW_MARKERS):
+        reasons.append(
+            "work-issue report missing workflow source "
+            "(workflows/work-issue.md)"
+        )
+    if not any(marker in lower for marker in WORK_ISSUE_TASK_MARKERS):
+        reasons.append(
+            "work-issue report missing task mode declaration (work-issue)"
+        )
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_work_issue_mode_isolation(report_text: str) -> dict:
+    """#139: reject review-merge/create-issue handoff fields in work-issue runs."""
+    section = _handoff_section_lines(report_text)
+    reasons = []
+    if section is None:
+        return {
+            "complete": True,
+            "downgraded": False,
+            "reasons": [],
+        }
+
+    labels = []
+    for line in section:
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            labels.append(stripped.split(":", 1)[0].strip().lower())
+
+    for name, aliases in WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS:
+        if any(
+            label.startswith(alias)
+            for label in labels
+            for alias in aliases
+        ):
+            reasons.append(
+                f"work-issue handoff must not include cross-mode field "
+                f"'{name}'"
+            )
+
+    legacy_workspace = any(
+        label.startswith("workspace mutations") for label in labels
+    )
+    precise_categories = any(
+        label.startswith("file edits by author")
+        for label in labels
+    )
+    if legacy_workspace and not precise_categories:
+        reasons.append(
+            "work-issue handoff must use precise mutation categories "
+            "instead of legacy 'Workspace mutations'"
+        )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_work_issue_final_report(report_text: str) -> dict:
+    """#139: composite verifier for work-issue final reports."""
+    checks = {
+        "workflow_source": assess_work_issue_workflow_source(report_text),
+        "mode_isolation": assess_work_issue_mode_isolation(report_text),
+    }
+
+    reasons = []
+    downgraded = False
+    for name, result in checks.items():
+        verdict = result.get("verdict")
+        if verdict in ("missing", "incomplete"):
+            downgraded = True
+            reasons.extend(result.get("reasons") or [])
+        elif result.get("downgraded") or not result.get("complete", True):
+            downgraded = True
+            reasons.extend(
+                f"{name}: {r}" for r in (result.get("reasons") or [])
+            )
+
+    grade = "A" if not downgraded else "downgraded"
+    return {
+        "grade": grade,
+        "downgraded": downgraded,
+        "checks": checks,
+        "reasons": reasons,
+        "complete": not downgraded,
+    }
+
+
 def assess_create_issue_mode_isolation(report_text: str) -> dict:
     """#337: reject work-issue/review-merge handoff fields in create-issue runs."""
     section = _handoff_section_lines(report_text)
@@ -4909,6 +5058,13 @@ def assess_final_report_validator(report_text, task_kind, **kwargs):
     return _validate(report_text, task_kind, **kwargs)
 
 
+def assess_review_final_report_schema(report_text, **kwargs):
+    """#391: reviewer final-report schema verification before session output."""
+    from review_final_report_schema import assess_review_final_report_schema as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 _QUEUE_STATUS_REPORT_HINT = re.compile(
     r"queue[- ]status|selected pr:\s*none|no pr selected|"
     r"queue[- ]status[- ]only",
@@ -5331,6 +5487,22 @@ def assess_validation_integrity_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_validation_cwd_proof_report(report_text, **kwargs):
+    """#398: validation commands require explicit worktree cwd and HEAD proof."""
+    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_already_landed_classification_report(report_text, **kwargs):
+    """#295: already-landed PRs are reconciliation-only, not review eligible."""
+    from reviewer_already_landed_classification import (
+        assess_already_landed_classification_report as _assess,
+    )
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_prior_blocker_skip_proof(report_text, **kwargs):
     """#318: require live blocker proof before skipping earlier open PRs."""
     from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess

reviewer_already_landed_classification.py

@@ -0,0 +1,143 @@
+"""Already-landed PR classification verifier for reviewer reports (#295)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
+
+_LANDED_CONTEXT_RE = re.compile(
+    r"already[- ]landed|ancestor proof\s*:\s*passed|"
+    r"eligibility class\s*:\s*already_landed",
+    re.IGNORECASE,
+)
+_ELIGIBILITY_CLASS_LINE_RE = re.compile(
+    r"eligibility class\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_INELIGIBLE_SELECTION_RE = re.compile(
+    r"\b(?:next|oldest)\s+eligible\s+pr\b",
+    re.IGNORECASE,
+)
+_REVIEW_ELIGIBLE_RE = re.compile(
+    r"\beligible for (?:review|merge)\b|\bready for (?:review|merge)\b",
+    re.IGNORECASE,
+)
+_PINNED_REVIEWED_HEAD_RE = re.compile(
+    r"\bpinned reviewed head\s*:",
+    re.IGNORECASE,
+)
+_CANDIDATE_HEAD_RE = re.compile(
+    r"\bcandidate head sha\s*:",
+    re.IGNORECASE,
+)
+_REVIEWED_HEAD_NONE_RE = re.compile(
+    r"\breviewed head sha\s*:\s*none\b",
+    re.IGNORECASE,
+)
+_VALIDATION_PROOF_RE = re.compile(
+    r"(?:validation passed|pytest.*passed|diff review passed|"
+    r"validated on pinned head)",
+    re.IGNORECASE,
+)
+_QUEUE_BLOCKED_RE = re.compile(
+    r"queue blocked by already[- ]landed|"
+    r"reconciliation(?:-only)?\s+(?:next step|required)",
+    re.IGNORECASE,
+)
+
+
+def _landed_context(report_text: str, eligibility_class: str | None) -> bool:
+    if (eligibility_class or "").strip().upper() == ALREADY_LANDED_ELIGIBILITY_CLASS:
+        return True
+    return bool(_LANDED_CONTEXT_RE.search(report_text or ""))
+
+
+def assess_already_landed_classification_report(
+    report_text: str,
+    *,
+    eligibility_class: str | None = None,
+    selected_pr_already_landed: bool | None = None,
+) -> dict[str, Any]:
+    """#295: already-landed PRs are reconciliation-only, not review eligible."""
+    text = report_text or ""
+    reasons: list[str] = []
+
+    landed = _landed_context(text, eligibility_class)
+    if selected_pr_already_landed is True:
+        landed = True
+    if not landed:
+        return {
+            "proven": True,
+            "block": False,
+            "landed_context": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    if _INELIGIBLE_SELECTION_RE.search(text):
+        reasons.append(
+            "already-landed PR must not be described as oldest/next eligible PR; "
+            "use 'Oldest open PR requiring action' and "
+            f"'Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}'"
+        )
+
+    if _REVIEW_ELIGIBLE_RE.search(text):
+        reasons.append(
+            "already-landed PR must not be described as eligible for review or merge"
+        )
+
+    class_match = _ELIGIBILITY_CLASS_LINE_RE.search(text)
+    if class_match:
+        declared = class_match.group(1).strip().upper()
+        if declared != ALREADY_LANDED_ELIGIBILITY_CLASS:
+            reasons.append(
+                "already-landed PR eligibility class must be "
+                f"{ALREADY_LANDED_ELIGIBILITY_CLASS}"
+            )
+    elif selected_pr_already_landed is True:
+        reasons.append(
+            f"already-landed selected PR must declare Eligibility class: "
+            f"{ALREADY_LANDED_ELIGIBILITY_CLASS}"
+        )
+
+    if _PINNED_REVIEWED_HEAD_RE.search(text):
+        if not _VALIDATION_PROOF_RE.search(text):
+            reasons.append(
+                "already-landed pre-review report must use Candidate head SHA, "
+                "not Pinned reviewed head, unless validation and diff review passed"
+            )
+
+    if selected_pr_already_landed is True and not _CANDIDATE_HEAD_RE.search(text):
+        if _PINNED_REVIEWED_HEAD_RE.search(text) or "head sha" in text.lower():
+            reasons.append(
+                "already-landed ancestry check must report Candidate head SHA"
+            )
+
+    if selected_pr_already_landed is True and not _REVIEWED_HEAD_NONE_RE.search(text):
+        if _PINNED_REVIEWED_HEAD_RE.search(text) and not _VALIDATION_PROOF_RE.search(text):
+            reasons.append(
+                "already-landed report must state 'Reviewed head SHA: none' "
+                "when validation did not run"
+            )
+
+    if selected_pr_already_landed is True and not _QUEUE_BLOCKED_RE.search(text):
+        reasons.append(
+            "already-landed queue blocker must state reconciliation requirement "
+            "or queue blocked wording"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "landed_context": True,
+        "reasons": reasons,
+        "safe_next_action": (
+            "classify as ALREADY_LANDED_RECONCILE_REQUIRED, use Candidate head SHA, "
+            "and stop normal review"
+            if not proven
+            else "proceed"
+        ),
+    }

reviewer_validation_cwd_proof.py

@@ -0,0 +1,233 @@
+"""Explicit worktree and cwd proof for PR review validation (#398)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+_PWD_RE = re.compile(
+    r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_HEAD_RE = re.compile(
+    r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE | re.MULTILINE,
+)
+_EXPECTED_HEAD_RE = re.compile(
+    r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_STATUS_RE = re.compile(
+    r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_VALIDATION_CMD_RE = re.compile(
+    r"validation\s+command\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE)
+_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE)
+_BASELINE_CWD_RE = re.compile(
+    r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BASELINE_SHA_RE = re.compile(
+    r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_BASELINE_CMD_RE = re.compile(
+    r"baseline\s+validation\s+command\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").rstrip("/")
+
+
+def _path_under_branches(path: str, project_root: str | None = None) -> bool:
+    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(project_root)
+        if normalized.startswith(f"{root}/"):
+            rel = normalized[len(root) + 1 :]
+            return rel == "branches" or rel.startswith("branches/")
+    return False
+
+
+def _expand_sha(sha: str) -> str:
+    return (sha or "").strip().lower()
+
+
+def _sha_matches(expected: str, observed: str) -> bool:
+    exp = _expand_sha(expected)
+    obs = _expand_sha(observed)
+    if not exp or not obs:
+        return False
+    if len(exp) == 40 and len(obs) == 40:
+        return exp == obs
+    return obs.startswith(exp) or exp.startswith(obs)
+
+
+def _command_has_explicit_cwd(command: str, cwd: str) -> bool:
+    text = (command or "").strip()
+    if not text:
+        return False
+    if _GIT_C_CMD_RE.search(text):
+        return True
+    if _CD_CMD_RE.search(text):
+        return True
+    if cwd and cwd in text:
+        return True
+    return False
+
+
+def assess_validation_cwd_proof_report(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Require cwd/HEAD proof before reviewer validation claims (#398)."""
+    text = report_text or ""
+    session = dict(validation_session or {})
+    reasons: list[str] = []
+    violations: list[str] = []
+
+    claims_validation = bool(
+        session.get("validation_ran")
+        or _VALIDATION_CMD_RE.search(text)
+        or session.get("command")
+    )
+    if not claims_validation:
+        return {
+            "proven": True,
+            "block": False,
+            "claims_validation": False,
+            "reasons": [],
+            "violations": [],
+            "safe_next_action": "proceed",
+        }
+
+    expected_head = (
+        session.get("expected_head_sha")
+        or session.get("candidate_head_sha")
+        or ""
+    ).strip()
+    if not expected_head:
+        match = _EXPECTED_HEAD_RE.search(text)
+        expected_head = (match.group(1) if match else "").strip()
+
+    observed_head = (session.get("observed_head_sha") or "").strip()
+    if not observed_head:
+        match = _HEAD_RE.search(text)
+        observed_head = (match.group(1) if match else "").strip()
+
+    cwd = (
+        session.get("working_directory")
+        or session.get("cwd")
+        or session.get("pwd")
+        or ""
+    ).strip()
+    if not cwd:
+        match = _PWD_RE.search(text)
+        cwd = (match.group(1) if match else "").strip().rstrip(",.;")
+
+    command = (session.get("command") or "").strip()
+    if not command:
+        match = _VALIDATION_CMD_RE.search(text)
+        command = (match.group(1) if match else "").strip().rstrip(".;")
+
+    if not cwd:
+        reasons.append(
+            "validation claimed without pwd/working-directory proof (#398)"
+        )
+    elif not _path_under_branches(cwd, project_root):
+        violations.append(
+            f"validation cwd {cwd!r} is not under branches/ (#398)"
+        )
+        reasons.append(
+            "reviewer validation must run from a branches/ worktree, "
+            "not the main checkout (#398)"
+        )
+
+    if not observed_head:
+        reasons.append(
+            "validation claimed without git rev-parse HEAD / observed HEAD SHA "
+            "proof (#398)"
+        )
+    elif expected_head and not _sha_matches(expected_head, observed_head):
+        violations.append(
+            f"observed HEAD {observed_head} does not match expected "
+            f"PR head {expected_head} (#398)"
+        )
+        reasons.append("validation HEAD SHA must match pinned PR head (#398)")
+
+    if not _STATUS_RE.search(text) and session.get("git_status") is None:
+        reasons.append(
+            "validation claimed without git status --short --branch proof (#398)"
+        )
+
+    if command and cwd and not _command_has_explicit_cwd(command, cwd):
+        if session.get("tool_working_directory") is not True:
+            reasons.append(
+                "validation command must use git -C <worktree>, "
+                "cd <worktree> && ..., or tool-provided cwd metadata (#398)"
+            )
+
+    baseline_ran = bool(
+        session.get("baseline_validation_ran")
+        or _BASELINE_CMD_RE.search(text)
+    )
+    if baseline_ran:
+        baseline_cwd = (session.get("baseline_worktree_path") or "").strip()
+        if not baseline_cwd:
+            match = _BASELINE_CWD_RE.search(text)
+            baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;")
+        if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root):
+            reasons.append(
+                "baseline validation claimed without baseline worktree cwd "
+                "under branches/ (#398)"
+            )
+        baseline_sha = (session.get("baseline_target_sha") or "").strip()
+        if not baseline_sha:
+            match = _BASELINE_SHA_RE.search(text)
+            baseline_sha = (match.group(1) if match else "").strip()
+        if not baseline_sha:
+            reasons.append(
+                "baseline validation claimed without baseline target SHA (#398)"
+            )
+        baseline_cmd = (session.get("baseline_command") or "").strip()
+        if not baseline_cmd:
+            match = _BASELINE_CMD_RE.search(text)
+            baseline_cmd = (match.group(1) if match else "").strip()
+        if not baseline_cmd:
+            reasons.append(
+                "baseline validation claimed without exact baseline command (#398)"
+            )
+
+    proven = not reasons and not violations
+    return {
+        "proven": proven,
+        "block": bool(violations) or not proven,
+        "claims_validation": True,
+        "expected_head_sha": expected_head or None,
+        "observed_head_sha": observed_head or None,
+        "working_directory": cwd or None,
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": (
+            "before validation record pwd, git rev-parse HEAD, git status, "
+            "expected PR head SHA; run commands with git -C or cd in the same line"
+            if not proven
+            else "proceed"
+        ),
+    }

role_session_router.py

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

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -574,6 +574,28 @@ Do not claim “full-suite failures are pre-existing” unless baseline proof is
 
 ## 23. Validation command proof rule
 
+Before any diff, test, or compile validation, record in the same command
+transcript or final report:
+
+* `pwd` or explicit working directory
+* `git rev-parse HEAD`
+* `git status --short --branch`
+* expected PR head SHA (candidate head SHA)
+
+Validation commands must use one of:
+
+* `git -C <review_worktree> ...`
+* `cd <review_worktree> && ...` in the same command
+* tool-provided explicit working-directory metadata
+
+Do not rely on inferred shell cwd from a prior command in a different block.
+
+`gitea_validate_review_final_report` rejects validation claims without
+cwd/HEAD proof when `validation_session` is supplied.
+
+Baseline validation must document baseline worktree path, baseline target SHA,
+cwd proof, exact baseline command, and baseline result using the same rules.
+
 Report the exact validation command as executed.
 
 Report the working directory where validation ran.
@@ -1096,6 +1118,7 @@ Controller Handoff:
 * Baseline worktree path:
 * Files reviewed:
 * Validation:
+* Validation cwd/HEAD proof:
 * Official validation integrity status:
 * Terminal review mutation:
 * Review decision:

skills/llm-project-workflow/workflows/work-issue.md

@@ -83,6 +83,20 @@ Examples:
 
 If capability cannot be proven, stop and produce a recovery handoff only.
 
+### 2A. Pre-task role routing (#139)
+
+Before issue selection or any mutation, resolve the composite author task:
+
+* `gitea_route_task_session(task_type="work-issue", remote=…)` — must return
+  `route_result: allowed_current_session` and `downstream_allowed: true`
+* `gitea_resolve_task_capability(task="work_issue", remote=…)` — must show
+  `required_role_kind: author` and `allowed_in_current_session: true`
+
+Hyphen (`work-issue`) and underscore (`work_issue`) aliases are equivalent.
+If routing returns `ambiguous_task_stop`, `wrong_role_stop`, or
+`route_to_author_session`, stop and produce a recovery handoff only — do not
+fall back to reviewer tools or guess a different profile.
+
 ## 3. Stop immediately on blocked infrastructure
 
 If any of the following appears, stop immediately:

task_capability_map.py

@@ -104,6 +104,22 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    "work_issue": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "work-issue": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "reconcile_landed_pr": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "reconcile-landed-pr": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
     "reconcile_already_landed_pr": {
         "permission": "gitea.pr.close",
         "role": "reconciler",

tests/test_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_final_report_validator.py

@@ -57,26 +57,47 @@ def _review_handoff(**overrides):
     return text
 
 
-def _reconcile_handoff(**overrides):
+def _reconcile_handoff(drop=(), **overrides):
     lines = [
         "## Controller Handoff",
         "",
         "- Task: reconcile already-landed PR #99",
         "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-        "- Role: reconciler",
-        "- Identity: sysadmin / prgs-author",
+        "- Role/profile: reconciler / prgs-reconciler",
+        "- Identity: sysadmin",
         "- Selected PR: #99",
-        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- PR live state: open",
+        "- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Target branch: master",
+        "- Target branch SHA: 5e023dc71b0e2b813a0b1eee6b0f42630c47a95c",
+        "- Ancestor proof: candidate head is ancestor of target branch SHA",
         "- Linked issue: #98",
         "- Linked issue live status: not verified in this session",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- Capabilities proven: gitea.read, gitea.pr.comment",
+        "- Missing capabilities: gitea.pr.close",
+        "- PR comments posted: 1 reconciliation comment on PR #99",
+        "- Issue comments posted: none",
+        "- PRs closed: none",
+        "- Issues closed: none",
         "- File edits by reconciler: none",
         "- Git ref mutations: git fetch prgs master",
+        "- Worktree mutations: none",
+        "- MCP/Gitea mutations: PR comment posted",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
         "- Reconciliation mutations: PR comment posted",
         "- Current status: reconciliation complete",
-        "- Safe next action: none",
+        "- Blocker: missing gitea.pr.close capability",
+        "- Safe next action: hand off to a profile with gitea.pr.close",
         "- No review/merge confirmation: confirmed",
     ]
-    text = "\n".join(lines)
+    kept = [
+        line
+        for line in lines
+        if not any(line.startswith(f"- {name}:") for name in drop)
+    ]
+    text = "\n".join(kept)
     for key, value in overrides.items():
         if f"- {key}:" in text:
             text = text.replace(f"- {key}:", f"- {key}: {value}")
@@ -194,6 +215,20 @@ class TestReviewPrRules(unittest.TestCase):
             )
         )
 
+    def test_already_landed_oldest_eligible_pr_blocks(self):
+        report = (
+            _review_handoff()
+            + "\nAlready landed.\nOldest eligible PR: PR #278."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.already_landed_eligible_wording"
+                for f in result["findings"]
+            )
+        )
+
     def test_already_landed_gate_blocks_review_terminal_states(self):
         cases = {
             "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
@@ -242,7 +277,6 @@ class TestReviewPrRules(unittest.TestCase):
         )
         result = assess_final_report_validator(report, "review_pr")
         self.assertEqual(result["grade"], "A")
-
     def test_git_fetch_must_not_be_readonly_only(self):
         report = (
             _review_handoff()
@@ -307,6 +341,159 @@ class TestReconciliationRules(unittest.TestCase):
         )
 
 
+class TestCanonicalReconcileSchema(unittest.TestCase):
+    """Issue #307: reconciliation workflows emit only the canonical schema."""
+
+    def test_comment_only_reconciliation_passes(self):
+        result = assess_final_report_validator(
+            _reconcile_handoff(),
+            "reconcile_already_landed",
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertEqual(result["findings"], [])
+
+    def test_successful_close_reconciliation_passes(self):
+        report = _reconcile_handoff().replace(
+            "- Capabilities proven: gitea.read, gitea.pr.comment",
+            "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
+        ).replace(
+            "- Missing capabilities: gitea.pr.close",
+            "- Missing capabilities: none",
+        ).replace(
+            "- PRs closed: none",
+            "- PRs closed: #99",
+        ).replace(
+            "- Blocker: missing gitea.pr.close capability",
+            "- Blocker: none",
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "A")
+
+    def test_blocked_reconciliation_passes(self):
+        report = _reconcile_handoff().replace(
+            "- Capabilities proven: gitea.read, gitea.pr.comment",
+            "- Capabilities proven: gitea.read",
+        ).replace(
+            "- Missing capabilities: gitea.pr.close",
+            "- Missing capabilities: gitea.pr.close, gitea.pr.comment",
+        ).replace(
+            "- PR comments posted: 1 reconciliation comment on PR #99",
+            "- PR comments posted: none",
+        ).replace(
+            "- MCP/Gitea mutations: PR comment posted",
+            "- MCP/Gitea mutations: none",
+        ).replace(
+            "- Reconciliation mutations: PR comment posted",
+            "- Reconciliation mutations: none",
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "A")
+
+    def test_missing_capabilities_field_required(self):
+        report = _reconcile_handoff(drop=("Missing capabilities",))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "downgraded")
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.controller_handoff"
+                and "Missing capabilities" in f["reason"]
+                for f in result["findings"]
+            )
+        )
+
+    def test_missing_ancestor_proof_and_candidate_sha_downgrade(self):
+        report = _reconcile_handoff(drop=("Ancestor proof", "Candidate head SHA"))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "downgraded")
+        reasons = " ".join(f["reason"] for f in result["findings"])
+        self.assertIn("Ancestor proof", reasons)
+        self.assertIn("Candidate head SHA", reasons)
+
+    def test_stale_issue_lock_proof_blocks(self):
+        report = _reconcile_handoff() + "\n- Issue lock proof: locked before diff"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "issue lock proof"
+                for f in result["findings"]
+            )
+        )
+
+    def test_stale_claim_comment_status_blocks(self):
+        report = _reconcile_handoff() + "\n- Claim/comment status: claimed"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "claim/comment status"
+                for f in result["findings"]
+            )
+        )
+
+    def test_stale_workspace_mutations_blocks_without_observed_mutations(self):
+        report = _reconcile_handoff() + "\n- Workspace mutations: None"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "workspace mutations"
+                for f in result["findings"]
+            )
+        )
+
+    def test_pr_number_opened_allowed_when_session_opened_pr(self):
+        report = _reconcile_handoff() + "\n- PR number opened: #12"
+        result = assess_final_report_validator(
+            report,
+            "reconcile_already_landed",
+            session_pr_opened=True,
+        )
+        self.assertFalse(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "pr number opened"
+                for f in result["findings"]
+            )
+        )
+
+    def test_pr_number_opened_still_blocked_without_session_proof(self):
+        report = _reconcile_handoff() + "\n- PR number opened: #12"
+        result = assess_final_report_validator(
+            report,
+            "reconcile_already_landed",
+            session_pr_opened=False,
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "pr number opened"
+                for f in result["findings"]
+            )
+        )
+
+    def test_action_log_fetch_requires_git_ref_mutation_entry(self):
+        report = _reconcile_handoff().replace(
+            "- Git ref mutations: git fetch prgs master",
+            "- Git ref mutations: none",
+        )
+        result = assess_final_report_validator(
+            report,
+            "reconcile_already_landed",
+            action_log=[{"command": "git fetch prgs master", "performed": True}],
+        )
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.git_fetch_readonly"
+                for f in result["findings"]
+            )
+        )
+
+
 class TestEntryPoint(unittest.TestCase):
     def test_unknown_task_kind_blocks(self):
         result = assess_final_report_validator("report", "unknown_mode")

tests/test_llm_workflow_split.py

@@ -100,6 +100,8 @@ def test_work_issue_workflow_contract():
     assert "canonical: true" in text
     assert "Do not merge your own PR" in text
     assert "## 21. PR creation rules" in text
+    assert "gitea_route_task_session" in text
+    assert "work-issue" in text
 
 
 def test_pr_queue_cleanup_workflow_contract():
@@ -139,6 +141,16 @@ def test_start_issue_template_references_work_issue_workflow():
     assert "workflows/work-issue.md" in text
 
 
+def test_reconcile_skill_registered_in_mcp_server():
+    from mcp_server import _PROJECT_SKILLS
+
+    assert "gitea-reconcile-landed-pr" in _PROJECT_SKILLS
+    steps = _PROJECT_SKILLS["gitea-reconcile-landed-pr"]["steps"]
+    joined = " ".join(steps).lower()
+    assert "gitea_scan_already_landed_open_prs" in joined
+    assert "gitea_assess_already_landed_reconciliation" in joined
+
+
 def test_pr_queue_cleanup_template_references_workflow():
     text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text(
         encoding="utf-8"
@@ -165,12 +177,24 @@ def test_final_report_validator_exported():
     assert callable(assess_final_report_validator)
 
 
+def test_review_final_report_schema_verifier_exported():
+    from review_proofs import assess_review_final_report_schema
+
+    assert callable(assess_review_final_report_schema)
+
+
 def test_create_issue_final_report_verifier_exported():
     from review_proofs import assess_create_issue_final_report
 
     assert callable(assess_create_issue_final_report)
 
 
+def test_work_issue_final_report_verifier_exported():
+    from review_proofs import assess_work_issue_final_report
+
+    assert callable(assess_work_issue_final_report)
+
+
 def test_merge_simulation_verifier_exported():
     from review_proofs import assess_merge_simulation_report
 
@@ -183,6 +207,12 @@ def test_validation_integrity_verifier_exported():
     assert callable(assess_validation_integrity_report)
 
 
+def test_validation_cwd_proof_verifier_exported():
+    from review_proofs import assess_validation_cwd_proof_report
+
+    assert callable(assess_validation_cwd_proof_report)
+
+
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 
@@ -243,6 +273,12 @@ def test_worktree_ownership_verifier_exported():
     assert callable(assess_worktree_ownership_report)
 
 
+def test_already_landed_classification_verifier_exported():
+    from review_proofs import assess_already_landed_classification_report
+
+    assert callable(assess_already_landed_classification_report)
+
+
 def test_pr_queue_cleanup_verifier_exported():
     from review_proofs import assess_pr_queue_cleanup_report
 

tests/test_mcp_server.py

@@ -3050,10 +3050,18 @@ class TestIssueLocking(unittest.TestCase):
         mock_api.return_value = [] # no open PRs
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
+        self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
+        self.assertEqual(res["work_lease"]["issue_number"], 196)
+        self.assertEqual(res["work_lease"]["branch"], "feat/issue-196-mutations")
+        self.assertIn("created_at", res["work_lease"])
+        self.assertIn("expires_at", res["work_lease"])
+        self.assertIn("last_heartbeat_at", res["work_lease"])
+        self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
         self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
         with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
             lock = json.load(f)
         self.assertIn("worktree_path", lock)
+        self.assertIn("work_lease", lock)
 
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
@@ -3094,6 +3102,51 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
+        mock_api.side_effect = [
+            [],
+            [{"name": "feat/issue-196-existing-work"}],
+        ]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already has matching branch", str(ctx.exception))
+
+    def test_lock_issue_blocks_active_same_operation_lease(self):
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump({
+                "issue_number": 196,
+                "branch_name": "feat/issue-196-other-work",
+                "worktree_path": "/tmp/other-worktree",
+                "work_lease": {
+                    "operation_type": "author_issue_work",
+                    "expires_at": "2999-01-01T00:00:00Z",
+                },
+            }, f)
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
+
+    def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump({
+                "issue_number": 196,
+                "branch_name": "feat/issue-196-other-work",
+                "worktree_path": "/tmp/other-worktree",
+                "work_lease": {
+                    "operation_type": "author_issue_work",
+                    "expires_at": "2000-01-01T00:00:00Z",
+                },
+            }, f)
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("Recovery review is required before takeover", str(ctx.exception))
+
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_lock_from_clean_scratch_worktree(self, _auth, _api):

tests/test_reconciliation_mcp_tools.py

@@ -0,0 +1,106 @@
+"""Tests for reconciliation MCP assessment tools (#301)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+from mcp_server import (
+    gitea_assess_already_landed_reconciliation,
+    gitea_scan_already_landed_open_prs,
+)
+
+AUTHOR_PROFILE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.comment",
+        "gitea.issue.comment",
+        "gitea.issue.close",
+    ],
+    "forbidden_operations": ["gitea.pr.close"],
+    "audit_label": "prgs-author",
+}
+
+OPEN_PR = {
+    "number": 278,
+    "title": "Landed (Closes #263)",
+    "body": "",
+    "state": "open",
+    "head": {"ref": "feat/x", "sha": "a" * 40},
+    "base": {"ref": "master"},
+}
+
+
+class TestReconciliationMcpTools(unittest.TestCase):
+    def setUp(self):
+        self.mock_api = patch("mcp_server.api_request").start()
+        self.mock_auth = patch(
+            "mcp_server.get_auth_header", return_value="token test"
+        ).start()
+        patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE).start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def test_assess_returns_plan_without_mutations(self):
+        self.mock_api.return_value = OPEN_PR
+        with patch(
+            "mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
+            return_value={
+                "reconciliation_allowed": True,
+                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+                "candidate_head_sha": OPEN_PR["head"]["sha"],
+                "target_branch_sha": "deadbeef",
+                "linked_issue": 263,
+                "review_merge_allowed": False,
+            },
+        ):
+            res = gitea_assess_already_landed_reconciliation(
+                pr_number=278, remote="prgs"
+            )
+        self.assertTrue(res["success"])
+        self.assertFalse(res["performed"])
+        self.assertEqual(res["plan"]["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
+        self.assertFalse(res["review_merge_allowed"])
+        patch_calls = [
+            c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
+        ]
+        self.assertEqual(patch_calls, [])
+
+    def test_scan_returns_candidates_with_pagination(self):
+        with patch(
+            "mcp_server.api_fetch_page",
+            return_value=([OPEN_PR], {
+                "page": 1,
+                "per_page": 50,
+                "is_final_page": True,
+                "has_more": False,
+                "next_page": None,
+            }),
+        ), patch(
+            "mcp_server.reconciliation_workflow.fetch_target_branch",
+            return_value={
+                "success": True,
+                "target_branch_sha": "deadbeef",
+                "git_fetch_command": "git fetch prgs master",
+            },
+        ), patch(
+            "mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
+            return_value={
+                "pr_number": 278,
+                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+                "reconciliation_allowed": True,
+            },
+        ):
+            res = gitea_scan_already_landed_open_prs(remote="prgs", limit=50)
+        self.assertTrue(res["success"])
+        self.assertEqual(res["candidate_count"], 1)
+        self.assertTrue(res["pagination"]["inventory_complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconciliation_workflow.py

@@ -0,0 +1,149 @@
+"""Tests for already-landed reconciliation workflow helpers (#301)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reconciliation_workflow
+from review_proofs import assess_reconcile_workflow_source
+
+
+class TestReconciliationAssessment(unittest.TestCase):
+    def test_already_landed_open_pr(self):
+        with patch(
+            "reconciliation_workflow.is_head_ancestor_of_ref",
+            return_value=True,
+        ):
+            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+                pr={
+                    "number": 278,
+                    "state": "open",
+                    "title": "Fix (Closes #263)",
+                    "body": "",
+                    "head": {"ref": "feat/x", "sha": "abc" * 13 + "a"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                    "git_fetch_command": "git fetch prgs master",
+                },
+            )
+        self.assertTrue(assessment["reconciliation_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+        )
+
+    def test_not_landed_pr(self):
+        with patch(
+            "reconciliation_workflow.is_head_ancestor_of_ref",
+            return_value=False,
+        ):
+            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+                pr={
+                    "number": 1,
+                    "state": "open",
+                    "title": "WIP",
+                    "body": "",
+                    "head": {"ref": "feat/y", "sha": "fff" * 13 + "f"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                },
+            )
+        self.assertFalse(assessment["reconciliation_allowed"])
+
+
+class TestReconciliationPlan(unittest.TestCase):
+    def test_full_close_when_close_pr_allowed(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={
+                "close_pr": True,
+                "comment_pr": True,
+                "close_issue": True,
+                "comment_issue": True,
+            },
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_FULL_RECONCILE,
+        )
+        self.assertTrue(plan["close_pr_allowed"])
+        self.assertFalse(plan["review_merge_allowed"])
+
+    def test_comment_then_stop_without_close(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={
+                "close_pr": False,
+                "comment_pr": True,
+            },
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_PARTIAL_COMMENT,
+        )
+
+    def test_recovery_handoff_without_comment_or_close(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={"close_pr": False, "comment_pr": False},
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF,
+        )
+
+    def test_not_landed_no_action(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": False,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED,
+                "reasons": ["not ancestor"],
+            },
+            capabilities={"close_pr": True},
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_NOT_LANDED,
+        )
+
+
+class TestReconcileWorkflowSource(unittest.TestCase):
+    def test_valid_report_passes(self):
+        report = (
+            "Task mode: reconcile-landed-pr\n"
+            "Workflow source: workflows/reconcile-landed-pr.md\n"
+        )
+        result = assess_reconcile_workflow_source(report)
+        self.assertTrue(result["complete"])
+
+    def test_missing_workflow_fails(self):
+        result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr")
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -121,6 +121,32 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertTrue(res["allowed_in_current_session"])
             self.assertFalse(res["different_mcp_namespace_required"])
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_work_issue_author_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("author-profile")):
+            for task in ("work_issue", "work-issue"):
+                res = mcp_server.gitea_resolve_task_capability(
+                    task=task, remote="prgs"
+                )
+                self.assertEqual(res["requested_task"], task)
+                self.assertEqual(
+                    res["required_operation_permission"], "gitea.pr.create"
+                )
+                self.assertEqual(res["required_role_kind"], "author")
+                self.assertTrue(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_work_issue_reviewer_profile_blocked(self, _auth, _api):
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="work_issue", remote="prgs"
+            )
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertEqual(res["required_role_kind"], "author")
+            self.assertTrue(res["different_mcp_namespace_required"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_issue_comment_task(self, _auth, _api):

tests/test_review_final_report_schema.py

@@ -0,0 +1,158 @@
+"""Tests for reviewer final-report schema verification (#391)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_final_report_schema import assess_review_final_report_schema  # noqa: E402
+
+
+def _minimal_review_report(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: review PR #203",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: reviewer",
+        "- Identity: sysadmin / prgs-reviewer",
+        "- Issue/PR: #182 / PR #203",
+        "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Files changed: review_proofs.py",
+        "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Mutations: review only",
+        "- File edits by reviewer: none",
+        "- Worktree/index mutations: none",
+        "- Git ref mutations: git fetch prgs master",
+        "- MCP/Gitea mutations: review submitted",
+        "- Review mutations: request_changes submitted",
+        "- Merge mutations: none",
+        "- Cleanup mutations: none",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
+        "- Current status: PR open",
+        "- Blockers: none",
+        "- Next: await author",
+        "- Safety: no self-review; no self-merge",
+        "- Selected PR: #203",
+        "- Reviewer eligibility: eligible",
+        "- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Worktree path: branches/review-203",
+        "- Worktree dirty: clean",
+        "- Unrelated local mutations: none",
+        "- Review decision: request_changes",
+        "- Merge result: not attempted",
+        "- Linked issue: #182",
+        "- Linked issue live status: open (gitea_view_issue)",
+        "- Cleanup status: none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        prefix = f"- {key}:"
+        replaced = False
+        new_lines = []
+        for line in text.splitlines():
+            if line.strip().startswith(prefix):
+                new_lines.append(f"{prefix} {value}")
+                replaced = True
+            else:
+                new_lines.append(line)
+        text = "\n".join(new_lines)
+        if not replaced:
+            text += f"\n{prefix} {value}"
+    return text
+
+
+class TestReviewFinalReportSchema(unittest.TestCase):
+    def test_clean_report_passes(self):
+        result = assess_review_final_report_schema(
+            _minimal_review_report(),
+            linked_issue_lock={"issue_number": 182, "pr_number": 203},
+        )
+        self.assertFalse(result["blocked"])
+        self.assertIn(result["grade"], ("A", "downgraded"))
+
+    def test_legacy_pinned_reviewed_head_blocks(self):
+        report = _minimal_review_report(**{"Pinned reviewed head": "abc123"})
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.legacy_stale_field", rule_ids)
+
+    def test_reviewed_head_without_validation_blocks(self):
+        report = _minimal_review_report().replace(
+            "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+            "- Validation: not run",
+        )
+        report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.reviewed_head_without_validation", rule_ids)
+
+    def test_merged_without_proof_blocks(self):
+        report = _minimal_review_report() + "\n\nPR #203 merged successfully."
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.merged_without_proof", rule_ids)
+
+    def test_issue_closed_without_live_proof_blocks(self):
+        report = _minimal_review_report(
+            **{
+                "Linked issue live status": "closed",
+                "Read-only diagnostics": "gitea_view_pr only",
+            }
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.issue_closed_without_live_proof", rule_ids)
+
+    def test_full_suite_pass_with_ignored_tests_blocks(self):
+        report = (
+            _minimal_review_report()
+            + "\nFull test suite passed with 0 failed; some tests ignored."
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.full_suite_pass_ignored_tests", rule_ids)
+
+    def test_blocked_handoff_replay_blocks(self):
+        report = (
+            _minimal_review_report(**{"Blockers": "capability stop"})
+            + "\nSafe next action: run gitea_merge_pr to finish."
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.blocked_handoff_replay", rule_ids)
+
+    def test_narrative_handoff_drift_blocks(self):
+        report = (
+            _minimal_review_report()
+            + "\n## Summary\nVerdict: APPROVE this PR."
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.narrative_handoff_drift", rule_ids)
+
+    def test_pending_vs_approved_blocks(self):
+        report = _minimal_review_report(
+            **{"Review decision": "PENDING official review submitted approve"}
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.pending_vs_approved", rule_ids)
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_review_final_report_schema as exported
+
+        self.assertTrue(callable(exported))
+
+
+class TestMcpValidateReviewFinalReport(unittest.TestCase):
+    def test_mcp_tool_callable(self):
+        import gitea_mcp_server
+
+        result = gitea_mcp_server.gitea_validate_review_final_report(
+            _minimal_review_report()
+        )
+        self.assertIn("grade", result)
+        self.assertIn("findings", result)
+        self.assertFalse(result["blocked"])

tests/test_review_proofs.py

@@ -40,6 +40,9 @@ from review_proofs import (  # noqa: E402
     assess_create_issue_final_report,
     assess_create_issue_mode_isolation,
     assess_create_issue_workflow_source,
+    assess_work_issue_final_report,
+    assess_work_issue_mode_isolation,
+    assess_work_issue_workflow_source,
     assess_edited_pr_inventory_coverage,
     assess_empty_queue_report,
     assess_fresh_issue_selection,
@@ -2320,6 +2323,52 @@ class TestCreateIssueFinalReport(unittest.TestCase):
         self.assertTrue(result["downgraded"])
 
 
+class TestWorkIssueFinalReport(unittest.TestCase):
+    """#139: work-issue final-report verifier coverage."""
+
+    def _good_report(self):
+        return "\n".join([
+            "Task mode: work-issue",
+            "Workflow source: skills/llm-project-workflow/workflows/work-issue.md",
+            "## Controller Handoff",
+            "- Task: work-issue",
+            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+            "- Role: author",
+            "- Identity: jcwalker3 / prgs-author",
+            "- Active profile: prgs-author",
+            "- Runtime context: prgs-author on prgs",
+            "- Selected issue: #139",
+            "- Branch name: feat/issue-139-role-aware-work-issue-routing",
+            "- PR number: not created in this session",
+            "- File edits by author: role_session_router.py",
+            "- Blockers: none",
+            "- Current status: implementing routing",
+            "- Safe next action: open PR",
+            "- Next: open PR",
+            "- Safety statement: no review/merge",
+        ])
+
+    def test_complete_work_issue_report_earns_a(self):
+        result = assess_work_issue_final_report(self._good_report())
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(result["downgraded"])
+
+    def test_missing_workflow_source_downgrades(self):
+        report = self._good_report().replace(
+            "workflows/work-issue.md", "SKILL.md only"
+        )
+        result = assess_work_issue_workflow_source(report)
+        self.assertTrue(result["downgraded"])
+
+    def test_review_field_in_handoff_downgrades(self):
+        report = self._good_report().replace(
+            "- Safety statement:",
+            "- Review decision: APPROVE\n- Safety statement:",
+        )
+        result = assess_work_issue_mode_isolation(report)
+        self.assertTrue(result["downgraded"])
+
+
 class TestValidationEnvironmentProof(unittest.TestCase):
     """Issue #311: exact validation command and execution environment."""
 

tests/test_reviewer_already_landed_classification.py

@@ -0,0 +1,81 @@
+import unittest
+
+from reviewer_already_landed_classification import (  # noqa: E402
+    ALREADY_LANDED_ELIGIBILITY_CLASS,
+    assess_already_landed_classification_report,
+)
+
+
+class TestAlreadyLandedClassification(unittest.TestCase):
+    def test_non_landed_report_passes(self):
+        result = assess_already_landed_classification_report(
+            "Selected the next eligible PR: PR #286."
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["landed_context"])
+
+    def test_oldest_eligible_pr_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "PR is already landed.\n"
+            "Oldest eligible PR: PR #278."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_next_eligible_pr_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Ancestor proof: passed\n"
+            "Selected the next eligible PR: PR #278."
+        )
+        self.assertFalse(result["proven"])
+
+    def test_eligible_for_review_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed on master and eligible for review next."
+        )
+        self.assertFalse(result["proven"])
+
+    def test_proper_landed_wording_passes(self):
+        result = assess_already_landed_classification_report(
+            "Oldest open PR requiring action: PR #278\n"
+            f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
+            "Candidate head SHA: abc123\n"
+            "Reviewed head SHA: none\n"
+            "Queue blocked by already-landed PR requiring reconciliation",
+            selected_pr_already_landed=True,
+        )
+        self.assertTrue(result["proven"])
+
+    def test_pinned_reviewed_head_without_validation_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed.\n"
+            "Pinned reviewed head: abc123def456",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_wrong_eligibility_class_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed.\n"
+            "Eligibility class: REVIEW_ELIGIBLE",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_missing_queue_blocker_blocks(self):
+        result = assess_already_landed_classification_report(
+            f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
+            "Candidate head SHA: abc123\n"
+            "Reviewed head SHA: none",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_already_landed_classification_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_validation_cwd_proof.py

@@ -0,0 +1,135 @@
+"""Tests for validation cwd/HEAD proof verifier (#398)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report  # noqa: E402
+
+ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
+WORKTREE = f"{ROOT}/branches/review-feat-issue-398"
+HEAD = "f5953549aad5e822f14f52d3ea3c6d7990109384"
+
+
+def _proof_backed_report() -> str:
+    return "\n".join([
+        f"Candidate head SHA: {HEAD}",
+        f"pwd: {WORKTREE}",
+        f"git rev-parse HEAD: {HEAD}",
+        "git status --short --branch: ## feat/issue-398...prgs/master",
+        f"Validation command: cd {WORKTREE} && venv/bin/python -m pytest tests/ -q",
+        "Result: 1497 passed, 6 skipped",
+    ])
+
+
+class TestValidationCwdProof(unittest.TestCase):
+    def test_no_validation_claim_passes(self):
+        result = assess_validation_cwd_proof_report("Review decision: approve")
+        self.assertTrue(result["proven"])
+
+    def test_missing_cwd_proof_fails(self):
+        result = assess_validation_cwd_proof_report(
+            f"Validation command: pytest tests/\nCandidate head SHA: {HEAD}",
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_main_checkout_cwd_blocks(self):
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                f"Candidate head SHA: {HEAD}",
+                f"pwd: {ROOT}",
+                f"git rev-parse HEAD: {HEAD}",
+                "git status --short --branch: ## master",
+                "Validation command: pytest tests/ -q",
+            ]),
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+            project_root=ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["violations"])
+
+    def test_wrong_head_blocks(self):
+        wrong = "a" * 40
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                f"Candidate head SHA: {HEAD}",
+                f"pwd: {WORKTREE}",
+                f"git rev-parse HEAD: {wrong}",
+                "git status --short --branch: clean",
+                f"Validation command: cd {WORKTREE} && pytest -q",
+            ]),
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+            project_root=ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["violations"])
+
+    def test_fully_proof_backed_passes(self):
+        result = assess_validation_cwd_proof_report(
+            _proof_backed_report(),
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+            project_root=ROOT,
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_baseline_without_cwd_fails(self):
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                _proof_backed_report(),
+                "Baseline validation command: pytest tests/ -q",
+            ]),
+            validation_session={
+                "validation_ran": True,
+                "expected_head_sha": HEAD,
+                "baseline_validation_ran": True,
+            },
+            project_root=ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("baseline" in r.lower() for r in result["reasons"])
+        )
+
+    def test_baseline_with_full_proof_passes(self):
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                _proof_backed_report(),
+                f"Baseline worktree: {ROOT}/branches/baseline-master-pr376",
+                f"Baseline target SHA: {HEAD}",
+                f"Baseline validation command: cd {ROOT}/branches/baseline-master-pr376 && pytest -q",
+            ]),
+            validation_session={
+                "validation_ran": True,
+                "expected_head_sha": HEAD,
+                "baseline_validation_ran": True,
+            },
+            project_root=ROOT,
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_final_report_validator_integration(self):
+        result = assess_final_report_validator(
+            "Validation command: pytest tests/ -q",
+            "review_pr",
+            validation_session={"validation_ran": True},
+        )
+        self.assertTrue(result["blocked"] or result["downgraded"])
+        self.assertTrue(
+            any(
+                f.get("rule_id") == "reviewer.validation_cwd_proof"
+                for f in result.get("findings") or []
+            )
+        )
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_validation_cwd_proof_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_session_router.py

@@ -192,6 +192,31 @@ class TestRoleSessionRouter(unittest.TestCase):
             )
         self.assertEqual(result.get("number"), 999)
 
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_work_issue_task_under_author_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            for task_type in ("work_issue", "work-issue"):
+                route = mcp_server.gitea_route_task_session(
+                    task_type=task_type, remote="prgs"
+                )
+                self.assertEqual(
+                    route["route_result"], role_session_router.ROUTE_ALLOWED
+                )
+                self.assertTrue(route["downstream_allowed"])
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_work_issue_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="work-issue", remote="prgs"
+            )
+        self.assertEqual(
+            route["route_result"], role_session_router.ROUTE_TO_AUTHOR
+        )
+        self.assertFalse(route["downstream_allowed"])
+
     @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
     def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):