cleveragents-core/tools/forgejo_audit.py

#!/usr/bin/env python3
"""Tier 0A — Forgejo behaviour audit (read-only).

Implements the audit described in
``.cursor/plans/improve_auto-agents_pipeline_a31d3945.plan.md`` (Tier 0A).

This script performs every read-only check that does not require creating
test PRs or branches against the production repo. Empirical pre-condition
verification (creating real test PRs to observe ``merge_when_checks_succeed``
behaviour, the ``head_commit_id`` 409 path, and umbrella auto-close) is
documented in the report as a manual follow-up — it intentionally cannot be
automated against a production repo.

Outputs
-------

A structured Markdown report on stdout (or to ``--output PATH``) suitable
for pasting into AGENTS.md under the "Merge invariant" section. Includes:

- Branch protection state for ``master`` (``dismiss_stale_approvals``,
  required status contexts, required approvals, push allow-list).
- Canonical merge-bot identity inferred from recent merge commits.
- ``merge_pr.ts`` configuration summary (Do strategy, head_commit_id usage,
  merge_when_checks_succeed semantics).
- Pre-condition verification checklist with status (``read-only-pass`` /
  ``empirical-todo`` / ``unknown``).

Usage
-----

::

    # Read-only audit, report to stdout
    python3 tools/forgejo_audit.py

    # Write the report to a specific path
    python3 tools/forgejo_audit.py --output ./audit-report.md

    # JSON output for programmatic consumption
    python3 tools/forgejo_audit.py --format json
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

# ─── Config ─────────────────────────────────────────────────────────────────

REPO_OWNER = os.environ.get("FORGEJO_OWNER", "cleveragents")
REPO_NAME = os.environ.get("FORGEJO_REPO", "cleveragents-core")
API_BASE = os.environ.get(
    "FORGEJO_API_BASE", "https://git.cleverthis.com/api/v1"
).rstrip("/")
DEFAULT_BRANCH = os.environ.get("FORGEJO_DEFAULT_BRANCH", "master")
REQUEST_TIMEOUT_SEC = 30


def load_token() -> str:
    """Pull GITEA_TOKEN from environment or .devcontainer/.env / .env."""
    token = os.environ.get("GITEA_TOKEN")
    if token:
        return token
    repo_root = Path(__file__).resolve().parent.parent
    for candidate in (repo_root / ".devcontainer/.env", repo_root / ".env"):
        if candidate.exists():
            for line in candidate.read_text().splitlines():
                m = re.match(r'^\s*GITEA_TOKEN\s*=\s*"?([^"#\s]+)"?', line)
                if m:
                    return m.group(1)
    raise RuntimeError(
        "GITEA_TOKEN not found in environment or .devcontainer/.env or .env"
    )


def api_get(path: str, token: str) -> Any:
    url = f"{API_BASE}{path}" if path.startswith("/") else path
    req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
    try:
        with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_SEC) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        # The branch protection endpoint returns 404 if no rule exists — caller handles.
        return {"_http_error": e.code, "_message": e.read().decode("utf-8", "replace")}


# ─── Audit subroutines ─────────────────────────────────────────────────────

def audit_branch_protection(token: str) -> dict[str, Any]:
    """Read ``branch_protections/master`` and summarise the load-bearing fields."""
    bp = api_get(
        f"/repos/{REPO_OWNER}/{REPO_NAME}/branch_protections/{DEFAULT_BRANCH}",
        token,
    )
    if isinstance(bp, dict) and "_http_error" in bp:
        return {
            "exists": False,
            "http_error": bp["_http_error"],
            "message": bp["_message"],
        }
    # Forgejo's branch protection schema uses snake_case — pull only the fields
    # the merge-driver and 2E will actually read.
    return {
        "exists": True,
        "branch_name": bp.get("branch_name"),
        "dismiss_stale_approvals": bp.get("dismiss_stale_approvals"),
        "required_approvals": bp.get("required_approvals"),
        "enable_status_check": bp.get("enable_status_check"),
        "status_check_contexts": bp.get("status_check_contexts") or [],
        "block_on_official_review_requests": bp.get(
            "block_on_official_review_requests"
        ),
        "block_on_outdated_branch": bp.get("block_on_outdated_branch"),
        "block_on_rejected_reviews": bp.get("block_on_rejected_reviews"),
        "enable_push": bp.get("enable_push"),
        "enable_push_whitelist": bp.get("enable_push_whitelist"),
        "push_whitelist_usernames": bp.get("push_whitelist_usernames") or [],
        "merge_whitelist_usernames": bp.get("merge_whitelist_usernames") or [],
        "approvals_whitelist_usernames": bp.get("approvals_whitelist_usernames")
        or [],
    }


def audit_merge_bot_identity(token: str, sample: int = 50) -> dict[str, Any]:
    """Walk recent master commits and tally distinct committers + their emails.

    Most-frequent committer on merge commits is the canonical merge-bot
    identity. Any direct push (the merge bot is *not* the committer) shows up
    as a separate identity.
    """
    commits = api_get(
        f"/repos/{REPO_OWNER}/{REPO_NAME}/commits"
        f"?sha={DEFAULT_BRANCH}&limit={sample}&page=1",
        token,
    )
    if not isinstance(commits, list):
        return {"error": "unable to read recent commits", "raw": commits}

    by_email: dict[str, dict[str, Any]] = {}
    for c in commits:
        commit = c.get("commit", {}) or {}
        committer = commit.get("committer", {}) or {}
        email = committer.get("email") or ""
        if not email:
            continue
        slot = by_email.setdefault(
            email,
            {
                "email": email,
                "name": committer.get("name", ""),
                "count": 0,
                "sample_subject": (commit.get("message") or "").splitlines()[0]
                if commit.get("message")
                else "",
            },
        )
        slot["count"] += 1

    ranked = sorted(by_email.values(), key=lambda d: -d["count"])
    canonical = ranked[0] if ranked else None
    return {
        "sample_size": sample,
        "actual_commits_seen": len(commits) if isinstance(commits, list) else 0,
        "canonical_merge_bot": canonical,
        "all_committers": ranked,
    }


def audit_merge_pr_ts() -> dict[str, Any]:
    """Statically inspect ``merge_pr.ts`` and extract the load-bearing fields."""
    path = (
        Path(__file__).resolve().parent.parent
        / ".opencode/skills/auto-agents-system/scripts/merge_pr.ts"
    )
    if not path.exists():
        return {"exists": False, "path": str(path)}
    src = path.read_text()
    findings: dict[str, Any] = {"exists": True, "path": str(path)}
    findings["uses_do_rebase"] = bool(re.search(r"Do:\s*'rebase'", src))
    findings["uses_do_merge"] = bool(re.search(r"Do:\s*'merge'", src))
    findings["uses_merge_when_checks_succeed_true"] = bool(
        re.search(r"merge_when_checks_succeed:\s*true", src)
    )
    findings["uses_merge_when_checks_succeed_false"] = bool(
        re.search(r"merge_when_checks_succeed:\s*false", src)
    )
    findings["passes_head_commit_id"] = bool(
        re.search(r"head_commit_id:\s*pr\.head\.sha", src)
    )
    findings["handles_409"] = "case 409" in src
    findings["handles_405"] = "case 405" in src
    findings["risk_summary"] = (
        "merge_when_checks_succeed=true is set unconditionally; "
        "if Forgejo JIT-rebases on auto-merge fire, the merged SHA is untested "
        "against current master. Verified empirically in 0A."
    )
    return findings


def audit_can_create_test_branch(token: str) -> dict[str, Any]:
    """Heuristic check that we have permission to create a test branch.

    Round 7 critique #8: the production repo must allow a long-running
    ``tests/integration-train`` test branch with branch protection mirroring
    master.
    """
    # We only check that the repo is writable by the token. The actual
    # branch-creation step is destructive and is left as a manual follow-up.
    repo = api_get(f"/repos/{REPO_OWNER}/{REPO_NAME}", token)
    if isinstance(repo, dict) and "_http_error" in repo:
        return {"can_check": False, "error": repo}
    permissions = (repo or {}).get("permissions", {}) or {}
    return {
        "can_check": True,
        "repo_full_name": (repo or {}).get("full_name"),
        "can_admin": permissions.get("admin", False),
        "can_push": permissions.get("push", False),
        "can_pull": permissions.get("pull", False),
        "manual_followup": (
            "Confirm manually that a tests/integration-train branch with "
            "branch protection mirroring master is acceptable for the 2D test "
            "plan. This requires admin access; record outcome in AGENTS.md."
        ),
    }


def audit_repo_settings(token: str) -> dict[str, Any]:
    """Surface repo-level toggles that affect the merge driver."""
    repo = api_get(f"/repos/{REPO_OWNER}/{REPO_NAME}", token)
    if isinstance(repo, dict) and "_http_error" in repo:
        return {"error": repo}
    return {
        "default_branch": (repo or {}).get("default_branch"),
        "default_delete_branch_after_merge": (repo or {}).get(
            "default_delete_branch_after_merge"
        ),
        "allow_rebase": (repo or {}).get("allow_rebase"),
        "allow_merge_commits": (repo or {}).get("allow_merge_commits"),
        "allow_squash_merge": (repo or {}).get("allow_squash_merge"),
        "allow_rebase_explicit": (repo or {}).get("allow_rebase_explicit"),
        "allow_rebase_update": (repo or {}).get("allow_rebase_update"),
    }


# ─── Pre-condition checklist ───────────────────────────────────────────────

PRECONDITIONS: list[dict[str, str]] = [
    {
        "id": "P1",
        "title": "Do='rebase' + merge_when_checks_succeed=false + head_commit_id "
        "merges the branch as-is (linear history, no server-side re-rebase).",
        "verification": "empirical",
        "how": "Create a fresh PR; rebase; capture HEAD SHA; call merge endpoint with "
        "those exact params; confirm the merged SHA on master equals the "
        "head_commit_id passed.",
    },
    {
        "id": "P2",
        "title": "head_commit_id optimistic lock returns HTTP 409 on stale value.",
        "verification": "empirical",
        "how": "Schedule a merge with a valid head_commit_id; push another commit to "
        "the head branch before the merge fires; confirm 409 returned.",
    },
    {
        "id": "P3",
        "title": "Originals auto-close as merged when an umbrella train branch merges "
        "(via --no-ff, head SHAs reachable in master).",
        "verification": "empirical",
        "how": "Verified during 2D rollout with a real two-PR train. If Forgejo does "
        "not auto-close, the merge driver explicitly PATCHes /pulls/<N> with "
        '{state: "closed"} for each constituent.',
    },
    {
        "id": "P4",
        "title": "nox -e lint/typecheck/unit_tests/integration_tests/e2e_tests/coverage_report "
        "is the complete set of required CI checks.",
        "verification": "read-only",
        "how": "Cross-check against branch_protections/master.status_check_contexts.",
    },
    {
        "id": "P5",
        "title": "DELETE /api/v1/repos/{owner}/{repo}/pulls/{index}/merge cancels a "
        "scheduled merge_when_checks_succeed.",
        "verification": "empirical",
        "how": "Schedule auto-merge on a fresh PR; DELETE the merge endpoint; confirm "
        "the schedule is gone (subsequent push does not trigger a merge).",
    },
]


# ─── Reporting ──────────────────────────────────────────────────────────────

def render_markdown_report(audit: dict[str, Any]) -> str:
    """Render the audit results as a Markdown report."""
    bp = audit["branch_protection"]
    bot = audit["merge_bot_identity"]
    mp = audit["merge_pr_ts"]
    test_branch = audit["test_branch_permissions"]
    repo = audit["repo_settings"]

    out: list[str] = []
    out.append("# Tier 0A — Forgejo Behaviour Audit Report")
    out.append("")
    out.append(f"_Generated: {audit['generated_at']}_")
    out.append(f"_Repo: {REPO_OWNER}/{REPO_NAME}_")
    out.append(f"_Tool: tools/forgejo_audit.py_")
    out.append("")

    out.append("## Branch protection (master)")
    out.append("")
    if not bp.get("exists"):
        out.append(
            f"- **No branch protection rule exists on `master`** "
            f"(HTTP {bp.get('http_error')}). Driver MUST refuse to start in "
            f"this state. Create a rule with required status checks before "
            f"shipping 0B."
        )
    else:
        out.append(f"- `dismiss_stale_approvals`: `{bp.get('dismiss_stale_approvals')}`")
        out.append(f"- `required_approvals`: `{bp.get('required_approvals')}`")
        out.append(f"- `enable_status_check`: `{bp.get('enable_status_check')}`")
        out.append(
            f"- `status_check_contexts`: "
            f"`{bp.get('status_check_contexts')}`"
        )
        out.append(
            f"- `block_on_official_review_requests`: "
            f"`{bp.get('block_on_official_review_requests')}`"
        )
        out.append(
            f"- `block_on_outdated_branch`: `{bp.get('block_on_outdated_branch')}`"
        )
        out.append(f"- `enable_push`: `{bp.get('enable_push')}`")
        out.append(
            f"- `enable_push_whitelist`: `{bp.get('enable_push_whitelist')}`"
        )
        out.append(
            f"- `push_whitelist_usernames`: "
            f"`{bp.get('push_whitelist_usernames')}`"
        )
        out.append(
            f"- `merge_whitelist_usernames`: "
            f"`{bp.get('merge_whitelist_usernames')}`"
        )
    out.append("")

    out.append("### 1A decision input (Jeff)")
    out.append("")
    flag = bp.get("dismiss_stale_approvals") if bp.get("exists") else None
    if flag is True:
        out.append(
            "- `dismiss_stale_approvals` is currently **ON**. The plan recommends "
            "flipping to `false` so rebases do not invalidate prior approvals. "
            "**Ask Jeff: flip the setting, or build the `reapprove_fallback`?** "
            "Record the decision below before 0B can ship."
        )
    elif flag is False:
        out.append(
            "- `dismiss_stale_approvals` is already **OFF**. No action needed; "
            "0B's throughput targets hold without the reapprove-if-clean-rebase "
            "fallback."
        )
    else:
        out.append(
            "- `dismiss_stale_approvals` could not be determined "
            "(no branch protection rule). Resolve before continuing."
        )
    out.append("- Jeff's decision (fill in): `pending | flip-to-false | keep-on`")
    out.append("- Date recorded: `____-__-__`")
    out.append("")

    out.append("## Canonical merge-bot identity")
    out.append("")
    canonical = bot.get("canonical_merge_bot")
    if canonical:
        out.append(
            f"- Most frequent committer in last {bot.get('sample_size')} master "
            f"commits: **`{canonical.get('name')}` <{canonical.get('email')}>** "
            f"({canonical.get('count')} commits)."
        )
        out.append(
            f"- Sample subject: `{canonical.get('sample_subject', '')[:80]}`"
        )
        out.append(
            "- This identity is the merge-bot. Any commit on master whose "
            "committer email differs is a direct push (Path B revert candidate, "
            "or unauthorized — 2E alerts)."
        )
        out.append("")
        out.append("- All distinct committers seen:")
        for c in bot.get("all_committers", []):
            out.append(
                f"  - `{c['name']}` <{c['email']}> — {c['count']} commits"
            )
    else:
        out.append(
            "- No committers found in recent master commits (unexpected). "
            "Re-run with a larger sample or investigate."
        )
    out.append("")

    out.append("## merge_pr.ts current behaviour")
    out.append("")
    if not mp.get("exists"):
        out.append("- **`merge_pr.ts` not found** at expected path.")
    else:
        out.append(f"- Path: `{mp.get('path')}`")
        out.append(f"- Uses `Do='rebase'`: `{mp.get('uses_do_rebase')}`")
        out.append(f"- Uses `Do='merge'`: `{mp.get('uses_do_merge')}`")
        out.append(
            f"- Sets `merge_when_checks_succeed=true`: "
            f"`{mp.get('uses_merge_when_checks_succeed_true')}`"
        )
        out.append(
            f"- Sets `merge_when_checks_succeed=false`: "
            f"`{mp.get('uses_merge_when_checks_succeed_false')}`"
        )
        out.append(
            f"- Passes `head_commit_id` (optimistic lock): "
            f"`{mp.get('passes_head_commit_id')}`"
        )
        out.append(f"- Handles HTTP 409: `{mp.get('handles_409')}`")
        out.append(f"- Handles HTTP 405: `{mp.get('handles_405')}`")
        out.append("")
        out.append(f"- **Risk:** {mp.get('risk_summary')}")
    out.append("")

    out.append("## Repo-level settings")
    out.append("")
    if "error" in repo:
        out.append(f"- Could not read repo settings: `{repo.get('error')}`")
    else:
        out.append(f"- `default_branch`: `{repo.get('default_branch')}`")
        out.append(
            f"- `default_delete_branch_after_merge`: "
            f"`{repo.get('default_delete_branch_after_merge')}` "
            f"(driver's auto/train/<ts> cleanup depends on this; if false, "
            f"0B.5 sweeps abandoned branches > 2 h old)"
        )
        out.append(f"- `allow_rebase`: `{repo.get('allow_rebase')}`")
        out.append(f"- `allow_merge_commits`: `{repo.get('allow_merge_commits')}`")
        out.append(f"- `allow_squash_merge`: `{repo.get('allow_squash_merge')}`")
    out.append("")

    out.append("## Test-branch permissions (round 7 critique #8)")
    out.append("")
    if test_branch.get("can_check"):
        out.append(
            f"- Token has admin: `{test_branch.get('can_admin')}`, "
            f"push: `{test_branch.get('can_push')}`, "
            f"pull: `{test_branch.get('can_pull')}` on "
            f"`{test_branch.get('repo_full_name')}`."
        )
        out.append(f"- {test_branch.get('manual_followup')}")
    else:
        out.append(
            f"- Could not check token permissions: `{test_branch.get('error')}`"
        )
    out.append("")

    out.append("## Pre-condition verification checklist")
    out.append("")
    out.append(
        "Pre-conditions 1, 2, and 4 from the plan are verified as part of 0A; "
        "P3 is verified during 2D rollout. The empirical checks (`empirical`) "
        "intentionally cannot be automated against a production repo — they "
        "are listed here so the operator can run them by hand and record the "
        "outcome."
    )
    out.append("")
    for pre in PRECONDITIONS:
        out.append(f"- **{pre['id']}** ({pre['verification']}): {pre['title']}")
        out.append(f"  - How: {pre['how']}")
        out.append(f"  - Outcome (fill in): `pending | confirmed | deviation`")
        out.append(f"  - Notes: ____")
    out.append("")

    out.append("## Auto-generated next steps")
    out.append("")
    out.append(
        "1. Run the empirical pre-condition checks (P1, P2, P5) against the "
        "real Forgejo instance with a throwaway PR. Record the outcomes above."
    )
    out.append(
        "2. Get Jeff's `dismiss_stale_approvals` decision and record it in 1A's "
        "section above."
    )
    out.append(
        "3. Confirm with Jeff whether a `tests/integration-train` branch with "
        "branch protection mirroring master is acceptable; record the outcome."
    )
    out.append(
        "4. Once all pre-conditions are recorded, paste this report into "
        "AGENTS.md under the 'Merge invariant' section, then proceed with "
        "`setup_labels` (rollout step 3)."
    )
    out.append("")
    return "\n".join(out)


# ─── Main ───────────────────────────────────────────────────────────────────

def run_audit(token: str) -> dict[str, Any]:
    return {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "repo": f"{REPO_OWNER}/{REPO_NAME}",
        "branch_protection": audit_branch_protection(token),
        "merge_bot_identity": audit_merge_bot_identity(token),
        "merge_pr_ts": audit_merge_pr_ts(),
        "test_branch_permissions": audit_can_create_test_branch(token),
        "repo_settings": audit_repo_settings(token),
        "preconditions": PRECONDITIONS,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Tier 0A — Forgejo behaviour audit (read-only)."
    )
    parser.add_argument(
        "--format",
        choices=("markdown", "json"),
        default="markdown",
        help="Output format (default: markdown).",
    )
    parser.add_argument(
        "--output",
        type=Path,
        help="Write report to PATH instead of stdout.",
    )
    args = parser.parse_args(argv)

    try:
        token = load_token()
    except RuntimeError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 2

    audit = run_audit(token)

    if args.format == "json":
        report = json.dumps(audit, indent=2, sort_keys=True)
    else:
        report = render_markdown_report(audit)

    if args.output:
        args.output.write_text(report + ("\n" if not report.endswith("\n") else ""))
        print(f"Wrote audit report to {args.output}", file=sys.stderr)
    else:
        sys.stdout.write(report)
        if not report.endswith("\n"):
            sys.stdout.write("\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())