feat(pre-push-safety-net): record admission vetoes and meet M1a on NodeGoat

Add group 9 to the approved task plan as the bounded owner amendment release
milestone M1a requires, implement the task-local recorded-veto evaluation flag,
and run the three pinned NodeGoat revisions under it.

The flag re-runs the production comparison and admission over a lifted copy of
the same evidence and records each veto instead of applying it. Recognized
completeness vetoes are lifted and logged; failures never are, because silence
from a failure is not a quiet fixed twin. It is off by default, accepted only on
explicit base/head replay, refused in Git stdin mode, labels every payload it
produces, and admits no defect even when a lifted candidate otherwise would.

M1a outcome, inside its one-week box and with expectations frozen on disk first:
the three entailed <lambda>2 witnesses are present at contributions.js 32-34; the
fixed twin was genuinely analyzed and quiet, answering its target question with
zero rows and no deadline reason while the same findings appear as base-only; the
unchanged revision was analyzed and recorded unchanged. The kill criterion did
not trigger.

Five vetoes fired and were recorded. M1a also corrected the diagnosis: with all
five lifted the detection still stops at operation_correspondence_unresolved,
because a newly introduced operation sits on a changed line and has no base
anchor. The unchanged revision resolves its base location on the same lines and
classifies as unchanged, which isolates correspondence as the mechanism. That
fifth veto was not in the earlier four-veto diagnosis and is now M1b's central
repair.

Diagnosis only. No defect admitted, no capability enabled, no gate changed, no
latency claim; the 900-second development budget cannot measure the hook gate.
Rollout and task 8.3 remain NO-GO.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/cli.py

@@ -132,6 +132,12 @@ def main(argv: list[str] | None = None) -> int:
     replay_parser.add_argument("--max-regions", type=int, default=500)
     replay_parser.add_argument("--mode", choices=("advisory", "blocking"), default="advisory")
     replay_parser.add_argument("--incomplete-coverage", choices=("allow", "block"), default="allow")
+    replay_parser.add_argument(
+        "--experimental-record-vetoes",
+        action="store_true",
+        help="explicit --base/--head replay only: record each comparison and admission veto per raw finding instead of "
+        "applying it; output is labeled experimental, changes no result and enables no hook capability",
+    )
 
     scan = subparsers.add_parser("scan", help="scan a local repository")
     scan.add_argument("path", type=Path)
@@ -235,9 +241,11 @@ def main(argv: list[str] | None = None) -> int:
             )
             if args.base and args.prior_hook:
                 parser.error("--prior-hook requires Git stdin mode")
+            if args.experimental_record_vetoes and not args.base:
+                parser.error("--experimental-record-vetoes requires explicit --base/--head replay; it is never a hook capability")
             prior_data = None
             if args.base:
-                delivery = replay(args.path, base=args.base, head=args.head, **options)
+                delivery = replay(args.path, base=args.base, head=args.head, record_vetoes=args.experimental_record_vetoes, **options)
             else:
                 # multiprocessing closes sys.stdin in its child; duplicate the Git pipe first.
                 with os.fdopen(os.dup(0), "rb") as stream:

src/openultrasast/push/report.py

@@ -41,9 +41,15 @@ class PushReport:
     change_context: ChangeContext | None = None
     change_contexts: tuple[ChangeContext, ...] = ()
     model_assistance: Mapping[str, object] = field(default_factory=dict)
+    # Labeled M1a recorded-veto evaluation. Present only under the explicit replay flag;
+    # it is retained beside the decision and never feeds admission or enforcement.
+    experimental: Mapping[str, object] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "model_assistance", MappingProxyType(dict(self.model_assistance)))
+        object.__setattr__(self, "experimental", MappingProxyType(dict(self.experimental)))
+        if self.experimental and not str(self.experimental.get("label", "")).startswith("EXPERIMENTAL"):
+            raise ValueError("experimental evaluation output must carry its experimental label")
         object.__setattr__(self, "provenance", MappingProxyType(dict(self.provenance)))
         object.__setattr__(self, "timings", MappingProxyType(dict(self.timings)))
         ids = tuple(defect.defect_id for defect in self.admission.defects)
@@ -126,6 +132,7 @@ def _write_artifact(report: PushReport, target: Path, temporary: Path, connectio
                     "timings": dict(report.timings),
                     "timing_scope": "Caller stage measurements; publication and rendering duration is returned in ReportDelivery.",
                     "resolution": report.resolution.to_payload() if report.resolution else None,
+                    "experimental": dict(report.experimental),
                 }
                 json.dump(payload, stream, ensure_ascii=True, allow_nan=False, indent=2)
                 stream.write("\n")
@@ -187,6 +194,20 @@ def _short(value: str, limit: int = 180) -> str:
     return json.dumps(clipped, ensure_ascii=True)[1:-1]
 
 
+def _experimental_line(experimental: Mapping[str, object]) -> str:
+    if experimental.get("status") != "evaluated":
+        return f"recorded-veto evaluation was not produced ({_short(str(experimental.get('reason', 'not_evaluated')))}); not an alert."
+    findings = experimental.get("findings")
+    count = len(findings) if isinstance(findings, list) else 0
+    vetoes = sum(len(f.get("vetoes", [])) for f in findings if isinstance(f, dict)) if isinstance(findings, list) else 0
+    base = experimental.get("base_only_findings")
+    quiet = len(base) if isinstance(base, list) else 0
+    return (
+        f"recorded-veto evaluation retained {count} raw finding(s) with {vetoes} recorded veto(es) and "
+        f"{quiet} base-only finding(s) in the detailed result; recorded, not applied; not an alert."
+    )
+
+
 def render_report(report: PushReport, *, artifact: Path | None, error: str | None = None) -> str:
     lines: list[str] = []
     defects = report.admission.defects
@@ -250,6 +271,8 @@ def render_report(report: PushReport, *, artifact: Path | None, error: str | Non
         notices.append("Details could not be saved. Retry with a writable artifact path and available reporting time.")
     if notices:
         lines.append("Notice: " + " ".join(notices))
+    if report.experimental:
+        lines.append("Experimental: " + _experimental_line(report.experimental))
     if artifact is not None:
         lines.append("Details: " + _short(str(artifact), 4096))
     if report.result.push_disposition == "block":

src/openultrasast/push/runner.py

@@ -8,7 +8,7 @@ import os
 import pickle
 import selectors
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from contextlib import nullcontext
 from dataclasses import asdict, replace
 from pathlib import Path
@@ -18,10 +18,10 @@ 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 ExecutionBudget
+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 ScanBudget, scan_repository
+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
@@ -32,6 +32,7 @@ from openultrasast.push.policy import (
     AdmissionCandidate,
     AdmissionResult,
     CapabilityKey,
+    DeltaComparison,
     admission_push_result,
     admit_candidates,
     compare_targeted_base,
@@ -40,6 +41,7 @@ from openultrasast.push.policy import (
 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, record_vetoes
 
 _T = TypeVar("_T")
 
@@ -168,6 +170,31 @@ def _admit(candidates: tuple[AdmissionCandidate, ...], provenance: dict[str, str
     return result
 
 
+def _record(
+    head: ModelScanResult,
+    delta: DeltaComparison,
+    context: ChangeContext,
+    comparison: PushComparison,
+    reasons: Sequence[str],
+    admission: AdmissionResult,
+    provenance: dict[str, str],
+) -> dict[str, object]:
+    """Experimental M1a: the same policy over lifted evidence, reported only, never admitted."""
+    registry = load_registry(current=provenance)
+    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,
+    )
+    return report.to_payload()
+
+
 def _discovery_identity() -> str:
     root = Path(__file__).resolve().parents[1]
     # Discovery invokes the shipped mapper, preprocessing, layout and family specs.
@@ -187,8 +214,15 @@ def _analyze(
     cache_dir: Path | None = None,
     execution_budget: ExecutionBudget,
     resolved: PushComparison | None = None,
+    record_vetoes: bool = False,
 ) -> PushReport:
-    """Run a single explicit comparison; exit policy never stands for coverage."""
+    """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")
@@ -208,6 +242,7 @@ def _analyze(
     admission = AdmissionResult((), (), ())
     comparison = resolved
     context = None
+    experimental: dict[str, object] = {"label": LABEL, "flag": FLAG, "status": "not_evaluated"} if record_vetoes else {}
     timings: dict[str, float] = {
         "deadline_seconds": config.deadline_seconds,
         "cancellation_allowance_seconds": config.cancellation_allowance_seconds,
@@ -340,13 +375,25 @@ def _analyze(
                 )
                 admission = _prepare(lambda: _admit(candidates, provenance), budget)
                 timings[stage + "_seconds"] = time.monotonic() - 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), budget
+                    )
+                    experimental = {**recorded, "status": "evaluated"}
+                    timings[stage + "_seconds"] = time.monotonic() - stage_started
                 stage, stage_started = "cleanup", time.monotonic()
         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"]
@@ -368,8 +415,16 @@ def _analyze(
         analyses = (ComparisonAnalysis(comparison, scopes, completed, coverage),)
     result = admission_push_result(admission, analyses, config=config)
     timings["total_seconds"] = time.monotonic() - started
-    report = PushReport(result, admission, tuple(records), provenance, timings, snapshots=tuple(manifests), change_context=context)
-    return report
+    return PushReport(
+        result,
+        admission,
+        tuple(records),
+        provenance,
+        timings,
+        snapshots=tuple(manifests),
+        change_context=context,
+        experimental=experimental,
+    )
 
 
 def _deliver(repository: Path, report: PushReport, artifact: Path, budget: ExecutionBudget) -> ReportDelivery:
@@ -390,6 +445,7 @@ def replay(
     max_regions: int = 500,
     cache_dir: Path | None = None,
     model_config: Path | None = None,
+    record_vetoes: bool = False,
 ) -> ReportDelivery:
     config = config or PushConfig()
     budget = ExecutionBudget(time.monotonic() + config.deadline_seconds, config.cancellation_allowance_seconds)
@@ -403,6 +459,7 @@ def replay(
         max_regions=max_regions,
         cache_dir=cache_dir,
         execution_budget=budget,
+        record_vetoes=record_vetoes,
     )
     from openultrasast.push.assistance import assist
 

src/openultrasast/push/vetoes.py

@@ -0,0 +1,434 @@
+"""Experimental recorded-veto evaluation for release milestone M1a.
+
+Record every comparison and admission veto per raw finding instead of applying it, so
+a maintainer can see which vetoes discard a detection the arbiter already reported.
+
+This module never admits a defect, enables a capability or changes the production
+result. The production comparison and admission have already run on the applied path.
+Here the same policy functions run again over a *lifted* view of the same evidence in
+which recognized completeness vetoes are removed and recorded. Failures are never
+lifted: deadline exhaustion, unanswered queries, unreadable input, empty graphs and
+unparsed files stay applied, because silence from failure is not a quiet fixed twin.
+Every payload produced here carries an experimental label.
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from collections.abc import Sequence
+from dataclasses import dataclass, replace
+
+from openultrasast.contracts import Contract
+from openultrasast.model.contracts import ChangeContext, ExecutionBudget, QuestionIdentity
+from openultrasast.model.scan import ModelScanResult
+from openultrasast.push.contracts import PushComparison
+from openultrasast.push.policy import (
+    AdmissionCandidate,
+    CandidateDelta,
+    CapabilityAdmission,
+    CapabilityKey,
+    EvidenceOperation,
+    _base_location,
+    _compare_evidence,
+    _context_boundaries,
+    _counterpart,
+    admit_candidates,
+)
+
+FLAG = "--experimental-record-vetoes"
+LABEL = (
+    "EXPERIMENTAL recorded-veto evaluation (release milestone M1a). Listed vetoes were recorded, not applied. "
+    "Nothing in this section is an alert, an admitted defect, a capability or a comparison result; the production "
+    "result and admission above remain authoritative."
+)
+
+# Outcome reasons that scan.py applies to an *answered* question after the fact. Lifting them
+# restores the answered rows; an unanswered or not-arbitrated question is never lifted.
+LIFTED_OUTCOME_REASONS = frozenset({"change_context_incomplete", "graph_incomplete"})
+# Physical vendor exclusion is retained; only its global completion veto is recorded here.
+LIFTED_DEGRADATION_REASONS = frozenset({"vendor_semantics_unresolved"})
+# Boundary kinds diagnosed in release-remediation-research.md as completeness vetoes rather
+# than input failures. The kind is one colon-separated segment of a boundary string.
+LIFTED_BOUNDARY_KINDS = frozenset(
+    {
+        "vendor_semantics_unresolved",
+        "ambiguous_line_correspondence",
+        "context_projection_unavailable",
+        "context_scope_empty",
+        "dynamic_external_or_depth_context_unresolved",
+        "context_location_unavailable",
+        "configuration_dependency_projection_unavailable",
+        "deleted_dependency_projection_unavailable",
+    }
+)
+# Admission reasons that this flag can never remove; they are recorded with that explanation.
+_EXPECTED_ADMISSION = {
+    "capability_unavailable": "no evaluated capability is enabled; the recorded-veto flag cannot enable one",
+    "capability_disabled": "capability declared but not enabled; the recorded-veto flag cannot enable one",
+    "capability_unevaluated": "capability not qualified; the recorded-veto flag cannot qualify one",
+}
+
+
+@dataclass(frozen=True)
+class RecordedVeto(Contract):
+    stage: str
+    veto: str
+    detail: str
+    lifted: bool
+
+
+@dataclass(frozen=True)
+class QuestionReceipt(Contract):
+    identity: QuestionIdentity
+    status: str
+    reason: str
+    rows: int | None
+    lifted: bool
+
+
+@dataclass(frozen=True)
+class OutcomeCount(Contract):
+    status: str
+    reason: str
+    count: int
+
+
+@dataclass(frozen=True)
+class ScanSummary(Contract):
+    findings: int
+    outcomes: tuple[OutcomeCount, ...]
+    lifted: tuple[OutcomeCount, ...]
+    degradations: tuple[str, ...]
+    lifted_degradations: tuple[str, ...]
+    boundaries_by_kind: tuple[OutcomeCount, ...]
+    changed_path_questions: tuple[QuestionReceipt, ...]
+
+
+@dataclass(frozen=True)
+class RecordedFinding(Contract):
+    family: str
+    site: str
+    rung: str
+    witness: str | None
+    question: QuestionIdentity | None
+    operation: EvidenceOperation | None
+    operation_method: str | None
+    base_location: str | None
+    production_novelty: str
+    production_reason: str
+    production_admission: tuple[str, ...]
+    recorded_novelty: str
+    recorded_reason: str
+    recorded_admission: tuple[str, ...]
+    vetoes: tuple[RecordedVeto, ...]
+
+
+@dataclass(frozen=True)
+class BaseOnlyFinding(Contract):
+    family: str
+    site: str
+    rung: str
+    witness: str | None
+    head_counterparts: tuple[QuestionReceipt, ...]
+
+
+@dataclass(frozen=True)
+class RecordedVetoReport(Contract):
+    label: str
+    flag: str
+    lifted_kinds: tuple[str, ...]
+    head: ScanSummary
+    base: ScanSummary | None
+    findings: tuple[RecordedFinding, ...]
+    base_only_findings: tuple[BaseOnlyFinding, ...]
+    lifted_coverage_reasons: tuple[str, ...]
+    remaining_coverage_reasons: tuple[str, ...]
+
+
+def boundary_kind(boundary: str) -> str | None:
+    """Return the recognized lifted kind inside a boundary string, if any."""
+    return next((segment for segment in boundary.split(":") if segment in LIFTED_BOUNDARY_KINDS), None)
+
+
+def _lift_scan(scan: ModelScanResult) -> tuple[ModelScanResult, Counter[str]]:
+    lifted: Counter[str] = Counter()
+    outcomes = []
+    for outcome in scan.question_outcomes:
+        if outcome.status == "unresolved" and outcome.reason in LIFTED_OUTCOME_REASONS and outcome.raw_rows_json is not None:
+            lifted["question:" + outcome.reason] += 1
+            outcomes.append(replace(outcome, status="completed", reason="recorded_veto:" + outcome.reason))
+        else:
+            outcomes.append(outcome)
+    degradations = []
+    for degradation in scan.degradations:
+        if str(degradation.get("reason")) in LIFTED_DEGRADATION_REASONS:
+            lifted["degradation:" + str(degradation.get("reason"))] += 1
+        else:
+            degradations.append(degradation)
+    scope = scan.scope
+    if scope is not None:
+        kept = []
+        for boundary in scope.unresolved_boundaries:
+            kind = boundary_kind(boundary)
+            if kind is None:
+                kept.append(boundary)
+            else:
+                lifted["boundary:" + kind] += 1
+        scope = replace(scope, unresolved_boundaries=tuple(kept))
+    return replace(scan, question_outcomes=tuple(outcomes), degradations=tuple(degradations), scope=scope), lifted
+
+
+def _lift_context(context: ChangeContext) -> tuple[ChangeContext, tuple[str, ...]]:
+    lifted = tuple(b for b in context.unresolved_boundaries if boundary_kind(b) is not None)
+    kept = tuple(b for b in context.unresolved_boundaries if boundary_kind(b) is None)
+    return replace(context, unresolved_boundaries=kept), lifted
+
+
+def _rows(raw: str | None) -> int | None:
+    if raw is None:
+        return None
+    rows = json.loads(raw)
+    return len(rows) if isinstance(rows, list) else None
+
+
+def _receipt(scan: ModelScanResult, lifted: ModelScanResult, identity: QuestionIdentity) -> QuestionReceipt | None:
+    original = next((o for o in scan.question_outcomes if o.identity == identity), None)
+    if original is None:
+        return None
+    after = next((o for o in lifted.question_outcomes if o.identity == identity), original)
+    return QuestionReceipt(identity, original.status, original.reason, _rows(original.raw_rows_json), after.status != original.status)
+
+
+def _summary(scan: ModelScanResult, lifted: ModelScanResult, counts: Counter[str], context: ChangeContext) -> ScanSummary:
+    outcomes = Counter((o.status, o.reason) for o in scan.question_outcomes)
+    boundaries = Counter(boundary_kind(b) or b.split(":")[0] for b in (scan.scope.unresolved_boundaries if scan.scope else ()))
+    changed = {context.decode_path(p) for p in context.changed_paths}
+    receipts = tuple(
+        receipt
+        for o in scan.question_outcomes
+        if o.identity.path in changed
+        for receipt in (_receipt(scan, lifted, o.identity),)
+        if receipt is not None
+    )
+    return ScanSummary(
+        len(scan.findings),
+        tuple(OutcomeCount(status, reason, count) for (status, reason), count in sorted(outcomes.items())),
+        tuple(
+            OutcomeCount(key.split(":", 1)[0], key.split(":", 1)[1], count)
+            for key, count in sorted(counts.items())
+            if not key.startswith("boundary:")
+        ),
+        tuple(dict.fromkeys(str(d.get("reason")) for d in scan.degradations)),
+        tuple(key.split(":", 1)[1] for key in sorted(counts) if key.startswith("degradation:")),
+        tuple(OutcomeCount("boundary", kind, count) for kind, count in sorted(boundaries.items())),
+        receipts,
+    )
+
+
+def _method(op: EvidenceOperation) -> str | None:
+    row = json.loads(op.detail)
+    value = row.get("sinkMethod") if op.mechanism == "taint" else row.get("opMethod") if op.mechanism == "dominance" else None
+    return value if isinstance(value, str) and value.strip() else None
+
+
+def _boundary_vetoes(stage: str, boundaries: Sequence[str]) -> list[RecordedVeto]:
+    return [RecordedVeto(stage, boundary_kind(b) or b, b, boundary_kind(b) is not None) for b in boundaries]
+
+
+def _degradation_vetoes(stage: str, scan: ModelScanResult) -> list[RecordedVeto]:
+    return [
+        RecordedVeto(
+            stage,
+            "degradation:" + str(d.get("reason")),
+            json.dumps(dict(d), sort_keys=True, default=str),
+            str(d.get("reason")) in LIFTED_DEGRADATION_REASONS,
+        )
+        for d in scan.degradations
+    ]
+
+
+def _question_veto(stage: str, receipt: QuestionReceipt | None, identity: QuestionIdentity) -> list[RecordedVeto]:
+    if receipt is None:
+        return [RecordedVeto(stage, "question_missing", identity.question_id, False)]
+    if receipt.status == "completed":
+        return []
+    detail = f"{identity.question_id} status={receipt.status} rows={receipt.rows}"
+    return [RecordedVeto(stage, receipt.reason, detail, receipt.lifted)]
+
+
+def _finding_vetoes(
+    op: EvidenceOperation | None,
+    *,
+    head: ModelScanResult,
+    lifted_head: ModelScanResult,
+    base: ModelScanResult | None,
+    lifted_base: ModelScanResult | None,
+    context: ChangeContext,
+    recorded: CandidateDelta,
+    admission: Sequence[str],
+) -> tuple[RecordedVeto, ...]:
+    vetoes: list[RecordedVeto] = []
+    if op is None:
+        vetoes.append(RecordedVeto("comparison", recorded.reason, "no completed head operation matches this finding after lifting", False))
+    else:
+        question = op.question
+        vetoes.extend(_question_veto("head_question", _receipt(head, lifted_head, question), question))
+        vetoes.extend(_boundary_vetoes("head_scan", _context_boundaries(head, question)))
+        vetoes.extend(_degradation_vetoes("head_scan", head))
+        method = _method(op)
+        if question.function is None or (method is not None and question.function != method):
+            vetoes.append(
+                RecordedVeto(
+                    "identity",
+                    "query_identity_differs_from_witness",
+                    f"question function={question.function!r}; witness method={method!r}; recorded, not a policy veto",
+                    False,
+                )
+            )
+        if base is None or lifted_base is None:
+            vetoes.append(RecordedVeto("base_scan", "base_unavailable", "no base scan was produced", False))
+        else:
+            counterpart = _counterpart(question, context)
+            vetoes.extend(_question_veto("base_question", _receipt(base, lifted_base, counterpart), counterpart))
+            vetoes.extend(_boundary_vetoes("base_scan", _context_boundaries(base)))
+            vetoes.extend(_degradation_vetoes("base_scan", base))
+            if base.scope is not None:
+                if not base.scope.population_complete:
+                    vetoes.append(RecordedVeto("base_scan", "population_incomplete", "base census not asserted complete", False))
+                for deferred in base.scope.deferred:
+                    if deferred.reason != "tier_zero":
+                        vetoes.append(RecordedVeto("base_scan", "deferred:" + deferred.reason, deferred.identity.question_id, False))
+            pending = [o for o in lifted_base.question_outcomes if o.status != "completed"]
+            if pending:
+                detail = f"{len(pending)} of {len(lifted_base.question_outcomes)} base questions not completed after lifting"
+                vetoes.append(RecordedVeto("base_scan", "questions_not_completed", detail, False))
+    vetoes.extend(_boundary_vetoes("change_context", context.unresolved_boundaries))
+    if op is not None and recorded.novelty == "unknown":
+        location = _base_location(op, context)
+        where = f"{location[0]}:{location[1]}" if location else "unresolved (operation on a changed line or ambiguous)"
+        vetoes.append(RecordedVeto("comparison", recorded.reason, "base location " + where, False))
+    for reason in admission:
+        explanation = _EXPECTED_ADMISSION.get(reason, "production admission reason on the lifted candidate")
+        vetoes.append(RecordedVeto("admission", reason, explanation, False))
+    return tuple(vetoes)
+
+
+def record_vetoes(
+    *,
+    head: ModelScanResult,
+    base: ModelScanResult | None,
+    context: ChangeContext,
+    comparison: PushComparison,
+    semantics: str,
+    production: Sequence[CandidateDelta],
+    production_admission: dict[str, tuple[str, ...]],
+    coverage_reasons: Sequence[str],
+    capabilities: Sequence[CapabilityAdmission] = (),
+    execution_budget: ExecutionBudget | None = None,
+) -> RecordedVetoReport:
+    """Re-run the production comparison and admission over lifted evidence and list what was lifted.
+
+    `production` and `production_admission` (keyed by delta.defect_id) are the applied results the
+    runner already produced; they are retained beside the recorded outcome for every finding.
+    """
+    lifted_head, head_counts = _lift_scan(head)
+    lifted_base, base_counts = _lift_scan(base) if base is not None else (None, Counter())
+    lifted_context, lifted_boundaries = _lift_context(context)
+    lifted_reasons = tuple(r for r in coverage_reasons if boundary_kind(r) is not None or r in LIFTED_OUTCOME_REASONS)
+    remaining = tuple(r for r in coverage_reasons if r not in lifted_reasons)
+    recorded = _compare_evidence(
+        lifted_head,
+        lifted_base,
+        context=lifted_context,
+        head_semantics=semantics,
+        base_semantics=semantics,
+        execution_budget=execution_budget,
+    )
+    candidates = tuple(
+        AdmissionCandidate(
+            delta,
+            CapabilityKey(
+                delta.head_operation.question.language if delta.head_operation else "unknown",
+                "unspecified",
+                "unspecified",
+                delta.family,
+                delta.head_operation.mechanism if delta.head_operation else "taint",
+                "unreviewed",
+                semantics,
+            ),
+            comparison,
+            remaining,
+        )
+        for delta in recorded
+    )
+    # Only the per-candidate reasons are retained. Any defect the lifted path would admit is
+    # deliberately dropped here: this section is a diagnostic record, never an alert channel.
+    admitted = admit_candidates(candidates, capabilities=capabilities)
+    recorded_admission = {d.candidate.delta.defect_id: d.reasons for d in admitted.dispositions}
+    applied = {(c.family, c.site, c.witness): c for c in production}
+    findings = []
+    for finding, delta in zip(head.findings, recorded, strict=True):
+        before = applied.get((finding.family, finding.site, finding.witness or None))
+        op = delta.head_operation
+        location = _base_location(op, context) if op else None
+        admission = recorded_admission.get(delta.defect_id, ())
+        findings.append(
+            RecordedFinding(
+                finding.family,
+                finding.site,
+                finding.rung.value,
+                finding.witness or None,
+                op.question if op else None,
+                op,
+                _method(op) if op else None,
+                f"{location[0]}:{location[1]}" if location else None,
+                before.novelty if before else "unknown",
+                before.reason if before else "production_candidate_missing",
+                production_admission.get(before.defect_id, ()) if before else (),
+                delta.novelty,
+                delta.reason,
+                admission,
+                _finding_vetoes(
+                    op,
+                    head=head,
+                    lifted_head=lifted_head,
+                    base=base,
+                    lifted_base=lifted_base,
+                    context=context,
+                    recorded=delta,
+                    admission=admission,
+                ),
+            )
+        )
+    base_only = []
+    if base is not None and lifted_base is not None:
+        head_sites = {(f.family, f.site) for f in head.findings}
+        for finding in base.findings:
+            if (finding.family, finding.site) in head_sites:
+                continue
+            path = finding.site.split(":", 1)[0]
+            counterparts = tuple(
+                receipt
+                for o in head.question_outcomes
+                if o.identity.path == path and o.identity.family == finding.family
+                for receipt in (_receipt(head, lifted_head, o.identity),)
+                if receipt is not None
+            )
+            base_only.append(BaseOnlyFinding(finding.family, finding.site, finding.rung.value, finding.witness or None, counterparts))
+    kinds = (
+        *("question:" + r for r in LIFTED_OUTCOME_REASONS),
+        *("degradation:" + r for r in LIFTED_DEGRADATION_REASONS),
+        *("boundary:" + k for k in LIFTED_BOUNDARY_KINDS),
+    )
+    return RecordedVetoReport(
+        LABEL,
+        FLAG,
+        tuple(sorted(kinds)),
+        _summary(head, lifted_head, head_counts, context),
+        _summary(base, lifted_base, base_counts, context) if base is not None and lifted_base is not None else None,
+        tuple(findings),
+        tuple(base_only),
+        tuple(dict.fromkeys((*lifted_reasons, *lifted_boundaries))),
+        remaining,
+    )