fix(online-agent): normalize external runtime capabilities

active-runtime/app/aios_capability_contract.py

@@ -0,0 +1,163 @@
+"""Canonical fail-closed capability semantics for task eligibility."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+
+_TASK_ALIASES: dict[str, frozenset[str]] = {
+    "repository_read": frozenset({"repo.read"}),
+    "git.read": frozenset({"repo.read"}),
+    "repo.read": frozenset({"repo.read"}),
+    "test_execution": frozenset({"test.execute"}),
+    "git.test": frozenset({"test.execute"}),
+    "test.execute": frozenset({"test.execute"}),
+    "governed_git_write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
+    "governed.git.write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
+    "governed.write.branch": frozenset({"governed.git.write.branch"}),
+    "governed.write.file": frozenset({"governed.git.write.file"}),
+    "shell": frozenset({"runtime.shell"}),
+    "git": frozenset({"runtime.git"}),
+    "python": frozenset({"runtime.python"}),
+    "code.read": frozenset({"legacy.code.read"}),
+    "code.review": frozenset({"legacy.code.review"}),
+    "code.write": frozenset({"legacy.code.write"}),
+    "code.write.patch": frozenset({"legacy.code.write.patch"}),
+    "test.python": frozenset({"legacy.test.python"}),
+    "test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
+    "git.commit": frozenset({"legacy.git.commit"}),
+    "git.diff.read": frozenset({"legacy.git.diff.read"}),
+    "git.force_push": frozenset({"legacy.git.force_push"}),
+    "deploy.production": frozenset({"legacy.deploy.production"}),
+    "secret.transform": frozenset({"legacy.secret.transform"}),
+    "secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
+    "runtime.shadow": frozenset({"legacy.runtime.shadow"}),
+    "review.security": frozenset({"legacy.review.security"}),
+    "review.api-contract": frozenset({"legacy.review.api-contract"}),
+    "gpu.train": frozenset({"legacy.gpu.train"}),
+    "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
+    "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+}
+
+_WORKER_ALIASES: dict[str, frozenset[str]] = {
+    "git.read": frozenset({"repo.read"}),
+    "repo.read": frozenset({"repo.read"}),
+    "repository_read": frozenset({"repo.read"}),
+    "git.test": frozenset({"test.execute"}),
+    "test.execute": frozenset({"test.execute"}),
+    "test_execution": frozenset({"test.execute"}),
+    "governed.write.branch": frozenset({"governed.git.write.branch"}),
+    "governed.git.write.branch": frozenset({"governed.git.write.branch"}),
+    "governed.write.file": frozenset({"governed.git.write.file"}),
+    "governed.git.write.file": frozenset({"governed.git.write.file"}),
+    "shell": frozenset({"runtime.shell"}),
+    "git": frozenset({"runtime.git"}),
+    "python": frozenset({"runtime.python"}),
+    "code.read": frozenset({"legacy.code.read"}),
+    "code.review": frozenset({"legacy.code.review"}),
+    "code.write": frozenset({"legacy.code.write"}),
+    "code.write.patch": frozenset({"legacy.code.write.patch"}),
+    "test.python": frozenset({"legacy.test.python"}),
+    "test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
+    "git.commit": frozenset({"legacy.git.commit"}),
+    "git.diff.read": frozenset({"legacy.git.diff.read"}),
+    "git.force_push": frozenset({"legacy.git.force_push"}),
+    "deploy.production": frozenset({"legacy.deploy.production"}),
+    "secret.transform": frozenset({"legacy.secret.transform"}),
+    "secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
+    "runtime.shadow": frozenset({"legacy.runtime.shadow"}),
+    "review.security": frozenset({"legacy.review.security"}),
+    "review.api-contract": frozenset({"legacy.review.api-contract"}),
+    "gpu.train": frozenset({"legacy.gpu.train"}),
+    "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
+    "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+}
+
+_UNKNOWN_PREFIX = "exact:"
+
+
+def _clean(value: object) -> str:
+    if not isinstance(value, str):
+        raise ValueError("capability must be a string")
+    text = value.strip().lower()
+    if not text:
+        raise ValueError("capability must not be empty")
+    return text
+
+
+def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
+    """Map task-facing names to one semantic vocabulary.
+
+    Unknown names remain explicit exact requirements. They are never treated
+    as wildcards or granted by a failed normalization.
+    """
+
+    result: set[str] = set()
+    for value in values:
+        text = _clean(value)
+        if "*" in text:
+            result.add(_UNKNOWN_PREFIX + text)
+        else:
+            result.update(_TASK_ALIASES.get(text, frozenset({_UNKNOWN_PREFIX + text})))
+    return frozenset(result)
+
+
+def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
+    """Map registered worker names to effective semantic capabilities."""
+
+    result: set[str] = set()
+    for value in values:
+        text = _clean(value)
+        if "*" in text:
+            continue
+        mapped = _WORKER_ALIASES.get(text)
+        if mapped is not None:
+            result.update(mapped)
+    if {
+        "governed.git.write.branch",
+        "governed.git.write.file",
+    }.issubset(result):
+        result.add("governed.git.write")
+    return frozenset(result)
+
+
+def has_capabilities(
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+) -> bool:
+    """Return the single eligibility decision used by poll and claim."""
+
+    required = normalize_required_capabilities(required_capabilities)
+    effective = normalize_worker_capabilities(worker_capabilities)
+    if any(value.startswith(_UNKNOWN_PREFIX) for value in required):
+        return False
+    return required.issubset(effective)
+
+
+def required_capabilities_subset(
+    granted_requirements: Iterable[object],
+    required_requirements: Iterable[object],
+) -> bool:
+    """Check semantic capability containment for delegation constraints."""
+    granted = normalize_required_capabilities(granted_requirements)
+    required = normalize_required_capabilities(required_requirements)
+    if any(value.startswith(_UNKNOWN_PREFIX) for value in granted | required):
+        return False
+    return required.issubset(granted)
+
+
+def normalized_capability_matrix(
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+) -> dict[str, object]:
+    """Expose non-secret decision evidence for tests and diagnostics."""
+
+    worker_values = tuple(worker_capabilities)
+    required_values = tuple(required_capabilities)
+    worker = normalize_worker_capabilities(worker_values)
+    required = normalize_required_capabilities(required_values)
+    return {
+        "worker_effective": sorted(worker),
+        "task_required": sorted(required),
+        "eligible": has_capabilities(worker_values, required_values),
+    }

active-runtime/app/aios_external_agent_runtime.py

@@ -0,0 +1,710 @@
+"""Credential-isolated external Agent runtime executor for Online Agent Tasks.
+
+Task payloads never choose commands. A root-owned runtime profile chooses the executable and
+I/O mode. The Online Agent worker writes a hash-bound request into an exchange directory; this
+uncredentialed process executes Hermes/OpenClaw and returns a bounded, sanitized result.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import time
+from typing import Any, Mapping, Sequence
+
+REQUEST_SCHEMA = "aios.external-agent-runtime.request.v1"
+RESULT_SCHEMA = "aios.external-agent-runtime.result.v1"
+PROFILE_SCHEMA = "aios.external-agent-runtime.profile.v1"
+SUPPORTED_RUNTIMES = frozenset({"hermes", "openclaw"})
+REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,254}$")
+RUNTIME_RE = re.compile(r"^[a-z][a-z0-9_-]{2,31}$")
+SENSITIVE_RE = re.compile(
+    r"(?i)(authorization\s*[:=]\s*bearer\s+\S+|"
+    r"(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
+)
+SECRET_ENV_MARKERS = (
+    "TOKEN",
+    "SECRET",
+    "PASSWORD",
+    "CREDENTIAL",
+    "AUTHORIZATION",
+    "PRIVATE_KEY",
+    "API_KEY",
+)
+PROFILE_ENV_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$")
+PROFILE_CREDENTIAL_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{2,127}$")
+PROFILE_CREDENTIAL_KEY_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$")
+PROVIDER_CREDENTIAL_ENV_NAMES = frozenset({"OPENAI_API_KEY"})
+PROFILE_FIXED_ENV_DENY = frozenset(
+    {
+        "PATH",
+        "HOME",
+        "NO_COLOR",
+        "LC_ALL",
+        "LANG",
+        "LD_PRELOAD",
+        "LD_LIBRARY_PATH",
+        "PYTHONPATH",
+        "NODE_OPTIONS",
+        "BASH_ENV",
+        "SHELL",
+        "IFS",
+    }
+)
+MAX_REQUEST_BYTES = 256 * 1024
+MAX_OUTPUT_BYTES = 128 * 1024
+MAX_PROFILE_BYTES = 64 * 1024
+MAX_CREDENTIAL_BYTES = 64 * 1024
+DEFAULT_PROFILE_ROOT = Path("/etc/aios/external-agent-runtime.d")
+DEFAULT_EXCHANGE_ROOT = Path("/var/lib/aios-external-agent-executor")
+ALLOWED_EXECUTABLE_ROOTS = (
+    Path("/usr/bin"),
+    Path("/usr/local/bin"),
+    Path("/opt/aios/bin"),
+    Path("/opt/hermes"),
+    Path("/opt/openclaw"),
+    Path("/opt/aios-hermes-agent"),
+    Path("/opt/aios/openclaw-engineering"),
+)
+
+
+class ExternalAgentRuntimeError(RuntimeError):
+    pass
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+
+
+def sha256(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def sanitize_text(value: str) -> str:
+    return SENSITIVE_RE.sub("[REDACTED]", str(value))
+
+
+def _bounded_text(value: str, limit: int = MAX_OUTPUT_BYTES) -> str:
+    data = sanitize_text(value).encode("utf-8", "replace")
+    if len(data) <= limit:
+        return data.decode("utf-8", "replace")
+    marker = b"\n...[TRUNCATED]...\n"
+    half = max(1, (limit - len(marker)) // 2)
+    return (data[:half] + marker + data[-half:]).decode("utf-8", "replace")
+
+
+def _safe_runtime(value: object) -> str:
+    runtime = str(value or "").strip().lower()
+    if RUNTIME_RE.fullmatch(runtime) is None or runtime not in SUPPORTED_RUNTIMES:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_RUNTIME_UNSUPPORTED")
+    return runtime
+
+
+def _assert_no_secret_env() -> None:
+    leaked = sorted(
+        key
+        for key, value in os.environ.items()
+        if value and any(marker in key.upper() for marker in SECRET_ENV_MARKERS)
+    )
+    # CREDENTIALS_DIRECTORY itself is systemd metadata; the directory is expected to be hidden.
+    leaked = [key for key in leaked if key != "CREDENTIALS_DIRECTORY"]
+    if leaked:
+        raise ExternalAgentRuntimeError(
+            "EXTERNAL_AGENT_CREDENTIAL_ENV_EXPOSED:" + ",".join(leaked)
+        )
+
+
+def _safe_executable(value: object) -> str:
+    if not isinstance(value, str) or not value.startswith("/"):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXECUTABLE_ABSOLUTE_REQUIRED")
+    path = Path(value)
+    if path.is_symlink():
+        resolved = path.resolve(strict=True)
+    else:
+        resolved = path.resolve(strict=True)
+    if not resolved.is_file() or not os.access(resolved, os.X_OK):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXECUTABLE_INVALID")
+    if not any(resolved == root or root in resolved.parents for root in ALLOWED_EXECUTABLE_ROOTS):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXECUTABLE_ROOT_DENY")
+    return str(resolved)
+
+
+def _safe_profile_script(value: object) -> str:
+    if not isinstance(value, str) or not value.startswith("/"):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_SCRIPT_ABSOLUTE_REQUIRED")
+    path = Path(value)
+    if path.is_symlink():
+        resolved = path.resolve(strict=True)
+    else:
+        resolved = path.resolve(strict=True)
+    if not resolved.is_file():
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_SCRIPT_INVALID")
+    if not any(resolved == root or root in resolved.parents for root in ALLOWED_EXECUTABLE_ROOTS):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_SCRIPT_ROOT_DENY")
+    return str(resolved)
+
+
+def _profile_environment(value: object) -> tuple[tuple[str, str], ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, Mapping):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ENVIRONMENT_INVALID")
+    result: list[tuple[str, str]] = []
+    for raw_name, raw_value in value.items():
+        if not isinstance(raw_name, str) or PROFILE_ENV_NAME_RE.fullmatch(raw_name) is None:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ENVIRONMENT_NAME_INVALID")
+        if raw_name in PROFILE_FIXED_ENV_DENY:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ENVIRONMENT_NAME_DENY")
+        if any(marker in raw_name.upper() for marker in SECRET_ENV_MARKERS):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_SECRET_ENV_DENY")
+        if not isinstance(raw_value, str) or "\x00" in raw_value or "\n" in raw_value or "\r" in raw_value:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ENVIRONMENT_VALUE_INVALID")
+        if len(raw_value) > 4096:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ENVIRONMENT_VALUE_TOO_LARGE")
+        result.append((raw_name, raw_value))
+    return tuple(sorted(result))
+
+
+def _profile_credential_environment(
+    value: object,
+) -> tuple[tuple[str, str, str, str], ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, Mapping):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_CREDENTIAL_ENV_INVALID")
+    result: list[tuple[str, str, str, str]] = []
+    for raw_env_name, raw_spec in value.items():
+        if (
+            not isinstance(raw_env_name, str)
+            or raw_env_name not in PROVIDER_CREDENTIAL_ENV_NAMES
+            or PROFILE_ENV_NAME_RE.fullmatch(raw_env_name) is None
+        ):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_ENV_NAME_DENY")
+        if not isinstance(raw_spec, Mapping):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_SPEC_INVALID")
+        credential = raw_spec.get("credential")
+        fmt = str(raw_spec.get("format") or "dotenv")
+        key = raw_spec.get("key")
+        if (
+            not isinstance(credential, str)
+            or PROFILE_CREDENTIAL_NAME_RE.fullmatch(credential) is None
+        ):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_NAME_INVALID")
+        if fmt not in {"dotenv", "raw"}:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_FORMAT_INVALID")
+        if fmt == "dotenv":
+            if not isinstance(key, str) or PROFILE_CREDENTIAL_KEY_RE.fullmatch(key) is None:
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_KEY_INVALID")
+        elif key is not None:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_KEY_UNEXPECTED")
+        result.append((raw_env_name, credential, fmt, str(key or "")))
+    return tuple(sorted(result))
+
+
+@dataclass(frozen=True)
+class RuntimeProfile:
+    runtime: str
+    argv: tuple[str, ...]
+    input_mode: str
+    output_mode: str
+    timeout_seconds: int
+    max_output_bytes: int
+    environment: tuple[tuple[str, str], ...]
+    credential_env: tuple[tuple[str, str, str, str], ...]
+
+    @classmethod
+    def load(
+        cls,
+        runtime: str,
+        profile_root: Path = DEFAULT_PROFILE_ROOT,
+    ) -> "RuntimeProfile":
+        runtime = _safe_runtime(runtime)
+        root = Path(profile_root)
+        if root.is_symlink():
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ROOT_SYMLINK_DENY")
+        path = root / f"{runtime}.json"
+        if path.is_symlink() or not path.is_file() or path.stat().st_size > MAX_PROFILE_BYTES:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_UNAVAILABLE")
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_INVALID") from exc
+        if not isinstance(value, Mapping) or value.get("schema") != PROFILE_SCHEMA:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_INVALID")
+        if _safe_runtime(value.get("runtime")) != runtime:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_RUNTIME_MISMATCH")
+        argv = value.get("argv")
+        if (
+            not isinstance(argv, Sequence)
+            or isinstance(argv, (str, bytes))
+            or not argv
+            or not all(isinstance(item, str) and item and "\x00" not in item for item in argv)
+        ):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_ARGV_INVALID")
+        executable = _safe_executable(argv[0])
+        normalized = [executable, *list(argv[1:])]
+        if any(item in {"sh", "bash", "-c", "--command"} for item in normalized[1:]):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_SHELL_MODE_DENY")
+        joined_argv = " ".join(normalized)
+        if re.search(r"(?i)(token|secret|password|api[_-]?key|authorization)\s*[:=]", joined_argv):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PROFILE_SECRET_ARG_DENY")
+        if Path(executable).name in {"node", "nodejs"}:
+            if len(normalized) < 2 or normalized[1].startswith("-"):
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_NODE_SCRIPT_REQUIRED")
+            normalized[1] = _safe_profile_script(normalized[1])
+            if any(item in {"-e", "--eval", "-p", "--print", "-r", "--require"} for item in normalized[1:]):
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_NODE_CODE_MODE_DENY")
+        input_mode = str(value.get("input_mode") or "stdin_text")
+        output_mode = str(value.get("output_mode") or "stdout_text")
+        if input_mode not in {"stdin_text", "stdin_json", "file_json"}:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_INPUT_MODE_INVALID")
+        if output_mode not in {"stdout_text", "stdout_json", "file_json"}:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_MODE_INVALID")
+        timeout = int(value.get("timeout_seconds") or 900)
+        max_output = int(value.get("max_output_bytes") or MAX_OUTPUT_BYTES)
+        if not 1 <= timeout <= 3600:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_TIMEOUT_INVALID")
+        if not 1024 <= max_output <= MAX_OUTPUT_BYTES:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_LIMIT_INVALID")
+        environment = _profile_environment(value.get("environment"))
+        credential_env = _profile_credential_environment(value.get("credential_env"))
+        return cls(
+            runtime,
+            tuple(normalized),
+            input_mode,
+            output_mode,
+            timeout,
+            max_output,
+            environment,
+            credential_env,
+        )
+
+
+def validate_request(value: Mapping[str, Any]) -> dict[str, Any]:
+    if value.get("schema") != REQUEST_SCHEMA:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_SCHEMA_INVALID")
+    request_id = str(value.get("request_id") or "")
+    if REQUEST_ID_RE.fullmatch(request_id) is None:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_ID_INVALID")
+    runtime = _safe_runtime(value.get("runtime"))
+    task_id = str(value.get("task_id") or "")
+    execution_digest = str(value.get("execution_digest") or "")
+    if not task_id or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{2,179}", task_id):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_TASK_ID_INVALID")
+    if re.fullmatch(r"[0-9a-f]{64}", execution_digest) is None:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXECUTION_DIGEST_INVALID")
+    payload = value.get("payload")
+    if not isinstance(payload, Mapping):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PAYLOAD_INVALID")
+    encoded = canonical(payload)
+    if len(encoded) > MAX_REQUEST_BYTES:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_PAYLOAD_TOO_LARGE")
+    goal = payload.get("goal")
+    instructions = payload.get("instructions")
+    if not isinstance(goal, str) or not goal.strip():
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_GOAL_REQUIRED")
+    if instructions is not None and not isinstance(instructions, str):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_INSTRUCTIONS_INVALID")
+    return {
+        "schema": REQUEST_SCHEMA,
+        "request_id": request_id,
+        "runtime": runtime,
+        "task_id": task_id,
+        "execution_digest": execution_digest,
+        "payload": dict(payload),
+    }
+
+
+def render_prompt(payload: Mapping[str, Any]) -> str:
+    goal = str(payload.get("goal") or "").strip()
+    instructions = str(payload.get("instructions") or "").strip()
+    criteria = payload.get("success_criteria") or []
+    constraints = payload.get("constraints") or {}
+    lines = [
+        "AIOS Online Agent Task",
+        "",
+        "Goal:",
+        goal,
+    ]
+    if instructions:
+        lines.extend(["", "Instructions:", instructions])
+    if criteria:
+        lines.extend(["", "Success criteria:", json.dumps(criteria, ensure_ascii=False, sort_keys=True)])
+    if constraints:
+        lines.extend(["", "Constraints:", json.dumps(constraints, ensure_ascii=False, sort_keys=True)])
+    lines.extend([
+        "",
+        "Return the final task result only. Do not claim repository, production, credential, or control-plane side effects unless they are explicitly represented in the task payload.",
+    ])
+    return "\n".join(lines)
+
+
+def _argv_with_files(
+    profile: RuntimeProfile,
+    input_path: Path,
+    output_path: Path,
+    prompt: str,
+) -> list[str]:
+    replacements = {
+        "{input_json}": str(input_path),
+        "{output_json}": str(output_path),
+        "{prompt}": prompt,
+    }
+    result = []
+    for item in profile.argv:
+        for marker, replacement in replacements.items():
+            item = item.replace(marker, replacement)
+        result.append(item)
+    if profile.input_mode == "file_json" and not any("{input_json}" in item for item in profile.argv):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_INPUT_PLACEHOLDER_REQUIRED")
+    if profile.output_mode == "file_json" and not any("{output_json}" in item for item in profile.argv):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_PLACEHOLDER_REQUIRED")
+    return result
+
+
+def _credential_environment(profile: RuntimeProfile) -> dict[str, str]:
+    if not profile.credential_env:
+        return {}
+    raw_root = os.environ.get("CREDENTIALS_DIRECTORY")
+    if not raw_root or not raw_root.startswith("/"):
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_DIRECTORY_MISSING")
+    root = Path(raw_root)
+    if root.is_symlink() or not root.is_dir():
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_DIRECTORY_INVALID")
+    result: dict[str, str] = {}
+    for env_name, credential, fmt, key in profile.credential_env:
+        path = root / credential
+        if path.is_symlink() or not path.is_file():
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_FILE_INVALID")
+        try:
+            if path.stat().st_size > MAX_CREDENTIAL_BYTES:
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_TOO_LARGE")
+            raw = path.read_bytes()
+        except OSError as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_READ_FAILED") from exc
+        if b"\x00" in raw:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_CONTENT_INVALID")
+        try:
+            text = raw.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_CONTENT_INVALID") from exc
+        selected = ""
+        if fmt == "raw":
+            selected = text.strip()
+        else:
+            found = False
+            for line in text.splitlines():
+                candidate = line.strip()
+                if not candidate or candidate.startswith("#"):
+                    continue
+                if candidate.startswith("export "):
+                    candidate = candidate[7:].lstrip()
+                if "=" not in candidate:
+                    continue
+                candidate_key, candidate_value = candidate.split("=", 1)
+                if candidate_key.strip() != key:
+                    continue
+                if found:
+                    raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_KEY_DUPLICATE")
+                selected = candidate_value.strip()
+                if len(selected) >= 2 and selected[0] == selected[-1] and selected[0] in {"'", '"'}:
+                    selected = selected[1:-1]
+                found = True
+            if not found:
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_KEY_MISSING")
+        if not selected or "\x00" in selected or "\n" in selected or "\r" in selected:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_VALUE_INVALID")
+        if len(selected) > MAX_CREDENTIAL_BYTES:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_CREDENTIAL_VALUE_TOO_LARGE")
+        result[env_name] = selected
+    return result
+
+
+def _parse_output(
+    profile: RuntimeProfile,
+    stdout: str,
+    output_path: Path,
+) -> tuple[str, object]:
+    if profile.output_mode == "file_json":
+        if output_path.is_symlink() or not output_path.is_file():
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_FILE_MISSING")
+        raw = output_path.read_bytes()
+        if len(raw) > profile.max_output_bytes:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_TOO_LARGE")
+        text = raw.decode("utf-8", "replace")
+        try:
+            parsed = json.loads(text)
+        except json.JSONDecodeError as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_JSON_INVALID") from exc
+        return _bounded_text(text, profile.max_output_bytes), parsed
+    if len(stdout.encode("utf-8", "replace")) > profile.max_output_bytes:
+        stdout = _bounded_text(stdout, profile.max_output_bytes)
+    if profile.output_mode == "stdout_json":
+        try:
+            parsed = json.loads(stdout)
+        except json.JSONDecodeError as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_OUTPUT_JSON_INVALID") from exc
+        return _bounded_text(stdout, profile.max_output_bytes), parsed
+    return _bounded_text(stdout, profile.max_output_bytes), None
+
+
+def execute_request(
+    request_path: Path,
+    result_path: Path,
+    *,
+    profile_root: Path = DEFAULT_PROFILE_ROOT,
+) -> None:
+    _assert_no_secret_env()
+    if request_path.is_symlink() or not request_path.is_file():
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_FILE_INVALID")
+    raw_request = request_path.read_bytes()
+    if len(raw_request) > MAX_REQUEST_BYTES:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_TOO_LARGE")
+    try:
+        request = validate_request(json.loads(raw_request.decode("utf-8")))
+    except json.JSONDecodeError as exc:
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_JSON_INVALID") from exc
+    profile = RuntimeProfile.load(request["runtime"], profile_root)
+
+    with tempfile.TemporaryDirectory(prefix=f"aios-{request['runtime']}-") as temporary:
+        temp_root = Path(temporary)
+        (temp_root / "home").mkdir()
+        input_path = temp_root / "input.json"
+        output_path = temp_root / "output.json"
+        input_path.write_bytes(canonical(request["payload"]))
+        prompt = render_prompt(request["payload"])
+        argv = _argv_with_files(profile, input_path, output_path, prompt)
+
+        stdin_data = None
+        if profile.input_mode == "stdin_json":
+            stdin_data = canonical(request["payload"]).decode("utf-8")
+        elif profile.input_mode == "stdin_text":
+            stdin_data = prompt
+
+        child_env = {
+            "PATH": "/usr/bin:/usr/local/bin:/bin",
+            "LC_ALL": "C.UTF-8",
+            "LANG": "C.UTF-8",
+            "HOME": str(temp_root / "home"),
+            "NO_COLOR": "1",
+        }
+        child_env.update(dict(profile.environment))
+        child_env.update(_credential_environment(profile))
+
+        started = time.monotonic()
+        try:
+            completed = subprocess.run(
+                argv,
+                input=stdin_data,
+                capture_output=True,
+                text=True,
+                check=False,
+                shell=False,
+                timeout=profile.timeout_seconds,
+                cwd=str(temp_root),
+                env=child_env,
+            )
+        except subprocess.TimeoutExpired as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_TIMEOUT") from exc
+
+        if completed.returncode != 0:
+            raise ExternalAgentRuntimeError(
+                f"EXTERNAL_AGENT_PROCESS_FAILED:{completed.returncode}:"
+                + _bounded_text(completed.stderr, 2048)
+            )
+
+        output_text, output_json = _parse_output(profile, completed.stdout, output_path)
+        result: dict[str, Any] = {
+            "schema": RESULT_SCHEMA,
+            "request_id": request["request_id"],
+            "request_sha256": sha256(canonical(request)),
+            "runtime": request["runtime"],
+            "task_id": request["task_id"],
+            "execution_digest": request["execution_digest"],
+            "exit_code": completed.returncode,
+            "duration_ms": int((time.monotonic() - started) * 1000),
+            "output_text": output_text,
+            "output_sha256": sha256(output_text.encode("utf-8", "replace")),
+            "stderr_sha256": sha256(completed.stderr.encode("utf-8", "replace")),
+            "credential_material_exposed": False,
+            "control_plane_access_granted": False,
+        }
+        if output_json is not None:
+            result["output_json"] = output_json
+        result["result_sha256"] = sha256(canonical(result))
+        result_path.parent.mkdir(parents=True, exist_ok=True)
+        fd, name = tempfile.mkstemp(prefix=f".{result_path.name}.", dir=result_path.parent)
+        try:
+            os.fchmod(fd, 0o640)
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                json.dump(result, handle, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+                handle.write("\n")
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(name, result_path)
+        finally:
+            if os.path.exists(name):
+                os.unlink(name)
+
+
+class ExternalAgentExchange:
+    """Worker-side idempotent request/result exchange."""
+
+    def __init__(self, root: Path = DEFAULT_EXCHANGE_ROOT) -> None:
+        self.root = Path(root)
+
+    def execute(
+        self,
+        *,
+        request_id: str,
+        runtime: str,
+        task_id: str,
+        execution_digest: str,
+        payload: Mapping[str, Any],
+        timeout_seconds: int = 1800,
+    ) -> dict[str, Any]:
+        request = validate_request(
+            {
+                "schema": REQUEST_SCHEMA,
+                "request_id": request_id,
+                "runtime": runtime,
+                "task_id": task_id,
+                "execution_digest": execution_digest,
+                "payload": dict(payload),
+            }
+        )
+        requests = self.root / "requests"
+        results = self.root / "results"
+        failed = self.root / "failed"
+        for directory in (requests, results, failed):
+            directory.mkdir(parents=True, exist_ok=True)
+        request_path = requests / f"{request_id}.json"
+        result_path = results / f"{request_id}.json"
+        failed_path = failed / f"{request_id}.json"
+        request_sha = sha256(canonical(request))
+
+        if result_path.is_file():
+            return self._validated_result(result_path, request, request_sha)
+        if failed_path.is_file():
+            value = json.loads(failed_path.read_text(encoding="utf-8"))
+            raise ExternalAgentRuntimeError(
+                str(value.get("error") or "EXTERNAL_AGENT_EXECUTOR_FAILED")
+            )
+
+        if request_path.exists():
+            try:
+                existing = json.loads(request_path.read_text(encoding="utf-8"))
+            except json.JSONDecodeError as exc:
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXISTING_REQUEST_INVALID") from exc
+            if sha256(canonical(existing)) != request_sha:
+                raise ExternalAgentRuntimeError("EXTERNAL_AGENT_REQUEST_ID_CONFLICT")
+        else:
+            fd, name = tempfile.mkstemp(prefix=f".{request_path.name}.", dir=requests)
+            try:
+                os.fchmod(fd, 0o640)
+                with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                    json.dump(request, handle, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+                    handle.write("\n")
+                    handle.flush()
+                    os.fsync(handle.fileno())
+                os.replace(name, request_path)
+            finally:
+                if os.path.exists(name):
+                    os.unlink(name)
+
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            if result_path.is_file():
+                return self._validated_result(result_path, request, request_sha)
+            if failed_path.is_file():
+                try:
+                    value = json.loads(failed_path.read_text(encoding="utf-8"))
+                except json.JSONDecodeError:
+                    value = {"error": "EXTERNAL_AGENT_EXECUTOR_FAILED"}
+                raise ExternalAgentRuntimeError(
+                    str(value.get("error") or "EXTERNAL_AGENT_EXECUTOR_FAILED")
+                )
+            time.sleep(0.25)
+        raise ExternalAgentRuntimeError("EXTERNAL_AGENT_EXCHANGE_TIMEOUT")
+
+    @staticmethod
+    def _validated_result(
+        result_path: Path,
+        request: Mapping[str, Any],
+        request_sha: str,
+    ) -> dict[str, Any]:
+        try:
+            value = json.loads(result_path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError as exc:
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_RESULT_INVALID") from exc
+        if (
+            not isinstance(value, Mapping)
+            or value.get("schema") != RESULT_SCHEMA
+            or value.get("request_id") != request["request_id"]
+            or value.get("request_sha256") != request_sha
+            or value.get("runtime") != request["runtime"]
+            or value.get("task_id") != request["task_id"]
+            or value.get("execution_digest") != request["execution_digest"]
+        ):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_RESULT_BINDING_INVALID")
+        material = dict(value)
+        declared = str(material.pop("result_sha256", ""))
+        if declared != sha256(canonical(material)):
+            raise ExternalAgentRuntimeError("EXTERNAL_AGENT_RESULT_DIGEST_INVALID")
+        return dict(value)
+
+
+def serve(
+    root: Path = DEFAULT_EXCHANGE_ROOT,
+    profile_root: Path = DEFAULT_PROFILE_ROOT,
+) -> None:
+    _assert_no_secret_env()
+    root = Path(root)
+    for directory in (root / "requests", root / "results", root / "failed"):
+        directory.mkdir(parents=True, exist_ok=True)
+    while True:
+        for request_path in sorted((root / "requests").glob("*.json")):
+            if request_path.is_symlink():
+                continue
+            result_path = root / "results" / request_path.name
+            failed_path = root / "failed" / request_path.name
+            try:
+                execute_request(request_path, result_path, profile_root=profile_root)
+                request_path.unlink(missing_ok=True)
+            except Exception as exc:
+                failed_path.write_text(
+                    json.dumps(
+                        {"error": sanitize_text(str(exc) or type(exc).__name__)},
+                        ensure_ascii=False,
+                        sort_keys=True,
+                    )
+                    + "\n",
+                    encoding="utf-8",
+                )
+                request_path.unlink(missing_ok=True)
+        time.sleep(0.25)
+
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--root", default=str(DEFAULT_EXCHANGE_ROOT))
+    parser.add_argument("--profile-root", default=str(DEFAULT_PROFILE_ROOT))
+    parser.add_argument("--request")
+    parser.add_argument("--result")
+    args = parser.parse_args()
+    if args.request and args.result:
+        execute_request(
+            Path(args.request),
+            Path(args.result),
+            profile_root=Path(args.profile_root),
+        )
+    else:
+        serve(Path(args.root), Path(args.profile_root))

active-runtime/app/aios_online_agent_execution.py

@@ -24,7 +24,10 @@ BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
 KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,159}$")
 PROTECTED_PREFIXES = ("main", "master", "refs/", "tags/", "tag/", "release/", "baseline/", "production/")
 MUTATION_ACTIONS = frozenset({"deterministic"})
-SUPPORTED_ACTIONS = frozenset({"deterministic", "source_capture"})
+READ_ONLY_ACTIONS = frozenset({"source_capture", "agent_runtime"})
+SUPPORTED_ACTIONS = frozenset({"deterministic", "source_capture", "agent_runtime"})
+AGENT_RUNTIME_ADAPTERS = frozenset({"hermes", "openclaw"})
+MAX_AGENT_RUNTIME_INPUT_BYTES = 64 * 1024
 
 
 class ExecutionProjectionError(RuntimeError):
@@ -172,6 +175,50 @@ def _source_signals(value: object) -> tuple[str, ...]:
     return tuple(result)
 
 
+def _agent_runtime_input(value: object) -> dict[str, Any]:
+    if not isinstance(value, Mapping):
+        _error("AGENT_RUNTIME_INPUT_REQUIRED")
+    raw = dict(value)
+    allowed = {"goal", "instructions", "success_criteria", "constraints"}
+    if set(raw) - allowed:
+        _error("AGENT_RUNTIME_INPUT_FIELDS_INVALID")
+    goal = _string(raw.get("goal"), "AGENT_RUNTIME_GOAL_REQUIRED")
+    instructions = raw.get("instructions", "")
+    if not isinstance(instructions, str):
+        _error("AGENT_RUNTIME_INSTRUCTIONS_INVALID")
+    criteria = raw.get("success_criteria", [])
+    if not isinstance(criteria, list) or len(criteria) > 32:
+        _error("AGENT_RUNTIME_SUCCESS_CRITERIA_INVALID")
+    normalized_criteria: list[dict[str, str]] = []
+    for item in criteria:
+        if not isinstance(item, Mapping):
+            _error("AGENT_RUNTIME_SUCCESS_CRITERIA_INVALID")
+        criterion_id = _string(item.get("id"), "AGENT_RUNTIME_SUCCESS_CRITERIA_INVALID")
+        description = _string(
+            item.get("description"), "AGENT_RUNTIME_SUCCESS_CRITERIA_INVALID"
+        )
+        normalized_criteria.append(
+            {"id": criterion_id[:120], "description": description[:1000]}
+        )
+    constraints = raw.get("constraints", {})
+    if not isinstance(constraints, Mapping):
+        _error("AGENT_RUNTIME_CONSTRAINTS_INVALID")
+    sensitive_markers = ("token", "secret", "password", "credential", "authorization", "api_key", "apikey")
+    for key in constraints:
+        lowered = str(key).lower()
+        if any(marker in lowered for marker in sensitive_markers):
+            _error("AGENT_RUNTIME_SECRET_FIELD_DENY")
+    normalized = {
+        "goal": goal[:12000],
+        "instructions": instructions[:32000],
+        "success_criteria": normalized_criteria,
+        "constraints": dict(constraints),
+    }
+    if len(_canonical(normalized).encode("utf-8")) > MAX_AGENT_RUNTIME_INPUT_BYTES:
+        _error("AGENT_RUNTIME_INPUT_TOO_LARGE")
+    return normalized
+
+
 def _capabilities(value: object) -> tuple[str, ...]:
     if value is None:
         return ()
@@ -288,6 +335,22 @@ def validate_execution_envelope(
         signals = list(_source_signals(envelope.get("source_signals") or []))
         if envelope.get("source_signals") != signals:
             _error("SOURCE_CAPTURE_SIGNALS_INVALID")
+        if envelope.get("runtime_input") not in (None, {}):
+            _error("SOURCE_CAPTURE_RUNTIME_INPUT_DENY")
+    elif action == "agent_runtime":
+        if changes or verification:
+            _error("AGENT_RUNTIME_MUTATION_DENY")
+        adapter = str(envelope.get("adapter") or "")
+        if adapter not in AGENT_RUNTIME_ADAPTERS:
+            _error("AGENT_RUNTIME_ADAPTER_INVALID")
+        runtime_input = _agent_runtime_input(envelope.get("runtime_input"))
+        if envelope.get("runtime_input") != runtime_input:
+            _error("AGENT_RUNTIME_INPUT_NOT_CANONICAL")
+        required_runtime_capability = f"agent.runtime.{adapter}"
+        if required_runtime_capability not in capability_scope:
+            _error("AGENT_RUNTIME_CAPABILITY_REQUIRED")
+        if envelope.get("source_signals") not in (None, []):
+            _error("AGENT_RUNTIME_SOURCE_SIGNALS_DENY")
     elif not changes:
         _error("CHANGE_SET_INVALID")
     declared = str(envelope.get("execution_digest") or "")
@@ -329,11 +392,15 @@ def project_execution_envelope(
         "schema": EXECUTION_SCHEMA,
         **binding.public(),
         "action": action,
-        "adapter": str(request.get("adapter") or ("deterministic" if action == "deterministic" else "source_capture")),
+        "adapter": str(
+            request.get("adapter")
+            or ("deterministic" if action == "deterministic" else "source_capture")
+        ),
         "capability_scope": list(capabilities),
         "change_set": list(request.get("change_set") or []),
         "verification": list(request.get("verification") or []),
         "source_signals": list(_source_signals(request.get("source_signals") or [])) if action == "source_capture" else [],
+        "runtime_input": _agent_runtime_input(request.get("runtime_input")) if action == "agent_runtime" else {},
         "governance": governance,
         "workspace": {
             "authority_git_dir": "/data/aios-gitops/repo.git",
@@ -352,7 +419,25 @@ def project_online_task(
 ) -> dict[str, Any]:
     request = dict(task_request)
     required = _capabilities(request.get("required_capabilities"))
-    execution = project_execution_envelope(binding, execution_request, required_capabilities=required)
+    execution_value = dict(execution_request or {})
+    if str(execution_value.get("action") or "source_capture") == "agent_runtime":
+        adapter = str(execution_value.get("adapter") or "")
+        if adapter not in AGENT_RUNTIME_ADAPTERS:
+            _error("AGENT_RUNTIME_ADAPTER_INVALID")
+        runtime_input = execution_value.get("runtime_input")
+        if runtime_input is None:
+            runtime_input = {
+                "goal": request.get("goal"),
+                "instructions": request.get("instructions") or "",
+                "success_criteria": request.get("success_criteria") or [],
+                "constraints": request.get("constraints") or {},
+            }
+        execution_value["runtime_input"] = runtime_input
+    execution = project_execution_envelope(
+        binding,
+        execution_value,
+        required_capabilities=required,
+    )
     metadata = dict(request.get("metadata") or {})
     if "execution" in metadata:
         validate_execution_envelope(_object(metadata["execution"], "EXECUTION_ENVELOPE_INVALID"), binding=binding)

active-runtime/app/aios_online_agent_worker.py

@@ -44,6 +44,10 @@ from aios_online_agent_checkpoint import (
     CheckpointError,
     CheckpointStore,
 )
+from aios_external_agent_runtime import (
+    ExternalAgentExchange,
+    ExternalAgentRuntimeError,
+)
 from aios_online_agent_compact_map import (
     CompactInterfaceMapError,
     build_compact_interface_map,
@@ -83,6 +87,23 @@ WORKER_CAPABILITIES = tuple(
         if item.strip()
     )
 )
+WORKER_ALLOWED_ADAPTERS = frozenset(
+    item.strip()
+    for item in os.environ.get(
+        "AIOS_WORKER_ALLOWED_ADAPTERS",
+        "deterministic,source_capture,compact_interface_map,codex",
+    ).split(",")
+    if item.strip()
+)
+EXTERNAL_AGENT_EXECUTOR_ROOT = Path(
+    os.environ.get(
+        "AIOS_EXTERNAL_AGENT_EXECUTOR_ROOT",
+        "/var/lib/aios-external-agent-executor",
+    )
+)
+EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS = int(
+    os.environ.get("AIOS_EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS", "1800")
+)
 MAX_TASK_BYTES = 512 * 1024
 MAX_FILE_BYTES = 2 * 1024 * 1024
 ISOLATED_EXECUTOR_ROOT = Path(
@@ -359,6 +380,7 @@ class ExecutionEnvelope:
     governance: dict[str, Any]
     workspace: dict[str, Any]
     digest: str
+    runtime_input: dict[str, Any] | None = None
 
     @classmethod
     def from_task(
@@ -403,7 +425,7 @@ class ExecutionEnvelope:
             adapter = str(adapter_value.get("name") or "")
         else:
             adapter = str(adapter_value or "")
-        if adapter not in {"deterministic", "source_capture", "compact_interface_map", "codex", "hermes"}:
+        if adapter not in {"deterministic", "source_capture", "compact_interface_map", "codex", "hermes", "openclaw"}:
             raise WorkerError("RUNTIME_ADAPTER_UNSUPPORTED", phase="validation")
         change_set = execution.get("change_set")
         if not isinstance(change_set, list) or len(change_set) > 128 or (action == "deterministic" and not change_set):
@@ -494,6 +516,7 @@ class ExecutionEnvelope:
             adapter=adapter,
             capability_scope=tuple(execution.get("capability_scope") or ()),
             source_signals=tuple(execution.get("source_signals") or ()),
+            runtime_input=dict(execution.get("runtime_input") or {}),
             change_set=tuple(normalized_changes),
             verification=tuple(normalized_verification),
             governance=governance_value,
@@ -689,15 +712,76 @@ class CodexExecutor(RuntimeAdapter):
         )
 
 
-class HermesExecutor(RuntimeAdapter):
-    name = "hermes"
+class ExternalAgentExecutor(RuntimeAdapter):
+    runtime_name: str
 
     def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
+        if envelope.action != "agent_runtime":
             raise WorkerError(
-            "HERMES_ENGINEERING_ADAPTER_UNAVAILABLE",
+                "EXTERNAL_AGENT_ACTION_INVALID",
                 phase="adapter",
-            details={"reason": "The live Hermes service is a user-chat runtime, not an engineering executor"},
+                details={"adapter": self.runtime_name, "action": envelope.action},
             )
+        if envelope.adapter != self.runtime_name:
+            raise WorkerError(
+                "EXTERNAL_AGENT_ADAPTER_BINDING_INVALID",
+                phase="adapter",
+            )
+        if not envelope.runtime_input:
+            raise WorkerError("EXTERNAL_AGENT_RUNTIME_INPUT_MISSING", phase="adapter")
+        request_id = (
+            f"external-{self.runtime_name}-"
+            + digest_json(
+                {
+                    "task_id": envelope.task_id,
+                    "job_id": envelope.job_id,
+                    "execution_digest": envelope.digest,
+                    "runtime": self.runtime_name,
+                }
+            )[:32]
+        )
+        exchange = ExternalAgentExchange(EXTERNAL_AGENT_EXECUTOR_ROOT)
+        try:
+            external = exchange.execute(
+                request_id=request_id,
+                runtime=self.runtime_name,
+                task_id=envelope.task_id,
+                execution_digest=envelope.digest,
+                payload=envelope.runtime_input,
+                timeout_seconds=EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS,
+            )
+        except ExternalAgentRuntimeError as exc:
+            raise WorkerError(
+                str(exc),
+                phase="external_agent",
+                details={"adapter": self.runtime_name},
+            ) from exc
+        audit(
+            "EXTERNAL_AGENT_RUNTIME_COMPLETED",
+            task_id=envelope.task_id,
+            adapter=self.runtime_name,
+            request_id=request_id,
+            output_sha256=external.get("output_sha256"),
+            duration_ms=external.get("duration_ms"),
+            credential_material_exposed=False,
+        )
+        return {
+            "changed_files": [],
+            "commands": [],
+            "tests": [],
+            "adapter": self.runtime_name,
+            "external_agent": external,
+        }
+
+
+class HermesExecutor(ExternalAgentExecutor):
+    name = "hermes"
+    runtime_name = "hermes"
+
+
+class OpenClawExecutor(ExternalAgentExecutor):
+    name = "openclaw"
+    runtime_name = "openclaw"
 
 
 def adapter_for(name: str) -> RuntimeAdapter:
@@ -707,6 +791,7 @@ def adapter_for(name: str) -> RuntimeAdapter:
         "compact_interface_map": CompactInterfaceMapExecutor,
         "codex": CodexExecutor,
         "hermes": HermesExecutor,
+        "openclaw": OpenClawExecutor,
     }[name]()
 
 
@@ -1575,6 +1660,69 @@ def build_compact_interface_map_result(
     return result
 
 
+def build_agent_runtime_result(
+    task_id: str,
+    envelope: ExecutionEnvelope,
+    execution: Mapping[str, Any],
+) -> dict[str, Any]:
+    external = execution.get("external_agent")
+    if not isinstance(external, Mapping):
+        raise WorkerError("EXTERNAL_AGENT_RESULT_INVALID", phase="result")
+    if external.get("runtime") != envelope.adapter:
+        raise WorkerError("EXTERNAL_AGENT_RESULT_RUNTIME_MISMATCH", phase="result")
+    if external.get("task_id") != task_id:
+        raise WorkerError("EXTERNAL_AGENT_RESULT_TASK_MISMATCH", phase="result")
+    if external.get("execution_digest") != envelope.digest:
+        raise WorkerError("EXTERNAL_AGENT_RESULT_DIGEST_MISMATCH", phase="result")
+    if external.get("credential_material_exposed") is not False:
+        raise WorkerError("EXTERNAL_AGENT_CREDENTIAL_ISOLATION_FAILED", phase="result")
+    result = {
+        "status": "SUCCEEDED",
+        "classification": "EXTERNAL_AGENT_RUNTIME_SUCCEEDED",
+        "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
+        "execution_mode": "agent_runtime",
+        "execution_backend": "generic-online-agent-worker-v2",
+        "agent_runtime": envelope.adapter,
+        "task_id": task_id,
+        "job_id": envelope.job_id,
+        "repository": envelope.repository,
+        "base_oid": envelope.base_oid,
+        "expected_branch_oid": envelope.expected_branch_oid,
+        "target_branch": envelope.target_branch,
+        "generation": envelope.generation,
+        "execution_digest": envelope.digest,
+        "output_text": external.get("output_text"),
+        "output_json": external.get("output_json"),
+        "output_sha256": external.get("output_sha256"),
+        "runtime_duration_ms": external.get("duration_ms"),
+        "external_request_id": external.get("request_id"),
+        "changed_files": [],
+        "tests": [],
+        "workspace_cleanup": "PASS",
+        "writer_lease": "NONE",
+        "secret_material_exposed": False,
+        "credential_material_exposed": False,
+        "control_plane_access_granted_to_runtime": False,
+        "direct_main_write": False,
+        "force_push": False,
+        "production_write": False,
+        "protected_ref_bypass": False,
+    }
+    encoded = json.dumps(
+        sanitize(result),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    if len(encoded) > 192 * 1024:
+        raise WorkerError(
+            "EXTERNAL_AGENT_TASK_RESULT_TOO_LARGE",
+            phase="result",
+            details={"result_bytes": len(encoded)},
+        )
+    result["result_bytes"] = len(encoded)
+    return result
+
 def build_source_capture_result(
     task_id: str,
     envelope: ExecutionEnvelope,
@@ -1624,10 +1772,12 @@ class WorkerSupervisor:
             if not isinstance(execution, Mapping):
                 raise WorkerError("EXECUTION_ENVELOPE_SCHEMA_VALIDATION", phase="validation")
             persisted = ExecutionEnvelope.from_task(task, worker_capabilities=WORKER_CAPABILITIES)
-            # The durable plan registry is authoritative for a task that was projected
-            # before its deterministic plan was registered.  Re-resolve without sending
-            # the stale persisted payload so the Bridge can apply the exact plan/CAS.
-            status, authoritative = self.bridge.project_execution(task)
+            # Deterministic tasks must re-resolve through the plan registry.  A
+            # read-only source-capture task has no plan to replace its payload: replay
+            # the already validated durable envelope so source signals remain part of
+            # the canonical digest and the worker does not invent a different capture.
+            projection_input = execution if persisted.action == "source_capture" else None
+            status, authoritative = self.bridge.project_execution(task, projection_input)
             if not 200 <= status < 300:
                 raise WorkerError("TASK_PERSISTED_BINDING_UNVERIFIED", phase="projection", details={"api_status": status, "error": authoritative.get("error")})
             canonical = ExecutionEnvelope.from_task(task, authoritative, WORKER_CAPABILITIES)
@@ -1824,6 +1974,43 @@ class WorkerSupervisor:
                     attempt=attempt,
                 )
 
+            if envelope.action == "agent_runtime" and phase in {
+                "agent_runtime_done",
+                "result_pending",
+                "result_submitted",
+            }:
+                candidate = state.get("result")
+                if isinstance(candidate, Mapping):
+                    result = dict(candidate)
+                    audit(
+                        "TASK_RESUMED_FROM_EXTERNAL_AGENT_RESULT",
+                        task_id=task_id,
+                        previous_attempt=state.get("resume_from_attempt"),
+                        attempt=attempt,
+                        phase=phase,
+                        adapter=envelope.adapter,
+                    )
+            if (
+                result is None
+                and envelope.action == "agent_runtime"
+                and phase == "verification_passed"
+                and isinstance(state.get("execution"), Mapping)
+            ):
+                result = build_agent_runtime_result(
+                    task_id, envelope, state["execution"]
+                )
+                state.update(
+                    {"phase": "agent_runtime_done", "result": result, "attempt": attempt}
+                )
+                save_task_state(task_id, state)
+                save_checkpoint(task_id, state, envelope, attempt)
+                audit(
+                    "TASK_RESUMED_AFTER_EXTERNAL_AGENT_EXECUTION",
+                    task_id=task_id,
+                    attempt=attempt,
+                    adapter=envelope.adapter,
+                )
+
             if (
                 result is None
                 and envelope.action == "deterministic"
@@ -1904,6 +2091,10 @@ class WorkerSupervisor:
                 # re-run deterministic verification.  No external side effect exists yet.
                 run_root = new_run_root()
                 try:
+                    if envelope.action == "agent_runtime":
+                        worktree = run_root / "agent-runtime-workspace"
+                        worktree.mkdir(mode=0o750, parents=True, exist_ok=True)
+                    else:
                         worktree = prepare_worktree(run_root, envelope)
                     state.update(
                         {
@@ -1938,6 +2129,17 @@ class WorkerSupervisor:
                                 "attempt": attempt,
                             }
                         )
+                    elif envelope.action == "agent_runtime":
+                        result = build_agent_runtime_result(
+                            task_id, envelope, execution
+                        )
+                        state.update(
+                            {
+                                "phase": "agent_runtime_done",
+                                "result": result,
+                                "attempt": attempt,
+                            }
+                        )
                     else:
                         heartbeat.set_stage("committing")
                         governed = governed_commit(
@@ -2105,6 +2307,14 @@ class WorkerSupervisor:
                 audit(event, task_id=task.get("task_id"), reason=exc.code, phase=exc.phase, details=exc.details)
                 continue
             audit("TASK_VISIBLE_TO_WORKER_CLAIMABLE", task_id=task.get("task_id"), envelope_source=source, action=envelope.action)
+            if envelope.adapter not in WORKER_ALLOWED_ADAPTERS:
+                audit(
+                    "TASK_SKIPPED_ADAPTER_MISMATCH",
+                    task_id=task.get("task_id"),
+                    adapter=envelope.adapter,
+                    allowed_adapters=sorted(WORKER_ALLOWED_ADAPTERS),
+                )
+                continue
             dynamic_tasks.append((task, envelope))
         dynamic_tasks.sort(key=_claim_order)
         for task, envelope in dynamic_tasks:

config/external-agent-runtime.d/hermes-config.yaml

@@ -0,0 +1,9 @@
+providers:
+  oneapi:
+    name: oneapi
+    base_url: http://172.18.0.2:3000/v1
+    key_env: OPENAI_API_KEY
+    default_model: deepseek-v4-flash
+model:
+  provider: oneapi
+  name: deepseek-v4-flash

config/external-agent-runtime.d/hermes.example.json

@@ -0,0 +1,13 @@
+{
+  "argv": [
+    "/usr/local/bin/hermes",
+    "run",
+    "--non-interactive"
+  ],
+  "input_mode": "stdin_text",
+  "max_output_bytes": 131072,
+  "output_mode": "stdout_text",
+  "runtime": "hermes",
+  "schema": "aios.external-agent-runtime.profile.v1",
+  "timeout_seconds": 900
+}

config/external-agent-runtime.d/hermes.profile.json

@@ -0,0 +1,28 @@
+{
+  "argv": [
+    "/opt/aios-hermes-agent/hermes-agent/.venv/bin/hermes",
+    "-z",
+    "{prompt}",
+    "--provider",
+    "oneapi",
+    "-m",
+    "deepseek-v4-flash",
+    "--no-restore-cwd"
+  ],
+  "credential_env": {
+    "OPENAI_API_KEY": {
+      "credential": "aios-external-agent-oneapi-env",
+      "format": "dotenv",
+      "key": "ONEAPI_REGULAR_TOKEN"
+    }
+  },
+  "environment": {
+    "HERMES_HOME": "/etc/aios/external-agent-runtime.d/hermes-home"
+  },
+  "input_mode": "stdin_text",
+  "max_output_bytes": 131072,
+  "output_mode": "stdout_text",
+  "runtime": "hermes",
+  "schema": "aios.external-agent-runtime.profile.v1",
+  "timeout_seconds": 180
+}

config/external-agent-runtime.d/openclaw-config.json

@@ -0,0 +1,41 @@
+{
+  "tools": {
+    "profile": "minimal"
+  },
+  "memory": {
+    "search": {
+      "enabled": false
+    }
+  },
+  "agents": {
+    "defaults": {
+      "workspace": "/var/lib/aios-external-agent-executor/openclaw-workspace",
+      "model": {
+        "primary": "openai/deepseek-v4-flash"
+      },
+      "thinkingDefault": "off"
+    },
+    "entries": {
+      "main": {
+        "default": true,
+        "workspace": "/var/lib/aios-external-agent-executor/openclaw-workspace"
+      }
+    }
+  },
+  "models": {
+    "providers": {
+      "openai": {
+        "baseUrl": "http://172.18.0.2:3000/v1",
+        "api": "openai-completions",
+        "models": [
+          {
+            "id": "deepseek-v4-flash",
+            "name": "deepseek-v4-flash",
+            "contextWindow": 32768,
+            "maxTokens": 256
+          }
+        ]
+      }
+    }
+  }
+}

config/external-agent-runtime.d/openclaw.example.json

@@ -0,0 +1,13 @@
+{
+  "argv": [
+    "/usr/local/bin/openclaw",
+    "run",
+    "--non-interactive"
+  ],
+  "input_mode": "stdin_text",
+  "max_output_bytes": 131072,
+  "output_mode": "stdout_text",
+  "runtime": "openclaw",
+  "schema": "aios.external-agent-runtime.profile.v1",
+  "timeout_seconds": 900
+}

config/external-agent-runtime.d/openclaw.profile.json

@@ -0,0 +1,37 @@
+{
+  "argv": [
+    "/usr/bin/node",
+    "/opt/aios/openclaw-engineering/v2026.7.2-beta.7/openclaw.mjs",
+    "agent",
+    "--local",
+    "--json",
+    "--agent",
+    "main",
+    "--thinking",
+    "off",
+    "--message",
+    "{prompt}",
+    "--model",
+    "openai/deepseek-v4-flash",
+    "--timeout",
+    "180"
+  ],
+  "credential_env": {
+    "OPENAI_API_KEY": {
+      "credential": "aios-external-agent-oneapi-env",
+      "format": "dotenv",
+      "key": "ONEAPI_REGULAR_TOKEN"
+    }
+  },
+  "environment": {
+    "OPENCLAW_CONFIG_PATH": "/etc/aios/external-agent-runtime.d/openclaw-config.json",
+    "OPENCLAW_HOME": "/var/lib/aios-external-agent-executor/openclaw-home",
+    "OPENCLAW_STATE_DIR": "/var/lib/aios-external-agent-executor/openclaw-state"
+  },
+  "input_mode": "stdin_text",
+  "max_output_bytes": 131072,
+  "output_mode": "stdout_json",
+  "runtime": "openclaw",
+  "schema": "aios.external-agent-runtime.profile.v1",
+  "timeout_seconds": 240
+}

deploy/systemd/aios-external-agent-executor.service

@@ -0,0 +1,40 @@
+[Unit]
+Description=AIOS credential-isolated external Agent runtime executor
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=aios-engineering-executor
+Group=aios-executor-exchange
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_external_agent_runtime.py --root /var/lib/aios-external-agent-executor --profile-root /etc/aios/external-agent-runtime.d
+Restart=always
+RestartSec=2
+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
+IPAddressDeny=127.0.0.0/8
+IPAddressDeny=::1/128
+IPAddressAllow=172.18.0.2/32
+LoadCredential=aios-external-agent-oneapi-env:/data/agent/secrets/oneapi-model-ready.env
+InaccessiblePaths=/etc/aios-gitops
+InaccessiblePaths=/data/aios-gitops
+InaccessiblePaths=/data/agent/secrets
+ReadOnlyPaths=/etc/aios/external-agent-runtime.d
+ReadOnlyPaths=/opt/aios-hermes-agent
+ReadOnlyPaths=/opt/aios/openclaw-engineering
+ReadWritePaths=/var/lib/aios-external-agent-executor
+UMask=0027
+
+[Install]
+WantedBy=multi-user.target

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

@@ -0,0 +1,48 @@
+[Unit]
+Description=AIOS Hermes Online Agent Worker
+After=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service aios-external-agent-executor.service
+Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service aios-external-agent-executor.service
+
+[Service]
+Type=simple
+User=aios-engineering-worker
+Group=aios-engineering-worker
+SupplementaryGroups=gitops aios-executor-exchange
+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-hermes-worker-cn-network-mkt
+Environment=AIOS_AGENT_TYPE=external-agent-runtime-worker
+Environment=AIOS_AGENT_VERSION=2026-08-18.1-hermes-runtime
+Environment=AIOS_WORKER_CAPABILITIES=agent.runtime.hermes
+Environment=AIOS_WORKER_ALLOWED_ADAPTERS=hermes
+Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_EXTERNAL_AGENT_EXECUTOR_ROOT=/var/lib/aios-external-agent-executor
+Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-hermes
+Environment=AIOS_WORKER_CHECKPOINT_DB=/var/lib/aios-online-agent-hermes/checkpoints.sqlite3
+Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-hermes/workspaces
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_online_agent_worker.py
+Restart=always
+RestartSec=5
+TimeoutStopSec=20
+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-hermes
+ReadWritePaths=/var/lib/aios-external-agent-executor
+UMask=0027
+
+[Install]
+WantedBy=multi-user.target

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

@@ -0,0 +1,48 @@
+[Unit]
+Description=AIOS OpenClaw Online Agent Worker
+After=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service aios-external-agent-executor.service
+Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service aios-external-agent-executor.service
+
+[Service]
+Type=simple
+User=aios-engineering-worker
+Group=aios-engineering-worker
+SupplementaryGroups=gitops aios-executor-exchange
+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-openclaw-worker-cn-network-mkt
+Environment=AIOS_AGENT_TYPE=external-agent-runtime-worker
+Environment=AIOS_AGENT_VERSION=2026-08-18.1-openclaw-runtime
+Environment=AIOS_WORKER_CAPABILITIES=agent.runtime.openclaw
+Environment=AIOS_WORKER_ALLOWED_ADAPTERS=openclaw
+Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_EXTERNAL_AGENT_EXECUTOR_ROOT=/var/lib/aios-external-agent-executor
+Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-openclaw
+Environment=AIOS_WORKER_CHECKPOINT_DB=/var/lib/aios-online-agent-openclaw/checkpoints.sqlite3
+Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-openclaw/workspaces
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_online_agent_worker.py
+Restart=always
+RestartSec=5
+TimeoutStopSec=20
+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-openclaw
+ReadWritePaths=/var/lib/aios-external-agent-executor
+UMask=0027
+
+[Install]
+WantedBy=multi-user.target

deploy/tmpfiles.d/aios-external-agent-runtime.conf

@@ -0,0 +1,12 @@
+d /var/lib/aios-external-agent-executor 2770 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-external-agent-executor/requests 2770 aios-engineering-worker aios-executor-exchange -
+d /var/lib/aios-external-agent-executor/results 2770 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-external-agent-executor/failed 2770 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-external-agent-executor/openclaw-workspace 0750 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-external-agent-executor/openclaw-state 0750 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-external-agent-executor/openclaw-home 0750 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-online-agent-hermes 0750 aios-engineering-worker aios-engineering-worker -
+d /var/lib/aios-online-agent-hermes/workspaces 0750 aios-engineering-worker aios-engineering-worker -
+d /var/lib/aios-online-agent-openclaw 0750 aios-engineering-worker aios-engineering-worker -
+d /var/lib/aios-online-agent-openclaw/workspaces 0750 aios-engineering-worker aios-engineering-worker -
+d /etc/aios/external-agent-runtime.d 0755 root root -

docs/superpowers/specs/2026-08-18-hermes-openclaw-online-agent-runtime.md

@@ -0,0 +1,142 @@
+# Hermes / OpenClaw Online Agent Runtime
+
+## Goal
+
+Promote Hermes and OpenClaw from shadow-only adapter descriptions to real Online Agent workers
+without creating a second Task/claim protocol.
+
+Both workers reuse the existing P1/P2 Online Agent lifecycle:
+
+```text
+Online Task projected
+→ capability-filtered poll
+→ automatic claim
+→ heartbeat
+→ durable checkpoint
+→ external Agent execution
+→ attempt-aware result submission
+→ official verification
+```
+
+They do **not** change the existing TaskPlane `OpenClawAdapter/HermesAdapter` shadow contract. Those
+adapters remain side-effect-free compatibility surfaces.
+
+## Execution action
+
+The canonical execution envelope adds a read-only action:
+
+```text
+action=agent_runtime
+adapter=hermes | openclaw
+```
+
+The action requires capability `agent.runtime.<adapter>`, has an empty change set, no verification
+commands, and `writer_lease=false`.
+
+`project_online_task()` copies only bounded task fields into the signed execution envelope:
+
+- goal;
+- instructions;
+- success criteria;
+- non-secret constraints.
+
+The task may not choose a command, executable, URL, credential path, environment, or shell string.
+
+## Worker separation
+
+The generic engineering worker defaults to:
+
+```text
+AIOS_WORKER_ALLOWED_ADAPTERS=deterministic,source_capture,compact_interface_map,codex
+```
+
+so it skips Hermes/OpenClaw tasks before claim.
+
+Dedicated services advertise:
+
+```text
+Hermes:
+  AIOS_WORKER_CAPABILITIES=agent.runtime.hermes
+  AIOS_WORKER_ALLOWED_ADAPTERS=hermes
+
+OpenClaw:
+  AIOS_WORKER_CAPABILITIES=agent.runtime.openclaw
+  AIOS_WORKER_ALLOWED_ADAPTERS=openclaw
+```
+
+This makes adapter selection a pre-claim capability/fleet decision instead of a post-claim failure.
+
+## Credential-isolated executor
+
+The Online Agent worker itself still owns only Task API/Bridge coordination credentials. It writes
+a hash-bound request to `/var/lib/aios-external-agent-executor`.
+
+`aios-external-agent-executor.service` runs as the existing uncredentialed engineering executor user.
+It has:
+
+- one provider-only `LoadCredential` projection for the model-ready OneAPI environment;
+- `/etc/aios-gitops`, `/data/aios-gitops`, and the original provider-secret directory inaccessible;
+- no repository path;
+- no shell execution;
+- loopback denied, preventing direct access to the AIOS Task/Bridge control plane;
+- only the staging OneAPI container address allowed for the installed Agent runtime/provider;
+- only the exchange directory writable.
+
+The root-owned runtime profile in `/etc/aios/external-agent-runtime.d/<runtime>.json` decides the
+absolute executable, fixed non-secret environment, provider-only credential mapping, and I/O mode.
+A task cannot override any of these values. The profile may select `OPENAI_API_KEY` only from the
+systemd credential named `aios-external-agent-oneapi-env`; it cannot name a credential, inject an
+environment variable, or carry an inline secret. The runtime executor parses only the selected
+`ONEAPI_REGULAR_TOKEN` entry and does not expose the source file or other provider tiers to the
+child.
+
+## Runtime profile
+
+A profile supports:
+
+- `argv`: fixed argv, absolute allowlisted executable;
+- `input_mode`: `stdin_text`, `stdin_json`, or `file_json`;
+- `output_mode`: `stdout_text`, `stdout_json`, or `file_json`;
+- timeout;
+- bounded output size.
+
+`{input_json}` and `{output_json}` placeholders are allowed only for file modes. `{prompt}` is a
+root-profile-only argv placeholder populated from the bounded, signed task projection.
+
+The repository includes the staging profile templates validated against the installed Hermes
+v0.20.0 and OpenClaw v2026.7.2-beta.7 CLIs. Before activation, the Agent must re-run the actual
+`--help`/version probes and install the profiles and static configs root-owned. If the installed
+CLI differs, update only the root-owned profile/config projection and record the observed command;
+do not modify the Task contract to compensate.
+
+The provider projection is staging-only and uses the existing model-ready OneAPI source. It is not
+an AIOS Task, Bridge, GitOps, SSH, or Writer credential. Live activation remains blocked if the
+model-ready source cannot be projected through systemd or if the selected provider credential is
+missing/quota-invalid. This worker never passes API keys through task metadata or arbitrary
+environment variables.
+
+## Retry/restart
+
+The external request ID is deterministic over Task/Job/execution digest/runtime. The exchange keeps
+the request/result across worker restart. A repeated attempt with the exact execution digest reuses
+the durable external result instead of invoking the Agent twice.
+
+The normal P1 result idempotency/checkpoint path remains authoritative for Task result submission.
+
+## Result
+
+Successful execution returns:
+
+```text
+status=SUCCEEDED
+classification=EXTERNAL_AGENT_RUNTIME_SUCCEEDED
+execution_mode=agent_runtime
+agent_runtime=hermes | openclaw
+writer_lease=NONE
+changed_files=[]
+credential_material_exposed=false
+control_plane_access_granted_to_runtime=false
+```
+
+This first runtime version is intentionally read-only with respect to the repository. Engineering
+changes continue to use assistant-authored deterministic plans and governed WriterLease/commit.

tests/test_agent_runtime_capability_contract.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime/app"))
+
+from aios_capability_contract import (  # noqa: E402
+    has_capabilities,
+    normalize_required_capabilities,
+    normalize_worker_capabilities,
+)
+
+
+@pytest.mark.parametrize("runtime", ["hermes", "openclaw"])
+def test_runtime_capability_is_known_to_task_and_worker_vocabulary(runtime: str) -> None:
+    capability = f"agent.runtime.{runtime}"
+    assert normalize_required_capabilities([capability]) == frozenset({capability})
+    assert normalize_worker_capabilities([capability]) == frozenset({capability})
+    assert has_capabilities([capability], [capability]) is True
+
+
+def test_runtime_capabilities_are_isolated_between_workers() -> None:
+    assert has_capabilities(["agent.runtime.hermes"], ["agent.runtime.openclaw"]) is False
+    assert has_capabilities(["agent.runtime.openclaw"], ["agent.runtime.hermes"]) is False
+
+
+def test_unknown_runtime_capability_remains_fail_closed() -> None:
+    assert has_capabilities(["agent.runtime.hermes"], ["agent.runtime.unknown"]) is False

tests/test_external_agent_online_runtime.py

@@ -0,0 +1,418 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+from aios_external_agent_runtime import (
+    ExternalAgentRuntimeError,
+    RuntimeProfile,
+    execute_request,
+)
+from aios_online_agent_execution import (
+    ExecutionProjectionError,
+    GovernedExecutionBinding,
+    project_online_task,
+    validate_execution_envelope,
+)
+import aios_online_agent_worker as worker
+
+
+BASE="a"*40
+
+
+def binding(task_id="task-hermes-runtime"):
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":task_id,
+        "job_id":"job-"+task_id,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":"agent/"+task_id,
+        "branch_mode":"CREATE_NEW",
+        "generation":1,
+        "idempotency_key":task_id+"-binding-v1",
+    })
+
+
+def task_request(task_id, runtime):
+    return {
+        "task_id":task_id,
+        "goal":"Summarize the requested operation.",
+        "instructions":"Return a concise result.",
+        "success_criteria":[{"id":"done","description":"A result is returned."}],
+        "constraints":{"network":"runtime-managed"},
+        "required_capabilities":[f"agent.runtime.{runtime}"],
+        "max_attempts":3,
+    }
+
+
+def projected(task_id="task-hermes-runtime", runtime="hermes"):
+    return project_online_task(
+        binding(task_id),
+        task_request(task_id,runtime),
+        {
+            "action":"agent_runtime",
+            "adapter":runtime,
+            "capability_scope":[f"agent.runtime.{runtime}"],
+        },
+    )
+
+
+def clear_secret_env(monkeypatch):
+    for key in list(os.environ):
+        upper=key.upper()
+        if any(marker in upper for marker in (
+            "TOKEN","SECRET","PASSWORD","CREDENTIAL","AUTHORIZATION","PRIVATE_KEY"
+        )):
+            monkeypatch.delenv(key,raising=False)
+    monkeypatch.setenv("CREDENTIALS_DIRECTORY","/definitely-not-present-aios-test")
+
+
+def make_helper(tmp_path):
+    helper=tmp_path/"runtime_helper.py"
+    helper.write_text(
+        "import json,sys\n"
+        "value=json.load(sys.stdin)\n"
+        "print(json.dumps({'answer':'ok','goal':value['goal']},sort_keys=True))\n",
+        encoding="utf-8",
+    )
+    return helper
+
+
+def make_profile(tmp_path,runtime,helper):
+    root=tmp_path/"profiles";root.mkdir(exist_ok=True)
+    (root/f"{runtime}.json").write_text(
+        json.dumps({
+            "schema":"aios.external-agent-runtime.profile.v1",
+            "runtime":runtime,
+            "argv":["/usr/bin/python3",str(helper)],
+            "input_mode":"stdin_json",
+            "output_mode":"stdout_json",
+            "timeout_seconds":30,
+            "max_output_bytes":65536,
+        }),
+        encoding="utf-8",
+    )
+    return root
+
+
+def make_text_request(tmp_path, runtime, payload=None):
+    request={
+        "schema":"aios.external-agent-runtime.request.v1",
+        "request_id":f"external-{runtime}-0123456789abcdef",
+        "runtime":runtime,
+        "task_id":f"task-{runtime}-runtime",
+        "execution_digest":"d"*64,
+        "payload":payload or {
+            "goal":"Return ok.",
+            "instructions":"No side effects.",
+            "success_criteria":[],
+            "constraints":{},
+        },
+    }
+    request_path=tmp_path/"request.json"
+    result_path=tmp_path/"result.json"
+    request_path.write_text(json.dumps(request),encoding="utf-8")
+    return request_path, result_path
+
+
+def test_agent_runtime_projection_is_read_only_and_task_bound():
+    value=projected()
+    execution=value["metadata"]["execution"]
+    assert execution["action"]=="agent_runtime"
+    assert execution["adapter"]=="hermes"
+    assert execution["runtime_input"]["goal"]==value["goal"]
+    assert execution["runtime_input"]["instructions"]==value["instructions"]
+    assert execution["governance"]["writer_lease"] is False
+    assert execution["change_set"]==[]
+    assert execution["verification"]==[]
+    validate_execution_envelope(
+        execution,
+        binding=binding(),
+        worker_capabilities=["agent.runtime.hermes"],
+    )
+
+
+def test_agent_runtime_projection_requires_runtime_capability():
+    with pytest.raises(ExecutionProjectionError,match="AGENT_RUNTIME_CAPABILITY_REQUIRED"):
+        project_online_task(
+            binding(),
+            task_request("task-hermes-runtime","hermes"),
+            {
+                "action":"agent_runtime",
+                "adapter":"hermes",
+                "capability_scope":["git.read"],
+            },
+        )
+
+
+def test_agent_runtime_projection_rejects_mutation():
+    with pytest.raises(ExecutionProjectionError,match="AGENT_RUNTIME_MUTATION_DENY"):
+        project_online_task(
+            binding(),
+            task_request("task-hermes-runtime","hermes"),
+            {
+                "action":"agent_runtime",
+                "adapter":"hermes",
+                "capability_scope":["agent.runtime.hermes"],
+                "change_set":[{"path":"x.py","operation":"write_file","content":"x=1\n"}],
+            },
+        )
+
+
+def test_agent_runtime_projection_rejects_secret_constraint_keys():
+    request=task_request("task-hermes-runtime","hermes")
+    request["constraints"]={"api_token":"do-not-project"}
+    with pytest.raises(ExecutionProjectionError,match="AGENT_RUNTIME_SECRET_FIELD_DENY"):
+        project_online_task(
+            binding(),
+            request,
+            {
+                "action":"agent_runtime",
+                "adapter":"hermes",
+                "capability_scope":["agent.runtime.hermes"],
+            },
+        )
+
+
+@pytest.mark.parametrize("runtime",["hermes","openclaw"])
+def test_external_executor_runs_root_owned_profile_without_shell(tmp_path,monkeypatch,runtime):
+    clear_secret_env(monkeypatch)
+    helper=make_helper(tmp_path)
+    profile_root=make_profile(tmp_path,runtime,helper)
+    request={
+        "schema":"aios.external-agent-runtime.request.v1",
+        "request_id":f"external-{runtime}-0123456789abcdef",
+        "runtime":runtime,
+        "task_id":f"task-{runtime}-runtime",
+        "execution_digest":"d"*64,
+        "payload":{
+            "goal":"Return ok.",
+            "instructions":"No side effects.",
+            "success_criteria":[],
+            "constraints":{},
+        },
+    }
+    request_path=tmp_path/"request.json"
+    result_path=tmp_path/"result.json"
+    request_path.write_text(json.dumps(request),encoding="utf-8")
+    execute_request(request_path,result_path,profile_root=profile_root)
+    result=json.loads(result_path.read_text())
+    assert result["runtime"]==runtime
+    assert result["exit_code"]==0
+    assert result["output_json"]["answer"]=="ok"
+    assert result["credential_material_exposed"] is False
+    assert result["control_plane_access_granted"] is False
+
+
+def test_runtime_profile_rejects_shell_mode(tmp_path):
+    root=tmp_path/"profiles";root.mkdir()
+    (root/"hermes.json").write_text(json.dumps({
+        "schema":"aios.external-agent-runtime.profile.v1",
+        "runtime":"hermes",
+        "argv":["/usr/bin/python3","-c","print('x')"],
+        "input_mode":"stdin_text",
+        "output_mode":"stdout_text",
+    }))
+    with pytest.raises(ExternalAgentRuntimeError,match="EXTERNAL_AGENT_SHELL_MODE_DENY"):
+        RuntimeProfile.load("hermes",root)
+
+
+def test_external_executor_projects_only_selected_provider_credential(tmp_path,monkeypatch):
+    clear_secret_env(monkeypatch)
+    helper=tmp_path/"credential_helper.py"
+    helper.write_text(
+        "import os\n"
+        "value=os.environ.get('OPENAI_API_KEY','')\n"
+        "print('AIOS_PROVIDER_PROJECTED_OK' if value == 'test-provider-key' else 'AIOS_PROVIDER_PROJECTED_FAIL')\n",
+        encoding="utf-8",
+    )
+    profile_root=tmp_path/"profiles";profile_root.mkdir()
+    (profile_root/"hermes.json").write_text(json.dumps({
+        "schema":"aios.external-agent-runtime.profile.v1",
+        "runtime":"hermes",
+        "argv":["/usr/bin/python3",str(helper)],
+        "input_mode":"stdin_text",
+        "output_mode":"stdout_text",
+        "environment":{"RUNTIME_PROFILE_MARKER":"staging"},
+        "credential_env":{
+            "OPENAI_API_KEY":{
+                "credential":"provider.env",
+                "format":"dotenv",
+                "key":"ONEAPI_REGULAR_TOKEN",
+            }
+        },
+    }),encoding="utf-8")
+    credential_dir=tmp_path/"credentials";credential_dir.mkdir()
+    (credential_dir/"provider.env").write_text(
+        "ONEAPI_REGULAR_TOKEN=test-provider-key\nOTHER_PROVIDER_TOKEN=not-projected\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("CREDENTIALS_DIRECTORY",str(credential_dir))
+    request_path,result_path=make_text_request(tmp_path,"hermes")
+    execute_request(request_path,result_path,profile_root=profile_root)
+    result=json.loads(result_path.read_text())
+    assert "AIOS_PROVIDER_PROJECTED_OK" in result["output_text"]
+    assert "test-provider-key" not in result["output_text"]
+    assert result["credential_material_exposed"] is False
+
+
+def test_runtime_profile_rejects_credential_path_traversal(tmp_path):
+    root=tmp_path/"profiles";root.mkdir()
+    (root/"hermes.json").write_text(json.dumps({
+        "schema":"aios.external-agent-runtime.profile.v1",
+        "runtime":"hermes",
+        "argv":["/usr/bin/python3","-V"],
+        "input_mode":"stdin_text",
+        "output_mode":"stdout_text",
+        "credential_env":{
+            "OPENAI_API_KEY":{
+                "credential":"../provider.env",
+                "format":"dotenv",
+                "key":"ONEAPI_REGULAR_TOKEN",
+            }
+        },
+    }),encoding="utf-8")
+    with pytest.raises(ExternalAgentRuntimeError,match="EXTERNAL_AGENT_CREDENTIAL_NAME_INVALID"):
+        RuntimeProfile.load("hermes",root)
+
+
+def test_runtime_profile_prompt_placeholder_is_root_owned(tmp_path,monkeypatch):
+    clear_secret_env(monkeypatch)
+    helper=tmp_path/"prompt_helper.py"
+    helper.write_text(
+        "import sys\n"
+        "print('AIOS_PROMPT_PLACEHOLDER_OK' if 'Goal:' in sys.argv[1] else 'AIOS_PROMPT_PLACEHOLDER_FAIL')\n",
+        encoding="utf-8",
+    )
+    profile_root=tmp_path/"profiles";profile_root.mkdir()
+    (profile_root/"hermes.json").write_text(json.dumps({
+        "schema":"aios.external-agent-runtime.profile.v1",
+        "runtime":"hermes",
+        "argv":["/usr/bin/python3",str(helper),"{prompt}"],
+        "input_mode":"stdin_text",
+        "output_mode":"stdout_text",
+    }),encoding="utf-8")
+    request_path,result_path=make_text_request(tmp_path,"hermes")
+    execute_request(request_path,result_path,profile_root=profile_root)
+    result=json.loads(result_path.read_text())
+    assert "AIOS_PROMPT_PLACEHOLDER_OK" in result["output_text"]
+
+
+@pytest.mark.parametrize("runtime,capability",[
+    ("hermes","agent.runtime.hermes"),
+    ("openclaw","agent.runtime.openclaw"),
+])
+def test_worker_external_adapter_builds_readonly_result(monkeypatch,tmp_path,runtime,capability):
+    mapped=projected(f"task-{runtime}-runtime",runtime)
+    envelope=worker.ExecutionEnvelope.from_task(
+        mapped,
+        worker_capabilities=(capability,),
+    )
+    class FakeExchange:
+        def __init__(self,*args,**kwargs): pass
+        def execute(self,**kwargs):
+            return {
+                "schema":"aios.external-agent-runtime.result.v1",
+                "request_id":kwargs["request_id"],
+                "request_sha256":"e"*64,
+                "runtime":runtime,
+                "task_id":kwargs["task_id"],
+                "execution_digest":kwargs["execution_digest"],
+                "exit_code":0,
+                "duration_ms":10,
+                "output_text":"done",
+                "output_sha256":"f"*64,
+                "stderr_sha256":"0"*64,
+                "credential_material_exposed":False,
+                "control_plane_access_granted":False,
+                "result_sha256":"1"*64,
+            }
+    monkeypatch.setattr(worker,"ExternalAgentExchange",FakeExchange)
+    monkeypatch.setattr(worker,"audit",lambda *a,**k:None)
+    executor=worker.HermesExecutor() if runtime=="hermes" else worker.OpenClawExecutor()
+    execution=executor.execute(envelope,tmp_path)
+    result=worker.build_agent_runtime_result(envelope.task_id,envelope,execution)
+    assert result["status"]=="SUCCEEDED"
+    assert result["classification"]=="EXTERNAL_AGENT_RUNTIME_SUCCEEDED"
+    assert result["execution_mode"]=="agent_runtime"
+    assert result["agent_runtime"]==runtime
+    assert result["writer_lease"]=="NONE"
+    assert result["changed_files"]==[]
+    assert result["control_plane_access_granted_to_runtime"] is False
+
+
+def test_worker_adapter_filter_defaults_exclude_external_runtimes():
+    assert "hermes" not in worker.WORKER_ALLOWED_ADAPTERS
+    assert "openclaw" not in worker.WORKER_ALLOWED_ADAPTERS
+
+
+def test_external_executor_systemd_has_no_control_credentials_or_repository_access():
+    unit=(ROOT/"deploy/systemd/aios-external-agent-executor.service").read_text()
+    assert "User=aios-engineering-executor" in unit
+    assert "LoadCredential=aios-external-agent-oneapi-env:/data/agent/secrets/oneapi-model-ready.env" in unit
+    assert "LoadCredentialEncrypted" not in unit
+    assert "InaccessiblePaths=/run/credentials" not in unit
+    assert "InaccessiblePaths=/etc/aios-gitops" in unit
+    assert "InaccessiblePaths=/data/aios-gitops" in unit
+    assert "InaccessiblePaths=/data/agent/secrets" in unit
+    assert "IPAddressDeny=127.0.0.0/8" in unit
+    assert "IPAddressDeny=::1/128" in unit
+    assert "IPAddressAllow=172.18.0.2/32" in unit
+
+
+@pytest.mark.parametrize(
+    "unit_name,capability,adapter,state_root",
+    [
+        (
+            "aios-hermes-online-agent-worker.service",
+            "agent.runtime.hermes",
+            "hermes",
+            "/var/lib/aios-online-agent-hermes",
+        ),
+        (
+            "aios-openclaw-online-agent-worker.service",
+            "agent.runtime.openclaw",
+            "openclaw",
+            "/var/lib/aios-online-agent-openclaw",
+        ),
+    ],
+)
+def test_dedicated_worker_units_are_capability_and_adapter_scoped(
+    unit_name,capability,adapter,state_root
+):
+    unit=(ROOT/"deploy/systemd"/unit_name).read_text()
+    assert f"Environment=AIOS_WORKER_CAPABILITIES={capability}" in unit
+    assert f"Environment=AIOS_WORKER_ALLOWED_ADAPTERS={adapter}" in unit
+    assert f"Environment=AIOS_WORKER_STATE_ROOT={state_root}" in unit
+    assert "AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
+    assert "LoadCredentialEncrypted=agent_token:" in unit
+
+
+def test_example_profiles_are_non_authoritative_examples_only():
+    for runtime in ("hermes","openclaw"):
+        value=json.loads(
+            (ROOT/"config/external-agent-runtime.d"/f"{runtime}.example.json").read_text()
+        )
+        assert value["schema"]=="aios.external-agent-runtime.profile.v1"
+        assert value["runtime"]==runtime
+        assert value["input_mode"]=="stdin_text"
+        assert value["output_mode"]=="stdout_text"
+
+
+def test_staging_profiles_are_provider_only_and_root_profile_bound():
+    hermes=json.loads((ROOT/"config/external-agent-runtime.d/hermes.profile.json").read_text())
+    openclaw=json.loads((ROOT/"config/external-agent-runtime.d/openclaw.profile.json").read_text())
+    for runtime,value in (("hermes",hermes),("openclaw",openclaw)):
+        assert value["runtime"]==runtime
+        assert set(value["credential_env"])=={"OPENAI_API_KEY"}
+        assert value["credential_env"]["OPENAI_API_KEY"]["credential"]=="aios-external-agent-oneapi-env"
+        assert "OPENAI_API_KEY" not in value.get("environment",{})

tests/test_online_agent_source_capture_replay.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime/app"))
+
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+import aios_online_agent_worker as worker
+from aios_online_agent_worker import WorkerSupervisor
+
+
+BASE = "a" * 40
+
+
+def _binding() -> GovernedExecutionBinding:
+    return GovernedExecutionBinding.from_mapping(
+        {
+            "task_id": "r42-source-capture-replay-test-20260818",
+            "job_id": "job-source-capture-replay-test",
+            "repository": "aios/aios",
+            "base_oid": BASE,
+            "expected_branch_oid": BASE,
+            "target_branch": "agent/source-capture-replay-test",
+            "branch_mode": "CREATE_NEW",
+            "generation": 1,
+            "idempotency_key": "source-capture-replay-test-v1",
+        }
+    )
+
+
+def test_source_capture_replays_the_durable_envelope_to_bridge():
+    binding = _binding()
+    execution = project_execution_envelope(
+        binding,
+        {
+            "action": "source_capture",
+            "adapter": "compact_interface_map",
+            "capability_scope": ["git.read", "git.test"],
+            "source_signals": ["model provider registry version pinning"],
+        },
+    )
+    task = {
+        **binding.public(),
+        "required_capabilities": ["git.read", "git.test"],
+        "metadata": {"execution": execution},
+    }
+
+    class Bridge:
+        def __init__(self):
+            self.calls = []
+
+        def project_execution(self, _task, projection_input=None):
+            self.calls.append(projection_input)
+            return 200, dict(projection_input or {})
+
+    supervisor = object.__new__(WorkerSupervisor)
+    supervisor.bridge = Bridge()
+    resolved, source = supervisor.resolve_envelope(task)
+
+    assert source == "persisted_verified"
+    assert resolved.digest == execution["execution_digest"]
+    assert supervisor.bridge.calls == [execution]
+
+
+def test_deterministic_tasks_still_use_plan_registry_without_payload_replay():
+    binding = _binding()
+    execution = project_execution_envelope(
+        binding,
+        {
+            "action": "deterministic",
+            "adapter": "deterministic",
+            "capability_scope": [
+                "git.read",
+                "git.test",
+                "governed.write.branch",
+                "governed.write.file",
+            ],
+            "change_set": [
+                {
+                    "path": "active-runtime/app/example.py",
+                    "operation": "write_file",
+                    "content": "x",
+                }
+            ],
+        },
+    )
+    task = {
+        **binding.public(),
+        "required_capabilities": [
+            "git.read",
+            "git.test",
+            "governed.write.branch",
+            "governed.write.file",
+        ],
+        "metadata": {"execution": execution},
+    }
+
+    class Bridge:
+        def __init__(self):
+            self.calls = []
+
+        def project_execution(self, _task, projection_input=None):
+            self.calls.append(projection_input)
+            return 200, dict(execution)
+
+    supervisor = object.__new__(WorkerSupervisor)
+    supervisor.bridge = Bridge()
+    resolved, source = supervisor.resolve_envelope(task)
+
+    assert source == "persisted_verified"
+    assert resolved.digest == execution["execution_digest"]
+    assert supervisor.bridge.calls == [None]