Gitea-Tools/author_proofs.py

"""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", []))
        ),
    }