platform-ci/platform_ci/delivery/selection.py

"""Shared evidence gathering for choosing tested bytes: used by the workflow's reuse step and by the trusted publisher.

Both callers ask the same question, so they run the same code: given the commit being released and the planned targets, which earlier
successful run built exactly this input? The expected build inputs are recomputed here from the forge's own facts (the commit's
tree, the declared version at that commit, and the planned target), never taken from an artifact, so an artifact cannot vouch for itself.
"""

import urllib.error
from dataclasses import dataclass
from pathlib import Path, PurePosixPath

from platform_ci.build_inputs import BuildInputsRecord, compute_build_inputs
from platform_ci.delivery.errors import DeliveryError
from platform_ci.delivery.forge import ForgeGitHistory, ForgeReader, RepositoryScope, fetch_bundles, inspect_run, prepare_job_started, read_file
from platform_ci.delivery.manifest import BuildInputs
from platform_ci.delivery.reuse import Rejection, ReleaseArtifact, ReuseQuery, SourceRun, select_reusable_artifact
from platform_ci.effective_version import ReleaseActivityError, release_activity_since
from platform_ci.plan import Target
from platform_ci.versions import UNDECLARED, UNVERSIONED, parse_declared_version, version_file_path
from platform_ci.workspace import PipelineError

HISTORY_PAGES = 20
COMPLETED = "completed"


def declared_version_at(forge: ForgeReader, scope: RepositoryScope, target: Target, ref: str) -> str:
   """Read the component's declared version at a commit through the forge; an absent version file is `undeclared`."""
   source = target.get("version_source")
   file_name = version_file_path(target["build_tool"], source)
   if not file_name:
      return UNVERSIONED
   try:
      content = read_file(forge, scope.repository, (PurePosixPath(target["directory"]) / file_name).as_posix(), ref)
   except urllib.error.HTTPError:
      return UNDECLARED
   return UNDECLARED if content is None else parse_declared_version(target["build_tool"], content, target["commands"], source)


def release_activity_for_run(forge: ForgeReader, scope: RepositoryScope, release_sha: str, run_id: int) -> bool:
   """Return whether the release commit's history holds a captured release merge or bump inside the 3 days before ONE run's build start (the develop SNAPSHOT question).

   The build start is the server's `started_at` of the run's prepare job and the history is the forge's commit listing of `release_sha`, so `prepare` (while it runs), the publisher, the tag planner
   and the write-back all compute the same answer from the same forge facts, never from the current time. Any failure raises `ReleaseActivityError`: the version is then unknown, not guessed.
   """
   try:
      since = release_activity_since(prepare_job_started(forge, scope, run_id))
      return ForgeGitHistory(forge, scope.repository, HISTORY_PAGES).has_release_activity(release_sha, since)
   except (DeliveryError, urllib.error.URLError, OSError, ValueError) as error:  # GitHistoryError is a DeliveryError
      raise ReleaseActivityError(f"the release activity of run {run_id} at {release_sha[:12]} could not be established: {error}") from error


def expected_inputs(forge: ForgeReader, scope: RepositoryScope, history: ForgeGitHistory, target: Target, release_sha: str) -> BuildInputs:
   """Recompute, from the forge's own facts, the build inputs a rebuild of the release commit would record."""
   return as_manifest_inputs(compute_build_inputs(target, history.tree_of(release_sha), declared_version_at(forge, scope, target, release_sha)))


def as_manifest_inputs(record: BuildInputsRecord) -> BuildInputs:
   """Convert the build-input record to the delivery package's typed form."""
   return BuildInputs(
      source_tree=record["source_tree"],
      input_digest=record["input_digest"],
      toolchain=tuple(sorted(record["toolchain"].items())),
      platform=record["platform"],
      config_digest=record["config_digest"],
      version=record["version"],
   )


def query_for(forge: ForgeReader, scope: RepositoryScope, history: ForgeGitHistory, target: Target, release_sha: str) -> ReuseQuery:
   """Build the reuse query for one planned target of the release commit."""
   return ReuseQuery(component=target["component"], target=target["name"], inputs=expected_inputs(forge, scope, history, target, release_sha), release_sha=release_sha)


@dataclass
class SourceRuns:
   """Authoritative records of candidate runs, built lazily and once: a run is downloaded only when something asks for it."""

   forge: ForgeReader
   scope: RepositoryScope
   scratch: Path
   _cache: dict[int, SourceRun | Rejection] | None = None

   def get(self, run_id: int) -> SourceRun | Rejection:
      """Return the run's verified record, or a rejection when anything about it cannot be proven (never a crash)."""
      if self._cache is None:
         self._cache = {}
      if run_id not in self._cache:
         self._cache[run_id] = self._build(run_id)
      return self._cache[run_id]

   def _build(self, run_id: int) -> SourceRun | Rejection:
      try:
         facts = inspect_run(self.forge, self.scope, run_id)
         bundles = fetch_bundles(self.forge, self.scope, facts, self.scratch) if facts.status == COMPLETED and facts.conclusion == "success" else ()
      except (PipelineError, DeliveryError, urllib.error.URLError, OSError, ValueError) as error:
         return Rejection(str(run_id), f"{type(error).__name__}: {error}" if isinstance(error, PipelineError) else type(error).__name__)
      return SourceRun(run_id=facts.run_id, attempt=facts.attempt, head_sha=facts.head_sha, status=facts.status, conclusion=facts.conclusion, created_at=facts.created_at, expected_targets=frozenset(bundle.manifest.target for bundle in bundles), bundles=bundles, event=facts.event)


def select_for_query(query: ReuseQuery, run_ids: list[int], runs: SourceRuns, history: ForgeGitHistory) -> tuple[ReleaseArtifact | None, list[Rejection]]:
   """Walk candidate runs in the given order and return the first artifact that provably equals the query, with every rejection met."""
   rejections: list[Rejection] = []
   for run_id in run_ids:
      source = runs.get(run_id)
      if isinstance(source, Rejection):
         rejections.append(source)
         continue
      decision = select_reusable_artifact(query, (source,), history)
      rejections.extend(decision.rejections)
      if decision.selected is not None:
         return decision.selected, rejections
   return None, rejections