33 lines
1.2 KiB
Python
33 lines
1.2 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:
|
|
if actual.get(left) != actual.get(right):
|
|
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)
|
|
|