platform-ci/platform_ci/plan.py

"""Turn a validated configuration and one event into the deterministic execution plan.

The plan is data for the fixed workflow, not generated workflow YAML. It is a pure function
of the configuration and the event: nothing here receives a repository name or owner.
"""

import json
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from fnmatch import fnmatchcase
from typing import NotRequired, TypedDict

from platform_ci.build_tools import BUILD_TOOLS, CONTAINER_EXECUTION, DOCKER_POOL_EXECUTION, DOCKER_PROFILE_ENVIRONMENT, PLATFORMS, POOL_EXECUTION, PROFILE_CONTAINER_OPTIONS, TOOLCHAIN_DISPLAY
from platform_ci.configuration import Component, ProjectConfiguration, branch_matches
from platform_ci.effective_version import effective_version, version_parts, with_version_property
from platform_ci.event_selection import EventFacts, channel_matches, selected
from platform_ci.runtimes import RuntimeCatalog
from platform_ci.workspace import PipelineError, require_commit_hash

PLAN_SCHEMA_VERSION = 1
DELIVERY_NOTE = "This workflow never signs, publishes or deploys and holds no release credential. It lists the requested publications; the separate trusted publisher verifies this run and delivers through its own bindings."

DeclaredVersionReader = Callable[["Target"], str]
"""Reads the declared version of a planned target (from a checkout or from forge facts); injected so this module never imports `versions` (which imports it)."""

ReleaseActivityReader = Callable[[], bool]
"""Answers whether the release commit's history, within the window ending at the SERVER's build start of the run, holds a captured release merge or bump (`effective_version`); injected
and lazy: it reads the forge only when a develop SNAPSHOT needs it, so no plan without that case reads a job or a commit."""

_BRANCH_REFERENCE_PREFIX = "refs/heads/"
_TAG_REFERENCE_PREFIX = "refs/tags/"


class ReportTarget(TypedDict):
   """One declared report in the plan JSON: `format` is `junit` or `file`; `paths` are globs relative to the component directory."""

   format: str
   paths: list[str]
   required: bool


class ImageTarget(TypedDict):
   """One image a target packages from its own verified output: `artifact` names the verified artifact the image may contain."""

   artifact: str
   dockerfile: str
   recipe: str
   context: str
   target: str
   node_version: str
   health_port: str
   ignore_unfixed: bool
   """The closed scan-policy field: findings with no available fix do not fail the scan; every other HIGH or CRITICAL finding still does."""


class CheckoutTarget(TypedDict):
   """One pinned private dependency fetched before the build: a logical binding, the repository, the exact commit and the path; never a credential."""

   binding: str
   repository: str
   ref: str
   path: str


class VersionSourceTarget(TypedDict):
   """The explicit native version source of a `script` component: a component-relative `.properties` file and the key whose literal value is the version."""

   file: str
   property: str


class Target(TypedDict):
   """One component on one platform: everything a verification job needs, as plain data.

   `execution` is `container` (a catalog `image` on the standard pool), `pool` or `pool-docker` (the image on the profile pool `runner`, run by a pool job; only `pool-docker`, which requires the docker profile, mounts the test daemon) or `host` (directly on the `runner`, with an
   empty `image`). `requires` lists the runtime profiles; `emulator` (android toolchain script components) selects the `pool` job with the KVM device option and no other
   privilege. `environment` is the fixed, non-secret variables the build tool needs.
   """

   name: str
   display: str
   component: str
   platform: str
   build_tool: str
   toolchain: str
   toolchain_version: str
   execution: str
   runner: str
   image: str
   requires: list[str]
   environment: dict[str, str]
   directory: str
   commands: list[list[str]]
   reports: dict[str, ReportTarget]
   artifacts: dict[str, list[str]]
   images: dict[str, ImageTarget]
   """Images packaged after verification and packaging in the same job from the verified outputs; empty for most targets."""
   checkouts: dict[str, CheckoutTarget]
   """Pinned private dependencies the platform fetches (credential only inside that step) before the build; empty for most targets."""
   version_source: NotRequired[VersionSourceTarget]
   """Present only when the component declares one: it is then part of this target, so of the config digest, and is read the same way at packaging and through the forge."""


class Verification(TypedDict):
   """Whether configuration requests verification for this event, and why. A requested verification runs exactly once.

   When `required` is false the targets are still listed (so the matrix is never empty) but the verify job is expected to skip.
   """

   required: bool
   reason: str


class PlannedPublication(TypedDict):
   """One requested publication that applies to this event: a component artifact, a logical destination and the eligible channels.

   `target` is the verified target whose artifact bundle supplies the files. No address, credential or owner is ever present.
   """

   name: str
   component: str
   artifact: str
   format: str
   platform: str
   target: str
   destination: str
   channels: list[str]
   deploy: str


class Delivery(TypedDict):
   """Release eligibility and requested publications declared by configuration.

   `implemented` is false because this workflow itself never delivers; the trusted publisher does. `publications` lists
   only the requests whose channels are eligible for this event, so it is empty for pull requests and ineligible pushes.
   """

   implemented: bool
   eligible_channels: list[str]
   publications: list[PlannedPublication]
   note: str


class ReuseEligibility(TypedDict):
   """Whether tested bytes of an identical earlier build may replace this run's verification, decided from the event alone.

   Eligible for a push or pull request whose verification is required. A pull request may reuse a completed push
   only; manual or callable runs always verify. Eligibility is not a decision to reuse;
   the reuse step must still prove identical inputs, tree, lineage and bundles, and otherwise the build runs.
   """

   eligible: bool
   reason: str


class Plan(TypedDict):
   """The complete expected work for one run."""

   schema: int
   source_revision: str
   source_tree: str
   reuse: ReuseEligibility
   platform_revision: str
   event: str
   verification: Verification
   targets: list[Target]
   delivery: Delivery


@dataclass(frozen=True)
class EventContext:
   """The event facts the plan depends on. Repository identity is deliberately absent."""

   name: str
   ref: str
   source_revision: str
   platform_revision: str
   default_branch: str = ""
   """The repository's default branch as reported by the event; required only for branch pushes."""
   source_tree: str = ""
   """The git tree id of `source_revision`; required only to decide reuse, never to build."""
   head_branch: str = ""
   """A pull request's head branch (`github.head_ref`), so event selection by release channel sees it; empty for every other event."""


def prepare_plan(project: ProjectConfiguration, event: EventContext, runtimes: RuntimeCatalog, declared_version: DeclaredVersionReader | None = None, release_activity: ReleaseActivityReader | None = None) -> Plan:
   """Expand components into targets and decide from configuration whether this invocation verifies.

   Pull requests, manual and callable requests, default branch pushes and release-eligible pushes always verify once.
   Other pushes verify only when `verification.push` selects them. The pipeline never guesses whether a pull request
   exists, so an unselected push is reported as not requested rather than as covered. Release eligibility is read from
   configuration and recorded. Each target's image comes from the runtime catalog; an unsupported or unqualified
   toolchain raises before any job exists, requested or not.
   """
   require_commit_hash(event.source_revision, "source revision")
   require_commit_hash(event.platform_revision, "platform revision")
   branch = _pushed_branch(event)
   facts = event_facts_of(event)
   # the same selection the trusted publisher applies: components and images restricted by channel run only for the events they select
   targets = plan_targets(project, runtimes, facts, declared_version, release_activity)
   eligible_channels = sorted(channel.name for channel in project.channels if event.name == "push" and channel_matches(channel, facts))
   verification = _decide_verification(project, event, branch, eligible_channels)
   return Plan(
      schema=PLAN_SCHEMA_VERSION,
      source_revision=event.source_revision,
      source_tree=event.source_tree,
      reuse=_reuse_eligibility(event, verification),
      platform_revision=event.platform_revision,
      event=event.name,
      verification=verification,
      targets=targets,
      delivery=Delivery(implemented=False, eligible_channels=eligible_channels, publications=publication_requests(project, eligible_channels), note=DELIVERY_NOTE),
   )


def _reuse_eligibility(event: EventContext, verification: Verification) -> ReuseEligibility:
   """Allow proven completed bytes for required push or PR verification; keep manual and callable runs verifying."""
   if event.name not in ("push", "pull_request"):
      return ReuseEligibility(eligible=False, reason=f"{event.name} events always verify: they are the original verification")
   if not verification["required"]:
      return ReuseEligibility(eligible=False, reason="verification is not requested, so there is nothing to replace")
   if not event.source_tree:
      return ReuseEligibility(eligible=False, reason="the source tree id is unknown, so equality with an earlier build cannot be proven")
   return ReuseEligibility(eligible=True, reason="a required push or pull request verification may be replaced by tested bytes of an identical completed build")


def event_facts_of(event: EventContext) -> EventFacts:
   """Return the authoritative facts of the event: its name, the push branch or pull request head branch, the pushed tag and the exact revision."""
   branch = _pushed_branch(event) or ""
   tag = event.ref.removeprefix(_TAG_REFERENCE_PREFIX) if event.name == "push" and event.ref.startswith(_TAG_REFERENCE_PREFIX) else ""
   return EventFacts(name=event.name, branch=branch or event.head_branch, tag=tag, source_revision=event.source_revision)


def plan_targets(
   project: ProjectConfiguration,
   runtimes: RuntimeCatalog,
   facts: EventFacts | None = None,
   declared_version: DeclaredVersionReader | None = None,
   release_activity: ReleaseActivityReader | None = None,
) -> list[Target]:
   """Expand components and platforms into targets; the same function serves the build and the publisher's equivalence proof.

   With `facts`, a component or image its `only_on_channels` / `not_on_channels` do not select for that event is left out (an excluded image also loses
   its generated archive and evidence outputs), and a component that opted in to event facts gets the fixed variables. Without `facts` nothing is
   filtered and no event variable is set: the trusted publisher and the build must both pass the facts of the event they mean.

   UNTESTED. For a push whose matching channel declares an `effective_version` rule, every Maven component the event selects gets the calculated version as
   `-D<property>=<version>` on every native command (also when it equals the declared version: the explicit property can activate a profile), so the build, the recorded build inputs and the recomputation by the publisher and the tag planner all see it.
   The original (non-push) verification of a Maven component governed by effective-version channels that agree on one property plans that same property with the DECLARED version (`_with_declared_version`),
   so an unchanged-version release has the target and `config_digest` of the tested commands while a changed version differs. That needs `declared_version`; without it such events keep the plain commands. A develop SNAPSHOT also needs
   `release_activity` (build start and history, both from the forge and bound to ONE run); every caller binds it to the run whose build it plans or recomputes. That needs `declared_version`,
   which reads the declared version of an UNMODIFIED target (from the checkout in `prepare`, from the forge in the publisher and tag planner); without it such a plan is refused. The reader is a static
   resolver (`versions._pom_version`), the fast-fail before any build: the proof against Maven's own resolution is the publication adapter, which requires every retained POM's coordinates to equal the
   recorded version (`delivery/adapters.py::_require_manifest_version`). No Maven process runs and nothing compiles here.
   """
   channels = project.channels
   targets: list[Target] = []
   for component in project.components:
      if facts is not None and not selected(component.only_on_channels, component.not_on_channels, channels, facts):
         continue
      excluded = frozenset(name for name, image in component.images.items() if facts is not None and not selected(image.only_on_channels, image.not_on_channels, channels, facts))
      rule = _effective_rule(project, component, facts)
      original_property = _verification_version_property(project, component) if rule is None and facts is not None and facts.name != "push" and declared_version is not None else ""
      for platform in component.platforms:
         target = _target(component, platform, runtimes, facts, excluded)
         if rule is not None and facts is not None:
            target = _with_effective_version(target, rule, facts, declared_version, release_activity)
         elif original_property and declared_version is not None:
            target = _with_declared_version(target, original_property, declared_version)
         targets.append(target)
   return targets


def _effective_rule(project: ProjectConfiguration, component: Component, facts: EventFacts | None) -> tuple[str, str, str] | None:
   """Return the (rule, property) a push applies to this Maven component through the channels the event matches (as the captured calculator ran for every eligible publish kind); two different ones conflict."""
   if facts is None or facts.name != "push" or component.build_tool != "maven":
      return None
   rules = {(channel.effective_version, channel.version_property, channel.rc_baseline) for channel in project.channels if channel.effective_version and channel_matches(channel, facts)}
   if len(rules) > 1:
      raise PipelineError(f"the event matches release channels with different effective version rules {sorted(rules)}, so component {component.name!r} has no single version")
   return next(iter(rules)) if rules else None


def _with_effective_version(target: Target, rule: tuple[str, str, str], facts: EventFacts, declared_version: DeclaredVersionReader | None, release_activity: ReleaseActivityReader | None = None) -> Target:
   """Apply the calculated version to the target's commands, then prove the build tool now resolves exactly that version (never relabelling).

   The property is ALWAYS appended, also when the calculated version equals the declared one: a Maven profile may activate on the mere presence of the explicit property, so dropping it is not proven to
   preserve the build. The original verification of such a component plans the same explicit property (`_with_declared_version`), so an unchanged version matches those tested commands.
   """
   if declared_version is None:
      raise PipelineError(f"target {target['name']!r} needs an effective version, but no reader of its declared version was supplied")
   if target.get("version_source") is not None:
      raise PipelineError(f"target {target['name']!r} reads its version from an explicit version_source, which an effective version property cannot replace")
   declared = declared_version(target)
   version = effective_version(rule[0], declared, facts.branch, facts.tag, rule[2], release_activity)
   applied: Target = {**target, "commands": with_version_property(target["commands"], rule[1], version)}
   resolved = declared_version(applied)
   if resolved != version:
      raise PipelineError(f"target {target['name']!r}: with -D{rule[1]}={version} the project resolves version {resolved!r}, not {version!r}; the version must be declared through that property")
   return applied


def _verification_version_property(project: ProjectConfiguration, component: Component) -> str:
   """Return the version property the configured effective-version channels agree on, or "" when none exists.

   Release planning applies the configured rule to every selected Maven component, including verification-only components. Original verification uses that same scope.
   Channels without an effective-version rule are ignored; conflicting properties are refused.
   """
   if component.build_tool != "maven":
      return ""
   properties = {channel.version_property for channel in project.channels if channel.effective_version}
   if len(properties) > 1:
      raise PipelineError(f"component {component.name!r} has effective version channels with different version properties {sorted(properties)}, so its verification has no single property to plan")
   return next(iter(properties), "")


def _with_declared_version(target: Target, property_name: str, declared_version: DeclaredVersionReader) -> Target:
   """Plan an original (non-push) verification with the same explicit `-D<property>=<declared>` the release build uses, so that an unchanged-version release has identical commands.

   The declared version must be in a captured format (as every rule requires for the release) and the project must resolve exactly that version through the property; otherwise the release
   could never be built either, and this is refused here, before the verification, rather than recorded as an unreleasable build.
   """
   if target.get("version_source") is not None:
      raise PipelineError(f"target {target['name']!r} reads its version from an explicit version_source, which an effective version property cannot replace")
   declared = declared_version(target)
   version_parts(declared)
   applied: Target = {**target, "commands": with_version_property(target["commands"], property_name, declared)}
   resolved = declared_version(applied)
   if resolved != declared:
      raise PipelineError(f"target {target['name']!r}: with -D{property_name}={declared} the project resolves version {resolved!r}, not {declared!r}; the version must be declared through that property")
   return applied


def publication_requests(project: ProjectConfiguration, eligible_channels: Sequence[str]) -> list[PlannedPublication]:
   """Return the configured publications restricted to the eligible channels; a publication with none of them is not requested."""
   requests: list[PlannedPublication] = []
   for publication in project.publications:
      channels = sorted(set(publication.channels) & set(eligible_channels))
      if channels:
         target_name = f"{publication.component}-{publication.platform}"
         requests.append(
            PlannedPublication(
               name=publication.name,
               component=publication.component,
               artifact=publication.artifact,
               format=publication.format,
               platform=publication.platform,
               target=target_name,
               destination=publication.destination,
               channels=channels,
               deploy=publication.deploy,
            )
         )
   return requests


def build_matrix(plan: Plan, execution: str = CONTAINER_EXECUTION) -> dict[str, list[dict[str, str]]]:
   """Return the native job matrix of the targets with one execution kind (`container` by default, `host` or `pool`).

   Each entry carries its target as a JSON string so any matrix engine can pass it through. Standard container, pool and host
   targets are separate matrices because they run in separate jobs: only the pool job may carry container options and mounts, so a
   pool entry also carries the container `options` its runtime profiles add (empty when none).
   """
   entries: list[dict[str, str]] = []
   for target in plan["targets"]:
      if target["execution"] != execution:
         continue
      entry = {"name": target["name"], "display": target["display"], "image": target["image"], "runner": target["runner"], "target": to_compact_json(target)}
      if execution in (POOL_EXECUTION, DOCKER_POOL_EXECUTION):
         entry["options"] = pool_container_options(target["requires"])
      if target["checkouts"]:
         entry["dependencies"] = "true"  # the job's checkout-dependencies step runs only for entries that carry this flag
      entries.append(entry)
   return {"include": entries}


def pool_container_options(profiles: Sequence[str]) -> str:
   """Return the container options the runtime profiles add in the pool job, as one space-separated string (empty when none)."""
   return " ".join(option for profile in profiles for option in PROFILE_CONTAINER_OPTIONS.get(profile, ()))


def to_compact_json(value: object) -> str:
   """Serialize deterministically on a single line, suitable for a workflow output."""
   return json.dumps(value, sort_keys=True, separators=(",", ":"))


def _pushed_branch(event: EventContext) -> str | None:
   """Return the pushed branch name, or None when the event is not a branch push."""
   if event.name != "push" or not event.ref.startswith(_BRANCH_REFERENCE_PREFIX):
      return None
   return event.ref.removeprefix(_BRANCH_REFERENCE_PREFIX)


def _decide_verification(project: ProjectConfiguration, event: EventContext, branch: str | None, eligible_channels: list[str]) -> Verification:
   """Apply the always-verify rules first, then the configured push selection."""
   if event.name != "push":
      return Verification(required=True, reason=f"{event.name} event")
   if branch is not None:
      return _decide_branch_push_verification(project, event, branch, eligible_channels)
   if event.ref.startswith(_TAG_REFERENCE_PREFIX):
      tag = event.ref.removeprefix(_TAG_REFERENCE_PREFIX)
      if eligible_channels:
         return Verification(required=True, reason=f"release-eligible tag push (channels: {', '.join(eligible_channels)})")
      if any(fnmatchcase(tag, pattern) for pattern in project.push.tags):
         return Verification(required=True, reason="tag push selected by verification.push.tags")
      return Verification(required=False, reason=f"push of tag {tag!r}, which verification.push.tags does not select")
   raise PipelineError(f"push ref {event.ref!r} is neither a branch nor a tag; the pipeline cannot decide whether to verify it")


def _decide_branch_push_verification(project: ProjectConfiguration, event: EventContext, branch: str, eligible_channels: list[str]) -> Verification:
   """Decide verification for a push to a branch: default branch, release-eligible, or selected by configuration."""
   if not event.default_branch:
      raise PipelineError("default branch metadata (CI_DEFAULT_BRANCH) is required to decide verification for a branch push")
   if branch == event.default_branch:
      return Verification(required=True, reason="push to the default branch")
   if eligible_channels:
      return Verification(required=True, reason=f"release-eligible push (channels: {', '.join(eligible_channels)})")
   if branch_matches(project.push.branches, branch):
      return Verification(required=True, reason="push to a branch selected by verification.push.branches")
   return Verification(required=False, reason=f"push to branch {branch!r}, which verification.push.branches does not select")


def _target(component: Component, platform: str, runtimes: RuntimeCatalog, facts: EventFacts | None = None, excluded_images: frozenset[str] = frozenset()) -> Target:
   """Expand one component and platform into a target with tokenized native commands and its qualified runtime.

   The runtime is resolved from the catalog by toolchain, version, platform and profiles alone; an unqualified one raises here.
   """
   tool = BUILD_TOOLS[component.build_tool]
   toolchain = component.toolchain or tool.toolchain
   runtime = runtimes.resolve_runtime(toolchain, component.toolchain_version, platform, component.requires)
   platform_execution = PLATFORMS[platform].execution
   execution = platform_execution
   if platform_execution == CONTAINER_EXECUTION and component.requires:
      execution = DOCKER_POOL_EXECUTION if "docker" in component.requires else POOL_EXECUTION
   profiles = f" / {'+'.join(component.requires)}" if component.requires else ""
   environment = dict(tool.environment) | (dict(tool.container_environment) if platform_execution == CONTAINER_EXECUTION else {})
   if execution == DOCKER_POOL_EXECUTION:
      environment |= dict(DOCKER_PROFILE_ENVIRONMENT)
   if component.event_facts and facts is not None:
      # fixed, non-secret and taken from the authoritative event, not the runner: part of the target, so of the build-input identity
      environment |= {"CI_SOURCE_REVISION": facts.source_revision, "CI_EVENT_NAME": facts.name, "CI_REF_NAME": facts.ref_name, "CI_REF_TYPE": facts.ref_type}
   generated = {f"image-{name}" for name in excluded_images} | {f"image-{name}-evidence" for name in excluded_images}
   target = Target(
      name=f"{component.name}-{platform}",
      display=f"{component.name} / {PLATFORMS[platform].display} / {TOOLCHAIN_DISPLAY[toolchain]} {component.toolchain_version}{profiles}",
      component=component.name,
      platform=platform,
      build_tool=component.build_tool,
      toolchain=toolchain,
      toolchain_version=component.toolchain_version,
      execution=execution,
      runner=runtime.runner,
      image=runtime.image,
      requires=list(component.requires),
      environment=environment,
      directory=component.directory,
      commands=_native_commands(component),
      reports={name: report for name, report in _report_targets(component).items() if name not in generated},
      artifacts={name: list(paths) for name, paths in component.artifacts.items() if name not in generated},
      images={
         name: ImageTarget(
            artifact=image.artifact,
            dockerfile=image.dockerfile,
            recipe=image.recipe,
            context=image.context,
            target=image.target,
            node_version=image.node_version,
            health_port=image.health_port,
            ignore_unfixed=image.ignore_unfixed,
         )
         for name, image in component.images.items()
         if name not in excluded_images
      },
      checkouts={checkout.name: CheckoutTarget(binding=checkout.binding, repository=checkout.repository, ref=checkout.ref, path=checkout.path) for checkout in component.checkouts},
   )
   if component.version_source is not None:
      target["version_source"] = VersionSourceTarget(file=component.version_source.file, property=component.version_source.property)
   return target


def has_execution(plan: Plan, execution: str) -> bool:
   """Return whether any planned target runs with the given execution kind (`container` or `host`)."""
   return any(target["execution"] == execution for target in plan["targets"])


def _native_commands(component: Component) -> list[list[str]]:
   """Return the native commands as separate tokens: the tool's install commands first, then the declared commands."""
   tool = BUILD_TOOLS[component.build_tool]
   if tool.helper_subcommands:
      return [list(command) for command in component.build_commands]
   if tool.subcommands:
      programs = dict(tool.subcommands)
      return [list(programs[command[0]]) for command in component.build_commands]
   if tool.program_in_arguments:
      return [list(command) for command in component.build_commands]
   return [[tool.executable, *command] for command in (*tool.install_commands, *component.build_commands)]


def _report_targets(component: Component) -> dict[str, ReportTarget]:
   """Return each report as plain JSON data: its format, its globs and whether a successful build must produce it."""
   return {name: ReportTarget(format=report.format, paths=list(report.paths), required=report.required) for name, report in component.reports.items()}