"""Compatibility hardening for Gitea validation discovery and request latency. Native Gitea PR runs can omit their head branch. This client reads actual exact-head runs without a branch filter and checks available PR identity. Older status-only adapters retain an explicitly labeled legacy fallback; exact-head native reads never synthesize a workflow run or merge proof. Large repositories can require longer than the historical 15-second client budget to commit files, apply changesets, or materialize a server-side squash merge. The resilient client therefore uses the already allowlisted 60-second maximum from ``GiteaSettings``. All HTTP method allowlisting, redirect handling, response budgets, access checks and non-forcing merge behavior remain in the base client and are not duplicated here. """ from __future__ import annotations import re import time from typing import Any import httpx from .gitea import GiteaClient as BaseGiteaClient from .gitea import GiteaError, GiteaSettings, RegisteredRepository, TokenProvider GITEA_REQUEST_TIMEOUT_SECONDS = 60.0 class GiteaClient(BaseGiteaClient): """Gitea client with exact-head validation fallback and bounded latency.""" def __init__( self, settings: GiteaSettings, token_provider: TokenProvider, *, transport: httpx.BaseTransport | None = None, ) -> None: resilient_settings = settings.model_copy( update={"timeout_seconds": GITEA_REQUEST_TIMEOUT_SECONDS} ) super().__init__(resilient_settings, token_provider, transport=transport) @staticmethod def matches_exact_head_action_run( run: dict[str, Any], *, branch: str, head_sha: str, pull_request_number: int | None = None, ) -> bool: if run.get("head_sha") != head_sha: return False head_branch = run.get("head_branch", "") event = run.get("event") if head_branch is None and event == "pull_request": head_branch = "" if event == "workflow_dispatch" and head_branch == f"refs/heads/{branch}": head_branch = branch if not isinstance(head_branch, str) or head_branch not in {"", branch}: return False if not head_branch and event != "pull_request": return False path = run.get("path") path_number: int | None = None if path is not None: if not isinstance(path, str) or len(path) > 512 or path.count("@") > 1: return False if "@" in path: ref = path.split("@", 1)[1] match = re.fullmatch(r"refs/pull/([1-9][0-9]*)/head", ref) if match: path_number = int(match[1]) if event != "pull_request" or path_number > 2_147_483_647: return False elif ref != f"refs/heads/{branch}" or event == "pull_request": return False raw_references = run.get("pull_requests") references = [] if raw_references is None else raw_references if not isinstance(references, list): return False numbers: list[int] = [] for reference in references: if not isinstance(reference, dict): return False number = reference.get("number") if number is not None: if type(number) is not int or not 1 <= number <= 2_147_483_647: return False numbers.append(number) head = reference.get("head") if head is not None: if not isinstance(head, dict): return False if head.get("ref") not in (None, "", branch) or head.get("sha") not in ( None, "", head_sha, ): return False if event == "pull_request" and pull_request_number is not None: if path_number not in {None, pull_request_number} or any( number != pull_request_number for number in numbers ): return False # Legacy adapters identify PR runs by a real branch and SHA. Native # blank-branch runs need affirmative PR number/ref evidence. if path_number != pull_request_number and pull_request_number not in numbers: return head_branch == branch and not path return True def exact_head_action_runs( self, repository: RegisteredRepository, *, branch: str, head_sha: str, pull_request_number: int | None = None, ) -> list[dict[str, Any]]: """Read bounded actual run evidence; never synthesize a native validation.""" if not re.fullmatch(r"(?:[0-9a-fA-F]{40}|[0-9a-fA-F]{64})", head_sha): raise GiteaError("GITEA_TARGET_INVALID", "exact Actions head SHA is invalid") if pull_request_number is not None and ( type(pull_request_number) is not int or not 1 <= pull_request_number <= 2_147_483_647 ): raise GiteaError("GITEA_TARGET_INVALID", "exact Actions PR number is invalid") output: list[dict[str, Any]] = [] seen: set[tuple[int, int]] = set() expected_count: int | None = None deadline = time.monotonic() + 20 for page in range(1, 11): if time.monotonic() >= deadline: raise GiteaError( "GITEA_ACTION_RUNS_TRUNCATED", "exact Actions read exceeded its bound" ) payload = self._request( self._path("repos", repository.owner, repository.name, "actions", "runs"), {"head_sha": head_sha, "page": page, "limit": 100}, timeout_seconds=min(10.0, max(0.1, deadline - time.monotonic())), max_attempts=1, ) count = payload.get("total_count") if isinstance(payload, dict) else None if isinstance(payload, dict) and "total_count" in payload: if type(count) is not int or not 0 <= count <= 1000: raise GiteaError("GITEA_ACTION_RUNS_INVALID", "invalid exact Actions count") if expected_count is not None and count != expected_count: raise GiteaError("GITEA_ACTION_RUNS_TRUNCATED", "exact Actions count changed") expected_count = count elif expected_count is not None: raise GiteaError("GITEA_ACTION_RUNS_TRUNCATED", "exact Actions count disappeared") if isinstance(payload, dict): payload = ( payload.get("workflow_runs") if "workflow_runs" in payload else payload.get("runs") ) runs = self._as_list(payload) if len(runs) > 100: raise GiteaError("GITEA_ACTION_RUNS_INVALID", "excess exact Actions page") for run in runs: identifier, attempt = run.get("id"), run.get("run_attempt", 0) number = run.get("run_number", identifier) if ( type(identifier) is not int or identifier < 1 or type(attempt) is not int or attempt < 0 or ("run_attempt" in run and attempt == 0) or type(number) is not int or number < 1 or (identifier, attempt) in seen ): raise GiteaError( "GITEA_ACTION_RUNS_INVALID", "invalid or repeated exact Actions run" ) seen.add((identifier, attempt)) output.append(run) if expected_count is not None: if len(output) > expected_count: raise GiteaError("GITEA_ACTION_RUNS_INVALID", "excess exact Actions runs") complete = len(output) == expected_count else: complete = len(runs) < 100 if complete: if time.monotonic() >= deadline: raise GiteaError( "GITEA_ACTION_RUNS_TRUNCATED", "exact Actions read exceeded its bound" ) matching = [ run for run in output if self.matches_exact_head_action_run( run, branch=branch, head_sha=head_sha, pull_request_number=pull_request_number, ) ] return sorted( matching, key=lambda run: ( run.get("run_number", run["id"]), run.get("run_attempt", 0), run["id"], ), reverse=True, ) if not runs: raise GiteaError("GITEA_ACTION_RUNS_TRUNCATED", "incomplete exact Actions page") raise GiteaError("GITEA_ACTION_RUNS_TRUNCATED", "exact Actions pages exceeded their bound") def action_runs( self, repository: RegisteredRepository, *, branch: str, page: int, limit: int, ) -> list[dict[str, Any]]: payload = self._request( self._path("repos", repository.owner, repository.name, "actions", "runs"), {"branch": branch, "page": page, "limit": limit}, ) if isinstance(payload, dict): if "workflow_runs" in payload: payload = payload.get("workflow_runs") elif "runs" in payload: payload = payload.get("runs") else: payload = None runs = self._as_list(payload) if runs: return runs # Only an actually empty Actions result can fall back. Malformed # collections still fail above, preserving strict response validation. try: branch_payload = self.branch(repository, branch) except GiteaError: return [] commit = branch_payload.get("commit") if not isinstance(commit, dict): return [] head_sha = str(commit.get("id") or commit.get("sha") or "") if len(head_sha) != 40 or any( character not in "0123456789abcdefABCDEF" for character in head_sha ): return [] native_runs = self.exact_head_action_runs(repository, branch=branch, head_sha=head_sha) if native_runs: offset = (page - 1) * limit return native_runs[offset : offset + limit] statuses = self.commit_statuses(repository, head_sha, 1, 100) if not statuses: return [] latest_by_context: dict[str, dict[str, Any]] = {} for item in statuses: status_context = str(item.get("context") or "__default__") previous = latest_by_context.get(status_context) if previous is None or str(item.get("created_at") or "") > str( previous.get("created_at") or "" ): latest_by_context[status_context] = item latest_states = { str(item.get("status") or "").casefold() for item in latest_by_context.values() } # Gitea can omit ``head_branch`` for pull_request runs, causing the # branch-filtered run list above to be empty even though the exact-head # commit statuses are available. Skipped jobs are neutral evidence; # require at least one success and reject every non-terminal state. if "success" not in latest_states or not latest_states <= {"success", "skipped"}: return [] return [ { "id": 0, "name": "Exact-head commit status validation", "status": "completed", "conclusion": "success", "event": "commit_status_fallback", "head_branch": branch, "head_sha": head_sha, "html_url": "", "validation_evidence": "commit-statuses", } ]