test: require issue-lock proof in author handoffs (#208 )

Add Issue lock proof to author HANDOFF_ROLE_FIELDS and update controller handoff tests. Strengthen create_pr tests to read and validate the issue lock file (branch match, Closes #N, reveal opt-in unchanged). Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-06 11:13:43 -04:00
8 changed files with 3 additions and 1225 deletions
@@ -120,11 +120,6 @@ def assess_capability_stop_report(
    capability_denied: bool = True,
 ) -> dict:
    """Validate final report purity after reviewer capability denial."""
    from review_proofs import (
        assess_empty_queue_report,
        parse_trust_gate_status_from_report,
    )
    text = report_text or ""
    lower = text.lower()
    violations = []
@@ -153,19 +148,13 @@ def assess_capability_stop_report(
        r"inventory empty",
        re.I,
    )
    parsed_status = parse_trust_gate_status_from_report(text)
    effective_status = trust_gate_status or parsed_status
    if empty_queue_patterns.search(text):
-        if effective_status != "trusted_empty":
+        if trust_gate_status != "trusted_empty":
            violations.append(
                "empty-queue claim after capability stop without "
                "pr_inventory_trust_gate.status == trusted_empty"
            )
    empty_queue = assess_empty_queue_report(text)
    if empty_queue.get("claimed") and not empty_queue.get("proven"):
        violations.extend(empty_queue.get("reasons") or [])
    ok, elig_violations = validate_eligibility_wording(text)
    violations.extend(elig_violations)
@@ -259,7 +259,6 @@ import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -2275,23 +2274,6 @@ def gitea_merge_pr(
    return result
 def _local_git_remote_url(remote_name: str) -> str | None:
    """Best-effort local ``git remote get-url`` for trust-gate corroboration."""
    try:
        proc = subprocess.run(
            ["git", "remote", "get-url", remote_name],
            capture_output=True,
            text=True,
            cwd=PROJECT_ROOT,
        )
        if proc.returncode != 0:
            return None
        url = (proc.stdout or "").strip()
        return url or None
    except Exception:
        return None
@mcp.tool()
 def gitea_review_pr(
    pr_number: int,
@@ -2359,8 +2341,6 @@ def gitea_review_pr(
    prs_found_count = 0
    pr_details_list = []
    inventory_msg = ""
    inventory_trust_gate = None
    prs: list = []
    h, o, r = _resolve(remote, host, org, repo)
    auth = None
@@ -2445,28 +2425,7 @@ def gitea_review_pr(
            if inventory_msg:
                report_lines.append(inventory_msg)
            else:
-                inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
+                report_lines.append("Open PRs found: 0")
                    prs if inventory_attempted else None,
                    remote=remote,
                    org=o,
                    repo=r,
                    state="open",
                    authenticated_profile=profile,
                    local_remote_url=_local_git_remote_url(remote),
                    has_finality_metadata=True,
                )
                report_lines.extend(
                    review_proofs.format_pr_inventory_trust_gate_report(
                        inventory_trust_gate
                    )
                )
                if inventory_trust_gate.get("status") == "trusted_empty":
                    report_lines.append("Open PRs found: 0 (trusted_empty)")
                else:
                    report_lines.append(
                        "Empty-queue claim blocked: "
                        "pr_inventory_trust_gate did not return trusted_empty"
                    )
    else:
        report_lines.append(inventory_msg)
@@ -664,8 +664,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
            report_text, review_decision_lock
        )
    empty_queue_report = assess_empty_queue_report(report_text)
    contamination_status = contamination.get("status", "unknown")
    checkout_proven = bool(checkout_proof.get("proven"))
    validation_claimable = bool(validation.get("claimable"))
@@ -789,11 +787,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
            "reviewer worktree safety proof missing or failed (#233)"
        )
        downgrade_reasons.extend(worktree.get("reasons", []))
    if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"):
        downgrade_reasons.append(
            "empty-queue report missing or failed trust-gate proof (#198)"
        )
        downgrade_reasons.extend(empty_queue_report.get("reasons", []))
    merge_allowed = (
        identity_eligible
@@ -853,12 +846,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
        "unrelated_mutations_avoided": bool(
            worktree.get("unrelated_mutations_avoided")
        ),
        "empty_queue_trust_gate_proven": (
            empty_queue_report.get("proven")
            if empty_queue_report.get("claimed")
            else True
        ),
        "empty_queue_trust_gate_status": empty_queue_report.get("status"),
    }
@@ -1029,40 +1016,13 @@ HANDOFF_ROLE_FIELDS = {
        ("Repositories checked", ("repositories checked", "repos checked")),
        ("Open PR counts", ("open pr counts", "open pr count",
                            "open prs per repo")),
        ("PR inventory trust gate", ("pr inventory trust gate",
                                     "pr_inventory_trust_gate.status",
                                     "trust gate status")),
        ("Trust gate reasons", ("trust gate reasons",
                                "pr_inventory_trust_gate.reason")),
        ("Trust gate corroborated", ("trust gate corroborated",
                                     "pr_inventory_trust_gate.corroborated")),
        ("Inventory profile", ("inventory profile", "inventory mcp profile")),
        ("Selected PR or reason", ("selected pr", "none selected",
                                   "reason none selected")),
        ("Inventory completeness", ("inventory complete", "inventory scoped",
                                    "inventory completeness")),
    ),
    "continuation": (
        ("Continuation mode", ("continuation mode", "continuation")),
        ("Existing PR", ("existing pr", "pr number")),
        ("PR author", ("pr author", "existing pr author")),
        ("Issue claim status", ("issue claim", "claim status",
                                "status:in-progress")),
        ("Branch", ("branch", "existing branch")),
        ("Old PR head", ("old pr head", "old head")),
        ("New PR head", ("new pr head", "new head")),
        ("Session authored PR", ("session authored pr", "authored pr")),
        ("Why continuation allowed", ("why continuation", "continuation allowed")),
    ),
 }
 # Canonical secret/provenance sweep for comparable continuation runs (#189).
 CANONICAL_SECRET_SWEEP_COMMAND = (
    "git diff prgs/master...HEAD | rg -i "
    "'(token|password|secret|api[_-]?key|authorization:)'"
 )
 CANONICAL_SECRET_SWEEP_SCOPE = "full feature-branch diff against prgs/master"
 def _handoff_section_lines(report_text):
    """Return the lines of the Controller Handoff section, or None."""
@@ -1352,616 +1312,6 @@ def pr_inventory_trust_gate(
    }
 def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
    parts = (full_repo or "").split("/", 1)
    if len(parts) == 2:
        return parts[0].strip() or None, parts[1].strip() or None
    return None, None
 def assess_reviewer_queue_inventory(
    repo_reports: list[dict] | None,
    required_repos: list[str] | None = None,
    *,
    user_context: str | None = None,
 ) -> dict:
    """Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
    Any repository reporting ``open_pr_count == 0`` must pass
    ``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
    claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
    """
    required = list(required_repos or [
        "Scaled-Tech-Consulting/Gitea-Tools",
        "Scaled-Tech-Consulting/mcp-control-plane",
    ])
    completeness = assess_inventory_completeness(repo_reports, required)
    trust_gates: dict[str, dict] = {}
    blockers: list[str] = []
    can_claim_empty = bool(completeness.get("complete"))
    for report in repo_reports or []:
        repo = (report.get("repo") or "").strip()
        count = report.get("open_pr_count")
        if not isinstance(count, int) or count != 0:
            continue
        org, repo_name = _split_repo_slug(repo)
        list_response = report.get("list_prs_response")
        if list_response is None:
            list_response = []
        gate = pr_inventory_trust_gate(
            list_response,
            remote=report.get("remote"),
            org=org,
            repo=repo_name,
            state=report.get("state_filter"),
            authenticated_profile=report.get("authenticated_profile"),
            local_remote_url=report.get("local_remote_url"),
            user_context=user_context or report.get("user_context"),
            corroboration_open_pr_counter=report.get(
                "corroboration_open_pr_counter"
            ),
            has_finality_metadata=report.get("pagination_complete") is True,
        )
        trust_gates[repo] = gate
        status = gate.get("status")
        if status != "trusted_empty":
            can_claim_empty = False
            blockers.append(
                f"repository '{repo}': empty PR list trust gate is "
                f"'{status}'; cannot claim 'no open PRs'"
            )
            blockers.extend(gate.get("reasons") or [])
    return {
        "complete": bool(completeness.get("complete")),
        "can_claim_empty_queue": can_claim_empty,
        "can_claim_exhaustive": (
            bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
        ),
        "inventory_reasons": list(completeness.get("reasons") or []),
        "trust_gates": trust_gates,
        "blockers": blockers,
        "reasons": list(completeness.get("reasons") or []) + blockers,
    }
 def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
    """Render trust-gate lines for MCP inventory output."""
    lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
    if gate.get("corroborated"):
        lines.append("pr_inventory_trust_gate.corroborated: true")
    for reason in gate.get("reasons") or []:
        lines.append(f"pr_inventory_trust_gate.reason: {reason}")
    return lines
 _EMPTY_QUEUE_CLAIM = re.compile(
    r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
    r"nothing to review|queue cleared|inventory empty|"
    r"open pr count:\s*0|workflow correctly stops with nothing",
    re.I,
 )
 _WEAK_EMPTY_QUEUE_CORROBORATION = re.compile(
    r"latest commit.*(?:merge|pr #)|merge of pr #|"
    r"master latest commit|recent merge proves",
    re.I,
 )
 _TRUST_GATE_STATUS_LINE = re.compile(
    r"pr_inventory_trust_gate\.status:\s*(\S+)",
    re.I,
 )
 def parse_trust_gate_status_from_report(report_text: str | None) -> str | None:
    """Extract ``pr_inventory_trust_gate.status`` from report text, if present."""
    match = _TRUST_GATE_STATUS_LINE.search(report_text or "")
    return match.group(1).strip().lower() if match else None
 def assess_empty_queue_report(
    report_text: str | None,
    *,
    trust_gate: dict | None = None,
    task_role: str | None = None,
    inventory_profile: str | None = None,
 ) -> dict:
    """Issue #198: empty-queue reports must cite the formal trust-gate result.
    Blocks reports that claim an empty queue without
    ``pr_inventory_trust_gate.status == trusted_empty``, required inventory
    metadata, or that rely on weak corroboration (e.g. a recent merge commit).
    """
    text = report_text or ""
    lower = text.lower()
    reasons: list[str] = []
    missing: list[str] = []
    if not _EMPTY_QUEUE_CLAIM.search(text):
        return {
            "claimed": False,
            "proven": True,
            "block": False,
            "status": None,
            "missing_fields": [],
            "reasons": [],
        }
    status = (
        (trust_gate or {}).get("status")
        or parse_trust_gate_status_from_report(text)
    )
    status_norm = (status or "").strip().lower() or None
    if not status_norm:
        missing.append("pr_inventory_trust_gate.status")
        reasons.append(
            "empty-queue report missing pr_inventory_trust_gate.status; "
            "fail closed"
        )
    elif status_norm != "trusted_empty":
        reasons.append(
            f"empty-queue report has trust-gate status '{status_norm}', "
            "not trusted_empty"
        )
    has_gate_reasons = (
        "pr_inventory_trust_gate.reason" in lower
        or bool((trust_gate or {}).get("reasons"))
    )
    if status_norm and status_norm != "trusted_empty" and not has_gate_reasons:
        missing.append("pr_inventory_trust_gate.reasons")
    if status_norm == "trusted_empty":
        if "pr_inventory_trust_gate.corroborated" not in lower and (
            trust_gate or {}
        ).get("corroborated") is not True:
            missing.append("pr_inventory_trust_gate.corroborated")
    inventory_markers = (
        "repository:",
        "remote:",
        "owner:",
        "state filter:",
        "state_filter:",
    )
    if not any(marker in lower for marker in inventory_markers):
        missing.append("inventory remote/owner/repo/state filter")
    profile_markers = (
        "mcp profile:",
        "mcp-profile:",
        "inventory profile:",
        "active profile:",
    )
    has_profile = (
        any(marker in lower for marker in profile_markers)
        or bool((inventory_profile or "").strip())
    )
    if not has_profile:
        missing.append("inventory MCP profile")
    if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text):
        if "pr_inventory_trust_gate.status: trusted_empty" not in lower:
            reasons.append(
                "weak corroboration (recent merge commit) cannot substitute "
                "for pr_inventory_trust_gate.status == trusted_empty"
            )
    role = (task_role or "").strip().lower()
    if role == "author" and re.search(
        r"reviewer queue|nothing to review|review backlog empty",
        text,
        re.I,
    ):
        reasons.append(
            "author-bound session presented reviewer queue inventory as a "
            "reviewer decision"
        )
    if missing:
        reasons.extend(
            f"empty-queue report missing required field: {field}"
            for field in missing
        )
    proven = not reasons and not missing
    return {
        "claimed": True,
        "proven": proven,
        "block": not proven,
        "status": status_norm,
        "missing_fields": missing,
        "reasons": reasons,
    }
 # ── Issue selection / continuation mode (#188) ───────────────────────────────
 ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr"
 ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr"
 ISSUE_SELECTION_IN_PROGRESS = "in_progress"
 ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit"
 ISSUE_SELECTION_EXCLUDED = "excluded"
 _NO_OPEN_PR_CLAIM = re.compile(
    r"no duplicate pr|no open pr|no pr open|no eligible pr|"
    r"no existing pr|without an open pr",
    re.IGNORECASE,
 )
 def classify_issue_for_selection(
    issue_number: int,
    *,
    labels: list[str] | None = None,
    open_prs: list[dict] | None = None,
    operator_continuation_requested: bool = False,
    continuation_issue_numbers: list[int] | None = None,
    excluded: bool = False,
 ) -> dict:
    """Classify one issue for author queue selection (#188)."""
    label_set = {str(l).lower() for l in (labels or [])}
    prs = list(open_prs or [])
    continuation_issues = set(continuation_issue_numbers or [])
    if excluded:
        status = ISSUE_SELECTION_EXCLUDED
        selectable_for_fresh_work = False
        reasons = ["issue explicitly excluded from selection"]
    elif "status:in-progress" in label_set:
        status = ISSUE_SELECTION_IN_PROGRESS
        selectable_for_fresh_work = False
        reasons = ["issue already marked status:in-progress"]
    elif prs and (
        operator_continuation_requested
        or issue_number in continuation_issues
    ):
        status = ISSUE_SELECTION_CONTINUATION_EXPLICIT
        selectable_for_fresh_work = False
        reasons = ["operator requested continuation for issue with open PR"]
    elif prs:
        status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR
        selectable_for_fresh_work = False
        reasons = [
            f"issue #{issue_number} already represented by open PR "
            f"#{prs[0].get('number')}"
        ]
    else:
        status = ISSUE_SELECTION_UNCLAIMED_NO_PR
        selectable_for_fresh_work = True
        reasons = []
    return {
        "issue_number": issue_number,
        "status": status,
        "selectable_for_fresh_work": selectable_for_fresh_work,
        "open_prs": prs,
        "reasons": reasons,
    }
 def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
    """Fail closed when fresh selection picks an issue with an open PR."""
    reasons = []
    for item in classifications or []:
        if item.get("selectable_for_fresh_work"):
            continue
        if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT:
            continue
        if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR:
            reasons.append(
                f"issue #{item.get('issue_number')} has open PR and was "
                "selected for fresh work without continuation mode"
            )
    return {
        "complete": not reasons,
        "downgraded": bool(reasons),
        "reasons": reasons,
    }
 def canonical_secret_sweep_report(*, clean: bool) -> dict:
    """Return a sweep report using the canonical command/scope (#189)."""
    return {
        "method": CANONICAL_SECRET_SWEEP_COMMAND,
        "scope": CANONICAL_SECRET_SWEEP_SCOPE,
        "clean": clean,
    }
 def assess_force_with_lease_push_report(
    report_text: str,
    *,
    old_remote_head: str | None = None,
    expected_lease_head: str | None = None,
    new_pushed_head: str | None = None,
    branch_pushed: str | None = None,
    used_force_with_lease: bool | None = None,
 ) -> dict:
    """Issue #189: force-with-lease pushes must disclose lease evidence."""
    if not used_force_with_lease:
        return {"complete": True, "downgraded": False, "reasons": []}
    text = report_text or ""
    lower = text.lower()
    reasons = []
    if "force-with-lease" not in lower and "force with lease" not in lower:
        reasons.append(
            "continuation push used force-with-lease but report does not say so"
        )
    for label, sha in (
        ("old remote", old_remote_head),
        ("lease", expected_lease_head),
        ("pushed", new_pushed_head),
    ):
        if not sha:
            reasons.append(f"force-with-lease proof missing {label} head SHA")
        elif not _FULL_SHA.match(sha.lower()):
            reasons.append(
                f"force-with-lease {label} head SHA is not a full 40-hex SHA"
            )
        elif sha.lower() not in lower:
            reasons.append(
                f"continuation report missing {label} head SHA in evidence"
            )
    if branch_pushed:
        if branch_pushed.lower() not in lower:
            reasons.append("continuation report missing pushed branch name")
        only_branch_tokens = (
            "only feature branch",
            "push branch only",
            "only the feature branch",
            "only pushed feature branch",
        )
        if not any(t in lower for t in only_branch_tokens):
            reasons.append(
                "continuation report missing confirmation that only the "
                "feature branch was pushed"
            )
    return {
        "complete": not reasons,
        "downgraded": bool(reasons),
        "reasons": reasons,
    }
 def assess_continuation_mode_report(
    report_text: str,
    *,
    pr_number: int | None = None,
    pr_author: str | None = None,
    issue_number: int | None = None,
    issue_claim_status: str | None = None,
    branch: str | None = None,
    old_head_sha: str | None = None,
    new_head_sha: str | None = None,
    session_authored_pr: bool | None = None,
    continuation_allowed_reason: str | None = None,
 ) -> dict:
    """Issue #188/#189: continuation mode must disclose full PR evidence."""
    text = report_text or ""
    lower = text.lower()
    reasons = []
    if not any(p in lower for p in ("continuation", "continue pr", "continue issue")):
        reasons.append("report does not declare continuation mode")
    if pr_number is not None:
        if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
            reasons.append(f"continuation report missing PR #{pr_number}")
    if pr_author and pr_author.lower() not in lower:
        reasons.append("continuation report missing PR author")
    if issue_number is not None:
        if (
            f"#{issue_number}" not in lower
            and f"issue #{issue_number}" not in lower
            and f"issue {issue_number}" not in lower
        ):
            reasons.append(f"continuation report missing issue #{issue_number}")
    if issue_claim_status:
        claim_lower = issue_claim_status.lower()
        claim_tokens = (
            claim_lower,
            "claim status",
            "issue claim",
            "status:in-progress",
        )
        if not any(t in lower for t in claim_tokens):
            reasons.append("continuation report missing issue claim status")
    if branch and branch.lower() not in lower:
        reasons.append("continuation report missing branch name")
    for label, sha in (("old", old_head_sha), ("new", new_head_sha)):
        if not sha:
            reasons.append(f"continuation proof missing {label} head SHA")
        elif not _FULL_SHA.match(sha.lower()):
            reasons.append(
                f"continuation {label} head SHA is not a full 40-hex SHA"
            )
        elif sha.lower() not in lower:
            reasons.append(
                f"continuation report missing {label} head SHA in evidence"
            )
    if session_authored_pr is not None:
        authored_tokens = ("session authored", "authored pr", "own pr", "my pr")
        if not any(t in lower for t in authored_tokens):
            reasons.append(
                "continuation report missing whether session authored the PR"
            )
    if continuation_allowed_reason:
        reason_lower = continuation_allowed_reason.lower()
        if (
            reason_lower not in lower
            and not any(w in lower for w in reason_lower.split()[:3])
        ):
            reasons.append("continuation report missing why continuation is allowed")
    return {
        "complete": not reasons,
        "downgraded": bool(reasons),
        "reasons": reasons,
    }
 def assess_contradictory_no_pr_claim(
    report_text: str,
    *,
    edited_pr_numbers: list[int] | None = None,
    issue_open_pr_map: dict[int, int] | None = None,
 ) -> dict:
    """Downgrade when report claims no open PR but later edits one."""
    text = report_text or ""
    lower = text.lower()
    reasons = []
    if not _NO_OPEN_PR_CLAIM.search(lower):
        return {"complete": True, "downgraded": False, "reasons": []}
    edited = list(edited_pr_numbers or [])
    for pr_num in edited:
        if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
            reasons.append(
                f"report claims no open PR but edited PR #{pr_num}"
            )
    for issue_num, pr_num in (issue_open_pr_map or {}).items():
        if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower:
            if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
                reasons.append(
                    f"report claims no open PR for issue #{issue_num} but "
                    f"PR #{pr_num} exists and was referenced"
                )
    return {
        "complete": not reasons,
        "downgraded": bool(reasons),
        "reasons": reasons,
    }
 def assess_edited_pr_inventory_coverage(
    report_text: str,
    *,
    edited_pr_numbers: list[int] | None = None,
    inventoried_pr_numbers: list[int] | None = None,
 ) -> dict:
    """Open PR inventory must include PRs the run later edits (#188)."""
    text = report_text or ""
    lower = text.lower()
    reasons = []
    inventoried = set(inventoried_pr_numbers or [])
    for pr_num in edited_pr_numbers or []:
        if pr_num in inventoried:
            continue
        if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower:
            reasons.append(
                f"edited PR #{pr_num} missing from open PR inventory"
            )
    return {
        "complete": not reasons,
        "downgraded": bool(reasons),
        "reasons": reasons,
    }
 def assess_issue_selection_final_report(
    report_text: str,
    *,
    mode: str = "fresh",
    classifications: list[dict] | None = None,
    continuation_proof: dict | None = None,
    edited_pr_numbers: list[int] | None = None,
    inventoried_pr_numbers: list[int] | None = None,
    issue_open_pr_map: dict[int, int] | None = None,
 ) -> dict:
    """Issue #188: composite A-bar for author issue-selection runs."""
    handoff_role = "continuation" if mode == "continuation" else "author"
    checks = {
        "controller_handoff": assess_controller_handoff(
            report_text, role=handoff_role
        ),
        "contradictory_no_pr": assess_contradictory_no_pr_claim(
            report_text,
            edited_pr_numbers=edited_pr_numbers,
            issue_open_pr_map=issue_open_pr_map,
        ),
        "edited_pr_inventory": assess_edited_pr_inventory_coverage(
            report_text,
            edited_pr_numbers=edited_pr_numbers,
            inventoried_pr_numbers=inventoried_pr_numbers,
        ),
    }
    if mode == "fresh":
        checks["fresh_selection"] = assess_fresh_issue_selection(classifications)
    else:
        proof = continuation_proof or {}
        checks["continuation_mode"] = assess_continuation_mode_report(
            report_text,
            pr_number=proof.get("pr_number"),
            pr_author=proof.get("pr_author"),
            issue_number=proof.get("issue_number"),
            issue_claim_status=proof.get("issue_claim_status"),
            branch=proof.get("branch"),
            old_head_sha=proof.get("old_head_sha"),
            new_head_sha=proof.get("new_head_sha"),
            session_authored_pr=proof.get("session_authored_pr"),
            continuation_allowed_reason=proof.get("continuation_allowed_reason"),
        )
        push_proof = proof.get("push_proof") or {}
        checks["force_with_lease_push"] = assess_force_with_lease_push_report(
            report_text,
            old_remote_head=push_proof.get("old_remote_head"),
            expected_lease_head=push_proof.get("expected_lease_head"),
            new_pushed_head=push_proof.get("new_pushed_head"),
            branch_pushed=push_proof.get("branch_pushed"),
            used_force_with_lease=push_proof.get("used_force_with_lease"),
        )
        sweep = proof.get("secret_sweep")
        if sweep is not None:
            checks["secret_sweep"] = assess_secret_sweep(sweep)
    reasons = []
    downgraded = False
    for name, result in checks.items():
        verdict = result.get("verdict")
        if verdict in ("missing", "incomplete"):
            downgraded = True
            reasons.extend(result.get("reasons") or [])
        elif result.get("proven") is False:
            downgraded = True
            reasons.extend(
                f"{name}: {r}" for r in (result.get("reasons") or [])
            )
        elif result.get("downgraded") or not result.get("complete", True):
            downgraded = True
            reasons.extend(
                f"{name}: {r}" for r in (result.get("reasons") or [])
            )
    return {
        "grade": "A" if not downgraded else "downgraded",
        "downgraded": downgraded,
        "checks": checks,
        "reasons": reasons,
        "complete": not downgraded,
    }
 def assess_duplicate_search_proof(report_text, matches):
    """#207: reject LLM duplicate summaries that omit known title matches."""
    return issue_duplicate_gate.assess_duplicate_search_proof(
@@ -379,17 +379,6 @@ Role-specific fields (append to the compact block):
  `Linked issue status:`, `Cleanup status:`
 - author tasks: `Selected issue:`, `Claim/comment status:`,
  `PR number opened:`, `No review/merge:` (explicit confirmation)
 - continuation tasks (#188/#189): `Continuation mode:`, `Existing PR:`,
  `PR author:`, `Issue claim status:`, `Branch:`, `Old PR head:`,
  `New PR head:`, `Session authored PR:`, `Why continuation allowed:` —
  when rebasing with `git push --force-with-lease`, also record
  `Old remote head:`, `Lease head:`, `Pushed head:`, and confirm
  `Push branch only:` (feature branch only). Use the canonical secret sweep
  from `review_proofs.CANONICAL_SECRET_SWEEP_COMMAND` so runs are
  comparable. Issues with open PRs are excluded from fresh selection unless
  the operator explicitly requests continuation
  (`review_proofs.classify_issue_for_selection`,
  `assess_issue_selection_final_report`)
 - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
  `Selected PR or reason none selected:`, `Inventory completeness:`
@@ -17,18 +17,6 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
  configured repos were not checked. This is not a complete queue inventory."
 - A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
  if the other configured repo was not inventoried.
 - PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
  the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
  `review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
  clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
  corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
  sufficient proof.
 - Empty-queue report wall (#198): if the final report claims "no open PRs",
  "queue empty", or "nothing to review", it must include verbatim:
  `pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
  remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
  commit is not valid corroboration. Author-bound sessions must not present
  reviewer queue inventory as a reviewer decision.
 Rules (llm-project-workflow):
 - Review in a SEPARATE detached review worktree, never the author's folder.
@@ -148,19 +148,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
        )
        self.assertFalse(result["pure"])
    def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
        report = (
            "Cannot perform reviewer task under current profile. "
            "No reviewer mutations performed.\n"
            "Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
            "pr_inventory_trust_gate.status: trusted_empty\n"
            "pr_inventory_trust_gate.corroborated: true\n"
            "Inventory profile: prgs-reviewer\n"
            "No open PRs in queue."
        )
        result = assess_capability_stop_terminal_report(report)
        self.assertTrue(result["pure"])
    def test_pure_terminal_report_passes(self):
        report = (
            "Cannot perform reviewer task under current profile. "
@@ -261,79 +261,12 @@ class TestPRQueueInventory(unittest.TestCase):
        for call in mock_api.call_args_list:
            self.assertEqual(call.args[0], "GET")
    @patch("mcp_server._local_git_remote_url")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
-    def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
+    def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
        self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
    ):
        mock_get_profile.return_value = {
            "profile_name": "gitea-author",
            "allowed_operations": ["read"],
            "forbidden_operations": [],
            "base_url": None,
        }
        mock_get_all.return_value = []
        mock_api.return_value = {"login": "jcwalker3"}
        mock_local_url.return_value = None
        result = gitea_review_pr(
            pr_number=1,
            event="APPROVE",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertFalse(result["success"])
        msg = result["message"]
        self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
        self.assertIn("Empty-queue claim blocked", msg)
        self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
    @patch("mcp_server._local_git_remote_url")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
    def test_empty_inventory_trusted_empty_when_gate_passes(
        self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
    ):
        mock_get_profile.return_value = {
            "profile_name": "gitea-author",
            "allowed_operations": ["read"],
            "forbidden_operations": [],
            "base_url": None,
        }
        mock_get_all.return_value = []
        mock_api.return_value = {"login": "jcwalker3"}
        mock_local_url.return_value = (
            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
        )
        result = gitea_review_pr(
            pr_number=1,
            event="APPROVE",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertFalse(result["success"])
        msg = result["message"]
        self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
        self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
    @patch("mcp_server._local_git_remote_url")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
    def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
        """Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
        mock_local_url.return_value = (
            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
        )
        for event in ["APPROVE", "REQUEST_CHANGES"]:
            mock_get_profile.return_value = {
                "profile_name": "gitea-author",
@@ -21,23 +21,11 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 from review_proofs import (  # noqa: E402
    ISSUE_SELECTION_CONTINUATION_EXPLICIT,
    ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
    assess_author_pr_report,
    assess_capability_evidence,
    assess_capability_proof,
    assess_contradictory_no_pr_claim,
    assess_continuation_mode_report,
    assess_force_with_lease_push_report,
    canonical_secret_sweep_report,
    CANONICAL_SECRET_SWEEP_COMMAND,
    assess_controller_handoff,
    assess_edited_pr_inventory_coverage,
    assess_empty_queue_report,
    assess_fresh_issue_selection,
    assess_inventory_completeness,
    assess_issue_selection_final_report,
    assess_reviewer_queue_inventory,
    assess_live_state_recheck,
    assess_review_mutation_final_report,
    assess_role_boundary,
@@ -46,7 +34,6 @@ from review_proofs import (  # noqa: E402
    assess_sweep_evidence,
    assess_validation_report,
    build_final_report,
    classify_issue_for_selection,
    pr_inventory_trust_gate,
    resolve_repos_from_user_reference,
    verify_pinned_head_checkout,
@@ -1032,10 +1019,6 @@ class TestControllerHandoff(unittest.TestCase):
        complete = self.BASE_HANDOFF + "\n" + "\n".join([
            "- Repositories checked: Gitea-Tools, mcp-control-plane",
            "- Open PR counts: 2 / 0",
            "- PR inventory trust gate: trusted_nonempty / trusted_empty",
            "- Trust gate reasons: none",
            "- Trust gate corroborated: true",
            "- Inventory profile: prgs-reviewer",
            "- Selected PR or reason: none eligible (self-authored)",
            "- Inventory completeness: complete, no pagination needed",
        ])
@@ -1228,157 +1211,6 @@ class TestPRInventoryTrustGate(unittest.TestCase):
        self.assertTrue(res["corroborated"])
 class TestAssessReviewerQueueInventory(unittest.TestCase):
    """Issue #196: trust gate wired into canonical queue inventory."""
    def _repo_report(self, **overrides):
        report = {
            "repo": "Scaled-Tech-Consulting/Gitea-Tools",
            "state_filter": "open",
            "pagination_complete": True,
            "open_pr_count": 0,
            "list_prs_response": [],
            "remote": "prgs",
            "authenticated_profile": {
                "profile_name": "prgs-reviewer",
                "allowed_operations": ["read", "gitea.read"],
            },
            "local_remote_url": (
                "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
            ),
        }
        report.update(overrides)
        return report
    def test_empty_without_corroboration_blocks_empty_queue_claim(self):
        result = assess_reviewer_queue_inventory([
            self._repo_report(pagination_complete=False),
            self._repo_report(
                repo="Scaled-Tech-Consulting/mcp-control-plane",
                open_pr_count=0,
                pagination_complete=False,
            ),
        ])
        self.assertFalse(result["can_claim_empty_queue"])
        self.assertIn("untrusted_empty", str(result["trust_gates"]))
    def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
        result = assess_reviewer_queue_inventory([
            self._repo_report(),
            self._repo_report(
                repo="Scaled-Tech-Consulting/mcp-control-plane",
                local_remote_url=(
                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
                    "mcp-control-plane.git"
                ),
            ),
        ])
        self.assertTrue(result["can_claim_empty_queue"])
        self.assertTrue(result["can_claim_exhaustive"])
    def test_user_context_indicating_open_prs_blocks_empty_claim(self):
        result = assess_reviewer_queue_inventory(
            [self._repo_report()],
            user_context="please review open PR #195 in the queue",
        )
        self.assertFalse(result["can_claim_empty_queue"])
        self.assertTrue(result["blockers"])
    def test_nonempty_inventory_skips_empty_trust_gate_block(self):
        result = assess_reviewer_queue_inventory([
            self._repo_report(open_pr_count=2),
            self._repo_report(
                repo="Scaled-Tech-Consulting/mcp-control-plane",
                open_pr_count=1,
            ),
        ])
        self.assertTrue(result["complete"])
        self.assertEqual(result["trust_gates"], {})
 class TestAssessEmptyQueueReport(unittest.TestCase):
    """Issue #198: empty-queue reports require formal trust-gate proof."""
    def _trusted_report(self, **extra):
        lines = [
            "Queue inventory complete.",
            "Repository: Scaled-Tech-Consulting/Gitea-Tools",
            "Open PR count: 0",
            "pr_inventory_trust_gate.status: trusted_empty",
            "pr_inventory_trust_gate.corroborated: true",
            "Inventory profile: prgs-reviewer",
            "Workflow correctly stops with nothing to review.",
        ]
        lines.extend(extra)
        return "\n".join(lines)
    def test_non_empty_report_not_claimed(self):
        result = assess_empty_queue_report("Reviewed PR #236 and merged.")
        self.assertFalse(result["claimed"])
        self.assertTrue(result["proven"])
    def test_empty_claim_without_trust_gate_blocked(self):
        report = (
            "Open PR count: 0\n"
            "Pagination complete: yes\n"
            "Queue cleared."
        )
        result = assess_empty_queue_report(report)
        self.assertTrue(result["claimed"])
        self.assertFalse(result["proven"])
        self.assertTrue(result["block"])
    def test_trusted_empty_report_with_required_fields_passes(self):
        result = assess_empty_queue_report(self._trusted_report())
        self.assertTrue(result["proven"])
    def test_weak_merge_commit_corroboration_blocked(self):
        report = (
            "Open PR count: 0\n"
            "Master latest commit is merge of PR #79 so queue is empty."
        )
        result = assess_empty_queue_report(report)
        self.assertFalse(result["proven"])
        self.assertTrue(
            any("weak corroboration" in r for r in result["reasons"])
        )
    def test_author_session_reviewer_queue_wording_blocked(self):
        result = assess_empty_queue_report(
            self._trusted_report(),
            task_role="author",
        )
        self.assertFalse(result["proven"])
    def test_build_final_report_downgrades_weak_empty_queue(self):
        final = build_final_report(
            checkout_proof=_good_checkout(),
            inventory=_good_inventory(),
            validation=_good_validation(),
            contamination=_good_contamination(),
            identity_eligible=True,
            merge_performed=False,
            issue_status_verified=True,
            capability_evidence=_good_capability_evidence(),
            sweep=_good_sweep(),
            live_state=_good_live_state(),
            role_boundary=_good_role_boundary(),
            review_mutation=_good_review_mutation(),
            controller_handoff=_good_handoff(),
            capability_proof=_good_capability_proof(),
            sweep_proof=_good_secret_sweep(),
            worktree_proof={
                "worktree_path": "/repo/branches/review-pr-1",
                "porcelain_status": "",
                "scratch_used": True,
                "scratch_path": "/repo/branches/review-pr-1",
            },
            report_text="Open PR count: 0. Queue cleared.",
        )
        self.assertNotEqual(final["grade"], "A")
        self.assertFalse(final["empty_queue_trust_gate_proven"])
 class TestCapabilityEvidence(unittest.TestCase):
    """#179 gap 1: capability claims need exact evidence."""
@@ -1673,254 +1505,5 @@ class TestAuthorReporting(unittest.TestCase):
        self.assertFalse(result["complete"])
 class TestIssueSelectionContinuation(unittest.TestCase):
    """Issue #188: continuation mode wall for issues with open PRs."""
    OLD_SHA = PINNED
    NEW_SHA = OTHER
    OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
    def test_open_pr_issue_excluded_from_fresh_selection(self):
        classified = classify_issue_for_selection(
            183, open_prs=self.OPEN_PR,
        )
        self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
        self.assertFalse(classified["selectable_for_fresh_work"])
        blocked = assess_fresh_issue_selection([classified])
        self.assertTrue(blocked["downgraded"])
    def test_explicit_continuation_allows_represented_issue(self):
        classified = classify_issue_for_selection(
            183,
            open_prs=self.OPEN_PR,
            operator_continuation_requested=True,
        )
        self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
        blocked = assess_fresh_issue_selection([classified])
        self.assertFalse(blocked["downgraded"])
    def test_contradictory_no_pr_claim_downgrades(self):
        report = (
            "Selected issue #183; no duplicate PR open. "
            "Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
        )
        result = assess_contradictory_no_pr_claim(
            report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
        )
        self.assertTrue(result["downgraded"])
    def test_edited_pr_must_appear_in_inventory(self):
        report = "Open PR inventory: PR #195 only."
        result = assess_edited_pr_inventory_coverage(
            report,
            edited_pr_numbers=[187],
            inventoried_pr_numbers=[195],
        )
        self.assertTrue(result["downgraded"])
    def test_continuation_report_requires_old_and_new_head(self):
        report = (
            "Issue #182 continuation mode. PR #186. "
            f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
            "Session authored PR: yes. Continuation allowed: operator requested."
        )
        result = assess_continuation_mode_report(
            report,
            pr_number=186,
            pr_author="jcwalker3",
            branch="feat/issue-182-controller-handoff",
            old_head_sha=self.OLD_SHA,
            new_head_sha=self.NEW_SHA,
            session_authored_pr=True,
            continuation_allowed_reason="operator requested continuation",
        )
        self.assertTrue(result["complete"])
    def test_issue_selection_final_report_continuation_earns_a(self):
        report = "\n".join([
            "Issue #182 continuation mode — no new issue claimed.",
            f"PR #186 updated: old head {self.OLD_SHA}, "
            f"new head {self.NEW_SHA}.",
            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
            "Session authored PR: yes.",
            "Continuation allowed: operator requested rebase.",
            "Open PR inventory included PR #186.",
            "## Controller Handoff",
            "- Task: continuation",
            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
            "- Role: author",
            "- Identity: prgs-author",
            "- Issue/PR: #182 / PR #186",
            "- Branch/SHA: feat/issue-182-controller-handoff",
            "- Files changed: review_proofs.py",
            "- Validation: tests passed",
            "- Mutations: push_branch",
            "- Workspace mutations: none",
            "- Current status: PR mergeable",
            "- Blockers: none",
            "- Next: review",
            "- Safety: no review/merge",
            "- Continuation mode: issue #182 continuation",
            "- Existing PR: #186",
            "- PR author: jcwalker3",
            "- Issue claim status: status:in-progress",
            "- Branch: feat/issue-182-controller-handoff",
            f"- Old PR head: {self.OLD_SHA}",
            f"- New PR head: {self.NEW_SHA}",
            "- Session authored PR: yes",
            "- Why continuation allowed: operator requested rebase",
        ])
        result = assess_issue_selection_final_report(
            report,
            mode="continuation",
            continuation_proof={
                "pr_number": 186,
                "pr_author": "jcwalker3",
                "issue_number": 182,
                "issue_claim_status": "status:in-progress",
                "branch": "feat/issue-182-controller-handoff",
                "old_head_sha": self.OLD_SHA,
                "new_head_sha": self.NEW_SHA,
                "session_authored_pr": True,
                "continuation_allowed_reason": "operator requested rebase",
            },
            edited_pr_numbers=[186],
            inventoried_pr_numbers=[186],
        )
        self.assertEqual(result["grade"], "A")
 class TestContinuationModeProofs(unittest.TestCase):
    """Issue #189: formal continuation proofs beyond #188 selection wall."""
    OLD_SHA = PINNED
    NEW_SHA = OTHER
    REMOTE_OLD = "601c608c00000000000000000000000000000000"
    LEASE_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
    PUSHED_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
    def test_force_with_lease_requires_lease_evidence(self):
        report = (
            "Issue #182 continuation. force-with-lease push to "
            "feat/issue-182-controller-handoff-enforcement."
        )
        result = assess_force_with_lease_push_report(
            report,
            old_remote_head=self.REMOTE_OLD,
            expected_lease_head=self.LEASE_SHA,
            new_pushed_head=self.PUSHED_SHA,
            branch_pushed="feat/issue-182-controller-handoff-enforcement",
            used_force_with_lease=True,
        )
        self.assertTrue(result["downgraded"])
    def test_force_with_lease_complete_with_full_evidence(self):
        report = (
            "Issue #182 continuation with git push --force-with-lease. "
            f"Old remote head {self.REMOTE_OLD}. "
            f"Lease head {self.LEASE_SHA}. "
            f"Pushed head {self.PUSHED_SHA}. "
            "Branch feat/issue-182-controller-handoff-enforcement. "
            "Push branch only: only the feature branch was pushed."
        )
        result = assess_force_with_lease_push_report(
            report,
            old_remote_head=self.REMOTE_OLD,
            expected_lease_head=self.LEASE_SHA,
            new_pushed_head=self.PUSHED_SHA,
            branch_pushed="feat/issue-182-controller-handoff-enforcement",
            used_force_with_lease=True,
        )
        self.assertTrue(result["complete"])
    def test_continuation_requires_issue_claim_status(self):
        report = (
            "Issue #182 continuation mode. PR #186. "
            f"old head {self.OLD_SHA} new head {self.NEW_SHA}. "
            "PR author jcwalker3. Branch feat/issue-182-test."
        )
        result = assess_continuation_mode_report(
            report,
            pr_number=186,
            pr_author="jcwalker3",
            issue_number=182,
            issue_claim_status="status:in-progress",
            branch="feat/issue-182-test",
            old_head_sha=self.OLD_SHA,
            new_head_sha=self.NEW_SHA,
        )
        self.assertTrue(result["downgraded"])
    def test_canonical_secret_sweep_helper(self):
        sweep = canonical_secret_sweep_report(clean=True)
        self.assertEqual(sweep["method"], CANONICAL_SECRET_SWEEP_COMMAND)
        self.assertTrue(sweep["clean"])
    def test_continuation_final_report_with_push_and_sweep_earns_a(self):
        branch = "feat/issue-182-controller-handoff-enforcement"
        report = "\n".join([
            "Issue #182 continuation mode — status:in-progress already set.",
            f"PR #186 updated via git push --force-with-lease.",
            f"Old PR head {self.OLD_SHA}; new PR head {self.NEW_SHA}.",
            f"Old remote head {self.REMOTE_OLD}.",
            f"Lease head {self.LEASE_SHA}. Pushed head {self.PUSHED_SHA}.",
            f"Branch {branch}. Push branch only: only the feature branch.",
            "PR author: jcwalker3. Session authored PR: yes.",
            "Continuation allowed: operator requested rebase.",
            "Open PR inventory included PR #186.",
            "## Controller Handoff",
            "- Task: continuation",
            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
            "- Role: author",
            "- Identity: prgs-author",
            "- Issue/PR: #182 / PR #186",
            f"- Branch/SHA: {branch}",
            "- Files changed: review_proofs.py",
            "- Validation: tests passed",
            "- Mutations: push_branch",
            "- Workspace mutations: none",
            "- Current status: PR mergeable",
            "- Blockers: none",
            "- Next: review",
            "- Safety: no review/merge",
            "- Continuation mode: issue #182 continuation",
            "- Existing PR: #186",
            "- PR author: jcwalker3",
            "- Issue claim status: status:in-progress",
            f"- Branch: {branch}",
            f"- Old PR head: {self.OLD_SHA}",
            f"- New PR head: {self.NEW_SHA}",
            "- Session authored PR: yes",
            "- Why continuation allowed: operator requested rebase",
        ])
        result = assess_issue_selection_final_report(
            report,
            mode="continuation",
            continuation_proof={
                "pr_number": 186,
                "pr_author": "jcwalker3",
                "issue_number": 182,
                "issue_claim_status": "status:in-progress",
                "branch": branch,
                "old_head_sha": self.OLD_SHA,
                "new_head_sha": self.NEW_SHA,
                "session_authored_pr": True,
                "continuation_allowed_reason": "operator requested rebase",
                "push_proof": {
                    "old_remote_head": self.REMOTE_OLD,
                    "expected_lease_head": self.LEASE_SHA,
                    "new_pushed_head": self.PUSHED_SHA,
                    "branch_pushed": branch,
                    "used_force_with_lease": True,
                },
                "secret_sweep": canonical_secret_sweep_report(clean=True),
            },
            edited_pr_numbers=[186],
            inventoried_pr_numbers=[186],
        )
        self.assertEqual(result["grade"], "A")
 if __name__ == "__main__":
    unittest.main()