"""Real, read-only CI-result fetcher — the GATED-LIVE seam for VERIFY (§3.3.2 #4). This module implements the production :data:`~agent_team.nodes.build_verify_subgraph.CiResultFetcher` that the build->verify subgraph's VERIFY node injects once the §3.3.2 CI trust-boundary clears ``/sh-security-review`` + the GPT-4.1 cross-review. Until then the subgraph runs with the INERT default fetcher (``_no_ci_result`` -> the gate BLOCKs and the task parks); binding THIS fetcher only gives the pure-code gate (:func:`agent_team.ci_gate.evaluate_ci_gate`) an authenticated conclusion to read — it never makes the LLM the pass authority. What it is (and, just as importantly, what it is NOT): * It is a **pure DATA fetcher.** Given the task ``state``, it reads ``state["run_id"]`` (and echoes ``state.get("diff_hash")``), calls the GitHub Actions REST API ``GET /repos/{owner}/{repo}/actions/runs/{run_id}`` with a **READ-ONLY** token, and returns the authenticated run ``conclusion`` as a mapping ``{"run_id", "conclusion", "diff_hash"}``. The gate owns the verdict; this module never derives pass/fail itself. * It is **fail-closed.** ANY error — missing ``run_id``, missing token, 404, auth failure, malformed JSON, a network/timeout error, an unexpected status — returns ``None``. A ``None`` result makes the gate BLOCK (never a silent pass), so a broken fetch parks the task for a human rather than shipping. * It **never writes.** No ``POST``/``PATCH``, no ``git``, no patch apply, no filesystem mutation. It performs exactly one read-only GET. * It uses **no cloud credentials and no token-federation** — only a GitHub read-only token, resolved at CALL time from the environment (matching the prevailing transport idiom in :mod:`agent_team.transport.github_live`). It NEVER reads a success/failure file the patch could have written (boundary 4). Prevailing HTTP approach: mirrors :mod:`agent_team.transport.github_live` — a thin ``requests`` session, deferred import (``requests`` is optional and may be absent pre-deploy), call-time token resolution, and an injectable ``client`` for testability. Unlike the poster, a missing token / missing ``requests`` here does NOT raise: it fails closed to ``None`` so the gate BLOCKs (the fetcher's whole contract is "no authenticated result -> None"). The dedicated read-only env var ``AGENT_TEAM_CI_READ_TOKEN`` is preferred, falling back to ``GITHUB_TOKEN``. """ from __future__ import annotations import logging import os from collections.abc import Mapping from typing import Any from agent_team.nodes.build_verify_subgraph import CiResultFetcher __all__ = [ "CI_READ_TOKEN_ENV", "GITHUB_API_ROOT", "build_ci_result_fetcher", "fetch_ci_result", ] _LOG = logging.getLogger("agent_team.ci_fetcher") # The dedicated read-only token env var (preferred), falling back to the generic # GITHUB_TOKEN (the idiom github_live uses). The token MUST be read-only — the # fetcher only ever GETs; a write-scoped token here would be unnecessary blast # radius (provisioning issues a read-only fine-grained PAT / read-only var). CI_READ_TOKEN_ENV = "AGENT_TEAM_CI_READ_TOKEN" _FALLBACK_TOKEN_ENV = "GITHUB_TOKEN" GITHUB_API_ROOT = "https://api.github.com" # Conservative default timeout for the single read-only GET. A hang must fail # closed (-> None -> gate BLOCK), never wedge the verifier. _DEFAULT_TIMEOUT_S = 15.0 def _resolve_read_token() -> str | None: """Resolve the read-only GitHub token at call time, or ``None``. Prefers :data:`CI_READ_TOKEN_ENV`, falls back to ``GITHUB_TOKEN``. Returns ``None`` when neither is set so the fetcher fails closed (the caller maps a missing token to a ``None`` result -> gate BLOCK), rather than raising. """ return os.environ.get(CI_READ_TOKEN_ENV) or os.environ.get(_FALLBACK_TOKEN_ENV) def _build_session(token: str) -> Any: """Lazily construct a read-only ``requests.Session`` (deferred optional import). Mirrors :func:`agent_team.transport.github_live._build_session` (deferred ``requests`` import, ``Authorization: Bearer`` + the API-version header). The session is used for exactly one GET; no write verbs are ever issued. Raises :class:`RuntimeError` only if ``requests`` is unavailable — the caller catches it and fails closed to ``None`` so a pre-deploy environment without the optional dependency simply BLOCKs (never a spurious pass). """ import requests # deferred: optional dependency (see module docstring) session = requests.Session() session.headers.update( { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", } ) return session def fetch_ci_result( state: Mapping[str, Any], *, owner: str, repo: str, client: Any = None, api_root: str = GITHUB_API_ROOT, timeout: float = _DEFAULT_TIMEOUT_S, ) -> dict[str, Any] | None: """Fetch the authenticated CI run conclusion as DATA, or ``None`` (fail-closed). Reads ``state["run_id"]`` (required) and echoes ``state.get("diff_hash")``, then GETs ``{api_root}/repos/{owner}/{repo}/actions/runs/{run_id}`` with the read-only token and returns:: {"run_id": , "conclusion": , "diff_hash": } Returns ``None`` on ANY failure — no ``run_id`` in state, no resolvable token, ``requests`` unavailable, a non-2xx status (404/401/403/...), a malformed/absent JSON body, a missing ``conclusion``, or a network/timeout error. ``None`` is the fail-closed signal: the pure-code gate treats it as "no authenticated result" and BLOCKs, so a broken fetch parks the task. This function NEVER derives the verdict (the gate owns pass/fail) and NEVER writes. ``client`` injects a pre-built ``requests``-like session for tests (any object with ``get(url, *, timeout) -> response`` exposing ``status_code`` and ``json()``). When omitted, a read-only session is built from the resolved token. """ run_id = state.get("run_id") if run_id is None or str(run_id) == "": _LOG.warning("ci_fetcher: no run_id in state; failing closed to None") return None run_id = str(run_id) echoed_hash = state.get("diff_hash") if client is None: token = _resolve_read_token() if not token: _LOG.warning( "ci_fetcher: no read-only token (%s/%s) set; failing closed to None", CI_READ_TOKEN_ENV, _FALLBACK_TOKEN_ENV, ) return None try: client = _build_session(token) except Exception: # noqa: BLE001 - any build failure fails closed _LOG.warning("ci_fetcher: could not build HTTP client; failing closed") return None url = f"{api_root.rstrip('/')}/repos/{owner}/{repo}/actions/runs/{run_id}" try: response = client.get(url, timeout=timeout) except Exception: # noqa: BLE001 - timeout / connection / any -> fail closed _LOG.warning("ci_fetcher: GET %s failed (network/timeout); failing closed", url) return None status = _status_of(response) if status is None or not (200 <= status < 300): _LOG.warning("ci_fetcher: run GET returned status %r; failing closed", status) return None data = _json_of(response) if not isinstance(data, dict): _LOG.warning("ci_fetcher: run body was not a JSON object; failing closed") return None conclusion = data.get("conclusion") # An in-progress run has conclusion=None; that is NOT an authenticated # verdict, so fail closed (the gate would BLOCK on it anyway, but returning # None keeps the "no result" contract clean and avoids echoing a non-verdict). if conclusion is None or not isinstance(conclusion, str): _LOG.info("ci_fetcher: run %s has no conclusion yet; failing closed", run_id) return None # Bind the returned run_id to the run actually fetched (the API echoes id); # fall back to the requested run_id. The gate independently re-checks this # against expected_run_id, so this is provenance, not the trust decision. fetched_id = data.get("id") result_run_id = str(fetched_id) if fetched_id is not None else run_id return { "run_id": result_run_id, "conclusion": conclusion, "diff_hash": echoed_hash, } def build_ci_result_fetcher( *, owner: str, repo: str, client: Any = None, api_root: str = GITHUB_API_ROOT, timeout: float = _DEFAULT_TIMEOUT_S, ) -> CiResultFetcher: """Build a :data:`CiResultFetcher` bound to ``owner``/``repo`` (GATED-LIVE seam). Returns a single-argument ``state -> mapping | None`` callable shaped exactly like the VERIFY node's injected ``ci_result_fetcher`` seam, closing over the target ``owner``/``repo`` (and the optional injected ``client`` / ``api_root`` / ``timeout``). Compose it with :func:`agent_team.nodes.build_verify_subgraph.bind_ci_result_fetcher` (or the opt-in :func:`agent_team.coordinator.gated_build_verify_wiring`) once the §3.3.2 gate clears. It is read-only and fails closed (see :func:`fetch_ci_result`); binding it does not enable any apply/verify behaviour, it only gives the gate an authenticated conclusion to read. """ def fetcher(state: Mapping[str, Any]) -> dict[str, Any] | None: return fetch_ci_result( state, owner=owner, repo=repo, client=client, api_root=api_root, timeout=timeout, ) return fetcher def _status_of(response: Any) -> int | None: """Read the HTTP status from a ``requests``-like response, or ``None``. Accepts ``status_code`` (``requests``) or ``status`` (a minimal fake). Returns ``None`` if neither is present so the caller fails closed rather than raising on an exotic object. """ for attr in ("status_code", "status"): value = getattr(response, attr, None) if value is not None: try: return int(value) except (TypeError, ValueError): return None return None def _json_of(response: Any) -> Any: """Parse the JSON body of a ``requests``-like response, or ``None`` on failure. A malformed/absent body must fail closed (-> ``None`` -> the caller returns ``None`` -> gate BLOCK), never raise into the verifier node. """ parser = getattr(response, "json", None) if not callable(parser): return None try: return parser() except Exception: # noqa: BLE001 - any JSON decode error fails closed return None