Given slice-level metrics for a baseline and candidate, return
continue, pause, or rollback with reasons.
Hidden answer: invariants, tests, and Python solution
Invariants: critical safety regressions override cost wins, every
slice is checked independently, lower-is-better and higher-is-better
metrics use opposite delta signs, and missing launch-critical
metrics pause the rollout, malformed metric values pause the
rollout, and thin slices need enough baseline and candidate examples
before the comparison is trusted. Test clean wins, cost-only wins,
safety regressions, latency regressions, missing slice metrics,
non-finite telemetry, invalid budgets, and underpowered slices.
import math
def finite_number(value):
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
def decide_voice_agent_release(baseline, candidate, budgets):
reasons = []
for slice_name, metric_budgets in budgets.items():
old = baseline.get(slice_name)
new = candidate.get(slice_name)
if old is None or new is None:
reasons.append(("pause", slice_name, "missing slice"))
continue
min_count = metric_budgets.get("_min_count", 0)
if not isinstance(min_count, int) or min_count < 0:
reasons.append(("pause", slice_name, "invalid min_count"))
continue
old_n = old.get("n", 0)
new_n = new.get("n", 0)
if (
not isinstance(old_n, int)
or not isinstance(new_n, int)
or isinstance(old_n, bool)
or isinstance(new_n, bool)
or old_n < 0
or new_n < 0
):
reasons.append(("pause", slice_name, "invalid sample count"))
continue
if old_n < min_count or new_n < min_count:
reasons.append(("pause", slice_name, "underpowered slice"))
continue
for metric, rule in metric_budgets.items():
if metric == "_min_count":
continue
if metric not in old or metric not in new:
reasons.append(("pause", slice_name, f"missing {metric}"))
continue
if not isinstance(rule, dict) or "direction" not in rule or "allowed_delta" not in rule:
reasons.append(("pause", slice_name, f"bad rule for {metric}"))
continue
direction = rule["direction"]
allowed = rule["allowed_delta"]
severity = rule.get("severity", "normal")
if not finite_number(old[metric]) or not finite_number(new[metric]):
reasons.append(("pause", slice_name, f"invalid {metric}"))
continue
if not finite_number(allowed) or allowed < 0:
reasons.append(("pause", slice_name, f"bad budget for {metric}"))
continue
if direction not in {"lower_is_better", "higher_is_better"}:
reasons.append(("pause", slice_name, f"bad direction for {metric}"))
continue
delta = new[metric] - old[metric]
regressed = (
direction == "lower_is_better" and delta > allowed
) or (
direction == "higher_is_better" and -delta > allowed
)
if regressed:
action = "rollback" if severity == "critical" else "pause"
reasons.append((action, slice_name, metric, round(delta, 4)))
if any(reason[0] == "rollback" for reason in reasons):
return "rollback", reasons
if reasons:
return "pause", reasons
return "continue", []