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
18 changed files with 2063 additions and 32 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()
+31 -4
View File
@@ -1,5 +1,4 @@
"""ChatGPT GitOps Bridge v2 contract and immutable runtime identity."""
"""ChatGPT GitOps Bridge v2 contract and governed release operations."""
from __future__ import annotations
from dataclasses import dataclass
@@ -19,6 +18,13 @@ BRIDGE_OPERATIONS = (
"getGitOpsJob",
"resumeGitOpsJob",
"reconcileGitOpsJob",
"createTagApprovalRequest",
"requestOrCreateTag",
"getTagPublicationStatus",
"buildProductionArtifact",
"executeApprovedProductionRelease",
"getReleaseConsistency",
"runReleaseAcceptance",
)
OPERATION_REQUIRED_FIELDS = {
@@ -29,6 +35,13 @@ OPERATION_REQUIRED_FIELDS = {
"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"),
}
@@ -92,7 +105,7 @@ class ControlPlaneReleaseManifest:
class BridgeV2Adapter:
"""Thin operation adapter; mutation handlers are existing governed seams."""
"""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)
@@ -112,6 +125,13 @@ class BridgeV2Adapter:
"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
@@ -120,7 +140,7 @@ class BridgeV2Adapter:
def _call(self, operation: str, request: Mapping[str, Any]) -> Any:
required = OPERATION_REQUIRED_FIELDS[operation]
if tuple(key for key in required if key not in request):
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:
@@ -134,3 +154,10 @@ class BridgeV2Adapter:
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()
+116 -20
View File
@@ -43,6 +43,7 @@ class ContinuationExecutionError(RuntimeError):
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: ...
@@ -51,13 +52,16 @@ class AuthorityPort(Protocol):
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, new: str) -> None: ...
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: ...
@@ -105,6 +109,12 @@ class AuthorityGitAdapter:
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:
@@ -148,6 +158,38 @@ class DetachedGitWorkspace:
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:
@@ -241,10 +283,16 @@ class GiteaContinuationRemote:
raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
return oid
def push_exact(self, branch: str, expected: str, new: str) -> None:
def push_exact(self, branch: str, expected: str | None, new: str) -> None:
self._require_projected()
self.repository_identity()
if self.read_git_ref(branch) != expected:
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)
@@ -267,13 +315,24 @@ class GiteaContinuationRemote:
def read_api_branch(self, branch: str) -> str | None:
owner, name = self.repository.split("/", 1)
value = self._api(f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")
if value is None:
return None
oid = value.get("commit", {}).get("id") if isinstance(value, dict) else None
if not isinstance(oid, str) or not OID_RE.fullmatch(oid):
raise ContinuationExecutionError("GITEA_API_BRANCH_INVALID")
return oid
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)
@@ -293,8 +352,10 @@ class ContinuationPermissions:
production_write: bool = False
main_write: bool = False
def validate(self) -> None:
if any(asdict(self).values()):
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")
@@ -385,20 +446,31 @@ class ContinuationExecutor:
"lease_seconds": request.lease_seconds,
}
def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> None:
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)
if normalized != {**self._request_mapping(request), "create_target_branch": False, "worktree_mode": "detached_at_expected_head", "cas_ref_update_required": True}:
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()
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):
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")
@@ -414,9 +486,11 @@ class ContinuationExecutor:
"campaign_id": request.task_id,
"task_id": request.task_id,
"repository": request.repository,
"operation": "CONTINUATION_COMMIT_FILES",
"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,
@@ -428,7 +502,26 @@ class ContinuationExecutor:
current = self.controller.store.read(submitted.job_id)
if JobState(current["state"]) == JobState.COMPLETED:
return self.controller.store._receipt(current, duplicate=True).as_dict()
self._validate_preflight(binding)
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:
@@ -458,7 +551,10 @@ class ContinuationExecutor:
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")
authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
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(
@@ -529,7 +625,7 @@ class ContinuationExecutor:
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 = request["expected_branch_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)
@@ -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)
@@ -286,6 +286,8 @@ class GitOpsJobController:
"new_branch_oid",
"authorization_recovery_digest",
"lineage_anchor_oid",
"base_oid",
"branch_mode",
"generation",
"expected_main_oid",
"pr_number",
@@ -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())
+194 -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:
@@ -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,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)
@@ -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