42 lines
1.7 KiB
Python
42 lines
1.7 KiB
Python
"""Read-only release reconciliation and durable follow-up intent."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReconciliationReport:
|
|
healthy: bool
|
|
drift: tuple[str, ...]
|
|
actions: tuple[str, ...]
|
|
|
|
|
|
class ReconciliationController:
|
|
CHECKS = (
|
|
("authority_main", "gitea_main", "main_drift"),
|
|
("local_tag_oid", "remote_tag_oid", "tag_drift"),
|
|
("remote_peeled_target", "approved_target_oid", "peeled_target_drift"),
|
|
("registry_version", "baseline_version", "registry_baseline_drift"),
|
|
("approved_artifact_digest", "deployed_artifact_digest", "artifact_drift"),
|
|
("declared_production_state", "service_healthy", "service_health_drift"),
|
|
)
|
|
|
|
def reconcile(self, actual: Mapping[str, Any]) -> ReconciliationReport:
|
|
drift: list[str] = []
|
|
for left, right, label in self.CHECKS:
|
|
left_value = actual.get(left)
|
|
right_value = actual.get(right)
|
|
expected_type = bool if left == "declared_production_state" else str
|
|
complete = (
|
|
left in actual
|
|
and right in actual
|
|
and isinstance(left_value, expected_type)
|
|
and isinstance(right_value, expected_type)
|
|
and (expected_type is bool or bool(left_value.strip()) and bool(right_value.strip()))
|
|
)
|
|
if not complete or left_value != right_value:
|
|
drift.append(label)
|
|
actions = () if not drift else ("open_reconciliation_task", "open_reconciliation_pr", "attach_evidence", "notify_owner")
|
|
return ReconciliationReport(not drift, tuple(drift), actions)
|