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

active-runtime/app/aios_chatgpt_actions_api.py

@@ -22,7 +22,6 @@ from gitea_tag_publication import (
     TagPublisher,
     redact,
     validate_version,
-    validate_gitea_api_url,
 )
 from orchestrator.action_monitor import ActionMonitor
 from silent_release_hardening import (
@@ -309,7 +308,8 @@ def resolve_commit_oid(value: str | None, field: str = 'target_oid') -> str:
         raise ValueError(f'{field} does not resolve to the exact approved commit') from exc
 
 def existing_tag_target(tag: str) -> str | None:
-    validate_version(tag)
+    if not re.fullmatch(r'V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$', str(tag or '')):
+        raise ValueError('invalid release version')
     try:
         result = run_authoritative_git(
             ['rev-parse', '--verify', f'refs/tags/{tag}^{{commit}}'],
@@ -510,15 +510,7 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({'error':redact(exc)},413)
                 except (AuthoritativeGitError, RuntimeError) as exc:
                     return self.sendj({'error':redact(exc)},503)
-                payload = {'from': frm, 'to': to, 'diff': diff}
-                payload.update({
-                    key: diff[key]
-                    for key in (
-                        'from_oid', 'to_oid', 'changed_paths', 'stats',
-                        'total_size_bytes', 'truncated', 'patch_limit_bytes',
-                    )
-                })
-                return self.sendj(payload)
+                return self.sendj({'from':frm,'to':to,'diff':diff})
             if path in ('/api/patch','/api/patches','/api/branches','/patch'):
                 snapshot = repository_snapshot(
                     git_dir=AUTHORITATIVE_GIT_DIR,
@@ -781,10 +773,7 @@ class H(BaseHTTPRequestHandler):
         a=self.need('tag:create');
         if not a: return
         ver=data.get('version') or data.get('tag')
-        try:
-            validate_version(ver)
-        except ValueError:
-            return self.sendj({'error':'version must match VxRxCx'},400)
+        if not re.match(r'^V\d+R\d+C\d+(-[A-Za-z0-9_.-]+)?$', ver or ''): return self.sendj({'error':'version must match VxRxCx'},400)
         if not data.get('target_oid') or not data.get('evidence_closure_oid') or not data.get('approval_plan_id'):
             return self.sendj({
                 'error':'production_tag_evidence_required',
@@ -958,7 +947,6 @@ def api_spec():
     return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],'write':['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl','POST /createReleaseCandidate','POST /submitOwnerApproval','POST /startDeployment'],'policy':{'bearer_required_for_write':True,'direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True,'silent_release_execution_enabled':SILENT_RELEASE_EXECUTION_ENABLED,'silent_release_shadow_default':True}}
 
 def main():
-    validate_gitea_api_url(os.getenv('AIOS_GITOPS_GITEA_API_URL', ''))
     print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
     ThreadingHTTPServer((HOST,PORT), H).serve_forever()
 if __name__ == '__main__': main()

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -46,75 +46,6 @@ OPERATION_REQUIRED_FIELDS = {
     "getControlPlaneSnapshot": (),
 }
 
-_OPERATION_PATHS = {
-    "createGovernedTask": "/v2/governed-task",
-    "acquireWriterLease": "/v2/writer-lease",
-    "commitFileGoverned": "/v2/commit-file",
-    "commitFilesGoverned": "/v2/commit-files",
-    "getGitOpsJob": "/v2/job",
-    "resumeGitOpsJob": "/v2/job/resume",
-    "reconcileGitOpsJob": "/v2/job/reconcile",
-    "getTagPublicationStatus": "/v2/tag/status",
-    "getReleaseConsistency": "/v2/release/consistency",
-    "runReleaseAcceptance": "/v2/release/acceptance",
-}
-
-
-def operation_contracts() -> dict[str, dict[str, object]]:
-    """Return the runtime contract consumed by adapters and schema tooling."""
-    return {
-        operation: {
-            "method": "POST",
-            "path": _OPERATION_PATHS.get(operation),
-            "required_fields": list(fields),
-        }
-        for operation, fields in OPERATION_REQUIRED_FIELDS.items()
-    }
-
-
-def render_openapi_contract_fragment() -> str:
-    required = "\n".join(
-        f"                - {field}"
-        for field in OPERATION_REQUIRED_FIELDS["commitFilesGoverned"]
-    )
-    return (
-        "  /v2/commit-files:\n"
-        "    post:\n"
-        "      operationId: commitFilesGoverned\n"
-        "      summary: Submit one atomic governed commit containing multiple files\n"
-        "      requestBody:\n"
-        "        required: true\n"
-        "        content:\n"
-        "          application/json:\n"
-        "            schema:\n"
-        "              type: object\n"
-        "              additionalProperties: false\n"
-        "              required:\n"
-        f"{required}\n"
-        "              properties:\n"
-        "                task_id: {type: string}\n"
-        "                repository: {type: string, example: aios/aios}\n"
-        "                expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}\n"
-        "                target_branch: {type: string, pattern: '^agent/[A-Za-z0-9._/-]+$'}\n"
-        "                generation: {type: integer, minimum: 1}\n"
-        "                idempotency_key: {type: string}\n"
-        "                files:\n"
-        "                  type: array\n"
-        "                  minItems: 1\n"
-        "                  items:\n"
-        "                    type: object\n"
-        "                    additionalProperties: false\n"
-        "                    required: [path, content_base64]\n"
-        "                    properties:\n"
-        "                      path: {type: string, minLength: 1, maxLength: 512}\n"
-        "                      content_base64: {type: string}\n"
-        "                message: {type: string, minLength: 1, maxLength: 200}\n"
-        "      responses:\n"
-        "        '200': {description: Governed commit job and verified branch result}\n"
-        "        '400': {description: Contract or policy rejection}\n"
-        "        '409': {description: Branch CAS or idempotency conflict}\n"
-    )
-
 
 class RuntimeDriftError(RuntimeError):
     pass
@@ -191,7 +122,6 @@ class BridgeV2Adapter:
             "legacy_writes_enabled": False,
             "governed_writes_enabled": True,
             "operations": list(BRIDGE_OPERATIONS),
-            "operation_contracts": operation_contracts(),
             "force_push": False,
             "main_write": False,
             "tag_write": False,

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -14,11 +14,10 @@ from typing import Any, Mapping
 
 from aios_authority_git_executor import AuthorityGitExecutor
 from aios_chatgpt_bridge_v2 import BridgeV2Adapter, BridgeV2Error, OPERATION_REQUIRED_FIELDS
-from gitea_tag_publication import validate_gitea_api_url
 from aios_continuation_executor import AuthorityGitAdapter, ContinuationCommit, ContinuationExecutor, ContinuationPermissions, ContinuationRuntimeBinding, DetachedGitWorkspace, GiteaContinuationRemote, ServerSideCredentialProjection
 from aios_continuation_lease import WriterLeaseRequestV2
 from aios_control_plane_snapshot import build_control_plane_snapshot
-from aios_gitops_job_controller import GitOpsJobController, JobState, JobStore, _RECOVERABLE
+from aios_gitops_job_controller import GitOpsJobController, JobStore
 from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
 from governed_production_release import build_executor_config
 
@@ -53,7 +52,7 @@ class RuntimeConfig:
     ssh_key_path: Path
     remote_name: str = "gitea-http"
     repository: str = "aios/aios"
-    api_base: str = ""
+    api_base: str = "https://g.cnnetmkt.top/git/api/v1"
     v2_task_id: str = "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809"
     v2_branch: str = "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809"
     target_version: str = "V2R2C1"
@@ -281,8 +280,6 @@ class GovernedRuntime:
             return controller.store._receipt(row, duplicate=True).as_dict()
         if action == "resume":
             return controller.resume(job_id).as_dict()
-        if JobState(row["state"]) not in _RECOVERABLE:
-            return controller.store._receipt(row, duplicate=True).as_dict()
         return self._executor_from_row(row, controller).reconcile(job_id)
 
     def _snapshot_task(self) -> dict[str, Any]:
@@ -421,7 +418,7 @@ def main() -> None:
         token_path,
         ssh_key_path,
         remote_name=os.environ.get("AIOS_GITEA_REMOTE", "gitea-http"),
-        api_base=validate_gitea_api_url(os.environ.get("AIOS_GITEA_API_BASE", "")),
+        api_base=os.environ.get("AIOS_GITEA_API_BASE", "https://g.cnnetmkt.top/git/api/v1"),
         target_version=os.environ.get("AIOS_CONTROL_PLANE_TARGET_VERSION", "V2R2C1"),
     )
     runtime = GovernedRuntime(config)

active-runtime/app/aios_gitops_job_controller.py

@@ -41,8 +41,6 @@ class JobState(str, Enum):
     REMOTE_SYNCING = "REMOTE_SYNCING"
     REMOTE_VERIFIED = "REMOTE_VERIFIED"
     COMPLETED = "COMPLETED"
-    FAILED = "FAILED"
-    CANCELLED = "CANCELLED"
     RECOVERABLE_FAILURE = "RECOVERABLE_FAILURE"
     RECONCILING = "RECONCILING"
     RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED"
@@ -61,12 +59,7 @@ _ORDER = {
     JobState.COMPLETED: 8,
 }
 _RECOVERABLE = {JobState.RECOVERABLE_FAILURE, JobState.RECONCILIATION_REQUIRED}
-_TERMINAL = {
-    JobState.COMPLETED,
-    JobState.FAILED,
-    JobState.CANCELLED,
-    JobState.MANUAL_OWNER_REVIEW_REQUIRED,
-}
+_TERMINAL = {JobState.COMPLETED, JobState.MANUAL_OWNER_REVIEW_REQUIRED}
 
 
 def utc_now() -> str:

active-runtime/app/aios_gitops_model_api.py

@@ -5,8 +5,6 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Any
 
-from release_version import validate_release_version
-
 ROOT = Path(os.getenv('AIOS_GITOPS_ROOT', '/data/aios-gitops'))
 WORKSPACE = Path(os.getenv('AIOS_GITOPS_WORKSPACE', str(ROOT / 'workspace')))
 DB_PATH = Path(os.getenv('AIOS_GITOPS_DB', str(ROOT / 'runtime' / 'gitops_model.sqlite3')))
@@ -220,11 +218,7 @@ class Handler(BaseHTTPRequestHandler):
                 a = self.require_write(['owner']);
                 if not a: return
                 ver=data.get('version') or data.get('tag')
-                try:
-            validate_release_version(ver)
-        except ValueError:
-            self.send_json({'error':'version must match VxRxCx'},400)
-            return
+                if not re.match(r'^V\d+R\d+C\d+(-[a-zA-Z0-9_.-]+)?$', ver or ''): self.send_json({'error':'version must match VxRxCx'},400); return
                 msg=data.get('message') or f'release {ver}'; out=git_cmd(['tag','-a',ver,'-m',msg]);
                 if data.get('push',False): git_cmd(['push','origin',ver])
                 level='V' if re.match(r'^V[1-9]\d*R0C0', ver) else ('R' if re.match(r'^V\d+R[1-9]\d*C0', ver) else 'C'); exec_sql('INSERT OR REPLACE INTO gitops_versions(version,tag,level,summary,status,created_by,created_at) VALUES(?,?,?,?,?,?,?)',(ver,ver,level,msg,'ready',a['username'],now())); exec_sql('UPDATE gitops_baseline SET active_version=?,active_tag=?,updated_at=? WHERE id=?',(ver,ver,now(),'current')); audit(a['username'],'git_tag',ver,{'level':level}); self.send_json({'status':'tagged','version':ver,'git':out}); return

active-runtime/app/aios_model_api_staging.py

@@ -23,7 +23,6 @@ from gitea_tag_publication import (
     TagPublisher,
     redact,
     validate_version,
-    validate_gitea_api_url,
 )
 from orchestrator.action_monitor import ActionMonitor
 from silent_release_hardening import (
@@ -320,7 +319,8 @@ def resolve_commit_oid(value: str | None, field: str = 'target_oid') -> str:
         raise ValueError(f'{field} does not resolve to the exact approved commit') from exc
 
 def existing_tag_target(tag: str) -> str | None:
-    validate_version(tag)
+    if not re.fullmatch(r'V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$', str(tag or '')):
+        raise ValueError('invalid release version')
     try:
         result = run_authoritative_git(
             ['rev-parse', '--verify', f'refs/tags/{tag}^{{commit}}'],
@@ -553,15 +553,7 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({'error':redact(exc)},413)
                 except (AuthoritativeGitError, RuntimeError) as exc:
                     return self.sendj({'error':redact(exc)},503)
-                payload = {'from': frm, 'to': to, 'diff': diff}
-                payload.update({
-                    key: diff[key]
-                    for key in (
-                        'from_oid', 'to_oid', 'changed_paths', 'stats',
-                        'total_size_bytes', 'truncated', 'patch_limit_bytes',
-                    )
-                })
-                return self.sendj(payload)
+                return self.sendj({'from':frm,'to':to,'diff':diff})
             if path in ('/api/patch','/api/patches','/api/branches','/patch'):
                 snapshot = repository_snapshot(
                     git_dir=AUTHORITATIVE_GIT_DIR,
@@ -830,10 +822,7 @@ class H(BaseHTTPRequestHandler):
         a=self.need('tag:create');
         if not a: return
         ver=data.get('version') or data.get('tag')
-        try:
-            validate_version(ver)
-        except ValueError:
-            return self.sendj({'error':'version must match VxRxCx'},400)
+        if not re.match(r'^V\d+R\d+C\d+(-[A-Za-z0-9_.-]+)?$', ver or ''): return self.sendj({'error':'version must match VxRxCx'},400)
         if not data.get('target_oid') or not data.get('evidence_closure_oid') or not data.get('approval_plan_id'):
             return self.sendj({
                 'error':'production_tag_evidence_required',
@@ -1027,7 +1016,6 @@ def api_spec():
 
 
 def main():
-    validate_gitea_api_url(os.getenv('AIOS_GITOPS_GITEA_API_URL', ''))
     print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
     ThreadingHTTPServer((HOST,PORT), H).serve_forever()
 if __name__ == '__main__': main()

active-runtime/app/authoritative_git.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 import os
 import re
 import subprocess
-import time
 import urllib.parse
 from pathlib import Path
 from typing import Iterable
@@ -307,76 +306,14 @@ def authoritative_diff(
     from_oid = validate_git_revision(from_revision, "from", git_dir=path)
     to_oid = validate_git_revision(to_revision, "to", git_dir=path)
     args = ["diff", "--no-ext-diff", "--no-textconv", from_oid, to_oid]
-    patch_limit = max(1, min(_MAX_OUTPUT, 256 * 1024))
-    process = subprocess.Popen(
-        ["git", "--git-dir", str(path), *args],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
-    )
-    chunks: list[bytes] = []
-    stored_bytes = 0
-    total_bytes = 0
-    started = time.monotonic()
-    try:
-        while process.stdout is not None:
-            chunk = process.stdout.read(64 * 1024)
-            if not chunk:
-                break
-            total_bytes += len(chunk)
-            if stored_bytes < patch_limit:
-                remaining = patch_limit - stored_bytes
-                selected = chunk[:remaining]
-                chunks.append(selected)
-                stored_bytes += len(selected)
-            if time.monotonic() - started > _TIMEOUT:
-                process.kill()
-                raise AuthoritativeGitError("GIT_REF_QUERY_TIMEOUT")
-        stderr = (
-            process.stderr.read(8192).decode("utf-8", errors="replace")
-            if process.stderr is not None
-            else ""
-        )
-        returncode = process.wait(timeout=max(0.1, _TIMEOUT - (time.monotonic() - started)))
-    except (OSError, subprocess.TimeoutExpired) as exc:
-        process.kill()
-        process.wait()
-        raise AuthoritativeGitError("GIT_REF_QUERY_FAILED") from exc
-    if returncode != 0:
-        raise AuthoritativeGitError(_redact(stderr.strip() or "GIT_REF_QUERY_FAILED")[:200])
-    try:
-        names_result = run_authoritative_git(
-            ["diff", "--no-ext-diff", "--no-textconv", "--name-status", from_oid, to_oid],
-            git_dir=path,
-        )
-        changed_paths = [
-            line.split("\t", 1)[-1]
-            for line in names_result.stdout.splitlines()
-            if line.strip()
-        ]
-    except AuthoritativeGitOutputTooLarge:
-        changed_paths = []
-    try:
-        stats_result = run_authoritative_git(
-            ["diff", "--no-ext-diff", "--no-textconv", "--stat", from_oid, to_oid],
-            git_dir=path,
-        )
-        stats = stats_result.stdout.strip()
-    except AuthoritativeGitOutputTooLarge:
-        stats = "GIT_STAT_OUTPUT_TRUNCATED"
-    patch = b"".join(chunks).decode("utf-8", errors="replace")
+    result = run_authoritative_git(args, git_dir=path)
     return {
         "code": 0,
-        "stdout": patch,
+        "stdout": result.stdout,
         "stderr": "",
         "cmd": ["git", f"--git-dir={path}", *args],
         "from_oid": from_oid,
         "to_oid": to_oid,
-        "changed_paths": changed_paths,
-        "stats": stats,
-        "total_size_bytes": total_bytes,
-        "truncated": total_bytes > patch_limit,
-        "patch_limit_bytes": patch_limit,
     }
 
 

active-runtime/app/gitea_tag_publication.py

@@ -8,7 +8,6 @@ import json
 import os
 import re
 import subprocess
-import sys
 import tempfile
 import time
 import urllib.error
@@ -24,15 +23,10 @@ try:
     from authoritative_git import AuthoritativeGitError, run_authoritative_git
 except ModuleNotFoundError:
     from .authoritative_git import AuthoritativeGitError, run_authoritative_git
-try:
-    from release_version import RELEASE_VERSION_RE, validate_release_version
-except ModuleNotFoundError:
-    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from release_version import RELEASE_VERSION_RE, validate_release_version
 
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
-VERSION_RE = RELEASE_VERSION_RE
+VERSION_RE = re.compile(r"^V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$")
 DEPLOYMENT_EVIDENCE_RE = re.compile(r"^(?:sha256:[0-9a-f]{64}|git:[0-9a-f]{40})$")
 FINGERPRINT_FIELDS = (
     "version",
@@ -89,14 +83,9 @@ class PublicationManualOwnerReviewRequired(
 
 
 def validate_version(version: str) -> str:
-    return validate_release_version(version)
-
-
-def validate_gitea_api_url(api_url: str) -> str:
-    value = str(api_url or "").strip().rstrip("/")
-    parsed = urllib.parse.urlsplit(value)
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
-        raise ValueError("Gitea API URL must be HTTP(S)")
+    value = str(version or "")
+    if not VERSION_RE.fullmatch(value):
+        raise ValueError("invalid release version")
     return value
 
 
@@ -422,11 +411,13 @@ class GiteaTagClient:
         timeout: float = 15,
         token_line: int = 1,
     ):
-        self.api_url = validate_gitea_api_url(api_url)
+        self.api_url = api_url.rstrip("/")
         self.repository = repository.strip("/")
         self.token_file = token_file
         self.timeout = timeout
         self.token_line = token_line
+        if not self.api_url.startswith(("https://", "http://")):
+            raise ValueError("Gitea API URL must be HTTP(S)")
         if self.repository.count("/") != 1:
             raise ValueError("Gitea repository must be owner/name")
 

active-runtime/app/governed_tag_adapter.py

@@ -34,8 +34,6 @@ from gitea_tag_publication import (
     TagPublisher,
 )
 
-from release_version import validate_release_version
-
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
 
@@ -46,10 +44,8 @@ def load_request() -> dict[str, Any]:
         raise ValueError("request_object_required")
     if value.get("schema") != "aios.production-executor-request.v1":
         raise ValueError("request_schema_invalid")
-    try:
-        validate_release_version(value.get("version", ""))
-    except ValueError as exc:
-        raise ValueError("version_invalid") from exc
+    if not re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", str(value.get("version", ""))):
+        raise ValueError("version_invalid")
     for name in ("target_oid", "evidence_closure_oid"):
         if not OID_RE.fullmatch(str(value.get(name, ""))):
             raise ValueError(f"{name}_invalid")

active-runtime/app/release_metadata_adapter.py

@@ -27,8 +27,6 @@ import urllib.request
 from pathlib import Path
 from typing import Any, Mapping
 
-from release_version import validate_release_version
-
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA_RE = re.compile(r"^[0-9a-f]{64}$")
 PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
@@ -88,10 +86,8 @@ def load_request() -> dict[str, Any]:
         raise ValueError("request_schema_invalid")
     if value.get("repository_identity") != "aios/aios":
         raise ValueError("repository_identity_invalid")
-    try:
-        validate_release_version(value.get("version", ""))
-    except ValueError as exc:
-        raise ValueError("version_invalid") from exc
+    if not re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", str(value.get("version", ""))):
+        raise ValueError("version_invalid")
     for name in ("target_oid", "evidence_closure_oid"):
         if not OID_RE.fullmatch(str(value.get(name, ""))):
             raise ValueError(f"{name}_invalid")

active-runtime/app/silent_release_controller.py

@@ -22,12 +22,11 @@ from enum import Enum
 from pathlib import Path
 from typing import Any, Protocol
 
-from release_version import validate_release_version
-
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 CANDIDATE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{2,127}$")
+VERSION_RE = re.compile(r"^V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$")
 SECRET_PATTERNS = (
     re.compile(r"(?i)(authorization\s*[:=]\s*)(?:bearer|token)\s+\S+"),
     re.compile(r"(?i)((?:access_)?token\s*[:=]\s*)\S+"),
@@ -770,10 +769,8 @@ class TagApprovalEnvelope:
     def validate(self) -> None:
         if self.action != "production_tag":
             raise ApprovalError("approval does not authorize production tag")
-        try:
-            validate_release_version(self.version)
-        except ValueError as exc:
-            raise ApprovalError("invalid release version") from exc
+        if not VERSION_RE.fullmatch(self.version):
+            raise ApprovalError("invalid release version")
         _validate_oid("tag_oid", self.tag_oid)
         _validate_oid("target_oid", self.target_oid)
         if _parse_utc(self.expires_at) <= datetime.now(timezone.utc):

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-chatgpt-bridge-v2-staging.service

@@ -4,25 +4,14 @@ After=network-online.target
 
 [Service]
 Type=simple
-User=aiman
-Group=gitops
-PermissionsStartOnly=true
-WorkingDirectory=@RUNTIME_ROOT@
+User=aios-bridge-staging
+Group=aios-bridge-staging
+WorkingDirectory=/opt/aios/gitops-control-plane/current
 Environment=AIOS_BRIDGE_SCHEMA_VERSION=aios.chatgpt-gitops-bridge.v2
 Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
-Environment=AIOS_GITOPS_WRITES_ENABLED=false
-Environment=AIOS_GITEA_API_BASE=https://g.cnnetmkt.top/git/api/v1
-Environment=AIOS_GITEA_TOKEN_FILE=/run/credentials/aios-chatgpt-bridge-v2-staging.service/gitea_token
-Environment=AIOS_GITEA_REMOTE=gitea-http
-Environment=AIOS_CONTROL_PLANE_TARGET_VERSION=V2R2C1
-Environment=AIOS_CONTROL_PLANE_TARGET_OID=@SOURCE_OID@
-Environment=AIOS_CONTROL_PLANE_ARTIFACT_SHA256=@ARTIFACT_SHA256@
-Environment=PYTHONPATH=@RUNTIME_ROOT@:@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os
 Environment=AIOS_DEPLOYMENT_STATE=STAGING
-LoadCredential=gitea_token:/run/credentials/aios-governance-executor.service/gitea-governance-executor-token
-ExecStartPre=/usr/bin/install -d -o aiman -g gitops -m 0750 /var/lib/aios-gitops/bridge-v2-staging
-ExecStart=/usr/bin/python3 -B @RUNTIME_ROOT@/app/aios_chatgpt_bridge_v2_server.py
+ExecStart=/usr/bin/python3 /opt/aios/gitops-control-plane/current/bridge-v2-staging-server.py
 NoNewPrivileges=true
 PrivateTmp=true
 ProtectSystem=strict

deploy/systemd/aios-gitops-model-api-v2r2c1-staging.service

@@ -7,9 +7,6 @@ Wants=network-online.target
 Type=simple
 User=aiman
 Group=gitops
-PermissionsStartOnly=true
-ExecStartPre=/usr/bin/install -d -o aiman -g gitops -m 0750 /var/lib/aios-gitops/model-api-staging
-ExecStartPre=/usr/bin/install -d -o aiman -g gitops -m 0750 /var/lib/aios-gitops/model-api-staging/silent-release
 WorkingDirectory=@RUNTIME_ROOT@
 Environment=PYTHONUNBUFFERED=1
 Environment=PYTHONDONTWRITEBYTECODE=1
@@ -30,16 +27,10 @@ Environment=AIOS_GITOPS_DEPLOYMENT_STATE=STAGING
 Environment=AIOS_GITOPS_MODEL_API_SERVICE=aios-gitops-model-api-v2r2c1-staging.service
 Environment=AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false
 Environment=AIOS_SILENT_RELEASE_STATE_ROOT=/var/lib/aios-gitops/model-api-staging/silent-release
-Environment=AIOS_GITOPS_GITEA_API_URL=https://g.cnnetmkt.top/git/api/v1
-Environment=AIOS_GITOPS_GITEA_REPOSITORY=aios/aios
-Environment=AIOS_GITOPS_GITEA_REMOTE=gitea-http
-Environment=AIOS_GITOPS_GITEA_TOKEN_FILE=/run/credentials/aios-gitops-model-api-v2r2c1-staging.service/gitea_token
-Environment=AIOS_GITOPS_GITEA_TOKEN_LINE=1
-LoadCredential=gitea_token:/run/credentials/aios-governance-executor.service/gitea-governance-executor-token
 Environment=AIOS_BRIDGE_V2_URL=http://127.0.0.1:18892
 Environment=AIOS_GITOPS_HOST=127.0.0.1
 Environment=AIOS_GITOPS_PORT=18893
-Environment=PYTHONPATH=@RUNTIME_ROOT@:@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os
+Environment=PYTHONPATH=@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os
 ExecStart=/usr/bin/python3 @RUNTIME_ROOT@/app/aios_model_api_staging.py
 Restart=on-failure
 RestartSec=2

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.

docs/gitops/openapi-chatgpt-actions.yaml

@@ -241,48 +241,6 @@ components:
       properties:
         candidate_id: {type: string}
 paths:
-  /v2/commit-files:
-    post:
-      operationId: commitFilesGoverned
-      summary: Submit one atomic governed commit containing multiple files
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              additionalProperties: false
-              required:
-                - task_id
-                - repository
-                - expected_branch_oid
-                - target_branch
-                - generation
-                - idempotency_key
-                - files
-                - message
-              properties:
-                task_id: {type: string}
-                repository: {type: string, example: aios/aios}
-                expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-                target_branch: {type: string, pattern: '^agent/[A-Za-z0-9._/-]+$'}
-                generation: {type: integer, minimum: 1}
-                idempotency_key: {type: string}
-                files:
-                  type: array
-                  minItems: 1
-                  items:
-                    type: object
-                    additionalProperties: false
-                    required: [path, content_base64]
-                    properties:
-                      path: {type: string, minLength: 1, maxLength: 512}
-                      content_base64: {type: string}
-                message: {type: string, minLength: 1, maxLength: 200}
-      responses:
-        '200': {description: Governed commit job and verified branch result}
-        '400': {description: Contract or policy rejection}
-        '409': {description: Branch CAS or idempotency conflict}
   /createReleaseIntent:
     post:
       operationId: createReleaseIntent

release_version.py

@@ -1,21 +0,0 @@
-"""Canonical AIOS release-version parser shared by runtime and release tools."""
-from __future__ import annotations
-
-import re
-
-
-# The suffix preserves already-supported annotated release names while the
-# canonical VnRnCn namespace remains accepted exactly.
-RELEASE_VERSION_PATTERN = r"^V[0-9]+R[0-9]+C[0-9]+(?:-[A-Za-z0-9_.-]+)?$"
-RELEASE_VERSION_RE = re.compile(RELEASE_VERSION_PATTERN)
-
-
-def parse_release_version(version: object) -> str:
-    value = str(version or "").strip()
-    if not RELEASE_VERSION_RE.fullmatch(value):
-        raise ValueError("invalid release version")
-    return value
-
-
-def validate_release_version(version: object) -> str:
-    return parse_release_version(version)

reports/action-contract-regression-remediation-20260825.md

@@ -1,75 +0,0 @@
-# Action Contract Regression Remediation — 2026-08-25
-
-## Scope
-
-Governed task: `fix-action-contracts-20260825-1407`
-Base: `3244f0f17f8845592ac94d585f5ebd65693c9f29`
-Target branch: `agent/fix-action-contracts-20260825-1407`
-
-## Verified regressions
-
-1. `getReleaseConsistency` and `runReleaseAcceptance` fail before evaluation with:
-   `Gitea API URL must be HTTP(S)`.
-   V2R2C1 release evidence at `a8b906d1bc24088c19647e98de6a679268f6d0a2`
-   proves Gitea API readback passed for target `998d30304787e31e5ae607c0bd86b5edb2b0e5ad`.
-   Therefore current staging/contract overlay configuration has regressed.
-
-2. `getReleaseSaga` fails with:
-   `Permission denied: '/var/lib/aios-gitops/silent-release'`.
-   Runtime service identity must have read/traverse access to the durable release state root.
-   Fix ownership/group/ACL at deployment time; do not broaden to world-writable permissions.
-
-3. `getTagPublicationStatus` rejects the authoritative approved version `V2R2C1`
-   as `invalid release version`, while `getGovernedWriteCapabilities` returns the same
-   version as the approved release scope. The two Action contracts are internally inconsistent.
-   Release-version validation must accept the canonical AIOS `^V[0-9]+R[0-9]+C[0-9]+$`
-   namespace (or, preferably, use one shared validator for all release Actions).
-
-4. `commitFilesGoverned` is unusable from the published Action schema.
-   Runtime capability metadata declares required fields:
-   `task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message`,
-   but the exposed tool contract has no `files` input. Regenerate/publish the Action schema from
-   the same operation contract source and add a contract-parity test.
-
-5. `reconcileGitOpsJob` returned `DURABLE_BINDING_INVALID` for a completed governed
-   single-file job that was otherwise read back as `COMPLETED`.
-   Reconciliation of a valid terminal job should be idempotent/read-only and must not require
-   a stale or absent active lease. Add a regression case for terminal completed jobs.
-
-6. `getDiff` returns `ResponseTooLargeError` / `GIT_OUTPUT_LIMIT_EXCEEDED` for large diffs.
-   Add server-side bounded output: path/stat summary plus truncation metadata, or pagination.
-   The Action must never fail merely because a valid repository diff is large.
-
-## Required implementation changes
-
-- Centralize release-version parsing and reuse it in tag status, candidate, consistency, and acceptance.
-- Validate/normalize Gitea API URL at service startup and fail health checks with a configuration-specific diagnostic.
-- Ensure the deployment unit projects a valid HTTP(S) Gitea API base URL into the staging overlay.
-- Correct durable release-state directory ownership/ACL for the runtime service account.
-- Generate the external Action schema from the runtime operation contract; parity-test required fields.
-- Make terminal reconciliation idempotent and independent of expired active writer leases.
-- Bound diff output and return `truncated=true` with deterministic metadata rather than transport failure.
-
-## Acceptance gates
-
-All must pass on the same deployed overlay:
-
-- `getReleaseConsistency(V2R2C1, 998d303..., a8b906d..., approved_plan)` returns a structured result, not URL error.
-- `runReleaseAcceptance(...)` reaches policy evaluation, not URL error.
-- `getTagPublicationStatus("V2R2C1")` returns structured publication state.
-- `getReleaseSaga(nonexistent_candidate)` returns NOT_FOUND/empty-state semantics, never filesystem permission failure.
-- Published `commitFilesGoverned` schema exposes `files` and a two-file isolated branch commit succeeds.
-- Reconcile of a completed governed job is idempotent.
-- Large diff returns bounded/truncated data rather than transport exception.
-- Existing protected-ref controls remain unchanged: no direct main/tag/baseline/production write.
-
-## Proven stable reference
-
-V2R2C1 evidence closure (`a8b906d1bc24088c19647e98de6a679268f6d0a2`) records:
-- Gitea Git/API readback PASS
-- staging runtime PASS
-- restart recovery PASS
-- 59 mandatory tests passed / 0 failed
-
-This evidence should be used as the compatibility baseline when repairing the current
-`contract-overlay-v2` / `scheme-a-323b615b...` staging layer.

runbooks/agent-fix-action-contracts-20260825.md

@@ -1,207 +0,0 @@
-# Agent Execution Contract — Action Contract Regression Repair
-
-## Objective
-
-Fully repair the current AIOS Action runtime/contract overlay regressions on the single server, without modifying protected refs directly and without introducing new infrastructure. The task is not complete until the deployed staging/action overlay passes all acceptance gates.
-
-## Authority and scope
-
-- Repository: `aios/aios`
-- Base at task creation: `3244f0f17f8845592ac94d585f5ebd65693c9f29`
-- Governed task: `fix-action-contracts-20260825-1407`
-- Target branch: `agent/fix-action-contracts-20260825-1407`
-- Protected refs must not be directly written: `main`, release tags, baseline registry, production runtime.
-- No new SaaS, database, agent framework, or monitoring platform may be introduced.
-
-## Non-interactive rules
-
-The Agent must run the whole flow unnattended:
-
-1. Collect facts.
-2. Determine root causes.
-3. Make the smallest correct changes.
-4. Run unit/integration/contract tests.
-5. Deploy to staging only.
-6. Restart only the affected non-production services.
-7. Run the full Action smoke suite.
-8. On failure, automatically collect logs, diagnose, fix, and retry.
-9. If a change causes a new regression, automatically roll back the staging change to the previous known-good configuration and record the failure.
-10. Stop only if an owner decision is required for a high-risk, irreversible, or production-boundary action.
-
-## Fact collection — required before editing
-
-Record all of the following in the final evidence bundle:
-
-- `id`, `hostname`, `uname -a`, `date --iso-seconds`
-- `git rev-parse --show HEAD` for authority worktree and running overlay source
-- `git status --porcelaine=v2`
-- `git remote -v`
-- current systemd service unit(s), drop-ins, Environment/EnvironmentFile sources for GitOps/Bredge/release services
-- `systemctl show <affected-service>` including `User=`, `Group=`, `Environment=`, `EnvironmentFile=`, `ExecStart=`
-- resolved Gitea API base URL after env/config expansion; redact secrets but not scheme/host
-- `stat -c, `namei -l`, `getfacl`, als -ld on `/var/lib/aios-gitops/silent-release` and its parents
-- find Action OpenAPI/schema generator and runtime operation definitions for `commitFilesGoverned`
-- find the release version validator(s) and all callers of `getTagPublicationStatus`
-- find `reconcileGitOpsJob` implementation and durable binding validation logic
-- find `getDiff  implementation and transport output limits
-
-Do not edit any file until the actual implementation and configuration sources have been located.
-
-## Repair 1 — Gitea API URL
-
-Symptoms:
-
-- `getReleaseConsistency` → `Gitea API URL must be HTTP(S)`
-- `runReleaseAcceptance` → `Gitea API URL must be HTTP(ÊX
-
-Required behavior:
-
-- Determine which env/config key is projected into the current contract overlay.
-- Compare the deployed overlay with V2R2C1 known-good configuration/deployment manifests.
-- Restore a valid `HTTP://` or `HTTPS://` Gitea API base URL.
-- Add startup config validation so invalid schemes fail at health/startup with a config-specific message, not during a release check.
-- Never hard-code a different Gitea host if the existing deployment already has a single authoritative source.
-
-## Repair 2 — durable release state permissions
-
-Symptom:
-
-- `getReleaseSaga` → `Permission denied: '/var/lib/aios-gitops/silent-release'`
-
-Required behavior:
-
-- Identify the actual runtime service User/Group.
-- Grant only the required traverse/read/write permissions for the durable state directories the service must use.
-- Prefer ownership + group mode or a specific ACL; keep the directory non-world-writable.
-- Persist the permission in deployment/install logic so restart/reboot does not regress.
-- Verify create/read/update of a staging-only saga state file under the service identity.
-
-## Repair 3 — release version contract
-
-Symptom:
-
-- `getTagPublicationStatus("V2R2C1")` → `invalid release version`
-- The same runtime exposes `V2R2C1` as `approved_release_version`.
-
-Required behavior:
-
-- Locate all version validators; consolidate onto one canonical parser.
-- Accept the AIOS canonical namespace `^V[0-9]+R[0-9]+C[0-9]+$`.
-- Preserve any additional already-supported legacy formats unless they are security-sensitive.
- - Add contract tests proving that the version accepted by capabilities is accepted by status, candidate, consistency, acceptance, and tag publication.
-
-## Repair 4 — `commitFilesGoverned` schema parity
-
-Symptom:
-
-- Runtime capability metadata requires `files`.
-- Published Action tool schema does not expose `files`.
-- Call reaches Bridge and fails `BRIDGE_REQUEST_FIELDS_MISSING`.
-
-Required behavior:
-
-- Find the canonical runtime operation contract.
-- Generate the external tool schema from the same source of truth; do not manually copy a parallel schema.
-- `files` must be required and must represent a list of file objects with at least `path` and `content_base64`.
-- Add a parity test that compares runtime required fields with published Action required fields.
-- Add a real governed commit test that writes two test files to an isolated `agent/` ibranch and verifies remote OID.
-
-## Repair 5 — terminal reconciliation
-
-Symptom:
-
-- A governed job read back as `COMPLETED' but `reconcileGitOpsJob` failed `DURABLE_BINDING_INVALID`.
-
-Required behavior:
-
-- Reconciling a terminal successful job must be idempotent.
-- An expired writer lease must not invalidate a durable binding for a job that is already terminal.
-- Active-job reconciliation may require a live lease; terminal read-back must not.
-- Add cases for COMPLETED, FAILED, CANCELLED / equivalent terminal states.
-
-## Repair 6 — bounded `getDiff` output
-
-Symptom:
-
-- Large valid diffs fail with `ResponseTooLargeError` or `GIT_OUTPUT_LIMIT_EXCEEDED`.
-
-Required behavior:
-
-- Do not load unbounded diff text into the Action response.
-- Return bounded summary data: commit identities, changed paths/stats, total bytes, and `truncated: true` when the full patch exceeds the safe limit.
-- If pagination already exists in the contract use it; otherwise prefer deterministic truncation over a breaking schema change.
-- Add tests for small, empty, and oversize diffs.
-
-## Staging deployment procedure
-
-1. Run the existing mandatory test suite before modification; record baseline.
-2. Apply code and config changes only on the governed branch/worktree.
-3. Run focused regression tests.
-4. Run the full mandatory suite.
-5. Build the staging contract overlay using the repository's existing build/deploy path.
-6. Deploy to staging/non-production only.
-7. If systemd units are changed: `systemctl daemon-reload`, then restart only the affected staging services.
-8. Wait for the existing health endpoint/runtime status to become healthy.
-9. Run the acceptance suite below.
-
-If any step fails:
-- collect `journalctl` logs and relevant app logs;
-- diagnose root cause;
-- automatically repair and retry when non-destructive;
-- if a new change breaks staging, restore the previous staging artifact/config and restart the affected service;
-- never silently continue after a failed gate.
-
-## Acceptance suite
-
-All of the following must pass on the same deployed staging overlay:
-
-- Runtime status is `active` and healthy after restart.
-- Gitea Git/API readback works from the runtime service identity.
-- `getReleaseConsistency(V2R2C1, 998d30304787e31e5ae607c0bd86b5edb2b0e5ad, a8b906d1bc24088c19647e98de6a679268f6d0a2, approved_plan) returns a structured result and reaches evaluation.
-- `runReleaseAcceptance` for the same identities reaches policy evaluation, not URL validation error.
-- `getTagPublicationStatus("V2R2C1")` returns structured state, not `invalid release version`.
-- `getReleaseSaga` for a nonexistent candidate returns NOT_FOUND/empty-state semantics, not filesystem permission failure.
-- `commitFilesGoverned` exposes its required `files` parameter in the published Action schema.
- - A real two-file governed commit to an isolated `agent/` branch succeeds and remote branch OID matches.
-- Resuming a COMPLETED job returns the existing `JOB_TERMINAL` semantics or idempotent success as designed.
-- Reconciling a COMPLETED governed job is successful/idempotent and does not require an active lease.
-- Small/empty `getDiff  still works.
-- Large `getDiff` returns bounded output with truncation metadata instead of a transport exception.
-- Existing protected-ref controls remain unchange: no direct main/tag/baseline/production write.
-
-## Restart / reboot persistence
-
-After the first full pass:
-
-- Restart the staging Action/Bridge/release services one more time without changing files.
-- Rerun the acceptance suite.
-- Permissions, Gitea URL, and schema must still be correct after restart.
-- If the repository already has a safe staging reboot test, run it as well.
-
-## Evidence and completion
-
-Create a single durable report in the repository, e.g. `reports/action-contract-repair-evidence-20260825.md`, containing:
-
-- pre-change facts
-- root cause for each regression
-- files changed and why
-- config/permission changes
-- test commands and exact results
-- staging deployment identity (commit/artifact/config digest)
-- restart-persistence results
-- final Action smoke results
-- any remaining risks
-
-Commit all source, test, deployment config, and runbook changes to the governed task branch. Do not mark the task SUCCEEDED until all acceptance gates pass on the actually deployed staging overlay.
-
-## Automatic rollback conditions
-
-Automatically roll back the staging deployment if any of the following occur:
-
-- service fails to become healthy after restart
-- protected-ref write boundary is broken
-- Gitea readback that worked before the change stops working
-- durable saga state cannot be read or recovered
-- mandatory test suite regresses below the pre-change baseline
-
-Do not automatically roll back an existing production release. Rollback here refers only to the staging overlay/service configuration changes made by this task.

scripts/build_v2r2c1_model_api_staging.py

@@ -5,7 +5,6 @@ import hashlib
 import json
 import os
 from pathlib import Path
-import sys
 import tarfile
 import tempfile
 
@@ -19,18 +18,6 @@ REQUIRED_APP = (
     'silent_release_controller.py',
     'silent_release_hardening.py',
     'aios_gitops_job_controller.py',
-    'aios_chatgpt_bridge_v2.py',
-    'aios_chatgpt_bridge_v2_server.py',
-    'governed_production_release.py',
-    'v2r2c1_production_adapter.py',
-    'v2r1c7_production_adapter.py',
-    'aios_governed_task_bootstrap.py',
-    'aios_continuation_executor.py',
-    'aios_continuation_lease.py',
-    'aios_authority_git_executor.py',
-    'aios_gitops_reconciliation_core.py',
-    'authorized_task_writer_provisioning.py',
-    'signer_contract_compatibility.py',
 )
 OPTIONAL_APP = (
     'aios_chatgpt_actions_api.py',
@@ -69,18 +56,11 @@ def main() -> int:
     source = args.source_root.resolve(strict=True)
     output = args.output_root.resolve()
     app = source / 'active-runtime' / 'app'
-    sys.path.insert(0, str(app))
-    from aios_chatgpt_bridge_v2 import render_openapi_contract_fragment
-
     runtime = source / 'runtime' / 'agent-os'
     docs = source / 'docs' / 'gitops'
     if not app.is_dir():
         raise SystemExit('active-runtime/app missing')
     paths = []
-    root_version = source / 'release_version.py'
-    if not root_version.is_file():
-        raise SystemExit('required source dependency missing: release_version.py')
-    paths.append((Path('release_version.py'), root_version))
     for name in REQUIRED_APP:
         path = app / name
         if not path.is_file():
@@ -97,8 +77,6 @@ def main() -> int:
         paths.append((Path('runtime-agent-os') / rel, path))
     openapi = docs / 'openapi-chatgpt-actions.yaml'
     if openapi.is_file():
-        if '  /v2/commit-files:' not in openapi.read_text(encoding='utf-8'):
-            raise SystemExit('published OpenAPI is missing generated commitFilesGoverned contract')
         paths.append((Path('docs') / 'gitops' / 'openapi-chatgpt-actions.yaml', openapi))
     for folder in ('config/model_registry', 'config/model_priority', 'patches', 'versions'):
         root = source / folder

scripts/release_version.sh

@@ -9,16 +9,10 @@ WORKSPACE="${AIOS_GITOPS_WORKSPACE:-/data/aios-gitops/workspace}"
 TAG_REMOTE="${AIOS_GITOPS_GITEA_REMOTE:-gitea}"
 BRANCH_REMOTE="${AIOS_GITOPS_BRANCH_REMOTE:-gitea}"
 
-SCRIPT_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)"
-if ! PYTHONPATH="$SCRIPT_ROOT:${PYTHONPATH:-}" python3 - "$VERSION" <<'PY'
-import sys
-from release_version import validate_release_version
-validate_release_version(sys.argv[1])
-PY
-then
-  echo "ERROR: version must match VxRxCx" >&2
-  exit 64
-fi
+case "$VERSION" in
+  V[0-9]*R[0-9]*C[0-9]*) ;;
+  *) echo "ERROR: version must match VxRxCx" >&2; exit 64 ;;
+esac
 
 for value in "$TARGET_OID" "$EVIDENCE_CLOSURE_OID"; do
   [[ "$value" =~ ^[0-9a-f]{40}$ ]] || {

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

services/aios-release-controller/aios_release_controller/manifest.py

@@ -1,9 +1,8 @@
 from __future__ import annotations
 import hashlib,json,re,tarfile
-from release_version import RELEASE_VERSION_RE
 from pathlib import Path
 from typing import Any
-OID_RE=re.compile(r"^[0-9a-f]{40}$"); SHA_RE=re.compile(r"^[0-9a-f]{64}$"); VERSION_RE=RELEASE_VERSION_RE
+OID_RE=re.compile(r"^[0-9a-f]{40}$"); SHA_RE=re.compile(r"^[0-9a-f]{64}$"); VERSION_RE=re.compile(r"^V[0-9]+R[0-9]+C[0-9]+(?:-[A-Za-z0-9_.-]+)?$")
 REQUIRED_SOURCE={"schema","version","repository","base_ref","base_oid","binding_mode","release_readiness","production_deployed","tag_created","baseline_changed","production_service","production_port","rollback_metadata","controller_schema","required_checks"}
 REQUEST_KEYS={"schema","version","repository","base_oid","target_oid","artifact_uri","artifact_sha256","artifact_size","evidence_closure_oid","approval_plan_id","execute","production_deployed","production_service","production_port","rollback_metadata","current_tag_oid","source_manifest_sha256"}
 class ManifestValidationError(ValueError):

tests/contract-integration-a.txt

@@ -1 +0,0 @@
-action-contract-integration-a

tests/contract-integration-b.txt

@@ -1 +0,0 @@
-action-contract-integration-b

tests/test_action_contract_regressions.py

@@ -1,116 +0,0 @@
-from __future__ import annotations
-
-import json
-import sys
-from pathlib import Path
-
-import pytest
-
-ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(ROOT))
-
-from aios_chatgpt_bridge_v2 import (  # noqa: E402
-    OPERATION_REQUIRED_FIELDS,
-    operation_contracts,
-    render_openapi_contract_fragment,
-)
-from aios_chatgpt_bridge_v2_server import GovernedRuntime  # noqa: E402
-from aios_gitops_job_controller import JobState  # noqa: E402
-from gitea_tag_publication import validate_gitea_api_url  # noqa: E402
-from release_version import validate_release_version  # noqa: E402
-
-
-def test_canonical_release_version_is_shared_and_accepts_v2r2c1():
-    assert validate_release_version("V2R2C1") == "V2R2C1"
-    assert validate_release_version("V2R2C1-repair") == "V2R2C1-repair"
-    with pytest.raises(ValueError):
-        validate_release_version("v2r2c1")
-
-
-def test_gitea_url_validation_is_startup_safe():
-    assert validate_gitea_api_url("https://g.cnnetmkt.top/git/api/v1") == (
-        "https://g.cnnetmkt.top/git/api/v1"
-    )
-    with pytest.raises(ValueError, match="HTTP"):
-        validate_gitea_api_url("g.cnnetmkt.top/git/api/v1")
-
-
-def test_commit_files_runtime_and_published_schema_have_same_required_fields():
-    runtime = operation_contracts()["commitFilesGoverned"]["required_fields"]
-    assert runtime == list(OPERATION_REQUIRED_FIELDS["commitFilesGoverned"])
-    fragment = render_openapi_contract_fragment()
-    published = (ROOT / "docs/gitops/openapi-chatgpt-actions.yaml").read_text()
-    assert fragment in published
-    for field in runtime:
-        assert f"                - {field}" in fragment
-    assert "required: [path, content_base64]" in fragment
-
-
-class _Store:
-    def __init__(self, row):
-        self.row = row
-        self.receipt_calls = 0
-
-    def read(self, _job_id):
-        return self.row
-
-    def _receipt(self, row, duplicate=False):
-        self.receipt_calls += 1
-        return type("Receipt", (), {"as_dict": lambda _self: {
-            "job_id": row["job_id"],
-            "state": row["state"],
-            "duplicate": duplicate,
-        }})()
-
-
-class _Controller:
-    def __init__(self, row):
-        self.store = _Store(row)
-
-
-def _runtime_for(row):
-    runtime = object.__new__(GovernedRuntime)
-    controller = _Controller(row)
-    runtime._controller_for_job = lambda _job_id: controller
-    runtime._executor_from_row = lambda *_args: (_ for _ in ()).throw(
-        AssertionError("terminal reconcile must not reconstruct an executor")
-    )
-    return runtime, controller
-
-
-@pytest.mark.parametrize(
-    "state",
-    [
-        JobState.COMPLETED.value,
-        JobState.FAILED.value,
-        JobState.CANCELLED.value,
-        JobState.MANUAL_OWNER_REVIEW_REQUIRED.value,
-    ],
-)
-def test_terminal_reconcile_is_idempotent_without_durable_binding(state):
-    row = {
-        "job_id": "job-terminal",
-        "state": state,
-        "request_json": json.dumps({}),
-    }
-    runtime, controller = _runtime_for(row)
-    result = GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile")
-    assert result["duplicate"] is True
-    assert controller.store.receipt_calls == 1
-
-
-def test_active_reconcile_still_uses_executor():
-    row = {
-        "job_id": "job-active",
-        "state": JobState.RECONCILIATION_REQUIRED.value,
-        "request_json": json.dumps({}),
-    }
-    runtime, controller = _runtime_for(row)
-    class Executor:
-        def reconcile(self, job_id):
-            return {"job_id": job_id, "state": "COMPLETED"}
-    runtime._executor_from_row = lambda *_args: Executor()
-    assert GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile") == {
-        "job_id": "job-active",
-        "state": "COMPLETED",
-    }

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"