def evaluate_zero_window(report: dict[str, Any], policy: ZeroWindowPolicy) -> dict[str, Any]:
"""Classify a ``report_deployment_evidence`` result against ``policy``.
Precedence: an observed fallback or dispatch to a legacy body is reported
even when evidence is incomplete; zero requires operator-attested completeness,
declared healthy dispatch capture for every replica, declared healthy
fallback attribution journal capture for every replica (the only journal
that records a legacy guard refusal as fallback), every in-window row
bound to the manifest's declared ``package_version``, the window and
Kernel-entry population meeting policy, and no fallback observed. Approved
composition refusals are carried, never counted as fallback or population.
``incomplete_evidence`` lists ``incomplete_reasons``.
"""
window = report["window"]
span = datetime.fromisoformat(window["end"]) - datetime.fromisoformat(window["start"])
native, fallbacks = report["native_entry_count"], report["fallback_event_count"]
dispatch_counts = report.get("dispatch_outcome_counts", {})
dispatch_legacy = dispatch_counts.get("fallback", 0) + dispatch_counts.get("direct_legacy", 0)
minima = policy.min_native_entries_by_entrypoint
groups = (
{row["entrypoint"]: row["count"] for row in report["native_entry_groups"]} if minima else {}
)
thresholds = (
[
{
"entrypoint": entrypoint,
"minimum": minimum,
"observed": groups.get(entrypoint, 0),
"met": groups.get(entrypoint, 0) >= minimum,
}
for entrypoint, minimum in sorted(minima.items())
]
if minima
else []
)
short_entrypoints = [row for row in thresholds if not row["met"]]
attempt_minima = policy.min_kernel_route_attempts_by_capability
attempt_thresholds = []
if attempt_minima is not None:
attempt_groups = report.get("kernel_route_attempt_groups", [])
for capability, minimum in sorted(attempt_minima.items()):
entries = {row["entrypoint"]: row["count"] for row in attempt_groups
if row["capability"] == capability}
observed = sum(entries.values())
both_continuations = (capability != "invocation.continuation"
or all(entries.get(entry, 0) > 0
for entry in ("resume", "resume_interrupt")))
attempt_thresholds.append({
"capability": capability, "minimum": minimum, "observed": observed,
"entrypoint_counts": entries,
"met": observed >= minimum and both_continuations,
})
short_capabilities = [row for row in attempt_thresholds if not row["met"]]
binding = package_version_binding(report)
fallback_capture = report.get("fallback_capture_status") or "unreported"
incomplete = [reason for reason, failed in (
("report_incomplete", report["completeness"] != "operator_attested"),
("dispatch_capture_unhealthy", report.get("dispatch_capture_status") != "healthy"),
("route_attempt_capture_unhealthy", attempt_minima is not None
and report.get("route_attempt_capture_status") != "healthy"),
(binding, binding != "package_version_bound"),
(f"fallback_capture_{fallback_capture}", fallback_capture != "healthy"),
) if failed]
if fallbacks:
verdict = "fallback_observed"
elif dispatch_legacy:
verdict = "legacy_body_observed"
elif not native:
verdict = "not_measurable"
elif incomplete:
verdict = "incomplete_evidence"
elif (span < policy.min_window or native < policy.min_native_entries
or short_entrypoints or short_capabilities):
verdict = "below_threshold"
else:
verdict = "zero_observed_within_policy"
result = {
"verdict": verdict,
"release_version": report["release_version"],
"deployment_id": report["deployment_id"],
"window": window,
"window_seconds": span.total_seconds(),
"native_entry_count": native,
"fallback_event_count": fallbacks,
"dispatch_legacy_body_outcome_count": dispatch_legacy,
"fallback_groups": report["fallback_groups"],
"approved_migration_refusal_count": report["approved_migration_refusal_count"],
"package_version_binding": binding,
"fallback_capture_status": fallback_capture,
"policy": {
"min_window_seconds": policy.min_window.total_seconds(),
"min_native_entries": policy.min_native_entries,
"approval_reference": policy.approval_reference,
},
"authority": AUTHORITY,
}
if verdict == "incomplete_evidence":
result["incomplete_reasons"] = incomplete
if minima is not None:
result["policy"]["min_native_entries_by_entrypoint"] = dict(minima)
result["entrypoint_thresholds"] = thresholds
if verdict == "below_threshold":
reasons = []
if span < policy.min_window:
reasons.append("window_below_minimum")
if native < policy.min_native_entries:
reasons.append("native_total_below_minimum")
reasons.extend(
f"native_entrypoint_{row['entrypoint']}_below_minimum" for row in short_entrypoints
)
result["below_threshold_reasons"] = reasons
if attempt_minima is not None:
result["policy"]["min_kernel_route_attempts_by_capability"] = dict(attempt_minima)
result["kernel_route_attempt_thresholds"] = attempt_thresholds
result["kernel_route_attempt_count"] = report.get("kernel_route_attempt_count", 0)
result["population_basis"] = "pre_admission_kernel_route_attempts"
if verdict == "below_threshold":
result.setdefault("below_threshold_reasons", []).extend(
f"kernel_route_attempt_{row['capability']}_below_minimum_or_missing_entrypoint"
for row in short_capabilities
)
return result