orchestrator/agent-team/agent_team/ci_gate.py

"""Pure-code CI pass/fail gate — the deterministic block decision (design §3.3.2).

This module is the §3.3.2 boundary #4: *"Pass/fail is a pure-code gate over
authenticated CI results, not the LLM verifier."* Mirroring secrev's "one
pure-code script owns the block decision", the gate reads the CI run
**conclusion** (already fetched, authenticated as the box read-only PAT via the
GitHub Checks/Actions API) keyed to a specific ``run_id`` + ``diff_hash`` and
makes a deterministic ``pass | fail | block`` decision. It consumes **only**
that authenticated, patch-independent conclusion; it never trusts a
success/failure file or artifact the patch could have written.

The gate also enforces the two box-side/CI defences that must hold before a
diff is even allowed to build (defence in depth — the CI side enforces the same
checks as a hard fail):

* **boundary #2** — the **trust-control-surface denylist**: a candidate diff
  may not touch ``.github/workflows/**``, IAM/policy IaC, branch-protection /
  ``CODEOWNERS`` / Dependabot config, or paths outside the task's declared
  scope. A match is canonicalized (symlink/``..`` resolution) and rejects
  renames into denied paths, so a path match cannot be bypassed by indirection.
* **boundary #3** — **diff integrity**: the gate recomputes the candidate diff
  hash and compares it to the ledger-recorded hash and to the hash CI verified,
  so a tampered or substituted diff fails closed.

This module is pure stdlib and imports the committed foundation contracts
verbatim (it redefines none of them). It performs **no** network I/O: the
authenticated CI conclusion is passed in as data (the caller fetches it via the
read-only PAT). Keeping the gate I/O-free is what makes the block decision
deterministic and unit-testable.

Decision semantics (a diff "ships" only on an unambiguous authenticated pass):

* :data:`GateDecision.PASS`  — the run concluded ``success`` for the exact
  ``run_id``/``diff_hash`` and every guard held; the verifier may open a draft
  PR.
* :data:`GateDecision.FAIL`  — the run concluded a recognised failure
  (``failure``/``timed_out``/``cancelled``/...); the verifier loops back to the
  builders.
* :data:`GateDecision.BLOCK` — a trust violation (denylist hit, hash mismatch,
  run-id mismatch, missing/ambiguous authenticated conclusion). This is an
  ALARM-worthy refuse-to-proceed, never a silent pass.

TRUST SOURCE (§3.3.2, QUESTION-1). ``state["run_id"]`` (the run id the fetcher
reads and the value compared against ``expected_run_id`` here) and
``state["diff_hash"]`` MUST be written ONLY by the trusted dispatcher / ledger at
dispatch time — NEVER by an LLM / builder / verifier node or by anything a
candidate diff can influence. The dispatcher records the run id it dispatched the
apply/verify workflow under (and the ledger-recorded diff hash); the LLM nodes
only read them. The gate's run-id EQUALITY check below (``actual_run_id`` vs the
dispatcher-supplied ``expected_run_id``) plus the fetcher's run_id format
validation are the defense if that assumption is ever broken: a tampered
``state["run_id"]`` would still have to equal the dispatcher's expected run id to
pass, and a malformed value fails closed.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from enum import Enum
from pathlib import PurePosixPath
from typing import Any, Mapping, Sequence

from agent_team.state_store import compute_content_hash

__all__ = [
    "DENYLIST_GLOBS",
    "CiGateError",
    "GateDecision",
    "GateResult",
    "denylist_violations",
    "diff_touched_paths",
    "evaluate_ci_gate",
    "verify_diff_hash",
]


class CiGateError(Exception):
    """Raised when the gate is called with structurally invalid inputs.

    Distinct from a :data:`GateDecision.BLOCK`: a ``BLOCK`` is a *valid* gate
    run that found a trust violation, whereas this exception means the caller
    handed the gate malformed data (e.g. a non-string diff). Fails loud rather
    than guessing.
    """


class GateDecision(Enum):
    """The deterministic gate outcome (§3.3.2 boundary #4)."""

    PASS = "pass"
    FAIL = "fail"
    BLOCK = "block"


# Trust-control-surface denylist (§3.3.2 boundary #2). A candidate diff that
# touches any of these is escalated to mandatory human + GPT cross-review, never
# auto-built — they are the mandatory-cross-review surface regardless. Globs are
# matched against POSIX-canonicalized repo-relative paths.
# INJ-03: this denylist is the UNION SUPERSET shared verbatim across all three
# trust-control copies — this tuple, the guard-job inline DENY_GLOBS, and the
# post-build inline DENY_GLOBS in ci/agent-team-apply-verify.yml. The three had
# drifted in BOTH directions (each carried entries the others lacked); they are
# now identical, and tests/test_apply_verify_workflow_hardening.py asserts the
# identity so any future drift fails CI. When editing one, edit all three.
DENYLIST_GLOBS: tuple[str, ...] = (
    # CI workflow definitions — the "pwn request" surface.
    ".github/workflows/**",
    ".github/actions/**",
    # Branch protection / ownership / dependency automation config.
    ".github/CODEOWNERS",
    "**/CODEOWNERS",
    "CODEOWNERS",
    ".github/dependabot.yml",
    ".github/dependabot.yaml",
    ".github/settings.yml",
    # IAM / policy IaC (CDK / SAM / Terraform / CloudFormation).
    "**/cdk.json",
    "**/template.yml",
    "**/template.yaml",
    "**/samconfig.toml",
    "**/*.tf",
    "**/*-stack.ts",
    "**/*_stack.py",
    "**/iam/**",
    "**/policies/**",
    "**/*iam*",
    "**/policy*.json",
    "**/*policy*.json",
    "**/*.pem",
    "**/*.key",
)

# Authenticated GitHub run conclusions that count as a recognised failure (the
# verifier loops back). Anything not in PASS/this set is ambiguous -> BLOCK.
_FAILURE_CONCLUSIONS: frozenset[str] = frozenset(
    {"failure", "timed_out", "cancelled", "action_required", "stale", "startup_failure"}
)

# The single authenticated conclusion that means "ship-able".
_SUCCESS_CONCLUSION = "success"

# ``git diff`` file-header line, e.g. ``diff --git a/foo.py b/foo.py``. We read
# the post-image (``b/``) path as the touched path and also surface the pre-image
# (``a/``) so a *rename into* a denied path is caught (boundary #2).
_DIFF_GIT_RE = re.compile(r"^diff --git a/(?P<a>.+?) b/(?P<b>.+?)\s*$")
# ``rename from``/``rename to`` lines carry the rename source/target explicitly.
_RENAME_FROM_RE = re.compile(r"^rename from (?P<path>.+?)\s*$")
_RENAME_TO_RE = re.compile(r"^rename to (?P<path>.+?)\s*$")


@dataclass
class GateResult:
    """The gate's deterministic verdict plus the evidence behind it.

    ``decision`` is the block decision the verifier node acts on. ``reasons``
    enumerates every concrete trigger (denylist hits, hash mismatch, the CI
    conclusion consumed) so the decision is auditable and an ALARM can quote it.
    ``run_id``/``diff_hash`` echo the keys the gate was bound to (provenance,
    §3.3.2). ``ci_conclusion`` is the authenticated conclusion actually
    consumed.
    """

    decision: GateDecision
    reasons: list[str] = field(default_factory=list)
    run_id: str | None = None
    diff_hash: str | None = None
    ci_conclusion: str | None = None

    @property
    def passed(self) -> bool:
        """``True`` only on an unambiguous authenticated pass."""
        return self.decision is GateDecision.PASS

    @property
    def blocked(self) -> bool:
        """``True`` on a trust violation (ALARM-worthy refuse-to-proceed)."""
        return self.decision is GateDecision.BLOCK


def _canonical_repo_path(raw: str) -> str:
    """Canonicalize a repo-relative path for denylist matching (§3.3.2).

    Strips a leading ``a/``/``b/`` git prefix, normalizes separators, resolves
    ``.``/``..`` segments without touching the filesystem (the diff describes
    paths that may not exist locally), and drops a leading ``/`` so the result
    is always repo-relative. A path that escapes the repo root via ``..`` is
    returned with a sentinel ``..`` prefix preserved so it cannot silently match
    *nothing* — the caller treats an escaping path as a denylist hit.
    """
    text = raw.strip().strip('"')
    # Drop a single git a//b/ prefix if present.
    if text.startswith(("a/", "b/")):
        text = text[2:]
    # PurePosixPath normalizes separators; resolve . and .. logically.
    parts: list[str] = []
    for segment in PurePosixPath(text).parts:
        if segment in ("", "."):
            continue
        if segment == "..":
            # Escaping the repo root — keep the marker so it never matches a
            # benign glob and is treated as suspicious by the caller.
            parts.append("..")
            continue
        parts.append(segment)
    return "/".join(parts)


def diff_touched_paths(unified_diff: str) -> list[str]:
    """Extract the canonicalized repo-relative paths a unified diff touches.

    Reads ``diff --git`` headers (both the ``a/`` pre-image and ``b/``
    post-image) plus explicit ``rename from``/``rename to`` lines, so a rename
    *into* a denied path is surfaced (§3.3.2 boundary #2 — "rejects renames into
    denied paths"). Returns a de-duplicated, sorted list of canonical paths.

    Raises :class:`CiGateError` if ``unified_diff`` is not a string.
    """
    if not isinstance(unified_diff, str):
        raise CiGateError(f"diff must be str, got {type(unified_diff).__name__}")

    touched: set[str] = set()
    for line in unified_diff.splitlines():
        m = _DIFF_GIT_RE.match(line)
        if m is not None:
            touched.add(_canonical_repo_path(m.group("a")))
            touched.add(_canonical_repo_path(m.group("b")))
            continue
        m = _RENAME_FROM_RE.match(line)
        if m is not None:
            touched.add(_canonical_repo_path(m.group("path")))
            continue
        m = _RENAME_TO_RE.match(line)
        if m is not None:
            touched.add(_canonical_repo_path(m.group("path")))
    touched.discard("")
    return sorted(touched)


def _glob_to_regex(glob: str) -> re.Pattern[str]:
    """Compile a denylist glob to an anchored regex.

    Supports ``**`` (any number of path segments, including zero), ``*`` (within
    a single segment), and ``?``. Everything else is matched literally. Matching
    is done on canonical POSIX repo-relative paths.
    """
    out: list[str] = ["^"]
    i = 0
    n = len(glob)
    while i < n:
        ch = glob[i]
        if ch == "*":
            if i + 1 < n and glob[i + 1] == "*":
                # ``**`` — any chars incl. ``/``. Swallow an optional trailing
                # ``/`` so ``dir/**`` also matches ``dir`` itself's children
                # without requiring a separator artifact.
                out.append(".*")
                i += 2
                if i < n and glob[i] == "/":
                    i += 1
                continue
            # single ``*`` — anything but a path separator.
            out.append("[^/]*")
            i += 1
            continue
        if ch == "?":
            out.append("[^/]")
            i += 1
            continue
        out.append(re.escape(ch))
        i += 1
    out.append("$")
    return re.compile("".join(out))


_DENYLIST_RES: tuple[tuple[str, re.Pattern[str]], ...] = tuple(
    (g, _glob_to_regex(g)) for g in DENYLIST_GLOBS
)


def denylist_violations(
    unified_diff: str,
    *,
    allowed_scope: Sequence[str] | None = None,
) -> list[str]:
    """Return the trust-control-surface violations in ``unified_diff`` (§3.3.2).

    A violation is any touched path that (a) matches a :data:`DENYLIST_GLOBS`
    entry, (b) escapes the repo root via ``..`` (path-indirection attempt), or
    (c) — when ``allowed_scope`` is given — falls outside the task's declared
    scope. ``allowed_scope`` is a sequence of canonical path prefixes
    (directories or exact files) the task is allowed to modify; a touched path
    outside every prefix is a violation ("files outside the task's declared
    scope").

    Returns a sorted list of human-readable reason strings; an empty list means
    the diff is clean for the denylist boundary.
    """
    violations: list[str] = []
    normalized_scope = (
        [_canonical_repo_path(p) for p in allowed_scope]
        if allowed_scope is not None
        else None
    )

    for path in diff_touched_paths(unified_diff):
        if ".." in PurePosixPath(path).parts:
            violations.append(f"path escapes repo root via '..': {path!r}")
            continue
        for glob, pattern in _DENYLIST_RES:
            if pattern.match(path):
                violations.append(f"denylisted path {path!r} matches glob {glob!r}")
                break
        else:
            if normalized_scope is not None and not _within_scope(
                path, normalized_scope
            ):
                violations.append(f"path {path!r} is outside the task's declared scope")
    return sorted(violations)


def _within_scope(path: str, scope: Sequence[str]) -> bool:
    """Return ``True`` if ``path`` is within one declared-scope prefix."""
    candidate = PurePosixPath(path)
    for prefix in scope:
        if not prefix:
            continue
        if path == prefix:
            return True
        prefix_path = PurePosixPath(prefix)
        try:
            candidate.relative_to(prefix_path)
            return True
        except ValueError:
            continue
    return False


def verify_diff_hash(
    candidate_diff: str,
    *,
    ledger_hash: str | None,
    ci_verified_hash: str | None = None,
) -> bool:
    """Verify the candidate diff hash matches the ledger (and CI) (§3.3.2 #3).

    Recomputes the content hash of ``candidate_diff`` (sha256, via the committed
    :func:`agent_team.state_store.compute_content_hash`) and compares it to the
    ``ledger_hash`` recorded by the builder and, when supplied, to the
    ``ci_verified_hash`` CI checked before applying. Comparison is
    constant-time. Returns ``True`` only when all supplied hashes agree; a
    ``None`` ledger hash is treated as a failure (there is nothing to bind to).

    Raises :class:`CiGateError` if ``candidate_diff`` is not a string.
    """
    if not isinstance(candidate_diff, str):
        raise CiGateError(
            f"candidate_diff must be str, got {type(candidate_diff).__name__}"
        )
    if not ledger_hash:
        return False

    actual = compute_content_hash(candidate_diff.encode("utf-8"))
    if not _consteq(actual, ledger_hash):
        return False
    if ci_verified_hash is not None and not _consteq(actual, ci_verified_hash):
        return False
    return True


def _consteq(a: str, b: str) -> bool:
    """Constant-time string compare (hashes are not secret, but be tidy)."""
    if len(a) != len(b):
        return False
    result = 0
    for x, y in zip(a, b):
        result |= ord(x) ^ ord(y)
    return result == 0


def evaluate_ci_gate(
    *,
    candidate_diff: str,
    ledger_hash: str | None,
    ci_result: Mapping[str, Any] | None,
    expected_run_id: str,
    allowed_scope: Sequence[str] | None = None,
) -> GateResult:
    """Make the deterministic pass/fail/block decision (§3.3.2 boundary #4).

    Inputs (all authenticated/patch-independent — the gate does no I/O):

    * ``candidate_diff`` — the builder's diff text (re-hashed here).
    * ``ledger_hash`` — the hash the builder recorded in the task ledger.
    * ``ci_result`` — the authenticated CI conclusion the caller fetched via the
      read-only PAT (GitHub Checks/Actions API). The gate reads only
      ``run_id``, ``conclusion``, and (optionally) ``diff_hash`` from it; it
      **never** reads a patch-written success file/artifact.
    * ``expected_run_id`` — the run id the verifier dispatched for this exact
      diff; the conclusion must be keyed to it (a stale/substituted run id is a
      BLOCK).
    * ``allowed_scope`` — optional declared-scope prefixes for the task.

    Decision order (a trust violation always wins over a CI verdict):

    1. **Denylist / scope** — any violation -> :data:`GateDecision.BLOCK`.
    2. **Diff-hash integrity** — hash mismatch (ledger or CI-verified) ->
       ``BLOCK``.
    3. **Authenticated conclusion** — missing result, a ``run_id`` that does not
       match ``expected_run_id``, or an unrecognised/ambiguous conclusion ->
       ``BLOCK``; a recognised failure -> :data:`GateDecision.FAIL`; ``success``
       -> :data:`GateDecision.PASS`.

    Returns a :class:`GateResult` with the decision and the reasons behind it.
    Raises :class:`CiGateError` on structurally invalid inputs.
    """
    if not isinstance(expected_run_id, str) or not expected_run_id:
        raise CiGateError("expected_run_id must be a non-empty string")

    reasons: list[str] = []

    # (1) Trust-control-surface denylist + declared scope. Highest priority:
    # these files are the mandatory-cross-review surface, never auto-built.
    violations = denylist_violations(candidate_diff, allowed_scope=allowed_scope)
    if violations:
        reasons.extend(violations)
        return GateResult(
            decision=GateDecision.BLOCK,
            reasons=reasons,
            run_id=expected_run_id,
            diff_hash=ledger_hash,
            ci_conclusion=None,
        )

    # (2) Diff-hash integrity: bind the decision to the exact bytes the ledger
    # recorded (and, if present, what CI verified before applying).
    ci_verified_hash = (
        str(ci_result.get("diff_hash"))
        if ci_result is not None and ci_result.get("diff_hash") is not None
        else None
    )
    if not verify_diff_hash(
        candidate_diff,
        ledger_hash=ledger_hash,
        ci_verified_hash=ci_verified_hash,
    ):
        reasons.append(
            "diff-hash mismatch: candidate diff does not match the ledger"
            + (" / CI-verified" if ci_verified_hash is not None else "")
            + " hash"
        )
        return GateResult(
            decision=GateDecision.BLOCK,
            reasons=reasons,
            run_id=expected_run_id,
            diff_hash=ledger_hash,
            ci_conclusion=None,
        )

    # (3) Authenticated, patch-independent CI conclusion.
    if ci_result is None:
        reasons.append("no authenticated CI result supplied")
        return GateResult(
            decision=GateDecision.BLOCK,
            reasons=reasons,
            run_id=expected_run_id,
            diff_hash=ledger_hash,
            ci_conclusion=None,
        )

    actual_run_id = ci_result.get("run_id")
    if str(actual_run_id) != expected_run_id:
        reasons.append(
            f"CI run-id mismatch: expected {expected_run_id!r}, "
            f"conclusion is keyed to {actual_run_id!r}"
        )
        return GateResult(
            decision=GateDecision.BLOCK,
            reasons=reasons,
            run_id=expected_run_id,
            diff_hash=ledger_hash,
            ci_conclusion=None,
        )

    conclusion = ci_result.get("conclusion")
    normalized = str(conclusion).strip().lower() if conclusion is not None else None

    if normalized == _SUCCESS_CONCLUSION:
        reasons.append("authenticated CI conclusion: success")
        return GateResult(
            decision=GateDecision.PASS,
            reasons=reasons,
            run_id=expected_run_id,
            diff_hash=ledger_hash,
            ci_conclusion=normalized,
        )

    if normalized in _FAILURE_CONCLUSIONS:
        reasons.append(f"authenticated CI conclusion: {normalized}")
        return GateResult(
            decision=GateDecision.FAIL,
            reasons=reasons,
            run_id=expected_run_id,
            diff_hash=ledger_hash,
            ci_conclusion=normalized,
        )

    # Unknown / null / still-running conclusion: refuse to proceed (never a
    # silent pass). e.g. ``None`` (in-progress), ``"neutral"``, ``"skipped"``.
    reasons.append(
        f"unrecognised/ambiguous CI conclusion {conclusion!r}; refusing to proceed"
    )
    return GateResult(
        decision=GateDecision.BLOCK,
        reasons=reasons,
        run_id=expected_run_id,
        diff_hash=ledger_hash,
        ci_conclusion=normalized,
    )