governance: add dormant single-owner activation prerequisites

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

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

@@ -0,0 +1,11 @@
+# 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.
+
+It intentionally has no live activation entry point. Both systemd services are
+static, refuse manual start, carry no credential, and keep all write and
+activation flags false. A separate `SINGLE_OWNER_GOVERNANCE_ATOMIC_ACTIVATION`
+approval is required before any live status publication or protection change.

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/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/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,13 @@
+[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 = []
+
+[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_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,30 @@
+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((ROOT / "deploy" / "systemd").glob("aios-governance-*.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)
+
+
+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()