Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
Four products now need the same operational discipline: a late-delivery warning model, a product ranker, a warehouse demand forecast, and a damaged-package photo classifier. Each uses different metrics, but each relies on immutable data evidence, validated candidates, controlled promotion, monitoring, and rollback.
This capstone assembles that discipline into one ML platform workflow. It isn't tied to a particular orchestrator or cloud vendor. A reviewer must be able to trace any live decision back to data, feature, model, policy, and promotion evidence. That requires stored receipts, not a mutable Boolean that happens to say passed.
Define the Shared Release Tuple
Models differ, but their release manifest can share a schema:
| 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 | demand-lags-v1 | parcel-rgb-224-center-crop-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 | blocked-listing and NDCG limits | peak underforecast-cost limit | usable-image and source-slice 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 |
The release tuple prevents an incident review from asking which threshold, feature transform, or gate policy happened to be active. Sculley et al. warn that ML systems accumulate debt through data dependencies, configuration, and feedback loops unless those boundaries are managed explicitly.[1]
Build the Portfolio Repository
Submit a small but inspectable platform surface:
text
1production-ml-platform/
2 contracts/
3 release_manifest.schema.json
4 promotion_policy.json
5 pipelines/
6 validate_snapshot.py
7 train_candidate.py
8 evaluate_candidate.py
9 promote_alias.py
10 registry/
11 releases.jsonl
12 monitoring/
13 live_windows.py
14 rollback_policy.py
15 receipts/
16 offline_gate_report.json
17 canary_monitor_report.json
18 alias_audit.jsonl
19 projects/
20 eta/
21 ranking/
22 forecast/
23 vision/
24 tests/
25 test_failed_gate_never_promotes.py
26 test_empty_monitor_window_holds.py
27 test_unregistered_receipt_never_promotes.py
28 test_alias_race_blocks_promotion.py
29 test_delayed_labels_block_promotion.py
30 test_rollback_restores_manifest.pyGoogle Cloud's MLOps architecture separates automated data/model validation, metadata, serving, monitoring, and continuous-training triggers around promotion.[2] Your repository needn't copy that platform, but it should prove each boundary through a deterministic local fixture and test.
Register a Candidate Without Moving Production
Training completion is evidence, not permission to change live behavior. Start by freezing the whole release tuple. The candidate names its previous release, artifact package, and promotion-policy version. The policy registry owns the immutable thresholds, so a candidate can't loosen its own gate. Fingerprints bind each evaluation to those exact objects before any traffic moves.
01-register-release-registry.py
1from dataclasses import asdict, dataclass, replace
2from hashlib import sha256
3import json
4
5@dataclass(frozen=True)
6class PromotionPolicy:
7 policy_id: str
8 required_offline_gates: tuple[str, ...]
9 required_schema: tuple[str, ...]
10 min_critical_slice_recall: float
11 max_cost_delta: float
12 min_canary_windows: int
13 max_error_rate: float
14 max_p95_latency_ms: int
15 max_late_warning_cost_delta: float
16
17@dataclass(frozen=True)
18class OfflineEvalRow:
19 entity_id: str
20 critical_slice: bool
21 late_delivery: int
22 predicted_late: int
23 baseline_cost: float
24 candidate_cost: float
25
26@dataclass(frozen=True)
27class CandidateArtifacts:
28 artifact_id: str
29 data_snapshot: str
30 feature_version: str
31 model_artifact: str
32 action_policy_version: str
33 observed_schema: tuple[str, ...]
34 training_entity_ids: tuple[str, ...]
35 evaluation_entity_ids: tuple[str, ...]
36 evaluation_rows: tuple[OfflineEvalRow, ...]
37
38@dataclass(frozen=True)
39class Release:
40 release_id: str
41 data_snapshot: str
42 feature_version: str
43 model_artifact: str
44 action_policy_version: str
45 promotion_policy_version: str
46 previous_release: str | None
47 candidate_artifact_id: str
48 candidate_artifact_fingerprint: str
49
50@dataclass(frozen=True)
51class OfflineEvidence:
52 evidence_id: str
53 candidate: str
54 artifact_id: str
55 artifact_fingerprint: str
56 release_fingerprint: str
57 promotion_policy_version: str
58 promotion_policy_fingerprint: str
59 missing_schema_fields: tuple[str, ...]
60 leaked_entity_ids: tuple[str, ...]
61 critical_slice_positives: int
62 critical_slice_true_positives: int
63 critical_slice_recall: float
64 cost_delta: float
65
66 def gates(self, policy: PromotionPolicy) -> dict[str, bool]:
67 return {
68 "schema_valid": not self.missing_schema_fields,
69 "no_leakage": not self.leaked_entity_ids,
70 "critical_slice_pass": (
71 self.critical_slice_positives > 0
72 and self.critical_slice_recall >= policy.min_critical_slice_recall
73 ),
74 "cost_improves": self.cost_delta <= policy.max_cost_delta,
75 }
76
77@dataclass(frozen=True)
78class OfflineReceipt:
79 receipt_id: str
80 candidate: str
81 production_before: str
82 promotion_policy_version: str | None
83 promotion_policy_fingerprint: str | None
84 offline_evidence_id: str | None
85 release_fingerprint: str | None
86 artifact_fingerprint: str | None
87 failed_gates: tuple[str, ...]
88 decision: str
89
90def immutable_fingerprint(value: object) -> str:
91 payload = json.dumps(asdict(value), sort_keys=True, separators=(",", ":"))
92 return sha256(payload.encode("utf-8")).hexdigest()
93
94POLICIES = {
95 "eta-promotion-v1": PromotionPolicy(
96 policy_id="eta-promotion-v1",
97 required_offline_gates=(
98 "schema_valid",
99 "no_leakage",
100 "critical_slice_pass",
101 "cost_improves",
102 ),
103 required_schema=(
104 "entity_id",
105 "event_time",
106 "late_delivery",
107 "predicted_late",
108 "baseline_cost",
109 "candidate_cost",
110 ),
111 min_critical_slice_recall=0.90,
112 max_cost_delta=0.0,
113 min_canary_windows=2,
114 max_error_rate=0.01,
115 max_p95_latency_ms=250,
116 max_late_warning_cost_delta=0.0,
117 )
118}
119
120stable_artifact = CandidateArtifacts(
121 artifact_id="eta-artifact-v0",
122 data_snapshot="carrier-events-through-2026-04-30",
123 feature_version="eta-features-v1",
124 model_artifact="delay-model-v0",
125 action_policy_version="eta-threshold-v1",
126 observed_schema=POLICIES["eta-promotion-v1"].required_schema,
127 training_entity_ids=(),
128 evaluation_entity_ids=(),
129 evaluation_rows=(),
130)
131recall_fail_artifact = CandidateArtifacts(
132 artifact_id="eta-artifact-recall-fail",
133 data_snapshot="carrier-events-through-2026-05-31",
134 feature_version="eta-features-v1",
135 model_artifact="delay-model-recall-fail",
136 action_policy_version="eta-threshold-v1",
137 observed_schema=POLICIES["eta-promotion-v1"].required_schema,
138 training_entity_ids=("train-1", "train-2"),
139 evaluation_entity_ids=("eval-1", "eval-2", "eval-3", "eval-4"),
140 evaluation_rows=(
141 OfflineEvalRow("eval-1", True, 1, 1, 25.0, 24.0),
142 OfflineEvalRow("eval-2", True, 1, 0, 25.0, 24.0),
143 OfflineEvalRow("eval-3", False, 0, 0, 25.0, 24.0),
144 OfflineEvalRow("eval-4", False, 0, 0, 25.0, 24.0),
145 ),
146)
147leaky_artifact = CandidateArtifacts(
148 artifact_id="eta-artifact-leaky",
149 data_snapshot="carrier-events-through-2026-05-31",
150 feature_version="eta-features-v1",
151 model_artifact="delay-model-leaky",
152 action_policy_version="eta-threshold-v1",
153 observed_schema=POLICIES["eta-promotion-v1"].required_schema,
154 training_entity_ids=("train-1", "eval-1"),
155 evaluation_entity_ids=("eval-1", "eval-2", "eval-3", "eval-4"),
156 evaluation_rows=(
157 OfflineEvalRow("eval-1", True, 1, 1, 25.0, 24.0),
158 OfflineEvalRow("eval-2", True, 1, 1, 25.0, 24.0),
159 OfflineEvalRow("eval-3", False, 0, 0, 25.0, 24.0),
160 OfflineEvalRow("eval-4", False, 0, 0, 25.0, 24.0),
161 ),
162)
163passing_artifact = CandidateArtifacts(
164 artifact_id="eta-artifact-v1",
165 data_snapshot="carrier-events-through-2026-05-31",
166 feature_version="eta-features-v1",
167 model_artifact="delay-model-v1",
168 action_policy_version="eta-threshold-v1",
169 observed_schema=POLICIES["eta-promotion-v1"].required_schema,
170 training_entity_ids=("train-1", "train-2"),
171 evaluation_entity_ids=("eval-1", "eval-2", "eval-3", "eval-4"),
172 evaluation_rows=(
173 OfflineEvalRow("eval-1", True, 1, 1, 25.0, 24.0),
174 OfflineEvalRow("eval-2", True, 1, 1, 25.0, 24.0),
175 OfflineEvalRow("eval-3", False, 0, 0, 25.0, 24.0),
176 OfflineEvalRow("eval-4", False, 0, 0, 25.0, 24.0),
177 ),
178)
179artifacts = {
180 artifact.artifact_id: artifact
181 for artifact in (
182 stable_artifact,
183 recall_fail_artifact,
184 leaky_artifact,
185 passing_artifact,
186 )
187}
188
189def release_from_artifact(
190 release_id: str,
191 artifact: CandidateArtifacts,
192 previous_release: str | None,
193) -> Release:
194 return Release(
195 release_id=release_id,
196 data_snapshot=artifact.data_snapshot,
197 feature_version=artifact.feature_version,
198 model_artifact=artifact.model_artifact,
199 action_policy_version=artifact.action_policy_version,
200 promotion_policy_version="eta-promotion-v1",
201 previous_release=previous_release,
202 candidate_artifact_id=artifact.artifact_id,
203 candidate_artifact_fingerprint=immutable_fingerprint(artifact),
204 )
205
206registry = {
207 "delivery-risk-v0": release_from_artifact(
208 "delivery-risk-v0", stable_artifact, None
209 ),
210 "delivery-risk-recall-fail": release_from_artifact(
211 "delivery-risk-recall-fail", recall_fail_artifact, "delivery-risk-v0"
212 ),
213 "delivery-risk-leaky": release_from_artifact(
214 "delivery-risk-leaky", leaky_artifact, "delivery-risk-v0"
215 ),
216 "delivery-risk-v1": release_from_artifact(
217 "delivery-risk-v1", passing_artifact, "delivery-risk-v0"
218 ),
219}
220aliases = {"production": "delivery-risk-v0"}
221offline_evidence: dict[str, OfflineEvidence] = {}
222offline_receipts: dict[str, OfflineReceipt] = {}
223
224print("registry:", list(registry))
225print("production:", aliases["production"])
226print(
227 "policy fingerprint bound:",
228 len(immutable_fingerprint(POLICIES["eta-promotion-v1"])) == 64,
229)Output
1registry: ['delivery-risk-v0', 'delivery-risk-recall-fail', 'delivery-risk-leaky', 'delivery-risk-v1']
2production: delivery-risk-v0
3policy fingerprint bound: TrueOffline gates aren't caller-owned booleans or thresholds. evaluate_candidate loads the registered artifact package, checks schema and split membership, derives recall and cost from its held-out rows, and binds the result to fingerprints for artifact, release, and policy. Only then can open_canary fetch that evidence by ID. Canary and production monitoring use the same pattern later.
02-evaluate-candidate-offline.py
1def release_matches_artifact(
2 release: Release,
3 artifact: CandidateArtifacts,
4) -> bool:
5 return (
6 release.data_snapshot,
7 release.feature_version,
8 release.model_artifact,
9 release.action_policy_version,
10 ) == (
11 artifact.data_snapshot,
12 artifact.feature_version,
13 artifact.model_artifact,
14 artifact.action_policy_version,
15 )
16
17def evidence_binding_failures(
18 candidate: Release,
19 evidence: OfflineEvidence,
20) -> list[str]:
21 failed: list[str] = []
22 policy = POLICIES.get(candidate.promotion_policy_version)
23 artifact = artifacts.get(candidate.candidate_artifact_id)
24 if evidence.candidate != candidate.release_id:
25 failed.append("offline_evidence_candidate_mismatch")
26 if evidence.promotion_policy_version != candidate.promotion_policy_version:
27 failed.append("offline_evidence_policy_version_mismatch")
28 if (
29 policy is None
30 or evidence.promotion_policy_fingerprint != immutable_fingerprint(policy)
31 ):
32 failed.append("offline_evidence_policy_fingerprint_mismatch")
33 if evidence.release_fingerprint != immutable_fingerprint(candidate):
34 failed.append("offline_evidence_release_fingerprint_mismatch")
35 if evidence.artifact_id != candidate.candidate_artifact_id:
36 failed.append("offline_evidence_artifact_id_mismatch")
37 if evidence.artifact_fingerprint != candidate.candidate_artifact_fingerprint:
38 failed.append("offline_evidence_artifact_fingerprint_mismatch")
39 if (
40 artifact is None
41 or immutable_fingerprint(artifact)
42 != candidate.candidate_artifact_fingerprint
43 ):
44 failed.append("registered_artifact_fingerprint_mismatch")
45 elif not release_matches_artifact(candidate, artifact):
46 failed.append("release_artifact_tuple_mismatch")
47 return failed
48
49def evaluate_candidate(candidate_id: str) -> OfflineEvidence:
50 candidate = registry[candidate_id]
51 policy = POLICIES[candidate.promotion_policy_version]
52 artifact = artifacts[candidate.candidate_artifact_id]
53 if immutable_fingerprint(artifact) != candidate.candidate_artifact_fingerprint:
54 raise ValueError("registered artifact changed after release registration")
55 if not release_matches_artifact(candidate, artifact):
56 raise ValueError("release tuple does not match candidate artifact")
57
58 missing_schema = tuple(
59 sorted(set(policy.required_schema) - set(artifact.observed_schema))
60 )
61 leaked_entities = tuple(
62 sorted(
63 set(artifact.training_entity_ids)
64 & set(artifact.evaluation_entity_ids)
65 )
66 )
67 critical_positives = [
68 row for row in artifact.evaluation_rows
69 if row.critical_slice and row.late_delivery == 1
70 ]
71 critical_true_positives = sum(
72 row.predicted_late == 1 for row in critical_positives
73 )
74 critical_recall = (
75 critical_true_positives / len(critical_positives)
76 if critical_positives
77 else 0.0
78 )
79 baseline_cost = sum(row.baseline_cost for row in artifact.evaluation_rows)
80 candidate_cost = sum(row.candidate_cost for row in artifact.evaluation_rows)
81 cost_delta = (
82 candidate_cost / baseline_cost - 1.0
83 if baseline_cost > 0
84 else float("inf")
85 )
86
87 evidence = OfflineEvidence(
88 evidence_id=f"offline-evidence-{len(offline_evidence) + 1}",
89 candidate=candidate_id,
90 artifact_id=artifact.artifact_id,
91 artifact_fingerprint=immutable_fingerprint(artifact),
92 release_fingerprint=immutable_fingerprint(candidate),
93 promotion_policy_version=policy.policy_id,
94 promotion_policy_fingerprint=immutable_fingerprint(policy),
95 missing_schema_fields=missing_schema,
96 leaked_entity_ids=leaked_entities,
97 critical_slice_positives=len(critical_positives),
98 critical_slice_true_positives=critical_true_positives,
99 critical_slice_recall=critical_recall,
100 cost_delta=cost_delta,
101 )
102 offline_evidence[evidence.evidence_id] = evidence
103 return evidence
104
105failing_evidence = evaluate_candidate("delivery-risk-recall-fail")
106leaky_evidence = evaluate_candidate("delivery-risk-leaky")
107passing_evidence = evaluate_candidate("delivery-risk-v1")
108policy = POLICIES["eta-promotion-v1"]
109print("failing gates:", failing_evidence.gates(policy))
110print("leaky gates:", leaky_evidence.gates(policy))
111print("passing gates:", passing_evidence.gates(policy))
112print("stored evidence ids:", list(offline_evidence))Output
1failing gates: {'schema_valid': True, 'no_leakage': True, 'critical_slice_pass': False, 'cost_improves': True}
2leaky gates: {'schema_valid': True, 'no_leakage': False, 'critical_slice_pass': True, 'cost_improves': True}
3passing gates: {'schema_valid': True, 'no_leakage': True, 'critical_slice_pass': True, 'cost_improves': True}
4stored evidence ids: ['offline-evidence-1', 'offline-evidence-2', 'offline-evidence-3']03-open-canary-after-offline-gates.py
1def open_canary(candidate_id: str, offline_evidence_id: str) -> OfflineReceipt:
2 candidate = registry.get(candidate_id)
3 production_before = aliases["production"]
4 failed = []
5 policy = POLICIES.get(candidate.promotion_policy_version) if candidate else None
6 evidence = offline_evidence.get(offline_evidence_id)
7 if candidate is None:
8 failed.append("candidate_not_registered")
9 elif policy is None:
10 failed.append("promotion_policy_not_registered")
11 elif evidence is None:
12 failed.append("offline_evidence_not_registered")
13 else:
14 failed.extend(evidence_binding_failures(candidate, evidence))
15 gates = evidence.gates(policy)
16 failed.extend(
17 gate for gate in policy.required_offline_gates if not gates.get(gate, False)
18 )
19 if candidate is not None and candidate.previous_release != production_before:
20 failed.append("previous_release_mismatch")
21 if aliases.get("canary") not in (None, candidate_id):
22 failed.append("another_canary_is_active")
23
24 receipt = OfflineReceipt(
25 receipt_id=f"offline-receipt-{len(offline_receipts) + 1}",
26 candidate=candidate_id,
27 production_before=production_before,
28 promotion_policy_version=(
29 candidate.promotion_policy_version if candidate is not None else None
30 ),
31 promotion_policy_fingerprint=(
32 evidence.promotion_policy_fingerprint if evidence is not None else None
33 ),
34 offline_evidence_id=offline_evidence_id if evidence is not None else None,
35 release_fingerprint=(
36 evidence.release_fingerprint if evidence is not None else None
37 ),
38 artifact_fingerprint=(
39 evidence.artifact_fingerprint if evidence is not None else None
40 ),
41 failed_gates=tuple(sorted(failed)),
42 decision="hold_offline" if failed else "open_canary",
43 )
44 offline_receipts[receipt.receipt_id] = receipt
45 if receipt.decision == "open_canary":
46 aliases["canary"] = candidate_id
47 return receipt
48
49bad_offline_receipt = open_canary(
50 "delivery-risk-recall-fail", failing_evidence.evidence_id
51)
52accepted_offline_receipt = open_canary("delivery-risk-v1", passing_evidence.evidence_id)
53print("failed:", bad_offline_receipt.decision, bad_offline_receipt.failed_gates)
54print("accepted:", accepted_offline_receipt.decision, accepted_offline_receipt.offline_evidence_id)
55print("aliases:", json.dumps(aliases, sort_keys=True))Output
1failed: hold_offline ('critical_slice_pass',)
2accepted: open_canary offline-evidence-3
3aliases: {"canary": "delivery-risk-v1", "production": "delivery-risk-v0"}A failed offline slice leaves production unchanged. Passing evaluation gates opens only the canary alias. Each attempt appends an immutable receipt with candidate, production base, evidence ID, and policy, release, and artifact fingerprints. Nothing in this cell can overwrite production. Caller-owned metrics, booleans, or thresholds never open canary traffic; only bound evidence derived from the registered artifact can.
Wait for Canary Evidence
Fast checks catch broken schemas, errors, and latency spikes. They can't prove prediction quality when labels arrive later. A canary rollout needs both kinds of evidence. The controller below stores each observation window inside an immutable receipt, refuses to promote after the first hour because late-delivery outcomes aren't ready yet, and handles an empty window list as a hold rather than crashing.
04-canary-window-contract.py
1@dataclass(frozen=True)
2class CanaryWindow:
3 window_id: str
4 release_id: str
5 observed_day: int
6 requests: int
7 error_rate: float
8 p95_latency_ms: int
9 delayed_labels_ready: bool
10 late_warning_cost_delta: float | None
11
12@dataclass(frozen=True)
13class CanaryReceipt:
14 receipt_id: str
15 candidate: str
16 production_before: str
17 offline_receipt_id: str
18 promotion_policy_version: str | None
19 promotion_policy_fingerprint: str | None
20 release_fingerprint: str | None
21 artifact_fingerprint: str | None
22 windows: tuple[CanaryWindow, ...]
23 failed_gates: tuple[str, ...]
24 decision: str
25
26canary_receipts: dict[str, CanaryReceipt] = {}
27
28print("canary policy windows:", POLICIES["eta-promotion-v1"].min_canary_windows)05-evaluate-canary-evidence.py
1def accepted_offline_binding_failures(
2 candidate: Release,
3 receipt: OfflineReceipt,
4) -> list[str]:
5 failed: list[str] = []
6 policy = POLICIES.get(candidate.promotion_policy_version)
7 evidence = (
8 offline_evidence.get(receipt.offline_evidence_id)
9 if receipt.offline_evidence_id is not None
10 else None
11 )
12 if receipt.candidate != candidate.release_id:
13 failed.append("offline_receipt_candidate_mismatch")
14 if receipt.decision != "open_canary":
15 failed.append("offline_receipt_not_accepted")
16 if receipt.promotion_policy_version != candidate.promotion_policy_version:
17 failed.append("offline_receipt_policy_version_mismatch")
18 if (
19 policy is None
20 or receipt.promotion_policy_fingerprint != immutable_fingerprint(policy)
21 ):
22 failed.append("offline_receipt_policy_fingerprint_mismatch")
23 if receipt.release_fingerprint != immutable_fingerprint(candidate):
24 failed.append("offline_receipt_release_fingerprint_mismatch")
25 if receipt.artifact_fingerprint != candidate.candidate_artifact_fingerprint:
26 failed.append("offline_receipt_artifact_fingerprint_mismatch")
27 if evidence is None:
28 failed.append("offline_evidence_not_registered")
29 else:
30 failed.extend(evidence_binding_failures(candidate, evidence))
31 return failed
32
33def evaluate_canary(
34 candidate_id: str,
35 offline_receipt_id: str,
36 windows: list[CanaryWindow],
37) -> CanaryReceipt:
38 candidate = registry.get(candidate_id)
39 policy = POLICIES.get(candidate.promotion_policy_version) if candidate else None
40 offline_receipt = offline_receipts.get(offline_receipt_id)
41 failed = []
42 abort_reasons = []
43 if candidate is None:
44 failed.append("candidate_not_registered")
45 abort_reasons.append("candidate_not_registered")
46 elif policy is None:
47 failed.append("promotion_policy_not_registered")
48 abort_reasons.append("promotion_policy_not_registered")
49 if aliases.get("canary") != candidate_id:
50 failed.append("canary_alias_missing")
51 if offline_receipt is None or candidate is None:
52 failed.append("accepted_offline_receipt_missing")
53 abort_reasons.append("accepted_offline_receipt_missing")
54 else:
55 binding_failures = accepted_offline_binding_failures(
56 candidate, offline_receipt
57 )
58 failed.extend(binding_failures)
59 abort_reasons.extend(binding_failures)
60 if (
61 candidate is not None
62 and candidate.previous_release != aliases["production"]
63 ):
64 failed.append("production_changed_during_canary")
65 abort_reasons.append("production_changed_during_canary")
66 if policy is not None and len(windows) < policy.min_canary_windows:
67 failed.append("observation_window_incomplete")
68 if len({window.window_id for window in windows}) != len(windows):
69 failed.append("duplicate_window_id")
70 abort_reasons.append("duplicate_window_id")
71 observed_days = [window.observed_day for window in windows]
72 if observed_days != sorted(set(observed_days)):
73 failed.append("window_order_invalid")
74 abort_reasons.append("window_order_invalid")
75 if any(window.release_id != candidate_id for window in windows):
76 failed.append("mixed_release_windows")
77 abort_reasons.append("mixed_release_windows")
78 if any(window.requests <= 0 for window in windows):
79 failed.append("request_count_missing")
80 abort_reasons.append("request_count_missing")
81 if policy is not None and any(window.error_rate > policy.max_error_rate for window in windows):
82 failed.append("error_rate_regression")
83 abort_reasons.append("error_rate_regression")
84 if policy is not None and any(
85 window.p95_latency_ms > policy.max_p95_latency_ms for window in windows
86 ):
87 failed.append("latency_regression")
88 abort_reasons.append("latency_regression")
89
90 latest = windows[-1] if windows else None
91 if latest is None or not latest.delayed_labels_ready:
92 failed.append("delayed_quality_not_ready")
93 elif (
94 policy is not None
95 and (
96 latest.late_warning_cost_delta is None
97 or latest.late_warning_cost_delta > policy.max_late_warning_cost_delta
98 )
99 ):
100 failed.append("late_warning_cost_regression")
101 abort_reasons.append("late_warning_cost_regression")
102
103 decision = (
104 "abort_canary"
105 if abort_reasons
106 else "hold_canary"
107 if failed
108 else "ready_for_promotion"
109 )
110 receipt = CanaryReceipt(
111 receipt_id=f"canary-receipt-{len(canary_receipts) + 1}",
112 candidate=candidate_id,
113 production_before=aliases["production"],
114 offline_receipt_id=offline_receipt_id,
115 promotion_policy_version=(
116 candidate.promotion_policy_version if candidate is not None else None
117 ),
118 promotion_policy_fingerprint=(
119 offline_receipt.promotion_policy_fingerprint
120 if offline_receipt is not None
121 else None
122 ),
123 release_fingerprint=(
124 offline_receipt.release_fingerprint
125 if offline_receipt is not None
126 else None
127 ),
128 artifact_fingerprint=(
129 offline_receipt.artifact_fingerprint
130 if offline_receipt is not None
131 else None
132 ),
133 windows=tuple(windows),
134 failed_gates=tuple(sorted(failed)),
135 decision=decision,
136 )
137 canary_receipts[receipt.receipt_id] = receipt
138 if decision == "abort_canary":
139 aliases.pop("canary", None)
140 return receipt
141
142first_hour = CanaryWindow("first-hour", "delivery-risk-v1", 0, 500, 0.002, 118, False, None)
143day_seven = CanaryWindow("day-seven", "delivery-risk-v1", 7, 4200, 0.003, 124, True, -0.08)
144
145empty_receipt = evaluate_canary("delivery-risk-v1", accepted_offline_receipt.receipt_id, [])
146early_receipt = evaluate_canary(
147 "delivery-risk-v1", accepted_offline_receipt.receipt_id, [first_hour]
148)
149ready_receipt = evaluate_canary(
150 "delivery-risk-v1", accepted_offline_receipt.receipt_id, [first_hour, day_seven]
151)
152print("empty decision:", empty_receipt.decision)
153print("early decision:", early_receipt.decision)
154print("ready decision:", ready_receipt.decision)06-print-canary-receipts.py
1print("empty:", empty_receipt.decision, empty_receipt.failed_gates)
2print("early:", early_receipt.decision, early_receipt.failed_gates)
3print(
4 "ready:",
5 ready_receipt.decision,
6 [window.window_id for window in ready_receipt.windows],
7 ready_receipt.windows[-1].late_warning_cost_delta,
8)Output
1empty: hold_canary ('delayed_quality_not_ready', 'observation_window_incomplete')
2early: hold_canary ('delayed_quality_not_ready', 'observation_window_incomplete')
3ready: ready_for_promotion ['first-hour', 'day-seven'] -0.08late_warning_cost_delta=-0.08 means the candidate reduced late-warning cost by eight percent relative to the previous release on this local fixture. It's a teaching threshold, not a universal production policy. Real teams choose windows and limits from product risk, traffic volume, and label delay.
An incomplete window returns hold_canary: gather more evidence without widening exposure. A measured latency, error-rate, or delayed-quality regression returns abort_canary and removes the canary alias. Corrupted or mismatched telemetry aborts too because the controller can't prove safe exposure. Missing evidence and negative evidence aren't the same operational state.
Promote Last, Then Prove Rollback
The final cell makes alias movement explicit. Promotion fetches a stored canary receipt by ID, follows it back to offline evidence, and recomputes the policy, release, and artifact bindings. A ready decision becomes stale if any bound object changes. Rollback follows the same rule: production metrics become a stored receipt before they can restore the previous release. Both paths recheck the live production alias immediately before movement.
07-promote-with-stored-receipt.py
1@dataclass(frozen=True)
2class AliasEvent:
3 action: str
4 from_release: str
5 to_release: str
6 evidence_receipt_id: str
7 reasons: tuple[str, ...]
8
9@dataclass(frozen=True)
10class ProductionReceipt:
11 receipt_id: str
12 window: CanaryWindow
13 failed_gates: tuple[str, ...]
14 decision: str
15
16audit_events: list[AliasEvent] = []
17production_receipts: dict[str, ProductionReceipt] = {}
18
19def promote(candidate_id: str, canary_receipt_id: str) -> dict[str, object]:
20 candidate = registry.get(candidate_id)
21 canary_receipt = canary_receipts.get(canary_receipt_id)
22 failed = []
23 if candidate is None:
24 failed.append("candidate_not_registered")
25 if aliases.get("canary") != candidate_id:
26 failed.append("canary_alias_missing")
27 if canary_receipt is None:
28 failed.append("canary_receipt_not_registered")
29 elif candidate is not None:
30 policy = POLICIES.get(candidate.promotion_policy_version)
31 if canary_receipt.candidate != candidate_id:
32 failed.append("canary_receipt_candidate_mismatch")
33 if canary_receipt.decision != "ready_for_promotion":
34 failed.append("canary_receipt_not_ready")
35 if (
36 canary_receipt.promotion_policy_version
37 != candidate.promotion_policy_version
38 ):
39 failed.append("canary_receipt_policy_version_mismatch")
40 if (
41 policy is None
42 or canary_receipt.promotion_policy_fingerprint
43 != immutable_fingerprint(policy)
44 ):
45 failed.append("canary_receipt_policy_fingerprint_mismatch")
46 if (
47 canary_receipt.release_fingerprint
48 != immutable_fingerprint(candidate)
49 ):
50 failed.append("canary_receipt_release_fingerprint_mismatch")
51 if (
52 canary_receipt.artifact_fingerprint
53 != candidate.candidate_artifact_fingerprint
54 ):
55 failed.append("canary_receipt_artifact_fingerprint_mismatch")
56
57 offline_receipt = offline_receipts.get(canary_receipt.offline_receipt_id)
58 if offline_receipt is None:
59 failed.append("offline_receipt_not_registered")
60 else:
61 failed.extend(
62 accepted_offline_binding_failures(candidate, offline_receipt)
63 )
64 if (
65 canary_receipt.promotion_policy_fingerprint
66 != offline_receipt.promotion_policy_fingerprint
67 or canary_receipt.release_fingerprint
68 != offline_receipt.release_fingerprint
69 or canary_receipt.artifact_fingerprint
70 != offline_receipt.artifact_fingerprint
71 ):
72 failed.append("canary_offline_binding_mismatch")
73 if (
74 candidate is not None
75 and candidate.previous_release != aliases["production"]
76 ):
77 failed.append("production_changed_since_canary_open")
78 if (
79 canary_receipt is not None
80 and canary_receipt.production_before != aliases["production"]
81 ):
82 failed.append("production_changed_since_canary_receipt")
83 if failed:
84 return {"action": "hold_promotion", "reasons": sorted(failed)}
85
86 previous = aliases["production"]
87 aliases["previous_production"] = previous
88 aliases["production"] = candidate_id
89 aliases.pop("canary")
90 event = AliasEvent("promote", previous, candidate_id, canary_receipt_id, ())
91 audit_events.append(event)
92 return asdict(event)
93
94print("promote helper ready")08-monitor-production-and-rollback.py
1def production_gate_results(window: CanaryWindow) -> tuple[list[str], list[str]]:
2 """Return (failed_gates, rollback_reasons).
3
4 Incomplete delayed labels are recorded but never authorize rollback.
5 Only measured regressions (errors, latency, labeled cost) can restore
6 the previous production alias.
7 """
8 release = registry.get(window.release_id)
9 if release is None:
10 return (["production_window_release_not_registered"], [])
11 policy = POLICIES.get(release.promotion_policy_version)
12 if policy is None:
13 return (["production_policy_not_registered"], [])
14 failed: list[str] = []
15 rollback_reasons: list[str] = []
16 if window.error_rate > policy.max_error_rate:
17 failed.append("error_rate_regression")
18 rollback_reasons.append("error_rate_regression")
19 if window.p95_latency_ms > policy.max_p95_latency_ms:
20 failed.append("latency_regression")
21 rollback_reasons.append("latency_regression")
22 if not window.delayed_labels_ready:
23 # Missing evidence: hold the monitor; do not treat as quality failure.
24 failed.append("delayed_quality_not_ready")
25 elif (
26 window.late_warning_cost_delta is None
27 or window.late_warning_cost_delta > policy.max_late_warning_cost_delta
28 ):
29 failed.append("late_warning_cost_regression")
30 rollback_reasons.append("late_warning_cost_regression")
31 return failed, rollback_reasons
32
33def evaluate_production(window: CanaryWindow) -> ProductionReceipt:
34 failed, rollback_reasons = production_gate_results(window)
35 release_mismatch = window.release_id != aliases["production"]
36 if release_mismatch:
37 failed.append("production_window_release_mismatch")
38 decision = (
39 "hold_rollback"
40 if release_mismatch
41 else "rollback_required"
42 if rollback_reasons
43 else "hold_monitor"
44 if failed
45 else "keep_production"
46 )
47 receipt = ProductionReceipt(
48 receipt_id=f"production-receipt-{len(production_receipts) + 1}",
49 window=window,
50 failed_gates=tuple(sorted(failed)),
51 decision=decision,
52 )
53 production_receipts[receipt.receipt_id] = receipt
54 return receipt
55
56def rollback_if_needed(production_receipt_id: str) -> dict[str, object]:
57 receipt = production_receipts.get(production_receipt_id)
58 if receipt is None:
59 return {"action": "hold_rollback", "reason": "production_receipt_not_registered"}
60 if receipt.decision == "hold_rollback":
61 return {"action": "hold_rollback", "reasons": list(receipt.failed_gates)}
62 if receipt.decision == "hold_monitor":
63 # Incomplete labels or other non-regression holds: leave production alone.
64 return {
65 "action": "hold_monitor",
66 "release": aliases["production"],
67 "reasons": list(receipt.failed_gates),
68 }
69 if receipt.window.release_id != aliases["production"]:
70 return {"action": "hold_rollback", "reason": "production_changed_since_monitor_receipt"}
71 if receipt.decision == "keep_production":
72 return {"action": "keep_production", "release": aliases["production"]}
73 if receipt.decision != "rollback_required":
74 return {"action": "hold_rollback", "reason": "production_receipt_decision_invalid"}
75
76 previous = aliases.get("previous_production")
77 if previous is None or previous not in registry:
78 return {"action": "hold_rollback", "reason": "previous_production_not_registered"}
79 failed_release = aliases["production"]
80 aliases["production"] = previous
81 aliases["rollback_from"] = failed_release
82 event = AliasEvent(
83 "rollback",
84 failed_release,
85 previous,
86 receipt.receipt_id,
87 receipt.failed_gates,
88 )
89 audit_events.append(event)
90 return asdict(event)
91
92print("rollback helper ready")09-attempt-promotion-paths.py
1print("fabricated promotion:", promote("delivery-risk-v1", "canary-receipt-missing"))
2print("early promotion:", promote("delivery-risk-v1", early_receipt.receipt_id))
3print("approved promotion:", promote("delivery-risk-v1", ready_receipt.receipt_id))10-detect-production-regression.py
1degraded = CanaryWindow("production-day-eight", "delivery-risk-v1", 8, 900, 0.004, 130, True, 0.14)
2degraded_receipt = evaluate_production(degraded)
3rollback_result = rollback_if_needed(degraded_receipt.receipt_id)
4print("production decision:", degraded_receipt.decision)
5print("rollback action:", rollback_result["action"])11-audit-alias-movements.py
1print("production receipt:", degraded_receipt.decision, degraded_receipt.failed_gates)
2print("rollback:", rollback_result)
3print("aliases:", json.dumps(aliases, sort_keys=True))
4print(
5 "audit:",
6 [(event.action, event.from_release, event.to_release, event.evidence_receipt_id) for event in audit_events],
7)Output
1production receipt: rollback_required ('late_warning_cost_regression',)
2rollback: {'action': 'rollback', 'from_release': 'delivery-risk-v1', 'to_release': 'delivery-risk-v0', 'evidence_receipt_id': 'production-receipt-1', 'reasons': ('late_warning_cost_regression',)}
3aliases: {"previous_production": "delivery-risk-v0", "production": "delivery-risk-v0", "rollback_from": "delivery-risk-v1"}
4audit: [('promote', 'delivery-risk-v0', 'delivery-risk-v1', 'canary-receipt-3'), ('rollback', 'delivery-risk-v1', 'delivery-risk-v0', 'production-receipt-1')]Rollback restores delivery-risk-v0, not its weights alone. That distinction matters because preprocessing, features, thresholds, and policy can all change serving behavior. Each audit event names the stored receipt that authorized its alias movement, so a later reviewer can reconstruct both promotion and rollback.
Production monitoring mirrors the canary hold-vs-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 must not look like a quality failure.
Join Fast and Delayed Monitoring
Live checks differ by product, but the promotion controller handles the same categories:
| 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 should pause wider promotion until enough outcomes arrive. After promotion, incomplete delayed 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.
Continuous training is appropriate when a schedule or monitored condition creates a candidate run. It should never skip data validation, offline comparisons, or a promotion record. The pipeline's value isn't automation alone; it's refusing untraceable changes.
Practice: break the release controller
Run the runnable examples again after each mutation. Predict which receipt or alias changes before reading output, then revert the mutation before trying the next one.
- 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 mutable caller-generated dictionary doesn't prove who evaluated the candidate, which windows were measured, or whether the evidence belongs to the current release. Promotion fetches a stored immutable receipt by ID, verifies its full evidence chain, and rechecks the production alias before moving traffic.
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 policy, release, and artifact fingerprints bind the receipt to exact immutable values. A same-ID replacement makes old evidence stale and blocks promotion until the new objects are evaluated.
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.
Submission checklist
| Artifact | Acceptance condition |
|---|---|
| release schema | identifies data, features, model, action policy, promotion policy, previous release, and candidate-artifact fingerprint |
| registry | contains immutable stable and candidate releases |
| 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 empty-window, failed-gate, fabricated-receipt, alias-race, and rollback paths |
This completes the conventional production ML portfolio. The next capstone returns to LLM products: document QA must apply the same lineage and release discipline to retrieved evidence and generated answers.