168 lines
8.4 KiB
Python
168 lines
8.4 KiB
Python
"""ChatGPT GitOps Bridge v2 contract and governed release operations."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Mapping
|
|
|
|
|
|
BRIDGE_SCHEMA_VERSION = "aios.chatgpt-gitops-bridge.v2"
|
|
BRIDGE_OPERATIONS = (
|
|
"getRuntimeStatus",
|
|
"getGovernedWriteCapabilities",
|
|
"createGovernedTask",
|
|
"acquireWriterLease",
|
|
"commitFileGoverned",
|
|
"commitFilesGoverned",
|
|
"getGitOpsJob",
|
|
"resumeGitOpsJob",
|
|
"reconcileGitOpsJob",
|
|
"createTagApprovalRequest",
|
|
"requestOrCreateTag",
|
|
"getTagPublicationStatus",
|
|
"buildProductionArtifact",
|
|
"executeApprovedProductionRelease",
|
|
"getReleaseConsistency",
|
|
"runReleaseAcceptance",
|
|
"getControlPlaneSnapshot",
|
|
)
|
|
|
|
OPERATION_REQUIRED_FIELDS = {
|
|
"createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
|
|
"acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
|
|
"commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
|
|
"commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
|
|
"getGitOpsJob": ("job_id",),
|
|
"resumeGitOpsJob": ("job_id", "expected_revision"),
|
|
"reconcileGitOpsJob": ("job_id", "expected_revision"),
|
|
"createTagApprovalRequest": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id", "owner_approval", "execute"),
|
|
"requestOrCreateTag": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id", "execute"),
|
|
"getTagPublicationStatus": ("version",),
|
|
"buildProductionArtifact": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id"),
|
|
"executeApprovedProductionRelease": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id", "execute"),
|
|
"getReleaseConsistency": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id"),
|
|
"runReleaseAcceptance": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id"),
|
|
"getControlPlaneSnapshot": (),
|
|
}
|
|
|
|
|
|
class RuntimeDriftError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class BridgeV2Error(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ControlPlaneReleaseManifest:
|
|
control_plane_release_version: str
|
|
control_plane_target_oid: str
|
|
artifact_sha256: str
|
|
runtime_path: Path
|
|
bridge_schema_version: str
|
|
repository_identity: str
|
|
authoritative_git_dir: Path
|
|
deployment_state: str
|
|
|
|
def as_dict(self) -> dict[str, object]:
|
|
return {
|
|
"control_plane_release_version": self.control_plane_release_version,
|
|
"control_plane_target_oid": self.control_plane_target_oid,
|
|
"artifact_sha256": self.artifact_sha256,
|
|
"runtime_path": str(self.runtime_path),
|
|
"bridge_schema_version": self.bridge_schema_version,
|
|
"repository_identity": self.repository_identity,
|
|
"authoritative_git_dir": str(self.authoritative_git_dir),
|
|
"deployment_state": self.deployment_state,
|
|
}
|
|
|
|
def validate(self, artifact: Path, *, observed_oid: str, approved_root: Path = Path("/opt/aios/gitops-control-plane")) -> dict[str, object]:
|
|
try:
|
|
runtime = self.runtime_path.resolve(strict=True)
|
|
root = approved_root.resolve(strict=True)
|
|
artifact_path = artifact.resolve(strict=True)
|
|
runtime.relative_to(root)
|
|
artifact_path.relative_to(runtime)
|
|
except (OSError, ValueError):
|
|
raise RuntimeDriftError("CONTROL_PLANE_RUNTIME_DRIFT") from None
|
|
mutable_markers = {"workspace", "workspaces", "worktree", "worktrees"}
|
|
if mutable_markers.intersection(part.lower() for part in runtime.parts):
|
|
raise RuntimeDriftError("CONTROL_PLANE_RUNTIME_DRIFT")
|
|
valid = (
|
|
len(self.control_plane_target_oid) == 40
|
|
and observed_oid == self.control_plane_target_oid
|
|
and len(self.artifact_sha256) == 64
|
|
and hashlib.sha256(artifact_path.read_bytes()).hexdigest() == self.artifact_sha256
|
|
and self.bridge_schema_version == BRIDGE_SCHEMA_VERSION
|
|
and self.repository_identity == "aios/aios"
|
|
and self.authoritative_git_dir == Path("/data/aios-gitops/repo.git")
|
|
and self.authoritative_git_dir.is_absolute()
|
|
and self.deployment_state == "STAGING"
|
|
)
|
|
if not valid:
|
|
raise RuntimeDriftError("CONTROL_PLANE_RUNTIME_DRIFT")
|
|
return {**self.as_dict(), "drift_status": "PASS"}
|
|
|
|
|
|
class BridgeV2Adapter:
|
|
"""Thin operation adapter; all mutating handlers are injected seams."""
|
|
|
|
def __init__(self, *, runtime_status: Mapping[str, Any], handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None) -> None:
|
|
self.runtime_status = dict(runtime_status)
|
|
self.handlers = dict(handlers or {})
|
|
|
|
def getRuntimeStatus(self) -> dict[str, Any]:
|
|
return {"bridge_schema_version": BRIDGE_SCHEMA_VERSION, **self.runtime_status}
|
|
|
|
def getGovernedWriteCapabilities(self) -> dict[str, Any]:
|
|
return {
|
|
"bridge_schema_version": BRIDGE_SCHEMA_VERSION,
|
|
"legacy_writes_enabled": False,
|
|
"governed_writes_enabled": True,
|
|
"operations": list(BRIDGE_OPERATIONS),
|
|
"force_push": False,
|
|
"main_write": False,
|
|
"tag_write": False,
|
|
"baseline_write": False,
|
|
"production_write": False,
|
|
"approved_release_operations": True,
|
|
"approved_release_scope": {
|
|
"version": "V2R2C1",
|
|
"target_oid": "998d30304787e31e5ae607c0bd86b5edb2b0e5ad",
|
|
"evidence_closure_oid": "a8b906d1bc24088c19647e98de6a679268f6d0a2",
|
|
"approval_plan_id": "sha256:c0df52008ae7f64cec5ee2d28947aa7b892066c670ab8295e4549140b1a7500e",
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def legacy_write_disabled() -> dict[str, object]:
|
|
return {"error": "LEGACY_WRITE_DISABLED", "replacement": "createGovernedTask", "governed_writes_available": True}
|
|
|
|
def _call(self, operation: str, request: Mapping[str, Any]) -> Any:
|
|
required = OPERATION_REQUIRED_FIELDS[operation]
|
|
if any(key not in request for key in required):
|
|
raise BridgeV2Error("BRIDGE_REQUEST_FIELDS_MISSING")
|
|
handler = self.handlers.get(operation)
|
|
if handler is None:
|
|
raise BridgeV2Error("GOVERNED_HANDLER_UNAVAILABLE")
|
|
return handler(dict(request))
|
|
|
|
def createGovernedTask(self, request: Mapping[str, Any]) -> Any: return self._call("createGovernedTask", request)
|
|
def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
|
|
def commitFileGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFileGoverned", request)
|
|
def commitFilesGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFilesGoverned", request)
|
|
def getGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("getGitOpsJob", request)
|
|
def resumeGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("resumeGitOpsJob", request)
|
|
def reconcileGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("reconcileGitOpsJob", request)
|
|
def createTagApprovalRequest(self, request: Mapping[str, Any]) -> Any: return self._call("createTagApprovalRequest", request)
|
|
def requestOrCreateTag(self, request: Mapping[str, Any]) -> Any: return self._call("requestOrCreateTag", request)
|
|
def getTagPublicationStatus(self, request: Mapping[str, Any]) -> Any: return self._call("getTagPublicationStatus", request)
|
|
def buildProductionArtifact(self, request: Mapping[str, Any]) -> Any: return self._call("buildProductionArtifact", request)
|
|
def executeApprovedProductionRelease(self, request: Mapping[str, Any]) -> Any: return self._call("executeApprovedProductionRelease", request)
|
|
def getReleaseConsistency(self, request: Mapping[str, Any]) -> Any: return self._call("getReleaseConsistency", request)
|
|
def runReleaseAcceptance(self, request: Mapping[str, Any]) -> Any: return self._call("runReleaseAcceptance", request)
|
|
|
|
def getControlPlaneSnapshot(self, request: Mapping[str, Any] | None = None) -> Any: return self._call("getControlPlaneSnapshot", request or {})
|