Release portable deterministic core 1.1.0: frozen formats, offline recovery and regression checks
ecad52f verified | from __future__ import annotations | |
| from dataclasses import dataclass | |
| from enum import Enum | |
| from .engine import Verdict, canonical_receipt_sha256 | |
| class ProjectState(str, Enum): | |
| ACTIVE = "active" | |
| FROZEN = "frozen" | |
| class LifecycleResult: | |
| state: ProjectState | |
| verdict: Verdict | |
| reason: str | |
| receipt: dict | |
| def _non_empty(value: str, name: str) -> str: | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ValueError(f"{name} must be non-empty") | |
| return value.strip() | |
| def _validate_bool(value: bool, name: str) -> None: | |
| if type(value) is not bool: | |
| raise TypeError(f"{name} must be bool") | |
| def _receipt( | |
| *, | |
| project_id: str, | |
| transition: str, | |
| from_state: ProjectState, | |
| to_state: ProjectState, | |
| checkpoint_digest: str | None, | |
| checkpoint_verified: bool | None, | |
| human_approved: bool | None, | |
| policy_version: str, | |
| reason: str, | |
| reopen_reason: str | None = None, | |
| ) -> dict: | |
| receipt = { | |
| "subsystem": "project_lifecycle", | |
| "policy_version": policy_version, | |
| "project_id": project_id, | |
| "transition": transition, | |
| "from_state": from_state.value, | |
| "to_state": to_state.value, | |
| "checkpoint_digest": checkpoint_digest, | |
| "checkpoint_verified": checkpoint_verified, | |
| "human_approved": human_approved, | |
| "reason": reason, | |
| "reopen_reason": reopen_reason, | |
| } | |
| receipt["receipt_sha256"] = canonical_receipt_sha256(receipt) | |
| return receipt | |
| def freeze_project( | |
| *, | |
| project_id: str, | |
| current_state: ProjectState, | |
| checkpoint_digest: str, | |
| checkpoint_verified: bool, | |
| human_approved: bool, | |
| policy_version: str, | |
| current_checkpoint_digest: str | None = None, | |
| ) -> LifecycleResult: | |
| project_id = _non_empty(project_id, "project_id") | |
| checkpoint_digest = _non_empty(checkpoint_digest, "checkpoint_digest") | |
| policy_version = _non_empty(policy_version, "policy_version") | |
| _validate_bool(checkpoint_verified, "checkpoint_verified") | |
| _validate_bool(human_approved, "human_approved") | |
| if current_state is ProjectState.FROZEN: | |
| if current_checkpoint_digest == checkpoint_digest: | |
| reason = "already_frozen_same_checkpoint" | |
| return LifecycleResult( | |
| state=ProjectState.FROZEN, | |
| verdict=Verdict.APPROVE, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="freeze", | |
| from_state=ProjectState.FROZEN, | |
| to_state=ProjectState.FROZEN, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=checkpoint_verified, | |
| human_approved=human_approved, | |
| policy_version=policy_version, | |
| reason=reason, | |
| ), | |
| ) | |
| reason = "frozen_checkpoint_mismatch" | |
| return LifecycleResult( | |
| state=ProjectState.FROZEN, | |
| verdict=Verdict.BLOCK, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="freeze", | |
| from_state=ProjectState.FROZEN, | |
| to_state=ProjectState.FROZEN, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=checkpoint_verified, | |
| human_approved=human_approved, | |
| policy_version=policy_version, | |
| reason=reason, | |
| ), | |
| ) | |
| if not checkpoint_verified: | |
| reason = "checkpoint_not_verified" | |
| return LifecycleResult( | |
| state=ProjectState.ACTIVE, | |
| verdict=Verdict.BLOCK, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="freeze", | |
| from_state=ProjectState.ACTIVE, | |
| to_state=ProjectState.ACTIVE, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=False, | |
| human_approved=human_approved, | |
| policy_version=policy_version, | |
| reason=reason, | |
| ), | |
| ) | |
| if not human_approved: | |
| reason = "human_approval_required_for_freeze" | |
| return LifecycleResult( | |
| state=ProjectState.ACTIVE, | |
| verdict=Verdict.ESCALATE, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="freeze", | |
| from_state=ProjectState.ACTIVE, | |
| to_state=ProjectState.ACTIVE, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=True, | |
| human_approved=False, | |
| policy_version=policy_version, | |
| reason=reason, | |
| ), | |
| ) | |
| reason = "verified_checkpoint_frozen" | |
| return LifecycleResult( | |
| state=ProjectState.FROZEN, | |
| verdict=Verdict.APPROVE, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="freeze", | |
| from_state=ProjectState.ACTIVE, | |
| to_state=ProjectState.FROZEN, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=True, | |
| human_approved=True, | |
| policy_version=policy_version, | |
| reason=reason, | |
| ), | |
| ) | |
| def authorize_mutation( | |
| *, | |
| project_id: str, | |
| current_state: ProjectState, | |
| policy_version: str, | |
| ) -> LifecycleResult: | |
| project_id = _non_empty(project_id, "project_id") | |
| policy_version = _non_empty(policy_version, "policy_version") | |
| if current_state is ProjectState.FROZEN: | |
| reason = "project_frozen_controlled_reopen_required" | |
| verdict = Verdict.BLOCK | |
| else: | |
| reason = "project_active" | |
| verdict = Verdict.APPROVE | |
| return LifecycleResult( | |
| state=current_state, | |
| verdict=verdict, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="mutation_gate", | |
| from_state=current_state, | |
| to_state=current_state, | |
| checkpoint_digest=None, | |
| checkpoint_verified=None, | |
| human_approved=None, | |
| policy_version=policy_version, | |
| reason=reason, | |
| ), | |
| ) | |
| def reopen_project( | |
| *, | |
| project_id: str, | |
| current_state: ProjectState, | |
| checkpoint_digest: str, | |
| reopen_reason: str, | |
| human_approved: bool, | |
| policy_version: str, | |
| ) -> LifecycleResult: | |
| project_id = _non_empty(project_id, "project_id") | |
| checkpoint_digest = _non_empty(checkpoint_digest, "checkpoint_digest") | |
| reopen_reason = _non_empty(reopen_reason, "reopen_reason") | |
| policy_version = _non_empty(policy_version, "policy_version") | |
| _validate_bool(human_approved, "human_approved") | |
| if current_state is not ProjectState.FROZEN: | |
| reason = "invalid_reopen_transition" | |
| return LifecycleResult( | |
| state=current_state, | |
| verdict=Verdict.BLOCK, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="reopen", | |
| from_state=current_state, | |
| to_state=current_state, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=True, | |
| human_approved=human_approved, | |
| policy_version=policy_version, | |
| reason=reason, | |
| reopen_reason=reopen_reason, | |
| ), | |
| ) | |
| if not human_approved: | |
| reason = "human_approval_required_for_reopen" | |
| return LifecycleResult( | |
| state=ProjectState.FROZEN, | |
| verdict=Verdict.ESCALATE, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="reopen", | |
| from_state=ProjectState.FROZEN, | |
| to_state=ProjectState.FROZEN, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=True, | |
| human_approved=False, | |
| policy_version=policy_version, | |
| reason=reason, | |
| reopen_reason=reopen_reason, | |
| ), | |
| ) | |
| reason = "controlled_reopen_approved" | |
| return LifecycleResult( | |
| state=ProjectState.ACTIVE, | |
| verdict=Verdict.APPROVE, | |
| reason=reason, | |
| receipt=_receipt( | |
| project_id=project_id, | |
| transition="reopen", | |
| from_state=ProjectState.FROZEN, | |
| to_state=ProjectState.ACTIVE, | |
| checkpoint_digest=checkpoint_digest, | |
| checkpoint_verified=True, | |
| human_approved=True, | |
| policy_version=policy_version, | |
| reason=reason, | |
| reopen_reason=reopen_reason, | |
| ), | |
| ) |