feat(control-plane): add governed host seed executor

active-runtime/app/aios_control_plane_host_seed_executor.py

@@ -0,0 +1,748 @@
+#!/usr/bin/env python3
+"""Server-local governed immutable Host seed executor.
+
+This executor is deliberately not a general-purpose privileged command runner.
+It polls one fixed Git ref/path, verifies a high-risk governed-task binding as
+the durable authorization evidence, materializes one exact candidate from the
+authoritative Git object database plus a declared patch/manifest, and atomically
+publishes the immutable release tree.
+
+It never starts/restarts services, never creates the continuation Task/Job,
+never writes Git refs, and never switches Production.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import argparse
+import fcntl
+import hashlib
+import json
+import os
+import pwd
+from pathlib import Path, PurePosixPath
+import re
+import shutil
+import stat
+import subprocess
+import tempfile
+from typing import Any, Mapping
+
+REQUEST_SCHEMA = "aios.control-plane-host-seed-request.v1"
+MANIFEST_SCHEMAS = {
+    "aios.control-plane-immutable-candidate.v5",
+    "aios.control-plane-immutable-candidate.v6",
+}
+OPERATION = "HOST_SEED_IMMUTABLE_CANDIDATE"
+REPOSITORY = "aios/aios"
+REQUEST_REF = "refs/heads/agent/host-seed-authorization-requests"
+REQUEST_PATH = "control-plane/host-seed/request.json"
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SHA_RE = re.compile(r"^[0-9a-f]{64}$")
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+RELEASE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{3,191}$")
+CANDIDATE_RE = re.compile(r"^V[0-9]+$")
+MAX_REQUEST_BYTES = 128 * 1024
+MAX_MANIFEST_BYTES = 4 * 1024 * 1024
+MAX_PATCH_BYTES = 2 * 1024 * 1024
+MAX_MACHINE_GOAL_BYTES = 512 * 1024
+MAX_BOOTSTRAP_BYTES = 512 * 1024
+
+
+class HostSeedError(RuntimeError):
+    pass
+
+
+def canonical(value: object) -> bytes:
+    return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
+
+
+def sha256(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def canonical_compact(value: object) -> bytes:
+    return json.dumps(
+        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+
+
+def _git(git_dir: Path, *args: str, input_bytes: bytes | None = None) -> bytes:
+    env = {"PATH": os.environ.get("PATH", "")}
+    proc = subprocess.run(
+        ["git", f"--git-dir={git_dir}", *args],
+        input=input_bytes,
+        capture_output=True,
+        timeout=120,
+        env=env,
+    )
+    if proc.returncode:
+        raise HostSeedError("GIT_" + (args[0] if args else "COMMAND").upper() + "_FAILED")
+    return proc.stdout
+
+
+def _git_text(git_dir: Path, *args: str) -> str:
+    return _git(git_dir, *args).decode("utf-8", "strict").strip()
+
+
+def _safe_repo_path(value: object, *, field: str) -> str:
+    if not isinstance(value, str) or not value or len(value) > 512:
+        raise HostSeedError(field + "_INVALID")
+    p = PurePosixPath(value)
+    if p.is_absolute() or any(part in {"", ".", ".."} for part in p.parts):
+        raise HostSeedError(field + "_INVALID")
+    normalized = p.as_posix()
+    if normalized.startswith(".git/") or normalized == ".git":
+        raise HostSeedError(field + "_INVALID")
+    return normalized
+
+
+def _read_git_blob(git_dir: Path, oid: str, path: str, *, maximum: int) -> bytes:
+    if not OID_RE.fullmatch(oid):
+        raise HostSeedError("PAYLOAD_COMMIT_INVALID")
+    raw = _git(git_dir, "show", f"{oid}:{path}")
+    if len(raw) <= 0 or len(raw) > maximum:
+        raise HostSeedError("GIT_BLOB_SIZE_INVALID")
+    return raw
+
+
+def _parse_json(raw: bytes, *, code: str) -> dict[str, Any]:
+    try:
+        value = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise HostSeedError(code) from exc
+    if not isinstance(value, dict):
+        raise HostSeedError(code)
+    return value
+
+
+def _parse_time(value: object, *, field: str) -> datetime:
+    if not isinstance(value, str) or not value:
+        raise HostSeedError(field + "_INVALID")
+    text = value[:-1] + "+00:00" if value.endswith("Z") else value
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError as exc:
+        raise HostSeedError(field + "_INVALID") from exc
+    if parsed.tzinfo is None:
+        raise HostSeedError(field + "_INVALID")
+    return parsed.astimezone(timezone.utc)
+
+
+def _regular_file_bytes(
+    path: Path, *, maximum: int, expected_uid: int | None = None
+) -> bytes:
+    if path.parent.is_symlink():
+        raise HostSeedError("AUTHORIZATION_BINDING_PARENT_SYMLINK_REJECTED")
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        fd = os.open(path, flags)
+    except OSError as exc:
+        raise HostSeedError("AUTHORIZATION_BINDING_UNAVAILABLE") from exc
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_size <= 0 or info.st_size > maximum:
+            raise HostSeedError("AUTHORIZATION_BINDING_INVALID")
+        if info.st_mode & 0o022:
+            raise HostSeedError("AUTHORIZATION_BINDING_WRITABLE_BY_GROUP_OR_OTHER")
+        if expected_uid is not None and info.st_uid != expected_uid:
+            raise HostSeedError("AUTHORIZATION_BINDING_OWNER_MISMATCH")
+        raw = os.read(fd, maximum + 1)
+        if len(raw) > maximum:
+            raise HostSeedError("AUTHORIZATION_BINDING_INVALID")
+        return raw
+    finally:
+        os.close(fd)
+
+
+def _atomic_json(path: Path, value: Mapping[str, Any], *, mode: int = 0o600) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise HostSeedError("STATE_SYMLINK_REJECTED")
+    fd, temporary = tempfile.mkstemp(prefix=".tmp-", dir=path.parent)
+    try:
+        os.fchmod(fd, mode)
+        with os.fdopen(fd, "wb", closefd=True) as stream:
+            stream.write(canonical(value))
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+        os.chmod(path, mode)
+        parent_fd = os.open(path.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)
+
+
+@dataclass(frozen=True)
+class Intent:
+    candidate: str
+    release_id: str
+    base_oid: str
+    payload_commit: str
+    patch_path: str
+    manifest_path: str
+    machine_goal_path: str
+    seed_bootstrap_path: str
+    artifact_sha256: str
+    manifest_sha256: str
+    machine_goal_sha256: str
+    release_tree_sha256: str
+    seed_bootstrap_sha256: str
+    target_host: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "candidate": self.candidate,
+            "release_id": self.release_id,
+            "base_oid": self.base_oid,
+            "payload_commit": self.payload_commit,
+            "patch_path": self.patch_path,
+            "manifest_path": self.manifest_path,
+            "machine_goal_path": self.machine_goal_path,
+            "seed_bootstrap_path": self.seed_bootstrap_path,
+            "artifact_sha256": self.artifact_sha256,
+            "manifest_sha256": self.manifest_sha256,
+            "machine_goal_sha256": self.machine_goal_sha256,
+            "release_tree_sha256": self.release_tree_sha256,
+            "seed_bootstrap_sha256": self.seed_bootstrap_sha256,
+            "target_host": self.target_host,
+        }
+
+    @property
+    def digest(self) -> str:
+        return sha256(canonical(self.as_dict()))
+
+
+@dataclass(frozen=True)
+class Request:
+    intent: Intent
+    authorization_task_id: str
+    authorization_binding_sha256: str
+    authorization_sha256: str
+    expires_at: datetime
+
+    @classmethod
+    def load(cls, raw: Mapping[str, Any], *, host_identity: str, now: datetime) -> "Request":
+        if set(raw) != {"schema", "operation", "repository", "intent", "authorization"}:
+            raise HostSeedError("REQUEST_FIELDS_INVALID")
+        if raw.get("schema") != REQUEST_SCHEMA or raw.get("operation") != OPERATION:
+            raise HostSeedError("REQUEST_CONTRACT_INVALID")
+        if raw.get("repository") != REPOSITORY:
+            raise HostSeedError("REQUEST_REPOSITORY_MISMATCH")
+        intent_raw = raw.get("intent")
+        if not isinstance(intent_raw, Mapping) or set(intent_raw) != {
+            "candidate", "release_id", "base_oid", "payload_commit", "patch_path",
+            "manifest_path", "machine_goal_path", "seed_bootstrap_path",
+            "artifact_sha256", "manifest_sha256", "machine_goal_sha256",
+            "release_tree_sha256", "seed_bootstrap_sha256", "target_host",
+        }:
+            raise HostSeedError("INTENT_FIELDS_INVALID")
+        candidate = str(intent_raw.get("candidate") or "")
+        release_id = str(intent_raw.get("release_id") or "")
+        base_oid = str(intent_raw.get("base_oid") or "")
+        payload_commit = str(intent_raw.get("payload_commit") or "")
+        artifact_sha = str(intent_raw.get("artifact_sha256") or "")
+        manifest_sha = str(intent_raw.get("manifest_sha256") or "")
+        machine_goal_sha = str(intent_raw.get("machine_goal_sha256") or "")
+        release_tree_sha = str(intent_raw.get("release_tree_sha256") or "")
+        seed_bootstrap_sha = str(intent_raw.get("seed_bootstrap_sha256") or "")
+        target_host = str(intent_raw.get("target_host") or "")
+        if not CANDIDATE_RE.fullmatch(candidate):
+            raise HostSeedError("CANDIDATE_INVALID")
+        if not RELEASE_RE.fullmatch(release_id):
+            raise HostSeedError("RELEASE_ID_INVALID")
+        if not OID_RE.fullmatch(base_oid) or not OID_RE.fullmatch(payload_commit):
+            raise HostSeedError("INTENT_OID_INVALID")
+        if not all(SHA_RE.fullmatch(x) for x in (
+            artifact_sha, manifest_sha, machine_goal_sha, release_tree_sha, seed_bootstrap_sha
+        )):
+            raise HostSeedError("INTENT_SHA_INVALID")
+        if target_host != host_identity:
+            raise HostSeedError("TARGET_HOST_MISMATCH")
+        intent = Intent(
+            candidate=candidate,
+            release_id=release_id,
+            base_oid=base_oid,
+            payload_commit=payload_commit,
+            patch_path=_safe_repo_path(intent_raw.get("patch_path"), field="PATCH_PATH"),
+            manifest_path=_safe_repo_path(intent_raw.get("manifest_path"), field="MANIFEST_PATH"),
+            machine_goal_path=_safe_repo_path(intent_raw.get("machine_goal_path"), field="MACHINE_GOAL_PATH"),
+            seed_bootstrap_path=_safe_repo_path(intent_raw.get("seed_bootstrap_path"), field="SEED_BOOTSTRAP_PATH"),
+            artifact_sha256=artifact_sha,
+            manifest_sha256=manifest_sha,
+            machine_goal_sha256=machine_goal_sha,
+            release_tree_sha256=release_tree_sha,
+            seed_bootstrap_sha256=seed_bootstrap_sha,
+            target_host=target_host,
+        )
+        auth = raw.get("authorization")
+        if not isinstance(auth, Mapping) or set(auth) != {
+            "task_id", "binding_sha256", "expires_at", "intent_sha256",
+            "authorization_sha256",
+        }:
+            raise HostSeedError("AUTHORIZATION_FIELDS_INVALID")
+        task_id = str(auth.get("task_id") or "")
+        binding_sha = str(auth.get("binding_sha256") or "")
+        authorization_sha = str(auth.get("authorization_sha256") or "")
+        if (
+            not TASK_RE.fullmatch(task_id)
+            or not SHA_RE.fullmatch(binding_sha)
+            or not SHA_RE.fullmatch(authorization_sha)
+        ):
+            raise HostSeedError("AUTHORIZATION_IDENTITY_INVALID")
+        if auth.get("intent_sha256") != intent.digest:
+            raise HostSeedError("AUTHORIZATION_INTENT_MISMATCH")
+        expires_text = str(auth.get("expires_at") or "")
+        expires_at = _parse_time(expires_text, field="AUTHORIZATION_EXPIRES_AT")
+        expected_authorization_sha = sha256(canonical({
+            "task_id": task_id,
+            "intent_sha256": intent.digest,
+            "expires_at": expires_text,
+        }))
+        if authorization_sha != expected_authorization_sha:
+            raise HostSeedError("AUTHORIZATION_DIGEST_MISMATCH")
+        if now >= expires_at:
+            raise HostSeedError("AUTHORIZATION_EXPIRED")
+        if (expires_at - now).total_seconds() > 3600:
+            raise HostSeedError("AUTHORIZATION_WINDOW_TOO_LARGE")
+        return cls(
+            intent=intent,
+            authorization_task_id=task_id,
+            authorization_binding_sha256=binding_sha,
+            authorization_sha256=authorization_sha,
+            expires_at=expires_at,
+        )
+
+
+@dataclass(frozen=True)
+class Config:
+    git_dir: Path
+    task_binding_root: Path
+    state_root: Path
+    release_root: Path
+    host_identity: str
+    task_binding_owner: str | None = "aios-bridge-staging"
+    request_ref: str = REQUEST_REF
+    request_path: str = REQUEST_PATH
+
+
+def _load_task_binding(config: Config, request: Request) -> dict[str, Any]:
+    safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", request.authorization_task_id).strip("-")[:120] or "task"
+    path = config.task_binding_root / f"{safe}.json"
+    expected_uid = None
+    if config.task_binding_owner:
+        try:
+            expected_uid = pwd.getpwnam(config.task_binding_owner).pw_uid
+        except KeyError as exc:
+            raise HostSeedError("AUTHORIZATION_BINDING_OWNER_UNKNOWN") from exc
+    raw = _regular_file_bytes(path, maximum=256 * 1024, expected_uid=expected_uid)
+    if sha256(raw) != request.authorization_binding_sha256:
+        raise HostSeedError("AUTHORIZATION_BINDING_HASH_MISMATCH")
+    binding = _parse_json(raw, code="AUTHORIZATION_BINDING_JSON_INVALID")
+    required = {
+        "task_id", "repository", "base_oid", "target_branch", "risk",
+        "description", "idempotency_key", "expected_branch_oid",
+        "branch_mode", "generation", "job_id",
+    }
+    if set(binding) != required:
+        raise HostSeedError("AUTHORIZATION_BINDING_FIELDS_INVALID")
+    if binding.get("task_id") != request.authorization_task_id:
+        raise HostSeedError("AUTHORIZATION_TASK_MISMATCH")
+    if binding.get("repository") != REPOSITORY:
+        raise HostSeedError("AUTHORIZATION_REPOSITORY_MISMATCH")
+    if binding.get("risk") != "high":
+        raise HostSeedError("AUTHORIZATION_RISK_MISMATCH")
+    if binding.get("description") != "HOST_SEED_AUTHORIZATION:" + request.authorization_sha256:
+        raise HostSeedError("AUTHORIZATION_DESCRIPTION_MISMATCH")
+    if not str(binding.get("target_branch") or "").startswith("agent/host-seed-authorization-"):
+        raise HostSeedError("AUTHORIZATION_BRANCH_MISMATCH")
+    if not isinstance(binding.get("generation"), int) or binding["generation"] < 1:
+        raise HostSeedError("AUTHORIZATION_GENERATION_INVALID")
+    return binding
+
+
+def _load_manifest(config: Config, request: Request) -> dict[str, Any]:
+    raw = _read_git_blob(
+        config.git_dir, request.intent.payload_commit, request.intent.manifest_path,
+        maximum=MAX_MANIFEST_BYTES,
+    )
+    if sha256(raw) != request.intent.manifest_sha256:
+        raise HostSeedError("MANIFEST_HASH_MISMATCH")
+    manifest = _parse_json(raw, code="MANIFEST_JSON_INVALID")
+    if manifest.get("schema") not in MANIFEST_SCHEMAS:
+        raise HostSeedError("MANIFEST_SCHEMA_REJECTED")
+    if manifest.get("candidate") != request.intent.candidate:
+        raise HostSeedError("MANIFEST_CANDIDATE_MISMATCH")
+    if manifest.get("repository") != REPOSITORY:
+        raise HostSeedError("MANIFEST_REPOSITORY_MISMATCH")
+    if manifest.get("base_oid") != request.intent.base_oid:
+        raise HostSeedError("MANIFEST_BASE_MISMATCH")
+    if manifest.get("release_id") != request.intent.release_id:
+        raise HostSeedError("MANIFEST_RELEASE_ID_MISMATCH")
+    members = manifest.get("package_members")
+    changes = manifest.get("changed_files")
+    if not isinstance(members, Mapping) or not isinstance(changes, list):
+        raise HostSeedError("MANIFEST_MEMBERS_INVALID")
+    return manifest
+
+
+def _load_identity_evidence(config: Config, request: Request) -> dict[str, Any]:
+    goal_raw = _read_git_blob(
+        config.git_dir, request.intent.payload_commit, request.intent.machine_goal_path,
+        maximum=MAX_MACHINE_GOAL_BYTES,
+    )
+    if sha256(goal_raw) != request.intent.machine_goal_sha256:
+        raise HostSeedError("MACHINE_GOAL_HASH_MISMATCH")
+    goal = _parse_json(goal_raw, code="MACHINE_GOAL_JSON_INVALID")
+    required_pairs = {
+        "candidate": request.intent.candidate,
+        "repository": REPOSITORY,
+        "base_oid": request.intent.base_oid,
+        "release_id": request.intent.release_id,
+        "artifact_sha256": request.intent.artifact_sha256,
+        "manifest_sha256": request.intent.manifest_sha256,
+        "release_tree_sha256": request.intent.release_tree_sha256,
+        "seed_bootstrap_sha256": request.intent.seed_bootstrap_sha256,
+    }
+    for key, expected in required_pairs.items():
+        if goal.get(key) != expected:
+            raise HostSeedError("MACHINE_GOAL_BINDING_MISMATCH:" + key)
+    if goal.get("host_seed_authorized") is not False:
+        raise HostSeedError("MACHINE_GOAL_HOST_SEED_FLAG_INVALID")
+    bootstrap = _read_git_blob(
+        config.git_dir, request.intent.payload_commit, request.intent.seed_bootstrap_path,
+        maximum=MAX_BOOTSTRAP_BYTES,
+    )
+    if sha256(bootstrap) != request.intent.seed_bootstrap_sha256:
+        raise HostSeedError("SEED_BOOTSTRAP_HASH_MISMATCH")
+    return goal
+
+
+def _load_patch(config: Config, request: Request) -> bytes:
+    patch = _read_git_blob(
+        config.git_dir, request.intent.payload_commit, request.intent.patch_path,
+        maximum=MAX_PATCH_BYTES,
+    )
+    if b"\x00" in patch:
+        raise HostSeedError("PATCH_BINARY_NUL_REJECTED")
+    return patch
+
+
+def _manifest_release_map(manifest: Mapping[str, Any]) -> dict[str, str]:
+    members = manifest["package_members"]
+    result: dict[str, str] = {}
+    prefix = "payload/repo/after/"
+    for name, digest in members.items():
+        if not isinstance(name, str) or not name.startswith(prefix) or not isinstance(digest, str) or not SHA_RE.fullmatch(digest):
+            continue
+        rel = _safe_repo_path(name[len(prefix):], field="RELEASE_MEMBER")
+        if rel in result:
+            raise HostSeedError("DUPLICATE_RELEASE_MEMBER")
+        result[rel] = digest
+    declared_count = manifest.get("release_member_count")
+    if not isinstance(declared_count, int) or declared_count != len(result):
+        raise HostSeedError("RELEASE_MEMBER_COUNT_MISMATCH")
+    return result
+
+
+def _expected_changes(manifest: Mapping[str, Any]) -> dict[str, tuple[str, str]]:
+    result: dict[str, tuple[str, str]] = {}
+    for entry in manifest["changed_files"]:
+        if not isinstance(entry, Mapping):
+            raise HostSeedError("CHANGED_FILE_ENTRY_INVALID")
+        rel = _safe_repo_path(entry.get("path"), field="CHANGED_PATH")
+        digest = str(entry.get("after_sha256") or "")
+        status = str(entry.get("status") or "")
+        if not SHA_RE.fullmatch(digest) or status not in {"modified", "added"} or rel in result:
+            raise HostSeedError("CHANGED_FILE_ENTRY_INVALID")
+        result[rel] = (digest, status)
+    if manifest.get("changed_file_count") != len(result):
+        raise HostSeedError("CHANGED_FILE_COUNT_MISMATCH")
+    return result
+
+
+def _regular_member(root: Path, relative: str) -> Path:
+    current = root
+    for part in PurePosixPath(relative).parts:
+        current = current / part
+        try:
+            info = current.lstat()
+        except FileNotFoundError as exc:
+            raise HostSeedError("RELEASE_MEMBER_MISSING:" + relative) from exc
+        if stat.S_ISLNK(info.st_mode):
+            raise HostSeedError("RELEASE_MEMBER_SYMLINK_REJECTED:" + relative)
+    if not current.is_file():
+        raise HostSeedError("RELEASE_MEMBER_NOT_REGULAR:" + relative)
+    return current
+
+
+def _materialize_and_verify(config: Config, request: Request, manifest: Mapping[str, Any], patch: bytes, workspace: Path) -> dict[str, str]:
+    if _git_text(config.git_dir, "cat-file", "-t", request.intent.base_oid) != "commit":
+        raise HostSeedError("BASE_COMMIT_NOT_FOUND")
+    if _git_text(config.git_dir, "cat-file", "-t", request.intent.payload_commit) != "commit":
+        raise HostSeedError("PAYLOAD_COMMIT_NOT_FOUND")
+    workspace.parent.mkdir(parents=True, exist_ok=True)
+    clone = subprocess.run(
+        ["git", "clone", "--no-checkout", "--shared", str(config.git_dir), str(workspace)],
+        capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
+    )
+    if clone.returncode:
+        raise HostSeedError("WORKSPACE_CLONE_FAILED")
+    checkout = subprocess.run(
+        ["git", "-C", str(workspace), "checkout", "--detach", request.intent.base_oid],
+        capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
+    )
+    if checkout.returncode:
+        raise HostSeedError("WORKSPACE_CHECKOUT_FAILED")
+    try:
+        proc = subprocess.run(
+            ["git", "-C", str(workspace), "apply", "--whitespace=nowarn", "-"],
+            input=patch, capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
+        )
+        if proc.returncode:
+            raise HostSeedError("PATCH_APPLY_FAILED")
+        changes = _expected_changes(manifest)
+        status_raw = subprocess.run(
+            ["git", "-C", str(workspace), "status", "--porcelain", "-z"],
+            check=True, capture_output=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
+        ).stdout
+        actual: set[str] = set()
+        chunks = status_raw.split(b"\0")
+        for chunk in chunks:
+            if not chunk:
+                continue
+            text = chunk.decode("utf-8", "surrogateescape")
+            if len(text) < 4:
+                raise HostSeedError("WORKTREE_STATUS_INVALID")
+            actual.add(text[3:].replace("\\", "/"))
+        if actual != set(changes):
+            raise HostSeedError("PATCH_CHANGESET_MISMATCH")
+        base_paths = {
+            _safe_repo_path(path, field="TRACKED_PATH")
+            for path in _git_text(config.git_dir, "ls-tree", "-r", "--name-only", request.intent.base_oid).splitlines()
+            if path
+        }
+        for rel, (digest, status) in changes.items():
+            if status == "modified" and rel not in base_paths:
+                raise HostSeedError("MODIFIED_PATH_NOT_IN_BASE:" + rel)
+            if status == "added" and rel in base_paths:
+                raise HostSeedError("ADDED_PATH_ALREADY_IN_BASE:" + rel)
+            p = _regular_member(workspace, rel)
+            if sha256(p.read_bytes()) != digest:
+                raise HostSeedError("CHANGED_FILE_HASH_MISMATCH:" + rel)
+        release_map = _manifest_release_map(manifest)
+        expected_release_paths = base_paths | {
+            rel for rel, (_digest, status) in changes.items() if status == "added"
+        }
+        if set(release_map) != expected_release_paths:
+            raise HostSeedError("RELEASE_MEMBER_SET_MISMATCH")
+        for rel, digest in release_map.items():
+            p = _regular_member(workspace, rel)
+            if sha256(p.read_bytes()) != digest:
+                raise HostSeedError("RELEASE_MEMBER_HASH_MISMATCH:" + rel)
+        tree_digest = sha256(canonical_compact({f"payload/repo/after/{k}": release_map[k] for k in sorted(release_map)}))
+        if tree_digest != request.intent.release_tree_sha256:
+            raise HostSeedError("RELEASE_TREE_HASH_MISMATCH")
+        if manifest.get("release_tree_sha256") not in {None, request.intent.release_tree_sha256}:
+            raise HostSeedError("MANIFEST_RELEASE_TREE_HASH_MISMATCH")
+        return release_map
+    except Exception:
+        raise
+
+
+def _publish_release(config: Config, request: Request, workspace: Path, release_map: Mapping[str, str]) -> Path:
+    if config.release_root.exists() and config.release_root.is_symlink():
+        raise HostSeedError("RELEASE_ROOT_SYMLINK_REJECTED")
+    final = config.release_root / request.intent.release_id
+    if final.exists():
+        if final.is_symlink() or not final.is_dir():
+            raise HostSeedError("RELEASE_DESTINATION_CONFLICT")
+        for rel, digest in release_map.items():
+            p = final / PurePosixPath(rel)
+            if p.is_symlink() or not p.is_file() or sha256(p.read_bytes()) != digest:
+                raise HostSeedError("EXISTING_RELEASE_HASH_MISMATCH")
+        return final
+
+    config.release_root.mkdir(parents=True, exist_ok=True)
+    os.chmod(config.release_root, 0o755)
+    staging = Path(tempfile.mkdtemp(prefix=".host-seed-", dir=config.release_root))
+    os.chmod(staging, 0o755)
+    try:
+        for rel, digest in release_map.items():
+            source = _regular_member(workspace, rel)
+            target = staging / PurePosixPath(rel)
+            target.parent.mkdir(parents=True, exist_ok=True)
+            current = target.parent
+            while current != staging:
+                os.chmod(current, 0o755)
+                current = current.parent
+            data = source.read_bytes()
+            if sha256(data) != digest:
+                raise HostSeedError("STAGING_SOURCE_DRIFT")
+            fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
+            try:
+                with os.fdopen(fd, "wb", closefd=True) as out:
+                    out.write(data)
+                    out.flush()
+                    os.fsync(out.fileno())
+            except Exception:
+                raise
+            source_mode = stat.S_IMODE(source.stat(follow_symlinks=False).st_mode)
+            os.chmod(target, source_mode & 0o755 or 0o644)
+        for rel, digest in release_map.items():
+            p = staging / PurePosixPath(rel)
+            if p.is_symlink() or sha256(p.read_bytes()) != digest:
+                raise HostSeedError("STAGING_VERIFY_FAILED")
+        os.replace(staging, final)
+        parent_fd = os.open(config.release_root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+        try:
+            os.fsync(parent_fd)
+        finally:
+            os.close(parent_fd)
+        return final
+    finally:
+        if staging.exists():
+            shutil.rmtree(staging)
+
+
+def execute(config: Config, *, now: datetime | None = None, require_root: bool = True) -> dict[str, Any]:
+    if require_root and os.geteuid() != 0:
+        raise HostSeedError("ROOT_REQUIRED")
+    current = now or datetime.now(timezone.utc)
+    config.state_root.mkdir(parents=True, exist_ok=True, mode=0o700)
+    lock_path = config.state_root / "executor.lock"
+    lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
+    try:
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+        ref_probe = subprocess.run(
+            ["git", f"--git-dir={config.git_dir}", "rev-parse", "--verify", config.request_ref],
+            capture_output=True, timeout=30, env={"PATH": os.environ.get("PATH", "")},
+        )
+        if ref_probe.returncode:
+            return {
+                "schema": "aios.control-plane-host-seed-poll.v1",
+                "state": "NO_REQUEST",
+                "request_ref": config.request_ref,
+                "service_mutation": False,
+                "task_creation": False,
+            }
+        request_oid = ref_probe.stdout.decode("utf-8", "strict").strip()
+        if not OID_RE.fullmatch(request_oid):
+            raise HostSeedError("REQUEST_REF_INVALID")
+        raw = _read_git_blob(config.git_dir, request_oid, config.request_path, maximum=MAX_REQUEST_BYTES)
+        request_value = _parse_json(raw, code="REQUEST_JSON_INVALID")
+        request = Request.load(request_value, host_identity=config.host_identity, now=current)
+        binding = _load_task_binding(config, request)
+        manifest = _load_manifest(config, request)
+        _load_identity_evidence(config, request)
+        patch = _load_patch(config, request)
+
+        op_id = request.intent.digest
+        receipt_path = config.state_root / "receipts" / f"{op_id}.json"
+        if receipt_path.exists():
+            receipt = _parse_json(_regular_file_bytes(receipt_path, maximum=256 * 1024), code="RECEIPT_INVALID")
+            if receipt.get("state") == "EXECUTED" and receipt.get("intent_sha256") == op_id:
+                return receipt
+            raise HostSeedError("OPERATION_STATE_CONFLICT")
+
+        _atomic_json(config.state_root / "operations" / f"{op_id}.json", {
+            "schema": "aios.control-plane-host-seed-operation.v1",
+            "state": "PENDING",
+            "intent_sha256": op_id,
+            "request_commit": request_oid,
+            "authorization_task_id": request.authorization_task_id,
+            "authorization_binding_sha256": request.authorization_binding_sha256,
+            "started_at": current.isoformat().replace("+00:00", "Z"),
+        })
+
+        workspace_root = config.state_root / "workspaces"
+        workspace_root.mkdir(parents=True, exist_ok=True)
+        workspace = workspace_root / f"seed-{op_id[:20]}"
+        if workspace.exists():
+            raise HostSeedError("WORKSPACE_ALREADY_EXISTS")
+        release_map: dict[str, str] | None = None
+        try:
+            release_map = _materialize_and_verify(config, request, manifest, patch, workspace)
+            destination = _publish_release(config, request, workspace, release_map)
+        finally:
+            if workspace.exists():
+                shutil.rmtree(workspace, ignore_errors=True)
+
+        receipt = {
+            "schema": "aios.control-plane-host-seed-receipt.v1",
+            "state": "EXECUTED",
+            "operation": OPERATION,
+            "intent_sha256": op_id,
+            "request_commit": request_oid,
+            "repository": REPOSITORY,
+            "target_host": request.intent.target_host,
+            "candidate": request.intent.candidate,
+            "release_id": request.intent.release_id,
+            "base_oid": request.intent.base_oid,
+            "payload_commit": request.intent.payload_commit,
+            "validation_artifact_sha256": request.intent.artifact_sha256,
+            "artifact_delivery_used": False,
+            "manifest_sha256": request.intent.manifest_sha256,
+            "machine_goal_sha256": request.intent.machine_goal_sha256,
+            "release_tree_sha256": request.intent.release_tree_sha256,
+            "seed_bootstrap_sha256": request.intent.seed_bootstrap_sha256,
+            "authorization_task_id": request.authorization_task_id,
+            "authorization_binding_sha256": request.authorization_binding_sha256,
+            "release_path": str(destination),
+            "release_member_count": len(release_map or {}),
+            "service_mutation": False,
+            "service_restart": False,
+            "projector_started": False,
+            "task_creation": False,
+            "production_write": False,
+            "main_write": False,
+            "tag_write": False,
+            "baseline_write": False,
+            "executed_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+        }
+        _atomic_json(receipt_path, receipt)
+        _atomic_json(config.state_root / "operations" / f"{op_id}.json", receipt)
+        return receipt
+    finally:
+        fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        os.close(lock_fd)
+
+
+def config_from_environment() -> Config:
+    return Config(
+        git_dir=Path(os.environ.get("AIOS_HOST_SEED_GIT_DIR", "/data/aios-gitops/repo.git")),
+        task_binding_root=Path(os.environ.get(
+            "AIOS_HOST_SEED_TASK_BINDING_ROOT",
+            "/var/lib/aios-gitops/bridge-v2-staging/task-bindings",
+        )),
+        state_root=Path(os.environ.get(
+            "AIOS_HOST_SEED_STATE_ROOT",
+            "/var/lib/aios-control-plane-host-seed-executor",
+        )),
+        release_root=Path(os.environ.get("AIOS_HOST_SEED_RELEASE_ROOT", "/opt/aios/releases")),
+        host_identity=os.environ.get("AIOS_HOST_IDENTITY", "cn-network-mkt"),
+        task_binding_owner=os.environ.get(
+            "AIOS_HOST_SEED_TASK_BINDING_OWNER", "aios-bridge-staging"
+        ) or None,
+        request_ref=os.environ.get("AIOS_HOST_SEED_REQUEST_REF", REQUEST_REF),
+        request_path=os.environ.get("AIOS_HOST_SEED_REQUEST_PATH", REQUEST_PATH),
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("command", choices=("execute",))
+    args = parser.parse_args()
+    if args.command == "execute":
+        result = execute(config_from_environment())
+        print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/aios_existing_branch_approval.py

@@ -1,352 +0,0 @@
-"""Cryptographic Owner-approval boundary for governed existing-branch publication.
-
-Only a complete signed envelope from an authoritative trust-root adapter is
-accepted.  References, chat assertions, and locally supplied keys are not an
-approval.  Missing public-key or revocation evidence fails closed.
-"""
-from __future__ import annotations
-
-import base64
-from dataclasses import dataclass
-from datetime import datetime, timezone
-import hashlib
-import json
-import os
-from pathlib import Path
-import re
-from typing import Any, Mapping, Protocol
-
-from cryptography.exceptions import InvalidSignature
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
-
-
-APPROVAL_SCHEMA = "aios.existing-branch-owner-approval.v1"
-PUBLISH_OPERATION = "publishExistingBranchGoverned"
-OID_RE = re.compile(r"^[0-9a-f]{40}$")
-DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
-IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._@:+/-]{0,254}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
-
-
-class OwnerApprovalRejected(RuntimeError):
-    """Stable, audit-safe error for a rejected Owner approval."""
-
-    def __init__(self, code: str) -> None:
-        super().__init__(code)
-        self.code = code
-
-
-def canonical_json(value: Any) -> bytes:
-    """Encode a signed payload without implementation-dependent spacing."""
-    try:
-        encoded = json.dumps(
-            value,
-            sort_keys=True,
-            separators=(",", ":"),
-            ensure_ascii=False,
-            allow_nan=False,
-        )
-    except (TypeError, ValueError) as exc:
-        raise OwnerApprovalRejected("OWNER_APPROVAL_CANONICALIZATION_REJECTED") from exc
-    return (encoded + "\n").encode("utf-8")
-
-
-def digest_json(value: Any) -> str:
-    return hashlib.sha256(canonical_json(value)).hexdigest()
-
-
-def _require_identifier(field: str, value: object) -> str:
-    if not isinstance(value, str) or not IDENTIFIER_RE.fullmatch(value):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return value
-
-
-def _require_digest(field: str, value: object) -> str:
-    if not isinstance(value, str) or not DIGEST_RE.fullmatch(value):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return value
-
-
-def _require_oid(field: str, value: object) -> str:
-    if not isinstance(value, str) or not OID_RE.fullmatch(value):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return value
-
-
-def _parse_utc(field: str, value: object) -> datetime:
-    if not isinstance(value, str):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    try:
-        instant = datetime.fromisoformat(value.replace("Z", "+00:00"))
-    except ValueError as exc:
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED") from exc
-    if instant.tzinfo is None:
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return instant.astimezone(timezone.utc)
-
-
-@dataclass(frozen=True, slots=True)
-class PublicationApprovalBinding:
-    repository: str
-    branch: str
-    expected_branch_oid: str
-    operation: str
-    request_digest: str
-    idempotency_digest: str
-
-    def validate(self) -> "PublicationApprovalBinding":
-        _require_identifier("repository", self.repository)
-        if not BRANCH_RE.fullmatch(self.branch):
-            raise OwnerApprovalRejected("OWNER_APPROVAL_BRANCH_REJECTED")
-        _require_oid("expected_branch_oid", self.expected_branch_oid)
-        if self.operation != PUBLISH_OPERATION:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_OPERATION_REJECTED")
-        _require_digest("request_digest", self.request_digest)
-        _require_digest("idempotency_digest", self.idempotency_digest)
-        return self
-
-    @property
-    def as_dict(self) -> dict[str, str]:
-        return {
-            "repository": self.repository,
-            "branch": self.branch,
-            "expected_branch_oid": self.expected_branch_oid,
-            "operation": self.operation,
-            "request_digest": self.request_digest,
-            "idempotency_digest": self.idempotency_digest,
-        }
-
-
-@dataclass(frozen=True, slots=True)
-class OwnerApprovalEnvelope:
-    approval_id: str
-    repository: str
-    branch: str
-    expected_branch_oid: str
-    operation: str
-    issued_at: str
-    not_before: str
-    expires_at: str
-    nonce: str
-    owner_identity: str
-    trust_domain: str
-    key_id: str
-    request_digest: str
-    idempotency_digest: str
-    signature: str
-
-    @property
-    def canonical_payload(self) -> dict[str, str]:
-        return {
-            "schema": APPROVAL_SCHEMA,
-            "approval_id": self.approval_id,
-            "repository": self.repository,
-            "branch": self.branch,
-            "expected_branch_oid": self.expected_branch_oid,
-            "operation": self.operation,
-            "issued_at": self.issued_at,
-            "not_before": self.not_before,
-            "expires_at": self.expires_at,
-            "nonce": self.nonce,
-            "owner_identity": self.owner_identity,
-            "trust_domain": self.trust_domain,
-            "key_id": self.key_id,
-            "request_digest": self.request_digest,
-            "idempotency_digest": self.idempotency_digest,
-        }
-
-    def canonical_bytes(self) -> bytes:
-        return canonical_json(self.canonical_payload)
-
-    @property
-    def fingerprint(self) -> str:
-        return hashlib.sha256(self.canonical_bytes()).hexdigest()
-
-    @property
-    def binding(self) -> PublicationApprovalBinding:
-        return PublicationApprovalBinding(
-            repository=self.repository,
-            branch=self.branch,
-            expected_branch_oid=self.expected_branch_oid,
-            operation=self.operation,
-            request_digest=self.request_digest,
-            idempotency_digest=self.idempotency_digest,
-        )
-
-    @classmethod
-    def from_mapping(cls, value: object) -> "OwnerApprovalEnvelope":
-        if not isinstance(value, Mapping):
-            raise OwnerApprovalRejected("OWNER_APPROVAL_ENVELOPE_REQUIRED")
-        expected_fields = {
-            "schema", "approval_id", "repository", "branch", "expected_branch_oid",
-            "operation", "issued_at", "not_before", "expires_at", "nonce",
-            "owner_identity", "trust_domain", "key_id", "request_digest",
-            "idempotency_digest", "signature",
-        }
-        if set(value) != expected_fields or value.get("schema") != APPROVAL_SCHEMA:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_ENVELOPE_REJECTED")
-        envelope = cls(
-            approval_id=_require_identifier("approval_id", value["approval_id"]),
-            repository=_require_identifier("repository", value["repository"]),
-            branch=str(value["branch"]),
-            expected_branch_oid=_require_oid("expected_branch_oid", value["expected_branch_oid"]),
-            operation=str(value["operation"]),
-            issued_at=str(value["issued_at"]),
-            not_before=str(value["not_before"]),
-            expires_at=str(value["expires_at"]),
-            nonce=_require_identifier("nonce", value["nonce"]),
-            owner_identity=_require_identifier("owner_identity", value["owner_identity"]),
-            trust_domain=_require_identifier("trust_domain", value["trust_domain"]),
-            key_id=_require_identifier("key_id", value["key_id"]),
-            request_digest=_require_digest("request_digest", value["request_digest"]),
-            idempotency_digest=_require_digest("idempotency_digest", value["idempotency_digest"]),
-            signature=str(value["signature"]),
-        )
-        envelope.binding.validate()
-        _parse_utc("issued_at", envelope.issued_at)
-        _parse_utc("not_before", envelope.not_before)
-        _parse_utc("expires_at", envelope.expires_at)
-        if not envelope.signature:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_SIGNATURE_REJECTED")
-        return envelope
-
-
-@dataclass(frozen=True, slots=True)
-class TrustRootEvidence:
-    """Public-key and revocation evidence from the authoritative trust root."""
-
-    key_id: str
-    owner_identity: str
-    trust_domain: str
-    public_key: bytes | None
-    authority_confirmed: bool
-    revocation_state: str
-
-
-class AuthoritativeTrustRootAdapter(Protocol):
-    def resolve(
-        self, *, key_id: str, owner_identity: str, trust_domain: str
-    ) -> TrustRootEvidence | None: ...
-
-
-class ReplayLedger(Protocol):
-    def consume(
-        self, approval: OwnerApprovalEnvelope, binding: PublicationApprovalBinding
-    ) -> None: ...
-
-
-class DurableNonceReplayLedger:
-    """Single-use nonce storage that remains conservative across restarts."""
-
-    def __init__(self, root: Path) -> None:
-        self.root = Path(root)
-
-    def consume(
-        self, approval: OwnerApprovalEnvelope, binding: PublicationApprovalBinding
-    ) -> None:
-        binding.validate()
-        if self.root.is_symlink():
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED")
-        try:
-            self.root.mkdir(parents=True, exist_ok=True)
-            if not self.root.is_dir() or self.root.is_symlink():
-                raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED")
-            name = hashlib.sha256(approval.nonce.encode("utf-8")).hexdigest()
-            record_path = self.root / name
-            flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
-            if hasattr(os, "O_NOFOLLOW"):
-                flags |= os.O_NOFOLLOW
-            descriptor = os.open(record_path, flags, 0o600)
-        except FileExistsError as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_REJECTED") from exc
-        except OwnerApprovalRejected:
-            raise
-        except OSError as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED") from exc
-        record = {
-            "schema": "aios.existing-branch-owner-approval-replay.v1",
-            "nonce_sha256": name,
-            "approval_fingerprint": approval.fingerprint,
-            "binding_digest": digest_json(binding.as_dict),
-        }
-        try:
-            with os.fdopen(descriptor, "wb", closefd=True) as stream:
-                stream.write(canonical_json(record))
-                stream.flush()
-                os.fsync(stream.fileno())
-            directory = os.open(self.root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
-            try:
-                os.fsync(directory)
-            finally:
-                os.close(directory)
-        except OSError as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED") from exc
-
-
-class OwnerApprovalVerifier:
-    def __init__(
-        self,
-        trust_root: AuthoritativeTrustRootAdapter,
-        *,
-        allowed_trust_domains: set[str] | frozenset[str],
-        maximum_lifetime_seconds: int = 900,
-    ) -> None:
-        if not allowed_trust_domains or maximum_lifetime_seconds < 1:
-            raise ValueError("OWNER_APPROVAL_VERIFIER_CONFIGURATION_REJECTED")
-        self.trust_root = trust_root
-        self.allowed_trust_domains = frozenset(allowed_trust_domains)
-        self.maximum_lifetime_seconds = maximum_lifetime_seconds
-
-    def verify(
-        self,
-        value: object,
-        binding: PublicationApprovalBinding,
-        *,
-        now: datetime | None = None,
-    ) -> OwnerApprovalEnvelope:
-        binding.validate()
-        approval = OwnerApprovalEnvelope.from_mapping(value)
-        if approval.binding != binding:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_BINDING_REJECTED")
-        instant = datetime.now(timezone.utc) if now is None else now
-        if instant.tzinfo is None:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        instant = instant.astimezone(timezone.utc)
-        issued_at = _parse_utc("issued_at", approval.issued_at)
-        not_before = _parse_utc("not_before", approval.not_before)
-        expires_at = _parse_utc("expires_at", approval.expires_at)
-        if not issued_at <= not_before <= expires_at:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        if not not_before <= instant <= expires_at:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        if (expires_at - issued_at).total_seconds() > self.maximum_lifetime_seconds:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        if approval.trust_domain not in self.allowed_trust_domains:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_DOMAIN_REJECTED")
-        evidence = self.trust_root.resolve(
-            key_id=approval.key_id,
-            owner_identity=approval.owner_identity,
-            trust_domain=approval.trust_domain,
-        )
-        if evidence is None or not evidence.authority_confirmed:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_EVIDENCE_MISSING")
-        if (
-            evidence.key_id != approval.key_id
-            or evidence.owner_identity != approval.owner_identity
-            or evidence.trust_domain != approval.trust_domain
-        ):
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_BINDING_REJECTED")
-        if evidence.revocation_state == "REVOKED":
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REVOKED")
-        if evidence.revocation_state != "ACTIVE":
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REVOCATION_EVIDENCE_MISSING")
-        if evidence.public_key is None or len(evidence.public_key) != 32:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_EVIDENCE_MISSING")
-        try:
-            signature = base64.b64decode(approval.signature, validate=True)
-            Ed25519PublicKey.from_public_bytes(evidence.public_key).verify(
-                signature, approval.canonical_bytes()
-            )
-        except (ValueError, TypeError, InvalidSignature) as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_SIGNATURE_REJECTED") from exc
-        return approval

active-runtime/app/aios_existing_branch_publisher.py

@@ -1,331 +0,0 @@
-"""Narrow governed publisher for an already-existing task branch.
-
-Its only writing port is a compare-and-swap publication after a complete
-signed Owner approval, a stable remote observation, and a single-use nonce
-claim.  It neither owns trust material nor accepts direct reference strings.
-"""
-from __future__ import annotations
-
-from dataclasses import dataclass
-import re
-from typing import Any, Mapping, Protocol
-
-from aios_existing_branch_approval import (
-    OwnerApprovalEnvelope,
-    OwnerApprovalRejected,
-    OwnerApprovalVerifier,
-    PublicationApprovalBinding,
-    ReplayLedger,
-    canonical_json,
-    digest_json,
-)
-
-
-REQUEST_SCHEMA = "aios.existing-branch-publication.v1"
-PUBLISH_OPERATION = "publishExistingBranchGoverned"
-OID_RE = re.compile(r"^[0-9a-f]{40}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
-
-
-class ExistingBranchPublicationRejected(RuntimeError):
-    def __init__(self, code: str) -> None:
-        super().__init__(code)
-        self.code = code
-
-
-class RemoteOidConflict(RuntimeError):
-    pass
-
-
-class RemoteObservationAmbiguous(RuntimeError):
-    pass
-
-
-@dataclass(frozen=True, slots=True)
-class ExistingBranchPublicationRequest:
-    repository: str
-    branch: str
-    expected_branch_oid: str
-    idempotency_key: str
-    payload: Mapping[str, Any]
-
-    @classmethod
-    def from_mapping(cls, value: object) -> "ExistingBranchPublicationRequest":
-        if not isinstance(value, Mapping):
-            raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
-        expected_fields = {
-            "schema",
-            "repository",
-            "branch",
-            "expected_branch_oid",
-            "operation",
-            "idempotency_key",
-            "payload",
-        }
-        if set(value) != expected_fields or value.get("schema") != REQUEST_SCHEMA:
-            raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
-        repository = value["repository"]
-        branch = value["branch"]
-        expected_oid = value["expected_branch_oid"]
-        key = value["idempotency_key"]
-        payload = value["payload"]
-        if (
-            not isinstance(repository, str)
-            or repository != "aios/aios"
-            or not BRANCH_RE.fullmatch(str(branch))
-            or not isinstance(expected_oid, str)
-            or not OID_RE.fullmatch(expected_oid)
-            or value.get("operation") != PUBLISH_OPERATION
-            or not isinstance(key, str)
-            or not key.strip()
-            or not isinstance(payload, Mapping)
-        ):
-            raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
-        try:
-            canonical_json(payload)
-        except OwnerApprovalRejected as exc:
-            raise ExistingBranchPublicationRejected("PUBLICATION_PAYLOAD_REJECTED") from exc
-        return cls(repository, str(branch), expected_oid, key, dict(payload))
-
-    @property
-    def request_digest(self) -> str:
-        return digest_json(
-            {
-                "schema": REQUEST_SCHEMA,
-                "repository": self.repository,
-                "branch": self.branch,
-                "expected_branch_oid": self.expected_branch_oid,
-                "operation": PUBLISH_OPERATION,
-                "payload": self.payload,
-            }
-        )
-
-    @property
-    def idempotency_digest(self) -> str:
-        return digest_json(
-            {
-                "idempotency_key": self.idempotency_key,
-                "request_digest": self.request_digest,
-            }
-        )
-
-    @property
-    def approval_binding(self) -> PublicationApprovalBinding:
-        return PublicationApprovalBinding(
-            repository=self.repository,
-            branch=self.branch,
-            expected_branch_oid=self.expected_branch_oid,
-            operation=PUBLISH_OPERATION,
-            request_digest=self.request_digest,
-            idempotency_digest=self.idempotency_digest,
-        )
-
-
-@dataclass(frozen=True, slots=True)
-class RemoteBranchObservation:
-    state: str
-    git_oid: str | None
-    api_oid: str | None
-    expected_commit_visible: bool | None
-
-
-@dataclass(frozen=True, slots=True)
-class RemotePublicationEvidence:
-    authority_oid: str | None
-    gitea_git_oid: str | None
-    gitea_api_oid: str | None
-    commit_visible: bool | None
-
-
-class ExistingBranchRemotePort(Protocol):
-    def observe(self, branch: str) -> RemoteBranchObservation: ...
-
-    def publish_compare_and_swap(
-        self, branch: str, expected_branch_oid: str, commit_oid: str
-    ) -> RemotePublicationEvidence: ...
-
-
-class AuthorityBranchPort(Protocol):
-    def read_branch_oid(self, branch: str) -> str | None: ...
-
-
-class CredentialProjection(Protocol):
-    def project(self) -> None: ...
-
-    def revoke(self) -> None: ...
-
-
-class PublicationWorkspace(Protocol):
-    def head_oid(self) -> str: ...
-
-    def commit_payload(self, payload: Mapping[str, Any]) -> str: ...
-
-
-class DetachedWorkspaceFactory(Protocol):
-    def open_detached(self, expected_branch_oid: str) -> PublicationWorkspace: ...
-
-
-class AuditSink(Protocol):
-    def record(self, event: Mapping[str, Any]) -> None: ...
-
-
-class ExistingBranchPublisher:
-    """Validate once, observe once, then make one bounded publication attempt."""
-
-    def __init__(
-        self,
-        *,
-        verifier: OwnerApprovalVerifier,
-        replay_ledger: ReplayLedger,
-        authority: AuthorityBranchPort,
-        remote: ExistingBranchRemotePort,
-        credentials: CredentialProjection,
-        workspaces: DetachedWorkspaceFactory,
-        audit: AuditSink,
-    ) -> None:
-        self.verifier = verifier
-        self.replay_ledger = replay_ledger
-        self.authority = authority
-        self.remote = remote
-        self.credentials = credentials
-        self.workspaces = workspaces
-        self.audit = audit
-
-    def _audit(
-        self,
-        *,
-        status: str,
-        request: ExistingBranchPublicationRequest,
-        approval: OwnerApprovalEnvelope | None = None,
-        commit_oid: str | None = None,
-    ) -> None:
-        event: dict[str, Any] = {
-            "event": "governed_existing_branch_publication",
-            "status": status,
-            "repository": request.repository,
-            "branch": request.branch,
-            "expected_branch_oid": request.expected_branch_oid,
-            "operation": PUBLISH_OPERATION,
-            "request_digest": request.request_digest,
-            "idempotency_digest": request.idempotency_digest,
-        }
-        if approval is not None:
-            event.update(
-                {
-                    "approval_id": approval.approval_id,
-                    "owner_identity": approval.owner_identity,
-                    "trust_domain": approval.trust_domain,
-                    "approval_fingerprint": approval.fingerprint,
-                }
-            )
-        if commit_oid is not None:
-            event["commit_oid"] = commit_oid
-        self.audit.record(event)
-
-    def _preflight_remote(
-        self, request: ExistingBranchPublicationRequest
-    ) -> RemoteBranchObservation | dict[str, str]:
-        try:
-            authority_oid = self.authority.read_branch_oid(request.branch)
-        except Exception:
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if authority_oid != request.expected_branch_oid:
-            return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-        try:
-            observation = self.remote.observe(request.branch)
-        except RemoteObservationAmbiguous:
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if observation.state == "AMBIGUOUS":
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if observation.state != "PRESENT":
-            return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-        if (
-            observation.git_oid is not None
-            and observation.api_oid is not None
-            and observation.git_oid != observation.api_oid
-        ):
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if (
-            observation.git_oid != request.expected_branch_oid
-            or observation.api_oid != request.expected_branch_oid
-        ):
-            return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-        if observation.expected_commit_visible is not True:
-            return {"status": "RECONCILIATION_REQUIRED"}
-        return observation
-
-    def publish(
-        self, request_value: object, approval_value: object
-    ) -> dict[str, str | bool]:
-        request = ExistingBranchPublicationRequest.from_mapping(request_value)
-        try:
-            approval = self.verifier.verify(approval_value, request.approval_binding)
-        except OwnerApprovalRejected:
-            self._audit(status="REJECTED", request=request)
-            raise
-        preflight = self._preflight_remote(request)
-        if isinstance(preflight, dict):
-            self._audit(status=preflight["status"], request=request, approval=approval)
-            return preflight
-        self.replay_ledger.consume(approval, request.approval_binding)
-        workspace = self.workspaces.open_detached(request.expected_branch_oid)
-        if workspace.head_oid() != request.expected_branch_oid:
-            self._audit(status="REJECTED", request=request, approval=approval)
-            raise ExistingBranchPublicationRejected("DETACHED_WORKSPACE_HEAD_REJECTED")
-        projected = False
-        try:
-            self.credentials.project()
-            projected = True
-            commit_oid = workspace.commit_payload(request.payload)
-            if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
-                raise ExistingBranchPublicationRejected("PUBLICATION_COMMIT_REJECTED")
-            try:
-                evidence = self.remote.publish_compare_and_swap(
-                    request.branch, request.expected_branch_oid, commit_oid
-                )
-            except RemoteOidConflict:
-                self._audit(
-                    status="MANUAL_OWNER_REVIEW_REQUIRED",
-                    request=request,
-                    approval=approval,
-                    commit_oid=commit_oid,
-                )
-                return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-            except RemoteObservationAmbiguous:
-                self._audit(
-                    status="RECONCILIATION_REQUIRED",
-                    request=request,
-                    approval=approval,
-                    commit_oid=commit_oid,
-                )
-                return {"status": "RECONCILIATION_REQUIRED"}
-            if (
-                evidence.authority_oid != commit_oid
-                or evidence.gitea_git_oid != commit_oid
-                or evidence.gitea_api_oid != commit_oid
-                or evidence.commit_visible is not True
-            ):
-                self._audit(
-                    status="RECONCILIATION_REQUIRED",
-                    request=request,
-                    approval=approval,
-                    commit_oid=commit_oid,
-                )
-                return {"status": "RECONCILIATION_REQUIRED"}
-            self._audit(
-                status="PUBLISHED",
-                request=request,
-                approval=approval,
-                commit_oid=commit_oid,
-            )
-            return {
-                "status": "PUBLISHED",
-                "commit_oid": commit_oid,
-                "authority_oid": evidence.authority_oid,
-                "gitea_git_oid": evidence.gitea_git_oid,
-                "gitea_api_oid": evidence.gitea_api_oid,
-                "commit_visible": evidence.commit_visible,
-            }
-        finally:
-            if projected:
-                self.credentials.revoke()

deploy/install/install_control_plane_host_seed_executor.sh

@@ -0,0 +1,52 @@
+#!/bin/sh
+set -eu
+
+if [ "$(id -u)" -ne 0 ]; then
+  echo "ROOT_REQUIRED" >&2
+  exit 20
+fi
+
+SOURCE_ROOT=${1:-}
+if [ -z "$SOURCE_ROOT" ]; then
+  echo "SOURCE_ROOT_REQUIRED" >&2
+  exit 21
+fi
+
+EXECUTOR="$SOURCE_ROOT/active-runtime/app/aios_control_plane_host_seed_executor.py"
+SERVICE="$SOURCE_ROOT/deploy/systemd/aios-control-plane-host-seed-executor.service"
+TIMER="$SOURCE_ROOT/deploy/systemd/aios-control-plane-host-seed-executor.timer"
+
+for file in "$EXECUTOR" "$SERVICE" "$TIMER"; do
+  if [ ! -f "$file" ] || [ -L "$file" ]; then
+    echo "SOURCE_FILE_INVALID:$file" >&2
+    exit 22
+  fi
+done
+
+AUTHORITY_GIT=/data/aios-gitops/repo.git
+REQUEST_REF=refs/heads/agent/host-seed-authorization-requests
+if [ ! -d "$AUTHORITY_GIT" ]; then
+  echo "AUTHORITATIVE_GIT_UNAVAILABLE" >&2
+  exit 23
+fi
+if ! id -u aios-bridge-staging >/dev/null 2>&1; then
+  echo "BRIDGE_SERVICE_ACCOUNT_UNAVAILABLE" >&2
+  exit 24
+fi
+if git --git-dir="$AUTHORITY_GIT" show-ref --verify --quiet "$REQUEST_REF"; then
+  echo "HOST_SEED_REQUEST_REF_ALREADY_PRESENT" >&2
+  exit 25
+fi
+
+install -d -m 0755 /usr/local/libexec
+install -m 0755 "$EXECUTOR" /usr/local/libexec/aios-control-plane-host-seed-executor
+install -m 0644 "$SERVICE" /etc/systemd/system/aios-control-plane-host-seed-executor.service
+install -m 0644 "$TIMER" /etc/systemd/system/aios-control-plane-host-seed-executor.timer
+install -d -m 0700 /var/lib/aios-control-plane-host-seed-executor
+
+systemctl daemon-reload
+systemctl enable --now aios-control-plane-host-seed-executor.timer
+systemctl is-enabled --quiet aios-control-plane-host-seed-executor.timer
+systemctl is-active --quiet aios-control-plane-host-seed-executor.timer
+
+echo "HOST_SEED_EXECUTOR_INSTALL=PASS"

deploy/install/rollback_control_plane_host_seed_executor.sh

@@ -0,0 +1,15 @@
+#!/bin/sh
+set -eu
+
+if [ "$(id -u)" -ne 0 ]; then
+  echo "ROOT_REQUIRED" >&2
+  exit 20
+fi
+
+systemctl disable --now aios-control-plane-host-seed-executor.timer 2>/dev/null || true
+rm -f /etc/systemd/system/aios-control-plane-host-seed-executor.timer
+rm -f /etc/systemd/system/aios-control-plane-host-seed-executor.service
+rm -f /usr/local/libexec/aios-control-plane-host-seed-executor
+systemctl daemon-reload
+
+echo "HOST_SEED_EXECUTOR_ROLLBACK=PASS"

deploy/systemd/aios-control-plane-host-seed-executor.service

@@ -0,0 +1,41 @@
+[Unit]
+Description=AIOS governed immutable Host seed executor
+After=local-fs.target
+
+[Service]
+Type=oneshot
+User=root
+Group=root
+UMask=0077
+Environment=PYTHONDONTWRITEBYTECODE=1
+Environment=AIOS_HOST_IDENTITY=cn-network-mkt
+Environment=AIOS_HOST_SEED_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_HOST_SEED_TASK_BINDING_ROOT=/var/lib/aios-gitops/bridge-v2-staging/task-bindings
+Environment=AIOS_HOST_SEED_TASK_BINDING_OWNER=aios-bridge-staging
+Environment=AIOS_HOST_SEED_STATE_ROOT=/var/lib/aios-control-plane-host-seed-executor
+Environment=AIOS_HOST_SEED_RELEASE_ROOT=/opt/aios/releases
+Environment=AIOS_HOST_SEED_REQUEST_REF=refs/heads/agent/host-seed-authorization-requests
+Environment=AIOS_HOST_SEED_REQUEST_PATH=control-plane/host-seed/request.json
+ExecStart=/usr/bin/python3 -B /usr/local/libexec/aios-control-plane-host-seed-executor execute
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+ProtectKernelLogs=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX
+IPAddressDeny=any
+CapabilityBoundingSet=CAP_DAC_OVERRIDE CAP_FOWNER CAP_CHOWN
+ReadOnlyPaths=/usr/local/libexec/aios-control-plane-host-seed-executor
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadOnlyPaths=/var/lib/aios-gitops/bridge-v2-staging/task-bindings
+ReadWritePaths=/var/lib/aios-control-plane-host-seed-executor
+ReadWritePaths=/opt/aios/releases

deploy/systemd/aios-control-plane-host-seed-executor.timer

@@ -0,0 +1,12 @@
+[Unit]
+Description=Poll governed immutable Host seed requests
+
+[Timer]
+OnBootSec=2min
+OnUnitActiveSec=30s
+AccuracySec=1s
+Persistent=true
+Unit=aios-control-plane-host-seed-executor.service
+
+[Install]
+WantedBy=timers.target

docs/superpowers/specs/2026-08-22-server-local-governed-host-seed-executor.md

@@ -0,0 +1,117 @@
+# Server-local governed immutable Host seed executor
+
+Date: 2026-08-22
+
+## Goal
+
+Provide a server-local, fail-closed way to publish an already validated
+immutable control-plane candidate without depending on Windows, an external
+Agent worker, SSH, or an arbitrary root command channel.
+
+The executor is deliberately narrower than deployment:
+
+- it reads only the authoritative local Git object database;
+- it consumes one fixed request ref and path;
+- it accepts only `HOST_SEED_IMMUTABLE_CANDIDATE`;
+- it requires a high-risk governed-task binding whose description binds the
+  exact candidate intent *and* authorization expiry;
+- it materializes `base_oid + candidate.patch`, verifies every release-member
+  SHA against the candidate manifest, and verifies the candidate Machine Goal;
+- it atomically publishes one immutable directory under `/opt/aios/releases`;
+- it never starts or restarts services, starts the Projector, creates the fixed
+  continuation Task/Job, switches Production, or writes `main`, tags, or the
+  baseline.
+
+## Canonical server path
+
+```text
+explicit Owner HOST_SEED authorization
+  -> high-risk governed authorization task binding
+  -> governed request commit on refs/heads/agent/host-seed-authorization-requests
+  -> root timer polls local authoritative Git
+  -> validate exact intent, expiry, binding SHA, candidate evidence
+  -> read payload commit from local Git
+  -> shared disposable clone at exact base
+  -> apply declared patch
+  -> verify exact changed set + all release-member bytes
+  -> atomic immutable publish
+  -> durable sanitized receipt
+```
+
+The Host seed request is not the fixed continuation Task. The fixed
+continuation Task remains a separate later gate.
+
+## Trust and authorization binding
+
+The request contains an immutable `intent` with:
+
+- candidate / release id;
+- base OID and payload commit;
+- fixed patch, manifest, Machine Goal and bootstrap paths;
+- validation artifact SHA;
+- Manifest, Machine Goal, release-tree and bootstrap SHAs;
+- exact target Host identity.
+
+`intent_sha256` is the canonical SHA of that object. Authorization adds
+`task_id` and `expires_at`; those are hashed into `authorization_sha256`.
+The server-side governed-task binding must contain:
+
+```text
+risk=high
+description=HOST_SEED_AUTHORIZATION:<authorization_sha256>
+target_branch=agent/host-seed-authorization-...
+```
+
+The request also binds the exact byte SHA of that server-side binding. This
+prevents extending the authorization window or changing the candidate without
+creating a new governed authorization binding.
+
+The executor additionally requires the binding file to be a non-symlink
+regular file, not group/world writable, and owned by the configured Bridge
+service account in production.
+
+## Delivery model
+
+The live delivery payload is server-native Git data, not the external ZIP.
+The payload commit carries:
+
+```text
+control-plane/candidates/<candidate>/candidate.patch
+control-plane/candidates/<candidate>/implementation_manifest_*.json
+control-plane/candidates/<candidate>/machine_goal_*.json
+control-plane/candidates/<candidate>/aios_control_plane_immutable_seed_*.py
+```
+
+The external ZIP SHA remains bound as validation evidence through Machine
+Goal, but the executor records `artifact_delivery_used=false`: it does not
+claim that ZIP bytes were copied to the Host.
+
+The release tree itself is reconstructed and verified from authoritative Git
+objects plus the declared patch.
+
+## Idempotency and recovery
+
+Before publishing, the executor writes a durable `PENDING` operation record
+keyed by the exact intent SHA. The final receipt is `EXECUTED`.
+
+If the immutable release directory already exists, every member must match the
+manifest before the operation is accepted as idempotent. Any drift fails
+closed.
+
+A missing request ref is a normal `NO_REQUEST` poll and has no side effects.
+
+## Systemd confinement
+
+The service:
+
+- runs as root only because immutable release publication requires it;
+- has no network access;
+- reads authoritative Git and Bridge task bindings read-only;
+- can write only its private state root and `/opt/aios/releases`;
+- has no service-control, Projector, Task, Git-ref, Production, tag or baseline
+  operation in its code path.
+
+The timer polls every 30 seconds. Installing/enabling this executor is itself a
+separate governed control-plane deployment. Installing it does **not** grant a
+Host seed authorization and does not seed any candidate while the fixed
+request ref is absent.

tests/test_control_plane_host_seed_executor.py

@@ -0,0 +1,328 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import shutil
+import subprocess
+
+import pytest
+
+
+MODULE_PATH = Path(__file__).parents[1] / "active-runtime" / "app" / "aios_control_plane_host_seed_executor.py"
+spec = importlib.util.spec_from_file_location("host_seed_executor", MODULE_PATH)
+m = importlib.util.module_from_spec(spec)
+assert spec and spec.loader
+import sys
+sys.modules[spec.name] = m
+spec.loader.exec_module(m)
+
+
+def run(*args: str, cwd: Path) -> str:
+    return subprocess.check_output(args, cwd=cwd, text=True).strip()
+
+
+def sha(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def canonical(value: object) -> bytes:
+    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
+
+
+def setup_case(tmp_path: Path, *, risk: str = "high", tamper_bootstrap: bool = False):
+    repo = tmp_path / "src"
+    bare = tmp_path / "repo.git"
+    repo.mkdir()
+    run("git", "init", "-b", "main", cwd=repo)
+    run("git", "config", "user.email", "test@example.invalid", cwd=repo)
+    run("git", "config", "user.name", "test", cwd=repo)
+    (repo / "a.txt").write_text("base\n", encoding="utf-8")
+    (repo / "bin.sh").write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
+    (repo / "bin.sh").chmod(0o755)
+    run("git", "add", "-A", cwd=repo)
+    run("git", "commit", "-m", "base", cwd=repo)
+    base = run("git", "rev-parse", "HEAD", cwd=repo)
+    run("git", "clone", "--bare", str(repo), str(bare), cwd=tmp_path)
+
+    (repo / "a.txt").write_text("changed\n", encoding="utf-8")
+    (repo / "c.txt").write_text("added\n", encoding="utf-8")
+    run("git", "add", "-N", "c.txt", cwd=repo)
+    patch = subprocess.check_output(["git", "diff", "--binary", base], cwd=repo)
+    after = {
+        "a.txt": sha((repo / "a.txt").read_bytes()),
+        "bin.sh": sha((repo / "bin.sh").read_bytes()),
+        "c.txt": sha((repo / "c.txt").read_bytes()),
+    }
+    package_members = {f"payload/repo/after/{k}": v for k, v in after.items()}
+    release_tree = sha(json.dumps({k: package_members[k] for k in sorted(package_members)}, sort_keys=True, separators=(",", ":")).encode())
+    manifest = {
+        "schema": "aios.control-plane-immutable-candidate.v6",
+        "candidate": "V9",
+        "repository": "aios/aios",
+        "base_oid": base,
+        "release_id": "control-plane-test-v9",
+        "changed_file_count": 2,
+        "changed_files": [
+            {"path": "a.txt", "status": "modified", "after_sha256": after["a.txt"]},
+            {"path": "c.txt", "status": "added", "after_sha256": after["c.txt"]},
+        ],
+        "release_member_count": 3,
+        "package_members": package_members,
+        "release_tree_sha256": release_tree,
+    }
+    manifest_raw = canonical(manifest)
+    bootstrap = b"#!/usr/bin/env python3\nprint('seed helper evidence only')\n"
+    bootstrap_sha = sha(bootstrap)
+    artifact_sha = "a" * 64
+    goal = {
+        "schema": "aios.control-plane-immutable-candidate-goal.v6",
+        "candidate": "V9",
+        "repository": "aios/aios",
+        "base_oid": base,
+        "release_id": "control-plane-test-v9",
+        "artifact_sha256": artifact_sha,
+        "manifest_sha256": sha(manifest_raw),
+        "release_tree_sha256": release_tree,
+        "seed_bootstrap_sha256": bootstrap_sha,
+        "host_seed_authorized": False,
+    }
+    goal_raw = canonical(goal)
+
+    # Reset the source worktree to base; delivery files live in a separate commit.
+    run("git", "reset", "--hard", base, cwd=repo)
+    delivery = repo / "control-plane" / "candidates" / "v9"
+    delivery.mkdir(parents=True)
+    (delivery / "candidate.patch").write_bytes(patch)
+    (delivery / "implementation_manifest_v9.json").write_bytes(manifest_raw)
+    (delivery / "machine_goal_v9.json").write_bytes(goal_raw)
+    (delivery / "aios_control_plane_immutable_seed_v9.py").write_bytes(
+        b"tampered\n" if tamper_bootstrap else bootstrap
+    )
+    run("git", "add", "-A", cwd=repo)
+    run("git", "commit", "-m", "candidate payload", cwd=repo)
+    payload_commit = run("git", "rev-parse", "HEAD", cwd=repo)
+    run("git", "push", str(bare), f"{payload_commit}:refs/heads/agent/candidate-payload", cwd=repo)
+
+    intent = {
+        "candidate": "V9",
+        "release_id": "control-plane-test-v9",
+        "base_oid": base,
+        "payload_commit": payload_commit,
+        "patch_path": "control-plane/candidates/v9/candidate.patch",
+        "manifest_path": "control-plane/candidates/v9/implementation_manifest_v9.json",
+        "machine_goal_path": "control-plane/candidates/v9/machine_goal_v9.json",
+        "seed_bootstrap_path": "control-plane/candidates/v9/aios_control_plane_immutable_seed_v9.py",
+        "artifact_sha256": artifact_sha,
+        "manifest_sha256": sha(manifest_raw),
+        "machine_goal_sha256": sha(goal_raw),
+        "release_tree_sha256": release_tree,
+        "seed_bootstrap_sha256": bootstrap_sha,
+        "target_host": "cn-network-mkt",
+    }
+    intent_digest = sha(canonical(intent))
+    task_id = "host-seed-authz-v9-cn-network-mkt"
+    now = datetime(2026, 8, 22, 12, 0, tzinfo=timezone.utc)
+    expires_text = (now + timedelta(minutes=30)).isoformat().replace("+00:00", "Z")
+    authorization_digest = sha(canonical({
+        "task_id": task_id,
+        "intent_sha256": intent_digest,
+        "expires_at": expires_text,
+    }))
+    binding = {
+        "task_id": task_id,
+        "repository": "aios/aios",
+        "base_oid": payload_commit,
+        "target_branch": "agent/host-seed-authorization-v9-cn-network-mkt",
+        "risk": risk,
+        "description": "HOST_SEED_AUTHORIZATION:" + authorization_digest,
+        "idempotency_key": "host-seed-authz-v9-1",
+        "expected_branch_oid": payload_commit,
+        "branch_mode": "CREATE_NEW",
+        "generation": 1,
+        "job_id": "job-" + "1" * 32,
+    }
+    binding_root = tmp_path / "bindings"
+    binding_root.mkdir()
+    binding_raw = json.dumps(binding, sort_keys=True).encode()
+    (binding_root / f"{task_id}.json").write_bytes(binding_raw)
+    (binding_root / f"{task_id}.json").chmod(0o600)
+
+    request = {
+        "schema": "aios.control-plane-host-seed-request.v1",
+        "operation": "HOST_SEED_IMMUTABLE_CANDIDATE",
+        "repository": "aios/aios",
+        "intent": intent,
+        "authorization": {
+            "task_id": task_id,
+            "binding_sha256": sha(binding_raw),
+            "expires_at": expires_text,
+            "intent_sha256": intent_digest,
+            "authorization_sha256": authorization_digest,
+        },
+    }
+    request_path = repo / "control-plane" / "host-seed" / "request.json"
+    request_path.parent.mkdir(parents=True, exist_ok=True)
+    request_path.write_bytes(canonical(request))
+    run("git", "add", str(request_path.relative_to(repo)), cwd=repo)
+    run("git", "commit", "-m", "host seed request", cwd=repo)
+    request_commit = run("git", "rev-parse", "HEAD", cwd=repo)
+    run("git", "push", str(bare), f"{request_commit}:{m.REQUEST_REF}", cwd=repo)
+
+    config = m.Config(
+        git_dir=bare,
+        task_binding_root=binding_root,
+        state_root=tmp_path / "state",
+        release_root=tmp_path / "releases",
+        host_identity="cn-network-mkt",
+        task_binding_owner=None,
+    )
+    return config, now, after, request
+
+
+def test_execute_publishes_exact_release_and_is_idempotent(tmp_path: Path):
+    config, now, after, request = setup_case(tmp_path)
+    receipt = m.execute(config, now=now, require_root=False)
+    assert receipt["state"] == "EXECUTED"
+    assert receipt["service_mutation"] is False
+    assert receipt["service_restart"] is False
+    assert receipt["projector_started"] is False
+    assert receipt["task_creation"] is False
+    assert receipt["production_write"] is False
+    final = config.release_root / "control-plane-test-v9"
+    assert final.is_dir() and not final.is_symlink()
+    assert final.stat().st_mode & 0o777 == 0o755
+    assert sha((final / "a.txt").read_bytes()) == after["a.txt"]
+    assert sha((final / "bin.sh").read_bytes()) == after["bin.sh"]
+    assert sha((final / "c.txt").read_bytes()) == after["c.txt"]
+    assert (final / "bin.sh").stat().st_mode & 0o111
+    again = m.execute(config, now=now, require_root=False)
+    assert again == receipt
+
+
+def test_execute_rejects_non_high_risk_authorization(tmp_path: Path):
+    config, now, _, _ = setup_case(tmp_path, risk="engineering")
+    with pytest.raises(m.HostSeedError, match="AUTHORIZATION_RISK_MISMATCH"):
+        m.execute(config, now=now, require_root=False)
+    assert not config.release_root.exists()
+
+
+def test_execute_rejects_tampered_seed_bootstrap(tmp_path: Path):
+    config, now, _, _ = setup_case(tmp_path, tamper_bootstrap=True)
+    with pytest.raises(m.HostSeedError, match="SEED_BOOTSTRAP_HASH_MISMATCH"):
+        m.execute(config, now=now, require_root=False)
+    assert not config.release_root.exists()
+
+
+def test_request_rejects_wrong_host(tmp_path: Path):
+    config, now, _, _ = setup_case(tmp_path)
+    config = m.Config(
+        git_dir=config.git_dir,
+        task_binding_root=config.task_binding_root,
+        state_root=config.state_root,
+        release_root=config.release_root,
+        host_identity="other-host",
+        task_binding_owner=None,
+    )
+    with pytest.raises(m.HostSeedError, match="TARGET_HOST_MISMATCH"):
+        m.execute(config, now=now, require_root=False)
+
+
+def test_existing_release_drift_is_fail_closed(tmp_path: Path):
+    config, now, _, _ = setup_case(tmp_path)
+    first = m.execute(config, now=now, require_root=False)
+    assert first["state"] == "EXECUTED"
+    (config.release_root / "control-plane-test-v9" / "a.txt").write_text("drift\n")
+    # remove durable receipt so existing-release verification path is exercised.
+    for p in (config.state_root / "receipts").glob("*.json"):
+        p.unlink()
+    with pytest.raises(m.HostSeedError, match="EXISTING_RELEASE_HASH_MISMATCH"):
+        m.execute(config, now=now, require_root=False)
+
+
+def test_no_request_ref_is_safe_noop(tmp_path: Path):
+    repo = tmp_path / "src"
+    bare = tmp_path / "repo.git"
+    repo.mkdir()
+    run("git", "init", "-b", "main", cwd=repo)
+    run("git", "config", "user.email", "test@example.invalid", cwd=repo)
+    run("git", "config", "user.name", "test", cwd=repo)
+    (repo / "a").write_text("x\n")
+    run("git", "add", "a", cwd=repo)
+    run("git", "commit", "-m", "base", cwd=repo)
+    run("git", "clone", "--bare", str(repo), str(bare), cwd=tmp_path)
+    cfg = m.Config(
+        git_dir=bare,
+        task_binding_root=tmp_path / "bindings",
+        state_root=tmp_path / "state",
+        release_root=tmp_path / "releases",
+        host_identity="cn-network-mkt",
+        task_binding_owner=None,
+    )
+    result = m.execute(cfg, now=datetime(2026, 8, 22, tzinfo=timezone.utc), require_root=False)
+    assert result["state"] == "NO_REQUEST"
+    assert result["service_mutation"] is False
+    assert result["task_creation"] is False
+    assert not cfg.release_root.exists()
+
+
+def test_authorization_expiry_cannot_be_extended_without_new_binding(tmp_path: Path):
+    config, now, _, _ = setup_case(tmp_path)
+    # Rewrite only the request expiry and recompute the request commit, but keep
+    # the original governed-task binding description. The authorization digest
+    # therefore no longer matches the binding and must fail closed.
+    repo = tmp_path / "src"
+    req_path = repo / "control-plane" / "host-seed" / "request.json"
+    value = json.loads(req_path.read_text())
+    value["authorization"]["expires_at"] = (now + timedelta(minutes=45)).isoformat().replace("+00:00", "Z")
+    value["authorization"]["authorization_sha256"] = sha(canonical({
+        "task_id": value["authorization"]["task_id"],
+        "intent_sha256": value["authorization"]["intent_sha256"],
+        "expires_at": value["authorization"]["expires_at"],
+    }))
+    req_path.write_bytes(canonical(value))
+    run("git", "add", str(req_path.relative_to(repo)), cwd=repo)
+    run("git", "commit", "-m", "tamper expiry", cwd=repo)
+    new_commit = run("git", "rev-parse", "HEAD", cwd=repo)
+    run("git", "push", "--force", str(config.git_dir), f"{new_commit}:{m.REQUEST_REF}", cwd=repo)
+    with pytest.raises(m.HostSeedError, match="AUTHORIZATION_DESCRIPTION_MISMATCH"):
+        m.execute(config, now=now, require_root=False)
+
+
+def test_systemd_contract_has_no_network_or_implicit_activation():
+    root = Path(__file__).parents[1]
+    service = (root / "deploy/systemd/aios-control-plane-host-seed-executor.service").read_text()
+    timer = (root / "deploy/systemd/aios-control-plane-host-seed-executor.timer").read_text()
+    assert "User=root" in service
+    assert "IPAddressDeny=any" in service
+    assert "RestrictAddressFamilies=AF_UNIX" in service
+    assert "ReadOnlyPaths=/data/aios-gitops/repo.git" in service
+    assert "ReadWritePaths=/opt/aios/releases" in service
+    assert "systemctl start" not in service
+    assert "projector" not in service.lower()
+    assert "OnUnitActiveSec=30s" in timer
+
+
+def test_release_root_symlink_is_rejected(tmp_path: Path):
+    config, now, _, _ = setup_case(tmp_path)
+    real = tmp_path / "real-releases"
+    real.mkdir()
+    config.release_root.symlink_to(real, target_is_directory=True)
+    with pytest.raises(m.HostSeedError, match="RELEASE_ROOT_SYMLINK_REJECTED"):
+        m.execute(config, now=now, require_root=False)
+
+def test_install_and_rollback_scripts_have_valid_bash_syntax():
+    root = Path(__file__).parents[1]
+    for relative in (
+        "deploy/install/install_control_plane_host_seed_executor.sh",
+        "deploy/install/rollback_control_plane_host_seed_executor.sh",
+    ):
+        subprocess.run(
+            ["/usr/bin/bash", "-n", str(root / relative)],
+            check=True,
+            text=True,
+            capture_output=True,
+        )
+

tests/test_existing_branch_owner_approval.py

@@ -1,285 +0,0 @@
-from __future__ import annotations
-
-import base64
-from datetime import datetime, timedelta, timezone
-import json
-
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-import pytest
-
-from aios_existing_branch_approval import (
-    DurableNonceReplayLedger,
-    OwnerApprovalEnvelope,
-    OwnerApprovalRejected,
-    OwnerApprovalVerifier,
-    TrustRootEvidence,
-)
-from aios_existing_branch_publisher import (
-    ExistingBranchPublicationRejected,
-    ExistingBranchPublisher,
-    RemoteBranchObservation,
-    RemotePublicationEvidence,
-)
-
-
-BASE_OID = "a" * 40
-COMMIT_OID = "b" * 40
-BRANCH = "agent/owner-approval-contract-20260810"
-
-
-class TrustRoot:
-    def __init__(self, public_key: bytes, *, revocation: str = "ACTIVE") -> None:
-        self.public_key = public_key
-        self.revocation = revocation
-        self.calls: list[str] = []
-
-    def resolve(self, *, key_id: str, owner_identity: str, trust_domain: str):
-        self.calls.append("trust-root")
-        return TrustRootEvidence(
-            key_id=key_id,
-            owner_identity=owner_identity,
-            trust_domain=trust_domain,
-            public_key=self.public_key,
-            authority_confirmed=True,
-            revocation_state=self.revocation,
-        )
-
-
-class Credentials:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-
-    def project(self) -> None:
-        self.events.append("credential-project")
-
-    def revoke(self) -> None:
-        self.events.append("credential-revoke")
-
-
-class Workspace:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-
-    def head_oid(self) -> str:
-        self.events.append("workspace-head")
-        return BASE_OID
-
-    def commit_payload(self, payload):
-        self.events.append("workspace-commit")
-        return COMMIT_OID
-
-
-class Workspaces:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-
-    def open_detached(self, expected_branch_oid: str) -> Workspace:
-        assert expected_branch_oid == BASE_OID
-        self.events.append("workspace-open")
-        return Workspace(self.events)
-
-
-class Remote:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-        self.observation = RemoteBranchObservation("PRESENT", BASE_OID, BASE_OID, True)
-        self.authority_oid = BASE_OID
-        self.pushes: list[tuple[str, str, str]] = []
-
-    def read_branch_oid(self, branch: str) -> str:
-        self.events.append("authority-observe")
-        assert branch == BRANCH
-        return self.authority_oid
-
-    def observe(self, branch: str) -> RemoteBranchObservation:
-        self.events.append("remote-observe")
-        assert branch == BRANCH
-        return self.observation
-
-    def publish_compare_and_swap(self, branch: str, expected: str, commit: str):
-        self.events.append("remote-publish")
-        self.pushes.append((branch, expected, commit))
-        return RemotePublicationEvidence(COMMIT_OID, COMMIT_OID, COMMIT_OID, True)
-
-
-class Audit:
-    def __init__(self) -> None:
-        self.rows: list[dict] = []
-
-    def record(self, event) -> None:
-        self.rows.append(dict(event))
-
-
-def request(*, payload=None, branch=BRANCH, expected_oid=BASE_OID):
-    return {
-        "schema": "aios.existing-branch-publication.v1",
-        "repository": "aios/aios",
-        "branch": branch,
-        "expected_branch_oid": expected_oid,
-        "operation": "publishExistingBranchGoverned",
-        "idempotency_key": "owner-approval-binding-20260810",
-        "payload": {"files": ["active-runtime/app/example.py"]} if payload is None else payload,
-    }
-
-
-def signed_approval(private: Ed25519PrivateKey, request_value, **replacements):
-    from aios_existing_branch_publisher import ExistingBranchPublicationRequest
-
-    parsed = ExistingBranchPublicationRequest.from_mapping(request_value)
-    now = datetime.now(timezone.utc)
-    raw = {
-        "schema": "aios.existing-branch-owner-approval.v1",
-        "approval_id": "approval-owner-binding-20260810",
-        "repository": parsed.repository,
-        "branch": parsed.branch,
-        "expected_branch_oid": parsed.expected_branch_oid,
-        "operation": "publishExistingBranchGoverned",
-        "issued_at": (now - timedelta(seconds=30)).isoformat(),
-        "not_before": (now - timedelta(seconds=1)).isoformat(),
-        "expires_at": (now + timedelta(minutes=5)).isoformat(),
-        "nonce": "nonce-owner-binding-20260810",
-        "owner_identity": "owner.aios",
-        "trust_domain": "aios-owner-root",
-        "key_id": "owner-key-20260810",
-        "request_digest": parsed.request_digest,
-        "idempotency_digest": parsed.idempotency_digest,
-        "signature": "AA==",
-    }
-    raw.update(replacements)
-    envelope = OwnerApprovalEnvelope.from_mapping(raw)
-    raw["signature"] = base64.b64encode(private.sign(envelope.canonical_bytes())).decode("ascii")
-    return raw
-
-
-def service(tmp_path, *, revocation="ACTIVE"):
-    private = Ed25519PrivateKey.generate()
-    public_key = private.public_key().public_bytes(
-        serialization.Encoding.Raw, serialization.PublicFormat.Raw
-    )
-    root = TrustRoot(public_key, revocation=revocation)
-    events: list[str] = []
-    remote = Remote(events)
-    audit = Audit()
-    publisher = ExistingBranchPublisher(
-        verifier=OwnerApprovalVerifier(root, allowed_trust_domains={"aios-owner-root"}),
-        replay_ledger=DurableNonceReplayLedger(tmp_path / "replay"),
-        authority=remote,
-        remote=remote,
-        credentials=Credentials(events),
-        workspaces=Workspaces(events),
-        audit=audit,
-    )
-    return private, root, events, remote, audit, publisher
-
-
-def test_valid_signed_owner_approval_is_verified_before_capability_projection(tmp_path):
-    private, root, events, remote, audit, publisher = service(tmp_path)
-    value = request()
-    approval = signed_approval(private, value)
-
-    result = publisher.publish(value, approval)
-
-    assert result["status"] == "PUBLISHED"
-    assert result["commit_oid"] == COMMIT_OID
-    assert remote.pushes == [(BRANCH, BASE_OID, COMMIT_OID)]
-    assert events == [
-        "authority-observe",
-        "remote-observe",
-        "workspace-open",
-        "workspace-head",
-        "credential-project",
-        "workspace-commit",
-        "remote-publish",
-        "credential-revoke",
-    ]
-    assert root.calls == ["trust-root"]
-    assert approval["signature"] not in json.dumps(audit.rows)
-    assert "signature" not in audit.rows[-1]
-
-
-@pytest.mark.parametrize(
-    "case",
-    [
-        "invalid_signature",
-        "expired",
-        "revoked",
-        "missing_key_evidence",
-        "missing_revocation_evidence",
-        "wrong_repository",
-        "wrong_branch",
-        "wrong_oid",
-        "wrong_operation",
-        "arbitrary_reference",
-    ],
-)
-def test_invalid_owner_approval_inputs_are_rejected_without_publication(tmp_path, case):
-    private, root, events, remote, _, publisher = service(
-        tmp_path, revocation="REVOKED" if case == "revoked" else "ACTIVE"
-    )
-    value = request()
-    approval = signed_approval(private, value)
-    if case == "invalid_signature":
-        approval["signature"] = "AAAA"
-    elif case == "expired":
-        approval = signed_approval(
-            private,
-            value,
-            issued_at="2000-01-01T00:00:00+00:00",
-            not_before="2000-01-01T00:00:01+00:00",
-            expires_at="2000-01-01T00:01:00+00:00",
-        )
-    elif case == "wrong_repository":
-        approval = signed_approval(private, value, repository="other/repository")
-    elif case == "wrong_branch":
-        approval = signed_approval(private, value, branch="agent/another-approved-branch")
-    elif case == "wrong_oid":
-        approval = signed_approval(private, value, expected_branch_oid="c" * 40)
-    elif case == "wrong_operation":
-        approval["operation"] = "differentOperation"
-    elif case == "arbitrary_reference":
-        approval = "owner-approval-reference-123"
-    elif case == "missing_key_evidence":
-        root.public_key = None
-    elif case == "missing_revocation_evidence":
-        root.revocation = "UNKNOWN"
-
-    with pytest.raises((OwnerApprovalRejected, ExistingBranchPublicationRejected)):
-        publisher.publish(value, approval)
-
-    assert remote.pushes == []
-    assert "credential-project" not in events
-    assert "workspace-open" not in events
-
-
-@pytest.mark.parametrize(
-    "changed",
-    [
-        lambda value: {**value, "branch": "agent/approval-replay-other-branch"},
-        lambda value: {**value, "expected_branch_oid": "c" * 40},
-        lambda value: {**value, "payload": {"files": ["other.py"]}},
-    ],
-)
-def test_signed_approval_cannot_be_replayed_for_another_branch_oid_or_payload(tmp_path, changed):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    original = request()
-    approval = signed_approval(private, original)
-
-    with pytest.raises(OwnerApprovalRejected, match="OWNER_APPROVAL_BINDING_REJECTED"):
-        publisher.publish(changed(original), approval)
-
-    assert remote.pushes == []
-    assert events == []
-
-
-def test_signed_approval_nonce_is_single_use(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    value = request()
-    approval = signed_approval(private, value)
-
-    assert publisher.publish(value, approval)["status"] == "PUBLISHED"
-    with pytest.raises(OwnerApprovalRejected, match="OWNER_APPROVAL_REPLAY_REJECTED"):
-        publisher.publish(value, approval)
-
-    assert len(remote.pushes) == 1
-    assert events.count("credential-project") == 1

tests/test_existing_branch_publication_security.py

@@ -1,72 +0,0 @@
-from __future__ import annotations
-
-import inspect
-
-from aios_existing_branch_publisher import (
-    ExistingBranchPublisher,
-    RemoteBranchObservation,
-)
-from test_existing_branch_owner_approval import (
-    BASE_OID,
-    COMMIT_OID,
-    request,
-    service,
-    signed_approval,
-)
-
-
-def test_remote_oid_conflict_requires_manual_owner_review_before_capability_projection(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    remote.observation = RemoteBranchObservation("PRESENT", "c" * 40, "c" * 40, True)
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-    assert events == ["authority-observe", "remote-observe"]
-    assert remote.pushes == []
-
-
-def test_ambiguous_remote_requires_reconciliation_after_one_observation(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    remote.observation = RemoteBranchObservation("AMBIGUOUS", None, None, None)
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {"status": "RECONCILIATION_REQUIRED"}
-    assert events == ["authority-observe", "remote-observe"]
-    assert remote.pushes == []
-
-
-def test_authority_oid_conflict_requires_manual_owner_review(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    remote.authority_oid = "c" * 40
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-    assert events == ["authority-observe"]
-    assert remote.pushes == []
-
-
-def test_success_reports_all_remote_agreement_fields(tmp_path):
-    private, _, _, _, _, publisher = service(tmp_path)
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {
-        "status": "PUBLISHED",
-        "commit_oid": COMMIT_OID,
-        "authority_oid": COMMIT_OID,
-        "gitea_git_oid": COMMIT_OID,
-        "gitea_api_oid": COMMIT_OID,
-        "commit_visible": True,
-    }
-
-
-def test_publisher_exposes_only_compare_and_swap_publication_path():
-    source = inspect.getsource(ExistingBranchPublisher).lower()
-    assert "force" not in source