Merge commit '4243b60ca32d79f1ee5e6b0f10d7570e9dea2937' into HEAD

author_duplicate_work_gate.py

@@ -1,198 +0,0 @@
-"""Early duplicate-work detection for author work-issue flows (#400)."""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-from issue_claim_heartbeat import (
-    _linked_open_pr,
-    _matching_branch_names,
-    classify_issue_claim,
-)
-
-STAGES = (
-    "claim",
-    "lock",
-    "worktree",
-    "edit",
-    "commit",
-    "push",
-    "create_pr",
-)
-
-ELIGIBILITY_OPEN_PR_EXISTS = "OPEN_PR_EXISTS"
-ELIGIBILITY_DUPLICATE_BRANCH_EXISTS = "DUPLICATE_BRANCH_EXISTS"
-ELIGIBILITY_ACTIVE_CLAIM = "ACTIVE_CLAIM_BY_OTHER"
-ELIGIBILITY_CLEAR = "CLEAR"
-
-
-def assess_author_duplicate_work(
-    issue_number: int,
-    *,
-    stage: str,
-    open_prs: list[dict] | None = None,
-    branch_names: list[str] | None = None,
-    claim_entry: dict | None = None,
-    matching_branches: list[str] | None = None,
-    allow_stale_takeover: bool = False,
-) -> dict[str, Any]:
-    """Fail closed when duplicate work is detected before author mutations (#400)."""
-    stage_norm = (stage or "").strip().lower()
-    if stage_norm not in STAGES:
-        return {
-            "allowed": False,
-            "block": True,
-            "eligibility_class": "INVALID_STAGE",
-            "stage": stage_norm or None,
-            "reasons": [f"unknown duplicate-work stage {stage!r}"],
-            "safe_next_action": f"use one of: {', '.join(STAGES)}",
-        }
-
-    prs = list(open_prs or [])
-    linked_pr = _linked_open_pr(int(issue_number), prs)
-    branches = list(
-        matching_branches
-        if matching_branches is not None
-        else _matching_branch_names(int(issue_number), list(branch_names or []))
-    )
-
-    reasons: list[str] = []
-    eligibility = ELIGIBILITY_CLEAR
-
-    if linked_pr:
-        eligibility = ELIGIBILITY_OPEN_PR_EXISTS
-        reasons.append(
-            f"open PR #{linked_pr.get('number')} already covers issue "
-            f"#{issue_number}"
-        )
-
-    if branches and stage_norm in {"claim", "lock", "worktree", "edit"}:
-        if eligibility == ELIGIBILITY_CLEAR:
-            eligibility = ELIGIBILITY_DUPLICATE_BRANCH_EXISTS
-        reasons.append(
-            f"remote branch(es) already exist for issue #{issue_number}: "
-            f"{', '.join(branches)}"
-        )
-
-    entry = claim_entry or {}
-    claim_status = (entry.get("status") or "").strip()
-    if claim_status in {"active", "awaiting_review"} and stage_norm == "claim":
-        if entry.get("reclaimable") and allow_stale_takeover:
-            pass
-        elif claim_status == "active" and not entry.get("reclaimable"):
-            if eligibility == ELIGIBILITY_CLEAR:
-                eligibility = ELIGIBILITY_ACTIVE_CLAIM
-            reasons.append(
-                f"issue #{issue_number} has active claim "
-                f"(status={claim_status})"
-            )
-    elif claim_status == "awaiting_review" and stage_norm == "claim":
-        if not linked_pr:
-            reasons.append(
-                f"issue #{issue_number} is awaiting_review but no linked open PR "
-                "was supplied for duplicate-work proof"
-            )
-
-    allowed = not reasons
-    outcome = "duplicate_work_prevented" if not allowed else "clear"
-    if stage_norm == "create_pr" and not allowed:
-        outcome = "duplicate_pr_prevented"
-    elif stage_norm in {"commit", "push"} and not allowed:
-        outcome = "duplicate_push_prevented" if stage_norm == "push" else "duplicate_commit_prevented"
-
-    return {
-        "allowed": allowed,
-        "block": not allowed,
-        "eligibility_class": eligibility if not allowed else ELIGIBILITY_CLEAR,
-        "stage": stage_norm,
-        "linked_open_pr": linked_pr.get("number") if linked_pr else None,
-        "matching_branches": branches,
-        "claim_status": claim_status or None,
-        "outcome": outcome,
-        "reasons": reasons,
-        "safe_next_action": (
-            "stop without edits/commit/push/PR; produce reconciliation handoff "
-            "preserving local work only"
-            if not allowed
-            else "proceed"
-        ),
-    }
-
-
-def build_claim_entry_from_classification(classification: dict) -> dict:
-    """Map ``classify_issue_claim`` output to gate claim metadata."""
-    return {
-        "status": classification.get("status"),
-        "reclaimable": classification.get("reclaimable"),
-        "linked_open_pr": classification.get("linked_open_pr"),
-    }
-
-
-_DUPLICATE_OUTCOME_RE = re.compile(
-    r"(duplicate\s+(?:pr|branch|commit|push)\s+prevented|"
-    r"duplicate\s+work\s+not\s+prevented|reconciliation\s+handoff)",
-    re.IGNORECASE,
-)
-
-
-def assess_work_issue_duplicate_prevention_report(report_text: str) -> dict:
-    """#400: work-issue reports must state duplicate-work prevention outcome."""
-    text = report_text or ""
-    if "duplicate work" not in text.lower() and "duplicate pr" not in text.lower():
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-    if _DUPLICATE_OUTCOME_RE.search(text):
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-    return {
-        "proven": False,
-        "block": True,
-        "reasons": [
-            "duplicate-work discussion must name a prevention outcome "
-            "(duplicate PR/branch/commit/push prevented, or duplicate work "
-            "not prevented, or reconciliation handoff)"
-        ],
-        "safe_next_action": "state exact duplicate-work prevention class in final report",
-    }
-
-
-def classify_and_assess(
-    issue: dict,
-    *,
-    stage: str,
-    comments: list[dict] | None = None,
-    open_prs: list[dict] | None = None,
-    branch_names: list[str] | None = None,
-    allow_stale_takeover: bool = False,
-) -> dict[str, Any]:
-    """Combine claim classification with duplicate-work gate assessment."""
-    issue_number = int(issue.get("number") or 0)
-    claim = classify_issue_claim(
-        issue=issue,
-        comments=comments or [],
-        open_prs=open_prs or [],
-        branch_names=branch_names or [],
-    )
-    assessment = assess_author_duplicate_work(
-        issue_number,
-        stage=stage,
-        open_prs=open_prs,
-        branch_names=branch_names,
-        claim_entry=build_claim_entry_from_classification(claim),
-        matching_branches=claim.get("matching_branches"),
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    return {
-        "issue_number": issue_number,
-        "claim": claim,
-        "duplicate_work": assessment,
-    }

author_mutation_worktree.py

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

docs/gitea-dual-namespace-deployment.md

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

docs/gitea-execution-profiles.md

@@ -226,12 +226,11 @@ explicit operator-directed closure of a contaminated PR). If `close_pr` ever
 resolves as unknown, agents must fail closed rather than fall back to the
 edit path.
 
-## Reconciler profile for already-landed open PRs (#304 / #310)
+## Reconciler profile for already-landed open PRs (#310)
 
-Normal author and reviewer profiles must not gain broad `gitea.pr.close`
-authority. Already-landed open PRs (head SHA is an ancestor of the target
-branch) need a dedicated reconciler profile such as `prgs-reconciler` with a
-narrow operation set:
+Normal author and reviewer profiles must not gain broad PR-close authority.
+Already-landed open PRs (head SHA is an ancestor of the target branch) need a
+dedicated reconciler profile such as `prgs-reconciler` with:
 
 - `gitea.read`
 - `gitea.pr.comment`
@@ -239,14 +238,8 @@ narrow operation set:
 - `gitea.issue.close`
 - `gitea.pr.close`
 
-Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
-`gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
-`gitea.repo.commit`.
-
-Launch a static `gitea-reconciler` MCP namespace with
-`GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
-`reconciler_profile.assess_reconciler_profile` (#304). Use the
-`gitea_reconcile_already_landed_pr` tool (#310). The resolver task is
+Use the `gitea-reconciler` MCP namespace (static profile launch) and the
+`gitea_reconcile_already_landed_pr` tool. The resolver task is
 `reconcile_already_landed_pr`. PR close is allowed only after live PR fetch,
 fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
 heads are not already landed cannot be closed through this path.

docs/llm-workflow-runbooks.md

@@ -273,20 +273,7 @@ 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. `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:
+mutations), `status:in-progress`, and claim comments. Full portable wording:
 [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 
 ## Global LLM Worktree Rule

docs/wiki/Workflow.md

@@ -17,10 +17,6 @@
 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).
@@ -35,4 +31,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,8 +92,7 @@ _ALREADY_LANDED_GATE_FIRED_RE = re.compile(
     re.IGNORECASE,
 )
 _ELIGIBLE_REVIEW_RE = re.compile(
-    r"eligible for (?:review|merge)|ready for (?:review|merge)|"
-    r"(?:next|oldest)\s+eligible\s+pr",
+    r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
     re.IGNORECASE,
 )
 _APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
@@ -122,37 +121,11 @@ _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",
-    "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",
+    "Linked issue live status",
     "Safe next action",
     "No review/merge confirmation",
 )
@@ -441,15 +414,13 @@ def _rule_reviewer_git_fetch_readonly(
                 "classify git fetch under Git ref mutations, not read-only diagnostics",
             )
         ]
-    if not git_ref_reported:
-        # Issue #307: an observed git fetch (report text or action log)
-        # must be listed under Git ref mutations.
+    if not git_ref_reported and _GIT_FETCH_RE.search(text):
         return [
             validator_finding(
                 "reviewer.git_fetch_readonly",
                 "downgrade",
                 "Git ref mutations",
-                "git fetch occurred without Git ref mutation classification",
+                "git fetch mentioned without Git ref mutation classification",
                 "record git fetch under Git ref mutations",
             )
         ]
@@ -697,18 +668,10 @@ def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]
     return []
 
 
-def _rule_reconcile_stale_author_fields(
-    report_text: str,
-    *,
-    session_pr_opened: bool = False,
-) -> list[dict[str, str]]:
+def _rule_reconcile_stale_author_fields(report_text: str) -> 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(
@@ -969,7 +932,6 @@ def assess_final_report_validator(
     mutations_observed: bool = False,
     local_edits: bool = False,
     issue_filing_lock: dict | None = None,
-    session_pr_opened: bool = False,
 ) -> dict[str, Any]:
     """Validate final-report text against task-specific proof rules (#327).
 
@@ -1024,7 +986,6 @@ def assess_final_report_validator(
         "action_log": action_log,
         "mutations_observed": mutations_observed,
         "local_edits": local_edits,
-        "session_pr_opened": session_pr_opened,
     }
 
     for rule in _RULES_BY_TASK.get(normalized_kind, ()):

gitea_mcp_server.py

@@ -21,7 +21,6 @@ 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
@@ -398,28 +397,6 @@ 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
@@ -449,33 +426,32 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 "file edits before mutation (fail closed). "
                 f"{_format_preflight_workspace_details(details)}"
             )
-    else:
-        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)}"
-            )
+        return
 
-        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)}"
-                )
+    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)}"
+        )
 
-    _enforce_branches_only_author_mutation(worktree_path)
+    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)}"
+            )
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -501,12 +477,8 @@ 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 author_duplicate_work_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
-import reconciler_profile  # noqa: E402
-import reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
 
@@ -514,132 +486,6 @@ 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:
@@ -1050,76 +896,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
-    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
-    return [_branch_entry_name(branch) for branch in branches]
-
-
-def _gather_author_duplicate_work_context(
-    issue_number: int,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    base = repo_api_url(h, o, r)
-    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
-    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
-    branch_names = _list_repo_branch_names(h, o, r, auth)
-    if exclude_branch_name:
-        branch_names = [
-            name for name in branch_names
-            if name != exclude_branch_name
-        ]
-    return {
-        "issue": issue,
-        "comments": comments,
-        "open_prs": open_prs,
-        "branch_names": branch_names,
-    }
-
-
-def _enforce_author_duplicate_work_gate(
-    issue_number: int,
-    stage: str,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Fail closed when duplicate work is detected (#400)."""
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    if assessment.get("block"):
-        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
-        raise RuntimeError(
-            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
-            f"{reasons} (fail closed)"
-        )
-    return result
-
-
 @mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
@@ -1157,16 +933,6 @@ 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(
@@ -1182,24 +948,36 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
+    # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
-        issue_number,
-        "lock",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=branch_name,
-    )
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    try:
+        prs = api_get_all(url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+
+    for pr in prs:
+        pr_head = pr.get("head", {}).get("ref", "")
+        pr_title = pr.get("title", "")
+        pr_body = pr.get("body", "")
+
+        if expected_pattern in pr_head:
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
+            )
+
+        patterns = [
+            f"closes #{issue_number}",
+            f"fixes #{issue_number}",
+        ]
+        text_to_check = f"{pr_title} {pr_body}".lower()
+        if any(p in text_to_check for p in patterns):
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (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,
@@ -1207,7 +985,6 @@ def gitea_lock_issue(
         "org": o,
         "repo": r,
         "worktree_path": resolved_worktree,
-        "work_lease": work_lease,
     }
 
     try:
@@ -1228,7 +1005,6 @@ 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"] = [
@@ -1285,7 +1061,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196) ──
@@ -1313,17 +1089,6 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
-    auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
-        int(locked_issue),
-        "create_pr",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=locked_branch,
-    )
-
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -1340,6 +1105,7 @@ def gitea_create_pr(
             f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
         )
 
+    auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
     meta = {"title": title, "head": head, "base": base}
@@ -3663,178 +3429,6 @@ 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,
@@ -4014,6 +3608,7 @@ def gitea_reconcile_already_landed_pr(
 
     return result
 
+
 @mcp.tool()
 def gitea_close_issue(
     issue_number: int,
@@ -4523,15 +4118,12 @@ def gitea_create_issue_comment(
 def _role_kind(allowed, forbidden) -> str:
     """Classify the active profile from its normalized operations.
 
-    'reconciler' can close already-landed PRs without review/author powers;
     'author' can create PRs / push branches; 'reviewer' can approve/merge;
     'reconciler' can close already-landed PRs via ``gitea.pr.close`` without
     review/author powers; 'mixed' can do both (a config smell — the
     reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do
     neither.
     """
-    if reconciler_profile.is_reconciler_profile(allowed, forbidden):
-        return "reconciler"
     def can(op):
         return gitea_config.check_operation(op, allowed, forbidden)[0]
     review = can("gitea.pr.approve") or can("gitea.pr.merge")
@@ -4799,28 +4391,6 @@ _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 "
@@ -5345,7 +4915,6 @@ _RUNTIME_CAPABILITY_TASKS = (
     "create_issue",
     "comment_issue",
     "create_pr",
-    "work_issue",
     "review_pr",
     "merge_pr",
     "close_issue",
@@ -5398,7 +4967,6 @@ 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",
@@ -5551,35 +5119,6 @@ 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,
-) -> 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,
-    )
-
-
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",
@@ -5698,10 +5237,6 @@ def gitea_get_runtime_context(
         "preflight_block_reasons": preflight["preflight_block_reasons"],
         "preflight_workspace": preflight.get("preflight_workspace"),
         "session_capabilities": session_capabilities,
-        "reconciler_profile_assessment": reconciler_profile.assess_reconciler_profile(
-            allowed, forbidden
-        ),
-        "role_kind": _role_kind(allowed, forbidden),
         "shell_health": native_mcp_preference.shell_health_status(),
     }
 
@@ -6035,14 +5570,6 @@ def gitea_mark_issue(
         )
 
     if action == "start":
-        _enforce_author_duplicate_work_gate(
-            issue_number,
-            "claim",
-            h=h,
-            o=o,
-            r=r,
-            auth=auth,
-        )
         with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
                       issue_number=issue_number,
                       request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -6124,65 +5651,6 @@ def gitea_post_heartbeat(
     )
 
 
-@mcp.tool()
-def gitea_assess_author_duplicate_work(
-    issue_number: int,
-    stage: str,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Read-only: assess duplicate-work risk before author mutations (#400).
-
-    Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
-    """
-    read_block = _profile_operation_gate("gitea.read")
-    if read_block:
-        return {
-            "success": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    return {
-        "success": True,
-        "performed": False,
-        "issue_number": issue_number,
-        "stage": stage,
-        "allowed": assessment.get("allowed"),
-        "block": assessment.get("block"),
-        "eligibility_class": assessment.get("eligibility_class"),
-        "outcome": assessment.get("outcome"),
-        "linked_open_pr": assessment.get("linked_open_pr"),
-        "matching_branches": assessment.get("matching_branches"),
-        "claim_status": assessment.get("claim_status"),
-        "reasons": assessment.get("reasons"),
-        "safe_next_action": assessment.get("safe_next_action"),
-        "claim": result.get("claim"),
-    }
-
-
 @mcp.tool()
 def gitea_reconcile_issue_claims(
     state: str = "open",

reconciler_profile.py

@@ -1,94 +0,0 @@
-"""Reconciler profile model for already-landed PR closure (#304).
-
-Defines the narrowly scoped operation set for a dedicated reconciler profile
-such as ``prgs-reconciler``. Close MCP tooling and ancestry gates ship in the
-#310 stack; this module validates profile shape only.
-"""
-
-from __future__ import annotations
-
-import gitea_config
-
-RECONCILER_REQUIRED_OPERATIONS = (
-    "gitea.read",
-    "gitea.pr.close",
-)
-
-RECONCILER_RECOMMENDED_OPERATIONS = (
-    "gitea.pr.comment",
-    "gitea.issue.comment",
-    "gitea.issue.close",
-)
-
-RECONCILER_FORBIDDEN_OPERATIONS = (
-    "gitea.pr.approve",
-    "gitea.pr.merge",
-    "gitea.pr.review",
-    "gitea.pr.create",
-    "gitea.branch.push",
-    "gitea.repo.commit",
-)
-
-
-def _normalized_allowed(allowed: list[str]) -> set[str]:
-    normalized: set[str] = set()
-    for entry in allowed or []:
-        try:
-            normalized.add(gitea_config.normalize_operation(entry))
-        except gitea_config.ConfigError:
-            continue
-    return normalized
-
-
-def _forbidden_in_allowed(allowed: list[str], ops: tuple[str, ...]) -> list[str]:
-    """Return forbidden ops that are explicitly listed in *allowed*."""
-    allowed_n = _normalized_allowed(allowed)
-    present: list[str] = []
-    for op in ops:
-        try:
-            if gitea_config.normalize_operation(op) in allowed_n:
-                present.append(op)
-        except gitea_config.ConfigError:
-            continue
-    return present
-
-
-def is_reconciler_profile(allowed: list[str], forbidden: list[str]) -> bool:
-    """Return True when *allowed*/*forbidden* describe a reconciler profile."""
-    def can(op: str) -> bool:
-        return gitea_config.check_operation(op, allowed, forbidden)[0]
-
-    if not can("gitea.pr.close"):
-        return False
-    if _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
-        return False
-    return True
-
-
-def assess_reconciler_profile(allowed: list[str], forbidden: list[str]) -> dict:
-    """Validate reconciler profile operations (read-only, fail closed)."""
-    allowed = list(allowed or [])
-    forbidden = list(forbidden or [])
-    reasons: list[str] = []
-
-    for op in RECONCILER_REQUIRED_OPERATIONS:
-        ok, _ = gitea_config.check_operation(op, allowed, forbidden)
-        if not ok:
-            reasons.append(f"missing required operation {op}")
-
-    for op in _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
-        reasons.append(f"forbidden operation must not be allowed: {op}")
-
-    missing_recommended = [
-        op for op in RECONCILER_RECOMMENDED_OPERATIONS
-        if not gitea_config.check_operation(op, allowed, forbidden)[0]
-    ]
-
-    return {
-        "is_reconciler_profile": is_reconciler_profile(allowed, forbidden),
-        "valid": not reasons and is_reconciler_profile(allowed, forbidden),
-        "reasons": reasons,
-        "missing_recommended_operations": missing_recommended,
-        "required_operations": list(RECONCILER_REQUIRED_OPERATIONS),
-        "forbidden_operations": list(RECONCILER_FORBIDDEN_OPERATIONS),
-    }

reconciliation_workflow.py

@@ -1,292 +0,0 @@
-"""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

@@ -1,319 +0,0 @@
-"""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,
-) -> 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,
-    )
-    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,11 +18,6 @@ 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}$")
@@ -1747,17 +1742,6 @@ 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"))
@@ -1937,11 +1921,6 @@ 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
@@ -2056,12 +2035,6 @@ 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 []
-        ),
     }
 
 
@@ -2328,30 +2301,6 @@ 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."""
@@ -3544,122 +3493,6 @@ 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_duplicate_prevention_report(report_text, **kwargs):
-    """#400: work-issue reports must classify duplicate-work prevention."""
-    from author_duplicate_work_gate import (
-        assess_work_issue_duplicate_prevention_report as _assess,
-    )
-
-    return _assess(report_text, **kwargs)
-
-
-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),
-        "duplicate_prevention": assess_work_issue_duplicate_prevention_report(
-            report_text
-        ),
-    }
-
-    reasons = []
-    downgraded = False
-    for name, result in checks.items():
-        verdict = result.get("verdict")
-        if result.get("block"):
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif 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 [])
-            )
-        elif result.get("proven") is False:
-            downgraded = True
-            reasons.extend(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)
@@ -5076,13 +4909,6 @@ 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",
@@ -5505,15 +5331,6 @@ def assess_validation_integrity_report(report_text, **kwargs):
     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

@@ -1,143 +0,0 @@
-"""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"
-        ),
-    }

role_session_router.py

@@ -76,7 +76,6 @@ AUTHOR_TASKS = frozenset({
     "delete_branch",
     "work_issue",
     "work-issue",
-    "reconcile_landed_pr",
 })
 
 RECONCILER_TASKS = frozenset({
@@ -105,7 +104,6 @@ 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/work-issue.md

@@ -83,20 +83,6 @@ 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:
@@ -277,24 +263,6 @@ Do not select an issue based only on memory from a previous session.
 
 Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue.
 
-Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true:
-
-* `claim` — before `gitea_mark_issue`
-* `lock` — before `gitea_lock_issue`
-* `worktree` / `edit` — before creating a worktree or editing files
-* `commit` — immediately before `git commit`
-* `push` — immediately before `git push`
-* `create_pr` — immediately before `gitea_create_pr` (also enforced server-side)
-
-`gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed.
-
-If a concurrent open PR appears after work begins:
-
-* before commit or push — stop and preserve local work without pushing
-* after push but before PR creation — produce a reconciliation handoff instead of opening a PR
-
-Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`).
-
 If an open PR already exists for the issue, do not implement duplicate work.
 
 Classify the issue as:

task_capability_map.py

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

@@ -1,102 +0,0 @@
-"""Tests for early author duplicate-work gate (#400)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from author_duplicate_work_gate import (  # noqa: E402
-    ELIGIBILITY_OPEN_PR_EXISTS,
-    assess_author_duplicate_work,
-    assess_work_issue_duplicate_prevention_report,
-)
-
-
-def _open_pr(number: int = 397, issue: int = 395) -> dict:
-    return {
-        "number": number,
-        "head": {"ref": f"feat/issue-{issue}-example"},
-        "title": f"feat: example (Closes #{issue})",
-        "body": f"Closes #{issue}",
-    }
-
-
-class TestAuthorDuplicateWorkGate(unittest.TestCase):
-    def test_clear_when_no_duplicates(self):
-        result = assess_author_duplicate_work(
-            400,
-            stage="claim",
-            open_prs=[],
-            branch_names=["master", "feat/issue-399-other"],
-        )
-        self.assertTrue(result["allowed"])
-        self.assertFalse(result["block"])
-
-    def test_open_pr_blocks_claim(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="claim",
-            open_prs=[_open_pr()],
-            branch_names=[],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["eligibility_class"], ELIGIBILITY_OPEN_PR_EXISTS)
-
-    def test_matching_branch_blocks_lock_not_create_pr(self):
-        branches = ["feat/issue-400-early-duplicate-work-gate"]
-        lock = assess_author_duplicate_work(
-            400,
-            stage="lock",
-            open_prs=[],
-            branch_names=branches,
-        )
-        self.assertFalse(lock["allowed"])
-
-        create_pr = assess_author_duplicate_work(
-            400,
-            stage="create_pr",
-            open_prs=[],
-            branch_names=branches,
-            matching_branches=[],
-        )
-        self.assertTrue(create_pr["allowed"])
-
-    def test_open_pr_blocks_create_pr_stage(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="create_pr",
-            open_prs=[_open_pr()],
-            branch_names=["feat/issue-395-proof-backed-review-handoff"],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["outcome"], "duplicate_pr_prevented")
-
-    def test_push_stage_blocks_on_concurrent_pr(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="push",
-            open_prs=[_open_pr()],
-            branch_names=[],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["outcome"], "duplicate_push_prevented")
-
-    def test_duplicate_prevention_report_requires_outcome(self):
-        bad = assess_work_issue_duplicate_prevention_report(
-            "Duplicate work detected for issue #395."
-        )
-        self.assertFalse(bad["proven"])
-
-        good = assess_work_issue_duplicate_prevention_report(
-            "Duplicate PR prevented; reconciliation handoff produced."
-        )
-        self.assertTrue(good["proven"])
-
-    def test_exported_from_review_proofs(self):
-        from review_proofs import assess_work_issue_duplicate_prevention_report as exported
-
-        self.assertTrue(callable(exported))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_author_mutation_worktree.py

@@ -1,109 +0,0 @@
-"""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,15 +55,7 @@ class TestCommitPayloads(unittest.TestCase):
         with open(self.config_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(CONFIG))
 
-        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_dir = tempfile.TemporaryDirectory()
         self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
 
         self.lock_file_path = "/tmp/gitea_issue_lock.json"
@@ -102,7 +94,6 @@ 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,47 +57,26 @@ def _review_handoff(**overrides):
     return text
 
 
-def _reconcile_handoff(drop=(), **overrides):
+def _reconcile_handoff(**overrides):
     lines = [
         "## Controller Handoff",
         "",
         "- Task: reconcile already-landed PR #99",
         "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-        "- Role/profile: reconciler / prgs-reconciler",
-        "- Identity: sysadmin",
+        "- Role: reconciler",
+        "- Identity: sysadmin / prgs-author",
         "- Selected PR: #99",
-        "- 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",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
         "- 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",
-        "- Blocker: missing gitea.pr.close capability",
-        "- Safe next action: hand off to a profile with gitea.pr.close",
+        "- Safe next action: none",
         "- No review/merge confirmation: confirmed",
     ]
-    kept = [
-        line
-        for line in lines
-        if not any(line.startswith(f"- {name}:") for name in drop)
-    ]
-    text = "\n".join(kept)
+    text = "\n".join(lines)
     for key, value in overrides.items():
         if f"- {key}:" in text:
             text = text.replace(f"- {key}:", f"- {key}: {value}")
@@ -215,20 +194,6 @@ 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"),
@@ -277,6 +242,7 @@ 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()
@@ -341,159 +307,6 @@ 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

@@ -95,19 +95,11 @@ def test_create_issue_workflow_contract():
     assert "## 9. Duplicate search before mutation" in text
 
 
-def test_author_duplicate_work_gate_exported():
-    from review_proofs import assess_work_issue_duplicate_prevention_report
-
-    assert callable(assess_work_issue_duplicate_prevention_report)
-
-
 def test_work_issue_workflow_contract():
     text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
     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():
@@ -147,16 +139,6 @@ 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"
@@ -183,24 +165,12 @@ 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
 
@@ -273,12 +243,6 @@ 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,18 +3050,10 @@ 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:
@@ -3102,51 +3094,6 @@ 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_reconciler_profile.py

@@ -1,87 +0,0 @@
-"""Tests for reconciler profile model (#304)."""
-
-import os
-import sys
-import unittest
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-import gitea_mcp_server as mcp_server
-import migrate_profiles
-import reconciler_profile
-
-
-PRGS_RECONCILER_ALLOWED = [
-    "gitea.read",
-    "gitea.pr.comment",
-    "gitea.issue.comment",
-    "gitea.issue.close",
-    "gitea.pr.close",
-]
-PRGS_RECONCILER_FORBIDDEN = [
-    "gitea.pr.approve",
-    "gitea.pr.merge",
-    "gitea.pr.create",
-    "gitea.branch.push",
-]
-
-
-class TestReconcilerProfileModel(unittest.TestCase):
-    def test_prgs_reconciler_shape_valid(self):
-        result = reconciler_profile.assess_reconciler_profile(
-            PRGS_RECONCILER_ALLOWED,
-            PRGS_RECONCILER_FORBIDDEN,
-        )
-        self.assertTrue(result["is_reconciler_profile"])
-        self.assertTrue(result["valid"])
-        self.assertEqual(result["reasons"], [])
-
-    def test_author_profile_not_reconciler(self):
-        allowed = [
-            "gitea.read",
-            "gitea.pr.create",
-            "gitea.branch.push",
-            "gitea.issue.comment",
-        ]
-        forbidden = ["gitea.pr.approve", "gitea.pr.merge"]
-        self.assertFalse(
-            reconciler_profile.is_reconciler_profile(allowed, forbidden)
-        )
-
-    def test_reviewer_profile_not_reconciler(self):
-        allowed = [
-            "gitea.read",
-            "gitea.pr.review",
-            "gitea.pr.approve",
-            "gitea.pr.merge",
-        ]
-        forbidden = ["gitea.pr.create", "gitea.branch.push"]
-        self.assertFalse(
-            reconciler_profile.is_reconciler_profile(allowed, forbidden)
-        )
-
-    def test_broad_close_on_author_invalid(self):
-        allowed = PRGS_RECONCILER_ALLOWED + ["gitea.pr.create"]
-        result = reconciler_profile.assess_reconciler_profile(
-            allowed,
-            PRGS_RECONCILER_FORBIDDEN,
-        )
-        self.assertFalse(result["valid"])
-        self.assertFalse(result["is_reconciler_profile"])
-
-    def test_role_kind_detects_reconciler(self):
-        role = mcp_server._role_kind(
-            PRGS_RECONCILER_ALLOWED,
-            PRGS_RECONCILER_FORBIDDEN,
-        )
-        self.assertEqual(role, "reconciler")
-
-    def test_migrate_infer_role_reconciler(self):
-        self.assertEqual(
-            migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
-            "reconciler",
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reconciliation_mcp_tools.py

@@ -1,106 +0,0 @@
-"""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

@@ -1,149 +0,0 @@
-"""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,32 +121,6 @@ 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

@@ -1,158 +0,0 @@
-"""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,9 +40,6 @@ 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,
@@ -2323,52 +2320,6 @@ 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

@@ -1,81 +0,0 @@
-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_role_session_router.py

@@ -192,31 +192,6 @@ 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):