feat(gitops): wire durable execution side effects

active-runtime/app/aios_authority_git_executor.py

@@ -0,0 +1,149 @@
+"""Fail-closed Authority Git side-effect boundary.
+
+Only this module may issue the narrowly scoped ref CAS primitives used by the
+durable executor. It never accepts force updates, arbitrary refs, tags, or
+direct production mutations.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+from typing import Mapping, Sequence
+
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+TASK_BRANCH_RE = re.compile(r"^refs/heads/agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+
+
+class AuthorityGitError(RuntimeError):
+    pass
+
+
+def _oid(value: str, field: str) -> str:
+    if not isinstance(value, str) or not OID_RE.fullmatch(value):
+        raise AuthorityGitError(f"{field}_INVALID")
+    return value
+
+
+@dataclass(frozen=True, slots=True)
+class AuthorityGitExecutor:
+    git_dir: Path
+    worktree_root: Path = Path("/var/lib/aios-gitops/worktrees")
+
+    def __post_init__(self) -> None:
+        if not self.git_dir.is_absolute() or self.git_dir.is_symlink():
+            raise AuthorityGitError("AUTHORITY_GIT_DIR_INVALID")
+        if not self.worktree_root.is_absolute() or self.worktree_root.is_symlink():
+            raise AuthorityGitError("WORKTREE_ROOT_INVALID")
+
+    def _git(self, args: Sequence[str], *, check: bool = True) -> str:
+        result = subprocess.run(
+            ["git", f"--git-dir={self.git_dir}", *args],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        if check and result.returncode != 0:
+            raise AuthorityGitError("AUTHORITY_GIT_COMMAND_FAILED")
+        return result.stdout.strip()
+
+    def read_ref(self, ref: str) -> str | None:
+        if not ref.startswith("refs/") or ref.startswith(("refs/tags/", "refs/remotes/")):
+            raise AuthorityGitError("REF_READ_SCOPE_DENIED")
+        value = self._git(["rev-parse", "--verify", ref], check=False)
+        return value if value and OID_RE.fullmatch(value) else None
+
+    def ancestry_check(self, older_oid: str, newer_oid: str) -> bool:
+        _oid(older_oid, "older_oid")
+        _oid(newer_oid, "newer_oid")
+        result = subprocess.run(
+            ["git", f"--git-dir={self.git_dir}", "merge-base", "--is-ancestor", older_oid, newer_oid],
+            capture_output=True,
+            check=False,
+        )
+        if result.returncode not in (0, 1):
+            raise AuthorityGitError("ANCESTRY_CHECK_FAILED")
+        return result.returncode == 0
+
+    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")
+        _oid(expected_oid, "expected_oid")
+        _oid(new_oid, "new_oid")
+        current = self.read_ref(branch)
+        if current != expected_oid:
+            if current == new_oid:
+                return new_oid
+            raise AuthorityGitError("AUTHORITY_BRANCH_EXPECTATION_MISMATCH")
+        if not self.ancestry_check(expected_oid, new_oid):
+            raise AuthorityGitError("NEW_OID_NOT_DESCENDANT")
+        self._git(["update-ref", branch, new_oid, expected_oid])
+        readback = self.read_ref(branch)
+        if readback != new_oid:
+            raise AuthorityGitError("AUTHORITY_CAS_READBACK_MISMATCH")
+        return readback
+
+    def ff_CAS_main_after_verified_protected_merge(self, expected_oid: str, new_oid: str) -> str:
+        _oid(expected_oid, "expected_oid")
+        _oid(new_oid, "new_oid")
+        if not self.ancestry_check(expected_oid, new_oid):
+            raise AuthorityGitError("MAIN_FF_REQUIRED")
+        current = self.read_ref("refs/heads/main")
+        if current != expected_oid:
+            if current == new_oid:
+                return new_oid
+            raise AuthorityGitError("AUTHORITY_MAIN_EXPECTATION_MISMATCH")
+        self._git(["update-ref", "refs/heads/main", new_oid, expected_oid])
+        readback = self.read_ref("refs/heads/main")
+        if readback != new_oid:
+            raise AuthorityGitError("AUTHORITY_MAIN_CAS_READBACK_MISMATCH")
+        return readback
+
+    def commit_file_in_isolated_worktree(
+        self,
+        *,
+        base_oid: str,
+        files: Mapping[str, bytes],
+        message: str,
+        worktree_name: str,
+    ) -> str:
+        _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")
+        root = self.worktree_root / worktree_name
+        if root.exists() or root.is_symlink():
+            raise AuthorityGitError("WORKTREE_ALREADY_EXISTS")
+        self.worktree_root.mkdir(parents=True, exist_ok=True)
+        self._git(["worktree", "add", "--detach", str(root), base_oid])
+        try:
+            for relative, content in files.items():
+                path = Path(relative)
+                if path.is_absolute() or ".." in path.parts or not path.parts or path.parts[0] == ".git":
+                    raise AuthorityGitError("FILE_PATH_INVALID")
+                target = root / path
+                target.parent.mkdir(parents=True, exist_ok=True)
+                target.write_bytes(content)
+            result = subprocess.run(
+                ["git", "-C", str(root), "add", "--", *files.keys()],
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            if result.returncode != 0:
+                raise AuthorityGitError("WORKTREE_ADD_FAILED")
+            result = subprocess.run(
+                ["git", "-C", str(root), "-c", "user.name=aios-durable-job", "-c", "user.email=aios-durable-job@localhost", "commit", "--no-verify", "-m", message],
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            if result.returncode != 0:
+                raise AuthorityGitError("WORKTREE_COMMIT_FAILED")
+            return subprocess.check_output(["git", "-C", str(root), "rev-parse", "HEAD"], text=True).strip()
+        finally:
+            subprocess.run(["git", f"--git-dir={self.git_dir}", "worktree", "remove", "--force", str(root)], capture_output=True, check=False)

active-runtime/app/aios_campaign_recovery.py

@@ -90,12 +90,15 @@ def load_recovered_authorization(path: str | Path) -> dict[str, Any]:
     return {
         "schema": receipt["schema"],
         "state": receipt["recovery_state"],
+        "campaign_id": receipt["campaign_id"],
         "original_event_id": receipt["original_event_id"],
+        "original_authorization_digest": receipt["original_authorization_digest"],
         "signed_grant_id": receipt["signed_grant_id"],
         "task_id": receipt["task_id"],
         "repository": receipt["repository"],
         "original_base_oid": receipt["original_base_oid"],
         "branch": receipt["branch"],
+        "expected_branch_oid": receipt["expected_branch_oid"],
         "digest": actual,
         "new_owner_interaction": False,
     }

active-runtime/app/aios_gitea_machine_transport.py

@@ -0,0 +1,204 @@
+"""Gitea HTTP/Git transport using only a projected systemd credential."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+from typing import Any, Mapping
+from urllib.error import HTTPError, URLError
+from urllib.parse import quote, urlencode
+from urllib.request import Request, urlopen
+
+
+class GiteaTransportError(RuntimeError):
+    pass
+
+
+def _safe_oid(value: str, field: str) -> str:
+    if not isinstance(value, str) or len(value) != 40 or any(ch not in "0123456789abcdef" for ch in value):
+        raise GiteaTransportError(f"{field}_INVALID")
+    return value
+
+
+@dataclass(slots=True)
+class GiteaMachineTransport:
+    api_base: str
+    repository_id: str
+    remote_url: str
+    authority_git_dir: Path
+    credential_name: str = "gitea-machine-token"
+    actor: str = "aios-autonomy-agent"
+    timeout: int = 20
+
+    def __post_init__(self) -> None:
+        if self.credential_name != "gitea-machine-token":
+            raise GiteaTransportError("CREDENTIAL_NAME_INVALID")
+        if "/" not in self.repository_id or self.repository_id.startswith("/"):
+            raise GiteaTransportError("REPOSITORY_INVALID")
+        if not self.api_base.startswith("https://") and not self.api_base.startswith("http://127.0.0.1"):
+            raise GiteaTransportError("GITEA_API_SCOPE_INVALID")
+        if not self.remote_url.startswith("https://"):
+            raise GiteaTransportError("GITEA_REMOTE_SCOPE_INVALID")
+
+    @property
+    def _credential_path(self) -> Path:
+        directory = os.environ.get("CREDENTIALS_DIRECTORY", "")
+        if not directory:
+            raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
+        path = Path(directory) / self.credential_name
+        if path.is_symlink() or not path.is_file():
+            raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
+        return path
+
+    def _token(self) -> str:
+        value = self._credential_path.read_text(encoding="utf-8").strip()
+        if not value:
+            raise GiteaTransportError("CREDENTIAL_EMPTY")
+        return value
+
+    def _request(self, method: str, path: str, body: Mapping[str, Any] | None = None) -> tuple[int, Any]:
+        data = None if body is None else json.dumps(dict(body), separators=(",", ":")).encode()
+        headers = {"Accept": "application/json", "Authorization": f"token {self._token()}"}
+        if data is not None:
+            headers["Content-Type"] = "application/json"
+        request = Request(self.api_base.rstrip("/") + path, data=data, headers=headers, method=method)
+        try:
+            with urlopen(request, timeout=self.timeout) as response:
+                raw = response.read()
+                return response.status, json.loads(raw.decode()) if raw else {}
+        except HTTPError as exc:
+            # Never include response bodies: Gitea may echo sensitive request context.
+            raise GiteaTransportError(f"GITEA_HTTP_{exc.code}") from exc
+        except (URLError, TimeoutError, OSError) as exc:
+            raise GiteaTransportError("GITEA_NETWORK_ERROR") from exc
+
+    def whoami(self) -> Mapping[str, Any]:
+        return self._request("GET", "/user")[1]
+
+    def repository_info(self) -> Mapping[str, Any]:
+        owner, name = self.repository_id.split("/", 1)
+        return self._request("GET", f"/repos/{quote(owner)}/{quote(name)}")[1]
+
+    def repository(self, owner: str, name: str) -> Mapping[str, Any]:
+        if f"{owner}/{name}" != self.repository_id:
+            raise GiteaTransportError("REPOSITORY_SCOPE_DENIED")
+        return self.repository_info()
+
+    def read_branch(self, branch: str) -> str | None:
+        owner, name = self.repository_id.split("/", 1)
+        try:
+            value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")[1]
+        except GiteaTransportError as exc:
+            if str(exc) == "GITEA_HTTP_404":
+                return None
+            raise
+        sha = value.get("commit", {}).get("id") if isinstance(value, dict) else None
+        return _safe_oid(sha, "remote_branch_oid")
+
+    def _askpass(self) -> tuple[Path, dict[str, str]]:
+        credential = self._credential_path
+        fd, name = tempfile.mkstemp(prefix="aios-gitea-askpass-", dir="/tmp")
+        path = Path(name)
+        os.fchmod(fd, 0o700)
+        script = "#!/bin/sh\ncase \"$1\" in *Username*) printf '%s\\n' \"$GITEA_MACHINE_ACTOR\" ;; *) cat \"$CREDENTIAL_FILE\" ;; esac\n"
+        with os.fdopen(fd, "w", encoding="utf-8") as stream:
+            stream.write(script)
+            stream.flush()
+            os.fsync(stream.fileno())
+        return path, {
+            "GIT_ASKPASS": str(path),
+            "GIT_TERMINAL_PROMPT": "0",
+            "GITEA_MACHINE_ACTOR": self.actor,
+            "CREDENTIAL_FILE": str(credential),
+        }
+
+    def _push_new_branch(self, branch: str, new_oid: str) -> None:
+        askpass, additions = self._askpass()
+        environment = os.environ.copy()
+        environment.update(additions)
+        try:
+            result = subprocess.run(
+                ["git", f"--git-dir={self.authority_git_dir}", "push", self.remote_url, f"{new_oid}:refs/heads/{branch}"],
+                text=True,
+                capture_output=True,
+                check=False,
+                env=environment,
+                timeout=self.timeout,
+            )
+            if result.returncode != 0:
+                raise GiteaTransportError("GITEA_BRANCH_CREATE_FAILED")
+        finally:
+            askpass.unlink(missing_ok=True)
+
+    def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str:
+        _safe_oid(expected_oid, "expected_oid")
+        _safe_oid(new_oid, "new_oid")
+        current = self.read_branch(branch)
+        if current == new_oid:
+            return new_oid
+        if current is None:
+            self._push_new_branch(branch, new_oid)
+            readback = self.read_branch(branch)
+            if readback != new_oid:
+                raise GiteaTransportError("GITEA_BRANCH_CREATE_READBACK_MISMATCH")
+            return readback
+        if current != expected_oid:
+            raise GiteaTransportError("GITEA_BRANCH_EXPECTATION_MISMATCH")
+        owner, name = self.repository_id.split("/", 1)
+        status, value = self._request(
+            "PUT",
+            f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}",
+            {"new_commit_id": new_oid, "old_commit_id": expected_oid, "force": False},
+        )
+        if status not in (200, 201, 204):
+            raise GiteaTransportError("GITEA_BRANCH_CAS_FAILED")
+        readback = self.read_branch(branch)
+        if readback != new_oid:
+            raise GiteaTransportError("GITEA_BRANCH_CAS_READBACK_MISMATCH")
+        return readback
+
+    def _pulls(self) -> list[Mapping[str, Any]]:
+        owner, name = self.repository_id.split("/", 1)
+        query = urlencode({"state": "all", "base_branch": "main", "limit": 100})
+        value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls?{query}")[1]
+        return value if isinstance(value, list) else []
+
+    def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]:
+        if target != "main":
+            raise GiteaTransportError("PR_BASE_SCOPE_DENIED")
+        head = self.read_branch(branch)
+        if head is None:
+            raise GiteaTransportError("PR_HEAD_MISSING")
+        matches = [p for p in self._pulls() if (p.get("head") or {}).get("ref") == branch and (p.get("base") or {}).get("ref") == target]
+        open_matches = [p for p in matches if p.get("state") == "open"]
+        if len(open_matches) > 1:
+            raise GiteaTransportError("MULTIPLE_MATCHING_PRS")
+        if open_matches:
+            if (open_matches[0].get("head") or {}).get("sha") != head:
+                raise GiteaTransportError("PR_HEAD_CONFLICT")
+            return open_matches[0]
+        owner, name = self.repository_id.split("/", 1)
+        return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls", {"head": branch, "base": target, "title": title, "body": body, "allow_maintainer_edit": False})[1]
+
+    def publish_status(self, branch: str, state: str, description: str) -> Mapping[str, Any]:
+        oid = self.read_branch(branch)
+        if oid is None:
+            raise GiteaTransportError("STATUS_BRANCH_MISSING")
+        owner, name = self.repository_id.split("/", 1)
+        return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/statuses/{oid}", {"state": state, "context": "aios/durable-release", "description": description})[1]
+
+    def protected_merge(self, pr_number: int) -> Mapping[str, Any]:
+        if not isinstance(pr_number, int) or pr_number < 1:
+            raise GiteaTransportError("PR_NUMBER_INVALID")
+        owner, name = self.repository_id.split("/", 1)
+        pr = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}")[1]
+        if pr.get("state") != "open" or (pr.get("base") or {}).get("ref") != "main":
+            raise GiteaTransportError("PR_MERGE_BINDING_INVALID")
+        if pr.get("mergeable") is False:
+            raise GiteaTransportError("PR_NOT_MERGEABLE")
+        body = {"do": "fast-forward-only", "force_merge": False, "merge_when_checks_succeed": False, "head_commit_id": (pr.get("head") or {}).get("sha")}
+        return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}/merge", body)[1]

active-runtime/app/aios_gitops_job_controller.py

@@ -275,7 +275,31 @@ class GitOpsJobController:
             raise JobError("PAYLOAD_DIGEST_INVALID")
         if request["branch"] in {"main", "master"} or request["branch"].startswith(("refs/", "tags/")):
             raise JobError("PROTECTED_BRANCH_REJECTED")
-        return {field: str(request[field]) for field in fields}
+        binding = {field: str(request[field]) for field in fields}
+        # Operation-specific, non-secret binding inputs must survive the
+        # durable round trip and participate in idempotency. Arbitrary request
+        # fields are intentionally discarded rather than persisted.
+        optional = (
+            "task_id",
+            "new_branch_oid",
+            "authorization_recovery_digest",
+            "lineage_anchor_oid",
+            "generation",
+            "expected_main_oid",
+            "pr_number",
+            "target_branch",
+        )
+        for field in optional:
+            value = request.get(field)
+            if value is None:
+                continue
+            if isinstance(value, bool) or not isinstance(value, (str, int)):
+                raise JobError("JOB_BINDING_OPTION_INVALID")
+            text = str(value).strip()
+            if not text or len(text) > 512:
+                raise JobError("JOB_BINDING_OPTION_INVALID")
+            binding[field] = text
+        return binding
 
     def submit(self, request: Mapping[str, Any], *, payload: bytes | None = None) -> JobReceipt:
         binding = self._binding(request)

active-runtime/app/aios_gitops_job_executor.py

@@ -0,0 +1,238 @@
+"""Durable Job side-effect executor for the AIOS campaign control plane."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+from dataclasses import dataclass
+import json
+from pathlib import Path
+from typing import Any, Mapping
+
+from aios_authority_git_executor import AuthorityGitExecutor, AuthorityGitError
+from aios_campaign_recovery import RecoveryEvidenceError, load_recovered_authorization
+from aios_gitea_machine_broker import CampaignCapabilityReceipt, GiteaMachineBroker, MachineBrokerError
+from aios_gitops_job_controller import GitOpsJobController, JobError, JobState, JobStore
+
+
+TASK_ID = "AIOS_AUTONOMOUS_RELEASE_CAMPAIGN_V1"
+CAMPAIGN_ID = "AIOS-AUTONOMOUS-RELEASE-CAMPAIGN-20260807"
+REPOSITORY = "aios/aios"
+BRANCH = "agent/autonomous-release-campaign-v1-20260807"
+EXPECTED_BRANCH_OID = "bba675fab3b7e0c5fba7066720252b3c91b9f20d"
+AUTHORIZATION_DIGEST = "6b0bf550eb702126e967eeff2d561fa808444b924d8a71f5399accad508ed3b2"
+LINEAGE_ANCHOR = "69b45e81dcd2c3707c367003b08c55fd361faae8"
+
+
+class JobExecutionError(RuntimeError):
+    pass
+
+
+@dataclass(slots=True)
+class DurableJobExecutor:
+    controller: GitOpsJobController
+    authority: AuthorityGitExecutor
+    broker: GiteaMachineBroker
+    recovery_path: Path
+
+    def _request(self, job_id: str) -> tuple[dict[str, Any], dict[str, Any]]:
+        row = self.controller.store.read(job_id)
+        try:
+            request = json.loads(row["request_json"])
+        except (KeyError, TypeError, json.JSONDecodeError) as exc:
+            raise JobExecutionError("JOB_REQUEST_INVALID") from exc
+        return row, dict(request.get("binding") or {})
+
+    def _validate_campaign(self, request: Mapping[str, Any]) -> str:
+        required = {
+            "task_id": TASK_ID,
+            "campaign_id": CAMPAIGN_ID,
+            "repository": REPOSITORY,
+            "branch": BRANCH,
+            "expected_branch_oid": EXPECTED_BRANCH_OID,
+            "authorization_recovery_digest": AUTHORIZATION_DIGEST,
+        }
+        for key, expected in required.items():
+            if request.get(key) != expected:
+                raise JobExecutionError(f"JOB_{key.upper()}_MISMATCH")
+        new_oid = request.get("new_branch_oid")
+        if not isinstance(new_oid, str) or len(new_oid) != 40 or any(ch not in "0123456789abcdef" for ch in new_oid):
+            raise JobExecutionError("NEW_BRANCH_OID_INVALID")
+        try:
+            recovery = load_recovered_authorization(self.recovery_path)
+        except RecoveryEvidenceError as exc:
+            raise JobExecutionError(str(exc)) from exc
+        if recovery["digest"] != AUTHORIZATION_DIGEST:
+            raise JobExecutionError("RECOVERY_DIGEST_MISMATCH")
+        for key, expected in required.items():
+            if key == "campaign_id":
+                actual = recovery.get(key)
+            elif key == "authorization_recovery_digest":
+                continue
+            else:
+                actual = recovery.get({"expected_branch_oid": "expected_branch_oid"}.get(key, key))
+            if actual is not None and actual != expected:
+                raise JobExecutionError("RECOVERY_BINDING_MISMATCH")
+        if recovery.get("expected_branch_oid") != EXPECTED_BRANCH_OID:
+            raise JobExecutionError("RECOVERY_BRANCH_OID_MISMATCH")
+        authority_head = self.authority.read_ref(f"refs/heads/{BRANCH}")
+        if authority_head not in {EXPECTED_BRANCH_OID, new_oid}:
+            raise JobExecutionError("AUTHORITY_BRANCH_EXPECTATION_MISMATCH")
+        if not self.authority.ancestry_check(EXPECTED_BRANCH_OID, new_oid):
+            raise JobExecutionError("NEW_OID_NOT_DESCENDANT")
+        identity = self.broker.verify_identity()
+        if identity.get("repository") != REPOSITORY or identity.get("login") == "aiman":
+            raise JobExecutionError("MACHINE_IDENTITY_SCOPE_INVALID")
+        return new_oid
+
+    def dry_run(self, job_id: str) -> dict[str, Any]:
+        _row, request = self._request(job_id)
+        new_oid = self._validate_campaign(request)
+        remote = self.broker.readback()
+        if remote is None:
+            action = "CREATE_REQUIRED"
+        elif remote == new_oid:
+            action = "ALREADY_IN_SYNC"
+        elif remote == EXPECTED_BRANCH_OID:
+            action = "FAST_FORWARD_REQUIRED"
+        else:
+            action = "MANUAL_OWNER_REVIEW_REQUIRED"
+        if action == "MANUAL_OWNER_REVIEW_REQUIRED":
+            raise JobExecutionError("GITEA_BRANCH_CONFLICT")
+        return {
+            "decision": "SAFE_TO_ADVANCE",
+            "authority_action": "CAS_REQUIRED",
+            "gitea_action": action,
+            "previous_branch_oid": EXPECTED_BRANCH_OID,
+            "final_branch_oid": new_oid,
+            "force_push": False,
+        }
+
+    def _recover_requested(self, job_id: str) -> dict[str, Any]:
+        row = self.controller.store.read(job_id)
+        if JobState(row["state"]) in {JobState.RECOVERABLE_FAILURE, JobState.RECONCILIATION_REQUIRED}:
+            self.controller.resume(job_id)
+            row = self.controller.store.read(job_id)
+        return row
+
+    def execute(self, job_id: str, *, dry_run: bool = False) -> dict[str, Any]:
+        row, request = self._request(job_id)
+        operation = request.get("operation")
+        if operation == "ADVANCE_EXISTING_CAMPAIGN_BRANCH":
+            if dry_run:
+                return self.dry_run(job_id)
+            return self._advance_campaign(job_id)
+        if operation == "COMMIT_FILE":
+            return self._commit_file(job_id, request)
+        if operation == "SYNC_TASK_BRANCH_TO_GITEA":
+            new_oid = self._validate_campaign(request)
+            final = self.broker.push_branch(new_oid)
+            return {"operation": operation, "remote_verified": final == new_oid, "final_network_oid": final}
+        if operation == "CREATE_PR":
+            return dict(self.broker.create_pr("AIOS autonomous release control plane", "Durable campaign continuation; exact head and recovery evidence are bound in the Job."))
+        if operation == "PUBLISH_CHECK":
+            return dict(self.broker.status("success", "Durable campaign control-plane checks passed"))
+        if operation == "PROTECTED_MERGE":
+            number = int(request.get("pr_number", "0"))
+            return dict(self.broker.merge(number))
+        if operation == "RECONCILE_AUTHORITY_MAIN":
+            old = request.get("expected_main_oid")
+            new = request.get("new_branch_oid")
+            if not isinstance(old, str) or not isinstance(new, str):
+                raise JobExecutionError("MAIN_RECONCILIATION_BINDING_INVALID")
+            return {"authority_main": self.authority.ff_CAS_main_after_verified_protected_merge(old, new)}
+        raise JobExecutionError("OPERATION_NOT_ALLOWED")
+
+    def _advance_campaign(self, job_id: str) -> dict[str, Any]:
+        self._recover_requested(job_id)
+        row, request = self._request(job_id)
+        new_oid = self._validate_campaign(request)
+        state = JobState(row["state"])
+        if state == JobState.REQUESTED:
+            receipt = self.controller.advance(job_id, JobState.AUTHORIZED, side_effect_state="AUTHORIZED", receipt={"authorization_digest": AUTHORIZATION_DIGEST, "task_id": TASK_ID})
+            row = self.controller.store.read(job_id)
+        if JobState(row["state"]) == JobState.AUTHORIZED:
+            self.controller.lease_job(job_id, lease_id=f"lease-{job_id}", lineage_anchor_oid=LINEAGE_ANCHOR, current_head_oid=EXPECTED_BRANCH_OID, generation=1, lease_expires_at="2099-01-01T00:00:00Z", expected_revision=int(row["revision"]))
+            row = self.controller.store.read(job_id)
+        if JobState(row["state"]) == JobState.LEASED:
+            self.controller.advance(job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY", receipt={"branch": BRANCH, "expected_oid": EXPECTED_BRANCH_OID})
+            row = self.controller.store.read(job_id)
+        if JobState(row["state"]) == JobState.WORKTREE_READY:
+            self.controller.advance(job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
+            row = self.controller.store.read(job_id)
+        authority_before = self.authority.read_ref(f"refs/heads/{BRANCH}")
+        if authority_before == EXPECTED_BRANCH_OID:
+            authority_oid = self.authority.advance_task_branch_CAS(BRANCH and f"refs/heads/{BRANCH}", EXPECTED_BRANCH_OID, new_oid)
+        elif authority_before == new_oid:
+            authority_oid = new_oid
+        else:
+            raise JobExecutionError("AUTHORITY_BRANCH_CONFLICT")
+        row = self.controller.store.read(job_id)
+        if JobState(row["state"]) == JobState.EXECUTING:
+            self.controller.advance(job_id, JobState.LOCAL_VERIFIED, side_effect_state="LOCAL_VERIFIED", receipt={"authority": {"branch": BRANCH, "previous_oid": EXPECTED_BRANCH_OID, "new_oid": new_oid, "CAS_success": True, "readback_oid": authority_oid}})
+        row = self.controller.store.read(job_id)
+        remote_before = self.broker.readback()
+        if remote_before != new_oid:
+            if remote_before not in {None, EXPECTED_BRANCH_OID}:
+                raise JobExecutionError("GITEA_BRANCH_CONFLICT")
+            if JobState(row["state"]) == JobState.LOCAL_VERIFIED:
+                self.controller.advance(job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS", receipt={"gitea": {"previous_oid": remote_before, "desired_oid": new_oid, "force_push": False}})
+            final_network = self.broker.push_branch(new_oid)
+        else:
+            final_network = new_oid
+        final_api = self.broker.readback()
+        if final_api != new_oid:
+            raise JobExecutionError("GITEA_BRANCH_READBACK_MISMATCH")
+        row = self.controller.store.read(job_id)
+        if JobState(row["state"]) in {JobState.LOCAL_VERIFIED, JobState.REMOTE_SYNCING}:
+            self.controller.advance(job_id, JobState.REMOTE_VERIFIED, side_effect_state="REMOTE_VERIFIED", receipt={"gitea": {"machine_actor": "aios-autonomy-agent", "previous_oid": remote_before, "desired_oid": new_oid, "final_network_oid": final_network, "final_API_oid": final_api, "force_push": False}})
+        row = self.controller.store.read(job_id)
+        if JobState(row["state"]) == JobState.REMOTE_VERIFIED:
+            self.controller.advance(job_id, JobState.COMPLETED, side_effect_state="REMOTE_VERIFIED", receipt={"authority_oid": authority_oid, "network_oid": final_network, "gitea_api_oid": final_api, "force_push": False})
+        return self.controller.store._receipt(self.controller.store.read(job_id), duplicate=True).as_dict()
+
+    def _commit_file(self, job_id: str, request: Mapping[str, Any]) -> dict[str, Any]:
+        row = self.controller.store.read(job_id)
+        payload_ref = request.get("payload_ref")
+        if not payload_ref:
+            raise JobExecutionError("PAYLOAD_REQUIRED")
+        payload = self.controller.store.payloads.read_verified(payload_ref, row["payload_digest"], int(request["payload_size"]))
+        try:
+            spec = json.loads(payload.decode())
+            path = str(spec["path"])
+            content = base64.b64decode(spec["content_base64"], validate=True)
+        except Exception as exc:
+            raise JobExecutionError("COMMIT_FILE_PAYLOAD_INVALID") from exc
+        new_oid = self.authority.commit_file_in_isolated_worktree(base_oid=row["expected_branch_oid"], files={path: content}, message="chore(gitops): durable commit-file", worktree_name=job_id)
+        return {"operation": "COMMIT_FILE", "new_oid": new_oid}
+
+
+def build_from_environment(job_root: Path) -> DurableJobExecutor:
+    from aios_gitea_machine_transport import GiteaMachineTransport
+    from aios_gitea_machine_broker import MachineCredentialProjection
+
+    account = __import__("os").environ.get("AIOS_GITEA_MACHINE_ACCOUNT", "aios-autonomy-agent")
+    api = __import__("os").environ.get("AIOS_GITEA_API", "https://g.cnnetmkt.top/git/api/v1")
+    remote = __import__("os").environ.get("AIOS_GITEA_REMOTE", "https://g.cnnetmkt.top/git/aios/aios.git")
+    authority_dir = Path(__import__("os").environ.get("AIOS_GITOPS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git"))
+    worktree_root = Path(__import__("os").environ.get("AIOS_GITOPS_WORKTREE_ROOT", "/var/lib/aios-gitops/worktrees"))
+    recovery = Path(__import__("os").environ.get("AIOS_RELEASE_RECOVERY_RECEIPT", "/var/lib/aios-gitops/release-campaigns/AIOS-AUTONOMOUS-RELEASE-CAMPAIGN-20260807/authorization-recovery.json"))
+    transport = GiteaMachineTransport(api, REPOSITORY, remote, authority_dir)
+    credential = MachineCredentialProjection.from_mapping({"scheme": "systemd-credential", "name": "gitea-machine-token", "account": account})
+    receipt = CampaignCapabilityReceipt(CAMPAIGN_ID, TASK_ID, REPOSITORY, BRANCH, EXPECTED_BRANCH_OID, frozenset({"exact_task_branch_push", "pr_create_update", "status_publish", "protected_merge", "remote_readback"}), frozenset({"direct_main_push", "force_push", "tag_update", "tag_delete", "baseline_write", "production_deploy"}))
+    return DurableJobExecutor(GitOpsJobController(JobStore(job_root)), AuthorityGitExecutor(authority_dir, worktree_root), GiteaMachineBroker(transport, credential, receipt), recovery)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--job-root", required=True, type=Path)
+    parser.add_argument("--job-id", required=True)
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args(argv)
+    result = build_from_environment(args.job_root).execute(args.job_id, dry_run=args.dry_run)
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/aios_gitops_writer_worker.py

@@ -29,10 +29,16 @@ def runtime_policy(environment: Mapping[str, str] | None = None) -> dict[str, bo
     return {"legacy_global_write": legacy, "durable_job_execution": durable, "repository_mutation_allowed": durable}
 
 
-def handle_request(controller: GitOpsJobController, request: Mapping[str, Any]) -> dict[str, Any]:
+def handle_request(controller: GitOpsJobController, request: Mapping[str, Any], executor: Any | None = None) -> dict[str, Any]:
     job_id = str(request.get("job_id") or "")
     if not job_id:
         raise JobError("JOB_ID_REQUIRED")
+    if request.get("action") == "execute":
+        if executor is None:
+            raise JobError("JOB_EXECUTOR_UNAVAILABLE")
+        result = executor.execute(job_id, dry_run=bool(request.get("dry_run", False)))
+        receipt = controller.store._receipt(controller.store.read(job_id), duplicate=True)
+        return {"job": result, **receipt.as_dict()}
     if request.get("action") == "resume":
         receipt = controller.resume(job_id)
     else:
@@ -41,7 +47,7 @@ def handle_request(controller: GitOpsJobController, request: Mapping[str, Any])
     return receipt.as_dict()
 
 
-def serve_socket(sock: socket.socket, controller: GitOpsJobController, stopping: threading.Event) -> None:
+def serve_socket(sock: socket.socket, controller: GitOpsJobController, stopping: threading.Event, executor: Any | None = None) -> None:
     sock.listen(32)
     sock.settimeout(1.0)
     while not stopping.is_set():
@@ -53,7 +59,7 @@ def serve_socket(sock: socket.socket, controller: GitOpsJobController, stopping:
             try:
                 raw = connection.makefile("rb").readline(1_048_576)
                 request = json.loads(raw.decode("utf-8"))
-                response = {"status": "ok", **handle_request(controller, request)}
+                response = {"status": "ok", **handle_request(controller, request, executor)}
             except ServiceUnavailable as exc:
                 response = {"status": "error", "error": str(exc), "retryable": True, "side_effect_state": "UNKNOWN"}
             except Exception as exc:
@@ -77,12 +83,14 @@ def main(argv: list[str] | None = None) -> int:
     if not policy["durable_job_execution"]:
         raise RuntimeError("DURABLE_JOB_EXECUTION_GATE_CLOSED")
     controller = GitOpsJobController(JobStore(args.job_root))
+    from aios_gitops_job_executor import build_from_environment
+    executor = build_from_environment(args.job_root)
     stopping = threading.Event()
     signal.signal(signal.SIGTERM, lambda *_: stopping.set())
     signal.signal(signal.SIGINT, lambda *_: stopping.set())
     sock = socket_from_systemd()
     try:
-        serve_socket(sock, controller, stopping)
+        serve_socket(sock, controller, stopping, executor)
     finally:
         sock.close()
     return 0

deploy/systemd/aios-gitops-writer.service

@@ -12,6 +12,12 @@ Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
 Environment=AIOS_GITOPS_JOB_ROOT=/var/lib/aios-gitops/jobs
 Environment=AIOS_GITEA_MACHINE_CREDENTIAL_REF=systemd-credential:gitea-machine-token
+Environment=AIOS_GITOPS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_GITOPS_WORKTREE_ROOT=/var/lib/aios-gitops/worktrees
+Environment=AIOS_GITEA_API=https://g.cnnetmkt.top/git/api/v1
+Environment=AIOS_GITEA_REMOTE=https://g.cnnetmkt.top/git/aios/aios.git
+Environment=AIOS_GITEA_MACHINE_ACCOUNT=aios-autonomy-agent
+Environment=AIOS_RELEASE_RECOVERY_RECEIPT=/var/lib/aios-gitops/release-campaigns/AIOS-AUTONOMOUS-RELEASE-CAMPAIGN-20260807/authorization-recovery.json
 LoadCredentialEncrypted=gitea-machine-token:/etc/aios/credentials/gitea-machine.token.cred
 Environment=PYTHONDONTWRITEBYTECODE=1
 ExecStart=/usr/bin/python3 -B /opt/aios-gitops/active-runtime/app/aios_gitops_writer_worker.py --job-root ${AIOS_GITOPS_JOB_ROOT}
@@ -21,7 +27,7 @@ NoNewPrivileges=no
 PrivateTmp=true
 ProtectSystem=strict
 ProtectHome=true
-ReadWritePaths=/var/lib/aios-gitops/jobs /var/lib/aios-gitops/payloads
+ReadWritePaths=/var/lib/aios-gitops/jobs /var/lib/aios-gitops/payloads /var/lib/aios-gitops/worktrees
 
 [Install]
 WantedBy=multi-user.target

tests/test_live_executor_contract.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+import aios_gitops_job_executor as executor_module
+from aios_authority_git_executor import AuthorityGitError, AuthorityGitExecutor
+from aios_gitea_machine_transport import GiteaMachineTransport
+from aios_gitea_machine_broker import CampaignCapabilityReceipt, GiteaMachineBroker, MachineCredentialProjection
+from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore, JobState
+from aios_gitops_writer_worker import handle_request
+
+
+OLD = "bba675fab3b7e0c5fba7066720252b3c91b9f20d"
+NEW = "3fbbc18fb2131bbbd9e511004a59ee95cf3e86b7"
+
+
+class FakeAuthority:
+    def __init__(self, descendant=True):
+        self.head = OLD
+        self.descendant = descendant
+        self.cas_calls = []
+
+    def read_ref(self, ref):
+        return self.head
+
+    def ancestry_check(self, older, newer):
+        return self.descendant and older == OLD and newer == NEW
+
+    def advance_task_branch_CAS(self, branch, expected, new):
+        if self.head != expected:
+            raise AuthorityGitError("AUTHORITY_BRANCH_EXPECTATION_MISMATCH")
+        self.cas_calls.append((branch, expected, new))
+        self.head = new
+        return new
+
+
+class FakeBroker:
+    def __init__(self):
+        self.remote = None
+        self.push_calls = []
+
+    def verify_identity(self):
+        return {"login": "aios-autonomy-agent", "repository": "aios/aios"}
+
+    def readback(self):
+        return self.remote
+
+    def push_branch(self, new):
+        self.push_calls.append(new)
+        self.remote = new
+        return new
+
+    def create_pr(self, *args):
+        return {"number": 1}
+
+    def status(self, *args):
+        return {"status": "success"}
+
+    def merge(self, *args):
+        return {"merged": True}
+
+
+def _request(**extra):
+    value = {
+        "campaign_id": executor_module.CAMPAIGN_ID,
+        "repository": executor_module.REPOSITORY,
+        "operation": "ADVANCE_EXISTING_CAMPAIGN_BRANCH",
+        "branch": executor_module.BRANCH,
+        "expected_branch_oid": OLD,
+        "payload_digest": "0" * 64,
+        "task_id": executor_module.TASK_ID,
+        "new_branch_oid": NEW,
+        "authorization_recovery_digest": executor_module.AUTHORIZATION_DIGEST,
+    }
+    value.update(extra)
+    return value
+
+
+def _executor(temp, authority=None, broker=None):
+    controller = GitOpsJobController(JobStore(temp))
+    job = controller.submit(_request())
+    executor = executor_module.DurableJobExecutor(controller, authority or FakeAuthority(), broker or FakeBroker(), Path(temp) / "recovery.json")
+    return controller, job, executor
+
+
+def _recovery():
+    return {
+        "digest": executor_module.AUTHORIZATION_DIGEST,
+        "campaign_id": executor_module.CAMPAIGN_ID,
+        "task_id": executor_module.TASK_ID,
+        "repository": executor_module.REPOSITORY,
+        "branch": executor_module.BRANCH,
+        "expected_branch_oid": OLD,
+    }
+
+
+def test_advance_existing_branch_success_and_duplicate_execute(monkeypatch):
+    monkeypatch.setattr(executor_module, "load_recovered_authorization", lambda _path: _recovery())
+    with tempfile.TemporaryDirectory() as temp:
+        controller, job, executor = _executor(temp)
+        first = executor.execute(job.job_id)
+        second = executor.execute(job.job_id)
+        assert first["state"] == JobState.COMPLETED.value
+        assert second["state"] == JobState.COMPLETED.value
+        assert executor.authority.cas_calls == [(f"refs/heads/{executor_module.BRANCH}", OLD, NEW)]
+        assert executor.broker.push_calls == [NEW]
+
+
+def test_authorization_digest_required(monkeypatch):
+    monkeypatch.setattr(executor_module, "load_recovered_authorization", lambda _path: _recovery())
+    with tempfile.TemporaryDirectory() as temp:
+        controller = GitOpsJobController(JobStore(temp))
+        job = controller.submit(_request(authorization_recovery_digest="bad"))
+        executor = executor_module.DurableJobExecutor(controller, FakeAuthority(), FakeBroker(), Path(temp) / "recovery.json")
+        with pytest.raises(executor_module.JobExecutionError, match="AUTHORIZATION_RECOVERY_DIGEST_MISMATCH"):
+            executor.dry_run(job.job_id)
+
+
+def test_wrong_branch_rejected(monkeypatch):
+    monkeypatch.setattr(executor_module, "load_recovered_authorization", lambda _path: _recovery())
+    with tempfile.TemporaryDirectory() as temp:
+        controller = GitOpsJobController(JobStore(temp))
+        job = controller.submit(_request(branch="agent/not-the-campaign"))
+        executor = executor_module.DurableJobExecutor(controller, FakeAuthority(), FakeBroker(), Path(temp) / "recovery.json")
+        with pytest.raises(executor_module.JobExecutionError, match="BRANCH_MISMATCH"):
+            executor.dry_run(job.job_id)
+
+
+def test_non_descendant_new_oid_rejected(monkeypatch):
+    monkeypatch.setattr(executor_module, "load_recovered_authorization", lambda _path: _recovery())
+    with tempfile.TemporaryDirectory() as temp:
+        authority = FakeAuthority(descendant=False)
+        controller, job, executor = _executor(temp, authority=authority)
+        with pytest.raises(executor_module.JobExecutionError, match="NEW_OID_NOT_DESCENDANT"):
+            executor.dry_run(job.job_id)
+
+
+def test_worker_execute_action_consumes_job():
+    with tempfile.TemporaryDirectory() as temp:
+        controller = GitOpsJobController(JobStore(temp))
+        job = controller.submit(_request())
+
+        class Stub:
+            def execute(self, job_id, *, dry_run=False):
+                assert job_id == job.job_id
+                assert dry_run is True
+                return {"decision": "SAFE_TO_ADVANCE"}
+
+        result = handle_request(controller, {"action": "execute", "job_id": job.job_id, "dry_run": True}, Stub())
+        assert result["job"]["decision"] == "SAFE_TO_ADVANCE"
+
+
+def test_machine_transport_branch_cas_is_not_force_push(monkeypatch):
+    transport = GiteaMachineTransport("http://127.0.0.1:3000/api/v1", "aios/aios", "https://g.cnnetmkt.top/git/aios/aios.git", Path("/tmp/repo.git"))
+    calls = []
+    def read_branch(_self, _branch):
+        return NEW if calls else OLD
+    monkeypatch.setattr(GiteaMachineTransport, "read_branch", read_branch)
+    monkeypatch.setattr(GiteaMachineTransport, "_request", lambda _self, method, path, body=None: calls.append((method, path, body)) or (200, {}))
+    assert transport.compare_and_swap_branch(executor_module.BRANCH, OLD, NEW) == NEW
+    assert calls[0][0] == "PUT"
+    assert calls[0][2]["force"] is False