From 6d3dd646cecd75747b9c7d2696311e2a94d24ec4 Mon Sep 17 00:00:00 2001 From: TaprootFreakAI <315477232+TaprootFreakAI@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:03:29 +0200 Subject: [PATCH 1/2] Approve allowlisted initial fork CI runs after fresh A38 validation. --- .github/actions/a38-guard/action.yml | 2 +- docs/a38-guard.md | 24 ++++ docs/pull-request-lifecycle.md | 2 +- src/agent_cli/a38_guard.py | 12 +- src/agent_cli/pr_guard_config.py | 23 +++- src/agent_cli/workflow_approval.py | 197 +++++++++++++++++++++++++++ 6 files changed, 255 insertions(+), 5 deletions(-) create mode 100644 src/agent_cli/workflow_approval.py diff --git a/.github/actions/a38-guard/action.yml b/.github/actions/a38-guard/action.yml index 6f1b6df..c8c0d60 100644 --- a/.github/actions/a38-guard/action.yml +++ b/.github/actions/a38-guard/action.yml @@ -5,7 +5,7 @@ description: > inputs: token: - description: GitHub token (contents read, issues write, pull-requests write, statuses write) + description: GitHub token (contents read, issues write, pull-requests write, statuses write; actions write when repository workflow approval is enabled) required: true repository: description: owner/name override (workflow_dispatch / standalone) diff --git a/docs/a38-guard.md b/docs/a38-guard.md index 29340a0..1a30370 100644 --- a/docs/a38-guard.md +++ b/docs/a38-guard.md @@ -102,6 +102,30 @@ Publish the migration proposal draft first per the [pull request lifecycle](pull The bot identifies the active policy revision in its comment. Download `.github/a38.json` from that exact revision before generating the report. For ordinary PRs, this is the base; for explicitly approved migrations, it is the head. +## Optional fork workflow approval + +A repository may opt in to bot-owned CI authorization in its trusted default-branch `.github/pr-guard.json`: + +```json +"workflow_approval": { + "enabled": true, + "workflows": [".github/workflows/ci.yml", ".github/workflows/security.yml"] +} +``` + +This is an optional top-level object alongside `schema` and `a38`. Omission disables the feature. Both fields are required when present; `enabled` is a boolean, and `workflows` is a duplicate-free list of at most 64 exact workflow YAML paths (nonempty when enabled). Globs and unknown fields fail closed. The proposed PR-head config cannot activate approval. Workflow selection and the token's `actions: write` permission belong to the adopting repository, not to a global runner policy. + +The bot approves only an **initial** `pull_request` run waiting in `completed` / `action_required`, with `run_attempt: 1`, for an allowlisted workflow on the exact current head, fork repository and branch. A fresh A38 `pass` under `enforce` is required. Failed or incomplete evidence, observe mode, excluded targets, closed PRs and same-repository PRs cannot trigger approval. Existing migration authorization remains required for policy/workflow/config changes. + +For each workflow, the newest matching run across **all** states wins. A queued, successful, failed or rerun attempt supersedes an older blocked run. The bot never calls a rerun, dispatch, cancel, merge, review-approval or environment-approval endpoint. Approval authorizes execution; it is not a test result or a Ready verdict. + +The run's PR association must match the current PR/head/base. For private forks whose API association array is empty, the fork branch must identify exactly one open PR, its head must include the current base, and the run must not predate the PR or a later recorded target/lifecycle change. Ambiguous association, incomplete pagination, API errors or denied permissions fail closed. Head/base, trusted config, latest author report and maintainer authorization are refreshed before every POST. GitHub provides no atomic compare-and-approve operation; these checks minimize, but cannot eliminate, a change racing the final API call. + +Enable `actions: write` in the trusted guard workflow (or equivalent Actions write access on a dedicated App token). The guard uses GitHub's [approve-workflow-run endpoint](https://docs.github.com/en/rest/actions/workflow-runs#approve-a-workflow-run-for-a-fork-pull-request), accepts only its documented `201` success, and never retries that POST. Insufficient permissions remain an explicit failure; changing the repository's fork protection setting is not a fallback. `--dry-run` previews candidates without any writes. Assessment JSON includes `workflow_approvals`; completed authorization also records `workflow:approve:` in `writes`. + +The trusted default-branch workflow and config must be installed before this feature is active. A head-only proposal does not grant itself permissions or authorize its own runs. Scheduled reconciliation catches runs created after the author report event. After authorization, inspect the actual independent GitHub checks through completion, including blocked `action_required` workflow runs that may be absent from the PR check rollup. + + ## Author report The author runs the full local job list from a clean checkout of the exact head. Keep the policy copy, report and logs outside the checkout: diff --git a/docs/pull-request-lifecycle.md b/docs/pull-request-lifecycle.md index f2adf78..cec0665 100644 --- a/docs/pull-request-lifecycle.md +++ b/docs/pull-request-lifecycle.md @@ -57,7 +57,7 @@ Stay draft until Ready for review is earned on the **exact clean signed final he 1. Full applicable tests for that head (repository rules and, when adopted, the complete A38 policy run and local verification). 2. For A38 adopters: author report publication, current-base (or exact approved head) policy checks, and the live join required by [a38.md](a38.md) and [a38-guard.md](a38-guard.md). -3. Independently required GitHub checks on this head (`skipped` and `cancelled` are not green unless the workflow documents that skip). +3. Independently required GitHub checks on this head (`skipped` and `cancelled` are not green unless the workflow documents that skip). Inspect both the PR check rollup and the current-head workflow-run inventory: `action_required` runs may be absent from the check rollup. A38 equivalence covers only the jobs in its active policy; it does not replace independently required security or other GitHub-only checks. Bot authorization to start a run is not a successful run. 4. Independent required reviews and approvals per the attached skills and the target repository's written rules. 5. Then the Ready comment / leave-draft steps those rules define (`isDraft=false`). diff --git a/src/agent_cli/a38_guard.py b/src/agent_cli/a38_guard.py index 043c1e1..d301e61 100644 --- a/src/agent_cli/a38_guard.py +++ b/src/agent_cli/a38_guard.py @@ -152,6 +152,8 @@ class Assessment: skip_publish: bool = False dry_run: bool = False writes: list[str] = field(default_factory=list) + workflow_approval_enabled: bool = False + workflow_approvals: list[dict[str, Any]] = field(default_factory=list) def to_json(self) -> dict[str, Any]: trusted = self.trusted_default_branch or self.default_branch @@ -184,6 +186,7 @@ def to_json(self) -> dict[str, Any]: "skip_publish": self.skip_publish, "dry_run": self.dry_run, "writes": list(self.writes), + "workflow_approvals": list(self.workflow_approvals), "comment_body": self.comment_body, } @@ -1304,6 +1307,7 @@ def _attach_trusted_config(assessment: Assessment, trusted: TrustedGuardConfig) assessment.config_fingerprint = trusted.fingerprint assessment.scope_decision = trusted.decision assessment.scope_reason = trusted.reason + assessment.workflow_approval_enabled = bool((trusted.config or {}).get("workflow_approval", {}).get("enabled", False)) def _out_of_scope_assessment( @@ -1666,8 +1670,14 @@ def reconcile_pull( assessment.writes.append("skipped:closed") elif dry_run: assessment.writes.append("dry-run") + if not assessment.closed: + from .workflow_approval import approve_workflow_runs + assessment.workflow_approvals = approve_workflow_runs(api, assessment, dry_run=True) return assessment - return publish_assessment(api, assessment) + published = publish_assessment(api, assessment) + from .workflow_approval import approve_workflow_runs + published.workflow_approvals = approve_workflow_runs(api, published) + return published except GuardError as exc: if "changed before publish" in str(exc): last_err = exc diff --git a/src/agent_cli/pr_guard_config.py b/src/agent_cli/pr_guard_config.py index 8383364..d589794 100644 --- a/src/agent_cli/pr_guard_config.py +++ b/src/agent_cli/pr_guard_config.py @@ -16,6 +16,8 @@ SCHEMA_ID = "pr-guard/v1" CONFIG_PATH = ".github/pr-guard.json" TOP_KEYS = frozenset({"schema", "a38"}) +WORKFLOW_APPROVAL_KEYS = frozenset({"enabled", "workflows"}) +WORKFLOW_PATH_RE = re.compile(r"^\.github/workflows/[A-Za-z0-9_-][A-Za-z0-9_.-]*\.(?:yml|yaml)$") A38_KEYS = frozenset({"enforce", "exclude", "default"}) SCOPE_MODES = frozenset({"enforce", "exclude"}) MAX_BRANCH_LIST = 256 @@ -108,7 +110,7 @@ def load_pr_guard_config(text: str) -> dict[str, Any]: payload = _loads_json(text) if not isinstance(payload, dict): raise PrGuardConfigError("pr-guard config must be a JSON object") - _require_keys(payload, TOP_KEYS, "pr-guard config") + _require_keys({k: v for k, v in payload.items() if k != "workflow_approval"}, TOP_KEYS, "pr-guard config") schema = payload["schema"] if not isinstance(schema, str) or schema != SCHEMA_ID: raise PrGuardConfigError(f"schema must be {SCHEMA_ID}") @@ -127,7 +129,7 @@ def load_pr_guard_config(text: str) -> dict[str, Any]: "a38.enforce and a38.exclude overlap: " + ", ".join(repr(name) for name in overlap) ) - return { + normalized = { "schema": SCHEMA_ID, "a38": { "enforce": enforce, @@ -135,6 +137,23 @@ def load_pr_guard_config(text: str) -> dict[str, Any]: "default": default, }, } + if "workflow_approval" in payload: + approval = payload["workflow_approval"] + if not isinstance(approval, dict): + raise PrGuardConfigError("workflow_approval must be an object") + _require_keys(approval, WORKFLOW_APPROVAL_KEYS, "workflow_approval") + enabled = approval["enabled"] + paths = approval["workflows"] + if type(enabled) is not bool: + raise PrGuardConfigError("workflow_approval.enabled must be boolean") + if not isinstance(paths, list) or len(paths) > 64: + raise PrGuardConfigError("workflow_approval.workflows must be an array of at most 64 paths") + if any(not isinstance(p, str) or len(p) > 255 or WORKFLOW_PATH_RE.fullmatch(p) is None for p in paths): + raise PrGuardConfigError("workflow_approval.workflows must contain exact workflow YAML paths") + if len(set(paths)) != len(paths) or (enabled and not paths): + raise PrGuardConfigError("workflow approval needs a nonempty, duplicate-free allowlist when enabled") + normalized["workflow_approval"] = {"enabled": enabled, "workflows": list(paths)} + return normalized def evaluate_a38_scope( diff --git a/src/agent_cli/workflow_approval.py b/src/agent_cli/workflow_approval.py new file mode 100644 index 0000000..86a75da --- /dev/null +++ b/src/agent_cli/workflow_approval.py @@ -0,0 +1,197 @@ +"""Opt-in approval of initial fork CI runs after a live A38 pass. + +This module authorizes execution, never retries a test or approves a review. +Configuration and workflow allowlists come only from the trusted default ref. +""" +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Mapping +from urllib.parse import urlencode + +MAX_RUNS = 1000 + + +def _timestamp(value: Any) -> datetime: + from .a38_guard import GuardError + if not isinstance(value, str): + raise GuardError("workflow approval timestamp missing") + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise GuardError("workflow approval timestamp invalid") from exc + if result.tzinfo is None: + raise GuardError("workflow approval timestamp must have a timezone") + return result.astimezone(timezone.utc) + + +def _runs(api: Any, repo: str, head: str) -> list[Mapping[str, Any]]: + from .a38_guard import GuardError + items: list[Mapping[str, Any]] = [] + total = None + for page in range(1, 11): + query = urlencode(dict(event="pull_request", head_sha=head, per_page=100, page=page)) + data = api.get_json(f"/repos/{repo}/actions/runs?{query}") + if not isinstance(data, Mapping) or type(data.get("total_count")) is not int: + raise GuardError("workflow run inventory missing total_count") + count = data["total_count"] + if count < 0 or count >= MAX_RUNS or (total is not None and count != total): + raise GuardError("workflow run inventory changed or exceeds bound") + total = count + batch = data.get("workflow_runs") + if not isinstance(batch, list) or any(not isinstance(r, Mapping) for r in batch): + raise GuardError("workflow run inventory invalid") + items.extend(batch) + ids = [r.get("id") for r in items] + if any(type(i) is not int or i <= 0 for i in ids) or len(set(ids)) != len(ids): + raise GuardError("workflow run inventory has invalid or duplicate IDs") + if len(items) == total: + return items + if len(items) > total or not batch: + raise GuardError("workflow run inventory incomplete") + raise GuardError("workflow run pagination bound exceeded") + + +def _matches(run: Mapping[str, Any], pull: Mapping[str, Any], paths: list[str]) -> bool: + head, base = pull["head"], pull["base"] + return ( + run.get("event") == "pull_request" + and run.get("head_sha") == head["sha"] + and run.get("head_branch") == head["ref"] + and isinstance(run.get("repository"), Mapping) + and run["repository"].get("full_name") == base["repo"]["full_name"] + and isinstance(run.get("head_repository"), Mapping) + and run["head_repository"].get("full_name") == head["repo"]["full_name"] + and run.get("path") in paths + ) + + +def _latest(runs: list[Mapping[str, Any]], pull: Mapping[str, Any], paths: list[str]) -> dict[str, Mapping[str, Any]]: + latest: dict[str, Mapping[str, Any]] = {} + for run in runs: + if not _matches(run, pull, paths): + continue + path = run["path"] + key = (_timestamp(run.get("created_at")), run["id"]) + previous = latest.get(path) + if previous is None or key > (_timestamp(previous.get("created_at")), previous["id"]): + latest[path] = run + return latest + + +def _pending(run: Mapping[str, Any]) -> bool: + # A later attempt is never auto-approved: this feature does not retry tests. + return (run.get("status") == "completed" and run.get("conclusion") == "action_required" + and type(run.get("run_attempt")) is int and run["run_attempt"] == 1) + + +def _belongs_to_pull(api: Any, run: Mapping[str, Any], pull: Mapping[str, Any]) -> None: + from .a38_guard import GuardError + repo = pull["base"]["repo"]["full_name"] + head = pull["head"] + links = run.get("pull_requests") + if not isinstance(links, list): + raise GuardError("workflow run pull request associations missing") + if _timestamp(run.get("created_at")) < _timestamp(pull.get("created_at")): + raise GuardError("workflow run predates this pull request") + if links: + if len(links) != 1 or not isinstance(links[0], Mapping): + raise GuardError("workflow run pull request association is ambiguous") + linked = links[0] + if (linked.get("number") != pull["number"] + or linked.get("head", {}).get("sha") != head["sha"] + or linked.get("base", {}).get("sha") != pull["base"]["sha"]): + raise GuardError("workflow run targets another pull request head or base") + return + # GitHub returns an empty associations array for private forks. Prove the + # fork branch identifies exactly this open PR, and that its head contains + # the current base (an older-base merge cannot change the measured tree). + pulls = api.paginate(f"/repos/{repo}/pulls?state=open") + matches = [p for p in pulls if isinstance(p, Mapping) and p.get("state") == "open" + and p.get("head", {}).get("ref") == head["ref"] + and p.get("head", {}).get("repo", {}).get("full_name") == head["repo"]["full_name"]] + if len(matches) != 1 or matches[0].get("number") != pull["number"]: + raise GuardError("fork workflow run cannot be uniquely associated with this pull request") + match = matches[0] + if match.get("head", {}).get("sha") != head["sha"] or match.get("base", {}).get("sha") != pull["base"]["sha"]: + raise GuardError("fork pull request changed during workflow approval") + comparison = api.get_json(f"/repos/{repo}/compare/{pull['base']['sha']}...{head['sha']}") + if not isinstance(comparison, Mapping) or comparison.get("status") not in {"ahead", "identical"}: + raise GuardError("fork workflow approval requires the current base to be included in the head") + events = api.paginate(f"/repos/{repo}/issues/{pull['number']}/events") + for event in events: + if not isinstance(event, Mapping): + raise GuardError("pull request event inventory invalid") + if event.get("event") in {"base_ref_changed", "base_ref_force_pushed", "reopened"}: + if _timestamp(event.get("created_at")) > _timestamp(run.get("created_at")): + raise GuardError("workflow run predates a pull request target or lifecycle change") + + +def approve_workflow_runs(api: Any, assessment: Any, *, dry_run: bool = False) -> list[dict[str, Any]]: + from .a38_guard import ( + GuardError, assess_pull, fetch_pull, resolve_trusted_guard_config, + _report_fingerprint, collect_comments, pick_latest_author_report, + migration_approval, + ) + if (not assessment.workflow_approval_enabled or assessment.closed or not assessment.ok + or assessment.status != "pass" or assessment.mode != "enforce"): + return [] + snap = fetch_pull(api, assessment.repo, assessment.pr) + trusted = resolve_trusted_guard_config(api, snap) + config = (trusted.config or {}).get("workflow_approval") + if not config or not config["enabled"] or snap.head_repo == snap.repo: + return [] + paths = config["workflows"] + + def fresh_pull() -> Mapping[str, Any]: + fresh = assess_pull(api, assessment.repo, assessment.pr, dry_run=True) + fields = ("head_sha", "base_sha", "base_ref", "head_repo", "config_revision", "config_fingerprint", + "report_fingerprint", "approval_fingerprint", "policy_sha") + if (not fresh.ok or fresh.closed or fresh.status != "pass" or fresh.mode != "enforce" + or any(getattr(fresh, f) != getattr(assessment, f) for f in fields)): + raise GuardError("A38 evidence or trusted configuration changed before workflow approval") + pull = api.get_json(f"/repos/{assessment.repo}/pulls/{assessment.pr}") + if (not isinstance(pull, Mapping) or pull.get("state") != "open" + or pull.get("head", {}).get("sha") != assessment.head_sha + or pull.get("base", {}).get("sha") != assessment.base_sha + or pull.get("base", {}).get("ref") != assessment.base_ref + or pull.get("head", {}).get("repo", {}).get("full_name") != assessment.head_repo + or not isinstance(pull.get("head", {}).get("ref"), str)): + raise GuardError("pull request changed before workflow approval") + return pull + + pull = fresh_pull() + candidates = _latest(_runs(api, assessment.repo, assessment.head_sha), pull, paths) + result = [] + for path, candidate in sorted(candidates.items()): + if not _pending(candidate): + continue + pull = fresh_pull() + # Re-list all states: a newer queued/passed/failed run supersedes an old + # blocked run, so never wake that old run and repeat a completed suite. + latest = _latest(_runs(api, assessment.repo, assessment.head_sha), pull, paths).get(path) + if latest is None or latest["id"] != candidate["id"] or not _pending(latest): + continue + run = api.get_json(f"/repos/{assessment.repo}/actions/runs/{candidate['id']}") + if not isinstance(run, Mapping) or run.get("id") != candidate["id"] or not _matches(run, pull, paths): + raise GuardError("workflow run identity changed before approval") + if not _pending(run): + continue + _belongs_to_pull(api, run, pull) + # Last live checks immediately before the irreversible authorization. + final = fetch_pull(api, assessment.repo, assessment.pr) + config_now = resolve_trusted_guard_config(api, final) + comments = collect_comments(api, assessment.repo, assessment.pr) + if (final != snap or config_now.config_revision != trusted.config_revision + or config_now.fingerprint != trusted.fingerprint + or _report_fingerprint(pick_latest_author_report(comments, final.author_id)) != assessment.report_fingerprint + or migration_approval(api, final) != assessment.approval_fingerprint): + raise GuardError("pull or author evidence changed before workflow approval") + if not dry_run: + status, _, _ = api.request("POST", f"/repos/{assessment.repo}/actions/runs/{run['id']}/approve", retry=False) + if status != 201: + raise GuardError(f"workflow approval HTTP {status}; Actions write permission is required") + assessment.writes.append(f"workflow:approve:{run['id']}") + result.append({"run_id": run["id"], "workflow": path, "head": assessment.head_sha, + "status": "planned" if dry_run else "approved"}) + return result From 3b95a06a34532959016c35272e02b36118ac9cb7 Mon Sep 17 00:00:00 2001 From: TaprootFreakAI <315477232+TaprootFreakAI@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:32:28 +0200 Subject: [PATCH 2/2] Reconcile PR readiness against complete CI and mergeability. --- .github/actions/a38-guard/action.yml | 2 +- docs/a38-guard.md | 85 ++- docs/pull-request-lifecycle.md | 8 + src/agent_cli/a38_guard.py | 9 + src/agent_cli/pr_guard_config.py | 50 +- src/agent_cli/pr_lifecycle.py | 269 ++++++++++ src/agent_cli/workflow_approval.py | 38 +- tests/test_pr_lifecycle.py | 248 +++++++++ tests/test_workflow_approval.py | 754 +++++++++++++++++++++++++++ tests/test_workflow_approval_core.py | 212 ++++++++ 10 files changed, 1660 insertions(+), 15 deletions(-) create mode 100644 src/agent_cli/pr_lifecycle.py create mode 100644 tests/test_pr_lifecycle.py create mode 100644 tests/test_workflow_approval.py create mode 100644 tests/test_workflow_approval_core.py diff --git a/.github/actions/a38-guard/action.yml b/.github/actions/a38-guard/action.yml index c8c0d60..6ef4fae 100644 --- a/.github/actions/a38-guard/action.yml +++ b/.github/actions/a38-guard/action.yml @@ -5,7 +5,7 @@ description: > inputs: token: - description: GitHub token (contents read, issues write, pull-requests write, statuses write; actions write when repository workflow approval is enabled) + description: GitHub token (contents read, issues write, pull-requests write, statuses write; actions/checks read for lifecycle, actions write for workflow approval) required: true repository: description: owner/name override (workflow_dispatch / standalone) diff --git a/docs/a38-guard.md b/docs/a38-guard.md index 1a30370..0259b6d 100644 --- a/docs/a38-guard.md +++ b/docs/a38-guard.md @@ -126,6 +126,89 @@ Enable `actions: write` in the trusted guard workflow (or equivalent Actions wri The trusted default-branch workflow and config must be installed before this feature is active. A head-only proposal does not grant itself permissions or authorize its own runs. Scheduled reconciliation catches runs created after the author report event. After authorization, inspect the actual independent GitHub checks through completion, including blocked `action_required` workflow runs that may be absent from the PR check rollup. +## Optional continuous readiness + +The same trusted repository configuration can enable CI and conflict monitoring: + +```json +"lifecycle": { + "enabled": true, + "auto_ready": true, + "required_workflows": [".github/workflows/ci.yml"], + "ignored_workflows": [".github/workflows/pr-guard.yml"], + "required_checks": {".github/workflows/ci.yml": ["Test"]}, + "conditional_workflows": [ + { + "workflow": ".github/workflows/security.yml", + "base_branches": ["release"], + "labels_any": ["full-ci"] + } + ] +} +``` + +All fields except `conditional_workflows` and `required_checks` are required when `lifecycle` is +present. Omission disables lifecycle writes. `auto_ready` requires enabled +workflow approval and a nonempty required-workflow list. Workflow paths are +exact, bounded and unique across these lists. A conditional workflow is required +when its target branch **or** any listed PR label matches; branches and labels +are exact, case-sensitive strings. Its condition must not be empty. This is +repository configuration, including which CI is expected for release PRs. +`required_checks` maps required or conditional workflow paths to exact job check +names (including expanded matrix names). When that workflow is required, each +listed check must actually finish successfully in its latest check suite. +Use this for workflows whose setup job can succeed while the test jobs skip; +an overall workflow success must not hide a missing or skipped required test. + +For **every open Ready PR**, confirmed merge conflicts or CI that is missing, +queued, waiting, running, blocked, cancelled or failed cause a Draft transition. +This applies even when the target is excluded from A38. Missing required +workflows are not an empty green result. Only completed, successful required +workflows satisfy CI. Optional workflows that intentionally skip are not counted +as successful required tests. Pending or failed independent check runs and commit +statuses also block Ready. The newest workflow run supersedes historical results; +both workflow inventories and checks are inspected, including approval-blocked +runs absent from GitHub's rollup. + +Ignore only repository control workflows that are not product CI, particularly +the guard itself: otherwise its in-progress check would always prevent Ready. +Required, conditional and ignored workflow paths cannot overlap. Conditional +jobs skipped inside a successful workflow do not make that workflow fail. + +Automatic Ready requires all required CI green, GitHub `mergeable: true`, a +fresh enforced A38 pass, and an authenticated bot-owned record of CI this bot +actually authorized on the current head/base. The recorded runs must still be +the latest matching runs. A manually started green suite alone does not authorize +promotion. An unknown merge status neither invents a conflict nor permits Ready. +This feature changes readiness only; it creates no review approvals and never +bypasses review requirements, branch protection or human merge. + +Each transition gets an EN/DE comment with the concrete reasons in collapsed +details. A durable intent is written before the mutation and updated after +success; the next scan repairs the comment if that update was interrupted. +Unchanged readiness creates no duplicate comment. Authorization records use +`PR-GUARD:CI-AUTH:v1`; transition records use `PR-GUARD:LIFECYCLE:v1`. Only the +authenticated bot's numeric user ID can supply these records. Dry run performs +no writes, including audit comments. + +The adopting workflow owns runner routing, `actions` and `checks` read access, +`pull-requests`/`issues`/`statuses` write access, and `actions: write` for initial +workflow approval. Serialize **all** event and scheduled invocations with one +repository-wide concurrency group and `cancel-in-progress: false`. Run trusted +`--all-open` reconciliation on a repository-configured schedule (for example every +five minutes). GitHub may delay scheduled execution; this is not a real-time SLA. +Privileged runs must never check out PR code. Bot readiness must not be wired to +automatic test retries; with `GITHUB_TOKEN`, the bot's own readiness/comment +events do not trigger new workflow runs. Other token integrations must ensure +their Ready handlers do not repeat already-requested CI. + +Head/base, configuration, evidence and CI are refreshed before promotion. +GitHub does not offer an atomic CI-and-readiness transaction; subsequent changes +are corrected by the next reconciliation. A head/base change returned by the +Ready mutation immediately restores Draft. API denial fails explicitly and is +not a successful transition. Neither this code nor its configuration is active +until the trusted installation is deployed. + ## Author report The author runs the full local job list from a clean checkout of the exact head. Keep the policy copy, report and logs outside the checkout: @@ -163,7 +246,7 @@ Supported events: - `pull_request_target`: opened, reopened, synchronize, edited, ready_for_review. - `issue_comment`: created, edited, deleted, for PRs only. -- Scheduled all-open reconciliation every 30 minutes on the trusted default branch. +- Scheduled all-open reconciliation on the trusted default branch; cadence is repository configuration. - `workflow_dispatch`: an explicit repository and PR number. Issue-only events and the bot's own comments are ignored. The installed workflow deliberately has no `pull_request_review` trigger because that event loads workflow code from PR context. After approving or dismissing a policy review, post a normal PR comment such as `A38 recheck` for immediate reassessment, or dispatch the default-branch workflow. Scheduled reconciliation catches other review/base changes. The CLI can consume submitted/edited/dismissed review events supplied by an external trusted event handler, but never grant elevated credentials to PR-context workflow code. Never check out the PR head in a privileged bot job. diff --git a/docs/pull-request-lifecycle.md b/docs/pull-request-lifecycle.md index cec0665..a064272 100644 --- a/docs/pull-request-lifecycle.md +++ b/docs/pull-request-lifecycle.md @@ -49,6 +49,14 @@ Hosted CI and other applicable checks may fail. There is no promise that CI neve Pending checks must be labeled **pending**. Do not fabricate a pass. +Repositories can enable the [guard's continuous readiness reconciliation](a38-guard.md#optional-continuous-readiness). +An open Ready PR returns to Draft with an explanatory comment when required CI +is missing, queued, running, blocked or failed, or GitHub confirms merge conflicts. +After the CI authorized by the bot succeeds, it can restore Ready only with +current A38 evidence and confirmed mergeability. Required workflows, conditional +CI scope, control-workflow exclusions and the polling schedule belong to the +adopting repository. This does not rerun tests, submit review approvals or merge. + If the repository's guard integration is known to be defective, require a **verified** rollout of the fixed integration before Ready for review. Do not instruct merging through red statuses. ## Ready for review diff --git a/src/agent_cli/a38_guard.py b/src/agent_cli/a38_guard.py index d301e61..74e5483 100644 --- a/src/agent_cli/a38_guard.py +++ b/src/agent_cli/a38_guard.py @@ -154,6 +154,8 @@ class Assessment: writes: list[str] = field(default_factory=list) workflow_approval_enabled: bool = False workflow_approvals: list[dict[str, Any]] = field(default_factory=list) + lifecycle_enabled: bool = False + lifecycle: dict[str, Any] = field(default_factory=dict) def to_json(self) -> dict[str, Any]: trusted = self.trusted_default_branch or self.default_branch @@ -187,6 +189,7 @@ def to_json(self) -> dict[str, Any]: "dry_run": self.dry_run, "writes": list(self.writes), "workflow_approvals": list(self.workflow_approvals), + "lifecycle": dict(self.lifecycle), "comment_body": self.comment_body, } @@ -1308,6 +1311,7 @@ def _attach_trusted_config(assessment: Assessment, trusted: TrustedGuardConfig) assessment.scope_decision = trusted.decision assessment.scope_reason = trusted.reason assessment.workflow_approval_enabled = bool((trusted.config or {}).get("workflow_approval", {}).get("enabled", False)) + assessment.lifecycle_enabled = bool((trusted.config or {}).get("lifecycle", {}).get("enabled", False)) def _out_of_scope_assessment( @@ -1341,6 +1345,7 @@ def _out_of_scope_assessment( skip_publish=False, dry_run=dry_run, ) + _attach_trusted_config(assessment, trusted) return assessment @@ -1671,10 +1676,14 @@ def reconcile_pull( elif dry_run: assessment.writes.append("dry-run") if not assessment.closed: + from .pr_lifecycle import reconcile_lifecycle + assessment.lifecycle = reconcile_lifecycle(api, assessment, dry_run=True) from .workflow_approval import approve_workflow_runs assessment.workflow_approvals = approve_workflow_runs(api, assessment, dry_run=True) return assessment published = publish_assessment(api, assessment) + from .pr_lifecycle import reconcile_lifecycle + published.lifecycle = reconcile_lifecycle(api, published) from .workflow_approval import approve_workflow_runs published.workflow_approvals = approve_workflow_runs(api, published) return published diff --git a/src/agent_cli/pr_guard_config.py b/src/agent_cli/pr_guard_config.py index d589794..4c1b338 100644 --- a/src/agent_cli/pr_guard_config.py +++ b/src/agent_cli/pr_guard_config.py @@ -110,7 +110,7 @@ def load_pr_guard_config(text: str) -> dict[str, Any]: payload = _loads_json(text) if not isinstance(payload, dict): raise PrGuardConfigError("pr-guard config must be a JSON object") - _require_keys({k: v for k, v in payload.items() if k != "workflow_approval"}, TOP_KEYS, "pr-guard config") + _require_keys({k: v for k, v in payload.items() if k not in {"workflow_approval", "lifecycle"}}, TOP_KEYS, "pr-guard config") schema = payload["schema"] if not isinstance(schema, str) or schema != SCHEMA_ID: raise PrGuardConfigError(f"schema must be {SCHEMA_ID}") @@ -153,6 +153,54 @@ def load_pr_guard_config(text: str) -> dict[str, Any]: if len(set(paths)) != len(paths) or (enabled and not paths): raise PrGuardConfigError("workflow approval needs a nonempty, duplicate-free allowlist when enabled") normalized["workflow_approval"] = {"enabled": enabled, "workflows": list(paths)} + if "lifecycle" in payload: + lifecycle = payload["lifecycle"] + if not isinstance(lifecycle, dict): + raise PrGuardConfigError("lifecycle must be an object") + _require_keys({k: v for k, v in lifecycle.items() if k not in {"conditional_workflows", "required_checks"}}, + frozenset({"enabled", "auto_ready", "required_workflows", "ignored_workflows"}), "lifecycle") + if type(lifecycle["enabled"]) is not bool or type(lifecycle["auto_ready"]) is not bool: + raise PrGuardConfigError("lifecycle enabled/auto_ready must be boolean") + for key in ("required_workflows", "ignored_workflows"): + paths = lifecycle[key] + if (not isinstance(paths, list) or len(paths) > 64 + or any(not isinstance(p, str) or len(p) > 255 or WORKFLOW_PATH_RE.fullmatch(p) is None for p in paths) + or len(set(paths)) != len(paths)): + raise PrGuardConfigError(f"lifecycle.{key} must be a bounded list of unique workflow YAML paths") + if set(lifecycle["required_workflows"]) & set(lifecycle["ignored_workflows"]): + raise PrGuardConfigError("required and ignored lifecycle workflows overlap") + conditions = lifecycle.get("conditional_workflows", []) + if not isinstance(conditions, list) or len(conditions) > 64: + raise PrGuardConfigError("conditional_workflows must be a bounded list") + condition_paths = set(lifecycle["required_workflows"] + lifecycle["ignored_workflows"]) + for condition in conditions: + if not isinstance(condition, dict): + raise PrGuardConfigError("conditional workflow must be an object") + _require_keys(condition, frozenset({"workflow", "base_branches", "labels_any"}), "conditional workflow") + path = condition["workflow"] + if not isinstance(path, str) or len(path) > 255 or WORKFLOW_PATH_RE.fullmatch(path) is None or path in condition_paths: + raise PrGuardConfigError("conditional workflow path invalid or duplicated") + condition_paths.add(path) + branches = _parse_branch_list(condition["base_branches"], "conditional workflow base_branches") + labels = condition["labels_any"] + if (not isinstance(labels, list) or len(labels) > 64 + or any(not isinstance(label, str) or not label or len(label) > 100 + or any(ord(c) < 32 for c in label) for label in labels) + or len(set(labels)) != len(labels) or (not branches and not labels)): + raise PrGuardConfigError("conditional workflow needs exact branches or labels") + if lifecycle["auto_ready"] and (not lifecycle["enabled"] or not lifecycle["required_workflows"] + or not normalized.get("workflow_approval", {}).get("enabled")): + raise PrGuardConfigError("auto_ready requires enabled lifecycle, workflow approval and required workflows") + checks = lifecycle.get("required_checks", {}) + eligible_paths = set(lifecycle["required_workflows"]) | {c["workflow"] for c in conditions} + if not isinstance(checks, dict) or set(checks) - eligible_paths: + raise PrGuardConfigError("required_checks must map required workflow paths to check names") + for names in checks.values(): + if (not isinstance(names, list) or not 1 <= len(names) <= 64 + or any(not isinstance(n, str) or not n or len(n) > 255 for n in names) + or len(set(names)) != len(names)): + raise PrGuardConfigError("required_checks needs bounded unique exact check names") + normalized["lifecycle"] = dict(lifecycle) return normalized diff --git a/src/agent_cli/pr_lifecycle.py b/src/agent_cli/pr_lifecycle.py new file mode 100644 index 0000000..9e56ec1 --- /dev/null +++ b/src/agent_cli/pr_lifecycle.py @@ -0,0 +1,269 @@ +"""Configured PR readiness, based on live CI rather than a cached green rollup. + +This reconciler never runs tests, submits reviews, or merges pull requests. +The caller must serialize all guard invocations for the repository. +""" +from __future__ import annotations + +import json +from typing import Any, Mapping + +from .workflow_approval import _field, _runs, _timestamp + +AUTH_MARKER = "" +STATE_MARKER = "" + + +def _own_record(api: Any, assessment: Any, marker: str) -> tuple[Mapping | None, dict]: + from .a38_guard import GuardError, collect_comments + own_id, _ = api.resolve_own_user() + comments = collect_comments(api, assessment.repo, assessment.pr) + records = [c for c in comments if _field(c, "user", "id") == own_id + and str(c.get("body", "")).startswith(marker + "\n")] + if len(records) > 1 and marker == AUTH_MARKER: + raise GuardError("ambiguous bot lifecycle audit comments") + if not records: + return None, {} + comment = max(records, key=lambda c: c["id"]) + try: + data = json.loads(comment["body"].split("```json\n", 1)[1].split("\n```", 1)[0]) + except (ValueError, IndexError, TypeError) as exc: + raise GuardError("invalid bot lifecycle audit comment") from exc + if not isinstance(data, dict): + raise GuardError("invalid bot lifecycle audit record") + return comment, data + + +def _save_record(api: Any, assessment: Any, marker: str, record: dict, + en: str, de: str, *, create: bool = False) -> None: + from .a38_guard import GuardError + existing, _ = _own_record(api, assessment, marker) + if create: + existing = None + body = (f"{marker}\nEN:\n{en}\n\nDE:\n{de}\n\n
\nDetails\n\n" + + "```json\n" + json.dumps(record, indent=2, sort_keys=True) + "\n```\n\n
") + if existing and existing.get("body") == body: + return + path = (f"/repos/{assessment.repo}/issues/comments/{existing['id']}" if existing + else f"/repos/{assessment.repo}/issues/{assessment.pr}/comments") + status, _, _ = api.request("PATCH" if existing else "POST", path, body={"body": body}, retry=False) + if not 200 <= status < 300: + raise GuardError(f"bot lifecycle audit comment HTTP {status}") + assessment.writes.append("lifecycle:comment") + + +def _complete_transition_comment(api: Any, assessment: Any, record: dict) -> None: + draft = record["state"] == "draft" + en = ("This pull request is back in Draft because CI is not fully green or merge conflicts exist." + if draft else "The authorized CI runs are green and no merge conflicts exist; this pull request is ready for review.") + de = ("Dieser Pull Request steht wieder auf Draft, weil die CI noch nicht vollständig grün ist oder Merge-Konflikte bestehen." + if draft else "Die freigegebenen CI-Läufe sind grün und es gibt keine Merge-Konflikte; dieser Pull Request ist bereit zum Review.") + _save_record(api, assessment, STATE_MARKER, {**record, "phase": "applied"}, en, de) + + +def record_workflow_approval(api: Any, assessment: Any, run: Mapping) -> None: + """Persist only an authorization whose POST returned 201, before the next one.""" + _, previous = _own_record(api, assessment, AUTH_MARKER) + identity = {"repo": assessment.repo, "pr": assessment.pr, + "head": assessment.head_sha, "base": assessment.base_sha} + runs = previous.get("runs", []) if all(previous.get(k) == v for k, v in identity.items()) else [] + runs = [r for r in runs if r.get("workflow") != run["path"]] + runs.append({"run_id": run["id"], "workflow": run["path"]}) + _save_record(api, assessment, AUTH_MARKER, {**identity, "runs": runs}, + "I have authorized the recorded CI runs; their results are still pending.", + "Ich habe die dokumentierten CI-Läufe freigegeben; ihre Ergebnisse stehen noch aus.") + + +def _checks(api: Any, repo: str, head: str) -> list[Mapping]: + from .a38_guard import GuardError + result: list[Mapping] = [] + total = None + for page in range(1, 11): + data = api.get_json(f"/repos/{repo}/commits/{head}/check-runs?filter=latest&per_page=100&page={page}") + count = _field(data, "total_count") + batch = _field(data, "check_runs") + if (type(count) is not int or not 0 <= count < 1000 + or (total is not None and count != total) + or not isinstance(batch, list) or any(not isinstance(c, Mapping) for c in batch)): + raise GuardError("CI check inventory invalid, changed, or exceeds bound") + total = count + result.extend(batch) + ids = [c.get("id") for c in result] + if any(type(i) is not int or i <= 0 for i in ids) or len(set(ids)) != len(ids): + raise GuardError("CI check inventory has invalid or duplicate IDs") + if len(result) == total: + return result + if len(result) > total or not batch: + raise GuardError("CI check inventory incomplete") + raise GuardError("CI check inventory exceeds page bound") + + +def ci_state(api: Any, assessment: Any, config: Mapping, pull: Mapping | None = None) -> tuple[list[str], dict[str, Mapping]]: + """Missing, waiting, running and failed required workflows all block Ready.""" + from .a38_guard import GuardError + required = set(config["required_workflows"]) + labels = {_field(label, "name") for label in (pull or {}).get("labels", [])} + for condition in config.get("conditional_workflows", []): + if assessment.base_ref in condition["base_branches"] or labels.intersection(condition["labels_any"]): + required.add(condition["workflow"]) + runs = _runs(api, assessment.repo, assessment.head_sha, event=None) + latest: dict[str, Mapping] = {} + ignored_suites = set() + superseded_suites = set() + for run in runs: + if run.get("head_sha") != assessment.head_sha: + raise GuardError("CI workflow inventory contains another head") + path = run.get("path") + if not isinstance(path, str): + raise GuardError("CI workflow inventory lacks a workflow path") + if path in config["ignored_workflows"]: + ignored_suites.add(run.get("check_suite_id")) + continue + # The head SHA can belong to several PRs; an explicit foreign PR link + # cannot provide evidence for this one. Empty private-fork links remain + # usable for observation, but never establish authorization ownership. + links = run.get("pull_requests") + if isinstance(links, list) and links and not any(_field(p, "number") == assessment.pr for p in links): + continue + previous = latest.get(path) + if previous is None or (_timestamp(run.get("created_at")), run["id"]) > (_timestamp(previous.get("created_at")), previous["id"]): + if previous: + superseded_suites.add(previous.get("check_suite_id")) + latest[path] = run + else: + superseded_suites.add(run.get("check_suite_id")) + reasons = [f"Missing required CI: {path}" for path in sorted(required) if path not in latest] + for path, run in sorted(latest.items()): + accepted = {"success"} if path in required else {"success", "skipped", "neutral"} + if run.get("status") != "completed" or run.get("conclusion") not in accepted: + reasons.append(f"CI not green: {path} ({run.get('conclusion') or run.get('status') or 'unknown'})") + # An old run's check suite must not override the latest workflow result. + excluded = (ignored_suites | superseded_suites) - {None} + excluded -= {r.get("check_suite_id") for r in latest.values()} + checks = _checks(api, assessment.repo, assessment.head_sha) + for path in sorted(required): + suite = _field(latest.get(path), "check_suite_id") + for name in config.get("required_checks", {}).get(path, []): + matches = [c for c in checks if suite is not None and _field(c, "check_suite", "id") == suite and c.get("name") == name] + check = max(matches, key=lambda c: c["id"]) if matches else {} + if check.get("status") != "completed" or check.get("conclusion") != "success": + reasons.append(f"Required CI check not green: {path} / {name}") + newest: dict[tuple, Mapping] = {} + for check in checks: + if _field(check, "check_suite", "id") in excluded: + continue + key = (_field(check, "app", "id"), check.get("name")) + if key not in newest or check["id"] > newest[key]["id"]: + newest[key] = check + for check in newest.values(): + # A workflow can intentionally skip individual conditional jobs while + # succeeding overall. Standalone skipped checks cannot establish green. + suite = _field(check, "check_suite", "id") + successful_suite = suite is not None and any(r.get("check_suite_id") == suite + and r.get("status") == "completed" and (r.get("conclusion") == "success" + or (path not in required and r.get("conclusion") in {"skipped", "neutral"})) for path, r in latest.items()) + acceptable = check.get("conclusion") == "success" or (successful_suite and check.get("conclusion") in {"skipped", "neutral"}) + if check.get("status") != "completed" or not acceptable: + reasons.append(f"CI check not green: {check.get('name')} ({check.get('conclusion') or check.get('status') or 'unknown'})") + statuses = api.paginate(f"/repos/{assessment.repo}/commits/{assessment.head_sha}/statuses") + seen = set() + for status in statuses: # GitHub returns newest first. + context = _field(status, "context") + if not isinstance(context, str): + raise GuardError("CI status inventory invalid") + if context not in seen and status.get("state") != "success": + reasons.append(f"CI status not green: {context} ({status.get('state') or 'unknown'})") + seen.add(context) + return reasons, latest + + +def _transition(api: Any, node: str, draft: bool) -> Mapping: + from .a38_guard import GuardError + operation = "convertPullRequestToDraft" if draft else "markPullRequestReadyForReview" + query = ("mutation($id: ID!) { " + operation + + "(input: {pullRequestId: $id}) { pullRequest { id isDraft headRefOid baseRefOid } } }") + status, data, _ = api.request("POST", "/graphql", body={"query": query, "variables": {"id": node}}, retry=False) + pull = _field(data, "data", operation, "pullRequest") + if status != 200 or _field(data, "errors") or _field(pull, "id") != node or _field(pull, "isDraft") is not draft: + raise GuardError(f"PR lifecycle mutation failed (HTTP {status})") + return pull + + +def reconcile_lifecycle(api: Any, assessment: Any, *, dry_run: bool = False) -> dict: + from .a38_guard import GuardError, assess_pull, fetch_pull, resolve_trusted_guard_config + if assessment.closed or not assessment.lifecycle_enabled: + return {} + snap = fetch_pull(api, assessment.repo, assessment.pr) + trusted = resolve_trusted_guard_config(api, snap) + config = (trusted.config or {}).get("lifecycle") + if not config or not config["enabled"]: + return {} + if snap.head_sha != assessment.head_sha or snap.base_sha != assessment.base_sha: + raise GuardError("pull changed before lifecycle assessment") + path = f"/repos/{assessment.repo}/pulls/{assessment.pr}" + pull = api.get_json(path) + if pull.get("state") != "open": + return {} + if type(pull.get("draft")) is not bool or not isinstance(pull.get("node_id"), str): + raise GuardError("pull lifecycle state missing") + # Recover the explanatory comment if an earlier process died after the + # mutation. The durable intent precedes it and never claims success early. + _, previous = _own_record(api, assessment, STATE_MARKER) + if (previous.get("phase") == "planned" and previous.get("head") == snap.head_sha + and previous.get("base") == snap.base_sha + and previous.get("state") == ("draft" if pull["draft"] else "ready") and not dry_run): + _complete_transition_comment(api, assessment, previous) + reasons, latest = ci_state(api, assessment, config, pull) + if pull.get("mergeable") is False: + reasons.insert(0, "Merge conflicts") + target = None + if not pull["draft"] and reasons: + target = "draft" + elif pull["draft"] and not reasons and pull.get("mergeable") is True and config["auto_ready"]: + _, authorization = _own_record(api, assessment, AUTH_MARKER) + identity = {"repo": assessment.repo, "pr": assessment.pr, "head": snap.head_sha, "base": snap.base_sha} + owned = authorization.get("runs", []) + if (all(authorization.get(k) == v for k, v in identity.items()) and owned + and all(_field(latest.get(r.get("workflow")), "id") == r.get("run_id") for r in owned)): + fresh = assess_pull(api, assessment.repo, assessment.pr, dry_run=True) + if fresh.ok and fresh.status == "pass" and fresh.mode == "enforce" and fresh.head_sha == snap.head_sha and fresh.base_sha == snap.base_sha: + target = "ready" + result = {"action": target or "unchanged", "reasons": reasons, "dry_run": dry_run} + if target is None or dry_run: + return result + # Re-read state/config and CI immediately before changing readiness. + final_snap = fetch_pull(api, assessment.repo, assessment.pr) + final_config = resolve_trusted_guard_config(api, final_snap) + final_pull = api.get_json(path) + if (final_snap != snap or final_config.fingerprint != trusted.fingerprint + or final_config.config_revision != trusted.config_revision + or final_pull.get("draft") != pull["draft"]): + raise GuardError("pull or configuration changed before lifecycle transition") + final_reasons, _ = ci_state(api, assessment, config, final_pull) + if final_pull.get("mergeable") is False: + final_reasons.insert(0, "Merge conflicts") + if target == "ready" and (final_reasons or final_pull.get("mergeable") is not True): + return {"action": "unchanged", "reasons": final_reasons, "dry_run": False} + if target == "draft" and not final_reasons: + return {"action": "unchanged", "reasons": [], "dry_run": False} + record = {"repo": assessment.repo, "pr": assessment.pr, "head": snap.head_sha, + "base": snap.base_sha, "state": target, "reasons": final_reasons, "phase": "planned"} + _save_record(api, assessment, STATE_MARKER, record, + "I am checking the final conditions for the documented readiness change.", + "Ich prüfe die letzten Voraussetzungen für die dokumentierte Statusänderung.", create=True) + if fetch_pull(api, assessment.repo, assessment.pr) != snap: + raise GuardError("pull changed after lifecycle intent; no readiness change") + if target == "ready": + fresh = assess_pull(api, assessment.repo, assessment.pr, dry_run=True) + fields = ("head_sha", "base_sha", "config_revision", "config_fingerprint", "report_fingerprint", "approval_fingerprint") + if (not fresh.ok or fresh.status != "pass" or fresh.mode != "enforce" + or any(getattr(fresh, f) != getattr(assessment, f) for f in fields)): + raise GuardError("A38 evidence changed before Ready transition") + changed = _transition(api, pull["node_id"], target == "draft") + assessment.writes.append(f"pull:{target}") + if target == "ready" and (changed.get("headRefOid") != snap.head_sha or changed.get("baseRefOid") != snap.base_sha): + _transition(api, pull["node_id"], True) + raise GuardError("pull changed during Ready transition; restored Draft") + _complete_transition_comment(api, assessment, record) + result["reasons"] = final_reasons + return result diff --git a/src/agent_cli/workflow_approval.py b/src/agent_cli/workflow_approval.py index 86a75da..b92607e 100644 --- a/src/agent_cli/workflow_approval.py +++ b/src/agent_cli/workflow_approval.py @@ -12,6 +12,14 @@ MAX_RUNS = 1000 +def _field(value: Any, *keys: str) -> Any: + for key in keys: + if not isinstance(value, Mapping): + return None + value = value.get(key) + return value + + def _timestamp(value: Any) -> datetime: from .a38_guard import GuardError if not isinstance(value, str): @@ -25,12 +33,15 @@ def _timestamp(value: Any) -> datetime: return result.astimezone(timezone.utc) -def _runs(api: Any, repo: str, head: str) -> list[Mapping[str, Any]]: +def _runs(api: Any, repo: str, head: str, *, event: str | None = "pull_request") -> list[Mapping[str, Any]]: from .a38_guard import GuardError items: list[Mapping[str, Any]] = [] total = None for page in range(1, 11): - query = urlencode(dict(event="pull_request", head_sha=head, per_page=100, page=page)) + params = dict(head_sha=head, per_page=100, page=page) + if event is not None: + params["event"] = event + query = urlencode(params) data = api.get_json(f"/repos/{repo}/actions/runs?{query}") if not isinstance(data, Mapping) or type(data.get("total_count")) is not int: raise GuardError("workflow run inventory missing total_count") @@ -99,8 +110,8 @@ def _belongs_to_pull(api: Any, run: Mapping[str, Any], pull: Mapping[str, Any]) raise GuardError("workflow run pull request association is ambiguous") linked = links[0] if (linked.get("number") != pull["number"] - or linked.get("head", {}).get("sha") != head["sha"] - or linked.get("base", {}).get("sha") != pull["base"]["sha"]): + or _field(linked, "head", "sha") != head["sha"] + or _field(linked, "base", "sha") != pull["base"]["sha"]): raise GuardError("workflow run targets another pull request head or base") return # GitHub returns an empty associations array for private forks. Prove the @@ -108,12 +119,12 @@ def _belongs_to_pull(api: Any, run: Mapping[str, Any], pull: Mapping[str, Any]) # the current base (an older-base merge cannot change the measured tree). pulls = api.paginate(f"/repos/{repo}/pulls?state=open") matches = [p for p in pulls if isinstance(p, Mapping) and p.get("state") == "open" - and p.get("head", {}).get("ref") == head["ref"] - and p.get("head", {}).get("repo", {}).get("full_name") == head["repo"]["full_name"]] + and _field(p, "head", "ref") == head["ref"] + and _field(p, "head", "repo", "full_name") == head["repo"]["full_name"]] if len(matches) != 1 or matches[0].get("number") != pull["number"]: raise GuardError("fork workflow run cannot be uniquely associated with this pull request") match = matches[0] - if match.get("head", {}).get("sha") != head["sha"] or match.get("base", {}).get("sha") != pull["base"]["sha"]: + if _field(match, "head", "sha") != head["sha"] or _field(match, "base", "sha") != pull["base"]["sha"]: raise GuardError("fork pull request changed during workflow approval") comparison = api.get_json(f"/repos/{repo}/compare/{pull['base']['sha']}...{head['sha']}") if not isinstance(comparison, Mapping) or comparison.get("status") not in {"ahead", "identical"}: @@ -152,11 +163,11 @@ def fresh_pull() -> Mapping[str, Any]: raise GuardError("A38 evidence or trusted configuration changed before workflow approval") pull = api.get_json(f"/repos/{assessment.repo}/pulls/{assessment.pr}") if (not isinstance(pull, Mapping) or pull.get("state") != "open" - or pull.get("head", {}).get("sha") != assessment.head_sha - or pull.get("base", {}).get("sha") != assessment.base_sha - or pull.get("base", {}).get("ref") != assessment.base_ref - or pull.get("head", {}).get("repo", {}).get("full_name") != assessment.head_repo - or not isinstance(pull.get("head", {}).get("ref"), str)): + or _field(pull, "head", "sha") != assessment.head_sha + or _field(pull, "base", "sha") != assessment.base_sha + or _field(pull, "base", "ref") != assessment.base_ref + or _field(pull, "head", "repo", "full_name") != assessment.head_repo + or not isinstance(_field(pull, "head", "ref"), str)): raise GuardError("pull request changed before workflow approval") return pull @@ -192,6 +203,9 @@ def fresh_pull() -> Mapping[str, Any]: if status != 201: raise GuardError(f"workflow approval HTTP {status}; Actions write permission is required") assessment.writes.append(f"workflow:approve:{run['id']}") + if assessment.lifecycle_enabled: + from .pr_lifecycle import record_workflow_approval + record_workflow_approval(api, assessment, run) result.append({"run_id": run["id"], "workflow": path, "head": assessment.head_sha, "status": "planned" if dry_run else "approved"}) return result diff --git a/tests/test_pr_lifecycle.py b/tests/test_pr_lifecycle.py new file mode 100644 index 0000000..fe08d26 --- /dev/null +++ b/tests/test_pr_lifecycle.py @@ -0,0 +1,248 @@ +"""CI completion, ownership, conflicts and transitions through the real guard.""" +import copy +import json +from urllib.parse import urlparse + +import pytest + +from agent_cli.a38_guard import GuardError, reconcile_pull +from agent_cli.pr_guard_config import load_pr_guard_config, PrGuardConfigError +from agent_cli.pr_lifecycle import AUTH_MARKER, STATE_MARKER +from test_a38_guard import HEAD, BASE, BASE2, BOT_ID, REPO +from test_workflow_approval_core import ApprovalAPI, PATH + +pytestmark = pytest.mark.no_pg +GUARD = ".github/workflows/guard.yml" + + +class LifecycleAPI(ApprovalAPI): + def __init__(self): + super().__init__() + self.config["lifecycle"] = {"enabled": True, "auto_ready": True, + "required_workflows": [PATH], "ignored_workflows": [GUARD]} + self.set_pr_guard_config(self.config) + self.pull.update(draft=False, mergeable=True, node_id="PR_example") + self.runs[0].update(check_suite_id=201, conclusion="success") + self.checks = [] + self.transitions = [] + self.graphql_error = False + self.mutate_during_transition = False + self.fail_comment_once = False + + def request_fn(self, method, url, body=None): + path = urlparse(url).path + if method == "GET" and path.endswith("/check-runs"): + return 200, {"total_count": len(self.checks), "check_runs": copy.deepcopy(self.checks)}, {} + if method == "POST" and path == "/graphql": + payload = json.loads(body) + assert payload["variables"] == {"id": "PR_example"} + assert "mergePullRequest" not in payload["query"] + operation = "convertPullRequestToDraft" if "convertPullRequestToDraft" in payload["query"] else "markPullRequestReadyForReview" + if self.graphql_error: + return 200, {"errors": [{"message": "denied"}]}, {} + self.pull["draft"] = operation == "convertPullRequestToDraft" + self.transitions.append(self.pull["draft"]) + if self.mutate_during_transition: + self.pull["head"]["sha"] = BASE2 + return 200, {"data": {operation: {"pullRequest": { + "id": "PR_example", "isDraft": self.pull["draft"], + "headRefOid": self.pull["head"]["sha"], "baseRefOid": BASE}}}}, {} + if method == "PATCH" and "/issues/comments/" in path and self.fail_comment_once: + payload = json.loads(body) + if STATE_MARKER in payload.get("body", "") and '"phase": "applied"' in payload["body"]: + self.fail_comment_once = False + return 503, {}, {} + return super().request_fn(method, url, body) + + def own_authorization(self, **changes): + record = {"repo": REPO, "pr": 1, "head": HEAD, "base": BASE, + "runs": [{"run_id": 101, "workflow": PATH}]} + record.update(changes) + self.comments.append({"id": 500, "user": {"id": BOT_ID}, + "body": AUTH_MARKER + "\n```json\n" + json.dumps(record) + "\n```"}) + + +@pytest.mark.parametrize("status,conclusion", [ + ("queued", None), ("waiting", None), ("in_progress", None), ("pending", None), + ("completed", "failure"), ("completed", "cancelled"), ("completed", "timed_out"), + ("completed", "action_required"), ("completed", "skipped"), ("completed", "neutral"), +]) +def test_not_fully_green_returns_ready_to_draft_once(status, conclusion): + fake = LifecycleAPI() + fake.runs[0].update(status=status, conclusion=conclusion) + result = reconcile_pull(fake.api(), REPO, 1) + assert result.lifecycle["action"] == "draft" + assert fake.pull["draft"] and fake.transitions == [True] + comments = [c for c in fake.comments if c["body"].startswith(STATE_MARKER)] + assert len(comments) == 1 and '"phase": "applied"' in comments[0]["body"] + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + assert len([c for c in fake.comments if c["body"].startswith(STATE_MARKER)]) == 1 + + +def test_missing_ci_is_not_vacuously_green(): + fake = LifecycleAPI() + fake.runs.clear() + result = reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + assert result.lifecycle["reasons"] == [f"Missing required CI: {PATH}"] + + +def test_conflict_demotes_even_with_all_ci_green_and_a38_excluded(): + fake = LifecycleAPI() + fake.pull["mergeable"] = False + fake.config["a38"]["default"] = "exclude" + fake.set_pr_guard_config(fake.config) + result = reconcile_pull(fake.api(), REPO, 1) + assert result.status == "not_applicable" + assert fake.transitions == [True] + assert result.lifecycle["reasons"] == ["Merge conflicts"] + + +def test_unknown_mergeability_never_promotes_or_invents_conflicts(): + fake = LifecycleAPI() + fake.pull["mergeable"] = None + fake.own_authorization() + reconcile_pull(fake.api(), REPO, 1) + fake.pull["draft"] = True + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [] + + +def test_approval_then_pending_then_green_promotes_once_without_rerunning(): + fake = LifecycleAPI() + fake.runs[0]["conclusion"] = "action_required" + first = reconcile_pull(fake.api(), REPO, 1) + assert first.workflow_approvals[0]["status"] == "approved" + assert fake.posts == [101] and fake.transitions == [True] + assert len([c for c in fake.comments if c["body"].startswith(AUTH_MARKER)]) == 1 + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + fake.runs[0].update(status="completed", conclusion="success") + assert reconcile_pull(fake.api(), REPO, 1).lifecycle["action"] == "ready" + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [101] and fake.transitions == [True, False] + + +@pytest.mark.parametrize("case", ["missing", "forged", "head", "base", "run", "report", "disabled"]) +def test_green_alone_does_not_authorize_auto_ready(case): + fake = LifecycleAPI() + fake.pull["draft"] = True + if case != "missing": + fake.own_authorization() + if case == "forged": + fake.comments[-1]["user"]["id"] = 77 + elif case in {"head", "base"}: + fake.comments[-1]["body"] = fake.comments[-1]["body"].replace(HEAD if case == "head" else BASE, BASE2) + elif case == "run": + fake.runs[0]["id"] = 102 + elif case == "report": + fake.comments = fake.comments[1:] + elif case == "disabled": + fake.config["lifecycle"]["auto_ready"] = False + fake.set_pr_guard_config(fake.config) + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [] + + +def test_guard_itself_and_superseded_failures_do_not_block(): + fake = LifecycleAPI() + fake.pull["draft"] = True + fake.own_authorization() + fake.runs.append(fake.run(id=99, check_suite_id=199, conclusion="failure", created_at="2026-09-04T00:00:00Z")) + fake.runs.append(fake.run(id=102, path=GUARD, check_suite_id=202, event="pull_request_target", status="in_progress", conclusion=None)) + fake.checks = [dict(id=10, name="old test", check_suite={"id": 199}, status="completed", conclusion="failure"), + dict(id=11, name="guard", check_suite={"id": 202}, status="in_progress", conclusion=None)] + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [False] + + +@pytest.mark.parametrize("status,conclusion", [("in_progress", None), ("completed", "failure")]) +def test_independent_check_blocks_ready(status, conclusion): + fake = LifecycleAPI() + fake.checks = [dict(id=11, name="security", check_suite={"id": 300}, status=status, conclusion=conclusion)] + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + + +def test_commit_status_pending_blocks(): + fake = LifecycleAPI() + fake.statuses.append({"sha": HEAD, "context": "external", "state": "pending"}) + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + + +@pytest.mark.parametrize("labels,base,expected", [([], "develop", False), (["full"], "develop", True), ([], "release", True)]) +def test_conditional_required_workflow_uses_repo_branch_or_label(labels, base, expected): + from agent_cli.pr_lifecycle import ci_state + from agent_cli.a38_guard import assess_pull + fake = LifecycleAPI() + fake.config["lifecycle"]["conditional_workflows"] = [{"workflow": ".github/workflows/security.yml", "base_branches": ["release"], "labels_any": ["full"]}] + assessment = assess_pull(fake.api(), REPO, 1) + assessment.base_ref = base + reasons, _ = ci_state(fake.api(), assessment, fake.config["lifecycle"], {"labels": [{"name": n} for n in labels]}) + assert any("Missing required CI" in r for r in reasons) is expected + + +@pytest.mark.parametrize("conclusion", [None, "skipped", "failure", "success"]) +def test_successful_workflow_cannot_hide_a_missing_or_skipped_required_test(conclusion): + fake = LifecycleAPI() + fake.config["lifecycle"]["required_checks"] = {PATH: ["Test"]} + fake.set_pr_guard_config(fake.config) + if conclusion: + fake.checks = [{"id": 22, "name": "Test", "check_suite": {"id": 201}, "status": "completed", "conclusion": conclusion}] + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == ([] if conclusion == "success" else [True]) + + +def test_dry_run_is_read_only(): + fake = LifecycleAPI() + fake.runs.clear() + result = reconcile_pull(fake.api(), REPO, 1, dry_run=True) + assert result.lifecycle["action"] == "draft" + assert not fake.transitions and not fake.writes + + +def test_graphql_error_is_not_a_successful_transition(): + fake = LifecycleAPI() + fake.runs.clear() + fake.graphql_error = True + with pytest.raises(GuardError, match="mutation failed"): + reconcile_pull(fake.api(), REPO, 1) + assert not fake.transitions + assert all('"phase": "applied"' not in c["body"] for c in fake.comments) + + +def test_changed_head_during_ready_is_restored_to_draft(): + fake = LifecycleAPI() + fake.pull["draft"] = True + fake.own_authorization() + fake.mutate_during_transition = True + with pytest.raises(GuardError, match="restored Draft"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [False, True] + + +def test_comment_failure_is_repaired_without_repeating_transition(): + fake = LifecycleAPI() + fake.runs.clear() + fake.fail_comment_once = True + with pytest.raises(GuardError, match="comment HTTP 503"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + reconcile_pull(fake.api(), REPO, 1) + assert fake.transitions == [True] + comments = [c for c in fake.comments if c["body"].startswith(STATE_MARKER)] + assert len(comments) == 1 and '"phase": "applied"' in comments[0]["body"] + + +@pytest.mark.parametrize("change", [ + {"enabled": "true"}, {"auto_ready": 1}, {"unknown": True}, + {"required_workflows": []}, {"required_workflows": [PATH, PATH]}, + {"required_workflows": [GUARD]}, {"ignored_workflows": ["*.yml"]}, +]) +def test_lifecycle_config_is_strict(change): + fake = LifecycleAPI() + fake.config["lifecycle"].update(change) + with pytest.raises(PrGuardConfigError): + load_pr_guard_config(json.dumps(fake.config)) diff --git a/tests/test_workflow_approval.py b/tests/test_workflow_approval.py new file mode 100644 index 0000000..cc950fa --- /dev/null +++ b/tests/test_workflow_approval.py @@ -0,0 +1,754 @@ +"""Workflow-approval opt-in: config, latest run, fork association, writes.""" + +from __future__ import annotations + +import copy +import json +from urllib.parse import parse_qs, urlparse + +import pytest + +from agent_cli import workflow_approval as wa +from agent_cli.a38_guard import ( + POLICY_APPROVAL_PREFIX, + Assessment, + GuardError, + assess_pull, + reconcile_pull, +) +from agent_cli.pr_guard_config import PrGuardConfigError, load_pr_guard_config +from test_a38_guard import ( + BASE, + BASE2, + HEAD, + REPO, + FakeAPI, + _pr_guard_config, + _report_comment, +) + +pytestmark = pytest.mark.no_pg + +PATH = ".github/workflows/test.yml" +OTHER = ".github/workflows/other.yaml" +FORK = "contributor/fork" +BRANCH = "feature" +MAINTAINER = 3030 +A38_BODY = { + "schema": "pr-guard/v1", + "a38": {"enforce": [], "exclude": [], "default": "enforce"}, +} + + +def _cfg(approval: dict) -> dict: + payload = copy.deepcopy(A38_BODY) + payload["workflow_approval"] = approval + return payload + + +def _link(*, number: int = 1, head: str = HEAD, base: str = BASE) -> dict: + return {"number": number, "head": {"sha": head}, "base": {"sha": base}} + + +class FakeApproval(FakeAPI): + """FakeAPI plus Actions inventory, approve POST, fork association extras.""" + + def __init__(self) -> None: + super().__init__() + self.pull["head"]["repo"]["full_name"] = FORK + self.pull["head"]["ref"] = BRANCH + self.pull["created_at"] = "2026-09-01T00:00:00Z" + self.config = _cfg({"enabled": True, "workflows": [PATH]}) + self.set_pr_guard_config(self.config) + self.add_author_report(_report_comment(), updated_at="2026-09-05T12:00:00Z", cid=21) + self.runs = [self.run()] + self.posts: list[int] = [] + self.actions_pages: list[int] = [] + self.compare_urls: list[str] = [] + self.event_reads = 0 + self.post_status = 201 + self.inventory = None + self.on_inventory = None + self.before_run_read = None + self.extra_pulls: list[dict] = [] + self.list_head_sha: str | None = None + self.events: list[dict] = [] + self.comparison = "ahead" + self.run_override = None + self.listed_pulls: list[dict] | None = None + + @staticmethod + def run(**changes: object) -> dict: + data: dict = { + "id": 101, + "path": PATH, + "event": "pull_request", + "head_sha": HEAD, + "head_branch": BRANCH, + "repository": {"full_name": REPO}, + "head_repository": {"full_name": FORK}, + "pull_requests": [], + "status": "completed", + "conclusion": "action_required", + "run_attempt": 1, + "created_at": "2026-09-05T11:00:00Z", + } + data.update(changes) + return data + + def request_fn(self, method: str, url: str, body: bytes | None = None): + path = urlparse(url).path + root = f"/repos/{REPO}" + if method == "GET" and path == f"{root}/actions/runs": + page = int((parse_qs(urlparse(url).query).get("page") or ["1"])[0]) + per_page = int((parse_qs(urlparse(url).query).get("per_page") or ["100"])[0]) + self.actions_pages.append(page) + if self.on_inventory is not None: + self.on_inventory(self, page) + if callable(self.inventory): + return 200, copy.deepcopy(self.inventory(page)), {} + if self.inventory is not None: + return 200, copy.deepcopy(self.inventory), {} + start = (page - 1) * per_page + chunk = self.runs[start : start + per_page] + return 200, {"total_count": len(self.runs), "workflow_runs": copy.deepcopy(chunk)}, {} + if path.startswith(f"{root}/actions/runs/"): + ident = int(path.split("/actions/runs/")[1].split("/")[0]) + run = next(r for r in self.runs if r["id"] == ident) + if method == "POST": + assert path.endswith("/approve") + self.posts.append(ident) + if self.post_status == 201: + run.update(status="queued", conclusion=None) + return self.post_status, {}, {} + if self.before_run_read is not None: + callback, self.before_run_read = self.before_run_read, None + callback(self) + payload = self.run_override if self.run_override is not None else run + return 200, copy.deepcopy(payload), {} + if method == "GET" and path == f"{root}/pulls": + if self.listed_pulls is not None: + return 200, copy.deepcopy(self.listed_pulls), {} + listed = copy.deepcopy(self.pull) + if self.list_head_sha is not None: + listed["head"]["sha"] = self.list_head_sha + return 200, [listed, *copy.deepcopy(self.extra_pulls)], {} + if method == "GET" and path.startswith(f"{root}/compare/"): + self.compare_urls.append(path) + return 200, {"status": self.comparison}, {} + if method == "GET" and path == f"{root}/issues/1/events": + self.event_reads += 1 + return 200, copy.deepcopy(self.events), {} + status, data, headers = super().request_fn(method, url, body) + return status, copy.deepcopy(data), headers + + +def _private_fork() -> FakeApproval: + fake = FakeApproval() + fake.pull["base"]["repo"]["private"] = True + fake.comments.clear() + fake.add_author_report( + _report_comment(private=True), updated_at="2026-09-05T12:00:00Z", cid=21 + ) + return fake + + +# --- config ----------------------------------------------------------------- + + +@pytest.mark.parametrize( + "raw", + [ + json.dumps(_cfg({"enabled": True, "workflows": [PATH], "retry": True})), + json.dumps({**A38_BODY, "extra": 1, "workflow_approval": {"enabled": True, "workflows": [PATH]}}), + json.dumps(_cfg({"enabled": "true", "workflows": [PATH]})), + json.dumps(_cfg({"enabled": 1, "workflows": [PATH]})), + json.dumps(_cfg({"enabled": True, "workflows": PATH})), + json.dumps(_cfg({"enabled": True, "workflows": [1]})), + json.dumps({**A38_BODY, "workflow_approval": [PATH]}), + json.dumps(_cfg({"enabled": True, "workflows": []})), + json.dumps(_cfg({"enabled": True, "workflows": [PATH, PATH]})), + json.dumps(_cfg({"enabled": False, "workflows": [PATH, PATH]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/*.yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/test-?.yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/ci[ab].yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/../x.yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/foo/bar.yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/foo/../../etc.yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": ["/etc/passwd.yml"]})), + json.dumps(_cfg({"enabled": True, "workflows": [".github/workflows/x.yml\\y.yml"]})), + json.dumps(_cfg({"enabled": True})), + json.dumps(_cfg({"workflows": [PATH]})), + json.dumps(_cfg({"enabled": True, "workflows": [PATH + "x" * 240]})), + json.dumps(_cfg({"enabled": True, "workflows": [f".github/workflows/w{i}.yml" for i in range(65)]})), + '{"schema":"pr-guard/v1","a38":{"enforce":[],"exclude":[],"default":"enforce"},' + '"workflow_approval":{"enabled":true,"enabled":false,"workflows":[".github/workflows/test.yml"]}}', + ], + ids=[ + "unknown-approval-field", + "unknown-top-field", + "enabled-string", + "enabled-int", + "workflows-string", + "workflows-int-item", + "approval-array", + "enabled-empty", + "duplicate-paths", + "duplicate-paths-disabled", + "glob-star", + "glob-question", + "glob-brackets", + "traversal-dotdot", + "nested-dir", + "traversal-nested", + "absolute-path", + "backslash", + "missing-workflows", + "missing-enabled", + "path-too-long", + "sixty-five-paths", + "duplicate-json-key", + ], +) +def test_invalid_workflow_approval_config_fails_closed(raw: str) -> None: + with pytest.raises(PrGuardConfigError): + load_pr_guard_config(raw) + + +def test_valid_strict_workflow_approval_config() -> None: + long_name = "a" * (255 - len(".github/workflows/") - len(".yml")) + long_path = f".github/workflows/{long_name}.yml" + assert len(long_path) == 255 + cfg = load_pr_guard_config( + json.dumps( + _cfg( + { + "enabled": True, + "workflows": [PATH, OTHER, long_path], + } + ) + ) + ) + assert cfg["workflow_approval"] == { + "enabled": True, + "workflows": [PATH, OTHER, long_path], + } + sixty_four = load_pr_guard_config( + json.dumps(_cfg({"enabled": True, "workflows": [f".github/workflows/w{i}.yml" for i in range(64)]})) + ) + assert len(sixty_four["workflow_approval"]["workflows"]) == 64 + disabled = load_pr_guard_config(json.dumps(_cfg({"enabled": False, "workflows": []}))) + assert disabled["workflow_approval"] == {"enabled": False, "workflows": []} + missing = load_pr_guard_config(json.dumps(A38_BODY)) + assert "workflow_approval" not in missing + + +def test_invalid_trusted_config_never_lists_actions() -> None: + fake = FakeApproval() + fake.set_pr_guard_config(_cfg({"enabled": True, "workflows": [".github/workflows/*.yml"]})) + with pytest.raises(GuardError, match="pr-guard"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.actions_pages == [] + assert fake.posts == [] + + +# --- opt-in / short-circuit ------------------------------------------------- + + +@pytest.mark.parametrize( + "kwargs", + [ + {"ok": False, "status": "fail", "mode": "enforce", "workflow_approval_enabled": True}, + {"ok": True, "status": "pass", "mode": "enforce", "workflow_approval_enabled": True, "closed": True}, + {"ok": True, "status": "pass", "mode": "observe", "workflow_approval_enabled": True}, + {"ok": True, "status": "pass", "mode": "enforce", "workflow_approval_enabled": False}, + {"ok": True, "status": "not_applicable", "mode": "enforce", "workflow_approval_enabled": True}, + ], +) +def test_approve_short_circuits_without_touching_api(kwargs: dict) -> None: + assert wa.approve_workflow_runs(object(), Assessment(**kwargs)) == [] + + +@pytest.mark.parametrize("case", ["missing", "disabled", "same_repo", "closed", "observe", "fail"]) +def test_opt_in_disabled_or_missing_never_calls_actions(case: str) -> None: + fake = FakeApproval() + if case == "missing": + fake.set_pr_guard_config(_pr_guard_config()) + elif case == "disabled": + fake.set_pr_guard_config(_cfg({"enabled": False, "workflows": [PATH]})) + elif case == "same_repo": + fake.pull["head"]["repo"]["full_name"] = REPO + elif case == "closed": + fake.pull["state"] = "closed" + elif case == "observe": + policy = json.loads(fake.files[(BASE, ".github/a38.json")]) + policy["mode"] = "observe" + fake.files[(BASE, ".github/a38.json")] = json.dumps(policy).encode() + else: + fake.comments.clear() + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals == [] + assert "workflow_approvals" in result.to_json() + assert result.to_json()["workflow_approvals"] == [] + assert fake.actions_pages == [] + assert fake.posts == [] + + +# --- reconcile integration -------------------------------------------------- + + +def test_reconcile_approves_and_exposes_writes_then_is_idempotent() -> None: + fake = FakeApproval() + result = reconcile_pull(fake.api(), REPO, 1) + assert result.ok + assert result.status == "pass" + expected = {"run_id": 101, "workflow": PATH, "head": HEAD, "status": "approved"} + assert result.workflow_approvals == [expected] + assert result.to_json()["workflow_approvals"] == [expected] + assert "workflow:approve:101" in result.writes + assert fake.posts == [101] + second = reconcile_pull(fake.api(), REPO, 1) + assert second.ok + assert second.workflow_approvals == [] + assert fake.posts == [101] + + +def test_dry_run_plans_without_post() -> None: + fake = FakeApproval() + result = reconcile_pull(fake.api(), REPO, 1, dry_run=True) + assert result.workflow_approvals == [ + {"run_id": 101, "workflow": PATH, "head": HEAD, "status": "planned"} + ] + assert result.to_json()["workflow_approvals"][0]["status"] == "planned" + assert fake.posts == [] + assert fake.writes == [] + assert fake.actions_pages # inventory is still read + + +def test_linked_association_approves_without_open_pr_fallback() -> None: + fake = FakeApproval() + fake.runs[0]["pull_requests"] = [_link()] + other = copy.deepcopy(fake.pull) + other["number"] = 2 + fake.extra_pulls = [other] + fake.comparison = "diverged" + fake.events = [{"event": "reopened", "created_at": "2026-09-05T11:30:00Z"}] + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals[0]["status"] == "approved" + assert fake.posts == [101] + assert fake.compare_urls == [] + assert fake.event_reads == 0 + + +@pytest.mark.parametrize("status", ["ahead", "identical"]) +def test_private_fork_empty_associations_unique_open_pr_and_base_ancestor(status: str) -> None: + fake = _private_fork() + fake.comparison = status + fake.events = [{"event": "reopened", "created_at": "2026-09-04T00:00:00Z"}] + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals == [ + {"run_id": 101, "workflow": PATH, "head": HEAD, "status": "approved"} + ] + assert fake.posts == [101] + assert any(f"{BASE}...{HEAD}" in url for url in fake.compare_urls) + assert fake.event_reads >= 1 + + +# --- latest-per-workflow / exact match / attempt ---------------------------- + + +@pytest.mark.parametrize( + "changes", + [ + {"status": "completed", "conclusion": "success"}, + {"status": "completed", "conclusion": "failure"}, + {"status": "queued", "conclusion": None}, + {"status": "in_progress", "conclusion": None}, + {"status": "completed", "conclusion": "cancelled"}, + {"status": "waiting", "conclusion": None}, + {"run_attempt": 2}, + ], +) +def test_newer_run_of_any_status_suppresses_old_action_required(changes: dict) -> None: + fake = FakeApproval() + fake.runs.append(fake.run(id=102, created_at="2026-09-05T11:01:00Z", **changes)) + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals == [] + assert fake.posts == [] + + +def test_newer_action_required_is_approved_over_older_success() -> None: + fake = FakeApproval() + fake.runs = [ + fake.run(id=100, created_at="2026-09-05T10:00:00Z", conclusion="success"), + fake.run(id=101, created_at="2026-09-05T11:00:00Z"), + ] + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals[0]["run_id"] == 101 + assert fake.posts == [101] + + +def test_same_timestamp_higher_id_is_latest() -> None: + fake = FakeApproval() + stamp = "2026-09-05T11:00:00Z" + fake.runs = [ + fake.run(id=101, created_at=stamp), + fake.run(id=102, created_at=stamp, conclusion="success"), + ] + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + fake.posts.clear() + fake.runs = [ + fake.run(id=101, created_at=stamp, conclusion="success"), + fake.run(id=102, created_at=stamp), + ] + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals[0]["run_id"] == 102 + assert fake.posts == [102] + + +@pytest.mark.parametrize( + "changes", + [ + {"event": "push"}, + {"event": "workflow_dispatch"}, + {"head_sha": BASE}, + {"head_branch": "other"}, + {"repository": {"full_name": FORK}}, + {"head_repository": {"full_name": REPO}}, + {"path": OTHER}, + {"run_attempt": 2}, + {"run_attempt": 3}, + {"status": "completed", "conclusion": "success"}, + ], +) +def test_inexact_or_retry_runs_are_never_approved(changes: dict) -> None: + fake = FakeApproval() + fake.runs = [fake.run(**changes)] + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals == [] + assert fake.posts == [] + + +def test_allowlist_approves_only_listed_workflow() -> None: + fake = FakeApproval() + fake.runs = [ + fake.run(id=101), + fake.run(id=202, path=OTHER), + ] + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals == [ + {"run_id": 101, "workflow": PATH, "head": HEAD, "status": "approved"} + ] + assert fake.posts == [101] + + +# --- pagination ------------------------------------------------------------- + + +def test_pagination_boundary_reads_page_two_for_the_pending_run() -> None: + fake = FakeApproval() + fake.runs = [fake.run(id=i, path=OTHER) for i in range(1, 101)] + fake.runs.append(fake.run(id=101)) + result = reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [101] + assert 2 in fake.actions_pages + assert result.workflow_approvals[0]["run_id"] == 101 + + +def test_pagination_exact_page_does_not_fetch_another_page() -> None: + fake = FakeApproval() + fake.runs = [fake.run(id=i, path=OTHER) for i in range(1, 100)] + fake.runs.append(fake.run(id=100)) + + def inventory(page: int) -> dict: + if page != 1: + raise AssertionError(f"unexpected page {page} when total_count is 100") + return {"total_count": 100, "workflow_runs": copy.deepcopy(fake.runs)} + + fake.inventory = inventory + result = reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [100] + assert fake.actions_pages == [1, 1] + assert result.workflow_approvals[0]["run_id"] == 100 + + +@pytest.mark.parametrize( + "inventory", + [ + {"total_count": 1000, "workflow_runs": []}, + {"total_count": -1, "workflow_runs": []}, + {"workflow_runs": []}, + {"total_count": True, "workflow_runs": []}, + {"total_count": 1, "workflow_runs": "runs"}, + {"total_count": 1, "workflow_runs": [{}]}, + {"total_count": 1, "workflow_runs": [{"id": "101"}]}, + {"total_count": 2, "workflow_runs": [FakeApproval.run(), FakeApproval.run()]}, + {"total_count": 1, "workflow_runs": [FakeApproval.run(), "x"]}, + ], + ids=[ + "at-max-bound", + "negative-count", + "missing-total", + "bool-total", + "runs-not-list", + "missing-id", + "string-id", + "duplicate-ids", + "non-mapping-item", + ], +) +def test_malformed_inventory_fails_closed_without_post(inventory: dict) -> None: + fake = FakeApproval() + fake.inventory = inventory + with pytest.raises(GuardError): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + + +def test_pagination_count_changed_between_pages_fails_closed() -> None: + fake = FakeApproval() + + def inventory(page: int) -> dict: + if page == 1: + return { + "total_count": 101, + "workflow_runs": [FakeApproval.run(id=i, path=OTHER) for i in range(1, 101)], + } + return {"total_count": 102, "workflow_runs": [FakeApproval.run(id=101)]} + + fake.inventory = inventory + with pytest.raises(GuardError, match="changed or exceeds"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + + +def test_pagination_duplicate_id_across_pages_fails_closed() -> None: + fake = FakeApproval() + + def inventory(page: int) -> dict: + if page == 1: + return { + "total_count": 101, + "workflow_runs": [FakeApproval.run(id=i, path=OTHER) for i in range(1, 101)], + } + return {"total_count": 101, "workflow_runs": [FakeApproval.run(id=1, path=OTHER)]} + + fake.inventory = inventory + with pytest.raises(GuardError, match="duplicate"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + + +def test_pagination_empty_page_and_bound_exceeded_fail_closed() -> None: + fake = FakeApproval() + + def incomplete(page: int) -> dict: + if page == 1: + return {"total_count": 2, "workflow_runs": [fake.run()]} + return {"total_count": 2, "workflow_runs": []} + + fake.inventory = incomplete + with pytest.raises(GuardError, match="incomplete"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + + def ten_short_pages(page: int) -> dict: + return { + "total_count": 150, + "workflow_runs": [FakeApproval.run(id=page * 100 + i, path=OTHER) for i in range(10)], + } + + fake = FakeApproval() + fake.inventory = ten_short_pages + with pytest.raises(GuardError, match="pagination bound"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + assert fake.actions_pages == list(range(1, 11)) + + +def test_runs_helper_rejects_count_change_and_duplicates() -> None: + class Scripted: + def get_json(self, path: str) -> dict: + page = int((parse_qs(urlparse(path).query).get("page") or ["1"])[0]) + if page == 1: + return { + "total_count": 3, + "workflow_runs": [FakeApproval.run(id=1), FakeApproval.run(id=2)], + } + return {"total_count": 4, "workflow_runs": [FakeApproval.run(id=3)]} + + with pytest.raises(GuardError, match="changed or exceeds"): + wa._runs(Scripted(), REPO, HEAD) + + class Dupes: + def get_json(self, path: str) -> dict: + return {"total_count": 2, "workflow_runs": [FakeApproval.run(), FakeApproval.run()]} + + with pytest.raises(GuardError, match="duplicate"): + wa._runs(Dupes(), REPO, HEAD) + + +# --- 403 / mutation / association failures ---------------------------------- + + +def test_approve_403_fails_loud_without_post_retry() -> None: + fake = FakeApproval() + fake.post_status = 403 + api = fake.api() + seen: list[dict] = [] + inner = api.request + + def wrapped(method: str, path: str, **kwargs: object): + if method.upper() == "POST" and "/approve" in str(path): + seen.append({"retry": kwargs.get("retry"), "path": path}) + return inner(method, path, **kwargs) + + api.request = wrapped # type: ignore[method-assign] + with pytest.raises(GuardError, match="HTTP 403"): + reconcile_pull(api, REPO, 1) + assert fake.posts == [101] + assert len(seen) == 1 + assert seen[0]["retry"] is False + assert fake.statuses[0]["state"] == "error" + + +def _add_maintainer_approval(fake: FakeApproval) -> None: + fake.files[(HEAD, ".github/a38.json")] = fake.files[(BASE, ".github/a38.json")] + fake.permissions["maintainer"] = {"permission": "write", "user": {"id": MAINTAINER}} + fake.reviews = [ + { + "id": 100, + "user": {"id": MAINTAINER, "login": "maintainer"}, + "state": "APPROVED", + "commit_id": HEAD, + "submitted_at": "2026-09-05T13:00:00Z", + "body": f"{POLICY_APPROVAL_PREFIX} head={HEAD} base={BASE}", + } + ] + + +@pytest.mark.parametrize("what", ["head", "base", "config", "report", "approval"]) +def test_mutation_before_write_is_caught_by_live_reassessment(what: str) -> None: + fake = FakeApproval() + if what == "approval": + fake.files[(HEAD, ".github/a38.json")] = fake.files[(BASE, ".github/a38.json")] + fake.permissions["maintainer"] = {"permission": "write", "user": {"id": MAINTAINER}} + fired = {"done": False} + + def mutate(f: FakeApproval, page: int) -> None: + if fired["done"] or page != 1: + return + fired["done"] = True + if what == "head": + f.pull["head"]["sha"] = BASE2 + elif what == "base": + f.pull["base"]["sha"] = BASE2 + elif what == "config": + # Same-SHA content edits are invisible: GitHubApi caches contents by commit. + f.ref_commits["develop"] = BASE2 + widened = _cfg({"enabled": True, "workflows": [PATH, OTHER]}) + f.files[(BASE2, ".github/pr-guard.json")] = json.dumps(widened).encode() + elif what == "report": + f.comments[0]["body"] += "\n" + else: + _add_maintainer_approval(f) + + fake.on_inventory = mutate + api = fake.api() + assessment = assess_pull(api, REPO, 1) + assert assessment.ok + assert assessment.status == "pass" + assert assessment.workflow_approval_enabled + with pytest.raises(GuardError, match="changed before workflow approval"): + wa.approve_workflow_runs(api, assessment) + assert fake.posts == [] + + +def test_get_run_identity_change_fails_closed() -> None: + fake = FakeApproval() + fake.run_override = fake.run(repository={"full_name": "other/public-app"}) + with pytest.raises(GuardError, match="identity changed"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + + +@pytest.mark.parametrize( + "what", + [ + "missing", + "not-list", + "ambiguous", + "cross-pr", + "cross-head", + "cross-base", + "predates", + ], +) +def test_missing_ambiguous_and_cross_pr_links_fail_closed(what: str) -> None: + fake = FakeApproval() + run = fake.runs[0] + if what == "missing": + del run["pull_requests"] + elif what == "not-list": + run["pull_requests"] = None + elif what == "ambiguous": + run["pull_requests"] = [_link(), _link(number=2)] + elif what == "cross-pr": + run["pull_requests"] = [_link(number=2)] + elif what == "cross-head": + run["pull_requests"] = [_link(head=BASE2)] + elif what == "cross-base": + run["pull_requests"] = [_link(base=BASE2)] + else: + run["pull_requests"] = [_link()] + run["created_at"] = "2026-08-01T00:00:00Z" + with pytest.raises(GuardError): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] + + +@pytest.mark.parametrize( + "what", + [ + "ambiguous-open", + "number-mismatch", + "list-sha-changed", + "diverged", + "behind", + "retargeted", + "force-pushed", + "reopened", + "compare-malformed", + "event-malformed", + ], +) +def test_private_fork_fallback_fails_closed_without_unique_stable_link(what: str) -> None: + fake = _private_fork() + if what == "ambiguous-open": + other = copy.deepcopy(fake.pull) + other["number"] = 2 + fake.extra_pulls = [other] + elif what == "number-mismatch": + listed = copy.deepcopy(fake.pull) + listed["number"] = 9 + fake.listed_pulls = [listed] + elif what == "list-sha-changed": + fake.list_head_sha = BASE2 + elif what == "diverged": + fake.comparison = "diverged" + elif what == "behind": + fake.comparison = "behind" + elif what == "retargeted": + fake.events = [{"event": "base_ref_changed", "created_at": "2026-09-05T11:30:00Z"}] + elif what == "force-pushed": + fake.events = [{"event": "base_ref_force_pushed", "created_at": "2026-09-05T11:30:00Z"}] + elif what == "reopened": + fake.events = [{"event": "reopened", "created_at": "2026-09-05T11:30:00Z"}] + elif what == "compare-malformed": + fake.comparison = "nope" + else: + fake.events = ["not-an-object"] + with pytest.raises(GuardError): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [] diff --git a/tests/test_workflow_approval_core.py b/tests/test_workflow_approval_core.py new file mode 100644 index 0000000..35984b8 --- /dev/null +++ b/tests/test_workflow_approval_core.py @@ -0,0 +1,212 @@ +"""Real guard reconciliation with an in-memory GitHub transport.""" +import copy +import json +from urllib.parse import urlparse, parse_qs + +import pytest + +from agent_cli.a38_guard import GuardError, reconcile_pull +from agent_cli.pr_guard_config import load_pr_guard_config, PrGuardConfigError +from test_a38_guard import FakeAPI, HEAD, BASE, BASE2, DEFAULT_TIP, REPO, _report_comment, _pr_guard_config + +pytestmark = pytest.mark.no_pg +PATH = ".github/workflows/test.yml" +FORK = "author/public-app" + + +class ApprovalAPI(FakeAPI): + def __init__(self): + super().__init__() + self.pull["head"]["repo"]["full_name"] = FORK + self.pull["head"]["ref"] = "feature" + self.pull["created_at"] = "2026-09-01T00:00:00Z" + self.config = _pr_guard_config() + self.config["workflow_approval"] = {"enabled": True, "workflows": [PATH]} + self.set_pr_guard_config(self.config) + self.add_author_report(_report_comment(), updated_at="2026-09-05T12:00:00Z", cid=21) + self.runs = [self.run()] + self.posts = [] + self.actions_gets = [] + self.post_status = 201 + self.before_run_read = None + self.extra_pulls = [] + self.events = [] + self.comparison = "ahead" + self.inventory = None + + @staticmethod + def run(**changes): + data = dict(id=101, path=PATH, workflow_id=10, event="pull_request", head_sha=HEAD, + head_branch="feature", repository={"full_name": REPO}, + head_repository={"full_name": FORK}, pull_requests=[], + status="completed", conclusion="action_required", run_attempt=1, + created_at="2026-09-05T11:00:00Z") + data.update(changes) + return data + + def request_fn(self, method, url, body=None): + path = urlparse(url).path + root = f"/repos/{REPO}" + if method == "GET" and path == root + "/actions/runs": + self.actions_gets.append(path) + data = self.inventory or {"total_count": len(self.runs), "workflow_runs": self.runs} + return 200, copy.deepcopy(data), {} + if path.startswith(root + "/actions/runs/"): + ident = int(path.split("/actions/runs/")[1].split("/")[0]) + run = next(r for r in self.runs if r["id"] == ident) + if method == "POST": + assert path.endswith("/approve"), "no rerun/dispatch/cancel endpoint allowed" + self.posts.append(ident) + if self.post_status == 201: + run.update(status="queued", conclusion=None) + return self.post_status, {}, {} + if self.before_run_read: + callback, self.before_run_read = self.before_run_read, None + callback(self) + return 200, copy.deepcopy(run), {} + if method == "GET" and path == root + "/pulls": + return 200, copy.deepcopy([self.pull, *self.extra_pulls]), {} + if method == "GET" and path.startswith(root + "/compare/"): + return 200, {"status": self.comparison}, {} + if method == "GET" and path == root + "/issues/1/events": + return 200, copy.deepcopy(self.events), {} + status, data, headers = super().request_fn(method, url, body) + return status, copy.deepcopy(data), headers + + +def test_real_reconcile_approves_initial_private_fork_run_and_is_idempotent(): + fake = ApprovalAPI() + result = reconcile_pull(fake.api(), REPO, 1) + assert result.ok + assert result.workflow_approvals == [{"run_id": 101, "workflow": PATH, "head": HEAD, "status": "approved"}] + assert "workflow:approve:101" in result.writes + assert fake.posts == [101] + assert reconcile_pull(fake.api(), REPO, 1).workflow_approvals == [] + assert fake.posts == [101] + + +def test_dry_run_previews_without_any_writes(): + fake = ApprovalAPI() + result = reconcile_pull(fake.api(), REPO, 1, dry_run=True) + assert result.workflow_approvals[0]["status"] == "planned" + assert not fake.posts and not fake.writes + + +@pytest.mark.parametrize("case", ["missing", "disabled", "report", "same_repo", "closed", "observe", "exclude"]) +def test_no_approval_or_actions_inventory_without_authorization(case): + fake = ApprovalAPI() + if case == "missing": + fake.set_pr_guard_config(None) + elif case == "disabled": + fake.config["workflow_approval"]["enabled"] = False + fake.set_pr_guard_config(fake.config) + elif case == "report": + fake.comments.clear() + elif case == "same_repo": + fake.pull["head"]["repo"]["full_name"] = REPO + elif case == "closed": + fake.pull["state"] = "closed" + elif case == "observe": + policy = json.loads(fake.files[(BASE, ".github/a38.json")]) + policy["mode"] = "observe" + fake.files[(BASE, ".github/a38.json")] = json.dumps(policy).encode() + else: + fake.config["a38"]["default"] = "exclude" + fake.set_pr_guard_config(fake.config) + result = reconcile_pull(fake.api(), REPO, 1) + assert result.workflow_approvals == [] + assert not fake.posts and not fake.actions_gets + + +@pytest.mark.parametrize("changes", [ + {"status": "queued", "conclusion": None}, {"conclusion": "success"}, {"conclusion": "failure"}, + {"run_attempt": 2}, {"event": "push"}, {"head_sha": BASE}, {"head_branch": "other"}, + {"repository": {"full_name": "other/repo"}}, {"head_repository": {"full_name": "other/fork"}}, + {"path": ".github/workflows/unknown.yml"}, +]) +def test_ineligible_runs_are_never_approved(changes): + fake = ApprovalAPI() + fake.runs = [fake.run(**changes)] + reconcile_pull(fake.api(), REPO, 1) + assert not fake.posts + + +@pytest.mark.parametrize("changes", [{"conclusion": "success"}, {"conclusion": "failure"}, + {"status": "queued", "conclusion": None}, {"run_attempt": 2}]) +def test_newer_run_suppresses_old_blocked_run_even_when_not_successful(changes): + fake = ApprovalAPI() + fake.runs.append(fake.run(id=102, created_at="2026-09-05T11:01:00Z", **changes)) + reconcile_pull(fake.api(), REPO, 1) + assert not fake.posts + + +def test_permission_denial_fails_and_is_not_retried(): + fake = ApprovalAPI() + fake.post_status = 403 + with pytest.raises(GuardError, match="Actions write"): + reconcile_pull(fake.api(), REPO, 1) + assert fake.posts == [101] + assert fake.statuses[0]["state"] == "error" + + +@pytest.mark.parametrize("what", ["head", "base", "config", "report"]) +def test_change_before_write_rejects_stale_authorization(what): + fake = ApprovalAPI() + def mutate(f): + if what == "head": + f.pull["head"]["sha"] = BASE2 + elif what == "base": + f.pull["base"]["sha"] = BASE2 + elif what == "config": + f.ref_commits["develop"] = BASE2 + else: + f.comments[0]["body"] += "\nChanged author evidence" + fake.before_run_read = mutate + with pytest.raises(GuardError): + reconcile_pull(fake.api(), REPO, 1) + assert not fake.posts + + +@pytest.mark.parametrize("what", ["ambiguous", "not_rebased", "retargeted", "reopened", "older", "foreign_link"]) +def test_private_fork_association_must_be_proven(what): + fake = ApprovalAPI() + if what == "ambiguous": + other = copy.deepcopy(fake.pull) + other["number"] = 2 + fake.extra_pulls = [other] + elif what == "not_rebased": + fake.comparison = "diverged" + elif what in {"retargeted", "reopened"}: + fake.events = [{"event": "base_ref_changed" if what == "retargeted" else "reopened", + "created_at": "2026-09-05T11:30:00Z"}] + elif what == "older": + fake.runs[0]["created_at"] = "2026-08-01T00:00:00Z" + else: + fake.runs[0]["pull_requests"] = [{"number": 2, "head": {"sha": HEAD}, "base": {"sha": BASE}}] + with pytest.raises(GuardError): + reconcile_pull(fake.api(), REPO, 1) + assert not fake.posts + + +@pytest.mark.parametrize("inventory", [{"total_count": 1000, "workflow_runs": []}, + {"total_count": 2, "workflow_runs": []}, {"total_count": True, "workflow_runs": []}, + {"total_count": 1, "workflow_runs": [{}]}, + {"total_count": 2, "workflow_runs": [ApprovalAPI.run(), ApprovalAPI.run()]}]) +def test_partial_or_malformed_inventory_cannot_authorize(inventory): + fake = ApprovalAPI() + fake.inventory = inventory + with pytest.raises(GuardError): + reconcile_pull(fake.api(), REPO, 1) + assert not fake.posts + + +@pytest.mark.parametrize("approval", [None, {}, {"enabled": "true", "workflows": [PATH]}, + {"enabled": True, "workflows": []}, {"enabled": True, "workflows": [PATH, PATH]}, + {"enabled": True, "workflows": [".github/workflows/*.yml"]}, + {"enabled": True, "workflows": [".github/workflows/../x.yml"]}, + {"enabled": True, "workflows": [PATH], "override": True}]) +def test_invalid_approval_config_fails_closed(approval): + config = _pr_guard_config() + config["workflow_approval"] = approval + with pytest.raises(PrGuardConfigError): + load_pr_guard_config(json.dumps(config))