Open_Ultra_SAST/src/openultrasast/push/runner.py

"""Bounded, local base/head replay through the existing repository driver."""

from __future__ import annotations

import hashlib
import logging
import multiprocessing
import os
import pickle
import selectors
import time
from collections import Counter
from collections.abc import Callable, Sequence
from contextlib import nullcontext
from dataclasses import asdict, replace
from pathlib import Path
from typing import Any, Literal, TypeVar, cast

from openultrasast.config import PushConfig
from openultrasast.cpg.artifact import digest_value
from openultrasast.cpg.backend import JoernBackend, engine_runtime_identity, joern_version
from openultrasast.mapping import analyze_entry_points
from openultrasast.model.contracts import ChangeContext, ExecutionBudget
from openultrasast.model.layout import layout_facts
from openultrasast.model.regions import ScanRegion, regions_for
from openultrasast.model.scan import ModelScanResult, ScanBudget, scan_repository
from openultrasast.model.shipped import declared_sources
from openultrasast.preprocess import build_file_target, enumerate_source_files
from openultrasast.push.cache import ArtifactCache, SemanticKeys
from openultrasast.push.contracts import ComparisonAnalysis, PushComparison, PushResult, SnapshotManifest
from openultrasast.push.eligibility import fingerprint, load_registry
from openultrasast.push.explain import language_reason
from openultrasast.push.policy import (
    ActionableDefect,
    AdmissionCandidate,
    AdmissionResult,
    CapabilityKey,
    DeltaComparison,
    admission_push_result,
    admit_candidates,
    compare_targeted_base,
    dependency_gaps,
    semantics_digest,
)
from openultrasast.push.quick import Basis, QuickTierResult, head_spans, run_quick_tier
from openultrasast.push.report import PushReport, ReportDelivery, ScanRecord, _completed_scan, deliver_report, render_report
from openultrasast.push.reuse import ReusingBackend, declaration_identity, discovery
from openultrasast.push.snapshot import SnapshotAdapter
from openultrasast.push.vetoes import FLAG, LABEL, load_declarations, record_vetoes

logger = logging.getLogger(__name__)

_T = TypeVar("_T")


def _prepare(function: Callable[[], _T], budget: ExecutionBudget) -> _T:
    """Bound owned discovery/metadata or explicit HTTP assistance on Linux.

    Read incrementally while the child writes; no pipe-buffer deadlock or unbounded
    blocking recv. The child never launches engines, subprocesses or project code.
    """
    if time.monotonic() >= budget.deadline_monotonic:
        raise TimeoutError("deadline_exhausted")
    reader, writer = os.pipe()

    def child() -> None:
        os.close(reader)
        try:
            payload: tuple[bool, Any]
            try:
                payload = (True, function())
            except Exception as error:
                payload = (False, type(error).__name__)
            with os.fdopen(writer, "wb") as stream:
                pickle.dump(payload, stream, protocol=5)
        finally:
            os._exit(0)

    process = multiprocessing.get_context("fork").Process(target=child)
    data = bytearray()
    try:
        process.start()
        os.close(writer)
        writer = -1
        with selectors.DefaultSelector() as selector:
            selector.register(reader, selectors.EVENT_READ)
            while selector.get_map():
                remaining = budget.deadline_monotonic - time.monotonic()
                if remaining <= 0:
                    raise TimeoutError("deadline_exhausted")
                for key, _ in selector.select(min(0.05, remaining)):
                    chunk = os.read(key.fd, 65536)
                    if not chunk:
                        selector.unregister(key.fileobj)
                    else:
                        data.extend(chunk)
                        if len(data) > 32 * 1024**2:
                            raise ValueError("preparation_output_limit")
        process.join(max(0, budget.deadline_monotonic - time.monotonic()))
        if process.is_alive() or process.exitcode != 0 or not data:
            raise RuntimeError("preparation_worker_failed")
        success, result = pickle.loads(data)  # trusted child, never repository-provided pickle
        if not success:
            raise RuntimeError("preparation_failed:" + result)
        return cast(_T, result)
    finally:
        os.close(reader)
        if writer >= 0:
            os.close(writer)
        if process.pid is not None:
            if process.is_alive():
                process.kill()
                process.join(max(0, budget.deadline_monotonic + budget.cancellation_allowance_seconds - time.monotonic()))
            if not process.is_alive():
                process.close()


def _discover(root: Path) -> tuple[tuple[ScanRegion, ...], tuple[bytes, ...]]:
    # Same proposer and declaration-based shipping rules as ordinary scans. Do not
    # call preprocess's live-HEAD probe on a materialized, intentionally Git-free tree.
    files = enumerate_source_files(root)
    targets = [build_file_target(root, path) for path in files]
    regions = regions_for(analyze_entry_points(root, targets), targets, shipped=declared_sources(root))
    # Conservatively seed every tracked declaration/non-source file. This is change
    # context only, never a graph allowlist or a second scope selector.
    sources = set(files)
    declarations = tuple(os.fsencode(str(p.relative_to(root))) for p in root.rglob("*") if p.is_file() and p not in sources)
    return regions, declarations


def _provenance(repository: Path, scan_budget: ScanBudget) -> dict[str, str]:
    root = Path(__file__).resolve().parents[1]

    def digest(paths: list[Path]) -> str:
        hasher = hashlib.sha256()
        for path in sorted(paths):
            hasher.update(str(path.relative_to(root)).encode())
            hasher.update(path.read_bytes())
        return hasher.hexdigest()

    version = joern_version()
    engine = "joern-" + (".".join(map(str, version)) if version else "unavailable_or_unidentified")
    runtime = engine_runtime_identity()
    integration = digest(list((root / "cpg").glob("*.py")))
    return {
        "repository": str(repository.absolute()),
        "engine_runtime": runtime,
        "engine": engine,
        "facts": digest(list((root / "ruleset").rglob("*.toml"))),
        "graph_layout": digest_value(
            {
                "facts": [asdict(row) for row in layout_facts()],
                "implementation": digest([root / "model/layout.py", root / "model/partitions.py"]),
            }
        ),
        "queries": digest(list((root / "cpg/queries").glob("*.sc"))),
        "policy": digest(list((root / "push").glob("*.py"))),
        "core": digest(list((root / "model").glob("*.py")) + list((root / "cpg").glob("*.py"))),
        "semantics": semantics_digest(
            engine_identity=engine, options={**asdict(scan_budget), "engine_runtime": runtime, "engine_integration": integration}
        ),
        "ranking_mode": "evidence",
        "model": "disabled",
        "capabilities": fingerprint(),
        "cache": "cold_no_reuse",
    }


def _admit(candidates: tuple[AdmissionCandidate, ...], provenance: dict[str, str]) -> AdmissionResult:
    registry = load_registry(current=provenance)
    result = admit_candidates(candidates, capabilities=registry.capabilities)
    if candidates and registry.reasons:
        result = replace(
            result,
            coverage_reasons=tuple(dict.fromkeys((*result.coverage_reasons, *("eligibility:" + reason for reason in registry.reasons)))),
        )
    return result


def _record(
    head: ModelScanResult,
    delta: DeltaComparison,
    context: ChangeContext,
    comparison: PushComparison,
    reasons: Sequence[str],
    admission: AdmissionResult,
    provenance: dict[str, str],
    declarations: Path | None = None,
) -> dict[str, object]:
    """Experimental M1a: the same policy over lifted evidence, reported only, never admitted.

    `declarations` supplies templates for the experimental evaluation rendering ONLY. They are
    never passed to admission, so nothing read from that file can enable a capability.
    """
    registry = load_registry(current=provenance)
    supplied = load_declarations(declarations) if declarations is not None else ()
    report = record_vetoes(
        head=head,
        base=delta.base_scan,
        context=context,
        comparison=comparison,
        semantics=delta.semantics,
        production=delta.candidates,
        production_admission={d.candidate.delta.defect_id: d.reasons for d in admission.dispositions},
        coverage_reasons=tuple(dict.fromkeys(reasons)),
        capabilities=registry.capabilities,
        declarations=supplied,
    )
    return report.to_payload()


def _uncovered_languages(manifest: SnapshotManifest, context: ChangeContext) -> list[str]:
    """`language_not_covered` reasons for changed languages with no engine frontend and no quick rules.

    Only a language this push changes is named: a Go helper nobody touched is not this push's gap.
    """
    from openultrasast.findings import PATTERN_RULES
    from openultrasast.model.partitions import FRONTENDS
    from openultrasast.preprocess import LANGUAGE_BY_EXTENSION

    covered = set(FRONTENDS) | {language for rule in PATTERN_RULES if rule.status != "disabled" for language in rule.languages}
    if any(not rule.languages for rule in PATTERN_RULES if rule.status != "disabled"):
        return []

    def language(path: str) -> str | None:
        return LANGUAGE_BY_EXTENSION.get(os.path.splitext(path)[1].lower())

    totals = Counter(language(os.fsdecode(bytes.fromhex(item.path_hex))) for item in manifest.files)
    changed = Counter(language(context.decode_path(path)) for path in context.changed_paths if path not in context.deleted_paths)
    return [
        language_reason(name, count, totals.get(name, 0))
        for name, count in sorted((k, v) for k, v in changed.items() if k is not None and k not in covered)
    ]


def _discovery_identity() -> str:
    root = Path(__file__).resolve().parents[1]
    # Discovery invokes the shipped mapper, preprocessing, layout and family specs.
    # Hash code bytes, never a mutable import name or a repository-provided version.
    return digest_value([(p.relative_to(root).as_posix(), hashlib.sha256(p.read_bytes()).hexdigest()) for p in sorted(root.rglob("*.py"))])


def _stage(timings: dict[str, float], stage: str, started: float) -> float:
    """Close one stage, record its cost and say so.

    A run that is killed, or that reports a later stage's timeout, otherwise leaves no account of
    where its budget went. Three PHP investigations were spent reading `cpg batch taint failed:
    timeout` as a slow query when the query answers 20 requests in 24 seconds, because nothing said
    what had already consumed the transaction.
    """
    now = time.monotonic()
    timings[stage + "_seconds"] = now - started
    logger.info("push stage %s finished in %.1fs", stage, now - started)
    return now


def _analyze(
    repository: Path,
    *,
    base: str,
    head: str,
    artifact: Path,
    config: PushConfig | None = None,
    backend: Any = None,
    max_regions: int = 500,
    cache_dir: Path | None = None,
    execution_budget: ExecutionBudget,
    resolved: PushComparison | None = None,
    record_vetoes: bool = False,
    declarations: Path | None = None,
    engine: Literal["inline", "off", "background"] = "inline",
    quick_basis: Basis = "comparison",
) -> PushReport:
    """Run a single explicit comparison; exit policy never stands for coverage.

    `record_vetoes` is the task-local M1a evaluation flag: after the production result is
    decided, the same policy is rerun over lifted evidence and every veto is recorded in
    the artifact's labeled `experimental` section. It changes no result, admission or
    eligibility, and the runner rejects it outside explicit replay.
    """
    config = config or PushConfig()
    if max_regions < 1:
        raise ValueError("max_regions must be positive")
    if cache_dir is not None and cache_dir.resolve().is_relative_to(repository.resolve()):
        raise ValueError("cache directory must be outside the analyzed repository")
    started = time.monotonic()
    budget = execution_budget
    scan_budget = ScanBudget(max_model_calls=0, max_regions=max_regions, order_by_evidence=True)
    if backend is None:
        backend = JoernBackend()
    cache = None
    cache_semantics = None
    reused: list[ReusingBackend] = []
    records: list[ScanRecord] = []
    manifests: list[SnapshotManifest] = []
    reasons: list[str] = []
    admission = AdmissionResult((), (), ())
    comparison = resolved
    context = None
    experimental: dict[str, object] = {"label": LABEL, "flag": FLAG, "status": "not_evaluated"} if record_vetoes else {}
    quick: QuickTierResult | None = None
    timings: dict[str, float] = {
        "deadline_seconds": config.deadline_seconds,
        "cancellation_allowance_seconds": config.cancellation_allowance_seconds,
    }
    provenance = {key: "unavailable_preparation_incomplete" for key in ("engine", "facts", "queries", "policy")}
    provenance.update(repository=str(repository.absolute()), model="disabled", requested_base=base, requested_head=head)
    stage = "resolution"
    stage_started = started
    try:
        adapter = SnapshotAdapter(repository, execution_budget=budget)
        comparison = resolved or adapter.resolve_replay(base, head)
        stage_started = _stage(timings, stage, stage_started)
        stage = "provenance"
        provenance.update(_prepare(lambda: _provenance(repository, scan_budget), budget))
        stage_started = _stage(timings, stage, stage_started)
        if cache_dir is not None:
            cache = ArtifactCache(cache_dir, max_bytes=config.cache_max_bytes)
            cache_semantics = SemanticKeys(
                provenance["facts"],
                provenance["queries"],
                digest_value({"query_parameters": "complete-joern-request-v1"}),
                digest_value({"ranking_mode": "evidence", "options": asdict(scan_budget)}),
                provenance["policy"],
                provenance["model"],
                provenance["capabilities"],
                config.mode,
            )
            provenance["cache"] = "validated_local_reuse"
        discovery_identity = digest_value({"code": _prepare(lambda: _discovery_identity(), budget), "facts": provenance["facts"]})
        stage, stage_started = "preparation", time.monotonic()
        with adapter.materialize(comparison.head_oid, budget=budget) as tip:
            manifests.append(tip.manifest)
            assert comparison.base_oid is not None
            old_context = (
                nullcontext(tip) if comparison.base_oid == comparison.head_oid else adapter.materialize(comparison.base_oid, budget=budget)
            )
            with old_context as old:
                manifests.append(old.manifest)
                for manifest in manifests:
                    reasons.extend("snapshot:" + boundary.reason for boundary in manifest.boundaries)
                timings["snapshot_seconds"] = time.monotonic() - stage_started
                timings["snapshot_reuses"] = int(comparison.base_oid == comparison.head_oid)
                discovery_started = time.monotonic()
                head_regions, head_declarations, head_hit = _prepare(
                    lambda: discovery(tip.root, tip.manifest, cache=cache, identity=discovery_identity, budget=budget, discover=_discover),
                    budget,
                )
                base_regions, base_declarations, base_hit = _prepare(
                    lambda: discovery(old.root, old.manifest, cache=cache, identity=discovery_identity, budget=budget, discover=_discover),
                    budget,
                )
                timings["discovery_seconds"] = time.monotonic() - discovery_started
                timings["discovery_hits"] = int(head_hit) + int(base_hit)
                head_backend = base_backend = backend
                if (
                    cache is not None
                    and cache_semantics is not None
                    and isinstance(backend, JoernBackend)
                    and all(m.complete for m in manifests)
                ):
                    head_backend = ReusingBackend(
                        backend,
                        cache,
                        cache_semantics,
                        declarations=declaration_identity(tip.manifest, head_declarations),
                        exclusions=provenance["graph_layout"],
                    )
                    base_backend = ReusingBackend(
                        backend,
                        cache,
                        cache_semantics,
                        declarations=declaration_identity(old.manifest, base_declarations),
                        exclusions=provenance["graph_layout"],
                    )
                    reused.extend((head_backend, base_backend))
                context = adapter.compare(comparison, declaration_paths=tuple(set(head_declarations + base_declarations)), budget=budget)
                context = replace(context, unresolved_boundaries=tuple(dict.fromkeys((*context.unresolved_boundaries, *reasons))))
                reasons.extend(context.unresolved_boundaries)
                reasons.extend(_uncovered_languages(tip.manifest, context))
                stage_started = _stage(timings, stage, stage_started)
                stage = "quick"
                # The fast first tier: pattern rules over the changed head lines, before the engine
                # and inside the same deadline (Requirement 9.3). Advisory, never admitted.
                try:
                    spans = head_spans(context)
                    basis_commit = comparison.base_oid if quick_basis == "last_commit" else None
                    quick = _prepare(lambda: run_quick_tier(tip.root, spans, basis=quick_basis, basis_commit=basis_commit), budget)
                except Exception as error:  # noqa: BLE001 -- a failed quick tier is a stated gap, never the end of the check
                    quick = QuickTierResult("failed", quick_basis, reason=type(error).__name__)
                    reasons.append("quick_rules_failed:" + type(error).__name__)
                stage_started = _stage(timings, stage, stage_started)
                if engine != "inline":
                    reasons.append("engine_" + engine)
                if engine == "inline":
                    stage = "head"
                    head_scan = scan_repository(
                        tip.root,
                        head_regions,
                        backend=head_backend,
                        budget=scan_budget,
                        execution_budget=budget,
                        ranking_mode="evidence",
                        unit="repository",
                        population_complete=all(m.complete for m in manifests),
                        change_context=context,
                    )
                    records.append(ScanRecord(comparison, "head", head_scan))
                    stage_started = _stage(timings, stage, stage_started)
                    stage = "base"
                    delta = compare_targeted_base(
                        old.root,
                        head=head_scan,
                        head_regions=head_regions,
                        base_regions=base_regions,
                        context=context,
                        backend=base_backend,
                        execution_budget=budget,
                        scan_budget=scan_budget,
                        head_semantics=provenance["semantics"],
                        base_semantics=provenance["semantics"],
                        cache=cache,
                        cache_semantics=cache_semantics,
                    )
                    if delta.base_scan is not None:
                        records.append(ScanRecord(comparison, "base", delta.base_scan))
                    reasons.extend(delta.coverage_reasons)
                    stage_started = _stage(timings, stage, stage_started)
                    stage = "admission"
                    candidates = tuple(
                        AdmissionCandidate(
                            c,
                            CapabilityKey(
                                c.head_operation.question.language if c.head_operation else "unknown",
                                "unspecified",
                                "unspecified",
                                c.family,
                                c.head_operation.mechanism if c.head_operation else "taint",
                                "unreviewed",
                                delta.semantics,
                            ),
                            comparison,
                            dependency_gaps(c, context=context, head=head_scan, base=delta.base_scan),
                        )
                        for c in delta.candidates
                    )
                    admission = _prepare(lambda: _admit(candidates, provenance), budget)
                    stage_started = _stage(timings, stage, stage_started)
                    if record_vetoes:
                        stage, stage_started = "recorded_vetoes", time.monotonic()
                        assert context is not None
                        recorded = _prepare(
                            lambda: _record(head_scan, delta, context, comparison, tuple(reasons), admission, provenance, declarations),
                            budget,
                        )
                        experimental = {**recorded, "status": "evaluated"}
                        stage_started = _stage(timings, stage, stage_started)
                stage = "cleanup"
        timings[stage + "_seconds"] = time.monotonic() - stage_started
    except Exception as error:
        reasons.append(stage + "_failed:" + type(error).__name__)
        timings[stage + "_seconds"] = time.monotonic() - stage_started
        if experimental.get("status") == "not_evaluated":
            experimental["reason"] = reasons[-1]
    if time.monotonic() >= budget.deadline_monotonic:
        reasons.append("deadline_exhausted")
        if experimental.get("status") == "not_evaluated":
            experimental.setdefault("reason", "deadline_exhausted")
    timings["graph_hits"] = sum(item.graph_hits for item in reused)
    timings["query_hits"] = sum(item.query_hits for item in reused)
    head_records = [r.scan for r in records if r.side == "head"]
    for record in records:
        if record.scan.scope is not None:
            reasons.extend(record.scan.scope.unresolved_boundaries)
    if not records or not all(_completed_scan(r.scan) for r in records):
        reasons.append("analysis_incomplete")
    if not any(r.side == "base" for r in records):
        reasons.append("base_comparison_unavailable")
    admission = replace(admission, coverage_reasons=tuple(dict.fromkeys((*admission.coverage_reasons, *reasons))))
    analyses: tuple[ComparisonAnalysis, ...] = ()
    if comparison is not None:
        scopes = tuple(scan.scope for scan in head_records if scan.scope is not None)
        completed = tuple(outcome.identity for scan in head_records for outcome in scan.question_outcomes if outcome.status == "completed")
        coverage: Literal["unavailable", "incomplete", "complete_within_scope"] = (
            "unavailable" if not records else "incomplete" if admission.coverage_reasons else "complete_within_scope"
        )
        analyses = (ComparisonAnalysis(comparison, scopes, completed, coverage),)
    result = admission_push_result(admission, analyses, config=config)
    timings["total_seconds"] = time.monotonic() - started
    return PushReport(
        result,
        admission,
        tuple(records),
        provenance,
        timings,
        snapshots=tuple(manifests),
        change_context=context,
        experimental=experimental,
        quick=(quick.to_payload(),) if quick is not None else (),
    )


def _last_commit_quick(
    repository: Path, comparison: PushComparison, *, artifact: Path, config: PushConfig, budget: ExecutionBudget
) -> PushReport | None:
    """Quick rules over the files the pushed head's last commit changed; None for a root commit."""
    try:
        parent = SnapshotAdapter(repository, execution_budget=budget).resolve_replay(comparison.head_oid + "^", comparison.head_oid)
    except Exception:  # noqa: BLE001 -- a root commit has no parent to compare with
        return None
    resolved = PushComparison(comparison.head_oid, parent.base_oid, "new_branch_last_commit", comparison.refs)
    assert resolved.base_oid is not None
    return _analyze(
        repository,
        base=resolved.base_oid,
        head=resolved.head_oid,
        artifact=artifact,
        config=config,
        execution_budget=budget,
        resolved=resolved,
        engine="off",
        quick_basis="last_commit",
    )


def _background(
    repository: Path,
    pending: Sequence[tuple[str | None, str]],
    artifact: Path,
    deadline: float,
    max_regions: int,
    cache_dir: Path | None,
) -> dict[str, object]:
    """Earlier background results to show, and at most one new background engine run (task 17.5)."""
    from openultrasast.push import background

    state: dict[str, object] = {"previous": background.previous(artifact, repository)}
    ready = [(base, head) for base, head in pending if base is not None]
    if ready:
        base, head = ready[0]
        assert base is not None
        state.update(
            background.start(
                repository,
                base=base,
                head=head,
                artifact=artifact,
                deadline_seconds=deadline,
                max_regions=max_regions,
                cache_dir=cache_dir,
            )
        )
        if len(ready) > 1:
            state["not_started"] = [h for _, h in ready[1:]]  # one engine at a time
    elif pending:
        state.update(status="not_started", reason="no base revision to compare against")
    return state


def _deliver(repository: Path, report: PushReport, artifact: Path, budget: ExecutionBudget) -> ReportDelivery:
    if artifact.resolve().is_relative_to(repository.resolve()):
        error = "artifact_inside_repository"
        return ReportDelivery(report.result, render_report(report, artifact=None, error=error), None, error)
    return deliver_report(report, artifact, execution_budget=budget)


def replay(
    repository: Path,
    *,
    base: str,
    head: str,
    artifact: Path,
    config: PushConfig | None = None,
    backend: Any = None,
    max_regions: int = 500,
    cache_dir: Path | None = None,
    model_config: Path | None = None,
    record_vetoes: bool = False,
    declarations: Path | None = None,
    engine: Literal["inline", "off", "background"] = "inline",
    background_deadline: float = 900.0,
) -> ReportDelivery:
    config = config or PushConfig()
    if declarations is not None and not record_vetoes:
        raise ValueError("experimental declarations require the recorded-veto evaluation flag")
    budget = ExecutionBudget(time.monotonic() + config.deadline_seconds, config.cancellation_allowance_seconds)
    report = _analyze(
        repository,
        base=base,
        head=head,
        artifact=artifact,
        config=config,
        backend=backend,
        max_regions=max_regions,
        cache_dir=cache_dir,
        execution_budget=budget,
        record_vetoes=record_vetoes,
        declarations=declarations,
        engine=engine,
    )
    pending = [(base, head)] if engine == "background" else []
    report = replace(report, background=_background(repository, pending, artifact, background_deadline, max_regions, cache_dir))
    from openultrasast.push.assistance import assist

    report = assist(report, config=model_config, budget=budget)
    return _deliver(repository, report, artifact, budget)


def push(
    repository: Path,
    *,
    updates: str | Callable[[], str],
    remote_name: str,
    remote_url: str,
    artifact: Path,
    config: PushConfig | None = None,
    backend: Any = None,
    max_regions: int = 500,
    cache_dir: Path | None = None,
    model_config: Path | None = None,
    execution_budget: ExecutionBudget | None = None,
    engine: Literal["inline", "off", "background"] = "inline",
    background_deadline: float = 900.0,
) -> ReportDelivery:
    """Consume one Git transaction. Resolution, every comparison and reporting share a deadline."""
    config = config or PushConfig()
    if max_regions < 1:
        raise ValueError("max_regions must be positive")
    if cache_dir is not None and cache_dir.resolve().is_relative_to(repository.resolve()):
        raise ValueError("cache directory must be outside the analyzed repository")
    started = time.monotonic()
    budget = execution_budget or ExecutionBudget(started + config.deadline_seconds, config.cancellation_allowance_seconds)
    reports: list[PushReport] = []
    analyses: list[ComparisonAnalysis] = []
    fallbacks: list[PushReport] = []
    reasons: list[str] = []
    resolution = None
    try:
        data = _prepare(updates, budget) if callable(updates) else updates
        if len(data.encode("utf-8")) > 1024 * 1024:
            raise ValueError("push_input_limit")
        adapter = SnapshotAdapter(repository, comparison_base=config.comparison_base, execution_budget=budget)
        resolution = adapter.resolve_updates(data)
        reasons.extend("resolution:" + row.disposition for row in resolution.updates if row.disposition not in ("ready", "deleted"))
        for comparison in resolution.comparisons:
            if comparison.base_oid is None or time.monotonic() >= budget.deadline_monotonic:
                analyses.append(ComparisonAnalysis(comparison, (), (), "unavailable"))
                reasons.append("base_comparison_unavailable" if comparison.base_oid is None else "deadline_exhausted")
                if comparison.base_oid is None:
                    # A new branch has no remote base. Say what was checked instead: the quick rules
                    # over the files its last commit changed (Requirement 9.2, 9.3).
                    fallback = _last_commit_quick(repository, comparison, artifact=artifact, config=config, budget=budget)
                    if fallback is not None:
                        fallbacks.append(fallback)
                continue
            report = _analyze(
                repository,
                base=comparison.base_oid,
                head=comparison.head_oid,
                artifact=artifact,
                config=config,
                backend=backend,
                max_regions=max_regions,
                cache_dir=cache_dir,
                execution_budget=budget,
                resolved=comparison,
                engine=engine,
            )
            reports.append(report)
            analyses.extend(report.result.analyses)
    except Exception as error:
        reasons.append("push_input_failed:" + type(error).__name__)
    if time.monotonic() >= budget.deadline_monotonic:
        reasons.append("deadline_exhausted")
    defects: dict[str, ActionableDefect] = {}
    for report in reports:
        for defect in report.admission.defects:
            previous = defects.get(defect.defect_id)
            if previous is None:
                defects[defect.defect_id] = defect
            else:
                fields = ("witnesses", "locations", "comparisons", "refs", "change_evidence", "consequences", "repairs", "evaluation_ids")
                defects[defect.defect_id] = replace(
                    previous,
                    **cast(dict[str, Any], {key: tuple(dict.fromkeys((*getattr(previous, key), *getattr(defect, key)))) for key in fields}),
                )
    admission = AdmissionResult(
        tuple(defects.values()),
        tuple(d for r in reports for d in r.admission.dispositions),
        tuple(dict.fromkeys((*reasons, *(g for r in reports for g in r.admission.coverage_reasons)))),
    )
    result = admission_push_result(admission, tuple(analyses), config=config)
    if resolution is not None and not resolution.comparisons and not reasons:
        result = PushResult((), "none", "not_applicable", "allow", ())
    provenance = dict(reports[0].provenance) if reports else {key: "not_run" for key in ("engine", "facts", "queries", "policy")}
    # Remote URLs can contain credentials. Retain identity without persisting secrets.
    provenance.update(
        repository=str(repository.absolute()),
        model="disabled",
        remote_name=remote_name or "unnamed",
        remote_url_sha256=hashlib.sha256(remote_url.encode()).hexdigest(),
    )
    timings = {f"comparison_{i}_{k}": v for i, r in enumerate(reports) for k, v in r.timings.items()}
    timings.update(
        total_seconds=time.monotonic() - started,
        deadline_seconds=config.deadline_seconds,
        cancellation_allowance_seconds=config.cancellation_allowance_seconds,
    )
    report = PushReport(
        result,
        admission,
        tuple(s for r in reports for s in r.scans),
        provenance,
        timings,
        resolution=resolution,
        snapshots=tuple(s for r in reports for s in r.snapshots),
        change_contexts=tuple(r.change_context for r in reports if r.change_context is not None),
        quick=tuple(q for r in (*reports, *fallbacks) for q in r.quick),
        background=_background(
            repository,
            [(r.change_context.base_revision, r.change_context.head_revision) for r in reports if r.change_context]
            if engine == "background"
            else [],
            artifact,
            background_deadline,
            max_regions,
            cache_dir,
        ),
    )
    from openultrasast.push.assistance import assist

    report = assist(report, config=model_config, budget=budget)
    return _deliver(repository, report, artifact, budget)