feat(codex): add trusted remote Codex brain runner

automation/remote_codex_health_probe.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+import time
+
+from services.aios_orchestrator_gateway.config import Config
+from services.aios_orchestrator_gateway.remote_codex import RemoteCodexBroker
+
+
+def atomic_receipt(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    fd, temporary = tempfile.mkstemp(prefix=".remote-codex-health-", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(value, handle, separators=(",", ":"), sort_keys=True)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.chmod(temporary, 0o600)
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def run(config_path: Path, receipt: Path, healthy_within: int) -> int:
+    config = Config.from_file(config_path)
+    if config.central_channel != "remote_codex":
+        raise RuntimeError("remote_codex_not_selected")
+    broker = RemoteCodexBroker(
+        config.database_path,
+        contract_commit=config.contract_commit,
+        lease_seconds=config.remote_job_lease_seconds,
+        max_retries=config.max_retries,
+    )
+    try:
+        snapshot = broker.health_snapshot(healthy_within_seconds=healthy_within)
+    finally:
+        broker.close()
+    if snapshot["healthy_runners"] < 1:
+        raise RuntimeError("remote_codex_runner_unhealthy")
+    atomic_receipt(receipt, {
+        "subject": "remote_codex",
+        "health": "HEALTHY",
+        "contract_commit": config.contract_commit,
+        "healthy_runners": snapshot["healthy_runners"],
+        "runner_count": snapshot["runner_count"],
+        "job_counts": snapshot["job_counts"],
+        "observed_at_epoch": int(time.time()),
+    })
+    print("REMOTE_CODEX_HEALTH=PASS")
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--gateway-config", required=True, type=Path)
+    parser.add_argument("--receipt", required=True, type=Path)
+    parser.add_argument("--healthy-within-seconds", type=int, default=300)
+    args = parser.parse_args()
+    try:
+        return run(args.gateway_config, args.receipt, args.healthy_within_seconds)
+    except Exception as exc:
+        reason = str(exc)
+        safe = reason if reason.replace("_", "").isalnum() and len(reason) <= 64 else type(exc).__name__
+        print(f"REMOTE_CODEX_HEALTH=BLOCKED reason={safe}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

config/aios-orchestrator-gateway.example.json

@@ -10,11 +10,22 @@
   "inbound_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
   "callback_token_file": "/etc/aios-orchestrator-gateway/callback.token",
   "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "central_channel": "codex_local",
+  "central_channel": "remote_codex",
   "codex_local": {
     "queue_dir": "/var/lib/aios-codex-brain/queue",
     "timeout_seconds": 300
   },
+  "remote_codex": {
+    "request_max_bytes": 262144,
+    "job_lease_seconds": 120,
+    "runners": [
+      {
+        "runner_id": "codex-brain-192-168-88-6",
+        "token_file": "/etc/aios-orchestrator-gateway/remote-codex-88-6.token",
+        "scopes": ["claim", "heartbeat", "decision", "fail"]
+      }
+    ]
+  },
   "workspace_agent": {
     "trigger_id": null,
     "access_token_file": null

openapi/aios-orchestrator-gateway.openapi.yaml

@@ -5,6 +5,99 @@ info:
 servers:
   - url: http://127.0.0.1:18920
 paths:
+  /v1/orchestrator/central/jobs/claim:
+    post:
+      summary: Claim one remote Codex decision job
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: '#/components/parameters/RunnerId'
+        - $ref: '#/components/parameters/IdempotencyKey'
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: false
+              required: [contract_commit]
+              properties:
+                contract_commit:
+                  type: string
+                  pattern: '^[0-9a-f]{40}$'
+      responses:
+        '200':
+          description: Job lease or no available job
+        '401':
+          description: Runner or token scope rejected
+        '409':
+          description: Contract or idempotency conflict
+  /v1/orchestrator/central/jobs/{job_id}/heartbeat:
+    post:
+      summary: Extend an active remote Codex job lease
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: '#/components/parameters/JobId'
+        - $ref: '#/components/parameters/RunnerId'
+        - $ref: '#/components/parameters/IdempotencyKey'
+      requestBody:
+        $ref: '#/components/requestBodies/RemoteLeaseRequest'
+      responses:
+        '202': {description: Lease extended}
+        '401': {description: Runner, scope, or lease credential rejected}
+        '409': {description: Lease is stale or owned by another runner}
+  /v1/orchestrator/central/jobs/{job_id}/decision:
+    post:
+      summary: Submit one strict decision for a leased job
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: '#/components/parameters/JobId'
+        - $ref: '#/components/parameters/RunnerId'
+        - $ref: '#/components/parameters/IdempotencyKey'
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: false
+              required: [lease_token, claim_idempotency_key, decision]
+              properties:
+                lease_token: {type: string, pattern: '^[0-9a-f]{64}$'}
+                claim_idempotency_key: {type: string}
+                decision: {$ref: '#/components/schemas/Decision'}
+      responses:
+        '202': {description: Decision accepted and applied exactly once}
+        '401': {description: Runner, scope, or lease credential rejected}
+        '409': {description: State, contract, lease, or idempotency conflict}
+        '422': {description: Decision schema or allowed work-unit violation}
+  /v1/orchestrator/central/jobs/{job_id}/fail:
+    post:
+      summary: Fail a leased job into retry or dead-letter
+      security:
+        - bearerAuth: []
+      parameters:
+        - $ref: '#/components/parameters/JobId'
+        - $ref: '#/components/parameters/RunnerId'
+        - $ref: '#/components/parameters/IdempotencyKey'
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: false
+              required: [lease_token, claim_idempotency_key, category]
+              properties:
+                lease_token: {type: string, pattern: '^[0-9a-f]{64}$'}
+                claim_idempotency_key: {type: string}
+                category: {type: string, pattern: '^[A-Za-z0-9_]{1,64}$'}
+      responses:
+        '202': {description: Failure recorded}
+        '401': {description: Runner, scope, or lease credential rejected}
+        '409': {description: Lease or idempotency conflict}
   /v1/orchestrator/events:
     post:
       security:
@@ -50,7 +143,7 @@ paths:
                   trigger_channel:
                     type: string
                     nullable: true
-                    enum: [workspace_agent, responses_api]
+                    enum: [workspace_agent, responses_api, codex_local, remote_codex]
                   trigger_http_status:
                     type: integer
                     nullable: true
@@ -83,6 +176,52 @@ paths:
         '404':
           description: Event not found
 components:
+  parameters:
+    JobId:
+      in: path
+      name: job_id
+      required: true
+      schema:
+        type: string
+        pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$'
+    RunnerId:
+      in: header
+      name: X-AIOS-Runner-ID
+      required: true
+      schema: {type: string, maxLength: 128}
+    IdempotencyKey:
+      in: header
+      name: Idempotency-Key
+      required: true
+      schema: {type: string, minLength: 16, maxLength: 128}
+  requestBodies:
+    RemoteLeaseRequest:
+      required: true
+      content:
+        application/json:
+          schema:
+            type: object
+            additionalProperties: false
+            required: [lease_token, claim_idempotency_key]
+            properties:
+              lease_token: {type: string, pattern: '^[0-9a-f]{64}$'}
+              claim_idempotency_key: {type: string}
+  schemas:
+    Decision:
+      type: object
+      additionalProperties: false
+      required: [schema_version, project_key, contract_commit, event_id, expected_state_version, verdict, state_updates, next_work_units, requires_human_input, stop_reason]
+      properties:
+        schema_version: {type: integer, enum: [1]}
+        project_key: {type: string}
+        contract_commit: {type: string, pattern: '^[0-9a-f]{40}$'}
+        event_id: {type: string}
+        expected_state_version: {type: integer, minimum: 0}
+        verdict: {type: string, enum: [BLOCKED, COMPLETE, DISPATCH, NOOP]}
+        state_updates: {type: array, items: {type: object}}
+        next_work_units: {type: array, items: {type: object}}
+        requires_human_input: {type: boolean}
+        stop_reason: {type: string, nullable: true}
   securitySchemes:
     bearerAuth:
       type: http

remote-runner/aios_remote_codex_brain.py

@@ -0,0 +1,459 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+import json
+import logging
+import os
+from pathlib import Path
+import signal
+import stat
+import subprocess
+import tempfile
+import threading
+import time
+from typing import Any
+import urllib.error
+import urllib.request
+import uuid
+
+from services.aios_orchestrator_gateway.errors import ConfigurationError, TriggerError, ValidationError
+from services.aios_orchestrator_gateway.models import DECISION_SCHEMA, Event
+from services.aios_orchestrator_gateway.remote_codex import validate_remote_decision
+from services.aios_orchestrator_gateway.securefs import atomic_write_json, read_secure_bytes, read_secure_json
+
+
+LOG = logging.getLogger("aios-remote-codex-brain")
+REQUIRED_EXEC_FLAGS = (
+    "--json", "--output-schema", "--output-last-message", "--sandbox",
+    "--config", "--cd", "--ephemeral",
+)
+APPROVED_CODEX_ROOTS = (
+    Path("/usr/lib/node_modules/@openai/codex"),
+    Path("/usr/local/lib/node_modules/@openai/codex"),
+    Path("/opt/openai/codex"),
+)
+EXPECTED_CODEX_VERSION = "0.144.6"
+
+
+@dataclass(frozen=True)
+class RunnerConfig:
+    gateway_url: str
+    runner_id: str
+    token_file: Path
+    contract_commit: str
+    codex_binary: Path
+    codex_home: Path
+    controller_readonly: Path
+    state_file: Path
+    timeout_seconds: int
+    poll_seconds: int
+    heartbeat_seconds: int
+
+    @classmethod
+    def from_file(cls, path: Path) -> "RunnerConfig":
+        raw = read_secure_json(path, expected_uid=0)
+        expected = {
+            "gateway_url", "runner_id", "token_file", "contract_commit",
+            "codex_binary", "codex_home", "controller_readonly", "state_file",
+            "timeout_seconds", "poll_seconds", "heartbeat_seconds",
+        }
+        if set(raw) != expected:
+            raise ConfigurationError("invalid_runner_config_fields")
+
+        def absolute(name: str) -> Path:
+            value = raw.get(name)
+            if not isinstance(value, str) or not Path(value).is_absolute():
+                raise ConfigurationError(f"invalid_{name}")
+            return Path(value)
+
+        url = raw.get("gateway_url")
+        if url != "http://127.0.0.1:28920/v1/orchestrator":
+            raise ConfigurationError("gateway_must_use_approved_loopback_tunnel")
+        runner_id = raw.get("runner_id")
+        contract = raw.get("contract_commit")
+        if not isinstance(runner_id, str) or not runner_id or len(runner_id) > 128:
+            raise ConfigurationError("invalid_runner_id")
+        if not isinstance(contract, str) or len(contract) != 40 or any(c not in "0123456789abcdef" for c in contract):
+            raise ConfigurationError("invalid_contract_commit")
+
+        def bounded(name: str, low: int, high: int) -> int:
+            value = raw.get(name)
+            if not isinstance(value, int) or not low <= value <= high:
+                raise ConfigurationError(f"invalid_{name}")
+            return value
+
+        return cls(
+            gateway_url=url, runner_id=runner_id, token_file=absolute("token_file"),
+            contract_commit=contract, codex_binary=absolute("codex_binary"),
+            codex_home=absolute("codex_home"),
+            controller_readonly=absolute("controller_readonly"),
+            state_file=absolute("state_file"),
+            timeout_seconds=bounded("timeout_seconds", 30, 3600),
+            poll_seconds=bounded("poll_seconds", 1, 300),
+            heartbeat_seconds=bounded("heartbeat_seconds", 5, 300),
+        )
+
+
+class RemoteCodexRunner:
+    def __init__(self, config: RunnerConfig):
+        self.config = config
+        self._stop = threading.Event()
+
+    def _environment(self) -> dict[str, str]:
+        return {
+            "CODEX_HOME": str(self.config.codex_home),
+            "HOME": str(self.config.codex_home.parent),
+            "PATH": f"{self.config.codex_binary.parent}:/usr/bin:/bin",
+            "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8",
+        }
+
+    @staticmethod
+    def _safe_regular(path: Path, *, owner_uid: int, mode: int | None = None) -> os.stat_result:
+        info = path.lstat()
+        if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+            raise ConfigurationError("unsafe_regular_file")
+        if info.st_uid != owner_uid:
+            raise ConfigurationError("unsafe_file_owner")
+        if mode is not None and stat.S_IMODE(info.st_mode) != mode:
+            raise ConfigurationError("unsafe_file_mode")
+        return info
+
+    def validate_official_cli(self) -> None:
+        binary = self.config.codex_binary
+        if not binary.is_absolute() or not binary.exists():
+            raise ConfigurationError("codex_cli_not_installed")
+        link_info = binary.lstat()
+        if link_info.st_uid != 0:
+            raise ConfigurationError("unsafe_codex_launcher")
+        if not stat.S_ISLNK(link_info.st_mode) and link_info.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+            raise ConfigurationError("unsafe_codex_launcher")
+        target = binary.resolve(strict=True)
+        target_info = target.lstat()
+        if not stat.S_ISREG(target_info.st_mode) or target_info.st_uid != 0:
+            raise ConfigurationError("unsafe_codex_target")
+        if target_info.st_mode & (stat.S_IWGRP | stat.S_IWOTH) or not os.access(target, os.X_OK):
+            raise ConfigurationError("unsafe_codex_target")
+        approved = next((root for root in APPROVED_CODEX_ROOTS if target.is_relative_to(root)), None)
+        if approved is None:
+            raise ConfigurationError("unapproved_codex_install_root")
+        if approved.name == "codex" and "node_modules" in approved.parts:
+            package = approved / "package.json"
+            value = read_secure_json(package, expected_uid=0)
+            if value.get("name") != "@openai/codex" or value.get("version") != EXPECTED_CODEX_VERSION:
+                raise ConfigurationError("unofficial_codex_package")
+            native = list(approved.glob("node_modules/@openai/codex-*/vendor/*/bin/codex"))
+            if len(native) != 1:
+                raise ConfigurationError("codex_native_binary_missing")
+            native_info = native[0].lstat()
+            if (
+                not stat.S_ISREG(native_info.st_mode) or native_info.st_uid != 0
+                or native_info.st_mode & (stat.S_IWGRP | stat.S_IWOTH)
+                or not os.access(native[0], os.X_OK)
+            ):
+                raise ConfigurationError("unsafe_codex_native_binary")
+        result = self._run_quiet([str(binary), "exec", "--help"], 30, capture_stdout=True)
+        if result.returncode != 0 or any(flag not in result.stdout for flag in REQUIRED_EXEC_FLAGS):
+            raise ConfigurationError("codex_cli_version_incompatible")
+
+    def validate_auth_metadata(self) -> None:
+        auth = self.config.codex_home / "auth.json"
+        self._safe_regular(auth, owner_uid=os.geteuid(), mode=0o600)
+
+    def login_status(self) -> None:
+        result = self._run_quiet(
+            [str(self.config.codex_binary), "login", "status"], 30, capture_stdout=True
+        )
+        if result.returncode != 0:
+            raise ConfigurationError("codex_authentication_required")
+        normalized = result.stdout.lower()
+        if "oneapi" in normalized or "api key" in normalized or "chatgpt" not in normalized:
+            raise ConfigurationError("codex_chatgpt_authentication_required")
+
+    def validate_runtime_boundary(self) -> None:
+        controller = self.config.controller_readonly
+        if not controller.is_dir() or controller.is_symlink() or os.access(controller, os.W_OK):
+            raise ConfigurationError("controller_not_read_only")
+        if (controller / ".git").exists():
+            raise ConfigurationError("controller_must_not_be_git_worktree")
+        state_parent = self.config.state_file.parent
+        state_parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        if state_parent.is_symlink():
+            raise ConfigurationError("unsafe_state_directory")
+
+    def health_check(self) -> None:
+        self.validate_official_cli()
+        self.validate_auth_metadata()
+        self.validate_runtime_boundary()
+        self.login_status()
+
+    def _run_quiet(self, argv: list[str], timeout: int, *, capture_stdout: bool = False) -> subprocess.CompletedProcess[str]:
+        try:
+            return subprocess.run(
+                argv, cwd=self.config.controller_readonly, env=self._environment(),
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.PIPE if capture_stdout else subprocess.DEVNULL,
+                stderr=subprocess.STDOUT if capture_stdout else subprocess.DEVNULL,
+                text=True, timeout=timeout, check=False,
+            )
+        except (OSError, subprocess.SubprocessError) as exc:
+            raise ConfigurationError("codex_cli_probe_failed") from exc
+
+    def _token(self) -> str:
+        info = self._safe_regular(self.config.token_file, owner_uid=os.geteuid(), mode=0o600)
+        if info.st_size > 4096:
+            raise ConfigurationError("runner_token_too_large")
+        raw = read_secure_bytes(self.config.token_file, expected_uid=os.geteuid(), max_bytes=4096)
+        if not raw or b"\n" in raw or b"\r" in raw:
+            raise ConfigurationError("invalid_runner_token")
+        return raw.decode("utf-8")
+
+    def _request(self, path: str, scope: str, body: dict[str, Any], key: str) -> dict[str, Any]:
+        encoded = json.dumps(body, separators=(",", ":"), sort_keys=True).encode()
+        request = urllib.request.Request(
+            f"{self.config.gateway_url}/{path.lstrip('/')}", data=encoded, method="POST",
+            headers={
+                "Authorization": f"Bearer {self._token()}",
+                "Content-Type": "application/json",
+                "Idempotency-Key": key,
+                "X-AIOS-Runner-ID": self.config.runner_id,
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                if response.status not in {200, 202}:
+                    raise TriggerError("gateway_request_rejected")
+                raw = response.read(1_048_577)
+        except (urllib.error.URLError, TimeoutError, OSError) as exc:
+            raise TriggerError(f"gateway_{scope}_unavailable") from exc
+        if len(raw) > 1_048_576:
+            raise TriggerError("gateway_response_too_large")
+        try:
+            value = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise TriggerError("gateway_invalid_response") from exc
+        if not isinstance(value, dict):
+            raise TriggerError("gateway_invalid_response")
+        return value
+
+    def _load_state(self) -> dict[str, Any] | None:
+        try:
+            return read_secure_json(self.config.state_file, expected_uid=os.geteuid())
+        except FileNotFoundError:
+            return None
+
+    def _save_state(self, value: dict[str, Any]) -> None:
+        atomic_write_json(self.config.state_file, value, mode=0o600)
+
+    def _clear_state(self) -> None:
+        try:
+            self.config.state_file.unlink()
+            directory_fd = os.open(self.config.state_file.parent, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        except FileNotFoundError:
+            pass
+
+    def _claim(self) -> dict[str, Any] | None:
+        claim_key = "claim-" + uuid.uuid4().hex
+        response = self._request(
+            "central/jobs/claim", "claim",
+            {"contract_commit": self.config.contract_commit}, claim_key,
+        )
+        if response.get("claimed") is False:
+            return None
+        required = {
+            "claimed", "job_id", "runner_id", "lease_version", "lease_until",
+            "lease_token", "event", "claim_idempotency_key",
+        }
+        if set(response) != required or response.get("runner_id") != self.config.runner_id:
+            raise TriggerError("invalid_claim_response")
+        event = Event.from_dict(response.get("event"))
+        if event.contract_commit != self.config.contract_commit:
+            raise ValidationError("contract_version_mismatch")
+        state = {
+            "phase": "CLAIMED", "job_id": response["job_id"],
+            "lease_token": response["lease_token"], "claim_idempotency_key": claim_key,
+            "event": event.to_dict(), "decision": None,
+        }
+        self._save_state(state)
+        return state
+
+    def _heartbeat(self, state: dict[str, Any], key: str | None = None) -> dict[str, Any]:
+        return self._request(
+            f"central/jobs/{state['job_id']}/heartbeat", "heartbeat",
+            {"lease_token": state["lease_token"],
+             "claim_idempotency_key": state["claim_idempotency_key"]},
+            key or ("heartbeat-" + uuid.uuid4().hex),
+        )
+
+    def _exec(self, event: Event, heartbeat_state: dict[str, Any]) -> dict[str, Any]:
+        prompt = (
+            "读取只读治理合同与本次事件。只输出符合Decision Schema的决策。"
+            "不得修改Git、部署生产、读取秘密或伪造证据；证据不足时输出BLOCKED。\n"
+            + event.canonical_json()
+        )
+        runtime_root = self.config.state_file.parent / "runtime"
+        runtime_root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        os.chmod(runtime_root, 0o700)
+        with tempfile.TemporaryDirectory(prefix="remote-decision-", dir=runtime_root) as directory:
+            root = Path(directory)
+            schema_path = root / "decision.schema.json"
+            output_path = root / "decision.json"
+            schema_path.write_text(json.dumps(DECISION_SCHEMA, separators=(",", ":")), encoding="utf-8")
+            schema_path.chmod(0o600)
+            argv = [
+                str(self.config.codex_binary), "exec", "--json",
+                "--output-schema", str(schema_path),
+                "--output-last-message", str(output_path),
+                "--sandbox", "read-only", "--config", 'approval_policy="never"',
+                "--cd", str(self.config.controller_readonly), "--ephemeral", "-",
+            ]
+            process = subprocess.Popen(
+                argv, cwd=self.config.controller_readonly, env=self._environment(),
+                stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                text=True, start_new_session=True,
+            )
+            heartbeat_error: list[Exception] = []
+            heartbeat_stop = threading.Event()
+
+            def heartbeat_loop() -> None:
+                while not heartbeat_stop.wait(self.config.heartbeat_seconds):
+                    try:
+                        self._heartbeat(heartbeat_state)
+                    except Exception as exc:
+                        heartbeat_error.append(exc)
+                        return
+
+            thread = threading.Thread(target=heartbeat_loop, daemon=True)
+            thread.start()
+            try:
+                process.communicate(prompt, timeout=self.config.timeout_seconds)
+            except subprocess.TimeoutExpired as exc:
+                self._terminate_group(process)
+                raise TriggerError("codex_exec_timeout") from exc
+            finally:
+                heartbeat_stop.set(); thread.join(timeout=5)
+            if heartbeat_error:
+                raise TriggerError("lease_heartbeat_failed")
+            if process.returncode != 0:
+                raise TriggerError("codex_exec_nonzero")
+            try:
+                if output_path.stat().st_size > 1_048_576:
+                    raise TriggerError("codex_output_too_large")
+                value = json.loads(output_path.read_text(encoding="utf-8"))
+            except (OSError, json.JSONDecodeError) as exc:
+                raise TriggerError("codex_exec_invalid_json") from exc
+            if not isinstance(value, dict):
+                raise TriggerError("codex_exec_invalid_json")
+            return validate_remote_decision(value, event).to_dict()
+
+    @staticmethod
+    def _terminate_group(process: subprocess.Popen[str]) -> None:
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+            process.wait(timeout=2)
+        except (ProcessLookupError, subprocess.TimeoutExpired):
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+            process.wait(timeout=5)
+
+    def _submit(self, state: dict[str, Any]) -> None:
+        key = state.get("decision_idempotency_key")
+        if not isinstance(key, str):
+            key = "decision-" + uuid.uuid4().hex
+            state["decision_idempotency_key"] = key
+            self._save_state(state)
+        response = self._request(
+            f"central/jobs/{state['job_id']}/decision", "decision",
+            {"lease_token": state["lease_token"],
+             "claim_idempotency_key": state["claim_idempotency_key"],
+             "decision": state["decision"]}, key,
+        )
+        if response.get("accepted") is not True or response.get("completed") is not True:
+            raise TriggerError("decision_not_confirmed")
+        self._clear_state()
+
+    def _fail(self, state: dict[str, Any], category: str) -> None:
+        key = state.get("fail_idempotency_key")
+        if not isinstance(key, str):
+            key = "fail-" + uuid.uuid4().hex
+            state["fail_idempotency_key"] = key
+            self._save_state(state)
+        self._request(
+            f"central/jobs/{state['job_id']}/fail", "fail",
+            {"lease_token": state["lease_token"],
+             "claim_idempotency_key": state["claim_idempotency_key"],
+             "category": category}, key,
+        )
+        self._clear_state()
+
+    def run_one(self) -> bool:
+        self.health_check()
+        state = self._load_state()
+        if state is None:
+            state = self._claim()
+            if state is None:
+                return False
+        try:
+            self._heartbeat(state, state.get("recovery_heartbeat_key") or "recover-" + uuid.uuid4().hex)
+            if state.get("phase") == "CLAIMED":
+                event = Event.from_dict(state.get("event"))
+                decision = self._exec(event, state)
+                state["phase"] = "DECIDED"
+                state["decision"] = decision
+                self._save_state(state)
+            if state.get("phase") != "DECIDED" or not isinstance(state.get("decision"), dict):
+                raise TriggerError("invalid_runner_state")
+            self._submit(state)
+            return True
+        except (ConfigurationError, TriggerError, ValidationError) as exc:
+            category = type(exc).__name__.upper()
+            try:
+                self._fail(state, category)
+            except Exception:
+                LOG.warning("remote_job_retained category=TRANSPORT_UNAVAILABLE")
+            raise
+
+    def run_forever(self) -> None:
+        while not self._stop.is_set():
+            try:
+                worked = self.run_one()
+            except Exception as exc:
+                LOG.warning("remote_cycle_failed category=%s", type(exc).__name__)
+                worked = False
+            self._stop.wait(0 if worked else self.config.poll_seconds)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True, type=Path)
+    parser.add_argument("--once", action="store_true")
+    parser.add_argument("--health", action="store_true")
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s")
+    try:
+        runner = RemoteCodexRunner(RunnerConfig.from_file(args.config))
+        if args.health:
+            runner.health_check()
+            print("REMOTE_CODEX_BRAIN_HEALTH=PASS")
+            return 0
+        if args.once:
+            runner.run_one()
+            return 0
+        runner.run_forever()
+        return 0
+    except Exception as exc:
+        reason = str(exc)
+        safe = reason if reason.replace("_", "").isalnum() and len(reason) <= 64 else type(exc).__name__
+        print(f"REMOTE_CODEX_BRAIN=BLOCKED reason={safe}")
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

remote-runner/install_remote_codex_brain.sh

@@ -0,0 +1,155 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+umask 0077
+
+SOURCE_ROOT="${SOURCE_ROOT:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}"
+CONTRACT_COMMIT="${CONTRACT_COMMIT:-cbe33ecab8e9794ba5f318bea43aeaf447a56fbc}"
+INSTALL_ROOT=/opt/aios-remote-codex-brain
+CONFIG_ROOT=/etc/aios-remote-codex-brain
+STATE_ROOT=/var/lib/aios-remote-codex-brain
+SERVICE_USER=aios-remote-codex
+CODEX_BINARY=/usr/bin/codex
+CODEX_HOME="$STATE_ROOT/.codex"
+CONTROLLER="$INSTALL_ROOT/controller-readonly"
+RUNNER_ID="codex-brain-192-168-88-6"
+EXPECTED_CODEX_VERSION=0.144.6
+
+fail() { printf 'STATUS=%s\nAUTOMATION_CLOSED=NO\n' "$1" >&2; exit 2; }
+[[ "$(id -u)" -eq 0 ]] || fail OWNER_INSTALL_REQUIRED
+for command in python3 node npm runuser systemctl systemd-analyze ssh sha256sum; do
+  command -v "$command" >/dev/null 2>&1 || fail "MISSING_COMMAND_${command}"
+done
+
+id "$SERVICE_USER" >/dev/null 2>&1 || useradd --system --home-dir "$STATE_ROOT" --shell /usr/sbin/nologin "$SERVICE_USER"
+install -d -o root -g root -m 0755 "$INSTALL_ROOT" "$INSTALL_ROOT/releases"
+install -d -o root -g "$SERVICE_USER" -m 0750 "$CONFIG_ROOT"
+install -d -o "$SERVICE_USER" -g "$SERVICE_USER" -m 0700 "$STATE_ROOT" "$CODEX_HOME" "$STATE_ROOT/runtime"
+
+[[ -e "$CODEX_BINARY" ]] || fail BLOCKED_CODEX_CLI_NOT_INSTALLED
+resolved="$(readlink -f "$CODEX_BINARY")"
+case "$resolved" in
+  /usr/lib/node_modules/@openai/codex/*|/usr/local/lib/node_modules/@openai/codex/*|/opt/openai/codex/*) ;;
+  *) fail BLOCKED_UNOFFICIAL_CODEX_CLI ;;
+esac
+python3 - "$CODEX_BINARY" "$resolved" <<'PY' || exit 2
+import os,stat,sys
+launcher=os.lstat(sys.argv[1]); target=os.lstat(sys.argv[2])
+if launcher.st_uid != 0: raise SystemExit(2)
+if not stat.S_ISLNK(launcher.st_mode) and launcher.st_mode & (stat.S_IWGRP|stat.S_IWOTH): raise SystemExit(2)
+if target.st_uid != 0 or target.st_mode & (stat.S_IWGRP|stat.S_IWOTH): raise SystemExit(2)
+if not stat.S_ISREG(target.st_mode) or not os.access(sys.argv[2],os.X_OK): raise SystemExit(2)
+PY
+package_root="$(npm root -g)/@openai/codex"
+[[ -f "$package_root/package.json" && ! -L "$package_root/package.json" ]] || fail BLOCKED_UNOFFICIAL_CODEX_CLI
+[[ "$(node -e "console.log(require('$package_root/package.json').name)")" = '@openai/codex' ]] || fail BLOCKED_UNOFFICIAL_CODEX_CLI
+[[ "$(node -e "console.log(require('$package_root/package.json').version)")" = "$EXPECTED_CODEX_VERSION" ]] || fail BLOCKED_CODEX_CLI_VERSION_INCOMPATIBLE
+mapfile -d '' native_targets < <(find "$package_root/node_modules/@openai" -path '*/vendor/*/bin/codex' -type f -perm /111 -print0)
+[[ "${#native_targets[@]}" -eq 1 ]] || fail BLOCKED_UNOFFICIAL_CODEX_CLI
+python3 - "${native_targets[0]}" <<'PY' || exit 2
+import os,stat,sys
+s=os.lstat(sys.argv[1])
+if not stat.S_ISREG(s.st_mode) or s.st_uid != 0 or s.st_mode & (stat.S_IWGRP|stat.S_IWOTH) or not os.access(sys.argv[1],os.X_OK): raise SystemExit(2)
+PY
+
+help_file="$(mktemp)"; trap 'rm -f "$help_file"' EXIT
+runuser -u "$SERVICE_USER" -- env -i HOME="$STATE_ROOT" CODEX_HOME="$CODEX_HOME" PATH=/usr/bin:/bin \
+  "$CODEX_BINARY" exec --help >"$help_file" 2>&1 || fail BLOCKED_CODEX_CLI_VERSION_INCOMPATIBLE
+for flag in --json --output-schema --output-last-message --sandbox --config --cd --ephemeral; do
+  grep -Fq -- "$flag" "$help_file" || fail BLOCKED_CODEX_CLI_VERSION_INCOMPATIBLE
+done
+rm -f "$help_file"; trap - EXIT
+
+AUTH_JSON="$CODEX_HOME/auth.json"
+[[ -e "$AUTH_JSON" ]] || fail BLOCKED_REMOTE_CODEX_AUTHENTICATION_REQUIRED
+python3 - "$AUTH_JSON" "$SERVICE_USER" <<'PY' || exit 2
+import os,pwd,stat,sys
+s=os.lstat(sys.argv[1]); uid=pwd.getpwnam(sys.argv[2]).pw_uid
+if stat.S_ISLNK(s.st_mode) or not stat.S_ISREG(s.st_mode) or s.st_uid != uid or stat.S_IMODE(s.st_mode) != 0o600: raise SystemExit(2)
+PY
+status_file="$(mktemp)"; trap 'rm -f "$status_file"' EXIT
+runuser -u "$SERVICE_USER" -- env -i HOME="$STATE_ROOT" CODEX_HOME="$CODEX_HOME" PATH=/usr/bin:/bin \
+  "$CODEX_BINARY" login status >"$status_file" 2>&1 || fail BLOCKED_REMOTE_CODEX_AUTHENTICATION_REQUIRED
+grep -Eqi 'chatgpt' "$status_file" || fail BLOCKED_REMOTE_CODEX_AUTHENTICATION_REQUIRED
+grep -Eqi 'one[ -]?api|api key' "$status_file" && fail BLOCKED_REMOTE_CODEX_AUTHENTICATION_REQUIRED
+rm -f "$status_file"; trap - EXIT
+
+for required in \
+  "$CONFIG_ROOT/runner.token" \
+  "$CONFIG_ROOT/tunnel_ed25519" \
+  "$CONFIG_ROOT/known_hosts"; do
+  [[ -f "$required" && ! -L "$required" ]] || fail BLOCKED_REMOTE_CODEX_TUNNEL_UNAVAILABLE
+  chown "$SERVICE_USER:$SERVICE_USER" "$required"
+  chmod 0600 "$required"
+done
+
+for required in \
+  remote-runner/aios_remote_codex_brain.py \
+  remote-runner/verify_remote_codex_brain.sh \
+  services/aios_orchestrator_gateway/__init__.py \
+  systemd/aios-remote-codex-brain.service \
+  systemd/aios-remote-codex-brain-health.service \
+  systemd/aios-remote-codex-brain-health.timer \
+  systemd/aios-remote-codex-tunnel.service \
+  .agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml; do
+  [[ -f "$SOURCE_ROOT/$required" && ! -L "$SOURCE_ROOT/$required" ]] || fail INVALID_SOURCE
+done
+
+digest="$(find "$SOURCE_ROOT/remote-runner" "$SOURCE_ROOT/services/aios_orchestrator_gateway" "$SOURCE_ROOT/systemd" -type f -print0 | sort -z | xargs -0 sha256sum | sha256sum | awk '{print $1}')"
+release="$INSTALL_ROOT/releases/$digest"
+if [[ ! -e "$release" ]]; then
+  install -d -o root -g root -m 0755 "$release" "$release/remote-runner" "$release/services" "$release/systemd"
+  cp -a "$SOURCE_ROOT/remote-runner/." "$release/remote-runner/"
+  cp -a "$SOURCE_ROOT/services/aios_orchestrator_gateway" "$release/services/"
+  cp -a "$SOURCE_ROOT/systemd/." "$release/systemd/"
+  find "$release" -type d -exec chmod 0755 {} +
+  find "$release" -type f -exec chmod 0644 {} +
+  chmod 0755 "$release/remote-runner/"*.py "$release/remote-runner/"*.sh
+  chown -R root:root "$release"
+fi
+ln -sfn "$release" "$INSTALL_ROOT/.current-$digest"
+mv -Tf "$INSTALL_ROOT/.current-$digest" "$INSTALL_ROOT/current"
+
+rm -rf "$CONTROLLER"
+install -d -o root -g root -m 0555 "$CONTROLLER/.agent/policies"
+install -o root -g root -m 0444 \
+  "$SOURCE_ROOT/.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml" \
+  "$CONTROLLER/.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml"
+
+cat >"$CONFIG_ROOT/config.json" <<EOF
+{
+  "gateway_url":"http://127.0.0.1:28920/v1/orchestrator",
+  "runner_id":"$RUNNER_ID",
+  "token_file":"$CONFIG_ROOT/runner.token",
+  "contract_commit":"$CONTRACT_COMMIT",
+  "codex_binary":"$CODEX_BINARY",
+  "codex_home":"$CODEX_HOME",
+  "controller_readonly":"$CONTROLLER",
+  "state_file":"$STATE_ROOT/runner-state.json",
+  "timeout_seconds":300,
+  "poll_seconds":5,
+  "heartbeat_seconds":30
+}
+EOF
+chown root:"$SERVICE_USER" "$CONFIG_ROOT/config.json"
+chmod 0640 "$CONFIG_ROOT/config.json"
+
+cat >"$CONFIG_ROOT/tunnel.env" <<'EOF'
+AIOS_GATEWAY_SSH_HOST=hw.imruyi.top
+AIOS_GATEWAY_SSH_PORT=22
+AIOS_GATEWAY_SSH_USER=aios-remote-codex-tunnel
+EOF
+chown root:"$SERVICE_USER" "$CONFIG_ROOT/tunnel.env"
+chmod 0640 "$CONFIG_ROOT/tunnel.env"
+
+for unit in aios-remote-codex-brain.service aios-remote-codex-brain-health.service aios-remote-codex-brain-health.timer aios-remote-codex-tunnel.service; do
+  install -o root -g root -m 0644 "$SOURCE_ROOT/systemd/$unit" "/etc/systemd/system/$unit"
+done
+systemd-analyze verify \
+  /etc/systemd/system/aios-remote-codex-brain.service \
+  /etc/systemd/system/aios-remote-codex-brain-health.service \
+  /etc/systemd/system/aios-remote-codex-brain-health.timer \
+  /etc/systemd/system/aios-remote-codex-tunnel.service
+systemctl daemon-reload
+systemctl enable --now aios-remote-codex-tunnel.service
+systemctl enable --now aios-remote-codex-brain.service aios-remote-codex-brain-health.timer
+printf 'REMOTE_CODEX_BRAIN_INSTALL=PASS\nRELEASE_DIGEST=%s\n' "$digest"

remote-runner/verify_remote_codex_brain.sh

@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+set -euo pipefail
+umask 0077
+
+CONFIG=/etc/aios-remote-codex-brain/config.json
+CURRENT=/opt/aios-remote-codex-brain/current
+[[ -f "$CONFIG" && ! -L "$CONFIG" ]] || { echo 'REMOTE_CODEX_VERIFY=BLOCKED'; exit 1; }
+[[ -L "$CURRENT" ]] || { echo 'REMOTE_CODEX_VERIFY=BLOCKED'; exit 1; }
+resolved="$(readlink -f "$CURRENT")"
+case "$resolved" in /opt/aios-remote-codex-brain/releases/*) ;; *) echo 'REMOTE_CODEX_VERIFY=BLOCKED'; exit 1;; esac
+systemctl is-active --quiet aios-remote-codex-tunnel.service
+systemctl is-active --quiet aios-remote-codex-brain.service
+systemctl is-enabled --quiet aios-remote-codex-brain-health.timer
+runuser -u aios-remote-codex -- env -i \
+  HOME=/var/lib/aios-remote-codex-brain \
+  CODEX_HOME=/var/lib/aios-remote-codex-brain/.codex \
+  PATH=/usr/bin:/bin \
+  PYTHONPATH="$CURRENT" \
+  /usr/bin/python3 "$CURRENT/remote-runner/aios_remote_codex_brain.py" --config "$CONFIG" --health
+printf 'REMOTE_CODEX_VERIFY=PASS\n'

scripts/install_aios_orchestrator_gateway.sh

@@ -43,15 +43,16 @@ for path in \
   "$SOURCE_ROOT/systemd/aios-codex-central-decision.service" \
   "$SOURCE_ROOT/systemd/aios-codex-central-health.service" \
   "$SOURCE_ROOT/systemd/aios-codex-central-health.timer" \
+  "$SOURCE_ROOT/automation/remote_codex_health_probe.py" \
   "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json"; do
   [[ -e "$path" ]] || fail "MISSING_SOURCE_${path}"
 done
 
 PYCACHE="/tmp/aios-orchestrator-install-pycache-$$"
 rm -rf "$PYCACHE"
-PYTHONPYCACHEPREFIX="$PYCACHE" python3 -m compileall -q "$SOURCE_ROOT/services" "$SOURCE_ROOT/scripts"
+PYTHONPYCACHEPREFIX="$PYCACHE" python3 -m compileall -q "$SOURCE_ROOT/services" "$SOURCE_ROOT/scripts" "$SOURCE_ROOT/automation"
 PYTHONDONTWRITEBYTECODE=1 PYTHONPATH="$SOURCE_ROOT" \
-  python3 -m unittest -v tests.test_aios_orchestrator_gateway
+  python3 -m unittest -v tests.test_aios_orchestrator_gateway tests.test_aios_remote_codex
 rm -rf "$PYCACHE"
 for script in "$SOURCE_ROOT"/scripts/*.sh; do bash -n "$script"; done
 python3 -m json.tool "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json" >/dev/null
@@ -92,7 +93,9 @@ SOURCE_FILES=(
   scripts/install_codex_local_central.sh
   automation/codex_central_decision.py
   automation/codex_central_health_probe.py
+  automation/remote_codex_health_probe.py
   tests/test_aios_orchestrator_gateway.py
+  tests/test_aios_remote_codex.py
   tests/test_aios_orchestrator_gateway_rollback.sh
   tests/test_aios_callback_recovery.py
   workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md
@@ -201,6 +204,10 @@ if config_path.is_file() and not config_path.is_symlink():
         (config.get("responses_fallback") or {}).get("api_key_file"),
         (config.get("openclaw_sync") or {}).get("bearer_token_file"),
     ]
+    candidates.extend(
+        item.get("token_file") for item in (config.get("remote_codex") or {}).get("runners", [])
+        if isinstance(item, dict)
+    )
     names.extend(name for name in candidates if isinstance(name, str) and name)
 for name in dict.fromkeys(names):
     path = pathlib.Path(name)
@@ -250,6 +257,9 @@ if cfg.responses_api_key_file:
     cfg.responses_api_key()
 if cfg.openclaw_sync_endpoint:
     cfg.openclaw_sync_token()
+for runner in cfg.remote_runner_credentials:
+    for scope in runner.scopes:
+        cfg.remote_runner_token(runner.runner_id, scope)
 print("CONFIG_VALIDATE=PASS")
 PY
 

services/aios_orchestrator_gateway/config.py

@@ -10,6 +10,16 @@ from .errors import ConfigurationError
 from .securefs import read_secure_bytes, read_secure_json, verify_secure_file
 
 
+REMOTE_RUNNER_SCOPES = frozenset({"claim", "heartbeat", "decision", "fail"})
+
+
+@dataclass(frozen=True)
+class RemoteRunnerCredential:
+    runner_id: str
+    token_file: Path
+    scopes: frozenset[str]
+
+
 @dataclass(frozen=True)
 class Config:
     host: str
@@ -45,6 +55,9 @@ class Config:
     git_poll_bare_repo: Path | None
     git_poll_ref_prefixes: tuple[str, ...]
     git_poll_interval_seconds: int
+    remote_request_max_bytes: int
+    remote_job_lease_seconds: int
+    remote_runner_credentials: tuple[RemoteRunnerCredential, ...]
 
     @classmethod
     def from_file(cls, path: Path) -> "Config":
@@ -58,7 +71,7 @@ class Config:
             "central_channel", "codex_local", "polling", "loop_controls",
             "openclaw_sync", "git_poll",
         }
-        if set(raw) != expected:
+        if frozenset(raw) not in {frozenset(expected), frozenset(expected | {"remote_codex"})}:
             raise ConfigurationError("invalid_config_fields")
 
         def path_value(key: str) -> Path:
@@ -96,13 +109,55 @@ class Config:
         loop = raw.get("loop_controls")
         sync = raw.get("openclaw_sync")
         git_poll = raw.get("git_poll")
-        if not all(isinstance(x, dict) for x in (workspace, responses, codex_local, polling, loop, sync, git_poll)):
+        remote_codex = raw.get("remote_codex", {
+            "request_max_bytes": 262144,
+            "job_lease_seconds": 120,
+            "runners": [],
+        })
+        if not all(isinstance(x, dict) for x in (workspace, responses, codex_local, polling, loop, sync, git_poll, remote_codex)):
             raise ConfigurationError("invalid_nested_config")
 
         central_channel = raw.get("central_channel")
-        if central_channel not in {"codex_local", "workspace_agent", "responses_api"}:
+        if central_channel not in {"codex_local", "remote_codex", "workspace_agent", "responses_api"}:
             raise ConfigurationError("invalid_central_channel")
 
+        if set(remote_codex) != {"request_max_bytes", "job_lease_seconds", "runners"}:
+            raise ConfigurationError("invalid_remote_codex_fields")
+        remote_request_max = integer(remote_codex, "request_max_bytes", 1024, 1_048_576)
+        remote_lease = integer(remote_codex, "job_lease_seconds", 30, 3600)
+        raw_runners = remote_codex.get("runners")
+        if not isinstance(raw_runners, list) or len(raw_runners) > 32:
+            raise ConfigurationError("invalid_remote_runners")
+        runner_credentials: list[RemoteRunnerCredential] = []
+        seen_runners: set[str] = set()
+        for item in raw_runners:
+            if not isinstance(item, dict) or set(item) != {"runner_id", "token_file", "scopes"}:
+                raise ConfigurationError("invalid_remote_runner")
+            runner_id = item.get("runner_id")
+            token_path = item.get("token_file")
+            scopes = item.get("scopes")
+            if (
+                not isinstance(runner_id, str) or not runner_id
+                or len(runner_id) > 128
+                or any(ch not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._:-" for ch in runner_id)
+                or runner_id in seen_runners
+            ):
+                raise ConfigurationError("invalid_remote_runner_id")
+            if not isinstance(token_path, str) or not Path(token_path).is_absolute():
+                raise ConfigurationError("invalid_remote_runner_token_file")
+            if (
+                not isinstance(scopes, list) or not scopes
+                or any(not isinstance(scope, str) or scope not in REMOTE_RUNNER_SCOPES for scope in scopes)
+                or len(scopes) != len(set(scopes))
+            ):
+                raise ConfigurationError("invalid_remote_runner_scopes")
+            seen_runners.add(runner_id)
+            runner_credentials.append(RemoteRunnerCredential(
+                runner_id=runner_id, token_file=Path(token_path), scopes=frozenset(scopes)
+            ))
+        if central_channel == "remote_codex" and not runner_credentials:
+            raise ConfigurationError("remote_codex_runner_required")
+
         trigger_id = workspace.get("trigger_id")
         workspace_token = workspace.get("access_token_file")
         if trigger_id is not None:
@@ -128,12 +183,15 @@ class Config:
 
         codex_queue = codex_local.get("queue_dir")
         codex_timeout = integer(codex_local, "timeout_seconds", 30, 3600)
-        if central_channel == "codex_local":
+        if central_channel in {"codex_local", "remote_codex"}:
             if trigger_id is not None or responses_key_file is not None:
                 raise ConfigurationError("parked_channel_must_be_disabled")
-            if not isinstance(codex_queue, str) or not Path(codex_queue).is_absolute():
-                raise ConfigurationError("invalid_codex_queue_dir")
-            codex_queue_dir = Path(codex_queue)
+            if central_channel == "codex_local":
+                if not isinstance(codex_queue, str) or not Path(codex_queue).is_absolute():
+                    raise ConfigurationError("invalid_codex_queue_dir")
+                codex_queue_dir = Path(codex_queue)
+            else:
+                codex_queue_dir = None
         else:
             codex_queue_dir = None
             if central_channel == "workspace_agent" and trigger_id is None:
@@ -206,6 +264,9 @@ class Config:
             git_poll_bare_repo=git_bare_repo,
             git_poll_ref_prefixes=git_prefixes,
             git_poll_interval_seconds=integer(git_poll, "interval_seconds", 5, 3600),
+            remote_request_max_bytes=remote_request_max,
+            remote_job_lease_seconds=remote_lease,
+            remote_runner_credentials=tuple(runner_credentials),
         )
 
     def _secret(self, path: Path) -> str:
@@ -235,3 +296,11 @@ class Config:
 
     def openclaw_sync_token(self) -> str | None:
         return self._secret(self.openclaw_sync_token_file) if self.openclaw_sync_token_file else None
+
+    def remote_runner_token(self, runner_id: str, scope: str) -> bytes:
+        for credential in self.remote_runner_credentials:
+            if credential.runner_id == runner_id:
+                if scope not in credential.scopes:
+                    raise ConfigurationError("remote_runner_scope_denied")
+                return self._secret(credential.token_file).encode("utf-8")
+        raise ConfigurationError("remote_runner_not_allowed")

services/aios_orchestrator_gateway/remote_codex.py

@@ -0,0 +1,461 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import hmac
+import json
+from pathlib import Path
+import secrets
+import sqlite3
+import threading
+import time
+from typing import Any
+
+from .errors import AuthenticationError, ConfigurationError, ConflictError, ValidationError
+from .models import Decision, Event
+
+
+REMOTE_SCOPES = frozenset({"claim", "heartbeat", "decision", "fail"})
+
+
+def require_idempotency_key(value: str | None) -> str:
+    if not isinstance(value, str) or not 16 <= len(value) <= 128:
+        raise ValidationError("invalid_idempotency_key")
+    if any(ch not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._:-" for ch in value):
+        raise ValidationError("invalid_idempotency_key")
+    return value
+
+
+def authenticate_runner(config: Any, runner_id: str | None, scope: str, header: str) -> bytes:
+    if scope not in REMOTE_SCOPES:
+        raise AuthenticationError("invalid_runner_scope")
+    if not isinstance(runner_id, str) or not runner_id:
+        raise AuthenticationError("missing_runner_id")
+    if not header.startswith("Bearer "):
+        raise AuthenticationError("missing_bearer")
+    try:
+        expected = config.remote_runner_token(runner_id, scope)
+    except ConfigurationError as exc:
+        raise AuthenticationError("runner_not_authorized") from exc
+    supplied = header[7:].encode("utf-8")
+    if not hmac.compare_digest(supplied, expected):
+        raise AuthenticationError("invalid_bearer")
+    return expected
+
+
+def validate_remote_decision(value: dict[str, Any], event: Event) -> Decision:
+    decision = Decision.from_dict(value)
+    if decision.contract_commit != event.contract_commit:
+        raise ValidationError("contract_version_mismatch")
+    if decision.event_id != event.event_id or decision.project_key != event.project_key:
+        raise ValidationError("event_identity_mismatch")
+    if decision.expected_state_version != event.state_version:
+        raise ValidationError("state_version_mismatch")
+    allowed = event.payload.get("allowed_work_unit_ids", [])
+    if not isinstance(allowed, list) or any(not isinstance(item, str) for item in allowed):
+        raise ValidationError("invalid_allowed_work_units")
+    if any(unit.id not in allowed for unit in decision.next_work_units):
+        raise ValidationError("work_unit_not_allowed")
+    return decision
+
+
+@dataclass(frozen=True)
+class RemoteLease:
+    job_id: str
+    runner_id: str
+    lease_version: int
+    lease_until: int
+    lease_token: str
+    event: dict[str, Any]
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "job_id": self.job_id,
+            "runner_id": self.runner_id,
+            "lease_version": self.lease_version,
+            "lease_until": self.lease_until,
+            "lease_token": self.lease_token,
+            "event": self.event,
+        }
+
+
+class RemoteCodexBroker:
+    """Durable, exact-once remote decision job broker.
+
+    Lease credentials are derived with HMAC from the scoped runner token and
+    are never logged or persisted in plaintext.
+    """
+
+    def __init__(self, database_path: Path, *, contract_commit: str,
+                 lease_seconds: int, max_retries: int):
+        self.database_path = Path(database_path)
+        self.contract_commit = contract_commit
+        self.lease_seconds = lease_seconds
+        self.max_retries = max_retries
+        self._local = threading.local()
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.database_path, timeout=30, isolation_level=None)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA busy_timeout=30000")
+        return connection
+
+    def _conn(self) -> sqlite3.Connection:
+        connection = getattr(self._local, "connection", None)
+        if connection is None:
+            connection = self._connect()
+            self._local.connection = connection
+        return connection
+
+    def close(self) -> None:
+        connection = getattr(self._local, "connection", None)
+        if connection is not None:
+            connection.close()
+            self._local.connection = None
+
+    def _initialize(self) -> None:
+        connection = self._connect()
+        try:
+            connection.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS remote_jobs (
+                    job_id TEXT PRIMARY KEY,
+                    event_id TEXT NOT NULL UNIQUE,
+                    event_digest TEXT NOT NULL,
+                    event_json TEXT NOT NULL,
+                    contract_commit TEXT NOT NULL,
+                    expected_state_version INTEGER NOT NULL,
+                    status TEXT NOT NULL,
+                    runner_id TEXT,
+                    lease_version INTEGER NOT NULL DEFAULT 0,
+                    lease_until INTEGER,
+                    attempts INTEGER NOT NULL DEFAULT 0,
+                    next_attempt_at INTEGER NOT NULL DEFAULT 0,
+                    failure_category TEXT,
+                    decision_digest TEXT,
+                    created_at INTEGER NOT NULL,
+                    updated_at INTEGER NOT NULL
+                );
+                CREATE INDEX IF NOT EXISTS idx_remote_jobs_claim
+                    ON remote_jobs(status, next_attempt_at, lease_until, created_at);
+                CREATE TABLE IF NOT EXISTS remote_requests (
+                    runner_id TEXT NOT NULL,
+                    scope TEXT NOT NULL,
+                    idempotency_key TEXT NOT NULL,
+                    request_digest TEXT NOT NULL,
+                    job_id TEXT,
+                    lease_version INTEGER,
+                    response_json TEXT NOT NULL,
+                    created_at INTEGER NOT NULL,
+                    PRIMARY KEY(runner_id, scope, idempotency_key)
+                );
+                CREATE TABLE IF NOT EXISTS remote_runners (
+                    runner_id TEXT PRIMARY KEY,
+                    last_seen_at INTEGER NOT NULL,
+                    last_scope TEXT NOT NULL
+                );
+                """
+            )
+        finally:
+            connection.close()
+
+    @staticmethod
+    def _digest(value: dict[str, Any]) -> str:
+        canonical = json.dumps(value, separators=(",", ":"), sort_keys=True).encode()
+        return hashlib.sha256(canonical).hexdigest()
+
+    @staticmethod
+    def _lease_token(token: bytes, job_id: str, runner_id: str,
+                     lease_version: int, claim_key: str) -> str:
+        message = f"{job_id}\0{runner_id}\0{lease_version}\0{claim_key}".encode()
+        return hmac.new(token, message, hashlib.sha256).hexdigest()
+
+    def enqueue(self, event: Event) -> tuple[bool, str]:
+        if event.contract_commit != self.contract_commit:
+            raise ConflictError("contract_version_mismatch")
+        now = int(time.time())
+        digest = event.digest()
+        connection = self._conn()
+        connection.execute("BEGIN IMMEDIATE")
+        try:
+            existing = connection.execute(
+                "SELECT event_digest FROM remote_jobs WHERE event_id=?", (event.event_id,)
+            ).fetchone()
+            if existing is not None:
+                if existing["event_digest"] != digest:
+                    raise ConflictError("remote_job_idempotency_conflict")
+                connection.execute("COMMIT")
+                return False, event.event_id
+            connection.execute(
+                """
+                INSERT INTO remote_jobs(
+                    job_id,event_id,event_digest,event_json,contract_commit,
+                    expected_state_version,status,next_attempt_at,created_at,updated_at
+                ) VALUES(?,?,?,?,?,?,'PENDING',?,?,?)
+                """,
+                (event.event_id, event.event_id, digest, event.canonical_json(),
+                 event.contract_commit, event.state_version, now, now, now),
+            )
+            connection.execute("COMMIT")
+            return True, event.event_id
+        except Exception:
+            connection.execute("ROLLBACK")
+            raise
+
+    def _touch_runner(self, connection: sqlite3.Connection, runner_id: str, scope: str, now: int) -> None:
+        connection.execute(
+            """
+            INSERT INTO remote_runners(runner_id,last_seen_at,last_scope) VALUES(?,?,?)
+            ON CONFLICT(runner_id) DO UPDATE SET
+                last_seen_at=excluded.last_seen_at,last_scope=excluded.last_scope
+            """, (runner_id, now, scope),
+        )
+
+    def _replay(self, connection: sqlite3.Connection, runner_id: str, scope: str,
+                key: str, digest: str) -> dict[str, Any] | None:
+        row = connection.execute(
+            "SELECT request_digest,response_json FROM remote_requests "
+            "WHERE runner_id=? AND scope=? AND idempotency_key=?",
+            (runner_id, scope, key),
+        ).fetchone()
+        if row is None:
+            return None
+        if row["request_digest"] != digest:
+            raise ConflictError("idempotency_payload_conflict")
+        return json.loads(row["response_json"])
+
+    def _record(self, connection: sqlite3.Connection, *, runner_id: str, scope: str,
+                key: str, digest: str, response: dict[str, Any], job_id: str | None,
+                lease_version: int | None, now: int) -> None:
+        connection.execute(
+            """
+            INSERT INTO remote_requests(
+                runner_id,scope,idempotency_key,request_digest,job_id,
+                lease_version,response_json,created_at
+            ) VALUES(?,?,?,?,?,?,?,?)
+            """,
+            (runner_id, scope, key, digest, job_id, lease_version,
+             json.dumps(response, separators=(",", ":"), sort_keys=True), now),
+        )
+
+    def claim(self, *, runner_id: str, idempotency_key: str,
+              runner_token: bytes) -> dict[str, Any]:
+        key = require_idempotency_key(idempotency_key)
+        request = {"runner_id": runner_id, "contract_commit": self.contract_commit}
+        digest = self._digest(request)
+        now = int(time.time())
+        connection = self._conn()
+        connection.execute("BEGIN IMMEDIATE")
+        try:
+            replay = self._replay(connection, runner_id, "claim", key, digest)
+            if replay is not None:
+                if replay == {"claimed": False}:
+                    connection.execute("COMMIT")
+                    return replay
+                row = connection.execute(
+                    "SELECT status,runner_id,lease_version,lease_until FROM remote_jobs WHERE job_id=?",
+                    (replay.get("job_id"),),
+                ).fetchone()
+                if row is not None and row["runner_id"] == runner_id and row["status"] == "LEASED":
+                    replay["lease_token"] = self._lease_token(
+                        runner_token, str(replay["job_id"]), runner_id,
+                        int(row["lease_version"]), key,
+                    )
+                    connection.execute("COMMIT")
+                    return replay
+                raise ConflictError("claim_replay_no_longer_active")
+
+            connection.execute(
+                """
+                UPDATE remote_jobs SET
+                    status=CASE WHEN attempts + 1 > ? THEN 'DEAD_LETTER' ELSE 'RETRY' END,
+                    attempts=attempts+1,runner_id=NULL,lease_until=NULL,
+                    failure_category='STALE_LEASE',next_attempt_at=?,updated_at=?
+                WHERE status='LEASED' AND lease_until < ?
+                """, (self.max_retries, now, now, now),
+            )
+            row = connection.execute(
+                """
+                SELECT * FROM remote_jobs
+                WHERE status IN ('PENDING','RETRY') AND next_attempt_at <= ?
+                  AND contract_commit=?
+                ORDER BY created_at LIMIT 1
+                """, (now, self.contract_commit),
+            ).fetchone()
+            self._touch_runner(connection, runner_id, "claim", now)
+            if row is None:
+                response = {"claimed": False}
+                self._record(connection, runner_id=runner_id, scope="claim", key=key,
+                             digest=digest, response=response, job_id=None,
+                             lease_version=None, now=now)
+                connection.execute("COMMIT")
+                return response
+            lease_version = int(row["lease_version"]) + 1
+            lease_until = now + self.lease_seconds
+            connection.execute(
+                """
+                UPDATE remote_jobs SET status='LEASED',runner_id=?,lease_version=?,
+                    lease_until=?,updated_at=? WHERE job_id=? AND status IN ('PENDING','RETRY')
+                """, (runner_id, lease_version, lease_until, now, row["job_id"]),
+            )
+            lease = RemoteLease(
+                job_id=str(row["job_id"]), runner_id=runner_id,
+                lease_version=lease_version, lease_until=lease_until,
+                lease_token=self._lease_token(runner_token, str(row["job_id"]), runner_id,
+                                              lease_version, key),
+                event=json.loads(row["event_json"]),
+            )
+            response = {"claimed": True, **lease.to_dict(), "claim_idempotency_key": key}
+            persisted_response = dict(response)
+            persisted_response.pop("lease_token")
+            self._record(connection, runner_id=runner_id, scope="claim", key=key,
+                         digest=digest, response=persisted_response, job_id=lease.job_id,
+                         lease_version=lease_version, now=now)
+            connection.execute("COMMIT")
+            return response
+        except Exception:
+            connection.execute("ROLLBACK")
+            raise
+
+    def _validate_lease(self, connection: sqlite3.Connection, *, job_id: str,
+                        runner_id: str, lease_token: str, claim_key: str,
+                        runner_token: bytes, allow_completed: bool = False) -> sqlite3.Row:
+        row = connection.execute("SELECT * FROM remote_jobs WHERE job_id=?", (job_id,)).fetchone()
+        if row is None:
+            raise ConflictError("remote_job_not_found")
+        if row["runner_id"] != runner_id:
+            raise ConflictError("remote_job_not_leased")
+        expected = self._lease_token(runner_token, job_id, runner_id,
+                                     int(row["lease_version"]), claim_key)
+        if not hmac.compare_digest(lease_token, expected):
+            raise AuthenticationError("invalid_lease_token")
+        if allow_completed and row["status"] == "COMPLETED":
+            return row
+        if row["status"] != "LEASED":
+            raise ConflictError("remote_job_not_leased")
+        if row["lease_until"] is None or int(row["lease_until"]) < int(time.time()):
+            raise ConflictError("remote_lease_expired")
+        return row
+
+    def heartbeat(self, *, job_id: str, runner_id: str, lease_token: str,
+                  claim_key: str, idempotency_key: str, runner_token: bytes) -> dict[str, Any]:
+        key = require_idempotency_key(idempotency_key)
+        request = {"job_id": job_id, "lease_token": lease_token, "claim_key": claim_key}
+        digest = self._digest(request); now = int(time.time())
+        connection = self._conn(); connection.execute("BEGIN IMMEDIATE")
+        try:
+            replay = self._replay(connection, runner_id, "heartbeat", key, digest)
+            if replay is not None:
+                connection.execute("COMMIT"); return replay
+            row = self._validate_lease(connection, job_id=job_id, runner_id=runner_id,
+                                       lease_token=lease_token, claim_key=claim_key,
+                                       runner_token=runner_token)
+            lease_until = now + self.lease_seconds
+            connection.execute("UPDATE remote_jobs SET lease_until=?,updated_at=? WHERE job_id=?",
+                               (lease_until, now, job_id))
+            self._touch_runner(connection, runner_id, "heartbeat", now)
+            response = {"accepted": True, "job_id": job_id,
+                        "lease_version": int(row["lease_version"]), "lease_until": lease_until}
+            self._record(connection, runner_id=runner_id, scope="heartbeat", key=key,
+                         digest=digest, response=response, job_id=job_id,
+                         lease_version=int(row["lease_version"]), now=now)
+            connection.execute("COMMIT"); return response
+        except Exception:
+            connection.execute("ROLLBACK"); raise
+
+    def event_for_submission(self, *, job_id: str, runner_id: str, lease_token: str,
+                             claim_key: str, runner_token: bytes) -> Event:
+        row = self._validate_lease(self._conn(), job_id=job_id, runner_id=runner_id,
+                                   lease_token=lease_token, claim_key=claim_key,
+                                   runner_token=runner_token, allow_completed=True)
+        return Event.from_dict(json.loads(row["event_json"]))
+
+    def complete(self, *, job_id: str, runner_id: str, idempotency_key: str,
+                 request_digest: str, decision_digest: str,
+                 state_version_after: int) -> dict[str, Any]:
+        key = require_idempotency_key(idempotency_key); now = int(time.time())
+        connection = self._conn(); connection.execute("BEGIN IMMEDIATE")
+        try:
+            replay = self._replay(connection, runner_id, "decision", key, request_digest)
+            if replay is not None:
+                connection.execute("COMMIT"); return replay
+            row = connection.execute("SELECT * FROM remote_jobs WHERE job_id=?", (job_id,)).fetchone()
+            if row is None:
+                raise ConflictError("remote_job_not_found")
+            if row["status"] == "COMPLETED":
+                if row["decision_digest"] != decision_digest:
+                    raise ConflictError("remote_decision_idempotency_conflict")
+            elif row["status"] == "LEASED" and row["runner_id"] == runner_id:
+                connection.execute(
+                    "UPDATE remote_jobs SET status='COMPLETED',decision_digest=?,lease_until=NULL,updated_at=? WHERE job_id=?",
+                    (decision_digest, now, job_id),
+                )
+            else:
+                raise ConflictError("remote_job_not_leased")
+            response = {"accepted": True, "job_id": job_id,
+                        "state_version": state_version_after, "completed": True}
+            self._record(connection, runner_id=runner_id, scope="decision", key=key,
+                         digest=request_digest, response=response, job_id=job_id,
+                         lease_version=int(row["lease_version"]), now=now)
+            self._touch_runner(connection, runner_id, "decision", now)
+            connection.execute("COMMIT"); return response
+        except Exception:
+            connection.execute("ROLLBACK"); raise
+
+    def fail(self, *, job_id: str, runner_id: str, lease_token: str, claim_key: str,
+             category: str, idempotency_key: str, runner_token: bytes,
+             retry_after: int) -> dict[str, Any]:
+        if not isinstance(category, str) or not category or len(category) > 64 or not category.replace("_", "").isalnum():
+            raise ValidationError("invalid_failure_category")
+        key = require_idempotency_key(idempotency_key)
+        request = {"job_id": job_id, "lease_token": lease_token,
+                   "claim_key": claim_key, "category": category}
+        digest = self._digest(request); now = int(time.time())
+        connection = self._conn(); connection.execute("BEGIN IMMEDIATE")
+        try:
+            replay = self._replay(connection, runner_id, "fail", key, digest)
+            if replay is not None:
+                connection.execute("COMMIT"); return replay
+            row = self._validate_lease(connection, job_id=job_id, runner_id=runner_id,
+                                       lease_token=lease_token, claim_key=claim_key,
+                                       runner_token=runner_token)
+            attempts = int(row["attempts"]) + 1
+            status = "DEAD_LETTER" if attempts > self.max_retries else "RETRY"
+            connection.execute(
+                """
+                UPDATE remote_jobs SET status=?,attempts=?,failure_category=?,
+                    next_attempt_at=?,runner_id=NULL,lease_until=NULL,updated_at=? WHERE job_id=?
+                """, (status, attempts, category, now + max(0, retry_after), now, job_id),
+            )
+            response = {"accepted": True, "job_id": job_id, "status": status,
+                        "attempts": attempts}
+            self._record(connection, runner_id=runner_id, scope="fail", key=key,
+                         digest=digest, response=response, job_id=job_id,
+                         lease_version=int(row["lease_version"]), now=now)
+            self._touch_runner(connection, runner_id, "fail", now)
+            connection.execute("COMMIT"); return response
+        except Exception:
+            connection.execute("ROLLBACK"); raise
+
+    def health_snapshot(self, *, healthy_within_seconds: int = 300) -> dict[str, Any]:
+        now = int(time.time())
+        connection = self._conn()
+        rows = connection.execute(
+            "SELECT runner_id,last_seen_at,last_scope FROM remote_runners ORDER BY runner_id"
+        ).fetchall()
+        counts = {
+            str(row["status"]): int(row["count"])
+            for row in connection.execute(
+                "SELECT status,COUNT(*) AS count FROM remote_jobs GROUP BY status"
+            ).fetchall()
+        }
+        return {
+            "channel": "remote_codex",
+            "contract_commit": self.contract_commit,
+            "healthy_runners": sum(1 for row in rows if now - int(row["last_seen_at"]) <= healthy_within_seconds),
+            "runner_count": len(rows),
+            "job_counts": counts,
+        }

services/aios_orchestrator_gateway/service.py

@@ -27,6 +27,12 @@ from .errors import (
     ValidationError,
 )
 from .models import Decision, Event, ID_RE
+from .remote_codex import (
+    RemoteCodexBroker,
+    authenticate_runner,
+    require_idempotency_key,
+    validate_remote_decision,
+)
 from .securefs import ensure_secure_directory, ensure_shared_directory, read_secure_json
 from .store import Store
 from .sync import OpenClawSynchronizer, lifecycle_record
@@ -61,6 +67,14 @@ class Runtime:
         self.codex_queue = None
         if config.central_channel == "codex_local" and config.codex_queue_dir is not None:
             self.codex_queue = CodexLocalQueue(config.codex_queue_dir)
+        self.remote_codex = None
+        if config.central_channel == "remote_codex":
+            self.remote_codex = RemoteCodexBroker(
+                config.database_path,
+                contract_commit=config.contract_commit,
+                lease_seconds=config.remote_job_lease_seconds,
+                max_retries=config.max_retries,
+            )
         self.synchronizer = OpenClawSynchronizer(
             config.openclaw_sync_dir,
             endpoint=config.openclaw_sync_endpoint,
@@ -87,6 +101,8 @@ class Runtime:
         if self.worker.is_alive():
             self.worker.join(timeout=15)
         self.store.close()
+        if self.remote_codex is not None:
+            self.remote_codex.close()
 
     def submit_event(self, payload: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
         event = Event.from_dict(payload)
@@ -114,6 +130,53 @@ class Runtime:
         self._enqueue_decision_lifecycle(decision, version)
         return version
 
+    def submit_remote_decision(
+        self,
+        *,
+        job_id: str,
+        runner_id: str,
+        lease_token: str,
+        claim_key: str,
+        idempotency_key: str,
+        runner_token: bytes,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        if self.remote_codex is None:
+            raise ConflictError("remote_codex_not_selected")
+        event = self.remote_codex.event_for_submission(
+            job_id=job_id,
+            runner_id=runner_id,
+            lease_token=lease_token,
+            claim_key=claim_key,
+            runner_token=runner_token,
+        )
+        decision = validate_remote_decision(payload, event)
+        request_digest = hashlib.sha256(
+            json.dumps(
+                {
+                    "job_id": job_id,
+                    "runner_id": runner_id,
+                    "lease_token": lease_token,
+                    "claim_key": claim_key,
+                    "decision": decision.to_dict(),
+                },
+                separators=(",", ":"),
+                sort_keys=True,
+            ).encode()
+        ).hexdigest()
+        decision_digest = hashlib.sha256(
+            json.dumps(decision.to_dict(), separators=(",", ":"), sort_keys=True).encode()
+        ).hexdigest()
+        version = self.submit_decision(decision.to_dict())
+        return self.remote_codex.complete(
+            job_id=job_id,
+            runner_id=runner_id,
+            idempotency_key=idempotency_key,
+            request_digest=request_digest,
+            decision_digest=decision_digest,
+            state_version_after=version,
+        )
+
     def _loop(self) -> None:
         next_summary = 0.0
         while not self.stop_event.wait(self.config.poll_interval_seconds):
@@ -261,6 +324,16 @@ class Runtime:
                         callback_state="AWAITING_CALLBACK",
                         callback_deadline_after=self.config.callback_timeout_seconds,
                     )
+                elif self.remote_codex is not None:
+                    self.remote_codex.enqueue(event)
+                    self.store.update_event_trigger(
+                        event.event_id,
+                        status="AWAITING_DECISION",
+                        channel="remote_codex",
+                        remote_id=event.event_id,
+                        callback_state="AWAITING_CALLBACK",
+                        callback_deadline_after=self.config.callback_timeout_seconds,
+                    )
                 elif self.workspace is not None:
                     try:
                         self.store.update_event_trigger(
@@ -608,12 +681,12 @@ class Handler(BaseHTTPRequestHandler):
         self.end_headers()
         self.wfile.write(body)
 
-    def _body(self) -> dict[str, Any]:
+    def _body(self, max_bytes: int = 1_048_576) -> dict[str, Any]:
         try:
             length = int(self.headers.get("Content-Length", "0"))
         except ValueError as exc:
             raise ValidationError("invalid_content_length") from exc
-        if length <= 0 or length > 1_048_576:
+        if length <= 0 or length > max_bytes:
             raise ValidationError("invalid_body_size")
         try:
             value = json.loads(self.rfile.read(length))
@@ -630,6 +703,19 @@ class Handler(BaseHTTPRequestHandler):
         if not hmac.compare_digest(header[7:], expected):
             raise AuthenticationError("invalid_bearer")
 
+    def _remote_auth(self, scope: str) -> tuple[str, bytes, str]:
+        if self.runtime.remote_codex is None:
+            raise ConflictError("remote_codex_not_selected")
+        runner_id = self.headers.get("X-AIOS-Runner-ID")
+        token = authenticate_runner(
+            self.runtime.config,
+            runner_id,
+            scope,
+            self.headers.get("Authorization", ""),
+        )
+        key = require_idempotency_key(self.headers.get("Idempotency-Key"))
+        return str(runner_id), token, key
+
     def do_GET(self) -> None:
         path = urlparse(self.path).path.rstrip("/") or "/"
         try:
@@ -667,6 +753,64 @@ class Handler(BaseHTTPRequestHandler):
     def do_POST(self) -> None:
         path = urlparse(self.path).path.rstrip("/")
         try:
+            if path == "/v1/orchestrator/central/jobs/claim":
+                runner_id, token, key = self._remote_auth("claim")
+                payload = self._body(self.runtime.config.remote_request_max_bytes)
+                if set(payload) != {"contract_commit"}:
+                    raise ValidationError("invalid_claim_fields")
+                if payload.get("contract_commit") != self.runtime.config.contract_commit:
+                    raise ConflictError("contract_version_mismatch")
+                response = self.runtime.remote_codex.claim(
+                    runner_id=runner_id, idempotency_key=key, runner_token=token
+                )
+                self._reply(200, response)
+                return
+            prefix = "/v1/orchestrator/central/jobs/"
+            if path.startswith(prefix):
+                tail = path[len(prefix):]
+                job_id, separator, action = tail.partition("/")
+                if not separator or not ID_RE.fullmatch(job_id):
+                    raise ValidationError("invalid_remote_job_path")
+                if action not in {"heartbeat", "decision", "fail"}:
+                    self._reply(404, {"error": "not_found"})
+                    return
+                runner_id, token, key = self._remote_auth(action)
+                payload = self._body(self.runtime.config.remote_request_max_bytes)
+                lease_token = payload.get("lease_token")
+                claim_key = payload.get("claim_idempotency_key")
+                if not isinstance(lease_token, str) or len(lease_token) != 64:
+                    raise ValidationError("invalid_lease_token")
+                claim_key = require_idempotency_key(claim_key)
+                if action == "heartbeat":
+                    if set(payload) != {"lease_token", "claim_idempotency_key"}:
+                        raise ValidationError("invalid_heartbeat_fields")
+                    response = self.runtime.remote_codex.heartbeat(
+                        job_id=job_id, runner_id=runner_id,
+                        lease_token=lease_token, claim_key=claim_key,
+                        idempotency_key=key, runner_token=token,
+                    )
+                elif action == "decision":
+                    if set(payload) != {"lease_token", "claim_idempotency_key", "decision"}:
+                        raise ValidationError("invalid_remote_decision_fields")
+                    decision = payload.get("decision")
+                    if not isinstance(decision, dict):
+                        raise ValidationError("invalid_remote_decision")
+                    response = self.runtime.submit_remote_decision(
+                        job_id=job_id, runner_id=runner_id,
+                        lease_token=lease_token, claim_key=claim_key,
+                        idempotency_key=key, runner_token=token, payload=decision,
+                    )
+                else:
+                    if set(payload) != {"lease_token", "claim_idempotency_key", "category"}:
+                        raise ValidationError("invalid_fail_fields")
+                    response = self.runtime.remote_codex.fail(
+                        job_id=job_id, runner_id=runner_id,
+                        lease_token=lease_token, claim_key=claim_key,
+                        category=payload.get("category"), idempotency_key=key,
+                        runner_token=token, retry_after=self.runtime.config.retry_base_seconds,
+                    )
+                self._reply(202, response)
+                return
             if path == "/v1/orchestrator/events":
                 self._authenticate(self.runtime.config.inbound_token())
                 created, event = self.runtime.submit_event(self._body())

services/aios_orchestrator_gateway/store.py

@@ -676,7 +676,7 @@ class Store:
         else:
             public_status = "RECEIVED"
         channel = row["channel"]
-        if channel not in {"workspace_agent", "responses_api", "codex_local"}:
+        if channel not in {"workspace_agent", "responses_api", "codex_local", "remote_codex"}:
             channel = None
         version = row["state_version_after"]
         return {

systemd/aios-remote-codex-brain-health.service

@@ -0,0 +1,32 @@
+[Unit]
+Description=AIOS remote Codex brain metadata-only health check
+After=aios-remote-codex-tunnel.service
+Requires=aios-remote-codex-tunnel.service
+ConditionPathExists=/etc/aios-remote-codex-brain/config.json
+
+[Service]
+Type=oneshot
+User=aios-remote-codex
+Group=aios-remote-codex
+UMask=0077
+Environment=PYTHONPATH=/opt/aios-remote-codex-brain/current
+Environment=CODEX_HOME=/var/lib/aios-remote-codex-brain/.codex
+ExecStart=/usr/bin/python3 /opt/aios-remote-codex-brain/current/remote-runner/aios_remote_codex_brain.py --config /etc/aios-remote-codex-brain/config.json --health
+TimeoutStartSec=60
+NoNewPrivileges=yes
+PrivateDevices=yes
+PrivateTmp=yes
+ProtectHome=yes
+ProtectSystem=strict
+ReadOnlyPaths=/opt/aios-remote-codex-brain /etc/aios-remote-codex-brain /usr/lib/node_modules/@openai/codex
+ReadWritePaths=/var/lib/aios-remote-codex-brain
+InaccessiblePaths=/data /root /home /etc/ssh /etc/docker /var/run/docker.sock
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM

systemd/aios-remote-codex-brain-health.timer

@@ -0,0 +1,12 @@
+[Unit]
+Description=Periodically verify the AIOS remote Codex brain
+
+[Timer]
+OnBootSec=2min
+OnUnitActiveSec=5min
+RandomizedDelaySec=30
+Persistent=true
+Unit=aios-remote-codex-brain-health.service
+
+[Install]
+WantedBy=timers.target

systemd/aios-remote-codex-brain.service

@@ -0,0 +1,45 @@
+[Unit]
+Description=AIOS trusted remote Codex brain runner
+After=network-online.target aios-remote-codex-tunnel.service
+Wants=network-online.target
+Requires=aios-remote-codex-tunnel.service
+ConditionPathExists=/etc/aios-remote-codex-brain/config.json
+ConditionPathExists=/var/lib/aios-remote-codex-brain/.codex/auth.json
+
+[Service]
+Type=simple
+User=aios-remote-codex
+Group=aios-remote-codex
+UMask=0077
+Environment=PYTHONPATH=/opt/aios-remote-codex-brain/current
+Environment=CODEX_HOME=/var/lib/aios-remote-codex-brain/.codex
+ExecStart=/usr/bin/python3 /opt/aios-remote-codex-brain/current/remote-runner/aios_remote_codex_brain.py --config /etc/aios-remote-codex-brain/config.json
+Restart=on-failure
+RestartSec=10
+TimeoutStopSec=20
+NoNewPrivileges=yes
+PrivateDevices=yes
+PrivateTmp=yes
+ProtectClock=yes
+ProtectControlGroups=yes
+ProtectHome=yes
+ProtectHostname=yes
+ProtectKernelLogs=yes
+ProtectKernelModules=yes
+ProtectKernelTunables=yes
+ProtectSystem=strict
+ReadOnlyPaths=/opt/aios-remote-codex-brain /etc/aios-remote-codex-brain /usr/lib/node_modules/@openai/codex
+ReadWritePaths=/var/lib/aios-remote-codex-brain
+InaccessiblePaths=/data /root /home /etc/ssh /etc/docker /var/run/docker.sock
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+
+[Install]
+WantedBy=multi-user.target

systemd/aios-remote-codex-tunnel.service

@@ -0,0 +1,35 @@
+[Unit]
+Description=Approved loopback-only tunnel to AIOS Orchestrator Gateway
+After=network-online.target
+Wants=network-online.target
+ConditionPathExists=/etc/aios-remote-codex-brain/tunnel.env
+ConditionPathExists=/etc/aios-remote-codex-brain/tunnel_ed25519
+ConditionPathExists=/etc/aios-remote-codex-brain/known_hosts
+
+[Service]
+Type=simple
+User=aios-remote-codex
+Group=aios-remote-codex
+EnvironmentFile=/etc/aios-remote-codex-brain/tunnel.env
+ExecStart=/usr/bin/ssh -NT -L 127.0.0.1:28920:127.0.0.1:18920 -p ${AIOS_GATEWAY_SSH_PORT} -i /etc/aios-remote-codex-brain/tunnel_ed25519 -o BatchMode=yes -o IdentitiesOnly=yes -o ExitOnForwardFailure=yes -o ServerAliveInterval=15 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=yes -o UserKnownHostsFile=/etc/aios-remote-codex-brain/known_hosts ${AIOS_GATEWAY_SSH_USER}@${AIOS_GATEWAY_SSH_HOST}
+Restart=always
+RestartSec=5
+NoNewPrivileges=yes
+PrivateDevices=yes
+PrivateTmp=yes
+ProtectHome=yes
+ProtectSystem=strict
+ReadOnlyPaths=/etc/aios-remote-codex-brain
+InaccessiblePaths=/data /root /home /etc/docker /var/run/docker.sock
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+
+[Install]
+WantedBy=multi-user.target

tests/test_aios_remote_codex.py

@@ -0,0 +1,377 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import logging
+import os
+from pathlib import Path
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+import urllib.request
+
+from services.aios_orchestrator_gateway.errors import (
+    AuthenticationError, ConfigurationError, ConflictError, TriggerError, ValidationError,
+)
+from services.aios_orchestrator_gateway.models import Decision, Event
+from services.aios_orchestrator_gateway.remote_codex import (
+    RemoteCodexBroker, authenticate_runner, validate_remote_decision,
+)
+from services.aios_orchestrator_gateway.config import Config
+from services.aios_orchestrator_gateway.service import Handler, Runtime, ThreadingHTTPServer
+
+
+CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
+ROOT = Path(__file__).resolve().parents[1]
+RUNNER_PATH = ROOT / "remote-runner/aios_remote_codex_brain.py"
+SPEC = importlib.util.spec_from_file_location("aios_remote_codex_brain", RUNNER_PATH)
+assert SPEC and SPEC.loader
+RUNNER = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = RUNNER
+SPEC.loader.exec_module(RUNNER)
+
+
+def event_payload(**changes):
+    value = {
+        "schema_version": 1,
+        "project_key": "v2r1-personal-weixin-production",
+        "contract_commit": CONTRACT,
+        "event_id": "evt-remote-001",
+        "source_agent": "system",
+        "event_type": "central_decision_requested",
+        "work_unit_id": "A0-REMOTE",
+        "status": "READY",
+        "state_version": 0,
+        "implementation_commit": None,
+        "result_commit": None,
+        "blocker": None,
+        "evidence_refs": [],
+        "payload": {"allowed_work_unit_ids": []},
+    }
+    value.update(changes)
+    return value
+
+
+def decision_payload(**changes):
+    value = {
+        "schema_version": 1,
+        "project_key": "v2r1-personal-weixin-production",
+        "contract_commit": CONTRACT,
+        "event_id": "evt-remote-001",
+        "expected_state_version": 0,
+        "verdict": "BLOCKED",
+        "state_updates": [],
+        "next_work_units": [],
+        "requires_human_input": False,
+        "stop_reason": "evidence_required",
+    }
+    value.update(changes)
+    return value
+
+
+class BrokerTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.db = Path(self.temp.name) / "state.sqlite3"
+        self.broker = RemoteCodexBroker(
+            self.db, contract_commit=CONTRACT, lease_seconds=60, max_retries=1
+        )
+        self.event = Event.from_dict(event_payload())
+        self.token = b"runner-secret-not-real"
+        self.broker.enqueue(self.event)
+
+    def tearDown(self):
+        self.broker.close()
+        self.temp.cleanup()
+
+    def claim(self, key=None):
+        key = key or ("claim-" + "0" * 16)
+        return self.broker.claim(
+            runner_id="brain-88-6", idempotency_key=key, runner_token=self.token
+        )
+
+    def test_remote_runner_allowlist(self):
+        config = SimpleNamespace(remote_runner_token=lambda runner, scope: self.token if runner == "brain-88-6" and scope == "claim" else (_ for _ in ()).throw(ConfigurationError("denied")))
+        self.assertEqual(authenticate_runner(config, "brain-88-6", "claim", "Bearer runner-secret-not-real"), self.token)
+        with self.assertRaises(AuthenticationError):
+            authenticate_runner(config, "other", "claim", "Bearer runner-secret-not-real")
+        with self.assertRaises(AuthenticationError):
+            authenticate_runner(config, "brain-88-6", "decision", "Bearer runner-secret-not-real")
+
+    def test_job_claim_exact_once(self):
+        first = self.claim()
+        second = self.claim()
+        self.assertEqual(first, second)
+        stored = self.broker._conn().execute(
+            "SELECT response_json FROM remote_requests WHERE scope='claim'"
+        ).fetchone()["response_json"]
+        self.assertNotIn(first["lease_token"], stored)
+
+    def test_duplicate_decision_idempotent(self):
+        lease = self.claim()
+        response = self.broker.complete(
+            job_id=lease["job_id"], runner_id="brain-88-6",
+            idempotency_key="decision-" + "0" * 16, request_digest="a" * 64,
+            decision_digest="b" * 64, state_version_after=1,
+        )
+        replay = self.broker.complete(
+            job_id=lease["job_id"], runner_id="brain-88-6",
+            idempotency_key="decision-" + "0" * 16, request_digest="a" * 64,
+            decision_digest="b" * 64, state_version_after=1,
+        )
+        self.assertEqual(response, replay)
+
+    def test_stale_state_rejected(self):
+        with self.assertRaisesRegex(ValidationError, "state_version_mismatch"):
+            validate_remote_decision(
+                decision_payload(expected_state_version=9), self.event
+            )
+
+    def test_contract_mismatch_rejected(self):
+        with self.assertRaisesRegex(ValidationError, "contract_version_mismatch"):
+            validate_remote_decision(
+                decision_payload(contract_commit="f" * 40), self.event
+            )
+
+    def test_lease_timeout_recovery(self):
+        first = self.claim()
+        self.broker._conn().execute(
+            "UPDATE remote_jobs SET lease_until=0 WHERE job_id=?", (first["job_id"],)
+        )
+        second = self.claim("claim-fedcba9876543210")
+        self.assertTrue(second["claimed"])
+        self.assertEqual(second["lease_version"], first["lease_version"] + 1)
+
+    def test_completed_job_not_replayed(self):
+        lease = self.claim()
+        self.broker.complete(
+            job_id=lease["job_id"], runner_id="brain-88-6",
+            idempotency_key="decision-" + "0" * 16, request_digest="a" * 64,
+            decision_digest="b" * 64, state_version_after=1,
+        )
+        empty = self.claim("claim-new-0123456789abcdef")
+        self.assertEqual(empty, {"claimed": False})
+
+    def test_failure_retries_then_dead_letters(self):
+        lease = self.claim()
+        first = self.broker.fail(
+            job_id=lease["job_id"], runner_id="brain-88-6",
+            lease_token=lease["lease_token"], claim_key=lease["claim_idempotency_key"],
+            category="CODEX_EXEC_NONZERO", idempotency_key="fail-" + "0" * 16,
+            runner_token=self.token, retry_after=0,
+        )
+        self.assertEqual(first["status"], "RETRY")
+        lease2 = self.claim("claim-second-0123456789abcdef")
+        second = self.broker.fail(
+            job_id=lease2["job_id"], runner_id="brain-88-6",
+            lease_token=lease2["lease_token"], claim_key=lease2["claim_idempotency_key"],
+            category="CODEX_EXEC_NONZERO", idempotency_key="fail-second-" + "0" * 16,
+            runner_token=self.token, retry_after=0,
+        )
+        self.assertEqual(second["status"], "DEAD_LETTER")
+
+
+class RemoteGatewayHTTPTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        for name in ("events", "decisions", "outbox", "etc"):
+            (self.root / name).mkdir(mode=0o700)
+        for name in ("sync", "commands"):
+            (self.root / name).mkdir(mode=0o2750)
+            (self.root / name).chmod(0o2750)
+        for name, value in (
+            ("inbound.token", "inbound-not-real"),
+            ("callback.token", "callback-not-real"),
+            ("remote.token", "runner-not-real"),
+        ):
+            path = self.root / "etc" / name
+            path.write_text(value, encoding="utf-8"); path.chmod(0o600)
+        raw = {
+            "host": "127.0.0.1", "port": 18920,
+            "project_key": "v2r1-personal-weixin-production",
+            "database_path": str(self.root / "state.sqlite3"),
+            "event_inbox_dir": str(self.root / "events"),
+            "decision_inbox_dir": str(self.root / "decisions"),
+            "dispatch_outbox_dir": str(self.root / "outbox"),
+            "openclaw_sync_dir": str(self.root / "sync"),
+            "command_queue_dir": str(self.root / "commands"),
+            "inbound_token_file": str(self.root / "etc/inbound.token"),
+            "callback_token_file": str(self.root / "etc/callback.token"),
+            "contract_commit": CONTRACT, "central_channel": "remote_codex",
+            "codex_local": {"queue_dir": str(self.root / "queue"), "timeout_seconds": 30},
+            "remote_codex": {
+                "request_max_bytes": 262144, "job_lease_seconds": 60,
+                "runners": [{"runner_id": "brain-88-6",
+                             "token_file": str(self.root / "etc/remote.token"),
+                             "scopes": ["claim", "heartbeat", "decision", "fail"]}],
+            },
+            "workspace_agent": {"trigger_id": None, "access_token_file": None},
+            "responses_fallback": {"api_key_file": None, "model": None},
+            "polling": {"interval_seconds": 1, "summary_interval_seconds": 10},
+            "loop_controls": {"callback_timeout_seconds": 30, "lease_seconds": 10,
+                              "retry_base_seconds": 1, "max_retries": 2,
+                              "max_same_blocker_retries": 2, "max_hops_per_gate": 20},
+            "openclaw_sync": {"endpoint": None, "token_file": None},
+            "git_poll": {"enabled": False, "bare_repo": None,
+                         "ref_prefixes": [], "interval_seconds": 30},
+        }
+        config_path = self.root / "etc/config.json"
+        config_path.write_text(json.dumps(raw), encoding="utf-8"); config_path.chmod(0o600)
+        os.environ["AIOS_ORCHESTRATOR_CONFIG_OWNER_UID"] = str(os.getuid())
+        os.environ["AIOS_ORCHESTRATOR_SECRET_OWNER_UID"] = str(os.getuid())
+        self.runtime = Runtime(Config.from_file(config_path))
+        Handler.runtime = self.runtime
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown(); self.server.server_close(); self.thread.join(timeout=5)
+        self.runtime.store.close()
+        if self.runtime.remote_codex is not None: self.runtime.remote_codex.close()
+        os.environ.pop("AIOS_ORCHESTRATOR_CONFIG_OWNER_UID", None)
+        os.environ.pop("AIOS_ORCHESTRATOR_SECRET_OWNER_UID", None)
+        self.temp.cleanup()
+
+    def post(self, path: str, body: dict, key: str) -> dict:
+        request = urllib.request.Request(
+            f"http://127.0.0.1:{self.server.server_port}{path}",
+            data=json.dumps(body).encode(), method="POST",
+            headers={"Authorization": "Bearer runner-not-real",
+                     "X-AIOS-Runner-ID": "brain-88-6",
+                     "Idempotency-Key": key, "Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(request, timeout=5) as response:
+            return json.loads(response.read())
+
+    def test_http_claim_and_decision_chain(self):
+        self.runtime.submit_event(event_payload())
+        self.runtime.process_new_events()
+        lease = self.post(
+            "/v1/orchestrator/central/jobs/claim",
+            {"contract_commit": CONTRACT}, "claim-http-0123456789abcdef",
+        )
+        self.assertTrue(lease["claimed"])
+        result = self.post(
+            f"/v1/orchestrator/central/jobs/{lease['job_id']}/decision",
+            {"lease_token": lease["lease_token"],
+             "claim_idempotency_key": lease["claim_idempotency_key"],
+             "decision": decision_payload()},
+            "decision-http-0123456789abcdef",
+        )
+        self.assertTrue(result["accepted"]); self.assertTrue(result["completed"])
+        status = self.runtime.store.event_status("evt-remote-001")
+        self.assertEqual(status["status"], "DECIDED")
+        self.assertTrue(status["decision_applied"])
+
+
+class RunnerTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.token = self.root / "runner.token"
+        self.token.write_text("not-a-real-token", encoding="utf-8")
+        self.token.chmod(0o600)
+        self.config = RUNNER.RunnerConfig(
+            gateway_url="http://127.0.0.1:28920/v1/orchestrator",
+            runner_id="brain-88-6", token_file=self.token,
+            contract_commit=CONTRACT, codex_binary=Path("/usr/bin/codex"),
+            codex_home=self.root / ".codex", controller_readonly=self.root / "controller",
+            state_file=self.root / "state/runner-state.json", timeout_seconds=30,
+            poll_seconds=1, heartbeat_seconds=5,
+        )
+        self.runner = RUNNER.RemoteCodexRunner(self.config)
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def state(self, phase="CLAIMED", decision=None):
+        return {
+            "phase": phase, "job_id": "evt-remote-001", "lease_token": "a" * 64,
+            "claim_idempotency_key": "claim-" + "0" * 16,
+            "event": event_payload(), "decision": decision,
+        }
+
+    def test_runner_restart_recovery(self):
+        state = self.state("DECIDED", decision_payload())
+        with mock.patch.object(self.runner, "health_check"), \
+             mock.patch.object(self.runner, "_load_state", return_value=state), \
+             mock.patch.object(self.runner, "_heartbeat"), \
+             mock.patch.object(self.runner, "_exec") as execute, \
+             mock.patch.object(self.runner, "_submit") as submit:
+            self.assertTrue(self.runner.run_one())
+        execute.assert_not_called(); submit.assert_called_once_with(state)
+
+    def test_tunnel_disconnect_recovery(self):
+        state = self.state("DECIDED", decision_payload())
+        with mock.patch.object(self.runner, "health_check"), \
+             mock.patch.object(self.runner, "_load_state", return_value=state), \
+             mock.patch.object(self.runner, "_heartbeat"), \
+             mock.patch.object(self.runner, "_submit", side_effect=TriggerError("gateway_decision_unavailable")), \
+             mock.patch.object(self.runner, "_fail", side_effect=TriggerError("gateway_fail_unavailable")), \
+             mock.patch.object(self.runner, "_clear_state") as clear:
+            with self.assertRaises(TriggerError): self.runner.run_one()
+        clear.assert_not_called()
+
+    def test_codex_login_required(self):
+        completed = subprocess.CompletedProcess([], 1, stdout="Not logged in")
+        with mock.patch.object(self.runner, "_run_quiet", return_value=completed):
+            with self.assertRaisesRegex(ConfigurationError, "authentication_required"):
+                self.runner.login_status()
+
+    def test_codex_exec_nonzero(self):
+        process = mock.Mock(returncode=1)
+        process.communicate.return_value = (None, None)
+        with mock.patch("subprocess.Popen", return_value=process), \
+             mock.patch.object(self.runner, "_heartbeat"):
+            self.config.state_file.parent.mkdir(parents=True)
+            with self.assertRaisesRegex(TriggerError, "codex_exec_nonzero"):
+                self.runner._exec(Event.from_dict(event_payload()), self.state())
+
+    def test_codex_exec_timeout(self):
+        process = mock.Mock(pid=1234)
+        process.communicate.side_effect = subprocess.TimeoutExpired([], 1)
+        with mock.patch("subprocess.Popen", return_value=process), \
+             mock.patch.object(self.runner, "_heartbeat"), \
+             mock.patch.object(self.runner, "_terminate_group") as terminate:
+            self.config.state_file.parent.mkdir(parents=True)
+            with self.assertRaisesRegex(TriggerError, "codex_exec_timeout"):
+                self.runner._exec(Event.from_dict(event_payload()), self.state())
+        terminate.assert_called_once_with(process)
+
+    def test_invalid_decision_rejected(self):
+        with self.assertRaises(ValidationError):
+            validate_remote_decision({"verdict": "BLOCKED"}, Event.from_dict(event_payload()))
+
+    def test_logs_redact_auth_prompt_output(self):
+        secret = "secret-token-value"
+        prompt = "full-private-prompt"
+        output = "raw-jsonl-output"
+        logger = logging.getLogger("aios-remote-codex-brain")
+        with self.assertLogs(logger, level="WARNING") as captured:
+            logger.warning("remote_cycle_failed category=%s", "TriggerError")
+        text = "\n".join(captured.output)
+        self.assertNotIn(secret, text); self.assertNotIn(prompt, text); self.assertNotIn(output, text)
+
+    def test_remote_runner_has_no_git_write(self):
+        unit = (ROOT / "systemd/aios-remote-codex-brain.service").read_text(encoding="utf-8")
+        self.assertIn("ProtectSystem=strict", unit)
+        self.assertIn("InaccessiblePaths=/data", unit)
+        self.assertNotIn("/data/aios-gitops", unit.split("ReadWritePaths=", 1)[-1])
+
+    def test_remote_runner_has_no_production_access(self):
+        unit = (ROOT / "systemd/aios-remote-codex-brain.service").read_text(encoding="utf-8")
+        self.assertIn("NoNewPrivileges=yes", unit)
+        self.assertIn("/var/run/docker.sock", unit)
+        self.assertNotIn("CAP_SYS_ADMIN", unit)
+
+
+if __name__ == "__main__":
+    unittest.main()