feat(agent): add autonomous intent worker v2.1

deploy/agent-runtime-openclaw.conf.example

@@ -0,0 +1,18 @@
+# Non-secret staging configuration for the OpenClaw engineering backend.
+#
+# Discover the exact binary with `command -v openclaw`.
+# Do not point this at the chat/WeChat Gateway HTTP endpoint.
+AIOS_AGENT_RUNTIME_BACKEND_ARGV_JSON='["/usr/bin/python3","/usr/local/lib/aios-agent-runtime-broker/openclaw_engineering_backend.py"]'
+
+# Required: exact absolute path discovered on hw.imruyi.top.
+AIOS_OPENCLAW_BIN=/REPLACE/WITH/ABSOLUTE/OPENCLAW/PATH
+
+# Required: provider/model verified on the installed OpenClaw build.
+AIOS_OPENCLAW_MODEL=REPLACE_WITH_VERIFIED_PROVIDER_MODEL
+
+# Non-secret map: provider environment variable -> systemd credential file name.
+# Example after separately projecting an encrypted credential:
+# AIOS_OPENCLAW_PROVIDER_CREDENTIAL_MAP_JSON='{"OPENAI_API_KEY":"openai_api_key"}'
+AIOS_OPENCLAW_PROVIDER_CREDENTIAL_MAP_JSON='{}'
+
+AIOS_OPENCLAW_EXEC_TIMEOUT_SECONDS=1800

deploy/install-autonomous-intent-worker-staging.sh

@@ -0,0 +1,66 @@
+#!/bin/sh
+set -eu
+
+if [ "$(id -u)" -ne 0 ]; then
+  echo "must run as root" >&2
+  exit 20
+fi
+
+SRC_ROOT="${1:-.}"
+WORKER_SRC="$SRC_ROOT/services/aios-autonomous-agent-worker"
+BROKER_SRC="$SRC_ROOT/services/aios-agent-runtime-broker"
+UNIT_SRC="$SRC_ROOT/deploy/systemd"
+
+grep -Fq 'AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:19998' \
+  "$UNIT_SRC/aios-autonomous-agent-worker.service" || {
+    echo "staging bridge binding missing" >&2
+    exit 21
+  }
+
+getent group aios-agent-runtime >/dev/null 2>&1 || groupadd --system aios-agent-runtime
+id aios-agent-runtime >/dev/null 2>&1 || \
+  useradd --system --gid aios-agent-runtime --home-dir /nonexistent --shell /usr/sbin/nologin aios-agent-runtime
+id aios-autonomous-agent-worker >/dev/null 2>&1 || \
+  useradd --system --home-dir /nonexistent --shell /usr/sbin/nologin aios-autonomous-agent-worker
+usermod -a -G aios-agent-runtime aios-autonomous-agent-worker
+
+install -d -m 0755 /usr/local/lib/aios-autonomous-agent-worker
+install -d -m 0755 /usr/local/lib/aios-agent-runtime-broker
+install -d -m 0750 -o aios-autonomous-agent-worker -g aios-autonomous-agent-worker \
+  /var/lib/aios-autonomous-agent-worker
+install -d -m 0770 -o aios-autonomous-agent-worker -g aios-agent-runtime \
+  /var/lib/aios-autonomous-agent-worker/workspaces
+
+for file in autonomous_worker.py intent_runtime.py runtime_client.py register.py runtime_selftest.py; do
+  install -m 0755 "$WORKER_SRC/$file" "/usr/local/lib/aios-autonomous-agent-worker/$file"
+done
+install -m 0755 "$BROKER_SRC/broker.py" /usr/local/lib/aios-agent-runtime-broker/broker.py
+install -m 0755 "$BROKER_SRC/openclaw_engineering_backend.py" /usr/local/lib/aios-agent-runtime-broker/openclaw_engineering_backend.py
+install -m 0644 "$BROKER_SRC/RUNTIME_CONTRACT.md" /usr/local/lib/aios-agent-runtime-broker/RUNTIME_CONTRACT.md
+
+install -m 0644 "$UNIT_SRC/aios-autonomous-agent-worker.service" /etc/systemd/system/
+install -m 0644 "$UNIT_SRC/aios-autonomous-agent-register.service" /etc/systemd/system/
+install -m 0644 "$UNIT_SRC/aios-agent-runtime-broker@.service" /etc/systemd/system/
+
+systemctl daemon-reload
+
+cat <<'EOF'
+Files installed. Services were NOT started or enabled.
+
+Required before activation:
+1. Discover `openclaw` with `command -v openclaw`.
+2. Verify `openclaw agent exec --help` exposes the required headless coding
+   options on the installed version.
+3. Create /etc/aios/agent-runtime-openclaw.conf from the non-secret example.
+4. Project provider/model credentials into the OpenClaw broker service with a
+   separate systemd encrypted credential. Never reuse the Online Agent Task
+   token or writer lease.
+5. Start only `aios-agent-runtime-broker@openclaw.service`.
+6. Run runtime_selftest.py and require a real worktree mutation with unchanged
+   Git HEAD.
+7. Only after self-test PASS register the autonomous identity and enable
+   continuous polling.
+
+The autonomous worker is staging-only and rejects any Governed Bridge other
+than http://127.0.0.1:19998.
+EOF

deploy/systemd/aios-agent-runtime-broker@.service

@@ -0,0 +1,42 @@
+[Unit]
+Description=AIOS isolated %i engineering runtime broker
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=aios-agent-runtime
+Group=aios-agent-runtime
+Environment=PYTHONUNBUFFERED=1
+Environment=AIOS_AGENT_RUNTIME_ADAPTER=%i
+Environment=AIOS_AGENT_RUNTIME_SOCKET=/run/aios-agent-runtime/%i.sock
+Environment=AIOS_AGENT_RUNTIME_WORKSPACE_ROOT=/var/lib/aios-autonomous-agent-worker/workspaces
+Environment=AIOS_AGENT_RUNTIME_ALLOWED_PEER_USER=aios-autonomous-agent-worker
+Environment=AIOS_AGENT_RUNTIME_CONNECTION_TIMEOUT_SECONDS=15
+Environment=AIOS_AGENT_RUNTIME_BACKEND_TIMEOUT_SECONDS=3600
+EnvironmentFile=/etc/aios/agent-runtime-%i.conf
+ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-agent-runtime-broker/broker.py
+Restart=always
+RestartSec=5
+RuntimeDirectory=aios-agent-runtime
+RuntimeDirectoryMode=0770
+UMask=0007
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadWritePaths=/var/lib/aios-autonomous-agent-worker/workspaces /run/aios-agent-runtime
+InaccessiblePaths=-/data/aios-gitops/repo.git -/etc/credstore.encrypted/aios-autonomous-agent-worker
+CapabilityBoundingSet=
+AmbientCapabilities=
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-agent-runtime-broker@openclaw.service.d/10-provider-credential.conf.example

@@ -0,0 +1,5 @@
+# Example only. Choose the credential name/provider used by the verified model.
+# The encrypted credential file contains secret material and must never be committed.
+[Service]
+# Example:
+# LoadCredentialEncrypted=openai_api_key:/etc/credstore.encrypted/aios-agent-runtime-openclaw/openai_api_key

deploy/systemd/aios-autonomous-agent-register.service

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

deploy/systemd/aios-autonomous-agent-worker.service

@@ -0,0 +1,56 @@
+[Unit]
+Description=AIOS Autonomous Intent Engineering Worker (staging)
+After=network-online.target aios-agent-task-api.service aios-agent-runtime-broker@openclaw.service aios-autonomous-agent-register.service
+Wants=network-online.target aios-agent-runtime-broker@openclaw.service
+Requires=aios-autonomous-agent-register.service
+
+[Service]
+Type=simple
+User=aios-autonomous-agent-worker
+Group=aios-autonomous-agent-worker
+SupplementaryGroups=aios-agent-runtime gitops
+LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-autonomous-agent-worker/agent_token
+Environment=PYTHONUNBUFFERED=1
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:19998
+Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-autonomous-20260815
+Environment=AIOS_AGENT_VERSION=2026-08-15.4-autonomous-intent
+Environment=AIOS_AUTONOMOUS_REPOSITORY=aios/aios
+Environment=AIOS_AUTONOMOUS_BASE_REF=refs/heads/main
+Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_AUTONOMOUS_STATE_ROOT=/var/lib/aios-autonomous-agent-worker
+Environment=AIOS_AUTONOMOUS_WORKSPACE_ROOT=/var/lib/aios-autonomous-agent-worker/workspaces
+Environment=AIOS_AUTONOMOUS_POLL_SECONDS=5
+Environment=AIOS_AUTONOMOUS_HEARTBEAT_SECONDS=10
+Environment=AIOS_AUTONOMOUS_LEASE_SECONDS=300
+Environment=AIOS_AUTONOMOUS_RUNTIME_TIMEOUT_SECONDS=3600
+Environment=AIOS_AUTONOMOUS_DEFAULT_ADAPTER=openclaw
+Environment=AIOS_AGENT_RUNTIME_SHARED_GROUP=aios-agent-runtime
+Environment=AIOS_HERMES_BROKER_SOCKET=/run/aios-agent-runtime/hermes.sock
+Environment=AIOS_OPENCLAW_BROKER_SOCKET=/run/aios-agent-runtime/openclaw.sock
+ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-autonomous-agent-worker/autonomous_worker.py
+Restart=always
+RestartSec=5
+TimeoutStopSec=30
+UMask=0027
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadWritePaths=/var/lib/aios-autonomous-agent-worker
+InaccessiblePaths=-/etc/credstore.encrypted
+CapabilityBoundingSet=
+AmbientCapabilities=
+
+[Install]
+WantedBy=multi-user.target

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

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

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

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

docs/autonomous-intent-worker-v2.md

@@ -0,0 +1,110 @@
+# Autonomous Intent Worker v2
+
+## Goal
+
+A Custom GPT or other trusted producer creates a prose engineering task with:
+
+- `goal`
+- `instructions`
+- `success_criteria`
+- `required_capabilities`
+
+A long-running autonomous worker claims only tasks explicitly marked with:
+
+- `intent.task`
+- `engineering.execute`
+
+and only when the requested engineering runtime capability is actually
+available, e.g. `agent.runtime.hermes`.
+
+The runtime edits a disposable worktree. The worker independently inspects the
+diff, scans for secrets, reruns verification, and persists through staging
+Governed GitOps.
+
+## Security domains
+
+The Online Agent worker owns task/lease/governed-write credentials.
+
+The runtime broker is a separate service identity. It authorizes the worker
+over Unix `SO_PEERCRED`, opens the worktree using `O_PATH|O_NOFOLLOW`, and
+passes a pinned `/proc/self/fd/<fd>` path to the engineering backend. The
+backend never receives the Agent Task token or writer lease.
+
+The worker rejects:
+- a runtime-created Git commit (`HEAD != base_oid`);
+- direct authoritative ref mutation/push;
+- sensitive credential/key paths;
+- obvious secret material;
+- non-staging Governed Bridge endpoints.
+
+## Staging boundary
+
+`autonomous_worker.py` intentionally rejects any Governed Write API base other
+than:
+
+`http://127.0.0.1:19998`
+
+Production/main/tag/baseline/force-push writes remain outside this worker.
+
+## Migration protection
+
+The worker does not infer intent from prose. Historical tasks without
+`intent.task` are skipped before claim, so they do not lose an attempt.
+
+The bootstrap task
+`implement-autonomous-intent-worker-v2-20260815` currently lacks the intent
+marker and is therefore intentionally not claimable by this worker. That is a
+control-plane/bootstrap concern, not a reason to weaken the runtime rule.
+
+## Runtime requirement
+
+A chat/WeChat Hermes endpoint is not enough. A compliant engineering backend
+must:
+1. accept `aios.online-agent.runtime-request.v1`;
+2. inspect and mutate the supplied disposable worktree;
+3. leave Git HEAD unchanged;
+4. return `aios.online-agent.runtime-result.v1`.
+
+The exact installed Hermes/Codex interface is host-specific and must be
+discovered before configuring the broker. No CLI flags are invented here.
+
+## Activation gate
+
+Do not register or enable the autonomous consumer until:
+1. a real engineering backend is found;
+2. broker is configured with that backend;
+3. `runtime_selftest.py` reports PASS;
+4. backend has no Agent Task token or writer lease.
+
+After activation, complete D/E/F/G:
+- D: native intent task, no `metadata.execution`;
+- E: goal-only engineering task, runtime finds files/root cause;
+- F: unavailable runtime capability remains unclaimed;
+- G: restart during real runtime execution, no duplicate commit/branch.
+
+
+## OpenClaw backend candidate
+
+A current OpenClaw installation can be used as an autonomous engineering
+runtime without making OpenClaw itself an AIOS Task API client.
+
+The bundled wrapper uses the official headless coding entry point:
+
+```text
+openclaw agent exec
+  --message-file -
+  --cwd <pinned-disposable-worktree>
+  --model <verified-provider/model>
+  --auth-env-only
+  --timeout <seconds>
+  --json
+```
+
+Reference used for implementation:
+`https://docs.openclaw.ai/cli/agent`
+
+The wrapper maps the stable OpenClaw JSON envelope into
+`aios.online-agent.runtime-result.v1`. The AIOS worker remains responsible for
+claim/heartbeat, independent diff verification and Governed GitOps.
+
+OpenClaw is advertised as `agent.runtime.openclaw`, not as Hermes or Codex.

services/aios-agent-runtime-broker/RUNTIME_CONTRACT.md

@@ -0,0 +1,93 @@
+# Autonomous Engineering Runtime Contract v1
+
+The Online Agent Worker and the Hermes/Codex engineering runtime are separate
+security domains.
+
+The worker owns:
+- Agent Task polling, atomic claim, task lease and heartbeat;
+- intent resolution and restart state;
+- authoritative repository/base selection;
+- independent diff inspection and secret checks;
+- verification command allowlisting and re-execution;
+- Governed GitOps task/lease/commit;
+- final result submission.
+
+The runtime broker and Hermes/Codex backend do **not** receive:
+- Agent Task API token;
+- task lease token;
+- governed writer lease or Bridge credential;
+- authority repository write access.
+
+## stdin request
+
+The backend configured behind the broker receives one JSON object:
+
+`schema = aios.online-agent.runtime-request.v1`
+
+It contains task goal/instructions/success criteria and a pinned disposable
+worktree path. The backend may inspect and modify that worktree. It must not
+commit or push.
+
+## stdout response
+
+The backend returns one JSON object:
+
+```json
+{
+  "schema": "aios.online-agent.runtime-result.v1",
+  "status": "COMPLETED",
+  "summary": "root cause and implementation summary",
+  "verification": [
+    {
+      "label": "focused test",
+      "argv": ["/usr/bin/python3", "-m", "pytest", "-q", "tests/test_x.py"],
+      "cwd": ".",
+      "timeout_seconds": 600
+    }
+  ],
+  "evidence": {
+    "root_cause": "..."
+  }
+}
+```
+
+`status` may also be `BLOCKED` or `FAILED`.
+
+The worker ignores any backend claim about changed files and independently reads
+Git state. It also reruns verification requests through its local command
+allowlist.
+
+## Real runtime requirement
+
+A chat/router/WeChat Hermes service is not an engineering backend. A compliant
+backend must be able to read and mutate the disposable worktree and emit the
+runtime result schema above.
+
+The exact Hermes/Codex CLI/API must be discovered from the installed runtime.
+Do not invent flags. `/etc/aios/agent-runtime-hermes.conf` contains only
+non-secret backend command configuration. Any model/runtime credential must be
+delivered to the broker/backend through its own systemd credential projection,
+never through the Online Agent worker token.
+
+
+## OpenClaw headless engineering backend
+
+For hosts that have a current OpenClaw CLI, use the bundled
+`openclaw_engineering_backend.py`.
+
+The wrapper invokes the documented headless coding entry point:
+
+`openclaw agent exec --message-file - --cwd <pinned-worktree> --model <model> --auth-env-only --timeout <seconds> --json`
+
+It deliberately does **not** use the Gateway/WeChat/chat route. `agent exec`
+runs an embedded, one-shot agent turn and scopes its workspace/tool working
+directory to `--cwd`.
+
+Provider credentials, if required, are projected into the broker service under
+separate systemd credentials and mapped to provider environment variables only
+for the OpenClaw child. The Online Agent Task token and writer lease remain
+absent.
+
+Before enabling the autonomous worker, run a disposable broker self-test that
+proves OpenClaw changes the requested worktree while Git HEAD remains at the
+base OID.

services/aios-agent-runtime-broker/broker.py

@@ -0,0 +1,297 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import array
+import json
+import os
+from pathlib import Path
+import pwd
+import re
+import socket
+import stat
+import subprocess
+from typing import Any, Mapping
+
+
+MAX_BYTES = 1024 * 1024
+ADAPTER = os.environ.get("AIOS_AGENT_RUNTIME_ADAPTER", "").strip()
+SOCKET_PATH = Path(os.environ.get("AIOS_AGENT_RUNTIME_SOCKET", ""))
+WORKSPACE_ROOT = Path(os.environ.get("AIOS_AGENT_RUNTIME_WORKSPACE_ROOT", ""))
+BACKEND_ARGV_JSON = os.environ.get("AIOS_AGENT_RUNTIME_BACKEND_ARGV_JSON", "")
+ALLOWED_PEER_UID = int(os.environ.get("AIOS_AGENT_RUNTIME_ALLOWED_PEER_UID", "-1"))
+ALLOWED_PEER_USER = os.environ.get("AIOS_AGENT_RUNTIME_ALLOWED_PEER_USER", "").strip()
+CONNECTION_TIMEOUT = float(os.environ.get("AIOS_AGENT_RUNTIME_CONNECTION_TIMEOUT_SECONDS", "15"))
+BACKEND_TIMEOUT = int(os.environ.get("AIOS_AGENT_RUNTIME_BACKEND_TIMEOUT_SECONDS", "3600"))
+BACKEND_ENV_ALLOWLIST = (
+    "AIOS_OPENCLAW_BIN",
+    "AIOS_OPENCLAW_MODEL",
+    "AIOS_OPENCLAW_PROVIDER_CREDENTIAL_MAP_JSON",
+    "AIOS_OPENCLAW_EXEC_TIMEOUT_SECONDS",
+)
+SENSITIVE_RE = re.compile(r"(?i)(authorization\s*:\s*bearer\s+\S+|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)")
+
+
+class BrokerError(RuntimeError):
+    def __init__(self, code: str, *, details: Mapping[str, Any] | None = None) -> None:
+        super().__init__(code)
+        self.code = code
+        self.details = dict(details or {})
+
+
+def peer_uid(connection: socket.socket) -> int:
+    if not hasattr(socket, "SO_PEERCRED"):
+        raise BrokerError("SO_PEERCRED_UNAVAILABLE")
+    raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+    pid, uid, gid = array.array("i", raw[:12])
+    return int(uid)
+
+
+def configured_peer_uid() -> int:
+    if ALLOWED_PEER_UID >= 0:
+        return ALLOWED_PEER_UID
+    if not ALLOWED_PEER_USER:
+        raise BrokerError("ALLOWED_PEER_UID_NOT_CONFIGURED")
+    try:
+        return int(pwd.getpwnam(ALLOWED_PEER_USER).pw_uid)
+    except KeyError as exc:
+        raise BrokerError("ALLOWED_PEER_USER_NOT_FOUND") from exc
+
+
+def authorize_peer(connection: socket.socket) -> None:
+    uid = peer_uid(connection)
+    expected = configured_peer_uid()
+    if uid != expected:
+        raise BrokerError("PEER_UID_REJECTED", details={"peer_uid": uid})
+
+
+def read_request(connection: socket.socket) -> dict[str, Any]:
+    connection.settimeout(CONNECTION_TIMEOUT)
+    total = 0
+    chunks: list[bytes] = []
+    while True:
+        try:
+            chunk = connection.recv(65536)
+        except socket.timeout as exc:
+            raise BrokerError("REQUEST_TIMEOUT") from exc
+        if not chunk:
+            break
+        total += len(chunk)
+        if total > MAX_BYTES:
+            raise BrokerError("REQUEST_TOO_LARGE")
+        chunks.append(chunk)
+    try:
+        value = json.loads(b"".join(chunks).decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise BrokerError("REQUEST_INVALID_JSON") from exc
+    if not isinstance(value, Mapping):
+        raise BrokerError("REQUEST_OBJECT_REQUIRED")
+    return dict(value)
+
+
+def backend_argv() -> list[str]:
+    try:
+        value = json.loads(BACKEND_ARGV_JSON)
+    except json.JSONDecodeError as exc:
+        raise BrokerError("BACKEND_ARGV_JSON_INVALID") from exc
+    if not isinstance(value, list) or not value or not all(isinstance(item, str) and item for item in value):
+        raise BrokerError("BACKEND_ARGV_INVALID")
+    executable = Path(value[0])
+    if not executable.is_absolute():
+        raise BrokerError("BACKEND_EXECUTABLE_INVALID")
+    try:
+        resolved = executable.resolve(strict=True)
+    except OSError as exc:
+        raise BrokerError("BACKEND_EXECUTABLE_INVALID") from exc
+    if not resolved.is_file():
+        raise BrokerError("BACKEND_EXECUTABLE_INVALID")
+    return list(value)
+
+
+def validate_request(value: Mapping[str, Any]) -> tuple[dict[str, Any], int]:
+    if value.get("schema") != "aios.online-agent.runtime-request.v1":
+        raise BrokerError("REQUEST_SCHEMA_INVALID")
+    if value.get("adapter") != ADAPTER or ADAPTER not in {"hermes", "codex", "openclaw"}:
+        raise BrokerError("REQUEST_ADAPTER_INVALID")
+    context = value.get("execution_context")
+    policy = value.get("policy")
+    if not isinstance(context, Mapping) or not isinstance(policy, Mapping):
+        raise BrokerError("REQUEST_CONTEXT_INVALID")
+    worktree_raw = context.get("worktree")
+    if not isinstance(worktree_raw, str):
+        raise BrokerError("REQUEST_WORKTREE_INVALID")
+    for field in (
+        "authoritative_repository_write",
+        "direct_git_ref_mutation",
+        "direct_git_push",
+        "direct_git_commit",
+        "main_write",
+        "force_push",
+        "production_write",
+        "tag_write",
+        "baseline_write",
+        "task_api_credential_visible",
+        "writer_lease_visible",
+    ):
+        if policy.get(field) is not False:
+            raise BrokerError("REQUEST_POLICY_REJECTED", details={"field": field})
+
+    root = WORKSPACE_ROOT.resolve()
+    candidate = Path(worktree_raw)
+    if not candidate.is_absolute():
+        raise BrokerError("REQUEST_WORKTREE_INVALID")
+    parent = candidate.parent.resolve()
+    if root == parent or root not in parent.parents:
+        raise BrokerError("REQUEST_WORKTREE_OUT_OF_SCOPE")
+
+    flags = getattr(os, "O_PATH", os.O_RDONLY) | getattr(os, "O_NOFOLLOW", 0) | os.O_DIRECTORY
+    try:
+        fd = os.open(str(candidate), flags)
+    except OSError as exc:
+        raise BrokerError("REQUEST_WORKTREE_OPEN_FAILED") from exc
+    st = os.fstat(fd)
+    if not stat.S_ISDIR(st.st_mode):
+        os.close(fd)
+        raise BrokerError("REQUEST_WORKTREE_NOT_DIRECTORY")
+    pinned = Path(f"/proc/self/fd/{fd}")
+    try:
+        resolved = pinned.resolve(strict=True)
+    except OSError:
+        os.close(fd)
+        raise BrokerError("REQUEST_WORKTREE_PIN_FAILED")
+    if root not in resolved.parents:
+        os.close(fd)
+        raise BrokerError("REQUEST_WORKTREE_OUT_OF_SCOPE")
+    return dict(value), fd
+
+
+def sanitize(value: Any) -> Any:
+    if isinstance(value, Mapping):
+        cleaned: dict[str, Any] = {}
+        for key, inner in value.items():
+            lowered = str(key).lower()
+            if any(marker in lowered for marker in ("token", "secret", "credential", "authorization", "password", "api_key", "apikey", "access_key")):
+                cleaned[str(key)] = "[REDACTED]"
+            else:
+                cleaned[str(key)] = sanitize(inner)
+        return cleaned
+    if isinstance(value, list):
+        return [sanitize(item) for item in value]
+    if isinstance(value, str):
+        return SENSITIVE_RE.sub("[REDACTED]", value)
+    return value
+
+
+def sanitized_failure(code: str, details: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    safe: dict[str, Any] = {}
+    for key, value in dict(details or {}).items():
+        lowered = str(key).lower()
+        if any(marker in lowered for marker in ("token", "secret", "credential", "authorization", "password", "api_key", "apikey", "access_key")):
+            safe[str(key)] = "[REDACTED]"
+        else:
+            safe[str(key)] = value
+    return {
+        "schema": "aios.online-agent.runtime-result.v1",
+        "status": "FAILED",
+        "summary": code,
+        "verification": [],
+        "evidence": safe,
+    }
+
+
+def execute_backend(request: Mapping[str, Any], worktree_fd: int) -> dict[str, Any]:
+    argv = backend_argv()
+    request = dict(request)
+    context = dict(request["execution_context"])
+    # Pass a pinned inode-backed path, not the original user-visible pathname.
+    context["worktree"] = f"/proc/self/fd/{worktree_fd}"
+    request["execution_context"] = context
+    payload = json.dumps(request, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    environment = {
+        "PATH": "/usr/local/bin:/usr/bin:/bin",
+        "LC_ALL": "C.UTF-8",
+        "AIOS_AGENT_RUNTIME_ADAPTER": ADAPTER,
+    }
+    for name in BACKEND_ENV_ALLOWLIST:
+        value = os.environ.get(name)
+        if value is not None:
+            environment[name] = value
+    credentials_dir = os.environ.get("CREDENTIALS_DIRECTORY")
+    if credentials_dir:
+        environment["CREDENTIALS_DIRECTORY"] = credentials_dir
+
+    try:
+        completed = subprocess.run(
+            argv,
+            input=payload,
+            cwd=f"/proc/self/fd/{worktree_fd}",
+            pass_fds=(worktree_fd,),
+            capture_output=True,
+            check=False,
+            timeout=BACKEND_TIMEOUT,
+            env=environment,
+        )
+    except subprocess.TimeoutExpired:
+        return sanitized_failure("AGENT_BACKEND_TIMEOUT")
+    except OSError as exc:
+        return sanitized_failure("AGENT_BACKEND_EXEC_FAILED", {"error": type(exc).__name__})
+    if completed.returncode != 0:
+        return sanitized_failure("AGENT_BACKEND_FAILED", {"exit_code": completed.returncode})
+    if len(completed.stdout) > MAX_BYTES:
+        return sanitized_failure("AGENT_BACKEND_RESULT_TOO_LARGE")
+    try:
+        value = json.loads(completed.stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return sanitized_failure("AGENT_BACKEND_RESULT_INVALID")
+    if not isinstance(value, Mapping) or value.get("schema") != "aios.online-agent.runtime-result.v1":
+        return sanitized_failure("AGENT_BACKEND_RESULT_SCHEMA_INVALID")
+    sanitized = sanitize(dict(value))
+    return sanitized if isinstance(sanitized, dict) else sanitized_failure("AGENT_BACKEND_RESULT_INVALID")
+
+
+def serve_once(listener: socket.socket) -> None:
+    connection, _ = listener.accept()
+    worktree_fd: int | None = None
+    try:
+        authorize_peer(connection)
+        request = read_request(connection)
+        request, worktree_fd = validate_request(request)
+        response = execute_backend(request, worktree_fd)
+    except BrokerError as exc:
+        response = sanitized_failure(exc.code, exc.details)
+    except Exception as exc:
+        response = sanitized_failure("BROKER_INTERNAL_ERROR", {"error": type(exc).__name__})
+    finally:
+        if worktree_fd is not None:
+            os.close(worktree_fd)
+    raw = json.dumps(response, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
+    try:
+        connection.settimeout(CONNECTION_TIMEOUT)
+        connection.sendall(raw[:MAX_BYTES])
+    finally:
+        connection.close()
+
+
+def main() -> int:
+    if ADAPTER not in {"hermes", "codex", "openclaw"}:
+        raise BrokerError("BROKER_ADAPTER_INVALID")
+    if not SOCKET_PATH.is_absolute() or not WORKSPACE_ROOT.is_absolute():
+        raise BrokerError("BROKER_PATH_INVALID")
+    SOCKET_PATH.parent.mkdir(parents=True, mode=0o750, exist_ok=True)
+    if SOCKET_PATH.exists() or SOCKET_PATH.is_symlink():
+        SOCKET_PATH.unlink()
+    listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    listener.bind(str(SOCKET_PATH))
+    os.chmod(SOCKET_PATH, 0o660)
+    listener.listen(8)
+    try:
+        while True:
+            serve_once(listener)
+    finally:
+        listener.close()
+        if SOCKET_PATH.exists():
+            SOCKET_PATH.unlink()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-agent-runtime-broker/openclaw_engineering_backend.py

@@ -0,0 +1,369 @@
+#!/usr/bin/env python3
+"""OpenClaw engineering backend for the AIOS autonomous runtime broker.
+
+This process is intentionally *not* an Online Agent Task worker. It receives
+one secret-free `aios.online-agent.runtime-request.v1` object on stdin, invokes
+OpenClaw's headless `agent exec` against the pinned disposable worktree, and
+returns one `aios.online-agent.runtime-result.v1` object on stdout.
+
+Online Task claim/heartbeat/result and Governed GitOps remain owned by the
+AIOS autonomous worker.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+from typing import Any, Mapping
+
+
+REQUEST_SCHEMA = "aios.online-agent.runtime-request.v1"
+RESULT_SCHEMA = "aios.online-agent.runtime-result.v1"
+MAX_INPUT_BYTES = 1024 * 1024
+MAX_OUTPUT_BYTES = 1024 * 1024
+DEFAULT_TIMEOUT = 1800
+ALLOWED_ADAPTERS = frozenset({"hermes", "codex", "openclaw"})
+SECRET_PATTERNS = (
+    re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"),
+    re.compile(r"\bghp_[A-Za-z0-9]{20,}\b"),
+    re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
+    re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
+)
+
+
+class BackendError(RuntimeError):
+    def __init__(self, code: str, *, details: Mapping[str, Any] | None = None) -> None:
+        super().__init__(code)
+        self.code = code
+        self.details = dict(details or {})
+
+
+def redact_text(value: str, limit: int = 8192) -> str:
+    value = value[:limit]
+    for pattern in SECRET_PATTERNS:
+        value = pattern.sub("[REDACTED]", value)
+    return value
+
+
+def result(
+    status: str,
+    summary: str,
+    *,
+    verification: list[dict[str, Any]] | None = None,
+    evidence: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    return {
+        "schema": RESULT_SCHEMA,
+        "status": status,
+        "summary": redact_text(summary) or status,
+        "verification": verification or [],
+        "evidence": dict(evidence or {}),
+    }
+
+
+def emit(value: Mapping[str, Any]) -> int:
+    raw = json.dumps(dict(value), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    sys.stdout.write(raw)
+    sys.stdout.write("\n")
+    sys.stdout.flush()
+    return 0
+
+
+def read_request() -> dict[str, Any]:
+    raw = sys.stdin.buffer.read(MAX_INPUT_BYTES + 1)
+    if len(raw) > MAX_INPUT_BYTES:
+        raise BackendError("REQUEST_TOO_LARGE")
+    try:
+        value = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise BackendError("REQUEST_INVALID_JSON") from exc
+    if not isinstance(value, Mapping):
+        raise BackendError("REQUEST_OBJECT_REQUIRED")
+    if value.get("schema") != REQUEST_SCHEMA:
+        raise BackendError("REQUEST_SCHEMA_INVALID")
+    return dict(value)
+
+
+def nonempty(value: Any, field: str, limit: int) -> str:
+    if not isinstance(value, str):
+        raise BackendError(f"{field.upper()}_INVALID")
+    value = value.strip()
+    if not value or len(value) > limit:
+        raise BackendError(f"{field.upper()}_INVALID")
+    return value
+
+
+def string_list(value: Any, field: str) -> list[str]:
+    if not isinstance(value, list) or not value or len(value) > 64:
+        raise BackendError(f"{field.upper()}_INVALID")
+    result_value: list[str] = []
+    for item in value:
+        result_value.append(nonempty(item, field, 4096))
+    return result_value
+
+
+def validate_request(value: Mapping[str, Any]) -> tuple[dict[str, Any], Path]:
+    adapter = str(value.get("adapter") or "")
+    if adapter not in ALLOWED_ADAPTERS:
+        raise BackendError("ADAPTER_INVALID")
+
+    task = value.get("task")
+    context = value.get("execution_context")
+    policy = value.get("policy")
+    if not isinstance(task, Mapping) or not isinstance(context, Mapping) or not isinstance(policy, Mapping):
+        raise BackendError("REQUEST_CONTEXT_INVALID")
+
+    task_id = nonempty(task.get("task_id"), "task_id", 180)
+    goal = nonempty(task.get("goal"), "goal", 16384)
+    instructions = nonempty(task.get("instructions"), "instructions", 65536)
+    success = string_list(task.get("success_criteria"), "success_criteria")
+
+    worktree_raw = context.get("worktree")
+    if not isinstance(worktree_raw, str) or not worktree_raw.startswith("/proc/self/fd/"):
+        raise BackendError("PINNED_WORKTREE_REQUIRED")
+    worktree = Path(worktree_raw)
+    try:
+        resolved = worktree.resolve(strict=True)
+    except OSError as exc:
+        raise BackendError("PINNED_WORKTREE_UNAVAILABLE") from exc
+    if not resolved.is_dir():
+        raise BackendError("PINNED_WORKTREE_INVALID")
+
+    for field in (
+        "authoritative_repository_write",
+        "direct_git_ref_mutation",
+        "direct_git_push",
+        "direct_git_commit",
+        "main_write",
+        "force_push",
+        "production_write",
+        "tag_write",
+        "baseline_write",
+        "task_api_credential_visible",
+        "writer_lease_visible",
+    ):
+        if policy.get(field) is not False:
+            raise BackendError("REQUEST_POLICY_REJECTED", details={"field": field})
+
+    return {
+        "adapter": adapter,
+        "task_id": task_id,
+        "goal": goal,
+        "instructions": instructions,
+        "success_criteria": success,
+        "repository": str(context.get("repository") or ""),
+        "base_oid": str(context.get("base_oid") or ""),
+        "target_branch": str(context.get("target_branch") or ""),
+    }, worktree
+
+
+def openclaw_binary() -> Path:
+    configured = os.environ.get("AIOS_OPENCLAW_BIN", "").strip()
+    if not configured:
+        raise BackendError("OPENCLAW_BINARY_NOT_CONFIGURED")
+    path = Path(configured)
+    if not path.is_absolute():
+        raise BackendError("OPENCLAW_BINARY_INVALID")
+    try:
+        resolved = path.resolve(strict=True)
+    except OSError as exc:
+        raise BackendError("OPENCLAW_BINARY_UNAVAILABLE") from exc
+    if not resolved.is_file() or not os.access(resolved, os.X_OK):
+        raise BackendError("OPENCLAW_BINARY_UNAVAILABLE")
+    return resolved
+
+
+def provider_environment() -> dict[str, str]:
+    """Build an environment containing only explicitly projected provider auth.
+
+    `--auth-env-only` tells OpenClaw not to load ambient config/auth stores.
+    The mapping is non-secret; values are read from systemd credential files.
+    """
+    raw = os.environ.get("AIOS_OPENCLAW_PROVIDER_CREDENTIAL_MAP_JSON", "{}")
+    try:
+        mapping = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise BackendError("PROVIDER_CREDENTIAL_MAP_INVALID") from exc
+    if not isinstance(mapping, Mapping):
+        raise BackendError("PROVIDER_CREDENTIAL_MAP_INVALID")
+
+    credentials_directory = os.environ.get("CREDENTIALS_DIRECTORY", "").strip()
+    environment = {
+        "PATH": "/usr/local/bin:/usr/bin:/bin",
+        "LC_ALL": "C.UTF-8",
+        "NO_COLOR": "1",
+        # Prevent optional login-shell env snapshot import into exec runs.
+        "OPENCLAW_EXEC_SHELL_SNAPSHOT": "0",
+    }
+    if mapping:
+        if not credentials_directory:
+            raise BackendError("PROVIDER_CREDENTIAL_DIRECTORY_UNAVAILABLE")
+        credential_root = Path(credentials_directory)
+        for env_name, credential_name in mapping.items():
+            if (
+                not isinstance(env_name, str)
+                or not re.fullmatch(r"[A-Z][A-Z0-9_]{2,63}", env_name)
+                or not isinstance(credential_name, str)
+                or not re.fullmatch(r"[A-Za-z0-9._-]{1,128}", credential_name)
+            ):
+                raise BackendError("PROVIDER_CREDENTIAL_MAP_INVALID")
+            path = credential_root / credential_name
+            if not path.is_file() or path.is_symlink():
+                raise BackendError("PROVIDER_CREDENTIAL_UNAVAILABLE", details={"name": credential_name})
+            secret = path.read_text(encoding="utf-8").strip()
+            if not secret:
+                raise BackendError("PROVIDER_CREDENTIAL_EMPTY", details={"name": credential_name})
+            environment[env_name] = secret
+    return environment
+
+
+def build_prompt(task: Mapping[str, Any]) -> str:
+    criteria = "\n".join(f"- {item}" for item in task["success_criteria"])
+    return f"""You are the autonomous engineering runtime for AIOS task {task['task_id']}.
+
+Repository: {task['repository']}
+Base OID: {task['base_oid']}
+Target branch (owned by AIOS Governed GitOps, not by you): {task['target_branch']}
+
+Goal:
+{task['goal']}
+
+Instructions:
+{task['instructions']}
+
+Success criteria:
+{criteria}
+
+Execution boundary:
+- Work only inside the current disposable worktree.
+- Inspect the repository yourself and determine the root cause and files to change.
+- Implement the smallest correct fix.
+- Run focused tests and repair failures when feasible.
+- Do NOT run `git commit`, `git push`, `git update-ref`, create branches/tags, or modify authoritative refs.
+- Do NOT attempt to access AIOS Agent Task credentials, writer leases, or Governed GitOps credentials.
+- Leave Git HEAD unchanged; AIOS independently inspects the diff, reruns verification, and performs the governed commit.
+- In your final answer, concisely state root cause, files changed, and tests you ran.
+"""
+
+
+def openclaw_timeout() -> int:
+    raw = os.environ.get("AIOS_OPENCLAW_EXEC_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT))
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise BackendError("OPENCLAW_TIMEOUT_INVALID") from exc
+    if not 60 <= value <= 7200:
+        raise BackendError("OPENCLAW_TIMEOUT_INVALID")
+    return value
+
+
+def invoke_openclaw(task: Mapping[str, Any], worktree: Path) -> dict[str, Any]:
+    binary = openclaw_binary()
+    model = nonempty(os.environ.get("AIOS_OPENCLAW_MODEL"), "openclaw_model", 256)
+    timeout_seconds = openclaw_timeout()
+    prompt = build_prompt(task)
+
+    argv = [
+        str(binary),
+        "agent",
+        "exec",
+        "--message-file",
+        "-",
+        "--cwd",
+        str(worktree),
+        "--model",
+        model,
+        "--auth-env-only",
+        "--timeout",
+        str(timeout_seconds),
+        "--json",
+    ]
+
+    try:
+        pinned_fd = int(worktree.name)
+    except ValueError as exc:
+        raise BackendError("PINNED_WORKTREE_FD_INVALID") from exc
+
+    try:
+        completed = subprocess.run(
+            argv,
+            input=prompt.encode("utf-8"),
+            cwd=str(worktree),
+            pass_fds=(pinned_fd,),
+            capture_output=True,
+            check=False,
+            timeout=timeout_seconds + 90,
+            env=provider_environment(),
+        )
+    except subprocess.TimeoutExpired:
+        return result(
+            "FAILED",
+            "OPENCLAW_HARD_TIMEOUT",
+            evidence={"backend": "openclaw-agent-exec", "timeout_seconds": timeout_seconds},
+        )
+    except OSError as exc:
+        return result(
+            "FAILED",
+            "OPENCLAW_EXEC_FAILED",
+            evidence={"backend": "openclaw-agent-exec", "error": type(exc).__name__},
+        )
+
+    stdout = completed.stdout[:MAX_OUTPUT_BYTES]
+    try:
+        envelope = json.loads(stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return result(
+            "FAILED",
+            "OPENCLAW_RESULT_INVALID",
+            evidence={
+                "backend": "openclaw-agent-exec",
+                "exit_code": completed.returncode,
+                "stderr_sha256": __import__("hashlib").sha256(completed.stderr).hexdigest(),
+            },
+        )
+
+    if not isinstance(envelope, Mapping):
+        return result("FAILED", "OPENCLAW_RESULT_INVALID", evidence={"backend": "openclaw-agent-exec"})
+
+    ok = envelope.get("ok") is True and envelope.get("status") == "ok" and completed.returncode == 0
+    final_text = str(envelope.get("final") or "").strip()
+    evidence = {
+        "backend": "openclaw-agent-exec",
+        "openclaw_status": envelope.get("status"),
+        "exit_code": completed.returncode,
+        "model": envelope.get("model"),
+        "provider": envelope.get("provider"),
+        "session_id_present": bool(envelope.get("sessionId")),
+        "code_mode_engaged": envelope.get("codeModeEngaged"),
+        "assistant_turns": envelope.get("assistantTurns"),
+        "tool_summary": envelope.get("toolSummary") if isinstance(envelope.get("toolSummary"), Mapping) else {},
+    }
+
+    if not ok:
+        error = envelope.get("error")
+        if isinstance(error, Mapping):
+            evidence["error_kind"] = error.get("kind")
+        summary = "OPENCLAW_" + str(envelope.get("status") or "ERROR").upper()
+        return result("FAILED", summary, evidence=evidence)
+
+    summary = final_text or "OpenClaw completed the engineering turn."
+    return result("COMPLETED", summary, evidence=evidence)
+
+
+def main() -> int:
+    try:
+        request = read_request()
+        task, worktree = validate_request(request)
+        return emit(invoke_openclaw(task, worktree))
+    except BackendError as exc:
+        return emit(result("BLOCKED", exc.code, evidence=exc.details))
+    except Exception as exc:
+        return emit(result("FAILED", "OPENCLAW_BACKEND_INTERNAL_ERROR", evidence={"error": type(exc).__name__}))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-autonomous-agent-worker/autonomous_worker.py

@@ -0,0 +1,822 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import base64
+from datetime import datetime, timezone
+import grp
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import tempfile
+import threading
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Mapping, Sequence
+
+from intent_runtime import IntentError, Resolution, is_claimable_intent, resolve
+from runtime_client import RuntimeErrorV2, collect_changes, invoke, request_payload, scan_secrets
+
+
+API_BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
+BRIDGE_BASE = os.environ.get("AIOS_GOVERNED_WRITE_API_URL", "http://127.0.0.1:19998").rstrip("/")
+AGENT_ID = os.environ.get("AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-autonomous-20260815")
+AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.4-autonomous-intent")
+REPOSITORY = os.environ.get("AIOS_AUTONOMOUS_REPOSITORY", "aios/aios")
+BASE_REF = os.environ.get("AIOS_AUTONOMOUS_BASE_REF", "refs/heads/main")
+AUTHORITY_GIT_DIR = Path(os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git"))
+STATE_ROOT = Path(os.environ.get("AIOS_AUTONOMOUS_STATE_ROOT", "/var/lib/aios-autonomous-agent-worker"))
+WORKSPACE_ROOT = Path(os.environ.get("AIOS_AUTONOMOUS_WORKSPACE_ROOT", str(STATE_ROOT / "workspaces")))
+SHARED_GROUP = os.environ.get("AIOS_AGENT_RUNTIME_SHARED_GROUP", "aios-agent-runtime")
+DEFAULT_ADAPTER = os.environ.get("AIOS_AUTONOMOUS_DEFAULT_ADAPTER", "hermes")
+HERMES_SOCKET = Path(os.environ.get("AIOS_HERMES_BROKER_SOCKET", "/run/aios-agent-runtime/hermes.sock"))
+CODEX_SOCKET = Path(os.environ.get("AIOS_CODEX_BROKER_SOCKET", "/run/aios-agent-runtime/codex.sock"))
+OPENCLAW_SOCKET = Path(os.environ.get("AIOS_OPENCLAW_BROKER_SOCKET", "/run/aios-agent-runtime/openclaw.sock"))
+POLL_SECONDS = int(os.environ.get("AIOS_AUTONOMOUS_POLL_SECONDS", "5"))
+HEARTBEAT_SECONDS = int(os.environ.get("AIOS_AUTONOMOUS_HEARTBEAT_SECONDS", "10"))
+LEASE_SECONDS = int(os.environ.get("AIOS_AUTONOMOUS_LEASE_SECONDS", "300"))
+RUNTIME_TIMEOUT_SECONDS = int(os.environ.get("AIOS_AUTONOMOUS_RUNTIME_TIMEOUT_SECONDS", "3600"))
+MAX_OUTPUT_BYTES = 64 * 1024
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SENSITIVE_RE = re.compile(
+    r"(?i)(authorization\s*:\s*bearer\s+\S+|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
+)
+
+
+class WorkerError(RuntimeError):
+    def __init__(self, code: str, *, phase: str, details: Mapping[str, Any] | None = None) -> None:
+        super().__init__(code)
+        self.code = code
+        self.phase = phase
+        self.details = dict(details or {})
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def sha256_json(value: Any) -> str:
+    return sha256_bytes(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8"))
+
+
+def sanitize(value: Any) -> Any:
+    if isinstance(value, Mapping):
+        result: dict[str, Any] = {}
+        for key, inner in value.items():
+            lowered = str(key).lower()
+            if any(word in lowered for word in ("token", "secret", "credential", "authorization", "password", "api_key", "apikey", "access_key")):
+                result[str(key)] = "[REDACTED]"
+            else:
+                result[str(key)] = sanitize(inner)
+        return result
+    if isinstance(value, list):
+        return [sanitize(item) for item in value]
+    if isinstance(value, tuple):
+        return [sanitize(item) for item in value]
+    if isinstance(value, str):
+        return SENSITIVE_RE.sub("[REDACTED]", value)
+    return value
+
+
+def audit(event_type: str, **fields: Any) -> None:
+    STATE_ROOT.mkdir(parents=True, mode=0o750, exist_ok=True)
+    path = STATE_ROOT / "events.jsonl"
+    record = {
+        "event_type": event_type,
+        "created_at": now(),
+        "agent_id": AGENT_ID,
+        "agent_version": AGENT_VERSION,
+        "credential_material_exposed": False,
+        "writer_lease_material_exposed": False,
+        **sanitize(fields),
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
+    os.chmod(path, 0o640)
+    print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
+
+
+def atomic_json(path: Path, value: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, mode=0o750, exist_ok=True)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise WorkerError("STATE_PATH_SYMLINK_REJECTED", phase="state")
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
+    try:
+        os.fchmod(fd, 0o640)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(sanitize(dict(value)), handle, ensure_ascii=False, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def state_path(task_id: str) -> Path:
+    safe = re.sub(r"[^A-Za-z0-9._:-]+", "-", task_id)[:180]
+    return STATE_ROOT / "tasks" / f"{safe}.json"
+
+
+def load_state(task_id: str) -> dict[str, Any] | None:
+    path = state_path(task_id)
+    if not path.is_file() or path.is_symlink():
+        return None
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+    return value if isinstance(value, dict) else None
+
+
+def save_state(task_id: str, value: Mapping[str, Any]) -> None:
+    atomic_json(state_path(task_id), value)
+
+
+def api_call(method: str, route: str, token: str, body: Mapping[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
+    payload = None if body is None else json.dumps(dict(body), ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
+    if payload is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(API_BASE + route, data=payload, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(request, timeout=20) as response:
+            value = json.loads(response.read().decode("utf-8"))
+            return response.status, value if isinstance(value, dict) else {}
+    except urllib.error.HTTPError as exc:
+        try:
+            value = json.loads(exc.read(4096).decode("utf-8", "replace"))
+        except Exception:
+            value = {"error": "HTTP_ERROR"}
+        return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
+    except (urllib.error.URLError, TimeoutError):
+        return 599, {"error": "AGENT_TASK_API_UNAVAILABLE"}
+
+
+def bridge_call(route: str, body: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
+    if BRIDGE_BASE != "http://127.0.0.1:19998":
+        raise WorkerError(
+            "NON_STAGING_BRIDGE_REJECTED",
+            phase="startup",
+            details={"bridge_base": BRIDGE_BASE},
+        )
+    payload = json.dumps(dict(body), ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    request = urllib.request.Request(
+        BRIDGE_BASE + route,
+        data=payload,
+        headers={"Accept": "application/json", "Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=60) as response:
+            value = json.loads(response.read().decode("utf-8"))
+            return response.status, value if isinstance(value, dict) else {}
+    except urllib.error.HTTPError as exc:
+        try:
+            value = json.loads(exc.read(4096).decode("utf-8", "replace"))
+        except Exception:
+            value = {"error": "HTTP_ERROR"}
+        return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
+    except (urllib.error.URLError, TimeoutError):
+        return 599, {"error": "GOVERNED_WRITE_API_UNAVAILABLE"}
+
+
+def load_agent_token() -> str:
+    directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
+    path = directory / "agent_token"
+    if not path.is_file():
+        raise WorkerError("AGENT_RUNTIME_CREDENTIAL_UNAVAILABLE", phase="startup")
+    token = path.read_text(encoding="utf-8").strip()
+    if len(token) < 24:
+        raise WorkerError("AGENT_RUNTIME_CREDENTIAL_INVALID", phase="startup")
+    return token
+
+
+def available_adapters() -> tuple[str, ...]:
+    values: list[str] = []
+    for name, path in (("hermes", HERMES_SOCKET), ("codex", CODEX_SOCKET), ("openclaw", OPENCLAW_SOCKET)):
+        try:
+            if path.is_socket():
+                values.append(name)
+        except OSError:
+            pass
+    return tuple(values)
+
+
+def capabilities() -> tuple[str, ...]:
+    values = {
+        "online_agent_task_api", "task.poll", "task.claim", "task.heartbeat", "task.result",
+        "intent.task", "engineering.consumer", "engineering.execute", "engineering.autonomous",
+        "workspace.worktree", "git.read", "git.diff", "git.test",
+        "governed.write.branch", "governed.write.file",
+    }
+    values.update(f"agent.runtime.{name}" for name in available_adapters())
+    return tuple(sorted(values))
+
+
+class LeaseHeartbeat:
+    def __init__(self, task_id: str, agent_token: str, lease_token: str) -> None:
+        self.task_id = task_id
+        self.agent_token = agent_token
+        self.lease_token = lease_token
+        self.stage = "claimed"
+        self.failed: str | None = None
+        self.stop = threading.Event()
+        self.thread = threading.Thread(target=self._run, daemon=True, name=f"heartbeat-{task_id}")
+
+    def set_stage(self, stage: str) -> None:
+        self.stage = stage
+        audit("TASK_PROGRESS_STAGE", task_id=self.task_id, stage=stage)
+
+    def send(self) -> None:
+        status, response = api_call(
+            "POST",
+            f"/v1/tasks/{self.task_id}/heartbeat",
+            self.agent_token,
+            {
+                "lease_token": self.lease_token,
+                "lease_seconds": LEASE_SECONDS,
+                "progress": {
+                    "stage": self.stage,
+                    "event_type": "TASK_STARTED" if self.stage == "started" else "TASK_PROGRESS",
+                    "execution_mode": "autonomous-intent-v2",
+                    "production_side_effects": False,
+                    "main_write": False,
+                },
+            },
+        )
+        if 200 <= status < 300:
+            audit(
+                "TASK_HEARTBEAT",
+                task_id=self.task_id,
+                stage=self.stage,
+                api_status=status,
+                task_status=response.get("status"),
+                lease_expires_at=response.get("lease_expires_at"),
+            )
+        else:
+            self.failed = str(response.get("error") or status)
+            audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=self.stage, error=self.failed)
+
+    def _run(self) -> None:
+        while not self.stop.wait(HEARTBEAT_SECONDS):
+            try:
+                self.send()
+            except Exception as exc:
+                self.failed = type(exc).__name__
+                audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, error=self.failed)
+
+    def start(self) -> None:
+        self.thread.start()
+
+    def close(self) -> None:
+        self.stop.set()
+        self.thread.join(timeout=3)
+
+
+def ensure_roots() -> None:
+    if WORKSPACE_ROOT.is_symlink() or STATE_ROOT.is_symlink() or AUTHORITY_GIT_DIR.is_symlink():
+        raise WorkerError("ROOT_SYMLINK_REJECTED", phase="startup")
+    if not AUTHORITY_GIT_DIR.is_dir():
+        raise WorkerError("AUTHORITY_GIT_DIR_UNAVAILABLE", phase="startup")
+    STATE_ROOT.mkdir(parents=True, mode=0o750, exist_ok=True)
+    WORKSPACE_ROOT.mkdir(parents=True, mode=0o750, exist_ok=True)
+
+
+def run_git(args: Sequence[str], *, cwd: Path, timeout: int = 180) -> subprocess.CompletedProcess[bytes]:
+    completed = subprocess.run(
+        ["/usr/bin/git", *args],
+        cwd=str(cwd),
+        capture_output=True,
+        check=False,
+        timeout=timeout,
+        env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
+    )
+    if completed.returncode != 0:
+        raise WorkerError(
+            "GIT_COMMAND_FAILED",
+            phase="workspace",
+            details={"operation": args[0] if args else "", "exit_code": completed.returncode},
+        )
+    return completed
+
+
+def prepare_worktree(task_id: str, base_oid: str) -> tuple[Path, Path]:
+    ensure_roots()
+    run_root = Path(tempfile.mkdtemp(prefix=f"{task_id[:48]}-", dir=str(WORKSPACE_ROOT)))
+    os.chmod(run_root, 0o750)
+    clone = run_root / "repository"
+    worktree = run_root / "worktree"
+    run_git(["clone", "--no-local", "--no-hardlinks", str(AUTHORITY_GIT_DIR), str(clone)], cwd=run_root, timeout=300)
+    run_git(["-C", str(clone), "rev-parse", "--verify", f"{base_oid}^{{commit}}"], cwd=run_root)
+    run_git(["-C", str(clone), "worktree", "add", "--detach", str(worktree), base_oid], cwd=run_root, timeout=300)
+    return run_root, worktree
+
+
+def share_run_root(run_root: Path) -> None:
+    try:
+        gid = grp.getgrnam(SHARED_GROUP).gr_gid
+    except KeyError as exc:
+        raise WorkerError("RUNTIME_SHARED_GROUP_UNAVAILABLE", phase="workspace") from exc
+    root = run_root.resolve()
+    if not root.is_dir() or run_root.is_symlink():
+        raise WorkerError("RUNTIME_RUN_ROOT_INVALID", phase="workspace")
+    for current, dirs, files in os.walk(root, topdown=True, followlinks=False):
+        current_path = Path(current)
+        for name in dirs + files:
+            candidate = current_path / name
+            if candidate.is_symlink():
+                raise WorkerError(
+                    "RUNTIME_WORKSPACE_SYMLINK_REJECTED",
+                    phase="workspace",
+                    details={"path": str(candidate.relative_to(root))},
+                )
+        for candidate in [current_path, *(current_path / name for name in dirs), *(current_path / name for name in files)]:
+            st = candidate.stat()
+            os.chown(candidate, -1, gid)
+            mode = st.st_mode & 0o777
+            os.chmod(candidate, mode | (0o070 if candidate.is_dir() else 0o060))
+
+
+def cleanup_run_root(run_root: Path) -> None:
+    root = WORKSPACE_ROOT.resolve()
+    candidate = run_root.resolve(strict=False)
+    if candidate == root or root not in candidate.parents:
+        raise WorkerError("CLEANUP_SCOPE_INVALID", phase="cleanup")
+    if candidate.exists():
+        shutil.rmtree(candidate)
+
+
+def allowed_programs() -> set[str]:
+    raw = os.environ.get("AIOS_AUTONOMOUS_ALLOWED_PROGRAMS_JSON", '["/usr/bin/git","/usr/bin/python3"]')
+    try:
+        value = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise WorkerError("ALLOWED_PROGRAMS_CONFIG_INVALID", phase="startup") from exc
+    if not isinstance(value, list) or not all(isinstance(item, str) and item.startswith("/") for item in value):
+        raise WorkerError("ALLOWED_PROGRAMS_CONFIG_INVALID", phase="startup")
+    return set(value)
+
+
+def validate_verification_argv(value: Any) -> list[str]:
+    if not isinstance(value, list) or not value or not all(isinstance(item, str) and item for item in value):
+        raise WorkerError("VERIFICATION_ARGV_INVALID", phase="verification")
+    argv = list(value)
+    if argv[0] not in allowed_programs():
+        raise WorkerError("VERIFICATION_PROGRAM_NOT_ALLOWED", phase="verification", details={"program": Path(argv[0]).name})
+    if argv[0].endswith("/git") and any(item in {"push", "update-ref", "receive-pack", "send-pack", "commit", "reset", "clean"} for item in argv[1:]):
+        raise WorkerError("VERIFICATION_GIT_MUTATION_REJECTED", phase="verification")
+    if argv[0].endswith("/python3") and "-c" in argv:
+        raise WorkerError("VERIFICATION_INLINE_PYTHON_REJECTED", phase="verification")
+    return argv
+
+
+def run_verification(label: str, argv: Sequence[str], *, cwd: Path, timeout: int) -> dict[str, Any]:
+    started = time.monotonic()
+    completed = subprocess.run(
+        list(argv),
+        cwd=str(cwd),
+        capture_output=True,
+        check=False,
+        timeout=timeout,
+        env={"PATH": "/usr/local/bin:/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
+    )
+    result = {
+        "label": label[:120],
+        "argv": [Path(item).name if idx == 0 else item for idx, item in enumerate(argv)],
+        "exit_code": completed.returncode,
+        "stdout_sha256": sha256_bytes(completed.stdout),
+        "stderr_sha256": sha256_bytes(completed.stderr),
+        "duration_ms": int((time.monotonic() - started) * 1000),
+    }
+    audit("VERIFICATION_RESULT", **result)
+    if completed.returncode != 0:
+        raise WorkerError(
+            "VERIFICATION_FAILED",
+            phase="verification",
+            details={"label": label[:120], "exit_code": completed.returncode},
+        )
+    return result
+
+
+def independent_verification(
+    worktree: Path,
+    changed_files: Sequence[str],
+    runtime_verification: Sequence[Mapping[str, Any]],
+) -> list[dict[str, Any]]:
+    results = [
+        run_verification(
+            "git_diff_check",
+            ["/usr/bin/git", "-C", str(worktree), "diff", "--check"],
+            cwd=worktree,
+            timeout=120,
+        )
+    ]
+    py_files = [item for item in changed_files if item.endswith(".py")]
+    if py_files:
+        results.append(
+            run_verification(
+                "changed_python_compile",
+                ["/usr/bin/python3", "-m", "py_compile", *py_files],
+                cwd=worktree,
+                timeout=300,
+            )
+        )
+    test_files = [
+        item for item in py_files
+        if item.startswith("tests/") or Path(item).name.startswith("test_")
+    ]
+    if test_files:
+        results.append(
+            run_verification(
+                "changed_pytest",
+                ["/usr/bin/python3", "-m", "pytest", "-q", *test_files],
+                cwd=worktree,
+                timeout=900,
+            )
+        )
+    seen: set[str] = set()
+    for item in runtime_verification:
+        argv = validate_verification_argv(item.get("argv"))
+        cwd_value = item.get("cwd", ".")
+        if cwd_value in {"", "."}:
+            cwd = worktree
+        else:
+            if not isinstance(cwd_value, str) or cwd_value.startswith(("/", "~")) or ".." in Path(cwd_value).parts:
+                raise WorkerError("VERIFICATION_CWD_INVALID", phase="verification")
+            cwd = (worktree / cwd_value).resolve()
+            if worktree.resolve() not in cwd.parents and cwd != worktree.resolve():
+                raise WorkerError("VERIFICATION_CWD_ESCAPE", phase="verification")
+        timeout = item.get("timeout_seconds", 300)
+        if not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 <= timeout <= 1800:
+            raise WorkerError("VERIFICATION_TIMEOUT_INVALID", phase="verification")
+        key = sha256_json({"argv": argv, "cwd": str(cwd), "timeout": timeout})
+        if key in seen:
+            continue
+        seen.add(key)
+        results.append(
+            run_verification(str(item.get("label") or "agent_requested_verification"), argv, cwd=cwd, timeout=timeout)
+        )
+    return results
+
+
+def governed_commit(
+    task_id: str,
+    resolution: Resolution,
+    worktree: Path,
+    changed_files: Sequence[str],
+) -> dict[str, Any]:
+    files: dict[str, str] = {}
+    for relative in changed_files:
+        path = (worktree / relative).resolve()
+        if worktree.resolve() not in path.parents or not path.is_file():
+            raise WorkerError("COMMIT_FILE_INVALID", phase="committing", details={"path": relative})
+        files[relative] = base64.b64encode(path.read_bytes()).decode("ascii")
+
+    governed_task_id = f"{task_id}-writer"
+    create_key = f"{task_id}-governed-v2"
+    commit_key = f"{task_id}-commit-v2"
+    status, created = bridge_call(
+        "/v2/governed-task",
+        {
+            "task_id": governed_task_id,
+            "repository": resolution.repository,
+            "base_oid": resolution.base_oid,
+            "target_branch": resolution.target_branch,
+            "risk": "engineering",
+            "description": f"Autonomous intent task {task_id}",
+            "idempotency_key": create_key,
+        },
+    )
+    if not 200 <= status < 300:
+        raise WorkerError(
+            "GOVERNED_TASK_CREATE_FAILED",
+            phase="committing",
+            details={"api_status": status, "error": created.get("error")},
+        )
+
+    generation = int(created.get("generation") or 1)
+    expected = str(created.get("expected_branch_oid") or resolution.base_oid)
+    branch_mode = str(created.get("branch_mode") or "CREATE_NEW")
+    status, lease = bridge_call(
+        "/v2/writer-lease",
+        {
+            "schema": "aios.task-writer-lease.v2",
+            "task_id": governed_task_id,
+            "repository": resolution.repository,
+            "branch_mode": branch_mode,
+            "base_oid": resolution.base_oid,
+            "expected_branch_oid": expected,
+            "target_branch": resolution.target_branch,
+            "generation": generation,
+            "lease_seconds": 900,
+        },
+    )
+    if not 200 <= status < 300:
+        raise WorkerError(
+            "WRITER_LEASE_ACQUIRE_FAILED",
+            phase="committing",
+            details={"api_status": status, "error": lease.get("error")},
+        )
+
+    status, committed = bridge_call(
+        "/v2/commit-files",
+        {
+            "task_id": governed_task_id,
+            "repository": resolution.repository,
+            "expected_branch_oid": expected,
+            "target_branch": resolution.target_branch,
+            "generation": generation,
+            "idempotency_key": commit_key,
+            "files": files,
+            "message": f"chore(agent): execute online intent task {task_id}"[:240],
+        },
+    )
+    if not 200 <= status < 300:
+        raise WorkerError(
+            "GOVERNED_COMMIT_FAILED",
+            phase="committing",
+            details={"api_status": status, "error": committed.get("error")},
+        )
+    commit_oid = committed.get("commit_oid")
+    if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
+        raise WorkerError("GOVERNED_COMMIT_OID_INVALID", phase="committing")
+    return {
+        "governed_task_id": governed_task_id,
+        "generation": generation,
+        "branch_mode": branch_mode,
+        "commit_oid": commit_oid,
+        "commit_visible": committed.get("commit_visible"),
+        "target_branch": resolution.target_branch,
+        "create_idempotency_key": create_key,
+        "commit_idempotency_key": commit_key,
+    }
+
+
+def socket_for(adapter: str) -> Path:
+    if adapter == "hermes":
+        return HERMES_SOCKET
+    if adapter == "codex":
+        return CODEX_SOCKET
+    if adapter == "openclaw":
+        return OPENCLAW_SOCKET
+    raise WorkerError("ADAPTER_INVALID", phase="runtime")
+
+
+def resolve_task(task: Mapping[str, Any], prior: Mapping[str, Any] | None = None) -> Resolution:
+    try:
+        return resolve(
+            task,
+            authority_git_dir=AUTHORITY_GIT_DIR,
+            repository=REPOSITORY,
+            base_ref=BASE_REF,
+            available_adapters=available_adapters(),
+            default_adapter=DEFAULT_ADAPTER,
+            prior=prior,
+        )
+    except IntentError as exc:
+        raise WorkerError(exc.code, phase="intent_resolution", details=exc.details) from exc
+
+
+def execute_claim(task: Mapping[str, Any], claim: Mapping[str, Any], token: str) -> int:
+    task_id = str(claim.get("task_id") or task.get("task_id") or "")
+    lease_token = claim.get("lease_token")
+    if not isinstance(lease_token, str) or len(lease_token) < 24:
+        raise WorkerError("LEASE_TOKEN_NOT_RETURNED", phase="claim")
+    attempt = int(claim.get("attempt") or 0)
+    state = load_state(task_id) or {"task_id": task_id, "phase": "claimed"}
+    prior = state.get("resolution") if isinstance(state.get("resolution"), Mapping) else None
+    resolution = resolve_task(task, prior=prior)
+    state["resolution"] = resolution.to_json()
+    state["attempt"] = attempt
+    save_state(task_id, state)
+
+    heartbeat = LeaseHeartbeat(task_id, token, lease_token)
+    heartbeat.start()
+    run_root: Path | None = None
+    try:
+        heartbeat.set_stage("started")
+        heartbeat.send()
+        audit("TASK_STARTED", task_id=task_id, attempt=attempt, adapter=resolution.adapter)
+
+        existing_commit = state.get("governed")
+        if isinstance(existing_commit, Mapping) and isinstance(existing_commit.get("commit_oid"), str):
+            result = {
+                "status": "SUCCEEDED",
+                "verification_state": "VERIFIED_BY_WORKER_READBACK_PENDING_SERVER",
+                "task_id": task_id,
+                "execution_mode": "autonomous-intent-v2",
+                "adapter": resolution.adapter,
+                "resolution_digest": resolution.resolution_digest,
+                "governed": dict(existing_commit),
+                "recovered_after_commit": True,
+                "secret_material_exposed": False,
+                "direct_main_write": False,
+                "force_push": False,
+                "production_write": False,
+            }
+        else:
+            heartbeat.set_stage("workspace_preparing")
+            run_root, worktree = prepare_worktree(task_id, resolution.base_oid)
+            share_run_root(run_root)
+            state["phase"] = "workspace_prepared"
+            save_state(task_id, state)
+
+            heartbeat.set_stage("agent_executing")
+            payload = request_payload(
+                adapter=resolution.adapter,
+                task_id=task_id,
+                goal=resolution.goal,
+                instructions=resolution.instructions,
+                success_criteria=resolution.success_criteria,
+                repository=resolution.repository,
+                base_oid=resolution.base_oid,
+                target_branch=resolution.target_branch,
+                worktree=worktree,
+            )
+            runtime = invoke(
+                socket_for(resolution.adapter),
+                payload,
+                timeout_seconds=RUNTIME_TIMEOUT_SECONDS,
+            )
+            heartbeat.set_stage("validating")
+            changed_files = collect_changes(worktree, resolution.base_oid)
+            scan_secrets(worktree, changed_files)
+            verification = independent_verification(worktree, changed_files, runtime.verification)
+
+            heartbeat.set_stage("committing")
+            governed = governed_commit(task_id, resolution, worktree, changed_files)
+            state.update({"phase": "commit_done", "governed": governed, "changed_files": list(changed_files)})
+            save_state(task_id, state)
+
+            result = {
+                "status": "SUCCEEDED",
+                "verification_state": "VERIFIED_BY_WORKER_READBACK_PENDING_SERVER",
+                "task_id": task_id,
+                "execution_mode": "autonomous-intent-v2",
+                "adapter": resolution.adapter,
+                "resolution_digest": resolution.resolution_digest,
+                "base_oid": resolution.base_oid,
+                "target_branch": resolution.target_branch,
+                "changed_files": list(changed_files),
+                "tests": verification,
+                "agent_summary": runtime.summary,
+                "agent_evidence": sanitize(runtime.evidence),
+                "governed": governed,
+                "secret_material_exposed": False,
+                "direct_main_write": False,
+                "force_push": False,
+                "production_write": False,
+            }
+
+        if heartbeat.failed:
+            raise WorkerError("HEARTBEAT_UNHEALTHY", phase="result_submitting", details={"error": heartbeat.failed})
+        heartbeat.set_stage("result_submitting")
+        result_key = f"{task_id}-result-v2"
+        status, response = api_call(
+            "POST",
+            f"/v1/tasks/{task_id}/result",
+            token,
+            {
+                "lease_token": lease_token,
+                "idempotency_key": result_key,
+                "result": result,
+            },
+        )
+        if not 200 <= status < 300:
+            raise WorkerError(
+                "TASK_RESULT_SUBMIT_FAILED",
+                phase="result_submitting",
+                details={"api_status": status, "error": response.get("error")},
+            )
+        audit("RESULT_SUBMITTED", task_id=task_id, task_status=response.get("status"), result_idempotency_key=result_key)
+        state["phase"] = "result_submitted"
+        save_state(task_id, state)
+        return 0
+    except (WorkerError, RuntimeErrorV2) as exc:
+        code = exc.code if hasattr(exc, "code") else type(exc).__name__
+        phase = exc.phase if isinstance(exc, WorkerError) else "runtime"
+        details = exc.details if hasattr(exc, "details") else {}
+        audit("AUTONOMOUS_EXECUTION_FAILED", task_id=task_id, error=code, phase=phase, details=details)
+        failure_status = "BLOCKED" if code in {
+            "AGENT_RUNTIME_UNAVAILABLE", "RUNTIME_UNAVAILABLE", "RUNTIME_BLOCKED",
+            "GOVERNED_WRITE_API_UNAVAILABLE", "NON_STAGING_BRIDGE_REJECTED",
+        } else "FAILED_RETRYABLE"
+        failure = {
+            "status": failure_status,
+            "verification_state": "UNVERIFIED",
+            "task_id": task_id,
+            "execution_mode": "autonomous-intent-v2",
+            "error": code,
+            "phase": phase,
+            "details": sanitize(details),
+            "secret_material_exposed": False,
+            "direct_main_write": False,
+            "force_push": False,
+            "production_write": False,
+        }
+        if not heartbeat.failed:
+            status, response = api_call(
+                "POST",
+                f"/v1/tasks/{task_id}/result",
+                token,
+                {
+                    "lease_token": lease_token,
+                    "idempotency_key": f"{task_id}-attempt-{attempt}-failure-v2",
+                    "result": failure,
+                },
+            )
+            audit("FAILURE_RESULT_SUBMITTED", task_id=task_id, api_status=status, task_status=response.get("status"))
+        return 20
+    finally:
+        heartbeat.close()
+        if run_root is not None:
+            try:
+                cleanup_run_root(run_root)
+                audit("WORKSPACE_CLEANED", task_id=task_id, cleanup="PASS")
+            except Exception as exc:
+                audit("WORKSPACE_CLEANUP_FAILED", task_id=task_id, error=type(exc).__name__)
+
+
+def worker_once(token: str) -> int:
+    status, poll = api_call("POST", "/v1/tasks/poll", token, {"limit": 100})
+    if status != 200:
+        audit("TASK_POLL_FAILED", api_status=status, error=poll.get("error"))
+        return 10
+
+    worker_caps = capabilities()
+    candidates: list[Mapping[str, Any]] = []
+    for item in poll.get("tasks", []):
+        if not isinstance(item, Mapping):
+            continue
+        if not is_claimable_intent(item, worker_caps):
+            continue
+        try:
+            # Read-only preflight. Do not consume an attempt if runtime/base is unavailable.
+            resolve_task(item, prior=(load_state(str(item.get("task_id") or "")) or {}).get("resolution"))
+        except WorkerError as exc:
+            audit("TASK_SKIPPED_PREFLIGHT", task_id=item.get("task_id"), error=exc.code)
+            continue
+        candidates.append(item)
+
+    if not candidates:
+        return 0
+
+    # Server priority/order remains authoritative; take the first visible compatible task.
+    task = candidates[0]
+    task_id = str(task.get("task_id") or "")
+    status, claim = api_call("POST", f"/v1/tasks/{task_id}/claim", token, {"lease_seconds": LEASE_SECONDS})
+    if status != 200:
+        audit("TASK_CLAIM_FAILED", task_id=task_id, api_status=status, error=claim.get("error"))
+        return 11
+    audit(
+        "TASK_CLAIMED",
+        task_id=task_id,
+        attempt=claim.get("attempt"),
+        revision=claim.get("revision"),
+        claim_mode="server_atomic",
+        execution_mode="autonomous-intent-v2",
+    )
+    merged = dict(task)
+    merged.update({key: value for key, value in claim.items() if key not in {"lease_token"}})
+    merged["task_id"] = task_id
+    return execute_claim(merged, claim, token)
+
+
+def main() -> int:
+    ensure_roots()
+    if BRIDGE_BASE != "http://127.0.0.1:19998":
+        raise WorkerError("NON_STAGING_BRIDGE_REJECTED", phase="startup", details={"bridge_base": BRIDGE_BASE})
+    token = load_agent_token()
+    audit(
+        "AUTONOMOUS_WORKER_READY",
+        poll_mode="continuous",
+        claim_mode="server_atomic",
+        task_scope="intent-only",
+        bridge_base=BRIDGE_BASE,
+        available_adapters=list(available_adapters()),
+        capabilities=list(capabilities()),
+        main_write=False,
+        production_write=False,
+        force_push=False,
+    )
+    while True:
+        try:
+            result = worker_once(token)
+        except Exception as exc:
+            audit("WORKER_LOOP_ERROR", error=type(exc).__name__)
+            result = 20
+        time.sleep(POLL_SECONDS if result == 0 else max(POLL_SECONDS, 5))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-autonomous-agent-worker/intent_runtime.py

@@ -0,0 +1,273 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+from typing import Any, Mapping, Sequence
+
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+REPOSITORY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{2,199}$")
+ALLOWED_ADAPTERS = frozenset({"hermes", "codex", "openclaw"})
+
+
+class IntentError(RuntimeError):
+    def __init__(self, code: str, *, details: Mapping[str, Any] | None = None) -> None:
+        super().__init__(code)
+        self.code = code
+        self.details = dict(details or {})
+
+
+def stable_digest(value: Any) -> str:
+    raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(raw).hexdigest()
+
+
+def _text(value: Any, field: str, limit: int) -> str:
+    if not isinstance(value, str):
+        raise IntentError(f"{field.upper()}_INVALID")
+    value = value.strip()
+    if not value or len(value) > limit:
+        raise IntentError(f"{field.upper()}_INVALID")
+    return value
+
+
+def _string_list(value: Any, field: str, *, required: bool) -> tuple[str, ...]:
+    if value is None and not required:
+        return ()
+    if not isinstance(value, list) or len(value) > 64 or (required and not value):
+        raise IntentError(f"{field.upper()}_INVALID")
+    return tuple(_text(item, field, 4096) for item in value)
+
+
+@dataclass(frozen=True)
+class IntentTask:
+    task_id: str
+    goal: str
+    instructions: str
+    success_criteria: tuple[str, ...]
+    required_capabilities: tuple[str, ...]
+    digest: str
+
+    @classmethod
+    def parse(cls, task: Mapping[str, Any]) -> "IntentTask":
+        task_id = str(task.get("task_id") or "")
+        if not TASK_ID_RE.fullmatch(task_id):
+            raise IntentError("TASK_ID_INVALID")
+
+        metadata = task.get("metadata")
+        if isinstance(metadata, Mapping) and isinstance(metadata.get("execution"), Mapping):
+            raise IntentError("STRUCTURED_TASK_NOT_INTENT")
+
+        goal = _text(task.get("goal"), "goal", 16384)
+        instructions = _text(
+            task.get("instructions") or "Inspect the repository, implement the goal, and satisfy the success criteria.",
+            "instructions",
+            65536,
+        )
+        success = _string_list(task.get("success_criteria"), "success_criteria", required=True)
+        capabilities = _string_list(task.get("required_capabilities") or [], "required_capabilities", required=False)
+
+        required_markers = {"intent.task", "engineering.execute"}
+        missing = sorted(required_markers - set(capabilities))
+        if missing:
+            raise IntentError("INTENT_CAPABILITY_MARKER_REQUIRED", details={"missing": missing})
+
+        canonical = {
+            "task_id": task_id,
+            "goal": goal,
+            "instructions": instructions,
+            "success_criteria": success,
+            "required_capabilities": sorted(capabilities),
+        }
+        return cls(
+            task_id=task_id,
+            goal=goal,
+            instructions=instructions,
+            success_criteria=success,
+            required_capabilities=capabilities,
+            digest=stable_digest(canonical),
+        )
+
+
+@dataclass(frozen=True)
+class Resolution:
+    schema: str
+    task_id: str
+    repository: str
+    base_ref: str
+    base_oid: str
+    target_branch: str
+    adapter: str
+    intent_digest: str
+    goal: str
+    instructions: str
+    success_criteria: tuple[str, ...]
+    required_capabilities: tuple[str, ...]
+    resolution_digest: str
+
+    def to_json(self) -> dict[str, Any]:
+        value = asdict(self)
+        value["success_criteria"] = list(self.success_criteria)
+        value["required_capabilities"] = list(self.required_capabilities)
+        return value
+
+    @classmethod
+    def from_json(cls, value: Mapping[str, Any], *, intent: IntentTask) -> "Resolution":
+        if value.get("schema") != "aios.autonomous-intent-resolution.v2":
+            raise IntentError("RESOLUTION_SCHEMA_INVALID")
+        if value.get("task_id") != intent.task_id or value.get("intent_digest") != intent.digest:
+            raise IntentError("RESOLUTION_TASK_DRIFT")
+        result = cls(
+            schema=str(value["schema"]),
+            task_id=str(value["task_id"]),
+            repository=str(value["repository"]),
+            base_ref=str(value["base_ref"]),
+            base_oid=str(value["base_oid"]),
+            target_branch=str(value["target_branch"]),
+            adapter=str(value["adapter"]),
+            intent_digest=str(value["intent_digest"]),
+            goal=str(value["goal"]),
+            instructions=str(value["instructions"]),
+            success_criteria=tuple(str(item) for item in value.get("success_criteria") or []),
+            required_capabilities=tuple(str(item) for item in value.get("required_capabilities") or []),
+            resolution_digest=str(value["resolution_digest"]),
+        )
+        validate_resolution(result)
+        if result.resolution_digest != resolution_digest(result):
+            raise IntentError("RESOLUTION_DIGEST_INVALID")
+        return result
+
+
+def read_base_oid(authority_git_dir: Path, base_ref: str) -> str:
+    if not base_ref.startswith("refs/heads/"):
+        raise IntentError("BASE_REF_INVALID")
+    if authority_git_dir.is_symlink() or not authority_git_dir.is_dir():
+        raise IntentError("AUTHORITY_GIT_DIR_INVALID")
+    completed = subprocess.run(
+        ["/usr/bin/git", "--git-dir", str(authority_git_dir), "rev-parse", "--verify", f"{base_ref}^{{commit}}"],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=30,
+        env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
+    )
+    oid = completed.stdout.strip()
+    if completed.returncode != 0 or not OID_RE.fullmatch(oid):
+        raise IntentError("BASE_OID_UNAVAILABLE")
+    return oid
+
+
+def derive_branch(task_id: str, digest: str) -> str:
+    slug = re.sub(r"[^A-Za-z0-9._-]+", "-", task_id).strip(".-").lower() or "task"
+    slug = slug[:100]
+    branch = f"agent/intent/{slug}-{digest[:12]}"
+    if len(branch) > 180:
+        branch = branch[:167] + "-" + digest[:12]
+    if ".." in branch or branch.endswith(("/", ".", ".lock")):
+        raise IntentError("TARGET_BRANCH_INVALID")
+    return branch
+
+
+def select_adapter(capabilities: Sequence[str], available: Sequence[str], default_adapter: str) -> str:
+    advertised = set(available)
+    if not advertised.issubset(ALLOWED_ADAPTERS):
+        raise IntentError("ADAPTER_CONFIGURATION_INVALID")
+    requested = {
+        item.split(".", 2)[-1]
+        for item in capabilities
+        if item.startswith("agent.runtime.")
+    }
+    requested.discard("")
+    if len(requested) > 1:
+        raise IntentError("MULTIPLE_AGENT_RUNTIMES_REQUIRED")
+    adapter = next(iter(requested)) if requested else default_adapter
+    if adapter not in ALLOWED_ADAPTERS:
+        raise IntentError("AGENT_RUNTIME_INVALID")
+    if adapter not in advertised:
+        raise IntentError("AGENT_RUNTIME_UNAVAILABLE", details={"adapter": adapter})
+    return adapter
+
+
+def resolution_digest(value: Resolution) -> str:
+    return stable_digest({
+        "schema": value.schema,
+        "task_id": value.task_id,
+        "repository": value.repository,
+        "base_ref": value.base_ref,
+        "base_oid": value.base_oid,
+        "target_branch": value.target_branch,
+        "adapter": value.adapter,
+        "intent_digest": value.intent_digest,
+        "goal": value.goal,
+        "instructions": value.instructions,
+        "success_criteria": value.success_criteria,
+        "required_capabilities": sorted(value.required_capabilities),
+    })
+
+
+def validate_resolution(value: Resolution) -> None:
+    if not TASK_ID_RE.fullmatch(value.task_id):
+        raise IntentError("TASK_ID_INVALID")
+    if not REPOSITORY_RE.fullmatch(value.repository) or value.repository.startswith(("/", ".")):
+        raise IntentError("REPOSITORY_INVALID")
+    if not value.base_ref.startswith("refs/heads/"):
+        raise IntentError("BASE_REF_INVALID")
+    if not OID_RE.fullmatch(value.base_oid):
+        raise IntentError("BASE_OID_INVALID")
+    if not value.target_branch.startswith("agent/intent/"):
+        raise IntentError("TARGET_BRANCH_POLICY_INVALID")
+    if value.adapter not in ALLOWED_ADAPTERS:
+        raise IntentError("AGENT_RUNTIME_INVALID")
+
+
+def resolve(
+    task: Mapping[str, Any],
+    *,
+    authority_git_dir: Path,
+    repository: str,
+    base_ref: str,
+    available_adapters: Sequence[str],
+    default_adapter: str,
+    prior: Mapping[str, Any] | None = None,
+) -> Resolution:
+    intent = IntentTask.parse(task)
+    if prior is not None:
+        return Resolution.from_json(prior, intent=intent)
+    if not REPOSITORY_RE.fullmatch(repository) or repository.startswith(("/", ".")):
+        raise IntentError("REPOSITORY_INVALID")
+    base_oid = read_base_oid(authority_git_dir, base_ref)
+    adapter = select_adapter(intent.required_capabilities, available_adapters, default_adapter)
+    provisional = Resolution(
+        schema="aios.autonomous-intent-resolution.v2",
+        task_id=intent.task_id,
+        repository=repository,
+        base_ref=base_ref,
+        base_oid=base_oid,
+        target_branch=derive_branch(intent.task_id, intent.digest),
+        adapter=adapter,
+        intent_digest=intent.digest,
+        goal=intent.goal,
+        instructions=intent.instructions,
+        success_criteria=intent.success_criteria,
+        required_capabilities=intent.required_capabilities,
+        resolution_digest="",
+    )
+    result = Resolution(
+        **{**asdict(provisional), "resolution_digest": resolution_digest(provisional)}
+    )
+    validate_resolution(result)
+    return result
+
+
+def is_claimable_intent(task: Mapping[str, Any], worker_capabilities: Sequence[str]) -> bool:
+    try:
+        intent = IntentTask.parse(task)
+    except IntentError:
+        return False
+    required = set(intent.required_capabilities)
+    return required.issubset(set(worker_capabilities))

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

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

services/aios-autonomous-agent-worker/runtime_client.py

@@ -0,0 +1,224 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+from pathlib import Path
+import re
+import socket
+import subprocess
+from typing import Any, Mapping, Sequence
+
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+MAX_BYTES = 1024 * 1024
+MAX_CHANGED_FILES = 256
+MAX_FILE_BYTES = 4 * 1024 * 1024
+SENSITIVE_NAMES = frozenset({
+    ".env", ".env.local", ".env.production", "id_rsa", "id_ed25519",
+    "credentials", "credentials.json",
+})
+SENSITIVE_SUFFIXES = (".key", ".p12", ".pfx")
+SECRET_PATTERNS = (
+    re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
+    re.compile(rb"\bAKIA[0-9A-Z]{16}\b"),
+    re.compile(rb"\bghp_[A-Za-z0-9]{30,}\b"),
+    re.compile(rb"\bsk-[A-Za-z0-9_-]{24,}\b"),
+)
+
+
+class RuntimeErrorV2(RuntimeError):
+    def __init__(self, code: str, *, details: Mapping[str, Any] | None = None) -> None:
+        super().__init__(code)
+        self.code = code
+        self.details = dict(details or {})
+
+
+def safe_relative_path(value: str) -> str:
+    if (
+        not value
+        or value.startswith(("/", "~"))
+        or "\\" in value
+        or "\x00" in value
+        or any(part in {"", ".", ".."} for part in value.split("/"))
+    ):
+        raise RuntimeErrorV2("CHANGED_PATH_INVALID", details={"path": value[:160]})
+    lowered = value.split("/")[-1].lower()
+    if value.startswith(".git/") or value in {".git", ".gitmodules"}:
+        raise RuntimeErrorV2("GIT_METADATA_CHANGE_REJECTED", details={"path": value})
+    if lowered in SENSITIVE_NAMES or lowered.endswith(SENSITIVE_SUFFIXES):
+        raise RuntimeErrorV2("SENSITIVE_PATH_REJECTED", details={"path": value})
+    return value
+
+
+def git(worktree: Path, args: Sequence[str], timeout: int = 60) -> bytes:
+    completed = subprocess.run(
+        ["/usr/bin/git", "-C", str(worktree), *args],
+        capture_output=True,
+        check=False,
+        timeout=timeout,
+        env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
+    )
+    if completed.returncode != 0:
+        raise RuntimeErrorV2("WORKTREE_GIT_FAILED", details={"operation": args[0] if args else ""})
+    return completed.stdout
+
+
+def _zpaths(raw: bytes) -> list[str]:
+    result: list[str] = []
+    for item in raw.split(b"\0"):
+        if not item:
+            continue
+        try:
+            result.append(item.decode("utf-8"))
+        except UnicodeDecodeError as exc:
+            raise RuntimeErrorV2("WORKTREE_PATH_ENCODING_INVALID") from exc
+    return result
+
+
+def collect_changes(worktree: Path, base_oid: str) -> list[str]:
+    if not OID_RE.fullmatch(base_oid):
+        raise RuntimeErrorV2("BASE_OID_INVALID")
+    head = git(worktree, ["rev-parse", "HEAD"]).decode("ascii", "strict").strip()
+    if head != base_oid:
+        raise RuntimeErrorV2("WORKTREE_HEAD_MUTATED", details={"expected": base_oid, "actual": head})
+    tracked = _zpaths(git(worktree, ["diff", "--name-only", "--no-renames", "-z", base_oid, "--"]))
+    untracked = _zpaths(git(worktree, ["ls-files", "--others", "--exclude-standard", "-z"]))
+    changed = sorted({safe_relative_path(item) for item in tracked + untracked})
+    if not changed:
+        raise RuntimeErrorV2("AGENT_PRODUCED_NO_CHANGES")
+    if len(changed) > MAX_CHANGED_FILES:
+        raise RuntimeErrorV2("CHANGED_FILE_LIMIT_EXCEEDED")
+    root = worktree.resolve()
+    for relative in changed:
+        path = (root / relative).resolve(strict=False)
+        if root not in path.parents or not path.is_file() or path.is_symlink():
+            raise RuntimeErrorV2("CHANGED_FILE_UNSUPPORTED", details={"path": relative})
+        if path.stat().st_size > MAX_FILE_BYTES:
+            raise RuntimeErrorV2("CHANGED_FILE_TOO_LARGE", details={"path": relative})
+    return changed
+
+
+def scan_secrets(worktree: Path, changed: Sequence[str]) -> None:
+    root = worktree.resolve()
+    for relative in changed:
+        path = (root / safe_relative_path(relative)).resolve()
+        raw = path.read_bytes()
+        for pattern in SECRET_PATTERNS:
+            if pattern.search(raw):
+                raise RuntimeErrorV2("SECRET_MATERIAL_REJECTED", details={"path": relative})
+
+
+@dataclass(frozen=True)
+class RuntimeResult:
+    status: str
+    summary: str
+    verification: tuple[dict[str, Any], ...]
+    evidence: dict[str, Any]
+
+    @classmethod
+    def parse(cls, value: Mapping[str, Any]) -> "RuntimeResult":
+        if value.get("schema") != "aios.online-agent.runtime-result.v1":
+            raise RuntimeErrorV2("RUNTIME_RESULT_SCHEMA_INVALID")
+        status = str(value.get("status") or "")
+        if status not in {"COMPLETED", "BLOCKED", "FAILED"}:
+            raise RuntimeErrorV2("RUNTIME_RESULT_STATUS_INVALID")
+        summary = str(value.get("summary") or "").strip()
+        if not summary or len(summary) > 8192:
+            raise RuntimeErrorV2("RUNTIME_RESULT_SUMMARY_INVALID")
+        verification = value.get("verification") or []
+        evidence = value.get("evidence") or {}
+        if not isinstance(verification, list) or len(verification) > 64:
+            raise RuntimeErrorV2("RUNTIME_VERIFICATION_INVALID")
+        if not isinstance(evidence, Mapping):
+            raise RuntimeErrorV2("RUNTIME_EVIDENCE_INVALID")
+        return cls(status, summary, tuple(dict(item) for item in verification if isinstance(item, Mapping)), dict(evidence))
+
+
+def request_payload(
+    *,
+    adapter: str,
+    task_id: str,
+    goal: str,
+    instructions: str,
+    success_criteria: Sequence[str],
+    repository: str,
+    base_oid: str,
+    target_branch: str,
+    worktree: Path,
+) -> dict[str, Any]:
+    if adapter not in {"hermes", "codex", "openclaw"}:
+        raise RuntimeErrorV2("ADAPTER_INVALID")
+    if not OID_RE.fullmatch(base_oid):
+        raise RuntimeErrorV2("BASE_OID_INVALID")
+    return {
+        "schema": "aios.online-agent.runtime-request.v1",
+        "adapter": adapter,
+        "task": {
+            "task_id": task_id,
+            "goal": goal,
+            "instructions": instructions,
+            "success_criteria": list(success_criteria),
+        },
+        "execution_context": {
+            "repository": repository,
+            "base_oid": base_oid,
+            "target_branch": target_branch,
+            "worktree": str(worktree),
+        },
+        "policy": {
+            "authoritative_repository_write": False,
+            "direct_git_ref_mutation": False,
+            "direct_git_push": False,
+            "direct_git_commit": False,
+            "main_write": False,
+            "force_push": False,
+            "production_write": False,
+            "tag_write": False,
+            "baseline_write": False,
+            "task_api_credential_visible": False,
+            "writer_lease_visible": False,
+        },
+    }
+
+
+def invoke(socket_path: Path, payload: Mapping[str, Any], timeout_seconds: int) -> RuntimeResult:
+    if not socket_path.is_absolute():
+        raise RuntimeErrorV2("RUNTIME_SOCKET_NOT_ABSOLUTE")
+    if not 1 <= timeout_seconds <= 7200:
+        raise RuntimeErrorV2("RUNTIME_TIMEOUT_INVALID")
+    raw = json.dumps(dict(payload), ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
+    if len(raw) > MAX_BYTES:
+        raise RuntimeErrorV2("RUNTIME_REQUEST_TOO_LARGE")
+
+    client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    client.settimeout(timeout_seconds)
+    try:
+        client.connect(str(socket_path))
+        client.sendall(raw)
+        client.shutdown(socket.SHUT_WR)
+        chunks: list[bytes] = []
+        total = 0
+        while True:
+            chunk = client.recv(65536)
+            if not chunk:
+                break
+            total += len(chunk)
+            if total > MAX_BYTES:
+                raise RuntimeErrorV2("RUNTIME_RESPONSE_TOO_LARGE")
+            chunks.append(chunk)
+    except (OSError, socket.timeout) as exc:
+        raise RuntimeErrorV2("RUNTIME_UNAVAILABLE", details={"socket": str(socket_path)}) from exc
+    finally:
+        client.close()
+    try:
+        value = json.loads(b"".join(chunks).decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeErrorV2("RUNTIME_RESPONSE_INVALID") from exc
+    if not isinstance(value, Mapping):
+        raise RuntimeErrorV2("RUNTIME_RESPONSE_INVALID")
+    result = RuntimeResult.parse(value)
+    if result.status == "BLOCKED":
+        raise RuntimeErrorV2("RUNTIME_BLOCKED", details={"summary": result.summary})
+    if result.status == "FAILED":
+        raise RuntimeErrorV2("RUNTIME_FAILED", details={"summary": result.summary})
+    return result

services/aios-autonomous-agent-worker/runtime_selftest.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+
+from runtime_client import collect_changes, invoke, request_payload, scan_secrets
+
+
+WORKSPACE_ROOT = Path(os.environ.get(
+    "AIOS_AUTONOMOUS_WORKSPACE_ROOT",
+    "/var/lib/aios-autonomous-agent-worker/workspaces",
+))
+ADAPTER = os.environ.get("AIOS_AUTONOMOUS_SELFTEST_ADAPTER", "openclaw")
+SOCKET = Path(os.environ.get(
+    "AIOS_AUTONOMOUS_SELFTEST_BROKER_SOCKET",
+    f"/run/aios-agent-runtime/{ADAPTER}.sock",
+))
+
+
+def run(argv: list[str], cwd: Path) -> str:
+    completed = subprocess.run(
+        argv,
+        cwd=str(cwd),
+        text=True,
+        capture_output=True,
+        check=False,
+        timeout=60,
+        env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
+    )
+    if completed.returncode != 0:
+        raise RuntimeError(f"SELFTEST_COMMAND_FAILED:{argv[0]}:{completed.returncode}")
+    return completed.stdout.strip()
+
+
+def main() -> int:
+    if not SOCKET.is_socket():
+        raise RuntimeError("AUTONOMOUS_ENGINEERING_BACKEND_UNAVAILABLE")
+    WORKSPACE_ROOT.mkdir(parents=True, mode=0o750, exist_ok=True)
+    run_root = Path(tempfile.mkdtemp(prefix="runtime-selftest-", dir=str(WORKSPACE_ROOT)))
+    repo = run_root / "worktree"
+    repo.mkdir()
+    try:
+        run(["/usr/bin/git", "init", str(repo)], run_root)
+        run(["/usr/bin/git", "-C", str(repo), "config", "user.email", "selftest@example.invalid"], run_root)
+        run(["/usr/bin/git", "-C", str(repo), "config", "user.name", "AIOS Runtime Selftest"], run_root)
+        marker = repo / "agent_runtime_selftest.txt"
+        marker.write_text("VALUE=1\n", encoding="utf-8")
+        run(["/usr/bin/git", "-C", str(repo), "add", "agent_runtime_selftest.txt"], run_root)
+        run(["/usr/bin/git", "-C", str(repo), "commit", "-m", "selftest base"], run_root)
+        base_oid = run(["/usr/bin/git", "-C", str(repo), "rev-parse", "HEAD"], run_root)
+
+        payload = request_payload(
+            adapter=ADAPTER,
+            task_id="autonomous-runtime-selftest",
+            goal="Change agent_runtime_selftest.txt so it contains exactly VALUE=2 followed by a newline.",
+            instructions=(
+                "Modify only agent_runtime_selftest.txt in the provided disposable worktree. "
+                "Do not commit, push, change Git refs, or access task/governed-write credentials."
+            ),
+            success_criteria=[
+                "agent_runtime_selftest.txt contains VALUE=2",
+                "Git HEAD remains unchanged",
+            ],
+            repository="selftest/disposable",
+            base_oid=base_oid,
+            target_branch="agent/intent/runtime-selftest",
+            worktree=repo,
+        )
+        result = invoke(SOCKET, payload, timeout_seconds=300)
+        changed = collect_changes(repo, base_oid)
+        scan_secrets(repo, changed)
+        content = marker.read_text(encoding="utf-8")
+        head = run(["/usr/bin/git", "-C", str(repo), "rev-parse", "HEAD"], run_root)
+        if content != "VALUE=2\n":
+            raise RuntimeError("RUNTIME_SELFTEST_CONTENT_MISMATCH")
+        if head != base_oid:
+            raise RuntimeError("RUNTIME_SELFTEST_HEAD_MUTATED")
+        if changed != ["agent_runtime_selftest.txt"]:
+            raise RuntimeError("RUNTIME_SELFTEST_UNEXPECTED_DIFF")
+        print(json.dumps({
+            "status": "PASS",
+            "adapter": ADAPTER,
+            "base_oid": base_oid,
+            "changed_files": changed,
+            "head_unchanged": True,
+            "runtime_summary": result.summary,
+            "task_api_token_visible_to_runtime": False,
+            "writer_lease_visible_to_runtime": False,
+        }, sort_keys=True))
+        return 0
+    finally:
+        shutil.rmtree(run_root, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

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

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

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

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

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

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

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

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

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

tests/test_autonomous_broker_security_v2.py

@@ -0,0 +1,201 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import sys
+import time
+
+import pytest
+
+
+BROKER_PATH = Path(__file__).parents[1] / "services" / "aios-agent-runtime-broker" / "broker.py"
+
+
+def load_broker():
+    spec = importlib.util.spec_from_file_location("broker_under_test", BROKER_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_peer_credentials_accept_configured_uid(monkeypatch) -> None:
+    broker = load_broker()
+    left, right = socket.socketpair()
+    try:
+        monkeypatch.setattr(broker, "ALLOWED_PEER_UID", os.getuid())
+        monkeypatch.setattr(broker, "ALLOWED_PEER_USER", "")
+        broker.authorize_peer(left)
+    finally:
+        left.close()
+        right.close()
+
+
+def test_peer_credentials_reject_other_uid(monkeypatch) -> None:
+    broker = load_broker()
+    left, right = socket.socketpair()
+    try:
+        monkeypatch.setattr(broker, "ALLOWED_PEER_UID", os.getuid() + 100000)
+        monkeypatch.setattr(broker, "ALLOWED_PEER_USER", "")
+        with pytest.raises(broker.BrokerError) as caught:
+            broker.authorize_peer(left)
+        assert caught.value.code == "PEER_UID_REJECTED"
+    finally:
+        left.close()
+        right.close()
+
+
+def test_request_read_timeout(monkeypatch) -> None:
+    broker = load_broker()
+    left, right = socket.socketpair()
+    try:
+        monkeypatch.setattr(broker, "CONNECTION_TIMEOUT", 0.05)
+        with pytest.raises(broker.BrokerError) as caught:
+            broker.read_request(left)
+        assert caught.value.code == "REQUEST_TIMEOUT"
+    finally:
+        left.close()
+        right.close()
+
+
+def test_worktree_fd_remains_pinned_after_path_replacement(tmp_path: Path, monkeypatch) -> None:
+    broker = load_broker()
+    root = tmp_path / "workspaces"
+    run_root = root / "run"
+    original = run_root / "worktree"
+    original.mkdir(parents=True)
+    (original / "marker.txt").write_text("original\n", encoding="utf-8")
+
+    request = {
+        "schema": "aios.online-agent.runtime-request.v1",
+        "adapter": "hermes",
+        "execution_context": {
+            "repository": "aios/aios",
+            "base_oid": "a" * 40,
+            "target_branch": "agent/intent/test",
+            "worktree": str(original),
+        },
+        "policy": {
+            "authoritative_repository_write": False,
+            "direct_git_ref_mutation": False,
+            "direct_git_push": False,
+            "direct_git_commit": False,
+            "main_write": False,
+            "force_push": False,
+            "production_write": False,
+            "tag_write": False,
+            "baseline_write": False,
+            "task_api_credential_visible": False,
+            "writer_lease_visible": False,
+        },
+    }
+    monkeypatch.setattr(broker, "ADAPTER", "hermes")
+    monkeypatch.setattr(broker, "WORKSPACE_ROOT", root)
+    _, fd = broker.validate_request(request)
+    try:
+        moved = run_root / "original-moved"
+        original.rename(moved)
+        original.mkdir()
+        (original / "marker.txt").write_text("replacement\n", encoding="utf-8")
+
+        pinned = Path(f"/proc/self/fd/{fd}")
+        (pinned / "through-fd.txt").write_text("pinned\n", encoding="utf-8")
+        assert (moved / "through-fd.txt").read_text(encoding="utf-8") == "pinned\n"
+        assert not (original / "through-fd.txt").exists()
+    finally:
+        os.close(fd)
+
+
+def test_backend_receives_no_task_api_token(tmp_path: Path, monkeypatch) -> None:
+    broker = load_broker()
+    worktree = tmp_path / "worktree"
+    worktree.mkdir()
+    backend = tmp_path / "backend.py"
+    backend.write_text(
+        """#!/usr/bin/env python3
+import json, os, pathlib, sys
+request = json.load(sys.stdin)
+assert 'AIOS_AGENT_TASK_TOKEN' not in os.environ
+worktree = pathlib.Path(request['execution_context']['worktree'])
+(worktree / 'changed.txt').write_text('ok\\n', encoding='utf-8')
+json.dump({
+  'schema':'aios.online-agent.runtime-result.v1',
+  'status':'COMPLETED',
+  'summary':'done',
+  'verification':[],
+  'evidence':{'token_visible':False, 'api_key':'runtime-secret-value'}
+}, sys.stdout)
+""",
+        encoding="utf-8",
+    )
+    backend.chmod(0o755)
+
+    monkeypatch.setattr(broker, "ADAPTER", "hermes")
+    monkeypatch.setattr(broker, "BACKEND_ARGV_JSON", json.dumps(["/usr/bin/python3", str(backend)]))
+    monkeypatch.setattr(broker, "BACKEND_TIMEOUT", 5)
+    monkeypatch.delenv("CREDENTIALS_DIRECTORY", raising=False)
+
+    fd = os.open(str(worktree), getattr(os, "O_PATH", os.O_RDONLY) | os.O_DIRECTORY)
+    try:
+        request = {
+            "schema": "aios.online-agent.runtime-request.v1",
+            "adapter": "hermes",
+            "execution_context": {"worktree": str(worktree)},
+        }
+        result = broker.execute_backend(request, fd)
+    finally:
+        os.close(fd)
+    assert result["status"] == "COMPLETED"
+    assert result["evidence"]["api_key"] == "[REDACTED]"
+    assert (worktree / "changed.txt").read_text(encoding="utf-8") == "ok\n"
+
+
+def test_broker_backend_env_does_not_inherit_task_token(tmp_path: Path, monkeypatch) -> None:
+    broker = load_broker()
+    worktree = tmp_path / "worktree-env"
+    worktree.mkdir()
+    backend = tmp_path / "backend-env.py"
+    backend.write_text(
+        """#!/usr/bin/env python3
+import json, os, sys
+assert 'AIOS_AGENT_TASK_TOKEN' not in os.environ
+assert os.environ.get('AIOS_OPENCLAW_MODEL') == 'test/model'
+assert os.environ.get('AIOS_OPENCLAW_BIN') == '/test/openclaw'
+json.dump({
+  'schema':'aios.online-agent.runtime-result.v1',
+  'status':'COMPLETED',
+  'summary':'environment isolated',
+  'verification':[],
+  'evidence':{'task_token_visible':False}
+}, sys.stdout)
+""",
+        encoding="utf-8",
+    )
+    backend.chmod(0o755)
+
+    monkeypatch.setattr(broker, "ADAPTER", "openclaw")
+    monkeypatch.setattr(
+        broker,
+        "BACKEND_ARGV_JSON",
+        json.dumps(["/usr/bin/python3", str(backend)]),
+    )
+    monkeypatch.setenv("AIOS_AGENT_TASK_TOKEN", "must-not-cross")
+    monkeypatch.setenv("AIOS_OPENCLAW_MODEL", "test/model")
+    monkeypatch.setenv("AIOS_OPENCLAW_BIN", "/test/openclaw")
+    monkeypatch.delenv("CREDENTIALS_DIRECTORY", raising=False)
+
+    fd = os.open(str(worktree), getattr(os, "O_PATH", os.O_RDONLY) | os.O_DIRECTORY)
+    try:
+        request = {
+            "schema": "aios.online-agent.runtime-request.v1",
+            "adapter": "openclaw",
+            "execution_context": {"worktree": str(worktree)},
+        }
+        result = broker.execute_backend(request, fd)
+    finally:
+        os.close(fd)
+    assert result["status"] == "COMPLETED"

tests/test_autonomous_intent_runtime_v2.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+from pathlib import Path
+import subprocess
+
+import pytest
+
+from intent_runtime import IntentError, IntentTask, is_claimable_intent, resolve
+
+
+def make_bare(tmp_path: Path) -> tuple[Path, str]:
+    work = tmp_path / "work"
+    bare = tmp_path / "repo.git"
+    subprocess.run(["git", "init", str(work)], check=True, capture_output=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.email", "test@example.invalid"], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "AIOS Test"], check=True)
+    (work / "README.md").write_text("base\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "README.md"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-m", "base"], check=True, capture_output=True)
+    subprocess.run(["git", "-C", str(work), "branch", "-M", "main"], check=True)
+    oid = subprocess.check_output(["git", "-C", str(work), "rev-parse", "HEAD"], text=True).strip()
+    subprocess.run(["git", "clone", "--bare", str(work), str(bare)], check=True, capture_output=True)
+    return bare, oid
+
+
+def intent_task(task_id: str = "custom-gpt-intent-20260815") -> dict[str, object]:
+    return {
+        "task_id": task_id,
+        "goal": "Fix the repository behavior described by the task.",
+        "instructions": "Inspect the code, find the root cause, implement and test the smallest correct fix.",
+        "success_criteria": ["Relevant tests pass", "git diff --check passes"],
+        "required_capabilities": [
+            "intent.task",
+            "engineering.execute",
+            "engineering.autonomous",
+            "git.read",
+            "git.diff",
+            "git.test",
+            "governed.write.file",
+            "governed.write.branch",
+            "workspace.worktree",
+            "agent.runtime.hermes",
+        ],
+        "metadata": {},
+    }
+
+
+def fixed_implementation_task() -> dict[str, object]:
+    return {
+        "task_id": "implement-autonomous-intent-worker-v2-20260815",
+        "goal": "Implement autonomous intent worker v2.",
+        "instructions": "Implement the worker.",
+        "success_criteria": ["Implementation succeeds"],
+        "required_capabilities": [
+            "engineering.execute",
+            "git.diff",
+            "git.read",
+            "git.test",
+            "governed.write.branch",
+            "governed.write.file",
+            "workspace.worktree",
+        ],
+        "metadata": {},
+    }
+
+
+def test_fixed_bootstrap_task_is_not_claimable_by_intent_worker() -> None:
+    worker_caps = intent_task()["required_capabilities"]
+    assert is_claimable_intent(fixed_implementation_task(), worker_caps) is False
+    with pytest.raises(IntentError) as caught:
+        IntentTask.parse(fixed_implementation_task())
+    assert caught.value.code == "INTENT_CAPABILITY_MARKER_REQUIRED"
+
+
+def test_native_custom_gpt_intent_task_is_claimable() -> None:
+    task = intent_task()
+    assert is_claimable_intent(task, task["required_capabilities"]) is True
+
+
+def test_missing_runtime_capability_fails_closed() -> None:
+    task = intent_task()
+    caps = [item for item in task["required_capabilities"] if item != "agent.runtime.hermes"]
+    assert is_claimable_intent(task, caps) is False
+
+
+def test_resolution_is_stable_across_restart(tmp_path: Path) -> None:
+    bare, oid = make_bare(tmp_path)
+    task = intent_task()
+    first = resolve(
+        task,
+        authority_git_dir=bare,
+        repository="aios/aios",
+        base_ref="refs/heads/main",
+        available_adapters=["hermes"],
+        default_adapter="hermes",
+    )
+    recovered = resolve(
+        task,
+        authority_git_dir=bare,
+        repository="aios/aios",
+        base_ref="refs/heads/main",
+        available_adapters=["hermes"],
+        default_adapter="hermes",
+        prior=first.to_json(),
+    )
+    assert first == recovered
+    assert recovered.base_oid == oid
+    assert recovered.target_branch.startswith("agent/intent/")
+
+
+def test_requested_codex_is_rejected_when_unavailable(tmp_path: Path) -> None:
+    bare, _ = make_bare(tmp_path)
+    task = intent_task("codex-only-20260815")
+    task["required_capabilities"] = [
+        item for item in task["required_capabilities"] if item != "agent.runtime.hermes"
+    ] + ["agent.runtime.codex"]
+    with pytest.raises(IntentError) as caught:
+        resolve(
+            task,
+            authority_git_dir=bare,
+            repository="aios/aios",
+            base_ref="refs/heads/main",
+            available_adapters=["hermes"],
+            default_adapter="hermes",
+        )
+    assert caught.value.code == "AGENT_RUNTIME_UNAVAILABLE"
+
+
+def test_openclaw_is_a_first_class_runtime_adapter(tmp_path: Path) -> None:
+    bare, _ = make_bare(tmp_path)
+    task = intent_task("openclaw-intent-20260815")
+    task["required_capabilities"] = [
+        item for item in task["required_capabilities"] if item != "agent.runtime.hermes"
+    ] + ["agent.runtime.openclaw"]
+    resolved = resolve(
+        task,
+        authority_git_dir=bare,
+        repository="aios/aios",
+        base_ref="refs/heads/main",
+        available_adapters=["openclaw"],
+        default_adapter="openclaw",
+    )
+    assert resolved.adapter == "openclaw"

tests/test_autonomous_runtime_client_v2.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import socket
+import subprocess
+import threading
+import time
+
+import pytest
+
+from runtime_client import RuntimeErrorV2, collect_changes, invoke, request_payload, scan_secrets
+
+
+def repo(tmp_path: Path) -> tuple[Path, str]:
+    work = tmp_path / "repo"
+    subprocess.run(["git", "init", str(work)], check=True, capture_output=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.email", "test@example.invalid"], check=True)
+    subprocess.run(["git", "-C", str(work), "config", "user.name", "AIOS Test"], check=True)
+    (work / "a.py").write_text("VALUE = 1\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "a.py"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-m", "base"], check=True, capture_output=True)
+    oid = subprocess.check_output(["git", "-C", str(work), "rev-parse", "HEAD"], text=True).strip()
+    return work, oid
+
+
+def test_collect_changes_reads_actual_git_state(tmp_path: Path) -> None:
+    work, oid = repo(tmp_path)
+    (work / "a.py").write_text("VALUE = 2\n", encoding="utf-8")
+    (work / "new.py").write_text("NEW = True\n", encoding="utf-8")
+    assert collect_changes(work, oid) == ["a.py", "new.py"]
+
+
+def test_agent_commit_is_rejected(tmp_path: Path) -> None:
+    work, oid = repo(tmp_path)
+    (work / "a.py").write_text("VALUE = 2\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(work), "add", "a.py"], check=True)
+    subprocess.run(["git", "-C", str(work), "commit", "-m", "agent commit"], check=True, capture_output=True)
+    with pytest.raises(RuntimeErrorV2) as caught:
+        collect_changes(work, oid)
+    assert caught.value.code == "WORKTREE_HEAD_MUTATED"
+
+
+def test_secret_material_is_rejected(tmp_path: Path) -> None:
+    work, oid = repo(tmp_path)
+    private_key_marker = "-----BEGIN PRIVATE " + "KEY-----"
+    (work / "new.txt").write_text(
+        private_key_marker + "\nabc\n", encoding="utf-8")
+    changed = collect_changes(work, oid)
+    with pytest.raises(RuntimeErrorV2) as caught:
+        scan_secrets(work, changed)
+    assert caught.value.code == "SECRET_MATERIAL_REJECTED"
+
+
+def test_runtime_protocol_round_trip(tmp_path: Path) -> None:
+    sock = tmp_path / "runtime.sock"
+    response = {
+        "schema": "aios.online-agent.runtime-result.v1",
+        "status": "COMPLETED",
+        "summary": "Implemented the requested change.",
+        "verification": [],
+        "evidence": {"runtime": "fake-hermes"},
+    }
+
+    def server() -> None:
+        listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        listener.bind(str(sock))
+        listener.listen(1)
+        conn, _ = listener.accept()
+        raw = b""
+        while True:
+            chunk = conn.recv(65536)
+            if not chunk:
+                break
+            raw += chunk
+        value = json.loads(raw.decode("utf-8"))
+        assert value["schema"] == "aios.online-agent.runtime-request.v1"
+        conn.sendall(json.dumps(response).encode("utf-8"))
+        conn.close()
+        listener.close()
+
+    thread = threading.Thread(target=server)
+    thread.start()
+    for _ in range(100):
+        if sock.exists():
+            break
+        time.sleep(0.01)
+    payload = request_payload(
+        adapter="hermes",
+        task_id="intent-e2e-20260815",
+        goal="Change VALUE",
+        instructions="Inspect and implement",
+        success_criteria=["Tests pass"],
+        repository="aios/aios",
+        base_oid="a" * 40,
+        target_branch="agent/intent/example",
+        worktree=tmp_path,
+    )
+    result = invoke(sock, payload, 5)
+    thread.join(timeout=2)
+    assert result.status == "COMPLETED"
+    assert result.evidence["runtime"] == "fake-hermes"

tests/test_openclaw_engineering_backend_v2.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+
+BACKEND_PATH = Path(__file__).parents[1] / "services" / "aios-agent-runtime-broker" / "openclaw_engineering_backend.py"
+
+
+def load_backend():
+    spec = importlib.util.spec_from_file_location("openclaw_backend_under_test", BACKEND_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def fake_openclaw(tmp_path: Path) -> Path:
+    path = tmp_path / "openclaw"
+    path.write_text(
+        """#!/usr/bin/env python3
+import json, os, pathlib, sys
+args = sys.argv[1:]
+assert args[:2] == ['agent', 'exec']
+def value(flag):
+    return args[args.index(flag) + 1]
+assert value('--message-file') == '-'
+cwd = pathlib.Path(value('--cwd'))
+assert cwd.resolve() == pathlib.Path.cwd().resolve()
+assert '--auth-env-only' in args
+assert '--json' in args
+assert os.environ.get('OPENCLAW_EXEC_SHELL_SNAPSHOT') == '0'
+assert 'AIOS_AGENT_TASK_TOKEN' not in os.environ
+prompt = sys.stdin.read()
+assert 'Do NOT run `git commit`' in prompt
+(cwd / 'changed_by_openclaw.txt').write_text('changed\\n', encoding='utf-8')
+json.dump({
+  'ok': True,
+  'status': 'ok',
+  'final': 'Changed the requested file and verified the result.',
+  'payloads': [{'text':'done'}],
+  'model': value('--model'),
+  'provider': 'test-provider',
+  'sessionId': 'test-session',
+  'assistantTurns': 2,
+  'codeModeEngaged': True,
+  'toolSummary': {'calls': 3, 'tools': ['read','write','exec'], 'totalToolTimeMs': 10}
+}, sys.stdout)
+""",
+        encoding="utf-8",
+    )
+    path.chmod(0o755)
+    return path
+
+
+def request(worktree: Path) -> dict[str, object]:
+    return {
+        "schema": "aios.online-agent.runtime-request.v1",
+        "adapter": "hermes",
+        "task": {
+            "task_id": "openclaw-backend-selftest",
+            "goal": "Make the requested change.",
+            "instructions": "Inspect, implement, and test.",
+            "success_criteria": ["The change is correct"],
+        },
+        "execution_context": {
+            "repository": "aios/aios",
+            "base_oid": "a" * 40,
+            "target_branch": "agent/intent/openclaw-backend-selftest",
+            "worktree": str(worktree),
+        },
+        "policy": {
+            "authoritative_repository_write": False,
+            "direct_git_ref_mutation": False,
+            "direct_git_push": False,
+            "direct_git_commit": False,
+            "main_write": False,
+            "force_push": False,
+            "production_write": False,
+            "tag_write": False,
+            "baseline_write": False,
+            "task_api_credential_visible": False,
+            "writer_lease_visible": False,
+        },
+    }
+
+
+def test_openclaw_agent_exec_contract(tmp_path: Path, monkeypatch) -> None:
+    backend = load_backend()
+    worktree = tmp_path / "worktree"
+    worktree.mkdir()
+    fake = fake_openclaw(tmp_path)
+    monkeypatch.setenv("AIOS_OPENCLAW_BIN", str(fake))
+    monkeypatch.setenv("AIOS_OPENCLAW_MODEL", "test/model")
+    monkeypatch.setenv("AIOS_OPENCLAW_PROVIDER_CREDENTIAL_MAP_JSON", "{}")
+    fd = os.open(str(worktree), getattr(os, "O_PATH", os.O_RDONLY) | os.O_DIRECTORY)
+    try:
+        value = request(worktree)
+        value["execution_context"]["worktree"] = f"/proc/self/fd/{fd}"
+        task, pinned = backend.validate_request(value)
+        result = backend.invoke_openclaw(task, pinned)
+    finally:
+        os.close(fd)
+    assert result["status"] == "COMPLETED"
+    assert result["evidence"]["backend"] == "openclaw-agent-exec"
+    assert result["evidence"]["model"] == "test/model"
+    assert (worktree / "changed_by_openclaw.txt").read_text(encoding="utf-8") == "changed\n"
+
+
+def test_provider_credential_is_projected_only_to_openclaw_child(tmp_path: Path, monkeypatch) -> None:
+    backend = load_backend()
+    credentials = tmp_path / "credentials"
+    credentials.mkdir()
+    (credentials / "openai_api_key").write_text("provider-secret\n", encoding="utf-8")
+    monkeypatch.setenv("CREDENTIALS_DIRECTORY", str(credentials))
+    monkeypatch.setenv(
+        "AIOS_OPENCLAW_PROVIDER_CREDENTIAL_MAP_JSON",
+        json.dumps({"OPENAI_API_KEY": "openai_api_key"}),
+    )
+    env = backend.provider_environment()
+    assert env["OPENAI_API_KEY"] == "provider-secret"
+    assert "AIOS_AGENT_TASK_TOKEN" not in env
+    assert env["OPENCLAW_EXEC_SHELL_SNAPSHOT"] == "0"
+
+
+def test_missing_model_fails_closed(tmp_path: Path, monkeypatch) -> None:
+    backend = load_backend()
+    worktree = tmp_path / "worktree"
+    worktree.mkdir()
+    monkeypatch.setenv("AIOS_OPENCLAW_BIN", str(fake_openclaw(tmp_path)))
+    monkeypatch.delenv("AIOS_OPENCLAW_MODEL", raising=False)
+    fd = os.open(str(worktree), getattr(os, "O_PATH", os.O_RDONLY) | os.O_DIRECTORY)
+    try:
+        value = request(worktree)
+        value["execution_context"]["worktree"] = f"/proc/self/fd/{fd}"
+        task, pinned = backend.validate_request(value)
+        with pytest.raises(backend.BackendError) as caught:
+            backend.invoke_openclaw(task, pinned)
+    finally:
+        os.close(fd)
+    assert caught.value.code == "OPENCLAW_MODEL_INVALID"
+
+
+def test_pinned_worktree_is_required(tmp_path: Path) -> None:
+    backend = load_backend()
+    value = request(tmp_path)
+    value["execution_context"]["worktree"] = str(tmp_path)
+    with pytest.raises(backend.BackendError) as caught:
+        backend.validate_request(value)
+    assert caught.value.code == "PINNED_WORKTREE_REQUIRED"