Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
A shipping service wants to replace delivery-risk-v0. One candidate makes every test row cheaper, yet misses one of two late deliveries in the critical slice. A single passed flag could promote the cheaper model while hiding the exact cases the service must catch.
A successful HTTP response tells you that a request completed, not that its prediction was useful. Ordinary software can also produce plausible wrong answers without throwing an exception; an ML service additionally needs checks on its data, learned transformations, and labeled outcomes. The first candidate below fails one of those checks even while reducing the assumed cost.
Follow six handoffs in this local exercise:
- Validate record schemas, identities, clocks, and finite values.
- Select features observed and available by prediction time.
- Fit preprocessing and a small model on admitted training rows; save and reload both together.
- Derive slice recall and assumed cost from evaluation rows.
- Bind the results to retained artifact, release, and policy fingerprints.
- Use injected monitoring windows to exercise local canary, promotion, and rollback decisions.
These are the lesson's boundaries, not a mandatory six-stage architecture for every ML system. The dictionaries and model files remain editable. Fingerprints support integrity checks against trusted expected values; they don't make storage immutable, authenticate the evaluator, or implement distributed transactions.
The ETA, ranking, forecast, and vision capstones already produce pieces of this release. Reuse their train/validation split, preprocessing, metric, and persistence ideas here. The running example is delivery warnings; the other products help you identify what varies. All data and telemetry are synthetic. The lab performs real local fitting, serialization, reload, and controller transitions, but it doesn't deploy a service or measure a live canary.
Define the shared release tuple
Start with a release that can be restored as one unit. Models differ, but their manifest can share a schema. Read the fields before looking at implementation details:
| Field | ETA example | Ranking example | Forecast example | Vision example |
|---|---|---|---|---|
| data snapshot | carrier events through cutoff | catalog and judged queries | daily counts through cutoff | return photos grouped by shipment |
| feature or preprocessing version | eta-features-v1 | ranking-features-v1 | calendar-regression-v1 | parcel-rgb-224-policy-v1 |
| model artifact | delay-model-v1 | market-ranker-v1 | warehouse-demand-v1 | damage-cnn-v1 |
| action policy | warning threshold | eligibility and slate rule | alert threshold | quality check and review threshold |
| promotion policy | slice recall and cost limits | eligibility and NDCG limits | capacity-decision cost and forecast error limits | quality rejection and case-level review limits |
| monitor | delayed labels and freshness | impressions and returns | residuals and alert review | photo quality and reviewer labels |
| previous release | delivery-risk-v0 | market-ranker-v0 | warehouse-demand-v0 | damage-cnn-v0 |
If a threshold, crop, or gate policy is missing from the release, an incident reviewer can't reconstruct what served a request. Sculley et al. identify data dependencies, configuration, and feedback loops among sources of ML technical debt.[1] Restore the components relevant to the product together: an image cropper for vision, an imputer for tabular features, or a tokenizer for a text model. The table proposes a shared release schema; it does not claim the preceding synthetic capstones trained all four listed artifacts.
The tuple still doesn't grant traffic authority. An alias is a mutable pointer such as production or canary; only authorized controller code should move it. Limited live exposure (a canary) follows offline gates. Rollback selects the previous tuple, not a fresh training run. In this lab, one versioned promotion policy also owns the canary and rollback thresholds; the model package contains the prediction threshold and a digest of the builder source.
Pause before the transition map: which boundary should reject a candidate with failed slice recall? Registration alone must never move production traffic.
Registration gives v1 an identity while the local production pointer stays on v0. An offline failure stops at hold_offline; a pass opens the local canary pointer. Supplied monitoring windows then exercise promotion and rollback. None of these pointers routes a real request.
Compute the failing slice by hand
Before any registry code, make the first decision by hand. delivery-risk-recall-fail is a deliberately bad candidate and the local production pointer selects delivery-risk-v0. These hand-entered rows isolate the gate's behavior; they're not outputs of the trained model we'll build next. Their cost columns are supplied units, independent of the warning-cost formula used later. The candidate is cheaper overall, but the critical slice contains the late deliveries that the warning policy exists to catch. Would you open a canary?
| entity | critical slice | late delivery | predicted late | baseline cost | candidate cost |
|---|---|---|---|---|---|
eval-1 | yes | 1 | 1 | 25 | 24 |
eval-2 | yes | 1 | 0 | 25 | 24 |
eval-3 | no | 0 | 0 | 25 | 24 |
eval-4 | no | 0 | 0 | 25 | 24 |
Only the first two rows are late positives in the critical slice. Count those rows before looking at total cost. Critical-slice recall is true positives on late critical rows divided by all late critical rows:
The policy floor is , so recall fails at . Cost did improve: , which clears max_cost_delta = 0.
The cheaper cost still doesn't buy a canary. One required gate is enough to hold the candidate offline, store the failed evidence, and leave production on delivery-risk-v0. That is the decision the controller must preserve when someone later tries to pass a friendly summary instead of these rows.
Build the package before trusting its manifest
A manifest that says no_leakage=True isn't evidence. Start with prediction requests, carrier scans, and delayed outcomes. The runnable pipeline source implements the full local path using Python's standard library; save it as assets/production_pipeline.py alongside the notebook or script running these cells.
For eval-1, the request time is May 28 at 12:00 UTC. Three scans compete:
| Scan observed | Scan available to the service | Delay | Eligible at noon? |
|---|---|---|---|
| 10:00 | 11:00 | 9 minutes | Yes |
| 11:30 | 12:30 | 99 minutes | No: arrived after prediction |
| 13:00 | 13:01 | 100 minutes | No: future observation |
The feature builder filters on both timestamps, then chooses the most recent eligible observation for that entity. Joining the latest scan today could insert information unavailable at yesterday's prediction. A backward as-of join selects an earlier or equal timestamp; it still needs sorted keys, entity grouping, availability filtering, duplicate handling, and any freshness policy.[2] This lab normalizes aware timestamps to UTC before rejecting same-instant scan revisions. A missing scan becomes a missing feature. It implements no maximum scan-age threshold; add and test one if the product requires it.
Training requests precede May 25, and their outcome labels must also have arrived by May 25. Evaluation requests occur on May 28; their labels are available by the May 31 evaluation snapshot. Label time determines whether a row may be used for fitting or scoring. It never makes the label an input feature.
The six training delays are 0, 1, 2, 8, 9, 10. Fit the imputation mean and scale on those rows only: center , range , transformed input . Here the divisor is the range, not a standard deviation. Computing these parameters from evaluation rows makes that split influence the estimator. Recomputing them on serving batches can also change a request's prediction when its batch membership changes. Save the training-fitted state and reuse it for evaluation and serving.[3]
A one-feature least-squares model fits a numeric score to the binary late label. The score isn't a calibrated probability; the fixed warning threshold is . No parameter or threshold is selected using the four evaluation labels. Scikit-learn's guidance applies equally to this hand-written estimator: fit transformations on training data, then reuse them unchanged for evaluation and serving.[3]
Trace the fit on these six rows. Their transformed features are and labels are , so and . The least-squares slope is ; the intercept is . The saved score therefore equals . At delays 9, 8, 1, and 0 it gives scores 0.98, 0.86, 0.02, and -0.10, which become warnings [1, 1, 0, 0]. That negative score is another reason to avoid calling this estimator a probability model.
A missing input is imputed to the training mean 5, giving score 0.5. The serving rule uses >= 0.5, so it warns at equality. This is the fixture's explicit fallback, not evidence that a missing scan predicts lateness. A real warning policy must evaluate missing-input cases and decide whether to warn, abstain, or use a separate fallback.
The JSON package stores feature names, fitted imputation and scaling, coefficients, threshold, training IDs, cutoff, a digest of the admitted training rows, builder-source digest, and Python version. The separate full-fixture digest covers the input records used for this retrospective lab, including later evaluation outcomes. Neither proves that the run occurred on those historical dates. With general estimators, persist fitted preprocessing with the estimator and record compatible dependencies; scikit-learn does not support cross-version loading and warns against untrusted pickle-based artifacts.[4] The numeric JSON loader checks both its expected digest and its own schema, finite parameters, positive scale, fixed threshold, and identities. A matching checksum alone cannot establish a valid model.
Run the pipeline and inspect its observable boundaries. The bad scans must not change eval-1 from 9 minutes to 99 or 100. Both serialized models must reproduce their pre-save predictions after reload.
from assets.production_pipeline import (
canonical_bytes, finite_number, fixture, load_bundle, nonblank,
point_in_time_rows, predict, run_pipeline, timestamp,
)
built = run_pipeline("model-packages")
print("point-in-time eval-1:", built["evaluation"][0]["delay_minutes"])
print("training-only center:", built["packages"]["v1"]["bundle"]["center"])
print("reloaded predictions:", [row["predicted_late"] for row in built["evaluation"]])
print("baseline / candidate cost:",
sum(row["baseline_cost"] for row in built["evaluation"]),
sum(row["candidate_cost"] for row in built["evaluation"]))
assert built["evaluation"][0]["delay_minutes"] == 9
assert built["packages"]["v1"]["bundle"]["center"] == 5point-in-time eval-1: 9
training-only center: 5
reloaded predictions: [1, 1, 0, 0]
baseline / candidate cost: 50.0 4.0Under the explicit cost policy, warning costs 2 units and missing a late delivery costs 25. An always-silent baseline costs 50 across two missed late deliveries; the candidate warns on both, costing 4. These are assumed costs on an easy synthetic fixture, not estimated savings. The fixture defines both critical_slice and late_delivery from the same delay threshold. A real critical cohort needs a product rule fixed before examining outcomes. Four separable rows demonstrate the plumbing, not generalization or production readiness.
Before registering anything, deliberately break the joins and serving contract. Duplicating an outcome must fail rather than multiply evaluation rows. Missing outcomes must not become negative labels. Corrupting saved bytes must fail before inference. Predict each rejection, then run the checks.
from hashlib import sha256
import json
from pathlib import Path
def expect_rejection(name, operation):
try:
operation()
except ValueError:
print(name + ": rejected")
else:
raise AssertionError(name + " unexpectedly passed")
requests, scans, outcomes = fixture()
expect_rejection("duplicate outcome", lambda: point_in_time_rows(
requests, scans, outcomes + [outcomes[0]], "2026-05-31T00:00:00Z"))
expect_rejection("immature labels", lambda: point_in_time_rows(
requests, scans, outcomes, "2026-05-29T00:00:00Z"))
expect_rejection("serving schema", lambda: predict(
built["packages"]["v1"]["bundle"], {"delay_hours": 9}))
package = built["packages"]["v1"]
path = Path(package["path"])
saved = path.read_bytes()
path.write_bytes(saved + b" ")
expect_rejection("changed model bytes", lambda: load_bundle(path, package["sha256"]))
path.write_bytes(saved)
missing_prediction = predict(package["bundle"], {"delay_minutes": None})
assert missing_prediction == predict(package["bundle"], {"delay_minutes": 5})
print("missing input uses training mean:", missing_prediction)
requests, scans, outcomes = fixture()
same_instant = {**scans[6], "observed_at": "2026-05-28T10:00:00+00:00", "delay_minutes": 88}
expect_rejection("equivalent scan clocks", lambda: point_in_time_rows(
requests, [same_instant] + scans, outcomes, "2026-05-31T00:00:00Z"))
expect_rejection("huge serving input", lambda: predict(package["bundle"], {"delay_minutes": 10**400}))
malformed_path = Path("model-packages/invalid.json")
for name, patch in (("checksum-valid zero scale", {"scale": 0}),
("checksum-valid NaN coefficient", {"slope": float("nan")})):
payload = json.dumps({**package["bundle"], **patch}).encode()
malformed_path.write_bytes(payload)
expect_rejection(name, lambda: load_bundle(malformed_path, sha256(payload).hexdigest()))
malformed_path.unlink()
assert package["bundle"]["source_sha256"] == built["training_sha256"]
assert built["training_sha256"] != built["source_sha256"]duplicate outcome: rejected
immature labels: rejected
serving schema: rejected
changed model bytes: rejected
missing input uses training mean: 1
equivalent scan clocks: rejected
huge serving input: rejected
checksum-valid zero scale: rejected
checksum-valid NaN coefficient: rejectedRegister a candidate without moving production
Register the release tuple before deriving evidence. This example hashes a canonical JSON serialization of its dataclass values. The policy, artifact, and release fingerprints let later calls check whether those registered values still match the values used by the evaluator.
If the policy changes under the same ID, promotion detects the mismatch when it rechecks the stored bindings. If model bytes change, reload checks them against the registered expected digest. No background process invalidates receipts immediately. A writer who can alter both the evidence and trusted expectations can defeat these consistency checks. Production needs controlled writers, retained artifacts and receipts, and atomic alias updates; the dictionaries here implement none of that infrastructure. The helper named immutable_fingerprint computes a digest, not an immutability guarantee.
The first cell turns the four-row fixture into registered artifacts and aliases. Run it, then check the output: four releases should exist while production remains on delivery-risk-v0. Registration prepares evidence; it doesn't authorize traffic.
from dataclasses import asdict, dataclass, fields, replace
from hashlib import sha256
import json
from math import isfinite
@dataclass(frozen=True)
class PromotionPolicy:
policy_id: str
required_offline_gates: tuple[str, ...]
required_schema: tuple[str, ...]
min_critical_slice_recall: float
max_cost_delta: float
min_canary_windows: int
max_error_rate: float
max_p95_latency_ms: int
max_late_warning_cost_delta: float
def __post_init__(self):
nonblank(self.policy_id, "policy_id")
expected_gates = {"schema_valid", "no_leakage", "critical_slice_pass", "cost_improves"}
if type(self.required_offline_gates) is not tuple:
raise ValueError("required_offline_gates must be a tuple")
for name in self.required_offline_gates:
nonblank(name, "gate name")
if len(self.required_offline_gates) != 4 or set(self.required_offline_gates) != expected_gates:
raise ValueError("this controller requires all four offline gates")
if type(self.required_schema) is not tuple or not self.required_schema:
raise ValueError("required_schema must be a nonempty tuple")
for name in self.required_schema:
nonblank(name, "schema field")
if len(set(self.required_schema)) != len(self.required_schema):
raise ValueError("duplicate required schema field")
for name in ("min_critical_slice_recall", "max_cost_delta", "max_error_rate",
"max_p95_latency_ms", "max_late_warning_cost_delta"):
finite_number(getattr(self, name), name)
if not 0 <= self.min_critical_slice_recall <= 1 or not 0 <= self.max_error_rate <= 1:
raise ValueError("recall and error-rate limits must be in [0, 1]")
if self.max_p95_latency_ms < 0 or self.max_cost_delta < -1 or self.max_late_warning_cost_delta < -1:
raise ValueError("invalid latency or relative-cost limit")
if type(self.min_canary_windows) is not int or self.min_canary_windows < 1:
raise ValueError("min_canary_windows must be a positive integer")
@dataclass(frozen=True)
class OfflineEvalRow:
entity_id: str
critical_slice: bool
late_delivery: int
predicted_late: int
baseline_cost: float
candidate_cost: float
event_time: str = "2026-05-28T12:00:00Z"
def __post_init__(self):
nonblank(self.entity_id, "entity_id")
timestamp(self.event_time)
if type(self.critical_slice) is not bool:
raise ValueError("critical_slice must be Boolean")
for name in ("late_delivery", "predicted_late"):
value = getattr(self, name)
if type(value) is not int or value not in (0, 1):
raise ValueError(f"{name} must be an integer zero or one")
for name in ("baseline_cost", "candidate_cost"):
value = getattr(self, name)
finite_number(value, name)
if value < 0:
raise ValueError(f"{name} must be nonnegative")
@dataclass(frozen=True)
class CandidateArtifacts:
artifact_id: str
data_snapshot: str
feature_version: str
model_artifact: str
action_policy_version: str
observed_schema: tuple[str, ...]
training_entity_ids: tuple[str, ...]
evaluation_entity_ids: tuple[str, ...]
evaluation_rows: tuple[OfflineEvalRow, ...]
def __post_init__(self):
for name in ("artifact_id", "data_snapshot", "feature_version", "model_artifact", "action_policy_version"):
nonblank(getattr(self, name), name)
for name in ("observed_schema", "training_entity_ids", "evaluation_entity_ids"):
values = getattr(self, name)
if type(values) is not tuple:
raise ValueError(f"{name} must be a tuple")
for value in values:
nonblank(value, name)
if len(set(values)) != len(values):
raise ValueError(f"duplicate value in {name}")
if type(self.evaluation_rows) is not tuple or any(type(row) is not OfflineEvalRow for row in self.evaluation_rows):
raise ValueError("evaluation_rows must be a tuple of validated OfflineEvalRow records")
@dataclass(frozen=True)
class Release:
release_id: str
data_snapshot: str
feature_version: str
model_artifact: str
action_policy_version: str
promotion_policy_version: str
previous_release: str | None
candidate_artifact_id: str
candidate_artifact_fingerprint: str
@dataclass(frozen=True)
class OfflineEvidence:
evidence_id: str
candidate: str
artifact_id: str
artifact_fingerprint: str
release_fingerprint: str
promotion_policy_version: str
promotion_policy_fingerprint: str
missing_schema_fields: tuple[str, ...]
leaked_entity_ids: tuple[str, ...]
critical_slice_positives: int
critical_slice_true_positives: int
critical_slice_recall: float | None
cost_delta: float | None
def gates(self, policy: PromotionPolicy) -> dict[str, bool]:
return {
"schema_valid": not self.missing_schema_fields,
"no_leakage": not self.leaked_entity_ids,
"critical_slice_pass": (
self.critical_slice_positives > 0
and self.critical_slice_recall >= policy.min_critical_slice_recall
),
"cost_improves": self.cost_delta is not None and self.cost_delta <= policy.max_cost_delta,
}
@dataclass(frozen=True)
class OfflineReceipt:
receipt_id: str
candidate: str
production_before: str
promotion_policy_version: str | None
promotion_policy_fingerprint: str | None
offline_evidence_id: str | None
release_fingerprint: str | None
artifact_fingerprint: str | None
failed_gates: tuple[str, ...]
decision: str
def immutable_fingerprint(value: object) -> str:
payload = json.dumps(asdict(value), sort_keys=True, separators=(",", ":"), allow_nan=False)
return sha256(payload.encode("utf-8")).hexdigest()
POLICIES = {
"eta-promotion-v1": PromotionPolicy(
policy_id="eta-promotion-v1",
required_offline_gates=(
"schema_valid",
"no_leakage",
"critical_slice_pass",
"cost_improves",
),
required_schema=(
"entity_id",
"event_time",
"critical_slice",
"late_delivery",
"predicted_late",
"baseline_cost",
"candidate_cost",
),
min_critical_slice_recall=0.90,
max_cost_delta=0.0,
min_canary_windows=2,
max_error_rate=0.01,
max_p95_latency_ms=250,
max_late_warning_cost_delta=0.0,
)
}
stable_artifact = CandidateArtifacts(
artifact_id="eta-artifact-v0",
data_snapshot=built["source_sha256"],
feature_version="eta-features-v1",
model_artifact=built["packages"]["v0"]["sha256"],
action_policy_version="eta-threshold-v1",
observed_schema=POLICIES["eta-promotion-v1"].required_schema,
training_entity_ids=tuple(row["entity_id"] for row in built["training"]),
evaluation_entity_ids=(),
evaluation_rows=(),
)
recall_fail_artifact = CandidateArtifacts(
artifact_id="eta-artifact-recall-fail",
data_snapshot="carrier-events-through-2026-05-31",
feature_version="eta-features-v1",
model_artifact="delay-model-recall-fail",
action_policy_version="eta-threshold-v1",
observed_schema=POLICIES["eta-promotion-v1"].required_schema,
training_entity_ids=("train-1", "train-2"),
evaluation_entity_ids=("eval-1", "eval-2", "eval-3", "eval-4"),
evaluation_rows=(
OfflineEvalRow("eval-1", True, 1, 1, 25.0, 24.0),
OfflineEvalRow("eval-2", True, 1, 0, 25.0, 24.0),
OfflineEvalRow("eval-3", False, 0, 0, 25.0, 24.0),
OfflineEvalRow("eval-4", False, 0, 0, 25.0, 24.0),
),
)
leaky_artifact = CandidateArtifacts(
artifact_id="eta-artifact-leaky",
data_snapshot="carrier-events-through-2026-05-31",
feature_version="eta-features-v1",
model_artifact="delay-model-leaky",
action_policy_version="eta-threshold-v1",
observed_schema=POLICIES["eta-promotion-v1"].required_schema,
training_entity_ids=("train-1", "eval-1"),
evaluation_entity_ids=("eval-1", "eval-2", "eval-3", "eval-4"),
evaluation_rows=(
OfflineEvalRow("eval-1", True, 1, 1, 25.0, 24.0),
OfflineEvalRow("eval-2", True, 1, 1, 25.0, 24.0),
OfflineEvalRow("eval-3", False, 0, 0, 25.0, 24.0),
OfflineEvalRow("eval-4", False, 0, 0, 25.0, 24.0),
),
)
passing_artifact = CandidateArtifacts(
artifact_id="eta-artifact-v1",
data_snapshot=built["source_sha256"],
feature_version="eta-features-v1",
model_artifact=built["packages"]["v1"]["sha256"],
action_policy_version="eta-threshold-v1",
observed_schema=POLICIES["eta-promotion-v1"].required_schema,
training_entity_ids=tuple(row["entity_id"] for row in built["training"]),
evaluation_entity_ids=("eval-1", "eval-2", "eval-3", "eval-4"),
evaluation_rows=tuple(
OfflineEvalRow(row["entity_id"], row["critical_slice"], row["late_delivery"],
row["predicted_late"], row["baseline_cost"], row["candidate_cost"],
row["predict_at"])
for row in built["evaluation"]
),
)
artifacts = {
artifact.artifact_id: artifact
for artifact in (
stable_artifact,
recall_fail_artifact,
leaky_artifact,
passing_artifact,
)
}
def release_from_artifact(
release_id: str,
artifact: CandidateArtifacts,
previous_release: str | None,
) -> Release:
return Release(
release_id=release_id,
data_snapshot=artifact.data_snapshot,
feature_version=artifact.feature_version,
model_artifact=artifact.model_artifact,
action_policy_version=artifact.action_policy_version,
promotion_policy_version="eta-promotion-v1",
previous_release=previous_release,
candidate_artifact_id=artifact.artifact_id,
candidate_artifact_fingerprint=immutable_fingerprint(artifact),
)
registry = {
"delivery-risk-v0": release_from_artifact(
"delivery-risk-v0", stable_artifact, None
),
"delivery-risk-recall-fail": release_from_artifact(
"delivery-risk-recall-fail", recall_fail_artifact, "delivery-risk-v0"
),
"delivery-risk-leaky": release_from_artifact(
"delivery-risk-leaky", leaky_artifact, "delivery-risk-v0"
),
"delivery-risk-v1": release_from_artifact(
"delivery-risk-v1", passing_artifact, "delivery-risk-v0"
),
}
aliases = {"production": "delivery-risk-v0"}
offline_evidence: dict[str, OfflineEvidence] = {}
offline_receipts: dict[str, OfflineReceipt] = {}
active_offline_receipts: dict[str, str] = {}
def reload_release(release_id: str):
release = registry[release_id]
package = next((item for item in built["packages"].values()
if item["sha256"] == release.model_artifact), None)
if package is None:
raise ValueError("release has no loadable model package")
return load_bundle(package["path"], release.model_artifact)
print("registry:", list(registry))
print("production:", aliases["production"])
print(
"policy fingerprint bound:",
len(immutable_fingerprint(POLICIES["eta-promotion-v1"])) == 64,
)registry: ['delivery-risk-v0', 'delivery-risk-recall-fail', 'delivery-risk-leaky', 'delivery-risk-v1']
production: delivery-risk-v0
policy fingerprint bound: TrueThe registry now contains candidates, but it contains no verdict about their rows. Predict what should happen when evaluation sees a recall failure, a leakage failure, and a clean candidate.
evaluate_candidate checks the registered row schema and split membership, derives recall and cost, and binds the result to artifact, release, and policy fingerprints. The trained candidate's rows came from the reload path; the two injected failures remain controller fixtures. In a real system, the evaluator must own inference and dataset access, not accept caller-written predictions as independent evidence.
def release_matches_artifact(
release: Release,
artifact: CandidateArtifacts,
) -> bool:
return (
release.data_snapshot,
release.feature_version,
release.model_artifact,
release.action_policy_version,
) == (
artifact.data_snapshot,
artifact.feature_version,
artifact.model_artifact,
artifact.action_policy_version,
)
def evidence_binding_failures(
candidate: Release,
evidence: OfflineEvidence,
) -> list[str]:
failed: list[str] = []
policy = POLICIES.get(candidate.promotion_policy_version)
artifact = artifacts.get(candidate.candidate_artifact_id)
if evidence.candidate != candidate.release_id:
failed.append("offline_evidence_candidate_mismatch")
if evidence.promotion_policy_version != candidate.promotion_policy_version:
failed.append("offline_evidence_policy_version_mismatch")
if (
policy is None
or evidence.promotion_policy_fingerprint != immutable_fingerprint(policy)
):
failed.append("offline_evidence_policy_fingerprint_mismatch")
if evidence.release_fingerprint != immutable_fingerprint(candidate):
failed.append("offline_evidence_release_fingerprint_mismatch")
if evidence.artifact_id != candidate.candidate_artifact_id:
failed.append("offline_evidence_artifact_id_mismatch")
if evidence.artifact_fingerprint != candidate.candidate_artifact_fingerprint:
failed.append("offline_evidence_artifact_fingerprint_mismatch")
if (
artifact is None
or immutable_fingerprint(artifact)
!= candidate.candidate_artifact_fingerprint
):
failed.append("registered_artifact_fingerprint_mismatch")
elif not release_matches_artifact(candidate, artifact):
failed.append("release_artifact_tuple_mismatch")
return failed
def evaluate_candidate(candidate_id: str) -> OfflineEvidence:
candidate = registry[candidate_id]
policy = POLICIES[candidate.promotion_policy_version]
artifact = artifacts[candidate.candidate_artifact_id]
if immutable_fingerprint(artifact) != candidate.candidate_artifact_fingerprint:
raise ValueError("registered artifact changed after release registration")
if not release_matches_artifact(candidate, artifact):
raise ValueError("release tuple does not match candidate artifact")
actual_schema = {field.name for field in fields(OfflineEvalRow)}
missing_schema = tuple(sorted(set(policy.required_schema) - actual_schema))
row_ids = [row.entity_id for row in artifact.evaluation_rows]
if len(row_ids) != len(set(row_ids)) or set(row_ids) != set(artifact.evaluation_entity_ids):
raise ValueError("evaluation row IDs do not match the declared split")
for row in artifact.evaluation_rows:
if row.late_delivery not in (0, 1) or row.predicted_late not in (0, 1):
raise ValueError("evaluation labels and predictions must be binary")
if any(not isfinite(value) or value < 0 for value in (row.baseline_cost, row.candidate_cost)):
raise ValueError("evaluation costs must be finite and nonnegative")
leaked_entities = tuple(
sorted(
set(artifact.training_entity_ids)
& set(artifact.evaluation_entity_ids)
)
)
critical_positives = [
row for row in artifact.evaluation_rows
if row.critical_slice and row.late_delivery == 1
]
critical_true_positives = sum(
row.predicted_late == 1 for row in critical_positives
)
critical_recall = (
critical_true_positives / len(critical_positives)
if critical_positives
else None
)
baseline_cost = sum(row.baseline_cost for row in artifact.evaluation_rows)
candidate_cost = sum(row.candidate_cost for row in artifact.evaluation_rows)
finite_number(baseline_cost, "total baseline cost")
finite_number(candidate_cost, "total candidate cost")
cost_delta = (
candidate_cost / baseline_cost - 1.0
if baseline_cost > 0
else None
)
if cost_delta is not None:
finite_number(cost_delta, "relative cost delta")
evidence = OfflineEvidence(
evidence_id=f"offline-evidence-{len(offline_evidence) + 1}",
candidate=candidate_id,
artifact_id=artifact.artifact_id,
artifact_fingerprint=immutable_fingerprint(artifact),
release_fingerprint=immutable_fingerprint(candidate),
promotion_policy_version=policy.policy_id,
promotion_policy_fingerprint=immutable_fingerprint(policy),
missing_schema_fields=missing_schema,
leaked_entity_ids=leaked_entities,
critical_slice_positives=len(critical_positives),
critical_slice_true_positives=critical_true_positives,
critical_slice_recall=critical_recall,
cost_delta=cost_delta,
)
offline_evidence[evidence.evidence_id] = evidence
return evidence
failing_evidence = evaluate_candidate("delivery-risk-recall-fail")
leaky_evidence = evaluate_candidate("delivery-risk-leaky")
passing_evidence = evaluate_candidate("delivery-risk-v1")
policy = POLICIES["eta-promotion-v1"]
print("failing gates:", failing_evidence.gates(policy))
print("leaky gates:", leaky_evidence.gates(policy))
print("passing gates:", passing_evidence.gates(policy))
print("stored evidence ids:", list(offline_evidence))failing gates: {'schema_valid': True, 'no_leakage': True, 'critical_slice_pass': False, 'cost_improves': True}
leaky gates: {'schema_valid': True, 'no_leakage': False, 'critical_slice_pass': True, 'cost_improves': True}
passing gates: {'schema_valid': True, 'no_leakage': True, 'critical_slice_pass': True, 'cost_improves': True}
stored evidence ids: ['offline-evidence-1', 'offline-evidence-2', 'offline-evidence-3']def open_canary(candidate_id: str, offline_evidence_id: str) -> OfflineReceipt:
candidate = registry.get(candidate_id)
production_before = aliases["production"]
failed = []
policy = POLICIES.get(candidate.promotion_policy_version) if candidate else None
evidence = offline_evidence.get(offline_evidence_id)
if candidate is None:
failed.append("candidate_not_registered")
elif policy is None:
failed.append("promotion_policy_not_registered")
elif evidence is None:
failed.append("offline_evidence_not_registered")
else:
failed.extend(evidence_binding_failures(candidate, evidence))
gates = evidence.gates(policy)
failed.extend(
gate for gate in policy.required_offline_gates if not gates.get(gate, False)
)
if candidate is not None and candidate.previous_release != production_before:
failed.append("previous_release_mismatch")
if aliases.get("canary") not in (None, candidate_id):
failed.append("another_canary_is_active")
receipt = OfflineReceipt(
receipt_id=f"offline-receipt-{len(offline_receipts) + 1}",
candidate=candidate_id,
production_before=production_before,
promotion_policy_version=(
candidate.promotion_policy_version if candidate is not None else None
),
promotion_policy_fingerprint=(
evidence.promotion_policy_fingerprint if evidence is not None else None
),
offline_evidence_id=offline_evidence_id if evidence is not None else None,
release_fingerprint=(
evidence.release_fingerprint if evidence is not None else None
),
artifact_fingerprint=(
evidence.artifact_fingerprint if evidence is not None else None
),
failed_gates=tuple(sorted(failed)),
decision="hold_offline" if failed else "open_canary",
)
offline_receipts[receipt.receipt_id] = receipt
if receipt.decision == "open_canary":
aliases["canary"] = candidate_id
active_offline_receipts[candidate_id] = receipt.receipt_id
return receipt
bad_offline_receipt = open_canary(
"delivery-risk-recall-fail", failing_evidence.evidence_id
)
accepted_offline_receipt = open_canary("delivery-risk-v1", passing_evidence.evidence_id)
print("failed:", bad_offline_receipt.decision, bad_offline_receipt.failed_gates)
print("accepted:", accepted_offline_receipt.decision, accepted_offline_receipt.offline_evidence_id)
print("aliases:", json.dumps(aliases, sort_keys=True))failed: hold_offline ('critical_slice_pass',)
accepted: open_canary offline-evidence-3
aliases: {"canary": "delivery-risk-v1", "production": "delivery-risk-v0"}A failed offline slice leaves production unchanged. The gate named no_leakage checks only train/evaluation entity overlap. It doesn't detect future features or related physical cases assigned different IDs. The builder checks its supplied feature and label clocks and fits preprocessing on training rows; it has no real parcel identity audit. The three evidence records distinguish a hand-entered recall failure, an overlap failure, and the candidate built from reloaded predictions. No positive critical cases means recall is None; zero baseline cost makes the relative cost delta None. Their gates hold rather than invent evidence from a zero denominator.
Now the controller can decide whether limited traffic may open. open_canary receives an evidence ID, follows it back to the registered candidate, and records candidate, production base, policy, release, and artifact fingerprints. Passing evaluation gates opens only the canary alias. Nothing in this cell can overwrite production.
The printed aliases now confirm that the bad candidate stayed offline and the trained candidate opened only canary. An ID selects stored evidence; it isn't a security credential. Only the trusted controller may create those records or mutate aliases.
Wait for canary evidence
Offline evidence answers one question: does the registered candidate clear static gates? It can't answer whether late deliveries cost more after real requests arrive. Fast checks catch broken schemas, errors, and latency spikes; delayed labels answer the quality question later. A canary needs both streams.
A canary limits initial exposure while checking a new release. An A/B test uses an assignment and analysis design to estimate an outcome difference. A canary can use randomized assignment and health aborts within an A/B experiment; the distinction is the question the evidence can answer, not mutually exclusive infrastructure. Google Cloud places these online checks after offline validation.[5]
Health and labeled outcomes can arrive on different schedules. A carrier's late-delivery label needs a defined event and availability rule; it need not wait for eventual delivery if a missed deadline already establishes lateness. The injected windows use three states under this authored policy:
hold_canary: service checks pass, but there are too few windows or the latest delayed-quality metric is missing. Keep the local canary pointer and gather evidence.abort_canary: any supplied window breaches a service limit, a mature window breaches the cost limit, or bindings or telemetry are invalid. Remove this candidate's local canary pointer.ready_for_promotion: required window, binding, and metric checks pass. This permits a later local promotion call, not a real deployment.
The example limits are 250 ms, 1% errors, at least two ordered windows, and a nonpositive relative cost delta. They are policy assumptions, not measured service-level agreements or statistically justified acceptance thresholds.
Here, the controller tests that decision logic using supplied synthetic summaries. It doesn't route requests, wait seven days, collect latency, or join production labels. A real monitor must join request ID, served release, prediction time, and outcome availability, then calculate rates over a stated cohort. It also needs minimum sample counts, label coverage, freshness limits, and uncertainty criteria chosen before the rollout. A comparison of two summaries alone isn't a causal cost estimate.
Start with the receipt contract. Before any metric is evaluated, notice which facts must travel with a window: release ID, observed day, request count, service health, and delayed-label state. Predict the decision for no windows and for one unlabeled hour.
@dataclass(frozen=True)
class CanaryWindow:
window_id: str
release_id: str
observed_day: int
requests: int
error_rate: float
p95_latency_ms: int
delayed_labels_ready: bool
late_warning_cost_delta: float | None
def __post_init__(self):
nonblank(self.window_id, "window_id")
nonblank(self.release_id, "release_id")
@dataclass(frozen=True)
class CanaryReceipt:
receipt_id: str
candidate: str
production_before: str
offline_receipt_id: str
promotion_policy_version: str | None
promotion_policy_fingerprint: str | None
release_fingerprint: str | None
artifact_fingerprint: str | None
windows: tuple[CanaryWindow, ...]
failed_gates: tuple[str, ...]
decision: str
canary_receipts: dict[str, CanaryReceipt] = {}
latest_canary_receipts: dict[str, str] = {}
def finite_metric(value) -> bool:
try:
finite_number(value, "metric")
except ValueError:
return False
return True
def valid_window(window: CanaryWindow) -> bool:
return (
type(window) is CanaryWindow
and type(window.requests) is int and window.requests > 0
and type(window.observed_day) is int and window.observed_day >= 0
and finite_metric(window.error_rate) and 0 <= window.error_rate <= 1
and finite_metric(window.p95_latency_ms) and window.p95_latency_ms >= 0
and (window.late_warning_cost_delta is None
or finite_metric(window.late_warning_cost_delta) and window.late_warning_cost_delta >= -1)
and type(window.delayed_labels_ready) is bool
)
print("canary policy windows:", POLICIES["eta-promotion-v1"].min_canary_windows)The contract exists, but it has no observations yet. Evaluation will now receive the accepted offline receipt plus an ordered list of windows. It can abort on corrupted or measured-negative evidence, but it must hold when the list is too short or labels are still missing.
def accepted_offline_binding_failures(
candidate: Release,
receipt: OfflineReceipt,
) -> list[str]:
failed: list[str] = []
policy = POLICIES.get(candidate.promotion_policy_version)
evidence = (
offline_evidence.get(receipt.offline_evidence_id)
if receipt.offline_evidence_id is not None
else None
)
if receipt.candidate != candidate.release_id:
failed.append("offline_receipt_candidate_mismatch")
if active_offline_receipts.get(candidate.release_id) != receipt.receipt_id:
failed.append("offline_receipt_superseded")
if receipt.decision != "open_canary":
failed.append("offline_receipt_not_accepted")
if receipt.promotion_policy_version != candidate.promotion_policy_version:
failed.append("offline_receipt_policy_version_mismatch")
if (
policy is None
or receipt.promotion_policy_fingerprint != immutable_fingerprint(policy)
):
failed.append("offline_receipt_policy_fingerprint_mismatch")
if receipt.release_fingerprint != immutable_fingerprint(candidate):
failed.append("offline_receipt_release_fingerprint_mismatch")
if receipt.artifact_fingerprint != candidate.candidate_artifact_fingerprint:
failed.append("offline_receipt_artifact_fingerprint_mismatch")
if evidence is None:
failed.append("offline_evidence_not_registered")
else:
failed.extend(evidence_binding_failures(candidate, evidence))
return failed
def evaluate_canary(
candidate_id: str,
offline_receipt_id: str,
windows: list[CanaryWindow],
) -> CanaryReceipt:
if type(windows) is not list or any(type(window) is not CanaryWindow for window in windows):
raise ValueError("windows must be a list of CanaryWindow records")
candidate = registry.get(candidate_id)
policy = POLICIES.get(candidate.promotion_policy_version) if candidate else None
offline_receipt = offline_receipts.get(offline_receipt_id)
failed = []
abort_reasons = []
if candidate is None:
failed.append("candidate_not_registered")
abort_reasons.append("candidate_not_registered")
elif policy is None:
failed.append("promotion_policy_not_registered")
abort_reasons.append("promotion_policy_not_registered")
if aliases.get("canary") != candidate_id:
failed.append("canary_alias_missing")
if offline_receipt is None or candidate is None:
failed.append("accepted_offline_receipt_missing")
abort_reasons.append("accepted_offline_receipt_missing")
else:
binding_failures = accepted_offline_binding_failures(
candidate, offline_receipt
)
failed.extend(binding_failures)
abort_reasons.extend(binding_failures)
if (
candidate is not None
and candidate.previous_release != aliases["production"]
):
failed.append("production_changed_during_canary")
abort_reasons.append("production_changed_during_canary")
if policy is not None and len(windows) < policy.min_canary_windows:
failed.append("observation_window_incomplete")
if len({window.window_id for window in windows}) != len(windows):
failed.append("duplicate_window_id")
abort_reasons.append("duplicate_window_id")
observed_days = [window.observed_day for window in windows if valid_window(window)]
if observed_days != sorted(set(observed_days)):
failed.append("window_order_invalid")
abort_reasons.append("window_order_invalid")
if any(window.release_id != candidate_id for window in windows):
failed.append("mixed_release_windows")
abort_reasons.append("mixed_release_windows")
if any(not valid_window(window) for window in windows):
failed.append("invalid_window_metrics")
abort_reasons.append("invalid_window_metrics")
if any(type(window.requests) is not int or window.requests <= 0 for window in windows):
failed.append("request_count_missing")
abort_reasons.append("request_count_missing")
if policy is not None and any(valid_window(window) and window.error_rate > policy.max_error_rate for window in windows):
failed.append("error_rate_regression")
abort_reasons.append("error_rate_regression")
if policy is not None and any(
valid_window(window) and window.p95_latency_ms > policy.max_p95_latency_ms for window in windows
):
failed.append("latency_regression")
abort_reasons.append("latency_regression")
latest = windows[-1] if windows else None
if latest is None or not latest.delayed_labels_ready or latest.late_warning_cost_delta is None:
failed.append("delayed_quality_not_ready")
if policy is not None and any(
valid_window(window) and window.delayed_labels_ready
and window.late_warning_cost_delta is not None
and window.late_warning_cost_delta > policy.max_late_warning_cost_delta
for window in windows
):
failed.append("late_warning_cost_regression")
abort_reasons.append("late_warning_cost_regression")
decision = (
"abort_canary"
if abort_reasons
else "hold_canary"
if failed
else "ready_for_promotion"
)
receipt = CanaryReceipt(
receipt_id=f"canary-receipt-{len(canary_receipts) + 1}",
candidate=candidate_id,
production_before=aliases["production"],
offline_receipt_id=offline_receipt_id,
promotion_policy_version=(
candidate.promotion_policy_version if candidate is not None else None
),
promotion_policy_fingerprint=(
offline_receipt.promotion_policy_fingerprint
if offline_receipt is not None
else None
),
release_fingerprint=(
offline_receipt.release_fingerprint
if offline_receipt is not None
else None
),
artifact_fingerprint=(
offline_receipt.artifact_fingerprint
if offline_receipt is not None
else None
),
windows=tuple(windows),
failed_gates=tuple(sorted(failed)),
decision=decision,
)
canary_receipts[receipt.receipt_id] = receipt
latest_canary_receipts[candidate_id] = receipt.receipt_id
if decision == "abort_canary" and aliases.get("canary") == candidate_id:
aliases.pop("canary", None)
return receipt
first_hour = CanaryWindow("first-hour", "delivery-risk-v1", 0, 500, 0.002, 118, False, None)
day_seven = CanaryWindow("day-seven", "delivery-risk-v1", 7, 4200, 0.003, 124, True, -0.08)
empty_receipt = evaluate_canary("delivery-risk-v1", accepted_offline_receipt.receipt_id, [])
early_receipt = evaluate_canary(
"delivery-risk-v1", accepted_offline_receipt.receipt_id, [first_hour]
)
ready_receipt = evaluate_canary(
"delivery-risk-v1", accepted_offline_receipt.receipt_id, [first_hour, day_seven]
)
print("empty decision:", empty_receipt.decision)
print("early decision:", early_receipt.decision)
print("ready decision:", ready_receipt.decision)The evaluator has created three receipts. Print them before interpreting the states: the empty list, the first hour, and the complete pair test different evidence boundaries.
print("empty:", empty_receipt.decision, empty_receipt.failed_gates)
print("early:", early_receipt.decision, early_receipt.failed_gates)
print(
"ready:",
ready_receipt.decision,
[window.window_id for window in ready_receipt.windows],
ready_receipt.windows[-1].late_warning_cost_delta,
)empty: hold_canary ('delayed_quality_not_ready', 'observation_window_incomplete')
early: hold_canary ('delayed_quality_not_ready', 'observation_window_incomplete')
ready: ready_for_promotion ['first-hour', 'day-seven'] -0.08late_warning_cost_delta=-0.08 is an injected eight-percent improvement for exercising the controller. It isn't computed from the four offline rows, whose assumed cost change was . Offline and online cohorts differ; real canary metrics must be derived from actual assigned traffic and mature outcomes. The zero-regression threshold here is a policy assumption, not a universal rule.
The empty and first-hour cases return hold_canary. The complete pair returns ready_for_promotion because service limits pass, the latest quality metric is present, and no supplied mature window has a cost regression. A later good delta cannot hide an earlier failing window in the same evaluation history.
A supplied latency, error-rate, or mature cost regression returns abort_canary and removes this candidate's local canary pointer. Invalid metrics or mismatched bindings also abort. Wrong record types are rejected before evaluation. Argo Rollouts distinguishes successful, failed, and inconclusive analysis runs, which can continue, abort, or pause a rollout.[6] A missing measurement and a failing measurement remain different states.
Promote last, then prove rollback
Canary readiness still isn't an alias move. Promotion fetches a stored receipt, follows it to offline evidence, recomputes policy, release, and artifact bindings, and checks the current production pointer. This is a sequential state check, not an atomic compare-and-swap. A changed pointer holds promotion; concurrent writers and a pointer that moves away and back require the generations and transaction described below.
Rollback selects a retained previous release. It doesn't fit new weights. Reload time, runtime readiness, routing convergence, and external feature compatibility determine how long recovery takes; there is no instantaneous or millisecond guarantee here. The next cells exercise missing and early receipts, a local promotion, and a cost-regression rollback. Their production alias remains an in-memory pointer.
First define the alias event and production-receipt shapes. The helper below doesn't move anything; it gives later decisions a place to record why an alias changed.
@dataclass(frozen=True)
class AliasEvent:
action: str
from_release: str
to_release: str
evidence_receipt_id: str
reasons: tuple[str, ...]
@dataclass(frozen=True)
class ProductionReceipt:
receipt_id: str
window: CanaryWindow
release_fingerprint: str | None
policy_fingerprint: str | None
rollback_target_fingerprint: str | None
failed_gates: tuple[str, ...]
decision: str
audit_events: list[AliasEvent] = []
production_receipts: dict[str, ProductionReceipt] = {}
latest_production_receipts: dict[str, str] = {}
production_start_days: dict[str, int] = {"delivery-risk-v0": 0}
latest_valid_production_days: dict[str, int] = {"delivery-risk-v0": 0}
def promote(candidate_id: str, canary_receipt_id: str) -> dict[str, object]:
candidate = registry.get(candidate_id)
canary_receipt = canary_receipts.get(canary_receipt_id)
failed = []
if candidate is None:
failed.append("candidate_not_registered")
if aliases.get("canary") != candidate_id:
failed.append("canary_alias_missing")
if canary_receipt is None:
failed.append("canary_receipt_not_registered")
elif candidate is not None:
if latest_canary_receipts.get(candidate_id) != canary_receipt_id:
failed.append("canary_receipt_superseded")
policy = POLICIES.get(candidate.promotion_policy_version)
if canary_receipt.candidate != candidate_id:
failed.append("canary_receipt_candidate_mismatch")
if canary_receipt.decision != "ready_for_promotion":
failed.append("canary_receipt_not_ready")
if (
canary_receipt.promotion_policy_version
!= candidate.promotion_policy_version
):
failed.append("canary_receipt_policy_version_mismatch")
if (
policy is None
or canary_receipt.promotion_policy_fingerprint
!= immutable_fingerprint(policy)
):
failed.append("canary_receipt_policy_fingerprint_mismatch")
if (
canary_receipt.release_fingerprint
!= immutable_fingerprint(candidate)
):
failed.append("canary_receipt_release_fingerprint_mismatch")
if (
canary_receipt.artifact_fingerprint
!= candidate.candidate_artifact_fingerprint
):
failed.append("canary_receipt_artifact_fingerprint_mismatch")
offline_receipt = offline_receipts.get(canary_receipt.offline_receipt_id)
if offline_receipt is None:
failed.append("offline_receipt_not_registered")
else:
failed.extend(
accepted_offline_binding_failures(candidate, offline_receipt)
)
if (
canary_receipt.promotion_policy_fingerprint
!= offline_receipt.promotion_policy_fingerprint
or canary_receipt.release_fingerprint
!= offline_receipt.release_fingerprint
or canary_receipt.artifact_fingerprint
!= offline_receipt.artifact_fingerprint
):
failed.append("canary_offline_binding_mismatch")
if (
candidate is not None
and candidate.previous_release != aliases["production"]
):
failed.append("production_changed_since_canary_open")
if (
canary_receipt is not None
and canary_receipt.production_before != aliases["production"]
):
failed.append("production_changed_since_canary_receipt")
if failed:
return {"action": "hold_promotion", "reasons": sorted(failed)}
try:
reload_release(candidate_id)
reload_release(aliases["production"])
except (ValueError, OSError):
return {"action": "hold_promotion", "reasons": ["model_reload_failed"]}
previous = aliases["production"]
aliases["previous_production"] = previous
aliases["production"] = candidate_id
production_start_days[candidate_id] = canary_receipt.windows[-1].observed_day
latest_valid_production_days[candidate_id] = production_start_days[candidate_id]
aliases.pop("canary")
event = AliasEvent("promote", previous, candidate_id, canary_receipt_id, ())
audit_events.append(event)
return asdict(event)
print("promote helper ready")Promotion logic is defined, but no alias has moved. Monitoring must first decide whether a production window is healthy, incomplete, or measurably worse. The next cell keeps incomplete labels as hold_monitor and reserves rollback for explicit regressions.
def production_gate_results(window: CanaryWindow) -> tuple[list[str], list[str]]:
"""Return (failed_gates, rollback_reasons).
Incomplete delayed labels are recorded but never authorize rollback.
Only measured regressions (errors, latency, labeled cost) can restore
the previous production alias.
"""
release = registry.get(window.release_id)
if release is None:
return (["production_window_release_not_registered"], [])
policy = POLICIES.get(release.promotion_policy_version)
if policy is None:
return (["production_policy_not_registered"], [])
if not valid_window(window):
return (["invalid_window_metrics"], [])
failed: list[str] = []
rollback_reasons: list[str] = []
if window.error_rate > policy.max_error_rate:
failed.append("error_rate_regression")
rollback_reasons.append("error_rate_regression")
if window.p95_latency_ms > policy.max_p95_latency_ms:
failed.append("latency_regression")
rollback_reasons.append("latency_regression")
if not window.delayed_labels_ready or window.late_warning_cost_delta is None:
# Missing evidence: hold the monitor; do not treat as quality failure.
failed.append("delayed_quality_not_ready")
elif (
window.late_warning_cost_delta is None
or window.late_warning_cost_delta > policy.max_late_warning_cost_delta
):
failed.append("late_warning_cost_regression")
rollback_reasons.append("late_warning_cost_regression")
return failed, rollback_reasons
def evaluate_production(window: CanaryWindow) -> ProductionReceipt:
if type(window) is not CanaryWindow:
raise ValueError("production monitoring requires a CanaryWindow record")
failed, rollback_reasons = production_gate_results(window)
release_mismatch = window.release_id != aliases["production"]
has_day = type(window.observed_day) is int and window.observed_day >= 0
older = has_day and window.observed_day < max(
production_start_days.get(window.release_id, 0),
latest_valid_production_days.get(window.release_id, 0),
)
if older:
failed.append("production_window_older_than_current_evidence")
if release_mismatch:
failed.append("production_window_release_mismatch")
decision = (
"hold_rollback"
if release_mismatch or older
else "rollback_required"
if rollback_reasons
else "hold_monitor"
if failed
else "keep_production"
)
receipt = ProductionReceipt(
receipt_id=f"production-receipt-{len(production_receipts) + 1}",
window=window,
release_fingerprint=(immutable_fingerprint(registry[window.release_id])
if window.release_id in registry else None),
policy_fingerprint=(immutable_fingerprint(POLICIES[registry[window.release_id].promotion_policy_version])
if window.release_id in registry
and registry[window.release_id].promotion_policy_version in POLICIES else None),
rollback_target_fingerprint=(immutable_fingerprint(registry[aliases["previous_production"]])
if aliases.get("previous_production") in registry else None),
failed_gates=tuple(sorted(failed)),
decision=decision,
)
production_receipts[receipt.receipt_id] = receipt
if not older:
latest_production_receipts[window.release_id] = receipt.receipt_id
if not release_mismatch and valid_window(window):
latest_valid_production_days[window.release_id] = window.observed_day
return receipt
def rollback_if_needed(production_receipt_id: str) -> dict[str, object]:
receipt = production_receipts.get(production_receipt_id)
if receipt is None:
return {"action": "hold_rollback", "reason": "production_receipt_not_registered"}
if latest_production_receipts.get(receipt.window.release_id) != production_receipt_id:
return {"action": "hold_rollback", "reason": "production_receipt_superseded"}
if receipt.decision == "hold_rollback":
return {"action": "hold_rollback", "reasons": list(receipt.failed_gates)}
if receipt.decision == "hold_monitor":
# Incomplete labels or other non-regression holds: leave production alone.
return {
"action": "hold_monitor",
"release": aliases["production"],
"reasons": list(receipt.failed_gates),
}
if receipt.window.release_id != aliases["production"]:
return {"action": "hold_rollback", "reason": "production_changed_since_monitor_receipt"}
if receipt.decision == "keep_production":
return {"action": "keep_production", "release": aliases["production"]}
if receipt.decision != "rollback_required":
return {"action": "hold_rollback", "reason": "production_receipt_decision_invalid"}
previous = aliases.get("previous_production")
if previous is None or previous not in registry:
return {"action": "hold_rollback", "reason": "previous_production_not_registered"}
failed_release = aliases["production"]
current = registry[failed_release]
policy = POLICIES.get(current.promotion_policy_version)
if (receipt.release_fingerprint != immutable_fingerprint(current)
or policy is None or receipt.policy_fingerprint != immutable_fingerprint(policy)
or previous != current.previous_release
or receipt.rollback_target_fingerprint != immutable_fingerprint(registry[previous])):
return {"action": "hold_rollback", "reason": "rollback_binding_changed"}
try:
reload_release(previous)
except (ValueError, OSError):
return {"action": "hold_rollback", "reason": "rollback_model_reload_failed"}
aliases["production"] = previous
production_start_days[previous] = receipt.window.observed_day
latest_valid_production_days[previous] = production_start_days[previous]
aliases["rollback_from"] = failed_release
event = AliasEvent(
"rollback",
failed_release,
previous,
receipt.receipt_id,
receipt.failed_gates,
)
audit_events.append(event)
return asdict(event)
print("rollback helper ready")Now exercise the promotion boundary. The missing receipt and early receipt should hold; only the stored ready receipt can move production from v0 to v1. Predict which alias and audit fields change before running the call.
print("fabricated promotion:", promote("delivery-risk-v1", "canary-receipt-missing"))
print("early promotion:", promote("delivery-risk-v1", early_receipt.receipt_id))
print("approved promotion:", promote("delivery-risk-v1", ready_receipt.receipt_id))
assert aliases["production"] == "delivery-risk-v1"
assert predict(reload_release(aliases["production"]), {"delay_minutes": 9}) == 1fabricated promotion: {'action': 'hold_promotion', 'reasons': ['canary_receipt_not_registered']}
early promotion: {'action': 'hold_promotion', 'reasons': ['canary_receipt_not_ready', 'canary_receipt_superseded']}
approved promotion: {'action': 'promote', 'from_release': 'delivery-risk-v0', 'to_release': 'delivery-risk-v1', 'evidence_receipt_id': 'canary-receipt-3', 'reasons': ()}The local production pointer selects the candidate after the approved call. Feed the monitor an invented day-eight window with a labeled cost regression. Service metrics still pass, so this rollback decision must come from the supplied quality metric and must reference the selected release.
degraded = CanaryWindow("production-day-eight", "delivery-risk-v1", 8, 900, 0.004, 130, True, 0.14)
degraded_receipt = evaluate_production(degraded)
rollback_result = rollback_if_needed(degraded_receipt.receipt_id)
print("production decision:", degraded_receipt.decision)
print("rollback action:", rollback_result["action"])
assert aliases["production"] == "delivery-risk-v0"
assert predict(reload_release(aliases["production"]), {"delay_minutes": 9}) == 0production decision: rollback_required
rollback action: rollbackPrint the receipt and aliases last. The audit should contain one promotion and one rollback, each tied to a stored receipt, with production back on v0.
print("production receipt:", degraded_receipt.decision, degraded_receipt.failed_gates)
print("rollback:", rollback_result)
print("aliases:", json.dumps(aliases, sort_keys=True))
print(
"audit:",
[(event.action, event.from_release, event.to_release, event.evidence_receipt_id) for event in audit_events],
)production receipt: rollback_required ('late_warning_cost_regression',)
rollback: {'action': 'rollback', 'from_release': 'delivery-risk-v1', 'to_release': 'delivery-risk-v0', 'evidence_receipt_id': 'production-receipt-1', 'reasons': ('late_warning_cost_regression',)}
aliases: {"previous_production": "delivery-risk-v0", "production": "delivery-risk-v0", "rollback_from": "delivery-risk-v1"}
audit: [('promote', 'delivery-risk-v0', 'delivery-risk-v1', 'canary-receipt-3'), ('rollback', 'delivery-risk-v1', 'delivery-risk-v0', 'production-receipt-1')]The local serving probe now reloads delivery-risk-v0 and changes its nine-minute-delay prediction back from warning to silence. This verifies package selection and reload, not a load-balancer rollback. Preprocessing, coefficients, and threshold travel together. A production rollout must also verify deployed revision identity, runtime compatibility, readiness, routing convergence, and any incompatible external feature dependencies.
Production monitoring mirrors the canary hold-versus-abort rule. A measured error-rate, latency, or labeled cost regression returns rollback_required and can restore the previous alias. Incomplete delayed labels return hold_monitor: keep production on the live release and wait for outcomes. Missing evidence isn't negative evidence, so it mustn't look like a quality failure.
That policy isn't a promise to wait indefinitely. A real system needs a deadline for missing labels or telemetry and a fallback when quality can't be measured. Invalid numbers like NaN aren't healthy metrics. This controller aborts a canary on malformed telemetry but holds production monitoring for investigation; the appropriate production fallback depends on the service's risk.
Test the failures that a happy path misses
A ready receipt can become stale even if its contents never change. A later hold supersedes it. An abort for candidate B must not remove candidate A's canary. A rollback target must still match the package that was recorded when the failure was evaluated. The following independent mutations restore controller state after each check so they don't accidentally authorize the next test.
from contextlib import contextmanager
from copy import deepcopy
@contextmanager
def controller_sandbox():
stores = [registry, artifacts, POLICIES, aliases, offline_evidence, offline_receipts,
active_offline_receipts, canary_receipts, latest_canary_receipts,
production_receipts, latest_production_receipts, production_start_days,
latest_valid_production_days]
saved = [deepcopy(store) for store in stores]
saved_audit = list(audit_events)
try:
yield
finally:
for store, original in zip(stores, saved):
store.clear()
store.update(original)
audit_events[:] = saved_audit
with controller_sandbox():
aliases.update(production="delivery-risk-v0", canary="delivery-risk-v1")
invalid = replace(day_seven, error_rate=float("nan"))
result = evaluate_canary("delivery-risk-v1", accepted_offline_receipt.receipt_id, [first_hour, invalid])
assert result.decision == "abort_canary"
assert "invalid_window_metrics" in result.failed_gates
assert "canary" not in aliases
print("NaN canary metrics: aborted")
with controller_sandbox():
aliases.update(production="delivery-risk-v0", canary="delivery-risk-v1")
evaluate_canary("unknown-candidate", "missing", [])
assert aliases["canary"] == "delivery-risk-v1"
evaluate_canary("delivery-risk-v1", accepted_offline_receipt.receipt_id, [first_hour])
result = promote("delivery-risk-v1", ready_receipt.receipt_id)
assert result["action"] == "hold_promotion"
assert "canary_receipt_superseded" in result["reasons"]
print("unrelated abort and stale ready receipt: contained")
with controller_sandbox():
aliases.update(production="delivery-risk-v1", previous_production="delivery-risk-v0")
incomplete = evaluate_production(replace(degraded, late_warning_cost_delta=None))
assert rollback_if_needed(incomplete.receipt_id)["action"] == "hold_monitor"
failure = evaluate_production(degraded)
POLICIES["eta-promotion-v1"] = replace(policy, max_p95_latency_ms=500)
assert rollback_if_needed(failure.receipt_id)["reason"] == "rollback_binding_changed"
assert aliases["production"] == "delivery-risk-v1"
print("missing cost and changed rollback policy: held")
with controller_sandbox():
aliases.update(production="delivery-risk-v1", previous_production="delivery-risk-v0")
failure = evaluate_production(degraded)
evaluate_production(replace(degraded, window_id="later-healthy", observed_day=9,
late_warning_cost_delta=-0.01))
assert rollback_if_needed(failure.receipt_id)["reason"] == "production_receipt_superseded"
print("old rollback receipt after newer monitor: held")
with controller_sandbox():
aliases.update(production="delivery-risk-v0", canary="delivery-risk-v1")
latest_canary_receipts["delivery-risk-v1"] = ready_receipt.receipt_id
path = Path(built["packages"]["v1"]["path"])
payload = path.read_bytes()
try:
path.write_bytes(payload + b" ")
assert promote("delivery-risk-v1", ready_receipt.receipt_id)["reasons"] == ["model_reload_failed"]
assert aliases["production"] == "delivery-risk-v0"
finally:
path.write_bytes(payload)
print("changed serialized candidate: held before alias move")
with controller_sandbox():
aliases.update(production="delivery-risk-v0", canary="delivery-risk-v1")
earlier_bad = replace(day_seven, window_id="day-six-regression", observed_day=6,
late_warning_cost_delta=0.25)
result = evaluate_canary("delivery-risk-v1", accepted_offline_receipt.receipt_id,
[earlier_bad, day_seven])
assert result.decision == "abort_canary"
assert "late_warning_cost_regression" in result.failed_gates
print("earlier mature regression after later improvement: aborted")
with controller_sandbox():
aliases.update(production="delivery-risk-v1", previous_production="delivery-risk-v0")
newer = evaluate_production(replace(degraded, window_id="newer-day-nine", observed_day=9,
late_warning_cost_delta=-0.01))
older = evaluate_production(replace(degraded, window_id="older-day-one", observed_day=1))
assert older.decision == "hold_rollback"
assert "production_window_older_than_current_evidence" in older.failed_gates
assert latest_production_receipts["delivery-risk-v1"] == newer.receipt_id
assert rollback_if_needed(older.receipt_id)["action"] == "hold_rollback"
assert aliases["production"] == "delivery-risk-v1"
malformed = evaluate_production(replace(degraded, window_id="malformed-day-ten",
observed_day=10, error_rate=float("nan")))
assert malformed.decision == "hold_monitor"
late_failure = evaluate_production(replace(degraded, window_id="late-day-eight"))
assert late_failure.decision == "hold_rollback"
assert latest_valid_production_days["delivery-risk-v1"] == 9
assert rollback_if_needed(late_failure.receipt_id)["action"] == "hold_rollback"
assert aliases["production"] == "delivery-risk-v1"
print("older monitor cannot replace newer evidence: held")
for name, patch in (("critical_slice", {"critical_slice": "false"}),
("late_delivery", {"late_delivery": True}),
("predicted_late", {"predicted_late": 1.0}),
("event_time", {"event_time": ""})):
try:
replace(passing_artifact.evaluation_rows[0], **patch)
except ValueError:
pass
else:
raise AssertionError(f"invalid {name} admitted to offline rows")
print("offline row types and event clocks: rejected")
with controller_sandbox():
aliases.update(production="delivery-risk-v0", canary="delivery-risk-v1")
result = evaluate_canary("delivery-risk-v1", accepted_offline_receipt.receipt_id,
[first_hour, replace(day_seven, p95_latency_ms=10**400)])
assert result.decision == "abort_canary"
assert "invalid_window_metrics" in result.failed_gates
print("huge canary metric: aborted without numeric overflow")NaN canary metrics: aborted
unrelated abort and stale ready receipt: contained
missing cost and changed rollback policy: held
old rollback receipt after newer monitor: held
changed serialized candidate: held before alias move
earlier mature regression after later improvement: aborted
older monitor cannot replace newer evidence: held
offline row types and event clocks: rejected
huge canary metric: aborted without numeric overflowThese tests establish single-process behavior. Checking an alias and then assigning it in Python doesn't prevent a concurrent controller from changing it between those operations. A production implementation needs a transaction or compare-and-swap against an alias generation, with the audit event committed atomically. Generations also prevent an old receipt from becoming valid again after an alias moves away and back. Give receipts a freshness limit and a rollout-attempt identity; this lab's active-receipt tracking isn't a distributed lock.
The monitor retains the latest valid observation day separately from its latest receipt. Malformed telemetry can hold monitoring, but cannot erase that day. A window with a valid day earlier than promotion or retained valid evidence holds without superseding the receipt pointer. Integer days cannot order updates within a day, establish an absolute measurement time, or distinguish repeated rollout attempts. Real window IDs, timestamps, availability, and attempt generations need their own contract. The caller must supply the full canary history here; submitting only later passing windows does not prove that earlier windows were healthy.
💡 Key insight: A hash binds evidence to bytes; it doesn't establish that the evidence is true. Independently measured outcomes and controlled writers are still necessary.
Join fast and delayed monitoring
The controller route stays the same while the evidence changes by product. Before reading the table, name one signal that arrives immediately, one that needs a label, and the event that should restore the previous release.
| Gate type | ETA | Ranking | Forecast | Vision |
|---|---|---|---|---|
| immediate data health | scan freshness | eligible candidate supply | latest counts loaded | photo quality |
| immediate service health | latency/errors | scoring latency | forecast API availability | image scoring latency |
| delayed quality | late warning cost | purchase/return experiment | MAE and peak residual | reviewer-confirmed damage |
| rollback event | stale warning spike | blocked listing exposure | broken alert flood | unsupported escalations |
For scoring systems with delayed labels, canary monitoring pauses wider promotion until enough outcomes arrive. After promotion, incomplete labels hold the production monitor rather than rolling back; only measured regressions restore the previous alias. A model that hasn't failed yet isn't the same as a model that has passed.
⚠️ Common mistake: Treating missing delayed labels as measured failure, or treating a passing canary as a business-lift estimate. Define a deadline and fallback for missing evidence; use an appropriate experiment design for outcome comparisons.
Continuous training is appropriate when a schedule or monitored condition creates a candidate run. Google Cloud's MLOps guidance places automated data validation, model validation, metadata, serving, monitoring, and training triggers around that promotion, with online canary or experiment checks before a candidate serves all traffic.[5] Retraining on a clock isn't a skip-the-gate button. Automation earns trust only when each handoff remains traceable.
Practice: break the release controller
Change one boundary, predict the decision, then run the relevant function inside controller_sandbox(). The happy-path assertions deliberately fail if you mutate their expected outcome; don't interpret that as a broken controller. Reset the fixture before the next mutation so each diagnosis starts from the same evidence.
- Change the third argument in the
delivery-risk-v1release_from_artifact(...)call from"delivery-risk-v0"to"delivery-risk-v-missing". - In
recall_fail_artifact, change the second critical row'spredicted_latevalue from0to1. Rerun and confirm recall is derived as1.0from rows rather than accepted as an argument. - Pass
[day_seven, first_hour]to oneevaluate_canarycall. Confirm that the receipt recordswindow_order_invalid. - Change the
day_sevenconstructor'sdelayed_labels_readyargument fromTruetoFalse. - Change the
day_sevenconstructor'slate_warning_cost_deltaargument from-0.08to0.05. - Replace
ready_receipt.receipt_idwith"canary-receipt-missing"in the approved promotion call. - Before approved promotion, set
aliases["production"] = "delivery-risk-v1"to simulate an out-of-band alias move. Confirm that promotion holds, then reset it to"delivery-risk-v0". - After
ready_receiptis stored but before promotion, setPOLICIES["eta-promotion-v1"] = replace(POLICIES["eta-promotion-v1"], max_p95_latency_ms=500). The ID stays the same, but promotion must hold on fingerprint mismatch. Rerun from the start to restore the policy. - After
ready_receiptis stored but before promotion, setartifacts["eta-artifact-v1"] = replace(passing_artifact, model_artifact="delay-model-v1-repacked"). Promotion must hold because the registered package no longer matches the artifact fingerprint. Rerun from the start to restore it. - Change the
degradedconstructor'slate_warning_cost_deltaargument from0.14to-0.01. - Change the
degradedconstructor's release ID from"delivery-risk-v1"to"delivery-risk-v0". Confirm that monitoring evidence for a different release can't move the current alias. - Replace
degraded_receipt.receipt_idwith"production-receipt-missing"in the rollback call. Which authorization check holds the alias? - Change the
degradedconstructor sodelayed_labels_ready=Falseandlate_warning_cost_delta=None(reset the cost delta and release ID to the original passing-service values first). Confirm the decision ishold_monitor,rollback_if_neededleaves production ondelivery-risk-v1, and no rollback audit event is appended.
Practice answer sketches
Why does a mismatched previous_release block canary traffic?
Answer
previous_release_mismatch fails before traffic moves. The candidate points at a rollback target that isn't the live production release, so restoring it wouldn't prove a return to current behavior.
What happens when the critical offline slice fails?
Answer
The offline receipt returns hold_offline. production stays on delivery-risk-v0, and no canary alias opens.
Why do missing delayed labels and a positive cost delta produce different canary decisions?
Answer
Missing labels mean quality hasn't been measured yet, so the receipt returns hold_canary. A positive delta means measured late-warning cost regressed, so the receipt returns abort_canary and removes the limited-exposure alias. Both block promotion, but only one proves the candidate failed.
Why can't promotion accept any dictionary whose decision field says ready_for_promotion?
Answer
A caller-generated dictionary doesn't establish its evaluator or observations. Promotion fetches stored evidence by ID and rechecks its bindings and the production pointer. This relies on trusted writers; a receipt ID is neither authentication nor an atomic update.
Why do fingerprints matter when policy and artifact IDs haven't changed?
Answer
An ID can be reused while thresholds or artifact contents change. The fingerprints bind the receipt to the values serialized during evaluation. Rechecks detect a same-ID replacement that no longer matches, so fresh evaluation is needed. They cannot prevent an authorized writer from replacing every object and expected fingerprint together.
Why does rollback fetch a stored production receipt instead of accepting caller-supplied reasons?
Answer
Caller-owned reasons can be fabricated or changed. A registered receipt binds the monitored release, window metrics, gate failures, and decision before rollback rechecks the live alias.
What happens when the degraded production delta becomes -0.01?
Answer
No rollback reason remains in this fixture. rollback_if_needed returns keep_production, so delivery-risk-v1 stays live. Production policies should restore a prior alias only when an explicit rollback condition fires.
What happens when a production window has healthy service metrics but delayed_labels_ready=False?
Answer
The monitor records delayed_quality_not_ready and returns hold_monitor. rollback_if_needed keeps delivery-risk-v1 live and appends no rollback event. Incomplete labels mean quality hasn't been measured yet, so they must not restore the previous alias the way a labeled cost regression would.
Trace any attempted alias move backward before you submit: candidate, artifact, offline evidence, canary receipt, production receipt, then alias event. If one link is missing, stale, or bound to a different release, the correct output is a hold. If measured evidence breaches a policy, the output is an abort or rollback with a receipt that explains why.
Submission: prove the handoffs
Extend the supplied builder and controller into the proposed layout below. These additional files are not created by the cells. Make each handoff inspectable: what was registered, which evidence passed or held, who could move an alias, and which receipt supported rollback. Local fixtures verify mechanisms; real rollout evidence requires actual cohorts and observations.
production-ml-platform/
contracts/
release_manifest.schema.json
promotion_policy.json
pipelines/
validate_snapshot.py
train_candidate.py
evaluate_candidate.py
promote_alias.py
registry/
releases.jsonl
monitoring/
live_windows.py
rollback_policy.py
receipts/
offline_gate_report.json
canary_monitor_report.json
alias_audit.jsonl
projects/
eta/
ranking/
forecast/
vision/
tests/
test_failed_gate_never_promotes.py
test_empty_monitor_window_holds.py
test_unregistered_receipt_never_promotes.py
test_alias_race_blocks_promotion.py
test_delayed_labels_block_promotion.py
test_rollback_restores_manifest.py| Artifact | Acceptance condition |
|---|---|
| release schema | identifies data, features, model, action policy, promotion policy, previous release, and candidate-artifact fingerprint |
| registry | retains stable and candidate release values with controlled writers |
| append-only receipts | record derived offline metrics plus policy, release, and artifact fingerprints, and why each candidate passed, held, aborted, promoted, or rolled back |
| alias promotion code | follows stored receipts back to bound evidence and rechecks production before moving alias |
| monitor policy | defines canary pause, promote, abort, rollback |
| tests | execute joins, train-only fitting, reload, malformed metrics, stale windows/receipts, alias changes, and rollback; test concurrent races when a transactional store is implemented |
The controller refuses an alias move unless it's bound to stored evidence. Document QA carries the same habit into a document-answering service: an answer needs cited policy text, or an abstention.