"""Loopback-only governed Bridge v2 HTTP gateway with durable handlers.""" from __future__ import annotations from dataclasses import dataclass, replace 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 import threading 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, JobError, JobState, JobStore from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap from governed_production_release import build_executor_config MAX_BODY = 2 * 1024 * 1024 ROUTES = { "/v2/runtime": "getRuntimeStatus", "/v2/capabilities": "getGovernedWriteCapabilities", "/v2/governed-task": "createGovernedTask", "/v2/writer-lease": "acquireWriterLease", "/v2/writer-lease/bootstrap": "bootstrapWriterLease", "/v2/writer-lease/renew": "renewWriterLease", "/v2/commit-file": "commitFileGoverned", "/v2/commit-files": "commitFilesGoverned", "/v2/job": "getGitOpsJob", "/v2/job/resume": "resumeGitOpsJob", "/v2/job/reconcile": "reconcileGitOpsJob", "/v2/tag/approval-request": "createTagApprovalRequest", "/v2/tag/publish": "requestOrCreateTag", "/v2/tag/status": "getTagPublicationStatus", "/v2/release/artifact": "buildProductionArtifact", "/v2/release/execute": "executeApprovedProductionRelease", "/v2/release/consistency": "getReleaseConsistency", "/v2/release/acceptance": "runReleaseAcceptance", } @dataclass(frozen=True, slots=True) class RuntimeConfig: target_oid: str authoritative_git_dir: Path state_root: Path v2_job_root: Path token_path: Path ssh_key_path: Path remote_name: str = "gitea-http" repository: str = "aios/aios" api_base: str = "https://g.cnnetmkt.top/git/api/v1" v2_task_id: str = "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809" v2_branch: str = "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809" target_version: str = "V2R2C1" registry_root: Path = Path("/var/lib/aios-gitops/runtime/registry-projection") baseline_path: Path = Path("/var/lib/aios-gitops-runtime-anchor/baseline/current.json") health_url: str = "http://127.0.0.1:18927/health" 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._recovery_lock = threading.RLock() self.bindings: dict[str, dict[str, Any]] = {} self.release = build_executor_config(config) 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 { "status": "active", "health": 200, "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": self.config.repository, "repository_identity": self.config.repository, "authoritative_git_dir": str(self.config.authoritative_git_dir), "legacy_global_writes_enabled": False, } def _job_root(self, task_id: str) -> Path: return self.config.state_root / "jobs" / _safe(task_id) def _build_runtime( self, lease: WriterLeaseRequestV2, *, job_id: str, lease_not_before: datetime | None = None, lease_expires_at: datetime | None = None, ) -> tuple[Any, GitOpsJobController, ContinuationExecutor]: resolved_job_id, resolved_controller = self._resolve_governed_job_id( lease.task_id, requested_job_id=job_id, expected={ "task_id": lease.task_id, "repository": lease.repository, "base_oid": lease.base_oid, "expected_branch_oid": lease.expected_branch_oid, "target_branch": lease.target_branch, "branch_mode": lease.branch_mode, "generation": lease.generation, }, ) if lease.job_id != resolved_job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") lease.validate(AuthorityGitAdapter(self.authority)) worktree_path = self.config.state_root / "worktrees" / _safe(lease.task_id) if worktree_path.exists(): workspace = DetachedGitWorkspace(worktree_path) stale_oid = workspace.head_oid() if stale_oid != lease.expected_branch_oid or not workspace.is_detached(): durable = self._matching_durable_lease(lease) row = resolved_controller.store.read(resolved_job_id) exact_recovery = ( lease.branch_mode == "CREATE_NEW" and durable is not None and str(durable[1].get("job_id") or "") == resolved_job_id and str(durable[1].get("state") or "") == "ACTIVE" and str(durable[1].get("current_head_oid") or "") == lease.expected_branch_oid and int(durable[1].get("generation")) == lease.generation and str(row.get("state") or "") == "EXECUTING" and str(row.get("side_effect_state") or "") == "IN_PROGRESS" and self.authority.read_ref("refs/heads/" + lease.target_branch) is None and workspace.is_detached() ) if not exact_recovery: raise RuntimeError("WORKTREE_HEAD_MISMATCH") self.authority.recover_failed_detached_worktree_exact( path=worktree_path, expected_oid=lease.expected_branch_oid, stale_oid=stale_oid, target_ref="refs/heads/" + lease.target_branch, ) 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 = resolved_controller 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, ) now = datetime.now(timezone.utc) binding = ContinuationRuntimeBinding( request=lease, job_id=resolved_job_id, workspace_path=worktree_path, lease_not_before=lease_not_before or now - timedelta(seconds=1), lease_expires_at=lease_expires_at or now + timedelta(seconds=lease.lease_seconds), original_receipt_valid=True, original_grant_consumed=True, permissions=ContinuationPermissions(), ) return binding, controller, executor def _normalize_runtime_lease(self, lease: WriterLeaseRequestV2) -> WriterLeaseRequestV2: # CREATE_NEW is a provisioning intent. Once the exact task-owned target # ref exists at the exact expected head, continuation must use # CONTINUE_EXISTING semantics. All binding drift remains fail-closed. if lease.branch_mode != "CREATE_NEW": return lease current_head = self.authority.read_ref("refs/heads/" + lease.target_branch) if current_head is None: return lease binding = self.bindings.get(lease.task_id) if binding is None: raise RuntimeError("LEASE_BINDING_CONFLICT") exact = { "task_id": lease.task_id, "job_id": lease.job_id, "repository": lease.repository, "base_oid": lease.base_oid, "expected_branch_oid": lease.expected_branch_oid, "target_branch": lease.target_branch, "generation": lease.generation, } for field, expected in exact.items(): raw = binding.get(field) if isinstance(binding, Mapping) else getattr(binding, field, None) if field == "generation": try: raw = int(raw) except (TypeError, ValueError): raise RuntimeError("LEASE_BINDING_CONFLICT") from None if raw != expected: raise RuntimeError("LEASE_BINDING_CONFLICT") if current_head != lease.expected_branch_oid: raise RuntimeError("EXPECTED_BRANCH_OID_CONFLICT") return replace(lease, branch_mode="CONTINUE_EXISTING") def _normalize_recovery_request(self, request: Mapping[str, Any]) -> dict[str, Any]: """Normalize only an exact already-provisioned CREATE_NEW recovery request.""" normalized = dict(request) if str(normalized.get("branch_mode") or "CONTINUE_EXISTING") != "CREATE_NEW": return normalized task_id = str(normalized.get("task_id") or "") requested_job_id = str(normalized.get("job_id") or normalized.get("governed_job_id") or "") repository = str(normalized.get("repository") or "") base_oid = str(normalized.get("base_oid") or "") expected_branch_oid = str(normalized.get("expected_branch_oid") or "") target_branch = str(normalized.get("target_branch") or "") try: generation = int(normalized.get("generation")) except (TypeError, ValueError): return normalized current_head = self.authority.read_ref("refs/heads/" + target_branch) if current_head is None: return normalized binding = self.bindings.get(task_id) if binding is None: raise RuntimeError("LEASE_BINDING_CONFLICT") def _binding_value(name: str) -> Any: return binding.get(name) if isinstance(binding, Mapping) else getattr(binding, name, None) exact = { "task_id": task_id, "job_id": requested_job_id or str(_binding_value("job_id") or ""), "repository": repository, "base_oid": base_oid, "expected_branch_oid": expected_branch_oid, "target_branch": target_branch, "generation": generation, } for field, expected in exact.items(): actual = _binding_value(field) if field == "generation": try: actual = int(actual) except (TypeError, ValueError): raise RuntimeError("LEASE_BINDING_CONFLICT") from None if actual != expected: raise RuntimeError("LEASE_BINDING_CONFLICT") if current_head != expected_branch_oid: raise RuntimeError("EXPECTED_BRANCH_OID_CONFLICT") normalized["branch_mode"] = "CONTINUE_EXISTING" return normalized def _resolve_governed_job_id( self, task_id: str, *, requested_job_id: str | None = None, durable_job_id: str | None = None, expected: Mapping[str, Any] | None = None, ) -> tuple[str, GitOpsJobController]: metadata = self.bindings.get(task_id) if not isinstance(metadata, Mapping): raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") job_id = str(metadata.get("job_id") or "").strip() if not job_id: raise RuntimeError("RUNTIME_LINEAGE_MISSING_JOB_ID") if not job_id.startswith("job-"): raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") supplied = str(requested_job_id or "").strip() if supplied and supplied != job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") durable = str(durable_job_id or "").strip() if durable and durable != job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") job_root = self._job_root(task_id) controller = GitOpsJobController(JobStore(job_root, payload_root=job_root / "payloads")) try: row = controller.store.read(job_id) raw = json.loads(str(row["request_json"])) except Exception as exc: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") from exc job_binding = raw.get("binding", raw) if not isinstance(job_binding, Mapping): raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") expected = dict(expected or {}) metadata_checks = { "task_id": str(metadata.get("task_id") or ""), "repository": str(metadata.get("repository") or ""), "base_oid": str(metadata.get("base_oid") or ""), "target_branch": str(metadata.get("target_branch") or ""), } for field, actual in metadata_checks.items(): wanted = str(expected.get(field) or (task_id if field == "task_id" else "")) if wanted and actual != wanted: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") try: metadata_generation = int(metadata.get("generation")) wanted_generation = int(expected.get("generation")) except (TypeError, ValueError) as exc: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") from exc if metadata_generation != wanted_generation: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") metadata_mode = str(metadata.get("branch_mode") or "") wanted_mode = str(expected.get("branch_mode") or "") mode_matches = metadata_mode == wanted_mode or ( metadata_mode == "CREATE_NEW" and wanted_mode == "CONTINUE_EXISTING" ) if not mode_matches: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") wanted_head = str(expected.get("expected_branch_oid") or "") allowed_heads = {str(metadata.get("expected_branch_oid") or "")} try: with controller.store._connect() as connection: lease_heads = connection.execute( "SELECT current_head_oid FROM branch_leases " "WHERE job_id=? AND branch=? AND generation=? AND state='ACTIVE'", (job_id, metadata_checks["target_branch"], metadata_generation), ).fetchall() except sqlite3.Error as exc: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") from exc allowed_heads.update(str(item[0] or "") for item in lease_heads) if wanted_head and wanted_head not in allowed_heads: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") checks = { "task_id": str(job_binding.get("task_id") or job_binding.get("campaign_id") or ""), "repository": str(job_binding.get("repository") or row["repository"] or ""), "base_oid": str(job_binding.get("base_oid") or job_binding.get("lineage_anchor_oid") or ""), "expected_branch_oid": str( job_binding.get("expected_branch_oid") or row["expected_branch_oid"] or "" ), "target_branch": str(job_binding.get("target_branch") or job_binding.get("branch") or row["branch"] or ""), } for field, actual in checks.items(): if field == "expected_branch_oid": # The Job's immutable expected head is the original task # anchor. After commit/restart the request head legitimately # advances through branch_leases.current_head_oid. wanted = str(metadata.get("expected_branch_oid") or "") else: wanted = str(expected.get(field) or (task_id if field == "task_id" else "")) if wanted and actual != wanted: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") # Legacy governed-task Job rows do not persist generation/branch_mode. # Their durable task-binding projection is therefore the authority for # those two fields (validated above). If a newer Job schema carries # them too, require exact agreement rather than silently preferring it. if "generation" in job_binding: try: if int(job_binding["generation"]) != wanted_generation: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") except (TypeError, ValueError) as exc: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") from exc if "branch_mode" in job_binding: job_mode = str(job_binding.get("branch_mode") or "") job_mode_matches = job_mode == wanted_mode or ( job_mode == "CREATE_NEW" and wanted_mode == "CONTINUE_EXISTING" ) if not job_mode_matches: raise RuntimeError("RUNTIME_GOVERNED_JOB_UNRESOLVED") return job_id, controller def _runtime(self, request: Mapping[str, Any]) -> tuple[Any, GitOpsJobController, ContinuationExecutor]: task_id = str(request["task_id"]) generation = int(request["generation"]) expected = { "task_id": task_id, "repository": str(request["repository"]), "base_oid": str(request["base_oid"]), "expected_branch_oid": str(request["expected_branch_oid"]), "target_branch": str(request["target_branch"]), "branch_mode": str(request.get("branch_mode") or "CONTINUE_EXISTING"), "generation": generation, } job_id, _ = self._resolve_governed_job_id( task_id, requested_job_id=str(request.get("job_id") or request.get("governed_job_id") or ""), expected=expected, ) lease = WriterLeaseRequestV2( task_id=task_id, job_id=job_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=generation, lease_seconds=int(request.get("lease_seconds", 900)), ) lease = self._normalize_runtime_lease(lease) key = (task_id, generation) if key in self.runtimes: existing = self.runtimes[key] if not self._same_binding(existing[0], lease): raise RuntimeError("LEASE_BINDING_CONFLICT") return existing result = self._build_runtime(lease, job_id=job_id) self.runtimes[key] = result self.controllers[lease.task_id] = result[1] 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, controller, _ = self._runtime(request) req = binding.request durable = self._matching_durable_lease(req) if durable is not None: if str(durable[1].get("job_id") or "") != binding.job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") try: durable_expires_at = datetime.fromisoformat( str(durable[1]["lease_expires_at"]).replace("Z", "+00:00") ).astimezone(timezone.utc) except (KeyError, TypeError, ValueError) as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc if durable_expires_at <= datetime.now(timezone.utc): return self.renew_writer_lease(request) return self._lease_response( binding, recovered=False, idempotent=True, durable_branch_lease_present=True, ) row = controller.store.read(binding.job_id) state = JobState(str(row["state"])) if state == JobState.REQUESTED: try: controller.advance( binding.job_id, JobState.AUTHORIZED, side_effect_state="AUTHORIZED", receipt={ "task_id": req.task_id, "job_id": binding.job_id, "original_grant_consumed": True, }, expected_revision=int(row["revision"]), ) except JobError: # A concurrent exact acquire may have won the CAS. Re-read and # accept only the same governed lineage below. pass row = controller.store.read(binding.job_id) state = JobState(str(row["state"])) if state == JobState.AUTHORIZED: try: controller.lease_job( binding.job_id, lease_id=f"lease-{binding.job_id}", lineage_anchor_oid=req.base_oid, current_head_oid=req.expected_branch_oid, generation=req.generation, lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(), expected_revision=int(row["revision"]), ) except JobError: # Resolve a possible concurrent winner through the exact # durable matcher; never manufacture or replace a lease. durable = self._matching_durable_lease(req) if durable is None: raise elif state != JobState.LEASED: raise RuntimeError("WRITER_LEASE_JOB_STATE_INVALID") durable = self._matching_durable_lease(req) if durable is None or str(durable[1].get("job_id") or "") != binding.job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") return self._lease_response( binding, recovered=False, idempotent=False, durable_branch_lease_present=True, ) @staticmethod def _durable_value( sources: list[Mapping[str, Any]], aliases: tuple[str, ...], *, field: str, ) -> Any: values: list[Any] = [] for source in sources: for alias in aliases: if source.get(alias) is not None: values.append(source[alias]) break if not values: return None normalized = {str(value) for value in values} if len(normalized) != 1: raise RuntimeError("DURABLE_BINDING_INVALID") return values[0] def _durable_lease_request( self, task_id: str, *, generation: int, expected_branch_oid: str, target_branch: str, ) -> dict[str, Any]: metadata = self.bindings.get(task_id) if not isinstance(metadata, dict) or str(metadata.get("task_id") or "") != task_id: raise RuntimeError("TASK_OWNERSHIP_INVALID") job_id = str(metadata.get("job_id") or "") if not job_id: raise RuntimeError("RUNTIME_LINEAGE_MISSING_JOB_ID") database = self._job_root(task_id) / "state.sqlite3" active_rows: list[sqlite3.Row] = [] if database.is_file() and not database.is_symlink(): try: with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection: connection.row_factory = sqlite3.Row active_rows = connection.execute( "SELECT job_id,branch,lineage_anchor_oid,current_head_oid,generation,state,revision,lease_expires_at " "FROM branch_leases WHERE job_id=? AND branch=? AND generation=? AND state='ACTIVE'", (job_id, str(metadata.get("target_branch") or ""), generation), ).fetchall() except sqlite3.Error as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc if len(active_rows) > 1: raise RuntimeError("LEASE_BINDING_CONFLICT") if active_rows: row = active_rows[0] if str(row["current_head_oid"] or "") != expected_branch_oid: raise RuntimeError("LEASE_BINDING_CONFLICT") if str(row["branch"] or "") != target_branch: raise RuntimeError("LEASE_BINDING_CONFLICT") authority_head = self.authority.read_ref("refs/heads/" + target_branch) if authority_head is None: active_mode = str(metadata.get("branch_mode") or "") if active_mode != "CREATE_NEW" or str(row["current_head_oid"] or "") != str(row["lineage_anchor_oid"] or ""): raise RuntimeError("EXPECTED_BRANCH_OID_CONFLICT") else: if str(authority_head) != str(row["current_head_oid"] or ""): raise RuntimeError("EXPECTED_BRANCH_OID_CONFLICT") active_mode = "CONTINUE_EXISTING" resolved_job_id, _ = self._resolve_governed_job_id( task_id, requested_job_id=job_id, durable_job_id=str(row["job_id"] or ""), expected={ "task_id": task_id, "repository": str(metadata.get("repository") or ""), "base_oid": str(row["lineage_anchor_oid"] or ""), "expected_branch_oid": str(row["current_head_oid"] or ""), "target_branch": str(row["branch"] or ""), "branch_mode": active_mode, "generation": int(row["generation"]), }, ) return { "task_id": task_id, "job_id": resolved_job_id, "repository": str(metadata.get("repository") or ""), "branch_mode": active_mode, "base_oid": str(row["lineage_anchor_oid"] or ""), "expected_branch_oid": str(row["current_head_oid"] or ""), "target_branch": str(row["branch"] or ""), "generation": int(row["generation"]), "revision": int(row["revision"]), "lease_expires_at": str(row["lease_expires_at"] or ""), } sources: list[Mapping[str, Any]] = [metadata] matching_rows: list[Mapping[str, Any]] = [] database = self._job_root(task_id) / "state.sqlite3" if database.is_file() and not database.is_symlink(): try: with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection: rows = connection.execute( "SELECT request_json FROM jobs ORDER BY updated_at DESC" ).fetchall() for (request_json,) in rows: try: raw = json.loads(request_json) except (TypeError, ValueError): raise RuntimeError("DURABLE_BINDING_INVALID") from None binding = raw.get("binding", raw) if ( isinstance(binding, dict) and str(binding.get("task_id") or binding.get("campaign_id") or "") == task_id ): raw_generation = binding.get("generation") if raw_generation is None: # The original GOVERNED_TASK_CREATE anchor is # durable task ownership, not a generation lease. continue try: binding_generation = int(raw_generation) except (TypeError, ValueError): raise RuntimeError("DURABLE_BINDING_INVALID") from None binding_target = str(binding.get("target_branch") or binding.get("branch") or "") binding_expected = str(binding.get("expected_branch_oid") or "") if ( binding_generation == generation and binding_expected == expected_branch_oid and binding_target == target_branch ): matching_rows.append(binding) except (OSError, sqlite3.Error) as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc if matching_rows: # Metadata is the task anchor, while the current-generation job row # is authoritative for generation-local branch mode. Historical # generation fields must never participate in this aggregate. metadata_anchor = dict(metadata) metadata_anchor.pop("generation", None) metadata_anchor.pop("branch_mode", None) sources = [metadata_anchor, *matching_rows] repository = self._durable_value(sources, ("repository",), field="repository") base_oid = self._durable_value(sources, ("base_oid", "lineage_anchor_oid"), field="base_oid") expected_oid = self._durable_value( sources, ("expected_branch_oid",), field="expected_branch_oid", ) target_branch = self._durable_value( sources, ("target_branch", "branch"), field="target_branch", ) branch_mode = self._durable_value(sources, ("branch_mode",), field="branch_mode") generation = self._durable_value(sources, ("generation",), field="generation") if branch_mode is None and base_oid and expected_oid: branch_mode = ( "CREATE_NEW" if base_oid == expected_oid and self.authority.read_ref("refs/heads/" + str(target_branch)) is None else "CONTINUE_EXISTING" ) if generation is None and branch_mode == "CREATE_NEW": generation = 1 try: generation = int(generation) except (TypeError, ValueError) as exc: raise RuntimeError("DURABLE_BINDING_GENERATION_INVALID") from exc if ( repository != self.config.repository or not self._valid_oid(str(base_oid or "")) or not self._valid_oid(str(expected_oid or "")) or not isinstance(target_branch, str) or not target_branch.startswith("agent/") or branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"} or generation < 1 ): raise RuntimeError("DURABLE_BINDING_INVALID") job_id, _ = self._resolve_governed_job_id( task_id, expected={ "task_id": task_id, "repository": str(repository), "base_oid": str(base_oid), "expected_branch_oid": str(expected_oid), "target_branch": target_branch, "branch_mode": str(branch_mode), "generation": int(generation), }, ) return { "task_id": task_id, "job_id": job_id, "repository": str(repository), "branch_mode": str(branch_mode), "base_oid": str(base_oid), "expected_branch_oid": str(expected_oid), "target_branch": target_branch, "generation": generation, } def _validated_recovery_request(self, request: Mapping[str, Any]) -> WriterLeaseRequestV2: if request.get("schema") not in WriterLeaseRequestV2.accepted_schemas: raise RuntimeError("LEASE_SCHEMA_INVALID") try: generation = int(request["generation"]) except (KeyError, TypeError, ValueError) as exc: raise RuntimeError("GENERATION_BINDING_CONFLICT") from exc if isinstance(request.get("generation"), bool) or generation < 1: raise RuntimeError("GENERATION_BINDING_CONFLICT") durable = self._durable_lease_request( str(request.get("task_id") or ""), generation=generation, expected_branch_oid=str(request.get("expected_branch_oid") or ""), target_branch=str(request.get("target_branch") or ""), ) requested_job_id = str(request.get("job_id") or request.get("governed_job_id") or "") if requested_job_id and requested_job_id != durable["job_id"]: raise RuntimeError("LEASE_BINDING_CONFLICT") for field in ("repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch"): if str(request.get(field) or "") != durable[field]: raise RuntimeError("LEASE_BINDING_CONFLICT") if isinstance(request.get("generation"), bool) or generation != durable["generation"]: raise RuntimeError("GENERATION_BINDING_CONFLICT") try: lease_seconds = int(request["lease_seconds"]) except (KeyError, TypeError, ValueError) as exc: raise RuntimeError("LEASE_DURATION_INVALID") from exc lease = WriterLeaseRequestV2( task_id=str(request["task_id"]), job_id=str(durable["job_id"]), repository=str(request["repository"]), branch_mode=str(request["branch_mode"]), base_oid=str(request["base_oid"]), expected_branch_oid=str(request["expected_branch_oid"]), target_branch=str(request["target_branch"]), generation=generation, lease_seconds=lease_seconds, ) lease.validate(AuthorityGitAdapter(self.authority)) return lease @staticmethod def _lease_response( binding: ContinuationRuntimeBinding, *, recovered: bool, idempotent: bool, bootstrapped: bool = False, runtime_rehydrated: bool = False, durable_branch_lease_present: bool = False, ) -> dict[str, Any]: request = binding.request return { "schema": WriterLeaseRequestV2.schema, "task_id": request.task_id, "job_id": binding.job_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, "lease_expires_at": binding.lease_expires_at.astimezone(timezone.utc).isoformat(), "create_target_branch": False, "worktree_mode": "detached_at_base_oid" if request.branch_mode == "CREATE_NEW" else "detached_at_expected_head", "cas_ref_update_required": True, "recovered": recovered, "bootstrapped": bootstrapped, "idempotent": idempotent, "runtime_rehydrated": runtime_rehydrated, "durable_branch_lease_present": durable_branch_lease_present, } def _matching_durable_lease( self, lease: WriterLeaseRequestV2 ) -> tuple[GitOpsJobController, dict[str, Any]] | None: database = self._job_root(lease.task_id) / "state.sqlite3" if not database.is_file() or database.is_symlink(): return None try: with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection: connection.row_factory = sqlite3.Row rows = connection.execute( """ SELECT bl.lease_id, bl.job_id, bl.branch, bl.lineage_anchor_oid, bl.current_head_oid, bl.generation, bl.lease_expires_at, bl.state, bl.revision AS lease_revision, j.repository AS job_repository, j.branch AS job_branch, j.expected_branch_oid AS job_expected_branch_oid, j.request_json FROM branch_leases AS bl JOIN jobs AS j ON j.job_id = bl.job_id WHERE bl.branch=? AND bl.lineage_anchor_oid=? AND bl.current_head_oid=? AND bl.generation=? """, ( lease.target_branch, lease.base_oid, lease.expected_branch_oid, lease.generation, ), ).fetchall() except (OSError, sqlite3.Error) as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc matches: list[dict[str, Any]] = [] for row in rows: try: raw = json.loads(row["request_json"]) except (TypeError, ValueError) as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc binding = raw.get("binding", raw) if not isinstance(binding, dict): continue try: generation = int(binding.get("generation", row["generation"])) except (TypeError, ValueError): continue task_metadata = self.bindings.get(lease.task_id) durable_mode = str( binding.get("branch_mode") or (task_metadata.get("branch_mode") if isinstance(task_metadata, Mapping) else "") or "" ) mode_matches = durable_mode == lease.branch_mode or ( durable_mode == "CREATE_NEW" and lease.branch_mode == "CONTINUE_EXISTING" ) if ( str(row["job_id"] or "") == lease.job_id and str(binding.get("task_id") or binding.get("campaign_id") or "") == lease.task_id and str(binding.get("repository") or row["job_repository"] or "") == lease.repository and mode_matches and str(binding.get("base_oid") or binding.get("lineage_anchor_oid") or row["lineage_anchor_oid"] or "") == lease.base_oid and str(row["current_head_oid"] or "") == lease.expected_branch_oid and str(binding.get("target_branch") or binding.get("branch") or row["job_branch"] or "") == lease.target_branch and generation == lease.generation ): matches.append(dict(row)) if not matches: return None if len(matches) != 1: raise RuntimeError("LEASE_BINDING_CONFLICT") controller = self.controllers.get(lease.task_id) if controller is None: controller = GitOpsJobController( JobStore(self._job_root(lease.task_id), payload_root=self._job_root(lease.task_id) / "payloads") ) self.controllers[lease.task_id] = controller return controller, matches[0] @staticmethod def _same_binding(binding: ContinuationRuntimeBinding, lease: WriterLeaseRequestV2) -> bool: request = binding.request return ( binding.job_id == lease.job_id and request.job_id == lease.job_id and request.task_id == lease.task_id and request.repository == lease.repository and request.branch_mode == lease.branch_mode and request.base_oid == lease.base_oid and request.expected_branch_oid == lease.expected_branch_oid and request.target_branch == lease.target_branch and request.generation == lease.generation ) def _recovery_lock_for(self) -> threading.RLock: lock = getattr(self, "_recovery_lock", None) if lock is None: lock = threading.RLock() self._recovery_lock = lock return lock def bootstrap_writer_lease(self, request: Mapping[str, Any]) -> dict[str, Any]: lease = self._validated_recovery_request(self._normalize_recovery_request(request)) key = (lease.task_id, lease.generation) now = datetime.now(timezone.utc) with self._recovery_lock_for(): durable = self._matching_durable_lease(lease) if durable is not None: controller, row = durable if str(row.get("state") or "") != "ACTIVE": raise RuntimeError("LEASE_NOT_RECOVERABLE") try: durable_expires_at = datetime.fromisoformat( str(row["lease_expires_at"]).replace("Z", "+00:00") ).astimezone(timezone.utc) except (TypeError, ValueError) as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc if durable_expires_at <= now: return self.renew_writer_lease(request) existing = self.runtimes.get(key) if existing is not None: if not self._same_binding(existing[0], lease): raise RuntimeError("LEASE_BINDING_CONFLICT") return self._lease_response( existing[0], recovered=False, bootstrapped=True, idempotent=True, runtime_rehydrated=False, durable_branch_lease_present=True, ) runtime = self._build_runtime( lease, job_id=lease.job_id, lease_not_before=now - timedelta(seconds=1), lease_expires_at=durable_expires_at, ) self.runtimes[key] = runtime self.controllers[lease.task_id] = controller return self._lease_response( runtime[0], recovered=False, bootstrapped=True, idempotent=True, runtime_rehydrated=True, durable_branch_lease_present=True, ) existing = self.runtimes.get(key) if existing is not None: binding, controller, executor = existing if not self._same_binding(binding, lease): raise RuntimeError("LEASE_BINDING_CONFLICT") if binding.lease_expires_at > now: return self._lease_response(binding, recovered=False, idempotent=True) binding = replace( binding, request=replace(binding.request, lease_seconds=lease.lease_seconds), lease_not_before=now - timedelta(seconds=1), lease_expires_at=now + timedelta(seconds=lease.lease_seconds), ) self.runtimes[key] = (binding, controller, executor) return self._lease_response( binding, recovered=False, bootstrapped=True, idempotent=False ) runtime = self._build_runtime( lease, job_id=lease.job_id, lease_not_before=now - timedelta(seconds=1), lease_expires_at=now + timedelta(seconds=lease.lease_seconds), ) self.runtimes[key] = runtime self.controllers[lease.task_id] = runtime[1] return self._lease_response( runtime[0], recovered=False, bootstrapped=True, idempotent=False ) def renew_writer_lease(self, request: Mapping[str, Any]) -> dict[str, Any]: lease = self._validated_recovery_request(self._normalize_recovery_request(request)) key = (lease.task_id, lease.generation) now = datetime.now(timezone.utc) with self._recovery_lock_for(): durable = self._matching_durable_lease(lease) if durable is None: raise RuntimeError("LEASE_NOT_FOUND") controller, row = durable if str(row.get("state") or "") != "ACTIVE": raise RuntimeError("LEASE_NOT_RECOVERABLE") try: durable_expires_at = datetime.fromisoformat( str(row["lease_expires_at"]).replace("Z", "+00:00") ).astimezone(timezone.utc) except (TypeError, ValueError) as exc: raise RuntimeError("DURABLE_BINDING_INVALID") from exc runtime = self.runtimes.get(key) if runtime is not None and not self._same_binding(runtime[0], lease): raise RuntimeError("LEASE_BINDING_CONFLICT") def bind_expiry(expires_at: datetime) -> tuple[Any, GitOpsJobController, ContinuationExecutor]: if runtime is None: return self._build_runtime( lease, job_id=lease.job_id, lease_not_before=now - timedelta(seconds=1), lease_expires_at=expires_at, ) binding, _, executor = runtime return ( replace( binding, request=replace(binding.request, lease_seconds=lease.lease_seconds), lease_not_before=now - timedelta(seconds=1), lease_expires_at=expires_at, ), controller, executor, ) if durable_expires_at > now: rehydrated = runtime is None renewed_runtime = bind_expiry(durable_expires_at) self.runtimes[key] = renewed_runtime self.controllers[lease.task_id] = controller return self._lease_response( renewed_runtime[0], recovered=False, idempotent=True, runtime_rehydrated=rehydrated, durable_branch_lease_present=True, ) renewed_expires_at = now + timedelta(seconds=lease.lease_seconds) try: controller.update_lease( str(row["lease_id"]), expected_revision=int(row["lease_revision"]), current_head_oid=lease.expected_branch_oid, generation=lease.generation, state="ACTIVE", lease_expires_at=renewed_expires_at.astimezone(timezone.utc).isoformat(), ) except JobError as exc: if str(exc) != "LEASE_REVISION_CAS_CONFLICT": raise raced = self._matching_durable_lease(lease) if raced is None: raise RuntimeError("LEASE_NOT_FOUND") from exc controller, raced_row = raced try: raced_expires_at = datetime.fromisoformat( str(raced_row["lease_expires_at"]).replace("Z", "+00:00") ).astimezone(timezone.utc) except (TypeError, ValueError) as parse_exc: raise RuntimeError("DURABLE_BINDING_INVALID") from parse_exc if raced_expires_at <= now: raise RuntimeError("LEASE_REVISION_CAS_CONFLICT") from exc rehydrated = runtime is None renewed_runtime = bind_expiry(raced_expires_at) self.runtimes[key] = renewed_runtime self.controllers[lease.task_id] = controller return self._lease_response( renewed_runtime[0], recovered=False, idempotent=True, runtime_rehydrated=rehydrated, durable_branch_lease_present=True, ) rehydrated = runtime is None renewed_runtime = bind_expiry(renewed_expires_at) self.runtimes[key] = renewed_runtime self.controllers[lease.task_id] = controller return self._lease_response( renewed_runtime[0], recovered=True, idempotent=False, runtime_rehydrated=rehydrated, durable_branch_lease_present=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, controller, executor = runtime requested_job_id = str(request.get("job_id") or request.get("governed_job_id") or "") if requested_job_id and requested_job_id != binding.job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") if binding.request.job_id != binding.job_id: raise RuntimeError("RUNTIME_LINEAGE_MISSING_JOB_ID") try: with controller.store._connect() as connection: connection.row_factory = sqlite3.Row exact_rows = connection.execute( "SELECT lease_id,job_id,lineage_anchor_oid,current_head_oid,generation,state,revision " "FROM branch_leases WHERE job_id=? AND branch=? AND generation=? AND state='ACTIVE'", (binding.job_id, binding.request.target_branch, binding.request.generation), ).fetchall() except sqlite3.Error as exc: raise RuntimeError("JOB_DB_UNAVAILABLE") from exc if len(exact_rows) != 1: raise RuntimeError("LEASE_BINDING_CONFLICT") precommit_lease = exact_rows[0] if str(precommit_lease["lineage_anchor_oid"] or "") != binding.request.base_oid: raise RuntimeError("LEASE_BINDING_CONFLICT") durable_head = str(precommit_lease["current_head_oid"] or "") if durable_head != binding.request.expected_branch_oid: # The only accepted advanced-head retry is the exact OID already # recorded by this same Job after remote verification. This is # the crash boundary where lease CAS succeeded but COMPLETED did # not persist; every other head remains fail-closed. current_job = controller.store.read(binding.job_id) current_state = str(current_job.get("state") or "") try: current_receipt = json.loads(str(current_job.get("receipt_json") or "{}")) except json.JSONDecodeError as exc: raise RuntimeError("GOVERNED_COMMIT_RESPONSE_INVALID") from exc retry_oid = str(current_receipt.get("authority_oid") or "") if ( current_state not in {"REMOTE_VERIFIED", "COMPLETED"} or re.fullmatch(r"[0-9a-f]{40}", retry_oid) is None or durable_head != retry_oid ): raise RuntimeError("LEASE_HEAD_RECONCILIATION_REQUIRED") normalized_request = self._normalize_runtime_lease(binding.request) if normalized_request != binding.request: binding = replace(binding, request=normalized_request) self.runtimes[(normalized_request.task_id, normalized_request.generation)] = (binding, controller, executor) if request["repository"] != binding.request.repository or 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) def advance_durable_head_before_completion(commit_oid: str) -> None: if re.fullmatch(r"[0-9a-f]{40}", commit_oid) is None: raise RuntimeError("GOVERNED_COMMIT_OID_INVALID") durable_head = str(precommit_lease["current_head_oid"] or "") if durable_head == commit_oid: return if durable_head != binding.request.expected_branch_oid: raise RuntimeError("LEASE_HEAD_RECONCILIATION_REQUIRED") try: controller.update_lease( str(precommit_lease["lease_id"]), expected_revision=int(precommit_lease["revision"]), current_head_oid=commit_oid, generation=binding.request.generation, ) except Exception as exc: if "LEASE_REVISION_CAS_CONFLICT" in str(exc): raise RuntimeError("LEASE_CAS_CONFLICT") from exc raise result = executor.execute( binding, commit, before_complete=advance_durable_head_before_completion, ) if not isinstance(result, Mapping): raise RuntimeError("GOVERNED_COMMIT_RESPONSE_INVALID") if str(result.get("job_id") or "") != binding.job_id: raise RuntimeError("LEASE_BINDING_CONFLICT") if str(result.get("state") or "") != "COMPLETED": # Preserve the canonical reconciliation state. A remote-pending # result is not a malformed successful commit response and must # never advance the durable current head. raise RuntimeError("RECONCILIATION_REQUIRED") receipt = result.get("receipt") if not isinstance(receipt, Mapping) or receipt.get("commit_visible") is not True: raise RuntimeError("REMOTE_COMMIT_NOT_VERIFIED") commit_oid = str(receipt.get("authority_oid") or "") if re.fullmatch(r"[0-9a-f]{40}", commit_oid) is None: raise RuntimeError("GOVERNED_COMMIT_OID_INVALID") try: with controller.store._connect() as connection: connection.row_factory = sqlite3.Row columns = { str(row["name"]) for row in connection.execute("PRAGMA table_info(branch_leases)").fetchall() } required = {"lease_id", "job_id", "branch", "generation", "state", "revision", "current_head_oid"} if not required.issubset(columns): raise RuntimeError("LEASE_STORE_SCHEMA_INVALID") where = ["job_id=?", "branch=?", "generation=?", "state='ACTIVE'"] args = [binding.job_id, binding.request.target_branch, binding.request.generation] if "task_id" in columns: where.append("task_id=?") args.append(binding.request.task_id) if "repository" in columns: where.append("repository=?") args.append(binding.request.repository) rows = connection.execute( "SELECT lease_id,job_id,current_head_oid,revision FROM branch_leases WHERE " + " AND ".join(where) + " ORDER BY revision DESC", tuple(args), ).fetchall() except sqlite3.Error as exc: raise RuntimeError("JOB_DB_UNAVAILABLE") from exc if len(rows) != 1: raise RuntimeError("LEASE_BINDING_CONFLICT") row = rows[0] durable_head = str(row["current_head_oid"] or "") if durable_head != commit_oid: raise RuntimeError("LEASE_HEAD_RECONCILIATION_REQUIRED") continued_request = replace( binding.request, branch_mode="CONTINUE_EXISTING", expected_branch_oid=commit_oid, ) continued_binding = replace(binding, request=continued_request) self.runtimes[(continued_request.task_id, continued_request.generation)] = ( continued_binding, controller, executor, ) return { **dict(result), "job_id": binding.job_id, "commit_oid": commit_oid, "commit_visible": True, "branch_mode": continued_request.branch_mode, "target_branch": continued_request.target_branch, "generation": continued_request.generation, } 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, "bootstrapWriterLease": self.bootstrap_writer_lease, "renewWriterLease": self.renew_writer_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(), "createTagApprovalRequest": self.release.approval_request, "requestOrCreateTag": self.release.publish_tag, "getTagPublicationStatus": self.release.tag_status, "buildProductionArtifact": self.release.build_artifact, "executeApprovedProductionRelease": self.release.execute_request, "getReleaseConsistency": self.release.consistency, "runReleaseAcceptance": self.release.acceptance, } def 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: # job_id remains optional on the request solely for the deployed trusted # worker compatibility path; the server resolves it from the durable # task binding and always validates/returns the exact governed parent. allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | {"schema", "job_id", "lease_seconds", "create_target_branch", "worktree_mode", "cas_ref_update_required", "expected_branch_oid", "generation", "branch_mode"} if operation == "commitFilesGoverned": allowed.add("files") allowed.update({"owner_approval", "execute"}) 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 operation in ("getRuntimeStatus", "getGovernedWriteCapabilities") 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, remote_name=os.environ.get("AIOS_GITEA_REMOTE", "gitea-http"), api_base=os.environ.get("AIOS_GITEA_API_BASE", "https://g.cnnetmkt.top/git/api/v1"), target_version=os.environ.get("AIOS_CONTROL_PLANE_TARGET_VERSION", "V2R2C1"), ) runtime = GovernedRuntime(config) serve( BridgeV2Adapter(runtime_status=runtime.runtime_status(), handlers=runtime.handlers()), port=int(os.environ.get("AIOS_BRIDGE_V2_PORT", "18892")), ) if __name__ == "__main__": main()