fix(agent): start continuous lease heartbeat thread

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

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