Gitea-Tools/audit_reconciliation_mode.py

"""Audit vs cleanup phase gates for reconciliation workflows (#419).

Audit/reconciliation tasks are read-only unless a separate cleanup phase is
explicitly authorized with exact capability proof, safety proof, and
before/after snapshots. Cleanup mutations must be classified in final reports;
audit reports must not claim ``no mutations`` when cleanup occurred.
"""

from __future__ import annotations

import re
from typing import Any

RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md"

PHASE_AUDIT = "audit"
PHASE_CLEANUP = "cleanup"

# Tasks that enter audit phase on capability resolution (read-only default).
AUDIT_PHASE_TASKS = frozenset({
    "reconcile-landed-pr",
    "reconcile_landed_pr",
    "reconcile_issue_claims",
    "reconcile_merged_cleanups",
})

# Mutation tasks forbidden during audit phase (fail closed).
AUDIT_FORBIDDEN_TASKS = frozenset({
    "delete_branch",
    "create_branch",
    "push_branch",
    "create_pr",
    "commit_files",
    "gitea_commit_files",
    "mark_issue",
    "lock_issue",
    "claim_issue",
    "close_pr",
    "close_issue",
    "create_issue",
    "merge_pr",
    "review_pr",
    "submit_pr_review",
    "comment_pr",
    "comment_issue",
    "set_issue_labels",
})

# Shell/git commands audit phase must not run.
AUDIT_FORBIDDEN_COMMAND_RE = re.compile(
    r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|"
    r"gitea_delete_branch|delete_remote_branch)",
    re.IGNORECASE,
)

_NO_MUTATIONS_RE = re.compile(
    r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)",
    re.IGNORECASE,
)
_CLEANUP_OCCURRED_RE = re.compile(
    r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|"
    r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|"
    r"cleanup\s+phase\s*:\s*(?!none\b)\S)",
    re.IGNORECASE,
)
_EXTERNAL_STATE_RE = re.compile(
    r"^\s*[-*]?\s*external[- ]state mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
)
_GIT_REF_RE = re.compile(
    r"^\s*[-*]?\s*git ref mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
)
_CLEANUP_MUTATIONS_RE = re.compile(
    r"^\s*[-*]?\s*cleanup mutations\s*:",
    re.IGNORECASE | re.MULTILINE,
)
_CLEANUP_PHASE_AUTH_RE = re.compile(
    r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
)
_DELETE_CAPABILITY_RE = re.compile(
    r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
)
_BEFORE_AFTER_RE = re.compile(
    r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:",
    re.IGNORECASE | re.MULTILINE,
)
_SAFETY_PROOF_RE = re.compile(
    r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true",
    re.IGNORECASE | re.MULTILINE,
)

_session: dict[str, Any] | None = None


def _blank_session() -> dict[str, Any]:
    return {
        "phase": PHASE_AUDIT,
        "entered_from_task": None,
        "cleanup_authorized": False,
        "cleanup_authorization": {},
    }


def current_phase() -> str | None:
    """Return active reconciliation phase or None when unset."""
    if not _session:
        return None
    return _session.get("phase")


def active_record() -> dict[str, Any] | None:
    """Return a copy of the session record, if any."""
    return dict(_session) if _session else None


def clear_phase() -> None:
    """Clear reconciliation phase state."""
    global _session
    _session = None


def enter_audit_phase(task: str) -> dict[str, Any]:
    """Enter read-only audit phase for a reconciliation task."""
    global _session
    normalized = (task or "").strip().lower()
    _session = _blank_session()
    _session["entered_from_task"] = normalized
    return dict(_session)


def authorize_cleanup_phase(
    *,
    operator_approved: bool = False,
    workflow_authorized: bool = False,
    delete_capability_proven: bool = False,
    safety_proof: dict[str, Any] | None = None,
    before_after_snapshot: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Authorize cleanup phase after explicit approval and safety proofs."""
    reasons: list[str] = []
    if not (operator_approved or workflow_authorized):
        reasons.append(
            "cleanup phase requires operator approval or explicit workflow "
            "authorization"
        )
    if not delete_capability_proven:
        reasons.append(
            "cleanup phase requires exact delete_branch capability proof "
            "(gitea.branch.delete)"
        )
    safety = dict(safety_proof or {})
    if not safety.get("safe_to_delete_remote") and not safety.get(
        "safe_to_remove_worktree"
    ):
        reasons.append(
            "cleanup phase requires proof that branch/worktree is safe to remove"
        )
    snapshot = dict(before_after_snapshot or {})
    if not snapshot.get("before") or not snapshot.get("after"):
        reasons.append(
            "cleanup phase requires before/after state snapshot"
        )

    if reasons:
        return {
            "authorized": False,
            "phase": current_phase() or PHASE_AUDIT,
            "reasons": reasons,
            "safe_next_action": (
                "remain in audit-only mode or supply operator approval, "
                "delete_branch capability proof, safety proof, and "
                "before/after snapshot before cleanup"
            ),
        }

    global _session
    if _session is None:
        _session = _blank_session()
    _session["phase"] = PHASE_CLEANUP
    _session["cleanup_authorized"] = True
    _session["cleanup_authorization"] = {
        "operator_approved": operator_approved,
        "workflow_authorized": workflow_authorized,
        "delete_capability_proven": delete_capability_proven,
        "safety_proof": safety,
        "before_after_snapshot": snapshot,
    }
    return {
        "authorized": True,
        "phase": PHASE_CLEANUP,
        "reasons": [],
        "cleanup_authorization": dict(_session["cleanup_authorization"]),
        "safe_next_action": "proceed with authorized cleanup mutations only",
    }


def check_audit_task_enters_phase(task: str) -> bool:
    """Return whether resolving *task* should enter audit phase."""
    return (task or "").strip().lower() in AUDIT_PHASE_TASKS


def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]:
    """Fail closed when a mutation task runs during audit phase."""
    normalized = (task or "").strip().lower()
    phase = current_phase()
    if phase != PHASE_AUDIT:
        return True, []
    if normalized in AUDIT_FORBIDDEN_TASKS:
        return False, [
            f"task '{normalized}' is forbidden in audit-only reconciliation "
            "mode: switch to an explicit cleanup phase with operator approval "
            "and exact delete_branch capability proof before cleanup mutations"
        ]
    return True, []


def check_cleanup_execution_allowed() -> tuple[bool, list[str]]:
    """Fail closed when cleanup execution is attempted without authorization."""
    phase = current_phase()
    if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"):
        return True, []
    if phase is None:
        return False, [
            "cleanup execution requires an active reconciliation session; "
            "resolve a reconciliation audit task first"
        ]
    return False, [
        "cleanup execution forbidden in audit-only reconciliation mode; "
        "call gitea_authorize_reconciliation_cleanup_phase with operator "
        "approval, delete_branch capability proof, safety proof, and "
        "before/after snapshot"
    ]


def classify_cleanup_mutation(action: str) -> str:
    """Map a cleanup action to the required mutation ledger category (#419)."""
    normalized = (action or "").strip().lower()
    if "delete_remote" in normalized or normalized in {
        "delete_branch",
        "gitea_delete_branch",
    }:
        return "external-state"
    if "branch" in normalized and "delete" in normalized:
        return "git-ref"
    if "worktree" in normalized or "remove_local" in normalized:
        return "cleanup"
    return "cleanup"


def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]:
    """Validate audit/cleanup reconciliation reports (fail closed)."""
    text = report_text or ""
    reasons: list[str] = []

    cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text))
    claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text))

    if cleanup_occurred and claims_no_mutations:
        reasons.append(
            "report claims no mutations but documents cleanup mutations; "
            "audit-only reports must not perform cleanup and cleanup reports "
            "must not claim no mutations"
        )

    if cleanup_occurred:
        if not _CLEANUP_PHASE_AUTH_RE.search(text):
            reasons.append(
                "cleanup mutations reported without "
                "'Cleanup phase authorized: true'"
            )
        if not _DELETE_CAPABILITY_RE.search(text):
            reasons.append(
                "cleanup mutations reported without delete_branch capability "
                "proof"
            )
        if not _BEFORE_AFTER_RE.search(text):
            reasons.append(
                "cleanup mutations reported without before/after state snapshot"
            )
        if not _SAFETY_PROOF_RE.search(text):
            reasons.append(
                "cleanup mutations reported without branch/worktree safety proof"
            )

        if re.search(r"delete_remote|remote branch.*delet", text, re.I):
            if not _EXTERNAL_STATE_RE.search(text):
                reasons.append(
                    "remote branch deletion must be classified under "
                    "External-state mutations"
                )
        if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I):
            if not _GIT_REF_RE.search(text):
                reasons.append(
                    "local branch deletion must be classified under "
                    "Git ref mutations"
                )
        if re.search(r"worktree.*remov|remove_local_worktree", text, re.I):
            if not _CLEANUP_MUTATIONS_RE.search(text):
                reasons.append(
                    "worktree removal must be classified under Cleanup mutations"
                )

    if (
        RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text
        or "reconcile-landed-pr" in text.lower()
    ):
        if cleanup_occurred and "audit phase" in text.lower():
            if "cleanup phase" not in text.lower():
                reasons.append(
                    "report mixes audit phase with cleanup mutations without "
                    "documenting cleanup phase transition"
                )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "cleanup_occurred": cleanup_occurred,
        "claims_no_mutations": claims_no_mutations,
        "safe_next_action": (
            "proceed"
            if proven
            else "fix audit/cleanup report: separate audit from cleanup phase, "
            "classify mutations, and do not claim no mutations after cleanup"
        ),
    }


def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]:
    """Block shell commands that perform cleanup during audit phase."""
    phase = current_phase()
    if phase != PHASE_AUDIT:
        return True, []
    cmd = (command or "").strip()
    if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd):
        return False, [
            f"command forbidden in audit-only reconciliation mode: {cmd!r}; "
            "authorize cleanup phase before branch/worktree deletion or push"
        ]
    return True, []