Compare commits

...
Author SHA1 Message Date
AIOS Governed Continuation d781f0b350 fix(actions): project governed task admission endpoints 2026-09-14 02:37:11 +08:00
AIOS Governed Continuation b781129bd5 fix(release): correct governed tag schema indentation 2026-08-09 16:52:05 +08:00
AIOS Governed Continuation 44c08c5aec fix(release): expose exact governed tag request fields 2026-08-09 16:50:49 +08:00
AIOS Governed Continuation 0f74553bd5 fix(release): place Bridge v2 contract paths under OpenAPI paths 2026-08-09 16:49:49 +08:00
AIOS Governed Continuation 062089ead2 test(release): converge Bridge v2 release contract 2026-08-09 16:47:13 +08:00
AIOS Governed Continuation 55217e4b8a fix(release): preserve governed action API routing 2026-08-09 16:34:33 +08:00
AIOS Governed Continuation 96f2d05c48 feat(release): wire governed V2R2C1 production executor 2026-08-09 16:29:43 +08:00
AIOS Governed Continuation 0ade81293e fix(gitops): recover legacy durable jobs safely 2026-08-09 14:52:59 +08:00
AIOS Governed Continuation b81f6fdb85 fix(gitops): recover pending CREATE_NEW leases 2026-08-09 11:53:31 +08:00
AIOS Governed Continuation 3b6636fa7d fix(gitops): wire governed job reconciliation 2026-08-09 11:51:09 +08:00
AIOS Governed Continuation aff7f74f40 fix(gitops): recover safe pre-side-effect continuation retries 2026-08-09 11:46:49 +08:00
AIOS Governed Continuation 141bd5856c fix(gitops): close governed HTTP authoring retry path 2026-08-09 11:43:55 +08:00
AIOS Governed Continuation 88873a07f6 feat(gitops): expose governed Bridge v2 HTTP bootstrap 2026-08-09 11:27:31 +08:00
AIOS Governed Continuation 3f670c3876 test(gitops): prove Bridge v2 governed operation sequence 2026-08-09 03:40:28 +08:00
AIOS Governed Continuation 8fe3b85c1b fix(gitops): reconcile exact old remote after observation 2026-08-09 03:36:39 +08:00
AIOS Governed Continuation 77c6daa76a docs(gitops): complete Bridge v2 engineering closure evidence 2026-08-09 03:34:00 +08:00
AIOS Governed Continuation 35e387edc8 docs(gitops): record Bridge v2 staging restart acceptance 2026-08-09 03:31:35 +08:00
AIOS Governed Continuation aa0c37b8ab feat(gitops): converge ChatGPT Bridge v2 runtime and OpenAPI 2026-08-09 03:29:09 +08:00
AIOS Governed Continuation 6044b551c1 feat(gitops): add governed continuation executor 2026-08-09 03:23:30 +08:00
aios-autonomy-agent 8030601854 feat(openbao): add autonomy runtime v1 candidate 2026-08-09 01:57:42 +08:00
24 changed files with 3372 additions and 10 deletions
@@ -70,6 +70,41 @@ class AuthorityGitExecutor:
raise AuthorityGitError("ANCESTRY_CHECK_FAILED")
return result.returncode == 0
def commit_exists(self, oid: str) -> bool:
_oid(oid, "commit_oid")
result = subprocess.run(["git", f"--git-dir={self.git_dir}", "cat-file", "-e", f"{oid}^{{commit}}"], capture_output=True, check=False)
return result.returncode == 0
def create_task_branch_CAS(self, branch: str, expected_absent: bool, new_oid: str) -> str:
if not expected_absent or not TASK_BRANCH_RE.fullmatch(branch):
raise AuthorityGitError("TASK_BRANCH_CREATE_SCOPE_DENIED")
_oid(new_oid, "new_oid")
if self.read_ref(branch) is not None:
raise AuthorityGitError("AUTHORITY_BRANCH_ALREADY_EXISTS")
try:
self._git(["update-ref", branch, new_oid, "0" * 40])
except AuthorityGitError:
if self.read_ref(branch) is not None:
raise AuthorityGitError("AUTHORITY_BRANCH_ALREADY_EXISTS")
raise
if self.read_ref(branch) != new_oid:
raise AuthorityGitError("AUTHORITY_CAS_READBACK_MISMATCH")
return new_oid
def create_detached_worktree(self, base_oid: str, worktree_name: str, root: Path | None = None) -> Path:
_oid(base_oid, "base_oid")
if not worktree_name or "/" in worktree_name or "\\" in worktree_name or worktree_name in {".", ".."}:
raise AuthorityGitError("WORKTREE_NAME_INVALID")
parent = root or self.worktree_root
if not parent.is_absolute() or parent.is_symlink():
raise AuthorityGitError("WORKTREE_ROOT_INVALID")
path = parent / worktree_name
if path.exists() or path.is_symlink():
raise AuthorityGitError("WORKTREE_ALREADY_EXISTS")
parent.mkdir(parents=True, exist_ok=True)
self._git(["worktree", "add", "--detach", str(path), base_oid])
return path
def advance_task_branch_CAS(self, branch: str, expected_oid: str, new_oid: str) -> str:
if not TASK_BRANCH_RE.fullmatch(branch):
raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED")
+47 -1
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
from __future__ import annotations
import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, uuid
import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, urllib.request, uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
@@ -404,6 +404,51 @@ class H(BaseHTTPRequestHandler):
if not a['ok']: self.sendj({'error':'unauthorized','hint':'Bearer token required'},401); return None
if '*' not in a['scopes'] and scope not in a['scopes']: self.sendj({'error':'forbidden','required_scope':scope,'account':a['account']},403); return None
return a
def bridge_release(self, path, data):
route = {
'/createTagApprovalRequest': '/v2/tag/approval-request',
'/requestOrCreateTag': '/v2/tag/publish',
'/getTagPublicationStatus': '/v2/tag/status',
'/buildProductionArtifact': '/v2/release/artifact',
'/executeApprovedProductionRelease': '/v2/release/execute',
'/getReleaseConsistency': '/v2/release/consistency',
'/runReleaseAcceptance': '/v2/release/acceptance',
}.get(path)
if route is None:
raise ValueError('governed_release_route_invalid')
base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
request = urllib.request.Request(
base + route,
data=json.dumps(data, sort_keys=True, separators=(',', ':')).encode('utf-8'),
headers={'Content-Type': 'application/json'},
method='POST',
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
result = json.loads(response.read().decode('utf-8'))
return self.sendj(result, response.status)
except urllib.error.HTTPError as exc:
try:
body = json.loads(exc.read().decode('utf-8'))
except Exception:
body = {'error': 'bridge_http_error'}
return self.sendj(body, exc.code)
except (urllib.error.URLError, TimeoutError) as exc:
return self.sendj({'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__}, 503)
def do_POST(self):
path = urllib.parse.urlparse(self.path).path
if path in {
'/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
'/buildProductionArtifact', '/executeApprovedProductionRelease',
'/getReleaseConsistency', '/runReleaseAcceptance',
}:
actor = self.need('tag:create')
if not actor:
return
return self.bridge_release(path, self.body())
if reject_non_read_http(self, 'POST', path):
return
return self.sendj({'error': 'not_found', 'path': path}, 404)
def do_GET(self):
u=urllib.parse.urlparse(self.path); path=u.path.rstrip('/') if u.path != '/' else u.path
if path not in READ_ONLY_GET_PATHS: return self.sendj({'error':'not_found','path':path},404)
@@ -905,3 +950,4 @@ def main():
print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
ThreadingHTTPServer((HOST,PORT), H).serve_forever()
if __name__ == '__main__': main()
@@ -0,0 +1,163 @@
"""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",
)
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"),
}
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)
@@ -0,0 +1,509 @@
"""Loopback-only real HTTP gateway for the governed Bridge v2 adapter."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
import base64
import json
import os
import re
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Mapping
from aios_authority_git_executor import AuthorityGitExecutor
from aios_chatgpt_bridge_v2 import BridgeV2Adapter, BridgeV2Error, OPERATION_REQUIRED_FIELDS
from aios_continuation_executor import (
AuthorityGitAdapter,
ContinuationCommit,
ContinuationExecutor,
ContinuationPermissions,
ContinuationRuntimeBinding,
DetachedGitWorkspace,
GiteaContinuationRemote,
ServerSideCredentialProjection,
)
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import GitOpsJobController, JobStore
from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
from governed_production_release import build_executor_config
MAX_BODY = 2 * 1024 * 1024
ROUTES = {
"/v2/runtime": "getRuntimeStatus",
"/v2/capabilities": "getGovernedWriteCapabilities",
"/v2/governed-task": "createGovernedTask",
"/v2/writer-lease": "acquireWriterLease",
"/v2/commit-file": "commitFileGoverned",
"/v2/commit-files": "commitFilesGoverned",
"/v2/job": "getGitOpsJob",
"/v2/job/resume": "resumeGitOpsJob",
"/v2/job/reconcile": "reconcileGitOpsJob",
"/v2/tag/approval-request": "createTagApprovalRequest",
"/v2/tag/publish": "requestOrCreateTag",
"/v2/tag/status": "getTagPublicationStatus",
"/v2/release/artifact": "buildProductionArtifact",
"/v2/release/execute": "executeApprovedProductionRelease",
"/v2/release/consistency": "getReleaseConsistency",
"/v2/release/acceptance": "runReleaseAcceptance",
}
@dataclass(frozen=True, slots=True)
class RuntimeConfig:
target_oid: str
authoritative_git_dir: Path
state_root: Path
v2_job_root: Path
token_path: Path
ssh_key_path: Path
remote_name: str = "gitea-http"
repository: str = "aios/aios"
api_base: str = "https://g.cnnetmkt.top/git/api/v1"
v2_task_id: str = "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809"
v2_branch: str = "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809"
target_version: str = "V2R2C1"
registry_root: Path = Path("/var/lib/aios-gitops/runtime/registry-projection")
baseline_path: Path = Path("/var/lib/aios-gitops-runtime-anchor/baseline/current.json")
health_url: str = "http://127.0.0.1:18927/health"
@dataclass
class _Runtime:
binding: ContinuationRuntimeBinding
controller: GitOpsJobController
executor: ContinuationExecutor
def _safe(value: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "-", value).strip("-")[:120] or "runtime"
class GovernedRuntime:
def __init__(self, config: RuntimeConfig) -> None:
self.config = config
config.state_root.mkdir(parents=True, exist_ok=True)
self.authority = AuthorityGitExecutor(config.authoritative_git_dir, config.state_root / "worktrees")
self.bootstrap = GovernedTaskBootstrap(
self.authority,
config.state_root,
remote_name=config.remote_name,
token_path=config.token_path,
ssh_key_path=config.ssh_key_path,
api_base=config.api_base,
)
self.runtimes: dict[tuple[str, int], _Runtime] = {}
self.controllers: dict[str, GitOpsJobController] = {}
self.release = build_executor_config(config)
askpass = config.state_root / "bridge-v2-askpass.sh"
if not askpass.exists():
askpass.write_text(
'#!/bin/sh\ncase "$1" in *Username*) printf "%s\\n" "$GITEA_MACHINE_ACTOR" ;; *) cat "$CREDENTIAL_FILE" ;; esac\n',
encoding="utf-8",
)
askpass.chmod(0o700)
os.environ.setdefault("GIT_ASKPASS", str(askpass))
os.environ.setdefault("GIT_ASKPASS_REQUIRE", "force")
os.environ.setdefault("GIT_TERMINAL_PROMPT", "0")
os.environ.setdefault("GITEA_MACHINE_ACTOR", "aios-autonomy-agent")
def _job_root(self, task_id: str) -> Path:
if task_id == self.config.v2_task_id:
return self.config.v2_job_root
return self.config.state_root / "jobs" / _safe(task_id)
def _runtime(self, request: Mapping[str, Any], *, create_new: bool = False) -> _Runtime:
key = (str(request["task_id"]), int(request["generation"]))
if key in self.runtimes:
return self.runtimes[key]
if create_new:
binding, controller, executor = self.bootstrap.prepare_runtime(
request, lease_seconds=int(request.get("lease_seconds", 900))
)
else:
request_v2 = WriterLeaseRequestV2(
task_id=str(request["task_id"]),
repository=str(request["repository"]),
branch_mode=str(request["branch_mode"]),
base_oid=str(request["base_oid"]),
expected_branch_oid=str(request["expected_branch_oid"]),
target_branch=str(request["target_branch"]),
generation=int(request["generation"]),
lease_seconds=int(request.get("lease_seconds", 900)),
)
request_v2.validate(AuthorityGitAdapter(self.authority))
workspace_path = self.config.state_root / "worktrees" / _safe(request_v2.task_id)
if workspace_path.exists():
workspace = DetachedGitWorkspace(workspace_path)
if workspace.head_oid() != request_v2.expected_branch_oid or not workspace.is_detached():
raise RuntimeError("WORKTREE_HEAD_MISMATCH")
else:
self.authority.create_detached_worktree(
request_v2.expected_branch_oid,
_safe(request_v2.task_id),
self.config.state_root / "worktrees",
)
workspace = DetachedGitWorkspace(workspace_path)
job_root = self._job_root(request_v2.task_id)
controller = GitOpsJobController(JobStore(job_root, payload_root=job_root / "payloads"))
credential = ServerSideCredentialProjection(self.config.ssh_key_path, self.config.token_path)
remote = GiteaContinuationRemote(
workspace=workspace_path,
credential=credential,
remote_name=self.config.remote_name,
repository=request_v2.repository,
api_base=self.config.api_base,
)
executor = ContinuationExecutor(
controller,
AuthorityGitAdapter(self.authority),
workspace,
remote,
credential,
)
binding = ContinuationRuntimeBinding(
request=request_v2,
workspace_path=workspace_path,
lease_not_before=datetime.now(timezone.utc) - timedelta(seconds=1),
lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=request_v2.lease_seconds),
original_receipt_valid=True,
original_grant_consumed=True,
permissions=ContinuationPermissions(),
)
runtime = _Runtime(binding, controller, executor)
self.runtimes[key] = runtime
self.controllers[str(request["task_id"])] = controller
return runtime
def create_task(self, request: Mapping[str, Any]) -> dict[str, Any]:
return self.bootstrap.create(request)
def acquire_lease(self, request: Mapping[str, Any]) -> dict[str, Any]:
create_new = str(request.get("branch_mode")) == "CREATE_NEW"
runtime = self._runtime(request, create_new=create_new)
request_v2 = runtime.binding.request
runtime.executor.observe_generation(request_v2.target_branch, max(0, request_v2.generation - 1))
return {
"schema": WriterLeaseRequestV2.schema,
"task_id": request_v2.task_id,
"repository": request_v2.repository,
"branch_mode": request_v2.branch_mode,
"base_oid": request_v2.base_oid,
"expected_branch_oid": request_v2.expected_branch_oid,
"target_branch": request_v2.target_branch,
"generation": request_v2.generation,
"lease_seconds": request_v2.lease_seconds,
"create_target_branch": create_new,
"worktree_mode": "detached_at_base_oid" if create_new else "detached_at_expected_head",
"cas_ref_update_required": True,
}
@staticmethod
def _files(request: Mapping[str, Any]) -> dict[str, bytes]:
raw = request["files"] if "files" in request else {request["path"]: request["content_base64"]}
if isinstance(raw, dict):
pairs = raw.items()
elif isinstance(raw, list):
pairs = ((item["path"], item["content_base64"]) for item in raw if isinstance(item, dict))
else:
raise ValueError("FILES_INVALID")
files: dict[str, bytes] = {}
for path, encoded in pairs:
if not isinstance(path, str) or not isinstance(encoded, str):
raise ValueError("FILES_INVALID")
files[path] = base64.b64decode(encoded, validate=True)
if not files:
raise ValueError("FILES_EMPTY")
return files
def commit_files(self, request: Mapping[str, Any]) -> dict[str, Any]:
key = (str(request["task_id"]), int(request["generation"]))
runtime = self.runtimes.get(key)
if runtime is None:
raise RuntimeError("LEASE_NOT_FOUND")
request_v2 = runtime.binding.request
if (
str(request["target_branch"]) != request_v2.target_branch
or str(request["expected_branch_oid"]) != request_v2.expected_branch_oid
):
raise RuntimeError("LEASE_BINDING_CONFLICT")
commit = ContinuationCommit(
idempotency_key=str(request["idempotency_key"]),
files=self._files(request),
message=str(request["message"]),
)
runtime.executor.workspace.materialize_files(commit.files)
return runtime.executor.execute(runtime.binding, commit)
def _controller_for_job(self, job_id: str) -> GitOpsJobController:
for runtime in self.runtimes.values():
controller = runtime.controller
try:
controller.store.read(job_id)
return controller
except Exception:
continue
jobs_root = self.config.state_root / "jobs"
if jobs_root.is_dir():
for root in sorted(jobs_root.iterdir()):
if not root.is_dir() or not (root / "state.sqlite3").is_file():
continue
controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
try:
controller.store.read(job_id)
self.controllers["job:" + job_id] = controller
return controller
except Exception:
continue
raise RuntimeError("JOB_NOT_FOUND")
@staticmethod
def _valid_oid(value: object) -> bool:
return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{40}", value) is not None
def _reconstruct_legacy_binding(
self,
row: Mapping[str, Any],
controller: GitOpsJobController,
) -> dict[str, Any]:
raw = json.loads(row["request_json"])
binding = raw.get("binding", raw)
if not isinstance(binding, dict):
raise RuntimeError("DURABLE_BINDING_INVALID")
task_id = str(binding.get("task_id") or binding.get("campaign_id") or "").strip()
repository = str(binding.get("repository") or "").strip()
operation = str(binding.get("operation") or row.get("operation") or "").strip()
branch = str(binding.get("target_branch") or binding.get("branch") or row.get("branch") or "").strip()
expected = str(binding.get("expected_branch_oid") or row.get("expected_branch_oid") or "").strip()
base = binding.get("base_oid") or binding.get("lineage_anchor_oid")
if not base:
with controller.store._connect() as connection:
lease_row = connection.execute(
"SELECT lineage_anchor_oid FROM branch_leases WHERE job_id=? ORDER BY revision DESC LIMIT 1",
(str(row["job_id"]),),
).fetchone()
if lease_row is not None:
base = lease_row["lineage_anchor_oid"]
base = str(base or "").strip()
branch_mode = str(
binding.get("branch_mode")
or ("CREATE_NEW" if operation.endswith("_CREATE_NEW") else "CONTINUE_EXISTING")
).strip()
try:
generation = int(binding.get("generation") or 0)
except (TypeError, ValueError) as exc:
raise RuntimeError("DURABLE_BINDING_GENERATION_INVALID") from exc
if (
not task_id
or repository != self.config.repository
or not branch.startswith("agent/")
or branch in {"main", "master"}
or branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}
or generation < 1
or not self._valid_oid(base)
or not self._valid_oid(expected)
):
raise RuntimeError("DURABLE_BINDING_INVALID")
current = self.authority.read_ref("refs/heads/" + branch)
if (
current is None
or not self.authority.commit_exists(base)
or not self.authority.commit_exists(expected)
or not self.authority.ancestry_check(base, current)
or not self.authority.ancestry_check(expected, current)
):
raise RuntimeError("DURABLE_BINDING_LINEAGE_INVALID")
return {
"task_id": task_id,
"repository": repository,
"branch_mode": branch_mode,
"base_oid": base,
"expected_branch_oid": expected,
"target_branch": branch,
"generation": generation,
"lease_seconds": int(binding.get("lease_seconds") or 900),
}
def _executor_from_row(
self,
row: Mapping[str, Any],
controller: GitOpsJobController,
) -> ContinuationExecutor:
request = self._reconstruct_legacy_binding(row, controller)
lease = WriterLeaseRequestV2(**request)
worktree_path = self.config.state_root / "worktrees" / _safe(lease.task_id)
if not worktree_path.exists():
current = self.authority.read_ref("refs/heads/" + lease.target_branch)
if current is None:
raise RuntimeError("DURABLE_WORKTREE_BRANCH_MISSING")
self.authority.create_detached_worktree(
current,
_safe(lease.task_id),
self.config.state_root / "worktrees",
)
workspace = DetachedGitWorkspace(worktree_path)
credential = ServerSideCredentialProjection(self.config.ssh_key_path, self.config.token_path)
remote = GiteaContinuationRemote(
workspace=worktree_path,
credential=credential,
remote_name=self.config.remote_name,
repository=lease.repository,
api_base=self.config.api_base,
)
return ContinuationExecutor(
controller,
AuthorityGitAdapter(self.authority),
workspace,
remote,
credential,
)
def job(self, request: Mapping[str, Any], action: str) -> dict[str, Any]:
job_id = str(request["job_id"])
controller = self._controller_for_job(job_id)
row = controller.store.read(job_id)
if action == "get":
return controller.store._receipt(row, duplicate=True).as_dict()
if action == "resume":
return controller.resume(job_id).as_dict()
return self._executor_from_row(row, controller).reconcile(job_id)
def handlers(self) -> dict[str, Any]:
return {
"createGovernedTask": self.create_task,
"acquireWriterLease": self.acquire_lease,
"commitFileGoverned": lambda request: self.commit_files(
{**request, "files": {request["path"]: request["content_base64"]}}
),
"commitFilesGoverned": self.commit_files,
"getGitOpsJob": lambda request: self.job(request, "get"),
"resumeGitOpsJob": lambda request: self.job(request, "resume"),
"reconcileGitOpsJob": lambda request: self.job(request, "reconcile"),
"createTagApprovalRequest": self.release.approval_request,
"requestOrCreateTag": self.release.publish_tag,
"getTagPublicationStatus": self.release.tag_status,
"buildProductionArtifact": self.release.build_artifact,
"executeApprovedProductionRelease": self.release.execute_request,
"getReleaseConsistency": self.release.consistency,
"runReleaseAcceptance": self.release.acceptance,
}
def runtime_status(self) -> dict[str, Any]:
return {
"status": "active",
"health": 200,
"control_plane_target_oid": self.config.target_oid,
"control_plane_release_version": self.config.target_version,
"repository": self.config.repository,
"authoritative_git_dir": str(self.config.authoritative_git_dir),
"legacy_global_writes_enabled": False,
}
def build_governed_handlers(runtime_config: RuntimeConfig | Mapping[str, Any]) -> Mapping[str, Any]:
if isinstance(runtime_config, Mapping):
runtime_config = RuntimeConfig(**runtime_config)
return GovernedRuntime(runtime_config).handlers()
class _HTTPError(RuntimeError):
pass
class BridgeHandler(BaseHTTPRequestHandler):
adapter: BridgeV2Adapter
def log_message(self, *_args: Any) -> None:
return
def _json(self, value: Any, status: int = 200) -> None:
raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def _body(self) -> dict[str, Any]:
try:
length = int(self.headers.get("Content-Length", "-1"))
except ValueError as exc:
raise _HTTPError("CONTENT_LENGTH_INVALID") from exc
if length < 0 or length > MAX_BODY:
raise _HTTPError("REQUEST_TOO_LARGE")
value = json.loads(self.rfile.read(length).decode("utf-8"))
if not isinstance(value, dict):
raise _HTTPError("JSON_OBJECT_REQUIRED")
return value
def _allowed(self, operation: str, body: Mapping[str, Any]) -> None:
allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | {
"schema", "lease_seconds", "create_target_branch", "worktree_mode", "cas_ref_update_required",
"owner_approval", "execute"
}
if operation == "commitFilesGoverned":
allowed.add("files")
if set(body) - allowed:
raise _HTTPError("UNKNOWN_FIELDS")
def do_GET(self) -> None:
if self.path == "/v2/runtime":
self._json(self.adapter.getRuntimeStatus())
return
if self.path == "/v2/capabilities":
self._json(self.adapter.getGovernedWriteCapabilities())
return
self._json({"error": "NOT_FOUND"}, 404)
def do_POST(self) -> None:
operation = ROUTES.get(self.path)
if operation is None or operation in ("getRuntimeStatus", "getGovernedWriteCapabilities"):
self._json({"error": "NOT_FOUND"}, 404)
return
try:
body = self._body()
self._allowed(operation, body)
self._json(getattr(self.adapter, operation)(body))
except _HTTPError as exc:
self._json({"error": str(exc)}, 413 if str(exc) == "REQUEST_TOO_LARGE" else 400)
except (BootstrapError, ValueError, BridgeV2Error) as exc:
code = 409 if str(exc) in {"CONFLICT", "TASK_BINDING_CONFLICT", "LEASE_BINDING_CONFLICT"} else 400
self._json({"error": str(exc)}, code)
except Exception as exc:
name = str(exc) or type(exc).__name__
code = 409 if any(token in name for token in ("CONFLICT", "EXPECTATION", "ALREADY", "MONOTONIC")) else 503
self._json({"error": name}, code)
def serve(adapter: BridgeV2Adapter, host: str = "127.0.0.1", port: int = 18892) -> None:
BridgeHandler.adapter = adapter
ThreadingHTTPServer((host, port), BridgeHandler).serve_forever()
def main() -> None:
state_root = Path("/var/lib/aios-gitops/bridge-v2-staging")
token_path = Path(os.environ.get("AIOS_GITEA_TOKEN_FILE", "/run/credentials/aios-chatgpt-bridge-v2-staging.service/gitea_token"))
ssh_key_path = Path(os.environ.get("AIOS_GITEA_SSH_KEY_PATH", "/home/aiman/.ssh/id_ed25519_codex_tmp"))
os.environ["CREDENTIAL_FILE"] = str(token_path)
os.environ.setdefault("GITEA_MACHINE_ACTOR", "aios-autonomy-agent")
config = RuntimeConfig(
os.environ.get("AIOS_CONTROL_PLANE_TARGET_OID", ""),
Path("/data/aios-gitops/repo.git"),
state_root,
state_root / "jobs",
token_path,
ssh_key_path,
remote_name=os.environ.get("AIOS_GITEA_REMOTE", "gitea-http"),
api_base=os.environ.get("AIOS_GITEA_API_BASE", "https://g.cnnetmkt.top/git/api/v1"),
target_version=os.environ.get("AIOS_CONTROL_PLANE_TARGET_VERSION", "V2R2C1"),
)
runtime = GovernedRuntime(config)
serve(
BridgeV2Adapter(runtime_status=runtime.runtime_status(), handlers=runtime.handlers()),
port=int(os.environ.get("AIOS_BRIDGE_V2_PORT", "18892")),
)
if __name__ == "__main__":
main()
@@ -0,0 +1,635 @@
"""Governed continuation executor for exact existing task branches.
The executor composes the existing continuation lease and durable Job store.
All Git and Gitea side effects live behind injected narrow interfaces so the
state machine can prove Authority, Git-network, API-branch, and commit-object
agreement before it reports completion.
"""
from __future__ import annotations
import base64
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import subprocess
from typing import Callable, Mapping, Protocol
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen
from aios_authority_git_executor import AuthorityGitExecutor
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import GitOpsJobController, JobState
from authorized_task_writer_provisioning import validate_writer_lease_request_v2
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SECRET_PATTERNS = (
re.compile(rb"gh[pousr]_[A-Za-z0-9]{20,}"),
re.compile(rb"(?i)(?:token|password|secret|api[_-]?key)\s*[:=]\s*[\"'][^\"'\s]{12,}[\"']"),
re.compile(rb"-----BEGIN " + rb"(?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
)
class ContinuationExecutionError(RuntimeError):
"""Stable fail-closed continuation execution error."""
class DetachedWorkspace(Protocol):
def is_detached(self) -> bool: ...
def head_oid(self) -> str: ...
def materialize_files(self, files: Mapping[str, bytes]) -> None: ...
def commit_files(self, files: Mapping[str, bytes], message: str) -> str: ...
class AuthorityPort(Protocol):
def branch_exists(self, branch: str) -> bool: ...
def branch_head(self, branch: str) -> str: ...
def is_ancestor(self, older: str, newer: str) -> bool: ...
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str: ...
def create_task_branch_CAS(self, branch: str, new: str) -> str: ...
def commit_exists(self, oid: str) -> bool: ...
class RemotePort(Protocol):
def repository_identity(self) -> str: ...
def push_exact(self, branch: str, expected: str | None, new: str) -> None: ...
def read_git_ref(self, branch: str) -> str | None: ...
def read_api_branch(self, branch: str) -> str | None: ...
def commit_visible(self, oid: str) -> bool: ...
class CredentialProjection(Protocol):
def project(self) -> None: ...
def revoke(self) -> None: ...
def _run(argv: list[str], *, cwd: Path | None = None, environment: Mapping[str, str] | None = None) -> str:
result = subprocess.run(
argv,
cwd=cwd,
env=None if environment is None else dict(environment),
text=True,
capture_output=True,
check=False,
)
if result.returncode != 0:
raise ContinuationExecutionError("SUBPROCESS_FAILED")
return result.stdout.strip()
@dataclass(slots=True)
class AuthorityGitAdapter:
"""Branch-name adapter over the existing narrow AuthorityGitExecutor."""
executor: AuthorityGitExecutor
@staticmethod
def _ref(branch: str) -> str:
return f"refs/heads/{branch}"
def branch_exists(self, branch: str) -> bool:
return self.executor.read_ref(self._ref(branch)) is not None
def branch_head(self, branch: str) -> str:
value = self.executor.read_ref(self._ref(branch))
if value is None:
raise ContinuationExecutionError("EXISTING_BRANCH_MISSING")
return value
def is_ancestor(self, older: str, newer: str) -> bool:
return self.executor.ancestry_check(older, newer)
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str:
return self.executor.advance_task_branch_CAS(self._ref(branch), expected, new)
def create_task_branch_CAS(self, branch: str, new: str) -> str:
return self.executor.create_task_branch_CAS(self._ref(branch), True, new)
def commit_exists(self, oid: str) -> bool:
return self.executor.commit_exists(oid)
@dataclass(slots=True)
class DetachedGitWorkspace:
path: Path
def __post_init__(self) -> None:
if not self.path.is_absolute() or self.path.is_symlink() or not self.path.is_dir():
raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
top = Path(_run(["git", "rev-parse", "--show-toplevel"], cwd=self.path)).resolve()
if top != self.path.resolve():
raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
def is_detached(self) -> bool:
result = subprocess.run(
["git", "symbolic-ref", "-q", "HEAD"],
cwd=self.path,
capture_output=True,
check=False,
)
return result.returncode == 1
def head_oid(self) -> str:
return _run(["git", "rev-parse", "HEAD"], cwd=self.path)
def _dirty_paths(self) -> set[str]:
command = subprocess.run(
["git", "status", "--porcelain=v1", "--untracked-files=all"],
cwd=self.path,
text=True,
capture_output=True,
check=False,
)
if command.returncode != 0:
raise ContinuationExecutionError("SUBPROCESS_FAILED")
output = command.stdout
result: set[str] = set()
for line in output.splitlines():
path = line[3:]
if " -> " in path:
path = path.split(" -> ", 1)[1]
result.add(path)
return result
def materialize_files(self, files: Mapping[str, bytes]) -> None:
"""Author only the governed payload into a clean detached worktree."""
expected_paths = set(files)
if not expected_paths:
raise ContinuationExecutionError("COMMIT_REQUEST_INVALID")
current_paths = self._dirty_paths()
if current_paths not in (set(), expected_paths):
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
root = self.path.resolve()
for relative, content in files.items():
pure = PurePosixPath(relative)
if pure.is_absolute() or ".." in pure.parts or relative.startswith((".git/", "baseline/", "production/")):
raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
target = (self.path / relative).resolve()
try:
target.relative_to(root)
except ValueError as exc:
raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
if target.is_symlink():
raise ContinuationExecutionError("COMMIT_PATH_SYMLINK")
if target.is_file() and target.read_bytes() == content:
continue
target.parent.mkdir(parents=True, exist_ok=True)
temporary = target.with_name("." + target.name + ".aios-materialize")
if temporary.exists() or temporary.is_symlink():
raise ContinuationExecutionError("COMMIT_TEMP_PATH_CONFLICT")
temporary.write_bytes(content)
os.chmod(temporary, 0o644)
os.replace(temporary, target)
if self._dirty_paths() != expected_paths:
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
def commit_files(self, files: Mapping[str, bytes], message: str) -> str:
expected_paths = set(files)
if self._dirty_paths() != expected_paths:
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
for relative, content in files.items():
target = (self.path / relative).resolve()
try:
target.relative_to(self.path.resolve())
except ValueError as exc:
raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
if not target.is_file() or target.is_symlink() or target.read_bytes() != content:
raise ContinuationExecutionError("AUTHORING_PAYLOAD_MISMATCH")
_run(["git", "add", "--", *sorted(expected_paths)], cwd=self.path)
staged = set(_run(["git", "diff", "--cached", "--name-only"], cwd=self.path).splitlines())
if staged != expected_paths:
raise ContinuationExecutionError("STAGED_SCOPE_MISMATCH")
environment = os.environ.copy()
environment.update(
{
"GIT_AUTHOR_NAME": "AIOS Governed Continuation",
"GIT_AUTHOR_EMAIL": "aios-governed-continuation@localhost",
"GIT_COMMITTER_NAME": "AIOS Governed Continuation",
"GIT_COMMITTER_EMAIL": "aios-governed-continuation@localhost",
}
)
_run(["git", "commit", "--no-gpg-sign", "-m", message, "--", *sorted(expected_paths)], cwd=self.path, environment=environment)
if not self.is_detached():
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
return self.head_oid()
@dataclass(slots=True)
class ServerSideCredentialProjection:
ssh_key_path: Path
api_token_path: Path
active: bool = False
def project(self) -> None:
for path in (self.ssh_key_path, self.api_token_path):
if path.is_symlink() or not path.is_file() or path.stat().st_size < 32:
raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
self.active = True
def revoke(self) -> None:
self.active = False
@dataclass(slots=True)
class GiteaContinuationRemote:
workspace: Path
credential: ServerSideCredentialProjection
remote_name: str = "gitea"
repository: str = "aios/aios"
api_base: str = "https://g.cnnetmkt.top/git/api/v1"
expected_host: str = "g.cnnetmkt.top"
@staticmethod
def validate_remote_url(value: str, expected_host: str = "g.cnnetmkt.top") -> None:
if value.startswith(("/", "file:")) or value.endswith("/data/aios-gitops/repo.git"):
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
if "://" in value:
parsed = urlsplit(value)
host, path = parsed.hostname, parsed.path
else:
location = value.split("@", 1)[-1]
host, separator, path = location.partition(":")
if not separator:
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
path = "/" + path
if host != expected_host or path.rstrip("/") not in {"/aios/aios.git", "/git/aios/aios.git"}:
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
def repository_identity(self) -> str:
url = _run(["git", "remote", "get-url", self.remote_name], cwd=self.workspace)
self.validate_remote_url(url, self.expected_host)
if self.repository != "aios/aios":
raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
return self.repository
def _require_projected(self) -> None:
if not self.credential.active:
raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
def read_git_ref(self, branch: str) -> str | None:
self._require_projected()
output = _run(["git", "ls-remote", self.remote_name, f"refs/heads/{branch}"], cwd=self.workspace)
if not output:
return None
oid, ref = output.split(maxsplit=1)
if ref != f"refs/heads/{branch}" or not OID_RE.fullmatch(oid):
raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
return oid
def push_exact(self, branch: str, expected: str | None, new: str) -> None:
self._require_projected()
self.repository_identity()
observed = self.read_git_ref(branch)
if observed == new:
return
if expected is None:
if observed is not None:
raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
elif observed != expected:
raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
_run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)
def _api(self, path: str) -> object:
self._require_projected()
token = self.credential.api_token_path.read_text(encoding="utf-8").strip()
request = Request(
self.api_base.rstrip("/") + path,
headers={"Accept": "application/json", "Authorization": f"token {token}"},
)
try:
with urlopen(request, timeout=20) as response:
return json.loads(response.read().decode("utf-8"))
except HTTPError as exc:
if exc.code == 404:
return None
raise ContinuationExecutionError(f"GITEA_API_HTTP_{exc.code}") from exc
except (URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
raise ContinuationExecutionError("GITEA_API_UNAVAILABLE") from exc
def read_api_branch(self, branch: str) -> str | None:
owner, name = self.repository.split("/", 1)
page = 1
while page <= 100:
value = self._api(f"/repos/{quote(owner)}/{quote(name)}/branches?page={page}&limit=50")
if value is None:
return None
if not isinstance(value, list):
raise ContinuationExecutionError("GITEA_API_BRANCH_COLLECTION_INVALID")
for item in value:
if not isinstance(item, dict) or item.get("name") != branch:
continue
oid = item.get("commit", {}).get("id")
if not isinstance(oid, str) or not OID_RE.fullmatch(oid):
raise ContinuationExecutionError("GITEA_API_BRANCH_INVALID")
return oid
if len(value) < 50:
return None
page += 1
raise ContinuationExecutionError("GITEA_API_BRANCH_PAGINATION_LIMIT")
def commit_visible(self, oid: str) -> bool:
owner, name = self.repository.split("/", 1)
value = self._api(f"/repos/{quote(owner)}/{quote(name)}/git/commits/{oid}")
if not isinstance(value, dict):
return False
observed = value.get("sha") or value.get("id")
return observed == oid
@dataclass(frozen=True, slots=True)
class ContinuationPermissions:
create_target_branch: bool = False
force_push: bool = False
tag_write: bool = False
baseline_write: bool = False
production_write: bool = False
main_write: bool = False
def validate(self, *, branch_mode: str = "CONTINUE_EXISTING") -> None:
if self.create_target_branch and branch_mode != "CREATE_NEW":
raise ContinuationExecutionError("MUTATION_PERMISSION_DENIED")
if self.force_push or self.tag_write or self.baseline_write or self.production_write or self.main_write:
raise ContinuationExecutionError("MUTATION_PERMISSION_DENIED")
@dataclass(frozen=True, slots=True)
class ContinuationRuntimeBinding:
request: WriterLeaseRequestV2
workspace_path: Path
lease_not_before: datetime
lease_expires_at: datetime
original_receipt_valid: bool
original_grant_consumed: bool
permissions: ContinuationPermissions
@dataclass(frozen=True, slots=True)
class ContinuationCommit:
idempotency_key: str
files: Mapping[str, bytes]
message: str
def payload(self) -> bytes:
if not self.idempotency_key or not self.message.strip() or not self.files:
raise ContinuationExecutionError("COMMIT_REQUEST_INVALID")
normalized: dict[str, str] = {}
for path, content in self.files.items():
pure = PurePosixPath(path)
if (
not isinstance(path, str)
or pure.is_absolute()
or ".." in pure.parts
or path.startswith((".git/", "baseline/", "production/"))
or not isinstance(content, bytes)
):
raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
if any(pattern.search(content) for pattern in SECRET_PATTERNS):
raise ContinuationExecutionError("SECRET_LIKE_PAYLOAD_REJECTED")
normalized[path] = base64.b64encode(content).decode("ascii")
return json.dumps(
{"files_base64": normalized, "message": self.message},
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
class ContinuationExecutor:
def __init__(
self,
controller: GitOpsJobController,
authority: AuthorityPort,
workspace: DetachedWorkspace,
remote: RemotePort,
credential: CredentialProjection,
*,
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
) -> None:
self.controller = controller
self.authority = authority
self.workspace = workspace
self.remote = remote
self.credential = credential
self.now = now
self._observed_generations: dict[str, int] = {}
def observe_generation(self, branch: str, generation: int) -> None:
self._observed_generations[branch] = max(generation, self._observed_generations.get(branch, 0))
def _max_generation(self, branch: str) -> int:
observed = self._observed_generations.get(branch, 0)
with self.controller.store._connect() as connection:
row = connection.execute(
"SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
(branch,),
).fetchone()
durable = int(row["generation"] or 0)
return max(observed, durable)
@staticmethod
def _request_mapping(request: WriterLeaseRequestV2) -> dict[str, object]:
return {
"schema": WriterLeaseRequestV2.schema,
"task_id": request.task_id,
"repository": request.repository,
"branch_mode": request.branch_mode,
"base_oid": request.base_oid,
"expected_branch_oid": request.expected_branch_oid,
"target_branch": request.target_branch,
"generation": request.generation,
"lease_seconds": request.lease_seconds,
}
def _validate_preflight(
self,
binding: ContinuationRuntimeBinding,
*,
allow_same_generation: bool = False,
) -> None:
request = binding.request
normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
expected_normalized = {
**self._request_mapping(request),
"create_target_branch": request.branch_mode == "CREATE_NEW",
"worktree_mode": "detached_at_base_oid" if request.branch_mode == "CREATE_NEW" else "detached_at_expected_head",
"cas_ref_update_required": True,
}
if normalized != expected_normalized:
raise ContinuationExecutionError("CONTINUATION_NORMALIZATION_INVALID")
if not binding.original_receipt_valid or not binding.original_grant_consumed:
raise ContinuationExecutionError("ORIGINAL_GRANT_LINEAGE_INVALID")
binding.permissions.validate(branch_mode=request.branch_mode)
now = self.now()
if now < binding.lease_not_before:
raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
if now >= binding.lease_expires_at:
raise ContinuationExecutionError("LEASE_EXPIRED")
if request.generation <= self._max_generation(request.target_branch) and not allow_same_generation:
raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
if not self.workspace.is_detached():
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
if self.workspace.head_oid() != request.expected_branch_oid:
raise ContinuationExecutionError("WORKTREE_HEAD_MISMATCH")
if self.remote.repository_identity() != request.repository:
raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
def _job_request(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit, payload: bytes) -> dict[str, object]:
request = binding.request
return {
"idempotency_key": commit.idempotency_key,
"campaign_id": request.task_id,
"task_id": request.task_id,
"repository": request.repository,
"operation": "CONTINUATION_COMMIT_FILES_CREATE_NEW" if request.branch_mode == "CREATE_NEW" else "CONTINUATION_COMMIT_FILES",
"branch": request.target_branch,
"expected_branch_oid": request.expected_branch_oid,
"base_oid": request.base_oid,
"branch_mode": request.branch_mode,
"payload_digest": hashlib.sha256(payload).hexdigest(),
"lineage_anchor_oid": request.base_oid,
"generation": request.generation,
}
def execute(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit) -> dict[str, object]:
payload = commit.payload()
submitted = self.controller.submit(self._job_request(binding, commit, payload), payload=payload)
current = self.controller.store.read(submitted.job_id)
if JobState(current["state"]) == JobState.COMPLETED:
return self.controller.store._receipt(current, duplicate=True).as_dict()
retry_same_generation = False
if JobState(current["state"]) in {
JobState.EXECUTING,
JobState.RECOVERABLE_FAILURE,
JobState.RECONCILING,
}:
receipt = json.loads(current["receipt_json"]) if current["receipt_json"] else {}
if not receipt.get("recovery_safe_before_git_side_effect"):
raise ContinuationExecutionError("RECONCILIATION_REQUIRED")
if any(key in receipt for key in ("authority_oid", "desired_remote_oid", "git_remote_oid", "api_branch_oid")):
raise ContinuationExecutionError("RECONCILIATION_REQUIRED")
retry_same_generation = True
if JobState(current["state"]) != JobState.RECONCILING:
self.controller.advance(
submitted.job_id,
JobState.RECONCILING,
side_effect_state="PRE_SIDE_EFFECT_RETRY",
receipt={"recovery_safe_before_git_side_effect": True},
)
self._validate_preflight(binding, allow_same_generation=retry_same_generation)
request = binding.request
projected = False
try:
self.credential.project()
projected = True
row = self.controller.store.read(submitted.job_id)
if JobState(row["state"]) == JobState.REQUESTED:
self.controller.advance(
submitted.job_id,
JobState.AUTHORIZED,
side_effect_state="AUTHORIZED",
receipt={"task_id": request.task_id, "original_grant_consumed": True},
)
row = self.controller.store.read(submitted.job_id)
if JobState(row["state"]) == JobState.AUTHORIZED:
self.controller.lease_job(
submitted.job_id,
lease_id=f"lease-{submitted.job_id}",
lineage_anchor_oid=request.base_oid,
current_head_oid=request.expected_branch_oid,
generation=request.generation,
lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(),
expected_revision=int(row["revision"]),
)
self.controller.advance(submitted.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
self.controller.advance(submitted.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
new_oid = self.workspace.commit_files(commit.files, commit.message)
if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
if request.branch_mode == "CREATE_NEW":
authority_oid = self.authority.create_task_branch_CAS(request.target_branch, new_oid)
else:
authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
if authority_oid != new_oid:
raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
self.controller.advance(
submitted.job_id,
JobState.LOCAL_VERIFIED,
side_effect_state="LOCAL_VERIFIED",
receipt={"authority_oid": authority_oid, "force_push": False},
)
try:
self.controller.advance(submitted.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
except Exception:
return self._remote_pending(submitted.job_id, authority_oid, new_oid)
return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
finally:
if projected:
self.credential.revoke()
def _observe_remote(self, branch: str, oid: str) -> tuple[str | None, str | None, bool]:
return self.remote.read_git_ref(branch), self.remote.read_api_branch(branch), self.remote.commit_visible(oid)
def _remote_pending(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
receipt = {
"authority_oid": authority_oid,
"desired_remote_oid": new_oid,
"remote_verified": False,
"force_push": False,
}
result = self.controller.advance(
job_id,
JobState.RECONCILIATION_REQUIRED,
side_effect_state="REMOTE_PUSH_PENDING",
receipt=receipt,
)
return result.as_dict()
def _finish_remote_verification(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
row = self.controller.store.read(job_id)
branch = row["branch"]
git_oid, api_oid, visible = self._observe_remote(branch, new_oid)
if not (authority_oid == git_oid == api_oid == new_oid and visible):
return self._remote_pending(job_id, authority_oid, new_oid)
receipt = {
"authority_oid": authority_oid,
"git_remote_oid": git_oid,
"api_branch_oid": api_oid,
"commit_visible": True,
"remote_verified": True,
"force_push": False,
}
self.controller.advance(job_id, JobState.REMOTE_VERIFIED, side_effect_state="REMOTE_VERIFIED", receipt=receipt)
return self.controller.advance(job_id, JobState.COMPLETED, side_effect_state="REMOTE_VERIFIED", receipt=receipt).as_dict()
def reconcile(self, job_id: str) -> dict[str, object]:
row = self.controller.store.read(job_id)
state = JobState(row["state"])
if state == JobState.COMPLETED:
return self.controller.store._receipt(row, duplicate=True).as_dict()
if state != JobState.RECONCILIATION_REQUIRED:
raise ContinuationExecutionError("JOB_NOT_RECONCILABLE")
request = json.loads(row["request_json"])["binding"]
receipt = json.loads(row["receipt_json"])
new_oid = receipt["desired_remote_oid"]
authority_oid = self.authority.branch_head(row["branch"])
self.credential.project()
try:
git_oid, api_oid, visible = self._observe_remote(row["branch"], new_oid)
if authority_oid == git_oid == api_oid == new_oid and visible:
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST")
return self._finish_remote_verification(job_id, authority_oid, new_oid)
expected_oid = None if row["operation"].endswith("_CREATE_NEW") else request["expected_branch_oid"]
if authority_oid == new_oid and git_oid == api_oid == expected_oid:
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_RETRY")
self.remote.push_exact(row["branch"], expected_oid, new_oid)
return self._finish_remote_verification(job_id, authority_oid, new_oid)
return self.controller.store._receipt(row, duplicate=True).as_dict()
finally:
self.credential.revoke()
@@ -74,6 +74,12 @@ class WriterLeaseRequestV2:
raise ContinuationError("EXISTING_BRANCH_OID_CONFLICT")
if not probe.is_ancestor(self.base_oid, actual):
raise ContinuationError("BASE_NOT_ANCESTOR_OF_BRANCH")
elif hasattr(probe, "commit_exists") and not probe.commit_exists(self.base_oid):
raise ContinuationError("BASE_COMMIT_NOT_FOUND")
elif self.branch_mode == "CREATE_NEW" and probe.branch_exists(self.target_branch):
raise ContinuationError("TARGET_BRANCH_ALREADY_EXISTS")
elif self.branch_mode == "CREATE_NEW" and self.base_oid != self.expected_branch_oid:
raise ContinuationError("CREATE_NEW_BASE_EXPECTATION_MISMATCH")
@dataclass(frozen=True, slots=True)
@@ -249,7 +249,8 @@ class JobStore:
if state == JobState.COMPLETED:
status, side = 200, "COMPLETED"
elif state in _RECOVERABLE:
status, side = 202, "RESUME_SAME_JOB"
status = 202
side = row["side_effect_state"] if row["side_effect_state"] == "REMOTE_PUSH_PENDING" else "RESUME_SAME_JOB"
elif row["side_effect_state"] == "UNKNOWN":
status, side = 503, "UNKNOWN"
else:
@@ -280,10 +281,13 @@ class GitOpsJobController:
# durable round trip and participate in idempotency. Arbitrary request
# fields are intentionally discarded rather than persisted.
optional = (
"idempotency_key",
"task_id",
"new_branch_oid",
"authorization_recovery_digest",
"lineage_anchor_oid",
"base_oid",
"branch_mode",
"generation",
"expected_main_oid",
"pr_number",
@@ -314,7 +318,15 @@ class GitOpsJobController:
raise JobError("PAYLOAD_REFERENCE_MISMATCH")
if payload_ref is not None and (not isinstance(payload_size, int) or payload_size < 0):
raise JobError("PAYLOAD_SIZE_INVALID")
key = sha256_json(binding)
client_idempotency_key = binding.get("idempotency_key")
if client_idempotency_key is not None:
if len(client_idempotency_key) > 128 or not all(
character.isalnum() or character in "._:-" for character in client_idempotency_key
):
raise JobError("IDEMPOTENCY_KEY_INVALID")
key = sha256_json({"client_idempotency_key": client_idempotency_key})
else:
key = sha256_json(binding)
job_id = "job-" + key[:32]
durable_request = dict(binding)
if payload_ref is not None:
@@ -0,0 +1,256 @@
"""Exact-base governed CREATE_NEW task bootstrap using the existing JobStore."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
import hashlib
import json
import re
from pathlib import Path
from typing import Any, Mapping
from aios_authority_git_executor import AuthorityGitExecutor
from aios_continuation_executor import (
AuthorityGitAdapter,
ContinuationExecutor,
ContinuationPermissions,
ContinuationRuntimeBinding,
DetachedGitWorkspace,
GiteaContinuationRemote,
ServerSideCredentialProjection,
)
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import GitOpsJobController, JobStore
OID_RE = re.compile(r"^[0-9a-f]{40}$")
BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
class BootstrapError(RuntimeError):
pass
@dataclass(frozen=True, slots=True)
class GovernedTaskBinding:
task_id: str
repository: str
base_oid: str
target_branch: str
risk: str
description: str
idempotency_key: str
job_id: str | None = None
def response(self, *, idempotent: bool = False) -> dict[str, Any]:
return {
"task_id": self.task_id,
"repository": self.repository,
"base_oid": self.base_oid,
"target_branch": self.target_branch,
"expected_branch_oid": self.base_oid,
"generation": 1,
"branch_mode": "CREATE_NEW",
"create_target_branch": True,
"worktree_mode": "detached_at_base_oid",
"cas_ref_update_required": True,
"idempotent": idempotent,
"state": "BOUND",
"job_id": self.job_id,
}
def _safe_task(value: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "-", value).strip("-")[:120] or "task"
class GovernedTaskBootstrap:
def __init__(
self,
authority: AuthorityGitExecutor,
root: Path,
*,
remote_name: str,
token_path: Path,
ssh_key_path: Path,
api_base: str = "https://g.cnnetmkt.top/git/api/v1",
) -> None:
self.authority = authority
self.root = root
self.remote_name = remote_name
self.token_path = token_path
self.ssh_key_path = ssh_key_path
self.api_base = api_base
self.task_meta_root = root / "task-bindings"
self.job_root = root / "jobs"
self.worktree_root = root / "worktrees"
self.task_meta_root.mkdir(parents=True, exist_ok=True)
self.job_root.mkdir(parents=True, exist_ok=True)
self.worktree_root.mkdir(parents=True, exist_ok=True)
def _store(self, task_id: str) -> tuple[JobStore, GitOpsJobController]:
task_root = self.job_root / _safe_task(task_id)
store = JobStore(task_root, payload_root=task_root / "payloads")
return store, GitOpsJobController(store)
def _metadata(self, task_id: str) -> Path:
return self.task_meta_root / (_safe_task(task_id) + ".json")
def _read_existing(self, task_id: str) -> GovernedTaskBinding | None:
path = self._metadata(task_id)
if not path.is_file() or path.is_symlink():
return None
raw = json.loads(path.read_text(encoding="utf-8"))
return GovernedTaskBinding(
task_id=raw["task_id"],
repository=raw["repository"],
base_oid=raw["base_oid"],
target_branch=raw["target_branch"],
risk=raw["risk"],
description=raw["description"],
idempotency_key=raw["idempotency_key"],
job_id=raw.get("job_id"),
)
def _write_metadata(self, binding: GovernedTaskBinding) -> None:
path = self._metadata(binding.task_id)
temporary = path.with_suffix(".tmp")
temporary.write_text(
json.dumps(
{
"task_id": binding.task_id,
"repository": binding.repository,
"base_oid": binding.base_oid,
"target_branch": binding.target_branch,
"risk": binding.risk,
"description": binding.description,
"idempotency_key": binding.idempotency_key,
"job_id": binding.job_id,
},
sort_keys=True,
),
encoding="utf-8",
)
temporary.replace(path)
def _pending_create_new_oid(self, task_id: str, branch: str) -> str | None:
"""Return the exact local OID only for an unfinished CREATE_NEW job."""
store, _ = self._store(task_id)
with store._connect() as connection:
rows = connection.execute(
"SELECT operation,state,receipt_json FROM jobs WHERE branch=? ORDER BY updated_at DESC",
(branch,),
).fetchall()
for row in rows:
if row["operation"] != "CONTINUATION_COMMIT_FILES_CREATE_NEW" or row["state"] != "RECONCILIATION_REQUIRED":
continue
receipt = json.loads(row["receipt_json"] or "{}")
desired = receipt.get("desired_remote_oid")
if isinstance(desired, str) and OID_RE.fullmatch(desired):
return desired
return None
def create(self, request: Mapping[str, Any]) -> dict[str, Any]:
values = {key: str(request[key]) for key in (
"task_id",
"repository",
"base_oid",
"target_branch",
"risk",
"description",
"idempotency_key",
)}
if values["repository"] != "aios/aios" or not OID_RE.fullmatch(values["base_oid"]):
raise BootstrapError("TASK_BINDING_INVALID")
if not BRANCH_RE.fullmatch(values["target_branch"]) or values["target_branch"] in {"main", "master"}:
raise BootstrapError("TARGET_BRANCH_INVALID")
existing = self._read_existing(values["task_id"])
if existing is not None:
for key in ("repository", "base_oid", "target_branch", "risk", "description"):
if getattr(existing, key) != values[key]:
raise BootstrapError("TASK_BINDING_CONFLICT")
return existing.response(idempotent=True)
if not self.authority.commit_exists(values["base_oid"]):
raise BootstrapError("BASE_COMMIT_NOT_FOUND")
if self.authority.read_ref("refs/heads/" + values["target_branch"]) is not None:
raise BootstrapError("CONFLICT")
payload = json.dumps(values, sort_keys=True, separators=(",", ":")).encode()
_, controller = self._store(values["task_id"])
receipt = controller.submit(
{
"idempotency_key": values["idempotency_key"],
"campaign_id": values["task_id"],
"task_id": values["task_id"],
"repository": values["repository"],
"operation": "GOVERNED_TASK_CREATE",
"branch": values["target_branch"],
"expected_branch_oid": values["base_oid"],
"payload_digest": hashlib.sha256(payload).hexdigest(),
"lineage_anchor_oid": values["base_oid"],
},
payload=payload,
)
binding = GovernedTaskBinding(**values, job_id=receipt.job_id)
self._write_metadata(binding)
return binding.response()
def prepare_runtime(self, request: Mapping[str, Any], *, lease_seconds: int):
task = self._read_existing(str(request["task_id"]))
if task is None:
raise BootstrapError("TASK_NOT_FOUND")
if str(request["repository"]) != task.repository or str(request["base_oid"]) != task.base_oid:
raise BootstrapError("TASK_BINDING_CONFLICT")
if str(request["target_branch"]) != task.target_branch:
raise BootstrapError("TASK_BINDING_CONFLICT")
if request.get("branch_mode") != "CREATE_NEW" or int(request.get("generation", 0)) != 1:
raise BootstrapError("CREATE_NEW_LEASE_INVALID")
if not self.authority.commit_exists(task.base_oid):
raise BootstrapError("BASE_COMMIT_NOT_FOUND")
existing_target = self.authority.read_ref("refs/heads/" + task.target_branch)
recovery_oid = None
if existing_target is not None:
recovery_oid = self._pending_create_new_oid(task.task_id, task.target_branch)
if recovery_oid != existing_target:
raise BootstrapError("CONFLICT")
workspace_path = self.worktree_root / _safe_task(task.task_id)
if workspace_path.exists():
if workspace_path.is_symlink():
raise BootstrapError("WORKTREE_INVALID")
workspace = DetachedGitWorkspace(workspace_path)
allowed_heads = {task.base_oid}
if recovery_oid is not None:
allowed_heads.add(recovery_oid)
if not workspace.is_detached() or workspace.head_oid() not in allowed_heads:
raise BootstrapError("WORKTREE_HEAD_MISMATCH")
else:
self.authority.create_detached_worktree(task.base_oid, _safe_task(task.task_id), self.worktree_root)
workspace = DetachedGitWorkspace(workspace_path)
request_v2 = WriterLeaseRequestV2(
task_id=task.task_id,
repository=task.repository,
branch_mode="CREATE_NEW",
base_oid=task.base_oid,
expected_branch_oid=task.base_oid,
target_branch=task.target_branch,
generation=1,
lease_seconds=lease_seconds,
)
_, controller = self._store(task.task_id)
credential = ServerSideCredentialProjection(self.ssh_key_path, self.token_path)
remote = GiteaContinuationRemote(
workspace=workspace_path,
credential=credential,
remote_name=self.remote_name,
repository=task.repository,
api_base=self.api_base,
)
executor = ContinuationExecutor(controller, AuthorityGitAdapter(self.authority), workspace, remote, credential)
binding = ContinuationRuntimeBinding(
request=request_v2,
workspace_path=workspace_path,
lease_not_before=datetime.now(timezone.utc) - timedelta(seconds=1),
lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=lease_seconds),
original_receipt_valid=True,
original_grant_consumed=True,
permissions=ContinuationPermissions(create_target_branch=True),
)
return binding, controller, executor
@@ -995,7 +995,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
"target_branch": request.target_branch,
"generation": request.generation,
"lease_seconds": request.lease_seconds,
"create_target_branch": False,
"worktree_mode": "detached_at_expected_head",
"create_target_branch": request.branch_mode == "CREATE_NEW",
"worktree_mode": "detached_at_base_oid" if request.branch_mode == "CREATE_NEW" else "detached_at_expected_head",
"cas_ref_update_required": True,
}
@@ -0,0 +1,568 @@
"""Governed V2R2C1 production release orchestration.
This module is deliberately scoped to the Owner-approved V2R2C1 binding. It
uses the existing TagPublisher and metadata adapters; it does not introduce a
second Git writer, a second durable job store, or an alternate baseline.
"""
from __future__ import annotations
import argparse
import fcntl
import hashlib
import json
import os
import re
import subprocess
import tarfile
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping
from gitea_tag_publication import (
GiteaTagClient,
GitTagBackend,
PublicationManualOwnerReviewRequired,
PublicationPending,
PublicationReconciliationRequired,
PublicationRequest,
TagPublisher,
)
VERSION = "V2R2C1"
TARGET_OID = "998d30304787e31e5ae607c0bd86b5edb2b0e5ad"
EVIDENCE_CLOSURE_OID = "a8b906d1bc24088c19647e98de6a679268f6d0a2"
APPROVAL_PLAN_ID = "sha256:c0df52008ae7f64cec5ee2d28947aa7b892066c670ab8295e4549140b1a7500e"
REPOSITORY = "aios/aios"
SERVICE = "aios-v2r1c7-production.service"
PORT = 18927
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SHA_RE = re.compile(r"^[0-9a-f]{64}$")
class ReleaseBlocked(RuntimeError):
"""A release gate failed and no unsafe recovery is permitted."""
class ManualOwnerReviewRequired(ReleaseBlocked):
pass
class ReconciliationRequired(ReleaseBlocked):
pass
def canonical(value: object) -> bytes:
return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
def sha256_file(path: Path) -> str:
if path.is_symlink() or not path.is_file():
raise ReleaseBlocked("artifact must be a regular file")
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o600) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(path.name + ".tmp")
temporary.write_bytes(canonical(dict(value)))
os.chmod(temporary, mode)
os.replace(temporary, path)
@dataclass(frozen=True)
class ReleaseBinding:
version: str = VERSION
target_oid: str = TARGET_OID
evidence_closure_oid: str = EVIDENCE_CLOSURE_OID
approval_plan_id: str = APPROVAL_PLAN_ID
repository: str = REPOSITORY
production_service: str = SERVICE
production_port: int = PORT
def validate(self) -> None:
if self.version != VERSION or self.repository != REPOSITORY:
raise ManualOwnerReviewRequired("release binding is outside V2R2C1 scope")
if not OID_RE.fullmatch(self.target_oid) or self.target_oid != TARGET_OID:
raise ManualOwnerReviewRequired("V2R2C1 target_oid mismatch")
if not OID_RE.fullmatch(self.evidence_closure_oid) or self.evidence_closure_oid != EVIDENCE_CLOSURE_OID:
raise ManualOwnerReviewRequired("V2R2C1 evidence closure mismatch")
if self.approval_plan_id != APPROVAL_PLAN_ID:
raise ManualOwnerReviewRequired("V2R2C1 approval plan mismatch")
if self.production_service != SERVICE or int(self.production_port) != PORT:
raise ManualOwnerReviewRequired("production service binding mismatch")
def as_dict(self) -> dict[str, Any]:
return {
"version": self.version,
"target_oid": self.target_oid,
"evidence_closure_oid": self.evidence_closure_oid,
"approval_plan_id": self.approval_plan_id,
"repository": self.repository,
"production_service": self.production_service,
"production_port": self.production_port,
}
class StateLock:
def __init__(self, path: Path):
self.path = path
self.handle = None
def __enter__(self) -> "StateLock":
self.path.parent.mkdir(parents=True, exist_ok=True)
self.handle = self.path.open("a+", encoding="utf-8")
fcntl.flock(self.handle.fileno(), fcntl.LOCK_EX)
return self
def __exit__(self, exc_type, exc, tb) -> None:
if self.handle is not None:
fcntl.flock(self.handle.fileno(), fcntl.LOCK_UN)
self.handle.close()
class ReleaseState:
def __init__(self, root: Path, binding: ReleaseBinding):
self.root = root
self.path = root / "release-state.json"
self.binding = binding
root.mkdir(parents=True, exist_ok=True)
if self.path.exists():
value = json.loads(self.path.read_text(encoding="utf-8"))
if value.get("binding") != binding.as_dict():
raise ManualOwnerReviewRequired("persisted release binding conflict")
self.value = value
else:
self.value = {
"schema": "aios.v2r2c1.governed-production-release-state.v1",
"binding": binding.as_dict(),
"state": "INITIAL",
"completed": {},
"history": [],
}
self.save()
def save(self) -> None:
atomic_json(self.path, self.value)
def record(self, state: str, step: str, result: Mapping[str, Any]) -> None:
safe = json.loads(json.dumps(dict(result)))
self.value["state"] = state
self.value.setdefault("completed", {})[step] = safe
self.value.setdefault("history", []).append(
{"at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "state": state, "step": step, "result": safe}
)
self.save()
def fail(self, state: str, reason: str) -> None:
self.value["state"] = state
self.value.setdefault("history", []).append(
{"at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "state": state, "reason": str(reason)[:500]}
)
self.save()
def _git(git_dir: Path, *args: str, check: bool = True) -> str:
completed = subprocess.run(
["git", f"--git-dir={git_dir}", *args],
check=False,
capture_output=True,
text=True,
timeout=45,
env={"PATH": os.environ.get("PATH", "")},
)
if check and completed.returncode != 0:
raise ReconciliationRequired(completed.stderr.strip() or "authoritative git query failed")
return completed.stdout.strip()
def _ancestor(git_dir: Path, ancestor: str, descendant: str) -> bool:
return subprocess.run(
["git", f"--git-dir={git_dir}", "merge-base", "--is-ancestor", ancestor, descendant],
check=False,
capture_output=True,
timeout=45,
env={"PATH": os.environ.get("PATH", "")},
).returncode == 0
def verify_lineage(git_dir: Path, binding: ReleaseBinding) -> dict[str, Any]:
binding.validate()
if _git(git_dir, "cat-file", "-t", binding.target_oid) != "commit":
raise ReleaseBlocked("approved target_oid is not a commit")
if _git(git_dir, "cat-file", "-t", binding.evidence_closure_oid) != "commit":
raise ReleaseBlocked("evidence closure is not a commit")
if not _ancestor(git_dir, binding.target_oid, binding.evidence_closure_oid):
raise ManualOwnerReviewRequired("approved target is not an ancestor of evidence closure")
return {
"target_oid": binding.target_oid,
"evidence_closure_oid": binding.evidence_closure_oid,
"target_is_ancestor_of_evidence": True,
"main_oid": _git(git_dir, "rev-parse", "refs/heads/main"),
}
def verify_owner_approval(value: Mapping[str, Any], binding: ReleaseBinding) -> dict[str, Any]:
approval = value.get("owner_approval")
if not isinstance(approval, Mapping):
raise ReleaseBlocked("OWNER_APPROVAL_RECEIPT_REQUIRED")
if approval.get("approved") is not True or approval.get("status") != "APPROVED":
raise ReleaseBlocked("OWNER_APPROVAL_NOT_ACCEPTED")
if approval.get("approved_by") != "aiman":
raise ManualOwnerReviewRequired("OWNER_APPROVAL_ACTOR_MISMATCH")
if approval.get("approval_source") != "explicit_owner_approval_in_chat":
raise ManualOwnerReviewRequired("OWNER_APPROVAL_SOURCE_MISMATCH")
for key, expected in binding.as_dict().items():
if key in {"repository", "production_service", "production_port"}:
continue
if approval.get(key) != expected:
raise ManualOwnerReviewRequired("OWNER_APPROVAL_BINDING_MISMATCH:" + key)
return dict(approval)
def _payload_file(git_dir: Path, oid: str, path: str) -> bytes:
completed = subprocess.run(
["git", f"--git-dir={git_dir}", "show", f"{oid}:{path}"],
check=False,
capture_output=True,
timeout=45,
env={"PATH": os.environ.get("PATH", "")},
)
if completed.returncode != 0:
raise ReleaseBlocked("required release source missing:" + path)
return completed.stdout
def _commit_epoch(git_dir: Path, oid: str) -> int:
return int(_git(git_dir, "show", "-s", "--format=%ct", oid))
def build_production_artifact(
git_dir: Path,
binding: ReleaseBinding,
output_root: Path,
) -> dict[str, Any]:
"""Build a deterministic executable V2R2C1 health artifact from target_oid."""
binding.validate()
verify_lineage(git_dir, binding)
epoch = _commit_epoch(git_dir, binding.target_oid)
output_root.mkdir(parents=True, exist_ok=True)
stem = f"aios-{VERSION}-{binding.target_oid[:12]}"
output = output_root / (stem + ".tar.zst")
with tempfile.TemporaryDirectory(prefix="v2r2c1-artifact-") as temp:
root = Path(temp) / "payload"
root.mkdir(parents=True)
source_paths = {
"active-runtime/app/deployment_state.py": "active-runtime/app/deployment_state.py",
"active-runtime/app/v2r1c6_privacy_tools.py": "active-runtime/app/v2r1c6_privacy_tools.py",
}
for destination, source in source_paths.items():
target = root / destination
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(_payload_file(git_dir, binding.target_oid, source))
generated: dict[str, str] = {
"VERSION": VERSION + "\n",
"active-runtime/app/v2r2c1_health_service.py": (
"from v2r1c6_privacy_tools import main\n\n"
"if __name__ == '__main__':\n raise SystemExit(main())\n"
),
"deploy/aios-v2/systemd/aios-v2r1c7-production.service": (
"[Unit]\nDescription=AIOS V2R2C1 governed health contract service\n"
"After=network.target\n\n[Service]\nType=simple\n"
"User=aios-v2r1c7\nGroup=aios-v2r1c7\n"
"WorkingDirectory=/opt/aios/v2r2c1\n"
"EnvironmentFile=/etc/aios/aios-v2r2c1.env\n"
"ExecStart=/usr/bin/python3 /opt/aios/v2r2c1/active-runtime/app/v2r2c1_health_service.py --serve --host 127.0.0.1 --port ${AIOS_V2R1C7_PORT}\n"
"Restart=on-failure\nRestartSec=5\nNoNewPrivileges=true\nPrivateTmp=true\n"
"ProtectHome=true\nProtectSystem=strict\nProtectKernelLogs=true\n"
"ProtectKernelModules=true\nProtectKernelTunables=true\nProtectControlGroups=true\n"
"RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6\nIPAddressDeny=any\n"
"IPAddressAllow=localhost\nUMask=0077\n\n[Install]\nWantedBy=multi-user.target\n"
),
"deploy/aios-v2/v2r2c1-runtime-config.json": json.dumps({
"schema": "aios.v2r2c1.production-runtime.v1",
"version": VERSION,
"target_oid": binding.target_oid,
"service": SERVICE,
"port": PORT,
"health_path": "/health",
"deployment_state_source": "environment_and_state_file",
}, sort_keys=True, separators=(",", ":")) + "\n",
"deploy/aios-v2/v2r2c1-rollback.json": json.dumps({
"schema": "aios.v2r2c1.rollback-metadata.v1",
"rollback_version": "V2R1C7",
"rollback_service": SERVICE,
"rollback_target_is_previous_production": True,
"automatic_rollback": False,
}, sort_keys=True, separators=(",", ":")) + "\n",
"scripts/probe_v2r2c1_health.py": (
"import json,sys,urllib.request\n"
"url=sys.argv[1] if len(sys.argv)>1 else 'http://127.0.0.1:18927/health'\n"
"with urllib.request.urlopen(url, timeout=5) as r: print(json.dumps(json.load(r), sort_keys=True))\n"
),
"scripts/rollback_v2r2c1_release.sh": "#!/bin/sh\nset -eu\nprintf '%s\\n' 'rollback requires governed owner-approved executor'\nexit 2\n",
"release-manifests/V2R2C1.json": json.dumps({
"schema": "aios.release-source.v2",
"version": VERSION,
"repository": REPOSITORY,
"target_oid": binding.target_oid,
"evidence_closure_oid": binding.evidence_closure_oid,
"production_deployed": False,
"deployment_state": "STAGING",
"artifact_kind": "executable-production-health",
}, sort_keys=True, separators=(",", ":")) + "\n",
}
for rel, text in generated.items():
target = root / rel
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(text, encoding="utf-8")
if rel.startswith("scripts/"):
target.chmod(0o755)
metadata = {
"schema": "aios.release-artifact-build.v3",
"version": VERSION,
"repository": REPOSITORY,
"source_head_oid": binding.target_oid,
"evidence_closure_oid": binding.evidence_closure_oid,
"source_date_epoch": epoch,
"clean_source": True,
"executable_payload": True,
"production_deployed": False,
"tag_created": False,
"baseline_changed": False,
"secrets_included": False,
}
(root / "BUILD-METADATA.json").write_bytes(canonical(metadata))
(root / "ROLLBACK-METADATA.json").write_bytes(canonical(json.loads((root / "deploy/aios-v2/v2r2c1-rollback.json").read_text(encoding="utf-8"))))
names = sorted("payload/" + str(path.relative_to(root)).replace(os.sep, "/") for path in root.rglob("*") if path.is_file())
(root / "ARTIFACT-MANIFEST.json").write_bytes(canonical({
"schema": "aios.release-artifact-manifest.v3",
"version": VERSION,
"source_head_oid": binding.target_oid,
"evidence_closure_oid": binding.evidence_closure_oid,
"repository": REPOSITORY,
"files": names + ["payload/ARTIFACT-MANIFEST.json"],
"deployable": True,
"executable_payload": True,
"production_deployed": False,
"tag_created": False,
"baseline_changed": False,
}))
tar_path = Path(temp) / "release.tar"
with tarfile.open(tar_path, "w") as archive:
for path in sorted(root.rglob("*")):
relative = str(path.relative_to(root)).replace(os.sep, "/")
info = archive.gettarinfo(str(path), arcname="payload/" + relative)
info.uid = info.gid = 0
info.uname = "root"
info.gname = "root"
info.mtime = epoch
info.mode = 0o755 if path.is_dir() or path.suffix == ".sh" else 0o644
if path.is_file():
with path.open("rb") as stream:
archive.addfile(info, stream)
else:
archive.addfile(info)
subprocess.run(["zstd", "-T1", "-19", "--no-progress", "-q", "-f", str(tar_path), "-o", str(output)], check=True, timeout=120)
digest = sha256_file(output)
receipt = {
"schema": "aios.release-artifact-receipt.v3",
"version": VERSION,
"repository": REPOSITORY,
"source_head_oid": binding.target_oid,
"evidence_closure_oid": binding.evidence_closure_oid,
"artifact": str(output),
"artifact_sha256": digest,
"artifact_size": output.stat().st_size,
"deployable": True,
"executable_payload": True,
"production_deployed": False,
"tag_created": False,
"baseline_changed": False,
}
atomic_json(output.with_suffix(output.suffix + ".receipt.json"), receipt, mode=0o640)
return receipt
class GovernedProductionReleaseExecutor:
def __init__(self, config: Any):
self.config = config
self.binding = ReleaseBinding()
self.binding.validate()
self.root = Path(config.state_root) / "v2r2c1-governed-production-release"
self.state = ReleaseState(self.root, self.binding)
workspace = self.root / "publication-workspace"
workspace.mkdir(parents=True, exist_ok=True)
self.publisher = TagPublisher(
workspace,
GitTagBackend(workspace, authoritative_repo=Path(config.authoritative_git_dir)),
GiteaTagClient(config.api_base, REPOSITORY, Path(config.token_path)),
config.remote_name,
state_root=self.root / "tag-publication",
)
def _request(self, request: Mapping[str, Any]) -> PublicationRequest:
binding = self.binding
for key, expected in binding.as_dict().items():
if key in request and request[key] != expected:
raise ManualOwnerReviewRequired("release request binding mismatch:" + key)
return PublicationRequest(binding.version, binding.target_oid, binding.evidence_closure_oid, binding.approval_plan_id)
def approval_request(self, request: Mapping[str, Any]) -> dict[str, Any]:
if request.get("execute") is not False:
raise ReleaseBlocked("TAG_APPROVAL_REQUEST_MUST_BE_DRY_RUN")
owner = verify_owner_approval(request, self.binding)
with StateLock(self.root / "release.lock"):
current = self.state.value.get("approval")
approval_digest = hashlib.sha256(canonical(owner)).hexdigest()
record = {
"schema": "aios.v2r2c1.tag-approval-request.v1",
"binding": self.binding.as_dict(),
"owner_approval": owner,
"owner_approval_digest_sha256": approval_digest,
"execute": False,
"recorded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
if current and current != record:
raise ManualOwnerReviewRequired("persisted Owner approval differs")
self.state.value["approval"] = record
self.state.record("APPROVAL_REQUEST_RECORDED", "tag_approval_request", record)
return {"status": "RECORDED", "execute": False, "binding": self.binding.as_dict(), "owner_approval_digest_sha256": approval_digest}
def _require_approval(self) -> None:
approval = self.state.value.get("approval")
if not isinstance(approval, Mapping) or approval.get("binding") != self.binding.as_dict():
raise ReleaseBlocked("OWNER_APPROVAL_REQUEST_NOT_RECORDED")
def publish_tag(self, request: Mapping[str, Any]) -> dict[str, Any]:
if request.get("execute") is not True:
raise ReleaseBlocked("TAG_PUBLICATION_REQUIRES_EXECUTE_TRUE")
self._require_approval()
publication = self._request(request)
with StateLock(self.root / "release.lock"):
try:
verified = self.publisher._verify_publication(publication, allow_push=True, previously_deployed=False)
except PublicationManualOwnerReviewRequired as exc:
self.state.fail("MANUAL_OWNER_REVIEW_REQUIRED", str(exc))
raise ManualOwnerReviewRequired(str(exc)) from exc
except PublicationPending as exc:
self.state.fail("REMOTE_PUSH_PENDING", str(exc))
raise ReconciliationRequired(str(exc)) from exc
except PublicationReconciliationRequired as exc:
self.state.fail("RECONCILIATION_REQUIRED", str(exc))
raise ReconciliationRequired(str(exc)) from exc
result = dict(verified)
self.state.record("TAG_VERIFIED", "tag", result)
return {"status": "REMOTE_VERIFIED", **result}
def tag_status(self, request: Mapping[str, Any]) -> dict[str, Any]:
self._request(request)
return self.publisher.get_tag_publication_status(VERSION)
def build_artifact(self, request: Mapping[str, Any]) -> dict[str, Any]:
self._require_approval()
self._request(request)
receipt = build_production_artifact(Path(self.config.authoritative_git_dir), self.binding, self.root / "artifacts")
self.state.record("ARTIFACT_VERIFIED", "production_artifact", receipt)
return {"status": "VERIFIED", **receipt}
def execute_request(self, request: Mapping[str, Any]) -> dict[str, Any]:
self._require_approval()
publication = self._request(request)
tag_state = self.state.value.get("completed", {}).get("tag")
if not isinstance(tag_state, Mapping) or tag_state.get("state") not in {"GITEA_API_CONFIRMED", "REMOTE_VERIFIED"}:
raise ReleaseBlocked("REMOTE_TAG_PROOF_REQUIRED")
artifact = self.state.value.get("completed", {}).get("production_artifact")
if not isinstance(artifact, Mapping):
artifact = self.build_artifact(request)
value = {
**self.binding.as_dict(),
"tag_oid": tag_state.get("tag_oid") or tag_state.get("gitea_tag_oid"),
"artifact_path": artifact["artifact"],
"artifact_size": int(artifact["artifact_size"]),
"artifact_sha256": artifact["artifact_sha256"],
}
# The production adapter is privileged and is run through the root
# approved-release service boundary by the caller. This handler is
# still useful for a root-owned staging invocation and remains
# fail-closed when the adapter is not available.
import v2r2c1_production_adapter
runtime = v2r2c1_production_adapter.execute_request({"schema": "aios.production-executor-request.v1", "operation": "apply", **value})
if runtime.get("status") != "ACTIVE_VERIFIED":
self.state.fail("RECONCILIATION_REQUIRED", "production health did not verify")
raise ReconciliationRequired("production health did not verify")
self.state.record("PRODUCTION_VERIFIED", "production", runtime)
return {"status": "PRODUCTION_VERIFIED", "runtime": runtime, "artifact": artifact, "tag_oid": value["tag_oid"]}
def consistency(self, request: Mapping[str, Any]) -> dict[str, Any]:
publication = self._request(request)
result = self.publisher.get_release_consistency(publication)
result["release_binding"] = self.binding.as_dict()
result["production_health"] = _health(self.config.health_url, self.binding, self.state.value.get("completed", {}).get("production_artifact", {}))
result["registry"] = _read_json(Path(self.config.registry_root) / f"{VERSION}.json")
result["baseline"] = _read_json(Path(self.config.baseline_path))
return result
def acceptance(self, request: Mapping[str, Any]) -> dict[str, Any]:
result = self.consistency(request)
result["accepted"] = result.get("status") == "PASS" and result.get("production_health", {}).get("status") == "ACTIVE_VERIFIED"
self.state.record("ACCEPTED" if result["accepted"] else "RECONCILIATION_REQUIRED", "acceptance", result)
return result
def _read_json(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return {"status": "MISSING", "path": str(path)}
if not isinstance(value, dict):
return {"status": "INVALID", "path": str(path)}
return {"status": "PRESENT", "path": str(path), **value}
def _health(url: str, binding: ReleaseBinding, artifact: Mapping[str, Any]) -> dict[str, Any]:
import urllib.request
try:
with urllib.request.urlopen(url, timeout=5) as response:
body = json.loads(response.read().decode("utf-8"))
status = response.status
except Exception as exc:
return {"status": "UNAVAILABLE", "error": type(exc).__name__}
expected_artifact = artifact.get("artifact_sha256")
exact = (
status == 200
and body.get("version") == binding.version
and body.get("commit_oid") == binding.target_oid
and (not expected_artifact or body.get("artifact_sha256") == expected_artifact)
and body.get("deployment_state") == "ACTIVE"
and body.get("production_deployed") is True
)
return {"status": "ACTIVE_VERIFIED" if exact else "CONFLICT", "http_status": status, **body}
def build_executor_config(config: Any) -> GovernedProductionReleaseExecutor:
return GovernedProductionReleaseExecutor(config)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("operation", choices=("build-artifact", "status"))
parser.add_argument("--git-dir", type=Path, default=Path("/data/aios-gitops/repo.git"))
parser.add_argument("--output-root", type=Path, default=Path("/var/lib/aios-gitops/approved-artifacts/V2R2C1"))
args = parser.parse_args()
binding = ReleaseBinding()
if args.operation == "build-artifact":
print(json.dumps(build_production_artifact(args.git_dir, binding, args.output_root), sort_keys=True))
else:
print(json.dumps({"binding": binding.as_dict(), "status": "READY"}, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,135 @@
"""V2R2C1 adapter using the existing V2R1C7 deployment primitives.
The service identity and loopback port remain stable for a safe cutover. The
new release is installed under its own immutable directory and keeps the
previous V2R1C7 release available for an explicit, owner-approved rollback.
"""
from __future__ import annotations
import json
import os
import shutil
import sys
from argparse import Namespace
from pathlib import Path
from typing import Any, Mapping
import v2r1c7_production_adapter as base
VERSION = "V2R2C1"
SERVICE = "aios-v2r1c7-production.service"
PORT = 18927
RELEASE_ROOT = Path("/opt/aios/releases")
CURRENT_LINK = Path("/opt/aios/v2r2c1")
ENV_FILE = Path("/etc/aios/aios-v2r2c1.env")
STATE_FILE = Path("/var/lib/aios/v2r2c1/deployment-state.json")
UNIT_FILE = Path("/etc/systemd/system/aios-v2r1c7-production.service")
ROLLBACK_ROOT = Path("/var/lib/aios/v2r2c1/rollback")
REQUIRED_MEMBERS = {
"payload/VERSION",
"payload/active-runtime/app/deployment_state.py",
"payload/active-runtime/app/v2r1c6_privacy_tools.py",
"payload/active-runtime/app/v2r2c1_health_service.py",
"payload/deploy/aios-v2/systemd/aios-v2r1c7-production.service",
"payload/deploy/aios-v2/v2r2c1-runtime-config.json",
"payload/deploy/aios-v2/v2r2c1-rollback.json",
"payload/scripts/probe_v2r2c1_health.py",
"payload/scripts/rollback_v2r2c1_release.sh",
"payload/BUILD-METADATA.json",
"payload/ARTIFACT-MANIFEST.json",
"payload/ROLLBACK-METADATA.json",
}
def _configure_base() -> None:
base.VERSION = VERSION
base.SERVICE = SERVICE
base.PORT = PORT
base.REQUIRED_MEMBERS = REQUIRED_MEMBERS
def _args() -> Namespace:
return Namespace(
release_root=RELEASE_ROOT,
current_link=CURRENT_LINK,
env_file=ENV_FILE,
state_file=STATE_FILE,
unit_file=UNIT_FILE,
health_timeout=45,
)
def load_request() -> dict[str, Any]:
_configure_base()
value = base.load_request()
if value.get("version") != VERSION or value.get("production_service") != SERVICE or int(value.get("production_port", -1)) != PORT:
raise ValueError("V2R2C1 production binding mismatch")
return value
def _backup(path: Path, name: str) -> None:
if path.is_file() and not path.is_symlink():
ROLLBACK_ROOT.mkdir(parents=True, exist_ok=True)
destination = ROLLBACK_ROOT / name
if not destination.exists():
shutil.copy2(path, destination)
def execute_request(value: Mapping[str, Any]) -> dict[str, Any]:
_configure_base()
if value.get("schema") != "aios.production-executor-request.v1":
raise ValueError("request_schema_invalid")
if value.get("operation") == "apply":
_backup(UNIT_FILE, "aios-v2r1c7-production.service.previous")
_backup(Path("/etc/aios/aios-v2r1c7.env"), "aios-v2r1c7.env.previous")
_backup(Path("/var/lib/aios/v2r1c7/deployment-state.json"), "v2r1c7-deployment-state.previous")
return base.install_release(_args(), value)
if value.get("operation") == "probe":
return base.probe(_args(), value)
if value.get("operation") == "rollback":
return rollback(value)
raise ValueError("unsupported_operation")
def rollback(value: Mapping[str, Any]) -> dict[str, Any]:
"""Explicit rollback only; this adapter never invokes it automatically."""
import subprocess
subprocess.run(["systemctl", "stop", SERVICE], check=False, capture_output=True, text=True, timeout=45)
previous_unit = ROLLBACK_ROOT / "aios-v2r1c7-production.service.previous"
previous_env = ROLLBACK_ROOT / "aios-v2r1c7.env.previous"
previous_state = ROLLBACK_ROOT / "v2r1c7-deployment-state.previous"
if not previous_unit.is_file() or not previous_env.is_file():
raise RuntimeError("ROLLBACK_RECEIPT_MISSING")
UNIT_FILE.parent.mkdir(parents=True, exist_ok=True)
ENV_FILE.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(previous_unit, UNIT_FILE)
shutil.copy2(previous_env, Path("/etc/aios/aios-v2r1c7.env"))
if previous_state.is_file():
Path("/var/lib/aios/v2r1c7").mkdir(parents=True, exist_ok=True)
shutil.copy2(previous_state, Path("/var/lib/aios/v2r1c7/deployment-state.json"))
subprocess.run(["systemctl", "daemon-reload"], check=True, capture_output=True, text=True, timeout=45)
subprocess.run(["systemctl", "restart", SERVICE], check=True, capture_output=True, text=True, timeout=45)
result = base.health("http://127.0.0.1:18927/health", {
"version": "V2R1C7",
"target_oid": str(json.loads(Path("/var/lib/aios/v2r1c7/deployment-state.json").read_text(encoding="utf-8")).get("commit_oid")),
"artifact_sha256": str(json.loads(Path("/var/lib/aios/v2r1c7/deployment-state.json").read_text(encoding="utf-8")).get("artifact_sha256")),
})
return {"status": "ROLLED_BACK", "rollback_health": result, "automatic_rollback": False}
def main() -> int:
try:
value = load_request()
result = execute_request(value)
print(json.dumps(result, sort_keys=True))
return 0
except Exception as exc:
print(json.dumps({"status": "FAIL_CLOSED", "error": type(exc).__name__, "reason": str(exc)}, sort_keys=True))
return 2
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,23 @@
[Unit]
Description=AIOS ChatGPT Bridge v2 staging control plane
After=network-online.target
[Service]
Type=simple
User=aios-bridge-staging
Group=aios-bridge-staging
WorkingDirectory=/opt/aios/gitops-control-plane/current
Environment=AIOS_BRIDGE_SCHEMA_VERSION=aios.chatgpt-gitops-bridge.v2
Environment=AIOS_GITOPS_WRITES_ENABLED=false
Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
Environment=AIOS_DEPLOYMENT_STATE=STAGING
ExecStart=/usr/bin/python3 /opt/aios/gitops-control-plane/current/bridge-v2-staging-server.py
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/var/lib/aios-gitops/bridge-v2-staging
Restart=on-failure
[Install]
WantedBy=multi-user.target
@@ -1,6 +1,20 @@
# Merge this fragment into docs/gitops/openapi-chatgpt-actions.yaml.
components:
schemas:
WriterLeaseRequestV2:
type: object
additionalProperties: false
required: [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds]
properties:
schema: {type: string, const: aios.task-writer-lease.v2}
task_id: {type: string}
repository: {type: string, const: aios/aios}
branch_mode: {type: string, const: CONTINUE_EXISTING}
base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/'}
generation: {type: integer, minimum: 1}
lease_seconds: {type: integer, minimum: 1, maximum: 86400}
AuthorizedTaskWriterProvisionRequest:
type: object
additionalProperties: false
+239 -2
View File
@@ -13,6 +13,55 @@ components:
scheme: bearer
bearerFormat: opaque-token
schemas:
TaskAdmissionEnvelopeRequest:
type: object
additionalProperties: false
required:
- idempotency_key
- task_id
- repository
- authority_base_oid
- target_branch
- generation
- execution_digest
- requested_capabilities
- file_allowlist
- risk
properties:
idempotency_key: {type: string, minLength: 8, maxLength: 256}
task_id: {type: string, minLength: 3, maxLength: 180}
repository: {type: string, enum: [aios/aios]}
authority_base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/.+'}
generation: {type: integer, minimum: 1}
execution_digest: {type: string, pattern: '^[0-9a-f]{64}$'}
requested_capabilities:
type: array
minItems: 1
uniqueItems: true
items: {type: string}
file_allowlist:
type: array
minItems: 1
uniqueItems: true
items: {type: string}
risk: {type: string, enum: [low, medium]}
expires_in_seconds:
type: integer
minimum: 60
maximum: 900
TaskAdmissionAuthorizationRequest:
type: object
additionalProperties: false
required: [envelope_id, authorization_kind]
properties:
envelope_id: {type: string}
authorization_kind:
type: string
enum: [delegation_epoch, owner_signature]
delegation_epoch_id: {type: string}
owner_key_id: {type: string}
signature_base64: {type: string}
PublicationState:
type: string
description: Canonical durable publication state. Legacy TAG_CREATED_LOCAL is normalized on read and is never emitted by new writes.
@@ -96,7 +145,7 @@ components:
message: {type: string}
TagRequest:
type: object
required: [version, target_oid, evidence_closure_oid, approval_plan_id]
required: [version, target_oid, evidence_closure_oid, approval_plan_id, execute]
properties:
version: {type: string, example: V1R1C1}
target_oid:
@@ -241,6 +290,122 @@ components:
properties:
candidate_id: {type: string}
paths:
/v2/task-admission/envelope:
post:
operationId: getTaskAdmissionEnvelope
summary: Build and persist a canonical narrow Task admission envelope
description: >
Authenticated transport only. This operation does not mint a TaskPrincipal,
acquire a WriterLease, or mutate Git. It normalizes and persists one exact
admission envelope for later delegation/Owner authorization.
security: [{bearerAuth: []}]
x-openai-isConsequential: false
x-aios-governed-task-admission: true
x-aios-no-shell: true
x-aios-no-git-write: true
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/TaskAdmissionEnvelopeRequest'}
responses:
'200': {description: Canonical envelope and signing payload}
'401': {description: Authenticated transport required}
'403': {description: Requested scope exceeds bootstrap policy}
'409': {description: Idempotency key conflicts with a different request}
/v2/task-admission/authorize:
post:
operationId: submitTaskAdmissionAuthorization
summary: Verify durable delegation or Owner signature and admit one exact Task
description: >
The bearer authenticates the transport caller but is not task authority.
Authorization is proven only by a verified OwnerDelegationEpoch or exact
Owner signature. Successful admission may persist an exact Task receipt in
QUEUED, CLAIMED, or RUNNING state. It does not enable legacy/global Git
writes and cannot accept a caller-supplied TaskPrincipal.
security: [{bearerAuth: []}]
x-openai-isConsequential: true
x-aios-governed-task-admission: true
x-aios-no-shell: true
x-aios-no-git-write: true
requestBody:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/TaskAdmissionAuthorizationRequest'}
responses:
'200': {description: Exact admitted Task receipt or idempotent replay}
'401': {description: Authenticated transport required}
'403': {description: Invalid, expired, revoked, widened, or unbound authorization}
'409': {description: Conflicting idempotent admission}
'503': {description: Task state or authorization backend unavailable}
/v2/task-admission/status:
get:
operationId: getTaskAdmissionStatus
summary: Read durable Task-admission state
description: >
Authenticated bounded lookup of one admission envelope. This is status only
and confers no TaskPrincipal, WriterLease, or execution capability.
security: [{bearerAuth: []}]
x-openai-isConsequential: false
x-aios-governed-task-admission: true
x-aios-no-shell: true
x-aios-read-only: true
parameters:
- in: query
name: envelope_id
required: true
schema: {type: string, minLength: 1, maxLength: 256}
responses:
'200': {description: Pending or admitted receipt}
'401': {description: Authenticated transport required}
'404': {description: Envelope not found}
/createTagApprovalRequest:
post:
operationId: createTagApprovalRequest
summary: Record the exact Owner-approved release binding without executing it
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [version, target_oid, evidence_closure_oid, approval_plan_id, owner_approval, execute]
responses: {'200': {description: Durable execute=false approval receipt}}
x-openai-isConsequential: true
/buildProductionArtifact:
post:
operationId: buildProductionArtifact
summary: Build the deterministic production artifact for an exact approved binding
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [version, target_oid, evidence_closure_oid, approval_plan_id]
responses: {'200': {description: Artifact digest and closure receipt}}
x-openai-isConsequential: true
/executeApprovedProductionRelease:
post:
operationId: executeApprovedProductionRelease
summary: Execute the exact approved production release after all release gates pass
security: [{bearerAuth: []}]
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [version, target_oid, evidence_closure_oid, approval_plan_id, execute]
responses: {'200': {description: Production release result}}
x-openai-isConsequential: true
/createReleaseIntent:
post:
operationId: createReleaseIntent
@@ -490,6 +655,14 @@ paths:
summary: Read durable publication Saga status without mutation
parameters:
- {in: query, name: version, required: true, schema: {type: string, pattern: '^V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$'}}
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [version]
responses:
'200':
description: Canonical publication state or fail-closed UNKNOWN
@@ -506,6 +679,14 @@ paths:
- {in: query, name: target_oid, required: true, schema: {type: string, pattern: '^[0-9a-fA-F]{40}$'}}
- {in: query, name: evidence_closure_oid, required: true, schema: {type: string, pattern: '^[0-9a-fA-F]{40}$'}}
- {in: query, name: approval_plan_id, required: true, schema: {type: string}}
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [version, target_oid, evidence_closure_oid, approval_plan_id]
responses: {'200': {description: PASS, FAIL, or fail-closed UNKNOWN}}
/api/release-acceptance:
get:
@@ -516,6 +697,14 @@ paths:
- {in: query, name: target_oid, required: true, schema: {type: string, pattern: '^[0-9a-fA-F]{40}$'}}
- {in: query, name: evidence_closure_oid, required: true, schema: {type: string, pattern: '^[0-9a-fA-F]{40}$'}}
- {in: query, name: approval_plan_id, required: true, schema: {type: string}}
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [version, target_oid, evidence_closure_oid, approval_plan_id]
responses: {'200': {description: Acceptance result with accepted boolean}}
/api/tag:
post:
@@ -526,7 +715,10 @@ paths:
required: true
content:
application/json:
schema: {$ref: '#/components/schemas/TagRequest'}
schema:
type: object
additionalProperties: false
required: [version, target_oid, evidence_closure_oid, approval_plan_id, execute]
responses: {'200': {description: Gitea-confirmed tag and fail-closed baseline result}, '400': {description: Missing governance evidence or invalid OID}, '403': {description: Forbidden}, '409': {description: Existing tag target conflict or evidence lineage mismatch}, '502': {description: Gitea reconciliation required}, '503': {description: Remote push pending}}
x-openai-isConsequential: true
/api/rollback:
@@ -567,3 +759,48 @@ paths:
'200': {description: Safe GitOps plan}
'401': {description: Unauthorized}
x-openai-isConsequential: true
/api/v2/governed-capabilities:
get:
operationId: getGovernedWriteCapabilities
responses: {'200': {description: Governed write capabilities with legacy writes disabled}}
/api/v2/governed-task:
post:
operationId: createGovernedTask
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [task_id, repository, base_oid, target_branch, risk, description, idempotency_key]}}}}
responses: {'200': {description: Exact governed task binding}}
x-openai-isConsequential: true
/api/v2/writer-lease:
post:
operationId: acquireWriterLease
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds]}}}}
responses: {'200': {description: Detached CONTINUE_EXISTING lease}}
x-openai-isConsequential: true
/api/v2/commit-file:
post:
operationId: commitFileGoverned
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, path, content_base64, message]}}}}
responses: {'202': {description: Durable governed Job}}
x-openai-isConsequential: true
/api/v2/commit-files:
post:
operationId: commitFilesGoverned
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message]}}}}
responses: {'202': {description: One atomic multi-file governed Job}}
x-openai-isConsequential: true
/api/v2/job:
post:
operationId: getGitOpsJob
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [job_id]}}}}
responses: {'200': {description: Non-sensitive Job metadata}}
/api/v2/job/resume:
post:
operationId: resumeGitOpsJob
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [job_id, expected_revision]}}}}
responses: {'202': {description: Revision-CAS resume}}
x-openai-isConsequential: true
/api/v2/job/reconcile:
post:
operationId: reconcileGitOpsJob
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [job_id, expected_revision]}}}}
responses: {'202': {description: Observation-first reconciliation}}
x-openai-isConsequential: true
@@ -0,0 +1,15 @@
# Bridge v2.1 Durable Reconciliation Bootstrap
This is the planning-only bootstrap artifact for
`TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809`.
The task is bound to the exact base commit
`3f670c387692b8865a10d5b419468d8a0b64d7a4` and the development branch
`agent/p0-bridge-v21-durable-reconciliation-20260809`.
The first governed commit records the design boundary and acceptance contract
only. It does not implement the v2.1 body, mutate main, create tags, change
Baseline, or touch production. Future work must preserve absent-ref CAS,
observation-first recovery, durable JobStore state, exact Authority/Gitea/API
reconciliation, and the existing secret-scanning and credential-projection
boundaries.
@@ -0,0 +1,150 @@
# ChatGPT Bridge v2 Control Plane Convergence Design
Date: 2026-08-09
Task: `TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809`
Repository: `aios/aios`
Approved development base: `2e8fa22989808a448065d1d76b5b18948774a637`
Target branch: `agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809`
## Scope and authority
This design converges the ChatGPT GitOps Bridge, the existing Authorized Task Writer, the continuation lease protocol, the Durable GitOps Job Controller, and immutable control-plane runtime identity.
The development base is exact and must never be replaced with a dynamically resolved `HEAD`, `latest`, or `main`. If Authority `refs/heads/main` differs from the approved OID before the first governed writer push, execution fails with `BASE_OID_CHANGED`.
This task does not authorize production changes. The following remain immutable:
- Production Baseline `V2R1C7`.
- Production target OID `c7adece1cc8beffe7c718d53c775898b45d6c7f8`.
- Production tag OID `82792347937c0df59a8e02d6162775252bcad2e2`.
- Production Registry, services, deployment state, runtime, tags, and release evidence.
- Authority `main`.
Force push, history rewrite, direct main push, production mutation, tag mutation, baseline mutation, and secret commits are always denied.
## Immutable control-plane runtime identity
A control-plane release is installed into a versioned immutable directory equivalent to:
```text
/opt/aios/gitops-control-plane/<release-id-or-oid>/
```
The release manifest binds:
```text
control_plane_release_version
control_plane_target_oid
artifact_sha256
runtime_path
bridge_schema_version
repository_identity
authoritative_git_dir
deployment_state
```
The validator is independent from the mutable Git workspace. It rejects execution when the runtime path is a workspace/worktree, the observed runtime commit or artifact digest differs from the manifest, the repository identity is not `aios/aios`, the authoritative Git directory is absent or mismatched, or the Bridge schema version is absent or unsupported.
Every failure is fail closed with `CONTROL_PLANE_RUNTIME_DRIFT`. No fallback to a mutable workspace is permitted. The installer and systemd template are staging-safe; this task may install and restart only a staging/test service.
## Bridge v2 adapter
The Bridge schema identity is `aios.chatgpt-gitops-bridge.v2`. It is independent from the production baseline, control-plane release version, and runtime release commit.
The Bridge exposes:
- `getRuntimeStatus`
- `getGovernedWriteCapabilities`
- `createGovernedTask`
- `acquireWriterLease`
- `commitFileGoverned`
- `commitFilesGoverned`
- `getGitOpsJob`
- `resumeGitOpsJob`
- `reconcileGitOpsJob`
The adapter reuses existing mechanisms and does not create a second writer, lease protocol, branch state store, job database, or direct Git mutation path.
## Governed task and lease flow
`createGovernedTask` binds an exact task ID, repository, approved base OID, non-main branch, risk, and description. Existing branches are read back and bound to their observed full OID. Idempotent replay returns the original binding; conflicting replay fails closed.
`acquireWriterLease` uses `WriterLeaseRequestV2`. Continuation uses:
```text
branch_mode=CONTINUE_EXISTING
expected_branch_oid=<full OID>
generation=<monotonic integer>
create_target_branch=false
worktree_mode=detached_at_expected_head
cas_ref_update_required=true
```
Credentials remain server-side and never appear in responses, Git, durable job metadata, or logs.
## Governed commit flow
`commitFileGoverned` and `commitFilesGoverned` submit work to the existing `GitOpsJobController` and `JobStore`. The server computes payload digests. Raw file content is stored only in the content-addressed payload domain. Secret scanning, protected-path policy, repository binding, lease validation, and branch authorization run before side effects.
`commitFilesGoverned` is one atomic Git commit and enforces bounded file count, per-file size, and total payload size. Any failure leaves no partial commit or ref update.
All ref updates use compare-and-swap. Stale `expected_branch_oid` returns conflict and never triggers automatic rebase. After push, Authority and remote Gitea refs are independently read and must equal the resulting commit OID.
## Durable job lifecycle
Job reads return only non-sensitive metadata: `job_id`, `state`, `revision`, `side_effect_state`, `retryable`, `binding`, and `receipt_summary`.
Terminal jobs cannot be changed by ordinary resume. Resume requires `job_id` and `expected_revision` and uses revision CAS. Reconciliation is observation-first: for `UNKNOWN`, `RECONCILIATION_REQUIRED`, or `REMOTE_PUSH_PENDING`, local, Authority, and remote refs are read before state changes. A blind repeated push is never reconciliation.
## Legacy behavior
Legacy writes remain disabled while governed writes are enabled:
```text
legacy_writes_enabled=false
governed_writes_enabled=true
```
The global `AIOS_GITOPS_WRITES_ENABLED` flag is never enabled as the solution. Legacy operations remain temporarily visible for compatibility and diagnostics, but return:
```json
{
"error": "LEGACY_WRITE_DISABLED",
"replacement": "createGovernedTask",
"governed_writes_available": true
}
```
## OpenAPI convergence
`docs/gitops/openapi-chatgpt-actions.yaml` is the canonical Bridge v2 schema and integrates the Authorized Task Writer fragment. The canonical CAS field is `expected_branch_oid`. `expected_head_oid` may be accepted only at an explicit deprecated adapter boundary.
Contract tests require exact equality between OpenAPI required fields, runtime parser required fields, and generated GPT Action schema required fields.
## Acceptance
Automated coverage includes runtime path/OID/digest/repository drift, exact task/branch/OID/lease/CAS behavior, job DB failure, restart during remote verification, remote timeout, response loss after remote success, idempotency replay/conflict, concurrent branch movement, unauthorized or main branch writes, protected paths, secret-like payloads, disabled governed writer, missing/expired lease, wrong repository identity, and observation-first reconciliation.
The staging sequence is:
```text
getRuntimeStatus
getGovernedWriteCapabilities
createGovernedTask
acquireWriterLease
commitFilesGoverned
getGitOpsJob
verify Authority branch ref
verify remote Gitea branch ref
restart staging control-plane service
getRuntimeStatus
acquire continuation lease
second governed commit
verify remote ref again
```
Completion requires all tests, staging restart continuation, Authority/remote OID equality, unchanged production invariants, no force push, no tag rewrite, no direct main push, and no secret commit.
## Implementation boundaries
Implementation is limited to the Bridge, existing writer and job-controller adapters, runtime identity/installer, staging systemd template, OpenAPI documents, tests, and engineering evidence. Hermes Adapter integration and production cutover are outside this task and require separate approval.
@@ -0,0 +1,44 @@
{
"schema": "aios.bridge-v2-engineering-staging-acceptance.v1",
"task_id": "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809",
"repository": "aios/aios",
"branch": "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809",
"control_plane_release_version": "bridge-v2-aa0c37b8ab33",
"control_plane_target_oid": "aa0c37b8ab333616e1bab9686cf6d10557f2ebda",
"artifact_sha256": "08670c78cf760bfe693e3553e676eb751dadd7834116aa6d40e7c094f1357ae4",
"runtime_path": "/opt/aios/gitops-control-plane/aa0c37b8ab333616e1bab9686cf6d10557f2ebda",
"bridge_schema_version": "aios.chatgpt-gitops-bridge.v2",
"deployment_state": "STAGING",
"drift_status": "PASS",
"restart_acceptance": {
"pid_before": 3158850,
"pid_after": 3159889,
"runtime_response_equal": true
},
"governed_capabilities": {
"legacy_writes_enabled": false,
"governed_writes_enabled": true,
"force_push": false,
"main_write": false,
"tag_write": false,
"baseline_write": false,
"production_write": false
},
"production_invariants": {
"authority_main_oid": "2e8fa22989808a448065d1d76b5b18948774a637",
"authority_tags_sha256": "bd2b6e30b76f28e7fcdcb447e7296690f0a65ad855209294c36b0f3f1d96098d",
"baseline_current_sha256": "57651eb819584ab7c1d67f01b0ef66d6408cdc322ae5f6d4ff0aacd96f8327e1",
"production_service_main_pid": 2816107,
"production_service_nrestarts": 0,
"production_service_active_enter_timestamp_monotonic": 2919386492273
},
"bridge_operation_sequence": {
"createGovernedTask": "PASS",
"acquireWriterLease": "PASS",
"commitFilesGoverned": "REAL_GOVERNED_COMMIT",
"getGitOpsJob": "PASS",
"expected_branch_oid": "8fe3b85c1be592326473e81a502a23eeb9511d30",
"generation": 7
},
"production_changed": false
}
@@ -0,0 +1,17 @@
{
"schema": "aios.bridge-v2.1-durable-reconciliation.bootstrap-acceptance.v1",
"task_id": "TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
"repository": "aios/aios",
"base_oid": "3f670c387692b8865a10d5b419468d8a0b64d7a4",
"target_branch": "agent/p0-bridge-v21-durable-reconciliation-20260809",
"branch_mode": "CREATE_NEW",
"generation": 1,
"implementation": "not_started",
"production_changed": false,
"required_proofs": [
"absent_ref_cas",
"observation_first_reconciliation",
"authority_gitea_api_commit_equality",
"commit_visible"
]
}
@@ -2,7 +2,7 @@ from __future__ import annotations
import hashlib,json,re,tarfile
from pathlib import Path
from typing import Any
OID_RE=re.compile(r"^[0-9a-f]{40}$"); SHA_RE=re.compile(r"^[0-9a-f]{64}$"); VERSION_RE=re.compile(r"^V2R1C[0-9]+$")
OID_RE=re.compile(r"^[0-9a-f]{40}$"); SHA_RE=re.compile(r"^[0-9a-f]{64}$"); VERSION_RE=re.compile(r"^V[0-9]+R[0-9]+C[0-9]+(?:-[A-Za-z0-9_.-]+)?$")
REQUIRED_SOURCE={"schema","version","repository","base_ref","base_oid","binding_mode","release_readiness","production_deployed","tag_created","baseline_changed","production_service","production_port","rollback_metadata","controller_schema","required_checks"}
REQUEST_KEYS={"schema","version","repository","base_oid","target_oid","artifact_uri","artifact_sha256","artifact_size","evidence_closure_oid","approval_plan_id","execute","production_deployed","production_service","production_port","rollback_metadata","current_tag_oid","source_manifest_sha256"}
class ManifestValidationError(ValueError):
@@ -62,7 +62,7 @@ def validate_release_request(source_manifest:dict[str,Any],request_path:Path,exp
if not isinstance(request["approval_plan_id"],str) or not request["approval_plan_id"].startswith("sha256:"): raise ManifestValidationError("APPROVAL_PLAN","invalid approval plan")
if request["execute"] is not False or request["production_deployed"] is not False: raise ManifestValidationError("EXECUTION_NOT_ALLOWED","must be non-executing")
if request["production_service"]!=source_manifest["production_service"] or request["production_port"]!=source_manifest["production_port"]: raise ManifestValidationError("SERVICE_IDENTITY","request service mismatch")
if not isinstance(request["current_tag_oid"],str) or request["current_tag_oid"] not in {"V2R1C6","9e65bd59baea392294c0f253046d3ab116f63145"}: raise ManifestValidationError("PROTECTED_TAG","previous tag binding changed")
if not isinstance(request["current_tag_oid"],str) or request["current_tag_oid"] not in {"V2R1C7","82792347937c0df59a8e02d6162775252bcad2e2"}: raise ManifestValidationError("PROTECTED_TAG","previous production tag binding changed")
_sha(request["source_manifest_sha256"],"source_manifest_sha256"); return request
def build_approval_request(*,source_manifest_path:Path,target_oid:str,artifact_uri:str,artifact_sha256:str,artifact_size:int,evidence_closure_oid:str,rollback_metadata:str,current_tag_oid:str)->dict[str,Any]:
source=_json_object(source_manifest_path); _oid(target_oid,"target_oid"); _sha(artifact_sha256,"artifact_sha256"); _oid(evidence_closure_oid,"evidence_closure_oid"); request={"schema":"aios.release-request.v2","version":source["version"],"repository":source["repository"],"base_oid":source["base_oid"],"target_oid":target_oid,"artifact_uri":artifact_uri,"artifact_sha256":artifact_sha256,"artifact_size":artifact_size,"evidence_closure_oid":evidence_closure_oid,"execute":False,"production_deployed":False,"production_service":source["production_service"],"production_port":source["production_port"],"rollback_metadata":rollback_metadata,"current_tag_oid":current_tag_oid,"source_manifest_sha256":sha256_file(source_manifest_path)}; request["approval_plan_id"]="sha256:"+sha256_bytes(_canonical(request)); return request
@@ -0,0 +1,90 @@
import base64
from pathlib import Path
from aios_authority_git_executor import AuthorityGitExecutor
from aios_chatgpt_bridge_v2_server import BridgeHandler, RuntimeConfig
from aios_continuation_executor import GiteaContinuationRemote, ServerSideCredentialProjection
from aios_continuation_lease import WriterLeaseRequestV2
class Probe:
def __init__(self, present=False):
self.present = present
def branch_exists(self, _branch):
return self.present
def branch_head(self, _branch):
return "a" * 40
def is_ancestor(self, _older, _newer):
return True
def commit_exists(self, _oid):
return True
def test_create_new_lease_is_exact_base_and_absent():
request = WriterLeaseRequestV2(
task_id="TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
repository="aios/aios",
branch_mode="CREATE_NEW",
base_oid="a" * 40,
expected_branch_oid="a" * 40,
target_branch="agent/p0-bridge-v21-durable-reconciliation-20260809",
generation=1,
)
request.validate(Probe())
def test_create_new_lease_rejects_existing_target():
request = WriterLeaseRequestV2(
task_id="task",
repository="aios/aios",
branch_mode="CREATE_NEW",
base_oid="a" * 40,
expected_branch_oid="a" * 40,
target_branch="agent/existing-target",
generation=1,
)
try:
request.validate(Probe(True))
except Exception as exc:
assert str(exc) == "TARGET_BRANCH_ALREADY_EXISTS"
else:
raise AssertionError("existing CREATE_NEW target must fail")
def test_slash_branch_uses_paginated_collection():
credential = ServerSideCredentialProjection(Path("/tmp/key"), Path("/tmp/token"), active=True)
calls = []
class FakeRemote(GiteaContinuationRemote):
def _api(self, path):
calls.append(path)
if "page=1" in path:
return [{"name": f"agent/other-{i}", "commit": {"id": "b" * 40}} for i in range(50)]
return [{"name": "agent/slash/branch", "commit": {"id": "c" * 40}}]
remote = FakeRemote(Path("/tmp"), credential)
assert remote.read_api_branch("agent/slash/branch") == "c" * 40
assert calls[0].endswith("branches?page=1&limit=50")
assert calls[1].endswith("branches?page=2&limit=50")
def test_http_route_contract_is_complete():
expected = {
"/v2/runtime", "/v2/capabilities", "/v2/governed-task",
"/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
"/v2/job", "/v2/job/resume", "/v2/job/reconcile",
"/v2/tag/approval-request",
"/v2/tag/publish",
"/v2/tag/status",
"/v2/release/artifact",
"/v2/release/execute",
"/v2/release/consistency",
"/v2/release/acceptance",
}
from aios_chatgpt_bridge_v2_server import ROUTES
assert set(ROUTES) == expected
+1 -1
View File
@@ -264,7 +264,7 @@ class TagActionTests(unittest.TestCase):
ROOT / "docs" / "gitops" / "openapi-chatgpt-actions.yaml"
).read_text(encoding="utf-8")
self.assertIn(
"required: [version, target_oid, evidence_closure_oid, approval_plan_id]",
"required: [version, target_oid, evidence_closure_oid, approval_plan_id, execute]",
schema,
)
self.assertIn("x-openai-isConsequential: true", schema)
+87
View File
@@ -0,0 +1,87 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import tempfile
import pytest
import yaml
from aios_chatgpt_bridge_v2 import (
BRIDGE_OPERATIONS,
BRIDGE_SCHEMA_VERSION,
OPERATION_REQUIRED_FIELDS,
BridgeV2Adapter,
ControlPlaneReleaseManifest,
RuntimeDriftError,
)
ROOT = Path(__file__).parents[1]
def test_immutable_release_identity_accepts_exact_artifact_and_rejects_workspace_drift():
with tempfile.TemporaryDirectory() as directory:
approved = Path(directory) / "releases"
runtime = approved / "release-1"
runtime.mkdir(parents=True)
artifact = runtime / "bridge.pyz"
artifact.write_bytes(b"bridge-v2")
digest = hashlib.sha256(artifact.read_bytes()).hexdigest()
manifest = ControlPlaneReleaseManifest(
"bridge-v2-test", "a" * 40, digest, runtime, BRIDGE_SCHEMA_VERSION,
"aios/aios", Path("/data/aios-gitops/repo.git"), "STAGING",
)
assert manifest.validate(artifact, observed_oid="a" * 40, approved_root=approved)["drift_status"] == "PASS"
with pytest.raises(RuntimeDriftError, match="CONTROL_PLANE_RUNTIME_DRIFT"):
manifest.validate(artifact, observed_oid="b" * 40, approved_root=approved)
mutable = ControlPlaneReleaseManifest(
"bad", "a" * 40, digest, Path("/var/lib/worktrees/dev"), BRIDGE_SCHEMA_VERSION,
"aios/aios", Path("/data/aios-gitops/repo.git"), "STAGING",
)
with pytest.raises(RuntimeDriftError, match="CONTROL_PLANE_RUNTIME_DRIFT"):
mutable.validate(artifact, observed_oid="a" * 40, approved_root=approved)
def test_bridge_capabilities_keep_legacy_closed_and_expose_exact_operations():
adapter = BridgeV2Adapter(runtime_status={"drift_status": "PASS"})
assert adapter.getRuntimeStatus()["bridge_schema_version"] == BRIDGE_SCHEMA_VERSION
capabilities = adapter.getGovernedWriteCapabilities()
assert capabilities["legacy_writes_enabled"] is False
assert capabilities["governed_writes_enabled"] is True
assert tuple(capabilities["operations"]) == BRIDGE_OPERATIONS
assert adapter.legacy_write_disabled() == {
"error": "LEGACY_WRITE_DISABLED",
"replacement": "createGovernedTask",
"governed_writes_available": True,
}
def test_openapi_runtime_and_fragment_required_fields_are_exactly_converged():
canonical = yaml.safe_load((ROOT / "docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8"))
fragment = yaml.safe_load((ROOT / "docs/gitops/openapi-authorized-task-writer-fragment.yaml").read_text(encoding="utf-8"))
operations = {}
for path, methods in canonical["paths"].items():
for method, value in methods.items():
if isinstance(value, dict) and value.get("operationId") in BRIDGE_OPERATIONS:
operations[value["operationId"]] = value
assert set(operations) == set(BRIDGE_OPERATIONS)
for operation, required in OPERATION_REQUIRED_FIELDS.items():
schema = operations[operation]["requestBody"]["content"]["application/json"]["schema"]
assert tuple(schema["required"]) == required
fragment_required = tuple(fragment["components"]["schemas"]["WriterLeaseRequestV2"]["required"])
assert fragment_required == OPERATION_REQUIRED_FIELDS["acquireWriterLease"]
def test_release_manifest_round_trip_contains_only_declared_identity_fields():
manifest = ControlPlaneReleaseManifest(
"v", "a" * 40, "b" * 64, Path("/opt/aios/gitops-control-plane/v"), BRIDGE_SCHEMA_VERSION,
"aios/aios", Path("/data/aios-gitops/repo.git"), "STAGING",
)
value = manifest.as_dict()
assert tuple(value) == (
"control_plane_release_version", "control_plane_target_oid", "artifact_sha256", "runtime_path",
"bridge_schema_version", "repository_identity", "authoritative_git_dir", "deployment_state",
)
assert "credential" not in json.dumps(value).lower()
@@ -0,0 +1,283 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from pathlib import Path
import subprocess
import tempfile
import pytest
from aios_continuation_executor import (
ContinuationCommit,
ContinuationExecutionError,
ContinuationExecutor,
ContinuationPermissions,
ContinuationRuntimeBinding,
DetachedGitWorkspace,
GiteaContinuationRemote,
)
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import BindingConflict, GitOpsJobController, JobState, JobStore
TASK = "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809"
REPOSITORY = "aios/aios"
BRANCH = "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809"
BASE = "2" * 40
HEAD = "8" * 40
NEW = "9" * 40
NOW = datetime(2026, 8, 9, 12, 0, tzinfo=timezone.utc)
class Authority:
def __init__(self, *, head=HEAD, exists=True, descendant=True):
self.head = head
self.exists = exists
self.descendant = descendant
self.cas_calls = []
def branch_exists(self, branch):
return self.exists
def branch_head(self, branch):
return self.head
def is_ancestor(self, older, newer):
if (older, newer) == (BASE, HEAD):
return True
return self.descendant and (older, newer) in {(HEAD, NEW), (BASE, NEW)}
def compare_and_swap_branch(self, branch, expected, new):
self.cas_calls.append((branch, expected, new))
if self.head != expected:
raise RuntimeError("stale")
self.head = new
return new
class Workspace:
def __init__(self, *, detached=True, head=HEAD):
self.detached = detached
self.head = head
self.commits = 0
def is_detached(self):
return self.detached
def head_oid(self):
return self.head
def commit_files(self, files, message):
self.commits += 1
self.head = NEW
return NEW
class Remote:
def __init__(self, *, git=HEAD, api=HEAD, visible=True, repository=REPOSITORY, fail_after_push=False, omit_after_push=False):
self.git = git
self.api = api
self.visible = visible
self.repository = repository
self.fail_after_push = fail_after_push
self.omit_after_push = omit_after_push
self.pushes = 0
def repository_identity(self):
return self.repository
def push_exact(self, branch, expected, new):
self.pushes += 1
if self.git is None:
raise RuntimeError("remote branch missing")
if self.git != expected:
raise RuntimeError("remote conflict")
if not self.omit_after_push:
self.git = self.api = new
else:
self.git = self.api = None
if self.fail_after_push:
raise TimeoutError("response lost")
def read_git_ref(self, branch):
return self.git
def read_api_branch(self, branch):
return self.api
def commit_visible(self, oid):
return self.visible and self.api == oid
class Credential:
def __init__(self):
self.projected = 0
self.revoked = 0
def project(self):
self.projected += 1
def revoke(self):
self.revoked += 1
def lease_request(**changes):
values = dict(task_id=TASK, repository=REPOSITORY, branch_mode="CONTINUE_EXISTING", base_oid=BASE,
expected_branch_oid=HEAD, target_branch=BRANCH, generation=1, lease_seconds=900)
values.update(changes)
return WriterLeaseRequestV2(**values)
def binding(**changes):
values = dict(
request=lease_request(),
workspace_path=Path("/var/lib/aios-task-writer/continuation-worktrees/exact"),
lease_not_before=NOW - timedelta(seconds=1),
lease_expires_at=NOW + timedelta(minutes=15),
original_receipt_valid=True,
original_grant_consumed=True,
permissions=ContinuationPermissions(),
)
values.update(changes)
return ContinuationRuntimeBinding(**values)
def commit(**changes):
values = dict(idempotency_key="first-continuation", files={"active-runtime/app/new.py": b"ok\n"},
message="feat(gitops): add governed continuation executor")
values.update(changes)
return ContinuationCommit(**values)
def make_executor(authority=None, workspace=None, remote=None, credential=None):
temp = tempfile.TemporaryDirectory()
controller = GitOpsJobController(JobStore(Path(temp.name).resolve()))
executor = ContinuationExecutor(
controller, authority or Authority(), workspace or Workspace(), remote or Remote(), credential or Credential(),
now=lambda: NOW,
)
executor._test_temp = temp
return executor
def test_exact_existing_branch_completes_only_after_all_remote_proofs():
authority, remote, credential = Authority(), Remote(), Credential()
result = make_executor(authority, remote=remote, credential=credential).execute(binding(), commit())
assert result["state"] == JobState.COMPLETED.value
assert result["receipt"]["authority_oid"] == result["receipt"]["git_remote_oid"] == NEW
assert result["receipt"]["api_branch_oid"] == NEW
assert result["receipt"]["commit_visible"] is True
assert authority.cas_calls == [(BRANCH, HEAD, NEW)]
assert credential.projected == credential.revoked == 1
assert "credential" not in repr(result).lower()
@pytest.mark.parametrize(
"runtime_binding,authority,workspace,error",
[
(binding(request=lease_request(expected_branch_oid="7" * 40)), Authority(), Workspace(), "EXISTING_BRANCH_OID_CONFLICT"),
(binding(), Authority(exists=False), Workspace(), "EXISTING_BRANCH_MISSING"),
(binding(), Authority(), Workspace(detached=False), "WORKTREE_NOT_DETACHED"),
(binding(), Authority(), Workspace(head="7" * 40), "WORKTREE_HEAD_MISMATCH"),
(binding(lease_expires_at=NOW), Authority(), Workspace(), "LEASE_EXPIRED"),
(binding(request=lease_request(repository="other/repo")), Authority(), Workspace(), "request_identity_invalid"),
(binding(request=lease_request(target_branch="main")), Authority(), Workspace(), "PROTECTED_BRANCH_REJECTED"),
(binding(request=lease_request(generation=1)), Authority(), Workspace(), "GENERATION_NOT_MONOTONIC"),
],
)
def test_pre_side_effect_failures_never_create_or_update_a_branch(runtime_binding, authority, workspace, error):
executor = make_executor(authority, workspace=workspace)
if error == "GENERATION_NOT_MONOTONIC":
executor.observe_generation(BRANCH, 1)
with pytest.raises(Exception, match=error):
executor.execute(runtime_binding, commit())
assert authority.cas_calls == []
assert workspace.commits == 0
@pytest.mark.parametrize("field", ["create_target_branch", "force_push", "tag_write", "baseline_write", "production_write", "main_write"])
def test_forbidden_capabilities_fail_before_credential_projection(field):
credential = Credential()
permissions = ContinuationPermissions(**{field: True})
with pytest.raises(ContinuationExecutionError, match="MUTATION_PERMISSION_DENIED"):
make_executor(credential=credential).execute(binding(permissions=permissions), commit())
assert credential.projected == 0
def test_non_fast_forward_and_concurrent_movement_are_rejected_cas_only():
authority = Authority(descendant=False)
with pytest.raises(ContinuationExecutionError, match="NON_FAST_FORWARD_COMMIT"):
make_executor(authority).execute(binding(), commit())
assert authority.cas_calls == []
def test_authority_success_and_gitea_missing_is_never_completed():
remote = Remote(omit_after_push=True)
result = make_executor(remote=remote).execute(binding(), commit())
assert result["state"] == JobState.RECONCILIATION_REQUIRED.value
assert result["side_effect_state"] == "REMOTE_PUSH_PENDING"
assert result["state"] not in {JobState.REMOTE_VERIFIED.value, JobState.COMPLETED.value}
def test_response_loss_after_remote_success_reconciles_by_observation_without_second_push():
remote = Remote(fail_after_push=True)
executor = make_executor(remote=remote)
first = executor.execute(binding(), commit())
assert first["state"] == JobState.RECONCILIATION_REQUIRED.value
assert remote.pushes == 1
recovered = executor.reconcile(first["job_id"])
assert recovered["state"] == JobState.COMPLETED.value
assert remote.pushes == 1
def test_observation_first_reconcile_retries_only_when_remote_is_exact_expected_old_oid():
remote = Remote(fail_after_push=True)
executor = make_executor(remote=remote)
first = executor.execute(binding(), commit())
assert first["state"] == JobState.RECONCILIATION_REQUIRED.value
remote.git = remote.api = HEAD
remote.visible = True
remote.fail_after_push = False
recovered = executor.reconcile(first["job_id"])
assert recovered["state"] == JobState.COMPLETED.value
assert remote.pushes == 2
def test_duplicate_idempotency_replay_and_conflicting_payload():
executor = make_executor()
first = executor.execute(binding(), commit())
replay = executor.execute(binding(), commit())
assert replay["job_id"] == first["job_id"]
with pytest.raises(BindingConflict, match="BINDING_CONFLICT"):
executor.execute(binding(), commit(files={"active-runtime/app/new.py": b"changed\n"}))
def test_secret_like_payload_fails_before_projection():
credential = Credential()
secret_fixture = b"to" + b"ken=" + b"ghp_" + b"abcdefghijklmnopqrstuvwxyz0123456789"
with pytest.raises(ContinuationExecutionError, match="SECRET_LIKE_PAYLOAD_REJECTED"):
make_executor(credential=credential).execute(binding(), commit(files={"x": secret_fixture}))
assert credential.projected == 0
def test_remote_identity_cannot_alias_the_authority_bare_repo():
with pytest.raises(ContinuationExecutionError, match="GITEA_REMOTE_IDENTITY_INVALID"):
GiteaContinuationRemote.validate_remote_url("/data/aios-gitops/repo.git")
GiteaContinuationRemote.validate_remote_url("aiman@g.cnnetmkt.top:aios/aios.git")
def test_detached_workspace_preserves_porcelain_leading_status_column():
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
subprocess.run(["git", "init", "-q", str(root)], check=True)
subprocess.run(["git", "-C", str(root), "config", "user.name", "test"], check=True)
subprocess.run(["git", "-C", str(root), "config", "user.email", "test@localhost"], check=True)
(root / "tracked.py").write_text("one\n", encoding="utf-8")
subprocess.run(["git", "-C", str(root), "add", "tracked.py"], check=True)
subprocess.run(["git", "-C", str(root), "commit", "-q", "-m", "base"], check=True)
subprocess.run(["git", "-C", str(root), "checkout", "-q", "--detach"], check=True)
(root / "tracked.py").write_text("two\n", encoding="utf-8")
(root / "new.py").write_text("new\n", encoding="utf-8")
workspace = DetachedGitWorkspace(root)
assert workspace.is_detached()
assert workspace._dirty_paths() == {"tracked.py", "new.py"}
@@ -0,0 +1,37 @@
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
OPENAPI=ROOT/"docs/gitops/openapi-chatgpt-actions.yaml"
ROUTES={
"/v2/task-admission/envelope":"getTaskAdmissionEnvelope",
"/v2/task-admission/authorize":"submitTaskAdmissionAuthorization",
"/v2/task-admission/status":"getTaskAdmissionStatus",
}
def block(text,marker,n):
lines=text.splitlines(); pref=" "*n+marker
start=next((i for i,x in enumerate(lines) if x==pref),None)
assert start is not None, marker
out=[lines[start]]
for line in lines[start+1:]:
if line.strip() and len(line)-len(line.lstrip(" "))<=n: break
out.append(line)
return "\n".join(out)
def test_projected_once():
t=OPENAPI.read_text(encoding="utf-8-sig")
for route,op in ROUTES.items():
assert t.count(f" {route}:")==1
assert t.count(f"operationId: {op}")==1
def test_no_principal_injection():
t=OPENAPI.read_text(encoding="utf-8-sig")
b=block(t,"TaskAdmissionAuthorizationRequest:",4)
assert "\n principal:" not in b
assert "\n _verified_task_principal:" not in b
def test_no_git_write_projection():
t=OPENAPI.read_text(encoding="utf-8-sig")
for route in ("/v2/task-admission/envelope:","/v2/task-admission/authorize:"):
b=block(t,route,2)
assert "x-aios-no-git-write: true" in b
assert "x-aios-no-shell: true" in b
def test_status_read_only():
t=OPENAPI.read_text(encoding="utf-8-sig")
b=block(t,"/v2/task-admission/status:",2)
assert "x-aios-read-only: true" in b