"""Durable state and bounded receipts for explicitly executed release runs. The store freezes a selective plan, claims each supported attempt before its executor is called, and records only bounded outcomes. Narrow executors remain separate from this persistence module. """ from __future__ import annotations import base64 import binascii from copy import deepcopy from dataclasses import dataclass from datetime import UTC, datetime import hashlib import json import os from pathlib import Path import re import stat import tempfile import threading from typing import Any from filelock import FileLock, Timeout as FileLockTimeout from .candidate_artifact import CandidateArtifactError, validate_release_channel SCHEMA = "govoplan.release-run" SCHEMA_VERSION = 4 LEGACY_SCHEMA_VERSIONS = {3} MAX_RECORD_BYTES = 2 * 1024 * 1024 MAX_PLAN_STEPS = 512 MAX_REPOSITORY_VERSIONS = 128 MAX_EVENTS = 256 MAX_COMMANDS = 2_048 MAX_LIST_LIMIT = 100 MAX_RUN_RECORDS = 512 MAX_CURSOR_BYTES = 1_024 RUN_ID_PATTERN = re.compile( r"^(?:rr-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}|rr-request-[0-9a-f]{64})$" ) STEP_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,159}$") REQUEST_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{7,127}$") REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") WORKSPACE_FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{64}$") VERSION_PATTERN = re.compile( r"^[0-9]+\.[0-9]+\.[0-9]+(?:[-+][0-9A-Za-z][0-9A-Za-z.-]*)?$" ) STEP_STATES = {"pending", "running", "succeeded", "failed", "interrupted"} PLAN_STATUSES = {"ready", "attention", "blocked"} PLANNED_STEP_STATUSES = {"planned", "needs-executor", "needs-version-metadata"} EVENT_TYPES = { "run_created", "run_resumed", "step_started", "step_succeeded", "step_failed", "step_interrupted", "step_reconciled", "step_retry_requested", } ATTEMPT_OUTCOMES = {"running", "succeeded", "failed", "interrupted"} RECONCILIATION_OUTCOMES = {"effect_absent", "effect_succeeded", "unresolved"} RECONCILIATION_CONFIRMATION = "RECONCILE" START_RECOVERY_RESERVE = 2 PROJECTION_TIMESTAMP = "9999-12-31T23:59:59Z" PROJECTION_RESULT_CODE = "r" * 64 RECEIPT_KINDS = { "catalog_candidate", "catalog_publication", "repository_state", } RECEIPT_ID_PATTERN = re.compile(r"^candidate-[0-9a-f]{32}$") GIT_OBJECT_PATTERN = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$") class ReleaseRunError(RuntimeError): """Base error for durable release run state.""" class ReleaseRunNotFound(ReleaseRunError): """Raised when a run does not exist.""" class ReleaseRunWorkspaceMismatch(ReleaseRunNotFound): """Raised when a valid record belongs to another workspace.""" class ReleaseRunConflict(ReleaseRunError): """Raised when a state transition is unsafe or no longer applicable.""" class ReleaseRunCorrupt(ReleaseRunError): """Raised when a persisted record fails its integrity or schema checks.""" @dataclass(frozen=True, slots=True) class ReleaseStepClaim: """Private executor claim result; attempt identifiers remain fingerprinted.""" run: dict[str, Any] claimed: bool outcome: str result_code: str | None result_receipt: dict[str, Any] | None class ReleaseRunStore: """Atomic, bounded local JSON store for release-run state.""" def __init__(self, root: Path, *, expected_workspace_fingerprint: str) -> None: # Keep the configured path spelling so a symlink root (or symlinked # parent) remains detectable instead of being erased by resolve(). self.root = Path(os.path.abspath(os.fspath(root.expanduser()))) if not WORKSPACE_FINGERPRINT_PATTERN.fullmatch( expected_workspace_fingerprint ): raise ReleaseRunConflict("Release workspace fingerprint is invalid.") self.expected_workspace_fingerprint = expected_workspace_fingerprint self._thread_lock = threading.RLock() self._lock_path = self.root / ".release-runs.lock" def create( self, *, request_id: str, input_snapshot: dict[str, Any], plan_snapshot: dict[str, Any], ) -> dict[str, Any]: request_fingerprint = _request_fingerprint(request_id) immutable_input = _validated_input(input_snapshot) if ( immutable_input["workspace_fingerprint"] != self.expected_workspace_fingerprint ): raise ReleaseRunConflict( "Release run input belongs to a different workspace." ) immutable_plan = _validated_plan(plan_snapshot) now = _timestamp() run_id = _creation_run_id(request_fingerprint) immutable = { "input": immutable_input, "plan": immutable_plan, } immutable["digest"] = _immutable_digest(immutable) steps = [ { "id": item["id"], "state": "pending", "attempt_count": 0, "attempt_fingerprint": None, "started_at": None, "finished_at": None, "result_code": None, "result_receipt": None, } for item in immutable_plan["dry_run_steps"] ] record: dict[str, Any] = { "schema": SCHEMA, "schema_version": SCHEMA_VERSION, "run_id": run_id, "create_request_fingerprint": request_fingerprint, "created_at": now, "updated_at": now, "immutable": immutable, "state": { "status": "planned", "steps": steps, "events": [], "next_event_sequence": 1, "processed_requests": [], "processed_attempts": [], }, } _append_event(record, event_type="run_created") record["updated_at"] = record["state"]["events"][-1]["at"] _refresh_status(record) record["record_digest"] = _record_digest(record) with self._locked(): self._ensure_root() existing = self._read_optional(run_id) if existing is not None: _validate_create_replay( existing, request_fingerprint=request_fingerprint, input_snapshot=immutable_input, ) return _view(existing) retirement = self._retirement_candidate() self._write_new(record) if retirement is not None: try: self._unlink_retained_record(retirement) except ReleaseRunError: # Do not leave the store beyond its hard record bound when # retirement fails before it can make room. self._unlink_new_record(self._path(run_id)) raise return _view(record) def find_created_run( self, *, request_id: str, input_snapshot: dict[str, Any] ) -> dict[str, Any] | None: """Return an exact prior create command before rebuilding its plan.""" request_fingerprint = _request_fingerprint(request_id) immutable_input = _validated_input(input_snapshot) if ( immutable_input["workspace_fingerprint"] != self.expected_workspace_fingerprint ): raise ReleaseRunConflict( "Release run input belongs to a different workspace." ) with self._locked(): record = self._read_optional(_creation_run_id(request_fingerprint)) if record is None: return None _validate_create_replay( record, request_fingerprint=request_fingerprint, input_snapshot=immutable_input, ) return _view(record) def get(self, run_id: str) -> dict[str, Any]: with self._locked(): record = self._read(run_id) return _view(record) def list(self, *, limit: int = 20) -> list[dict[str, Any]]: """Return the first page for backwards-compatible local callers.""" return self.list_page(limit=limit)["runs"] def list_page( self, *, limit: int = 20, cursor: str | None = None ) -> dict[str, Any]: """Return a deterministic descending page, including corrupt entries.""" limit = min(max(int(limit), 1), MAX_LIST_LIMIT) cursor_key = _decode_list_cursor( cursor, expected_workspace_fingerprint=self.expected_workspace_fingerprint, ) with self._locked(): entries: list[tuple[tuple[int, str, str, str], dict[str, Any]]] = [] for path in self._record_paths(): try: record = self._read(path.stem) except ReleaseRunWorkspaceMismatch: continue except (ReleaseRunCorrupt, ReleaseRunNotFound): entries.append( ( (0, "", "", path.stem), { "run_id": path.stem, "status": "unavailable", "error": "Run record is unavailable or failed integrity validation.", }, ) ) continue view = _view(record) immutable_input = view["immutable"]["input"] entries.append( ( ( 1, view["created_at"], view["created_at"], view["run_id"], ), { "run_id": view["run_id"], "created_at": view["created_at"], "updated_at": view["updated_at"], "status": view["state"]["status"], "channel": immutable_input["channel"], "repo_versions": immutable_input["repo_versions"], "recommended_next": view["recommended_next"], }, ) ) entries.sort(key=lambda item: item[0], reverse=True) if cursor_key is not None: entries = [entry for entry in entries if entry[0] < cursor_key] page = entries[: limit + 1] has_more = len(page) > limit visible = page[:limit] next_cursor = ( _encode_list_cursor( visible[-1][0], workspace_fingerprint=self.expected_workspace_fingerprint, ) if has_more and visible else None ) return { "runs": [summary for _key, summary in visible], "next_cursor": next_cursor, } def resume(self, run_id: str, *, request_id: str) -> dict[str, Any]: request_fingerprint = _request_fingerprint(request_id) with self._locked(): record = self._read(run_id) if _request_seen(record, request_fingerprint, "resume", None, None): return _view(record) running_steps = [ step for step in record["state"]["steps"] if step["state"] == "running" ] if not running_steps: raise ReleaseRunConflict( "A release run can only be resumed while a step is running." ) _ensure_command_capacity(record) for step in running_steps: attempt = _attempt_for_fingerprint( record, step["attempt_fingerprint"], step_id=step["id"] ) attempt["outcome"] = "interrupted" attempt["result_code"] = "process_interrupted" step["state"] = "interrupted" step["finished_at"] = _timestamp() step["result_code"] = "process_interrupted" _append_event( record, event_type="step_interrupted", step_id=step["id"], result_code="process_interrupted", ) _remember_request( record, request_fingerprint, "resume", None, None, attempt_fingerprint=running_steps[0]["attempt_fingerprint"], ) _append_event(record, event_type="run_resumed") self._persist_update(record) return _view(record) def retry_step( self, run_id: str, step_id: str, *, request_id: str, ) -> dict[str, Any]: _validate_step_id(step_id) request_fingerprint = _request_fingerprint(request_id) with self._locked(): record = self._read(run_id) if _request_seen(record, request_fingerprint, "retry", step_id, None): return _view(record) step, plan_step = _step_pair(record, step_id) if step["state"] not in {"failed", "interrupted"}: raise ReleaseRunConflict( "Only a failed or interrupted release step can be prepared for retry." ) if step["state"] == "interrupted" and bool(plan_step.get("mutating")): raise ReleaseRunConflict( "Record the observed effect of an interrupted mutating step before retry." ) _ensure_command_capacity(record) attempt_fingerprint = step["attempt_fingerprint"] retry_reason = ( "known_failure" if step["state"] == "failed" else "read_only_interruption" ) step.update( { "state": "pending", "attempt_fingerprint": None, "started_at": None, "finished_at": None, "result_code": None, "result_receipt": None, } ) _remember_request( record, request_fingerprint, "retry", step_id, None, attempt_fingerprint=attempt_fingerprint, ) _append_event( record, event_type="step_retry_requested", step_id=step_id, result_code=retry_reason, ) self._persist_update(record) return _view(record) def reconcile_step( self, run_id: str, step_id: str, *, request_id: str, outcome: str, confirmation: str, result_receipt: dict[str, Any] | None = None, ) -> dict[str, Any]: """Record an operator-observed outcome for an interrupted mutation.""" _validate_step_id(step_id) if outcome not in RECONCILIATION_OUTCOMES: raise ReleaseRunConflict("Release reconciliation outcome is invalid.") if confirmation != RECONCILIATION_CONFIRMATION: raise ReleaseRunConflict( f"Type {RECONCILIATION_CONFIRMATION} to record reconciliation." ) receipt = _validated_result_receipt(result_receipt) if receipt is not None and outcome != "effect_succeeded": raise ReleaseRunConflict( "Only a reconciled successful effect can carry a result receipt." ) request_fingerprint = _request_fingerprint(request_id) with self._locked(): record = self._read(run_id) if _request_seen( record, request_fingerprint, "reconcile", step_id, outcome ): step, _plan_step = _step_pair(record, step_id) if step.get("result_receipt") != receipt: raise ReleaseRunConflict( "The reconciliation request already has a different result receipt." ) return _view(record) step, plan_step = _step_pair(record, step_id) if step["state"] != "interrupted" or not bool( plan_step.get("mutating") ): raise ReleaseRunConflict( "Only an interrupted mutating step can be reconciled." ) attempt_fingerprint = step["attempt_fingerprint"] if outcome == "unresolved": if any( request["operation"] == "reconcile" and request["parameter"] == "unresolved" and request["attempt_fingerprint"] == attempt_fingerprint for request in record["state"]["processed_requests"] ): raise ReleaseRunConflict( "An unresolved outcome is already recorded for this attempt." ) _ensure_command_capacity(record, reserve_after=1) _ensure_unresolved_recovery_byte_capacity(record, step_id) else: _ensure_command_capacity(record) if outcome == "effect_absent": step.update( { "state": "pending", "attempt_fingerprint": None, "started_at": None, "finished_at": None, "result_code": None, "result_receipt": None, } ) elif outcome == "effect_succeeded": step.update( { "state": "succeeded", "finished_at": _timestamp(), "result_code": "reconciled_effect_succeeded", "result_receipt": receipt, } ) _remember_request( record, request_fingerprint, "reconcile", step_id, outcome, attempt_fingerprint=attempt_fingerprint, ) _append_event( record, event_type="step_reconciled", step_id=step_id, result_code=outcome, ) self._persist_update(record) return _view(record) def start_step( self, run_id: str, step_id: str, *, attempt_id: str, ) -> dict[str, Any]: """Claim one step while preserving the original store API.""" return self.claim_step( run_id, step_id, attempt_id=attempt_id ).run def claim_step( self, run_id: str, step_id: str, *, attempt_id: str, ) -> ReleaseStepClaim: """Durably claim a step and distinguish a delayed attempt replay.""" _validate_step_id(step_id) attempt_fingerprint = _request_fingerprint(attempt_id) with self._locked(): record = self._read(run_id) existing_attempt = _find_attempt(record, attempt_fingerprint) if existing_attempt is not None: if existing_attempt["step_id"] != step_id: raise ReleaseRunConflict( "The attempt identifier was already used for another release step." ) step, _plan_step = _step_pair(record, step_id) return ReleaseStepClaim( run=_view(record), claimed=False, outcome=existing_attempt["outcome"], result_code=existing_attempt["result_code"], result_receipt=deepcopy(step.get("result_receipt")), ) step, plan_step = _step_pair(record, step_id) available, reason = _step_available(record, step_id) if not available: raise ReleaseRunConflict(reason) _ensure_start_recovery_capacity(record) _ensure_attempt_capacity(record) step.update( { "state": "running", "attempt_count": step["attempt_count"] + 1, "attempt_fingerprint": attempt_fingerprint, "started_at": _timestamp(), "finished_at": None, "result_code": None, "result_receipt": None, } ) record["state"]["processed_attempts"].append( { "fingerprint": attempt_fingerprint, "step_id": step_id, "outcome": "running", "result_code": None, } ) _append_event(record, event_type="step_started", step_id=step_id) _ensure_start_recovery_byte_capacity( record, step_id, mutating=bool(plan_step.get("mutating")), ) self._persist_update(record) return ReleaseStepClaim( run=_view(record), claimed=True, outcome="running", result_code=None, result_receipt=None, ) def finish_step( self, run_id: str, step_id: str, *, attempt_id: str, succeeded: bool, result_code: str, result_receipt: dict[str, Any] | None = None, ) -> dict[str, Any]: """Record a bounded executor outcome without persisting process output.""" _validate_step_id(step_id) attempt_fingerprint = _request_fingerprint(attempt_id) result_code = _safe_code(result_code) receipt = _validated_result_receipt(result_receipt) if receipt is not None and not succeeded: raise ReleaseRunConflict("A failed release step cannot carry a result receipt.") target_state = "succeeded" if succeeded else "failed" with self._locked(): record = self._read(run_id) step, _plan_step = _step_pair(record, step_id) attempt = _find_attempt(record, attempt_fingerprint) if attempt is None or attempt["step_id"] != step_id: raise ReleaseRunConflict( "The release attempt identifier was not claimed for this step." ) if attempt["outcome"] != "running": if ( attempt["outcome"] == target_state and attempt["result_code"] == result_code and step.get("result_receipt") == receipt ): return _view(record) raise ReleaseRunConflict( "The release attempt already has a different recorded outcome." ) if ( step["state"] != "running" or step["attempt_fingerprint"] != attempt_fingerprint ): raise ReleaseRunConflict( "The release step is not running with this attempt identifier." ) step.update( { "state": target_state, "finished_at": _timestamp(), "result_code": result_code, "result_receipt": receipt, } ) attempt["outcome"] = target_state attempt["result_code"] = result_code _append_event( record, event_type="step_succeeded" if succeeded else "step_failed", step_id=step_id, result_code=result_code, ) self._persist_update(record) return _view(record) def interrupt_step( self, run_id: str, step_id: str, *, attempt_id: str, result_code: str = "executor_outcome_unknown", ) -> dict[str, Any]: """Freeze an executor exception as an ambiguous, reconcilable attempt.""" _validate_step_id(step_id) attempt_fingerprint = _request_fingerprint(attempt_id) result_code = _safe_code(result_code) if result_code != "executor_outcome_unknown": raise ReleaseRunConflict("Executor interruption result code is invalid.") with self._locked(): record = self._read(run_id) step, _plan_step = _step_pair(record, step_id) attempt = _find_attempt(record, attempt_fingerprint) if attempt is None or attempt["step_id"] != step_id: raise ReleaseRunConflict( "The release attempt identifier was not claimed for this step." ) if attempt["outcome"] == "interrupted": if attempt["result_code"] == result_code: return _view(record) raise ReleaseRunConflict( "The release attempt already has a different interruption outcome." ) if ( attempt["outcome"] != "running" or step["state"] != "running" or step["attempt_fingerprint"] != attempt_fingerprint ): raise ReleaseRunConflict( "The release step is not running with this attempt identifier." ) step.update( { "state": "interrupted", "finished_at": _timestamp(), "result_code": result_code, "result_receipt": None, } ) attempt["outcome"] = "interrupted" attempt["result_code"] = result_code _append_event( record, event_type="step_interrupted", step_id=step_id, result_code=result_code, ) self._persist_update(record) return _view(record) def current_attempt_fingerprint(self, run_id: str, step_id: str) -> str: """Return the opaque current fingerprint for server-side artifact recovery.""" _validate_step_id(step_id) with self._locked(): record = self._read(run_id) step, _plan_step = _step_pair(record, step_id) fingerprint = step.get("attempt_fingerprint") if not isinstance(fingerprint, str): raise ReleaseRunConflict( "Release step has no current attempt to reconcile." ) return fingerprint def _persist_update(self, record: dict[str, Any]) -> None: self._assert_workspace(record) record["updated_at"] = _timestamp() _refresh_status(record) record["record_digest"] = _record_digest(record) _validate_record(record) self._atomic_write(self._path(record["run_id"]), record) def _write_new(self, record: dict[str, Any]) -> None: _validate_record(record) path = self._path(record["run_id"]) if path.exists() or path.is_symlink(): raise ReleaseRunConflict("The generated release run already exists.") self._atomic_write(path, record) def _read(self, run_id: str) -> dict[str, Any]: path = self._path(run_id) flags = os.O_RDONLY if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW try: descriptor = os.open(path, flags) except FileNotFoundError as exc: raise ReleaseRunNotFound("Release run was not found.") from exc except OSError as exc: raise ReleaseRunCorrupt( "Release run record could not be opened safely." ) from exc try: metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode): raise ReleaseRunCorrupt("Release run record is not a regular file.") if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid(): raise ReleaseRunCorrupt( "Release run record is not owned by the console operator." ) if metadata.st_mode & 0o077: raise ReleaseRunCorrupt( "Release run record permissions are broader than 0600." ) if metadata.st_size > MAX_RECORD_BYTES: raise ReleaseRunCorrupt("Release run record exceeds its size limit.") with os.fdopen(descriptor, "rb") as handle: descriptor = -1 encoded = handle.read(MAX_RECORD_BYTES + 1) if len(encoded) > MAX_RECORD_BYTES: raise ReleaseRunCorrupt("Release run record exceeds its size limit.") payload = json.loads(encoded.decode("utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise ReleaseRunCorrupt("Release run record is unreadable.") from exc finally: if descriptor >= 0: try: os.close(descriptor) except OSError as exc: raise ReleaseRunCorrupt( "Release run record could not be closed safely." ) from exc _validate_record(payload, expected_run_id=run_id) self._assert_workspace(payload) if payload["schema_version"] in LEGACY_SCHEMA_VERSIONS: payload = _upgrade_record(payload) self._atomic_write(path, payload) return payload def _read_optional(self, run_id: str) -> dict[str, Any] | None: try: return self._read(run_id) except ReleaseRunNotFound: return None def _record_paths(self) -> list[Path]: self._ensure_root() try: return [ path for path in self.root.iterdir() if path.name.endswith(".json") and RUN_ID_PATTERN.fullmatch(path.stem) ] except OSError as exc: raise ReleaseRunCorrupt( "Release run storage could not be enumerated safely." ) from exc def _retirement_candidate(self) -> Path | None: """Select one verified terminal record without deleting it yet.""" paths = self._record_paths() if len(paths) < MAX_RUN_RECORDS: return None if len(paths) > MAX_RUN_RECORDS: raise ReleaseRunConflict( "Release run storage exceeds its retention limit; inspect and " "repair the private store before creating another run." ) completed: list[tuple[tuple[str, str, str], Path]] = [] for path in paths: try: record = self._read(path.stem) except (ReleaseRunCorrupt, ReleaseRunNotFound, ReleaseRunWorkspaceMismatch): # Unavailable evidence is deliberately never deleted implicitly. continue if record["state"]["status"] == "completed": completed.append( ( ( record["updated_at"], record["created_at"], record["run_id"], ), path, ) ) completed.sort(key=lambda item: item[0]) if not completed: raise ReleaseRunConflict( "Release run retention limit is reached; complete or explicitly " "remove retained runs before creating another run." ) return completed[0][1] def _unlink_retained_record(self, path: Path) -> None: try: metadata = path.lstat() if ( not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid()) or metadata.st_mode & 0o077 ): raise ReleaseRunCorrupt( "Completed release run could not be retired safely." ) path.unlink() _fsync_directory(self.root) except ReleaseRunError: raise except OSError as exc: raise ReleaseRunCorrupt( "Completed release run could not be retired safely." ) from exc def _unlink_new_record(self, path: Path) -> None: try: path.unlink() _fsync_directory(self.root) except OSError as exc: raise ReleaseRunCorrupt( "Release run creation could not be rolled back safely after a " "retention failure. Inspect the private store before continuing." ) from exc def _assert_workspace(self, record: dict[str, Any]) -> None: if ( record["immutable"]["input"]["workspace_fingerprint"] != self.expected_workspace_fingerprint ): raise ReleaseRunWorkspaceMismatch("Release run was not found.") def _path(self, run_id: str) -> Path: if not RUN_ID_PATTERN.fullmatch(run_id): raise ReleaseRunNotFound("Release run was not found.") return self.root / f"{run_id}.json" def _ensure_root(self) -> None: _reject_symlink_components(self.root) _create_private_directory_tree(self.root) _reject_symlink_components(self.root) if self.root.is_symlink() or not self.root.is_dir(): raise ReleaseRunCorrupt("Release run storage root is not a directory.") root_metadata = self.root.stat() if hasattr(os, "geteuid") and root_metadata.st_uid != os.geteuid(): raise ReleaseRunCorrupt( "Release run storage is not owned by the console operator." ) try: self.root.chmod(0o700) except OSError as exc: raise ReleaseRunCorrupt( "Release run storage permissions could not be secured." ) from exc if self.root.stat().st_mode & 0o077: raise ReleaseRunCorrupt( "Release run storage permissions are broader than 0700." ) _fsync_directory(self.root) def _atomic_write(self, path: Path, record: dict[str, Any]) -> None: encoded = _canonical_json(record) + b"\n" if len(encoded) > MAX_RECORD_BYTES: raise ReleaseRunConflict("Release run record exceeds its size limit.") descriptor = -1 temporary_path: str | None = None try: descriptor, temporary_path = tempfile.mkstemp( dir=self.root, prefix=".release-run-", suffix=".tmp" ) os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "wb") as handle: descriptor = -1 handle.write(encoded) handle.flush() os.fsync(handle.fileno()) os.replace(temporary_path, path) temporary_path = None written_metadata = path.lstat() if ( not stat.S_ISREG(written_metadata.st_mode) or written_metadata.st_mode & 0o077 or ( hasattr(os, "geteuid") and written_metadata.st_uid != os.geteuid() ) ): raise ReleaseRunCorrupt( "Release run record permissions could not be secured." ) _fsync_directory(self.root) except ReleaseRunError: raise except OSError as exc: raise ReleaseRunCorrupt( "Release run record could not be persisted safely." ) from exc finally: if descriptor >= 0: try: os.close(descriptor) except OSError as exc: raise ReleaseRunCorrupt( "Release run record could not be persisted safely." ) from exc if temporary_path is not None: try: os.unlink(temporary_path) except OSError: pass def _locked(self): # type: ignore[no-untyped-def] self._ensure_root() self._validate_lock_path() return _CombinedLock(self._thread_lock, FileLock(str(self._lock_path))) def _validate_lock_path(self) -> None: try: metadata = self._lock_path.lstat() except FileNotFoundError: return except OSError as exc: raise ReleaseRunCorrupt( "Release run lock could not be inspected safely." ) from exc if ( stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid()) ): raise ReleaseRunCorrupt("Release run lock path is unsafe.") class _CombinedLock: def __init__(self, thread_lock: threading.RLock, file_lock: FileLock) -> None: self.thread_lock = thread_lock self.file_lock = file_lock def __enter__(self) -> None: self.thread_lock.acquire() try: self.file_lock.acquire(timeout=10) except FileLockTimeout as exc: self.thread_lock.release() raise ReleaseRunConflict( "Release run storage is busy; retry the operation." ) from exc except OSError as exc: self.thread_lock.release() raise ReleaseRunCorrupt( "Release run lock could not be acquired safely." ) from exc except BaseException: self.thread_lock.release() raise def __exit__(self, *exc_info: object) -> None: try: try: self.file_lock.release() except OSError as exc: raise ReleaseRunCorrupt( "Release run lock could not be released safely." ) from exc finally: self.thread_lock.release() def _validated_input(value: dict[str, Any]) -> dict[str, Any]: if not isinstance(value, dict): raise ReleaseRunConflict("Release run input must be an object.") allowed = { "channel", "repo_versions", "online", "remote_tags", "public_catalog", "include_migrations", "workspace_fingerprint", } if set(value) != allowed: raise ReleaseRunConflict("Release run input fields are invalid.") channel = value.get("channel") repo_versions = value.get("repo_versions") try: channel = validate_release_channel(channel) except CandidateArtifactError: raise ReleaseRunConflict("Release channel is invalid.") if ( not isinstance(repo_versions, dict) or not repo_versions or len(repo_versions) > MAX_REPOSITORY_VERSIONS ): raise ReleaseRunConflict("At least one repository version is required.") normalized_versions: dict[str, str] = {} for repo, version in sorted(repo_versions.items()): if not isinstance(repo, str) or not REPOSITORY_PATTERN.fullmatch(repo): raise ReleaseRunConflict("Release repository name is invalid.") if not isinstance(version, str) or not VERSION_PATTERN.fullmatch(version): raise ReleaseRunConflict("Release target version is invalid.") normalized_versions[repo] = version result: dict[str, Any] = { "channel": channel, "repo_versions": normalized_versions, } for field in ("online", "remote_tags", "public_catalog", "include_migrations"): if not isinstance(value.get(field), bool): raise ReleaseRunConflict(f"Release run input {field} must be boolean.") result[field] = value[field] workspace_fingerprint = value.get("workspace_fingerprint") if not isinstance( workspace_fingerprint, str ) or not WORKSPACE_FINGERPRINT_PATTERN.fullmatch(workspace_fingerprint): raise ReleaseRunConflict("Release workspace fingerprint is invalid.") result["workspace_fingerprint"] = workspace_fingerprint return result def _validated_plan(value: dict[str, Any]) -> dict[str, Any]: if not isinstance(value, dict): raise ReleaseRunConflict("Release plan snapshot must be an object.") copied = deepcopy(value) base_fields = { "generated_at", "target_channel", "status", "units", "compatibility", "dry_run_steps", "notes", "gate_findings", "recommended_action", "source_preflight_ready", } if set(copied) not in (base_fields, base_fields | {"runtime_binding"}): raise ReleaseRunConflict("Release plan snapshot fields are invalid.") if "runtime_binding" in copied: runtime_binding = copied["runtime_binding"] validated_runtime = _validated_result_receipt(runtime_binding) if ( validated_runtime != runtime_binding or runtime_binding.get("kind") != "repository_state" or runtime_binding.get("repo") != "govoplan" or runtime_binding.get("target_tag") != "" or runtime_binding.get("tag_object") is not None or runtime_binding.get("worktree_clean") is not True ): raise ReleaseRunConflict( "Release plan runtime binding is invalid." ) if not _is_timestamp(copied.get("generated_at")): raise ReleaseRunConflict("Release plan timestamp is invalid.") if copied.get("status") not in PLAN_STATUSES: raise ReleaseRunConflict("Release plan status is invalid.") try: target_channel = validate_release_channel(copied.get("target_channel")) except CandidateArtifactError: raise ReleaseRunConflict("Release plan channel is invalid.") if target_channel != copied["target_channel"]: raise ReleaseRunConflict("Release plan channel is invalid.") if not isinstance(copied.get("units"), list): raise ReleaseRunConflict("Release plan units are invalid.") if not isinstance(copied.get("compatibility"), list) or not isinstance( copied.get("gate_findings"), list ): raise ReleaseRunConflict("Release plan findings are invalid.") if not isinstance(copied.get("notes"), list) or not all( isinstance(note, str) for note in copied["notes"] ): raise ReleaseRunConflict("Release plan notes are invalid.") if not isinstance(copied.get("source_preflight_ready"), bool): raise ReleaseRunConflict("Release plan preflight state is invalid.") recommendation = copied.get("recommended_action") if recommendation is not None and not isinstance(recommendation, dict): raise ReleaseRunConflict("Release plan recommendation is invalid.") steps = copied.get("dry_run_steps") if ( not isinstance(steps, list) or not steps or len(steps) > MAX_PLAN_STEPS ): raise ReleaseRunConflict("Release plan step collection is invalid.") seen: set[str] = set() for step in steps: if not isinstance(step, dict) or set(step) not in ({ "id", "title", "detail", "command", "cwd", "mutating", "repo", "status", }, { "id", "title", "detail", "command", "cwd", "mutating", "repo", "status", "source_binding", }): raise ReleaseRunConflict("Release plan step is invalid.") step_id = step.get("id") _validate_step_id(step_id) if step_id in seen: raise ReleaseRunConflict("Release plan step identifiers must be unique.") seen.add(step_id) if not isinstance(step.get("mutating", False), bool): raise ReleaseRunConflict("Release plan step mutation flag is invalid.") if step.get("status") not in PLANNED_STEP_STATUSES: raise ReleaseRunConflict("Release plan step status is invalid.") for field, maximum in (("title", 500), ("detail", 2_000)): if not isinstance(step.get(field), str) or len(step[field]) > maximum: raise ReleaseRunConflict("Release plan step text is invalid.") for field in ("command", "cwd", "repo"): if step.get(field) is not None and ( not isinstance(step[field], str) or len(step[field]) > 8_192 ): raise ReleaseRunConflict("Release plan step metadata is invalid.") if "source_binding" in step: binding = step["source_binding"] if binding is not None: validated_binding = _validated_result_receipt(binding) if ( validated_binding != binding or binding.get("kind") != "repository_state" or ( step.get("repo") is not None and binding.get("repo") != step.get("repo") ) ): raise ReleaseRunConflict( "Release plan source binding is invalid." ) encoded = _canonical_json(copied) if len(encoded) > MAX_RECORD_BYTES: raise ReleaseRunConflict("Release plan snapshot exceeds its size limit.") return copied def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: try: if not isinstance(value, dict): raise ValueError if set(value) != { "schema", "schema_version", "run_id", "create_request_fingerprint", "created_at", "updated_at", "immutable", "state", "record_digest", }: raise ValueError if ( value.get("schema") != SCHEMA or type(value.get("schema_version")) is not int or value["schema_version"] not in {SCHEMA_VERSION, *LEGACY_SCHEMA_VERSIONS} ): raise ValueError schema_version = value["schema_version"] run_id = value.get("run_id") if not isinstance(run_id, str) or not RUN_ID_PATTERN.fullmatch(run_id): raise ValueError if expected_run_id is not None and run_id != expected_run_id: raise ValueError if not _is_sha256(value.get("create_request_fingerprint")): raise ValueError if run_id != _creation_run_id(value["create_request_fingerprint"]): raise ValueError for key in ("created_at", "updated_at"): if not _is_timestamp(value.get(key)): raise ValueError created_at = value["created_at"] updated_at = value["updated_at"] if updated_at < created_at: raise ValueError immutable = value.get("immutable") if not isinstance(immutable, dict) or set(immutable) != { "input", "plan", "digest", }: raise ValueError validated_input = _validated_input(immutable["input"]) validated_plan = _validated_plan(immutable["plan"]) if immutable["input"] != validated_input or immutable["plan"] != validated_plan: raise ValueError if immutable.get("digest") != _immutable_digest(immutable): raise ValueError if validated_plan["target_channel"] != validated_input["channel"]: raise ValueError planned_units = validated_plan["units"] planned_versions = { unit.get("repo"): unit.get("target_version") for unit in planned_units if isinstance(unit, dict) and isinstance(unit.get("repo"), str) and isinstance(unit.get("target_version"), str) } if planned_versions != validated_input["repo_versions"] or len( planned_versions ) != len(planned_units): raise ValueError state = value.get("state") if not isinstance(state, dict) or set(state) != { "status", "steps", "events", "next_event_sequence", "processed_requests", "processed_attempts", }: raise ValueError steps = state.get("steps") plan_steps = validated_plan["dry_run_steps"] plan_step_ids = {plan_step["id"] for plan_step in plan_steps} if not isinstance(steps, list) or len(steps) != len(plan_steps): raise ValueError for step, plan_step in zip(steps, plan_steps, strict=True): expected_step_fields = { "id", "state", "attempt_count", "attempt_fingerprint", "started_at", "finished_at", "result_code", } if schema_version >= 4: expected_step_fields.add("result_receipt") if ( not isinstance(step, dict) or set(step) != expected_step_fields or step.get("id") != plan_step["id"] ): raise ValueError if step.get("state") not in STEP_STATES: raise ValueError if type(step.get("attempt_count")) is not int or step["attempt_count"] < 0: raise ValueError fingerprint = step.get("attempt_fingerprint") if fingerprint is not None and not _is_sha256(fingerprint): raise ValueError for timestamp in (step.get("started_at"), step.get("finished_at")): if timestamp is not None and ( not isinstance(timestamp, str) or len(timestamp) > 40 ): raise ValueError result_code = step.get("result_code") if result_code is not None: _safe_code(result_code) receipt = _validated_result_receipt(step.get("result_receipt")) if receipt != step.get("result_receipt"): raise ValueError _validate_step_state_fields(step) for timestamp in (step.get("started_at"), step.get("finished_at")): if timestamp is not None and not ( created_at <= timestamp <= updated_at ): raise ValueError first_incomplete_seen = False for step in steps: if first_incomplete_seen and step["state"] != "pending": raise ValueError if step["state"] != "succeeded": first_incomplete_seen = True events = state.get("events") if not isinstance(events, list) or len(events) > MAX_EVENTS: raise ValueError previous_sequence = 0 previous_event_at: str | None = None for event in events: if not isinstance(event, dict) or event.get("type") not in EVENT_TYPES: raise ValueError expected_event_keys = {"sequence", "at", "type"} if event.get("step_id") is not None: expected_event_keys.add("step_id") if event.get("result_code") is not None: expected_event_keys.add("result_code") if set(event) != expected_event_keys: raise ValueError sequence = event.get("sequence") if type(sequence) is not int or sequence <= previous_sequence: raise ValueError previous_sequence = sequence if not _is_timestamp(event.get("at")): raise ValueError if not (created_at <= event["at"] <= updated_at): raise ValueError if previous_event_at is not None and event["at"] < previous_event_at: raise ValueError previous_event_at = event["at"] if event.get("step_id") is not None: _validate_step_id(event["step_id"]) if event["step_id"] not in plan_step_ids: raise ValueError if event.get("result_code") is not None: _safe_code(event["result_code"]) _validate_event_fields(event) next_sequence = state.get("next_event_sequence") if type(next_sequence) is not int or next_sequence <= previous_sequence: raise ValueError requests = state.get("processed_requests") if not isinstance(requests, list) or len(requests) > MAX_COMMANDS: raise ValueError for request in requests: if ( not isinstance(request, dict) or set(request) != { "fingerprint", "operation", "step_id", "parameter", "attempt_fingerprint", } or not _is_sha256(request.get("fingerprint")) or not _is_sha256(request.get("attempt_fingerprint")) or request.get("operation") not in {"resume", "retry", "reconcile"} ): raise ValueError if request.get("step_id") is not None: _validate_step_id(request["step_id"]) if request["step_id"] not in plan_step_ids: raise ValueError if (request["operation"] == "resume") != (request["step_id"] is None): raise ValueError parameter = request["parameter"] if request["operation"] == "reconcile": if parameter not in RECONCILIATION_OUTCOMES: raise ValueError elif parameter is not None: raise ValueError if len({request["fingerprint"] for request in requests}) != len(requests): raise ValueError attempts = state.get("processed_attempts") if not isinstance(attempts, list) or len(attempts) > MAX_COMMANDS: raise ValueError for attempt in attempts: if ( not isinstance(attempt, dict) or set(attempt) != {"fingerprint", "step_id", "outcome", "result_code"} or not _is_sha256(attempt.get("fingerprint")) or attempt.get("outcome") not in ATTEMPT_OUTCOMES ): raise ValueError _validate_step_id(attempt.get("step_id")) if attempt["step_id"] not in plan_step_ids: raise ValueError attempt_result = attempt.get("result_code") if attempt["outcome"] == "running": if attempt_result is not None: raise ValueError else: if attempt_result is None: raise ValueError _safe_code(attempt_result) if ( attempt["outcome"] == "interrupted" and attempt_result not in {"process_interrupted", "executor_outcome_unknown"} ): raise ValueError if len({attempt["fingerprint"] for attempt in attempts}) != len(attempts): raise ValueError attempts_by_fingerprint = { attempt["fingerprint"]: attempt for attempt in attempts } for request in requests: request_attempt = attempts_by_fingerprint.get( request["attempt_fingerprint"] ) if request_attempt is None: raise ValueError if ( request["step_id"] is not None and request_attempt["step_id"] != request["step_id"] ): raise ValueError if request["operation"] == "resume" and request_attempt[ "outcome" ] != "interrupted": raise ValueError if request["operation"] == "retry" and request_attempt[ "outcome" ] not in {"failed", "interrupted"}: raise ValueError if request["operation"] == "reconcile" and request_attempt[ "outcome" ] != "interrupted": raise ValueError attempts_per_step = { step_id: sum( attempt["step_id"] == step_id for attempt in attempts ) for step_id in plan_step_ids } for step in steps: if step["attempt_count"] != attempts_per_step[step["id"]]: raise ValueError fingerprint = step["attempt_fingerprint"] if fingerprint is None: continue attempt = attempts_by_fingerprint.get(fingerprint) if attempt is None or attempt["step_id"] != step["id"]: raise ValueError expected_outcome = step["state"] if step["result_code"] == "reconciled_effect_succeeded": if step["state"] != "succeeded" or attempt["outcome"] != "interrupted": raise ValueError elif attempt["outcome"] != expected_outcome: raise ValueError if state.get("status") != _status(value): raise ValueError if value.get("record_digest") != _record_digest(value): raise ValueError except ( KeyError, StopIteration, TypeError, ValueError, ReleaseRunConflict, ) as exc: raise ReleaseRunCorrupt( "Release run record failed schema or integrity validation." ) from exc def _view(record: dict[str, Any]) -> dict[str, Any]: view = deepcopy(record) plan_steps = view["immutable"]["plan"]["dry_run_steps"] mutable_steps = view["state"]["steps"] enriched_steps: list[dict[str, Any]] = [] for index, (step, plan_step) in enumerate( zip(mutable_steps, plan_steps, strict=True) ): available, disabled_reason = _step_available(record, step["id"]) retry_available = step["state"] == "failed" or ( step["state"] == "interrupted" and not bool(plan_step.get("mutating")) ) enriched_steps.append( { **plan_step, **{ key: value for key, value in step.items() if key != "attempt_fingerprint" }, "order": index + 1, "available": available, "retry_available": retry_available, "reconciliation_required": step["state"] == "interrupted" and bool(plan_step.get("mutating")), "disabled_reason": disabled_reason, } ) view["state"]["steps"] = enriched_steps view["state"].pop("processed_requests", None) view["state"].pop("processed_attempts", None) view["recommended_next"] = _recommended_next(record) return view def _step_available(record: dict[str, Any], step_id: str) -> tuple[bool, str]: step, plan_step = _step_pair(record, step_id) if record["immutable"]["plan"].get("status") == "blocked": return False, "The immutable plan contains release-gate blockers." if step["state"] != "pending": descriptions = { "running": "This step is already running.", "succeeded": "This step is complete.", "failed": "Request an explicit retry before running this step again.", "interrupted": "Reconcile the interrupted attempt before retry.", } return False, descriptions.get(step["state"], "This step is unavailable.") planned_status = plan_step.get("status", "planned") if planned_status != "planned": return ( False, f"The plan marks this step {planned_status}; no executor is available.", ) steps = record["state"]["steps"] index = next(index for index, item in enumerate(steps) if item["id"] == step_id) incomplete = next( (item for item in steps[:index] if item["state"] != "succeeded"), None ) if incomplete is not None: return False, f"Complete {incomplete['id']} first." return True, "" def _recommended_next(record: dict[str, Any]) -> dict[str, Any]: plan = record["immutable"]["plan"] if plan.get("status") == "blocked": recommendation = plan.get("recommended_action") if isinstance(recommendation, dict): return { "id": recommendation.get("id") or "resolve_plan_blocker", "step_id": None, "title": recommendation.get("title") or "Resolve release gate", "detail": recommendation.get("detail") or "The frozen plan is blocked.", "remediation": recommendation.get("remediation") or "Resolve the blocker and create a new run from a fresh plan.", "available": False, } return { "id": "resolve_plan_blocker", "step_id": None, "title": "Resolve release gate", "detail": "The frozen plan is blocked.", "remediation": "Resolve the blocker and create a new run from a fresh plan.", "available": False, } for step, plan_step in zip( record["state"]["steps"], plan["dry_run_steps"], strict=True ): if step["state"] == "running": return _step_recommendation( "wait_for_step", step, plan_step, "Wait for the active executor or resume the run after an interruption.", False, ) if step["state"] == "failed": return _step_recommendation( "retry_step", step, plan_step, "Review the bounded result code, then explicitly prepare this known failed step for retry.", True, ) if step["state"] == "interrupted": if bool(plan_step.get("mutating")): return _step_recommendation( "reconcile_step", step, plan_step, "Reconcile the local and remote effect before explicitly preparing a retry.", True, ) return _step_recommendation( "retry_step", step, plan_step, "The read-only attempt was interrupted and can be explicitly prepared for retry.", True, ) if step["state"] == "pending": available, reason = _step_available(record, step["id"]) return _step_recommendation( "execute_step" if available else "unavailable_step", step, plan_step, "Use the existing preview and narrowly confirmed executor controls." if available else reason, available, ) return { "id": "run_complete", "step_id": None, "title": "Release run complete", "detail": "Every recorded step succeeded.", "remediation": "Review publication and installation evidence before closing the release.", "available": False, } def _step_recommendation( recommendation_id: str, step: dict[str, Any], plan_step: dict[str, Any], remediation: str, available: bool, ) -> dict[str, Any]: return { "id": recommendation_id, "step_id": step["id"], "title": plan_step.get("title") or step["id"], "detail": plan_step.get("detail") or "Release plan step.", "remediation": remediation, "available": available, } def _step_pair( record: dict[str, Any], step_id: str ) -> tuple[dict[str, Any], dict[str, Any]]: for step, plan_step in zip( record["state"]["steps"], record["immutable"]["plan"]["dry_run_steps"], strict=True, ): if step["id"] == step_id: return step, plan_step raise ReleaseRunNotFound("Release run step was not found.") def _append_event( record: dict[str, Any], *, event_type: str, step_id: str | None = None, result_code: str | None = None, at: str | None = None, ) -> None: if event_type not in EVENT_TYPES: raise ReleaseRunConflict("Release event type is invalid.") event: dict[str, Any] = { "sequence": record["state"]["next_event_sequence"], "at": at if at is not None else _timestamp(), "type": event_type, } if step_id is not None: _validate_step_id(step_id) event["step_id"] = step_id if result_code is not None: event["result_code"] = _safe_code(result_code) record["state"]["next_event_sequence"] += 1 record["state"]["events"].append(event) record["state"]["events"] = record["state"]["events"][-MAX_EVENTS:] def _ensure_start_recovery_byte_capacity( record: dict[str, Any], step_id: str, *, mutating: bool ) -> None: """Prove that every required completion/recovery write still fits.""" started = deepcopy(record) _project_persist(started) projections = [started] succeeded = _project_finish(started, step_id, succeeded=True) failed = _project_finish(started, step_id, succeeded=False) projections.extend((succeeded, failed, _project_retry(failed, step_id))) interrupted = _project_resume(started) projections.append(interrupted) if mutating: projections.extend( _project_reconcile(interrupted, step_id, outcome=outcome) for outcome in ("effect_absent", "effect_succeeded") ) else: projections.append(_project_retry(interrupted, step_id)) _ensure_projected_records_fit( projections, message=( "Release run has insufficient serialized recovery capacity for a " "new attempt; create a new run." ), ) def _ensure_unresolved_recovery_byte_capacity( record: dict[str, Any], step_id: str ) -> None: """Accept an advisory unresolved write only if a terminal write remains.""" unresolved = _project_reconcile(record, step_id, outcome="unresolved") projections = [unresolved] projections.extend( _project_reconcile(unresolved, step_id, outcome=outcome) for outcome in ("effect_absent", "effect_succeeded") ) _ensure_projected_records_fit( projections, message=( "The unresolved outcome would consume capacity reserved for a " "terminal reconciliation." ), ) def _project_finish( record: dict[str, Any], step_id: str, *, succeeded: bool ) -> dict[str, Any]: projected = deepcopy(record) step, _plan_step = _step_pair(projected, step_id) attempt = _attempt_for_fingerprint( projected, step["attempt_fingerprint"], step_id=step_id ) target_state = "succeeded" if succeeded else "failed" step.update( { "state": target_state, "finished_at": PROJECTION_TIMESTAMP, "result_code": PROJECTION_RESULT_CODE, "result_receipt": _projection_receipt() if succeeded else None, } ) attempt["outcome"] = target_state attempt["result_code"] = PROJECTION_RESULT_CODE _append_event( projected, event_type="step_succeeded" if succeeded else "step_failed", step_id=step_id, result_code=PROJECTION_RESULT_CODE, at=PROJECTION_TIMESTAMP, ) _project_persist(projected) return projected def _projection_receipt() -> dict[str, Any]: return { "kind": "catalog_publication", "candidate_id": "candidate-" + "c" * 32, "catalog_sha256": "c" * 64, "keyring_sha256": "c" * 64, "publication_commit_sha": "c" * 64, "publication_tag_object_sha": "c" * 64, "publication_tag_commit_sha": "c" * 64, "branch": "b" * 255, "tag_name": "t" * 160, "remote": "origin", "remote_sha256": "c" * 64, } def _project_resume(record: dict[str, Any]) -> dict[str, Any]: projected = deepcopy(record) running_steps = [ step for step in projected["state"]["steps"] if step["state"] == "running" ] for step in running_steps: attempt = _attempt_for_fingerprint( projected, step["attempt_fingerprint"], step_id=step["id"] ) attempt["outcome"] = "interrupted" attempt["result_code"] = "process_interrupted" step["state"] = "interrupted" step["finished_at"] = PROJECTION_TIMESTAMP step["result_code"] = "process_interrupted" _append_event( projected, event_type="step_interrupted", step_id=step["id"], result_code="process_interrupted", at=PROJECTION_TIMESTAMP, ) _remember_request( projected, "0" * 64, "resume", None, None, attempt_fingerprint=running_steps[0]["attempt_fingerprint"], ) _append_event(projected, event_type="run_resumed", at=PROJECTION_TIMESTAMP) _project_persist(projected) return projected def _project_retry(record: dict[str, Any], step_id: str) -> dict[str, Any]: projected = deepcopy(record) step, _plan_step = _step_pair(projected, step_id) attempt_fingerprint = step["attempt_fingerprint"] retry_reason = ( "known_failure" if step["state"] == "failed" else "read_only_interruption" ) step.update( { "state": "pending", "attempt_fingerprint": None, "started_at": None, "finished_at": None, "result_code": None, "result_receipt": None, } ) _remember_request( projected, "1" * 64, "retry", step_id, None, attempt_fingerprint=attempt_fingerprint, ) _append_event( projected, event_type="step_retry_requested", step_id=step_id, result_code=retry_reason, at=PROJECTION_TIMESTAMP, ) _project_persist(projected) return projected def _project_reconcile( record: dict[str, Any], step_id: str, *, outcome: str ) -> dict[str, Any]: projected = deepcopy(record) step, _plan_step = _step_pair(projected, step_id) attempt_fingerprint = step["attempt_fingerprint"] if outcome == "effect_absent": step.update( { "state": "pending", "attempt_fingerprint": None, "started_at": None, "finished_at": None, "result_code": None, "result_receipt": None, } ) elif outcome == "effect_succeeded": step.update( { "state": "succeeded", "finished_at": PROJECTION_TIMESTAMP, "result_code": "reconciled_effect_succeeded", "result_receipt": _projection_receipt(), } ) _remember_request( projected, "2" * 64 if outcome == "unresolved" else "3" * 64, "reconcile", step_id, outcome, attempt_fingerprint=attempt_fingerprint, ) _append_event( projected, event_type="step_reconciled", step_id=step_id, result_code=outcome, at=PROJECTION_TIMESTAMP, ) _project_persist(projected) return projected def _project_persist(record: dict[str, Any]) -> None: record["updated_at"] = PROJECTION_TIMESTAMP _refresh_status(record) record["record_digest"] = "f" * 64 def _ensure_projected_records_fit( records: list[dict[str, Any]], *, message: str ) -> None: if any( len(_canonical_json(projected)) + 1 > MAX_RECORD_BYTES for projected in records ): raise ReleaseRunConflict(message) def _ensure_command_capacity( record: dict[str, Any], *, reserve_after: int = 0 ) -> None: if ( len(record["state"]["processed_requests"]) + 1 + reserve_after > MAX_COMMANDS ): raise ReleaseRunConflict( "Release run command history reached its safety limit; create a new run." ) def _ensure_start_recovery_capacity(record: dict[str, Any]) -> None: if ( len(record["state"]["processed_requests"]) + START_RECOVERY_RESERVE > MAX_COMMANDS ): raise ReleaseRunConflict( "Release run has insufficient reserved recovery capacity for a new attempt; create a new run." ) def _ensure_attempt_capacity(record: dict[str, Any]) -> None: if len(record["state"]["processed_attempts"]) >= MAX_COMMANDS: raise ReleaseRunConflict( "Release run attempt history reached its safety limit; create a new run." ) def _remember_request( record: dict[str, Any], fingerprint: str, operation: str, step_id: str | None, parameter: str | None, *, attempt_fingerprint: str | None, ) -> None: _ensure_command_capacity(record) record["state"]["processed_requests"].append( { "fingerprint": fingerprint, "operation": operation, "step_id": step_id, "parameter": parameter, "attempt_fingerprint": attempt_fingerprint, } ) def _request_seen( record: dict[str, Any], fingerprint: str, operation: str, step_id: str | None, parameter: str | None, ) -> bool: matches = [ request for request in record["state"]["processed_requests"] if request["fingerprint"] == fingerprint ] if not matches: return False if any( request["operation"] == operation and request["step_id"] == step_id and request["parameter"] == parameter for request in matches ): return True raise ReleaseRunConflict( "The request identifier was already used for a different run-state action." ) def _find_attempt( record: dict[str, Any], fingerprint: str ) -> dict[str, Any] | None: return next( ( attempt for attempt in record["state"]["processed_attempts"] if attempt["fingerprint"] == fingerprint ), None, ) def _attempt_for_fingerprint( record: dict[str, Any], fingerprint: str | None, *, step_id: str ) -> dict[str, Any]: if fingerprint is None: raise ReleaseRunCorrupt( "Running release step has no persisted attempt identifier." ) attempt = _find_attempt(record, fingerprint) if attempt is None or attempt["step_id"] != step_id: raise ReleaseRunCorrupt( "Running release step has no matching persisted attempt." ) return attempt def _refresh_status(record: dict[str, Any]) -> None: record["state"]["status"] = _status(record) def _status(record: dict[str, Any]) -> str: if record["immutable"]["plan"].get("status") == "blocked": return "blocked" states = [step["state"] for step in record["state"]["steps"]] if states and all(state == "succeeded" for state in states): return "completed" if "running" in states: return "running" if any(state in {"failed", "interrupted"} for state in states): return "attention" if states: first_pending = next( step for step, mutable in zip( record["immutable"]["plan"]["dry_run_steps"], record["state"]["steps"], strict=True, ) if mutable["state"] == "pending" ) if first_pending.get("status", "planned") != "planned": return "blocked" return "planned" def _immutable_digest(immutable: dict[str, Any]) -> str: return hashlib.sha256( _canonical_json( {"input": immutable.get("input"), "plan": immutable.get("plan")} ) ).hexdigest() def _record_digest(record: dict[str, Any]) -> str: return hashlib.sha256( _canonical_json( {key: value for key, value in record.items() if key != "record_digest"} ) ).hexdigest() def _creation_run_id(request_fingerprint: str) -> str: if not _is_sha256(request_fingerprint): raise ReleaseRunConflict("Run creation request identifier is invalid.") return f"rr-request-{request_fingerprint}" def _encode_list_cursor( key: tuple[int, str, str, str], *, workspace_fingerprint: str ) -> str: payload = { "v": 1, "workspace": workspace_fingerprint, "key": list(key), } encoded = base64.urlsafe_b64encode(_canonical_json(payload)).decode("ascii") return encoded.rstrip("=") def _decode_list_cursor( cursor: str | None, *, expected_workspace_fingerprint: str ) -> tuple[int, str, str, str] | None: if cursor is None or cursor == "": return None if not isinstance(cursor, str) or len(cursor) > MAX_CURSOR_BYTES: raise ReleaseRunConflict("Release run list cursor is invalid.") try: padding = "=" * (-len(cursor) % 4) encoded = base64.b64decode( cursor + padding, altchars=b"-_", validate=True, ) payload = json.loads(encoded.decode("utf-8")) if not isinstance(payload, dict): raise ValueError key = payload.get("key") if ( set(payload) != {"v", "workspace", "key"} or payload.get("v") != 1 or payload.get("workspace") != expected_workspace_fingerprint or not isinstance(key, list) or len(key) != 4 or type(key[0]) is not int or key[0] not in {0, 1} or not all(isinstance(item, str) for item in key[1:]) or not RUN_ID_PATTERN.fullmatch(key[3]) ): raise ValueError if key[0] == 1: if not _is_timestamp(key[1]) or not _is_timestamp(key[2]): raise ValueError elif key[1] or key[2]: raise ValueError except ( binascii.Error, UnicodeDecodeError, ValueError, json.JSONDecodeError, ) as exc: raise ReleaseRunConflict("Release run list cursor is invalid.") from exc return key[0], key[1], key[2], key[3] def _validate_create_replay( record: dict[str, Any], *, request_fingerprint: str, input_snapshot: dict[str, Any], ) -> None: if record["create_request_fingerprint"] != request_fingerprint: raise ReleaseRunConflict( "The run creation identifier is already mapped to another record." ) if record["immutable"]["input"] != input_snapshot: raise ReleaseRunConflict( "The run creation identifier was already used with different inputs." ) def _request_fingerprint(request_id: str) -> str: if not isinstance(request_id, str) or not REQUEST_ID_PATTERN.fullmatch(request_id): raise ReleaseRunConflict("Run-state request identifier is invalid.") return hashlib.sha256(request_id.encode("utf-8")).hexdigest() def _validate_step_id(step_id: Any) -> None: if not isinstance(step_id, str) or not STEP_ID_PATTERN.fullmatch(step_id): raise ReleaseRunConflict("Release plan step identifier is invalid.") def _safe_code(value: str) -> str: if ( not isinstance(value, str) or len(value) > 64 or not re.fullmatch(r"[a-z][a-z0-9_]*", value) ): raise ReleaseRunConflict("Release result code is invalid.") return value def _validated_result_receipt( value: dict[str, Any] | None, ) -> dict[str, Any] | None: if value is None: return None if not isinstance(value, dict) or value.get("kind") not in RECEIPT_KINDS: raise ReleaseRunConflict("Release result receipt is invalid.") if value["kind"] == "catalog_candidate": if set(value) != {"kind", "candidate_id", "catalog_sha256"}: raise ReleaseRunConflict("Release result receipt is invalid.") candidate_id = value.get("candidate_id") digest = value.get("catalog_sha256") if ( not isinstance(candidate_id, str) or not RECEIPT_ID_PATTERN.fullmatch(candidate_id) or not _is_sha256(digest) ): raise ReleaseRunConflict("Release result receipt identity is invalid.") return dict(value) if value["kind"] == "catalog_publication": if set(value) != { "kind", "candidate_id", "catalog_sha256", "keyring_sha256", "publication_commit_sha", "publication_tag_object_sha", "publication_tag_commit_sha", "branch", "tag_name", "remote", "remote_sha256", }: raise ReleaseRunConflict("Catalog publication receipt is invalid.") candidate_id = value.get("candidate_id") commit = value.get("publication_commit_sha") tagged_commit = value.get("publication_tag_commit_sha") if ( not isinstance(candidate_id, str) or not RECEIPT_ID_PATTERN.fullmatch(candidate_id) or not _is_sha256(value.get("catalog_sha256")) or not _is_sha256(value.get("keyring_sha256")) or not isinstance(commit, str) or not GIT_OBJECT_PATTERN.fullmatch(commit) or not isinstance(value.get("publication_tag_object_sha"), str) or not GIT_OBJECT_PATTERN.fullmatch( value["publication_tag_object_sha"] ) or not isinstance(tagged_commit, str) or not GIT_OBJECT_PATTERN.fullmatch(tagged_commit) or tagged_commit != commit or not isinstance(value.get("branch"), str) or not value["branch"] or len(value["branch"]) > 255 or not isinstance(value.get("tag_name"), str) or not value["tag_name"] or len(value["tag_name"]) > 160 or value.get("remote") != "origin" or not _is_sha256(value.get("remote_sha256")) ): raise ReleaseRunConflict( "Catalog publication receipt identity is invalid." ) return dict(value) if set(value) != { "kind", "repo", "head", "branch", "remote", "remote_sha256", "worktree_clean", "target_tag", "tag_object", }: raise ReleaseRunConflict("Repository result receipt is invalid.") if ( not isinstance(value.get("repo"), str) or not REPOSITORY_PATTERN.fullmatch(value["repo"]) or not isinstance(value.get("head"), str) or not GIT_OBJECT_PATTERN.fullmatch(value["head"]) or not isinstance(value.get("branch"), str) or not value["branch"] or len(value["branch"]) > 255 or value.get("remote") != "origin" or not _is_sha256(value.get("remote_sha256")) or type(value.get("worktree_clean")) is not bool or not isinstance(value.get("target_tag"), str) or len(value["target_tag"]) > 160 or ( value.get("tag_object") is not None and ( not isinstance(value["tag_object"], str) or not GIT_OBJECT_PATTERN.fullmatch(value["tag_object"]) ) ) ): raise ReleaseRunConflict("Repository result receipt identity is invalid.") return dict(value) def _upgrade_record(record: dict[str, Any]) -> dict[str, Any]: upgraded = deepcopy(record) if upgraded.get("schema_version") == 3: for step in upgraded["state"]["steps"]: step["result_receipt"] = None upgraded["schema_version"] = SCHEMA_VERSION upgraded["record_digest"] = _record_digest(upgraded) _validate_record(upgraded) return upgraded def _is_sha256(value: Any) -> bool: return isinstance(value, str) and bool(re.fullmatch(r"[0-9a-f]{64}", value)) def _is_timestamp(value: Any) -> bool: if not isinstance(value, str) or not re.fullmatch( r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z", value ): return False try: datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") except ValueError: return False return True def _validate_step_state_fields(step: dict[str, Any]) -> None: state = step["state"] attempt_count = step["attempt_count"] fingerprint = step["attempt_fingerprint"] started_at = step["started_at"] finished_at = step["finished_at"] result_code = step["result_code"] result_receipt = step.get("result_receipt") if started_at is not None and not _is_timestamp(started_at): raise ValueError if finished_at is not None and not _is_timestamp(finished_at): raise ValueError if started_at is not None and finished_at is not None and finished_at < started_at: raise ValueError if state == "pending": if any( value is not None for value in ( fingerprint, started_at, finished_at, result_code, result_receipt, ) ): raise ValueError return if attempt_count < 1 or fingerprint is None or started_at is None: raise ValueError if state == "running": if ( finished_at is not None or result_code is not None or result_receipt is not None ): raise ValueError return if finished_at is None or result_code is None: raise ValueError if state != "succeeded" and result_receipt is not None: raise ValueError if state == "interrupted" and result_code not in { "process_interrupted", "executor_outcome_unknown", }: raise ValueError def _validate_event_fields(event: dict[str, Any]) -> None: event_type = event["type"] has_step = "step_id" in event has_result = "result_code" in event if event_type in {"run_created", "run_resumed"}: if has_step or has_result: raise ValueError return if not has_step: raise ValueError if event_type == "step_started": if has_result: raise ValueError return if not has_result: raise ValueError if event_type == "step_reconciled" and event["result_code"] not in ( RECONCILIATION_OUTCOMES ): raise ValueError if ( event_type == "step_interrupted" and event["result_code"] not in {"process_interrupted", "executor_outcome_unknown"} ): raise ValueError def _create_private_directory_tree(path: Path) -> None: missing: list[Path] = [] current = path while not current.exists(): missing.append(current) if current == current.parent: break current = current.parent _validate_authority_components(current) for directory in reversed(missing): try: os.mkdir(directory, mode=0o700) except FileExistsError: pass except OSError as exc: raise ReleaseRunCorrupt( "Release run storage directory could not be created safely." ) from exc _reject_symlink_components(directory) try: metadata = directory.lstat() if not stat.S_ISDIR(metadata.st_mode): raise ReleaseRunCorrupt( "Release run storage path is not a directory." ) if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid(): raise ReleaseRunCorrupt( "Release run storage is not owned by the console operator." ) directory.chmod(0o700) except OSError as exc: raise ReleaseRunCorrupt( "Release run storage permissions could not be secured." ) from exc if directory.stat().st_mode & 0o077: raise ReleaseRunCorrupt( "Release run storage permissions are broader than 0700." ) _fsync_directory(directory) _fsync_directory(directory.parent) _validate_authority_components(path.parent) def _validate_authority_components(path: Path) -> None: home = Path(os.path.abspath(os.fspath(Path.home()))) if path == home or path.is_relative_to(home): current = home parts = path.relative_to(home).parts candidates = [current] else: current = Path(path.anchor) parts = path.parts[1:] candidates = [] for part in parts: current /= part candidates.append(current) for current in candidates: try: metadata = current.lstat() except OSError as exc: raise ReleaseRunCorrupt( "Release run storage ancestry could not be validated." ) from exc if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): raise ReleaseRunCorrupt( "Release run storage ancestry is not a safe directory path." ) if hasattr(os, "geteuid") and metadata.st_uid not in {0, os.geteuid()}: raise ReleaseRunCorrupt( "Release run storage ancestry has an untrusted owner." ) writable_by_others = bool(metadata.st_mode & 0o022) sticky = bool(metadata.st_mode & stat.S_ISVTX) if writable_by_others and not sticky: raise ReleaseRunCorrupt( "Release run storage ancestry is writable by another principal." ) def _fsync_directory(path: Path) -> None: flags = os.O_RDONLY if hasattr(os, "O_DIRECTORY"): flags |= os.O_DIRECTORY try: descriptor = os.open(path, flags) try: os.fsync(descriptor) finally: os.close(descriptor) except OSError as exc: raise ReleaseRunCorrupt( "Release run storage metadata could not be persisted." ) from exc def _reject_symlink_components(path: Path) -> None: current = Path(path.anchor) for part in path.parts[1:]: current /= part if current.is_symlink(): raise ReleaseRunCorrupt( "Release run storage must not contain symbolic-link components." ) def _canonical_json(value: Any) -> bytes: try: return json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False, ).encode("utf-8") except (TypeError, ValueError) as exc: raise ReleaseRunConflict("Release run contains non-JSON data.") from exc def _timestamp() -> str: return ( datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") )