114 lines
3.3 KiB
Python
Executable File
114 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Durable state model for the AIOS autonomous release campaign."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
from typing import Any
|
|
|
|
STATES = (
|
|
"DEVELOPMENT",
|
|
"VALIDATING",
|
|
"PR_READY",
|
|
"MERGING",
|
|
"MAIN_RECONCILING",
|
|
"POSTMERGE_BUILDING",
|
|
"STAGING",
|
|
"EVIDENCE_CLOSING",
|
|
"BINDING_ACCEPTING",
|
|
"TAGGING",
|
|
"DEPLOYING",
|
|
"VERIFYING",
|
|
"REGISTRY_FINALIZING",
|
|
"BASELINE_FINALIZING",
|
|
"PRODUCTION_DEPLOYED",
|
|
)
|
|
RECOVERABLE = "RECOVERABLE_FAILURE"
|
|
RECONCILIATION = "RECONCILIATION_REQUIRED"
|
|
MANUAL = "MANUAL_OWNER_REVIEW_REQUIRED"
|
|
|
|
NEXT = {STATES[i]: STATES[i + 1] for i in range(len(STATES) - 1)}
|
|
|
|
MANUAL_REASONS = frozenset({
|
|
"formal_same_name_tag_conflict",
|
|
"force_push_required",
|
|
"published_history_rewrite_required",
|
|
"non_fast_forward_main_divergence",
|
|
"private_signing_material_exposure_required",
|
|
"repository_identity_conflict",
|
|
"campaign_revoked",
|
|
"operation_outside_campaign_scope",
|
|
})
|
|
|
|
|
|
class OrchestratorError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class ReleaseState:
|
|
version: str
|
|
state: str = "DEVELOPMENT"
|
|
last_stable_state: str = "DEVELOPMENT"
|
|
recoverable_reason: str | None = None
|
|
|
|
def advance(self) -> None:
|
|
if self.state not in NEXT:
|
|
raise OrchestratorError(f"cannot_advance:{self.state}")
|
|
self.last_stable_state = self.state
|
|
self.state = NEXT[self.state]
|
|
self.recoverable_reason = None
|
|
|
|
def recoverable_failure(self, reason: str) -> None:
|
|
self.last_stable_state = self.state
|
|
self.state = RECOVERABLE
|
|
self.recoverable_reason = reason
|
|
|
|
def reconciliation_required(self, reason: str) -> None:
|
|
self.last_stable_state = self.state
|
|
self.state = RECONCILIATION
|
|
self.recoverable_reason = reason
|
|
|
|
def manual_review(self, reason: str) -> None:
|
|
if reason not in MANUAL_REASONS:
|
|
raise OrchestratorError("reason_not_manual_red_line")
|
|
self.last_stable_state = self.state
|
|
self.state = MANUAL
|
|
self.recoverable_reason = reason
|
|
|
|
def resume(self) -> None:
|
|
if self.state not in {RECOVERABLE, RECONCILIATION}:
|
|
raise OrchestratorError("state_not_resumable")
|
|
self.state = self.last_stable_state
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema": "aios.autonomous-release-state.v1",
|
|
"version": self.version,
|
|
"state": self.state,
|
|
"last_stable_state": self.last_stable_state,
|
|
"recoverable_reason": self.recoverable_reason,
|
|
"owner_interaction_required": self.state == MANUAL,
|
|
}
|
|
|
|
|
|
def atomic_state(path: Path, state: ReleaseState) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8", closefd=True) as stream:
|
|
json.dump(state.as_dict(), stream, sort_keys=True, indent=2)
|
|
stream.write("\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(name, path)
|
|
finally:
|
|
try:
|
|
os.unlink(name)
|
|
except FileNotFoundError:
|
|
pass
|