Model Versioning & Deployment

from dataclasses import asdict, dataclass, fields, replace
import hashlib
import json

@dataclass(frozen=True)
class ReleaseBundle:
    service: str
    answer_model: str
    evidence_gate: str
    prompt_compile_run: str
    tokenizer: str
    prompt_version: str
    policy_version: str
    corpus_version: str
    serving_image: str
    input_schema: str
    eval_suite: str
    evaluator_version: str
    decoding: str

    def __post_init__(self) -> None:
        if any(type(getattr(self, field.name)) is not str or not getattr(self, field.name).strip() for field in fields(self)):
            raise ValueError("release fields must be nonempty strings")

stable = ReleaseBundle(
    service="deploy-answerer",
    answer_model="hosted-model@2026-04-01",
    evidence_gate="releaseops-evidence-classifier-v2@sha256:41d8",
    prompt_compile_run="manual-template-v1",
    tokenizer="deploy-tokenizer@sha256:91aa",
    prompt_version="deploy-answerer-v1.json",
    policy_version="[email protected]",
    corpus_version="releaseops-records@sha256:corpus-5",
    serving_image="registry.example/deploy-answerer@sha256:image-a",
    input_schema="deploy-answer.v2",
    eval_suite="deploy-grounding-suite@sha256:suite-7",
    evaluator_version="claim-evidence-eval-v2",
    decoding="temperature=0",
)

candidate = replace(
    stable,
    prompt_compile_run="dspy-mipro-v2-run-052",
    prompt_version="compiled_a_hosted-2026-04-01.json",
    serving_image="registry.example/deploy-answerer@sha256:image-b",
)

print(f"stable_prompt={stable.prompt_version}")
print(f"candidate_prompt={candidate.prompt_version}")
print(f"candidate_compile_run={candidate.prompt_compile_run}")

def release_id(bundle: ReleaseBundle) -> str:
    payload = json.dumps(asdict(bundle), sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    return f"{bundle.service}@sha256:{digest}"

stable_id = release_id(stable)
candidate_id = release_id(candidate)
prompt_patch_id = release_id(
    replace(candidate, prompt_version="compiled_a_hosted-2026-04-01.b.json")
)

print(f"stable={stable_id}")
print(f"candidate={candidate_id}")
print(f"prompt_patch={prompt_patch_id}")
print(f"prompt_patch_is_new_release={prompt_patch_id != candidate_id}")

# A name in a manifest is not a digest of the file behind that name.
from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as directory:
    prompt_file = Path(directory) / stable.prompt_version
    prompt_file.write_text('{"instructions":"first fixture"}')
    first_bytes_digest = hashlib.sha256(prompt_file.read_bytes()).hexdigest()
    prompt_file.write_text('{"instructions":"changed fixture"}')
    second_bytes_digest = hashlib.sha256(prompt_file.read_bytes()).hexdigest()
    print("same_manifest_id_after_overwrite=", release_id(stable) == stable_id, sep="")
    print("prompt_bytes_changed=", first_bytes_digest != second_bytes_digest, sep="")

from copy import deepcopy

class ReleaseRegistry:
    def __init__(self) -> None:
        self._bundles: dict[str, ReleaseBundle] = {}
        self._aliases: dict[str, str] = {}

    def register(self, bundle: ReleaseBundle) -> str:
        bundle_id = release_id(bundle)
        existing = self._bundles.get(bundle_id)
        if existing is not None and existing != bundle:
            raise ValueError("release digest collision")
        self._bundles[bundle_id] = deepcopy(bundle)
        return bundle_id

    def move_alias(self, alias: str, bundle_id: str, *, expected: str | None = None) -> None:
        if type(alias) is not str or not alias.strip():
            raise ValueError("alias must be a nonempty string")
        if bundle_id not in self._bundles:
            raise KeyError(f"unregistered release: {bundle_id}")
        if self._aliases.get(alias) != expected:
            raise ValueError("stale alias update")
        self._aliases[alias] = bundle_id

    def resolve(self, alias: str) -> str:
        return self._aliases[alias]

    def bundle(self, bundle_id: str) -> ReleaseBundle:
        return deepcopy(self._bundles[bundle_id])

registry = ReleaseRegistry()
assert registry.register(stable) == stable_id
assert registry.register(candidate) == candidate_id
registry.move_alias("production", stable_id)

print(f"registered={len(registry._bundles)}")
print(f"production={registry.resolve('production')}")
print(f"candidate_registered={candidate_id in registry._bundles}")

from datetime import datetime, timedelta, timezone
import math

NOW = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc)

@dataclass(frozen=True)
class OfflineEvidence:
    release_id: str
    eval_suite: str
    evaluator_version: str
    input_schema: str
    evaluation_report: str
    supported_evidence_f1: float
    unsupported_serve_count: int
    p95_latency_ms: int
    rollback_slice_f1: float
    evaluated_count: int
    support_checked_count: int
    latency_sample_count: int
    completed_at: datetime

@dataclass(frozen=True)
class Decision:
    allowed: bool
    reason: str

def finite_between(value: object, low: float, high: float) -> bool:
    if type(value) not in (int, float):
        return False
    try:
        return math.isfinite(value) and low <= value <= high
    except OverflowError:
        return False

def fresh(completed_at: datetime, as_of: datetime, max_age: timedelta) -> bool:
    if not isinstance(completed_at, datetime) or not isinstance(as_of, datetime):
        return False
    if completed_at.utcoffset() is None or as_of.utcoffset() is None:
        return False
    return timedelta(0) <= as_of - completed_at <= max_age

def offline_gate(
    bundle: ReleaseBundle, evidence: OfflineEvidence, as_of: datetime
) -> Decision:
    if type(bundle) is not ReleaseBundle or type(evidence) is not OfflineEvidence:
        return Decision(False, "invalid release or evidence record")
    if evidence.release_id != release_id(bundle):
        return Decision(False, "evidence belongs to another release")
    if evidence.eval_suite != bundle.eval_suite:
        return Decision(False, "evaluation suite changed")
    if evidence.evaluator_version != bundle.evaluator_version:
        return Decision(False, "evaluator version changed")
    if evidence.input_schema != bundle.input_schema:
        return Decision(False, "schema mismatch")
    if type(evidence.evaluation_report) is not str or not evidence.evaluation_report.strip():
        return Decision(False, "evaluation report missing")
    if not fresh(evidence.completed_at, as_of, timedelta(hours=24)):
        return Decision(False, "report is stale, future-dated, or lacks a timezone")
    if type(evidence.evaluated_count) is not int or evidence.evaluated_count != 200:
        return Decision(False, "suite requires 200 evaluated cases")
    if any(type(count) is not int or count != evidence.evaluated_count for count in (
        evidence.support_checked_count, evidence.latency_sample_count
    )):
        return Decision(False, "support or latency coverage incomplete")
    if not all(finite_between(score, 0, 1) for score in (
        evidence.supported_evidence_f1, evidence.rollback_slice_f1
    )) or not finite_between(evidence.p95_latency_ms, 0, math.inf):
        return Decision(False, "invalid metric")
    if type(evidence.unsupported_serve_count) is not int or not 0 <= evidence.unsupported_serve_count <= evidence.support_checked_count:
        return Decision(False, "unsupported count must be a valid integer")
    if evidence.supported_evidence_f1 < 0.92:
        return Decision(False, "supported_evidence_f1 below 0.92")
    if evidence.rollback_slice_f1 < 0.90:
        return Decision(False, "rollback slice below 0.90")
    if evidence.unsupported_serve_count != 0:
        return Decision(False, "unsupported answer was served")
    if evidence.p95_latency_ms > 500:
        return Decision(False, "p95 latency exceeds 500 ms")
    return Decision(True, "offline gate passed")

candidate_offline = OfflineEvidence(
    release_id=candidate_id,
    eval_suite=candidate.eval_suite,
    evaluator_version=candidate.evaluator_version,
    input_schema=candidate.input_schema,
    evaluation_report="reports/candidate-suite-7-redacted.json",
    supported_evidence_f1=0.93,
    unsupported_serve_count=0,
    p95_latency_ms=472,
    rollback_slice_f1=0.91,
    evaluated_count=200,
    support_checked_count=200,
    latency_sample_count=200,
    completed_at=NOW - timedelta(hours=2),
)
weaker_candidate = replace(candidate_offline, supported_evidence_f1=0.89)
changed_evaluator = replace(candidate_offline, evaluator_version="claim-evidence-eval-v3")

print(f"candidate={offline_gate(candidate, candidate_offline, NOW)}")
print(f"weak_metric={offline_gate(candidate, weaker_candidate, NOW)}")
print(f"changed_evaluator={offline_gate(candidate, changed_evaluator, NOW)}")

def open_canary(
    bundle: ReleaseBundle, evidence: OfflineEvidence, as_of: datetime
) -> Decision:
    decision = offline_gate(bundle, evidence, as_of)
    if decision.allowed:
        registry.move_alias("canary", release_id(bundle))
    return decision

canary_decision = open_canary(candidate, candidate_offline, NOW)

print(f"canary_opened={canary_decision.allowed}")
print(f"canary={registry.resolve('canary')}")
print(f"production_still_stable={registry.resolve('production') == stable_id}")

@dataclass(frozen=True)
class RecordedTrace:
    request_id: str
    origin_release: str
    prompt_version: str
    evidence_version: str
    evidence_gate: str
    tool_sequence: tuple[str, ...]
    evidence_supports_claim: bool
    served_unsupported_claim: bool
    temperature: float = 0.0
    seed: int = 0
    model_snapshot: str = ""

def prompt_would_serve(bundle: ReleaseBundle, evidence_supports_claim: bool) -> bool:
    if type(evidence_supports_claim) is not bool:
        raise ValueError("fixture support label must be a boolean")
    # Historical v1.1 bypassed the gate; known-good v1 and the candidate don't.
    if bundle.prompt_version == "deploy-answerer-v1.1-regression.json":
        return True
    if bundle.prompt_version in {stable.prompt_version, candidate.prompt_version}:
        return evidence_supports_claim
    raise ValueError("no simulated policy for this prompt")

def replay_against(trace: RecordedTrace, bundle: ReleaseBundle) -> dict[str, object]:
    if trace.evidence_version != bundle.corpus_version:
        raise ValueError("replay evidence snapshot changed")
    if trace.model_snapshot != bundle.answer_model:
        raise ValueError("replay model pin changed")
    if trace.evidence_gate != bundle.evidence_gate:
        raise ValueError("replay evidence gate changed")
    would_serve = prompt_would_serve(bundle, trace.evidence_supports_claim)
    reproduces = would_serve and not trace.evidence_supports_claim
    return {
        "candidate_release": release_id(bundle),
        "prompt_version": bundle.prompt_version,
        "tool_sequence": trace.tool_sequence,
        "pins": (trace.temperature, trace.seed, trace.model_snapshot),
        "candidate_reproduces_failure": reproduces,
    }

historical_incident = replace(
    stable,
    prompt_compile_run="regressed-template-v1.1",
    prompt_version="deploy-answerer-v1.1-regression.json",
    serving_image="registry.example/deploy-answerer@sha256:image-regressed",
)
failed_trace = RecordedTrace(
    request_id="req_205",
    origin_release=release_id(historical_incident),
    prompt_version=historical_incident.prompt_version,
    evidence_version="releaseops-records@sha256:corpus-5",
    evidence_gate=stable.evidence_gate,
    tool_sequence=("lookup_release", "fetch_record", "answer"),
    evidence_supports_claim=False,
    served_unsupported_claim=True,
    temperature=0.0,
    seed=7,
    model_snapshot=stable.answer_model,
)
historical_replay = replay_against(failed_trace, historical_incident)
stable_replay = replay_against(failed_trace, stable)
candidate_replay = replay_against(failed_trace, candidate)

print(f"origin_release={failed_trace.origin_release}")
print(f"historical_reproduces_failure={historical_replay['candidate_reproduces_failure']}")
print(f"stable_reproduces_failure={stable_replay['candidate_reproduces_failure']}")
print(f"replayed_against={candidate_replay['candidate_release']}")
print(f"candidate_prompt={candidate_replay['prompt_version']}")
print(f"recorded_step_names={candidate_replay['tool_sequence']}")
print(f"pins={candidate_replay['pins']}")
print(f"candidate_reproduces_failure={candidate_replay['candidate_reproduces_failure']}")

assert historical_replay["candidate_reproduces_failure"] is True
assert stable_replay["candidate_reproduces_failure"] is False
assert candidate_replay["candidate_reproduces_failure"] is False

import re

@dataclass(frozen=True)
class ShadowEnvelope:
    candidate_release: str
    stable_release: str
    sanitized_text: str
    side_effects_enabled: bool
    response_visible_to_user: bool
    evidence_supports_claim: bool
    observed_stable_route: str

@dataclass(frozen=True)
class ShadowComparison:
    route_agreement: bool
    candidate_route_supported: bool
    candidate_would_serve: bool
    stable_route: str
    candidate_route: str
    dropped: bool

def make_shadow(
    text: str, evidence_supports_claim: bool, observed_stable_route: str
) -> ShadowEnvelope:
    if type(text) is not str or type(evidence_supports_claim) is not bool:
        raise ValueError("need text and a boolean fixture support label")
    if observed_stable_route not in {"serve", "abstain"}:
        raise ValueError("unknown stable route")
    sanitized = re.sub(r"SHIP-\d+", "[SHIP_ID]", text)
    return ShadowEnvelope(
        candidate_release=registry.resolve("canary"),
        stable_release=registry.resolve("production"),
        sanitized_text=sanitized,
        side_effects_enabled=False,
        response_visible_to_user=False,
        evidence_supports_claim=evidence_supports_claim,
        observed_stable_route=observed_stable_route,
    )

def compare_shadow(envelope: ShadowEnvelope, queue_full: bool = False) -> ShadowComparison:
    if any(type(value) is not bool for value in (
        queue_full, envelope.side_effects_enabled, envelope.response_visible_to_user,
        envelope.evidence_supports_claim,
    )) or envelope.observed_stable_route not in {"serve", "abstain"}:
        raise ValueError("invalid shadow metadata or route")
    if queue_full or envelope.side_effects_enabled or envelope.response_visible_to_user:
        return ShadowComparison(False, False, False, "dropped", "dropped", True)
    # Stable already answered the user; shadow only scores the candidate copy.
    stable_route = envelope.observed_stable_route
    candidate_bundle = registry.bundle(envelope.candidate_release)
    candidate_would_serve = prompt_would_serve(
        candidate_bundle, envelope.evidence_supports_claim
    )
    candidate_route = "serve" if candidate_would_serve else "abstain"
    route_agreement = candidate_route == stable_route
    candidate_route_supported = candidate_route == "abstain" or envelope.evidence_supports_claim
    return ShadowComparison(
        route_agreement=route_agreement,
        candidate_route_supported=candidate_route_supported,
        candidate_would_serve=candidate_would_serve,
        stable_route=stable_route,
        candidate_route=candidate_route,
        dropped=False,
    )

shadow = make_shadow(
    "Is payment-tests approved for SHIP-4412 without a ReleaseOps record?",
    evidence_supports_claim=False,
    observed_stable_route="abstain",
)
comparison = compare_shadow(shadow)
dropped = compare_shadow(shadow, queue_full=True)

print(f"shadow_text={shadow.sanitized_text}")
print(f"candidate={shadow.candidate_release}")
print(f"side_effects_enabled={shadow.side_effects_enabled}")
print(f"response_visible={shadow.response_visible_to_user}")
print(f"candidate_route={comparison.candidate_route}")
print(f"route_agreement={comparison.route_agreement}")
print(f"candidate_route_supported={comparison.candidate_route_supported}")
print(f"dropped_when_queue_full={dropped.dropped}")

assert comparison.candidate_would_serve is False
assert comparison.route_agreement is True
assert comparison.candidate_route_supported is True
assert dropped.dropped is True

def bucket(conversation_id: str) -> int:
    if type(conversation_id) is not str or not conversation_id.strip():
        raise ValueError("conversation ID must be a nonempty string")
    digest = hashlib.sha256(conversation_id.encode("utf-8")).hexdigest()
    return int(digest[:8], 16) % 100

conversation_assignments: dict[str, str] = {}
aborted_releases: set[str] = set()

def assigned_release(conversation_id: str, canary_percent: int) -> str:
    if type(conversation_id) is not str or not conversation_id.strip():
        raise ValueError("conversation ID is required")
    if type(canary_percent) is not int or not 0 <= canary_percent <= 100:
        raise ValueError("canary percentage must be an integer from 0 to 100")
    existing = conversation_assignments.get(conversation_id)
    if existing is not None and existing not in aborted_releases:
        return existing
    alias = "canary" if bucket(conversation_id) < canary_percent else "production"
    bundle_id = registry.resolve(alias)
    if bundle_id in aborted_releases:
        bundle_id = registry.resolve("production")
    if bundle_id in aborted_releases:
        raise RuntimeError("no healthy release is available")
    conversation_assignments[conversation_id] = bundle_id
    return bundle_id

canary_thread = next(
    f"thread-{index}" for index in range(1000) if bucket(f"thread-{index}") < 10
)
assignments = [assigned_release(canary_thread, canary_percent=10) for _ in range(3)]
stable_thread = next(
    f"thread-{index}" for index in range(1000) if bucket(f"thread-{index}") >= 10
)
stable_before_widen = assigned_release(stable_thread, canary_percent=10)
stable_after_widen = assigned_release(stable_thread, canary_percent=100)

print(f"canary_thread={canary_thread}")
print(f"bucket={bucket(canary_thread)}")
print(f"same_release_each_turn={len(set(assignments)) == 1}")
print(f"assigned_to_candidate={assignments[0] == candidate_id}")
print(f"existing_stable_thread_pinned_after_widen={stable_before_widen == stable_after_widen == stable_id}")

@dataclass(frozen=True)
class LiveWindow:
    name: str
    release_id: str
    rollout_id: str
    started_at: datetime
    ended_at: datetime
    request_count: int
    support_checked_count: int
    unsupported_serve_count: int
    latency_sample_count: int
    p95_latency_ms: int
    transport_checked_count: int
    error_count: int
    shadow_drop_rate: float
    supported_request_count: int
    useful_supported_serve_count: int
    rollback_supported_request_count: int
    rollback_useful_serve_count: int

    @property
    def error_rate(self) -> float:
        return self.error_count / self.request_count

@dataclass(frozen=True)
class LiveDecision:
    status: str
    reason: str

def live_gate(
    window: LiveWindow, expected_release: str, expected_rollout: str, as_of: datetime
) -> LiveDecision:
    if type(window) is not LiveWindow:
        return LiveDecision("HOLD", "invalid window record")
    if (window.release_id, window.rollout_id) != (expected_release, expected_rollout):
        return LiveDecision("HOLD", "window belongs to another release or rollout")
    counts = (window.request_count, window.support_checked_count, window.unsupported_serve_count)
    if any(type(n) is not int or n < 0 for n in counts):
        return LiveDecision("HOLD", "invalid count")
    if not window.unsupported_serve_count <= window.support_checked_count <= window.request_count:
        return LiveDecision("HOLD", "inconsistent counts")
    if window.unsupported_serve_count > 0:
        return LiveDecision("ABORT", "unsupported serve count must remain zero")
    if (
        not fresh(window.ended_at, as_of, timedelta(minutes=5))
        or not isinstance(window.started_at, datetime)
        or window.started_at.utcoffset() is None
        or window.started_at >= window.ended_at
    ):
        return LiveDecision("HOLD", "invalid or stale observation interval")
    if any(type(count) is not int or count < 0 for count in (
        window.latency_sample_count, window.transport_checked_count, window.error_count
    )):
        return LiveDecision("HOLD", "invalid measurement count")
    if not (window.error_count <= window.transport_checked_count <= window.request_count
            and window.latency_sample_count <= window.request_count):
        return LiveDecision("HOLD", "inconsistent measurement counts")
    if window.request_count == 0:
        return LiveDecision("HOLD", "fewer than 1000 requests observed")
    if not (
        finite_between(window.p95_latency_ms, 0, math.inf)
        and finite_between(window.shadow_drop_rate, 0, 1)
    ):
        return LiveDecision("HOLD", "invalid metric")
    if window.p95_latency_ms > 550:
        return LiveDecision("ABORT", "live latency exceeded 550 ms")
    if window.error_rate > 0.01:
        return LiveDecision("ABORT", "error rate exceeded 1%")
    if window.shadow_drop_rate > 0.02:
        return LiveDecision("HOLD", "shadow telemetry incomplete")
    if window.support_checked_count != window.request_count:
        return LiveDecision("HOLD", "support evaluation incomplete")
    if window.latency_sample_count != window.request_count or window.transport_checked_count != window.request_count:
        return LiveDecision("HOLD", "latency or transport coverage incomplete")
    if window.request_count < 1_000:
        return LiveDecision("HOLD", "fewer than 1000 requests observed")
    usefulness_counts = (
        window.supported_request_count, window.useful_supported_serve_count,
        window.rollback_supported_request_count, window.rollback_useful_serve_count,
    )
    if any(type(count) is not int or count < 0 for count in usefulness_counts):
        return LiveDecision("HOLD", "invalid usefulness count")
    if not (
        window.useful_supported_serve_count <= window.supported_request_count <= window.request_count
        and window.rollback_useful_serve_count <= window.rollback_supported_request_count <= window.supported_request_count
        and window.rollback_useful_serve_count <= window.useful_supported_serve_count
        and window.useful_supported_serve_count - window.rollback_useful_serve_count <= window.supported_request_count - window.rollback_supported_request_count
    ):
        return LiveDecision("HOLD", "inconsistent usefulness counts")
    if window.supported_request_count < 100 or window.rollback_supported_request_count < 50:
        return LiveDecision("HOLD", "too few answerable or rollback cases")
    if (10 * window.useful_supported_serve_count < 9 * window.supported_request_count
            or 10 * window.rollback_useful_serve_count < 9 * window.rollback_supported_request_count):
        return LiveDecision("ABORT", "useful-answer floor below 90%")
    return LiveDecision("PASS", "live window passed")

rollout_id = "rollout-052"
window_1_percent = LiveWindow(
    "1%", candidate_id, rollout_id, NOW - timedelta(minutes=15), NOW,
    request_count=1_200, support_checked_count=1_200, unsupported_serve_count=0,
    latency_sample_count=1_200, p95_latency_ms=481,
    transport_checked_count=1_200, error_count=2, shadow_drop_rate=0.001,
    supported_request_count=800, useful_supported_serve_count=752,
    rollback_supported_request_count=100, rollback_useful_serve_count=92,
)
window_10_percent = replace(
    window_1_percent, name="10%", request_count=8_000, support_checked_count=8_000,
    unsupported_serve_count=112, latency_sample_count=8_000, p95_latency_ms=493,
    transport_checked_count=8_000, error_count=24,
    supported_request_count=6_000, useful_supported_serve_count=5_640,
    rollback_supported_request_count=500, rollback_useful_serve_count=465,
)

print(f"one_percent={live_gate(window_1_percent, candidate_id, rollout_id, NOW)}")
print(f"ten_percent={live_gate(window_10_percent, candidate_id, rollout_id, NOW)}")
print(f"ten_percent_unsupported_count={window_10_percent.unsupported_serve_count}")

bad_latencies = (None, True, "472", float("nan"), float("inf"), 10 ** 1000, -1)
offline_rejected = 0
live_held = 0
for latency in bad_latencies:
    offline = offline_gate(candidate, replace(candidate_offline, p95_latency_ms=latency), NOW)
    live = live_gate(replace(window_1_percent, p95_latency_ms=latency), candidate_id, rollout_id, NOW)
    assert offline == Decision(False, "invalid metric")
    assert live == LiveDecision("HOLD", "invalid metric")
    offline_rejected += 1
    live_held += 1
print(f"malformed_metrics={len(bad_latencies)} offline_rejected={offline_rejected} live_held={live_held}")

canary_percent = 10
failed_window = live_gate(window_10_percent, candidate_id, rollout_id, NOW)
if failed_window.status == "ABORT":
    aborted_releases.add(candidate_id)
    canary_percent = 0
    # Zero traffic isn't enough: any consumer of registry.resolve("canary") must leave the failed ID.
    registry.move_alias("canary", stable_id, expected=candidate_id)

print(f"canary_percent_after_abort={canary_percent}")
print(f"canary_alias_after_abort={registry.resolve('canary')}")
print(f"production_after_abort={registry.resolve('production')}")
print(f"pinned_canary_thread_restored_stable={assigned_release(canary_thread, canary_percent) == stable_id}")

assert registry.resolve("canary") == stable_id
assert registry.resolve("production") == stable_id

# Independent hypothetical: production had moved before the bad window arrived.
# This isn't authorization to retry the candidate rejected above.
rollback_drill = deepcopy(registry)
previous_production = rollback_drill.resolve("production")
rollback_drill.move_alias("production", candidate_id, expected=previous_production)
post_promotion_incident = replace(window_10_percent, name="100%")

if live_gate(post_promotion_incident, candidate_id, rollout_id, NOW).status == "ABORT":
    rollback_drill.move_alias("production", previous_production, expected=candidate_id)

print(f"drill_production_after_rollback={rollback_drill.resolve('production')}")
print(f"drill_restored_stable={rollback_drill.resolve('production') == stable_id}")
print(f"actual_production_unchanged={registry.resolve('production') == stable_id}")

@dataclass(frozen=True)
class ReleaseEvent:
    stage: str
    release_id: str
    decision: str
    evidence: str

events = (
    ReleaseEvent("register", candidate_id, "RECORDED", "manifest_sha"),
    ReleaseEvent("offline_gate", candidate_id, "PASSED", candidate_offline.evaluation_report),
    ReleaseEvent("canary_1_percent", candidate_id, "PASSED", "live:window-001"),
    ReleaseEvent("canary_10_percent", candidate_id, "ABORTED", "live:window-010"),
    ReleaseEvent("production", stable_id, "UNCHANGED", "rollback:not-needed"),
)

for event in events:
    print(f"{event.stage}:{event.decision}:{event.evidence}")
print("release_decision=REJECT_CANDIDATE_AFTER_CANARY_REGRESSION")
print(f"active_production={registry.resolve('production')}")