Gitea-Tools/review_proofs.py

"""Fail-closed proof helpers for the blind PR queue review workflow (#173).

A blind queue review must *prove* its preconditions instead of asserting
them. Each helper here evaluates one proof from issue #173's required
behaviors and returns a plain dict verdict; ``build_final_report`` combines
them and only awards an "A" grade when every proof is present. Missing
evidence always downgrades or blocks — never upgrades.

These helpers are pure (no network, no git calls): the workflow gathers the
raw facts (Gitea PR head SHA, ``git rev-parse HEAD``, listing pagination
state, pytest output, whoami results) and passes them in, so the same logic
is usable from prompts, harness assertions, and tests. The MCP-level gates
(mcp-control-plane #68/#76) remain authoritative for mutations; nothing
here weakens or replaces them.
"""

import re

_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")


# Repo name disambiguation rules for blind PR queue review.
# User phrases referencing the "MCP Gitea tool" (or similar) must resolve to
# Gitea-Tools repo, not be confused with mcp-control-plane.
# If ambiguous (e.g. just "open PRs"), check both configured repos.
REPO_ALIASES = {
    "gitea-tools": "Scaled-Tech-Consulting/Gitea-Tools",
    "gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
    "mcp gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
    "gitea mcp tool": "Scaled-Tech-Consulting/Gitea-Tools",
    "gitea-tools repo": "Scaled-Tech-Consulting/Gitea-Tools",
    "mcp-control-plane": "Scaled-Tech-Consulting/mcp-control-plane",
    "mcp control plane": "Scaled-Tech-Consulting/mcp-control-plane",
}


def resolve_repos_from_user_reference(
    reference: str, configured: list[str] | None = None
) -> list[str]:
    """Resolve a user reference string to the list of target repos to inventory.

    - Exact aliases for Gitea-Tools map only to Gitea-Tools.
    - mcp-control-plane aliases map only to it.
    - Empty, ambiguous, or general "open PRs" default to all configured repos
      (both by default).
    - Returns subset of configured; never invents new repos.
    """
    if configured is None:
        configured = [
            "Scaled-Tech-Consulting/Gitea-Tools",
            "Scaled-Tech-Consulting/mcp-control-plane",
        ]
    if not reference or not reference.strip():
        return list(configured)

    ref_lower = reference.lower()
    matched = []
    for alias, full_name in REPO_ALIASES.items():
        if alias in ref_lower:
            if full_name not in matched:
                matched.append(full_name)
    if matched:
        # return only the matched ones that are in configured, preserving order
        return [r for r in configured if r in matched]
    # no specific alias match → check all (complete inventory required)
    return list(configured)


SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
    "evidence missing: report contamination as unknown and "
    "choose another PR or stop"
)


def _normalize_ref(ref):
    """Normalize a git ref for base comparison.

    Strips ``refs/heads/`` and ``refs/remotes/<remote>/`` prefixes and a
    plain ``<remote>/`` prefix so ``master``, ``prgs/master`` and
    ``refs/remotes/prgs/master`` all compare equal. Returns '' for empty
    input (which then fails closed in the caller).
    """
    ref = (ref or "").strip()
    if ref.startswith("refs/heads/"):
        ref = ref[len("refs/heads/"):]
    elif ref.startswith("refs/remotes/"):
        ref = ref[len("refs/remotes/"):]
        ref = ref.split("/", 1)[1] if "/" in ref else ref
    elif "/" in ref:
        # Remote-tracking shorthand like 'prgs/master'.
        ref = ref.split("/", 1)[1]
    return ref


def verify_pinned_head_checkout(pinned_head_sha, local_head_sha,
                                pr_base_ref, diff_base_ref):
    """Required behaviors 1–2: prove HEAD == pinned PR head, correct base.

    Both SHAs must be full 40-hex and identical — an abbreviated or prefix
    match is not proof. The diff base must normalize to the PR base branch.
    Returns {'proven', 'block', 'reasons', 'pinned_head_sha',
    'local_head_sha'}; ``block`` is True whenever the proof fails, meaning
    review and merge must stop before proceeding.
    """
    reasons = []
    pinned = (pinned_head_sha or "").strip().lower()
    local = (local_head_sha or "").strip().lower()

    if not pinned:
        reasons.append("pinned PR head SHA missing; fail closed")
    elif not _FULL_SHA.match(pinned):
        reasons.append("pinned PR head SHA is not a full 40-hex SHA; fail closed")

    if not local:
        reasons.append("local checkout HEAD SHA missing; fail closed")
    elif not _FULL_SHA.match(local):
        reasons.append(
            "local checkout HEAD SHA is not a full 40-hex SHA "
            "(abbreviated SHAs are not proof); fail closed"
        )

    if pinned and local and _FULL_SHA.match(pinned) and _FULL_SHA.match(local):
        if pinned != local:
            reasons.append(
                "local checkout HEAD does not match the pinned PR head SHA; "
                "stop before review/merge"
            )

    base = _normalize_ref(pr_base_ref)
    diff_base = _normalize_ref(diff_base_ref)
    if not base:
        reasons.append("PR base ref missing; fail closed")
    if not diff_base:
        reasons.append("diff base ref missing; fail closed")
    if base and diff_base and base != diff_base:
        reasons.append(
            f"diff base '{diff_base}' is not the PR base branch '{base}'"
        )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "pinned_head_sha": pinned or None,
        "local_head_sha": local or None,
    }


def assess_inventory_completeness(repo_reports, required_repos):
    """Required behavior 3: the queue inventory must prove completeness.

    *repo_reports* is a list of dicts, one per repository listed:
    ``{'repo', 'state_filter', 'pagination_complete', 'open_pr_count'}``.
    The inventory is complete only when every required repository has a
    report that states its filter, proves pagination was handled, and gives
    a total open-PR count. ``can_claim_exhaustive`` — permission to say
    "these are the only open PRs" — is granted only for a complete
    inventory.
    """
    reasons = []
    by_repo = {}
    for report in repo_reports or []:
        name = (report.get("repo") or "").strip()
        if name:
            by_repo[name] = report

    totals = {}
    for repo in required_repos:
        report = by_repo.get(repo)
        if report is None:
            reasons.append(f"repository '{repo}' was not inventoried")
            continue
        if not (report.get("state_filter") or "").strip():
            reasons.append(f"repository '{repo}': PR state filter not stated")
        if report.get("pagination_complete") is not True:
            reasons.append(
                f"repository '{repo}': pagination not proven complete; "
                "a truncated listing cannot support an exhaustive claim"
            )
        count = report.get("open_pr_count")
        if not isinstance(count, int) or count < 0:
            reasons.append(
                f"repository '{repo}': total open PR count not reported"
            )
        else:
            totals[repo] = count

    complete = not reasons and bool(required_repos)
    if not required_repos:
        reasons.append("no required repositories configured; fail closed")
    return {
        "complete": complete,
        "can_claim_exhaustive": complete,
        "reasons": reasons,
        "total_open_prs": totals,
    }


def assess_validation_report(report):
    """Required behavior 4: validation reporting needs exact evidence.

    *report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
    ``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
    ``justification``), ``canonical_command``, ``deviation_justification``.

    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).
    - 'strong'  — full evidence.
    """
    reasons = []
    command = (report.get("command") or "").strip()

    if not command:
        return {
            "verdict": "invalid",
            "claimable": False,
            "reasons": ["no validation command stated; result cannot be claimed"],
        }
    if report.get("output_read") is not True:
        return {
            "verdict": "invalid",
            "claimable": False,
            "reasons": [
                "command output was not read to completion; a validation "
                "result cannot be claimed"
            ],
        }

    counts = [report.get("passed"), report.get("failed"), report.get("skipped")]
    if any(not isinstance(c, int) for c in counts):
        reasons.append(
            "test counts (passed/failed/skipped) missing; report is incomplete"
        )

    for ignored in report.get("ignored_paths") or []:
        path = ignored.get("path") or "<unnamed>"
        if not (ignored.get("justification") or "").strip():
            reasons.append(
                f"ignored path '{path}' has no justification for why it is "
                "safe to ignore"
            )

    canonical = (report.get("canonical_command") or "").strip()
    if canonical and command != canonical:
        if not (report.get("deviation_justification") or "").strip():
            reasons.append(
                "command differs from the repository's canonical validation "
                "command and the deviation is not justified"
            )

    verdict = "strong" if not reasons else "weak"
    return {"verdict": verdict, "claimable": True, "reasons": reasons}


def assess_self_review_contamination(reviewer_identity, pr_author,
                                     session_authored=None,
                                     evidence_source=None):
    """Required behavior 5: contamination is evidence-backed, never assumed.

    Distinguishes the two cases the issue calls out:
    - reviewer identity == PR author → contaminated (the identities
      themselves are the evidence);
    - same-session authorship → contaminated only with an explicit
      ``evidence_source``; a bare claim (either way) yields 'unknown'.

    'unknown' is a stop signal: choose another PR or stop — it is never
    silently treated as clean or as contaminated.
    """
    reviewer = (reviewer_identity or "").strip()
    author = (pr_author or "").strip()
    evidence = (evidence_source or "").strip()

    if not reviewer or not author:
        return {
            "status": "unknown",
            "reasons": ["reviewer identity or PR author unknown; fail closed"],
            "safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
        }

    if reviewer == author:
        return {
            "status": "contaminated",
            "reasons": [
                f"authenticated reviewer '{reviewer}' is the PR author; "
                "self-review is blocked"
            ],
            "safe_next_action": "choose another PR or stop",
        }

    if session_authored is True:
        if evidence:
            return {
                "status": "contaminated",
                "reasons": [
                    "this session authored/touched the PR branch "
                    f"(evidence: {evidence})"
                ],
                "safe_next_action": "choose another PR or stop",
            }
        return {
            "status": "unknown",
            "reasons": [
                "same-session authorship was claimed without an evidence "
                "source; contamination may not be declared by assumption"
            ],
            "safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
        }

    if session_authored is False and evidence:
        return {
            "status": "clean",
            "reasons": [
                f"reviewer '{reviewer}' differs from author '{author}' and "
                f"session non-authorship is evidenced ({evidence})"
            ],
            "safe_next_action": "proceed",
        }

    return {
        "status": "unknown",
        "reasons": [
            "session authorship was not established with evidence; "
            "contamination status is unknown"
        ],
        "safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
    }


def assess_capability_evidence(capability_claims):
    """#179 gap 1: a capability claim needs exact evidence, not assertion.

    *capability_claims* is a list of ``{'task', 'allowed',
    'evidence_source'}`` dicts, one per capability the report claims (e.g.
    review_pr, merge_pr). Each claim must name its task, be allowed, and
    cite an exact evidence source (``gitea_resolve_task_capability`` output
    or equivalent runtime-context evidence). No claims at all fails closed.
    """
    reasons = []
    claims = capability_claims or []
    if not claims:
        reasons.append(
            "no capability evidence provided; capability checks may not be "
            "claimed as passed"
        )
    for claim in claims:
        task = (claim.get("task") or "").strip() or "<unnamed task>"
        if claim.get("allowed") is not True:
            reasons.append(
                f"capability '{task}' is not proven allowed; fail closed"
            )
        if not (claim.get("evidence_source") or "").strip():
            reasons.append(
                f"capability '{task}' claimed without exact evidence "
                "(cite gitea_resolve_task_capability output or equivalent)"
            )
    proven = not reasons
    return {"proven": proven, "reasons": reasons, "claims": len(claims)}


def assess_sweep_evidence(sweep):
    """#179 gap 2: secret/provenance sweeps must state exact method + scope.

    *sweep* keys: ``command`` (the exact command, script, grep pattern, or
    named sweep method), ``scope`` (what was scanned, e.g. 'full PR diff
    against prgs/master'), ``clean`` (bool result). A vague summary without
    the exact method is downgraded; a missing sweep fails closed.
    """
    if not sweep:
        return {
            "verdict": "missing",
            "proven": False,
            "reasons": ["no secret/provenance sweep reported; fail closed"],
        }
    reasons = []
    if not (sweep.get("command") or "").strip():
        reasons.append(
            "sweep method/command not stated exactly (command, script, "
            "pattern, or named sweep method required)"
        )
    if not (sweep.get("scope") or "").strip():
        reasons.append("sweep scope not stated (what diff/files were scanned)")
    if not isinstance(sweep.get("clean"), bool):
        reasons.append("sweep result not stated as clean/not-clean")
    verdict = "exact" if not reasons else "vague"
    return {
        "verdict": verdict,
        "proven": verdict == "exact",
        "reasons": reasons,
        "clean": sweep.get("clean") if isinstance(sweep.get("clean"), bool)
        else None,
    }


def assess_live_state_recheck(recheck):
    """#179 gap 3: explicit live-state recheck before review/merge mutation.

    *recheck* keys: ``pr_state``, ``pinned_head_sha``, ``live_head_sha``,
    ``pinned_base_ref``, ``live_base_ref``, ``blocking_change_requests``.
    Proven only when the PR is still open, the live head equals the pinned
    head (full 40-hex), the base branch is unchanged, and blocking review
    state was checked and is absent. Not performing the recheck fails
    closed and blocks mutation.
    """
    if not recheck:
        return {
            "proven": False,
            "block": True,
            "reasons": [
                "final live-state recheck not performed before mutation; "
                "fail closed"
            ],
        }
    reasons = []
    if (recheck.get("pr_state") or "").strip().lower() != "open":
        reasons.append(
            f"PR state is '{recheck.get('pr_state')}', not open; stop"
        )

    pinned = (recheck.get("pinned_head_sha") or "").strip().lower()
    live = (recheck.get("live_head_sha") or "").strip().lower()
    if not (_FULL_SHA.match(pinned) and _FULL_SHA.match(live)):
        reasons.append(
            "pinned/live head SHAs missing or not full 40-hex; fail closed"
        )
    elif pinned != live:
        reasons.append(
            "live head SHA no longer equals the pinned head; re-pin and "
            "re-validate before mutation"
        )

    base_pinned = _normalize_ref(recheck.get("pinned_base_ref"))
    base_live = _normalize_ref(recheck.get("live_base_ref"))
    if not base_pinned or not base_live:
        reasons.append("base refs missing from live-state recheck; fail closed")
    elif base_pinned != base_live:
        reasons.append(
            f"base branch changed from '{base_pinned}' to '{base_live}'"
        )

    blocking = recheck.get("blocking_change_requests")
    if blocking is None:
        reasons.append(
            "blocking review state not checked; fail closed"
        )
    elif blocking:
        reasons.append(
            "an undismissed REQUEST_CHANGES / blocking review state remains "
            "unresolved"
        )

    proven = not reasons
    return {"proven": proven, "block": not proven, "reasons": reasons}


def assess_role_boundary(task_role, namespaces_used, justification=None):
    """#179 gap 4: stay in the task's namespace unless use is justified.

    *task_role* is 'reviewer' or 'author'; *namespaces_used* lists every
    MCP namespace the run called (e.g. ['gitea-reviewer', 'gitea-author']).
    A namespace whose name does not contain the task role is foreign; using
    one is clean only with an explicit justification in the report.
    """
    role = (task_role or "").strip().lower()
    reasons = []
    if not role:
        reasons.append("task role not stated; fail closed")
    if namespaces_used is None:
        reasons.append("namespaces used were not reported; fail closed")

    foreign = []
    for namespace in namespaces_used or []:
        if role and role not in (namespace or "").lower():
            foreign.append(namespace)
    if foreign and not (justification or "").strip():
        reasons.append(
            f"{role or 'task'} run used foreign namespace(s) "
            f"{foreign} without justification"
        )

    proven = not reasons
    return {
        "proven": proven,
        "reasons": reasons,
        "foreign_namespaces": foreign,
        "justified": bool((justification or "").strip()),
    }


def build_final_report(checkout_proof, inventory, validation, contamination,
                       identity_eligible, merge_performed,
                       issue_status_verified,
                       capability_evidence=None, sweep=None, live_state=None,
                       role_boundary=None):
    """Required behavior 6 + acceptance criteria: one report, distinct proofs.

    Combines the individual proof verdicts into the final-report fields the
    issue requires, computes ``merge_allowed`` from the proofs, and grades
    the run:

    - 'A'          — every proof present (merge itself is optional).
    - 'downgraded' — one or more proofs missing/weak; do not merge.
    - 'blocked'    — a *violation*: a merge was claimed although the proofs
      did not allow one.

    #179 raises the A bar: the report must also carry exact capability
    evidence (``assess_capability_evidence``), an exact secret/provenance
    sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
    (``assess_live_state_recheck`` — also required for ``merge_allowed``),
    and a clean role boundary (``assess_role_boundary``). Omitting any of
    them downgrades; a merge without the live recheck is a violation.
    """
    contamination_status = contamination.get("status", "unknown")
    checkout_proven = bool(checkout_proof.get("proven"))
    validation_claimable = bool(validation.get("claimable"))
    validation_strong = validation.get("verdict") == "strong"

    capability_evidence = capability_evidence or {
        "proven": False,
        "reasons": ["capability evidence not provided (#179)"],
    }
    sweep = sweep or {
        "verdict": "missing",
        "proven": False,
        "reasons": ["secret/provenance sweep evidence not provided (#179)"],
    }
    live_state = live_state or {
        "proven": False,
        "block": True,
        "reasons": ["pre-mutation live-state recheck not provided (#179)"],
    }
    role_boundary = role_boundary or {
        "proven": False,
        "reasons": ["role-boundary/namespace usage not reported (#179)"],
    }
    capability_proven = bool(capability_evidence.get("proven"))
    sweep_proven = bool(sweep.get("proven"))
    live_state_proven = bool(live_state.get("proven"))
    role_boundary_clean = bool(role_boundary.get("proven"))

    downgrade_reasons = []
    if not identity_eligible:
        downgrade_reasons.append("identity/profile not eligible for review")
    if not inventory.get("complete"):
        downgrade_reasons.append("open-PR inventory completeness not proven")
        downgrade_reasons.extend(inventory.get("reasons", []))
    if not checkout_proven:
        downgrade_reasons.append("pinned-head checkout not proven")
        downgrade_reasons.extend(checkout_proof.get("reasons", []))
    if not validation_strong:
        downgrade_reasons.append(
            f"validation evidence is {validation.get('verdict', 'missing')}"
        )
        downgrade_reasons.extend(validation.get("reasons", []))
    if contamination_status != "clean":
        downgrade_reasons.append(
            f"session contamination status is '{contamination_status}'"
        )
    if not issue_status_verified:
        downgrade_reasons.append("linked issue status not verified")
    if not capability_proven:
        downgrade_reasons.append("exact capability evidence missing (#179)")
        downgrade_reasons.extend(capability_evidence.get("reasons", []))
    if not sweep_proven:
        downgrade_reasons.append(
            f"secret/provenance sweep evidence is "
            f"{sweep.get('verdict', 'missing')} (#179)"
        )
        downgrade_reasons.extend(sweep.get("reasons", []))
    if not live_state_proven:
        downgrade_reasons.append(
            "pre-mutation live-state recheck missing or failed (#179)"
        )
        downgrade_reasons.extend(live_state.get("reasons", []))
    if not role_boundary_clean:
        downgrade_reasons.append("role/namespace boundary not clean (#179)")
        downgrade_reasons.extend(role_boundary.get("reasons", []))

    merge_allowed = (
        identity_eligible
        and checkout_proven
        and contamination_status == "clean"
        and validation_claimable
        and validation.get("verdict") != "invalid"
        # #179: no merge without a proven final live-state recheck.
        and live_state_proven
    )

    violations = []
    if merge_performed and not merge_allowed:
        violations.append(
            "merge was performed/claimed although the proofs did not allow "
            "one; this run is blocked, not graded"
        )

    if violations:
        grade = "blocked"
    elif downgrade_reasons:
        grade = "downgraded"
    else:
        grade = "A"

    return {
        "grade": grade,
        "violations": violations,
        "downgrade_reasons": downgrade_reasons,
        "identity_eligible": bool(identity_eligible),
        "pr_author_distinct_from_reviewer":
            contamination_status in ("clean",),
        "session_contamination": contamination_status,
        "inventory_complete": bool(inventory.get("complete")),
        "validated_on_pinned_head": checkout_proven and validation_claimable,
        "validation_passed":
            validation_claimable and validation.get("verdict") != "invalid",
        "validation_verdict": validation.get("verdict"),
        "merge_allowed": merge_allowed,
        "merge_performed": bool(merge_performed),
        "issue_status_verified": bool(issue_status_verified),
        "capability_evidence_proven": capability_proven,
        "sweep_verdict": sweep.get("verdict"),
        "live_state_recheck_proven": live_state_proven,
        "role_boundary_clean": role_boundary_clean,
    }