83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
"""Immutable release-lineage contracts used by activation and recovery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass
|
|
from typing import Any, Mapping
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReleaseLineage:
|
|
task_id: str
|
|
candidate_id: str
|
|
source_commit: str
|
|
version: str
|
|
required_capabilities: tuple[str, ...]
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
FIXED_V2R2C2 = ReleaseLineage(
|
|
task_id="r42-release-candidate-v2r2c2-live-20260819",
|
|
candidate_id="r42-release-candidate-v2r2c2-20260819",
|
|
source_commit="71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
|
|
version="V2R2C2",
|
|
required_capabilities=("release:candidate",),
|
|
)
|
|
|
|
|
|
class ReleaseLineageError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def validate_fixed_v2r2c2_task(task: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Validate identity/source/version without mutating Task state.
|
|
|
|
This contract intentionally does not bind status/attempt/revision, because
|
|
those are lifecycle state and may change through governed retry/recovery.
|
|
"""
|
|
|
|
if str(task.get("task_id") or "") != FIXED_V2R2C2.task_id:
|
|
raise ReleaseLineageError("FIXED_TASK_ID_DRIFT")
|
|
|
|
capabilities = task.get("required_capabilities")
|
|
if (
|
|
not isinstance(capabilities, list)
|
|
or tuple(capabilities) != FIXED_V2R2C2.required_capabilities
|
|
):
|
|
raise ReleaseLineageError("FIXED_TASK_CAPABILITY_DRIFT")
|
|
|
|
constraints = task.get("constraints")
|
|
if not isinstance(constraints, Mapping):
|
|
raise ReleaseLineageError("FIXED_TASK_CONSTRAINTS_MISSING")
|
|
release = constraints.get("release_candidate")
|
|
if not isinstance(release, Mapping):
|
|
raise ReleaseLineageError("FIXED_RELEASE_BINDING_MISSING")
|
|
|
|
actual = {
|
|
"candidate_id": str(release.get("candidate_id") or ""),
|
|
"source_commit": str(release.get("source_commit") or ""),
|
|
"version": str(release.get("version") or ""),
|
|
}
|
|
expected = {
|
|
"candidate_id": FIXED_V2R2C2.candidate_id,
|
|
"source_commit": FIXED_V2R2C2.source_commit,
|
|
"version": FIXED_V2R2C2.version,
|
|
}
|
|
if actual != expected:
|
|
raise ReleaseLineageError(
|
|
"FIXED_RELEASE_LINEAGE_DRIFT:"
|
|
+ repr({"expected": expected, "actual": actual})
|
|
)
|
|
|
|
return {
|
|
"schema": "aios.release-lineage.validation.v1",
|
|
"status": "VERIFIED",
|
|
"lineage": FIXED_V2R2C2.as_dict(),
|
|
"task_status": task.get("status"),
|
|
"attempt": task.get("attempt"),
|
|
"max_attempts": task.get("max_attempts"),
|
|
"revision": task.get("revision"),
|
|
}
|