fix(agent): start continuous lease heartbeat thread

active-runtime/app/aios_control_plane_host_seed_executor.py

@@ -1,748 +0,0 @@
-#!/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())

deploy/install/install_control_plane_host_seed_executor.sh

@@ -1,52 +0,0 @@
-#!/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

@@ -1,15 +0,0 @@
-#!/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

@@ -1,41 +0,0 @@
-[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

@@ -1,12 +0,0 @@
-[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

deploy/systemd/aios-online-agent-worker-20260815.service

@@ -0,0 +1,46 @@
+[Unit]
+Description=AIOS generic governed Online Agent Worker (2026-08-15)
+After=aios-agent-task-api.service aios-online-agent-worker-register-20260815.service
+Wants=aios-online-agent-worker-register-20260815.service
+
+[Service]
+Type=simple
+User=aios-engineering-worker
+Group=aios-engineering-worker
+SupplementaryGroups=gitops
+LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
+Environment=PYTHONUNBUFFERED=1
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-generic-20260815
+Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
+Environment=AIOS_AGENT_VERSION=2026-08-15.3-generic
+Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-worker-20260815
+Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-worker-20260815/workspaces
+Environment=AIOS_WORKER_POLL_SECONDS=5
+Environment=AIOS_WORKER_HEARTBEAT_SECONDS=10
+Environment=AIOS_WORKER_LEASE_SECONDS=300
+ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-online-agent-worker-20260815/online_agent_worker.py
+Restart=always
+RestartSec=5
+TimeoutStopSec=15
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadWritePaths=/var/lib/aios-online-agent-worker-20260815
+UMask=0027
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-online-agent-worker-register-20260815.service

@@ -0,0 +1,36 @@
+[Unit]
+Description=AIOS generic Online Agent register-or-recover (2026-08-15)
+After=network-online.target aios-gitops-model-api.service aios-agent-task-api.service
+Wants=network-online.target
+Before=aios-online-agent-worker-20260815.service
+
+[Service]
+Type=oneshot
+RemainAfterExit=yes
+User=root
+Group=root
+LoadCredential=bridge_control_token:/run/credentials/aios-gitops-model-api.service/agent_task_control_token
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-generic-20260815
+Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
+Environment=AIOS_AGENT_VERSION=2026-08-15.3-generic
+Environment=AIOS_WORKER_ENCRYPTED_CREDENTIAL=/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
+Environment=AIOS_WORKER_REGISTRATION_STATE=/var/lib/aios-online-agent-worker-20260815/registration
+ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-online-agent-worker-20260815/register.py
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadWritePaths=/etc/credstore.encrypted /var/lib/aios-online-agent-worker-20260815
+UMask=0077
+TimeoutStartSec=120
+
+[Install]
+WantedBy=multi-user.target

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

@@ -1,117 +0,0 @@
-# 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.

services/aios-online-agent-worker/README.md

@@ -0,0 +1,54 @@
+# Generic Governed Online Agent Worker
+
+This worker is a pull-based, restartable consumer for `aios.agent-task.v1`.
+The service configuration contains only stable runtime identity, capability
+names, API endpoints, heartbeat timing, and the credential projection. It does
+not contain a task id, repository base OID, target branch, or plan-file path.
+
+## Dynamic task contract
+
+Engineering tasks carry `metadata.execution` with schema
+`aios.online-agent.execution.v2`. The payload supplies the repository, exact
+base OID, target branch, detached-worktree policy, adapter name, explicit file
+change set, verification commands, and governed commit idempotency keys.
+Task prose is descriptive only and is never passed to a shell.
+
+The worker accepts only structured operations:
+
+- `deterministic`: allow-listed file changes and test commands;
+- `codex`: fail-closed until a task supplies an explicit Codex file/command
+  policy;
+- `hermes`: fail-closed because the current Hermes runtime is a user-chat
+  service, not an engineering executor.
+
+All command execution uses `shell=False`, fixed program allowlists, relative
+paths beneath an ephemeral detached worktree, and output redaction. Direct
+`git branch`, `update-ref`, `push`, `commit`, `reset`, `clean`, main,
+production, tag, and baseline mutations are rejected by the worker.
+
+## Governed write path
+
+The worker calls the Bridge in this order:
+
+1. `POST /v2/governed-task`;
+2. `POST /v2/writer-lease`;
+3. `POST /v2/commit-files`;
+4. read-only authority-ref visibility verification.
+
+CREATE_NEW is represented by the task payload and is normalized by the Bridge
+lease authority. The worker never creates an authoritative ref itself.
+
+## Lifecycle and recovery
+
+The runtime registers with `register_or_recover`, receives its agent token
+once, and stores only the encrypted systemd credential. The consumer loops over
+poll, atomically claims one task, sends the first heartbeat, emits
+`TASK_STARTED`, and continues heartbeats while executing.
+
+State is atomically stored per task without agent or lease token material. If
+the process stops after a governed commit but before result submission, the
+next lease attempt reuses the stable governed idempotency key and submits the
+stored commit evidence without issuing a second commit.
+
+The control plane must independently verify submitted results; worker-local
+success is not production or mainline authorization.

services/aios-online-agent-worker/aios_online_agent_worker/__init__.py

@@ -0,0 +1,3 @@
+"""Generic governed Online Agent Worker runtime."""
+
+__all__ = ["worker"]

services/aios-online-agent-worker/aios_online_agent_worker/worker.py

@@ -0,0 +1,937 @@
+#!/usr/bin/env python3
+"""Generic, restartable Online Agent Worker.
+
+The worker is deliberately task-agnostic.  A task's signed/durable payload is
+the only source of repository, branch, change, verification, and governance
+parameters.  There is no task id, base OID, branch name, or plan file in the
+runtime configuration.
+
+The worker performs local work only in an ephemeral detached worktree.  All
+authoritative ref updates go through the governed Bridge API; direct branch,
+update-ref, push, main, production, tag, and baseline mutations are rejected.
+"""
+
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from abc import ABC, abstractmethod
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import tempfile
+import threading
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Mapping
+from urllib.parse import unquote
+
+
+API_BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
+BRIDGE_BASE = os.environ.get("AIOS_GOVERNED_WRITE_API_URL", "http://127.0.0.1:18892").rstrip("/")
+AGENT_ID = os.environ.get(
+    "AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-generic-20260815"
+)
+AGENT_TYPE = os.environ.get("AIOS_AGENT_TYPE", "trusted-online-engineering-worker")
+AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.3-generic")
+AUTHORITY_GIT_DIR = Path(
+    os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git")
+)
+STATE_ROOT = Path(
+    os.environ.get("AIOS_WORKER_STATE_ROOT", "/var/lib/aios-online-agent-worker-20260815")
+)
+WORKSPACE_ROOT = Path(
+    os.environ.get(
+        "AIOS_WORKER_WORKSPACE_ROOT", "/var/lib/aios-online-agent-worker-20260815/workspaces"
+    )
+)
+GIT_CONFIG_GLOBAL = STATE_ROOT / "gitconfig"
+POLL_SECONDS = int(os.environ.get("AIOS_WORKER_POLL_SECONDS", "5"))
+HEARTBEAT_SECONDS = int(os.environ.get("AIOS_WORKER_HEARTBEAT_SECONDS", "20"))
+LEASE_SECONDS = int(os.environ.get("AIOS_WORKER_LEASE_SECONDS", "300"))
+MAX_TASK_BYTES = 512 * 1024
+MAX_FILE_BYTES = 2 * 1024 * 1024
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
+PATH_RE = re.compile(r"^[^\\\x00-\x1f\x7f]+$")
+SENSITIVE_RE = re.compile(
+    r"(?i)(authorization\s*:\s*bearer\s+\S+|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
+)
+
+
+class WorkerError(RuntimeError):
+    def __init__(self, code: str, *, phase: str, details: Mapping[str, Any] | None = None) -> None:
+        super().__init__(code)
+        self.code = code
+        self.phase = phase
+        self.details = dict(details or {})
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def digest_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def digest_json(value: Any) -> str:
+    return digest_bytes(
+        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    )
+
+
+def sanitize(value: Any) -> Any:
+    if isinstance(value, Mapping):
+        cleaned: dict[str, Any] = {}
+        for key, inner in value.items():
+            lowered = str(key).lower()
+            if any(marker in lowered for marker in ("token", "secret", "credential", "password", "authorization")):
+                cleaned[str(key)] = "[REDACTED]"
+            else:
+                cleaned[str(key)] = sanitize(inner)
+        return cleaned
+    if isinstance(value, list):
+        return [sanitize(item) for item in value]
+    if isinstance(value, tuple):
+        return [sanitize(item) for item in value]
+    if isinstance(value, str):
+        return SENSITIVE_RE.sub("[REDACTED]", value)
+    return value
+
+
+def excerpt(value: str, limit: int = 800) -> str:
+    value = str(sanitize(value))
+    return value if len(value) <= limit else value[: limit // 2] + "...[TRUNCATED]..." + value[-limit // 2 :]
+
+
+def audit(event_type: str, **fields: Any) -> None:
+    STATE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
+    path = STATE_ROOT / "events.jsonl"
+    record = {
+        "event_type": event_type,
+        "created_at": now(),
+        "agent_id": AGENT_ID,
+        "credential_material_exposed": False,
+        "lease_material_exposed": False,
+        **sanitize(fields),
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
+    os.chmod(path, 0o640)
+    print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
+
+
+def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o640) -> None:
+    path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise WorkerError("STATE_PATH_SYMLINK_REJECTED", phase="state")
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
+    try:
+        os.fchmod(fd, mode)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(sanitize(dict(value)), handle, ensure_ascii=False, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+        os.chmod(path, mode)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def safe_relative_path(value: Any, *, field: str = "path") -> str:
+    if not isinstance(value, str) or not value or len(value) > 240:
+        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
+    if unquote(value) != value or not PATH_RE.fullmatch(value):
+        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
+    if value.startswith(("/", "~")) or re.match(r"^[A-Za-z]:", value):
+        raise WorkerError(f"{field.upper()}_ABSOLUTE", phase="validation")
+    parts = value.replace("\\", "/").split("/")
+    if any(part in {"", ".", ".."} for part in parts):
+        raise WorkerError(f"{field.upper()}_TRAVERSAL", phase="validation")
+    return "/".join(parts)
+
+
+def resolve_under(root: Path, relative: str, *, write: bool = False) -> Path:
+    relative = safe_relative_path(relative)
+    root = root.resolve()
+    candidate = (root / relative).resolve(strict=False)
+    if candidate != root and root not in candidate.parents:
+        raise WorkerError("WORKSPACE_ESCAPE", phase="validation")
+    current = root
+    for part in Path(relative).parts:
+        current /= part
+        if current.exists() and current.is_symlink():
+            raise WorkerError("WORKSPACE_SYMLINK_REJECTED", phase="validation")
+    if write and candidate.exists() and candidate.is_symlink():
+        raise WorkerError("WRITE_SYMLINK_REJECTED", phase="validation")
+    return candidate
+
+
+def validate_oid(value: Any, field: str = "oid") -> str:
+    if not isinstance(value, str) or not OID_RE.fullmatch(value):
+        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
+    return value
+
+
+def validate_sha(value: Any, field: str = "sha256") -> str:
+    if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
+        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
+    return value
+
+
+def validate_branch(value: Any) -> str:
+    if not isinstance(value, str) or not 3 <= len(value) <= 180:
+        raise WorkerError("TARGET_BRANCH_INVALID", phase="validation")
+    if value in {"main", "master", "HEAD"} or value.startswith(
+        ("main/", "master/", "refs/", "tags/", "tag/", "release/", "releases/", "baseline/", "production/")
+    ):
+        raise WorkerError("PROTECTED_BRANCH_REJECTED", phase="validation")
+    if (
+        value.startswith("/")
+        or value.endswith(("/", ".", ".lock"))
+        or ".." in value
+        or "//" in value
+        or "@{" in value
+        or any(ch in value for ch in " ~^:?*[\\")
+        or any(part in {"", ".", ".."} for part in value.split("/"))
+        or not BRANCH_RE.fullmatch(value)
+    ):
+        raise WorkerError("TARGET_BRANCH_INVALID", phase="validation")
+    return value
+
+
+def validate_program(argv: Any, *, internal_git: bool = False) -> list[str]:
+    if not isinstance(argv, list) or not argv or not all(isinstance(item, str) and item for item in argv):
+        raise WorkerError("STRUCTURED_ARGV_REQUIRED", phase="validation")
+    command = argv[0]
+    if command not in {"/usr/bin/git", "/usr/bin/python3"}:
+        raise WorkerError("PROGRAM_NOT_ALLOWLISTED", phase="validation", details={"program": Path(command).name})
+    if command.endswith("python3"):
+        if any(item in {"-c", "-", "--command"} for item in argv):
+            raise WorkerError("PYTHON_INLINE_CODE_REJECTED", phase="validation")
+        if "-m" in argv:
+            index = argv.index("-m")
+            if index + 1 >= len(argv) or argv[index + 1] not in {"pytest", "py_compile"}:
+                raise WorkerError("PYTHON_MODULE_NOT_ALLOWLISTED", phase="validation")
+    if command.endswith("git") and not internal_git:
+        rejected = {"push", "update-ref", "receive-pack", "send-pack", "reset", "clean", "branch", "commit", "merge", "rebase"}
+        if any(item in rejected for item in argv[1:]):
+            raise WorkerError("GIT_MUTATION_NOT_ALLOWLISTED", phase="validation")
+    return list(argv)
+
+
+def run_command(
+    label: str,
+    argv: list[str],
+    *,
+    cwd: Path,
+    timeout: int,
+    internal_git: bool = False,
+    allowed_returncodes: tuple[int, ...] = (0,),
+) -> dict[str, Any]:
+    validate_program(argv, internal_git=internal_git)
+    if not 1 <= timeout <= 3600:
+        raise WorkerError("COMMAND_TIMEOUT_INVALID", phase=label)
+    started = time.monotonic()
+    try:
+        completed = subprocess.run(
+            argv,
+            cwd=str(cwd),
+            text=True,
+            capture_output=True,
+            check=False,
+            shell=False,
+            timeout=timeout,
+            env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)},
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise WorkerError("COMMAND_TIMEOUT", phase=label, details={"timeout_seconds": timeout}) from exc
+    result = {
+        "label": label,
+        "argv": [Path(item).name if index == 0 else item for index, item in enumerate(argv)],
+        "exit_code": completed.returncode,
+        "stdout_sha256": digest_bytes(completed.stdout.encode("utf-8", "replace")),
+        "stderr_sha256": digest_bytes(completed.stderr.encode("utf-8", "replace")),
+        "stdout_excerpt": excerpt(completed.stdout),
+        "stderr_excerpt": excerpt(completed.stderr),
+        "duration_ms": int((time.monotonic() - started) * 1000),
+    }
+    audit("COMMAND_RESULT", **result)
+    if completed.returncode not in allowed_returncodes:
+        raise WorkerError("COMMAND_FAILED", phase=label, details={"exit_code": completed.returncode, "stderr": excerpt(completed.stderr, 400)})
+    return result
+
+
+@dataclass(frozen=True)
+class ExecutionEnvelope:
+    task_id: str
+    repository: str
+    base_oid: str
+    target_branch: str
+    adapter: str
+    change_set: tuple[dict[str, Any], ...]
+    verification: tuple[dict[str, Any], ...]
+    governance: dict[str, Any]
+    workspace: dict[str, Any]
+    digest: str
+
+    @classmethod
+    def from_task(cls, task: Mapping[str, Any]) -> "ExecutionEnvelope":
+        task_id = str(task.get("task_id") or "")
+        if not ID_RE.fullmatch(task_id):
+            raise WorkerError("TASK_ID_INVALID", phase="validation")
+        metadata = task.get("metadata")
+        if not isinstance(metadata, Mapping):
+            raise WorkerError("TASK_METADATA_REQUIRED", phase="validation")
+        execution = metadata.get("execution")
+        if not isinstance(execution, Mapping):
+            raise WorkerError("DYNAMIC_EXECUTION_PAYLOAD_REQUIRED", phase="validation")
+        if execution.get("schema") != "aios.online-agent.execution.v2":
+            raise WorkerError("EXECUTION_SCHEMA_INVALID", phase="validation")
+        repository = execution.get("repository")
+        if not isinstance(repository, str) or not repository:
+            raise WorkerError("REPOSITORY_INVALID", phase="validation")
+        base_oid = validate_oid(execution.get("base_oid"), "base_oid")
+        target_branch = validate_branch(execution.get("target_branch"))
+        workspace = execution.get("workspace")
+        if not isinstance(workspace, Mapping):
+            raise WorkerError("WORKSPACE_POLICY_REQUIRED", phase="validation")
+        workspace_value = dict(workspace)
+        if workspace_value.get("authority_git_dir") != str(AUTHORITY_GIT_DIR):
+            raise WorkerError("AUTHORITY_ROOT_INVALID", phase="validation")
+        if workspace_value.get("read_only_repository") is not True:
+            raise WorkerError("REPOSITORY_WRITE_POLICY_INVALID", phase="validation")
+        if workspace_value.get("cleanup_policy") not in {"always", "cleanup_on_success"}:
+            raise WorkerError("WORKSPACE_CLEANUP_POLICY_INVALID", phase="validation")
+        adapter_value = execution.get("adapter")
+        if isinstance(adapter_value, Mapping):
+            adapter = str(adapter_value.get("name") or "")
+        else:
+            adapter = str(adapter_value or "")
+        if adapter not in {"deterministic", "codex", "hermes"}:
+            raise WorkerError("RUNTIME_ADAPTER_UNSUPPORTED", phase="validation")
+        change_set = execution.get("change_set")
+        if not isinstance(change_set, list) or not change_set or len(change_set) > 128:
+            raise WorkerError("CHANGE_SET_INVALID", phase="validation")
+        normalized_changes: list[dict[str, Any]] = []
+        for item in change_set:
+            if not isinstance(item, Mapping):
+                raise WorkerError("CHANGE_OPERATION_INVALID", phase="validation")
+            operation = str(item.get("operation") or "")
+            relative = safe_relative_path(item.get("path"))
+            normalized = dict(item)
+            normalized["operation"] = operation
+            normalized["path"] = relative
+            if operation == "write_file":
+                content = item.get("content")
+                if not isinstance(content, str) or len(content.encode("utf-8")) > MAX_FILE_BYTES:
+                    raise WorkerError("WRITE_CONTENT_INVALID", phase="validation")
+            elif operation == "replace_text":
+                if not isinstance(item.get("old"), str) or not item.get("old") or not isinstance(item.get("new"), str):
+                    raise WorkerError("REPLACE_TEXT_INVALID", phase="validation")
+                if len(str(item["old"]).encode("utf-8")) > MAX_FILE_BYTES:
+                    raise WorkerError("REPLACE_TEXT_INVALID", phase="validation")
+            elif operation == "delete_file":
+                pass
+            else:
+                raise WorkerError("CHANGE_OPERATION_UNSUPPORTED", phase="validation", details={"operation": operation})
+            expected = item.get("expected_sha256")
+            if expected is not None:
+                validate_sha(expected, "expected_sha256")
+            normalized_changes.append(normalized)
+        verification = execution.get("verification") or []
+        if not isinstance(verification, list) or len(verification) > 128:
+            raise WorkerError("VERIFICATION_INVALID", phase="validation")
+        normalized_verification: list[dict[str, Any]] = []
+        for item in verification:
+            if not isinstance(item, Mapping):
+                raise WorkerError("VERIFICATION_COMMAND_INVALID", phase="validation")
+            label = str(item.get("label") or "verification")
+            argv = validate_program(item.get("argv"))
+            cwd_value = item.get("cwd")
+            cwd = "." if cwd_value in {None, "", "."} else safe_relative_path(cwd_value, field="cwd")
+            timeout = item.get("timeout_seconds", 120)
+            if not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 <= timeout <= 3600:
+                raise WorkerError("VERIFICATION_TIMEOUT_INVALID", phase="validation")
+            normalized_verification.append({"label": label, "argv": argv, "cwd": cwd, "timeout_seconds": timeout})
+        governance = execution.get("governance")
+        if not isinstance(governance, Mapping):
+            raise WorkerError("GOVERNANCE_POLICY_REQUIRED", phase="validation")
+        governance_value = dict(governance)
+        governed_task_id = str(governance_value.get("task_id") or f"{task_id}-writer")
+        if not ID_RE.fullmatch(governed_task_id):
+            raise WorkerError("GOVERNED_TASK_ID_INVALID", phase="validation")
+        idempotency_key = str(governance_value.get("idempotency_key") or f"{task_id}-governed-v1")
+        commit_key = str(governance_value.get("commit_idempotency_key") or f"{task_id}-commit-v1")
+        for field, value in (("idempotency_key", idempotency_key), ("commit_idempotency_key", commit_key)):
+            if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,159}", value):
+                raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
+        governance_value.update(
+            {
+                "task_id": governed_task_id,
+                "idempotency_key": idempotency_key,
+                "commit_idempotency_key": commit_key,
+                "message": str(governance_value.get("message") or f"chore(gitops): execute task {task_id}"),
+            }
+        )
+        for field in ("main_write", "force_push", "production_write", "tag_write", "baseline_write", "protected_ref_bypass"):
+            if governance_value.get(field, False) is not False:
+                raise WorkerError("GOVERNANCE_POLICY_REJECTED", phase="validation", details={"field": field})
+        allowed_files = governance_value.get("allowed_files")
+        if allowed_files is not None:
+            if not isinstance(allowed_files, list):
+                raise WorkerError("ALLOWED_FILES_INVALID", phase="validation")
+            allowed = {safe_relative_path(item, field="allowed_file") for item in allowed_files}
+            if not {item["path"] for item in normalized_changes}.issubset(allowed):
+                raise WorkerError("CHANGE_OUTSIDE_ALLOWED_FILES", phase="validation")
+            governance_value["allowed_files"] = sorted(allowed)
+        return cls(
+            task_id=task_id,
+            repository=repository,
+            base_oid=base_oid,
+            target_branch=target_branch,
+            adapter=adapter,
+            change_set=tuple(normalized_changes),
+            verification=tuple(normalized_verification),
+            governance=governance_value,
+            workspace=workspace_value,
+            digest=digest_json(execution),
+        )
+
+
+class RuntimeAdapter(ABC):
+    name: str
+
+    @abstractmethod
+    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
+        raise NotImplementedError
+
+
+class DeterministicExecutor(RuntimeAdapter):
+    name = "deterministic"
+
+    @staticmethod
+    def _write_atomic(path: Path, content: str) -> None:
+        path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
+        fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                handle.write(content)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temporary, path)
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
+
+    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
+        changed: set[str] = set()
+        for change in envelope.change_set:
+            relative = str(change["path"])
+            path = resolve_under(worktree, relative, write=True)
+            expected = change.get("expected_sha256")
+            current = path.read_bytes() if path.is_file() else None
+            if expected is not None and (current is None or digest_bytes(current) != expected):
+                raise WorkerError("WORKTREE_SOURCE_DRIFT", phase="implementing", details={"path": relative})
+            operation = change["operation"]
+            if operation == "write_file":
+                self._write_atomic(path, str(change["content"]))
+            elif operation == "replace_text":
+                if current is None:
+                    raise WorkerError("REPLACE_SOURCE_MISSING", phase="implementing", details={"path": relative})
+                original = current.decode("utf-8")
+                old = str(change["old"])
+                count = original.count(old)
+                expected_count = int(change.get("count", 1))
+                if count != expected_count:
+                    raise WorkerError("REPLACE_TEXT_MATCH_COUNT_INVALID", phase="implementing", details={"path": relative, "count": count})
+                self._write_atomic(path, original.replace(old, str(change["new"])))
+            elif operation == "delete_file":
+                if path.exists():
+                    path.unlink()
+            changed.add(relative)
+            audit("REPOSITORY_CHANGE_APPLIED", operation=operation, path=relative, sha256=digest_bytes(path.read_bytes()) if path.exists() else None)
+        commands: list[dict[str, Any]] = []
+        if changed:
+            commands.append(run_command("git_intent_to_add", ["/usr/bin/git", "-C", str(worktree), "add", "-N", "--", *sorted(changed)], cwd=worktree, timeout=120, internal_git=True))
+        commands.append(run_command("git_diff_check", ["/usr/bin/git", "-C", str(worktree), "diff", "--check"], cwd=worktree, timeout=120))
+        commands.append(run_command("git_diff_stat", ["/usr/bin/git", "-C", str(worktree), "diff", "--no-ext-diff", "--stat"], cwd=worktree, timeout=120))
+        tests: list[dict[str, Any]] = []
+        for item in envelope.verification:
+            cwd = worktree if item["cwd"] == "." else resolve_under(worktree, item["cwd"])
+            tests.append(run_command(item["label"], item["argv"], cwd=cwd, timeout=item["timeout_seconds"]))
+        return {
+            "changed_files": sorted(changed),
+            "commands": commands,
+            "tests": tests,
+            "adapter": self.name,
+        }
+
+
+class CodexExecutor(RuntimeAdapter):
+    name = "codex"
+
+    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
+        raise WorkerError(
+            "CODEX_ADAPTER_REQUIRES_EXPLICIT_POLICY",
+            phase="adapter",
+            details={"reason": "Codex is not allowed to mutate a governed worktree without an explicit file/command policy"},
+        )
+
+
+class HermesExecutor(RuntimeAdapter):
+    name = "hermes"
+
+    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
+        raise WorkerError(
+            "HERMES_ENGINEERING_ADAPTER_UNAVAILABLE",
+            phase="adapter",
+            details={"reason": "The live Hermes service is a user-chat runtime, not an engineering executor"},
+        )
+
+
+def adapter_for(name: str) -> RuntimeAdapter:
+    return {"deterministic": DeterministicExecutor, "codex": CodexExecutor, "hermes": HermesExecutor}[name]()
+
+
+class HttpClient:
+    @staticmethod
+    def call(base: str, method: str, route: str, token: str | None, body: Mapping[str, Any] | None = None, timeout: int = 30) -> tuple[int, dict[str, Any]]:
+        payload = None if body is None else json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        headers = {"Accept": "application/json"}
+        if token:
+            headers["Authorization"] = "Bearer " + token
+        if payload is not None:
+            headers["Content-Type"] = "application/json"
+        request = urllib.request.Request(base + route, data=payload, headers=headers, method=method)
+        try:
+            with urllib.request.urlopen(request, timeout=timeout) as response:
+                value = json.loads(response.read().decode("utf-8"))
+                return response.status, value if isinstance(value, dict) else {}
+        except urllib.error.HTTPError as exc:
+            raw = exc.read(8192).decode("utf-8", "replace")
+            try:
+                value = json.loads(raw)
+            except json.JSONDecodeError:
+                value = {"error": "HTTP_ERROR"}
+            return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
+        except (urllib.error.URLError, TimeoutError):
+            return 599, {"error": "SERVICE_UNAVAILABLE"}
+
+
+class AgentTaskClient:
+    def __init__(self, token: str) -> None:
+        self.token = token
+
+    def poll(self) -> list[dict[str, Any]]:
+        status, response = HttpClient.call(API_BASE, "POST", "/v1/tasks/poll", self.token, {"limit": 100})
+        if status != 200:
+            raise WorkerError("TASK_POLL_FAILED", phase="poll", details={"api_status": status, "error": response.get("error")})
+        tasks = response.get("tasks")
+        return [dict(item) for item in tasks if isinstance(item, Mapping)] if isinstance(tasks, list) else []
+
+    def claim(self, task_id: str) -> tuple[int, dict[str, Any]]:
+        return HttpClient.call(API_BASE, "POST", f"/v1/tasks/{task_id}/claim", self.token, {"lease_seconds": LEASE_SECONDS})
+
+    def heartbeat(self, task_id: str, lease_token: str, stage: str) -> tuple[int, dict[str, Any]]:
+        return HttpClient.call(
+            API_BASE,
+            "POST",
+            f"/v1/tasks/{task_id}/heartbeat",
+            self.token,
+            {"lease_token": lease_token, "lease_seconds": LEASE_SECONDS, "progress": {"stage": stage, "worker_version": AGENT_VERSION}},
+        )
+
+    def result(self, task_id: str, lease_token: str, idempotency_key: str, result: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
+        return HttpClient.call(
+            API_BASE,
+            "POST",
+            f"/v1/tasks/{task_id}/result",
+            self.token,
+            {"lease_token": lease_token, "idempotency_key": idempotency_key, "result": dict(result)},
+        )
+
+
+class BridgeClient:
+    def call(self, route: str, body: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
+        return HttpClient.call(BRIDGE_BASE, "POST", route, None, body, timeout=60)
+
+
+class LeaseHeartbeat:
+    def __init__(self, client: AgentTaskClient, task_id: str, lease_token: str) -> None:
+        self.client = client
+        self.task_id = task_id
+        self.lease_token = lease_token
+        self.stage = "claimed"
+        self.failed: str | None = None
+        self.stop_event = threading.Event()
+        self.lock = threading.Lock()
+        self.thread = threading.Thread(target=self._run, name="online-agent-heartbeat", daemon=True)
+
+    def first(self) -> None:
+        status, response = self.client.heartbeat(self.task_id, self.lease_token, self.stage)
+        if not 200 <= status < 300:
+            self.failed = str(response.get("error") or status)
+            raise WorkerError("TASK_HEARTBEAT_FAILED", phase="heartbeat", details={"api_status": status, "error": self.failed})
+        audit("TASK_HEARTBEAT", task_id=self.task_id, stage=self.stage, api_status=status)
+
+    def start(self) -> None:
+        self.thread.start()
+
+    def set_stage(self, stage: str) -> None:
+        with self.lock:
+            self.stage = stage
+
+    def _run(self) -> None:
+        while not self.stop_event.wait(HEARTBEAT_SECONDS):
+            with self.lock:
+                stage = self.stage
+            try:
+                status, response = self.client.heartbeat(self.task_id, self.lease_token, stage)
+                if not 200 <= status < 300:
+                    self.failed = str(response.get("error") or status)
+                    audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, api_status=status, error=self.failed)
+                else:
+                    audit("TASK_HEARTBEAT", task_id=self.task_id, stage=stage, api_status=status)
+            except Exception as exc:  # pragma: no cover - defensive runtime guard
+                self.failed = type(exc).__name__
+                audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, error=self.failed)
+
+    def ensure(self) -> None:
+        if self.failed:
+            raise WorkerError("TASK_LEASE_LOST", phase="heartbeat", details={"error": self.failed})
+
+    def close(self) -> None:
+        self.stop_event.set()
+        if self.thread.ident is not None:
+            self.thread.join(timeout=2)
+
+
+def load_agent_token() -> str:
+    directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
+    path = directory / "agent_token"
+    if not path.is_file() or path.is_symlink():
+        raise WorkerError("AGENT_RUNTIME_CREDENTIAL_UNAVAILABLE", phase="startup")
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 24:
+        raise WorkerError("AGENT_RUNTIME_CREDENTIAL_INVALID", phase="startup")
+    return value
+
+
+def ensure_roots() -> None:
+    if AUTHORITY_GIT_DIR.is_symlink() or STATE_ROOT.is_symlink() or WORKSPACE_ROOT.is_symlink():
+        raise WorkerError("ROOT_SYMLINK_REJECTED", phase="startup")
+    if not AUTHORITY_GIT_DIR.is_dir():
+        raise WorkerError("AUTHORITY_GIT_DIR_UNAVAILABLE", phase="startup")
+    STATE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
+    WORKSPACE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
+    if GIT_CONFIG_GLOBAL.is_symlink():
+        raise WorkerError("GIT_CONFIG_SYMLINK_REJECTED", phase="startup")
+    expected = f"[safe]\n\tdirectory = {AUTHORITY_GIT_DIR}\n"
+    if GIT_CONFIG_GLOBAL.exists() and GIT_CONFIG_GLOBAL.read_text(encoding="utf-8") != expected:
+        raise WorkerError("GIT_CONFIG_BINDING_CONFLICT", phase="startup")
+    if not GIT_CONFIG_GLOBAL.exists():
+        GIT_CONFIG_GLOBAL.write_text(expected, encoding="utf-8")
+        os.chmod(GIT_CONFIG_GLOBAL, 0o640)
+
+
+def task_state_path(task_id: str) -> Path:
+    if not ID_RE.fullmatch(task_id):
+        raise WorkerError("TASK_ID_INVALID", phase="state")
+    return STATE_ROOT / "tasks" / f"{task_id}.json"
+
+
+def load_task_state(task_id: str, execution_digest: str) -> dict[str, Any] | None:
+    path = task_state_path(task_id)
+    if not path.is_file() or path.is_symlink():
+        return None
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        raise WorkerError("TASK_STATE_INVALID", phase="state")
+    if not isinstance(value, dict):
+        raise WorkerError("TASK_STATE_INVALID", phase="state")
+    if value.get("execution_digest") not in {None, execution_digest}:
+        raise WorkerError("TASK_PAYLOAD_DRIFT", phase="recovery")
+    return value
+
+
+def save_task_state(task_id: str, value: Mapping[str, Any]) -> None:
+    atomic_json(task_state_path(task_id), value)
+
+
+def new_run_root() -> Path:
+    ensure_roots()
+    root = Path(tempfile.mkdtemp(prefix="task-", dir=str(WORKSPACE_ROOT)))
+    os.chmod(root, 0o750)
+    return root
+
+
+def cleanup_run_root(path: Path) -> None:
+    root = WORKSPACE_ROOT.resolve()
+    candidate = path.resolve(strict=False)
+    if candidate == root or root not in candidate.parents:
+        raise WorkerError("CLEANUP_SCOPE_INVALID", phase="cleanup")
+    if candidate.exists() or candidate.is_symlink():
+        shutil.rmtree(candidate)
+
+
+def prepare_worktree(run_root: Path, base_oid: str) -> Path:
+    clone = run_root / "repository"
+    worktree = run_root / "worktree"
+    run_command("workspace_clone", ["/usr/bin/git", "clone", "--no-local", "--no-hardlinks", str(AUTHORITY_GIT_DIR), str(clone)], cwd=run_root, timeout=300, internal_git=True)
+    run_command("workspace_base_probe", ["/usr/bin/git", "-C", str(clone), "rev-parse", "--verify", base_oid + "^{commit}"], cwd=run_root, timeout=60)
+    run_command("workspace_worktree_create", ["/usr/bin/git", "-C", str(clone), "worktree", "add", "--detach", str(worktree), base_oid], cwd=run_root, timeout=300, internal_git=True)
+    return worktree
+
+
+def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeClient, heartbeat: LeaseHeartbeat) -> dict[str, Any]:
+    governance = envelope.governance
+    create_body = {
+        "task_id": governance["task_id"],
+        "repository": envelope.repository,
+        "base_oid": envelope.base_oid,
+        "target_branch": envelope.target_branch,
+        "risk": str(governance.get("risk") or "engineering"),
+        "description": str(governance.get("description") or f"Governed execution for {envelope.task_id}"),
+        "idempotency_key": governance["idempotency_key"],
+    }
+    status, response = bridge.call("/v2/governed-task", create_body)
+    if not 200 <= status < 300:
+        raise WorkerError("GOVERNED_TASK_CREATE_FAILED", phase="committing", details={"api_status": status, "error": response.get("error")})
+    generation = int(response.get("generation") or 0)
+    expected_branch_oid = str(response.get("expected_branch_oid") or envelope.base_oid)
+    branch_mode = str(response.get("branch_mode") or "CREATE_NEW")
+    lease_body = {
+        "schema": "aios.task-writer-lease.v2",
+        "task_id": governance["task_id"],
+        "repository": envelope.repository,
+        "branch_mode": branch_mode,
+        "base_oid": envelope.base_oid,
+        "expected_branch_oid": expected_branch_oid,
+        "target_branch": envelope.target_branch,
+        "generation": generation,
+        "lease_seconds": int(governance.get("lease_seconds") or 900),
+    }
+    status, lease_response = bridge.call("/v2/writer-lease", lease_body)
+    if not 200 <= status < 300:
+        raise WorkerError("WRITER_LEASE_ACQUIRE_FAILED", phase="committing", details={"api_status": status, "error": lease_response.get("error")})
+    heartbeat.ensure()
+    files: dict[str, str] = {}
+    paths = governance.get("allowed_files") or [item["path"] for item in envelope.change_set]
+    for relative in sorted({safe_relative_path(item, field="allowed_file") for item in paths}):
+        path = resolve_under(worktree, relative)
+        if not path.is_file():
+            raise WorkerError("COMMIT_FILE_MISSING", phase="committing", details={"path": relative})
+        files[relative] = base64.b64encode(path.read_bytes()).decode("ascii")
+    commit_body = {
+        "task_id": governance["task_id"],
+        "repository": envelope.repository,
+        "expected_branch_oid": expected_branch_oid,
+        "target_branch": envelope.target_branch,
+        "generation": generation,
+        "idempotency_key": governance["commit_idempotency_key"],
+        "files": files,
+        "message": governance["message"],
+    }
+    status, commit_response = bridge.call("/v2/commit-files", commit_body)
+    if not 200 <= status < 300:
+        raise WorkerError("GOVERNED_COMMIT_FAILED", phase="committing", details={"api_status": status, "error": commit_response.get("error")})
+    commit_oid = commit_response.get("commit_oid")
+    if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
+        raise WorkerError("GOVERNED_COMMIT_OID_INVALID", phase="committing")
+    visible_oid: str | None = None
+    for _ in range(5):
+        probe = subprocess.run(
+            ["/usr/bin/git", "--git-dir", str(AUTHORITY_GIT_DIR), "rev-parse", "--verify", f"refs/heads/{envelope.target_branch}"],
+            text=True,
+            capture_output=True,
+            check=False,
+            env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)},
+            timeout=30,
+        )
+        if probe.returncode == 0:
+            visible_oid = probe.stdout.strip()
+            if visible_oid == commit_oid:
+                break
+        time.sleep(1)
+    if visible_oid != commit_oid:
+        raise WorkerError("GIT_VISIBILITY_MISMATCH", phase="verification", details={"commit_oid": commit_oid, "visible_oid": visible_oid})
+    return {
+        "governed_task_id": governance["task_id"],
+        "branch_mode": branch_mode,
+        "generation": generation,
+        "target_branch": envelope.target_branch,
+        "commit_oid": commit_oid,
+        "visible_oid": visible_oid,
+        "commit_idempotency_key": governance["commit_idempotency_key"],
+        "api_commit_visible": commit_response.get("commit_visible"),
+        "direct_main_write": False,
+        "force_push": False,
+        "production_write": False,
+        "protected_ref_bypass": False,
+    }
+
+
+def build_result(task_id: str, envelope: ExecutionEnvelope, execution: Mapping[str, Any], governed: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "status": "SUCCEEDED",
+        "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
+        "task_id": task_id,
+        "execution_backend": "generic-online-agent-worker-v2",
+        "adapter": envelope.adapter,
+        "execution_digest": envelope.digest,
+        "repository": envelope.repository,
+        "base_oid": envelope.base_oid,
+        "target_branch": envelope.target_branch,
+        "changed_files": execution.get("changed_files", []),
+        "commands": [{k: item.get(k) for k in ("label", "exit_code", "stdout_sha256", "stderr_sha256", "duration_ms")} for item in execution.get("commands", [])],
+        "tests": [{k: item.get(k) for k in ("label", "exit_code", "stdout_sha256", "stderr_sha256", "duration_ms")} for item in execution.get("tests", [])],
+        "governed": dict(governed),
+        "workspace_cleanup": "PASS",
+        "secret_material_exposed": False,
+        "direct_main_write": False,
+        "force_push": False,
+        "production_write": False,
+        "protected_ref_bypass": False,
+    }
+
+
+class WorkerSupervisor:
+    def __init__(self, token: str) -> None:
+        self.agent = AgentTaskClient(token)
+        self.bridge = BridgeClient()
+
+    def process_claim(self, task: Mapping[str, Any], claim: Mapping[str, Any]) -> None:
+        task_id = str(claim.get("task_id") or task.get("task_id") or "")
+        lease_token = claim.get("lease_token")
+        if not isinstance(lease_token, str) or len(lease_token) < 24:
+            raise WorkerError("LEASE_TOKEN_NOT_RETURNED", phase="claim")
+        audit("TASK_CLAIMED", task_id=task_id, attempt=claim.get("attempt"), revision=claim.get("revision"))
+        heartbeat = LeaseHeartbeat(self.agent, task_id, lease_token)
+        try:
+            heartbeat.first()
+            heartbeat.set_stage("started")
+            heartbeat.start()
+            audit("TASK_STARTED", task_id=task_id, attempt=claim.get("attempt"))
+            envelope = ExecutionEnvelope.from_task(task)
+            state = load_task_state(task_id, envelope.digest) or {"task_id": task_id, "execution_digest": envelope.digest, "phase": "claimed"}
+            governed = state.get("governed") if isinstance(state.get("governed"), Mapping) else None
+            if isinstance(governed, Mapping) and isinstance(governed.get("commit_oid"), str) and OID_RE.fullmatch(governed["commit_oid"]):
+                result = state.get("result")
+                if not isinstance(result, Mapping):
+                    result = {"status": "SUCCEEDED", "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW", "task_id": task_id, "execution_digest": envelope.digest, "governed": dict(governed), "recovered_without_duplicate_commit": True, "direct_main_write": False, "production_write": False, "force_push": False, "protected_ref_bypass": False}
+                audit("TASK_RECOVERED_COMMIT", task_id=task_id, commit_oid=governed["commit_oid"], duplicate_commit=False)
+            else:
+                run_root = new_run_root()
+                try:
+                    worktree = prepare_worktree(run_root, envelope.base_oid)
+                    state.update({"phase": "workspace_prepared", "execution_digest": envelope.digest, "attempt": claim.get("attempt")})
+                    save_task_state(task_id, state)
+                    adapter = adapter_for(envelope.adapter)
+                    execution = adapter.execute(envelope, worktree)
+                    state.update({"phase": "verification_passed", "execution": execution})
+                    save_task_state(task_id, state)
+                    heartbeat.ensure()
+                    heartbeat.set_stage("committing")
+                    governed = governed_commit(envelope, worktree, self.bridge, heartbeat)
+                    result = build_result(task_id, envelope, execution, governed)
+                    state.update({"phase": "commit_done", "governed": governed, "result": result})
+                    save_task_state(task_id, state)
+                finally:
+                    cleanup_run_root(run_root)
+                    audit("WORKSPACE_CLEANED", task_id=task_id, cleanup="PASS")
+            heartbeat.ensure()
+            heartbeat.set_stage("result")
+            result_key = str(result.get("governed", {}).get("commit_idempotency_key") or f"{task_id}-result-{envelope.digest[:16]}")
+            status, response = self.agent.result(task_id, lease_token, result_key, result)
+            if not 200 <= status < 300:
+                raise WorkerError("TASK_RESULT_SUBMIT_FAILED", phase="result", details={"api_status": status, "error": response.get("error")})
+            audit("TASK_RESULT_SUBMITTED", task_id=task_id, api_status=status, result_idempotency_key=result_key, commit_oid=result.get("governed", {}).get("commit_oid"))
+        except WorkerError as exc:
+            audit("TASK_FAILED", task_id=task_id, error=exc.code, phase=exc.phase, details=exc.details)
+            failure_key = f"{task_id}-attempt-{claim.get('attempt', 0)}-failure"
+            failure = {
+                "status": "FAILED",
+                "verification_state": "UNVERIFIED",
+                "task_id": task_id,
+                "execution_backend": "generic-online-agent-worker-v2",
+                "error": exc.code,
+                "phase": exc.phase,
+                "details": exc.details,
+                "direct_main_write": False,
+                "force_push": False,
+                "production_write": False,
+                "protected_ref_bypass": False,
+                "secret_material_exposed": False,
+            }
+            try:
+                status, response = self.agent.result(task_id, lease_token, failure_key, failure)
+                audit("TASK_FAILURE_RESULT_SUBMITTED", task_id=task_id, api_status=status, error=response.get("error"))
+            except Exception as submit_exc:  # pragma: no cover - defensive runtime guard
+                audit("TASK_FAILURE_RESULT_SUBMIT_FAILED", task_id=task_id, error=type(submit_exc).__name__)
+        finally:
+            heartbeat.close()
+
+    def once(self) -> int:
+        try:
+            tasks = self.agent.poll()
+            audit("TASK_POLL", visible_count=len(tasks))
+        except WorkerError as exc:
+            audit("TASK_POLL_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
+            return 10
+        dynamic_tasks: list[dict[str, Any]] = []
+        for task in tasks:
+            metadata = task.get("metadata")
+            execution = metadata.get("execution") if isinstance(metadata, Mapping) else None
+            if not isinstance(execution, Mapping) or execution.get("schema") != "aios.online-agent.execution.v2":
+                audit("TASK_SKIPPED_NON_DYNAMIC", task_id=task.get("task_id"), reason="DYNAMIC_EXECUTION_PAYLOAD_REQUIRED")
+                continue
+            dynamic_tasks.append(task)
+        for task in dynamic_tasks:
+            task_id = str(task.get("task_id") or "")
+            status, claim = self.agent.claim(task_id)
+            if status == 409:
+                audit("TASK_CLAIM_SKIPPED", task_id=task_id, api_status=status, error=claim.get("error"))
+                continue
+            if status != 200:
+                audit("TASK_CLAIM_FAILED", task_id=task_id, api_status=status, error=claim.get("error"))
+                continue
+            self.process_claim(task, claim)
+            return 0
+        return 0
+
+    def run_forever(self) -> int:
+        ensure_roots()
+        audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["deterministic", "codex", "hermes"], runtime_only_credentials=True)
+        while True:
+            self.once()
+            time.sleep(POLL_SECONDS)
+
+
+def main() -> int:
+    try:
+        token = load_agent_token()
+        supervisor = WorkerSupervisor(token)
+        if os.environ.get("AIOS_WORKER_ONCE") == "1":
+            ensure_roots()
+            return supervisor.once()
+        return supervisor.run_forever()
+    except WorkerError as exc:
+        audit("WORKER_START_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
+        return 20
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-online-agent-worker/register.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Register or recover the generic Online Agent using runtime-only secrets."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import urllib.error
+import urllib.request
+from datetime import datetime, timezone
+from typing import Any
+
+
+BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
+AGENT_ID = os.environ.get("AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-generic-20260815")
+AGENT_TYPE = os.environ.get("AIOS_AGENT_TYPE", "trusted-online-engineering-worker")
+AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.3-generic")
+PROJECTION_PROVIDER = "BridgeV2RuntimeCredentialProjection"
+STATE_DIR = Path(os.environ.get("AIOS_WORKER_REGISTRATION_STATE", "/var/lib/aios-online-agent-worker-20260815/registration"))
+CREDENTIAL_PATH = Path(os.environ.get("AIOS_WORKER_ENCRYPTED_CREDENTIAL", "/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token"))
+IDENTITY_PATH = STATE_DIR / "identity.json"
+AUDIT_PATH = STATE_DIR / "events.jsonl"
+CAPABILITIES = [
+    "online_agent_task_api", "task.poll", "task.claim", "task.heartbeat", "task.result",
+    "engineering.consumer", "engineering.execute", "workspace.worktree", "git.read", "git.diff",
+    "git.test", "governed.write.branch", "governed.write.file",
+]
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def audit(event_type: str, **fields: Any) -> None:
+    STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
+    record = {"event_type": event_type, "created_at": now(), "agent_id": AGENT_ID, "credential_material_exposed": False, **fields}
+    with AUDIT_PATH.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
+    os.chmod(AUDIT_PATH, 0o640)
+    print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
+
+
+def api_call(method: str, route: str, token: str, body: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
+    payload = None if body is None else json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
+    if payload is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(BASE + route, data=payload, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(request, timeout=15) as response:
+            value = json.loads(response.read().decode("utf-8"))
+            return response.status, value if isinstance(value, dict) else {}
+    except urllib.error.HTTPError as exc:
+        try:
+            value = json.loads(exc.read(2048).decode("utf-8"))
+        except Exception:
+            value = {"error": "HTTP_ERROR"}
+        return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
+    except (urllib.error.URLError, TimeoutError):
+        return 599, {"error": "AGENT_TASK_API_UNAVAILABLE"}
+
+
+def read_identity() -> dict[str, Any] | None:
+    if not IDENTITY_PATH.is_file() or IDENTITY_PATH.is_symlink():
+        return None
+    try:
+        value = json.loads(IDENTITY_PATH.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+    return value if isinstance(value, dict) else None
+
+
+def identity_matches(value: dict[str, Any] | None) -> bool:
+    return bool(value and value.get("agent_id") == AGENT_ID and value.get("agent_type") == AGENT_TYPE and value.get("agent_version") == AGENT_VERSION and value.get("credential_projection_provider") == PROJECTION_PROVIDER and value.get("task_scope") == "dynamic")
+
+
+def encrypted_credential_is_valid() -> bool:
+    if not CREDENTIAL_PATH.is_file() or CREDENTIAL_PATH.is_symlink():
+        return False
+    result = subprocess.run(["systemd-creds", "decrypt", "--with-key=host", str(CREDENTIAL_PATH), "/dev/null"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=15)
+    return result.returncode == 0
+
+
+def write_identity() -> None:
+    value = {"agent_id": AGENT_ID, "agent_type": AGENT_TYPE, "agent_version": AGENT_VERSION, "task_scope": "dynamic", "credential_projection_provider": PROJECTION_PROVIDER, "credential_delivery": "systemd-LoadCredentialEncrypted", "task_api_url": BASE, "task_api_schema": "aios.agent-task.v1", "execution_backend": "generic-online-agent-worker-v2", "runtime_only_secret_delivery": True, "production_write": False, "main_write": False, "force_push": False, "tag_write": False, "baseline_write": False, "writer_lease": True, "created_at": now()}
+    STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
+    fd, temporary = tempfile.mkstemp(prefix="identity.", dir=str(STATE_DIR))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(value, handle, ensure_ascii=False, sort_keys=True)
+            handle.write("\n")
+        os.chmod(temporary, 0o640)
+        os.replace(temporary, IDENTITY_PATH)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def encrypt_agent_token(agent_token: str) -> None:
+    CREDENTIAL_PATH.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    fd, temporary = tempfile.mkstemp(prefix=".online-agent-token.", dir=str(CREDENTIAL_PATH.parent))
+    os.close(fd)
+    os.unlink(temporary)
+    try:
+        result = subprocess.run(["systemd-creds", "encrypt", "--with-key=host", "--name=agent_token", "/dev/stdin", temporary], input=(agent_token + "\n").encode("utf-8"), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=20)
+        if result.returncode != 0:
+            raise RuntimeError("AGENT_CREDENTIAL_ENCRYPT_FAILED")
+        os.chmod(temporary, 0o600)
+        os.replace(temporary, CREDENTIAL_PATH)
+        os.chmod(CREDENTIAL_PATH, 0o600)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def register_or_recover() -> str:
+    STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
+    existing = read_identity()
+    if CREDENTIAL_PATH.exists():
+        if not identity_matches(existing):
+            raise RuntimeError("AGENT_IDENTITY_BINDING_CONFLICT")
+        if not encrypted_credential_is_valid():
+            raise RuntimeError("AGENT_ENCRYPTED_CREDENTIAL_INVALID")
+        audit("AGENT_RECOVERED", credential_delivery="systemd-LoadCredentialEncrypted", credential_projection_provider=PROJECTION_PROVIDER, register_or_recover=True)
+        return "RECOVERED"
+    if existing is not None:
+        raise RuntimeError("AGENT_CREDENTIAL_MISSING_AFTER_PRIOR_REGISTRATION")
+    credential_directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
+    control_path = credential_directory / "bridge_control_token"
+    if not control_path.is_file():
+        raise RuntimeError("BRIDGE_RUNTIME_CONTROL_CREDENTIAL_UNAVAILABLE")
+    control_token = control_path.read_text(encoding="utf-8").strip()
+    if len(control_token) < 32:
+        raise RuntimeError("BRIDGE_RUNTIME_CONTROL_CREDENTIAL_INVALID")
+    status, response = api_call("POST", "/v1/agents/register", control_token, {"agent_id": AGENT_ID, "agent_type": AGENT_TYPE, "agent_version": AGENT_VERSION, "capabilities": CAPABILITIES, "metadata": {"task_scope": "dynamic", "credential_projection_provider": PROJECTION_PROVIDER, "runtime_only_secret_delivery": True, "task_api_schema": "aios.agent-task.v1", "execution_backend": "generic-online-agent-worker-v2", "production_write": False, "main_write": False, "force_push": False, "tag_write": False, "baseline_write": False, "writer_lease": True}})
+    if status == 409 and response.get("error") == "AGENT_ALREADY_EXISTS":
+        raise RuntimeError("AGENT_ALREADY_EXISTS_RECOVERY_CREDENTIAL_UNAVAILABLE")
+    if status != 201:
+        raise RuntimeError("AGENT_REGISTER_FAILED_" + str(response.get("error") or status))
+    agent_token = response.get("agent_token")
+    if not isinstance(agent_token, str) or len(agent_token) < 24:
+        raise RuntimeError("AGENT_TOKEN_NOT_RETURNED")
+    encrypt_agent_token(agent_token)
+    write_identity()
+    audit("AGENT_REGISTERED", credential_delivery="systemd-LoadCredentialEncrypted", credential_projection_provider=PROJECTION_PROVIDER, register_or_recover=True)
+    return "REGISTERED"
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(0 if register_or_recover() in {"REGISTERED", "RECOVERED"} else 20)
+    except Exception as exc:
+        audit("AGENT_REGISTRATION_FAILED", error=type(exc).__name__)
+        raise SystemExit(20)

services/aios-online-agent-worker/test_worker_contract.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from aios_online_agent_worker.worker import ExecutionEnvelope, WorkerError, sanitize  # noqa: E402
+
+
+BASE = "a" * 40
+
+
+def payload(task_id: str, branch: str, *, base_oid: str = BASE) -> dict[str, object]:
+    return {
+        "task_id": task_id,
+        "metadata": {
+            "execution": {
+                "schema": "aios.online-agent.execution.v2",
+                "repository": "aios/aios",
+                "base_oid": base_oid,
+                "target_branch": branch,
+                "adapter": "deterministic",
+                "workspace": {
+                    "authority_git_dir": str(Path(os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git"))),
+                    "read_only_repository": True,
+                    "cleanup_policy": "always",
+                },
+                "change_set": [
+                    {
+                        "operation": "write_file",
+                        "path": "tests/generated_marker.py",
+                        "content": "def test_marker():\n    assert True\n",
+                        "expected_sha256": None,
+                    }
+                ],
+                "verification": [
+                    {"label": "compile", "argv": ["/usr/bin/python3", "-m", "py_compile", "tests/generated_marker.py"], "cwd": ".", "timeout_seconds": 60}
+                ],
+                "governance": {
+                    "task_id": f"{task_id}-writer",
+                    "idempotency_key": f"{task_id}-governed-v1",
+                    "commit_idempotency_key": f"{task_id}-commit-v1",
+                    "allowed_files": ["tests/generated_marker.py"],
+                    "main_write": False,
+                    "force_push": False,
+                    "production_write": False,
+                    "tag_write": False,
+                    "baseline_write": False,
+                    "protected_ref_bypass": False,
+                },
+            }
+        },
+    }
+
+
+def test_dynamic_payload_is_not_bound_to_task_or_base() -> None:
+    first = ExecutionEnvelope.from_task(payload("task-alpha-20260815", "agent/alpha-20260815"))
+    second = ExecutionEnvelope.from_task(payload("task-beta-20260815", "agent/beta-20260815", base_oid="b" * 40))
+    assert first.task_id != second.task_id
+    assert first.base_oid != second.base_oid
+    assert first.target_branch != second.target_branch
+
+
+def test_protected_branch_is_rejected() -> None:
+    try:
+        ExecutionEnvelope.from_task(payload("task-main-20260815", "main"))
+    except WorkerError as exc:
+        assert exc.code == "PROTECTED_BRANCH_REJECTED"
+    else:
+        raise AssertionError("protected branch accepted")
+
+
+def test_task_payload_does_not_persist_secret_values() -> None:
+    value = sanitize({"agent_token": "agt-secret", "nested": {"authorization": "Bearer secret"}})
+    assert value == {"agent_token": "[REDACTED]", "nested": {"authorization": "[REDACTED]"}}
+
+
+def test_execution_digest_is_stable() -> None:
+    value = payload("task-digest-20260815", "agent/digest-20260815")
+    assert ExecutionEnvelope.from_task(value).digest == ExecutionEnvelope.from_task(value).digest

tests/test_control_plane_host_seed_executor.py

@@ -1,328 +0,0 @@
-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,
-        )
-