feat(author): add capability, secret sweep, and author PR report proof helpers and tests (#183 )

feat(author): harden reporting and claim capability per #183
- Add 'mark_issue' to TASK_MAP in resolve_task_capability for exact mutation proof (closes gap on unknown treated as allowed). - Add assess_controller_handoff and integrate into build_final_report (downgrades missing handoff). - Add tests for handoff and update existing. - Update SKILL.md and review-pr.md to mandate exact 'Controller Handoff' section, exact sweep evidence, PR head SHA in reports. - Author profile used throughout; no review/merge. Refs #183
2026-07-05 15:36:12 -04:00 · 2026-07-05 15:35:32 -04:00
15 changed files with 288 additions and 1985 deletions
@@ -1,217 +0,0 @@
 """Fail-closed branch-identity proofs for author workflows (#177).
 Author-side counterpart of the reviewer proofs in ``review_proofs.py``
 (#173). During the #173 implementation itself, a commit landed on local
 ``master`` because the shared checkout's branch moved mid-session (origin
 incident of #177). These helpers turn that from an after-the-fact repair
 into a fail-closed gate: an author workflow must prove its local git state
 before staging, committing, or pushing.
 The helpers are pure (no git calls): the workflow gathers the raw facts
 (``git branch --show-current``, ``git rev-parse HEAD``, the push refspec,
 the branch named in the issue claim) and passes them in, so the same logic
 works from prompts, harness assertions, and tests. Shared-worktree branch
 switches by other sessions are treated as expected events to detect, not
 exceptional ones. Nothing here weakens the review/merge/permission gates.
 """
 PROTECTED_BRANCHES = frozenset(
    {"master", "main", "develop", "development", "dev"}
 )
 def _clean(name):
    return (name or "").strip()
 def verify_branch_for_commit(current_branch, intended_branch):
    """Required behavior 1: prove the branch before staging/committing.
    Proven only when both names are present, the intended branch is not a
    protected branch, and the current branch equals the intended one (which
    also rules out being on any protected branch). Returns {'proven',
    'block', 'reasons', 'current_branch', 'intended_branch'}.
    """
    reasons = []
    current = _clean(current_branch)
    intended = _clean(intended_branch)
    if not current:
        reasons.append(
            "current branch unknown (detached HEAD or state not read); "
            "fail closed"
        )
    if not intended:
        reasons.append("intended feature branch not stated; fail closed")
    if intended and intended in PROTECTED_BRANCHES:
        reasons.append(
            f"intended branch '{intended}' is a protected branch; author "
            "work must target a feature branch"
        )
    if current and current in PROTECTED_BRANCHES:
        reasons.append(
            f"current branch '{current}' is a protected branch; committing "
            "here is blocked"
        )
    if current and intended and current != intended:
        reasons.append(
            f"current branch '{current}' is not the intended feature branch "
            f"'{intended}'; stop before staging/committing"
        )
    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "current_branch": current or None,
        "intended_branch": intended or None,
    }
 def detect_branch_drift(branch_at_validation, head_at_validation,
                        current_branch, current_head):
    """Required behaviors 2–3: stop when branch or HEAD moved mid-session.
    Compares the branch name and HEAD SHA captured at validation time with
    the state observed immediately before commit/push. Any difference —
    including an external branch switch in a shared worktree — is drift and
    blocks until reconciled. Missing state fails closed.
    """
    reasons = []
    branch_then = _clean(branch_at_validation)
    branch_now = _clean(current_branch)
    head_then = _clean(head_at_validation).lower()
    head_now = _clean(current_head).lower()
    if not branch_then or not head_then:
        reasons.append("validation-time branch/HEAD not recorded; fail closed")
    if not branch_now or not head_now:
        reasons.append("current branch/HEAD not read; fail closed")
    if branch_then and branch_now and branch_then != branch_now:
        reasons.append(
            f"branch changed from '{branch_then}' to '{branch_now}' since "
            "validation — possible external branch switch in a shared "
            "worktree; stop and reconcile before committing"
        )
    if head_then and head_now and head_then != head_now:
        reasons.append(
            "HEAD moved since validation; re-validate on the current HEAD "
            "before committing"
        )
    drifted = bool(reasons)
    return {"drifted": drifted, "block": drifted, "reasons": reasons}
 def verify_push_target(current_branch, remote_target_branch, intended_branch):
    """Acceptance: a push needs local, remote, and intended branches to match.
    Proven only when all three names are present, equal, and not a
    protected branch — a feature-branch workflow never pushes a protected
    branch, and never pushes to a refspec other than its own branch.
    """
    reasons = []
    current = _clean(current_branch)
    remote_target = _clean(remote_target_branch)
    intended = _clean(intended_branch)
    if not current:
        reasons.append("current branch unknown; fail closed")
    if not remote_target:
        reasons.append("remote target branch not stated; fail closed")
    if not intended:
        reasons.append("intended feature branch not stated; fail closed")
    for label, name in (("current", current), ("remote target", remote_target),
                        ("intended", intended)):
        if name and name in PROTECTED_BRANCHES:
            reasons.append(
                f"{label} branch '{name}' is a protected branch; author "
                "pushes to protected branches are blocked"
            )
    if current and remote_target and current != remote_target:
        reasons.append(
            f"push target '{remote_target}' does not match the local branch "
            f"'{current}'"
        )
    if current and intended and current != intended:
        reasons.append(
            f"local branch '{current}' does not match the intended feature "
            f"branch '{intended}'"
        )
    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
    }
 def assess_protected_branch_commit(commit_branch, pushed=False,
                                   repair_reported=True):
    """Required behavior 4: handle an accidental protected-branch commit.
    If a commit landed on a protected branch: it must never be pushed, a
    repair is required, and the repair must be *reported* — silently
    continuing after (or without) repair is a violation, as is having
    pushed the accident.
    """
    branch = _clean(commit_branch)
    accident = branch in PROTECTED_BRANCHES
    violations = []
    if accident:
        if pushed:
            violations.append(
                f"accidental commit on protected branch '{branch}' was "
                "pushed; protected-branch pushes are forbidden"
            )
        if not repair_reported:
            violations.append(
                "protected-branch commit repair was not reported; the "
                "workflow must surface the accident and the repair steps, "
                "never silently continue"
            )
    return {
        "accident": accident,
        "must_not_push": accident,
        "repair_required": accident,
        "violations": violations,
    }
 def build_commit_push_report(commit_proof, drift, push_proof, accident=None):
    """Acceptance: final report carries branch proof before commit and push.
    Combines the individual proofs; any failed proof, detected drift, or
    accident violation makes the status 'blocked' — the workflow stops and
    reports instead of continuing.
    """
    accident = accident or {"accident": False, "violations": []}
    violations = list(accident.get("violations", []))
    blocked = (
        not commit_proof.get("proven")
        or drift.get("drifted")
        or not push_proof.get("proven")
        or bool(violations)
    )
    return {
        "status": "blocked" if blocked else "ok",
        "branch_proof_before_commit": bool(commit_proof.get("proven")),
        "branch_proof_before_push": bool(push_proof.get("proven")),
        "drift_detected": bool(drift.get("drifted")),
        "protected_branch_accident": bool(accident.get("accident")),
        "violations": violations,
        "reasons": (
            list(commit_proof.get("reasons", []))
            + list(drift.get("reasons", []))
            + list(push_proof.get("reasons", []))
        ),
    }
@@ -16,70 +16,10 @@ Configuration (mcp_config.json):
 import os
 import re
 import sys
 import json
 import functools
 import contextlib
 import subprocess
 LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None):
    """Record the resolved capability context to fail-closed lock file."""
    data = {
        "initial_profile": profile_name,
        "initial_identity": identity,
        "current_profile": profile_name,
        "current_identity": identity,
        "remote": remote,
        "task": task,
        "role_pivot_authorized": False,
        "role_pivot_record": None,
    }
    try:
        with open(LOCK_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f)
    except Exception:
        pass
 def verify_mutation_authority(remote: str | None, host: str | None = None, required_role: str = "reviewer"):
    """Verify that the current mutation matches the locked capability context."""
    if not os.path.exists(LOCK_FILE):
        raise RuntimeError("Mutation authority lock is missing (fail closed)")
    try:
        with open(LOCK_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        raise RuntimeError(f"Could not read mutation authority lock: {e} (fail closed)")
    if data.get("remote") != remote:
        raise RuntimeError(
            f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)"
        )
    profile = get_profile()
    active_profile = profile.get("profile_name")
    h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
    active_identity = _authenticated_username(h) if h else None
    locked_profile = data.get("current_profile")
    locked_identity = data.get("current_identity")
    if active_profile != locked_profile or active_identity != locked_identity:
        raise RuntimeError(
            f"Mutation profile '{active_profile}' or identity '{active_identity}' "
            f"does not match locked authority (profile: '{locked_profile}', identity: '{locked_identity}') (fail closed)"
        )
    # Check reviewer/author role pivot boundaries
    if required_role == "reviewer" and "author" in str(data.get("initial_profile")).lower() and "reviewer" in str(active_profile).lower():
        if not data.get("role_pivot_authorized"):
            raise RuntimeError(
                "Attempted reviewer mutation from author session without authorized role pivot (fail closed)"
            )
 # Resolve the project root. MCP clients must launch this script directly with
 # the venv interpreter (venv/bin/python3) — see the config example above. We do
 # NOT os.execv() to re-point the interpreter: replacing the process after the
@@ -91,100 +31,6 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
 if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)
 PREFLIGHT_FILE = "/tmp/gitea_preflight_check.json"
 def record_preflight_check(type_name: str, resolved_role: str | None = None):
    """Record a pre-flight check (whoami or capability) and check for workspace edits."""
    import time
    is_dirty = False
    in_test = "pytest" in sys.modules or "unittest" in sys.modules
    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
        is_dirty = False
    else:
        try:
            res = subprocess.run(
                ["git", "status", "--porcelain"],
                capture_output=True, text=True, cwd=PROJECT_ROOT
            )
            for line in res.stdout.splitlines():
                if line and not line.startswith("??"):
                    is_dirty = True
                    break
        except Exception:
            pass
    data = {}
    if os.path.exists(PREFLIGHT_FILE):
        try:
            with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            pass
    if is_dirty:
        if type_name == "whoami" and not data.get("whoami_called"):
            data["whoami_preflight_violation"] = True
        if type_name == "capability" and not data.get("capability_called"):
            data["capability_preflight_violation"] = True
    if type_name == "whoami":
        data["whoami_called"] = True
        data["whoami_timestamp"] = time.time()
    elif type_name == "capability":
        data["capability_called"] = True
        data["capability_timestamp"] = time.time()
        if resolved_role:
            data["role"] = resolved_role
    try:
        with open(PREFLIGHT_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f)
    except Exception:
        pass
 def verify_preflight_purity(remote: str | None = None):
    """Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
    in_test = "pytest" in sys.modules or "unittest" in sys.modules
    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
        return
    if not os.path.exists(PREFLIGHT_FILE):
        raise RuntimeError("Pre-flight order violation: Identity and capability verification were skipped (fail closed)")
    try:
        with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        raise RuntimeError(f"Could not read pre-flight check record: {e} (fail closed)")
    if not data.get("whoami_called"):
        raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
    if not data.get("capability_called"):
        raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
    if data.get("whoami_preflight_violation"):
        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
    if data.get("capability_preflight_violation"):
        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
    is_dirty = False
    if os.environ.get("GITEA_TEST_ENVIRONMENT") == "1" and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
        is_dirty = False
    else:
        try:
            res = subprocess.run(
                ["git", "status", "--porcelain"],
                capture_output=True, text=True, cwd=PROJECT_ROOT
            )
            for line in res.stdout.splitlines():
                if line and not line.startswith("??"):
                    is_dirty = True
                    break
        except Exception:
            pass
    if data.get("role") == "reviewer" and is_dirty:
        raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 from gitea_auth import (  # noqa: E402
@@ -479,7 +325,6 @@ def gitea_create_issue(
        dict with 'number' of the created issue ('url' only with the reveal opt-in).
    """
    h, o, r = _resolve(remote, host, org, repo)
    verify_preflight_purity(remote)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/issues"
    try:
@@ -495,84 +340,6 @@ def gitea_create_issue(
    return _with_optional_url({"number": data["number"]}, data.get("html_url"))
@mcp.tool()
 def gitea_lock_issue(
    issue_number: int,
    branch_name: str,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
 ) -> dict:
    """Lock exactly one Gitea issue and its branch name to ensure durable tracking.
    Args:
        issue_number: The tracking issue number.
        branch_name: The branch name (must match (fix|feat|docs|chore)/issue-<issue_number>-<desc>).
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override Gitea host.
        org: Override Org.
        repo: Override Repo.
    """
    # 1. Enforce branch name includes issue number
    expected_pattern = f"issue-{issue_number}"
    if expected_pattern not in branch_name:
        raise ValueError(
            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
        )
    # 2. Check if the issue already has an open PR (reuse protection)
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    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)"
            )
    data = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "remote": remote,
        "org": o,
        "repo": r,
    }
    try:
        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f)
    except Exception as e:
        raise RuntimeError(f"Could not write issue lock file: {e}")
    return {
        "success": True,
        "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
        "issue_number": issue_number,
        "branch_name": branch_name,
    }
@mcp.tool()
 def gitea_create_pr(
    title: str,
@@ -600,42 +367,6 @@ def gitea_create_pr(
        dict with 'number' of the created PR ('url' only with the reveal opt-in).
    """
    h, o, r = _resolve(remote, host, org, repo)
    verify_preflight_purity(remote)
    # ── Issue Lock Validation (Issue #194 / #196) ──
    if not os.path.exists(ISSUE_LOCK_FILE):
        raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
    try:
        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
            lock_data = json.load(f)
    except Exception as e:
        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
    locked_issue = lock_data.get("issue_number")
    locked_branch = lock_data.get("branch_name")
    if head != locked_branch:
        raise ValueError(
            f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
        )
    # Check for forbidden terms anywhere in title/body
    forbidden_terms = ["equivalent", "related", "same as"]
    text_to_check = f"{title} {body}".lower()
    for term in forbidden_terms:
        if term in text_to_check:
            raise ValueError(
                f"PR title or body contains forbidden term '{term}' (fail closed)"
            )
    # Ensure Closes #<locked_issue> or Fixes #<locked_issue> is present exactly
    closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE)
    if not closes_pattern.search(text_to_check):
        raise ValueError(
            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}
@@ -1245,7 +976,6 @@ def gitea_submit_pr_review(
        authenticated user, profile name, PR author, PR number, head SHA
        checked, and the reasons/gates passed or blocked. Never secrets.
    """
    verify_preflight_purity(remote)
    action = (action or "").strip().lower()
    result = {
        "requested_action": action,
@@ -1261,12 +991,6 @@ def gitea_submit_pr_review(
    }
    reasons = result["reasons"]
    try:
        verify_mutation_authority(remote, host, required_role="reviewer")
    except RuntimeError as e:
        reasons.append(str(e))
        return result
    # Gate 1 — valid review action (no mutation on unknown action).
    if action not in _REVIEW_ACTIONS:
        reasons.append(
@@ -1476,7 +1200,6 @@ def gitea_commit_files(
        dict with success status and commit/branch information.
    """
    h, o, r = _resolve(remote, host, org, repo)
    verify_preflight_purity(remote)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/contents"
@@ -1570,7 +1293,6 @@ def gitea_merge_pr(
        reasons/gates passed or blocked, and merge result / merge commit if
        available. Never secrets.
    """
    verify_preflight_purity(remote)
    do = (do or "").strip().lower()
    result = {
        "performed": False,
@@ -1589,12 +1311,6 @@ def gitea_merge_pr(
    }
    reasons = result["reasons"]
    try:
        verify_mutation_authority(remote, host, required_role="reviewer")
    except RuntimeError as e:
        reasons.append(str(e))
        return result
    # Gate 1 — valid merge method (no API call on a bad method).
    if do not in _MERGE_METHODS:
        reasons.append(
@@ -2591,26 +2307,26 @@ _PROJECT_SKILLS = {
            "committed.",
        ],
    },
-    "jenkins-mcp": {
+    "jenkins-readonly": {
        "description": "Read-only Jenkins CI inspection (jobs, builds, "
-                       "logs).",
+                       "logs). Actual server name: jenkins-mcp (see mcp-control-plane).",
        "when_to_use": "Checking CI state once Jenkins MCP tools exist.",
        "required_operations": ["jenkins.read"],
        "status": "designed-not-implemented",
-        "notes": "Server code exists in mcp-control-plane as jenkins-mcp (read tools + gated trigger); registration pending. To register for discoverability in clients (Codex/Gemini/Grok/etc.): add to client MCP config under the jenkins-mcp name, reconnect/reload the client session after registration. Report SKIPPED if not connected. Do not substitute shell/API. Trigger requires dedicated profile (see #56).",
+        "notes": "Server code exists in mcp-control-plane as jenkins-mcp (read tools + gated trigger); registration pending (#55); docs in Gitea-Tools use historical name. Report SKIPPED if not connected. Do not substitute shell/API. Trigger requires dedicated profile (see #56).",
        "steps": [
            "Confirm a Jenkins MCP server is connected (jenkins-mcp); if not, report "
            "SKIPPED.",
            "Use read-only operations only; never trigger unless using dedicated profile + confirmation.",
        ],
    },
-    "glitchtip-mcp": {
+    "glitchtip-readonly": {
-        "description": "Read-only GlitchTip error/event inspection.",
+        "description": "Read-only GlitchTip error/event inspection. Actual server name: glitchtip-mcp (see mcp-control-plane).",
        "when_to_use": "Investigating reported errors once GlitchTip MCP "
                       "tools exist.",
        "required_operations": ["glitchtip.read"],
        "status": "designed-not-implemented",
-        "notes": "Server code exists in mcp-control-plane as glitchtip-mcp (read-only tools); registration pending. To register for discoverability in clients (Codex/Gemini/Grok/etc.): add to client MCP config under the glitchtip-mcp name, reconnect/reload the client session after registration. Filing orchestrator is partial in mcp-control-plane (see #57). Report SKIPPED if not connected. Filing to Gitea is separate orchestrator, not in this server.",
+        "notes": "Server code exists in mcp-control-plane as glitchtip-mcp (read-only tools); registration pending (#55); filing orchestrator is partial in mcp-control-plane (see #57). Report SKIPPED if not connected. Filing to Gitea is separate orchestrator, not in this server.",
        "steps": [
            "Confirm a GlitchTip MCP server is connected (glitchtip-mcp); if not, report "
            "SKIPPED.",
@@ -2843,7 +2559,6 @@ def gitea_whoami(
    """
    if remote not in REMOTES:
        raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
    record_preflight_check("whoami")
    h = host or REMOTES[remote]["host"]
    auth = _auth(h)
    url = gitea_url(h, "/api/v1/user")
@@ -3272,25 +2987,6 @@ def gitea_activate_profile(
    after_profile = get_profile()["profile_name"]
    after_identity = _authenticated_username(h) if h else None
    # 4.5 Record pivot in mutation authority lock
    if os.path.exists(LOCK_FILE):
        try:
            with open(LOCK_FILE, "r", encoding="utf-8") as f:
                lock_data = json.load(f)
            lock_data["current_profile"] = after_profile
            lock_data["current_identity"] = after_identity
            lock_data["role_pivot_authorized"] = True
            lock_data["role_pivot_record"] = {
                "from_profile": before_profile,
                "to_profile": after_profile,
                "from_identity": before_identity,
                "to_identity": after_identity
            }
            with open(LOCK_FILE, "w", encoding="utf-8") as f:
                json.dump(lock_data, f)
        except Exception:
            pass
    # 5. Audit the switch if auditing is on
    _audit(
        "activate_profile",
@@ -3641,6 +3337,10 @@ def gitea_resolve_task_capability(
            "permission": "gitea.issue.write",
            "role": "author",
        },
        "mark_issue": {
            "permission": "gitea.issue.write",
            "role": "author",
        },
        "create_branch": {
            "permission": "gitea.branch.create",
            "role": "author",
@@ -3680,7 +3380,6 @@ def gitea_resolve_task_capability(
    required_permission = TASK_MAP[task]["permission"]
    required_role = TASK_MAP[task]["role"]
    record_preflight_check("capability", required_role)
    profile = get_profile()
    config = gitea_config.load_config()
@@ -3781,8 +3480,6 @@ def gitea_resolve_task_capability(
            "STOP: the active profile cannot perform the requested task; "
            "follow exact_safe_next_action instead of improvising.")
    record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
    return {
        "requested_task": task,
        "required_operation_permission": required_permission,
@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
 if os.path.exists(venv_python) and sys.executable != venv_python:
    os.execv(venv_python, [venv_python] + sys.argv)
-from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
+from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url
 def main(argv=None):
@@ -60,29 +60,6 @@ def main(argv=None):
    host, org, repo = resolve_remote(args)
    # ── Mutation Authority context wall check (Issue #194) ──
    LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
    if os.path.exists(LOCK_FILE):
        try:
            with open(LOCK_FILE, "r", encoding="utf-8") as f:
                lock_data = json.load(f)
            # Resolve current CLI profile
            cli_profile = get_profile().get("profile_name")
            locked_profile = lock_data.get("current_profile")
            if cli_profile != locked_profile:
                print(
                    f"Mismatched active profile vs mutation profile (CLI override rejected): "
                    f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)",
                    file=sys.stderr
                )
                return 3
        except Exception as e:
            print(f"Mutation authority check failed: {e}", file=sys.stderr)
            return 3
    body = args.body
    if args.body_file:
        if args.body_file == "-":
@@ -202,13 +202,15 @@ def assess_validation_report(report):
    *report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
    ``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
-    ``justification``), ``canonical_command``, ``deviation_justification``.
+    ``justification``), ``canonical_command``, ``deviation_justification``,
    ``is_stdout_capture_fix``, ``normal_pytest_summary``.
    Verdicts:
    - 'invalid' — the result may not be claimed at all (no command stated,
      or the command output was never read).
    - 'weak'    — claimable but downgraded (missing counts, unjustified
-      ignored paths, unexplained deviation from the canonical command).
+      ignored paths, unexplained deviation from the canonical command, or
      missing normal summary on stdout capture fix).
    - 'strong'  — full evidence.
    """
    reasons = []
@@ -252,6 +254,14 @@ def assess_validation_report(report):
                "command and the deviation is not justified"
            )
    # For stdout-capture fixes (e.g. Issue #178)
    if report.get("is_stdout_capture_fix") is True:
        if not report.get("normal_pytest_summary"):
            reasons.append(
                "stdout-capture fix validation is missing normal pytest summary output "
                "or still requires junitxml workaround"
            )
    verdict = "strong" if not reasons else "weak"
    return {"verdict": verdict, "claimable": True, "reasons": reasons}
@@ -330,97 +340,10 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
    }
 def assess_role_boundary(proof):
    """Assess reviewer/author role separation for blind queue workflows.
    Issue #175 blocks a reviewer queue task from silently becoming author
    implementation work. The workflow may use both namespaces only when that
    mixed use is explicit, justified, and non-mutating; author mutations after
    a reviewer queue task require an explicit operator authorization.
    *proof* keys:
    ``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example
    'blind_pr_queue_review'), ``reviewer_namespace_used``,
    ``author_namespace_used``, ``author_mutations`` (list), ``review_mutations``
    (list), ``operator_authorized_author_work``, ``mixed_namespace_justification``,
    ``scratch_evidence_claimed``, and ``scratch_evidence_durable``.
    """
    proof = proof or {}
    task_role = (proof.get("task_role") or "").strip().lower()
    task_kind = (proof.get("task_kind") or "").strip().lower()
    author_mutations = list(proof.get("author_mutations") or [])
    review_mutations = list(proof.get("review_mutations") or [])
    reviewer_used = bool(proof.get("reviewer_namespace_used"))
    author_used = bool(proof.get("author_namespace_used"))
    authorized = bool(proof.get("operator_authorized_author_work"))
    mixed_justification = (
        proof.get("mixed_namespace_justification") or ""
    ).strip()
    scratch_claimed = bool(proof.get("scratch_evidence_claimed"))
    scratch_durable = bool(proof.get("scratch_evidence_durable"))
    reasons = []
    violations = []
    if task_role not in {"reviewer", "author"}:
        reasons.append("task role missing or unknown; role boundary unproven")
    if task_role == "reviewer":
        if author_mutations and not authorized:
            violations.append(
                "reviewer task performed author mutations without explicit "
                "operator authorization"
            )
        if author_used and not mixed_justification:
            reasons.append(
                "reviewer task used author namespace without an explicit "
                "justification"
            )
        if task_kind == "blind_pr_queue_review" and author_mutations:
            if not authorized:
                violations.append(
                    "blind PR queue review silently pivoted into author "
                    "implementation"
                )
    elif task_role == "author":
        if review_mutations:
            violations.append(
                "author task performed reviewer-only mutations"
            )
    if reviewer_used and author_used and not mixed_justification:
        reasons.append(
            "mixed reviewer+author namespace use was not reported as a "
            "role-boundary event"
        )
    if scratch_claimed and not scratch_durable:
        reasons.append(
            "scratch-only notes were claimed as durable evidence"
        )
    if violations:
        status = "violation"
        safe_next_action = "stop; report role-boundary violation"
    elif reasons:
        status = "warning"
        safe_next_action = "downgrade final report; do not claim A-level proof"
    else:
        status = "clean"
        safe_next_action = "proceed"
    return {
        "status": status,
        "clean": status == "clean",
        "reasons": reasons,
        "violations": violations,
        "safe_next_action": safe_next_action,
    }
 def build_final_report(checkout_proof, inventory, validation, contamination,
                       identity_eligible, merge_performed,
-                       issue_status_verified, role_boundary=None):
+                       issue_status_verified, controller_handoff=None,
                       capability_proof=None, sweep_proof=None):
    """Required behavior 6 + acceptance criteria: one report, distinct proofs.
    Combines the individual proof verdicts into the final-report fields the
@@ -436,12 +359,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
    checkout_proven = bool(checkout_proof.get("proven"))
    validation_claimable = bool(validation.get("claimable"))
    validation_strong = validation.get("verdict") == "strong"
    role_boundary = role_boundary or {
        "status": "warning",
        "reasons": ["role-boundary proof missing"],
        "violations": [],
    }
    role_status = role_boundary.get("status", "warning")
    downgrade_reasons = []
    if not identity_eligible:
@@ -461,17 +378,26 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
        downgrade_reasons.append(
            f"session contamination status is '{contamination_status}'"
        )
    if role_status != "clean":
        downgrade_reasons.append(f"role boundary status is '{role_status}'")
        downgrade_reasons.extend(role_boundary.get("reasons", []))
    if not issue_status_verified:
        downgrade_reasons.append("linked issue status not verified")
    handoff = assess_controller_handoff(controller_handoff or "")
    if not handoff.get("present"):
        downgrade_reasons.append("Controller Handoff section missing or incomplete")
        downgrade_reasons.extend(handoff.get("reasons", []))
    if capability_proof and not capability_proof.get("proven"):
        downgrade_reasons.append("capability proof verification failed")
        downgrade_reasons.extend(capability_proof.get("reasons", []))
    if sweep_proof and not sweep_proof.get("proven"):
        downgrade_reasons.append("secret/provenance sweep proof verification failed")
        downgrade_reasons.extend(sweep_proof.get("reasons", []))
    merge_allowed = (
        identity_eligible
        and checkout_proven
        and contamination_status == "clean"
        and role_status == "clean"
        and validation_claimable
        and validation.get("verdict") != "invalid"
    )
@@ -482,7 +408,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
            "merge was performed/claimed although the proofs did not allow "
            "one; this run is blocked, not graded"
        )
    violations.extend(role_boundary.get("violations", []))
    if violations:
        grade = "blocked"
@@ -499,7 +424,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
        "pr_author_distinct_from_reviewer":
            contamination_status in ("clean",),
        "session_contamination": contamination_status,
        "role_boundary": role_status,
        "inventory_complete": bool(inventory.get("complete")),
        "validated_on_pinned_head": checkout_proven and validation_claimable,
        "validation_passed":
@@ -508,263 +432,111 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
        "merge_allowed": merge_allowed,
        "merge_performed": bool(merge_performed),
        "issue_status_verified": bool(issue_status_verified),
        "controller_handoff_present": handoff.get("present", False),
    }
-# ── Controller Handoff validation (Issue #182) ────────────────────────────────
+def assess_controller_handoff(report_text: str) -> dict:
-#
+    """Required for author and reviewer final reports (#183).
 # Every final report must end with a compact section titled exactly
 # "Controller Handoff". Each required field is a (canonical name, aliases)
 # pair; a field counts as present when any alias starts a bullet/label line
 # inside the handoff section.
-HANDOFF_HEADING = "Controller Handoff"
+    The report must contain an exactly titled 'Controller Handoff' section
-
+    (compact format preferred). Missing it downgrades the report.
 HANDOFF_BASE_FIELDS = (
    ("Task", ("task",)),
    ("Repo", ("repo", "repository", "repo/state")),
    ("Role", ("role",)),
    ("Identity", ("identity",)),
    ("Issue/PR", ("issue/pr", "issues/prs", "issue", "pr")),
    ("Branch/SHA", ("branch/sha", "branch", "head sha")),
    ("Files changed", ("files changed", "changed", "files")),
    ("Validation", ("validation",)),
    ("Mutations", ("mutations",)),
    ("Workspace mutations", ("workspace mutations",)),
    ("Current status", ("current status", "status")),
    ("Blockers", ("blockers",)),
    ("Next", ("next",)),
    ("Safety", ("safety",)),
 )
 HANDOFF_ROLE_FIELDS = {
    "review": (
        ("Selected PR", ("selected pr",)),
        ("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
        ("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
        ("Review decision", ("review decision", "decision")),
        ("Merge result", ("merge result",)),
        ("Linked issue status", ("linked issue status", "linked issue")),
        ("Cleanup status", ("cleanup status", "cleanup")),
    ),
    "author": (
        ("Selected issue", ("selected issue",)),
        ("Claim/comment status", ("claim/comment status", "claim status",
                                  "claim")),
        ("PR number opened", ("pr number opened", "pr opened", "pr number")),
        ("No review/merge confirmation", ("no review/merge",
                                          "no review or merge")),
    ),
    "inventory": (
        ("Repositories checked", ("repositories checked", "repos checked")),
        ("Open PR counts", ("open pr counts", "open pr count",
                            "open prs per repo")),
        ("Selected PR or reason", ("selected pr", "none selected",
                                   "reason none selected")),
        ("Inventory completeness", ("inventory complete", "inventory scoped",
                                    "inventory completeness")),
    ),
 }
 def _handoff_section_lines(report_text):
    """Return the lines of the Controller Handoff section, or None."""
    lines = (report_text or "").splitlines()
    start = None
    for i, line in enumerate(lines):
        bare = line.strip().lstrip("#").strip().rstrip(":")
        if bare == HANDOFF_HEADING:
            start = i + 1
            break
    if start is None:
        return None
    return lines[start:]
 def assess_controller_handoff(report_text, role=None, local_edits=False):
    """Issue #182: final reports without a Controller Handoff downgrade.
    Verdicts:
    - 'missing'    — no exactly-titled section; the report is downgraded.
    - 'incomplete' — section present but required fields absent (listed).
    - 'complete'   — all base fields plus the role-specific fields present.
    *role* is 'review', 'author', 'inventory', or None (base fields only).
    The handoff supplements the full report; this helper never validates
    the full report body, only the continuation summary.
    """
-    import re
+    if not report_text:
-    section = _handoff_section_lines(report_text)
+        return {"present": False, "reasons": ["no report text"]}
-    if section is None:
+
    text = str(report_text)
    if "Controller Handoff" not in text:
        return {
-            "verdict": "missing",
+            "present": False,
            "downgraded": True,
            "missing_fields": [name for name, _ in HANDOFF_BASE_FIELDS],
            "reasons": [
-                "final report has no section titled exactly "
+                "final report missing exactly-titled 'Controller Handoff' section"
                f"'{HANDOFF_HEADING}'"
            ],
        }
-
+    return {"present": True, "reasons": []}
    labels = []
    fields_dict = {}
    for line in section:
        stripped = line.strip().lstrip("-*").strip()
        if ":" in stripped:
            k, v = stripped.split(":", 1)
            label = k.strip().lower()
            labels.append(label)
            fields_dict[label] = v.strip()
    required = list(HANDOFF_BASE_FIELDS)
    required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
    missing = []
    for name, aliases in required:
        if not any(label.startswith(alias)
                   for label in labels for alias in aliases):
            missing.append(name)
    if missing:
        return {
            "verdict": "incomplete",
            "downgraded": True,
            "missing_fields": missing,
            "reasons": [f"handoff missing required field: {m}"
                        for m in missing],
        }
    # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
    for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
        val = fields_dict.get(alias)
        if val:
            numbers = re.findall(r"\d+", val)
            has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
            if len(numbers) != 1 or has_forbidden:
                field_name = "Selected issue/PR"
                for name, aliases_list in required:
                    if alias in aliases_list:
                        field_name = name
                        break
                return {
                    "verdict": "incomplete",
                    "downgraded": True,
                    "missing_fields": [field_name],
                    "reasons": [
                        f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
                    ],
                }
    if local_edits:
        workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
        if not workspace_mutations_val or workspace_mutations_val == "none":
            return {
                "verdict": "incomplete",
                "downgraded": True,
                "missing_fields": ["Workspace mutations"],
                "reasons": [
                    "Workspace mutations cannot be 'none' or empty when local edits exist"
                ],
            }
    return {
        "verdict": "complete",
        "downgraded": False,
        "missing_fields": [],
        "reasons": [],
    }
 def assess_capability_proof(resolved_capabilities: dict) -> dict:
    """Required behavior: every mutation must have exact capability proof.
-# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
+    If a mutation task is unknown, unresolved, or lacks explicit resolver evidence,
-#
+    the workflow must fail closed / be downgraded.
 # A reviewer agent may not convert an empty PR list response into a definitive
 # "no open PRs" conclusion unless the inventory result is independently proven
 # trustworthy.
 def pr_inventory_trust_gate(
    list_prs_response: list | None,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    state: str | None = None,
    authenticated_profile: dict | None = None,
    local_remote_url: str | None = None,
    user_context: str | None = None,
    corroboration_open_pr_counter: int | None = None,
    has_finality_metadata: bool = False,
 ) -> dict:
    """Evaluate whether an empty PR list is trusted or untrusted.
    Returns a dict with 'status', 'reasons', and 'corroborated'.
    """
    if list_prs_response is None or not isinstance(list_prs_response, list):
        return {
            "status": "inventory_error",
            "reasons": ["PR list response is invalid (not a list or None)"],
            "corroborated": False,
        }
    if len(list_prs_response) > 0:
        return {
            "status": "trusted_nonempty",
            "reasons": [],
            "corroborated": False,
        }
    reasons = []
-
+    if not resolved_capabilities:
    # 1. Exact remote, owner, repo, and state filter resolved correctly
    if not remote or remote not in ("dadeschools", "prgs"):
        reasons.append("remote instance is invalid or unresolved")
    if not org or not org.strip():
        reasons.append("owner/org is invalid or unresolved")
    if not repo or not repo.strip():
        reasons.append("repository name is invalid or unresolved")
    if state != "open":
        reasons.append("state filter is not 'open'")
    # 2. Authenticated profile permission check
    if not authenticated_profile or not isinstance(authenticated_profile, dict):
        reasons.append("authenticated profile is missing or invalid")
    else:
        allowed = authenticated_profile.get("allowed_operations") or []
        if "gitea.read" not in allowed and "read" not in allowed:
            reasons.append("authenticated profile lacks read permissions")
    # 3. Pagination/finality metadata or independent read path corroboration
    corroborated = False
    if has_finality_metadata:
        corroborated = True
    elif corroboration_open_pr_counter == 0:
        corroborated = True
    else:
        reasons.append("pagination finality not proven and open_pr_counter corroboration is missing or non-zero")
    # 4. Local checkout remote URL matching the target repo
    if not local_remote_url or not isinstance(local_remote_url, str):
        reasons.append("local checkout remote URL is missing or invalid")
    else:
        expected = f"{org}/{repo}".lower()
        if expected not in local_remote_url.lower():
            reasons.append(f"local remote URL does not match target repository '{org}/{repo}'")
    # 5. User context check (indicators that PRs should exist)
    if user_context and isinstance(user_context, str):
        indicators = ["pr #", "pull request #", "open pr", "pr queue"]
        found = [ind for ind in indicators if ind in user_context.lower()]
        if found:
            reasons.append(f"user context indicates open PRs should exist (matched: {', '.join(found)})")
    if reasons:
        return {
-            "status": "untrusted_empty",
+            "proven": False,
-            "reasons": reasons,
+            "reasons": ["no capability proof resolved; fail closed"],
            "corroborated": corroborated,
        }
    for task, cap in resolved_capabilities.items():
        allowed = cap.get("allowed_in_current_session")
        op = cap.get("required_operation_permission")
        if not allowed:
            reasons.append(f"task '{task}' requires permission '{op}' which is not allowed in current session")
        # If the resolved result indicates an unknown task or missing flag
        if "unknown" in str(cap.get("requested_task", "")).lower() or cap.get("allowed_in_current_session") is None:
            reasons.append(f"task '{task}' could not be resolved to a known capability")
    proven = not reasons
    return {"proven": proven, "reasons": reasons}
 def assess_secret_sweep(sweep_report: dict) -> dict:
    """Required behavior: secret/provenance sweeps must state exact command/method and scope.
    *sweep_report* keys: ``method``, ``scope``, ``clean`` (bool).
    Returns dict with ``proven`` (bool), ``verdict`` ('strong', 'weak', 'invalid'), and ``reasons`` list.
    """
    if not sweep_report:
        return {
            "proven": False,
            "verdict": "invalid",
            "reasons": ["no secret/provenance sweep report provided"],
        }
    reasons = []
    method = (sweep_report.get("method") or "").strip()
    scope = (sweep_report.get("scope") or "").strip()
    if not method:
        reasons.append("secret sweep report missing exact scan command, pattern, or method")
    if not scope:
        reasons.append("secret sweep report missing scanned scope")
    if sweep_report.get("clean") is not True:
        reasons.append("secret sweep report did not confirm diff is clean")
    verdict = "weak" if reasons else "strong"
    proven = not reasons and sweep_report.get("clean") is True
    return {
-        "status": "trusted_empty",
+        "proven": proven,
-        "reasons": [],
+        "verdict": verdict,
-        "corroborated": corroborated,
+        "reasons": reasons,
    }
 def assess_author_pr_report(pr_report: dict) -> dict:
    """Required behavior: author PR creation report must include PR number, branch, and exact head SHA.
    *pr_report* keys: ``pr_number`` (int), ``branch`` (str), ``head_sha`` (str).
    """
    if not pr_report:
        return {
            "complete": False,
            "reasons": ["no PR report provided"],
        }
    reasons = []
    pr_number = pr_report.get("pr_number")
    branch = (pr_report.get("branch") or "").strip()
    head_sha = (pr_report.get("head_sha") or "").strip()
    if not isinstance(pr_number, int) or pr_number <= 0:
        reasons.append("PR number is missing or invalid")
    if not branch:
        reasons.append("PR branch name is missing")
    if not head_sha:
        reasons.append("PR head SHA is missing")
    elif not _FULL_SHA.match(head_sha):
        reasons.append("PR head SHA is not a full 40-hex commit SHA")
    complete = not reasons
    return {
        "complete": complete,
        "reasons": reasons,
    }
@@ -40,16 +40,6 @@ start_ref="${2:-prgs/master}"
 # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
 if [[ "$allow_unlinked" -eq 0 ]]; then
  if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
    echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
    exit 2
  fi
  locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
  if [[ "$branch" != "$locked_branch" ]]; then
    echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
    exit 2
  fi
  if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
     || [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
    :
@@ -150,33 +150,12 @@ Worktree folder = branch with `/` replaced by `-`
 6. Implement the narrow scope only — no unrelated refactors or formatting churn.
 7. Add/update focused tests when behavior changes.
 8. Run the checks (tests, compile/lint, `git diff --check`, secret scan).
-   Record the branch name and `HEAD` SHA at validation time — the drift
+9. Commit with an issue-linked message.
-   check in step 9 compares against exactly this state.
+10. Push the branch.
-9. **Branch proof before commit (#177):** prove and state, immediately
+11. Open a PR to `master`.
-   before staging/committing (`author_proofs.verify_branch_for_commit`,
+12. **If you are the author, stop before review/merge.**
-   `author_proofs.detect_branch_drift`):
+13. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
-   - current branch (`git branch --show-current`) equals the intended
+14. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
     feature branch from the issue claim
   - current branch is not `master`, `main`, `develop`, `development`, or
     `dev`
   - branch and `HEAD` have not changed since validation (step 8) — in a
     shared checkout another session may switch branches mid-session;
     treat that as expected and **stop before committing** when detected
   If any check fails, stop and reconcile; do not commit.
 10. Commit with an issue-linked message.
 11. **Branch proof before push (#177):** prove that the local branch, the
    push target branch, and the intended issue branch all match, and that
    none of them is a protected branch
    (`author_proofs.verify_push_target`). If a commit accidentally landed
    on a protected branch, do **not** push: report the accident and the
    exact repair steps (`author_proofs.assess_protected_branch_commit`) —
    never silently continue after a repair.
 12. Push the branch.
 13. Open a PR to `master`. The final report must include the branch proofs
    from steps 9 and 11 (`author_proofs.build_commit_push_report`).
 14. **If you are the author, stop before review/merge.**
 15. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
 16. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
    - why the PR merge path could not be used
    - exact commits pushed
    - PR metadata state
@@ -217,27 +196,19 @@ Worktree folder = branch with `/` replaced by `-`
   Both configured repos must be reported with state filter, pagination proof,
   and open-PR count (`review_proofs.assess_inventory_completeness` and
   `resolve_repos_from_user_reference`).
-7. **Role-boundary proof (#175):** a reviewer queue task must not silently
+7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
-   become author implementation. If no eligible PR exists, stop with the
+8. Run the tests. Validation reporting must include the exact command and
   queue report. Do not claim issues, create branches, commit, push, or open
   PRs unless the operator explicitly retasks the run as author work. Mixed
   reviewer+author namespace use must be reported with a justification, and
   scratch-only notes are not durable evidence unless posted or committed
   intentionally (`review_proofs.assess_role_boundary`).
 8. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
 9. Run the tests. Validation reporting must include the exact command and
   exact results: pass/fail, counts of tests passed/skipped/failed, any
   ignored paths and why they are safe to ignore, and whether the command
   differs from the repository's canonical validation command. Only claim a
   validation result after the command has completed and its output has
   been read (`review_proofs.assess_validation_report`).
-10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
+9. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
-11. The final report must distinguish (`review_proofs.build_final_report`):
+10. The final report must distinguish (`review_proofs.build_final_report`):
    identity eligible; PR author different from reviewer; session
-    contamination absent (with evidence); role boundary clean; validation
+    contamination absent (with evidence); validation performed on the pinned
-    performed on the pinned head; merge performed; issue status verified. If
+    head; merge performed; issue status verified. If any proof is missing,
-    any proof is missing, stop or downgrade the result instead of merging
+    stop or downgrade the result instead of merging confidently.
    confidently.
 ## G. Merge / cleanup workflow
@@ -316,52 +287,33 @@ Ready-to-copy templates live in [`templates/`](templates/):
 ## K. Controller Handoff (required, every task)
-Every LLM task **must end with a `Controller Handoff`** — whether the
+Every LLM task **must end with a `Controller Handoff`** (exact title) — whether the
 task was implementation, review, merge, issue triage, documentation,
 discussion-only, or blocked planning. It lets a controller LLM understand the
 current state immediately, without rereading the conversation.
 The section title must be exactly "Controller Handoff" (or "Controller Handoff Summary" for long form). Reports without it are downgraded (see review_proofs.assess_controller_handoff).
 **The compact format is the default.** It is written for controller-LLM
 readability, not as a full human status report. PR bodies still carry the
 full review detail — the handoff never replaces PR documentation.
-Compact format (default, canonical field set per issue #182):
+Compact format (default):
 ```md
 ## Controller Handoff
 - Task:
- Repo:
+- Repo/state:
- Role:
+- Issues/PRs:
- Identity:
+- Changed:
 - Issue/PR:
 - Branch/SHA:
 - Files changed:
 - Validation:
 - Mutations:
 - Current status:
 - Blockers:
 - Review:
 - Next:
 - Safety:
 ```
 Role-specific fields (append to the compact block):
 - review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
  `Pinned reviewed head:`, `Review decision:`, `Merge result:`,
  `Linked issue status:`, `Cleanup status:`
 - author tasks: `Selected issue:`, `Claim/comment status:`,
  `PR number opened:`, `No review/merge:` (explicit confirmation)
 - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
  `Selected PR or reason none selected:`, `Inventory completeness:`
 The section title must be exactly `Controller Handoff`.
 `review_proofs.assess_controller_handoff()` validates this section; reports
 missing it (or missing required fields) are downgraded. The handoff never
 replaces the full report — it is the compact continuation summary at the end,
 and the full report must still carry exact validation results and mutation
 confirmation.
 The `Safety:` line is never omitted; it is usually:
 ```text
@@ -35,11 +35,5 @@ Then run the cleanup template (worktree-cleanup.md):
 - delete remote branch, remove local branch + worktree folder
 - fetch/prune; confirm main checkout is clean and current (0 0).
-Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
+Handoff: reviewer identity, merge result + commit, cleanup done, issue closed, PR metadata state/merged flag/hash, remote master hash, post-merge verification method used & verification results.
 §K (long form — a merge is always high-risk), including the review/merge role
 fields (Selected PR, Reviewer eligibility, Pinned reviewed head, Review
 decision, Merge result, Linked issue status, Cleanup status) plus: merge
 commit, PR metadata state/merged flag/hash, remote master hash, and the
 post-merge verification method used & verification results. Reports missing
 the handoff are downgraded (review_proofs.assess_controller_handoff).
 ```
@@ -23,10 +23,6 @@ Rules (llm-project-workflow):
 - You must NOT be the PR author. If the authenticated user == PR author, stop.
  A different LLM-Agent-SHA does NOT make you a different actor — only a
  different authenticated Gitea user does (docs/llm-agent-sha.md).
 - Do not pivot from a reviewer queue task into author implementation unless
  the operator explicitly retasks the run. If author namespace was used, the
  final report must justify why; author mutations after reviewer queue work
  without explicit authorization are a role-boundary violation.
 - Do not merge if any check fails.
 Steps:
@@ -43,10 +39,6 @@ Steps:
   cannot evidence whether this session authored/touched the PR branch,
   report contamination as UNKNOWN (not contaminated, not clean) and choose
   another PR or stop.
   Role-boundary claims must also be evidence-backed (#175): report whether
   reviewer namespace, author namespace, author mutations, or review mutations
   occurred. Use `review_proofs.assess_role_boundary`; if it is not clean,
   downgrade or stop instead of claiming an A-level run.
 5. scripts/worktree-review <pr-head-branch>          # detached, branches/review-*
   cd branches/review-<pr-head-branch-slug>
 6. Checkout proof (#173) — prove and state, before any diff review or
@@ -73,10 +65,7 @@ Steps:
   - MCP-Profile: <profile name>
   - Eligibility: passed/failed
-Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
+Handoff: reviewer identity, PR author, scope verdict, checks + results, decision —
-§K (compact by default; long form if a merge happened or a gate blocked you),
+formatted per SKILL.md §K (compact by default; long form if a merge happened
-including the review/merge role fields: Selected PR, Reviewer eligibility,
+or a gate blocked you); if you could not merge, name the exact gate.
 Pinned reviewed head, Review decision, Merge result, Linked issue status,
 Cleanup status. If you could not merge, name the exact gate. Reports missing
 the handoff are downgraded (review_proofs.assess_controller_handoff).
 ```
@@ -26,19 +26,8 @@ Steps:
   cd branches/<type>-issue-<n>-<slug>
 6. Implement the narrow scope only; add/update focused tests if behavior changes.
 7. Checks: run the test suite, compile/lint changed files, git diff --check,
-   and scan the diff for secrets. Record the branch name and HEAD SHA at
+   and scan the diff for secrets.
-   validation time.
+8. Commit (issue-linked message), push the branch, open a PR to master.
 8. Branch proof before commit (#177) — prove and state:
   - git branch --show-current == the intended issue branch from step 5
   - the branch is NOT master/main/develop/development/dev
   - branch and HEAD unchanged since step 7 (another session can switch a
     shared checkout mid-session; if drift is detected, STOP and reconcile
     before committing)
   If a commit accidentally lands on a protected branch: do NOT push;
   report the accident and the exact repair steps — never silently continue.
 9. Commit (issue-linked message). Branch proof before push (#177): local
   branch == push target branch == intended issue branch, none protected.
   Then push the branch and open a PR to master.
   *The PR body MUST use closing keywords like `Closes #N` or `Fixes #N` to close the issue; do NOT use `Implements #N` or `Refs #N` for closing, as Gitea will not auto-close it.*
   Include an "LLM Handoff Metadata" block in the PR body (attribution only;
   never an eligibility input — docs/llm-agent-sha.md):
@@ -51,12 +40,9 @@ Steps:
   - Branch: <branch>
   - Worktree: <worktree path>
   - Self-review allowed: no
-10. Stop before review/merge — you are the author.
+9. Stop before review/merge — you are the author.
-Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
+Handoff: issue #, branch, worktree path, files changed, checks + results, PR URL —
-§K (compact; long form only on the high-risk triggers), including the author
+formatted as the compact Controller Handoff (SKILL.md §K; long form only on
-role fields: Selected issue, Claim/comment status, PR number opened, and an
+the high-risk triggers); Review line: "Review needed — PR is open".
 explicit "No review/merge:" confirmation — plus branch, worktree path, files
 changed, checks + results. Next line: "Review needed — PR is open". Reports
 missing the handoff are downgraded (review_proofs.assess_controller_handoff).
 ```
@@ -1,234 +0,0 @@
 """Tests for author-side branch-identity proofs (Issue #177).
 Issue #177 (author-side counterpart of the #173 reviewer proofs) requires
 author workflows to *prove* local git state before staging, committing, or
 pushing, instead of discovering drift after the fact:
 1. The current branch equals the intended feature branch and is never a
   protected branch (master/main/develop/development/dev).
 2. Branch or HEAD drift between validation and commit — including external
   branch switches in a shared worktree — stops the workflow.
 3. A push requires local branch, remote target branch, and intended issue
   branch to all match.
 4. An accidental commit on a protected branch must not be pushed and its
   repair must be reported, never silently continued.
 These are the harness assertions from the issue's Required behavior 5.
 """
 import sys
 import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 from author_proofs import (  # noqa: E402
    PROTECTED_BRANCHES,
    assess_protected_branch_commit,
    build_commit_push_report,
    detect_branch_drift,
    verify_branch_for_commit,
    verify_push_target,
 )
 FEATURE = "feat/issue-177-branch-drift-proofs"
 HEAD_1 = "64dc334a92685b7b6a1fdb7ffe363f02a69f5dbd"
 HEAD_2 = "ccc5ef79dfe629853e144763238593bd808d57e0"
 class TestProtectedBranches(unittest.TestCase):
    def test_known_protected_names(self):
        for name in ("master", "main", "develop", "development", "dev"):
            self.assertIn(name, PROTECTED_BRANCHES)
 class TestVerifyBranchForCommit(unittest.TestCase):
    """Required behavior 1: prove the branch before staging/committing."""
    def test_on_intended_feature_branch_is_proven(self):
        proof = verify_branch_for_commit(FEATURE, FEATURE)
        self.assertTrue(proof["proven"])
        self.assertFalse(proof["block"])
    def test_commit_attempted_while_on_master_is_blocked(self):
        # Harness assertion (behavior 5, bullet 1).
        proof = verify_branch_for_commit("master", FEATURE)
        self.assertFalse(proof["proven"])
        self.assertTrue(proof["block"])
        self.assertTrue(any("master" in r for r in proof["reasons"]))
    def test_every_protected_branch_is_blocked_as_current(self):
        for name in PROTECTED_BRANCHES:
            proof = verify_branch_for_commit(name, FEATURE)
            self.assertTrue(proof["block"], name)
    def test_intended_branch_may_not_be_protected(self):
        proof = verify_branch_for_commit("master", "master")
        self.assertFalse(proof["proven"])
        self.assertTrue(proof["block"])
    def test_wrong_feature_branch_is_blocked(self):
        proof = verify_branch_for_commit("feat/issue-178-other-work", FEATURE)
        self.assertFalse(proof["proven"])
        self.assertTrue(proof["block"])
    def test_missing_current_branch_fails_closed(self):
        proof = verify_branch_for_commit("", FEATURE)
        self.assertTrue(proof["block"])
    def test_missing_intended_branch_fails_closed(self):
        proof = verify_branch_for_commit(FEATURE, None)
        self.assertTrue(proof["block"])
 class TestBranchDrift(unittest.TestCase):
    """Required behaviors 2 + 3: drift between validation and commit stops
    the workflow."""
    def test_no_drift_when_branch_and_head_unchanged(self):
        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1)
        self.assertFalse(drift["drifted"])
        self.assertFalse(drift["block"])
    def test_branch_drift_between_validation_and_commit_is_blocked(self):
        # Harness assertion (behavior 5, bullet 2).
        drift = detect_branch_drift(FEATURE, HEAD_1, "feat/other", HEAD_1)
        self.assertTrue(drift["drifted"])
        self.assertTrue(drift["block"])
    def test_shared_worktree_branch_switch_is_detected(self):
        # Harness assertion (behavior 5, bullet 4): an external session
        # switching the shared checkout to another branch (e.g. master)
        # must be detected as drift, not treated as exceptional noise.
        drift = detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
        self.assertTrue(drift["drifted"])
        self.assertTrue(drift["block"])
        self.assertTrue(any("switch" in r.lower() for r in drift["reasons"]))
    def test_head_moved_since_validation_is_blocked(self):
        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_2)
        self.assertTrue(drift["drifted"])
        self.assertTrue(drift["block"])
        self.assertTrue(any("HEAD" in r for r in drift["reasons"]))
    def test_missing_state_fails_closed(self):
        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, None)
        self.assertTrue(drift["drifted"])
        self.assertTrue(drift["block"])
 class TestVerifyPushTarget(unittest.TestCase):
    """Required behavior 1 (push leg) + acceptance: push needs proof that
    local, remote, and intended branches all match."""
    def test_matching_local_remote_and_intended_is_proven(self):
        proof = verify_push_target(FEATURE, FEATURE, FEATURE)
        self.assertTrue(proof["proven"])
        self.assertFalse(proof["block"])
    def test_push_target_mismatch_is_blocked(self):
        # Harness assertion (behavior 5, bullet 3).
        proof = verify_push_target(FEATURE, "feat/issue-178-other-work", FEATURE)
        self.assertFalse(proof["proven"])
        self.assertTrue(proof["block"])
    def test_local_branch_differs_from_intended_is_blocked(self):
        proof = verify_push_target("feat/other", FEATURE, FEATURE)
        self.assertTrue(proof["block"])
    def test_pushing_a_protected_branch_is_blocked(self):
        proof = verify_push_target("master", "master", "master")
        self.assertFalse(proof["proven"])
        self.assertTrue(proof["block"])
    def test_missing_remote_target_fails_closed(self):
        proof = verify_push_target(FEATURE, "", FEATURE)
        self.assertTrue(proof["block"])
 class TestProtectedBranchAccident(unittest.TestCase):
    """Required behavior 4: accidental protected-branch commits must not be
    pushed and their repair must be reported."""
    def test_feature_branch_commit_is_not_an_accident(self):
        result = assess_protected_branch_commit(FEATURE)
        self.assertFalse(result["accident"])
        self.assertEqual(result["violations"], [])
    def test_commit_on_master_is_an_accident_and_must_not_push(self):
        result = assess_protected_branch_commit(
            "master", pushed=False, repair_reported=True
        )
        self.assertTrue(result["accident"])
        self.assertTrue(result["must_not_push"])
        self.assertEqual(result["violations"], [])
        self.assertTrue(result["repair_required"])
    def test_pushing_the_accident_is_a_violation(self):
        result = assess_protected_branch_commit(
            "master", pushed=True, repair_reported=True
        )
        self.assertTrue(any("push" in v.lower() for v in result["violations"]))
    def test_silent_repair_is_a_violation(self):
        # Harness assertion (behavior 5, bullet 5): the repair path must not
        # silently continue without reporting.
        result = assess_protected_branch_commit(
            "master", pushed=False, repair_reported=False
        )
        self.assertTrue(any("report" in v.lower() for v in result["violations"]))
 class TestCommitPushReport(unittest.TestCase):
    """Acceptance criteria: the final report includes branch proof before
    commit and before push, and blocks instead of continuing."""
    def _report(self, **overrides):
        kwargs = {
            "commit_proof": verify_branch_for_commit(FEATURE, FEATURE),
            "drift": detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1),
            "push_proof": verify_push_target(FEATURE, FEATURE, FEATURE),
            "accident": assess_protected_branch_commit(FEATURE),
        }
        kwargs.update(overrides)
        return build_commit_push_report(**kwargs)
    def test_fully_proven_report_is_ok(self):
        report = self._report()
        self.assertEqual(report["status"], "ok")
        self.assertTrue(report["branch_proof_before_commit"])
        self.assertTrue(report["branch_proof_before_push"])
        self.assertFalse(report["drift_detected"])
        self.assertEqual(report["violations"], [])
    def test_commit_proof_failure_blocks(self):
        report = self._report(
            commit_proof=verify_branch_for_commit("master", FEATURE)
        )
        self.assertEqual(report["status"], "blocked")
        self.assertFalse(report["branch_proof_before_commit"])
    def test_drift_blocks(self):
        report = self._report(
            drift=detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
        )
        self.assertEqual(report["status"], "blocked")
        self.assertTrue(report["drift_detected"])
    def test_push_proof_failure_blocks(self):
        report = self._report(
            push_proof=verify_push_target(FEATURE, "feat/other", FEATURE)
        )
        self.assertEqual(report["status"], "blocked")
        self.assertFalse(report["branch_proof_before_push"])
    def test_accident_violations_block(self):
        report = self._report(
            accident=assess_protected_branch_commit(
                "master", pushed=False, repair_reported=False
            )
        )
        self.assertEqual(report["status"], "blocked")
        self.assertTrue(report["violations"])
 if __name__ == "__main__":
    unittest.main()
@@ -5,7 +5,6 @@ the MCP protocol) with mocked API responses.
 """
 import json
 import os
 os.environ["GITEA_TEST_ENVIRONMENT"] = "1"
 import sys
 import unittest
 from unittest.mock import patch, MagicMock
@@ -34,16 +33,10 @@ from mcp_server import (  # noqa: E402
    gitea_submit_pr_review,
    gitea_list_issue_comments,
    gitea_create_issue_comment,
    gitea_lock_issue,
 )
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 import mcp_server
 # Globally disable verification check for existing isolated unit tests
 _real_verify = mcp_server.verify_mutation_authority
 mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -92,13 +85,10 @@ class TestCreatePR(unittest.TestCase):
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("os.path.exists", return_value=True)
+    def test_creates_pr(self, _auth, mock_api):
    @patch("builtins.open")
    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
        mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
        with patch.dict(os.environ, {}, clear=True):
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
        self.assertEqual(result["number"], 3)
        self.assertNotIn("url", result)
        payload = mock_api.call_args[0][3]
@@ -107,13 +97,10 @@ class TestCreatePR(unittest.TestCase):
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("os.path.exists", return_value=True)
+    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
    @patch("builtins.open")
    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
        mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
        self.assertIn("pulls/3", result["url"])
@@ -2309,255 +2296,3 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
            "gitea.issue.comment", reviewer["allowed_operations"],
            reviewer.get("forbidden_operations", []))
        self.assertTrue(ok)
 class TestVerifyMutationAuthority(unittest.TestCase):
    """Test verification lock logic under various configurations."""
    def setUp(self):
        self.patch_profile = patch("mcp_server.get_profile")
        self.mock_profile = self.patch_profile.start()
        self.patch_username = patch("mcp_server._authenticated_username")
        self.mock_username = self.patch_username.start()
        # Restore real function for these tests
        self._old_verify = mcp_server.verify_mutation_authority
        mcp_server.verify_mutation_authority = _real_verify
    def tearDown(self):
        self.patch_profile.stop()
        self.patch_username.stop()
        mcp_server.verify_mutation_authority = self._old_verify
        # Clean up lock file
        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
            os.remove("/tmp/gitea_mutation_authority.lock")
    def test_missing_lock_fails_closed(self):
        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
            os.remove("/tmp/gitea_mutation_authority.lock")
        with self.assertRaises(RuntimeError) as ctx:
            _real_verify("prgs")
        self.assertIn("lock is missing", str(ctx.exception))
    def test_mismatched_remote_fails(self):
        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
            json.dump({
                "remote": "dadeschools",
                "current_profile": "prgs-reviewer",
                "current_identity": "sysadmin"
            }, f)
        with self.assertRaises(RuntimeError) as ctx:
            _real_verify("prgs")
        self.assertIn("does not match locked remote", str(ctx.exception))
    def test_mismatched_profile_fails(self):
        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
            json.dump({
                "remote": "prgs",
                "current_profile": "prgs-author",
                "current_identity": "jcwalker3"
            }, f)
        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
        self.mock_username.return_value = "sysadmin"
        with self.assertRaises(RuntimeError) as ctx:
            _real_verify("prgs")
        self.assertIn("does not match locked authority", str(ctx.exception))
    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
            json.dump({
                "remote": "prgs",
                "initial_profile": "prgs-author",
                "initial_identity": "jcwalker3",
                "current_profile": "prgs-reviewer",
                "current_identity": "sysadmin",
                "role_pivot_authorized": False
            }, f)
        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
        self.mock_username.return_value = "sysadmin"
        with self.assertRaises(RuntimeError) as ctx:
            _real_verify("prgs", required_role="reviewer")
        self.assertIn("without authorized role pivot", str(ctx.exception))
    def test_allowed_when_match(self):
        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
            json.dump({
                "remote": "prgs",
                "initial_profile": "prgs-reviewer",
                "initial_identity": "sysadmin",
                "current_profile": "prgs-reviewer",
                "current_identity": "sysadmin",
                "role_pivot_authorized": False
            }, f)
        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
        self.mock_username.return_value = "sysadmin"
        # Should pass without exception
        _real_verify("prgs")
 class TestIssueLocking(unittest.TestCase):
    """Test issue locking and PR gating constraints."""
    def tearDown(self):
        if os.path.exists("/tmp/gitea_issue_lock.json"):
            os.remove("/tmp/gitea_issue_lock.json")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_success(self, _auth, mock_api):
        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.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
    def test_lock_issue_mismatch_branch_fails(self):
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
        self.assertIn("must contain locked issue pattern", str(ctx.exception))
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
        mock_api.return_value = [{
            "number": 200,
            "head": {"ref": "feat/issue-196-boundary"},
            "title": "Some PR",
            "body": "No closes ref"
        }]
        with self.assertRaises(ValueError) as ctx:
            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.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
        mock_api.return_value = [{
            "number": 200,
            "head": {"ref": "feat/other-branch"},
            "title": "Some PR",
            "body": "fixes #196"
        }]
        with self.assertRaises(ValueError) as ctx:
            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.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_missing_lock_fails(self, _auth):
        if os.path.exists("/tmp/gitea_issue_lock.json"):
            os.remove("/tmp/gitea_issue_lock.json")
        with self.assertRaises(RuntimeError) as ctx:
            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
        self.assertIn("Issue lock is missing", str(ctx.exception))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_branch_mismatch_fails(self, _auth):
        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
        with self.assertRaises(ValueError) as ctx:
            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
        self.assertIn("does not match locked branch", str(ctx.exception))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_forbidden_terms_fails(self, _auth):
        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
        for term in ("equivalent to #196", "related to #196", "same as #196"):
            with self.assertRaises(ValueError) as ctx:
                gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
            self.assertIn("contains forbidden term", str(ctx.exception))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_missing_closes_ref_fails(self, _auth):
        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
        with self.assertRaises(ValueError) as ctx:
            gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
        self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
 class TestPreflightVerification(unittest.TestCase):
    """Test workspace edits and pre-flight ordering verification."""
    def setUp(self):
        self.preflight_path = "/tmp/gitea_preflight_check.json"
        if os.path.exists(self.preflight_path):
            os.remove(self.preflight_path)
        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
    def tearDown(self):
        if os.path.exists(self.preflight_path):
            os.remove(self.preflight_path)
        os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
    @patch("subprocess.run")
    def test_record_preflight_detects_violation_whoami(self, mock_run):
        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
        from mcp_server import record_preflight_check, verify_preflight_purity
        record_preflight_check("whoami")
        with open(self.preflight_path, "r") as f:
            data = json.load(f)
        self.assertTrue(data.get("whoami_preflight_violation"))
        self.assertTrue(data.get("whoami_called"))
        data["capability_called"] = True
        with open(self.preflight_path, "w") as f:
            json.dump(data, f)
        with self.assertRaises(RuntimeError) as ctx:
            verify_preflight_purity("prgs")
        self.assertIn("Workspace file edits occurred before gitea_whoami", str(ctx.exception))
    @patch("subprocess.run")
    def test_record_preflight_detects_violation_capability(self, mock_run):
        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
        from mcp_server import record_preflight_check, verify_preflight_purity
        record_preflight_check("capability", resolved_role="author")
        with open(self.preflight_path, "r") as f:
            data = json.load(f)
        self.assertTrue(data.get("capability_preflight_violation"))
        self.assertTrue(data.get("capability_called"))
        self.assertEqual(data.get("role"), "author")
        data["whoami_called"] = True
        with open(self.preflight_path, "w") as f:
            json.dump(data, f)
        with self.assertRaises(RuntimeError) as ctx:
            verify_preflight_purity("prgs")
        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability", str(ctx.exception))
    @patch("subprocess.run")
    def test_verify_preflight_reviewer_edits_blocked(self, mock_run):
        with open(self.preflight_path, "w") as f:
            json.dump({
                "whoami_called": True,
                "capability_called": True,
                "role": "reviewer"
            }, f)
        mock_run.return_value = MagicMock(stdout="M review_proofs.py\n")
        from mcp_server import verify_preflight_purity
        with self.assertRaises(RuntimeError) as ctx:
            verify_preflight_purity("prgs")
        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
    @patch("subprocess.run")
    def test_verify_preflight_clean_reviewer_allowed(self, mock_run):
        with open(self.preflight_path, "w") as f:
            json.dump({
                "whoami_called": True,
                "capability_called": True,
                "role": "reviewer"
            }, f)
        mock_run.return_value = MagicMock(stdout="")
        from mcp_server import verify_preflight_purity
        verify_preflight_purity("prgs")
    def test_verify_preflight_skipped_fails(self):
        from mcp_server import verify_preflight_purity
        with self.assertRaises(RuntimeError) as ctx:
            verify_preflight_purity("prgs")
        self.assertIn("verification were skipped", str(ctx.exception))
@@ -46,8 +46,8 @@ EXPECTED_SKILLS = [
    "gitea-resolve-task-capability",
    "profile-switching",
    "redaction-security-review",
-    "jenkins-mcp",
+    "jenkins-readonly",
-    "glitchtip-mcp",
+    "glitchtip-readonly",
    "release-operator",
 ]
@@ -234,8 +234,8 @@ class TestProjectSkills(GuideTestBase):
        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
            r = mcp_list_project_skills()
        by_name = {s["name"]: s for s in r["skills"]}
-        self.assertNotEqual(by_name["jenkins-mcp"]["status"], "available")
+        self.assertNotEqual(by_name["jenkins-readonly"]["status"], "available")
-        self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
+        self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available")
    def test_no_urls_in_registry(self):
        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
@@ -245,17 +245,6 @@ class TestProjectSkills(GuideTestBase):
        self.assertNotIn("http://", blob)
        self.assertNotIn("keychain:", blob)
    def test_enabled_but_no_usable_tools_negative_assertion(self):
        """Negative assertion for 'enabled but no usable tools' (per issue #146)."""
        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
            r = mcp_list_project_skills()
        by_name = {s["name"]: s for s in r["skills"]}
        # jenkins-mcp is designed-not-implemented; even if "enabled" in config,
        # it should not be usable/available to current profile without tools.
        self.assertIn("jenkins-mcp", by_name)
        self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented")
        self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
 # ---------------------------------------------------------------------------
 # mcp_get_skill_guide
@@ -2,8 +2,6 @@
 Mocks api_request and credentials.
 """
 import io
 import os
 import sys
 import unittest
 from unittest.mock import patch
@@ -29,11 +27,6 @@ FAKE_PR_DATA = {
 class TestArgParsing(unittest.TestCase):
    def setUp(self):
        self.exists_patcher = patch("os.path.exists", return_value=False)
        self.exists_patcher.start()
        self.addCleanup(self.exists_patcher.stop)
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_missing_pr_number_exits(self, _auth):
        with self.assertRaises(SystemExit):
@@ -42,11 +35,6 @@ class TestArgParsing(unittest.TestCase):
 class TestAPIPayload(unittest.TestCase):
    def setUp(self):
        self.exists_patcher = patch("os.path.exists", return_value=False)
        self.exists_patcher.start()
        self.addCleanup(self.exists_patcher.stop)
    @patch("review_pr.api_request")
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_payload_fields_and_workflow(self, _auth, mock_api):
@@ -111,70 +99,5 @@ class TestAPIPayload(unittest.TestCase):
        self.assertIn("gitea_merge_pr", msg)
 class TestMutationAuthorityLock(unittest.TestCase):
    """Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
    @patch("review_pr.get_profile")
    def test_cli_blocked_on_profile_mismatch(self, mock_get_profile):
        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
        original_exists = os.path.exists
        def conditional_exists(path):
            if "gitea_mutation_authority.lock" in str(path):
                return True
            return original_exists(path)
        original_open = open
        def conditional_open(file, *args, **kwargs):
            if "gitea_mutation_authority.lock" in str(file):
                return io.StringIO('{"current_profile": "prgs-author"}')
            return original_open(file, *args, **kwargs)
        from _pytest.monkeypatch import MonkeyPatch
        import io
        buf = io.StringIO()
        monkeypatch = MonkeyPatch()
        monkeypatch.setattr(sys, "stderr", buf)
        with patch("os.path.exists", side_effect=conditional_exists), \
             patch("builtins.open", side_effect=conditional_open):
            try:
                rc = review_pr.main([
                    "--pr-number", "81", "--event", "APPROVE",
                ])
            finally:
                monkeypatch.undo()
        self.assertEqual(rc, 3)
        msg = buf.getvalue().lower()
        self.assertIn("cli override rejected", msg)
    @patch("review_pr.get_profile")
    @patch("review_pr.api_request")
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
        mock_api.side_effect = [FAKE_PR_DATA, {}]
        original_exists = os.path.exists
        def conditional_exists(path):
            if "gitea_mutation_authority.lock" in str(path):
                return True
            return original_exists(path)
        original_open = open
        def conditional_open(file, *args, **kwargs):
            if "gitea_mutation_authority.lock" in str(file):
                return io.StringIO('{"current_profile": "prgs-reviewer"}')
            return original_open(file, *args, **kwargs)
        with patch("os.path.exists", side_effect=conditional_exists), \
             patch("builtins.open", side_effect=conditional_open):
            rc = review_pr.main([
                "--pr-number", "81", "--event", "APPROVE",
            ])
        self.assertEqual(rc, 0)
 if __name__ == "__main__":
    unittest.main()
@@ -21,13 +21,14 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 from review_proofs import (  # noqa: E402
    assess_author_pr_report,
    assess_capability_proof,
    assess_controller_handoff,
    assess_inventory_completeness,
-    assess_role_boundary,
+    assess_secret_sweep,
    assess_self_review_contamination,
    assess_validation_report,
    build_final_report,
    pr_inventory_trust_gate,
    resolve_repos_from_user_reference,
    verify_pinned_head_checkout,
 )
@@ -101,21 +102,6 @@ def _good_contamination():
    )
 def _good_role_boundary():
    return assess_role_boundary(
        {
            "task_role": "reviewer",
            "task_kind": "blind_pr_queue_review",
            "reviewer_namespace_used": True,
            "author_namespace_used": False,
            "author_mutations": [],
            "review_mutations": [],
            "operator_authorized_author_work": False,
            "scratch_evidence_claimed": False,
        }
    )
 class TestCheckoutProof(unittest.TestCase):
    """Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -332,6 +318,21 @@ class TestValidationReporting(unittest.TestCase):
        )
        self.assertEqual(result["verdict"], "strong")
    def test_stdout_capture_fix_without_summary_is_weak(self):
        result = _good_validation(
            is_stdout_capture_fix=True,
            normal_pytest_summary=False
        )
        self.assertEqual(result["verdict"], "weak")
        self.assertTrue(any("stdout-capture" in r for r in result["reasons"]))
    def test_stdout_capture_fix_with_summary_is_strong(self):
        result = _good_validation(
            is_stdout_capture_fix=True,
            normal_pytest_summary=True
        )
        self.assertEqual(result["verdict"], "strong")
 class TestSelfReviewContamination(unittest.TestCase):
    """Required behavior 5: contamination claims need evidence."""
@@ -385,74 +386,6 @@ class TestSelfReviewContamination(unittest.TestCase):
        self.assertEqual(result["status"], "unknown")
 class TestRoleBoundary(unittest.TestCase):
    """Issue #175: reviewer queue tasks must not pivot into author work."""
    def test_reviewer_queue_without_author_mutations_is_clean(self):
        result = _good_role_boundary()
        self.assertEqual(result["status"], "clean")
        self.assertEqual(result["violations"], [])
    def test_reviewer_queue_author_mutation_without_authorization_violates(self):
        result = assess_role_boundary(
            {
                "task_role": "reviewer",
                "task_kind": "blind_pr_queue_review",
                "reviewer_namespace_used": True,
                "author_namespace_used": True,
                "author_mutations": ["claim issue #171", "push branch"],
                "operator_authorized_author_work": False,
                "mixed_namespace_justification": (
                    "author namespace was used for implementation"
                ),
            }
        )
        self.assertEqual(result["status"], "violation")
        self.assertTrue(any("pivot" in r for r in result["violations"]))
    def test_mixed_namespace_use_without_justification_is_warning(self):
        result = assess_role_boundary(
            {
                "task_role": "reviewer",
                "task_kind": "blind_pr_queue_review",
                "reviewer_namespace_used": True,
                "author_namespace_used": True,
                "author_mutations": [],
            }
        )
        self.assertEqual(result["status"], "warning")
        self.assertTrue(
            any("mixed" in r.lower() for r in result["reasons"])
        )
    def test_author_task_cannot_perform_review_mutations(self):
        result = assess_role_boundary(
            {
                "task_role": "author",
                "reviewer_namespace_used": False,
                "author_namespace_used": True,
                "review_mutations": ["approve PR"],
            }
        )
        self.assertEqual(result["status"], "violation")
        self.assertTrue(
            any("reviewer-only" in r for r in result["violations"])
        )
    def test_scratch_only_notes_are_not_durable_evidence(self):
        result = assess_role_boundary(
            {
                "task_role": "reviewer",
                "task_kind": "blind_pr_queue_review",
                "reviewer_namespace_used": True,
                "scratch_evidence_claimed": True,
                "scratch_evidence_durable": False,
            }
        )
        self.assertEqual(result["status"], "warning")
        self.assertTrue(any("scratch-only" in r for r in result["reasons"]))
 class TestFinalReport(unittest.TestCase):
    """Required behavior 6 + acceptance criteria: the report must
    distinguish each proof, and only a fully proven run earns an "A"."""
@@ -466,7 +399,7 @@ class TestFinalReport(unittest.TestCase):
            "identity_eligible": True,
            "merge_performed": False,
            "issue_status_verified": True,
-            "role_boundary": _good_role_boundary(),
+            "controller_handoff": "Controller Handoff\n- Task: test",
        }
        kwargs.update(overrides)
        return build_final_report(**kwargs)
@@ -479,7 +412,6 @@ class TestFinalReport(unittest.TestCase):
        self.assertTrue(report["identity_eligible"])
        self.assertTrue(report["pr_author_distinct_from_reviewer"])
        self.assertEqual(report["session_contamination"], "clean")
        self.assertEqual(report["role_boundary"], "clean")
        self.assertTrue(report["validated_on_pinned_head"])
        self.assertFalse(report["merge_performed"])
        self.assertTrue(report["issue_status_verified"])
@@ -552,36 +484,23 @@ class TestFinalReport(unittest.TestCase):
        self.assertNotEqual(report["grade"], "A")
        self.assertFalse(report["merge_allowed"])
-    def test_missing_role_boundary_downgrades_and_blocks_merge(self):
+    def test_failed_capability_proof_downgrades(self):
-        kwargs = {
+        cap_proof = {
-            "checkout_proof": _good_checkout(),
+            "proven": False,
-            "inventory": _good_inventory(),
+            "reasons": ["task mark_issue not allowed"]
            "validation": _good_validation(),
            "contamination": _good_contamination(),
            "identity_eligible": True,
            "merge_performed": False,
            "issue_status_verified": True,
        }
-        report = build_final_report(**kwargs)
+        report = self._report(capability_proof=cap_proof)
        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["merge_allowed"])
+        self.assertTrue(any("capability" in r for r in report["downgrade_reasons"]))
        self.assertEqual(report["role_boundary"], "warning")
-    def test_role_boundary_violation_blocks_report(self):
+    def test_failed_sweep_proof_downgrades(self):
-        boundary = assess_role_boundary(
+        sweep_proof = {
-            {
+            "proven": False,
-                "task_role": "reviewer",
+            "reasons": ["method missing"]
-                "task_kind": "blind_pr_queue_review",
+        }
-                "reviewer_namespace_used": True,
+        report = self._report(sweep_proof=sweep_proof)
-                "author_namespace_used": True,
+        self.assertNotEqual(report["grade"], "A")
-                "author_mutations": ["create PR"],
+        self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
                "operator_authorized_author_work": False,
                "mixed_namespace_justification": "implementation pivot",
            }
        )
        report = self._report(role_boundary=boundary)
        self.assertEqual(report["grade"], "blocked")
        self.assertFalse(report["merge_allowed"])
 class TestStdoutIsolation(unittest.TestCase):
@@ -696,244 +615,103 @@ class TestRepoNameDisambiguation(unittest.TestCase):
 class TestControllerHandoff(unittest.TestCase):
-    """Issue #182: final reports must end with a Controller Handoff."""
+    """#183: every final report must contain the Controller Handoff section."""
-    BASE_HANDOFF = "\n".join([
+    def test_report_with_handoff_passes(self):
-        "## Controller Handoff",
+        report = "some details\n\nController Handoff\n- Task: foo"
-        "",
+        result = assess_controller_handoff(report)
-        "- Task: implement issue #182",
+        self.assertTrue(result["present"])
        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
        "- Role: author",
        "- Identity: jcwalker3 / prgs-author",
        "- Issue/PR: #182 / PR #999",
        "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
        "- Files changed: review_proofs.py",
        "- Validation: 700 passed, 6 skipped",
        "- Mutations: one PR opened",
        "- Workspace mutations: none",
        "- Current status: PR open",
        "- Blockers: none",
        "- Next: review PR #999",
        "- Safety: no self-review; no self-merge; no secrets",
    ])
-    def test_report_without_handoff_is_downgraded(self):
+    def test_report_without_handoff_downgrades(self):
-        result = assess_controller_handoff("long report text, no handoff")
+        report = "long details without the section"
-        self.assertEqual(result["verdict"], "missing")
+        result = assess_controller_handoff(report)
-        self.assertTrue(result["downgraded"])
+        self.assertFalse(result["present"])
-
+        self.assertTrue(any("Controller Handoff" in r for r in result["reasons"]))
    def test_wrong_title_is_downgraded(self):
        text = self.BASE_HANDOFF.replace(
            "## Controller Handoff", "## Handoff Summary")
        result = assess_controller_handoff(text)
        self.assertEqual(result["verdict"], "missing")
    def test_complete_base_handoff_passes(self):
        result = assess_controller_handoff(
            "full report body...\n\n" + self.BASE_HANDOFF)
        self.assertEqual(result["verdict"], "complete")
        self.assertFalse(result["downgraded"])
    def test_missing_base_fields_are_listed(self):
        text = "\n".join(
            line for line in self.BASE_HANDOFF.splitlines()
            if not line.startswith(("- Mutations:", "- Safety:")))
        result = assess_controller_handoff(text)
        self.assertEqual(result["verdict"], "incomplete")
        self.assertTrue(result["downgraded"])
        self.assertIn("Mutations", result["missing_fields"])
        self.assertIn("Safety", result["missing_fields"])
    def test_review_role_requires_review_fields(self):
        result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
        self.assertEqual(result["verdict"], "incomplete")
        self.assertIn("Pinned reviewed head", result["missing_fields"])
        self.assertIn("Merge result", result["missing_fields"])
        complete = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Selected PR: #999",
            "- Reviewer eligibility: passed",
            "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
            "- Review decision: approve",
            "- Merge result: merged",
            "- Linked issue status: closed",
            "- Cleanup status: branch deleted",
        ])
        result = assess_controller_handoff(complete, role="review")
        self.assertEqual(result["verdict"], "complete")
    def test_author_role_requires_author_fields(self):
        complete = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Selected issue: #182",
            "- Claim/comment status: comment-claimed",
            "- PR number opened: #999",
            "- No review/merge: confirmed",
        ])
        result = assess_controller_handoff(complete, role="author")
        self.assertEqual(result["verdict"], "complete")
        result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
        self.assertEqual(result["verdict"], "incomplete")
        self.assertIn("No review/merge confirmation", result["missing_fields"])
    def test_author_role_rejects_equivalent_or_multiple_issues(self):
        # 1. equivalent reference blocked
        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Selected issue: Issue #194 / #196 equivalent",
            "- Claim/comment status: comment-claimed",
            "- PR number opened: #999",
            "- No review/merge: confirmed",
        ])
        res = assess_controller_handoff(incomplete_eq, role="author")
        self.assertEqual(res["verdict"], "incomplete")
        self.assertIn("Selected issue", res["missing_fields"])
        # 2. multiple issues blocked
        incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Selected issue: #194, #196",
            "- Claim/comment status: comment-claimed",
            "- PR number opened: #999",
            "- No review/merge: confirmed",
        ])
        res = assess_controller_handoff(incomplete_multi, role="author")
        self.assertEqual(res["verdict"], "incomplete")
        self.assertIn("Selected issue", res["missing_fields"])
    def test_author_role_rejects_fuzzy_pr_number(self):
        incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Selected issue: #196",
            "- Claim/comment status: comment-claimed",
            "- PR number opened: PR #203 / #204 equivalent",
            "- No review/merge: confirmed",
        ])
        res = assess_controller_handoff(incomplete_pr, role="author")
        self.assertEqual(res["verdict"], "incomplete")
        self.assertIn("PR number opened", res["missing_fields"])
    def test_inventory_role_requires_inventory_fields(self):
        complete = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Repositories checked: Gitea-Tools, mcp-control-plane",
            "- Open PR counts: 2 / 0",
            "- Selected PR or reason: none eligible (self-authored)",
            "- Inventory completeness: complete, no pagination needed",
        ])
        result = assess_controller_handoff(complete, role="inventory")
        self.assertEqual(result["verdict"], "complete")
    def test_skill_doc_declares_handoff_requirement(self):
        # Doc-contract: SKILL.md must keep requiring the exact section and
        # naming this validator, or the convention silently rots.
        skill = (
            __import__("pathlib").Path(__file__).resolve().parent.parent
            / "skills" / "llm-project-workflow" / "SKILL.md"
        ).read_text(encoding="utf-8")
        self.assertIn("## Controller Handoff", skill)
        self.assertIn("assess_controller_handoff", skill)
        self.assertIn("issue #182", skill)
    def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
        # 1. Workspace mutations: none is rejected when local_edits is True
        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Selected issue: #196",
            "- Claim/comment status: comment-claimed",
            "- PR number opened: #203",
            "- No review/merge: confirmed",
        ])
        res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
        self.assertEqual(res["verdict"], "incomplete")
        self.assertIn("Workspace mutations", res["missing_fields"])
        # 2. Workspace mutations: edited files is allowed when local_edits is True
        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
            "- Selected issue: #196",
            "- Claim/comment status: comment-claimed",
            "- PR number opened: #203",
            "- No review/merge: confirmed",
        ])
        res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
        self.assertEqual(res2["verdict"], "complete")
-class TestPRInventoryTrustGate(unittest.TestCase):
+class TestAuthorReporting(unittest.TestCase):
-    """Issue #194: unit tests for the PR inventory trust gate."""
+    """Harness assertions for author reporting and capability/sweep proofs (#183)."""
-    def setUp(self):
+    def test_good_capability_proof_passes(self):
-        self.profile = {
+        resolved = {
-            "profile_name": "prgs-reviewer",
+            "mark_issue": {
-            "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"],
+                "requested_task": "mark_issue",
                "required_operation_permission": "gitea.issue.write",
                "allowed_in_current_session": True,
            }
        }
-        self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+        result = assess_capability_proof(resolved)
        self.assertTrue(result["proven"])
-    def test_trusted_nonempty(self):
+    def test_missing_capability_proof_fails_closed(self):
-        res = pr_inventory_trust_gate([{"number": 1}])
+        result = assess_capability_proof({})
-        self.assertEqual(res["status"], "trusted_nonempty")
+        self.assertFalse(result["proven"])
-        self.assertFalse(res["corroborated"])
+        self.assertTrue(any("fail closed" in r for r in result["reasons"]))
-    def test_inventory_error_none_or_not_list(self):
+    def test_unresolved_capability_proof_fails_closed(self):
-        self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error")
+        # unknown task resolving to None/unknown requested task
-        self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error")
+        resolved = {
            "mark_issue": {
                "requested_task": "unknown_task",
                "required_operation_permission": None,
                "allowed_in_current_session": None,
            }
        }
        result = assess_capability_proof(resolved)
        self.assertFalse(result["proven"])
        self.assertTrue(any("could not be resolved" in r for r in result["reasons"]))
-    def test_untrusted_empty_no_pagination_or_corroboration(self):
+    def test_good_secret_sweep_passes(self):
-        res = pr_inventory_trust_gate(
+        sweep = {
-            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            "method": "git diff | grep -iE 'token|secret'",
-            state="open", authenticated_profile=self.profile,
+            "scope": "staged diff relative to master",
-            local_remote_url=self.local_url, user_context=None,
+            "clean": True,
-            corroboration_open_pr_counter=None, has_finality_metadata=False
+        }
-        )
+        result = assess_secret_sweep(sweep)
-        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertTrue(result["proven"])
-        self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"])
+        self.assertEqual(result["verdict"], "strong")
-    def test_untrusted_empty_profile_permission_mismatch(self):
+    def test_vague_secret_sweep_without_method_is_weak(self):
-        bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]}
+        sweep = {
-        res = pr_inventory_trust_gate(
+            "method": "",
-            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            "scope": "staged diff",
-            state="open", authenticated_profile=bad_profile,
+            "clean": True,
-            local_remote_url=self.local_url, user_context=None,
+        }
-            corroboration_open_pr_counter=0, has_finality_metadata=False
+        result = assess_secret_sweep(sweep)
-        )
+        self.assertFalse(result["proven"])
-        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertEqual(result["verdict"], "weak")
-        self.assertIn("authenticated profile lacks read permissions", res["reasons"])
+        self.assertTrue(any("method" in r or "scan" in r for r in result["reasons"]))
-    def test_untrusted_empty_remote_url_mismatch(self):
+    def test_unconfirmed_secret_sweep_is_invalid(self):
-        bad_url = "https://gitea.prgs.cc/other-org/other-repo.git"
+        sweep = {
-        res = pr_inventory_trust_gate(
+            "method": "git diff | grep",
-            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            "scope": "staged diff",
-            state="open", authenticated_profile=self.profile,
+            "clean": False,
-            local_remote_url=bad_url, user_context=None,
+        }
-            corroboration_open_pr_counter=0, has_finality_metadata=False
+        result = assess_secret_sweep(sweep)
-        )
+        self.assertFalse(result["proven"])
-        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertEqual(result["verdict"], "weak")
        self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"])
-    def test_untrusted_empty_user_context_indicates_prs(self):
+    def test_good_author_pr_report_passes(self):
-        res = pr_inventory_trust_gate(
+        pr = {
-            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            "pr_number": 185,
-            state="open", authenticated_profile=self.profile,
+            "branch": "feat/issue-184-repo-name-disambiguation",
-            local_remote_url=self.local_url, user_context="please check open PR #181",
+            "head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
-            corroboration_open_pr_counter=0, has_finality_metadata=False
+        }
-        )
+        result = assess_author_pr_report(pr)
-        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertTrue(result["complete"])
        self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"]))
-    def test_trusted_empty_with_corroboration(self):
+    def test_incomplete_author_pr_report_fails(self):
-        res = pr_inventory_trust_gate(
+        pr = {
-            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            "pr_number": 0,
-            state="open", authenticated_profile=self.profile,
+            "branch": "",
-            local_remote_url=self.local_url, user_context=None,
+            "head_sha": "e2bccba",
-            corroboration_open_pr_counter=0, has_finality_metadata=False
+        }
-        )
+        result = assess_author_pr_report(pr)
-        self.assertEqual(res["status"], "trusted_empty")
+        self.assertFalse(result["complete"])
-        self.assertTrue(res["corroborated"])
+        self.assertTrue(any("number" in r for r in result["reasons"]))
-
+        self.assertTrue(any("branch" in r for r in result["reasons"]))
-    def test_trusted_empty_with_finality_metadata(self):
+        self.assertTrue(any("SHA" in r for r in result["reasons"]))
        res = pr_inventory_trust_gate(
            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
            state="open", authenticated_profile=self.profile,
            local_remote_url=self.local_url, user_context=None,
            corroboration_open_pr_counter=None, has_finality_metadata=True
        )
        self.assertEqual(res["status"], "trusted_empty")
        self.assertTrue(res["corroborated"])
 if __name__ == "__main__":
@@ -14,39 +14,11 @@ BRANCHES = REPO / "branches"
 def run(script, *args):
-    branch = None
+    proc = subprocess.run(
-    for arg in args:
+        ["bash", str(SCRIPTS / script), *args],
-        if not arg.startswith("-"):
+        capture_output=True, text=True, cwd=str(REPO),
-            branch = arg
+    )
-            break
+    return proc.returncode, proc.stdout, proc.stderr
    lock_file = Path("/tmp/gitea_issue_lock.json")
    created_lock = False
    if script == "worktree-start" and branch:
        import re
        import json
        m = re.search(r"issue-(\d+)", branch)
        if not m:
            m = re.search(r"pr-(\d+)", branch)
        issue_num = int(m.group(1)) if m else 999
        lock_file.write_text(json.dumps({
            "issue_number": issue_num,
            "branch_name": branch,
            "remote": "prgs",
            "org": "Scaled-Tech-Consulting",
            "repo": "Gitea-Tools"
        }), encoding="utf-8")
        created_lock = True
    try:
        proc = subprocess.run(
            ["bash", str(SCRIPTS / script), *args],
            capture_output=True, text=True, cwd=str(REPO),
        )
        return proc.returncode, proc.stdout, proc.stderr
    finally:
        if created_lock and lock_file.exists():
            lock_file.unlink()
 class TestWorktreeStart(unittest.TestCase):