"""Runtime deployment-state provider for governed health contracts.""" from __future__ import annotations import datetime as dt import json import os import re from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping _OID = re.compile(r"^[0-9a-f]{40}$") _SHA = re.compile(r"^[0-9a-f]{64}$") _STATES = {"STAGED", "ACTIVE", "ROLLED_BACK"} @dataclass(frozen=True) class RuntimeDeploymentState: version: str commit_oid: str artifact_sha256: str deployment_state: str @property def production_deployed(self) -> bool: return self.deployment_state == "ACTIVE" def health_payload(self) -> dict[str, Any]: return { "status": "ok", "version": self.version, "commit_oid": self.commit_oid, "artifact_sha256": self.artifact_sha256, "deployment_state": self.deployment_state, "production_deployed": self.production_deployed, "health_timestamp": dt.datetime.now(dt.timezone.utc).isoformat(), } @classmethod def from_environment(cls, environ: Mapping[str, str] | None = None) -> "RuntimeDeploymentState": env = dict(environ or os.environ) values: dict[str, Any] = {} state_file = env.get("DEPLOYMENT_STATE_FILE", "").strip() if state_file: path = Path(state_file) if path.is_file() and not path.is_symlink(): loaded = json.loads(path.read_text(encoding="utf-8")) if not isinstance(loaded, dict): raise ValueError("deployment_state_file_object_required") values.update(loaded) file_deployment_state = values.get("deployment_state") values.update({k: v for k, v in { "version": env.get("RELEASE_VERSION", values.get("version", "")), "commit_oid": env.get("TARGET_OID", values.get("commit_oid", "")), "artifact_sha256": env.get("ARTIFACT_SHA256", values.get("artifact_sha256", "")), "deployment_state": env.get("DEPLOYMENT_STATE", values.get("deployment_state", "")), }.items() if v != ""}) if file_deployment_state not in (None, ""): values["deployment_state"] = file_deployment_state version = str(values.get("version", "")) commit_oid = str(values.get("commit_oid", "")) artifact_sha256 = str(values.get("artifact_sha256", "")) deployment_state = str(values.get("deployment_state", "")) if not version or not _OID.fullmatch(commit_oid) or not _SHA.fullmatch(artifact_sha256): raise ValueError("deployment_binding_invalid") if deployment_state not in _STATES: raise ValueError("deployment_state_invalid") return cls(version, commit_oid, artifact_sha256, deployment_state)