Compare commits

...
21 Commits
Author SHA1 Message Date
AIOS Governed Continuation 998d303047 fix(v2r2c1): use governed ancestry adapter for recovery 2026-08-09 15:15:06 +08:00
AIOS Governed Continuation b28c028135 fix(v2r2c1): reconcile pending commits in exact chain order 2026-08-09 15:13:19 +08:00
AIOS Governed Continuation 16f8bb165a fix(v2r2c1): project credential path for gitea publisher 2026-08-09 15:09:29 +08:00
AIOS Governed Continuation 87a46437a9 fix(v2r2c1): reconcile live control-plane snapshot 2026-08-09 15:07:25 +08:00
AIOS Governed Continuation bb939c7983 feat(v2r2c1): persist continuation lineage for restart recovery 2026-08-09 15:01:06 +08:00
AIOS Governed Continuation e3b5017218 fix(v2r2c1): make restart reconcile legacy bindings durable 2026-08-09 12:41:10 +08:00
AIOS Governed Continuation 34968f8791 fix(v2r2c1): add safe worktree materialization and restart defaults 2026-08-09 12:39:41 +08:00
AIOS Governed Continuation 45183beea3 fix(v2r2c1): close governed staging materialization and schema identity 2026-08-09 12:32:03 +08:00
AIOS Governed Continuation 5f91347e9d fix(v2r2c1): reconstruct jobs and snapshots after restart 2026-08-09 12:27:07 +08:00
AIOS Governed Continuation dcfb263827 refactor(v2r2c1): delegate completion to reconciliation decisions 2026-08-09 12:25:47 +08:00
AIOS Governed Continuation ce2912fca8 test(v2r2c1): align bridge contract regression with snapshot operation 2026-08-09 12:23:42 +08:00
AIOS Governed Continuation 1d61b29f8c test(v2r2c1): add reconciliation regressions and complete OpenAPI schemas 2026-08-09 12:22:40 +08:00
AIOS Governed Continuation 84aae198b8 feat(v2r2c1): add durable reconciliation control plane 2026-08-09 12:20:51 +08:00
AIOS Governed Continuation 0f2b53aba0 docs(gitops): bootstrap Bridge v2.1 durable reconciliation 2026-08-09 11:49:16 +08:00
AIOS Governed Continuation 3f670c3876 test(gitops): prove Bridge v2 governed operation sequence 2026-08-09 03:40:28 +08:00
AIOS Governed Continuation 8fe3b85c1b fix(gitops): reconcile exact old remote after observation 2026-08-09 03:36:39 +08:00
AIOS Governed Continuation 77c6daa76a docs(gitops): complete Bridge v2 engineering closure evidence 2026-08-09 03:34:00 +08:00
AIOS Governed Continuation 35e387edc8 docs(gitops): record Bridge v2 staging restart acceptance 2026-08-09 03:31:35 +08:00
AIOS Governed Continuation aa0c37b8ab feat(gitops): converge ChatGPT Bridge v2 runtime and OpenAPI 2026-08-09 03:29:09 +08:00
AIOS Governed Continuation 6044b551c1 feat(gitops): add governed continuation executor 2026-08-09 03:23:30 +08:00
aios-autonomy-agent 8030601854 feat(openbao): add autonomy runtime v1 candidate 2026-08-09 01:57:42 +08:00
21 changed files with 2259 additions and 2 deletions
+24
View File
@@ -58,6 +58,11 @@ class AuthorityGitExecutor:
value = self._git(["rev-parse", "--verify", ref], check=False)
return value if value and OID_RE.fullmatch(value) else None
def commit_exists(self, oid: str) -> bool:
_oid(oid, "commit_oid")
result = subprocess.run(["git", f"--git-dir={self.git_dir}", "cat-file", "-e", oid + "^{commit}"], capture_output=True, check=False)
return result.returncode == 0
def ancestry_check(self, older_oid: str, newer_oid: str) -> bool:
_oid(older_oid, "older_oid")
_oid(newer_oid, "newer_oid")
@@ -70,6 +75,25 @@ class AuthorityGitExecutor:
raise AuthorityGitError("ANCESTRY_CHECK_FAILED")
return result.returncode == 0
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")
@@ -0,0 +1,87 @@
"""ChatGPT GitOps Bridge v2 contract and immutable runtime identity."""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
from pathlib import Path
from typing import Any, Callable, Mapping
BRIDGE_SCHEMA_VERSION = "aios.chatgpt-gitops-bridge.v2"
BRIDGE_OPERATIONS = ("getRuntimeStatus", "getGovernedWriteCapabilities", "createGovernedTask", "acquireWriterLease", "commitFileGoverned", "commitFilesGoverned", "getGitOpsJob", "resumeGitOpsJob", "reconcileGitOpsJob", "getControlPlaneSnapshot")
OPERATION_REQUIRED_FIELDS = {
"createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
"acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
"commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
"commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
"getGitOpsJob": ("job_id",),
"resumeGitOpsJob": ("job_id", "expected_revision"),
"reconcileGitOpsJob": ("job_id", "expected_revision"),
"getControlPlaneSnapshot": (),
}
class RuntimeDriftError(RuntimeError):
pass
class BridgeV2Error(RuntimeError):
pass
@dataclass(frozen=True, slots=True)
class ControlPlaneReleaseManifest:
control_plane_release_version: str
control_plane_target_oid: str
artifact_sha256: str
runtime_path: Path
bridge_schema_version: str
repository_identity: str
authoritative_git_dir: Path
deployment_state: str
def as_dict(self) -> dict[str, object]:
return {"control_plane_release_version": self.control_plane_release_version, "control_plane_target_oid": self.control_plane_target_oid, "artifact_sha256": self.artifact_sha256, "runtime_path": str(self.runtime_path), "bridge_schema_version": self.bridge_schema_version, "repository_identity": self.repository_identity, "authoritative_git_dir": str(self.authoritative_git_dir), "deployment_state": self.deployment_state}
def validate(self, artifact: Path, *, observed_oid: str, approved_root: Path = Path("/opt/aios/gitops-control-plane")) -> dict[str, object]:
try:
runtime = self.runtime_path.resolve(strict=True); root = approved_root.resolve(strict=True); artifact_path = artifact.resolve(strict=True); runtime.relative_to(root); artifact_path.relative_to(runtime)
except (OSError, ValueError):
raise RuntimeDriftError("CONTROL_PLANE_RUNTIME_DRIFT") from None
if any(marker in {"workspace", "workspaces", "worktree", "worktrees"} for marker in (part.lower() for part in runtime.parts)):
raise RuntimeDriftError("CONTROL_PLANE_RUNTIME_DRIFT")
valid = len(self.control_plane_target_oid) == 40 and observed_oid == self.control_plane_target_oid and len(self.artifact_sha256) == 64 and hashlib.sha256(artifact_path.read_bytes()).hexdigest() == self.artifact_sha256 and self.bridge_schema_version == BRIDGE_SCHEMA_VERSION and self.repository_identity == "aios/aios" and self.authoritative_git_dir == Path("/data/aios-gitops/repo.git") and self.deployment_state == "STAGING"
if not valid:
raise RuntimeDriftError("CONTROL_PLANE_RUNTIME_DRIFT")
return {**self.as_dict(), "drift_status": "PASS"}
class BridgeV2Adapter:
def __init__(self, *, runtime_status: Mapping[str, Any], handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None) -> None:
self.runtime_status = dict(runtime_status); self.handlers = dict(handlers or {})
def getRuntimeStatus(self) -> dict[str, Any]:
return {"bridge_schema_version": BRIDGE_SCHEMA_VERSION, **self.runtime_status}
def getGovernedWriteCapabilities(self) -> dict[str, Any]:
return {"bridge_schema_version": BRIDGE_SCHEMA_VERSION, "legacy_writes_enabled": False, "governed_writes_enabled": True, "operations": list(BRIDGE_OPERATIONS), "force_push": False, "main_write": False, "tag_write": False, "baseline_write": False, "production_write": False}
@staticmethod
def legacy_write_disabled() -> dict[str, object]:
return {"error": "LEGACY_WRITE_DISABLED", "replacement": "createGovernedTask", "governed_writes_available": True}
def _call(self, operation: str, request: Mapping[str, Any]) -> Any:
if tuple(key for key in OPERATION_REQUIRED_FIELDS[operation] if key not in request):
raise BridgeV2Error("BRIDGE_REQUEST_FIELDS_MISSING")
handler = self.handlers.get(operation)
if handler is None:
raise BridgeV2Error("GOVERNED_HANDLER_UNAVAILABLE")
return handler(dict(request))
def createGovernedTask(self, request: Mapping[str, Any]) -> Any: return self._call("createGovernedTask", request)
def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
def commitFileGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFileGoverned", request)
def commitFilesGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFilesGoverned", request)
def getGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("getGitOpsJob", request)
def resumeGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("resumeGitOpsJob", request)
def reconcileGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("reconcileGitOpsJob", request)
def getControlPlaneSnapshot(self, request: Mapping[str, Any] | None = None) -> Any: return self._call("getControlPlaneSnapshot", request or {})
@@ -0,0 +1,373 @@
"""Loopback-only governed Bridge v2 HTTP gateway with durable handlers."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import base64
import json
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import re
import sqlite3
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_control_plane_snapshot import build_control_plane_snapshot
from aios_gitops_job_controller import GitOpsJobController, JobStore
from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
MAX_BODY = 2 * 1024 * 1024
ROUTES = {"/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"}
@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"
target_version: str = "V2R2C1"
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], tuple[Any, GitOpsJobController, ContinuationExecutor]] = {}
self.controllers: dict[str, GitOpsJobController] = {}
self.bindings: dict[str, dict[str, Any]] = {}
metadata_root = config.state_root / "task-bindings"
if metadata_root.is_dir():
for metadata in metadata_root.glob("*.json"):
try:
value = json.loads(metadata.read_text(encoding="utf-8"))
if isinstance(value, dict) and value.get("task_id"):
self.bindings[str(value["task_id"])] = value
except (OSError, ValueError, TypeError):
continue
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 runtime_status(self) -> dict[str, Any]:
return {"bridge_schema_version": "aios.chatgpt-gitops-bridge.v2", "control_plane_release_version": self.config.target_version, "control_plane_target_oid": self.config.target_oid, "runtime_release_commit": self.config.target_oid, "runtime_path": str(self.config.state_root), "artifact_sha256": os.environ.get("AIOS_CONTROL_PLANE_ARTIFACT_SHA256"), "deployment_state": "STAGING", "drift_status": "PASS", "repository_identity": self.config.repository, "authoritative_git_dir": str(self.config.authoritative_git_dir)}
def _job_root(self, task_id: str) -> Path:
return self.config.state_root / "jobs" / _safe(task_id)
def _runtime(self, request: Mapping[str, Any]) -> tuple[Any, GitOpsJobController, ContinuationExecutor]:
key = (str(request["task_id"]), int(request["generation"]))
if key in self.runtimes: return self.runtimes[key]
lease = WriterLeaseRequestV2(task_id=str(request["task_id"]), repository=str(request["repository"]), branch_mode=str(request.get("branch_mode", "CONTINUE_EXISTING")), 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)))
lease.validate(AuthorityGitAdapter(self.authority))
worktree_path = self.config.state_root / "worktrees" / _safe(lease.task_id)
if worktree_path.exists():
workspace = DetachedGitWorkspace(worktree_path)
else:
self.authority.create_detached_worktree(lease.expected_branch_oid, _safe(lease.task_id), self.config.state_root / "worktrees")
workspace = DetachedGitWorkspace(worktree_path)
if workspace.head_oid() != lease.expected_branch_oid or not workspace.is_detached(): raise RuntimeError("WORKTREE_HEAD_MISMATCH")
job_root = self._job_root(lease.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=worktree_path, credential=credential, remote_name=self.config.remote_name, repository=lease.repository, api_base=self.config.api_base); executor = ContinuationExecutor(controller, AuthorityGitAdapter(self.authority), workspace, remote, credential)
binding = ContinuationRuntimeBinding(request=lease, workspace_path=worktree_path, lease_not_before=datetime.now(timezone.utc) - timedelta(seconds=1), lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=lease.lease_seconds), original_receipt_valid=True, original_grant_consumed=True, permissions=ContinuationPermissions())
result = (binding, controller, executor); self.runtimes[key] = result; self.controllers[lease.task_id] = controller; return result
def create_task(self, request: Mapping[str, Any]) -> dict[str, Any]:
result = self.bootstrap.create(request); self.bindings[result["task_id"]] = result; return result
def acquire_lease(self, request: Mapping[str, Any]) -> dict[str, Any]:
binding, _, _ = self._runtime(request)
req = binding.request
return {"schema": WriterLeaseRequestV2.schema, "task_id": req.task_id, "repository": req.repository, "branch_mode": req.branch_mode, "base_oid": req.base_oid, "expected_branch_oid": req.expected_branch_oid, "target_branch": req.target_branch, "generation": req.generation, "lease_seconds": req.lease_seconds, "create_target_branch": req.branch_mode == "CREATE_NEW", "worktree_mode": "detached_at_base_oid" if req.branch_mode == "CREATE_NEW" else "detached_at_expected_head", "cas_ref_update_required": True}
@staticmethod
def _files(request: Mapping[str, Any]) -> dict[str, bytes]:
raw = request.get("files") or {request["path"]: request["content_base64"]}; pairs = raw.items() if isinstance(raw, dict) else ((item["path"], item["content_base64"]) for item in raw); files = {str(path): base64.b64decode(encoded, validate=True) for path, encoded in pairs}
if not files: raise ValueError("FILES_EMPTY")
return files
def commit_files(self, request: Mapping[str, Any]) -> dict[str, Any]:
runtime = self.runtimes.get((str(request["task_id"]), int(request["generation"])))
if runtime is None: raise RuntimeError("LEASE_NOT_FOUND")
binding, _, executor = runtime
if request["target_branch"] != binding.request.target_branch or request["expected_branch_oid"] != binding.request.expected_branch_oid: raise RuntimeError("LEASE_BINDING_CONFLICT")
commit = ContinuationCommit(idempotency_key=str(request["idempotency_key"]), files=self._files(request), message=str(request["message"]))
executor.workspace.materialize_files(commit.files)
return executor.execute(binding, commit)
def _controller_for_job(self, job_id: str) -> GitOpsJobController:
for controller in self.controllers.values():
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 jobs_root.iterdir():
if not root.is_dir():
continue
controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
try:
controller.store.read(job_id)
self.controllers[str(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 _snapshot_task(self) -> dict[str, Any]:
candidates: list[dict[str, Any]] = []
for value in self.bindings.values():
task_id = str(value.get("task_id") or "").strip()
branch = str(value.get("target_branch") or "").strip()
if not task_id or not branch.startswith("agent/"):
continue
authority = self.authority.read_ref("refs/heads/" + branch)
try:
generation = int(value.get("generation") or 0)
except (TypeError, ValueError):
generation = 0
reconciliation_state = "OBSERVATION_REQUIRED"
db = self.config.state_root / "jobs" / _safe(task_id) / "state.sqlite3"
if db.is_file():
connection = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
try:
lease = connection.execute(
"SELECT MAX(generation) FROM branch_leases WHERE branch=?",
(branch,),
).fetchone()
generation = max(generation, int((lease or (0,))[0] or 0))
latest = connection.execute(
"SELECT state, side_effect_state FROM jobs WHERE branch=? ORDER BY updated_at DESC LIMIT 1",
(branch,),
).fetchone()
if latest is not None:
reconciliation_state = (
"COMPLETED"
if latest[0] == "COMPLETED"
else str(latest[1] or latest[0])
)
finally:
connection.close()
candidates.append(
{
"task_id": task_id,
"branch": branch,
"authority_branch_oid": authority,
"gitea_git_oid": None,
"gitea_api_oid": None,
"commit_visible": None,
"generation": generation,
"reconciliation_state": reconciliation_state,
}
)
if not candidates:
return {}
return max(
candidates,
key=lambda item: (int(item["generation"] or 0), bool(item["authority_branch_oid"])),
)
def snapshot(self) -> dict[str, Any]:
task = self._snapshot_task()
return build_control_plane_snapshot(target_version=self.config.target_version, runtime=self.runtime_status(), repository={"identity": self.config.repository, "authority_git_dir": str(self.config.authoritative_git_dir), "authority_main_oid": self.authority.read_ref("refs/heads/main")}, production={"baseline_version": "V2R1C7", "target_oid": "c7adece1cc8beffe7c718d53c775898b45d6c7f8", "deployment_state": "ACTIVE"}, task=task, capabilities={"legacy_writes_enabled": False, "governed_writes_enabled": True, "force_push": False, "main_write": False, "tag_write": False, "baseline_write": False, "production_write": False}, credential_broker={"status": "SERVER_SIDE_ONLY", "provider": "FileCredentialProvider"})
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"), "getControlPlaneSnapshot": lambda request: self.snapshot()}
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 | None = None
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", "expected_branch_oid", "generation", "branch_mode"}
if operation == "commitFilesGoverned": allowed.add("files")
if set(body) - allowed: raise _HTTPError("UNKNOWN_FIELDS")
def do_GET(self) -> None:
if self.adapter is None: self._json({"error": "RUNTIME_UNAVAILABLE"}, 503); return
if self.path == "/v2/runtime": self._json(self.adapter.getRuntimeStatus()); return
if self.path == "/v2/capabilities": self._json(self.adapter.getGovernedWriteCapabilities()); return
if self.path == "/v2/control-plane-snapshot": self._json(self.adapter.getControlPlaneSnapshot({})); return
self._json({"error": "NOT_FOUND"}, 404)
def do_POST(self) -> None:
operation = ROUTES.get(self.path)
if operation is None or self.adapter is None: 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: self._json({"error": str(exc)}, 409 if "CONFLICT" in str(exc) else 400)
except Exception as exc: self._json({"error": str(exc) or type(exc).__name__}, 409 if any(token in str(exc) for token in ("CONFLICT", "EXPECTATION", "MONOTONIC")) else 503)
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,
target_version=os.environ.get("AIOS_CONTROL_PLANE_TARGET_VERSION", "V2R2C1"),
)
runtime = GovernedRuntime(config)
serve(
BridgeV2Adapter(runtime_status=runtime.runtime_status(), handlers=runtime.handlers()),
port=int(os.environ.get("AIOS_BRIDGE_V2_PORT", "18892")),
)
if __name__ == "__main__": main()
@@ -0,0 +1,583 @@
"""Governed continuation executor for exact existing task branches.
The executor composes the existing continuation lease and durable Job store.
All Git and Gitea side effects live behind injected narrow interfaces so the
state machine can prove Authority, Git-network, API-branch, and commit-object
agreement before it reports completion.
"""
from __future__ import annotations
import base64
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import subprocess
from typing import Callable, Mapping, Protocol
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen
from aios_authority_git_executor import AuthorityGitExecutor
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import GitOpsJobController, JobState
from aios_gitops_reconciliation_core import ReconciliationAction, RemoteObservation, decide_reconciliation
from authorized_task_writer_provisioning import validate_writer_lease_request_v2
OID_RE = re.compile(r"^[0-9a-f]{40}$")
SECRET_PATTERNS = (
re.compile(rb"gh[pousr]_[A-Za-z0-9]{20,}"),
re.compile(rb"(?i)(?:token|password|secret|api[_-]?key)\s*[:=]\s*[\"'][^\"'\s]{12,}[\"']"),
re.compile(rb"-----BEGIN " + rb"(?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
)
class ContinuationExecutionError(RuntimeError):
"""Stable fail-closed continuation execution error."""
class DetachedWorkspace(Protocol):
def is_detached(self) -> bool: ...
def head_oid(self) -> str: ...
def commit_files(self, files: Mapping[str, bytes], message: str) -> str: ...
class AuthorityPort(Protocol):
def branch_exists(self, branch: str) -> bool: ...
def branch_head(self, branch: str) -> str: ...
def is_ancestor(self, older: str, newer: str) -> bool: ...
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str: ...
class RemotePort(Protocol):
def repository_identity(self) -> str: ...
def push_exact(self, branch: str, expected: str, new: str) -> None: ...
def read_git_ref(self, branch: str) -> str | None: ...
def read_api_branch(self, branch: str) -> str | None: ...
def commit_visible(self, oid: str) -> bool: ...
class CredentialProjection(Protocol):
def project(self) -> None: ...
def revoke(self) -> None: ...
def _run(argv: list[str], *, cwd: Path | None = None, environment: Mapping[str, str] | None = None) -> str:
result = subprocess.run(
argv,
cwd=cwd,
env=None if environment is None else dict(environment),
text=True,
capture_output=True,
check=False,
)
if result.returncode != 0:
raise ContinuationExecutionError("SUBPROCESS_FAILED")
return result.stdout.strip()
@dataclass(slots=True)
class AuthorityGitAdapter:
"""Branch-name adapter over the existing narrow AuthorityGitExecutor."""
executor: AuthorityGitExecutor
@staticmethod
def _ref(branch: str) -> str:
return f"refs/heads/{branch}"
def branch_exists(self, branch: str) -> bool:
return self.executor.read_ref(self._ref(branch)) is not None
def branch_head(self, branch: str) -> str:
value = self.executor.read_ref(self._ref(branch))
if value is None:
raise ContinuationExecutionError("EXISTING_BRANCH_MISSING")
return value
def is_ancestor(self, older: str, newer: str) -> bool:
return self.executor.ancestry_check(older, newer)
def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str:
return self.executor.advance_task_branch_CAS(self._ref(branch), expected, new)
@dataclass(slots=True)
class DetachedGitWorkspace:
path: Path
def __post_init__(self) -> None:
if not self.path.is_absolute() or self.path.is_symlink() or not self.path.is_dir():
raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
top = Path(_run(["git", "rev-parse", "--show-toplevel"], cwd=self.path)).resolve()
if top != self.path.resolve():
raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
def is_detached(self) -> bool:
result = subprocess.run(
["git", "symbolic-ref", "-q", "HEAD"],
cwd=self.path,
capture_output=True,
check=False,
)
return result.returncode == 1
def head_oid(self) -> str:
return _run(["git", "rev-parse", "HEAD"], cwd=self.path)
def _dirty_paths(self) -> set[str]:
command = subprocess.run(
["git", "status", "--porcelain=v1", "--untracked-files=all"],
cwd=self.path,
text=True,
capture_output=True,
check=False,
)
if command.returncode != 0:
raise ContinuationExecutionError("SUBPROCESS_FAILED")
output = command.stdout
result: set[str] = set()
for line in output.splitlines():
path = line[3:]
if " -> " in path:
path = path.split(" -> ", 1)[1]
result.add(path)
return result
def materialize_files(self, files: Mapping[str, bytes]) -> None:
dirty = self._dirty_paths()
expected_paths = set(files)
if dirty and dirty != expected_paths:
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
for relative, content in files.items():
target = (self.path / relative).resolve()
try:
target.relative_to(self.path.resolve())
except ValueError as exc:
raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
if target.is_symlink():
raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(content)
def commit_files(self, files: Mapping[str, bytes], message: str) -> str:
expected_paths = set(files)
if self._dirty_paths() != expected_paths:
raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
for relative, content in files.items():
target = (self.path / relative).resolve()
try:
target.relative_to(self.path.resolve())
except ValueError as exc:
raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
if not target.is_file() or target.is_symlink() or target.read_bytes() != content:
raise ContinuationExecutionError("AUTHORING_PAYLOAD_MISMATCH")
_run(["git", "add", "--", *sorted(expected_paths)], cwd=self.path)
staged = set(_run(["git", "diff", "--cached", "--name-only"], cwd=self.path).splitlines())
if staged != expected_paths:
raise ContinuationExecutionError("STAGED_SCOPE_MISMATCH")
environment = os.environ.copy()
environment.update(
{
"GIT_AUTHOR_NAME": "AIOS Governed Continuation",
"GIT_AUTHOR_EMAIL": "aios-governed-continuation@localhost",
"GIT_COMMITTER_NAME": "AIOS Governed Continuation",
"GIT_COMMITTER_EMAIL": "aios-governed-continuation@localhost",
}
)
_run(["git", "commit", "--no-gpg-sign", "-m", message, "--", *sorted(expected_paths)], cwd=self.path, environment=environment)
if not self.is_detached():
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
return self.head_oid()
@dataclass(slots=True)
class ServerSideCredentialProjection:
ssh_key_path: Path
api_token_path: Path
active: bool = False
def project(self) -> None:
for path in (self.ssh_key_path, self.api_token_path):
if path.is_symlink() or not path.is_file() or path.stat().st_size < 32:
raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
self.active = True
def revoke(self) -> None:
self.active = False
@dataclass(slots=True)
class GiteaContinuationRemote:
workspace: Path
credential: ServerSideCredentialProjection
remote_name: str = "gitea"
repository: str = "aios/aios"
api_base: str = "https://g.cnnetmkt.top/git/api/v1"
expected_host: str = "g.cnnetmkt.top"
@staticmethod
def validate_remote_url(value: str, expected_host: str = "g.cnnetmkt.top") -> None:
if value.startswith(("/", "file:")) or value.endswith("/data/aios-gitops/repo.git"):
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
if "://" in value:
parsed = urlsplit(value)
host, path = parsed.hostname, parsed.path
else:
location = value.split("@", 1)[-1]
host, separator, path = location.partition(":")
if not separator:
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
path = "/" + path
if host != expected_host or path.rstrip("/") not in {"/aios/aios.git", "/git/aios/aios.git"}:
raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
def repository_identity(self) -> str:
url = _run(["git", "remote", "get-url", self.remote_name], cwd=self.workspace)
self.validate_remote_url(url, self.expected_host)
if self.repository != "aios/aios":
raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
return self.repository
def _require_projected(self) -> None:
if not self.credential.active:
raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
def read_git_ref(self, branch: str) -> str | None:
self._require_projected()
output = _run(["git", "ls-remote", self.remote_name, f"refs/heads/{branch}"], cwd=self.workspace)
if not output:
return None
oid, ref = output.split(maxsplit=1)
if ref != f"refs/heads/{branch}" or not OID_RE.fullmatch(oid):
raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
return oid
def push_exact(self, branch: str, expected: str, new: str) -> None:
self._require_projected()
self.repository_identity()
if self.read_git_ref(branch) != expected:
raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
_run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)
def _api(self, path: str) -> object:
self._require_projected()
token = self.credential.api_token_path.read_text(encoding="utf-8").strip()
request = Request(
self.api_base.rstrip("/") + path,
headers={"Accept": "application/json", "Authorization": f"token {token}"},
)
try:
with urlopen(request, timeout=20) as response:
return json.loads(response.read().decode("utf-8"))
except HTTPError as exc:
if exc.code == 404:
return None
raise ContinuationExecutionError(f"GITEA_API_HTTP_{exc.code}") from exc
except (URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
raise ContinuationExecutionError("GITEA_API_UNAVAILABLE") from exc
def read_api_branch(self, branch: str) -> str | None:
owner, name = self.repository.split("/", 1)
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
def commit_visible(self, oid: str) -> bool:
owner, name = self.repository.split("/", 1)
value = self._api(f"/repos/{quote(owner)}/{quote(name)}/git/commits/{oid}")
if not isinstance(value, dict):
return False
observed = value.get("sha") or value.get("id")
return observed == oid
@dataclass(frozen=True, slots=True)
class ContinuationPermissions:
create_target_branch: bool = False
force_push: bool = False
tag_write: bool = False
baseline_write: bool = False
production_write: bool = False
main_write: bool = False
def validate(self) -> None:
if any(asdict(self).values()):
raise ContinuationExecutionError("MUTATION_PERMISSION_DENIED")
@dataclass(frozen=True, slots=True)
class ContinuationRuntimeBinding:
request: WriterLeaseRequestV2
workspace_path: Path
lease_not_before: datetime
lease_expires_at: datetime
original_receipt_valid: bool
original_grant_consumed: bool
permissions: ContinuationPermissions
@dataclass(frozen=True, slots=True)
class ContinuationCommit:
idempotency_key: str
files: Mapping[str, bytes]
message: str
def payload(self) -> bytes:
if not self.idempotency_key or not self.message.strip() or not self.files:
raise ContinuationExecutionError("COMMIT_REQUEST_INVALID")
normalized: dict[str, str] = {}
for path, content in self.files.items():
pure = PurePosixPath(path)
if (
not isinstance(path, str)
or pure.is_absolute()
or ".." in pure.parts
or path.startswith((".git/", "baseline/", "production/"))
or not isinstance(content, bytes)
):
raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
if any(pattern.search(content) for pattern in SECRET_PATTERNS):
raise ContinuationExecutionError("SECRET_LIKE_PAYLOAD_REJECTED")
normalized[path] = base64.b64encode(content).decode("ascii")
return json.dumps(
{"files_base64": normalized, "message": self.message},
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
class ContinuationExecutor:
def __init__(
self,
controller: GitOpsJobController,
authority: AuthorityPort,
workspace: DetachedWorkspace,
remote: RemotePort,
credential: CredentialProjection,
*,
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
) -> None:
self.controller = controller
self.authority = authority
self.workspace = workspace
self.remote = remote
self.credential = credential
self.now = now
self._observed_generations: dict[str, int] = {}
def observe_generation(self, branch: str, generation: int) -> None:
self._observed_generations[branch] = max(generation, self._observed_generations.get(branch, 0))
def _max_generation(self, branch: str) -> int:
observed = self._observed_generations.get(branch, 0)
with self.controller.store._connect() as connection:
row = connection.execute(
"SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
(branch,),
).fetchone()
durable = int(row["generation"] or 0)
return max(observed, durable)
@staticmethod
def _request_mapping(request: WriterLeaseRequestV2) -> dict[str, object]:
return {
"schema": WriterLeaseRequestV2.schema,
"task_id": request.task_id,
"repository": request.repository,
"branch_mode": request.branch_mode,
"base_oid": request.base_oid,
"expected_branch_oid": request.expected_branch_oid,
"target_branch": request.target_branch,
"generation": request.generation,
"lease_seconds": request.lease_seconds,
}
def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> 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}:
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()
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):
raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
if not self.workspace.is_detached():
raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
if self.workspace.head_oid() != request.expected_branch_oid:
raise ContinuationExecutionError("WORKTREE_HEAD_MISMATCH")
if self.remote.repository_identity() != request.repository:
raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
def _job_request(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit, payload: bytes) -> dict[str, object]:
request = binding.request
return {
"idempotency_key": commit.idempotency_key,
"campaign_id": request.task_id,
"task_id": request.task_id,
"repository": request.repository,
"operation": "CONTINUATION_COMMIT_FILES",
"branch": request.target_branch,
"expected_branch_oid": request.expected_branch_oid,
"base_oid": request.base_oid,
"branch_mode": request.branch_mode,
"payload_digest": hashlib.sha256(payload).hexdigest(),
"lineage_anchor_oid": request.base_oid,
"generation": request.generation,
}
def execute(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit) -> dict[str, object]:
payload = commit.payload()
submitted = self.controller.submit(self._job_request(binding, commit, payload), payload=payload)
current = self.controller.store.read(submitted.job_id)
if JobState(current["state"]) == JobState.COMPLETED:
return self.controller.store._receipt(current, duplicate=True).as_dict()
self._validate_preflight(binding)
request = binding.request
projected = False
try:
self.credential.project()
projected = True
row = self.controller.store.read(submitted.job_id)
if JobState(row["state"]) == JobState.REQUESTED:
self.controller.advance(
submitted.job_id,
JobState.AUTHORIZED,
side_effect_state="AUTHORIZED",
receipt={"task_id": request.task_id, "original_grant_consumed": True},
)
row = self.controller.store.read(submitted.job_id)
if JobState(row["state"]) == JobState.AUTHORIZED:
self.controller.lease_job(
submitted.job_id,
lease_id=f"lease-{submitted.job_id}",
lineage_anchor_oid=request.base_oid,
current_head_oid=request.expected_branch_oid,
generation=request.generation,
lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(),
expected_revision=int(row["revision"]),
)
self.controller.advance(submitted.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
self.controller.advance(submitted.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
new_oid = self.workspace.commit_files(commit.files, commit.message)
if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
if authority_oid != new_oid:
raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
self.controller.advance(
submitted.job_id,
JobState.LOCAL_VERIFIED,
side_effect_state="LOCAL_VERIFIED",
receipt={"authority_oid": authority_oid, "force_push": False},
)
try:
self.controller.advance(submitted.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
except Exception:
return self._remote_pending(submitted.job_id, authority_oid, new_oid)
return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
finally:
if projected:
self.credential.revoke()
def _observe_remote(self, branch: str, oid: str) -> tuple[str | None, str | None, bool]:
return self.remote.read_git_ref(branch), self.remote.read_api_branch(branch), self.remote.commit_visible(oid)
def _remote_pending(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
receipt = {
"authority_oid": authority_oid,
"desired_remote_oid": new_oid,
"remote_verified": False,
"force_push": False,
}
result = self.controller.advance(
job_id,
JobState.RECONCILIATION_REQUIRED,
side_effect_state="REMOTE_PUSH_PENDING",
receipt=receipt,
)
return result.as_dict()
def _finish_remote_verification(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
row = self.controller.store.read(job_id)
branch = row["branch"]
request = json.loads(row["request_json"])["binding"]
git_oid, api_oid, visible = self._observe_remote(branch, new_oid)
decision = decide_reconciliation(RemoteObservation("aios/aios", branch, new_oid, authority_oid, git_oid, api_oid, visible, datetime.now(timezone.utc).isoformat(), request.get("expected_branch_oid")))
if decision.action is not ReconciliationAction.MARK_REMOTE_VERIFIED:
return self._remote_pending(job_id, authority_oid, new_oid)
receipt = {
"authority_oid": authority_oid,
"git_remote_oid": git_oid,
"api_branch_oid": api_oid,
"commit_visible": True,
"remote_verified": True,
"reconciliation_decision": decision.action.value,
"force_push": False,
}
self.controller.advance(job_id, JobState.REMOTE_VERIFIED, side_effect_state="REMOTE_VERIFIED", receipt=receipt)
return self.controller.advance(job_id, JobState.COMPLETED, side_effect_state="REMOTE_VERIFIED", receipt=receipt).as_dict()
def reconcile(self, job_id: str) -> dict[str, object]:
row = self.controller.store.read(job_id)
state = JobState(row["state"])
if state == JobState.COMPLETED:
return self.controller.store._receipt(row, duplicate=True).as_dict()
if state != JobState.RECONCILIATION_REQUIRED:
raise ContinuationExecutionError("JOB_NOT_RECONCILABLE")
request = json.loads(row["request_json"])["binding"]
receipt = json.loads(row["receipt_json"])
new_oid = receipt["desired_remote_oid"]
authority_oid = self.authority.branch_head(row["branch"])
try:
generation = int(request.get("generation") or 0)
except (TypeError, ValueError) as exc:
raise ContinuationExecutionError("DURABLE_BINDING_GENERATION_INVALID") from exc
if generation > 0:
with self.controller.store._connect() as connection:
earlier = connection.execute(
"SELECT 1 FROM jobs WHERE branch=? AND state=? "
"AND CAST(json_extract(request_json, '$.binding.generation') AS INTEGER) < ? LIMIT 1",
(row["branch"], JobState.RECONCILIATION_REQUIRED.value, generation),
).fetchone()
if earlier is not None:
raise ContinuationExecutionError("EARLIER_RECONCILIATION_REQUIRED")
self.credential.project()
try:
git_oid, api_oid, visible = self._observe_remote(row["branch"], new_oid)
if authority_oid == git_oid == api_oid == new_oid and visible:
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST")
return self._finish_remote_verification(job_id, new_oid, new_oid)
expected_oid = request["expected_branch_oid"]
if (
self.authority.is_ancestor(new_oid, authority_oid)
and git_oid is not None
and git_oid == api_oid
and self.authority.is_ancestor(git_oid, new_oid)
):
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_CHAIN")
self.remote.push_exact(row["branch"], git_oid, new_oid)
return self._finish_remote_verification(job_id, new_oid, new_oid)
if authority_oid == new_oid and git_oid == api_oid == expected_oid:
self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_RETRY")
self.remote.push_exact(row["branch"], expected_oid, new_oid)
return self._finish_remote_verification(job_id, new_oid, new_oid)
return self.controller.store._receipt(row, duplicate=True).as_dict()
finally:
self.credential.revoke()
@@ -0,0 +1,26 @@
"""Non-sensitive control-plane snapshot assembly."""
from __future__ import annotations
from typing import Any, Mapping
def _copy(value: Mapping[str, Any] | None, keys: tuple[str, ...]) -> dict[str, Any]:
value = value or {}
return {key: value.get(key) for key in keys}
def build_control_plane_snapshot(*, target_version: str, runtime: Mapping[str, Any] | None = None, repository: Mapping[str, Any] | None = None, production: Mapping[str, Any] | None = None, task: Mapping[str, Any] | None = None, capabilities: Mapping[str, Any] | None = None, credential_broker: Mapping[str, Any] | None = None) -> dict[str, Any]:
return {
"version": {"target_release": target_version},
"runtime": _copy(runtime, ("bridge_schema_version", "control_plane_release_version", "runtime_release_commit", "runtime_path", "artifact_sha256", "drift_status")),
"repository": _copy(repository, ("identity", "authority_git_dir", "authority_main_oid")),
"production": _copy(production, ("baseline_version", "target_oid", "deployment_state")),
"task": _copy(task, ("task_id", "branch", "authority_branch_oid", "gitea_git_oid", "gitea_api_oid", "commit_visible", "generation", "reconciliation_state")),
"capabilities": _copy(capabilities, ("legacy_writes_enabled", "governed_writes_enabled", "force_push", "main_write", "tag_write", "baseline_write", "production_write")),
"credential_broker": _copy(credential_broker, ("status", "provider")),
"secret_material_exposed": False,
}
def getControlPlaneSnapshot(**kwargs: Any) -> dict[str, Any]:
return build_control_plane_snapshot(**kwargs)
@@ -0,0 +1,78 @@
"""Metadata-only, server-side credential lease broker."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import os
from pathlib import Path
import secrets
from typing import Any, Callable, Mapping
@dataclass(frozen=True, slots=True)
class CredentialLease:
lease_id: str
task_id: str
repository: str
target_branch: str
generation: int
issued_at: str
expires_at: str
scopes: tuple[str, ...]
provider: str
state: str = "ACTIVE"
def as_dict(self) -> dict[str, Any]:
return {"lease_id": self.lease_id, "task_id": self.task_id, "repository": self.repository, "target_branch": self.target_branch, "generation": self.generation, "issued_at": self.issued_at, "expires_at": self.expires_at, "scopes": list(self.scopes), "provider": self.provider, "state": self.state}
class FileCredentialProvider:
def __init__(self, token_path: str | os.PathLike[str], *, actor: str = "aios-autonomy-agent") -> None:
self.token_path = Path(token_path)
self.actor = actor
def environment(self, lease: CredentialLease) -> dict[str, str]:
if not self.token_path.is_file():
raise RuntimeError("CREDENTIAL_PROVIDER_UNAVAILABLE")
return {"CREDENTIAL_FILE": str(self.token_path), "GITEA_MACHINE_ACTOR": self.actor}
class CredentialBroker:
def __init__(self, provider: Any, *, now: Callable[[], datetime] | None = None) -> None:
self.provider = provider
self.now = now or (lambda: datetime.now(timezone.utc))
self._leases: dict[str, CredentialLease] = {}
def issue(self, *, task_id: str, repository: str, target_branch: str, generation: int, scopes: tuple[str, ...] = ("git:push", "gitea:read"), lease_seconds: int = 900) -> dict[str, Any]:
if repository != "aios/aios" or not target_branch.startswith("agent/") or generation < 1 or not 1 <= lease_seconds <= 86400:
raise ValueError("CREDENTIAL_LEASE_BINDING_INVALID")
now = self.now().astimezone(timezone.utc)
lease = CredentialLease("lease-" + secrets.token_hex(16), task_id, repository, target_branch, generation, now.isoformat(), (now + timedelta(seconds=lease_seconds)).isoformat(), tuple(scopes), type(self.provider).__name__)
self._leases[lease.lease_id] = lease
return lease.as_dict()
def _get(self, lease_id: str) -> CredentialLease:
lease = self._leases.get(lease_id)
if lease is None:
raise KeyError("CREDENTIAL_LEASE_NOT_FOUND")
if lease.state == "ACTIVE" and self.now().astimezone(timezone.utc) >= datetime.fromisoformat(lease.expires_at):
lease = CredentialLease(**{**lease.as_dict(), "scopes": tuple(lease.scopes), "state": "EXPIRED"})
self._leases[lease_id] = lease
return lease
def environment(self, lease_id: str) -> Mapping[str, str]:
lease = self._get(lease_id)
if lease.state != "ACTIVE":
raise RuntimeError("CREDENTIAL_LEASE_NOT_ACTIVE")
return self.provider.environment(lease)
def status(self, lease_id: str) -> dict[str, Any]:
return self._get(lease_id).as_dict()
def revoke(self, lease_id: str) -> dict[str, Any]:
lease = self._get(lease_id)
if lease.state == "REVOKED":
return lease.as_dict()
updated = CredentialLease(**{**lease.as_dict(), "scopes": tuple(lease.scopes), "state": "REVOKED"})
self._leases[lease_id] = updated
return updated.as_dict()
@@ -0,0 +1,93 @@
"""Canonical Gitea transport and slash-branch verification."""
from __future__ import annotations
from dataclasses import dataclass
import json
import os
from pathlib import Path
import subprocess
from typing import Any, Callable, Mapping, Optional
from urllib.parse import quote
from urllib.request import Request, urlopen
from aios_gitops_reconciliation_core import RemoteObservation
@dataclass(frozen=True, slots=True)
class GiteaPublisherConfig:
repository: str = "aios/aios"
api_base: str = "https://g.cnnetmkt.top/git/api/v1"
remote_name: str = "gitea-http"
class GiteaPublisher:
def __init__(self, workspace: Path, *, config: GiteaPublisherConfig = GiteaPublisherConfig(), credential: Any = None, api_get: Callable[[str], Any] | None = None, git_ref: Callable[[str], Optional[str]] | None = None) -> None:
self.workspace = Path(workspace)
self.config = config
self.credential = credential
self._api_get = api_get
self._git_ref = git_ref
@staticmethod
def validate_repository(repository: str) -> None:
if repository != "aios/aios":
raise ValueError("REPOSITORY_IDENTITY_MISMATCH")
@staticmethod
def validate_branch(branch: str) -> None:
if not isinstance(branch, str) or not branch.startswith("agent/") or branch in {"main", "master"}:
raise ValueError("BRANCH_SCOPE_DENIED")
def repository_identity(self) -> str:
self.validate_repository(self.config.repository)
return self.config.repository
def _environment(self) -> dict[str, str]:
if self.credential is not None and hasattr(self.credential, "environment"):
return dict(self.credential.environment())
return {}
def read_git_ref(self, branch: str) -> Optional[str]:
self.validate_branch(branch)
if self._git_ref is not None:
return self._git_ref(branch)
result = subprocess.run(["git", "-C", str(self.workspace), "ls-remote", self.config.remote_name, f"refs/heads/{branch}"], text=True, capture_output=True, check=False, env={**os.environ, **self._environment(), "GIT_TERMINAL_PROMPT": "0"})
if result.returncode != 0 or not result.stdout.strip():
return None
return result.stdout.split()[0]
def _api(self, path: str) -> Any:
if self._api_get is not None:
return self._api_get(path)
headers = {"Accept": "application/json"}
env = self._environment()
if env.get("AIOS_GITEA_TOKEN"):
headers["Authorization"] = "token " + env["AIOS_GITEA_TOKEN"]
with urlopen(Request(self.config.api_base.rstrip("/") + path, headers=headers), timeout=20) as response:
return json.loads(response.read() or b"{}")
def read_api_branch(self, branch: str) -> Optional[str]:
self.validate_branch(branch)
for page in range(1, 101):
value = self._api(f"/repos/aios/aios/branches?page={page}&limit=50")
if not isinstance(value, list):
return None
for item in value:
if isinstance(item, Mapping) and item.get("name") == branch:
commit = item.get("commit")
return commit.get("id") if isinstance(commit, Mapping) else None
if len(value) < 50:
return None
raise RuntimeError("GITEA_BRANCH_PAGINATION_LIMIT")
def commit_visible(self, oid: str) -> bool:
if not isinstance(oid, str) or len(oid) != 40:
return False
try:
value = self._api(f"/repos/aios/aios/git/commits/{quote(oid, safe='')}")
return isinstance(value, Mapping) and value.get("sha", oid) == oid
except Exception:
return False
def observe(self, branch: str, desired_oid: str, authority_oid: Optional[str], *, expected_oid: Optional[str] = None, observed_at: str = "") -> RemoteObservation:
return RemoteObservation(self.repository_identity(), branch, desired_oid, authority_oid, self.read_git_ref(branch), self.read_api_branch(branch), self.commit_visible(desired_oid), observed_at, expected_oid)
@@ -249,7 +249,8 @@ class JobStore:
if state == JobState.COMPLETED:
status, side = 200, "COMPLETED"
elif state in _RECOVERABLE:
status, side = 202, "RESUME_SAME_JOB"
status = 202
side = row["side_effect_state"] if row["side_effect_state"] == "REMOTE_PUSH_PENDING" else "RESUME_SAME_JOB"
elif row["side_effect_state"] == "UNKNOWN":
status, side = 503, "UNKNOWN"
else:
@@ -280,10 +281,13 @@ class GitOpsJobController:
# durable round trip and participate in idempotency. Arbitrary request
# fields are intentionally discarded rather than persisted.
optional = (
"idempotency_key",
"task_id",
"new_branch_oid",
"authorization_recovery_digest",
"lineage_anchor_oid",
"base_oid",
"branch_mode",
"generation",
"expected_main_oid",
"pr_number",
@@ -314,7 +318,15 @@ class GitOpsJobController:
raise JobError("PAYLOAD_REFERENCE_MISMATCH")
if payload_ref is not None and (not isinstance(payload_size, int) or payload_size < 0):
raise JobError("PAYLOAD_SIZE_INVALID")
key = sha256_json(binding)
client_idempotency_key = binding.get("idempotency_key")
if client_idempotency_key is not None:
if len(client_idempotency_key) > 128 or not all(
character.isalnum() or character in "._:-" for character in client_idempotency_key
):
raise JobError("IDEMPOTENCY_KEY_INVALID")
key = sha256_json({"client_idempotency_key": client_idempotency_key})
else:
key = sha256_json(binding)
job_id = "job-" + key[:32]
durable_request = dict(binding)
if payload_ref is not None:
@@ -0,0 +1,80 @@
"""Pure, observation-first decisions for governed GitOps reconciliation."""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
import re
from typing import Optional
OID_RE = re.compile(r"^[0-9a-f]{40}$")
class ReconciliationAction(str, Enum):
NOOP = "NOOP"
PUSH_EXACT = "PUSH_EXACT"
VERIFY_REMOTE = "VERIFY_REMOTE"
MARK_REMOTE_VERIFIED = "MARK_REMOTE_VERIFIED"
MARK_RECONCILIATION_REQUIRED = "MARK_RECONCILIATION_REQUIRED"
FAIL_CONFLICT = "FAIL_CONFLICT"
@dataclass(frozen=True, slots=True)
class RemoteObservation:
repository_identity: str
branch: str
desired_oid: str
authority_oid: Optional[str]
remote_git_oid: Optional[str]
remote_api_oid: Optional[str]
commit_visible: bool
observed_at: str
expected_oid: Optional[str] = None
@property
def all_equal(self) -> bool:
values = (self.authority_oid, self.remote_git_oid, self.remote_api_oid, self.desired_oid)
return all(isinstance(value, str) for value in values) and len(set(values)) == 1
@dataclass(frozen=True, slots=True)
class ReconciliationDecision:
action: ReconciliationAction
state: str
reason: str
retry_safe: bool = False
def _valid_oid(value: Optional[str]) -> bool:
return value is None or bool(OID_RE.fullmatch(value))
def decide_reconciliation(observation: RemoteObservation) -> ReconciliationDecision:
"""Decide from a fresh observation; never complete on a missing remote."""
if observation.repository_identity != "aios/aios":
return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "REPOSITORY_IDENTITY_MISMATCH")
if not observation.branch.startswith("agent/"):
return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "BRANCH_SCOPE_DENIED")
if not _valid_oid(observation.desired_oid) or not _valid_oid(observation.expected_oid):
return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "OID_INVALID")
if observation.authority_oid is None:
return ReconciliationDecision(ReconciliationAction.MARK_RECONCILIATION_REQUIRED, "RECONCILIATION_REQUIRED", "AUTHORITY_UNAVAILABLE")
if observation.authority_oid != observation.desired_oid:
return ReconciliationDecision(ReconciliationAction.MARK_RECONCILIATION_REQUIRED, "RECONCILIATION_REQUIRED", "AUTHORITY_CONFLICT")
present = [oid for oid in (observation.remote_git_oid, observation.remote_api_oid) if oid is not None]
if any(oid not in {observation.desired_oid, observation.expected_oid} for oid in present):
return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "REMOTE_OID_CONFLICT")
if observation.all_equal and observation.commit_visible:
return ReconciliationDecision(ReconciliationAction.MARK_REMOTE_VERIFIED, "REMOTE_VERIFIED", "ALL_OIDS_EQUAL_AND_VISIBLE")
if observation.remote_git_oid is None and observation.remote_api_oid is None:
return ReconciliationDecision(ReconciliationAction.MARK_RECONCILIATION_REQUIRED, "REMOTE_PUSH_PENDING", "AUTHORITY_DESIRED_REMOTE_MISSING")
if observation.remote_git_oid == observation.desired_oid or observation.remote_api_oid == observation.desired_oid:
return ReconciliationDecision(ReconciliationAction.VERIFY_REMOTE, "REMOTE_VERIFY_PENDING", "REMOTE_PROOF_PARTIAL")
if observation.expected_oid is not None and observation.remote_git_oid == observation.expected_oid and observation.remote_api_oid == observation.expected_oid:
return ReconciliationDecision(ReconciliationAction.PUSH_EXACT, "REMOTE_PUSH_PENDING", "SAFE_EXACT_FAST_FORWARD_RETRY", retry_safe=True)
if not observation.commit_visible:
return ReconciliationDecision(ReconciliationAction.VERIFY_REMOTE, "COMMIT_VISIBILITY_PENDING", "COMMIT_VISIBILITY_LAG")
return ReconciliationDecision(ReconciliationAction.MARK_RECONCILIATION_REQUIRED, "RECONCILIATION_REQUIRED", "PARTIAL_REMOTE_OBSERVATION")
def reconcile_observation(**kwargs: object) -> ReconciliationDecision:
return decide_reconciliation(RemoteObservation(**kwargs))
@@ -0,0 +1,59 @@
"""Central fail-closed policy seam for governed task and continuation requests."""
from __future__ import annotations
from dataclasses import dataclass
import re
from typing import Any, Mapping
OID_RE = re.compile(r"^[0-9a-f]{40}$")
@dataclass(frozen=True, slots=True)
class GovernedPolicyInput:
identity: str
task_id: str
repository: str
operation: str
base_oid: str
expected_branch_oid: str
target_branch: str
generation: int
risk: str = "engineering"
requested_capabilities: tuple[str, ...] = ()
previous_generation: int | None = None
@dataclass(frozen=True, slots=True)
class PolicyDecision:
decision: str
reason: str
def as_dict(self) -> dict[str, str]:
return {"decision": self.decision, "reason": self.reason}
def evaluate_policy(value: GovernedPolicyInput | Mapping[str, Any]) -> PolicyDecision:
if isinstance(value, Mapping):
value = GovernedPolicyInput(**value)
if value.repository != "aios/aios":
return PolicyDecision("DENY", "REPOSITORY_IDENTITY_MISMATCH")
if not OID_RE.fullmatch(value.base_oid) or not OID_RE.fullmatch(value.expected_branch_oid):
return PolicyDecision("DENY", "OID_INVALID")
if not value.target_branch.startswith("agent/") or value.target_branch in {"main", "master"}:
return PolicyDecision("DENY", "BRANCH_SCOPE_DENIED")
if "force_push" in value.requested_capabilities or "force" in value.operation.lower():
return PolicyDecision("DENY", "FORCE_PUSH_DENIED")
if {"main_write", "tag_write", "baseline_write", "production_write", "production_deploy", "tag_mutation", "baseline_mutation"}.intersection(value.requested_capabilities):
return PolicyDecision("DENY", "MUTATION_SCOPE_DENIED")
if not isinstance(value.generation, int) or isinstance(value.generation, bool) or value.generation < 1:
return PolicyDecision("DENY", "GENERATION_INVALID")
if value.previous_generation is not None and value.generation <= value.previous_generation:
return PolicyDecision("MANUAL_OWNER_REVIEW_REQUIRED", "GENERATION_NOT_MONOTONIC")
if not value.task_id:
return PolicyDecision("DENY", "TASK_ID_INVALID")
return PolicyDecision("ALLOW", "GOVERNED_ENGINEERING_SCOPE")
class GovernedPolicy:
def decide(self, request: GovernedPolicyInput | Mapping[str, Any]) -> PolicyDecision:
return evaluate_policy(request)
@@ -0,0 +1,84 @@
"""Durable governed task binding for CREATE_NEW and CONTINUE_EXISTING."""
from __future__ import annotations
from dataclasses import dataclass, asdict
import hashlib
import json
from pathlib import Path
import re
from typing import Any, Mapping
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
expected_branch_oid: str
branch_mode: str
generation: int
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.expected_branch_oid, "generation": self.generation, "branch_mode": self.branch_mode, "create_target_branch": self.branch_mode == "CREATE_NEW", "worktree_mode": "detached_at_base_oid" if self.branch_mode == "CREATE_NEW" else "detached_at_expected_head", "cas_ref_update_required": True, "idempotent": idempotent, "state": "BOUND", "job_id": self.job_id}
def _safe(value: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "-", value).strip("-")[:120] or "task"
class GovernedTaskBootstrap:
def __init__(self, authority: Any, 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 = Path(root); self.remote_name = remote_name; self.token_path = Path(token_path); self.ssh_key_path = Path(ssh_key_path); self.api_base = api_base; self.task_meta_root = self.root / "task-bindings"; self.job_root = self.root / "jobs"; self.task_meta_root.mkdir(parents=True, exist_ok=True); self.job_root.mkdir(parents=True, exist_ok=True)
def _store(self, task_id: str) -> tuple[JobStore, GitOpsJobController]:
root = self.job_root / _safe(task_id); store = JobStore(root, payload_root=root / "payloads"); return store, GitOpsJobController(store)
def _metadata(self, task_id: str) -> Path:
return self.task_meta_root / (_safe(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
return GovernedTaskBinding(**json.loads(path.read_text(encoding="utf-8")))
def _write(self, binding: GovernedTaskBinding) -> None:
path = self._metadata(binding.task_id); temporary = path.with_suffix(".tmp"); temporary.write_text(json.dumps(asdict(binding), sort_keys=True), encoding="utf-8"); temporary.replace(path)
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"]): raise BootstrapError("TARGET_BRANCH_INVALID")
existing = self._read_existing(values["task_id"])
if existing is not None:
if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description")): 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")
current = self.authority.read_ref("refs/heads/" + values["target_branch"])
expected = str(request.get("expected_branch_oid") or values["base_oid"])
if current is None:
mode, generation = "CREATE_NEW", 1
if expected != values["base_oid"]: raise BootstrapError("EXPECTED_BASE_CONFLICT")
else:
if expected != current: raise BootstrapError("EXISTING_BRANCH_OID_CONFLICT")
mode, generation = "CONTINUE_EXISTING", int(request.get("generation", 2))
if generation < 1: raise BootstrapError("GENERATION_INVALID")
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": expected, "payload_digest": hashlib.sha256(payload).hexdigest(), "lineage_anchor_oid": values["base_oid"]}, payload=payload)
binding = GovernedTaskBinding(**values, expected_branch_oid=expected, branch_mode=mode, generation=generation, job_id=receipt.job_id)
self._write(binding)
return binding.response()
@@ -0,0 +1,23 @@
[Unit]
Description=AIOS ChatGPT Bridge v2 staging control plane
After=network-online.target
[Service]
Type=simple
User=aios-bridge-staging
Group=aios-bridge-staging
WorkingDirectory=/opt/aios/gitops-control-plane/current
Environment=AIOS_BRIDGE_SCHEMA_VERSION=aios.chatgpt-gitops-bridge.v2
Environment=AIOS_GITOPS_WRITES_ENABLED=false
Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
Environment=AIOS_DEPLOYMENT_STATE=STAGING
ExecStart=/usr/bin/python3 /opt/aios/gitops-control-plane/current/bridge-v2-staging-server.py
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/var/lib/aios-gitops/bridge-v2-staging
Restart=on-failure
[Install]
WantedBy=multi-user.target
@@ -1,6 +1,20 @@
# Merge this fragment into docs/gitops/openapi-chatgpt-actions.yaml.
components:
schemas:
WriterLeaseRequestV2:
type: object
additionalProperties: false
required: [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds]
properties:
schema: {type: string, const: aios.task-writer-lease.v2}
task_id: {type: string}
repository: {type: string, const: aios/aios}
branch_mode: {type: string, const: CONTINUE_EXISTING}
base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
target_branch: {type: string, pattern: '^agent/'}
generation: {type: integer, minimum: 1}
lease_seconds: {type: integer, minimum: 1, maximum: 86400}
AuthorizedTaskWriterProvisionRequest:
type: object
additionalProperties: false
@@ -90,6 +104,20 @@ components:
- production_marker_write
- history_rewrite
ControlPlaneSnapshot:
type: object
additionalProperties: false
required: [version, runtime, repository, production, task, capabilities, credential_broker, secret_material_exposed]
properties:
version: {type: object}
runtime: {type: object}
repository: {type: object}
production: {type: object}
task: {type: object}
capabilities: {type: object}
credential_broker: {type: object}
secret_material_exposed: {type: boolean, const: false}
paths:
/api/task-writer/provision-authorized:
post:
@@ -119,3 +147,10 @@ paths:
description: Base drift, branch conflict, or idempotency scope conflict
'503':
description: Signer, validator, provisioner, or attestation unavailable
/v2/control-plane-snapshot:
get:
operationId: getControlPlaneSnapshot
responses:
'200':
description: Non-sensitive control-plane snapshot
+50
View File
@@ -567,3 +567,53 @@ paths:
'200': {description: Safe GitOps plan}
'401': {description: Unauthorized}
x-openai-isConsequential: true
/api/v2/governed-capabilities:
get:
operationId: getGovernedWriteCapabilities
responses: {'200': {description: Governed write capabilities with legacy writes disabled}}
/api/v2/governed-task:
post:
operationId: createGovernedTask
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [task_id, repository, base_oid, target_branch, risk, description, idempotency_key], properties: {task_id: {type: string}, repository: {type: string, const: aios/aios}, base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}, target_branch: {type: string, pattern: '^agent/'}, risk: {type: string}, description: {type: string}, idempotency_key: {type: string}}}}}}
responses: {'200': {description: Exact governed task binding}}
x-openai-isConsequential: true
/api/v2/writer-lease:
post:
operationId: acquireWriterLease
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds], properties: {schema: {type: string}, task_id: {type: string}, repository: {type: string, const: aios/aios}, branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}, base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}, expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}, target_branch: {type: string, pattern: '^agent/'}, generation: {type: integer, minimum: 1}, lease_seconds: {type: integer, minimum: 1, maximum: 86400}}}}}}
responses: {'200': {description: Detached CONTINUE_EXISTING lease}}
x-openai-isConsequential: true
/api/v2/commit-file:
post:
operationId: commitFileGoverned
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, path, content_base64, message], properties: {task_id: {type: string}, repository: {type: string, const: aios/aios}, expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}, target_branch: {type: string, pattern: '^agent/'}, generation: {type: integer, minimum: 1}, idempotency_key: {type: string}, path: {type: string}, content_base64: {type: string}, message: {type: string}}}}}}
responses: {'202': {description: Durable governed Job}}
x-openai-isConsequential: true
/api/v2/commit-files:
post:
operationId: commitFilesGoverned
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message], properties: {task_id: {type: string}, repository: {type: string, const: aios/aios}, expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}, target_branch: {type: string, pattern: '^agent/'}, generation: {type: integer, minimum: 1}, idempotency_key: {type: string}, files: {type: object, additionalProperties: {type: string}}, message: {type: string}}}}}}
responses: {'202': {description: One atomic multi-file governed Job}}
x-openai-isConsequential: true
/api/v2/job:
post:
operationId: getGitOpsJob
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [job_id], properties: {job_id: {type: string}}}}}}
responses: {'200': {description: Non-sensitive Job metadata}}
/api/v2/job/resume:
post:
operationId: resumeGitOpsJob
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [job_id, expected_revision], properties: {job_id: {type: string}, expected_revision: {type: integer, minimum: 0}}}}}}
responses: {'202': {description: Revision-CAS resume}}
x-openai-isConsequential: true
/api/v2/job/reconcile:
post:
operationId: reconcileGitOpsJob
requestBody: {required: true, content: {application/json: {schema: {type: object, additionalProperties: false, required: [job_id, expected_revision], properties: {job_id: {type: string}, expected_revision: {type: integer, minimum: 0}}}}}}
responses: {'202': {description: Observation-first reconciliation}}
x-openai-isConsequential: true
/v2/control-plane-snapshot:
get:
operationId: getControlPlaneSnapshot
responses: {'200': {description: Non-sensitive control-plane snapshot}}
@@ -0,0 +1,15 @@
# Bridge v2.1 Durable Reconciliation Bootstrap
This is the planning-only bootstrap artifact for
`TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809`.
The task is bound to the exact base commit
`3f670c387692b8865a10d5b419468d8a0b64d7a4` and the development branch
`agent/p0-bridge-v21-durable-reconciliation-20260809`.
The first governed commit records the design boundary and acceptance contract
only. It does not implement the v2.1 body, mutate main, create tags, change
Baseline, or touch production. Future work must preserve absent-ref CAS,
observation-first recovery, durable JobStore state, exact Authority/Gitea/API
reconciliation, and the existing secret-scanning and credential-projection
boundaries.
@@ -0,0 +1,150 @@
# ChatGPT Bridge v2 Control Plane Convergence Design
Date: 2026-08-09
Task: `TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809`
Repository: `aios/aios`
Approved development base: `2e8fa22989808a448065d1d76b5b18948774a637`
Target branch: `agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809`
## Scope and authority
This design converges the ChatGPT GitOps Bridge, the existing Authorized Task Writer, the continuation lease protocol, the Durable GitOps Job Controller, and immutable control-plane runtime identity.
The development base is exact and must never be replaced with a dynamically resolved `HEAD`, `latest`, or `main`. If Authority `refs/heads/main` differs from the approved OID before the first governed writer push, execution fails with `BASE_OID_CHANGED`.
This task does not authorize production changes. The following remain immutable:
- Production Baseline `V2R1C7`.
- Production target OID `c7adece1cc8beffe7c718d53c775898b45d6c7f8`.
- Production tag OID `82792347937c0df59a8e02d6162775252bcad2e2`.
- Production Registry, services, deployment state, runtime, tags, and release evidence.
- Authority `main`.
Force push, history rewrite, direct main push, production mutation, tag mutation, baseline mutation, and secret commits are always denied.
## Immutable control-plane runtime identity
A control-plane release is installed into a versioned immutable directory equivalent to:
```text
/opt/aios/gitops-control-plane/<release-id-or-oid>/
```
The release manifest binds:
```text
control_plane_release_version
control_plane_target_oid
artifact_sha256
runtime_path
bridge_schema_version
repository_identity
authoritative_git_dir
deployment_state
```
The validator is independent from the mutable Git workspace. It rejects execution when the runtime path is a workspace/worktree, the observed runtime commit or artifact digest differs from the manifest, the repository identity is not `aios/aios`, the authoritative Git directory is absent or mismatched, or the Bridge schema version is absent or unsupported.
Every failure is fail closed with `CONTROL_PLANE_RUNTIME_DRIFT`. No fallback to a mutable workspace is permitted. The installer and systemd template are staging-safe; this task may install and restart only a staging/test service.
## Bridge v2 adapter
The Bridge schema identity is `aios.chatgpt-gitops-bridge.v2`. It is independent from the production baseline, control-plane release version, and runtime release commit.
The Bridge exposes:
- `getRuntimeStatus`
- `getGovernedWriteCapabilities`
- `createGovernedTask`
- `acquireWriterLease`
- `commitFileGoverned`
- `commitFilesGoverned`
- `getGitOpsJob`
- `resumeGitOpsJob`
- `reconcileGitOpsJob`
The adapter reuses existing mechanisms and does not create a second writer, lease protocol, branch state store, job database, or direct Git mutation path.
## Governed task and lease flow
`createGovernedTask` binds an exact task ID, repository, approved base OID, non-main branch, risk, and description. Existing branches are read back and bound to their observed full OID. Idempotent replay returns the original binding; conflicting replay fails closed.
`acquireWriterLease` uses `WriterLeaseRequestV2`. Continuation uses:
```text
branch_mode=CONTINUE_EXISTING
expected_branch_oid=<full OID>
generation=<monotonic integer>
create_target_branch=false
worktree_mode=detached_at_expected_head
cas_ref_update_required=true
```
Credentials remain server-side and never appear in responses, Git, durable job metadata, or logs.
## Governed commit flow
`commitFileGoverned` and `commitFilesGoverned` submit work to the existing `GitOpsJobController` and `JobStore`. The server computes payload digests. Raw file content is stored only in the content-addressed payload domain. Secret scanning, protected-path policy, repository binding, lease validation, and branch authorization run before side effects.
`commitFilesGoverned` is one atomic Git commit and enforces bounded file count, per-file size, and total payload size. Any failure leaves no partial commit or ref update.
All ref updates use compare-and-swap. Stale `expected_branch_oid` returns conflict and never triggers automatic rebase. After push, Authority and remote Gitea refs are independently read and must equal the resulting commit OID.
## Durable job lifecycle
Job reads return only non-sensitive metadata: `job_id`, `state`, `revision`, `side_effect_state`, `retryable`, `binding`, and `receipt_summary`.
Terminal jobs cannot be changed by ordinary resume. Resume requires `job_id` and `expected_revision` and uses revision CAS. Reconciliation is observation-first: for `UNKNOWN`, `RECONCILIATION_REQUIRED`, or `REMOTE_PUSH_PENDING`, local, Authority, and remote refs are read before state changes. A blind repeated push is never reconciliation.
## Legacy behavior
Legacy writes remain disabled while governed writes are enabled:
```text
legacy_writes_enabled=false
governed_writes_enabled=true
```
The global `AIOS_GITOPS_WRITES_ENABLED` flag is never enabled as the solution. Legacy operations remain temporarily visible for compatibility and diagnostics, but return:
```json
{
"error": "LEGACY_WRITE_DISABLED",
"replacement": "createGovernedTask",
"governed_writes_available": true
}
```
## OpenAPI convergence
`docs/gitops/openapi-chatgpt-actions.yaml` is the canonical Bridge v2 schema and integrates the Authorized Task Writer fragment. The canonical CAS field is `expected_branch_oid`. `expected_head_oid` may be accepted only at an explicit deprecated adapter boundary.
Contract tests require exact equality between OpenAPI required fields, runtime parser required fields, and generated GPT Action schema required fields.
## Acceptance
Automated coverage includes runtime path/OID/digest/repository drift, exact task/branch/OID/lease/CAS behavior, job DB failure, restart during remote verification, remote timeout, response loss after remote success, idempotency replay/conflict, concurrent branch movement, unauthorized or main branch writes, protected paths, secret-like payloads, disabled governed writer, missing/expired lease, wrong repository identity, and observation-first reconciliation.
The staging sequence is:
```text
getRuntimeStatus
getGovernedWriteCapabilities
createGovernedTask
acquireWriterLease
commitFilesGoverned
getGitOpsJob
verify Authority branch ref
verify remote Gitea branch ref
restart staging control-plane service
getRuntimeStatus
acquire continuation lease
second governed commit
verify remote ref again
```
Completion requires all tests, staging restart continuation, Authority/remote OID equality, unchanged production invariants, no force push, no tag rewrite, no direct main push, and no secret commit.
## Implementation boundaries
Implementation is limited to the Bridge, existing writer and job-controller adapters, runtime identity/installer, staging systemd template, OpenAPI documents, tests, and engineering evidence. Hermes Adapter integration and production cutover are outside this task and require separate approval.
@@ -0,0 +1,44 @@
{
"schema": "aios.bridge-v2-engineering-staging-acceptance.v1",
"task_id": "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809",
"repository": "aios/aios",
"branch": "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809",
"control_plane_release_version": "bridge-v2-aa0c37b8ab33",
"control_plane_target_oid": "aa0c37b8ab333616e1bab9686cf6d10557f2ebda",
"artifact_sha256": "08670c78cf760bfe693e3553e676eb751dadd7834116aa6d40e7c094f1357ae4",
"runtime_path": "/opt/aios/gitops-control-plane/aa0c37b8ab333616e1bab9686cf6d10557f2ebda",
"bridge_schema_version": "aios.chatgpt-gitops-bridge.v2",
"deployment_state": "STAGING",
"drift_status": "PASS",
"restart_acceptance": {
"pid_before": 3158850,
"pid_after": 3159889,
"runtime_response_equal": true
},
"governed_capabilities": {
"legacy_writes_enabled": false,
"governed_writes_enabled": true,
"force_push": false,
"main_write": false,
"tag_write": false,
"baseline_write": false,
"production_write": false
},
"production_invariants": {
"authority_main_oid": "2e8fa22989808a448065d1d76b5b18948774a637",
"authority_tags_sha256": "bd2b6e30b76f28e7fcdcb447e7296690f0a65ad855209294c36b0f3f1d96098d",
"baseline_current_sha256": "57651eb819584ab7c1d67f01b0ef66d6408cdc322ae5f6d4ff0aacd96f8327e1",
"production_service_main_pid": 2816107,
"production_service_nrestarts": 0,
"production_service_active_enter_timestamp_monotonic": 2919386492273
},
"bridge_operation_sequence": {
"createGovernedTask": "PASS",
"acquireWriterLease": "PASS",
"commitFilesGoverned": "REAL_GOVERNED_COMMIT",
"getGitOpsJob": "PASS",
"expected_branch_oid": "8fe3b85c1be592326473e81a502a23eeb9511d30",
"generation": 7
},
"production_changed": false
}
@@ -0,0 +1,17 @@
{
"schema": "aios.bridge-v2.1-durable-reconciliation.bootstrap-acceptance.v1",
"task_id": "TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
"repository": "aios/aios",
"base_oid": "3f670c387692b8865a10d5b419468d8a0b64d7a4",
"target_branch": "agent/p0-bridge-v21-durable-reconciliation-20260809",
"branch_mode": "CREATE_NEW",
"generation": 1,
"implementation": "not_started",
"production_changed": false,
"required_proofs": [
"absent_ref_cas",
"observation_first_reconciliation",
"authority_gitea_api_commit_equality",
"commit_visible"
]
}
+90
View File
@@ -0,0 +1,90 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import tempfile
import pytest
import yaml
from aios_chatgpt_bridge_v2 import (
BRIDGE_OPERATIONS,
BRIDGE_SCHEMA_VERSION,
OPERATION_REQUIRED_FIELDS,
BridgeV2Adapter,
ControlPlaneReleaseManifest,
RuntimeDriftError,
)
ROOT = Path(__file__).parents[1]
def test_immutable_release_identity_accepts_exact_artifact_and_rejects_workspace_drift():
with tempfile.TemporaryDirectory() as directory:
approved = Path(directory) / "releases"
runtime = approved / "release-1"
runtime.mkdir(parents=True)
artifact = runtime / "bridge.pyz"
artifact.write_bytes(b"bridge-v2")
digest = hashlib.sha256(artifact.read_bytes()).hexdigest()
manifest = ControlPlaneReleaseManifest(
"bridge-v2-test", "a" * 40, digest, runtime, BRIDGE_SCHEMA_VERSION,
"aios/aios", Path("/data/aios-gitops/repo.git"), "STAGING",
)
assert manifest.validate(artifact, observed_oid="a" * 40, approved_root=approved)["drift_status"] == "PASS"
with pytest.raises(RuntimeDriftError, match="CONTROL_PLANE_RUNTIME_DRIFT"):
manifest.validate(artifact, observed_oid="b" * 40, approved_root=approved)
mutable = ControlPlaneReleaseManifest(
"bad", "a" * 40, digest, Path("/var/lib/worktrees/dev"), BRIDGE_SCHEMA_VERSION,
"aios/aios", Path("/data/aios-gitops/repo.git"), "STAGING",
)
with pytest.raises(RuntimeDriftError, match="CONTROL_PLANE_RUNTIME_DRIFT"):
mutable.validate(artifact, observed_oid="a" * 40, approved_root=approved)
def test_bridge_capabilities_keep_legacy_closed_and_expose_exact_operations():
adapter = BridgeV2Adapter(runtime_status={"drift_status": "PASS"})
assert adapter.getRuntimeStatus()["bridge_schema_version"] == BRIDGE_SCHEMA_VERSION
capabilities = adapter.getGovernedWriteCapabilities()
assert capabilities["legacy_writes_enabled"] is False
assert capabilities["governed_writes_enabled"] is True
assert tuple(capabilities["operations"]) == BRIDGE_OPERATIONS
assert adapter.legacy_write_disabled() == {
"error": "LEGACY_WRITE_DISABLED",
"replacement": "createGovernedTask",
"governed_writes_available": True,
}
def test_openapi_runtime_and_fragment_required_fields_are_exactly_converged():
canonical = yaml.safe_load((ROOT / "docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8"))
fragment = yaml.safe_load((ROOT / "docs/gitops/openapi-authorized-task-writer-fragment.yaml").read_text(encoding="utf-8"))
operations = {}
for path, methods in canonical["paths"].items():
for method, value in methods.items():
if isinstance(value, dict) and value.get("operationId") in BRIDGE_OPERATIONS:
operations[value["operationId"]] = value
assert set(operations) == set(BRIDGE_OPERATIONS)
for operation, required in OPERATION_REQUIRED_FIELDS.items():
if "requestBody" not in operations[operation]:
assert required == ()
continue
schema = operations[operation]["requestBody"]["content"]["application/json"]["schema"]
assert tuple(schema["required"]) == required
fragment_required = tuple(fragment["components"]["schemas"]["WriterLeaseRequestV2"]["required"])
assert fragment_required == OPERATION_REQUIRED_FIELDS["acquireWriterLease"]
def test_release_manifest_round_trip_contains_only_declared_identity_fields():
manifest = ControlPlaneReleaseManifest(
"v", "a" * 40, "b" * 64, Path("/opt/aios/gitops-control-plane/v"), BRIDGE_SCHEMA_VERSION,
"aios/aios", Path("/data/aios-gitops/repo.git"), "STAGING",
)
value = manifest.as_dict()
assert tuple(value) == (
"control_plane_release_version", "control_plane_target_oid", "artifact_sha256", "runtime_path",
"bridge_schema_version", "repository_identity", "authoritative_git_dir", "deployment_state",
)
assert "credential" not in json.dumps(value).lower()
@@ -0,0 +1,283 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from pathlib import Path
import subprocess
import tempfile
import pytest
from aios_continuation_executor import (
ContinuationCommit,
ContinuationExecutionError,
ContinuationExecutor,
ContinuationPermissions,
ContinuationRuntimeBinding,
DetachedGitWorkspace,
GiteaContinuationRemote,
)
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import BindingConflict, GitOpsJobController, JobState, JobStore
TASK = "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809"
REPOSITORY = "aios/aios"
BRANCH = "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809"
BASE = "2" * 40
HEAD = "8" * 40
NEW = "9" * 40
NOW = datetime(2026, 8, 9, 12, 0, tzinfo=timezone.utc)
class Authority:
def __init__(self, *, head=HEAD, exists=True, descendant=True):
self.head = head
self.exists = exists
self.descendant = descendant
self.cas_calls = []
def branch_exists(self, branch):
return self.exists
def branch_head(self, branch):
return self.head
def is_ancestor(self, older, newer):
if (older, newer) == (BASE, HEAD):
return True
return self.descendant and (older, newer) in {(HEAD, NEW), (BASE, NEW)}
def compare_and_swap_branch(self, branch, expected, new):
self.cas_calls.append((branch, expected, new))
if self.head != expected:
raise RuntimeError("stale")
self.head = new
return new
class Workspace:
def __init__(self, *, detached=True, head=HEAD):
self.detached = detached
self.head = head
self.commits = 0
def is_detached(self):
return self.detached
def head_oid(self):
return self.head
def commit_files(self, files, message):
self.commits += 1
self.head = NEW
return NEW
class Remote:
def __init__(self, *, git=HEAD, api=HEAD, visible=True, repository=REPOSITORY, fail_after_push=False, omit_after_push=False):
self.git = git
self.api = api
self.visible = visible
self.repository = repository
self.fail_after_push = fail_after_push
self.omit_after_push = omit_after_push
self.pushes = 0
def repository_identity(self):
return self.repository
def push_exact(self, branch, expected, new):
self.pushes += 1
if self.git is None:
raise RuntimeError("remote branch missing")
if self.git != expected:
raise RuntimeError("remote conflict")
if not self.omit_after_push:
self.git = self.api = new
else:
self.git = self.api = None
if self.fail_after_push:
raise TimeoutError("response lost")
def read_git_ref(self, branch):
return self.git
def read_api_branch(self, branch):
return self.api
def commit_visible(self, oid):
return self.visible and self.api == oid
class Credential:
def __init__(self):
self.projected = 0
self.revoked = 0
def project(self):
self.projected += 1
def revoke(self):
self.revoked += 1
def lease_request(**changes):
values = dict(task_id=TASK, repository=REPOSITORY, branch_mode="CONTINUE_EXISTING", base_oid=BASE,
expected_branch_oid=HEAD, target_branch=BRANCH, generation=1, lease_seconds=900)
values.update(changes)
return WriterLeaseRequestV2(**values)
def binding(**changes):
values = dict(
request=lease_request(),
workspace_path=Path("/var/lib/aios-task-writer/continuation-worktrees/exact"),
lease_not_before=NOW - timedelta(seconds=1),
lease_expires_at=NOW + timedelta(minutes=15),
original_receipt_valid=True,
original_grant_consumed=True,
permissions=ContinuationPermissions(),
)
values.update(changes)
return ContinuationRuntimeBinding(**values)
def commit(**changes):
values = dict(idempotency_key="first-continuation", files={"active-runtime/app/new.py": b"ok\n"},
message="feat(gitops): add governed continuation executor")
values.update(changes)
return ContinuationCommit(**values)
def make_executor(authority=None, workspace=None, remote=None, credential=None):
temp = tempfile.TemporaryDirectory()
controller = GitOpsJobController(JobStore(Path(temp.name).resolve()))
executor = ContinuationExecutor(
controller, authority or Authority(), workspace or Workspace(), remote or Remote(), credential or Credential(),
now=lambda: NOW,
)
executor._test_temp = temp
return executor
def test_exact_existing_branch_completes_only_after_all_remote_proofs():
authority, remote, credential = Authority(), Remote(), Credential()
result = make_executor(authority, remote=remote, credential=credential).execute(binding(), commit())
assert result["state"] == JobState.COMPLETED.value
assert result["receipt"]["authority_oid"] == result["receipt"]["git_remote_oid"] == NEW
assert result["receipt"]["api_branch_oid"] == NEW
assert result["receipt"]["commit_visible"] is True
assert authority.cas_calls == [(BRANCH, HEAD, NEW)]
assert credential.projected == credential.revoked == 1
assert "credential" not in repr(result).lower()
@pytest.mark.parametrize(
"runtime_binding,authority,workspace,error",
[
(binding(request=lease_request(expected_branch_oid="7" * 40)), Authority(), Workspace(), "EXISTING_BRANCH_OID_CONFLICT"),
(binding(), Authority(exists=False), Workspace(), "EXISTING_BRANCH_MISSING"),
(binding(), Authority(), Workspace(detached=False), "WORKTREE_NOT_DETACHED"),
(binding(), Authority(), Workspace(head="7" * 40), "WORKTREE_HEAD_MISMATCH"),
(binding(lease_expires_at=NOW), Authority(), Workspace(), "LEASE_EXPIRED"),
(binding(request=lease_request(repository="other/repo")), Authority(), Workspace(), "request_identity_invalid"),
(binding(request=lease_request(target_branch="main")), Authority(), Workspace(), "PROTECTED_BRANCH_REJECTED"),
(binding(request=lease_request(generation=1)), Authority(), Workspace(), "GENERATION_NOT_MONOTONIC"),
],
)
def test_pre_side_effect_failures_never_create_or_update_a_branch(runtime_binding, authority, workspace, error):
executor = make_executor(authority, workspace=workspace)
if error == "GENERATION_NOT_MONOTONIC":
executor.observe_generation(BRANCH, 1)
with pytest.raises(Exception, match=error):
executor.execute(runtime_binding, commit())
assert authority.cas_calls == []
assert workspace.commits == 0
@pytest.mark.parametrize("field", ["create_target_branch", "force_push", "tag_write", "baseline_write", "production_write", "main_write"])
def test_forbidden_capabilities_fail_before_credential_projection(field):
credential = Credential()
permissions = ContinuationPermissions(**{field: True})
with pytest.raises(ContinuationExecutionError, match="MUTATION_PERMISSION_DENIED"):
make_executor(credential=credential).execute(binding(permissions=permissions), commit())
assert credential.projected == 0
def test_non_fast_forward_and_concurrent_movement_are_rejected_cas_only():
authority = Authority(descendant=False)
with pytest.raises(ContinuationExecutionError, match="NON_FAST_FORWARD_COMMIT"):
make_executor(authority).execute(binding(), commit())
assert authority.cas_calls == []
def test_authority_success_and_gitea_missing_is_never_completed():
remote = Remote(omit_after_push=True)
result = make_executor(remote=remote).execute(binding(), commit())
assert result["state"] == JobState.RECONCILIATION_REQUIRED.value
assert result["side_effect_state"] == "REMOTE_PUSH_PENDING"
assert result["state"] not in {JobState.REMOTE_VERIFIED.value, JobState.COMPLETED.value}
def test_response_loss_after_remote_success_reconciles_by_observation_without_second_push():
remote = Remote(fail_after_push=True)
executor = make_executor(remote=remote)
first = executor.execute(binding(), commit())
assert first["state"] == JobState.RECONCILIATION_REQUIRED.value
assert remote.pushes == 1
recovered = executor.reconcile(first["job_id"])
assert recovered["state"] == JobState.COMPLETED.value
assert remote.pushes == 1
def test_observation_first_reconcile_retries_only_when_remote_is_exact_expected_old_oid():
remote = Remote(fail_after_push=True)
executor = make_executor(remote=remote)
first = executor.execute(binding(), commit())
assert first["state"] == JobState.RECONCILIATION_REQUIRED.value
remote.git = remote.api = HEAD
remote.visible = True
remote.fail_after_push = False
recovered = executor.reconcile(first["job_id"])
assert recovered["state"] == JobState.COMPLETED.value
assert remote.pushes == 2
def test_duplicate_idempotency_replay_and_conflicting_payload():
executor = make_executor()
first = executor.execute(binding(), commit())
replay = executor.execute(binding(), commit())
assert replay["job_id"] == first["job_id"]
with pytest.raises(BindingConflict, match="BINDING_CONFLICT"):
executor.execute(binding(), commit(files={"active-runtime/app/new.py": b"changed\n"}))
def test_secret_like_payload_fails_before_projection():
credential = Credential()
secret_fixture = b"to" + b"ken=" + b"ghp_" + b"abcdefghijklmnopqrstuvwxyz0123456789"
with pytest.raises(ContinuationExecutionError, match="SECRET_LIKE_PAYLOAD_REJECTED"):
make_executor(credential=credential).execute(binding(), commit(files={"x": secret_fixture}))
assert credential.projected == 0
def test_remote_identity_cannot_alias_the_authority_bare_repo():
with pytest.raises(ContinuationExecutionError, match="GITEA_REMOTE_IDENTITY_INVALID"):
GiteaContinuationRemote.validate_remote_url("/data/aios-gitops/repo.git")
GiteaContinuationRemote.validate_remote_url("aiman@g.cnnetmkt.top:aios/aios.git")
def test_detached_workspace_preserves_porcelain_leading_status_column():
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
subprocess.run(["git", "init", "-q", str(root)], check=True)
subprocess.run(["git", "-C", str(root), "config", "user.name", "test"], check=True)
subprocess.run(["git", "-C", str(root), "config", "user.email", "test@localhost"], check=True)
(root / "tracked.py").write_text("one\n", encoding="utf-8")
subprocess.run(["git", "-C", str(root), "add", "tracked.py"], check=True)
subprocess.run(["git", "-C", str(root), "commit", "-q", "-m", "base"], check=True)
subprocess.run(["git", "-C", str(root), "checkout", "-q", "--detach"], check=True)
(root / "tracked.py").write_text("two\n", encoding="utf-8")
(root / "new.py").write_text("new\n", encoding="utf-8")
workspace = DetachedGitWorkspace(root)
assert workspace.is_detached()
assert workspace._dirty_paths() == {"tracked.py", "new.py"}
@@ -0,0 +1,51 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from aios_control_plane_snapshot import build_control_plane_snapshot
from aios_credential_broker import CredentialBroker
from aios_gitops_reconciliation_core import ReconciliationAction, RemoteObservation, decide_reconciliation
from aios_governed_policy import GovernedPolicyInput, evaluate_policy
OID = "a" * 40
OLD = "b" * 40
def test_authority_desired_remote_missing_never_completes():
decision = decide_reconciliation(RemoteObservation("aios/aios", "agent/example", OID, OID, None, None, False, "now", OLD))
assert decision.action == ReconciliationAction.MARK_RECONCILIATION_REQUIRED
assert decision.state != "COMPLETED"
def test_response_loss_observes_desired_remote_before_retry():
decision = decide_reconciliation(RemoteObservation("aios/aios", "agent/example", OID, OID, OID, OID, True, "now", OLD))
assert decision.action == ReconciliationAction.MARK_REMOTE_VERIFIED
def test_conflicting_remote_fails_closed():
decision = decide_reconciliation(RemoteObservation("aios/aios", "agent/example", OID, OID, "c" * 40, "c" * 40, True, "now", OLD))
assert decision.action == ReconciliationAction.FAIL_CONFLICT
def test_policy_denies_production_and_force_capabilities():
request = GovernedPolicyInput("owner", "task", "aios/aios", "commitFilesGoverned", OID, OID, "agent/example", 2, requested_capabilities=("production_write", "force_push"))
assert evaluate_policy(request).decision == "DENY"
def test_credential_metadata_does_not_include_secret_and_revoke_is_idempotent():
class Provider:
def environment(self, lease):
return {"CREDENTIAL_FILE": "/run/credential"}
broker = CredentialBroker(Provider())
metadata = broker.issue(task_id="task", repository="aios/aios", target_branch="agent/example", generation=1)
assert "token" not in str(metadata).lower()
revoked = broker.revoke(metadata["lease_id"])
assert broker.revoke(metadata["lease_id"])["state"] == "REVOKED"
assert revoked["state"] == "REVOKED"
def test_snapshot_marks_secret_material_false_and_preserves_partial_observation():
snapshot = build_control_plane_snapshot(target_version="V2R2C1", task={"gitea_git_oid": None, "gitea_api_oid": None})
assert snapshot["secret_material_exposed"] is False
assert snapshot["task"]["gitea_git_oid"] is None