225 lines
8.5 KiB
Python
225 lines
8.5 KiB
Python
"""Machine-verifiable acceptance contract for dormant V2R1C4 delivery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Any, Mapping
|
|
|
|
from .dormant import (
|
|
DeploymentMode,
|
|
DormantReleaseCandidate,
|
|
DormantReleaseGate,
|
|
ExecutionFlags,
|
|
)
|
|
from .errors import ReleaseBlocked
|
|
from .models import is_git_oid, is_sha256, sha256_json
|
|
|
|
|
|
class AcceptanceStatus(str, Enum):
|
|
READY = "READY"
|
|
BLOCKED = "BLOCKED"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DormantAcceptanceInput:
|
|
"""Immutable evidence supplied by the governed execution agent.
|
|
|
|
Git ancestry and four-way reference checks are performed by the execution
|
|
agent and represented as explicit booleans. This package never infers remote
|
|
consistency from a local repository alone.
|
|
"""
|
|
|
|
version: str
|
|
pull_request: int
|
|
candidate_oid: str
|
|
candidate_tree_oid: str
|
|
integration_oid: str
|
|
integration_tree_oid: str
|
|
artifact_sha256: str
|
|
manifest_sha256: str
|
|
package_closure_sha256: str
|
|
evidence_closure_oid: str
|
|
rollback_artifact_sha256: str
|
|
production_config_sha256: str
|
|
test_evidence_sha256: str
|
|
tests_passed: int
|
|
tests_failed: int
|
|
audited_skips: int
|
|
candidate_four_way_ref_closure: bool
|
|
integration_four_way_ref_closure: bool
|
|
candidate_reachable_from_integration: bool
|
|
worktree_clean: bool
|
|
gitops_writes_enabled: bool
|
|
task_plane_execution_enabled: bool
|
|
silent_release_execution_enabled: bool
|
|
protected_dispatch_transport: str = "sandbox"
|
|
native_openclaw_enabled: bool = False
|
|
native_codex_enabled: bool = False
|
|
automatic_release_enabled: bool = False
|
|
production_side_effects: bool = False
|
|
activation_deferred: bool = True
|
|
deployment_mode: str = DeploymentMode.DORMANT_SHADOW_ONLY.value
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.version or self.pull_request < 1:
|
|
raise ValueError("version and a positive pull request number are required")
|
|
for name in (
|
|
"candidate_oid",
|
|
"candidate_tree_oid",
|
|
"integration_oid",
|
|
"integration_tree_oid",
|
|
"evidence_closure_oid",
|
|
):
|
|
if not is_git_oid(getattr(self, name)):
|
|
raise ValueError(f"{name} must be a full lowercase 40-character Git OID")
|
|
for name in (
|
|
"artifact_sha256",
|
|
"manifest_sha256",
|
|
"package_closure_sha256",
|
|
"rollback_artifact_sha256",
|
|
"production_config_sha256",
|
|
"test_evidence_sha256",
|
|
):
|
|
if not is_sha256(getattr(self, name)):
|
|
raise ValueError(f"{name} must be a lowercase SHA-256 digest")
|
|
for name in ("tests_passed", "tests_failed", "audited_skips"):
|
|
value = getattr(self, name)
|
|
if isinstance(value, bool) or value < 0:
|
|
raise ValueError(f"{name} must be a non-negative integer")
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: Mapping[str, Any]) -> "DormantAcceptanceInput":
|
|
allowed = set(cls.__dataclass_fields__)
|
|
unknown = sorted(set(value) - allowed)
|
|
if unknown:
|
|
raise ValueError(f"unknown acceptance fields: {', '.join(unknown)}")
|
|
required = {
|
|
name
|
|
for name, field in cls.__dataclass_fields__.items()
|
|
if field.default is field.default_factory
|
|
}
|
|
missing = sorted(required - set(value))
|
|
if missing:
|
|
raise ValueError(f"missing acceptance fields: {', '.join(missing)}")
|
|
return cls(**dict(value))
|
|
|
|
def release_candidate(self) -> DormantReleaseCandidate:
|
|
return DormantReleaseCandidate(
|
|
version=self.version,
|
|
pull_request=self.pull_request,
|
|
integration_oid=self.integration_oid,
|
|
integration_tree_oid=self.integration_tree_oid,
|
|
artifact_sha256=self.artifact_sha256,
|
|
manifest_sha256=self.manifest_sha256,
|
|
package_closure_sha256=self.package_closure_sha256,
|
|
evidence_closure_oid=self.evidence_closure_oid,
|
|
rollback_artifact_sha256=self.rollback_artifact_sha256,
|
|
production_config_sha256=self.production_config_sha256,
|
|
execution_flags=ExecutionFlags(
|
|
gitops_writes_enabled=self.gitops_writes_enabled,
|
|
task_plane_execution_enabled=self.task_plane_execution_enabled,
|
|
silent_release_execution_enabled=self.silent_release_execution_enabled,
|
|
),
|
|
deployment_mode=DeploymentMode(self.deployment_mode),
|
|
protected_dispatch_transport=self.protected_dispatch_transport,
|
|
native_openclaw_enabled=self.native_openclaw_enabled,
|
|
native_codex_enabled=self.native_codex_enabled,
|
|
automatic_release_enabled=self.automatic_release_enabled,
|
|
production_side_effects=self.production_side_effects,
|
|
activation_deferred=self.activation_deferred,
|
|
)
|
|
|
|
def payload(self) -> Mapping[str, Any]:
|
|
return {
|
|
name: getattr(self, name)
|
|
for name in self.__dataclass_fields__
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DormantAcceptanceResult:
|
|
status: AcceptanceStatus
|
|
acceptance_digest: str
|
|
candidate_digest: str | None
|
|
blocking_code: str | None
|
|
blocking_message: str | None
|
|
evidence: Mapping[str, Any]
|
|
|
|
def payload(self) -> Mapping[str, Any]:
|
|
return {
|
|
"status": self.status.value,
|
|
"acceptance_digest": self.acceptance_digest,
|
|
"candidate_digest": self.candidate_digest,
|
|
"blocking_code": self.blocking_code,
|
|
"blocking_message": self.blocking_message,
|
|
"evidence": dict(self.evidence),
|
|
}
|
|
|
|
|
|
class DormantAcceptanceGate:
|
|
"""Fail-closed pre-merge gate for a dormant release candidate."""
|
|
|
|
def verify(self, evidence: DormantAcceptanceInput) -> DormantReleaseCandidate:
|
|
if evidence.version != "V2R1C4":
|
|
raise ReleaseBlocked("VERSION_NOT_V2R1C4", "this acceptance contract is bound to V2R1C4")
|
|
if evidence.candidate_oid == evidence.integration_oid:
|
|
raise ReleaseBlocked(
|
|
"INTEGRATION_MERGE_REQUIRED",
|
|
"integration must be a distinct governed merge commit",
|
|
)
|
|
if not evidence.candidate_four_way_ref_closure:
|
|
raise ReleaseBlocked(
|
|
"CANDIDATE_REF_CLOSURE_REQUIRED",
|
|
"candidate local, bare, remote and Gitea references are not closed",
|
|
)
|
|
if not evidence.integration_four_way_ref_closure:
|
|
raise ReleaseBlocked(
|
|
"INTEGRATION_REF_CLOSURE_REQUIRED",
|
|
"integration local, bare, remote and Gitea references are not closed",
|
|
)
|
|
if not evidence.candidate_reachable_from_integration:
|
|
raise ReleaseBlocked(
|
|
"CANDIDATE_LINEAGE_REQUIRED",
|
|
"candidate is not reachable from the integration commit",
|
|
)
|
|
if not evidence.worktree_clean:
|
|
raise ReleaseBlocked("WORKTREE_NOT_CLEAN", "acceptance worktree contains uncommitted changes")
|
|
if evidence.tests_failed:
|
|
raise ReleaseBlocked("TEST_FAILURES_PRESENT", "candidate test failures are present")
|
|
if evidence.tests_passed < 1:
|
|
raise ReleaseBlocked("TEST_EVIDENCE_REQUIRED", "at least one passing test is required")
|
|
candidate = evidence.release_candidate()
|
|
DormantReleaseGate().verify_candidate(candidate)
|
|
return candidate
|
|
|
|
def evaluate(self, evidence: DormantAcceptanceInput) -> DormantAcceptanceResult:
|
|
acceptance_digest = sha256_json(evidence.payload())
|
|
try:
|
|
candidate = self.verify(evidence)
|
|
except (ReleaseBlocked, ValueError) as exc:
|
|
return DormantAcceptanceResult(
|
|
status=AcceptanceStatus.BLOCKED,
|
|
acceptance_digest=acceptance_digest,
|
|
candidate_digest=None,
|
|
blocking_code=getattr(exc, "code", "INVALID_ACCEPTANCE_INPUT"),
|
|
blocking_message=str(exc),
|
|
evidence=evidence.payload(),
|
|
)
|
|
return DormantAcceptanceResult(
|
|
status=AcceptanceStatus.READY,
|
|
acceptance_digest=acceptance_digest,
|
|
candidate_digest=candidate.digest(),
|
|
blocking_code=None,
|
|
blocking_message=None,
|
|
evidence=evidence.payload(),
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"AcceptanceStatus",
|
|
"DormantAcceptanceInput",
|
|
"DormantAcceptanceResult",
|
|
"DormantAcceptanceGate",
|
|
]
|