fix(online-agent): normalize external runtime capabilities

active-runtime/app/aios_agent_task_execution_rebind.py

@@ -0,0 +1,722 @@
+"""Owner-controlled CAS-bound execution-envelope rebind for Online Agent Tasks.
+
+This service exists for a narrow recovery case: a Task was projected/claimed with a read-only
+source_capture envelope before its already-registered deterministic plan became authoritative.
+
+It never creates Tasks/Jobs/branches and never accepts arbitrary metadata patches. It only replaces
+metadata.execution + the derived execution_projection for the same governed binding, after an
+official NEEDS_REWORK transition has returned the Task to QUEUED with no active lease.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+from typing import Any, Callable, Mapping, Sequence
+import urllib.error
+import urllib.request
+
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    validate_execution_envelope,
+)
+
+TASK_RE = re.compile(r"^[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}$")
+SHA_RE = re.compile(r"^[0-9a-f]{64}$")
+MAX_BODY = 64 * 1024
+AUTHORIZATION_ID_ENV = "AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID"
+
+
+class ExecutionRebindError(RuntimeError):
+    def __init__(self, code: str, status: int = 409) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+def _canonical(value: object) -> bytes:
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode("utf-8")
+
+
+def _sha(value: object) -> str:
+    return hashlib.sha256(_canonical(value)).hexdigest()
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _require_string(
+    value: object,
+    code: str,
+    pattern: re.Pattern[str] | None = None,
+) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ExecutionRebindError(code, 400)
+    value = value.strip()
+    if pattern is not None and pattern.fullmatch(value) is None:
+        raise ExecutionRebindError(code, 400)
+    return value
+
+
+def _required_authorization_id() -> str:
+    value = os.environ.get(AUTHORIZATION_ID_ENV, "").strip()
+    if not value:
+        raise RuntimeError("TASK_EXECUTION_REBIND_AUTHORIZATION_ID_MISSING")
+    if KEY_RE.fullmatch(value) is None:
+        raise RuntimeError("TASK_EXECUTION_REBIND_AUTHORIZATION_ID_INVALID")
+    return value
+
+
+@dataclass(frozen=True)
+class ExecutionRebindRequest:
+    expected_status: str
+    expected_revision: int
+    expected_attempt: int
+    previous_verification_sha256: str
+    expected_old_execution_digest: str
+    expected_new_execution_digest: str
+    authorization_id: str
+    idempotency_key: str
+    reason: str
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "ExecutionRebindRequest":
+        allowed = {
+            "expected_status",
+            "expected_revision",
+            "expected_attempt",
+            "previous_verification_sha256",
+            "expected_old_execution_digest",
+            "expected_new_execution_digest",
+            "authorization_id",
+            "idempotency_key",
+            "reason",
+        }
+        if set(value) != allowed:
+            raise ExecutionRebindError("EXECUTION_REBIND_FIELDS_INVALID", 400)
+
+        status = _require_string(value["expected_status"], "EXPECTED_STATUS_INVALID")
+        if status != "QUEUED":
+            raise ExecutionRebindError("EXPECTED_STATUS_INVALID", 400)
+
+        revision = value["expected_revision"]
+        attempt = value["expected_attempt"]
+        if (
+            not isinstance(revision, int)
+            or isinstance(revision, bool)
+            or revision < 0
+            or not isinstance(attempt, int)
+            or isinstance(attempt, bool)
+            or attempt < 1
+        ):
+            raise ExecutionRebindError("EXECUTION_REBIND_INTEGER_INVALID", 400)
+
+        verification = _require_string(
+            value["previous_verification_sha256"], "VERIFICATION_SHA256_INVALID"
+        )
+        old_digest = _require_string(
+            value["expected_old_execution_digest"], "OLD_EXECUTION_DIGEST_INVALID"
+        )
+        new_digest = _require_string(
+            value["expected_new_execution_digest"], "NEW_EXECUTION_DIGEST_INVALID"
+        )
+        for digest, code in (
+            (verification, "VERIFICATION_SHA256_INVALID"),
+            (old_digest, "OLD_EXECUTION_DIGEST_INVALID"),
+            (new_digest, "NEW_EXECUTION_DIGEST_INVALID"),
+        ):
+            if SHA_RE.fullmatch(digest) is None:
+                raise ExecutionRebindError(code, 400)
+        if old_digest == new_digest:
+            raise ExecutionRebindError("EXECUTION_REBIND_DIGEST_UNCHANGED", 400)
+
+        authorization = _require_string(
+            value["authorization_id"], "AUTHORIZATION_ID_INVALID", KEY_RE
+        )
+        key = _require_string(value["idempotency_key"], "IDEMPOTENCY_KEY_INVALID", KEY_RE)
+        reason = _require_string(value["reason"], "EXECUTION_REBIND_REASON_INVALID")
+        if len(reason) > 512:
+            raise ExecutionRebindError("EXECUTION_REBIND_REASON_INVALID", 400)
+
+        return cls(
+            status,
+            revision,
+            attempt,
+            verification,
+            old_digest,
+            new_digest,
+            authorization,
+            key,
+            reason,
+        )
+
+    def public(self) -> dict[str, Any]:
+        return {
+            "expected_status": self.expected_status,
+            "expected_revision": self.expected_revision,
+            "expected_attempt": self.expected_attempt,
+            "previous_verification_sha256": self.previous_verification_sha256,
+            "expected_old_execution_digest": self.expected_old_execution_digest,
+            "expected_new_execution_digest": self.expected_new_execution_digest,
+            "authorization_id": self.authorization_id,
+            "idempotency_key": self.idempotency_key,
+            "reason": self.reason,
+        }
+
+
+class CanonicalBridgeExecutionResolver:
+    """Resolve the already-registered plan from the canonical governed Bridge."""
+
+    def __init__(self, base_url: str, control_token: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.control_token = control_token
+
+    def __call__(
+        self,
+        task_id: str,
+        required_capabilities: Sequence[str],
+    ) -> dict[str, Any]:
+        body = json.dumps(
+            {
+                "task_id": task_id,
+                "required_capabilities": list(required_capabilities),
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        request = urllib.request.Request(
+            self.base_url + "/v2/execution-envelope",
+            method="POST",
+            data=body,
+            headers={
+                "Authorization": "Bearer " + self.control_token,
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=15) as response:
+                raw = response.read()
+                status = response.status
+        except urllib.error.HTTPError as exc:
+            raw = exc.read()
+            status = exc.code
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ExecutionRebindError("CANONICAL_EXECUTION_RESPONSE_INVALID", 502) from exc
+        if not isinstance(value, Mapping):
+            raise ExecutionRebindError("CANONICAL_EXECUTION_RESPONSE_INVALID", 502)
+        if status != 200:
+            raise ExecutionRebindError(
+                str(value.get("error") or "CANONICAL_EXECUTION_RESOLUTION_FAILED"),
+                502 if status >= 500 else 409,
+            )
+        return dict(value)
+
+
+class ExecutionRebindStore:
+    """Official owner-controlled mutation with CAS, plan resolution and durable audit."""
+
+    def __init__(
+        self,
+        db_path: Path,
+        resolver: Callable[[str, Sequence[str]], Mapping[str, Any]],
+    ) -> None:
+        self.db_path = Path(db_path)
+        self.resolver = resolver
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.db_path, timeout=30.0)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA busy_timeout=30000")
+        return connection
+
+    @staticmethod
+    def _columns(
+        connection: sqlite3.Connection,
+        table: str,
+    ) -> dict[str, sqlite3.Row]:
+        return {
+            str(row["name"]): row
+            for row in connection.execute(f'PRAGMA table_info("{table}")')
+        }
+
+    @staticmethod
+    def _json_value(raw: object, code: str) -> dict[str, Any]:
+        if isinstance(raw, Mapping):
+            return dict(raw)
+        if not isinstance(raw, str):
+            raise ExecutionRebindError(code, 409)
+        try:
+            value = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ExecutionRebindError(code, 409) from exc
+        if not isinstance(value, Mapping):
+            raise ExecutionRebindError(code, 409)
+        return dict(value)
+
+    @staticmethod
+    def _verification_hash(raw: object) -> str:
+        if raw in (None, ""):
+            value: object = {}
+        elif isinstance(raw, str):
+            try:
+                value = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise ExecutionRebindError("TASK_VERIFICATION_INVALID", 409) from exc
+        else:
+            value = raw
+        return _sha(value)
+
+    @staticmethod
+    def _metadata_column(columns: Mapping[str, sqlite3.Row]) -> str:
+        for name in ("metadata_json", "metadata"):
+            if name in columns:
+                return name
+        raise ExecutionRebindError("TASK_METADATA_SCHEMA_UNSUPPORTED", 503)
+
+    @staticmethod
+    def _required_capabilities(
+        task: sqlite3.Row,
+        columns: Mapping[str, sqlite3.Row],
+        old_execution: Mapping[str, Any],
+    ) -> list[str]:
+        for field in ("required_capabilities_json", "required_capabilities"):
+            if field not in columns:
+                continue
+            raw = task[field]
+            if raw in (None, ""):
+                continue
+            if isinstance(raw, str):
+                try:
+                    value = json.loads(raw)
+                except json.JSONDecodeError:
+                    value = [item for item in raw.split(",") if item.strip()]
+            else:
+                value = raw
+            if isinstance(value, list) and all(isinstance(item, str) for item in value):
+                return [item.strip() for item in value if item.strip()]
+        value = old_execution.get("capability_scope")
+        if isinstance(value, list) and all(isinstance(item, str) for item in value):
+            return [item for item in value if item]
+        raise ExecutionRebindError("TASK_CAPABILITIES_UNAVAILABLE", 503)
+
+    @staticmethod
+    def _event_table(
+        connection: sqlite3.Connection,
+    ) -> tuple[str, dict[str, sqlite3.Row]]:
+        tables = {
+            row["name"]
+            for row in connection.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'"
+            )
+        }
+        for name in ("task_events", "events"):
+            if name not in tables:
+                continue
+            columns = ExecutionRebindStore._columns(connection, name)
+            if "task_id" in columns and ("event_type" in columns or "type" in columns):
+                return name, columns
+        raise ExecutionRebindError("TASK_EVENT_TABLE_UNSUPPORTED", 503)
+
+    @staticmethod
+    def _insert_event(
+        connection: sqlite3.Connection,
+        task_id: str,
+        payload: Mapping[str, Any],
+    ) -> None:
+        table, columns = ExecutionRebindStore._event_table(connection)
+        values: dict[str, Any] = {"task_id": task_id}
+        if "event_type" in columns:
+            values["event_type"] = "TASK_EXECUTION_REBIND_AUTHORIZED"
+        else:
+            values["type"] = "TASK_EXECUTION_REBIND_AUTHORIZED"
+        payload_text = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        if "payload_json" in columns:
+            values["payload_json"] = payload_text
+        elif "payload" in columns:
+            values["payload"] = payload_text
+        else:
+            raise ExecutionRebindError("TASK_EVENT_PAYLOAD_UNSUPPORTED", 503)
+        if "created_at" in columns:
+            values["created_at"] = _now()
+        if "agent_id" in columns:
+            values["agent_id"] = "owner-control"
+        if "actor_id" in columns:
+            values["actor_id"] = "owner-control"
+        names = list(values)
+        quoted = ",".join('"' + name + '"' for name in names)
+        placeholders = ",".join("?" for _ in names)
+        connection.execute(
+            f'INSERT INTO "{table}" ({quoted}) VALUES ({placeholders})',
+            tuple(values[name] for name in names),
+        )
+
+    @staticmethod
+    def _ensure_rebind_table(connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS task_execution_rebinds (
+              task_id TEXT NOT NULL,
+              idempotency_key TEXT NOT NULL,
+              request_sha256 TEXT NOT NULL,
+              authorization_id TEXT NOT NULL,
+              previous_verification_sha256 TEXT NOT NULL,
+              old_execution_digest TEXT NOT NULL,
+              new_execution_digest TEXT NOT NULL,
+              expected_revision INTEGER NOT NULL,
+              applied_revision INTEGER NOT NULL,
+              attempt INTEGER NOT NULL,
+              reason TEXT NOT NULL,
+              created_at TEXT NOT NULL,
+              PRIMARY KEY(task_id,idempotency_key)
+            )
+            """
+        )
+
+    def _load_and_validate(
+        self,
+        connection: sqlite3.Connection,
+        task_id: str,
+        request: ExecutionRebindRequest,
+    ) -> tuple[sqlite3.Row, dict[str, sqlite3.Row], str, dict[str, Any], dict[str, Any]]:
+        columns = self._columns(connection, "tasks")
+        required = {
+            "task_id",
+            "status",
+            "attempt",
+            "revision",
+            "claimed_by",
+            "lease_token_hash",
+            "lease_expires_at",
+            "verification_json",
+        }
+        if not required.issubset(columns):
+            raise ExecutionRebindError("TASK_SCHEMA_EXECUTION_REBIND_UNSUPPORTED", 503)
+        metadata_column = self._metadata_column(columns)
+        task = connection.execute(
+            "SELECT * FROM tasks WHERE task_id=?",
+            (task_id,),
+        ).fetchone()
+        if task is None:
+            raise ExecutionRebindError("TASK_NOT_FOUND", 404)
+
+        if (
+            str(task["status"]) != request.expected_status
+            or int(task["revision"]) != request.expected_revision
+            or int(task["attempt"]) != request.expected_attempt
+        ):
+            raise ExecutionRebindError("EXECUTION_REBIND_CAS_MISMATCH", 409)
+        if task["lease_token_hash"] not in (None, "") or task["lease_expires_at"] not in (None, ""):
+            raise ExecutionRebindError("EXECUTION_REBIND_ACTIVE_LEASE_DENY", 409)
+
+        verification_sha = self._verification_hash(task["verification_json"])
+        if verification_sha != request.previous_verification_sha256:
+            raise ExecutionRebindError("EXECUTION_REBIND_VERIFICATION_DRIFT", 409)
+
+        metadata = self._json_value(
+            task[metadata_column],
+            "TASK_METADATA_INVALID",
+        )
+        execution = metadata.get("execution")
+        if not isinstance(execution, Mapping):
+            raise ExecutionRebindError("TASK_EXECUTION_METADATA_MISSING", 409)
+        old_execution = validate_execution_envelope(dict(execution))
+        if old_execution["execution_digest"] != request.expected_old_execution_digest:
+            raise ExecutionRebindError("EXECUTION_REBIND_OLD_DIGEST_DRIFT", 409)
+        if old_execution.get("action") != "source_capture":
+            raise ExecutionRebindError("EXECUTION_REBIND_SOURCE_CAPTURE_REQUIRED", 409)
+
+        projection = metadata.get("execution_projection")
+        if isinstance(projection, Mapping):
+            projected_digest = str(projection.get("execution_digest") or "")
+            if projected_digest and projected_digest != request.expected_old_execution_digest:
+                raise ExecutionRebindError("EXECUTION_REBIND_PROJECTION_DRIFT", 409)
+        return task, columns, metadata_column, metadata, old_execution
+
+    def rebind(
+        self,
+        task_id: str,
+        request: ExecutionRebindRequest,
+    ) -> dict[str, Any]:
+        task_id = _require_string(task_id, "TASK_ID_INVALID", TASK_RE)
+        required_authorization = _required_authorization_id()
+        if request.authorization_id != required_authorization:
+            raise ExecutionRebindError("EXECUTION_REBIND_AUTHORIZATION_ID_MISMATCH", 403)
+
+        request_sha = _sha(request.public())
+
+        # Idempotent replay must win before CAS validation because a successful
+        # rebind necessarily increments the Task revision and changes the persisted envelope.
+        with self._connect() as connection:
+            self._ensure_rebind_table(connection)
+            replay = connection.execute(
+                "SELECT * FROM task_execution_rebinds WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise ExecutionRebindError("EXECUTION_REBIND_IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {
+                    "task_id": task_id,
+                    "status": "QUEUED",
+                    "attempt": int(replay["attempt"]),
+                    "revision": int(replay["applied_revision"]),
+                    "old_execution_digest": replay["old_execution_digest"],
+                    "new_execution_digest": replay["new_execution_digest"],
+                    "authorization_id": replay["authorization_id"],
+                    "idempotent": True,
+                }
+
+        # Read/CAS snapshot first. Canonical Bridge resolution happens outside the DB write lock.
+        with self._connect() as connection:
+            task, columns, _, _, old_execution = self._load_and_validate(
+                connection, task_id, request
+            )
+            capabilities = self._required_capabilities(task, columns, old_execution)
+
+        binding = GovernedExecutionBinding.from_mapping(old_execution)
+        resolved = dict(self.resolver(task_id, capabilities))
+        new_execution = validate_execution_envelope(resolved, binding=binding)
+        if new_execution.get("action") != "deterministic":
+            raise ExecutionRebindError("EXECUTION_REBIND_DETERMINISTIC_PLAN_REQUIRED", 409)
+        if new_execution["execution_digest"] != request.expected_new_execution_digest:
+            raise ExecutionRebindError("EXECUTION_REBIND_NEW_DIGEST_DRIFT", 409)
+        if new_execution["execution_digest"] == old_execution["execution_digest"]:
+            raise ExecutionRebindError("EXECUTION_REBIND_DIGEST_UNCHANGED", 409)
+
+        # Re-read under BEGIN IMMEDIATE and apply only if the exact snapshot is unchanged.
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._ensure_rebind_table(connection)
+
+            replay = connection.execute(
+                "SELECT * FROM task_execution_rebinds WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise ExecutionRebindError("EXECUTION_REBIND_IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {
+                    "task_id": task_id,
+                    "status": "QUEUED",
+                    "attempt": int(replay["attempt"]),
+                    "revision": int(replay["applied_revision"]),
+                    "old_execution_digest": replay["old_execution_digest"],
+                    "new_execution_digest": replay["new_execution_digest"],
+                    "authorization_id": replay["authorization_id"],
+                    "idempotent": True,
+                }
+
+            task, columns, metadata_column, metadata, current_old = self._load_and_validate(
+                connection, task_id, request
+            )
+            if current_old["execution_digest"] != old_execution["execution_digest"]:
+                raise ExecutionRebindError("EXECUTION_REBIND_OLD_DIGEST_DRIFT", 409)
+
+            new_metadata = dict(metadata)
+            new_metadata["execution"] = new_execution
+            new_metadata["execution_projection"] = {
+                "schema": "aios.online-agent.task-projection.v1",
+                "task_id": binding.task_id,
+                "job_id": binding.job_id,
+                "execution_digest": new_execution["execution_digest"],
+            }
+            applied_revision = request.expected_revision + 1
+            previous_claimed_by = task["claimed_by"]
+
+            cursor = connection.execute(
+                f"""
+                UPDATE tasks
+                   SET "{metadata_column}"=?, revision=?, claimed_by=NULL,
+                       lease_token_hash=NULL, lease_expires_at=NULL
+                 WHERE task_id=? AND status=? AND revision=? AND attempt=?
+                   AND claimed_by IS ? AND lease_token_hash IS NULL AND lease_expires_at IS NULL
+                """,
+                (
+                    json.dumps(new_metadata, sort_keys=True, separators=(",", ":")),
+                    applied_revision,
+                    task_id,
+                    request.expected_status,
+                    request.expected_revision,
+                    request.expected_attempt,
+                    previous_claimed_by,
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise ExecutionRebindError("EXECUTION_REBIND_CAS_MISMATCH", 409)
+
+            event_payload = {
+                "schema": "aios.task-execution-rebind.event.v1",
+                "task_id": task_id,
+                "job_id": binding.job_id,
+                "authorization_id": request.authorization_id,
+                "idempotency_key": request.idempotency_key,
+                "reason": request.reason,
+                "previous_verification_sha256": request.previous_verification_sha256,
+                "old_execution_digest": old_execution["execution_digest"],
+                "new_execution_digest": new_execution["execution_digest"],
+                "old_action": old_execution.get("action"),
+                "new_action": new_execution.get("action"),
+                "previous_revision": request.expected_revision,
+                "applied_revision": applied_revision,
+                "attempt": request.expected_attempt,
+                "request_sha256": request_sha,
+                "previous_claimed_by": previous_claimed_by,
+                "stale_claim_normalized": previous_claimed_by not in (None, ""),
+            }
+            self._insert_event(connection, task_id, event_payload)
+            connection.execute(
+                """
+                INSERT INTO task_execution_rebinds(
+                  task_id,idempotency_key,request_sha256,authorization_id,
+                  previous_verification_sha256,old_execution_digest,new_execution_digest,
+                  expected_revision,applied_revision,attempt,reason,created_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    task_id,
+                    request.idempotency_key,
+                    request_sha,
+                    request.authorization_id,
+                    request.previous_verification_sha256,
+                    old_execution["execution_digest"],
+                    new_execution["execution_digest"],
+                    request.expected_revision,
+                    applied_revision,
+                    request.expected_attempt,
+                    request.reason,
+                    _now(),
+                ),
+            )
+            connection.commit()
+            return {
+                "task_id": task_id,
+                "status": "QUEUED",
+                "attempt": request.expected_attempt,
+                "revision": applied_revision,
+                "old_execution_digest": old_execution["execution_digest"],
+                "new_execution_digest": new_execution["execution_digest"],
+                "authorization_id": request.authorization_id,
+                "idempotent": False,
+            }
+
+
+class ExecutionRebindHandler(BaseHTTPRequestHandler):
+    store: ExecutionRebindStore | None = None
+    control_token: str = ""
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def _json(self, value: Mapping[str, Any], status: int) -> None:
+        data = json.dumps(
+            dict(value), sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def _authorized(self) -> bool:
+        header = self.headers.get("Authorization", "")
+        return (
+            header.startswith("Bearer ")
+            and bool(self.control_token)
+            and hmac.compare_digest(header[7:], self.control_token)
+        )
+
+    def do_GET(self) -> None:
+        if self.path == "/health":
+            self._json(
+                {"status": "ok", "service": "aios-agent-task-execution-rebind"},
+                200,
+            )
+        else:
+            self._json({"error": "NOT_FOUND"}, 404)
+
+    def do_POST(self) -> None:
+        match = re.fullmatch(r"/v1/tasks/([^/]+)/execution-rebind", self.path)
+        if match is None:
+            self._json({"error": "NOT_FOUND"}, 404)
+            return
+        if not self._authorized():
+            self._json({"error": "UNAUTHORIZED"}, 401)
+            return
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            self._json({"error": "CONTENT_LENGTH_INVALID"}, 400)
+            return
+        if length < 1 or length > MAX_BODY:
+            self._json({"error": "REQUEST_SIZE_INVALID"}, 413)
+            return
+        try:
+            value = json.loads(self.rfile.read(length).decode("utf-8"))
+            if not isinstance(value, Mapping):
+                raise ValueError
+            request = ExecutionRebindRequest.from_mapping(value)
+            if self.store is None:
+                raise ExecutionRebindError("RUNTIME_UNAVAILABLE", 503)
+            result = self.store.rebind(match.group(1), request)
+            self._json(result, 200 if result["idempotent"] else 201)
+        except ExecutionRebindError as exc:
+            self._json({"error": exc.code}, exc.status)
+        except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
+            self._json({"error": "REQUEST_INVALID"}, 400)
+
+
+def load_control_token(path: Path) -> str:
+    if path.is_symlink() or not path.is_file():
+        raise RuntimeError("TASK_EXECUTION_REBIND_CONTROL_CREDENTIAL_UNAVAILABLE")
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 32:
+        raise RuntimeError("TASK_EXECUTION_REBIND_CONTROL_CREDENTIAL_INVALID")
+    return value
+
+
+def serve() -> None:
+    db = Path(os.environ.get("AIOS_AGENT_TASK_DB", "/var/lib/aios-agent-task/state.sqlite3"))
+    _required_authorization_id()
+    token_path = Path(
+        os.environ.get(
+            "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+            str(
+                Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+                / "agent_task_control_token"
+            ),
+        )
+    )
+    token = load_control_token(token_path)
+    bridge = os.environ.get(
+        "AIOS_GOVERNED_WRITE_API_URL",
+        "http://127.0.0.1:18892",
+    )
+    host = os.environ.get("AIOS_TASK_EXECUTION_REBIND_HOST", "127.0.0.1")
+    port = int(os.environ.get("AIOS_TASK_EXECUTION_REBIND_PORT", "18942"))
+    ExecutionRebindHandler.store = ExecutionRebindStore(
+        db,
+        CanonicalBridgeExecutionResolver(bridge, token),
+    )
+    ExecutionRebindHandler.control_token = token
+    ThreadingHTTPServer((host, port), ExecutionRebindHandler).serve_forever()
+
+
+if __name__ == "__main__":
+    serve()

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_compact_map.py

@@ -42,7 +42,7 @@ DEFAULT_TERMS = {
     "backoff", "circuit", "health", "liveness", "scope", "tenant", "project", "trace",
     "message", "checkpoint", "recovery", "idempotency", "dispatch", "assignment",
 }
-CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
+FLEET_CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
     "FLEET_AGENT_REGISTRATION": ("agent", "register", "worker", "capability", "version"),
     "FLEET_CAPACITY_SCHEDULING": ("capacity", "slot", "concurrency", "scheduler", "queue"),
     "FLEET_FAIRNESS": ("fair", "priority", "queue", "tenant", "project", "starvation"),
@@ -52,6 +52,30 @@ CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
     "FLEET_IDEMPOTENT_DISPATCH": ("idempot", "dispatch", "claim", "lease", "result", "attempt"),
     "FLEET_FAILURE_RECOVERY": ("checkpoint", "resume", "recover", "retry", "cleanup", "receipt"),
 }
+
+MODEL_GOVERNANCE_CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
+    "MODEL_PROVIDER_REGISTRY": ("model", "provider", "registry", "endpoint", "capability"),
+    "MODEL_SELECTION_POLICY": ("model", "select", "route", "policy", "capability"),
+    "MODEL_VERSION_PINNING": ("model", "version", "pin", "revision", "snapshot"),
+    "MODEL_BUDGET_ACCOUNTING": ("budget", "token", "cost", "quota", "usage"),
+    "MODEL_FALLBACK_POLICY": ("fallback", "retry", "route", "provider", "model"),
+    "MODEL_SCOPE_ISOLATION": ("tenant", "project", "scope", "policy", "model"),
+    "MODEL_CREDENTIAL_BOUNDARY": ("credential", "secret", "provider", "tool", "token"),
+    "MODEL_TRACE_AUDIT": ("trace", "audit", "model", "provider", "request"),
+    "MODEL_EVALUATION_GATE": ("eval", "evaluation", "approval", "quality", "gate"),
+    "MODEL_ROUTING_RESILIENCE": ("route", "retry", "fallback", "timeout", "health"),
+}
+
+PROFILE_CLASSIFICATION = {
+    "fleet": "P2_INTERFACE_MAP_CAPTURED",
+    "model_governance": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
+}
+
+MODEL_PROFILE_MARKERS = {
+    "model", "provider", "llm", "inference", "routing", "router", "budget",
+    "token", "cost", "quota", "fallback", "evaluation", "eval", "approval",
+    "version", "pinning",
+}
 SENSITIVE_RE = re.compile(
     r"(?i)(authorization\s*:\s*bearer\s+\S+|"
     r"(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*['\"]?[^'\"\s,;]+)"
@@ -92,6 +116,27 @@ def _signal_terms(signals: Sequence[str]) -> set[str]:
     return set(sorted(terms)[:96])
 
 
+def _profile(source_signals: Sequence[str]) -> str:
+    tokens: set[str] = set()
+    for signal in source_signals:
+        if not isinstance(signal, str):
+            continue
+        tokens.update(
+            token.lower()
+            for token in re.findall(r"[A-Za-z][A-Za-z0-9_.-]{2,63}", signal)
+        )
+    model_score = len(tokens & MODEL_PROFILE_MARKERS)
+    return "model_governance" if model_score >= 2 else "fleet"
+
+
+def _contract_terms(profile: str) -> Mapping[str, tuple[str, ...]]:
+    if profile == "model_governance":
+        return MODEL_GOVERNANCE_CONTRACT_TERMS
+    if profile == "fleet":
+        return FLEET_CONTRACT_TERMS
+    raise CompactInterfaceMapError("INTERFACE_MAP_PROFILE_INVALID")
+
+
 def _candidate_files(root: Path, terms: set[str]) -> list[dict[str, Any]]:
     root = root.resolve()
     candidates: list[dict[str, Any]] = []
@@ -190,10 +235,11 @@ def _surface_ref(symbol: Mapping[str, Any]) -> str:
 def _contract_map(
     symbols: Sequence[Mapping[str, Any]],
     modules: Sequence[Mapping[str, Any]],
+    contract_terms: Mapping[str, tuple[str, ...]],
 ) -> dict[str, Any]:
     result: dict[str, Any] = {}
     module_paths = [str(item["path"]) for item in modules]
-    for contract, terms in CONTRACT_TERMS.items():
+    for contract, terms in contract_terms.items():
         scored: list[tuple[int, str]] = []
         for symbol in symbols:
             haystack = _surface_ref(symbol).lower()
@@ -216,37 +262,67 @@ def _contract_map(
     return result
 
 
-def _gap_evidence(symbols: Sequence[Mapping[str, Any]]) -> tuple[list[str], list[str], list[str]]:
+def _gap_evidence(
+    symbols: Sequence[Mapping[str, Any]],
+    profile: str,
+) -> tuple[list[str], list[str], list[str]]:
     haystack = "\n".join(_surface_ref(item).lower() for item in symbols)
 
-    invariants = [
-        "Online Task claim/heartbeat remains the authoritative worker ownership boundary.",
-        "Governed mutation remains separated from read-only analysis and requires WriterLease.",
-        "P1 attempt-aware result idempotency and durable checkpoint recovery are reusable fleet primitives.",
-        "Scope/trace/message lineage must remain attached when work is scheduled or reassigned.",
-    ]
-    failure_windows: list[str] = []
-    if "scheduler" not in haystack:
-        failure_windows.append("No first-class durable fleet scheduler symbol detected; worker-local polling may still be the scheduling mechanism.")
-    if "capacity" not in haystack and "slot" not in haystack:
-        failure_windows.append("No explicit durable worker-capacity/slot primitive detected.")
-    if "fair" not in haystack and "starvation" not in haystack:
-        failure_windows.append("No explicit fairness/starvation policy symbol detected.")
-    if "circuit" not in haystack:
-        failure_windows.append("No explicit circuit-breaker primitive detected for repeated worker/task failures.")
-    if "reassign" not in haystack:
-        failure_windows.append("No explicit worker-reassignment primitive detected beyond lease expiry/retry semantics.")
+    if profile == "model_governance":
+        invariants = [
+            "Online Task claim/heartbeat remains the authoritative worker ownership boundary.",
+            "Provider/model credentials remain outside uncredentialed planner/executor processes.",
+            "P0-A tenant/project scope and P0-C trace/audit lineage must constrain model routing.",
+            "P1 retry/checkpoint semantics remain authoritative for model-call execution recovery.",
+        ]
+        failure_windows: list[str] = []
+        if "provider" not in haystack and "model" not in haystack:
+            failure_windows.append("No first-class model/provider registry surface detected.")
+        if "budget" not in haystack and "cost" not in haystack and "quota" not in haystack:
+            failure_windows.append("No explicit durable model budget/cost/quota accounting surface detected.")
+        if "fallback" not in haystack and "route" not in haystack:
+            failure_windows.append("No explicit model fallback/routing policy surface detected.")
+        if "eval" not in haystack and "approval" not in haystack:
+            failure_windows.append("No explicit evaluation/approval gate surface detected for model changes.")
+        if "version" not in haystack and "pin" not in haystack:
+            failure_windows.append("No explicit model-version pinning surface detected.")
+        desired = (
+            ("trace", "P0-C trace/audit lineage"),
+            ("scope", "P0-A tenant/project scope isolation"),
+            ("credential", "P0-C credential governance"),
+            ("checkpoint", "P1 durable execution recovery"),
+            ("message", "P0-B MessageEnvelope/handoff"),
+            ("retry", "P1 retry/idempotency"),
+        )
+    else:
+        invariants = [
+            "Online Task claim/heartbeat remains the authoritative worker ownership boundary.",
+            "Governed mutation remains separated from read-only analysis and requires WriterLease.",
+            "P1 attempt-aware result idempotency and durable checkpoint recovery are reusable fleet primitives.",
+            "Scope/trace/message lineage must remain attached when work is scheduled or reassigned.",
+        ]
+        failure_windows = []
+        if "scheduler" not in haystack:
+            failure_windows.append("No first-class durable fleet scheduler symbol detected; worker-local polling may still be the scheduling mechanism.")
+        if "capacity" not in haystack and "slot" not in haystack:
+            failure_windows.append("No explicit durable worker-capacity/slot primitive detected.")
+        if "fair" not in haystack and "starvation" not in haystack:
+            failure_windows.append("No explicit fairness/starvation policy symbol detected.")
+        if "circuit" not in haystack:
+            failure_windows.append("No explicit circuit-breaker primitive detected for repeated worker/task failures.")
+        if "reassign" not in haystack:
+            failure_windows.append("No explicit worker-reassignment primitive detected beyond lease expiry/retry semantics.")
+        desired = (
+            ("checkpoint", "P1 durable checkpoint/resume"),
+            ("message", "P0-B MessageEnvelope/handoff"),
+            ("trace", "P0-C trace correlation"),
+            ("scope", "P0-A scope isolation"),
+            ("lease", "claim/WriterLease ownership"),
+            ("heartbeat", "worker liveness"),
+            ("idempot", "retry-safe dispatch/result"),
+        )
 
     reuse_points: list[str] = []
-    desired = (
-        ("checkpoint", "P1 durable checkpoint/resume"),
-        ("message", "P0-B MessageEnvelope/handoff"),
-        ("trace", "P0-C trace correlation"),
-        ("scope", "P0-A scope isolation"),
-        ("lease", "claim/WriterLease ownership"),
-        ("heartbeat", "worker liveness"),
-        ("idempot", "retry-safe dispatch/result"),
-    )
     for token, label in desired:
         match = next((item for item in symbols if token in _surface_ref(item).lower()), None)
         if match is not None:
@@ -287,6 +363,7 @@ def build_compact_interface_map(
     if re.fullmatch(r"[0-9a-f]{40}", source_commit) is None:
         raise CompactInterfaceMapError("INTERFACE_MAP_SOURCE_COMMIT_INVALID")
 
+    profile = _profile(source_signals)
     terms = _signal_terms(source_signals)
     candidates = _candidate_files(root, terms)
     if not candidates:
@@ -363,9 +440,11 @@ def build_compact_interface_map(
             }
         )
 
-    invariants, failure_windows, reuse_points = _gap_evidence(all_symbols)
+    invariants, failure_windows, reuse_points = _gap_evidence(all_symbols, profile)
     result = {
         "schema": SCHEMA,
+        "profile": profile,
+        "classification": PROFILE_CLASSIFICATION[profile],
         "source_commit": source_commit,
         "implementation_map": {
             "modules": modules,
@@ -379,6 +458,6 @@ def build_compact_interface_map(
             "reuse_points": reuse_points,
         },
         "bounded_excerpts": excerpts,
-        "contracts": _contract_map(all_symbols, modules),
+        "contracts": _contract_map(all_symbols, modules, _contract_terms(profile)),
     }
     return _trim_to_budget(result)

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]:
-        raise WorkerError(
-            "HERMES_ENGINEERING_ADAPTER_UNAVAILABLE",
-            phase="adapter",
-            details={"reason": "The live Hermes service is a user-chat runtime, not an engineering executor"},
+        if envelope.action != "agent_runtime":
+            raise WorkerError(
+                "EXTERNAL_AGENT_ACTION_INVALID",
+                phase="adapter",
+                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]()
 
 
@@ -1526,9 +1611,18 @@ def build_compact_interface_map_result(
     contracts = interface_map.get("contracts")
     if not isinstance(implementation_map, Mapping) or not isinstance(excerpts, list) or not isinstance(contracts, Mapping):
         raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
+    profile = str(interface_map.get("profile") or "fleet")
+    classification = str(interface_map.get("classification") or "")
+    allowed_classifications = {
+        "fleet": "P2_INTERFACE_MAP_CAPTURED",
+        "model_governance": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
+    }
+    if allowed_classifications.get(profile) != classification:
+        raise WorkerError("COMPACT_INTERFACE_MAP_PROFILE_BINDING_INVALID", phase="result")
     result = {
         "status": "SUCCEEDED",
-        "classification": "P2_INTERFACE_MAP_CAPTURED",
+        "classification": classification,
+        "interface_map_profile": profile,
         "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
         "execution_mode": "compact_interface_map",
         "execution_backend": "generic-online-agent-worker-v2",
@@ -1566,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,
@@ -1615,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)
@@ -1815,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"
@@ -1895,7 +2091,11 @@ class WorkerSupervisor:
                 # re-run deterministic verification.  No external side effect exists yet.
                 run_root = new_run_root()
                 try:
-                    worktree = prepare_worktree(run_root, envelope)
+                    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(
                         {
                             "phase": "workspace_prepared",
@@ -1929,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(
@@ -2096,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:

active-runtime/app/task_plane/fleet.py

@@ -363,7 +363,7 @@ class FleetStateStore:
             row = self.conn.execute(
                 """
                 SELECT COUNT(*) FROM fleet_dispatches
-                 WHERE worker_id=? AND state IN ('RESERVED','CLAIMED') AND reserved_until>?
+                 WHERE worker_id=? AND state='RESERVED' AND reserved_until>?
                 """,
                 (worker_id, now),
             ).fetchone()
@@ -412,6 +412,13 @@ class FleetStateStore:
                 """,
                 (task_id,),
             ).fetchone()
+            # Capacity is rechecked inside the same durable reservation transaction.
+            # This closes the cross-thread/process race between eligibility scoring and reserve.
+            policy_row = self.conn.execute(
+                "SELECT max_slots FROM fleet_worker_capacity WHERE worker_id=?",
+                (worker_id,),
+            ).fetchone()
+            max_slots = int(policy_row[0]) if policy_row else 1
             if existing and str(existing[6]) in {"RESERVED", "CLAIMED"} and float(existing[5]) > now:
                 if str(existing[1]) != worker_id or int(existing[2]) != task_revision:
                     raise FleetConflict("FLEET_TASK_ALREADY_RESERVED")
@@ -419,6 +426,32 @@ class FleetStateStore:
                     str(existing[0]), task_id, worker_id, task_revision, str(existing[3]),
                     str(existing[4]), float(existing[5]), score
                 )
+
+            claim_columns = {
+                str(row[1])
+                for row in self.conn.execute("PRAGMA table_info(claims)").fetchall()
+            }
+            active_claims = 0
+            if {"worker_id", "status"}.issubset(claim_columns):
+                sql = "SELECT COUNT(*) FROM claims WHERE worker_id=? AND status='ACTIVE'"
+                params: list[Any] = [worker_id]
+                if "lease_expires_at" in claim_columns:
+                    sql += " AND lease_expires_at>?"
+                    params.append(now)
+                row = self.conn.execute(sql, tuple(params)).fetchone()
+                active_claims = int(row[0] if row else 0)
+            row = self.conn.execute(
+                """
+                SELECT COUNT(*) FROM fleet_dispatches
+                 WHERE worker_id=? AND task_id<>?
+                   AND state='RESERVED' AND reserved_until>?
+                """,
+                (worker_id, task_id, now),
+            ).fetchone()
+            active_reservations = int(row[0] if row else 0)
+            if active_claims + active_reservations >= max_slots:
+                raise FleetConflict("FLEET_WORKER_CAPACITY_EXHAUSTED")
+
             self.conn.execute(
                 """
                 INSERT INTO fleet_dispatches(
@@ -682,19 +715,23 @@ class FleetScheduler:
         return getter(worker_id)
 
     def _active_claims(self, worker_id: str, now: float) -> int:
-        columns = {
-            str(row[1])
-            for row in self.store.conn.execute("PRAGMA table_info(claims)").fetchall()
-        }
-        required = {"worker_id", "status"}
-        if not required.issubset(columns):
-            raise FleetError("FLEET_CLAIMS_SCHEMA_UNSUPPORTED")
-        sql = "SELECT COUNT(*) FROM claims WHERE worker_id=? AND status='ACTIVE'"
-        params: list[Any] = [worker_id]
-        if "lease_expires_at" in columns:
-            sql += " AND lease_expires_at>?"
-            params.append(now)
-        row = self.store.conn.execute(sql, tuple(params)).fetchone()
+        # The TaskPlane store owns one SQLite connection in the current runtime.
+        # Fleet reads must share its lock just like fleet writes do, otherwise two
+        # concurrent pullers can issue overlapping operations on the same handle.
+        with self.state.lock:
+            columns = {
+                str(row[1])
+                for row in self.store.conn.execute("PRAGMA table_info(claims)").fetchall()
+            }
+            required = {"worker_id", "status"}
+            if not required.issubset(columns):
+                raise FleetError("FLEET_CLAIMS_SCHEMA_UNSUPPORTED")
+            sql = "SELECT COUNT(*) FROM claims WHERE worker_id=? AND status='ACTIVE'"
+            params: list[Any] = [worker_id]
+            if "lease_expires_at" in columns:
+                sql += " AND lease_expires_at>?"
+                params.append(now)
+            row = self.store.conn.execute(sql, tuple(params)).fetchone()
         return int(row[0] if row else 0)
 
     @staticmethod
@@ -764,6 +801,52 @@ class FleetScheduler:
             and _task_id(task)
         ]
 
+    def rank_for_worker(
+        self,
+        worker_id: str,
+        *,
+        actor_context: Any = None,
+        now: float | None = None,
+    ) -> list[object]:
+        """Return fleet-ranked tasks without reserving or claiming anything."""
+        now = float(self.now_fn() if now is None else now)
+        worker_id = _text(worker_id)
+        if not worker_id:
+            return []
+        ranked: list[object] = []
+        for task in self._tasks(actor_context):
+            eligible, _, _ = self._eligible_worker(worker_id, task, now)
+            if eligible:
+                ranked.append(task)
+        return sorted(ranked, key=lambda task: self._task_score(task, now))
+
+    def reserve_for_worker(
+        self,
+        worker_id: str,
+        task: object,
+        *,
+        now: float | None = None,
+    ) -> FleetDispatchDecision:
+        """Reserve exactly one already-ranked task, rechecking eligibility and capacity."""
+        now = float(self.now_fn() if now is None else now)
+        worker_id = _text(worker_id)
+        eligible, _, policy = self._eligible_worker(worker_id, task, now)
+        if not eligible:
+            raise FleetConflict("FLEET_WORKER_NOT_ELIGIBLE")
+        metadata = _task_metadata(task)
+        scope_weight = _float(metadata.get("fleet_scope_weight"), 1.0, 0.05, 100.0)
+        return self.state.reserve(
+            task_id=_task_id(task),
+            worker_id=worker_id,
+            task_revision=_task_revision(task),
+            scope_key=_task_scope_key(task),
+            failure_domain=policy.failure_domain,
+            score=self._task_score(task, now),
+            reservation_seconds=self.reservation_seconds,
+            now=now,
+            scope_weight=scope_weight,
+        )
+
     def select(
         self,
         worker_ids: Sequence[str],

active-runtime/app/task_plane/scheduler.py

@@ -84,3 +84,193 @@ from .fleet import (
     FleetStateStore,
     FleetWorkerPolicy,
 )
+# R4.2 P2-B fleet-aware pull scheduling.
+#
+# The original TaskScheduler remains the security oracle and claim-CAS implementation.
+# Fleet ordering only controls the order presented to that oracle.
+import inspect as _fleet_inspect
+import threading as _fleet_threading
+from collections.abc import Mapping as _FleetMapping
+
+from .worker_registry import WorkerRegistry as _FleetWorkerRegistry
+from .errors import NotFoundError as _FleetNotFoundError
+
+_LegacyTaskScheduler = TaskScheduler
+_LegacyScheduleParameters = _fleet_inspect.signature(
+    _LegacyTaskScheduler.schedule_for_worker
+).parameters
+_LegacyNextParameters = _fleet_inspect.signature(
+    _LegacyTaskScheduler.next_queued_task
+).parameters
+
+
+class _FleetOrderedStoreProxy:
+    def __init__(self, delegate, tasks):
+        self._delegate = delegate
+        self._tasks = tuple(tasks)
+
+    def list_tasks(self, *args, **kwargs):
+        return list(self._tasks)
+
+    def __getattr__(self, name):
+        return getattr(self._delegate, name)
+
+
+def _fleet_claim_field(claim, name, default=None):
+    if isinstance(claim, _FleetMapping):
+        return claim.get(name, default)
+    return getattr(claim, name, default)
+
+
+class TaskScheduler(_LegacyTaskScheduler):
+    """Fleet-integrated pull scheduler preserving legacy security and claim semantics."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        store = getattr(self, "store", args[0] if args else None)
+        if store is None:
+            raise FleetError("FLEET_SCHEDULER_STORE_REQUIRED")
+        registry = getattr(self, "workers", None)
+        if registry is None:
+            registry = _FleetWorkerRegistry(store)
+        self.fleet_scheduler = FleetScheduler(store, registry)
+        self._fleet_local = _fleet_threading.local()
+
+    def _fleet_legacy_next(self, worker_id, tasks, *args, **kwargs):
+        proxy = _FleetOrderedStoreProxy(self.store, tasks)
+        legacy = _LegacyTaskScheduler(proxy)
+        legacy_kwargs = dict(kwargs)
+        if "actor_context" not in _LegacyNextParameters:
+            legacy_kwargs.pop("actor_context", None)
+        return legacy.next_queued_task(worker_id, *args, **legacy_kwargs)
+
+    def next_queued_task(self, worker_id, *args, **kwargs):
+        actor_context = kwargs.get(
+            "actor_context", getattr(self._fleet_local, "actor_context", None)
+        )
+
+        # Preserve legacy fail-closed worker checks before fleet ordering.
+        probe_kwargs = dict(kwargs)
+        if "actor_context" not in _LegacyNextParameters:
+            probe_kwargs.pop("actor_context", None)
+        legacy_probe = _LegacyTaskScheduler(self.store).next_queued_task(
+            worker_id, *args, **probe_kwargs
+        )
+        if legacy_probe is None:
+            return None
+
+        ranked = self.fleet_scheduler.rank_for_worker(
+            worker_id, actor_context=actor_context
+        )
+        # Preserve FleetScheduler fairness order. The legacy scheduler is used
+        # only as a fail-closed security oracle: present one fleet-ranked task
+        # at a time so its internal sort cannot reorder across scopes.
+        # Keep the legacy priority dimension ahead of Fleet fairness for
+        # compatibility with existing TaskPlane scheduling contracts; the
+        # Fleet order remains authoritative among equal-priority tasks.
+        def _legacy_priority(task):
+            plan = getattr(task, "task_plan", task)
+            metadata = getattr(plan, "metadata", {}) or {}
+            if not isinstance(metadata, _FleetMapping):
+                return 0
+            try:
+                return int(metadata.get("priority", 0))
+            except (TypeError, ValueError):
+                return 0
+
+        ranked = sorted(ranked, key=lambda task: -_legacy_priority(task))
+        for ranked_candidate in ranked:
+            try:
+                candidate = self._fleet_legacy_next(
+                    worker_id, [ranked_candidate], *args, **kwargs
+                )
+            except _FleetNotFoundError:
+                continue
+            if candidate is None:
+                continue
+
+            ranked_id = getattr(
+                ranked_candidate,
+                "task_id",
+                getattr(getattr(ranked_candidate, "task_plan", None), "task_id", None),
+            )
+            candidate_id = getattr(
+                candidate,
+                "task_id",
+                getattr(getattr(candidate, "task_plan", None), "task_id", None),
+            )
+            if candidate_id != ranked_id:
+                raise FleetConflict("FLEET_LEGACY_ORACLE_ORDER_VIOLATION")
+
+            try:
+                decision = self.fleet_scheduler.reserve_for_worker(
+                    worker_id, candidate
+                )
+            except FleetConflict:
+                continue
+            self._fleet_local.pending = decision
+            return candidate
+        return None
+
+    def schedule_for_worker(self, worker_id, *args, **kwargs):
+        actor_context = kwargs.get("actor_context")
+        legacy_kwargs = dict(kwargs)
+        if "actor_context" not in _LegacyScheduleParameters:
+            legacy_kwargs.pop("actor_context", None)
+        self._fleet_local.actor_context = actor_context
+        self._fleet_local.pending = None
+        try:
+            claim = _LegacyTaskScheduler.schedule_for_worker(
+                self, worker_id, *args, **legacy_kwargs
+            )
+            decision = getattr(self._fleet_local, "pending", None)
+            if claim is None:
+                if decision is not None:
+                    self.fleet_scheduler.state.release(
+                        decision.task_id, state="RELEASED"
+                    )
+                return None
+            if decision is not None:
+                claim_id = _fleet_claim_field(claim, "claim_id")
+                claim_task_id = _fleet_claim_field(claim, "task_id", decision.task_id)
+                if not claim_id:
+                    self.fleet_scheduler.state.release(
+                        decision.task_id, state="RELEASED"
+                    )
+                    raise FleetError("FLEET_CLAIM_ID_REQUIRED")
+                if claim_task_id != decision.task_id:
+                    self.fleet_scheduler.state.release(
+                        decision.task_id, state="RELEASED"
+                    )
+                    raise FleetConflict("FLEET_CLAIM_TASK_MISMATCH")
+                self.fleet_scheduler.state.bind_claim(
+                    decision.reservation_id, str(claim_id)
+                )
+            return claim
+        except Exception:
+            decision = getattr(self._fleet_local, "pending", None)
+            if decision is not None:
+                self.fleet_scheduler.state.release(
+                    decision.task_id, state="RELEASED"
+                )
+            raise
+        finally:
+            self._fleet_local.pending = None
+            self._fleet_local.actor_context = None
+
+    def record_worker_failure(self, worker_id, failure_class, **kwargs):
+        return self.fleet_scheduler.state.record_worker_failure(
+            worker_id, failure_class, **kwargs
+        )
+
+    def record_worker_success(self, worker_id, **kwargs):
+        return self.fleet_scheduler.state.record_worker_success(
+            worker_id, **kwargs
+        )
+
+    def quarantine_task_worker(
+        self, task_id, worker_id, failure_class, **kwargs
+    ):
+        return self.fleet_scheduler.state.record_task_worker_failure(
+            task_id, worker_id, failure_class, **kwargs
+        )

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-agent-task-execution-rebind.service

@@ -0,0 +1,37 @@
+[Unit]
+Description=AIOS owner-controlled Task execution-rebind API
+After=aios-agent-task-api.service
+Wants=aios-agent-task-api.service
+
+[Service]
+Type=simple
+User=aios-agent-task
+Group=aios-agent-task
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+EnvironmentFile=/etc/aios-gitops/agent-task-execution-rebind.env
+Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_TASK_EXECUTION_REBIND_HOST=127.0.0.1
+Environment=AIOS_TASK_EXECUTION_REBIND_PORT=18942
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_execution_rebind.py
+Restart=on-failure
+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
+ReadWritePaths=/var/lib/aios-agent-task
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

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-17-r42-p2b-fleet-order-preservation-repair.md

@@ -0,0 +1,17 @@
+# R4.2 P2-B Fleet Order Preservation Repair
+
+P2-C live acceptance exposed a real integration defect in the verified P2-B candidate. The fleet
+scheduler correctly produced durable cross-scope fairness order, but the integration passed the
+entire ranked list to the legacy scheduler. The legacy scheduler then applied its own priority /
+created_at / task_id sort and silently discarded fleet ordering.
+
+The repair preserves the legacy scheduler as the fail-closed security oracle while changing how it
+is called: one fleet-ranked candidate is presented at a time. A candidate rejected by legacy
+security/policy advances to the next fleet candidate. A candidate returned by the oracle must have
+the same task ID as the candidate presented, otherwise execution fails closed.
+
+Task ownership remains unchanged: the original schedule_for_worker/claim_task CAS remains
+authoritative, and a fleet reservation is still only bound after the claim succeeds.
+
+This repair does not modify FleetScheduler state semantics, capacity, WriterLease, checkpoint,
+protected refs or Production behavior.

docs/superpowers/specs/2026-08-17-r42-p2b-fleet-runtime-integration.md

@@ -0,0 +1,23 @@
+# R4.2 P2-B Fleet Runtime Integration
+
+P2-B connects the verified P2 fleet primitives to the existing TaskPlane pull scheduler without
+replacing its security or ownership logic.
+
+The integration uses two stages. `FleetScheduler.rank_for_worker()` performs only read-only
+capacity/fairness/fault-isolation ranking. The original `TaskScheduler.next_queued_task()` is then
+run over that ordered list and remains the fail-closed security oracle for worker liveness,
+capabilities, data classification, execution zone and other existing policy checks. Only a task
+accepted by the original scheduler receives a short durable fleet reservation.
+
+The original `TaskScheduler.schedule_for_worker()` remains the sole claim path and continues to call
+the existing TaskPlane `claim_task` CAS. After a successful claim, the claim ID is bound to the
+fleet reservation. A claim error releases the reservation. A fleet reservation is never treated as
+task ownership.
+
+P2-B also makes the reservation capacity check atomic inside the fleet SQLite transaction so two
+concurrent pullers cannot oversubscribe the same worker slot. Circuit and task/worker quarantine
+state are enforced on the real pull path. Explicit execution-failure signals can be recorded through
+the integrated scheduler's failure/success/quarantine methods; policy/claim failures are not
+automatically mislabeled as worker crashes.
+
+No new daemon, database or production write path is introduced.

docs/superpowers/specs/2026-08-18-generic-compact-interface-map-profiles.md

@@ -0,0 +1,18 @@
+# Generic Compact Interface-Map Profiles
+
+The compact interface-map runtime is a reusable read-only control-plane capability. It keeps the
+canonical `source_capture` action and `compact_interface_map` adapter, but selects a bounded semantic
+profile from the task's source signals.
+
+The existing `fleet` profile remains backward compatible and returns
+`P2_INTERFACE_MAP_CAPTURED` with the `FLEET_*` contracts.
+
+The new `model_governance` profile is selected when model/provider/routing/budget/evaluation terms
+are present. It returns `MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED` and model-specific contracts for
+provider registry, selection policy, version pinning, budget accounting, fallback, scope isolation,
+credential boundary, trace/audit, evaluation gates and routing resilience.
+
+Profile and classification are emitted by the compact-map module and revalidated by the worker
+before result submission. Unknown or mismatched profile/classification pairs fail closed.
+
+No new Bridge action, daemon, database, WriterLease, mutation path or credential surface is added.

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.

docs/superpowers/specs/2026-08-18-owner-controlled-execution-rebind.md

@@ -0,0 +1,36 @@
+# Owner-Controlled Same-Binding Execution Rebind
+
+## Purpose
+
+`aios_agent_task_execution_rebind.py` closes a projection/plan race without replacing the Task,
+Governed Job, target branch, generation, or idempotency lineage.
+
+The only supported transition is:
+
+```text
+official NEEDS_REWORK
+→ Task QUEUED with no active lease
+→ persisted source_capture envelope
+→ canonical Bridge resolves an already-registered deterministic plan
+→ CAS-bound metadata.execution + execution_projection replacement
+→ audit event
+→ normal worker claim
+```
+
+The service rejects arbitrary metadata patches. It derives the new envelope from canonical Bridge
+`/v2/execution-envelope`, validates every governed binding coordinate against the persisted envelope,
+requires `source_capture → deterministic`, requires exact old/new execution digests supplied by the
+Owner recovery request, and increments the Task revision under `BEGIN IMMEDIATE`.
+
+It records `TASK_EXECUTION_REBIND_AUTHORIZED` and an idempotent `task_execution_rebinds` receipt.
+No Task/Job/branch replacement is created and the worker still performs the real claim/heartbeat,
+WriterLease and governed commit.
+
+The endpoint is loopback-only:
+
+`POST /v1/tasks/{task_id}/execution-rebind`
+
+and requires the existing Owner control credential plus
+`AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID`.
+
+Direct/manual Task database edits remain forbidden; the service is the audited mutation boundary.

tests/task_plane/public_interface/test_task_plane_fleet_order_preservation.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[3]
+ACTIVE_RUNTIME = ROOT / "active-runtime"
+if str(ACTIVE_RUNTIME) not in sys.path:
+    sys.path.insert(0, str(ACTIVE_RUNTIME))
+
+from app.task_plane.scheduler import TaskScheduler
+
+
+def _security_contract():
+    path = Path(__file__).with_name("test_task_plane_security_contract.py")
+    spec = importlib.util.spec_from_file_location("p2b_order_security_contract", path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def _create_task(service, contract, task_id, tenant, project):
+    return service.create_task(
+        {
+            "task_id": task_id,
+            "issuer": {"type": "agent", "id": contract.ADMIN_IDENTITY},
+            "required_capabilities": ["code.read"],
+            "scope_type": "tenant",
+            "tenant_id": tenant,
+            "project_id": project,
+            "metadata": {"fleet_priority": 0},
+        }
+    )
+
+
+def _task_id(claim):
+    return getattr(
+        claim,
+        "task_id",
+        claim.get("task_id") if isinstance(claim, dict) else None,
+    )
+
+
+def test_P2B_FLEET_ORDER_is_preserved_through_legacy_claim_oracle():
+    contract = _security_contract()
+    service = contract._service()
+    service.register_worker(contract._worker_payload(max_concurrency=3))
+    worker_id = contract.WORKER_ID
+
+    scheduler = TaskScheduler(service.store)
+    scheduler.fleet_scheduler.configure_worker(
+        worker_id, max_slots=3, failure_domain="zone-a"
+    )
+
+    _create_task(service, contract, "p2b-order-a1", "tenant-a", "project-a")
+    _create_task(service, contract, "p2b-order-a2", "tenant-a", "project-a")
+    _create_task(service, contract, "p2b-order-b1", "tenant-b", "project-b")
+
+    first = scheduler.schedule_for_worker(worker_id)
+    second = scheduler.schedule_for_worker(worker_id)
+
+    assert _task_id(first) == "p2b-order-a1"
+    assert _task_id(second) == "p2b-order-b1"
+
+    rows = service.store.conn.execute(
+        "SELECT scope_key,dispatch_count FROM fleet_scope_state ORDER BY scope_key"
+    ).fetchall()
+    assert len(rows) >= 2

tests/task_plane/public_interface/test_task_plane_fleet_runtime_integration.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import importlib.util
+from pathlib import Path
+import sqlite3
+import sys
+import threading
+import time
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[3]
+ACTIVE_RUNTIME = ROOT / "active-runtime"
+if str(ACTIVE_RUNTIME) not in sys.path:
+    sys.path.insert(0, str(ACTIVE_RUNTIME))
+
+from app.task_plane.fleet import FleetConflict, FleetScheduler
+from app.task_plane.errors import NotFoundError
+from app.task_plane.scheduler import TaskScheduler
+
+
+def _security_contract():
+    path = Path(__file__).with_name("test_task_plane_security_contract.py")
+    spec = importlib.util.spec_from_file_location("p2b_security_contract", path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_P2B_REAL_DISPATCH_claim_binds_fleet_reservation_and_circuit():
+    contract = _security_contract()
+    service = contract._service()
+    service.register_worker(contract._worker_payload(max_concurrency=2))
+    worker_id = contract.WORKER_ID
+
+    service.create_task(
+        {
+            "task_id": "task-p2b-real-1",
+            "issuer": {"type": "agent", "id": contract.ADMIN_IDENTITY},
+            "required_capabilities": ["code.read"],
+        }
+    )
+
+    scheduler = TaskScheduler(service.store)
+    scheduler.fleet_scheduler.configure_worker(
+        worker_id,
+        max_slots=2,
+        failure_domain="zone-real",
+        labels={"runtime": "task-plane"},
+    )
+    first = scheduler.schedule_for_worker(worker_id)
+    assert first is not None
+    assert first.task_id == "task-p2b-real-1"
+
+    row = service.store.conn.execute(
+        "SELECT state,claim_id,worker_id FROM fleet_dispatches WHERE task_id=?",
+        ("task-p2b-real-1",),
+    ).fetchone()
+    assert tuple(row) == ("CLAIMED", first.claim_id, worker_id)
+
+    # A real circuit blocks the actual scheduler even though one capacity slot remains.
+    failure_now = time.time()
+    scheduler.record_worker_failure(
+        worker_id, "runtime_crash", threshold=1, now=failure_now
+    )
+    service.create_task(
+        {
+            "task_id": "task-p2b-real-2",
+            "issuer": {"type": "agent", "id": contract.ADMIN_IDENTITY},
+            "required_capabilities": ["code.read"],
+        }
+    )
+    with pytest.raises(NotFoundError, match="no queued tasks available"):
+        scheduler.schedule_for_worker(worker_id)
+
+    scheduler.record_worker_success(worker_id, now=failure_now + 1.0)
+    second = scheduler.schedule_for_worker(worker_id)
+    assert second is not None
+    assert second.task_id == "task-p2b-real-2"
+
+
+@dataclass
+class Plan:
+    task_id: str
+    required_capabilities: list[str] = field(default_factory=lambda: ["code.read"])
+    metadata: dict = field(default_factory=dict)
+    scope_type: str = "tenant"
+    tenant_id: str = "tenant-a"
+    project_id: str | None = "project-a"
+    user_id: str | None = None
+
+
+@dataclass
+class Task:
+    task_id: str
+    revision: int = 1
+    status: str = "QUEUED"
+    created_at: float = 1.0
+    task_plan: Plan | None = None
+
+    def __post_init__(self):
+        if self.task_plan is None:
+            self.task_plan = Plan(self.task_id)
+
+
+@dataclass
+class Worker:
+    worker_id: str
+    capabilities: list[str] = field(default_factory=lambda: ["code.read"])
+
+
+class Store:
+    def __init__(self):
+        self.conn = sqlite3.connect(":memory:", check_same_thread=False)
+        self._lock = threading.RLock()
+        self.conn.execute(
+            "CREATE TABLE claims("
+            "claim_id TEXT PRIMARY KEY,task_id TEXT,worker_id TEXT,status TEXT,"
+            "lease_expires_at REAL)"
+        )
+        self.tasks = []
+        self.workers = {}
+
+    def list_tasks(self, _ctx=None):
+        return list(self.tasks)
+
+    def get_worker(self, worker_id):
+        return self.workers[worker_id]
+
+
+class Registry:
+    def __init__(self):
+        self.online = set()
+
+    def is_online(self, worker_id, now=None):
+        return worker_id in self.online
+
+
+def _fleet():
+    store = Store()
+    registry = Registry()
+    store.workers["w-a"] = Worker("w-a")
+    registry.online.add("w-a")
+    fleet = FleetScheduler(store, registry, now_fn=lambda: 1000.0)
+    fleet.configure_worker("w-a", max_slots=1, failure_domain="zone-a")
+    return store, fleet
+
+
+def test_P2B_CAPACITY_RESERVATION_is_atomic_under_concurrent_pullers():
+    store, fleet = _fleet()
+    tasks = [Task("task-a"), Task("task-b")]
+    store.tasks = tasks
+    barrier = threading.Barrier(2)
+    outcomes = []
+
+    def reserve(task):
+        barrier.wait()
+        try:
+            fleet.reserve_for_worker("w-a", task, now=1000.0)
+            outcomes.append("reserved")
+        except FleetConflict:
+            outcomes.append("conflict")
+
+    threads = [threading.Thread(target=reserve, args=(task,)) for task in tasks]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert sorted(outcomes) == ["conflict", "reserved"]
+    assert fleet.state.reserved_slots("w-a", 1000.0) == 1
+
+
+def test_P2B_RANK_is_read_only_until_legacy_security_accepts_candidate():
+    store, fleet = _fleet()
+    store.tasks = [Task("task-a"), Task("task-b")]
+    ranked = fleet.rank_for_worker("w-a", now=1000.0)
+    assert [task.task_id for task in ranked] == ["task-a", "task-b"]
+    assert fleet.state.reserved_slots("w-a", 1000.0) == 0

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_agent_task_execution_rebind.py

@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+from pathlib import Path
+import hashlib
+import json
+import sqlite3
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_agent_task_execution_rebind import (
+    ExecutionRebindError,
+    ExecutionRebindRequest,
+    ExecutionRebindStore,
+)
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+AUTHORIZATION="owner-execution-rebind-20260818"
+TASK_ID="task-rebind-a"
+JOB_ID="job-rebind-a"
+BASE="a"*40
+BRANCH="agent/task-rebind-a"
+IDEMPOTENCY="task-rebind-a-binding-v1"
+
+
+@pytest.fixture(autouse=True)
+def _auth(monkeypatch):
+    monkeypatch.setenv(
+        "AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID",
+        AUTHORIZATION,
+    )
+
+
+def canon(value):
+    return hashlib.sha256(
+        json.dumps(value,sort_keys=True,separators=(",",":")).encode()
+    ).hexdigest()
+
+
+def binding():
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "branch_mode":"CREATE_NEW",
+        "generation":1,
+        "idempotency_key":IDEMPOTENCY,
+    })
+
+
+def old_envelope():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action":"source_capture",
+            "adapter":"compact_interface_map",
+            "capability_scope":[
+                "git.read","git.test","governed.write.branch","governed.write.file"
+            ],
+        },
+    )
+
+
+def new_envelope():
+    return 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":"VALUE=1\\n",
+                }
+            ],
+            "verification":[
+                {
+                    "label":"unit",
+                    "argv":["/usr/bin/python3","-m","pytest","-q"],
+                    "cwd":".",
+                    "timeout_seconds":120,
+                }
+            ],
+            "governance":{
+                "commit_idempotency_key":"task-rebind-a-commit-v1",
+            },
+        },
+    )
+
+
+def make_db(tmp_path, *, status="QUEUED", revision=5, attempt=2, lease=False):
+    path=tmp_path/"state.sqlite3"
+    con=sqlite3.connect(path)
+    con.executescript("""
+    CREATE TABLE tasks(
+      task_id TEXT PRIMARY KEY,
+      status TEXT NOT NULL,
+      attempt INTEGER NOT NULL,
+      max_attempts INTEGER NOT NULL,
+      revision INTEGER NOT NULL,
+      claimed_by TEXT,
+      lease_token_hash TEXT,
+      lease_expires_at TEXT,
+      verification_json TEXT,
+      result_json TEXT,
+      metadata_json TEXT,
+      required_capabilities_json TEXT
+    );
+    CREATE TABLE task_events(
+      event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+      task_id TEXT,
+      event_type TEXT,
+      agent_id TEXT,
+      payload_json TEXT,
+      created_at TEXT
+    );
+    """)
+    verification={"decision":"NEEDS_REWORK","reason":"WRONG_ENVELOPE_BEFORE_PLAN"}
+    old=old_envelope()
+    metadata={
+        "execution":old,
+        "execution_projection":{
+            "schema":"aios.online-agent.task-projection.v1",
+            "task_id":TASK_ID,
+            "job_id":JOB_ID,
+            "execution_digest":old["execution_digest"],
+        },
+        "preserved":{"value":"unchanged"},
+    }
+    con.execute(
+        "INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        (
+            TASK_ID,status,attempt,3,revision,
+            "previous-worker" if not lease else "active-worker",
+            "lease-hash" if lease else None,
+            "2099-01-01T00:00:00Z" if lease else None,
+            json.dumps(verification),
+            json.dumps({"status":"SUCCEEDED"}),
+            json.dumps(metadata),
+            json.dumps([
+                "git.read","git.test","governed.write.branch","governed.write.file"
+            ]),
+        ),
+    )
+    con.commit();con.close()
+    return path,verification
+
+
+def request(verification, *, revision=5, old=None, new=None, key="rebind-task-a-v1"):
+    old=old or old_envelope()
+    new=new or new_envelope()
+    return ExecutionRebindRequest.from_mapping({
+        "expected_status":"QUEUED",
+        "expected_revision":revision,
+        "expected_attempt":2,
+        "previous_verification_sha256":canon(verification),
+        "expected_old_execution_digest":old["execution_digest"],
+        "expected_new_execution_digest":new["execution_digest"],
+        "authorization_id":AUTHORIZATION,
+        "idempotency_key":key,
+        "reason":"projection-before-plan race recovery",
+    })
+
+
+def resolver(_task_id,_caps):
+    return new_envelope()
+
+
+def test_rebind_cas_updates_only_execution_projection_and_audits(tmp_path):
+    path,verification=make_db(tmp_path)
+    result=ExecutionRebindStore(path,resolver).rebind(
+        TASK_ID,request(verification)
+    )
+    assert result["status"]=="QUEUED"
+    assert result["revision"]==6
+    assert result["idempotent"] is False
+
+    con=sqlite3.connect(path)
+    row=con.execute(
+        "SELECT status,attempt,revision,claimed_by,metadata_json FROM tasks"
+    ).fetchone()
+    assert row[:4]==("QUEUED",2,6,None)
+    metadata=json.loads(row[4])
+    assert metadata["preserved"]=={"value":"unchanged"}
+    assert metadata["execution"]["action"]=="deterministic"
+    assert metadata["execution"]["execution_digest"]==new_envelope()["execution_digest"]
+    assert metadata["execution_projection"]["execution_digest"]==new_envelope()["execution_digest"]
+
+    event=con.execute(
+        "SELECT event_type,payload_json FROM task_events"
+    ).fetchone()
+    assert event[0]=="TASK_EXECUTION_REBIND_AUTHORIZED"
+    payload=json.loads(event[1])
+    assert payload["old_action"]=="source_capture"
+    assert payload["new_action"]=="deterministic"
+    assert payload["stale_claim_normalized"] is True
+    assert con.execute("SELECT COUNT(*) FROM task_execution_rebinds").fetchone()[0]==1
+
+
+def test_rebind_idempotent_replay(tmp_path):
+    path,verification=make_db(tmp_path)
+    store=ExecutionRebindStore(path,resolver)
+    req=request(verification)
+    first=store.rebind(TASK_ID,req)
+    second=store.rebind(TASK_ID,req)
+    assert first["idempotent"] is False
+    assert second["idempotent"] is True
+    assert second["new_execution_digest"]==first["new_execution_digest"]
+
+
+def test_rebind_rejects_active_lease(tmp_path):
+    path,verification=make_db(tmp_path,lease=True)
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_ACTIVE_LEASE_DENY"):
+        ExecutionRebindStore(path,resolver).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_rejects_verification_drift(tmp_path):
+    path,_=make_db(tmp_path)
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_VERIFICATION_DRIFT"):
+        ExecutionRebindStore(path,resolver).rebind(
+            TASK_ID,request({"decision":"NEEDS_REWORK","reason":"other"})
+        )
+
+
+def test_rebind_rejects_nonqueued_task(tmp_path):
+    path,verification=make_db(tmp_path,status="VERIFYING")
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_CAS_MISMATCH"):
+        ExecutionRebindStore(path,resolver).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_rejects_non_deterministic_canonical_resolution(tmp_path):
+    path,verification=make_db(tmp_path)
+    def wrong(_task,_caps):
+        return old_envelope()
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_DETERMINISTIC_PLAN_REQUIRED"):
+        ExecutionRebindStore(path,wrong).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_rejects_binding_drift(tmp_path):
+    path,verification=make_db(tmp_path)
+    different=project_execution_envelope(
+        GovernedExecutionBinding.from_mapping({
+            "task_id":TASK_ID,
+            "job_id":"job-other",
+            "repository":"aios/aios",
+            "base_oid":BASE,
+            "expected_branch_oid":BASE,
+            "target_branch":BRANCH,
+            "branch_mode":"CREATE_NEW",
+            "generation":1,
+            "idempotency_key":IDEMPOTENCY,
+        }),
+        {
+            "action":"deterministic",
+            "capability_scope":["git.read"],
+            "change_set":[{"path":"x.py","operation":"write_file","content":"x=1\\n"}],
+        },
+    )
+    def wrong(_task,_caps):
+        return different
+    with pytest.raises(Exception):
+        ExecutionRebindStore(path,wrong).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_request_is_narrow_and_digest_bound():
+    verification={"decision":"NEEDS_REWORK"}
+    data=request(verification).public()
+    data["arbitrary_metadata"]={"x":1}
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_FIELDS_INVALID"):
+        ExecutionRebindRequest.from_mapping(data)
+
+
+def test_execution_rebind_unit_is_unprivileged_and_owner_credential_bound():
+    unit=(ROOT/"deploy/systemd/aios-agent-task-execution-rebind.service").read_text()
+    assert "User=aios-agent-task" in unit
+    assert "Group=aios-agent-task" in unit
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
+    assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
+    assert "User=root" not in unit

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_compact_interface_map_profiles.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+import json
+import sys
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime/app"
+sys.path.insert(0, str(APP))
+
+from aios_online_agent_compact_map import build_compact_interface_map
+import aios_online_agent_worker as worker
+from aios_online_agent_worker import build_compact_interface_map_result
+
+
+def _repo(tmp_path: Path) -> Path:
+    root = tmp_path / "repo"
+    (root / "active-runtime/app").mkdir(parents=True)
+    (root / "tests").mkdir(parents=True)
+    (root / "active-runtime/app/model_router.py").write_text(
+        """
+class ProviderRegistry:
+    def register_provider(self, provider_id, model_version):
+        return provider_id, model_version
+
+class ModelRouter:
+    def select_model(self, tenant_id, budget_tokens, fallback_provider):
+        return fallback_provider
+
+def record_model_cost(trace_id, token_count, cost_usd):
+    return trace_id, token_count, cost_usd
+
+def evaluate_model_version(model_id, version, approval_gate):
+    return approval_gate
+""".strip() + "\\n",
+        encoding="utf-8",
+    )
+    (root / "tests/test_model_router.py").write_text(
+        """
+def test_model_budget_and_fallback():
+    assert True
+""".strip() + "\\n",
+        encoding="utf-8",
+    )
+    return root
+
+
+def _envelope(base_oid: str):
+    return SimpleNamespace(
+        base_oid=base_oid,
+        job_id="job-model-governance",
+        repository="aios/aios",
+        expected_branch_oid=base_oid,
+        target_branch="agent/model-governance-map",
+        generation=1,
+        digest="e" * 64,
+    )
+
+
+def test_model_governance_profile_contracts_and_classification(tmp_path):
+    root = _repo(tmp_path)
+    value = build_compact_interface_map(
+        root,
+        [
+            "model provider registry routing budget token cost fallback",
+            "evaluation approval version pinning tenant project credential trace",
+        ],
+        source_commit="a" * 40,
+    )
+    assert value["profile"] == "model_governance"
+    assert value["classification"] == "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED"
+    assert set(value["contracts"]) == {
+        "MODEL_PROVIDER_REGISTRY",
+        "MODEL_SELECTION_POLICY",
+        "MODEL_VERSION_PINNING",
+        "MODEL_BUDGET_ACCOUNTING",
+        "MODEL_FALLBACK_POLICY",
+        "MODEL_SCOPE_ISOLATION",
+        "MODEL_CREDENTIAL_BOUNDARY",
+        "MODEL_TRACE_AUDIT",
+        "MODEL_EVALUATION_GATE",
+        "MODEL_ROUTING_RESILIENCE",
+    }
+    assert "FLEET_FAIRNESS" not in value["contracts"]
+
+
+def test_fleet_profile_is_backward_compatible(tmp_path):
+    root = _repo(tmp_path)
+    value = build_compact_interface_map(
+        root,
+        ["agent fleet scheduler worker capacity fairness"],
+        source_commit="b" * 40,
+    )
+    assert value["profile"] == "fleet"
+    assert value["classification"] == "P2_INTERFACE_MAP_CAPTURED"
+    assert "FLEET_AGENT_REGISTRATION" in value["contracts"]
+
+
+def test_worker_binds_model_profile_to_model_classification():
+    interface_map = {
+        "schema": "aios.engineering.interface-map.v1",
+        "profile": "model_governance",
+        "classification": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
+        "source_commit": "c" * 40,
+        "implementation_map": {"modules": [], "classes": [], "functions": [], "tests": []},
+        "bounded_excerpts": [],
+        "contracts": {"MODEL_PROVIDER_REGISTRY": {"status": "GAP", "surfaces": []}},
+        "result_bytes": 512,
+    }
+    result = build_compact_interface_map_result(
+        "r42-model-governance-interface-map-20260818",
+        _envelope("c" * 40),
+        {"interface_map": interface_map},
+    )
+    assert result["classification"] == "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED"
+    assert result["interface_map_profile"] == "model_governance"
+    assert result["writer_lease"] == "NONE"
+
+
+def test_worker_rejects_profile_classification_mismatch():
+    interface_map = {
+        "schema": "aios.engineering.interface-map.v1",
+        "profile": "model_governance",
+        "classification": "P2_INTERFACE_MAP_CAPTURED",
+        "source_commit": "c" * 40,
+        "implementation_map": {"modules": [], "classes": [], "functions": [], "tests": []},
+        "bounded_excerpts": [],
+        "contracts": {},
+        "result_bytes": 512,
+    }
+    with pytest.raises(worker.WorkerError, match="COMPACT_INTERFACE_MAP_PROFILE_BINDING_INVALID"):
+        build_compact_interface_map_result(
+            "r42-model-governance-interface-map-20260818",
+            _envelope("c" * 40),
+            {"interface_map": interface_map},
+        )

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]