evidence: close live activation hardening candidate

Source: patch/isolated-writer-token-policy-v2-20260724
Source commit: b1e240dda3
Target base: df099420d2
Production runtime changes: not authorized
Tag/Baseline/Registry operations: not authorized

active-runtime/app/aios_isolated_writer_api.py

@@ -5,13 +5,148 @@ import json
 import os
 import subprocess
 from pathlib import Path
+import urllib.parse
 
 import aios_chatgpt_actions_api as api
 import aios_isolated_writer_policy as policy
+from isolated_writer_authz import AuthorizationError, authorize
+from isolated_writer_commit_file import AtomicCommitFile, CommitFileError, OperationLedger
+from isolated_writer_token_registry import TokenRegistry
 
 _reject = api.reject_non_read_http
 _write_file = api.write_file
 _sync_branch = api.sync_branch
+_original_do_get = api.H.do_GET
+api.OPENAPI = Path(os.getenv("AIOS_GITOPS_ISOLATED_WRITER_OPENAPI", str(api.OPENAPI)))
+
+
+def _registry() -> TokenRegistry:
+    return TokenRegistry(Path(os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"]))
+
+
+def _atomic_writer() -> AtomicCommitFile:
+    return AtomicCommitFile(
+        workspace=api.WS,
+        bare_repo=api.BARE_REPO,
+        ledger=OperationLedger(Path(os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"])),
+        remote=os.getenv("AIOS_GITOPS_GITEA_PUSH_REMOTE", ""),
+        audit=policy.append_audit_record,
+    )
+
+
+def _bearer(handler) -> str:
+    value = handler.headers.get("Authorization", "")
+    return value[7:].strip() if value.startswith("Bearer ") else ""
+
+
+def _authorize(handler, *, method: str, path: str, branch=None, target_path=None):
+    grant = authorize(
+        _registry(),
+        _bearer(handler),
+        method=method,
+        path=path,
+        source_ip=str(handler.client_address[0]),
+        workspace=api.WS,
+        branch=branch,
+        target_path=target_path,
+    )
+    grant.record["_source_ip"] = str(handler.client_address[0])
+    return grant
+
+
+def _authorization_error(handler, exc: AuthorizationError):
+    return handler.sendj({"error": exc.code}, exc.status)
+
+
+def _isolated_runtime_status(handler):
+    branch = api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"]
+    head = api.sh(["rev-parse", "HEAD"], True)["stdout"]
+    return handler.sendj({
+        "service": os.getenv(
+            "AIOS_GITOPS_SERVICE_AUDIENCE",
+            "aios-isolated-writer.service",
+        ),
+        "status": "active",
+        "runtime_role": os.getenv("AIOS_GITOPS_RUNTIME_ROLE"),
+        "workspace_role": os.getenv("AIOS_GITOPS_WORKSPACE_ROLE"),
+        "writes_enabled": os.getenv("AIOS_GITOPS_WRITES_ENABLED", "").lower()
+        in {"1", "true", "yes", "on"},
+        "allowed_branch": policy.allowed_branch(),
+        "approved_base_oid": policy.allowed_base_oid(),
+        "workspace_branch": branch,
+        "workspace_head_oid": head,
+        "repository_identity": os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY"),
+        "token_policy_version": 2,
+        "atomic_commit_file": True,
+        "path_policy_sha256": policy.configured_path_policy_sha256(),
+    })
+
+
+def _isolated_do_get(self):
+    path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+    if path in {"/health", "/api/health"}:
+        return self.sendj({
+            "status": "ok",
+            "service": os.getenv(
+                "AIOS_GITOPS_SERVICE_AUDIENCE",
+                "aios-isolated-writer.service",
+            ),
+        })
+    if path not in {"/api/runtime-status", "/api/baseline", "/api/patches"}:
+        return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
+    try:
+        _authorize(self, method="GET", path=path)
+    except AuthorizationError as exc:
+        return _authorization_error(self, exc)
+    if path == "/api/runtime-status":
+        return _isolated_runtime_status(self)
+    return _original_do_get(self)
+
+
+def _read_commit_file_body(handler):
+    try:
+        length = int(handler.headers.get("Content-Length", "0"))
+    except ValueError:
+        raise CommitFileError("writer_content_length_rejected") from None
+    if length <= 0:
+        raise CommitFileError("writer_request_object_required")
+    if length > 1_100_000:
+        raise CommitFileError("writer_request_too_large", 413)
+    raw = handler.rfile.read(length)
+    try:
+        value = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise CommitFileError("writer_request_json_rejected") from None
+    if not isinstance(value, dict):
+        raise CommitFileError("writer_request_object_required")
+    return value
+
+
+def _isolated_do_post(self):
+    path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+    if path != "/api/commit-file":
+        return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
+    try:
+        data = _read_commit_file_body(self)
+        branch = str(data.get("branch") or "")
+        target_path = str(data.get("path") or "")
+        grant = _authorize(
+            self,
+            method="POST",
+            path=path,
+            branch=branch,
+            target_path=target_path,
+        )
+        if api.content_has_secret(str(data.get("content") or "")):
+            raise CommitFileError("writer_secret_content_rejected", 403)
+        result = _atomic_writer().execute(grant, data)
+        return self.sendj(result)
+    except AuthorizationError as exc:
+        return _authorization_error(self, exc)
+    except CommitFileError as exc:
+        return self.sendj({"error": exc.code}, exc.status)
+    except PermissionError:
+        return self.sendj({"error": "isolated_writer_policy_rejected"}, 403)
 
 
 def reject_non_read_http(handler, method: str, path: str) -> bool:
@@ -173,6 +308,8 @@ api.H.branch, api.H.file, api.H.commit, api.H.patch = (
     _checked_commit,
     _checked_patch,
 )
+api.H.do_GET = _isolated_do_get
+api.H.do_POST = _isolated_do_post
 
 if __name__ == "__main__":
     policy.require_service_invariants(api.WS, api.BARE_REPO)

active-runtime/app/aios_isolated_writer_policy.py

@@ -9,10 +9,9 @@ import re
 from pathlib import Path, PurePosixPath, PureWindowsPath
 from urllib.parse import unquote
 
-WRITE_ENDPOINTS = frozenset({
-    "/api/branch", "/api/create-branch", "/api/file", "/api/commit-file",
-    "/api/commit", "/api/patch", "/api/create-patch",
-})
+import isolated_writer_path_policy as path_policy
+
+WRITE_ENDPOINTS = frozenset({"/api/commit-file"})
 ALLOWED_ROOTS = (
     "active-runtime",
     "docs",
@@ -69,6 +68,15 @@ def require_service_invariants(workspace: Path, bare_repo: Path) -> None:
         raise PermissionError("isolated_writer_role_required")
     if os.getenv("AIOS_GITOPS_WORKSPACE_ROLE", "") != "isolated-writer":
         raise PermissionError("isolated_writer_workspace_role_required")
+    if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
+        raise PermissionError("isolated_writer_token_policy_v2_required")
+    if not os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", "").strip():
+        raise PermissionError("isolated_writer_service_audience_required")
+    if not os.getenv("AIOS_GITOPS_TOKEN_REGISTRY_V2", "").strip():
+        raise PermissionError("isolated_writer_token_registry_required")
+    if not os.getenv("AIOS_GITOPS_WRITER_OPERATION_LEDGER", "").strip():
+        raise PermissionError("isolated_writer_operation_ledger_required")
+    path_policy.configured_path_policy()
     for name in _DENY_FLAGS:
         if os.getenv(name, "").strip().lower() not in _FALSE:
             raise PermissionError("isolated_writer_forbidden_capability_flag")
@@ -182,6 +190,7 @@ def _audit_sanitize(value):
                         "private_key",
                     )
                 )
+                and str(key).lower() != "token_id"
                 else _audit_sanitize(inner)
             )
             for key, inner in value.items()
@@ -239,9 +248,14 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
     ):
         raise PermissionError("isolated_writer_path_rejected")
     first = parts[0]
+    service_exact_path = False
     if first in DENIED_ROOTS:
-        raise PermissionError("isolated_writer_protected_path_rejected")
-    if value != "README.md" and first not in ALLOWED_ROOTS:
+        try:
+            path_policy.require_service_path("commit_file.atomic", value)
+            service_exact_path = True
+        except path_policy.PathPolicyError as exc:
+            raise PermissionError(str(exc)) from None
+    if value != "README.md" and first not in ALLOWED_ROOTS and not service_exact_path:
         raise PermissionError("isolated_writer_path_outside_allowed_roots")
     if workspace is not None:
         lexical_root = Path(os.path.abspath(str(workspace)))
@@ -261,6 +275,22 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
     return value
 
 
+def validate_content_for_path(path: str, content: str) -> None:
+    if path.split("/", 1)[0] not in DENIED_ROOTS:
+        return
+    try:
+        path_policy.validate_content("commit_file.atomic", path, content)
+    except path_policy.PathPolicyError as exc:
+        raise PermissionError(str(exc)) from None
+
+
+def configured_path_policy_sha256() -> str | None:
+    try:
+        return path_policy.configured_path_policy_sha256()
+    except path_policy.PathPolicyError as exc:
+        raise PermissionError(str(exc)) from None
+
+
 def validate_commit_paths(paths: list[str], workspace: Path | None = None) -> None:
     if enforcement_required():
         for path in paths:

active-runtime/app/isolated_writer_authz.py

@@ -0,0 +1,167 @@
+"""Fail-closed operation authorization for an isolated Git writer."""
+from __future__ import annotations
+
+import ipaddress
+import os
+import re
+import subprocess
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from isolated_writer_token_registry import TokenRegistry
+
+
+ROUTE_OPERATIONS = {
+    ("GET", "/api/runtime-status"): "runtime.status.read",
+    ("GET", "/api/baseline"): "baseline.read",
+    ("GET", "/api/patches"): "patches.read",
+    ("POST", "/api/commit-file"): "commit_file.atomic",
+}
+OID_RE = re.compile(r"[0-9a-f]{40}")
+
+
+class AuthorizationError(PermissionError):
+    def __init__(self, code: str, status: int = 403):
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+@dataclass(frozen=True)
+class AuthorizedGrant:
+    record: dict[str, Any]
+    operation: str
+    use_count: int
+
+    @property
+    def token_id(self) -> str:
+        return str(self.record["token_id"])
+
+    @property
+    def task_id(self) -> str:
+        return str(self.record["task_id"])
+
+
+def operation_for(method: str, path: str) -> str:
+    operation = ROUTE_OPERATIONS.get((method.upper(), path))
+    if not operation:
+        raise AuthorizationError("isolated_writer_operation_forbidden")
+    return operation
+
+
+def _parse_time(value: Any) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except (TypeError, ValueError):
+        raise AuthorizationError("writer_token_time_invalid") from None
+    if parsed.tzinfo is None:
+        raise AuthorizationError("writer_token_time_invalid")
+    return parsed.astimezone(timezone.utc)
+
+
+def _git_is_ancestor(workspace: Path, base: str, branch: str) -> bool:
+    result = subprocess.run(
+        ["git", "merge-base", "--is-ancestor", base, branch],
+        cwd=workspace,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    return result.returncode == 0
+
+
+def _validate_grant(
+    grant: dict[str, Any],
+    *,
+    audience: str,
+    source_ip: str,
+    operation: str,
+    branch: str | None,
+    path: str | None,
+    workspace: Path,
+    current_time: datetime,
+) -> None:
+    if int(grant.get("schema_version", 0)) != 2:
+        raise AuthorizationError("writer_token_policy_v2_required", 401)
+    if str(grant.get("audience") or "") != audience:
+        raise AuthorizationError("writer_token_audience_rejected")
+    try:
+        peer = ipaddress.ip_address(source_ip)
+        networks = [ipaddress.ip_network(str(value), strict=True) for value in grant.get("source_cidrs", [])]
+    except ValueError:
+        raise AuthorizationError("writer_token_source_rejected") from None
+    if not networks or not any(peer in network for network in networks):
+        raise AuthorizationError("writer_token_source_rejected")
+    created = _parse_time(grant.get("created_at"))
+    not_before = _parse_time(grant.get("not_before"))
+    expires = _parse_time(grant.get("expires_at"))
+    if expires <= not_before or (expires - created).total_seconds() > 1800:
+        raise AuthorizationError("writer_token_lifetime_rejected")
+    if current_time < not_before:
+        raise AuthorizationError("writer_token_not_yet_valid", 401)
+    if current_time >= expires:
+        raise AuthorizationError("writer_token_expired", 401)
+    if bool(grant.get("revoked")):
+        raise AuthorizationError("writer_token_revoked", 401)
+    operations = grant.get("allowed_operations")
+    if not isinstance(operations, list) or operation not in operations:
+        raise AuthorizationError("writer_token_operation_rejected")
+    expected_branch = str(grant.get("allowed_branch") or "")
+    configured_branch = os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", "")
+    if expected_branch != configured_branch:
+        raise AuthorizationError("writer_token_branch_rejected")
+    if branch is not None and branch != expected_branch:
+        raise AuthorizationError("writer_token_branch_rejected")
+    if path is not None:
+        paths = grant.get("allowed_paths")
+        if not isinstance(paths, list) or path not in paths:
+            raise AuthorizationError("writer_token_path_rejected")
+    base = str(grant.get("approved_base_oid") or "").lower()
+    if not OID_RE.fullmatch(base):
+        raise AuthorizationError("writer_token_base_rejected")
+    if not _git_is_ancestor(workspace, base, expected_branch):
+        raise AuthorizationError("writer_token_base_not_ancestor")
+
+
+def authorize(
+    registry: TokenRegistry,
+    token: str,
+    *,
+    method: str,
+    path: str,
+    source_ip: str,
+    workspace: Path,
+    branch: str | None = None,
+    target_path: str | None = None,
+    current_time: datetime | None = None,
+) -> AuthorizedGrant:
+    if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
+        raise AuthorizationError("writer_token_policy_v2_not_required")
+    if not token:
+        raise AuthorizationError("unauthorized", 401)
+    grant = registry.find(token)
+    if grant is None:
+        raise AuthorizationError("unauthorized", 401)
+    operation = operation_for(method, path)
+    _validate_grant(
+        grant,
+        audience=os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""),
+        source_ip=source_ip,
+        operation=operation,
+        branch=branch,
+        path=target_path,
+        workspace=workspace,
+        current_time=current_time or datetime.now(timezone.utc),
+    )
+    limits = grant.get("max_uses")
+    if not isinstance(limits, dict):
+        raise AuthorizationError("writer_token_use_limit_missing")
+    try:
+        used = registry.consume(str(grant["token_id"]), operation, int(limits[operation]))
+    except (KeyError, TypeError, ValueError):
+        raise AuthorizationError("writer_token_use_limit_missing") from None
+    except PermissionError as exc:
+        code = str(exc)
+        raise AuthorizationError(code, 401 if code == "writer_token_revoked" else 403) from None
+    return AuthorizedGrant(grant, operation, used)

active-runtime/app/isolated_writer_commit_file.py

@@ -0,0 +1,403 @@
+"""Atomic, idempotent commit-file Saga for the isolated writer."""
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import subprocess
+import tempfile
+import threading
+from pathlib import Path
+from typing import Any
+
+import aios_isolated_writer_policy as policy
+
+
+OID_RE = re.compile(r"[0-9a-f]{40}")
+IDEMPOTENCY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+STATES = frozenset({
+    "PENDING", "FILE_WRITTEN", "COMMITTED", "PUSHED", "VERIFIED", "COMPLETED",
+    "FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED",
+})
+_LOCKS: dict[str, threading.Lock] = {}
+_LOCKS_GUARD = threading.Lock()
+_LEDGER_INIT_LOCK = threading.Lock()
+
+
+class CommitFileError(RuntimeError):
+    def __init__(self, code: str, status: int = 400):
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+class SimulatedCrash(RuntimeError):
+    pass
+
+
+def canonical_digest(data: dict[str, Any]) -> str:
+    payload = {
+        "branch": data["branch"],
+        "path": data["path"],
+        "content_sha256": hashlib.sha256(data["content"].encode("utf-8")).hexdigest(),
+        "message": data["message"],
+        "expected_head_oid": data["expected_head_oid"],
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+class OperationLedger:
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        if self.path.parent.is_symlink() or self.path.is_symlink():
+            raise PermissionError("writer_operation_ledger_symlink_rejected")
+        with _LEDGER_INIT_LOCK:
+            with self.connect() as db:
+                db.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS writer_operations (
+                        operation_id TEXT PRIMARY KEY,
+                        token_id TEXT NOT NULL,
+                        task_id TEXT NOT NULL,
+                        idempotency_key TEXT NOT NULL UNIQUE,
+                        operation_digest TEXT NOT NULL,
+                        branch TEXT NOT NULL,
+                        path TEXT NOT NULL,
+                        content_sha256 TEXT NOT NULL,
+                        pre_head_oid TEXT NOT NULL,
+                        commit_oid TEXT,
+                        push_status TEXT,
+                        verification_status TEXT,
+                        state TEXT NOT NULL,
+                        created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        error_code TEXT
+                    )
+                    """
+                )
+        os.chmod(self.path, 0o600)
+
+    def connect(self) -> sqlite3.Connection:
+        db = sqlite3.connect(self.path, timeout=30)
+        db.row_factory = sqlite3.Row
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        return db
+
+    def get_or_create(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
+        key = data["idempotency_key"]
+        with self.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute(
+                "SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
+            ).fetchone()
+            if row:
+                db.commit()
+                record = dict(row)
+                if record["operation_digest"] != digest:
+                    raise CommitFileError("writer_idempotency_conflict", 409)
+                if (
+                    record["token_id"] != grant.token_id
+                    or record["task_id"] != grant.task_id
+                ):
+                    raise CommitFileError("writer_idempotency_owner_conflict", 409)
+                return record
+            operation_id = hashlib.sha256(
+                f"{grant.task_id}\0{key}".encode("utf-8")
+            ).hexdigest()
+            content_sha = hashlib.sha256(data["content"].encode("utf-8")).hexdigest()
+            db.execute(
+                """
+                INSERT INTO writer_operations(
+                    operation_id,token_id,task_id,idempotency_key,operation_digest,
+                    branch,path,content_sha256,pre_head_oid,state
+                ) VALUES(?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    operation_id, grant.token_id, grant.task_id, key, digest,
+                    data["branch"], data["path"], content_sha,
+                    data["expected_head_oid"], "PENDING",
+                ),
+            )
+            db.commit()
+        return self.get(key)
+
+    def get(self, key: str) -> dict[str, Any]:
+        with self.connect() as db:
+            row = db.execute(
+                "SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
+            ).fetchone()
+        if row is None:
+            raise KeyError(key)
+        return dict(row)
+
+    def update(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
+        if state not in STATES:
+            raise ValueError("invalid writer operation state")
+        allowed = {"commit_oid", "push_status", "verification_status", "error_code"}
+        assignments = ["state=?", "updated_at=CURRENT_TIMESTAMP"]
+        values: list[Any] = [state]
+        for name, value in fields.items():
+            if name not in allowed:
+                raise ValueError("invalid ledger field")
+            assignments.append(f"{name}=?")
+            values.append(value)
+        values.append(key)
+        with self.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            db.execute(
+                f"UPDATE writer_operations SET {','.join(assignments)} WHERE idempotency_key=?",
+                values,
+            )
+            db.commit()
+        return self.get(key)
+
+
+class AtomicCommitFile:
+    def __init__(
+        self,
+        *,
+        workspace: Path,
+        bare_repo: Path,
+        ledger: OperationLedger,
+        remote: str,
+        audit,
+        crash_after_state: str | None = None,
+    ):
+        self.workspace = workspace
+        self.bare_repo = bare_repo
+        self.ledger = ledger
+        self.remote = remote
+        self.audit = audit
+        self.crash_after_state = crash_after_state
+
+    def _git(self, *args: str, check: bool = True) -> str:
+        result = subprocess.run(
+            ["git", *args], cwd=self.workspace, text=True, capture_output=True
+        )
+        if check and result.returncode != 0:
+            raise CommitFileError("writer_git_operation_failed", 500)
+        return result.stdout.strip()
+
+    def _transition(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
+        record = self.ledger.update(key, state, **fields)
+        if self.crash_after_state == state:
+            raise SimulatedCrash(state)
+        return record
+
+    def _validate(self, data: Any) -> dict[str, Any]:
+        if not isinstance(data, dict):
+            raise CommitFileError("writer_request_object_required")
+        required = ("branch", "path", "content", "message", "expected_head_oid", "idempotency_key")
+        if any(field not in data for field in required):
+            raise CommitFileError("writer_request_incomplete")
+        if data.get("force") or data.get("force_push"):
+            raise CommitFileError("isolated_writer_force_rejected", 403)
+        if set(data) != set(required):
+            raise CommitFileError("writer_request_fields_rejected")
+        item = {field: data[field] for field in required}
+        if item["branch"] != policy.allowed_branch():
+            raise CommitFileError("isolated_writer_branch_rejected", 403)
+        try:
+            item["path"] = policy.validate_path(item["path"], self.workspace)
+        except PermissionError as exc:
+            raise CommitFileError(str(exc), 403) from None
+        if not isinstance(item["content"], str) or "\x00" in item["content"]:
+            raise CommitFileError("writer_content_rejected")
+        try:
+            policy.validate_content_for_path(item["path"], item["content"])
+        except PermissionError as exc:
+            raise CommitFileError(str(exc), 403) from None
+        if len(item["content"].encode("utf-8")) > 1_048_576:
+            raise CommitFileError("writer_content_too_large", 413)
+        message = item["message"]
+        if (
+            not isinstance(message, str)
+            or not 1 <= len(message) <= 200
+            or "\x00" in message
+            or "\n" in message
+            or "\r" in message
+        ):
+            raise CommitFileError("writer_message_rejected")
+        if not isinstance(item["expected_head_oid"], str) or not OID_RE.fullmatch(item["expected_head_oid"]):
+            raise CommitFileError("writer_expected_head_rejected")
+        if not isinstance(item["idempotency_key"], str) or not IDEMPOTENCY_RE.fullmatch(item["idempotency_key"]):
+            raise CommitFileError("writer_idempotency_key_rejected")
+        return item
+
+    def _atomic_write(self, relative: str, content: str) -> None:
+        target = self.workspace / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        fd, temporary = tempfile.mkstemp(prefix=".writer-", dir=target.parent)
+        try:
+            with os.fdopen(fd, "wb", closefd=True) as stream:
+                stream.write(content.encode("utf-8"))
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(temporary, target)
+            parent_fd = os.open(target.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+            try:
+                os.fsync(parent_fd)
+            finally:
+                os.close(parent_fd)
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
+
+    def _verify_remote(self, branch: str, oid: str) -> None:
+        bare = subprocess.run(
+            ["git", "--git-dir", str(self.bare_repo), "rev-parse", f"refs/heads/{branch}"],
+            text=True, capture_output=True,
+        )
+        if bare.returncode != 0 or bare.stdout.strip() != oid:
+            raise CommitFileError("writer_authoritative_bare_oid_mismatch", 500)
+        remote = subprocess.run(
+            ["git", "ls-remote", "--heads", self.remote, f"refs/heads/{branch}"],
+            cwd=self.workspace, text=True, capture_output=True,
+        )
+        if remote.returncode != 0:
+            raise CommitFileError("writer_gitea_verification_failed", 503)
+        fields = remote.stdout.strip().split()
+        if len(fields) != 2:
+            raise CommitFileError("writer_gitea_ref_missing", 503)
+        if fields[0] != oid:
+            raise CommitFileError("writer_gitea_oid_conflict", 409)
+
+    def execute(self, grant: Any, raw_data: Any) -> dict[str, Any]:
+        data = self._validate(raw_data)
+        digest = canonical_digest(data)
+        branch = data["branch"]
+        with _LOCKS_GUARD:
+            lock = _LOCKS.setdefault(branch, threading.Lock())
+        lock_path = self.ledger.path.parent / f"{hashlib.sha256(branch.encode()).hexdigest()}.lock"
+        with lock, open(lock_path, "a+b") as branch_lock:
+            fcntl.flock(branch_lock.fileno(), fcntl.LOCK_EX)
+            return self._execute_locked(grant, data, digest)
+
+    def _execute_locked(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
+        key = data["idempotency_key"]
+        record = self.ledger.get_or_create(grant, data, digest)
+        if record["state"] == "COMPLETED":
+            return self._result(record, idempotent=True)
+        if self._git("rev-parse", "--abbrev-ref", "HEAD") != data["branch"]:
+            record = self._transition(
+                key, "MANUAL_REVIEW_REQUIRED", error_code="writer_current_branch_conflict"
+            )
+            self._audit(grant, data, digest, record, "manual_review_required")
+            raise CommitFileError("writer_current_branch_conflict", 409)
+        head = self._git("rev-parse", "HEAD")
+        commit_oid = record.get("commit_oid")
+        if head != record["pre_head_oid"] and head != commit_oid:
+            record = self._transition(
+                key, "MANUAL_REVIEW_REQUIRED", error_code="writer_branch_head_conflict"
+            )
+            self._audit(grant, data, digest, record, "manual_review_required")
+            raise CommitFileError("writer_branch_head_conflict", 409)
+        try:
+            state = record["state"]
+            if state == "MANUAL_REVIEW_REQUIRED":
+                raise CommitFileError("writer_manual_review_required", 409)
+            if state == "FAILED_RECOVERABLE":
+                if record.get("push_status") == "confirmed":
+                    state = "PUSHED"
+                else:
+                    state = "COMMITTED" if commit_oid else "FILE_WRITTEN"
+            if state == "PENDING":
+                self._atomic_write(data["path"], data["content"])
+                record = self._transition(key, "FILE_WRITTEN")
+                state = record["state"]
+            if state == "FILE_WRITTEN":
+                actual_sha = hashlib.sha256(
+                    (self.workspace / data["path"]).read_bytes()
+                ).hexdigest()
+                if actual_sha != record["content_sha256"]:
+                    raise CommitFileError("writer_file_content_conflict", 409)
+                staged_before = self._git("diff", "--cached", "--name-only", check=False).splitlines()
+                if staged_before:
+                    raise CommitFileError("writer_extra_staged_files_rejected", 409)
+                self._git("add", "--", data["path"])
+                staged = self._git("diff", "--cached", "--name-only").splitlines()
+                if staged != [data["path"]]:
+                    raise CommitFileError("writer_extra_staged_files_rejected", 409)
+                self._git("commit", "-m", data["message"])
+                commit_oid = self._git("rev-parse", "HEAD")
+                record = self._transition(key, "COMMITTED", commit_oid=commit_oid)
+                state = record["state"]
+            if state == "COMMITTED":
+                commit_oid = str(record["commit_oid"])
+                pushed = subprocess.run(
+                    ["git", "push", "--porcelain", self.remote, f"HEAD:refs/heads/{data['branch']}"],
+                    cwd=self.workspace, text=True, capture_output=True,
+                )
+                if pushed.returncode != 0:
+                    self._transition(
+                        key, "FAILED_RECOVERABLE",
+                        push_status="failed", error_code="writer_gitea_push_failed",
+                    )
+                    raise CommitFileError("writer_gitea_push_failed", 503)
+                record = self._transition(key, "PUSHED", push_status="confirmed")
+                state = record["state"]
+            if state == "PUSHED":
+                commit_oid = str(record["commit_oid"])
+                self._verify_remote(data["branch"], commit_oid)
+                record = self._transition(
+                    key, "VERIFIED", verification_status="bare+gitea"
+                )
+                state = record["state"]
+            if state == "VERIFIED":
+                record = self._transition(key, "COMPLETED", error_code=None)
+            self._audit(grant, data, digest, record, "completed")
+            return self._result(record, idempotent=False)
+        except SimulatedCrash:
+            raise
+        except CommitFileError as exc:
+            latest = self.ledger.get(key)
+            if latest["state"] not in {"FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED"}:
+                failure_state = (
+                    "MANUAL_REVIEW_REQUIRED"
+                    if exc.code in {
+                        "writer_gitea_oid_conflict",
+                        "writer_file_content_conflict",
+                        "writer_current_branch_conflict",
+                    }
+                    else "FAILED_RECOVERABLE"
+                )
+                latest = self._transition(key, failure_state, error_code=exc.code)
+            self._audit(grant, data, digest, latest, "failed")
+            raise
+
+    def _audit(self, grant: Any, data: dict[str, Any], digest: str, record: dict[str, Any], result: str) -> None:
+        self.audit({
+            "token_id": grant.token_id,
+            "task_id": grant.task_id,
+            "audience": grant.record["audience"],
+            "source_ip": grant.record.get("_source_ip", "socket-peer"),
+            "operation": grant.operation,
+            "branch": data["branch"],
+            "path": data["path"],
+            "idempotency_key": data["idempotency_key"],
+            "operation_digest": digest,
+            "path_policy_sha256": policy.configured_path_policy_sha256(),
+            "pre_head_oid": record["pre_head_oid"],
+            "commit_oid": record.get("commit_oid"),
+            "result": result,
+        })
+
+    @staticmethod
+    def _result(record: dict[str, Any], *, idempotent: bool) -> dict[str, Any]:
+        return {
+            "status": "completed",
+            "state": record["state"],
+            "operation_id": record["operation_id"],
+            "branch": record["branch"],
+            "path": record["path"],
+            "commit_oid": record.get("commit_oid"),
+            "push_status": record.get("push_status"),
+            "verification_status": record.get("verification_status"),
+            "idempotent": idempotent,
+        }

active-runtime/app/isolated_writer_path_policy.py

@@ -0,0 +1,422 @@
+"""Strict service-level exact-path policy for isolated Git writers."""
+from __future__ import annotations
+
+import grp
+import hashlib
+import json
+import os
+import re
+import stat
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path, PurePosixPath, PureWindowsPath
+from typing import Any
+from urllib.parse import unquote
+
+
+MAX_POLICY_BYTES = 65_536
+POLICY_FIELDS = frozenset(
+    {
+        "schema_version",
+        "policy_id",
+        "audience",
+        "repository_identity",
+        "allowed_branch",
+        "allowed_operations",
+    }
+)
+OPERATION_FIELDS = frozenset({"exact_paths", "max_content_bytes", "content_profile"})
+SUPPORTED_OPERATIONS = frozenset({"commit_file.atomic"})
+SUPPORTED_CONTENT_PROFILES = frozenset({"systemd-codex-review-worker-v1"})
+_SECRET_RE = re.compile(
+    r"(?i)(authorization\s*:|bearer\s+|(?:token|password|secret)\s*[:=])"
+)
+_SYSTEMD_NAME_RE = re.compile(r"[A-Za-z][A-Za-z0-9]*\Z")
+_SYSTEMD_MAX_LINE_BYTES = 4096
+_SYSTEMD_ALLOWED_DIRECTIVES = {
+    "Unit": frozenset({"Description", "After", "Wants"}),
+    "Service": frozenset(
+        {
+            "Type",
+            "User",
+            "Group",
+            "WorkingDirectory",
+            "Environment",
+            "ExecStart",
+            "Restart",
+            "RestartSec",
+            "NoNewPrivileges",
+            "PrivateTmp",
+            "PrivateDevices",
+            "ProtectSystem",
+            "ProtectHome",
+            "ProtectKernelTunables",
+            "ProtectKernelModules",
+            "ProtectControlGroups",
+            "LockPersonality",
+            "RestrictSUIDSGID",
+            "RestrictRealtime",
+            "MemoryDenyWriteExecute",
+            "RuntimeDirectory",
+            "StateDirectory",
+            "CacheDirectory",
+            "ReadOnlyPaths",
+            "ReadWritePaths",
+            "UMask",
+        }
+    ),
+    "Install": frozenset({"WantedBy"}),
+}
+_SYSTEMD_MULTI_DIRECTIVES = frozenset(
+    {
+        ("Service", "Environment"),
+        ("Service", "ReadOnlyPaths"),
+        ("Service", "ReadWritePaths"),
+    }
+)
+_SYSTEMD_EXACT_SINGLE_VALUES = {
+    ("Unit", "Description"): "AIOS Codex Review/Test Worker",
+    ("Unit", "After"): "network-online.target",
+    ("Unit", "Wants"): "network-online.target",
+    ("Service", "Type"): "simple",
+    ("Service", "User"): "aiman",
+    ("Service", "Group"): "gitops",
+    (
+        "Service",
+        "WorkingDirectory",
+    ): "/usr/libexec/aios-codex-review-test-worker/current",
+    (
+        "Service",
+        "ExecStart",
+    ): "/usr/bin/python3 -m task_plane.workers.codex_review_test",
+    ("Service", "Restart"): "on-failure",
+    ("Service", "RestartSec"): "2",
+    ("Service", "NoNewPrivileges"): "true",
+    ("Service", "PrivateTmp"): "true",
+    ("Service", "PrivateDevices"): "true",
+    ("Service", "ProtectSystem"): "strict",
+    ("Service", "ProtectHome"): "true",
+    ("Service", "ProtectKernelTunables"): "true",
+    ("Service", "ProtectKernelModules"): "true",
+    ("Service", "ProtectControlGroups"): "true",
+    ("Service", "LockPersonality"): "true",
+    ("Service", "RestrictSUIDSGID"): "true",
+    ("Service", "RestrictRealtime"): "true",
+    ("Service", "MemoryDenyWriteExecute"): "true",
+    ("Service", "RuntimeDirectory"): "aios-codex-review-test-worker",
+    ("Service", "StateDirectory"): "aios-codex-review-test-worker",
+    ("Service", "CacheDirectory"): "aios-codex-review-test-worker",
+    ("Service", "UMask"): "0077",
+    ("Install", "WantedBy"): "multi-user.target",
+}
+_SYSTEMD_EXACT_MULTI_VALUES = {
+    ("Service", "Environment"): (
+        "AIOS_CODEX_WORKER_MODE=shadow",
+        "AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
+        "PYTHONDONTWRITEBYTECODE=1",
+    ),
+    ("Service", "ReadOnlyPaths"): (
+        "/data/aios-gitops/repo.git",
+        "/usr/libexec/aios-codex-review-test-worker",
+    ),
+    ("Service", "ReadWritePaths"): (
+        "/run/aios-codex-review-test-worker",
+        "/var/lib/aios-codex-review-test-worker",
+        "/var/cache/aios-codex-review-test-worker",
+    ),
+}
+
+
+class PathPolicyError(PermissionError):
+    pass
+
+
+@dataclass(frozen=True)
+class OperationPolicy:
+    exact_paths: tuple[str, ...]
+    max_content_bytes: int
+    content_profile: str
+
+
+@dataclass(frozen=True)
+class ServicePathPolicy:
+    schema_version: int
+    policy_id: str
+    audience: str
+    repository_identity: str
+    allowed_branch: str
+    allowed_operations: dict[str, OperationPolicy]
+    sha256: str
+
+    def operation(self, name: str) -> OperationPolicy:
+        try:
+            return self.allowed_operations[name]
+        except KeyError:
+            raise PathPolicyError("isolated_writer_service_operation_rejected") from None
+
+    def require_exact_path(self, operation: str, path: str) -> OperationPolicy:
+        selected = self.operation(operation)
+        if path not in selected.exact_paths:
+            raise PathPolicyError("isolated_writer_service_path_rejected")
+        return selected
+
+
+def _canonical_json(value: Any) -> bytes:
+    return (
+        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        + "\n"
+    ).encode("utf-8")
+
+
+def validate_relative_path_syntax(path: str) -> str:
+    if not isinstance(path, str):
+        raise PathPolicyError("isolated_writer_path_rejected")
+    if not path or path != path.strip() or "\x00" in path or "\\" in path:
+        raise PathPolicyError("isolated_writer_path_rejected")
+    try:
+        decoded = unquote(path, errors="strict")
+    except (UnicodeDecodeError, ValueError):
+        raise PathPolicyError("isolated_writer_path_rejected") from None
+    if decoded != path:
+        raise PathPolicyError("isolated_writer_path_rejected")
+    posix = PurePosixPath(path)
+    windows = PureWindowsPath(path)
+    parts = path.split("/")
+    if (
+        posix.is_absolute()
+        or windows.is_absolute()
+        or bool(windows.drive)
+        or any(part in {"", ".", ".."} for part in parts)
+    ):
+        raise PathPolicyError("isolated_writer_path_rejected")
+    return path
+
+
+def _require_fields(value: dict[str, Any], allowed: frozenset[str], code: str) -> None:
+    if set(value) != allowed:
+        raise PathPolicyError(code)
+
+
+def _parse_policy(document: Any, canonical: bytes) -> ServicePathPolicy:
+    if not isinstance(document, dict):
+        raise PathPolicyError("isolated_writer_path_policy_object_required")
+    _require_fields(document, POLICY_FIELDS, "isolated_writer_path_policy_fields_rejected")
+    if document["schema_version"] != 1:
+        raise PathPolicyError("isolated_writer_path_policy_schema_rejected")
+    text_fields = (
+        "policy_id",
+        "audience",
+        "repository_identity",
+        "allowed_branch",
+    )
+    if any(
+        not isinstance(document[field], str) or not document[field].strip()
+        for field in text_fields
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_value_rejected")
+    raw_operations = document["allowed_operations"]
+    if not isinstance(raw_operations, dict) or not raw_operations:
+        raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
+    if not set(raw_operations).issubset(SUPPORTED_OPERATIONS):
+        raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
+    operations: dict[str, OperationPolicy] = {}
+    for name, raw in raw_operations.items():
+        if not isinstance(raw, dict):
+            raise PathPolicyError("isolated_writer_path_policy_operation_rejected")
+        _require_fields(
+            raw,
+            OPERATION_FIELDS,
+            "isolated_writer_path_policy_operation_fields_rejected",
+        )
+        paths = raw["exact_paths"]
+        if (
+            not isinstance(paths, list)
+            or not paths
+            or any(not isinstance(item, str) for item in paths)
+        ):
+            raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
+        validated = tuple(validate_relative_path_syntax(item) for item in paths)
+        if len(set(validated)) != len(validated):
+            raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
+        maximum = raw["max_content_bytes"]
+        if not isinstance(maximum, int) or isinstance(maximum, bool) or not 1 <= maximum <= 1_048_576:
+            raise PathPolicyError("isolated_writer_path_policy_size_rejected")
+        profile = raw["content_profile"]
+        if profile not in SUPPORTED_CONTENT_PROFILES:
+            raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
+        operations[name] = OperationPolicy(validated, maximum, profile)
+    return ServicePathPolicy(
+        schema_version=1,
+        policy_id=document["policy_id"],
+        audience=document["audience"],
+        repository_identity=document["repository_identity"],
+        allowed_branch=document["allowed_branch"],
+        allowed_operations=operations,
+        sha256=hashlib.sha256(canonical).hexdigest(),
+    )
+
+
+def load_path_policy(
+    path: Path,
+    *,
+    expected_owner_uid: int = 0,
+    expected_group_gid: int | None = None,
+    expected_parent_mode: int = 0o750,
+    expected_file_mode: int = 0o640,
+) -> ServicePathPolicy:
+    lexical = Path(os.path.abspath(str(path)))
+    if lexical.is_symlink():
+        raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
+    try:
+        parent_info = lexical.parent.stat(follow_symlinks=False)
+        file_info = lexical.stat(follow_symlinks=False)
+    except FileNotFoundError:
+        raise PathPolicyError("isolated_writer_path_policy_missing") from None
+    if stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode):
+        raise PathPolicyError("isolated_writer_path_policy_parent_rejected")
+    if stat.S_ISLNK(file_info.st_mode) or not stat.S_ISREG(file_info.st_mode):
+        raise PathPolicyError("isolated_writer_path_policy_file_rejected")
+    if lexical.resolve(strict=True) != lexical:
+        raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
+    if (
+        parent_info.st_uid != expected_owner_uid
+        or stat.S_IMODE(parent_info.st_mode) != expected_parent_mode
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_parent_permissions_rejected")
+    if (
+        file_info.st_uid != expected_owner_uid
+        or stat.S_IMODE(file_info.st_mode) != expected_file_mode
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_permissions_rejected")
+    if expected_group_gid is not None and (
+        parent_info.st_gid != expected_group_gid or file_info.st_gid != expected_group_gid
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_group_rejected")
+    if file_info.st_size <= 0 or file_info.st_size > MAX_POLICY_BYTES:
+        raise PathPolicyError("isolated_writer_path_policy_file_size_rejected")
+    raw = lexical.read_bytes()
+    try:
+        document = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise PathPolicyError("isolated_writer_path_policy_json_rejected") from None
+    canonical = _canonical_json(document)
+    if raw != canonical:
+        raise PathPolicyError("isolated_writer_path_policy_not_canonical")
+    return _parse_policy(document, canonical)
+
+
+@lru_cache(maxsize=1)
+def configured_path_policy() -> ServicePathPolicy | None:
+    configured = os.getenv("AIOS_GITOPS_PATH_POLICY_FILE", "").strip()
+    if not configured:
+        return None
+    if not Path(configured).is_absolute():
+        raise PathPolicyError("isolated_writer_path_policy_path_rejected")
+    try:
+        gitops_gid = grp.getgrnam("gitops").gr_gid
+    except KeyError:
+        raise PathPolicyError("isolated_writer_path_policy_group_rejected") from None
+    loaded = load_path_policy(Path(configured), expected_group_gid=gitops_gid)
+    if loaded.audience != os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""):
+        raise PathPolicyError("isolated_writer_path_policy_audience_rejected")
+    if loaded.repository_identity != os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", ""):
+        raise PathPolicyError("isolated_writer_path_policy_repository_rejected")
+    if loaded.allowed_branch != os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", ""):
+        raise PathPolicyError("isolated_writer_path_policy_branch_rejected")
+    return loaded
+
+
+def clear_path_policy_cache() -> None:
+    configured_path_policy.cache_clear()
+
+
+def require_service_path(operation: str, path: str) -> OperationPolicy:
+    loaded = configured_path_policy()
+    if loaded is None:
+        raise PathPolicyError("isolated_writer_service_path_policy_required")
+    return loaded.require_exact_path(operation, path)
+
+
+def configured_path_policy_sha256() -> str | None:
+    loaded = configured_path_policy()
+    return None if loaded is None else loaded.sha256
+
+
+def validate_content(operation: str, path: str, content: str) -> None:
+    selected = require_service_path(operation, path)
+    try:
+        encoded = content.encode("utf-8", errors="strict")
+    except (AttributeError, UnicodeEncodeError):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
+    if len(encoded) > selected.max_content_bytes:
+        raise PathPolicyError("isolated_writer_service_content_too_large")
+    if selected.content_profile == "systemd-codex-review-worker-v1":
+        _validate_systemd_codex_review_worker(content)
+        return
+    raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
+
+
+def _validate_systemd_codex_review_worker(content: str) -> None:
+    if not isinstance(content, str):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    try:
+        content.encode("utf-8", errors="strict")
+    except UnicodeEncodeError:
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
+    if (
+        "\x00" in content
+        or "\r" in content
+        or _SECRET_RE.search(content)
+        or not content.endswith("\n")
+    ):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+
+    sections: set[str] = set()
+    values: dict[tuple[str, str], list[str]] = {}
+    section: str | None = None
+    for line in content.splitlines():
+        if len(line.encode("utf-8")) > _SYSTEMD_MAX_LINE_BYTES:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        if not line:
+            continue
+        if line[0].isspace() or line.startswith(("#", ";")):
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        if line.startswith("[") and line.endswith("]"):
+            candidate = line[1:-1]
+            if (
+                not _SYSTEMD_NAME_RE.fullmatch(candidate)
+                or candidate not in _SYSTEMD_ALLOWED_DIRECTIVES
+                or candidate in sections
+            ):
+                raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+            sections.add(candidate)
+            section = candidate
+            continue
+        if section is None or "=" not in line:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        directive, value = line.split("=", 1)
+        if (
+            not _SYSTEMD_NAME_RE.fullmatch(directive)
+            or directive not in _SYSTEMD_ALLOWED_DIRECTIVES[section]
+            or not value
+            or value != value.strip()
+        ):
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        key = (section, directive)
+        if key in values and key not in _SYSTEMD_MULTI_DIRECTIVES:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        values.setdefault(key, []).append(value)
+
+    if sections != set(_SYSTEMD_ALLOWED_DIRECTIVES):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    if set(values) != set(_SYSTEMD_EXACT_SINGLE_VALUES) | set(
+        _SYSTEMD_EXACT_MULTI_VALUES
+    ):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    for key, expected in _SYSTEMD_EXACT_SINGLE_VALUES.items():
+        if values.get(key) != [expected]:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    for key, expected in _SYSTEMD_EXACT_MULTI_VALUES.items():
+        observed = values.get(key, [])
+        if len(observed) != len(set(observed)) or tuple(observed) != expected:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")

active-runtime/app/isolated_writer_token_registry.py

@@ -0,0 +1,241 @@
+"""Digest-only, task-scoped token registry for isolated writers."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+import grp
+import pwd
+import secrets
+import sqlite3
+import stat
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+_REGISTRY_INIT_LOCK = threading.Lock()
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def token_digest(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+class TokenRegistry:
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self._prepare()
+
+    def _prepare(self) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        if self.path.parent.is_symlink() or self.path.is_symlink():
+            raise PermissionError("writer_token_registry_symlink_rejected")
+        with _REGISTRY_INIT_LOCK:
+            with self._connect() as db:
+                db.executescript(
+                    """
+                    CREATE TABLE IF NOT EXISTS token_grants (
+                        token_id TEXT PRIMARY KEY,
+                        token_digest TEXT NOT NULL UNIQUE,
+                        grant_json TEXT NOT NULL,
+                        revoked INTEGER NOT NULL DEFAULT 0,
+                        created_at TEXT NOT NULL
+                    );
+                    CREATE TABLE IF NOT EXISTS token_uses (
+                        token_id TEXT NOT NULL,
+                        operation TEXT NOT NULL,
+                        use_count INTEGER NOT NULL DEFAULT 0,
+                        PRIMARY KEY(token_id, operation),
+                        FOREIGN KEY(token_id) REFERENCES token_grants(token_id)
+                    );
+                    """
+                )
+        os.chmod(self.path, 0o600)
+
+    def _connect(self) -> sqlite3.Connection:
+        db = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+        db.row_factory = sqlite3.Row
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        db.execute("PRAGMA foreign_keys=ON")
+        return db
+
+    def issue(self, grant: dict[str, Any], token: str | None = None) -> tuple[str, dict[str, Any]]:
+        value = token or secrets.token_urlsafe(32)
+        if len(value.encode("utf-8")) < 32:
+            raise ValueError("writer_token_entropy_insufficient")
+        record = dict(grant)
+        record["schema_version"] = 2
+        record["token_digest"] = token_digest(value)
+        record["revoked"] = False
+        record.setdefault("created_at", utc_now())
+        required = (
+            "token_id",
+            "audience",
+            "task_id",
+            "source_cidrs",
+            "allowed_branch",
+            "approved_base_oid",
+            "allowed_operations",
+            "allowed_paths",
+            "not_before",
+            "expires_at",
+            "max_uses",
+            "issued_by",
+        )
+        if any(not record.get(field) for field in required):
+            raise ValueError("writer_token_grant_incomplete")
+        encoded = json.dumps(record, sort_keys=True, separators=(",", ":"))
+        with self._connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            try:
+                db.execute(
+                    """
+                    INSERT INTO token_grants(token_id, token_digest, grant_json, revoked, created_at)
+                    VALUES(?,?,?,?,?)
+                    """,
+                    (
+                        str(record["token_id"]),
+                        str(record["token_digest"]),
+                        encoded,
+                        0,
+                        str(record["created_at"]),
+                    ),
+                )
+                db.execute("COMMIT")
+            except Exception:
+                db.execute("ROLLBACK")
+                raise
+        return value, record
+
+    def find(self, token: str) -> dict[str, Any] | None:
+        candidate = token_digest(token)
+        with self._connect() as db:
+            rows = db.execute(
+                "SELECT token_digest, grant_json, revoked FROM token_grants"
+            ).fetchall()
+        match = None
+        for row in rows:
+            if hmac.compare_digest(str(row["token_digest"]), candidate):
+                match = dict(json.loads(str(row["grant_json"])))
+                match["revoked"] = bool(row["revoked"])
+        return match
+
+    def consume(self, token_id: str, operation: str, maximum: int) -> int:
+        if maximum < 1:
+            raise PermissionError("writer_token_operation_exhausted")
+        with self._connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            try:
+                row = db.execute(
+                    "SELECT revoked FROM token_grants WHERE token_id=?",
+                    (token_id,),
+                ).fetchone()
+                if row is None or bool(row["revoked"]):
+                    raise PermissionError("writer_token_revoked")
+                used_row = db.execute(
+                    "SELECT use_count FROM token_uses WHERE token_id=? AND operation=?",
+                    (token_id, operation),
+                ).fetchone()
+                used = int(used_row["use_count"]) if used_row else 0
+                if used >= maximum:
+                    raise PermissionError("writer_token_operation_exhausted")
+                current = used + 1
+                db.execute(
+                    """
+                    INSERT INTO token_uses(token_id, operation, use_count)
+                    VALUES(?,?,?)
+                    ON CONFLICT(token_id, operation)
+                    DO UPDATE SET use_count=excluded.use_count
+                    """,
+                    (token_id, operation, current),
+                )
+                db.execute("COMMIT")
+                return current
+            except Exception:
+                db.execute("ROLLBACK")
+                raise
+
+    def revoke(self, token_id: str) -> bool:
+        with self._connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            changed = db.execute(
+                "UPDATE token_grants SET revoked=1 WHERE token_id=? AND revoked=0",
+                (token_id,),
+            ).rowcount
+            db.execute("COMMIT")
+        return bool(changed)
+
+    def usage(self, token_id: str, operation: str) -> int:
+        with self._connect() as db:
+            row = db.execute(
+                "SELECT use_count FROM token_uses WHERE token_id=? AND operation=?",
+                (token_id, operation),
+            ).fetchone()
+        return int(row["use_count"]) if row else 0
+
+
+def _write_secret_file(
+    path: Path,
+    token: str,
+    *,
+    owner: str | None = None,
+    group: str | None = None,
+) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise PermissionError("writer_token_file_symlink_rejected")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    fd = os.open(path, flags, 0o600)
+    try:
+        if not stat.S_ISREG(os.fstat(fd).st_mode):
+            raise PermissionError("writer_token_file_not_regular")
+        if owner is not None or group is not None:
+            uid = pwd.getpwnam(owner).pw_uid if owner is not None else -1
+            gid = grp.getgrnam(group).gr_gid if group is not None else -1
+            os.fchown(fd, uid, gid)
+        os.write(fd, token.encode("utf-8"))
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--registry", required=True)
+    sub = parser.add_subparsers(dest="command", required=True)
+    issue = sub.add_parser("issue")
+    issue.add_argument("--grant-file", required=True)
+    issue.add_argument("--token-file", required=True)
+    issue.add_argument("--owner")
+    issue.add_argument("--group")
+    revoke = sub.add_parser("revoke")
+    revoke.add_argument("--token-id", required=True)
+    args = parser.parse_args(argv)
+    registry = TokenRegistry(args.registry)
+    if args.command == "issue":
+        grant = json.loads(Path(args.grant_file).read_text(encoding="utf-8"))
+        token, record = registry.issue(grant)
+        _write_secret_file(
+            Path(args.token_file),
+            token,
+            owner=args.owner,
+            group=args.group,
+        )
+        print(json.dumps({"token_id": record["token_id"], "created": True}, sort_keys=True))
+        return 0
+    changed = registry.revoke(args.token_id)
+    print(json.dumps({"token_id": args.token_id, "revoked": changed}, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

deploy/systemd/aios-governance-machine-status-producer.service

@@ -0,0 +1,26 @@
+[Unit]
+Description=AIOS dormant single-owner machine-status prerequisite
+RefuseManualStop=no
+
+[Service]
+Type=oneshot
+User=aiman
+Group=gitops
+Environment=AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false
+Environment=AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+ExecStart=/usr/bin/python3 -m aios_single_owner_governance.service --component machine-status-producer --self-check --checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+RestrictSUIDSGID=true
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadWritePaths=/var/lib/aios-gitops/runtime/single-owner-governance

deploy/systemd/aios-governance-owner-review-adapter.service

@@ -0,0 +1,26 @@
+[Unit]
+Description=AIOS dormant single-owner review adapter prerequisite
+RefuseManualStop=no
+
+[Service]
+Type=oneshot
+User=aiman
+Group=gitops
+Environment=AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false
+Environment=AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+ExecStart=/usr/bin/python3 -m aios_single_owner_governance.service --component owner-review-adapter --self-check --checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+RestrictSUIDSGID=true
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadWritePaths=/var/lib/aios-gitops/runtime/single-owner-governance

deploy/systemd/aios-governance-single-owner-activation.service

@@ -0,0 +1,20 @@
+[Unit]
+Description=AIOS single-owner governance activation verifier
+ConditionPathExists=/etc/aios-gitops/single-owner-activation-bindings.json
+
+[Service]
+Type=oneshot
+User=root
+Group=gitops
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+ExecStart=/usr/bin/aios-single-owner-governance --verify --bindings /etc/aios-gitops/single-owner-activation-bindings.json --state-root /var/lib/aios-gitops/governance-activation --lock-path /run/aios-gitops/governance-activation.lock
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectHome=true
+ProtectSystem=strict
+ReadWritePaths=/var/lib/aios-gitops/governance-activation /run/aios-gitops
+RestrictSUIDSGID=true
+UMask=0027
+
+# Deliberately not installable. Deployment remains disabled/inactive;
+# activation is an explicit approved CLI transaction, never an always-on unit.

deploy/tmpfiles.d/aios-single-owner-governance.conf

@@ -0,0 +1,4 @@
+d /var/lib/aios-gitops/governance-activation 0750 root gitops -
+d /var/lib/aios-gitops/governance-activation/wal 0750 root gitops -
+d /var/lib/aios-gitops/governance-activation/receipts 0750 root gitops -
+d /run/aios-gitops 0750 root gitops -

docs/gitops/isolated-writer-path-policy-v1.md

@@ -0,0 +1,68 @@
+# Isolated Writer Exact-Path Policy V1
+
+This policy is an additional, service-owned authorization layer for
+`POST /api/commit-file`. It never expands a Token Policy V2 grant.
+
+An atomic write is permitted only when all of the following agree:
+
+1. socket peer, audience, operation, branch, base and exact path in the
+   Token Policy V2 grant;
+2. service audience, repository identity, branch, operation and exact path
+   in `AIOS_GITOPS_PATH_POLICY_FILE`;
+3. lexical path, resolved containment and symlink checks;
+4. the configured content-size and content-profile checks;
+5. operation digest, expected branch HEAD and durable operation ledger.
+
+Any failed layer returns a fail-closed policy error. Coarse `/api/file` and
+`/api/commit` routes remain forbidden.
+
+## File contract
+
+The configured path must be absolute. The parent directory must be
+`root:gitops 0750`; the regular, non-symlink file must be
+`root:gitops 0640`. JSON must use UTF-8 canonical encoding with sorted keys,
+compact separators and one trailing newline. Unknown fields, unsupported
+operations or profiles, duplicate paths, oversized policies and
+group/world-writable files are rejected during Writer startup.
+
+Example:
+
+```json
+{"allowed_branch":"patch/codex-review-test-worker-20260724","allowed_operations":{"commit_file.atomic":{"content_profile":"systemd-codex-review-worker-v1","exact_paths":["deploy/systemd/aios-codex-review-test-worker.service"],"max_content_bytes":32768}},"audience":"aios-codex-review-worker-writer.service","policy_id":"codex-review-worker-path-policy-v1","repository_identity":"aios/aios","schema_version":1}
+```
+
+The SHA-256 of the canonical policy is exposed in the isolated runtime
+status and written to the operation audit record. The request body cannot
+select or override the policy file.
+
+## Content profile
+
+`systemd-codex-review-worker-v1` is an exact, fail-closed unit grammar. The
+only sections are `[Unit]`, `[Service]` and `[Install]`. Each section has a
+directive allowlist; every single-value directive and every repeated
+`Environment`, `ReadOnlyPaths` and `ReadWritePaths` entry must match the
+governed Shadow unit contract exactly.
+
+The profile fixes the Python module
+`task_plane.workers.codex_review_test`, `User=aiman`, `Group=gitops`,
+the controlled `/usr/libexec/aios-codex-review-test-worker/current`
+working directory, all hardening switches and the three permitted
+environment variables. It grants read access only to the authoritative
+bare repository and the versioned worker artifact, and write access only
+to the worker runtime, state and cache directories.
+
+Unknown sections or directives are rejected, including lifecycle hooks,
+credential injection, environment imports, root images, bind mounts,
+capability expansion, devices, sockets and top-level writable paths. The
+parser also rejects duplicate sections, duplicate single-value directives,
+empty resets, continuation lines, CRLF, NUL, non-UTF-8 surrogate content,
+overlong lines and Unicode lookalike directive names. It does not use
+`ConfigParser(strict=False)` or silently merge repeated values.
+
+Production credentials, if later authorized, must be supplied by a
+separately governed deployment drop-in. They are not permitted in the
+source unit or this content profile.
+
+Policy changes require a separate maintenance branch and Owner-reviewed
+ordinary merge commit. A task-scoped Writer must not edit its own policy
+file.

docs/gitops/openapi-chatgpt-actions.yaml

@@ -51,6 +51,31 @@ components:
         path: {type: string, example: docs/gitops/chatgpt-change.md}
         content: {type: string}
         message: {type: string}
+    AtomicFileCommitRequest:
+      type: object
+      additionalProperties: false
+      required: [branch, path, content, message, expected_head_oid, idempotency_key]
+      properties:
+        branch:
+          type: string
+          pattern: '^patch/[A-Za-z0-9._/-]+$'
+        path:
+          type: string
+          minLength: 1
+          maxLength: 512
+        content:
+          type: string
+          maxLength: 1048576
+        message:
+          type: string
+          minLength: 1
+          maxLength: 200
+        expected_head_oid:
+          type: string
+          pattern: '^[0-9a-f]{40}$'
+        idempotency_key:
+          type: string
+          pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$'
     CommitRequest:
       type: object
       properties:
@@ -428,6 +453,25 @@ paths:
             schema: {$ref: '#/components/schemas/FileCommitRequest'}
       responses: {'200': {description: File committed}, '400': {description: Secret rejected or invalid path}, '401': {description: Unauthorized}}
       x-openai-isConsequential: true
+  /api/commit-file:
+    post:
+      operationId: commitFileAtomic
+      summary: Atomically write one allowlisted file, commit, push, and verify using Token Policy V2
+      security: [{bearerAuth: []}]
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema: {$ref: '#/components/schemas/AtomicFileCommitRequest'}
+      responses:
+        '200': {description: Completed or idempotently replayed}
+        '400': {description: Invalid canonical request}
+        '401': {description: Missing, invalid, expired, or revoked Token Policy V2 grant}
+        '403': {description: Audience, source, branch, operation, path, or policy denied}
+        '409': {description: Idempotency conflict or manual review required}
+        '413': {description: Request too large}
+        '503': {description: Push or remote verification unavailable; recoverable}
+      x-openai-isConsequential: true
   /api/commit:
     post:
       operationId: commitChanges

evidence/single-owner-governance/AIOS-SINGLE-OWNER-LIVE-ACTIVATION-HARDENING-20260729T043010Z.json

@@ -0,0 +1,25 @@
+{
+  "artifact_path": "/var/lib/aios-gitops/runtime/tasks/AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729/artifacts/hardening-71def340e7cc34f4f19918dcf8d3c847abbe4fbf/live-activation-71def340e7cc34f4f19918dcf8d3c847abbe4fbf.tar.gz",
+  "artifact_sha256": "32bc49c207ec454b6fd81b0484f5b257bb789481b37175995029a2ec78cc62fa",
+  "copied_live_db_acceptance": "PASS",
+  "implementation_commit_oid": "71def340e7cc34f4f19918dcf8d3c847abbe4fbf",
+  "implementation_tree_oid": "3cc61bfbf69cbe0a24d54a0e53a6c50bcc130066",
+  "independent_validator": "ALLOW_CANDIDATE",
+  "production_mutation": false,
+  "result": "PASS",
+  "rollback_artifact_path": "/var/lib/aios-gitops/runtime/tasks/AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729/artifacts/hardening-71def340e7cc34f4f19918dcf8d3c847abbe4fbf/live-activation-rollback-71def340e7cc34f4f19918dcf8d3c847abbe4fbf.tar.gz",
+  "rollback_artifact_sha256": "4efe1ea73d4fd72e5364fc91a16ad5e9d1e6ebe78d8aa466f7b11c4915b08403",
+  "runtime_source_sha256": {
+    "services/aios-single-owner-governance/aios_single_owner_governance/activation.py": "48846a58fa3afb2b7c8ffe6e88504a9333863932df867f33ba0321ad3bac6003",
+    "services/aios-single-owner-governance/aios_single_owner_governance/gitea_protected_branch.py": "00775efe3f3d631d88379eff4ed2dcfaf386a2fa43111abe961e75109e80d768",
+    "services/aios-single-owner-governance/aios_single_owner_governance/journal.py": "a8f121245e04487d99c99fb03ada83f0c907723f8eb18552872ff46dcf5a7873",
+    "services/aios-single-owner-governance/aios_single_owner_governance/state_machine.py": "0b08a2c378d143cb9031a25132afc7b7d825bdc03d11efe10610341d4b70167e",
+    "services/aios-single-owner-governance/aios_single_owner_governance/validator.py": "754639846cb15cb997d612a1fe0ceb0224ddef5bbc3ccac9802a3747f122dac1"
+  },
+  "subtests_passed": 17,
+  "task_id": "AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729",
+  "test_evidence_path": "/var/lib/aios-gitops/runtime/tasks/AIOS-SINGLE-OWNER-GOVERNANCE-ACTIVATION-20260729/artifacts/hardening-71def340e7cc34f4f19918dcf8d3c847abbe4fbf/test-evidence-71def340e7cc34f4f19918dcf8d3c847abbe4fbf.json",
+  "test_evidence_sha256": "c1633732767f3f051ffb01404d12d05a6168a4b01b91efdb34a64988d6dee684",
+  "tests_failed": 0,
+  "tests_passed": 90
+}

services/aios-single-owner-governance/README.md

@@ -0,0 +1,51 @@
+# AIOS Single-Owner Governance Prerequisites
+
+This package is a dormant candidate produced under the PR #29 prerequisite
+authorization. It validates exact-head Owner approvals and machine gates,
+models protection-policy compare-and-swap with compensation, builds evidence,
+and exercises every mutation path only against disposable fixtures.
+
+The prerequisite shadow services remain static and refuse live mutation. This
+branch adds a separate, one-shot activation executor with `--plan`,
+`--activate`, `--verify`, `--rollback`, and `--status` commands. It is not an
+always-on writer and never enables global GitOps writes.
+
+The executor is fail-closed: every consequential command is bound to an exact
+approval merge, governance OID, evidence closure, Owner review, component
+digests, full branch-protection snapshot, operation ID, and plan ID. It writes
+a fsynced WAL before the single-row SQLite CAS, verifies the full applied row,
+and only rolls back with a target-to-previous CAS. Audit records are
+hash-chained and credentials are redacted.
+
+## Hardening Contract
+
+Fresh activation is authorized only before the approved `expires_at` instant.
+An already-recorded `PREPARED` transaction may recover after expiry only to
+reach a safe terminal state using the same exact bindings; `--verify` and
+`--rollback` remain available for that recorded transaction after expiry.
+`ROLLED_BACK` and `FAILED` are terminal states and must never be replayed as a
+new activation.
+
+The approved execution binding must pin the implementation commit and its
+exact tree, the installed hashes of the five runtime source files
+(`activation.py`, `gitea_protected_branch.py`, `journal.py`, `validator.py`,
+and `state_machine.py`), and independent SHA-256 values for the deployment
+artifact, rollback artifact, and test evidence. It must also bind an evidence
+closure that is a commit with the exact direct parent and exactly one approved
+closure file, including that file's Git blob OID and SHA-256.
+
+Every required machine status is accepted only when its exact status record,
+creator identity, producer provenance, target URL, description digest, head
+OID, state, and observation time match the approved binding. A fixture status
+or a dormant self-check is not machine-status producer provenance.
+
+This contract hardens validation; it does not authorize deployment. The
+existing governed approval rules still apply, and installation remains
+disabled and inactive until the required merged approval authorizes one
+explicit execution.
+
+Installation must remain disabled and inactive until a merged
+`SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION`
+approval authorizes one explicit execution. The approved runtime paths are
+`/var/lib/aios-gitops/governance-activation/` and
+`/run/aios-gitops/governance-activation.lock`.

services/aios-single-owner-governance/activation-bindings.schema.json

@@ -0,0 +1,115 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "additionalProperties": false,
+  "properties": {
+    "activation_approval_merge_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "activation_approval_pr_number": {"minimum": 1, "type": "integer"},
+    "activation_manifest_blob_oid": {"pattern": "^[0-9a-f]{40,64}$", "type": "string"},
+    "activation_manifest_path": {"pattern": "^approvals/.+\\.json$", "type": "string"},
+    "activation_manifest_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "artifact_path": {"pattern": "^/", "type": "string"},
+    "artifact_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "approval_db_path": {"minLength": 1, "type": "string"},
+    "approval_request_head_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "backend": {"enum": ["file", "sqlite"]},
+    "backend_path": {"minLength": 1, "type": "string"},
+    "branch": {"const": "governance/approvals"},
+    "evidence_closure_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "evidence_closure_path": {"pattern": "^(?!/)(?!.*\\.\\.).+", "type": "string"},
+    "evidence_closure_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "evidence_git_dir": {"minLength": 1, "type": "string"},
+    "executor": {"not": {"const": "owner"}, "type": "string"},
+    "expected_branch_protection_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "expected_gitea_schema_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "expected_governance_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "expected_state_machine_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "expected_validator_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "expires_at": {"format": "date-time", "type": "string"},
+    "forbidden_operations": {"additionalProperties": {"const": false}, "type": "object"},
+    "governance_git_dir": {"minLength": 1, "type": "string"},
+    "governance_ref": {"const": "refs/heads/governance/approvals"},
+    "implementation_commit_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "implementation_tree_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "machine_status_producer_id": {"minimum": 1, "type": "integer"},
+    "machine_status_producer_name": {"minLength": 1, "type": "string"},
+    "machine_checks": {
+      "additionalProperties": false,
+      "properties": {
+        "aios/branch-protection": {"$ref": "#/$defs/machineCheck"},
+        "aios/evidence-closure": {"$ref": "#/$defs/machineCheck"},
+        "aios/secret-scan": {"$ref": "#/$defs/machineCheck"},
+        "aios/state-machine": {"$ref": "#/$defs/machineCheck"},
+        "aios/validator": {"$ref": "#/$defs/machineCheck"}
+      },
+      "required": ["aios/branch-protection", "aios/evidence-closure", "aios/secret-scan", "aios/state-machine", "aios/validator"],
+      "type": "object"
+    },
+    "observed_evidence_closure_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "observed_governance_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+    "observed_owner_review_oid": {"type": ["integer", "string"]},
+    "observed_state_machine_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "observed_validator_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "operation_id": {"minLength": 1, "type": "string"},
+    "owner_review_oid": {"type": ["integer", "string"]},
+    "repo_id": {"minimum": 1, "type": "integer"},
+    "requester": {"not": {"const": "owner"}, "type": "string"},
+    "rollback_artifact_path": {"pattern": "^/", "type": "string"},
+    "rollback_artifact_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "runtime_sources": {
+      "items": {
+        "additionalProperties": false,
+        "properties": {
+          "git_path": {"minLength": 1, "type": "string"},
+          "runtime_path": {"pattern": "^/", "type": "string"},
+          "sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"}
+        },
+        "required": ["git_path", "runtime_path", "sha256"],
+        "type": "object"
+      },
+      "maxItems": 5,
+      "minItems": 5,
+      "type": "array"
+    },
+    "state_machine_path": {"minLength": 1, "type": "string"},
+    "target_branch_protection": {"type": "object"},
+    "task_id": {"minLength": 1, "type": "string"},
+    "test_evidence_path": {"pattern": "^/", "type": "string"},
+    "test_evidence_sha256": {"pattern": "^[0-9a-f]{64}$", "type": "string"},
+    "validator_path": {"minLength": 1, "type": "string"},
+    "wip": {"const": false}
+  },
+  "$defs": {
+    "machineCheck": {
+      "additionalProperties": false,
+      "properties": {
+        "head_oid": {"pattern": "^[0-9a-f]{40}$", "type": "string"},
+        "status_id": {"minimum": 1, "type": "integer"},
+        "creator_id": {"minimum": 1, "type": "integer"},
+        "creator_login": {"minLength": 1, "type": "string"},
+        "state": {"const": "success"}
+      },
+      "required": ["state", "head_oid", "status_id", "creator_id", "creator_login"],
+      "type": "object"
+    }
+  },
+  "required": [
+    "task_id", "operation_id", "expected_governance_oid", "observed_governance_oid",
+    "expected_validator_sha256", "observed_validator_sha256", "validator_path",
+    "expected_state_machine_sha256", "observed_state_machine_sha256", "state_machine_path",
+    "expected_branch_protection_sha256", "expected_gitea_schema_sha256",
+    "activation_approval_merge_oid", "activation_approval_pr_number", "approval_request_head_oid",
+    "activation_manifest_path", "activation_manifest_blob_oid", "activation_manifest_sha256",
+    "evidence_closure_oid", "observed_evidence_closure_oid", "evidence_git_dir",
+    "evidence_closure_path", "evidence_closure_sha256",
+    "implementation_commit_oid", "implementation_tree_oid", "runtime_sources",
+    "artifact_path", "artifact_sha256", "rollback_artifact_path", "rollback_artifact_sha256",
+    "test_evidence_path", "test_evidence_sha256",
+    "machine_status_producer_id", "machine_status_producer_name",
+    "owner_review_oid", "observed_owner_review_oid", "approval_db_path", "expires_at",
+    "governance_git_dir", "governance_ref", "machine_checks", "forbidden_operations",
+    "target_branch_protection", "backend", "backend_path", "repo_id", "branch",
+    "requester", "executor", "wip"
+  ],
+  "title": "AIOS Single Owner Live Activation Bindings",
+  "type": "object"
+}

services/aios-single-owner-governance/aios_single_owner_governance/__init__.py

@@ -0,0 +1,20 @@
+"""Dormant single-owner governance activation prerequisites."""
+
+from .models import (
+    ApprovalContext,
+    GovernanceState,
+    MachineGate,
+    ProtectionSnapshot,
+    Review,
+)
+from .validator import SingleOwnerApprovalValidator, ValidationDenied
+
+__all__ = [
+    "ApprovalContext",
+    "GovernanceState",
+    "MachineGate",
+    "ProtectionSnapshot",
+    "Review",
+    "SingleOwnerApprovalValidator",
+    "ValidationDenied",
+]

services/aios-single-owner-governance/aios_single_owner_governance/activation.py

@@ -0,0 +1,860 @@
+from __future__ import annotations
+
+import argparse
+from contextlib import contextmanager
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+import subprocess
+from typing import Any, Iterator, Mapping, Protocol
+
+try:
+    import fcntl
+except ImportError:  # pragma: no cover
+    fcntl = None
+
+from .gitea_protected_branch import (
+    CASConflict,
+    FileProtectedBranchAdapter,
+    SQLiteProtectedBranchAdapter,
+    calculate_snapshot_sha256,
+)
+from .journal import JournalStore, atomic_write_json, read_json, redact, sha256_json
+from . import state_machine as state_machine_module
+from . import validator as validator_module
+
+
+DEFAULT_STATE_ROOT = Path("/var/lib/aios-gitops/governance-activation")
+DEFAULT_LOCK_PATH = Path("/run/aios-gitops/governance-activation.lock")
+LIVE_GITEA_DB = Path("/data/aios-gitops/runtime/gitea-bin/data/gitea.db")
+LIVE_AUTHORITATIVE_GIT = Path("/data/aios-gitops/repo.git")
+HEX40 = re.compile(r"^[0-9a-f]{40}$")
+HEX64 = re.compile(r"^[0-9a-f]{64}$")
+REQUIRED_BINDINGS = {
+    "task_id",
+    "operation_id",
+    "expected_governance_oid",
+    "expected_validator_sha256",
+    "expected_state_machine_sha256",
+    "expected_branch_protection_sha256",
+    "activation_approval_merge_oid",
+    "evidence_closure_oid",
+    "owner_review_oid",
+    "observed_governance_oid",
+    "observed_validator_sha256",
+    "observed_state_machine_sha256",
+    "observed_evidence_closure_oid",
+    "observed_owner_review_oid",
+    "approval_request_head_oid",
+    "activation_approval_pr_number",
+    "expires_at",
+    "validator_path",
+    "state_machine_path",
+    "governance_git_dir",
+    "governance_ref",
+    "evidence_git_dir",
+    "approval_db_path",
+    "expected_gitea_schema_sha256",
+    "backend",
+    "backend_path",
+    "repo_id",
+    "branch",
+    "activation_manifest_path",
+    "activation_manifest_blob_oid",
+    "activation_manifest_sha256",
+    "implementation_commit_oid",
+    "implementation_tree_oid",
+    "runtime_sources",
+    "artifact_path",
+    "artifact_sha256",
+    "rollback_artifact_path",
+    "rollback_artifact_sha256",
+    "test_evidence_path",
+    "test_evidence_sha256",
+    "evidence_closure_path",
+    "evidence_closure_sha256",
+    "machine_status_producer_id",
+    "machine_status_producer_name",
+}
+REQUIRED_MACHINE_CONTEXTS = {
+    "aios/validator",
+    "aios/state-machine",
+    "aios/branch-protection",
+    "aios/secret-scan",
+    "aios/evidence-closure",
+}
+REQUIRED_RUNTIME_GIT_PATHS = {
+    "services/aios-single-owner-governance/aios_single_owner_governance/activation.py",
+    "services/aios-single-owner-governance/aios_single_owner_governance/gitea_protected_branch.py",
+    "services/aios-single-owner-governance/aios_single_owner_governance/journal.py",
+    "services/aios-single-owner-governance/aios_single_owner_governance/state_machine.py",
+    "services/aios-single-owner-governance/aios_single_owner_governance/validator.py",
+}
+
+
+class InjectedCrash(BaseException):
+    pass
+
+
+class Adapter(Protocol):
+    def read_current(self) -> dict[str, Any]: ...
+    def calculate_snapshot_sha256(self, snapshot: Mapping[str, Any]) -> str: ...
+    def apply_atomic(self, expected_sha256: str, target: Mapping[str, Any]) -> dict[str, Any]: ...
+    def verify_applied(self, expected_sha256: str) -> bool: ...
+    def rollback_atomic(self, expected_target_sha256: str, previous: Mapping[str, Any]) -> dict[str, Any]: ...
+    def verify_rollback(self, expected_sha256: str) -> bool: ...
+
+
+@contextmanager
+def exclusive_lock(path: Path) -> Iterator[None]:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+    flags = os.O_RDWR | os.O_CREAT
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    fd = os.open(path, flags, 0o640)
+    try:
+        os.fchmod(fd, 0o640)
+        if fcntl is not None:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        yield
+    finally:
+        if fcntl is not None:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        os.close(fd)
+
+
+def _require_hex(name: str, value: object, lengths: tuple[int, ...]) -> None:
+    if not isinstance(value, str) or len(value) not in lengths or not re.fullmatch(r"[0-9a-f]+", value):
+        raise RuntimeError(f"{name.upper()}_INVALID")
+
+
+def _sha256_file(path: Path) -> str:
+    if not path.is_absolute() or not path.is_file() or path.is_symlink():
+        raise RuntimeError("VERIFIED_COMPONENT_PATH_INVALID")
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def _git(git_dir: Path, *args: str) -> str:
+    if not git_dir.is_absolute() or not git_dir.exists() or git_dir.is_symlink():
+        raise RuntimeError("GIT_OBJECT_STORE_INVALID")
+    result = subprocess.run(
+        ["git", f"--git-dir={git_dir}", *args],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise RuntimeError("GIT_EVIDENCE_VERIFICATION_FAILED")
+    return result.stdout.strip()
+
+
+def _git_raw(git_dir: Path, *args: str) -> bytes:
+    if not git_dir.is_absolute() or not git_dir.exists() or git_dir.is_symlink():
+        raise RuntimeError("GIT_OBJECT_STORE_INVALID")
+    result = subprocess.run(
+        ["git", f"--git-dir={git_dir}", *args], capture_output=True, check=False
+    )
+    if result.returncode != 0:
+        raise RuntimeError("GIT_EVIDENCE_VERIFICATION_FAILED")
+    return result.stdout
+
+
+def _parse_utc(value: object) -> datetime:
+    if not isinstance(value, str):
+        raise RuntimeError("EXPIRY_INVALID")
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise RuntimeError("EXPIRY_INVALID") from exc
+    if parsed.tzinfo is None:
+        raise RuntimeError("EXPIRY_INVALID")
+    return parsed.astimezone(timezone.utc)
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _approved_execution_binding(bindings: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "task_id": bindings["task_id"],
+        "operation_id": bindings["operation_id"],
+        "expected_validator_sha256": bindings["expected_validator_sha256"],
+        "expected_state_machine_sha256": bindings["expected_state_machine_sha256"],
+        "expected_branch_protection_sha256": bindings["expected_branch_protection_sha256"],
+        "expected_gitea_schema_sha256": bindings["expected_gitea_schema_sha256"],
+        "target_branch_protection_sha256": calculate_snapshot_sha256(
+            bindings["target_branch_protection"]
+        ),
+        "evidence_closure_oid": bindings["evidence_closure_oid"],
+        "evidence_closure_path": bindings["evidence_closure_path"],
+        "evidence_closure_sha256": bindings["evidence_closure_sha256"],
+        "implementation_commit_oid": bindings["implementation_commit_oid"],
+        "implementation_tree_oid": bindings["implementation_tree_oid"],
+        "runtime_sources": bindings["runtime_sources"],
+        "artifact_path": bindings["artifact_path"],
+        "artifact_sha256": bindings["artifact_sha256"],
+        "rollback_artifact_path": bindings["rollback_artifact_path"],
+        "rollback_artifact_sha256": bindings["rollback_artifact_sha256"],
+        "test_evidence_path": bindings["test_evidence_path"],
+        "test_evidence_sha256": bindings["test_evidence_sha256"],
+        "machine_status_producer_id": bindings["machine_status_producer_id"],
+        "machine_status_producer_name": bindings["machine_status_producer_name"],
+        "backend": bindings["backend"],
+        "repo_id": bindings["repo_id"],
+        "branch": bindings["branch"],
+        "requester": bindings["requester"],
+        "executor": bindings["executor"],
+        "forbidden_operations": bindings["forbidden_operations"],
+    }
+
+
+def _same_path(left: Path, right: Path) -> bool:
+    try:
+        return left.resolve(strict=True) == right.resolve(strict=True) and os.path.samefile(left, right)
+    except (FileNotFoundError, OSError):
+        return False
+
+
+def _verify_authority_paths(bindings: Mapping[str, Any]) -> None:
+    if bindings["backend"] != "sqlite":
+        return
+    approval_db = Path(str(bindings["approval_db_path"]))
+    backend_db = Path(str(bindings["backend_path"]))
+    if not _same_path(approval_db, backend_db):
+        raise RuntimeError("APPROVAL_DB_BACKEND_DB_IDENTITY_MISMATCH")
+    live = _same_path(backend_db, LIVE_GITEA_DB)
+    if live:
+        if not _same_path(Path(str(bindings["governance_git_dir"])), LIVE_AUTHORITATIVE_GIT):
+            raise RuntimeError("AUTHORITATIVE_GOVERNANCE_GIT_REQUIRED")
+        if not _same_path(Path(str(bindings["evidence_git_dir"])), LIVE_AUTHORITATIVE_GIT):
+            raise RuntimeError("AUTHORITATIVE_EVIDENCE_GIT_REQUIRED")
+        return
+    acceptance_root = os.environ.get("AIOS_GOVERNANCE_ACTIVATION_ACCEPTANCE_ROOT")
+    if not acceptance_root:
+        raise RuntimeError("NON_LIVE_SQLITE_BACKEND_DENIED")
+    root = Path(acceptance_root).resolve(strict=True)
+    if not str(root).startswith("/run/aios-live-activation-acceptance-"):
+        raise RuntimeError("ACCEPTANCE_ROOT_INVALID")
+    for path in (approval_db, backend_db, Path(str(bindings["governance_git_dir"])), Path(str(bindings["evidence_git_dir"]))):
+        try:
+            path.resolve(strict=True).relative_to(root)
+        except (FileNotFoundError, ValueError) as exc:
+            raise RuntimeError("ACCEPTANCE_AUTHORITY_ESCAPE_DENIED") from exc
+
+
+def _load_approved_manifest(bindings: Mapping[str, Any], git_dir: Path) -> dict[str, Any]:
+    path = str(bindings["activation_manifest_path"])
+    if not path.startswith("approvals/") or not path.endswith(".json") or ".." in Path(path).parts:
+        raise RuntimeError("ACTIVATION_MANIFEST_PATH_INVALID")
+    merge_oid = str(bindings["activation_approval_merge_oid"])
+    blob_oid = _git(git_dir, "rev-parse", f"{merge_oid}:{path}")
+    if blob_oid != bindings["activation_manifest_blob_oid"]:
+        raise RuntimeError("ACTIVATION_MANIFEST_BLOB_MISMATCH")
+    raw_result = subprocess.run(
+        ["git", f"--git-dir={git_dir}", "show", f"{merge_oid}:{path}"],
+        capture_output=True,
+        check=False,
+    )
+    if raw_result.returncode != 0:
+        raise RuntimeError("ACTIVATION_MANIFEST_READ_FAILED")
+    if hashlib.sha256(raw_result.stdout).hexdigest() != bindings["activation_manifest_sha256"]:
+        raise RuntimeError("ACTIVATION_MANIFEST_SHA256_MISMATCH")
+    try:
+        manifest = json.loads(raw_result.stdout)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError("ACTIVATION_MANIFEST_INVALID") from exc
+    if not isinstance(manifest, dict):
+        raise RuntimeError("ACTIVATION_MANIFEST_INVALID")
+    return manifest
+
+
+def _require_approval_unexpired(bindings: Mapping[str, Any]) -> None:
+    if _utc_now() >= _parse_utc(bindings["expires_at"]):
+        raise RuntimeError("APPROVAL_EXPIRED")
+
+
+def _safe_git_path(value: object) -> str:
+    path = str(value)
+    if not path or Path(path).is_absolute() or ".." in Path(path).parts:
+        raise RuntimeError("APPROVED_GIT_PATH_INVALID")
+    return path
+
+
+def _verify_approved_entities(bindings: Mapping[str, Any], git_dir: Path) -> None:
+    implementation_oid = str(bindings["implementation_commit_oid"])
+    _git(git_dir, "cat-file", "-e", f"{implementation_oid}^{{commit}}")
+    if _git(git_dir, "rev-parse", f"{implementation_oid}^{{tree}}") != bindings["implementation_tree_oid"]:
+        raise RuntimeError("IMPLEMENTATION_TREE_MISMATCH")
+    sources = bindings["runtime_sources"]
+    if not isinstance(sources, list) or not sources:
+        raise RuntimeError("RUNTIME_SOURCES_INVALID")
+    seen_git: set[str] = set()
+    seen_runtime: set[str] = set()
+    loaded_runtime_paths = {
+        "services/aios-single-owner-governance/aios_single_owner_governance/activation.py": Path(__file__).resolve(),
+        "services/aios-single-owner-governance/aios_single_owner_governance/gitea_protected_branch.py": Path(__import__(SQLiteProtectedBranchAdapter.__module__, fromlist=["x"]).__file__).resolve(),
+        "services/aios-single-owner-governance/aios_single_owner_governance/journal.py": Path(__import__(JournalStore.__module__, fromlist=["x"]).__file__).resolve(),
+        "services/aios-single-owner-governance/aios_single_owner_governance/state_machine.py": Path(state_machine_module.__file__).resolve(),
+        "services/aios-single-owner-governance/aios_single_owner_governance/validator.py": Path(validator_module.__file__).resolve(),
+    }
+    for source in sources:
+        if not isinstance(source, dict) or set(source) != {"git_path", "runtime_path", "sha256"}:
+            raise RuntimeError("RUNTIME_SOURCE_BINDING_INVALID")
+        git_path = _safe_git_path(source["git_path"])
+        runtime_path = str(source["runtime_path"])
+        _require_hex("runtime_source_sha256", source["sha256"], (64,))
+        if git_path in seen_git or runtime_path in seen_runtime:
+            raise RuntimeError("RUNTIME_SOURCE_BINDING_DUPLICATE")
+        seen_git.add(git_path)
+        seen_runtime.add(runtime_path)
+        if Path(runtime_path).resolve() != loaded_runtime_paths.get(git_path):
+            raise RuntimeError("RUNTIME_SOURCE_PATH_MISMATCH")
+        if hashlib.sha256(_git_raw(git_dir, "show", f"{implementation_oid}:{git_path}")).hexdigest() != source["sha256"]:
+            raise RuntimeError("IMPLEMENTATION_SOURCE_DIGEST_MISMATCH")
+        if _sha256_file(Path(runtime_path)) != source["sha256"]:
+            raise RuntimeError("RUNTIME_SOURCE_DIGEST_MISMATCH")
+    if seen_git != REQUIRED_RUNTIME_GIT_PATHS:
+        raise RuntimeError("RUNTIME_SOURCE_SET_MISMATCH")
+    for prefix in ("artifact", "rollback_artifact", "test_evidence"):
+        if _sha256_file(Path(str(bindings[f"{prefix}_path"]))) != bindings[f"{prefix}_sha256"]:
+            raise RuntimeError("APPROVED_ENTITY_DIGEST_MISMATCH")
+
+    closure_oid = str(bindings["evidence_closure_oid"])
+    if _git(git_dir, "cat-file", "-t", closure_oid) != "commit":
+        raise RuntimeError("EVIDENCE_CLOSURE_NOT_COMMIT")
+    parents = _git(git_dir, "rev-list", "--parents", "-n", "1", closure_oid).split()
+    if parents != [closure_oid, implementation_oid]:
+        raise RuntimeError("EVIDENCE_CLOSURE_PARENT_MISMATCH")
+    closure_path = _safe_git_path(bindings["evidence_closure_path"])
+    if hashlib.sha256(_git_raw(git_dir, "show", f"{closure_oid}:{closure_path}")).hexdigest() != bindings["evidence_closure_sha256"]:
+        raise RuntimeError("EVIDENCE_CLOSURE_SHA256_MISMATCH")
+    if _git(git_dir, "diff", "--name-status", implementation_oid, closure_oid) != f"A\t{closure_path}":
+        raise RuntimeError("EVIDENCE_CLOSURE_DIFF_INVALID")
+
+
+def _verify_external_evidence(
+    bindings: Mapping[str, Any], *, require_unexpired: bool = True
+) -> None:
+    _verify_authority_paths(bindings)
+    validator_sha = _sha256_file(Path(str(bindings["validator_path"])))
+    state_machine_sha = _sha256_file(Path(str(bindings["state_machine_path"])))
+    if validator_sha != bindings["expected_validator_sha256"] or validator_sha != bindings["observed_validator_sha256"]:
+        raise RuntimeError("VALIDATOR_DIGEST_MISMATCH")
+    if state_machine_sha != bindings["expected_state_machine_sha256"] or state_machine_sha != bindings["observed_state_machine_sha256"]:
+        raise RuntimeError("STATE_MACHINE_DIGEST_MISMATCH")
+
+    governance_git_dir = Path(str(bindings["governance_git_dir"]))
+    governance_oid = _git(governance_git_dir, "rev-parse", "--verify", str(bindings["governance_ref"]))
+    if governance_oid != bindings["expected_governance_oid"] or governance_oid != bindings["observed_governance_oid"]:
+        raise RuntimeError("WRONG_GOVERNANCE_OID_DENIED")
+    merge_oid = str(bindings["activation_approval_merge_oid"])
+    _git(governance_git_dir, "cat-file", "-e", f"{merge_oid}^{{commit}}")
+    if subprocess.run(
+        ["git", f"--git-dir={governance_git_dir}", "merge-base", "--is-ancestor", merge_oid, governance_oid],
+        capture_output=True,
+        check=False,
+    ).returncode != 0:
+        raise RuntimeError("ACTIVATION_APPROVAL_NOT_IN_GOVERNANCE_REF")
+    manifest = _load_approved_manifest(bindings, governance_git_dir)
+    if manifest.get("approval_type") != "SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION":
+        raise RuntimeError("ACTIVATION_APPROVAL_TYPE_MISMATCH")
+    if manifest.get("execution_binding") != _approved_execution_binding(bindings):
+        raise RuntimeError("APPROVED_EXECUTION_BINDING_MISMATCH")
+    if manifest.get("execution_binding_sha256") != sha256_json(manifest["execution_binding"]):
+        raise RuntimeError("APPROVED_EXECUTION_BINDING_SHA256_MISMATCH")
+    if manifest.get("expires_at") != bindings["expires_at"]:
+        raise RuntimeError("APPROVAL_EXPIRY_BINDING_MISMATCH")
+
+    manifest_bound_fields = (
+        "implementation_commit_oid", "implementation_tree_oid", "runtime_sources",
+        "artifact_path", "artifact_sha256", "rollback_artifact_path",
+        "rollback_artifact_sha256", "test_evidence_path", "test_evidence_sha256",
+        "evidence_closure_oid", "evidence_closure_path", "evidence_closure_sha256",
+        "machine_status_producer_id", "machine_status_producer_name",
+    )
+    if any(manifest.get(field) != bindings[field] for field in manifest_bound_fields):
+        raise RuntimeError("APPROVED_ARTIFACT_BINDING_MISMATCH")
+
+    evidence_git_dir = Path(str(bindings["evidence_git_dir"]))
+    evidence_oid = str(bindings["evidence_closure_oid"])
+    if evidence_oid != bindings["observed_evidence_closure_oid"]:
+        raise RuntimeError("WRONG_EVIDENCE_CLOSURE_DENIED")
+    _verify_approved_entities(bindings, evidence_git_dir)
+
+    db_path = Path(str(bindings["approval_db_path"]))
+    if not db_path.is_absolute() or not db_path.is_file() or db_path.is_symlink():
+        raise RuntimeError("APPROVAL_DATABASE_INVALID")
+    expiry = _parse_utc(bindings["expires_at"])
+    if require_unexpired:
+        _require_approval_unexpired(bindings)
+    with sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) as connection:
+        connection.row_factory = sqlite3.Row
+        pr = connection.execute(
+            "SELECT pr.issue_id,pr.has_merged,pr.merged_commit_id,pr.merged_unix,i.is_closed "
+            "FROM pull_request pr JOIN issue i ON i.id=pr.issue_id "
+            "WHERE i.repo_id=? AND pr.\"index\"=? AND pr.base_branch=?",
+            (int(bindings["repo_id"]), int(bindings["activation_approval_pr_number"]), str(bindings["branch"])),
+        ).fetchone()
+        if pr is None or int(pr["has_merged"]) != 1 or int(pr["is_closed"]) != 1:
+            raise RuntimeError("ACTIVATION_APPROVAL_PR_NOT_MERGED")
+        if pr["merged_commit_id"] != bindings["activation_approval_merge_oid"]:
+            raise RuntimeError("ACTIVATION_APPROVAL_MERGE_MISMATCH")
+        if int(pr["merged_unix"] or 0) <= 0 or datetime.fromtimestamp(int(pr["merged_unix"]), timezone.utc) > expiry:
+            raise RuntimeError("ACTIVATION_APPROVAL_MERGED_AFTER_EXPIRY")
+        review = connection.execute(
+            "SELECT r.id,r.type,r.reviewer_id,r.commit_id,r.stale,r.dismissed,r.created_unix,u.lower_name "
+            "FROM review r JOIN user u ON u.id=r.reviewer_id WHERE r.id=? AND r.issue_id=?",
+            (int(bindings["owner_review_oid"]), int(pr["issue_id"])),
+        ).fetchone()
+        if review is None or int(review["id"]) != int(bindings["observed_owner_review_oid"]):
+            raise RuntimeError("WRONG_APPROVAL_OID_DENIED")
+        if (
+            int(review["type"]) != 1
+            or int(review["reviewer_id"]) != 2
+            or review["lower_name"] != "owner"
+            or review["commit_id"] != bindings["approval_request_head_oid"]
+            or int(review["stale"]) != 0
+            or int(review["dismissed"]) != 0
+            or datetime.fromtimestamp(int(review["created_unix"]), timezone.utc) >= expiry
+        ):
+            raise RuntimeError("OWNER_APPROVAL_INVALID")
+        statuses = connection.execute(
+            "SELECT cs.id,cs.context,cs.state,cs.sha,cs.creator_id,u.lower_name AS creator_login "
+            "FROM commit_status cs LEFT JOIN user u ON u.id=cs.creator_id "
+            "WHERE cs.repo_id=? AND cs.sha=? ORDER BY cs.id DESC",
+            (int(bindings["repo_id"]), str(bindings["approval_request_head_oid"])),
+        ).fetchall()
+        observed: dict[str, sqlite3.Row] = {}
+        for row in statuses:
+            if row["context"] in REQUIRED_MACHINE_CONTEXTS:
+                observed.setdefault(row["context"], row)
+        if set(observed) != REQUIRED_MACHINE_CONTEXTS:
+            raise RuntimeError("MACHINE_GATE_SET_MISMATCH")
+        for context, row in observed.items():
+            if row["state"] != "success" or row["sha"] != bindings["approval_request_head_oid"]:
+                raise RuntimeError("MACHINE_GATE_FAILED")
+            if (
+                int(row["creator_id"] or 0) != int(bindings["machine_status_producer_id"])
+                or row["creator_login"] != bindings["machine_status_producer_name"]
+            ):
+                raise RuntimeError("MACHINE_GATE_PRODUCER_MISMATCH")
+            expected = bindings["machine_checks"][context]
+            actual = {
+                "state": row["state"], "head_oid": row["sha"], "status_id": int(row["id"]),
+                "creator_id": int(row["creator_id"]), "creator_login": row["creator_login"],
+            }
+            if expected != actual:
+                raise RuntimeError("MACHINE_GATE_BINDING_MISMATCH")
+
+
+def validate_bindings(
+    bindings: Mapping[str, Any], *, require_unexpired: bool = True
+) -> None:
+    missing = sorted(REQUIRED_BINDINGS - set(bindings))
+    if missing:
+        raise RuntimeError("MISSING_BINDINGS:" + ",".join(missing))
+    for name in (
+        "expected_governance_oid",
+        "activation_approval_merge_oid",
+        "evidence_closure_oid",
+        "implementation_commit_oid",
+        "implementation_tree_oid",
+    ):
+        _require_hex(name, bindings[name], (40,))
+    for name in (
+        "expected_validator_sha256",
+        "expected_state_machine_sha256",
+        "expected_branch_protection_sha256",
+        "observed_validator_sha256",
+        "observed_state_machine_sha256",
+        "expected_gitea_schema_sha256",
+        "activation_manifest_sha256",
+        "artifact_sha256",
+        "rollback_artifact_sha256",
+        "test_evidence_sha256",
+        "evidence_closure_sha256",
+    ):
+        _require_hex(name, bindings[name], (64,))
+    _require_hex("activation_manifest_blob_oid", bindings["activation_manifest_blob_oid"], (40, 64))
+    if not isinstance(bindings["owner_review_oid"], (int, str)) or str(bindings["owner_review_oid"]).strip() == "":
+        raise RuntimeError("OWNER_REVIEW_OID_INVALID")
+    if bindings["governance_ref"] != "refs/heads/governance/approvals":
+        raise RuntimeError("GOVERNANCE_REF_INVALID")
+    if bindings["branch"] != "governance/approvals":
+        raise RuntimeError("BRANCH_BINDING_DENIED")
+    if bindings["backend"] not in {"file", "sqlite"}:
+        raise RuntimeError("UNKNOWN_PROTECTION_BACKEND")
+    if bindings["observed_governance_oid"] != bindings["expected_governance_oid"]:
+        raise RuntimeError("WRONG_GOVERNANCE_OID_DENIED")
+    if bindings["observed_evidence_closure_oid"] != bindings["evidence_closure_oid"]:
+        raise RuntimeError("WRONG_EVIDENCE_CLOSURE_DENIED")
+    if str(bindings["observed_owner_review_oid"]) != str(bindings["owner_review_oid"]):
+        raise RuntimeError("WRONG_APPROVAL_OID_DENIED")
+    if bindings.get("requester") == "owner":
+        raise RuntimeError("REQUESTER_SELF_APPROVAL_DENIED")
+    if bindings.get("executor") == "owner":
+        raise RuntimeError("EXECUTOR_SELF_APPROVAL_DENIED")
+    if bindings.get("wip") is not False:
+        raise RuntimeError("WIP_BLOCKED")
+    checks = bindings.get("machine_checks")
+    if not isinstance(checks, dict) or set(checks) != REQUIRED_MACHINE_CONTEXTS:
+        raise RuntimeError("MACHINE_GATE_SET_MISMATCH")
+    if not isinstance(bindings["machine_status_producer_id"], int) or bindings["machine_status_producer_id"] < 1:
+        raise RuntimeError("MACHINE_STATUS_PRODUCER_INVALID")
+    if not isinstance(bindings["machine_status_producer_name"], str) or not bindings["machine_status_producer_name"]:
+        raise RuntimeError("MACHINE_STATUS_PRODUCER_INVALID")
+    for context, check in checks.items():
+        if (
+            not isinstance(check, dict)
+            or set(check) != {"state", "head_oid", "status_id", "creator_id", "creator_login"}
+            or check.get("state") != "success"
+            or check.get("head_oid") != bindings["approval_request_head_oid"]
+            or not isinstance(check.get("status_id"), int)
+            or check.get("status_id") < 1
+            or check.get("creator_id") != bindings["machine_status_producer_id"]
+            or check.get("creator_login") != bindings["machine_status_producer_name"]
+        ):
+            raise RuntimeError(f"MACHINE_GATE_FAILED:{context}")
+    forbidden = bindings.get("forbidden_operations", {})
+    if any(bool(value) for value in forbidden.values()):
+        raise RuntimeError("FORBIDDEN_OPERATION_REQUESTED")
+    if os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower() != "false":
+        raise RuntimeError("GLOBAL_GITOPS_WRITES_MUST_REMAIN_FALSE")
+    if not isinstance(bindings.get("target_branch_protection"), dict):
+        raise RuntimeError("TARGET_BRANCH_PROTECTION_REQUIRED")
+    _verify_external_evidence(bindings, require_unexpired=require_unexpired)
+
+
+def binding_payload_hash(bindings: Mapping[str, Any]) -> str:
+    stable = dict(bindings)
+    stable.pop("crash_at", None)
+    return sha256_json(redact(stable))
+
+
+def plan_id(bindings: Mapping[str, Any], previous: Mapping[str, Any], target: Mapping[str, Any]) -> str:
+    stable = dict(bindings)
+    stable.pop("crash_at", None)
+    return sha256_json(
+        {
+            "bindings": redact(stable),
+            "previous": dict(previous),
+            "target": dict(target),
+        }
+    )
+
+
+class ActivationEngine:
+    def __init__(self, adapter: Adapter, state_root: Path, lock_path: Path) -> None:
+        self.adapter = adapter
+        self.journal = JournalStore(state_root)
+        self.lock_path = lock_path
+
+    def plan(self, bindings: Mapping[str, Any]) -> dict[str, Any]:
+        validate_bindings(bindings)
+        previous = self.adapter.read_current()
+        previous_sha = calculate_snapshot_sha256(previous)
+        if previous_sha != bindings["expected_branch_protection_sha256"]:
+            raise CASConflict("EXPECTED_BRANCH_PROTECTION_SHA256_MISMATCH")
+        target = dict(bindings["target_branch_protection"])
+        # Validate without writing by using the public target validator through a
+        # disposable comparison: adapters enforce the same checks again in CAS.
+        from .gitea_protected_branch import validate_target
+
+        validate_target(previous, target)
+        result = {
+            "status": "PLAN_READY",
+            "task_id": bindings["task_id"],
+            "operation_id": bindings["operation_id"],
+            "plan_id": plan_id(bindings, previous, target),
+            "payload_hash": binding_payload_hash(bindings),
+            "previous_sha256": previous_sha,
+            "target_sha256": calculate_snapshot_sha256(target),
+        }
+        return result
+
+    def _recover(self, record: dict[str, Any]) -> dict[str, Any]:
+        current = self.adapter.read_current()
+        current_sha = calculate_snapshot_sha256(current)
+        previous_sha = record["previous_sha256"]
+        target_sha = record["target_sha256"]
+        if record["phase"] == "PREPARED":
+            if current_sha == previous_sha:
+                return {**record, "recovery": "RESUME_BEFORE_APPLY"}
+            if current_sha == target_sha:
+                record = {**record, "phase": "APPLIED", "recovery": "RESUME_AFTER_APPLY"}
+                self.journal.write(record["operation_id"], record)
+                return record
+            raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
+        if record["phase"] in {"APPLIED", "COMMITTED"} and current_sha not in {target_sha, previous_sha}:
+            raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
+        return record
+
+    def _ensure_closure(self, record: dict[str, Any], event: str) -> None:
+        expected_sha = (
+            record["target_sha256"] if record["phase"] == "COMMITTED" else record["previous_sha256"]
+        )
+        if calculate_snapshot_sha256(self.adapter.read_current()) != expected_sha:
+            raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
+        atomic_write_json(
+            self.journal.root / "current-snapshot.json",
+            record["target"] if record["phase"] == "COMMITTED" else record["previous"],
+        )
+        atomic_write_json(self.journal.root / "rollback-snapshot.json", record["previous"])
+        self.journal.write_receipt(record["operation_id"], record)
+        if not self.journal.audit.has_event(record["operation_id"], event):
+            payload = {"event": event}
+            if event == "SINGLE_OWNER_ACTIVATION_COMMITTED":
+                payload.update(
+                    payload_hash=record["payload_hash"],
+                    target_sha256=record["target_sha256"],
+                )
+            self.journal.audit.append(record["operation_id"], payload)
+        atomic_write_json(self.journal.root / "state.json", record)
+
+    def activate(self, bindings: Mapping[str, Any], supplied_plan_id: str) -> dict[str, Any]:
+        with exclusive_lock(self.lock_path):
+            self.journal.audit.verify()
+            operation_id = str(bindings["operation_id"])
+            payload_hash = binding_payload_hash(bindings)
+            record = self.journal.load(operation_id)
+            if record:
+                if record.get("payload_hash") != payload_hash:
+                    raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
+                if record.get("plan_id") != supplied_plan_id:
+                    raise RuntimeError("PLAN_ID_MISMATCH")
+                if record.get("phase") == "ROLLED_BACK":
+                    raise RuntimeError("OPERATION_TERMINAL_ROLLED_BACK")
+                if record.get("phase") == "FAILED":
+                    raise RuntimeError("OPERATION_TERMINAL_FAILED")
+                prepared_at = _parse_utc(record.get("prepared_at"))
+                if prepared_at >= _parse_utc(bindings["expires_at"]):
+                    raise RuntimeError("RECOVERY_AUTHORIZATION_INVALID")
+                record = self._recover(record)
+                if record["phase"] == "COMMITTED":
+                    self._ensure_closure(record, "SINGLE_OWNER_ACTIVATION_COMMITTED")
+                    return {"status": "ALREADY_COMMITTED", **record}
+            else:
+                validate_bindings(bindings)
+                planned = self.plan(bindings)
+                if supplied_plan_id != planned["plan_id"]:
+                    raise RuntimeError("PLAN_ID_MISMATCH")
+            previous = self.adapter.read_current()
+            target = dict(bindings["target_branch_protection"])
+            if record is None:
+                if calculate_snapshot_sha256(previous) != planned["previous_sha256"]:
+                    raise CASConflict("PREPARED_BOUNDARY_SNAPSHOT_CHANGED")
+                prepared_at = _utc_now()
+                if prepared_at >= _parse_utc(bindings["expires_at"]):
+                    raise RuntimeError("APPROVAL_EXPIRED")
+                record = {
+                    "task_id": bindings["task_id"],
+                    "operation_id": operation_id,
+                    "transaction_id": operation_id,
+                    "payload_hash": payload_hash,
+                    "plan_id": supplied_plan_id,
+                    "phase": "PREPARED",
+                    "prepared_at": prepared_at.isoformat().replace("+00:00", "Z"),
+                    "previous": previous,
+                    "target": target,
+                    "previous_sha256": calculate_snapshot_sha256(previous),
+                    "target_sha256": calculate_snapshot_sha256(target),
+                    "activation_approval_merge_oid": bindings["activation_approval_merge_oid"],
+                    "evidence_closure_oid": bindings["evidence_closure_oid"],
+                    "owner_review_oid": bindings["owner_review_oid"],
+                }
+                self.journal.write(operation_id, record)
+                atomic_write_json(self.journal.root / "current-snapshot.json", previous)
+                atomic_write_json(self.journal.root / "rollback-snapshot.json", previous)
+            crash_at = bindings.get("crash_at")
+            if crash_at == "before_apply":
+                raise InjectedCrash("before_apply")
+            try:
+                current_sha = calculate_snapshot_sha256(self.adapter.read_current())
+                if current_sha == record["previous_sha256"]:
+                    applied = self.adapter.apply_atomic(record["previous_sha256"], record["target"])
+                    record["target"] = applied
+                    record["target_sha256"] = calculate_snapshot_sha256(applied)
+                elif current_sha != record["target_sha256"]:
+                    raise CASConflict("CAS_CONFLICT")
+                if crash_at == "during_apply":
+                    raise InjectedCrash("during_apply")
+                record["phase"] = "APPLIED"
+                self.journal.write(operation_id, record)
+                if crash_at == "after_apply_before_receipt":
+                    raise InjectedCrash("after_apply_before_receipt")
+                if not self.adapter.verify_applied(record["target_sha256"]):
+                    raise RuntimeError("POST_ACTIVATION_VERIFY_FAILED")
+                record["phase"] = "COMMITTED"
+                record["single_owner_policy_active"] = True
+                record["break_glass_active"] = False
+                self.journal.write(operation_id, record)
+                if crash_at == "after_commit_wal":
+                    raise InjectedCrash("after_commit_wal")
+                self._ensure_closure(record, "SINGLE_OWNER_ACTIVATION_COMMITTED")
+                return {"status": "ACTIVATED", **record}
+            except InjectedCrash:
+                raise
+            except Exception as exc:
+                failure = {**record, "phase": "FAILED", "error": type(exc).__name__}
+                current_sha = calculate_snapshot_sha256(self.adapter.read_current())
+                if current_sha == record["target_sha256"]:
+                    self.adapter.rollback_atomic(record["target_sha256"], record["previous"])
+                    if not self.adapter.verify_rollback(record["previous_sha256"]):
+                        raise RuntimeError("ROLLBACK_VERIFY_FAILED") from exc
+                    failure["phase"] = "ROLLED_BACK"
+                self.journal.write(operation_id, failure)
+                self.journal.write_receipt(operation_id, failure)
+                self.journal.audit.append(operation_id, {"event": failure["phase"], "error": type(exc).__name__})
+                raise
+
+    def verify(self, bindings: Mapping[str, Any], operation_id: str) -> dict[str, Any]:
+        record = self.journal.load(operation_id)
+        if not record:
+            raise RuntimeError("TRANSACTION_NOT_FOUND")
+        if record.get("payload_hash") != binding_payload_hash(bindings):
+            raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
+        self.journal.audit.verify()
+        current_sha = calculate_snapshot_sha256(self.adapter.read_current())
+        passed = record.get("phase") == "COMMITTED" and current_sha == record.get("target_sha256")
+        if not passed:
+            raise RuntimeError("ACTIVATION_VERIFY_FAILED")
+        return {"status": "VERIFIED", "operation_id": operation_id, "target_sha256": current_sha}
+
+    def rollback(
+        self, bindings: Mapping[str, Any], operation_id: str, supplied_plan_id: str
+    ) -> dict[str, Any]:
+        with exclusive_lock(self.lock_path):
+            self.journal.audit.verify()
+            record = self.journal.load(operation_id)
+            if not record:
+                raise RuntimeError("TRANSACTION_NOT_FOUND")
+            if record.get("payload_hash") != binding_payload_hash(bindings):
+                raise RuntimeError("OPERATION_ID_PAYLOAD_MISMATCH_DENIED")
+            if record.get("plan_id") != supplied_plan_id:
+                raise RuntimeError("PLAN_ID_MISMATCH")
+            for field in (
+                "activation_approval_merge_oid",
+                "evidence_closure_oid",
+                "owner_review_oid",
+            ):
+                if str(record.get(field)) != str(bindings[field]):
+                    raise RuntimeError("ROLLBACK_AUTHORIZATION_MISMATCH_DENIED")
+            current_sha = calculate_snapshot_sha256(self.adapter.read_current())
+            if current_sha == record["previous_sha256"]:
+                record["phase"] = "ROLLED_BACK"
+                record["single_owner_policy_active"] = False
+                self.journal.write(operation_id, record)
+                self._ensure_closure(record, "ACTIVATION_ROLLED_BACK")
+                return {"status": "ALREADY_ROLLED_BACK", "operation_id": operation_id}
+            if current_sha != record["target_sha256"]:
+                raise RuntimeError("MANUAL_INTERVENTION_REQUIRED")
+            self.adapter.rollback_atomic(record["target_sha256"], record["previous"])
+            if not self.adapter.verify_rollback(record["previous_sha256"]):
+                raise RuntimeError("ROLLBACK_VERIFY_FAILED")
+            if bindings.get("crash_at") == "after_rollback_cas":
+                raise InjectedCrash("after_rollback_cas")
+            record["phase"] = "ROLLED_BACK"
+            record["single_owner_policy_active"] = False
+            self.journal.write(operation_id, record)
+            if bindings.get("crash_at") == "after_rollback_wal":
+                raise InjectedCrash("after_rollback_wal")
+            self._ensure_closure(record, "ACTIVATION_ROLLED_BACK")
+            return {"status": "ROLLED_BACK", "operation_id": operation_id}
+
+    def status(self, operation_id: str | None = None) -> dict[str, Any]:
+        audit_head = self.journal.audit.verify()
+        if operation_id:
+            record = self.journal.load(operation_id)
+            if not record:
+                raise RuntimeError("TRANSACTION_NOT_FOUND")
+            return {"status": "OK", "audit_head": audit_head, "transaction": record}
+        state_path = self.journal.root / "state.json"
+        return {
+            "status": "OK",
+            "audit_head": audit_head,
+            "state": read_json(state_path) if state_path.exists() else None,
+            "transactions": list(self.journal.iter_transactions()),
+        }
+
+
+def load_bindings(path: Path) -> dict[str, Any]:
+    return read_json(path)
+
+
+def adapter_from_bindings(bindings: Mapping[str, Any]) -> Adapter:
+    backend = bindings.get("backend", "file")
+    backend_path = Path(str(bindings.get("backend_path", "")))
+    if not backend_path.is_absolute():
+        raise RuntimeError("BACKEND_PATH_MUST_BE_ABSOLUTE")
+    if backend == "file":
+        return FileProtectedBranchAdapter(backend_path)
+    if backend == "sqlite":
+        return SQLiteProtectedBranchAdapter(
+            backend_path,
+            repo_id=int(bindings.get("repo_id", 1)),
+            branch=str(bindings.get("branch", "governance/approvals")),
+            expected_schema_sha256=str(bindings["expected_gitea_schema_sha256"]),
+        )
+    raise RuntimeError("UNKNOWN_PROTECTION_BACKEND")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    action = parser.add_mutually_exclusive_group(required=True)
+    for name in ("plan", "activate", "verify", "rollback", "status"):
+        action.add_argument(f"--{name}", action="store_true")
+    parser.add_argument("--bindings", type=Path)
+    parser.add_argument("--plan-id")
+    parser.add_argument("--transaction")
+    parser.add_argument("--state-root", type=Path, default=DEFAULT_STATE_ROOT)
+    parser.add_argument("--lock-path", type=Path, default=DEFAULT_LOCK_PATH)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    if args.status and args.bindings is None:
+        # Status without bindings is intentionally limited to durable local state.
+        journal = JournalStore(args.state_root)
+        result = {
+            "status": "OK",
+            "audit_head": journal.audit.verify(),
+            "transactions": list(journal.iter_transactions()),
+        }
+        print(json.dumps(result, sort_keys=True))
+        return 0
+    if args.bindings is None:
+        raise SystemExit("--bindings is required")
+    bindings = load_bindings(args.bindings)
+    engine = ActivationEngine(adapter_from_bindings(bindings), args.state_root, args.lock_path)
+    if args.plan:
+        result = engine.plan(bindings)
+    elif args.activate:
+        if not args.plan_id:
+            raise SystemExit("--plan-id is required")
+        result = engine.activate(bindings, args.plan_id)
+    elif args.verify:
+        result = engine.verify(bindings, args.transaction or str(bindings["operation_id"]))
+    elif args.rollback:
+        if not args.plan_id:
+            raise SystemExit("--plan-id is required")
+        result = engine.rollback(
+            bindings, args.transaction or str(bindings["operation_id"]), args.plan_id
+        )
+    else:
+        result = engine.status(args.transaction)
+    print(json.dumps(redact(result), sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-single-owner-governance/aios_single_owner_governance/evidence.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+from typing import Mapping
+
+
+def canonical_json(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+@dataclass(frozen=True)
+class EvidenceClosure:
+    evidence: Mapping[str, str]
+    closure_sha256: str
+
+
+class ActivationEvidenceClosureBuilder:
+    def build(self, evidence: Mapping[str, str]) -> EvidenceClosure:
+        required = {
+            "activation_prerequisites_commit_oid",
+            "activation_prerequisites_tree_oid",
+            "artifact_sha256",
+            "rollback_artifact_sha256",
+            "validator_sha256",
+            "state_machine_sha256",
+            "branch_protection_adapter_sha256",
+            "rollback_configuration_sha256",
+            "test_evidence_sha256",
+            "deployment_manifest_sha256",
+            "dormant_unit_manifest_sha256",
+            "dormant_deployment_evidence_sha256",
+        }
+        if set(evidence) != required:
+            raise ValueError("evidence set mismatch")
+        for value in evidence.values():
+            if len(value) != 40 and len(value) != 64:
+                raise ValueError("invalid evidence digest")
+        return EvidenceClosure(
+            dict(evidence), hashlib.sha256(canonical_json(evidence)).hexdigest()
+        )
+
+
+class ApprovalMirrorStatusVerifier:
+    def verify(
+        self,
+        expected_oid: str,
+        *,
+        gitea_oid: str,
+        quarantine_oid: str,
+        authoritative_oid: str,
+        controller_visible_oid: str,
+        fsck_connectivity: str,
+    ) -> bool:
+        return (
+            len(expected_oid) == 40
+            and {gitea_oid, quarantine_oid, authoritative_oid, controller_visible_oid}
+            == {expected_oid}
+            and fsck_connectivity == "PASS"
+        )

services/aios-single-owner-governance/aios_single_owner_governance/fixture.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from contextlib import AbstractContextManager
+from pathlib import Path
+import shutil
+import tempfile
+
+from .models import ProtectionSnapshot
+from .protection import MemoryProtectionStore
+
+
+class DisposableGiteaFixture(AbstractContextManager):
+    def __init__(self, initial: ProtectionSnapshot) -> None:
+        self.root = Path(tempfile.mkdtemp(prefix="aios-governance-fixture-"))
+        self.protection = MemoryProtectionStore(initial)
+        self.destroyed = False
+
+    def __enter__(self) -> "DisposableGiteaFixture":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        resolved = self.root.resolve()
+        if not resolved.name.startswith("aios-governance-fixture-"):
+            raise RuntimeError("fixture cleanup path guard failed")
+        shutil.rmtree(resolved)
+        self.destroyed = True

services/aios-single-owner-governance/aios_single_owner_governance/gitea_protected_branch.py

@@ -0,0 +1,260 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sqlite3
+import time
+from typing import Any, Mapping
+
+from .journal import atomic_write_json, read_json, sha256_json
+
+
+class CASConflict(RuntimeError):
+    pass
+
+
+class UnsafePolicy(RuntimeError):
+    pass
+
+
+def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
+    return sha256_json(dict(snapshot))
+
+
+def validate_target(previous: Mapping[str, Any], target: Mapping[str, Any]) -> None:
+    if set(previous) != set(target):
+        raise UnsafePolicy("PROTECTION_ROW_SHAPE_CHANGED")
+    for field in ("id", "repo_id", "branch_name"):
+        if previous.get(field) != target.get(field):
+            raise UnsafePolicy(f"PROTECTION_IDENTITY_CHANGED:{field}")
+    if target.get("branch_name") != "governance/approvals":
+        raise UnsafePolicy("BRANCH_BINDING_DENIED")
+    if int(target.get("can_push", 1)) != 0:
+        raise UnsafePolicy("DIRECT_PUSH_MUST_REMAIN_DISABLED")
+    if int(target.get("required_approvals", 0)) != 1:
+        raise UnsafePolicy("EXACTLY_ONE_APPROVAL_REQUIRED")
+    if int(target.get("enable_approvals_whitelist", 0)) != 1:
+        raise UnsafePolicy("APPROVAL_WHITELIST_MUST_REMAIN_ENABLED")
+    if json.loads(target.get("approvals_whitelist_user_i_ds", "[]")) != [2]:
+        raise UnsafePolicy("OWNER_APPROVAL_WHITELIST_REQUIRED")
+    if int(target.get("enable_merge_whitelist", 0)) != 1:
+        raise UnsafePolicy("MERGE_WHITELIST_MUST_REMAIN_ENABLED")
+    if target.get("merge_whitelist_user_i_ds") != previous.get("merge_whitelist_user_i_ds"):
+        raise UnsafePolicy("MERGE_EXECUTOR_WHITELIST_CHANGE_DENIED")
+    if int(target.get("dismiss_stale_approvals", 0)) != 1:
+        raise UnsafePolicy("DISMISS_STALE_APPROVALS_REQUIRED")
+    if int(target.get("enable_status_check", 0)) != 1:
+        raise UnsafePolicy("STATUS_CHECKS_REQUIRED")
+    required_contexts = {
+        "aios/validator",
+        "aios/state-machine",
+        "aios/branch-protection",
+        "aios/secret-scan",
+        "aios/evidence-closure",
+    }
+    try:
+        contexts = set(json.loads(target.get("status_check_contexts", "[]")))
+    except (TypeError, ValueError) as exc:
+        raise UnsafePolicy("STATUS_CHECK_CONTEXTS_INVALID") from exc
+    if contexts != required_contexts:
+        raise UnsafePolicy("STATUS_CHECK_CONTEXTS_MISMATCH")
+    if int(target.get("block_on_rejected_reviews", 0)) != 1:
+        raise UnsafePolicy("REJECTED_REVIEWS_MUST_BLOCK")
+    if int(target.get("block_on_outdated_branch", 0)) != 1:
+        raise UnsafePolicy("OUTDATED_BRANCH_MUST_BLOCK")
+    if target.get("require_signed_commits") != previous.get("require_signed_commits"):
+        raise UnsafePolicy("SIGNATURE_POLICY_CHANGE_DENIED")
+
+
+class FileProtectedBranchAdapter:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    @staticmethod
+    def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
+        return calculate_snapshot_sha256(snapshot)
+
+    def read_current(self) -> dict[str, Any]:
+        return read_json(self.path)
+
+    def compare_expected(self, expected_sha256: str) -> bool:
+        return calculate_snapshot_sha256(self.read_current()) == expected_sha256
+
+    def apply_atomic(
+        self, expected_sha256: str, target: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        previous = self.read_current()
+        if calculate_snapshot_sha256(previous) != expected_sha256:
+            raise CASConflict("CAS_CONFLICT")
+        validate_target(previous, target)
+        atomic_write_json(self.path, dict(target))
+        applied = self.read_current()
+        if calculate_snapshot_sha256(applied) != calculate_snapshot_sha256(target):
+            raise CASConflict("CAS_VERIFY_FAILED")
+        return applied
+
+    def verify_applied(self, expected_sha256: str) -> bool:
+        return self.compare_expected(expected_sha256)
+
+    def rollback_atomic(
+        self, expected_target_sha256: str, previous: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        current = self.read_current()
+        if calculate_snapshot_sha256(current) != expected_target_sha256:
+            raise CASConflict("ROLLBACK_CONFLICT")
+        atomic_write_json(self.path, dict(previous))
+        restored = self.read_current()
+        if calculate_snapshot_sha256(restored) != calculate_snapshot_sha256(previous):
+            raise CASConflict("ROLLBACK_VERIFY_FAILED")
+        return restored
+
+    def verify_rollback(self, expected_sha256: str) -> bool:
+        return self.compare_expected(expected_sha256)
+
+
+class SQLiteProtectedBranchAdapter:
+    ALLOWED_UPDATE_FIELDS = (
+        "can_push",
+        "enable_merge_whitelist",
+        "merge_whitelist_user_i_ds",
+        "enable_status_check",
+        "status_check_contexts",
+        "enable_approvals_whitelist",
+        "approvals_whitelist_user_i_ds",
+        "required_approvals",
+        "block_on_rejected_reviews",
+        "block_on_official_review_requests",
+        "block_on_outdated_branch",
+        "dismiss_stale_approvals",
+        "ignore_stale_approvals",
+        "updated_unix",
+    )
+
+    def __init__(
+        self,
+        db_path: Path,
+        *,
+        repo_id: int = 1,
+        branch: str = "governance/approvals",
+        expected_schema_sha256: str,
+    ) -> None:
+        self.db_path = db_path
+        self.repo_id = repo_id
+        self.branch = branch
+        self.expected_schema_sha256 = expected_schema_sha256
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.db_path, timeout=10)
+        connection.row_factory = sqlite3.Row
+        return connection
+
+    def schema_sha256(self, connection: sqlite3.Connection | None = None) -> str:
+        own = connection is None
+        connection = connection or self._connect()
+        try:
+            rows = [dict(row) for row in connection.execute("PRAGMA table_info(protected_branch)")]
+            return sha256_json(rows)
+        finally:
+            if own:
+                connection.close()
+
+    def _check_schema(self, connection: sqlite3.Connection) -> None:
+        if not isinstance(self.expected_schema_sha256, str) or len(self.expected_schema_sha256) != 64:
+            raise RuntimeError("GITEA_SCHEMA_FINGERPRINT_REQUIRED")
+        if self.schema_sha256(connection) != self.expected_schema_sha256:
+            raise RuntimeError("GITEA_SCHEMA_FINGERPRINT_MISMATCH")
+
+    def _read(self, connection: sqlite3.Connection) -> dict[str, Any]:
+        row = connection.execute(
+            "SELECT * FROM protected_branch WHERE repo_id=? AND branch_name=?",
+            (self.repo_id, self.branch),
+        ).fetchone()
+        if row is None:
+            raise RuntimeError("PROTECTED_BRANCH_ROW_NOT_FOUND")
+        return dict(row)
+
+    def read_current(self) -> dict[str, Any]:
+        with self._connect() as connection:
+            self._check_schema(connection)
+            return self._read(connection)
+
+    @staticmethod
+    def calculate_snapshot_sha256(snapshot: Mapping[str, Any]) -> str:
+        return calculate_snapshot_sha256(snapshot)
+
+    def compare_expected(self, expected_sha256: str) -> bool:
+        return calculate_snapshot_sha256(self.read_current()) == expected_sha256
+
+    def apply_atomic(
+        self, expected_sha256: str, target: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        connection = self._connect()
+        try:
+            connection.execute("BEGIN IMMEDIATE")
+            self._check_schema(connection)
+            previous = self._read(connection)
+            if calculate_snapshot_sha256(previous) != expected_sha256:
+                raise CASConflict("CAS_CONFLICT")
+            validate_target(previous, target)
+            values = dict(target)
+            values["updated_unix"] = int(values.get("updated_unix") or time.time())
+            assignments = ",".join(f"{field}=?" for field in self.ALLOWED_UPDATE_FIELDS)
+            parameters = [values[field] for field in self.ALLOWED_UPDATE_FIELDS]
+            parameters.extend((self.repo_id, self.branch))
+            cursor = connection.execute(
+                f"UPDATE protected_branch SET {assignments} WHERE repo_id=? AND branch_name=?",
+                parameters,
+            )
+            if cursor.rowcount != 1:
+                raise CASConflict("CAS_ROWCOUNT_INVALID")
+            applied = self._read(connection)
+            expected_target = dict(target)
+            expected_target["updated_unix"] = values["updated_unix"]
+            if calculate_snapshot_sha256(applied) != calculate_snapshot_sha256(expected_target):
+                raise CASConflict("CAS_VERIFY_FAILED")
+            connection.commit()
+            return applied
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
+    def verify_applied(self, expected_sha256: str) -> bool:
+        return self.compare_expected(expected_sha256)
+
+    def rollback_atomic(
+        self, expected_target_sha256: str, previous: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        connection = self._connect()
+        try:
+            connection.execute("BEGIN IMMEDIATE")
+            self._check_schema(connection)
+            current = self._read(connection)
+            if calculate_snapshot_sha256(current) != expected_target_sha256:
+                raise CASConflict("ROLLBACK_CONFLICT")
+            for field in ("id", "repo_id", "branch_name"):
+                if current.get(field) != previous.get(field):
+                    raise CASConflict("ROLLBACK_IDENTITY_MISMATCH")
+            assignments = ",".join(f"{field}=?" for field in self.ALLOWED_UPDATE_FIELDS)
+            parameters = [previous[field] for field in self.ALLOWED_UPDATE_FIELDS]
+            parameters.extend((self.repo_id, self.branch))
+            cursor = connection.execute(
+                f"UPDATE protected_branch SET {assignments} WHERE repo_id=? AND branch_name=?",
+                parameters,
+            )
+            if cursor.rowcount != 1:
+                raise CASConflict("ROLLBACK_ROWCOUNT_INVALID")
+            restored = self._read(connection)
+            if calculate_snapshot_sha256(restored) != calculate_snapshot_sha256(previous):
+                raise CASConflict("ROLLBACK_VERIFY_FAILED")
+            connection.commit()
+            return restored
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
+    def verify_rollback(self, expected_sha256: str) -> bool:
+        return self.compare_expected(expected_sha256)

services/aios-single-owner-governance/aios_single_owner_governance/journal.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import tempfile
+from typing import Any, Iterable
+
+
+SENSITIVE_FRAGMENTS = (
+    "authorization",
+    "cookie",
+    "credential",
+    "password",
+    "private_key",
+    "secret",
+    "token",
+)
+
+
+def canonical_json(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def sha256_json(value: object) -> str:
+    return hashlib.sha256(canonical_json(value)).hexdigest()
+
+
+def redact(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            str(key): (
+                "[REDACTED]"
+                if any(fragment in str(key).lower() for fragment in SENSITIVE_FRAGMENTS)
+                else redact(item)
+            )
+            for key, item in value.items()
+        }
+    if isinstance(value, list):
+        return [redact(item) for item in value]
+    if isinstance(value, tuple):
+        return [redact(item) for item in value]
+    return value
+
+
+def ensure_state_directory(path: Path) -> None:
+    if path.is_symlink():
+        raise RuntimeError("STATE_DIRECTORY_SYMLINK_DENIED")
+    path.mkdir(parents=True, exist_ok=True, mode=0o750)
+    path.chmod(0o750)
+    mode = path.stat().st_mode & 0o777
+    if mode != 0o750:
+        raise RuntimeError("STATE_DIRECTORY_MODE_INVALID")
+
+
+def atomic_write_json(path: Path, value: object) -> None:
+    ensure_state_directory(path.parent)
+    raw = json.dumps(redact(value), sort_keys=True, indent=2).encode("utf-8") + b"\n"
+    with tempfile.NamedTemporaryFile(
+        dir=path.parent, prefix=f".{path.name}.", delete=False
+    ) as handle:
+        temporary = Path(handle.name)
+        handle.write(raw)
+        handle.flush()
+        os.fsync(handle.fileno())
+    temporary.chmod(0o640)
+    os.replace(temporary, path)
+    path.chmod(0o640)
+    directory_fd = os.open(path.parent, os.O_RDONLY)
+    try:
+        os.fsync(directory_fd)
+    finally:
+        os.close(directory_fd)
+
+
+def read_json(path: Path) -> dict[str, Any]:
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(raw, dict):
+        raise RuntimeError("STATE_DOCUMENT_NOT_OBJECT")
+    return raw
+
+
+class AuditChain:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    def _records(self) -> list[dict[str, Any]]:
+        if not self.path.exists():
+            return []
+        records: list[dict[str, Any]] = []
+        for line in self.path.read_text(encoding="utf-8").splitlines():
+            if line.strip():
+                record = json.loads(line)
+                if not isinstance(record, dict):
+                    raise RuntimeError("AUDIT_RECORD_INVALID")
+                records.append(record)
+        return records
+
+    def verify(self) -> str:
+        previous = "0" * 64
+        for index, record in enumerate(self._records(), 1):
+            payload = record.get("payload")
+            if record.get("sequence") != index:
+                raise RuntimeError("AUDIT_SEQUENCE_INVALID")
+            if record.get("previous_event_hash") != previous:
+                raise RuntimeError("AUDIT_PREVIOUS_HASH_INVALID")
+            payload_hash = sha256_json(payload)
+            if record.get("payload_hash") != payload_hash:
+                raise RuntimeError("AUDIT_PAYLOAD_HASH_INVALID")
+            material = {
+                "sequence": index,
+                "previous_event_hash": previous,
+                "payload_hash": payload_hash,
+                "transaction_id": record.get("transaction_id"),
+            }
+            event_hash = sha256_json(material)
+            if record.get("event_hash") != event_hash:
+                raise RuntimeError("AUDIT_EVENT_HASH_INVALID")
+            previous = event_hash
+        return previous
+
+    def append(self, transaction_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        ensure_state_directory(self.path.parent)
+        previous = self.verify()
+        sequence = len(self._records()) + 1
+        clean_payload = redact(payload)
+        payload_hash = sha256_json(clean_payload)
+        material = {
+            "sequence": sequence,
+            "previous_event_hash": previous,
+            "payload_hash": payload_hash,
+            "transaction_id": transaction_id,
+        }
+        record = {
+            **material,
+            "payload": clean_payload,
+            "event_hash": sha256_json(material),
+        }
+        fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o640)
+        try:
+            os.write(fd, canonical_json(record) + b"\n")
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+        self.path.chmod(0o640)
+        return record
+
+    def has_event(self, transaction_id: str, event: str) -> bool:
+        self.verify()
+        return any(
+            record.get("transaction_id") == transaction_id
+            and isinstance(record.get("payload"), dict)
+            and record["payload"].get("event") == event
+            for record in self._records()
+        )
+
+
+class JournalStore:
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.wal = root / "wal"
+        self.receipts = root / "receipts"
+        ensure_state_directory(root)
+        ensure_state_directory(self.wal)
+        ensure_state_directory(self.receipts)
+        self.audit = AuditChain(root / "audit.jsonl")
+
+    def transaction_path(self, operation_id: str) -> Path:
+        if not operation_id or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for char in operation_id):
+            raise RuntimeError("OPERATION_ID_INVALID")
+        return self.wal / f"{operation_id}.json"
+
+    def receipt_path(self, operation_id: str) -> Path:
+        return self.receipts / f"{operation_id}.json"
+
+    def load(self, operation_id: str) -> dict[str, Any] | None:
+        path = self.transaction_path(operation_id)
+        return read_json(path) if path.exists() else None
+
+    def write(self, operation_id: str, value: dict[str, Any]) -> None:
+        atomic_write_json(self.transaction_path(operation_id), value)
+
+    def write_receipt(self, operation_id: str, value: dict[str, Any]) -> None:
+        atomic_write_json(self.receipt_path(operation_id), value)
+
+    def iter_transactions(self) -> Iterable[dict[str, Any]]:
+        for path in sorted(self.wal.glob("*.json")):
+            yield read_json(path)

services/aios-single-owner-governance/aios_single_owner_governance/lock.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+try:
+    import fcntl
+except ImportError:  # pragma: no cover - exercised by Windows validation
+    fcntl = None
+    import msvcrt
+
+
+class MaintenanceLock:
+    """Process fence used only by fixture and later approved activation executors."""
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self._handle = None
+
+    def __enter__(self) -> "MaintenanceLock":
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._handle = self.path.open("a+")
+        if fcntl is not None:
+            try:
+                fcntl.flock(self._handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                self._handle.close()
+                self._handle = None
+                raise
+        else:
+            self._handle.seek(0)
+            try:
+                msvcrt.locking(self._handle.fileno(), msvcrt.LK_NBLCK, 1)
+            except OSError as exc:
+                self._handle.close()
+                self._handle = None
+                raise BlockingIOError("maintenance lock held") from exc
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        if self._handle is not None:
+            if fcntl is not None:
+                fcntl.flock(self._handle.fileno(), fcntl.LOCK_UN)
+            else:
+                self._handle.seek(0)
+                msvcrt.locking(self._handle.fileno(), msvcrt.LK_UNLCK, 1)
+            self._handle.close()
+            self._handle = None

services/aios-single-owner-governance/aios_single_owner_governance/models.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Mapping
+
+
+class GovernanceState(str, Enum):
+    HUMAN_OWNER_APPROVAL_PENDING = "HUMAN_OWNER_APPROVAL_PENDING"
+    OWNER_APPROVAL_VERIFIED = "OWNER_APPROVAL_VERIFIED"
+    MACHINE_GATES_VERIFIED = "MACHINE_GATES_VERIFIED"
+    AUTO_MERGE_READY = "AUTO_MERGE_READY"
+    OWNER_APPROVED_EXECUTION_READY = "OWNER_APPROVED_EXECUTION_READY"
+
+
+@dataclass(frozen=True)
+class Review:
+    reviewer: str
+    reviewer_id: int
+    state: str
+    head_oid: str
+    reviewed_at: datetime
+    stale: bool = False
+    dismissed: bool = False
+
+    def __post_init__(self) -> None:
+        if self.reviewed_at.tzinfo is None:
+            raise ValueError("reviewed_at must be timezone-aware")
+
+
+@dataclass(frozen=True)
+class MachineGate:
+    context: str
+    state: str
+    head_oid: str
+    observed_at: datetime
+
+    def __post_init__(self) -> None:
+        if self.observed_at.tzinfo is None:
+            raise ValueError("observed_at must be timezone-aware")
+
+
+@dataclass(frozen=True)
+class ApprovalContext:
+    current_head_oid: str
+    requester: str
+    executor: str
+    owner: str
+    owner_id: int
+    reviews: tuple[Review, ...]
+    machine_gates: tuple[MachineGate, ...]
+    required_machine_contexts: tuple[str, ...]
+    expires_at: datetime
+    wip: bool
+    now: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+
+    def __post_init__(self) -> None:
+        if self.expires_at.tzinfo is None or self.now.tzinfo is None:
+            raise ValueError("approval timestamps must be timezone-aware")
+
+
+@dataclass(frozen=True)
+class ProtectionSnapshot:
+    branch: str
+    required_approvals: int
+    approval_whitelist: tuple[str, ...]
+    merge_whitelist: tuple[str, ...]
+    dismiss_stale_approvals: bool
+    require_status_checks: bool
+    status_contexts: tuple[str, ...]
+    direct_push: bool
+    force_push: bool
+    branch_deletion: bool
+    version: int
+
+    def as_dict(self) -> Mapping[str, Any]:
+        return {
+            "branch": self.branch,
+            "required_approvals": self.required_approvals,
+            "approval_whitelist": list(self.approval_whitelist),
+            "merge_whitelist": list(self.merge_whitelist),
+            "dismiss_stale_approvals": self.dismiss_stale_approvals,
+            "require_status_checks": self.require_status_checks,
+            "status_contexts": list(self.status_contexts),
+            "direct_push": self.direct_push,
+            "force_push": self.force_push,
+            "branch_deletion": self.branch_deletion,
+            "version": self.version,
+        }

services/aios-single-owner-governance/aios_single_owner_governance/policy.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+
+class OperationDenied(PermissionError):
+    pass
+
+
+class DormantOperationGuard:
+    """Fail-closed operation boundary for the prerequisite deployment."""
+
+    DENIED = frozenset({
+        "protected_branch.direct_push",
+        "governance.direct_push",
+        "tag.direct_write",
+        "git.force_push",
+        "production.without_owner",
+        "branch_protection.live_update",
+        "single_owner_policy.activate",
+    })
+
+    def authorize(self, operation: str, *, fixture: bool = False) -> None:
+        if operation in self.DENIED or not fixture:
+            raise OperationDenied(operation)

services/aios-single-owner-governance/aios_single_owner_governance/protection.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+from typing import Protocol
+
+from .models import ProtectionSnapshot
+
+
+class CASConflict(RuntimeError):
+    pass
+
+
+class DormantViolation(RuntimeError):
+    pass
+
+
+def snapshot_sha256(snapshot: ProtectionSnapshot) -> str:
+    raw = json.dumps(snapshot.as_dict(), sort_keys=True, separators=(",", ":")).encode()
+    return hashlib.sha256(raw).hexdigest()
+
+
+class ProtectionStore(Protocol):
+    fixture: bool
+
+    def read(self, branch: str) -> ProtectionSnapshot: ...
+    def compare_and_swap(
+        self, branch: str, expected_sha256: str, target: ProtectionSnapshot
+    ) -> bool: ...
+
+
+@dataclass(frozen=True)
+class ProtectionPlan:
+    branch: str
+    expected_previous_sha256: str
+    previous: ProtectionSnapshot
+    target: ProtectionSnapshot
+    target_sha256: str
+
+
+class BranchProtectionCASMigrationAdapter:
+    """Candidate adapter. Mutations are restricted to disposable fixtures."""
+
+    def plan(
+        self,
+        current: ProtectionSnapshot,
+        target: ProtectionSnapshot,
+        expected_previous_sha256: str,
+    ) -> ProtectionPlan:
+        actual = snapshot_sha256(current)
+        if actual != expected_previous_sha256:
+            raise CASConflict("EXPECTED_PREVIOUS_SHA256_MISMATCH")
+        if current.branch != target.branch:
+            raise CASConflict("BRANCH_BINDING_MISMATCH")
+        if target.direct_push or target.force_push or target.branch_deletion:
+            raise DormantViolation("UNSAFE_TARGET_POLICY")
+        return ProtectionPlan(
+            current.branch, actual, current, target, snapshot_sha256(target)
+        )
+
+    def apply_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
+        if not store.fixture:
+            raise DormantViolation("LIVE_PROTECTION_MUTATION_DENIED")
+        if not store.compare_and_swap(
+            plan.branch, plan.expected_previous_sha256, plan.target
+        ):
+            raise CASConflict("CONCURRENT_PROTECTION_CHANGE")
+        return store.read(plan.branch)
+
+    def rollback_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
+        if not store.fixture:
+            raise DormantViolation("LIVE_PROTECTION_ROLLBACK_DENIED")
+        if not store.compare_and_swap(plan.branch, plan.target_sha256, plan.previous):
+            raise CASConflict("ROLLBACK_CONFLICT")
+        return store.read(plan.branch)
+
+
+class MemoryProtectionStore:
+    fixture = True
+
+    def __init__(self, snapshot: ProtectionSnapshot) -> None:
+        self._snapshot = snapshot
+
+    def read(self, branch: str) -> ProtectionSnapshot:
+        if self._snapshot.branch != branch:
+            raise KeyError(branch)
+        return self._snapshot
+
+    def compare_and_swap(
+        self, branch: str, expected_sha256: str, target: ProtectionSnapshot
+    ) -> bool:
+        if self._snapshot.branch != branch:
+            return False
+        if snapshot_sha256(self._snapshot) != expected_sha256:
+            return False
+        self._snapshot = target
+        return True

services/aios-single-owner-governance/aios_single_owner_governance/rollback.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+
+from .models import ProtectionSnapshot
+from .protection import MemoryProtectionStore, ProtectionPlan, snapshot_sha256
+
+
+@dataclass(frozen=True)
+class RollbackConfiguration:
+    branch: str
+    previous_sha256: str
+    target_sha256: str
+    payload: dict
+
+
+class RollbackConfigurationManager:
+    def build(self, plan: ProtectionPlan) -> RollbackConfiguration:
+        return RollbackConfiguration(
+            plan.branch,
+            plan.expected_previous_sha256,
+            plan.target_sha256,
+            {"previous": plan.previous.as_dict(), "target": plan.target.as_dict()},
+        )
+
+    def install_fixture(self, root: Path, config: RollbackConfiguration) -> Path:
+        if not root.name.startswith("aios-governance-fixture-"):
+            raise PermissionError("rollback configuration may only be installed in fixture root")
+        root.mkdir(parents=True, exist_ok=True)
+        destination = root / "rollback-configuration.json"
+        raw = json.dumps(config.payload, sort_keys=True, indent=2).encode() + b"\n"
+        with tempfile.NamedTemporaryFile(dir=root, prefix=".rollback.", delete=False) as handle:
+            handle.write(raw)
+            handle.flush()
+            temporary = Path(handle.name)
+        temporary.chmod(0o600)
+        temporary.replace(destination)
+        return destination
+
+    def verify(self, config: RollbackConfiguration, plan: ProtectionPlan) -> bool:
+        return (
+            config.previous_sha256 == snapshot_sha256(plan.previous)
+            and config.target_sha256 == snapshot_sha256(plan.target)
+        )
+
+
+@dataclass
+class ProtectionCompensationSaga:
+    adapter: object
+    store: MemoryProtectionStore
+    plan: ProtectionPlan
+    applied: bool = False
+    rolled_back: bool = False
+
+    def execute_fixture(self) -> ProtectionSnapshot:
+        result = self.adapter.apply_fixture(self.store, self.plan)
+        self.applied = True
+        return result
+
+    def compensate_fixture(self) -> ProtectionSnapshot:
+        if not self.applied:
+            return self.store.read(self.plan.branch)
+        result = self.adapter.rollback_fixture(self.store, self.plan)
+        self.rolled_back = True
+        return result

services/aios-single-owner-governance/aios_single_owner_governance/service.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import tempfile
+
+
+def runtime_state(component: str, prior_invocations: int = 0) -> dict:
+    return {
+        "component": component,
+        "deployment_mode": "DORMANT_SHADOW_ONLY",
+        "enabled": False,
+        "active": False,
+        "production_routing": False,
+        "production_mutation": False,
+        "global_gitops_writes_enabled": False,
+        "prior_invocations": prior_invocations,
+        "restart_recovery": prior_invocations > 0,
+    }
+
+
+def checkpoint_state(component: str, root: Path) -> dict:
+    root.mkdir(parents=True, exist_ok=True)
+    destination = root / f"{component}.json"
+    prior = 0
+    if destination.exists():
+        previous = json.loads(destination.read_text(encoding="utf-8"))
+        if previous.get("component") != component or previous.get("deployment_mode") != "DORMANT_SHADOW_ONLY":
+            raise RuntimeError("checkpoint identity mismatch")
+        prior = int(previous.get("invocations", 0))
+    state = runtime_state(component, prior)
+    checkpoint = {**state, "invocations": prior + 1}
+    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=root, delete=False) as handle:
+        json.dump(checkpoint, handle, sort_keys=True)
+        handle.write("\n")
+        handle.flush()
+        temporary = Path(handle.name)
+    temporary.replace(destination)
+    return state
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--component", required=True, choices=("owner-review-adapter", "machine-status-producer"))
+    parser.add_argument("--self-check", action="store_true")
+    parser.add_argument("--activate", action="store_true")
+    parser.add_argument("--checkpoint-dir", type=Path)
+    args = parser.parse_args(argv)
+    if args.activate:
+        parser.error("activation is not authorized by the prerequisites candidate")
+    if os.environ.get("AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED", "false").lower() != "false":
+        parser.error("candidate must remain disabled")
+    if os.environ.get("AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE", "false").lower() != "false":
+        parser.error("activation must remain inactive")
+    if os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower() != "false":
+        parser.error("production mutation must remain disabled")
+    state = (
+        checkpoint_state(args.component, args.checkpoint_dir)
+        if args.checkpoint_dir is not None
+        else runtime_state(args.component)
+    )
+    print(json.dumps(state, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-single-owner-governance/aios_single_owner_governance/state_machine.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+from pathlib import Path
+import tempfile
+
+from .models import GovernanceState
+
+
+class InvalidTransition(RuntimeError):
+    pass
+
+
+_NEXT = {
+    GovernanceState.HUMAN_OWNER_APPROVAL_PENDING: GovernanceState.OWNER_APPROVAL_VERIFIED,
+    GovernanceState.OWNER_APPROVAL_VERIFIED: GovernanceState.MACHINE_GATES_VERIFIED,
+    GovernanceState.MACHINE_GATES_VERIFIED: GovernanceState.AUTO_MERGE_READY,
+    GovernanceState.AUTO_MERGE_READY: GovernanceState.OWNER_APPROVED_EXECUTION_READY,
+}
+
+
+@dataclass
+class GovernanceStateMachine:
+    state: GovernanceState = GovernanceState.HUMAN_OWNER_APPROVAL_PENDING
+
+    def transition(self, target: GovernanceState) -> GovernanceState:
+        if _NEXT.get(self.state) != target:
+            raise InvalidTransition(f"{self.state.value}->{target.value}")
+        self.state = target
+        return self.state
+
+    def restore(self, checkpoint: GovernanceState) -> None:
+        if checkpoint not in GovernanceState:
+            raise InvalidTransition("unknown checkpoint")
+        self.state = checkpoint
+
+    def save_fixture(self, path: Path) -> None:
+        if not any(part.startswith("aios-governance-fixture-") for part in path.resolve().parts):
+            raise PermissionError("state checkpoint writes are fixture-only")
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
+            json.dump({"state": self.state.value}, handle, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            temporary = Path(handle.name)
+        temporary.replace(path)
+
+    @classmethod
+    def load_fixture(cls, path: Path) -> "GovernanceStateMachine":
+        if not any(part.startswith("aios-governance-fixture-") for part in path.resolve().parts):
+            raise PermissionError("state checkpoint reads are fixture-only")
+        raw = json.loads(path.read_text(encoding="utf-8"))
+        return cls(GovernanceState(raw["state"]))

services/aios-single-owner-governance/aios_single_owner_governance/status.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+
+class DormantStatusProducer(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class StatusRecord:
+    context: str
+    state: str
+    head_oid: str
+    observed_at: str
+
+
+class MachineStatusProducer:
+    def __init__(self, *, enabled: bool = False, fixture: bool = False) -> None:
+        self.enabled = enabled
+        self.fixture = fixture
+
+    def produce(self, context: str, state: str, head_oid: str) -> StatusRecord:
+        if not self.fixture:
+            raise DormantStatusProducer("LIVE_STATUS_PUBLICATION_DENIED")
+        if self.enabled:
+            raise DormantStatusProducer("DORMANT_CANDIDATE_MUST_NOT_BE_ENABLED")
+        if context not in {
+            "aios/machine-gates", "aios/secret-scan", "aios/evidence-closure"
+        }:
+            raise ValueError("unknown status context")
+        if state not in {"success", "failure"} or len(head_oid) != 40:
+            raise ValueError("invalid status record")
+        return StatusRecord(
+            context, state, head_oid,
+            datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+        )

services/aios-single-owner-governance/aios_single_owner_governance/validator.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .models import ApprovalContext, Review
+
+
+class ValidationDenied(RuntimeError):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
+@dataclass(frozen=True)
+class ValidationResult:
+    owner_review: Review
+    machine_contexts: tuple[str, ...]
+
+
+class SingleOwnerApprovalValidator:
+    """Pure candidate validator. It has no Gitea or production write path."""
+
+    def validate(self, context: ApprovalContext) -> ValidationResult:
+        if context.now >= context.expires_at:
+            raise ValidationDenied("APPROVAL_EXPIRED")
+        if context.wip:
+            raise ValidationDenied("WIP_BLOCKED")
+        if context.requester == context.owner:
+            raise ValidationDenied("REQUESTER_SELF_APPROVAL")
+        if context.executor == context.owner:
+            raise ValidationDenied("EXECUTOR_SELF_APPROVAL")
+
+        approved = [review for review in context.reviews if review.state == "APPROVED"]
+        owner_reviews = [
+            review for review in approved
+            if review.reviewer == context.owner and review.reviewer_id == context.owner_id
+        ]
+        if len(owner_reviews) != 1:
+            raise ValidationDenied("OWNER_APPROVAL_REQUIRED")
+        owner_review = owner_reviews[0]
+        if owner_review.stale:
+            raise ValidationDenied("STALE_OWNER_APPROVAL")
+        if owner_review.dismissed:
+            raise ValidationDenied("DISMISSED_OWNER_APPROVAL")
+        if owner_review.head_oid != context.current_head_oid:
+            raise ValidationDenied("OWNER_APPROVAL_NOT_LATEST_HEAD")
+        if owner_review.reviewed_at >= context.expires_at:
+            raise ValidationDenied("OWNER_APPROVAL_EXPIRED")
+        if any(
+            review.state == "APPROVED"
+            and review.reviewer != context.owner
+            and review.reviewer_id != context.owner_id
+            for review in context.reviews
+        ):
+            # Non-owner approvals may exist for legacy mechanics, but never
+            # satisfy or replace the human authorization decision.
+            pass
+
+        gates = {gate.context: gate for gate in context.machine_gates}
+        if set(gates) != set(context.required_machine_contexts):
+            raise ValidationDenied("MACHINE_GATE_SET_MISMATCH")
+        for gate in gates.values():
+            if gate.state != "success":
+                raise ValidationDenied("MACHINE_GATE_FAILED")
+            if gate.head_oid != context.current_head_oid:
+                raise ValidationDenied("MACHINE_GATE_NOT_LATEST_HEAD")
+            if gate.observed_at >= context.expires_at:
+                raise ValidationDenied("MACHINE_GATE_EXPIRED")
+        return ValidationResult(owner_review, tuple(sorted(gates)))

services/aios-single-owner-governance/pyproject.toml

@@ -0,0 +1,16 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "aios-single-owner-governance"
+version = "0.1.0"
+requires-python = ">=3.11"
+dependencies = []
+
+[project.scripts]
+aios-single-owner-governance = "aios_single_owner_governance.activation:main"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["aios_single_owner_governance*"]

tests/single_owner_governance/__init__.py

@@ -0,0 +1 @@
+

tests/single_owner_governance/helpers.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from aios_single_owner_governance.models import (
+    ApprovalContext,
+    MachineGate,
+    ProtectionSnapshot,
+    Review,
+)
+
+HEAD = "a" * 40
+NOW = datetime(2026, 7, 28, 19, 0, tzinfo=timezone.utc)
+CONTEXTS = ("aios/machine-gates", "aios/secret-scan", "aios/evidence-closure")
+
+
+def owner_review(**changes) -> Review:
+    values = dict(
+        reviewer="owner", reviewer_id=2, state="APPROVED", head_oid=HEAD,
+        reviewed_at=NOW - timedelta(minutes=1), stale=False, dismissed=False,
+    )
+    values.update(changes)
+    return Review(**values)
+
+
+def gates(*, state="success", head_oid=HEAD):
+    return tuple(MachineGate(name, state, head_oid, NOW) for name in CONTEXTS)
+
+
+def context(**changes) -> ApprovalContext:
+    values = dict(
+        current_head_oid=HEAD, requester="agent", executor="merge-executor",
+        owner="owner", owner_id=2, reviews=(owner_review(),), machine_gates=gates(),
+        required_machine_contexts=CONTEXTS, expires_at=NOW + timedelta(hours=1),
+        wip=False, now=NOW,
+    )
+    values.update(changes)
+    return ApprovalContext(**values)
+
+
+def legacy_protection(version=1) -> ProtectionSnapshot:
+    return ProtectionSnapshot(
+        "governance/approvals", 2, ("codex", "gpt"), ("codex", "gpt"),
+        True, False, (), False, False, False, version,
+    )
+
+def single_owner_protection(version=2) -> ProtectionSnapshot:
+    return ProtectionSnapshot(
+        "governance/approvals", 1, ("owner",), ("owner",), True, True,
+        CONTEXTS, False, False, False, version,
+    )

tests/single_owner_governance/test_dormant_invariants.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from pathlib import Path
+import unittest
+
+from aios_single_owner_governance.fixture import DisposableGiteaFixture
+from aios_single_owner_governance.protection import (
+    BranchProtectionCASMigrationAdapter,
+    DormantViolation,
+    snapshot_sha256,
+)
+from aios_single_owner_governance.policy import DormantOperationGuard, OperationDenied
+
+from .helpers import legacy_protection, single_owner_protection
+
+
+class LiveStore:
+    fixture = False
+
+    def read(self, branch):
+        return legacy_protection()
+
+    def compare_and_swap(self, branch, expected_sha256, target):
+        raise AssertionError("live compare_and_swap must never be invoked")
+
+
+class DormantInvariantTests(unittest.TestCase):
+    def test_live_protection_apply_and_rollback_are_denied(self):
+        adapter = BranchProtectionCASMigrationAdapter()
+        old, new = legacy_protection(), single_owner_protection()
+        plan = adapter.plan(old, new, snapshot_sha256(old))
+        with self.assertRaises(DormantViolation):
+            adapter.apply_fixture(LiveStore(), plan)
+        with self.assertRaises(DormantViolation):
+            adapter.rollback_fixture(LiveStore(), plan)
+
+    def test_unsafe_policy_cannot_be_planned(self):
+        adapter = BranchProtectionCASMigrationAdapter()
+        old = legacy_protection()
+        for unsafe in (
+            single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "direct_push": True}),
+            single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "force_push": True}),
+            single_owner_protection().__class__(**{**single_owner_protection().as_dict(), "approval_whitelist": ("owner",), "merge_whitelist": ("owner",), "status_contexts": tuple(single_owner_protection().status_contexts), "branch_deletion": True}),
+        ):
+            with self.assertRaises(DormantViolation):
+                adapter.plan(old, unsafe, snapshot_sha256(old))
+
+    def test_disposable_fixture_is_destroyed(self):
+        fixture = DisposableGiteaFixture(legacy_protection())
+        root = fixture.root
+        with fixture:
+            self.assertTrue(root.exists())
+        self.assertTrue(fixture.destroyed)
+        self.assertFalse(root.exists())
+
+    def test_permanent_and_pre_activation_writes_are_denied(self):
+        guard = DormantOperationGuard()
+        for operation in (
+            "protected_branch.direct_push",
+            "governance.direct_push",
+            "tag.direct_write",
+            "git.force_push",
+            "production.without_owner",
+            "branch_protection.live_update",
+            "single_owner_policy.activate",
+        ):
+            with self.subTest(operation=operation), self.assertRaises(OperationDenied):
+                guard.authorize(operation)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/single_owner_governance/test_evidence_status_service.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from aios_single_owner_governance.evidence import (
+    ActivationEvidenceClosureBuilder,
+    ApprovalMirrorStatusVerifier,
+)
+from aios_single_owner_governance.service import main, runtime_state
+from aios_single_owner_governance.status import (
+    DormantStatusProducer,
+    MachineStatusProducer,
+)
+
+from .helpers import HEAD
+
+
+class EvidenceStatusServiceTests(unittest.TestCase):
+    def test_complete_evidence_set_builds_canonical_closure(self):
+        evidence = {
+            "activation_prerequisites_commit_oid": "a" * 40,
+            "activation_prerequisites_tree_oid": "b" * 40,
+            "artifact_sha256": "c" * 64,
+            "rollback_artifact_sha256": "d" * 64,
+            "validator_sha256": "e" * 64,
+            "state_machine_sha256": "f" * 64,
+            "branch_protection_adapter_sha256": "0" * 64,
+            "rollback_configuration_sha256": "1" * 64,
+            "test_evidence_sha256": "2" * 64,
+            "deployment_manifest_sha256": "3" * 64,
+            "dormant_unit_manifest_sha256": "4" * 64,
+            "dormant_deployment_evidence_sha256": "5" * 64,
+        }
+        closure = ActivationEvidenceClosureBuilder().build(evidence)
+        self.assertEqual(len(closure.closure_sha256), 64)
+        with self.assertRaises(ValueError):
+            ActivationEvidenceClosureBuilder().build({})
+
+    def test_mirror_verifier_requires_all_four_refs_and_fsck(self):
+        verifier = ApprovalMirrorStatusVerifier()
+        self.assertTrue(verifier.verify(
+            HEAD, gitea_oid=HEAD, quarantine_oid=HEAD,
+            authoritative_oid=HEAD, controller_visible_oid=HEAD,
+            fsck_connectivity="PASS",
+        ))
+        self.assertFalse(verifier.verify(
+            HEAD, gitea_oid=HEAD, quarantine_oid=HEAD,
+            authoritative_oid="b" * 40, controller_visible_oid=HEAD,
+            fsck_connectivity="PASS",
+        ))
+
+    def test_machine_status_is_fixture_only_and_dormant(self):
+        with self.assertRaises(DormantStatusProducer):
+            MachineStatusProducer().produce("aios/machine-gates", "success", HEAD)
+        with self.assertRaises(DormantStatusProducer):
+            MachineStatusProducer(enabled=True, fixture=True).produce(
+                "aios/machine-gates", "success", HEAD
+            )
+        record = MachineStatusProducer(fixture=True).produce(
+            "aios/evidence-closure", "success", HEAD
+        )
+        self.assertEqual(record.head_oid, HEAD)
+
+    def test_service_self_check_is_dormant_and_activate_is_rejected(self):
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            self.assertEqual(main(["--component", "owner-review-adapter", "--self-check"]), 0)
+        state = json.loads(output.getvalue())
+        self.assertFalse(state["active"])
+        self.assertFalse(state["production_mutation"])
+        with contextlib.redirect_stderr(io.StringIO()):
+            with self.assertRaises(SystemExit):
+                main(["--component", "owner-review-adapter", "--activate"])
+
+    def test_write_flags_cannot_be_enabled(self):
+        for name in (
+            "AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED",
+            "AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE",
+            "AIOS_GITOPS_WRITES_ENABLED",
+        ):
+            with self.subTest(name=name), patch.dict(os.environ, {name: "true"}, clear=False):
+                with contextlib.redirect_stderr(io.StringIO()):
+                    with self.assertRaises(SystemExit):
+                        main(["--component", "machine-status-producer", "--self-check"])
+
+    def test_service_entrypoint_restart_recovery_across_processes(self):
+        with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
+            command = [
+                sys.executable, "-m", "aios_single_owner_governance.service",
+                "--component", "owner-review-adapter", "--self-check",
+                "--checkpoint-dir", root,
+            ]
+            first = subprocess.run(command, text=True, capture_output=True, check=True)
+            second = subprocess.run(command, text=True, capture_output=True, check=True)
+            first_state = json.loads(first.stdout)
+            second_state = json.loads(second.stdout)
+            self.assertFalse(first_state["restart_recovery"])
+            self.assertTrue(second_state["restart_recovery"])
+            self.assertEqual(second_state["prior_invocations"], 1)
+            self.assertFalse(second_state["production_mutation"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/single_owner_governance/test_live_activation.py

@@ -0,0 +1,771 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sqlite3
+import subprocess
+import hashlib
+import os
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from jsonschema import Draft202012Validator, ValidationError
+
+from aios_single_owner_governance.activation import (
+    ActivationEngine,
+    InjectedCrash,
+    _verify_approved_entities,
+    validate_bindings,
+)
+import aios_single_owner_governance.validator as validator_module
+import aios_single_owner_governance.state_machine as state_machine_module
+import aios_single_owner_governance.activation as activation_module
+from aios_single_owner_governance.gitea_protected_branch import (
+    CASConflict,
+    FileProtectedBranchAdapter,
+    SQLiteProtectedBranchAdapter,
+    UnsafePolicy,
+    calculate_snapshot_sha256,
+    validate_target,
+)
+from aios_single_owner_governance.journal import sha256_json
+
+
+MACHINE_CONTEXTS = {
+    "aios/validator",
+    "aios/state-machine",
+    "aios/branch-protection",
+    "aios/secret-scan",
+    "aios/evidence-closure",
+}
+
+
+def write_json(path: Path, value: object) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def legacy_row() -> dict:
+    return {
+        "id": 7,
+        "repo_id": 1,
+        "branch_name": "governance/approvals",
+        "can_push": 0,
+        "enable_whitelist": 0,
+        "whitelist_user_i_ds": "null",
+        "whitelist_team_i_ds": "null",
+        "enable_merge_whitelist": 1,
+        "whitelist_deploy_keys": 0,
+        "merge_whitelist_user_i_ds": "[42,43]",
+        "merge_whitelist_team_i_ds": "null",
+        "enable_status_check": 0,
+        "status_check_contexts": "[]",
+        "enable_approvals_whitelist": 1,
+        "approvals_whitelist_user_i_ds": "[42,43]",
+        "approvals_whitelist_team_i_ds": "null",
+        "required_approvals": 2,
+        "block_on_rejected_reviews": 1,
+        "block_on_official_review_requests": 1,
+        "block_on_outdated_branch": 1,
+        "dismiss_stale_approvals": 1,
+        "ignore_stale_approvals": 0,
+        "require_signed_commits": 0,
+        "protected_file_patterns": "",
+        "unprotected_file_patterns": "",
+        "created_unix": 1,
+        "updated_unix": 10,
+    }
+
+
+def owner_row() -> dict:
+    row = legacy_row()
+    row.update(
+        {
+            "approvals_whitelist_user_i_ds": "[2]",
+            "required_approvals": 1,
+            "enable_status_check": 1,
+            "status_check_contexts": json.dumps(sorted(MACHINE_CONTEXTS), separators=(",", ":")),
+            "updated_unix": 11,
+        }
+    )
+    return row
+
+
+def git_output(git_dir: Path, *args: str, stdin: str | None = None) -> str:
+    return subprocess.run(
+        ["git", f"--git-dir={git_dir}", *args], input=stdin, text=True,
+        capture_output=True, check=True,
+        env={
+            **__import__("os").environ,
+            "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.invalid",
+            "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.invalid",
+        },
+    ).stdout.strip()
+
+
+def commit_files(
+    git_dir: Path, files: dict[str, str], message: str, *, parent: str | None = None
+) -> tuple[str, str]:
+    root: dict[str, object] = {}
+    for path, raw in files.items():
+        node = root
+        parts = path.split("/")
+        for directory in parts[:-1]:
+            node = node.setdefault(directory, {})  # type: ignore[assignment]
+        node[parts[-1]] = git_output(git_dir, "hash-object", "-w", "--stdin", stdin=raw)
+
+    def build_tree(node: dict[str, object]) -> str:
+        lines = []
+        for name in sorted(node):
+            value = node[name]
+            if isinstance(value, dict):
+                lines.append(f"040000 tree {build_tree(value)}\t{name}")
+            else:
+                lines.append(f"100644 blob {value}\t{name}")
+        return git_output(git_dir, "mktree", stdin="\n".join(lines) + "\n")
+
+    tree = build_tree(root)
+    args = ["commit-tree", tree]
+    if parent is not None:
+        args.extend(("-p", parent))
+    return git_output(git_dir, *args, stdin=message + "\n"), tree
+
+
+def evidence_fixture(tmp_path: Path, execution_binding: dict) -> dict:
+    git_dir = tmp_path / "evidence.git"
+    subprocess.run(["git", "init", "--bare", str(git_dir)], capture_output=True, check=True)
+    package_root = Path(activation_module.__file__).resolve().parent
+    source_files: dict[str, str] = {}
+    runtime_sources = []
+    for name in (
+        "activation.py", "gitea_protected_branch.py", "journal.py",
+        "state_machine.py", "validator.py",
+    ):
+        source = package_root / name
+        raw = source.read_text(encoding="utf-8")
+        git_path = (
+            "services/aios-single-owner-governance/"
+            f"aios_single_owner_governance/{name}"
+        )
+        source_files[git_path] = raw
+        runtime_sources.append({
+            "git_path": git_path,
+            "runtime_path": str(source.resolve()),
+            "sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
+        })
+    implementation_oid, implementation_tree = commit_files(
+        git_dir, source_files, "implementation"
+    )
+    closure_path = "evidence/live-activation-closure.json"
+    closure_raw = '{"result":"PASS"}\n'
+    closure_oid, _ = commit_files(
+        git_dir, {**source_files, closure_path: closure_raw},
+        "closure", parent=implementation_oid
+    )
+    closure_sha = hashlib.sha256(closure_raw.encode()).hexdigest()
+    entities = {}
+    for name, raw in {
+        "artifact": b"approved-artifact\n",
+        "rollback_artifact": b"approved-rollback\n",
+        "test_evidence": b"approved-tests\n",
+    }.items():
+        path = tmp_path / "artifacts" / f"{name}.bin"
+        path.parent.mkdir(exist_ok=True)
+        path.write_bytes(raw)
+        entities[f"{name}_path"] = str(path.resolve())
+        entities[f"{name}_sha256"] = hashlib.sha256(raw).hexdigest()
+    head_oid = git_output(git_dir, "hash-object", "-w", "--stdin", stdin="head\n")
+    expiry = datetime.now(timezone.utc) + timedelta(hours=1)
+    producer_id = 99
+    producer_name = "trusted-machine-producer"
+    execution_binding = {
+        **execution_binding,
+        "evidence_closure_oid": closure_oid,
+        "evidence_closure_path": closure_path,
+        "evidence_closure_sha256": closure_sha,
+        "implementation_commit_oid": implementation_oid,
+        "implementation_tree_oid": implementation_tree,
+        "runtime_sources": runtime_sources,
+        "machine_status_producer_id": producer_id,
+        "machine_status_producer_name": producer_name,
+        **entities,
+    }
+    manifest = {
+        "approval_type": "SINGLE_OWNER_GOVERNANCE_LIVE_ACTIVATION_IMPLEMENTATION_AND_EXECUTION",
+        "expires_at": expiry.isoformat().replace("+00:00", "Z"),
+        "execution_binding": execution_binding,
+        "execution_binding_sha256": sha256_json(execution_binding),
+        **{key: execution_binding[key] for key in (
+            "implementation_commit_oid", "implementation_tree_oid", "runtime_sources",
+            "artifact_path", "artifact_sha256", "rollback_artifact_path",
+            "rollback_artifact_sha256", "test_evidence_path", "test_evidence_sha256",
+            "evidence_closure_oid", "evidence_closure_path", "evidence_closure_sha256",
+            "machine_status_producer_id", "machine_status_producer_name",
+        )},
+    }
+    manifest_raw = json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n"
+    manifest_blob = git_output(git_dir, "hash-object", "-w", "--stdin", stdin=manifest_raw)
+    leaf = git_output(git_dir, "mktree", stdin=f"100644 blob {manifest_blob}\tactivation.json\n")
+    approvals = git_output(git_dir, "mktree", stdin=f"040000 tree {leaf}\tsingle-owner\n")
+    tree = git_output(git_dir, "mktree", stdin=f"040000 tree {approvals}\tapprovals\n")
+    merge_oid = git_output(git_dir, "commit-tree", tree, stdin="approved\n")
+    git_output(git_dir, "update-ref", "refs/heads/governance/approvals", merge_oid)
+
+    db = tmp_path / "approval.db"
+    with sqlite3.connect(db) as connection:
+        connection.executescript(
+            "CREATE TABLE issue(id INTEGER,repo_id INTEGER,is_closed INTEGER);"
+            "CREATE TABLE pull_request(issue_id INTEGER,\"index\" INTEGER,base_branch TEXT,has_merged INTEGER,merged_commit_id TEXT,merged_unix INTEGER);"
+            "CREATE TABLE user(id INTEGER,lower_name TEXT);"
+            "CREATE TABLE review(id INTEGER,type INTEGER,reviewer_id INTEGER,issue_id INTEGER,commit_id TEXT,stale INTEGER,dismissed INTEGER,created_unix INTEGER);"
+            "CREATE TABLE commit_status(id INTEGER,repo_id INTEGER,state TEXT,sha TEXT,context TEXT,creator_id INTEGER);"
+        )
+        connection.execute("INSERT INTO issue VALUES(10,1,1)")
+        merged_unix = int(datetime.now(timezone.utc).timestamp())
+        connection.execute("INSERT INTO pull_request VALUES(10,31,'governance/approvals',1,?,?)", (merge_oid, merged_unix))
+        connection.execute("INSERT INTO user VALUES(2,'owner')")
+        connection.execute("INSERT INTO user VALUES(?,?)", (producer_id, producer_name))
+        connection.execute(
+            "INSERT INTO review VALUES(52,1,2,10,?,0,0,?)",
+            (head_oid, int(datetime.now(timezone.utc).timestamp())),
+        )
+        machine_checks = {}
+        for index, context in enumerate(sorted(MACHINE_CONTEXTS), 101):
+            connection.execute(
+                "INSERT INTO commit_status VALUES(?,1,'success',?,?,?)",
+                (index, head_oid, context, producer_id),
+            )
+            machine_checks[context] = {
+                "state": "success", "head_oid": head_oid, "status_id": index,
+                "creator_id": producer_id, "creator_login": producer_name,
+            }
+    return {
+        "git_dir": git_dir, "merge_oid": merge_oid, "closure_oid": closure_oid,
+        "head_oid": head_oid, "db": db, "expiry": expiry.isoformat().replace("+00:00", "Z"),
+        "manifest_path": "approvals/single-owner/activation.json",
+        "manifest_blob": manifest_blob,
+        "manifest_sha256": hashlib.sha256(manifest_raw.encode()).hexdigest(),
+        "execution_binding": execution_binding,
+        "machine_checks": machine_checks,
+        **execution_binding,
+    }
+
+
+def bindings(tmp_path: Path, adapter: FileProtectedBranchAdapter, **overrides) -> dict:
+    validator_path = Path(validator_module.__file__).resolve()
+    state_machine_path = Path(state_machine_module.__file__).resolve()
+    validator_sha = hashlib.sha256(validator_path.read_bytes()).hexdigest()
+    state_machine_sha = hashlib.sha256(state_machine_path.read_bytes()).hexdigest()
+    execution_binding = {
+        "task_id": "AIOS-LIVE-ACTIVATION-TEST",
+        "operation_id": "op-live-001",
+        "expected_validator_sha256": validator_sha,
+        "expected_state_machine_sha256": state_machine_sha,
+        "expected_branch_protection_sha256": adapter.calculate_snapshot_sha256(legacy_row()),
+        "expected_gitea_schema_sha256": "0" * 64,
+        "target_branch_protection_sha256": calculate_snapshot_sha256(owner_row()),
+        "backend": "file",
+        "repo_id": 1,
+        "branch": "governance/approvals",
+        "requester": "agent",
+        "executor": "activation-executor",
+        "forbidden_operations": {
+            "main": False, "governance_direct": False, "integration": False,
+            "tags": False, "force_push": False,
+        },
+    }
+    evidence = evidence_fixture(tmp_path, execution_binding)
+    value = {
+        "task_id": "AIOS-LIVE-ACTIVATION-TEST",
+        "operation_id": "op-live-001",
+        "expected_governance_oid": evidence["merge_oid"],
+        "observed_governance_oid": evidence["merge_oid"],
+        "expected_validator_sha256": validator_sha,
+        "observed_validator_sha256": validator_sha,
+        "expected_state_machine_sha256": state_machine_sha,
+        "observed_state_machine_sha256": state_machine_sha,
+        "validator_path": str(validator_path),
+        "state_machine_path": str(state_machine_path),
+        "governance_git_dir": str(evidence["git_dir"].resolve()),
+        "governance_ref": "refs/heads/governance/approvals",
+        "evidence_git_dir": str(evidence["git_dir"].resolve()),
+        "approval_db_path": str(evidence["db"].resolve()),
+        "expected_branch_protection_sha256": adapter.calculate_snapshot_sha256(legacy_row()),
+        "activation_approval_merge_oid": evidence["merge_oid"],
+        "approval_request_head_oid": evidence["head_oid"],
+        "activation_approval_pr_number": 31,
+        "activation_manifest_path": evidence["manifest_path"],
+        "activation_manifest_blob_oid": evidence["manifest_blob"],
+        "activation_manifest_sha256": evidence["manifest_sha256"],
+        "expires_at": evidence["expiry"],
+        "evidence_closure_oid": evidence["closure_oid"],
+        "observed_evidence_closure_oid": evidence["closure_oid"],
+        **{key: evidence[key] for key in (
+            "implementation_commit_oid", "implementation_tree_oid", "runtime_sources",
+            "artifact_path", "artifact_sha256", "rollback_artifact_path",
+            "rollback_artifact_sha256", "test_evidence_path", "test_evidence_sha256",
+            "evidence_closure_path", "evidence_closure_sha256",
+            "machine_status_producer_id", "machine_status_producer_name",
+        )},
+        "owner_review_oid": 52,
+        "observed_owner_review_oid": 52,
+        "requester": "agent",
+        "executor": "activation-executor",
+        "wip": False,
+        "machine_checks": evidence["machine_checks"],
+        "forbidden_operations": {
+            "main": False,
+            "governance_direct": False,
+            "integration": False,
+            "tags": False,
+            "force_push": False,
+        },
+        "target_branch_protection": owner_row(),
+        "backend": "file",
+        "backend_path": str(adapter.path.resolve()),
+        "repo_id": 1,
+        "branch": "governance/approvals",
+        "expected_gitea_schema_sha256": "0" * 64,
+    }
+    value.update(overrides)
+    return value
+
+
+def harness(tmp_path: Path):
+    backend = tmp_path / "protected.json"
+    write_json(backend, legacy_row())
+    adapter = FileProtectedBranchAdapter(backend)
+    engine = ActivationEngine(adapter, tmp_path / "state", tmp_path / "run" / "activation.lock")
+    request = bindings(tmp_path, adapter)
+    return backend, adapter, engine, request
+
+
+def test_live_plan_success(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    result = engine.plan(request)
+    assert result["status"] == "PLAN_READY"
+    assert len(result["plan_id"]) == 64
+
+
+def test_activation_bindings_schema_matches_runtime_contract(tmp_path: Path) -> None:
+    _, _, _, request = harness(tmp_path)
+    schema_path = Path(activation_module.__file__).resolve().parents[1] / "activation-bindings.schema.json"
+    schema = json.loads(schema_path.read_text(encoding="utf-8"))
+    Draft202012Validator.check_schema(schema)
+    validator = Draft202012Validator(schema)
+    validator.validate(request)
+    invalid = json.loads(json.dumps(request))
+    invalid["machine_checks"][sorted(MACHINE_CONTEXTS)[0]]["status_id"] = 0
+    with pytest.raises(ValidationError):
+        validator.validate(invalid)
+
+
+def test_runtime_source_must_be_actual_loaded_module_not_decoy(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    source = Path(request["runtime_sources"][0]["runtime_path"])
+    decoy = tmp_path / "decoy.py"
+    decoy.write_bytes(source.read_bytes())
+    request["runtime_sources"][0]["runtime_path"] = str(decoy.resolve())
+    with pytest.raises(RuntimeError, match="RUNTIME_SOURCE_PATH_MISMATCH"):
+        _verify_approved_entities(request, Path(request["evidence_git_dir"]))
+
+
+@pytest.mark.parametrize("path_field", ["artifact_path", "rollback_artifact_path", "test_evidence_path"])
+def test_approved_entity_digest_is_enforced(tmp_path: Path, path_field: str) -> None:
+    _, _, engine, request = harness(tmp_path)
+    Path(request[path_field]).write_bytes(b"tampered\n")
+    with pytest.raises(RuntimeError, match="APPROVED_ENTITY_DIGEST_MISMATCH"):
+        engine.plan(request)
+
+
+def test_evidence_closure_must_be_direct_child_commit(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    git_dir = Path(request["evidence_git_dir"])
+    unrelated, _ = commit_files(
+        git_dir,
+        {request["evidence_closure_path"]: '{"result":"PASS"}\n'},
+        "unrelated",
+    )
+    request["evidence_closure_oid"] = unrelated
+    request["observed_evidence_closure_oid"] = unrelated
+    with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
+        engine.plan(request)
+
+
+def test_pr_merge_after_expiry_is_denied(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    with sqlite3.connect(request["approval_db_path"]) as connection:
+        connection.execute(
+            "UPDATE pull_request SET merged_unix=?",
+            (int(_parse_test_utc(request["expires_at"]).timestamp()) + 1,),
+        )
+    with pytest.raises(RuntimeError, match="ACTIVATION_APPROVAL_MERGED_AFTER_EXPIRY"):
+        engine.plan(request)
+
+
+def _parse_test_utc(value: str) -> datetime:
+    return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+
+def test_machine_status_must_match_bound_producer_and_observed_row(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    context = sorted(MACHINE_CONTEXTS)[0]
+    request["machine_checks"][context]["status_id"] += 1000
+    with pytest.raises(RuntimeError, match="MACHINE_GATE_BINDING_MISMATCH"):
+        engine.plan(request)
+
+
+def test_latest_machine_status_from_unapproved_producer_is_denied(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    context = sorted(MACHINE_CONTEXTS)[0]
+    with sqlite3.connect(request["approval_db_path"]) as connection:
+        connection.execute("INSERT INTO user VALUES(100,'untrusted-producer')")
+        connection.execute(
+            "INSERT INTO commit_status VALUES(999,1,'success',?,?,100)",
+            (request["approval_request_head_oid"], context),
+        )
+    with pytest.raises(RuntimeError, match="MACHINE_GATE_PRODUCER_MISMATCH"):
+        engine.plan(request)
+
+
+def test_live_activate_verify_status_success(tmp_path: Path) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    activated = engine.activate(request, planned["plan_id"])
+    assert activated["status"] == "ACTIVATED"
+    assert adapter.read_current() == owner_row()
+    assert engine.verify(request, request["operation_id"])["status"] == "VERIFIED"
+    assert engine.status(request["operation_id"])["transaction"]["phase"] == "COMMITTED"
+    assert (engine.journal.root / "current-snapshot.json").exists()
+    assert (engine.journal.root / "rollback-snapshot.json").exists()
+
+
+def test_live_rollback_success(tmp_path: Path) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ROLLED_BACK"
+    assert adapter.read_current() == legacy_row()
+    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ALREADY_ROLLED_BACK"
+
+
+def test_terminal_rollback_cannot_be_reactivated(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    engine.rollback(request, request["operation_id"], planned["plan_id"])
+    with pytest.raises(RuntimeError, match="OPERATION_TERMINAL_ROLLED_BACK"):
+        engine.activate(request, planned["plan_id"])
+
+
+def test_verify_and_rollback_remain_available_after_approval_expiry(
+    tmp_path: Path, monkeypatch
+) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    future = datetime.now(timezone.utc) + timedelta(hours=2)
+
+    class FutureDateTime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return future if tz is not None else future.replace(tzinfo=None)
+
+    monkeypatch.setattr(activation_module, "datetime", FutureDateTime)
+    assert engine.verify(request, request["operation_id"])["status"] == "VERIFIED"
+    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ROLLED_BACK"
+    assert adapter.read_current() == legacy_row()
+
+
+def test_prepared_crash_can_recover_after_approval_expiry(tmp_path: Path, monkeypatch) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    request["crash_at"] = "before_apply"
+    planned = engine.plan(request)
+    with pytest.raises(InjectedCrash):
+        engine.activate(request, planned["plan_id"])
+    request.pop("crash_at")
+    future = datetime.now(timezone.utc) + timedelta(hours=2)
+
+    class FutureDateTime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return future if tz is not None else future.replace(tzinfo=None)
+
+    monkeypatch.setattr(activation_module, "datetime", FutureDateTime)
+    assert engine.activate(request, planned["plan_id"])["status"] == "ACTIVATED"
+    assert adapter.read_current() == owner_row()
+
+
+def test_fresh_activation_rechecks_expiry_at_prepared_boundary(
+    tmp_path: Path, monkeypatch
+) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    expiry = _parse_test_utc(request["expires_at"])
+    moments = iter((
+        expiry - timedelta(seconds=2),
+        expiry - timedelta(seconds=1),
+        expiry + timedelta(seconds=1),
+    ))
+    monkeypatch.setattr(activation_module, "_utc_now", lambda: next(moments))
+    with pytest.raises(RuntimeError, match="APPROVAL_EXPIRED"):
+        engine.activate(request, planned["plan_id"])
+    assert adapter.read_current() == legacy_row()
+    assert engine.journal.load(request["operation_id"]) is None
+
+
+def test_fresh_activation_rejects_snapshot_drift_at_prepared_boundary(
+    tmp_path: Path, monkeypatch
+) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    original_read = adapter.read_current
+    reads = 0
+
+    def drifting_read() -> dict:
+        nonlocal reads
+        reads += 1
+        if reads == 2:
+            drifted = original_read()
+            drifted["can_push"] = 1
+            drifted["updated_unix"] = 777
+            return drifted
+        return original_read()
+
+    monkeypatch.setattr(adapter, "read_current", drifting_read)
+    with pytest.raises(CASConflict, match="PREPARED_BOUNDARY_SNAPSHOT_CHANGED"):
+        engine.activate(request, planned["plan_id"])
+    assert original_read() == legacy_row()
+    assert engine.journal.load(request["operation_id"]) is None
+
+
+def test_rollback_uses_durable_authorization_after_live_status_rerun(tmp_path: Path) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    context = sorted(MACHINE_CONTEXTS)[0]
+    with sqlite3.connect(request["approval_db_path"]) as connection:
+        connection.execute(
+            "INSERT INTO commit_status VALUES(999,1,'success',?,?,?)",
+            (
+                request["approval_request_head_oid"], context,
+                request["machine_status_producer_id"],
+            ),
+        )
+    result = engine.rollback(request, request["operation_id"], planned["plan_id"])
+    assert result["status"] == "ROLLED_BACK"
+    assert adapter.read_current() == legacy_row()
+
+
+def test_missing_approval_denied(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    request.pop("owner_review_oid")
+    with pytest.raises(RuntimeError, match="MISSING_BINDINGS"):
+        engine.plan(request)
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "message"),
+    [
+        ("observed_owner_review_oid", 99, "WRONG_APPROVAL_OID_DENIED"),
+        ("observed_governance_oid", "f" * 40, "WRONG_GOVERNANCE_OID_DENIED"),
+        ("observed_evidence_closure_oid", "f" * 40, "WRONG_EVIDENCE_CLOSURE_DENIED"),
+    ],
+)
+def test_wrong_binding_denied(tmp_path: Path, field: str, value: object, message: str) -> None:
+    _, _, engine, request = harness(tmp_path)
+    request[field] = value
+    with pytest.raises(RuntimeError, match=message):
+        engine.plan(request)
+
+
+def test_cas_conflict_denied(tmp_path: Path) -> None:
+    backend, _, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    changed = legacy_row()
+    changed["updated_unix"] = 12
+    write_json(backend, changed)
+    with pytest.raises(CASConflict):
+        engine.activate(request, planned["plan_id"])
+
+
+@pytest.mark.parametrize("crash_at", ["before_apply", "during_apply", "after_apply_before_receipt", "after_commit_wal"])
+def test_process_crash_restart_recovery(tmp_path: Path, crash_at: str) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    request["crash_at"] = crash_at
+    planned = engine.plan(request)
+    with pytest.raises(InjectedCrash):
+        engine.activate(request, planned["plan_id"])
+    request.pop("crash_at")
+    result = engine.activate(request, planned["plan_id"])
+    expected_status = "ALREADY_COMMITTED" if crash_at == "after_commit_wal" else "ACTIVATED"
+    assert result["status"] == expected_status
+    assert adapter.read_current() == owner_row()
+    assert engine.journal.receipt_path(request["operation_id"]).exists()
+    assert engine.journal.audit.has_event(request["operation_id"], "SINGLE_OWNER_ACTIVATION_COMMITTED")
+
+
+def test_post_activation_verify_failure_rolls_back(tmp_path: Path, monkeypatch) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    monkeypatch.setattr(adapter, "verify_applied", lambda _: False)
+    with pytest.raises(RuntimeError, match="POST_ACTIVATION_VERIFY_FAILED"):
+        engine.activate(request, planned["plan_id"])
+    assert adapter.read_current() == legacy_row()
+    assert engine.journal.load(request["operation_id"])["phase"] == "ROLLED_BACK"
+
+
+def test_operation_id_idempotency(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    assert engine.activate(request, planned["plan_id"])["status"] == "ALREADY_COMMITTED"
+
+
+@pytest.mark.parametrize("crash_at", ["after_rollback_cas", "after_rollback_wal"])
+def test_rollback_crash_restart_repairs_closure(tmp_path: Path, crash_at: str) -> None:
+    _, adapter, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    request["crash_at"] = crash_at
+    with pytest.raises(InjectedCrash):
+        engine.rollback(request, request["operation_id"], planned["plan_id"])
+    request.pop("crash_at")
+    assert engine.rollback(request, request["operation_id"], planned["plan_id"])["status"] == "ALREADY_ROLLED_BACK"
+    assert adapter.read_current() == legacy_row()
+    assert engine.journal.receipt_path(request["operation_id"]).exists()
+    assert engine.journal.audit.has_event(request["operation_id"], "ACTIVATION_ROLLED_BACK")
+
+
+def test_operation_id_payload_mismatch_denied(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    changed = dict(request)
+    approval_alias = tmp_path / "approval-alias.db"
+    os.link(request["approval_db_path"], approval_alias)
+    changed["approval_db_path"] = str(approval_alias)
+    with pytest.raises(RuntimeError, match="PAYLOAD_MISMATCH"):
+        engine.activate(changed, planned["plan_id"])
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [("operation_id", "op-live-REPLAY"), ("task_id", "REPLAY-TASK")],
+)
+def test_approved_execution_binding_replay_denied(tmp_path: Path, field: str, value: str) -> None:
+    _, _, engine, request = harness(tmp_path)
+    request[field] = value
+    with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
+        engine.plan(request)
+
+
+def test_expiry_must_come_from_approved_manifest(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    request["expires_at"] = (datetime.now(timezone.utc) + timedelta(hours=12)).isoformat().replace("+00:00", "Z")
+    with pytest.raises(RuntimeError, match="APPROVAL_EXPIRY_BINDING_MISMATCH"):
+        engine.plan(request)
+
+
+def test_target_snapshot_must_match_approved_manifest(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    request["target_branch_protection"] = {**owner_row(), "updated_unix": 99}
+    with pytest.raises(RuntimeError, match="APPROVED_EXECUTION_BINDING_MISMATCH"):
+        engine.plan(request)
+
+
+def test_approval_database_must_be_backend_database(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    other_db = tmp_path / "other.db"
+    other_db.write_bytes(Path(request["approval_db_path"]).read_bytes())
+    request.update(backend="sqlite", backend_path=str(other_db.resolve()))
+    with pytest.raises(RuntimeError, match="APPROVAL_DB_BACKEND_DB_IDENTITY_MISMATCH"):
+        engine.plan(request)
+
+
+def test_audit_chain_continuity_and_credential_redaction(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    request["api_token"] = "must-not-appear"
+    planned = engine.plan(request)
+    engine.activate(request, planned["plan_id"])
+    assert len(engine.journal.audit.verify()) == 64
+    raw = (engine.journal.root / "audit.jsonl").read_text(encoding="utf-8")
+    assert "must-not-appear" not in raw
+    assert "[REDACTED]" not in raw  # only hashes/events are persisted to audit
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda request: request.update(requester="owner"),
+        lambda request: request.update(executor="owner"),
+        lambda request: request.update(wip=True),
+        lambda request: request.update(machine_checks={"anything": {"state": "success", "head_oid": request["approval_request_head_oid"]}}),
+        lambda request: request.update(forbidden_operations={"main": True}),
+        lambda request: request.update(forbidden_operations={"tags": True}),
+        lambda request: request.update(forbidden_operations={"force_push": True}),
+    ],
+)
+def test_negative_governance_acceptance(tmp_path: Path, mutation) -> None:
+    _, _, engine, request = harness(tmp_path)
+    mutation(request)
+    with pytest.raises(RuntimeError):
+        engine.plan(request)
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda target: target.update(enable_approvals_whitelist=0),
+        lambda target: target.update(enable_merge_whitelist=0),
+        lambda target: target.update(status_check_contexts="[]"),
+    ],
+)
+def test_unsafe_policy_enable_fields_denied(tmp_path: Path, mutation) -> None:
+    _, _, _, request = harness(tmp_path)
+    mutation(request["target_branch_protection"])
+    with pytest.raises(UnsafePolicy):
+        validate_target(legacy_row(), request["target_branch_protection"])
+
+
+@pytest.mark.parametrize("invalid_owner", ["codex", "gpt"])
+def test_owner_review_must_come_from_gitea_database(tmp_path: Path, invalid_owner: str) -> None:
+    _, _, engine, request = harness(tmp_path)
+    with sqlite3.connect(request["approval_db_path"]) as connection:
+        connection.execute("UPDATE user SET lower_name=? WHERE id=2", (invalid_owner,))
+    with pytest.raises(RuntimeError, match="OWNER_APPROVAL_INVALID"):
+        engine.plan(request)
+
+
+def test_stale_owner_review_denied_from_gitea_database(tmp_path: Path) -> None:
+    _, _, engine, request = harness(tmp_path)
+    with sqlite3.connect(request["approval_db_path"]) as connection:
+        connection.execute("UPDATE review SET stale=1 WHERE id=52")
+    with pytest.raises(RuntimeError, match="OWNER_APPROVAL_INVALID"):
+        engine.plan(request)
+
+
+def create_sqlite_fixture(path: Path) -> None:
+    row = legacy_row()
+    columns = []
+    for key, value in row.items():
+        kind = "INTEGER" if isinstance(value, int) else "TEXT"
+        columns.append(f"{key} {kind}")
+    with sqlite3.connect(path) as connection:
+        connection.execute(f"CREATE TABLE protected_branch ({','.join(columns)})")
+        connection.execute(
+            f"INSERT INTO protected_branch ({','.join(row)}) VALUES ({','.join('?' for _ in row)})",
+            list(row.values()),
+        )
+
+
+def test_sqlite_full_row_cas_and_rollback(tmp_path: Path) -> None:
+    db = tmp_path / "gitea.db"
+    create_sqlite_fixture(db)
+    probe = SQLiteProtectedBranchAdapter(db, expected_schema_sha256="0" * 64)
+    adapter = SQLiteProtectedBranchAdapter(db, expected_schema_sha256=probe.schema_sha256())
+    previous = adapter.read_current()
+    applied = adapter.apply_atomic(calculate_snapshot_sha256(previous), owner_row())
+    assert applied == owner_row()
+    restored = adapter.rollback_atomic(calculate_snapshot_sha256(applied), previous)
+    assert restored == previous

tests/single_owner_governance/test_state_cas_rollback.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+from pathlib import Path
+import tempfile
+import unittest
+
+from aios_single_owner_governance.fixture import DisposableGiteaFixture
+from aios_single_owner_governance.lock import MaintenanceLock
+from aios_single_owner_governance.models import GovernanceState
+from aios_single_owner_governance.protection import (
+    BranchProtectionCASMigrationAdapter,
+    CASConflict,
+    MemoryProtectionStore,
+    snapshot_sha256,
+)
+from aios_single_owner_governance.rollback import (
+    ProtectionCompensationSaga,
+    RollbackConfigurationManager,
+)
+from aios_single_owner_governance.state_machine import GovernanceStateMachine, InvalidTransition
+
+from .helpers import legacy_protection, single_owner_protection
+from .helpers import context
+from aios_single_owner_governance.validator import SingleOwnerApprovalValidator, ValidationDenied
+
+
+class StateCasRollbackTests(unittest.TestCase):
+    def test_ordinary_owner_approved_merge_simulation(self):
+        result = SingleOwnerApprovalValidator().validate(context())
+        self.assertEqual(result.owner_review.reviewer, "owner")
+        machine = GovernanceStateMachine()
+        for target in (
+            GovernanceState.OWNER_APPROVAL_VERIFIED,
+            GovernanceState.MACHINE_GATES_VERIFIED,
+            GovernanceState.AUTO_MERGE_READY,
+            GovernanceState.OWNER_APPROVED_EXECUTION_READY,
+        ):
+            machine.transition(target)
+        self.assertEqual(machine.state, GovernanceState.OWNER_APPROVED_EXECUTION_READY)
+
+    def test_validator_failure_is_stateless_and_recoverable(self):
+        validator = SingleOwnerApprovalValidator()
+        with self.assertRaises(ValidationDenied):
+            validator.validate(context(wip=True))
+        self.assertEqual(validator.validate(context()).owner_review.reviewer, "owner")
+
+    def test_state_machine_happy_path_and_restart_recovery(self):
+        machine = GovernanceStateMachine()
+        for target in (
+            GovernanceState.OWNER_APPROVAL_VERIFIED,
+            GovernanceState.MACHINE_GATES_VERIFIED,
+            GovernanceState.AUTO_MERGE_READY,
+            GovernanceState.OWNER_APPROVED_EXECUTION_READY,
+        ):
+            machine.transition(target)
+        restarted = GovernanceStateMachine(machine.state)
+        self.assertEqual(restarted.state, GovernanceState.OWNER_APPROVED_EXECUTION_READY)
+
+    def test_state_machine_rollback_to_checkpoint(self):
+        machine = GovernanceStateMachine()
+        machine.transition(GovernanceState.OWNER_APPROVAL_VERIFIED)
+        checkpoint = machine.state
+        machine.transition(GovernanceState.MACHINE_GATES_VERIFIED)
+        machine.restore(checkpoint)
+        self.assertEqual(machine.state, GovernanceState.OWNER_APPROVAL_VERIFIED)
+
+    def test_service_restart_from_each_checkpoint(self):
+        with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
+            checkpoint = Path(root) / "state.json"
+            for state in GovernanceState:
+                with self.subTest(state=state):
+                    GovernanceStateMachine(state).save_fixture(checkpoint)
+                    restarted = GovernanceStateMachine.load_fixture(checkpoint)
+                    self.assertEqual(restarted.state, state)
+
+    def test_invalid_transition_denied(self):
+        with self.assertRaises(InvalidTransition):
+            GovernanceStateMachine().transition(GovernanceState.AUTO_MERGE_READY)
+
+    def test_branch_protection_cas_success_and_rollback(self):
+        adapter = BranchProtectionCASMigrationAdapter()
+        old, new = legacy_protection(), single_owner_protection()
+        plan = adapter.plan(old, new, snapshot_sha256(old))
+        store = MemoryProtectionStore(old)
+        self.assertEqual(adapter.apply_fixture(store, plan), new)
+        self.assertEqual(adapter.rollback_fixture(store, plan), old)
+
+    def test_branch_protection_cas_conflict_fails_closed(self):
+        adapter = BranchProtectionCASMigrationAdapter()
+        old, changed, new = legacy_protection(), legacy_protection(7), single_owner_protection()
+        plan = adapter.plan(old, new, snapshot_sha256(old))
+        store = MemoryProtectionStore(changed)
+        with self.assertRaises(CASConflict):
+            adapter.apply_fixture(store, plan)
+
+    def test_partial_activation_recovery_saga(self):
+        adapter = BranchProtectionCASMigrationAdapter()
+        old, new = legacy_protection(), single_owner_protection()
+        plan = adapter.plan(old, new, snapshot_sha256(old))
+        saga = ProtectionCompensationSaga(adapter, MemoryProtectionStore(old), plan)
+        saga.execute_fixture()
+        self.assertEqual(saga.compensate_fixture(), old)
+        self.assertTrue(saga.applied and saga.rolled_back)
+
+    def test_rollback_configuration_fixture_install(self):
+        adapter = BranchProtectionCASMigrationAdapter()
+        old, new = legacy_protection(), single_owner_protection()
+        plan = adapter.plan(old, new, snapshot_sha256(old))
+        manager = RollbackConfigurationManager()
+        config = manager.build(plan)
+        with DisposableGiteaFixture(old) as fixture:
+            path = manager.install_fixture(fixture.root, config)
+            self.assertTrue(path.exists())
+            self.assertTrue(manager.verify(config, plan))
+
+    def test_maintenance_lock_fences_concurrent_holder(self):
+        with tempfile.TemporaryDirectory(prefix="aios-governance-fixture-") as root:
+            path = Path(root) / "maintenance.lock"
+            with MaintenanceLock(path):
+                with self.assertRaises(BlockingIOError):
+                    with MaintenanceLock(path):
+                        pass
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/single_owner_governance/test_systemd_contract.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from pathlib import Path
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+class SystemdContractTests(unittest.TestCase):
+    def test_units_are_static_dormant_and_credential_free(self):
+        units = sorted(
+            unit for unit in (ROOT / "deploy" / "systemd").glob("aios-governance-*.service")
+            if unit.name != "aios-governance-single-owner-activation.service"
+        )
+        self.assertEqual(len(units), 2)
+        for unit in units:
+            text = unit.read_text(encoding="utf-8")
+            with self.subTest(unit=unit.name):
+                self.assertIn("AIOS_SINGLE_OWNER_GOVERNANCE_ENABLED=false", text)
+                self.assertIn("AIOS_SINGLE_OWNER_ACTIVATION_ACTIVE=false", text)
+                self.assertIn("AIOS_GITOPS_WRITES_ENABLED=false", text)
+                self.assertIn("ProtectSystem=strict", text)
+                self.assertNotIn("EnvironmentFile=", text)
+                self.assertNotIn("LoadCredential=", text)
+                self.assertNotIn("WantedBy=", text)
+                self.assertNotIn("[Install]", text)
+                self.assertIn("--checkpoint-dir /var/lib/aios-gitops/runtime/single-owner-governance", text)
+                self.assertNotIn("/data/aios-gitops/production", text)
+
+    def test_activation_verifier_is_static_and_has_no_credential_or_install_hook(self):
+        unit = ROOT / "deploy" / "systemd" / "aios-governance-single-owner-activation.service"
+        text = unit.read_text(encoding="utf-8")
+        self.assertIn("AIOS_GITOPS_WRITES_ENABLED=false", text)
+        self.assertIn("--verify", text)
+        self.assertIn("ProtectSystem=strict", text)
+        self.assertNotIn("EnvironmentFile=", text)
+        self.assertNotIn("LoadCredential=", text)
+        self.assertNotIn("WantedBy=", text)
+        self.assertNotIn("[Install]", text)
+        self.assertNotIn("--activate", text)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/single_owner_governance/test_validator.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import timedelta
+import unittest
+
+from aios_single_owner_governance.models import Review
+from aios_single_owner_governance.validator import SingleOwnerApprovalValidator, ValidationDenied
+
+from .helpers import HEAD, NOW, context, gates, owner_review
+
+
+class ValidatorTests(unittest.TestCase):
+    def setUp(self):
+        self.validator = SingleOwnerApprovalValidator()
+
+    def denied(self, code, **changes):
+        with self.assertRaises(ValidationDenied) as caught:
+            self.validator.validate(context(**changes))
+        self.assertEqual(caught.exception.code, code)
+
+    def test_owner_current_head_and_machine_gates_pass(self):
+        result = self.validator.validate(context())
+        self.assertEqual(result.owner_review.reviewer, "owner")
+        self.assertEqual(len(result.machine_contexts), 3)
+
+    def test_requester_self_approval_denied(self):
+        self.denied("REQUESTER_SELF_APPROVAL", requester="owner")
+
+    def test_executor_self_approval_denied(self):
+        self.denied("EXECUTOR_SELF_APPROVAL", executor="owner")
+
+    def test_codex_as_owner_denied(self):
+        codex = owner_review(reviewer="codex", reviewer_id=42)
+        self.denied("OWNER_APPROVAL_REQUIRED", reviews=(codex,))
+
+    def test_gpt_as_owner_denied(self):
+        gpt = owner_review(reviewer="gpt", reviewer_id=43)
+        self.denied("OWNER_APPROVAL_REQUIRED", reviews=(gpt,))
+
+    def test_non_owner_approval_denied(self):
+        other = owner_review(reviewer="other", reviewer_id=77)
+        self.denied("OWNER_APPROVAL_REQUIRED", reviews=(other,))
+
+    def test_stale_owner_approval_denied(self):
+        self.denied("STALE_OWNER_APPROVAL", reviews=(owner_review(stale=True),))
+
+    def test_dismissed_owner_approval_denied(self):
+        self.denied("DISMISSED_OWNER_APPROVAL", reviews=(owner_review(dismissed=True),))
+
+    def test_old_head_owner_approval_denied(self):
+        self.denied("OWNER_APPROVAL_NOT_LATEST_HEAD", reviews=(owner_review(head_oid="b" * 40),))
+
+    def test_wip_denied(self):
+        self.denied("WIP_BLOCKED", wip=True)
+
+    def test_machine_gate_failure_denied(self):
+        self.denied("MACHINE_GATE_FAILED", machine_gates=gates(state="failure"))
+
+    def test_secret_scan_failure_denied(self):
+        failed = list(gates())
+        failed[1] = replace(failed[1], state="failure")
+        self.denied("MACHINE_GATE_FAILED", machine_gates=tuple(failed))
+
+    def test_machine_gate_wrong_head_denied(self):
+        self.denied("MACHINE_GATE_NOT_LATEST_HEAD", machine_gates=gates(head_oid="c" * 40))
+
+    def test_expired_request_denied(self):
+        self.denied("APPROVAL_EXPIRED", now=NOW + timedelta(hours=2))
+
+    def test_review_after_expiry_denied(self):
+        late = owner_review(reviewed_at=NOW + timedelta(hours=2))
+        self.denied("OWNER_APPROVAL_EXPIRED", reviews=(late,))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_isolated_writer_exact_path_policy.py

@@ -0,0 +1,417 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest import mock
+
+import aios_isolated_writer_policy as writer_policy
+import isolated_writer_path_policy as path_policy
+from isolated_writer_authz import AuthorizationError, _validate_grant
+
+
+TARGET = "deploy/systemd/aios-codex-review-test-worker.service"
+BRANCH = "patch/codex-review-test-worker-20260724"
+AUDIENCE = "aios-codex-review-worker-writer.service"
+
+VALID_UNIT = """\
+[Unit]
+Description=AIOS Codex Review/Test Worker
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=aiman
+Group=gitops
+WorkingDirectory=/usr/libexec/aios-codex-review-test-worker/current
+Environment=AIOS_CODEX_WORKER_MODE=shadow
+Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+Environment=PYTHONDONTWRITEBYTECODE=1
+ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+MemoryDenyWriteExecute=true
+RuntimeDirectory=aios-codex-review-test-worker
+StateDirectory=aios-codex-review-test-worker
+CacheDirectory=aios-codex-review-test-worker
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadOnlyPaths=/usr/libexec/aios-codex-review-test-worker
+ReadWritePaths=/run/aios-codex-review-test-worker
+ReadWritePaths=/var/lib/aios-codex-review-test-worker
+ReadWritePaths=/var/cache/aios-codex-review-test-worker
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target
+"""
+
+
+def document(*, paths=None, **updates):
+    value = {
+        "schema_version": 1,
+        "policy_id": "codex-review-worker-path-policy-v1",
+        "audience": AUDIENCE,
+        "repository_identity": "aios/aios",
+        "allowed_branch": BRANCH,
+        "allowed_operations": {
+            "commit_file.atomic": {
+                "exact_paths": paths or [TARGET],
+                "max_content_bytes": 32768,
+                "content_profile": "systemd-codex-review-worker-v1",
+            }
+        },
+    }
+    value.update(updates)
+    return value
+
+
+class ExactPathPolicyTests(unittest.TestCase):
+    def setUp(self):
+        self.saved = dict(os.environ)
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.policy_dir = self.root / "policies"
+        self.policy_dir.mkdir(mode=0o750)
+        self.policy_file = self.policy_dir / "writer.json"
+        self.workspace = self.root / "workspace"
+        self.workspace.mkdir(mode=0o700)
+        os.environ.update(
+            {
+                "AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
+                "AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
+                "AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios",
+                "AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
+            }
+        )
+
+    def tearDown(self):
+        path_policy.clear_path_policy_cache()
+        os.environ.clear()
+        os.environ.update(self.saved)
+        self.temp.cleanup()
+
+    def write_policy(self, value=None):
+        payload = document() if value is None else value
+        raw = (
+            json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+            + "\n"
+        )
+        self.policy_file.write_text(raw, encoding="utf-8")
+        self.policy_file.chmod(0o640)
+        return path_policy.load_path_policy(
+            self.policy_file,
+            expected_owner_uid=os.getuid(),
+            expected_group_gid=os.getgid(),
+        )
+
+    def configured(self, loaded):
+        return mock.patch.object(
+            path_policy,
+            "configured_path_policy",
+            return_value=loaded,
+        )
+
+    def test_exact_systemd_path_and_exact_token_path_are_allowed(self):
+        loaded = self.write_policy()
+        with self.configured(loaded):
+            self.assertEqual(TARGET, writer_policy.validate_path(TARGET, self.workspace))
+            writer_policy.validate_content_for_path(TARGET, VALID_UNIT)
+
+    def test_token_path_cannot_expand_service_policy(self):
+        loaded = self.write_policy(document(paths=["deploy/systemd/other.service"]))
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_service_path_rejected"
+        ):
+            writer_policy.validate_path(TARGET, self.workspace)
+
+    def test_service_policy_cannot_expand_token_path(self):
+        now = datetime.now(timezone.utc)
+        grant = {
+            "schema_version": 2,
+            "audience": AUDIENCE,
+            "source_cidrs": ["127.0.0.1/32"],
+            "created_at": (now - timedelta(seconds=5)).isoformat(),
+            "not_before": (now - timedelta(seconds=2)).isoformat(),
+            "expires_at": (now + timedelta(minutes=5)).isoformat(),
+            "revoked": False,
+            "allowed_operations": ["commit_file.atomic"],
+            "allowed_branch": BRANCH,
+            "allowed_paths": ["deploy/systemd/other.service"],
+            "approved_base_oid": "a" * 40,
+        }
+        with mock.patch(
+            "isolated_writer_authz._git_is_ancestor", return_value=True
+        ), self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
+            _validate_grant(
+                grant,
+                audience=AUDIENCE,
+                source_ip="127.0.0.1",
+                operation="commit_file.atomic",
+                branch=BRANCH,
+                path=TARGET,
+                workspace=self.workspace,
+                current_time=now,
+            )
+
+    def test_other_service_file_is_rejected(self):
+        loaded = self.write_policy()
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_service_path_rejected"
+        ):
+            writer_policy.validate_path(
+                "deploy/systemd/other.service", self.workspace
+            )
+
+    def test_path_traversal_is_rejected(self):
+        for value in ("deploy/systemd/../x", "../deploy/systemd/x"):
+            with self.subTest(value=value), self.assertRaises(PermissionError):
+                writer_policy.validate_path(value, self.workspace)
+
+    def test_url_encoded_traversal_is_rejected(self):
+        with self.assertRaises(PermissionError):
+            writer_policy.validate_path(
+                "deploy/systemd/%2e%2e/x", self.workspace
+            )
+
+    def test_symlink_policy_file_is_rejected(self):
+        self.write_policy()
+        link = self.policy_dir / "link.json"
+        link.symlink_to(self.policy_file)
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError, "isolated_writer_path_policy_symlink_rejected"
+        ):
+            path_policy.load_path_policy(
+                link,
+                expected_owner_uid=os.getuid(),
+                expected_group_gid=os.getgid(),
+            )
+
+    def test_world_writable_policy_file_is_rejected(self):
+        self.write_policy()
+        self.policy_file.chmod(0o646)
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_permissions_rejected",
+        ):
+            path_policy.load_path_policy(
+                self.policy_file,
+                expected_owner_uid=os.getuid(),
+                expected_group_gid=os.getgid(),
+            )
+
+    def test_unknown_policy_field_is_rejected(self):
+        value = document()
+        value["unexpected"] = True
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_fields_rejected",
+        ):
+            self.write_policy(value)
+
+    def test_audience_mismatch_is_rejected(self):
+        loaded = self.write_policy()
+        os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
+        os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "other.service"
+        path_policy.clear_path_policy_cache()
+        with mock.patch.object(
+            path_policy, "load_path_policy", return_value=loaded
+        ), self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_audience_rejected",
+        ):
+            path_policy.configured_path_policy()
+
+    def test_branch_mismatch_is_rejected(self):
+        loaded = self.write_policy()
+        os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
+        os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/other"
+        path_policy.clear_path_policy_cache()
+        with mock.patch.object(
+            path_policy, "load_path_policy", return_value=loaded
+        ), self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_branch_rejected",
+        ):
+            path_policy.configured_path_policy()
+
+    def assert_profile_rejected(self, content):
+        loaded = self.write_policy()
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_systemd_profile_rejected"
+        ):
+            writer_policy.validate_content_for_path(TARGET, content)
+
+    def test_disallowed_execution_and_credential_directives_are_rejected(self):
+        directives = (
+            "ExecStartPre=/usr/bin/id",
+            "ExecStartPost=/usr/bin/id",
+            "ExecReload=/usr/bin/id",
+            "ExecStop=/usr/bin/id",
+            "ExecStopPost=/usr/bin/id",
+            "ExecCondition=/usr/bin/true",
+            "EnvironmentFile=/run/worker.env",
+            "PassEnvironment=PATH",
+            "LoadCredential=worker:/run/worker",
+            "SetCredential=worker:value",
+            "ImportCredential=worker",
+            "StandardInput=socket",
+            "StandardOutput=socket",
+        )
+        for directive in directives:
+            with self.subTest(directive=directive):
+                self.assert_profile_rejected(
+                    VALID_UNIT.replace(
+                        "Restart=on-failure",
+                        f"Restart=on-failure\n{directive}",
+                    )
+                )
+
+    def test_disallowed_namespace_and_capability_directives_are_rejected(self):
+        directives = (
+            "RootDirectory=/srv/worker",
+            "RootImage=/srv/worker.raw",
+            "ReadWritePaths=/",
+            "BindPaths=/etc:/etc",
+            "BindReadOnlyPaths=/etc:/etc",
+            "TemporaryFileSystem=/etc",
+            "CapabilityBoundingSet=CAP_SYS_ADMIN",
+            "AmbientCapabilities=CAP_SYS_ADMIN",
+            "DeviceAllow=/dev/sda rw",
+            "SupplementaryGroups=docker",
+            "Sockets=unsafe.socket",
+        )
+        for directive in directives:
+            with self.subTest(directive=directive):
+                self.assert_profile_rejected(
+                    VALID_UNIT.replace(
+                        "Restart=on-failure",
+                        f"Restart=on-failure\n{directive}",
+                    )
+                )
+
+    def test_duplicate_single_directives_and_sections_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace("User=aiman", "User=aiman\nUser=aiman"),
+            VALID_UNIT.replace(
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test\n"
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
+            ),
+            VALID_UNIT + "\n[Service]\nType=simple\n",
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_empty_reset_continuation_crlf_nul_and_long_lines_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace(
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
+                "ExecStart=",
+            ),
+            VALID_UNIT.replace(
+                "Restart=on-failure",
+                "Restart=on-failure\n continuation-injection",
+            ),
+            VALID_UNIT.replace("\n", "\r\n"),
+            VALID_UNIT.replace("Restart=on-failure", "Restart=on-failure\x00"),
+            VALID_UNIT.replace(
+                "Description=AIOS Codex Review/Test Worker",
+                "Description=" + ("A" * 4097),
+            ),
+        )
+        for value in bad_values:
+            with self.subTest(kind=repr(value[:80])):
+                self.assert_profile_rejected(value)
+
+    def test_unknown_environment_and_production_execution_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace(
+                "Environment=PYTHONDONTWRITEBYTECODE=1",
+                "Environment=UNKNOWN_KEY=value",
+            ),
+            VALID_UNIT.replace(
+                "AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
+                "AIOS_TASK_PLANE_EXECUTION_ENABLED=true",
+            ),
+            VALID_UNIT.replace(
+                "Environment=PYTHONDONTWRITEBYTECODE=1",
+                "Environment=Authorization: Bearer placeholder",
+            ),
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_identity_working_directory_and_path_prefix_bypasses_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace("User=aiman", "User=root"),
+            VALID_UNIT.replace(
+                "/usr/libexec/aios-codex-review-test-worker/current",
+                "/data/aios-gitops/worktrees/worker",
+            ),
+            VALID_UNIT.replace(
+                "ReadWritePaths=/run/aios-codex-review-test-worker",
+                "ReadWritePaths=/etc",
+            ),
+            VALID_UNIT.replace(
+                "ReadOnlyPaths=/data/aios-gitops/repo.git",
+                "ReadOnlyPaths=/data",
+            ),
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_unknown_section_directive_and_unicode_lookalike_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace("[Install]", "[Socket]"),
+            VALID_UNIT.replace(
+                "Restart=on-failure",
+                "Restart=on-failure\nProtectClock=true",
+            ),
+            VALID_UNIT.replace("ExecStart=", "\u0415xecStart="),
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_invalid_unicode_surrogate_and_missing_final_newline_are_rejected(self):
+        for value in (VALID_UNIT + "\ud800", VALID_UNIT.rstrip("\n")):
+            with self.subTest(value=repr(value[-30:])):
+                self.assert_profile_rejected(value)
+
+    def test_writer_request_cannot_modify_policy_file(self):
+        with self.assertRaises(PermissionError):
+            writer_policy.validate_path(
+                "etc/aios-gitops/writer-path-policies/codex-review-worker.json",
+                self.workspace,
+            )
+
+    def test_coarse_file_and_commit_endpoints_remain_forbidden(self):
+        for endpoint in ("/api/file", "/api/commit"):
+            with self.subTest(endpoint=endpoint), self.assertRaisesRegex(
+                PermissionError, "isolated_writer_operation_forbidden"
+            ):
+                writer_policy.validate_endpoint(
+                    "POST", endpoint, self.workspace, self.root
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_isolated_writer_policy.py

@@ -15,6 +15,10 @@ class IsolatedWriterPolicyTests(unittest.TestCase):
         os.environ["AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT"] = "required"
         os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/aios-agent-task-control-plane-20260724"
         os.environ["AIOS_GITOPS_ALLOWED_BASE_OID"] = "a8e97868976b0dce17b7cb8487eadc1d9c687622"
+        os.environ["AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED"] = "2"
+        os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "test-isolated-writer.service"
+        os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"] = "/tmp/test-writer-token-registry.sqlite3"
+        os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"] = "/tmp/test-writer-operations.sqlite3"
 
     def tearDown(self):
         os.environ.clear()

tests/test_isolated_writer_token_policy_v2.py

@@ -0,0 +1,471 @@
+import hashlib
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "active-runtime" / "app"))
+
+import aios_isolated_writer_policy as policy
+from isolated_writer_authz import AuthorizationError, authorize, operation_for
+from isolated_writer_commit_file import (
+    AtomicCommitFile,
+    CommitFileError,
+    OperationLedger,
+    SimulatedCrash,
+)
+from isolated_writer_token_registry import TokenRegistry, token_digest
+
+
+BRANCH = "patch/codex-review-test-worker-20260724"
+AUDIENCE = "aios-codex-review-worker-writer.service"
+TASK_ID = "AIOS-CODEX-WRITER-SMOKE-20260724"
+TARGET_PATH = "docs/reliability/codex-review-worker-writer-smoke.json"
+
+
+def run(cwd, *args):
+    return subprocess.check_output(args, cwd=cwd, text=True).strip()
+
+
+@dataclass
+class DummyGrant:
+    token_id: str = "test-token-id"
+    task_id: str = TASK_ID
+    operation: str = "commit_file.atomic"
+    record: dict = None
+
+    def __post_init__(self):
+        if self.record is None:
+            self.record = {"audience": AUDIENCE, "_source_ip": "127.0.0.1"}
+
+
+class RepositoryCase(unittest.TestCase):
+    def setUp(self):
+        self.saved = dict(os.environ)
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.authoritative = self.root / "authoritative.git"
+        self.gitea = self.root / "gitea.git"
+        self.seed = self.root / "seed"
+        self.workspace = self.root / "writer"
+        subprocess.run(["git", "init", "--bare", str(self.authoritative)], check=True, capture_output=True)
+        subprocess.run(["git", "init", str(self.seed)], check=True, capture_output=True)
+        subprocess.run(["git", "config", "user.email", "writer@example.invalid"], cwd=self.seed, check=True)
+        subprocess.run(["git", "config", "user.name", "Writer Test"], cwd=self.seed, check=True)
+        (self.seed / "README.md").write_text("base\n", encoding="utf-8")
+        subprocess.run(["git", "add", "README.md"], cwd=self.seed, check=True)
+        subprocess.run(["git", "commit", "-m", "base"], cwd=self.seed, check=True, capture_output=True)
+        subprocess.run(["git", "branch", "-M", "main"], cwd=self.seed, check=True)
+        subprocess.run(["git", "remote", "add", "origin", str(self.authoritative)], cwd=self.seed, check=True)
+        subprocess.run(["git", "push", "origin", "main"], cwd=self.seed, check=True, capture_output=True)
+        self.base = run(self.seed, "git", "rev-parse", "HEAD")
+        subprocess.run(["git", "checkout", "-b", BRANCH], cwd=self.seed, check=True, capture_output=True)
+        subprocess.run(["git", "push", "origin", BRANCH], cwd=self.seed, check=True, capture_output=True)
+        subprocess.run(
+            ["git", "--git-dir", str(self.authoritative), "worktree", "add", str(self.workspace), BRANCH],
+            check=True,
+            capture_output=True,
+        )
+        subprocess.run(["git", "config", "user.email", "writer@example.invalid"], cwd=self.workspace, check=True)
+        subprocess.run(["git", "config", "user.name", "Writer Test"], cwd=self.workspace, check=True)
+        subprocess.run(["git", "clone", "--bare", str(self.authoritative), str(self.gitea)], check=True, capture_output=True)
+        subprocess.run(["git", "remote", "add", "gitea-http", str(self.gitea)], cwd=self.workspace, check=True)
+        self.registry = TokenRegistry(self.root / "registry.sqlite3")
+        os.environ.update({
+            "AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
+            "AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED": "2",
+            "AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
+            "AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
+        })
+
+    def tearDown(self):
+        os.environ.clear()
+        os.environ.update(self.saved)
+        self.temp.cleanup()
+
+    def grant(self, **updates):
+        now = datetime.now(timezone.utc)
+        grant = {
+            "token_id": "smoke-" + hashlib.sha256(os.urandom(16)).hexdigest()[:12],
+            "audience": AUDIENCE,
+            "task_id": TASK_ID,
+            "source_cidrs": ["127.0.0.1/32", "::1/128"],
+            "allowed_branch": BRANCH,
+            "approved_base_oid": self.base,
+            "allowed_operations": [
+                "runtime.status.read",
+                "baseline.read",
+                "patches.read",
+                "commit_file.atomic",
+            ],
+            "allowed_paths": [TARGET_PATH],
+            "not_before": (now - timedelta(seconds=2)).isoformat(),
+            "expires_at": (now + timedelta(minutes=20)).isoformat(),
+            "max_uses": {
+                "runtime.status.read": 2,
+                "baseline.read": 2,
+                "patches.read": 2,
+                "commit_file.atomic": 4,
+            },
+            "issued_by": "owner-authorized-local-maintenance-executor",
+            "created_at": (now - timedelta(seconds=3)).isoformat(),
+        }
+        grant.update(updates)
+        return grant
+
+    def issue(self, **updates):
+        return self.registry.issue(self.grant(**updates))
+
+    def auth(self, token, **updates):
+        args = {
+            "method": "GET",
+            "path": "/api/runtime-status",
+            "source_ip": "127.0.0.1",
+            "workspace": self.workspace,
+        }
+        args.update(updates)
+        return authorize(self.registry, token, **args)
+
+    def request(self, key="writer-smoke-0001", content='{"status":"READY"}\n'):
+        return {
+            "branch": BRANCH,
+            "path": TARGET_PATH,
+            "content": content,
+            "message": "chore(task-plane): verify Codex writer endpoint",
+            "expected_head_oid": run(self.workspace, "git", "rev-parse", "HEAD"),
+            "idempotency_key": key,
+        }
+
+    def writer(self, crash=None, remote="gitea-http", audit=None):
+        return AtomicCommitFile(
+            workspace=self.workspace,
+            bare_repo=self.authoritative,
+            ledger=OperationLedger(self.root / "operations.sqlite3"),
+            remote=remote,
+            audit=audit or (lambda record: None),
+            crash_after_state=crash,
+        )
+
+
+class TokenPolicyV2Tests(RepositoryCase):
+    def test_v1_coarse_scope_token_is_rejected(self):
+        token, record = self.issue()
+        record["schema_version"] = 1
+        with self.registry._connect() as db:
+            db.execute(
+                "UPDATE token_grants SET grant_json=? WHERE token_id=?",
+                (json.dumps(record), record["token_id"]),
+            )
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_policy_v2_required"):
+            self.auth(token)
+
+    def test_audience_mismatch_is_rejected(self):
+        token, _ = self.issue(audience="other.service")
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_audience_rejected"):
+            self.auth(token)
+
+    def test_source_ip_mismatch_is_rejected(self):
+        token, _ = self.issue(source_cidrs=["192.0.2.0/24"])
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_source_rejected"):
+            self.auth(token)
+
+    def test_branch_mismatch_is_rejected(self):
+        token, _ = self.issue(allowed_branch="patch/other")
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_branch_rejected"):
+            self.auth(token)
+
+    def test_operation_mismatch_is_rejected(self):
+        token, _ = self.issue(allowed_operations=["baseline.read"])
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_operation_rejected"):
+            self.auth(token)
+
+    def test_path_mismatch_is_rejected(self):
+        token, _ = self.issue()
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
+            self.auth(
+                token,
+                method="POST",
+                path="/api/commit-file",
+                branch=BRANCH,
+                target_path="docs/other.json",
+            )
+
+    def test_expired_token_is_rejected(self):
+        now = datetime.now(timezone.utc)
+        token, _ = self.issue(
+            created_at=(now - timedelta(minutes=10)).isoformat(),
+            not_before=(now - timedelta(minutes=9)).isoformat(),
+            expires_at=(now - timedelta(seconds=1)).isoformat(),
+        )
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_expired"):
+            self.auth(token)
+
+    def test_not_before_token_is_rejected(self):
+        now = datetime.now(timezone.utc)
+        token, _ = self.issue(
+            created_at=now.isoformat(),
+            not_before=(now + timedelta(minutes=1)).isoformat(),
+            expires_at=(now + timedelta(minutes=5)).isoformat(),
+        )
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_not_yet_valid"):
+            self.auth(token)
+
+    def test_revoked_token_is_rejected(self):
+        token, grant = self.issue()
+        self.assertTrue(self.registry.revoke(grant["token_id"]))
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_revoked"):
+            self.auth(token)
+
+    def test_operation_use_count_is_atomic_and_exhausted(self):
+        token, _ = self.issue(max_uses={
+            "runtime.status.read": 1,
+            "baseline.read": 1,
+            "patches.read": 1,
+            "commit_file.atomic": 1,
+        })
+        self.auth(token)
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_operation_exhausted"):
+            self.auth(token)
+
+    def test_concurrent_use_count_never_exceeds_limit(self):
+        token, _ = self.issue(max_uses={
+            "runtime.status.read": 2,
+            "baseline.read": 1,
+            "patches.read": 1,
+            "commit_file.atomic": 1,
+        })
+        successes, failures = [], []
+
+        def invoke():
+            try:
+                successes.append(self.auth(token))
+            except AuthorizationError as exc:
+                failures.append(exc.code)
+
+        threads = [threading.Thread(target=invoke) for _ in range(6)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertEqual(2, len(successes))
+        self.assertEqual(4, failures.count("writer_token_operation_exhausted"))
+
+    def test_registry_does_not_store_plaintext_token(self):
+        token, _ = self.issue()
+        registry_bytes = b"".join(
+            item.read_bytes()
+            for item in self.registry.path.parent.glob(self.registry.path.name + "*")
+            if item.is_file()
+        )
+        self.assertNotIn(token.encode(), registry_bytes)
+        with self.registry._connect() as db:
+            stored = db.execute("SELECT token_digest FROM token_grants").fetchone()[0]
+        self.assertEqual(token_digest(token), stored)
+
+    def test_coarse_routes_have_no_operation_mapping(self):
+        for path in ("/api/file", "/api/commit", "/api/tag", "/api/rollback", "/api/model"):
+            with self.subTest(path=path), self.assertRaises(AuthorizationError) as captured:
+                operation_for("POST", path)
+            self.assertEqual(403, captured.exception.status)
+
+    def test_route_mapping_is_fixed_not_request_controlled(self):
+        self.assertEqual("runtime.status.read", operation_for("GET", "/api/runtime-status"))
+        self.assertEqual("baseline.read", operation_for("GET", "/api/baseline"))
+        self.assertEqual("patches.read", operation_for("GET", "/api/patches"))
+        self.assertEqual("commit_file.atomic", operation_for("POST", "/api/commit-file"))
+
+    def test_x_forwarded_for_cannot_replace_socket_peer(self):
+        token, _ = self.issue(source_cidrs=["203.0.113.9/32"])
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_source_rejected"):
+            self.auth(token, source_ip="127.0.0.1")
+
+
+class AtomicCommitFileTests(RepositoryCase):
+    def test_commit_file_succeeds_and_verifies_bare_and_gitea(self):
+        result = self.writer().execute(DummyGrant(), self.request())
+        self.assertEqual("COMPLETED", result["state"])
+        self.assertEqual(
+            result["commit_oid"],
+            run(self.root, "git", "--git-dir", str(self.gitea), "rev-parse", f"refs/heads/{BRANCH}"),
+        )
+
+    def test_same_idempotency_key_returns_same_commit(self):
+        request = self.request()
+        first = self.writer().execute(DummyGrant(), request)
+        count = run(self.workspace, "git", "rev-list", "--count", "HEAD")
+        second = self.writer().execute(DummyGrant(), request)
+        self.assertEqual(first["commit_oid"], second["commit_oid"])
+        self.assertEqual(count, run(self.workspace, "git", "rev-list", "--count", "HEAD"))
+        self.assertTrue(second["idempotent"])
+
+    def test_same_key_with_different_payload_returns_409(self):
+        request = self.request()
+        self.writer().execute(DummyGrant(), request)
+        request["content"] = '{"status":"DIFFERENT"}\n'
+        with self.assertRaises(CommitFileError) as captured:
+            self.writer().execute(DummyGrant(), request)
+        self.assertEqual(409, captured.exception.status)
+
+    def test_same_key_from_different_task_returns_409(self):
+        request = self.request()
+        self.writer().execute(DummyGrant(), request)
+        other = DummyGrant(token_id="other-token", task_id="OTHER-TASK")
+        with self.assertRaisesRegex(CommitFileError, "writer_idempotency_owner_conflict"):
+            self.writer().execute(other, request)
+
+    def test_file_written_crash_is_recoverable(self):
+        request = self.request()
+        with self.assertRaises(SimulatedCrash):
+            self.writer(crash="FILE_WRITTEN").execute(DummyGrant(), request)
+        result = self.writer().execute(DummyGrant(), request)
+        self.assertEqual("COMPLETED", result["state"])
+
+    def test_committed_crash_is_recoverable_without_second_commit(self):
+        request = self.request()
+        with self.assertRaises(SimulatedCrash):
+            self.writer(crash="COMMITTED").execute(DummyGrant(), request)
+        committed = run(self.workspace, "git", "rev-parse", "HEAD")
+        result = self.writer().execute(DummyGrant(), request)
+        self.assertEqual(committed, result["commit_oid"])
+
+    def test_pushed_crash_only_reverifies(self):
+        request = self.request()
+        with self.assertRaises(SimulatedCrash):
+            self.writer(crash="PUSHED").execute(DummyGrant(), request)
+        result = self.writer().execute(DummyGrant(), request)
+        self.assertEqual("COMPLETED", result["state"])
+
+    def test_push_failure_is_failed_recoverable(self):
+        request = self.request()
+        with self.assertRaisesRegex(CommitFileError, "writer_gitea_push_failed"):
+            self.writer(remote="missing-remote").execute(DummyGrant(), request)
+        row = OperationLedger(self.root / "operations.sqlite3").get(request["idempotency_key"])
+        self.assertEqual("FAILED_RECOVERABLE", row["state"])
+
+    def test_branch_head_conflict_requires_manual_review(self):
+        request = self.request()
+        ledger = OperationLedger(self.root / "operations.sqlite3")
+        ledger.get_or_create(DummyGrant(), request, __import__("isolated_writer_commit_file").canonical_digest(request))
+        (self.workspace / "README.md").write_text("drift\n", encoding="utf-8")
+        subprocess.run(["git", "commit", "-am", "drift"], cwd=self.workspace, check=True, capture_output=True)
+        with self.assertRaisesRegex(CommitFileError, "writer_branch_head_conflict"):
+            self.writer().execute(DummyGrant(), request)
+        self.assertEqual("MANUAL_REVIEW_REQUIRED", ledger.get(request["idempotency_key"])["state"])
+
+    def test_extra_staged_file_is_rejected(self):
+        (self.workspace / "README.md").write_text("staged\n", encoding="utf-8")
+        subprocess.run(["git", "add", "README.md"], cwd=self.workspace, check=True)
+        with self.assertRaisesRegex(CommitFileError, "writer_extra_staged_files_rejected"):
+            self.writer().execute(DummyGrant(), self.request())
+
+    def test_path_traversal_and_protected_roots_are_rejected(self):
+        for path in (
+            "../docs/x", "%2e%2e/docs/x", "/etc/x", r"C:\temp\x",
+            r"\\server\share\x", "baseline/current.json", "runtime/x",
+        ):
+            request = self.request(key="writer-path-" + hashlib.sha256(path.encode()).hexdigest()[:12])
+            request["path"] = path
+            with self.subTest(path=path), self.assertRaises((PermissionError, CommitFileError)):
+                self.writer().execute(DummyGrant(), request)
+
+    def test_audit_contains_ids_but_not_content_or_authorization(self):
+        records = []
+        request = self.request(content='{"value":"non-secret-body-marker"}\n')
+        self.writer(audit=records.append).execute(DummyGrant(), request)
+        encoded = json.dumps(records)
+        self.assertIn("test-token-id", encoded)
+        self.assertNotIn("non-secret-body-marker", encoded)
+        self.assertNotIn("Authorization", encoded)
+
+    def test_concurrent_same_idempotency_creates_one_commit(self):
+        request = self.request()
+        results, errors = [], []
+
+        def invoke():
+            try:
+                results.append(self.writer().execute(DummyGrant(), request))
+            except Exception as exc:
+                errors.append(exc)
+
+        threads = [threading.Thread(target=invoke) for _ in range(4)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertFalse(errors)
+        self.assertEqual(1, len({item["commit_oid"] for item in results}))
+
+    def test_different_operations_are_serialized_by_branch_lock(self):
+        active = 0
+        maximum = 0
+        guard = threading.Lock()
+
+        class Probe(AtomicCommitFile):
+            def _execute_locked(inner, grant, data, digest):
+                nonlocal active, maximum
+                with guard:
+                    active += 1
+                    maximum = max(maximum, active)
+                time.sleep(0.05)
+                with guard:
+                    active -= 1
+                return {"state": "PROBED"}
+
+        probe = Probe(
+            workspace=self.workspace,
+            bare_repo=self.authoritative,
+            ledger=OperationLedger(self.root / "probe.sqlite3"),
+            remote="gitea-http",
+            audit=lambda record: None,
+        )
+        requests = [self.request(key=f"writer-lock-{index:04d}") for index in range(3)]
+        threads = [threading.Thread(target=probe.execute, args=(DummyGrant(), item)) for item in requests]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertEqual(1, maximum)
+
+    def test_invalid_expected_head_and_idempotency_are_rejected(self):
+        request = self.request()
+        request["expected_head_oid"] = "short"
+        with self.assertRaisesRegex(CommitFileError, "writer_expected_head_rejected"):
+            self.writer().execute(DummyGrant(), request)
+        request = self.request()
+        request["idempotency_key"] = "../bad"
+        with self.assertRaisesRegex(CommitFileError, "writer_idempotency_key_rejected"):
+            self.writer().execute(DummyGrant(), request)
+
+    def test_invalid_message_and_nul_content_are_rejected(self):
+        request = self.request()
+        request["message"] = "bad\nmessage"
+        with self.assertRaisesRegex(CommitFileError, "writer_message_rejected"):
+            self.writer().execute(DummyGrant(), request)
+        request = self.request()
+        request["content"] = "bad\x00content"
+        with self.assertRaisesRegex(CommitFileError, "writer_content_rejected"):
+            self.writer().execute(DummyGrant(), request)
+
+    def test_force_flags_are_403_and_unknown_fields_are_rejected(self):
+        for field in ("force", "force_push"):
+            request = self.request(key=f"writer-force-{field}")
+            request[field] = True
+            with self.subTest(field=field), self.assertRaises(CommitFileError) as captured:
+                self.writer().execute(DummyGrant(), request)
+            self.assertEqual(403, captured.exception.status)
+        request = self.request(key="writer-extra-field")
+        request["operation"] = "commit_file.atomic"
+        with self.assertRaisesRegex(CommitFileError, "writer_request_fields_rejected"):
+            self.writer().execute(DummyGrant(), request)
+
+
+if __name__ == "__main__":
+    unittest.main()