fix(gitops): centralize implementation branch policy

active-runtime/app/aios_authority_git_executor.py

@@ -11,6 +11,8 @@ from dataclasses import dataclass
 import os
 from pathlib import Path
 import re
+
+from aios_target_branch_policy import validate_target_branch
 import subprocess
 import tempfile
 from typing import Mapping, Sequence
@@ -95,7 +97,8 @@ class AuthorityGitExecutor:
         return path
 
     def advance_task_branch_CAS(self, branch: str, expected_oid: str, new_oid: str) -> str:
-        if not TASK_BRANCH_RE.fullmatch(branch):
+        branch_path = branch[len("refs/heads/"):] if branch.startswith("refs/heads/") else ""
+        if not branch.startswith("refs/heads/") or not validate_target_branch(branch_path):
             raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED")
         _oid(expected_oid, "expected_oid")
         _oid(new_oid, "new_oid")

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -9,6 +9,8 @@ import os
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 import re
+
+from aios_target_branch_policy import validate_target_branch
 import sqlite3
 from typing import Any, Mapping
 
@@ -210,8 +212,7 @@ class GovernedRuntime:
         if (
             not task_id
             or repository != self.config.repository
-            or not branch.startswith("agent/")
-            or branch in {"main", "master"}
+            or not validate_target_branch(branch)
             or branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}
             or generation < 1
             or not self._valid_oid(base)
@@ -287,7 +288,7 @@ class GovernedRuntime:
         for value in self.bindings.values():
             task_id = str(value.get("task_id") or "").strip()
             branch = str(value.get("target_branch") or "").strip()
-            if not task_id or not branch.startswith("agent/"):
+            if not task_id or not validate_target_branch(branch):
                 continue
             authority = self.authority.read_ref("refs/heads/" + branch)
             try:

active-runtime/app/aios_continuation_lease.py

@@ -10,6 +10,8 @@ from __future__ import annotations
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 import re
+
+from aios_target_branch_policy import validate_target_branch
 from typing import Protocol
 
 
@@ -60,7 +62,7 @@ class WriterLeaseRequestV2:
             raise ContinuationError("branch_mode_invalid")
         _oid(self.base_oid, "base_oid")
         _oid(self.expected_branch_oid, "expected_branch_oid")
-        if self.target_branch in {"main", "master"} or self.target_branch.startswith(("refs/", "tags/")):
+        if not validate_target_branch(self.target_branch):
             raise ContinuationError("protected_branch_rejected")
         if not isinstance(self.generation, int) or isinstance(self.generation, bool) or self.generation < 1:
             raise ContinuationError("generation_invalid")

active-runtime/app/aios_credential_broker.py

@@ -8,6 +8,8 @@ from pathlib import Path
 import secrets
 from typing import Any, Callable, Mapping
 
+from aios_target_branch_policy import validate_target_branch
+
 
 @dataclass(frozen=True, slots=True)
 class CredentialLease:
@@ -44,7 +46,7 @@ class CredentialBroker:
         self._leases: dict[str, CredentialLease] = {}
 
     def issue(self, *, task_id: str, repository: str, target_branch: str, generation: int, scopes: tuple[str, ...] = ("git:push", "gitea:read"), lease_seconds: int = 900) -> dict[str, Any]:
-        if repository != "aios/aios" or not target_branch.startswith("agent/") or generation < 1 or not 1 <= lease_seconds <= 86400:
+        if repository != "aios/aios" or not validate_target_branch(target_branch) or generation < 1 or not 1 <= lease_seconds <= 86400:
             raise ValueError("CREDENTIAL_LEASE_BINDING_INVALID")
         now = self.now().astimezone(timezone.utc)
         lease = CredentialLease("lease-" + secrets.token_hex(16), task_id, repository, target_branch, generation, now.isoformat(), (now + timedelta(seconds=lease_seconds)).isoformat(), tuple(scopes), type(self.provider).__name__)

active-runtime/app/aios_gitea_publisher.py

@@ -11,6 +11,7 @@ from urllib.parse import quote
 from urllib.request import Request, urlopen
 
 from aios_gitops_reconciliation_core import RemoteObservation
+from aios_target_branch_policy import validate_target_branch
 
 
 @dataclass(frozen=True, slots=True)
@@ -35,7 +36,7 @@ class GiteaPublisher:
 
     @staticmethod
     def validate_branch(branch: str) -> None:
-        if not isinstance(branch, str) or not branch.startswith("agent/") or branch in {"main", "master"}:
+        if not validate_target_branch(branch):
             raise ValueError("BRANCH_SCOPE_DENIED")
 
     def repository_identity(self) -> str:

active-runtime/app/aios_gitops_job_controller.py

@@ -18,6 +18,8 @@ import tempfile
 from datetime import datetime, timezone
 from typing import Any, Callable, Mapping
 
+from aios_target_branch_policy import validate_target_branch
+
 
 class JobError(RuntimeError):
     pass
@@ -274,7 +276,7 @@ class GitOpsJobController:
             raise JobError("EXPECTED_BRANCH_OID_INVALID")
         if len(digest) != 64 or any(ch not in "0123456789abcdef" for ch in digest):
             raise JobError("PAYLOAD_DIGEST_INVALID")
-        if request["branch"] in {"main", "master"} or request["branch"].startswith(("refs/", "tags/")):
+        if not validate_target_branch(request["branch"]):
             raise JobError("PROTECTED_BRANCH_REJECTED")
         binding = {field: str(request[field]) for field in fields}
         # Operation-specific, non-secret binding inputs must survive the
@@ -383,7 +385,7 @@ class GitOpsJobController:
                     raise JobError("JOB_NOT_FOUND")
                 if int(row["revision"]) != expected_revision:
                     raise JobError("REVISION_CAS_CONFLICT")
-                if row["branch"] in {"main", "master"}:
+                if not validate_target_branch(row["branch"]):
                     raise JobError("PROTECTED_BRANCH_REJECTED")
                 now = utc_now()
                 connection.execute("INSERT INTO branch_leases(lease_id,job_id,branch,lineage_anchor_oid,current_head_oid,generation,lease_expires_at,state,revision) VALUES(?,?,?,?,?,?,?,?,?)", (lease_id, job_id, row["branch"], lineage_anchor_oid, current_head_oid, generation, lease_expires_at, "ACTIVE", 0))

active-runtime/app/aios_gitops_reconciliation_core.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 from dataclasses import dataclass
 from enum import Enum
 import re
+
+from aios_target_branch_policy import validate_target_branch
 from typing import Optional
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
@@ -52,7 +54,7 @@ def decide_reconciliation(observation: RemoteObservation) -> ReconciliationDecis
     """Decide from a fresh observation; never complete on a missing remote."""
     if observation.repository_identity != "aios/aios":
         return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "REPOSITORY_IDENTITY_MISMATCH")
-    if not observation.branch.startswith("agent/"):
+    if not validate_target_branch(observation.branch):
         return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "BRANCH_SCOPE_DENIED")
     if not _valid_oid(observation.desired_oid) or not _valid_oid(observation.expected_oid):
         return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "OID_INVALID")

active-runtime/app/aios_governed_policy.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 import re
+
+from aios_target_branch_policy import validate_target_branch
 from typing import Any, Mapping
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
@@ -39,7 +41,7 @@ def evaluate_policy(value: GovernedPolicyInput | Mapping[str, Any]) -> PolicyDec
         return PolicyDecision("DENY", "REPOSITORY_IDENTITY_MISMATCH")
     if not OID_RE.fullmatch(value.base_oid) or not OID_RE.fullmatch(value.expected_branch_oid):
         return PolicyDecision("DENY", "OID_INVALID")
-    if not value.target_branch.startswith("agent/") or value.target_branch in {"main", "master"}:
+    if not validate_target_branch(value.target_branch):
         return PolicyDecision("DENY", "BRANCH_SCOPE_DENIED")
     if "force_push" in value.requested_capabilities or "force" in value.operation.lower():
         return PolicyDecision("DENY", "FORCE_PUSH_DENIED")

active-runtime/app/aios_governed_task_bootstrap.py

@@ -11,7 +11,7 @@ from typing import Any, Mapping
 from aios_gitops_job_controller import GitOpsJobController, JobStore
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+from aios_target_branch_policy import BRANCH_RE, validate_target_branch
 
 
 class BootstrapError(RuntimeError):
@@ -61,7 +61,7 @@ class GovernedTaskBootstrap:
     def create(self, request: Mapping[str, Any]) -> dict[str, Any]:
         values = {key: str(request[key]) for key in ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")}
         if values["repository"] != "aios/aios" or not OID_RE.fullmatch(values["base_oid"]): raise BootstrapError("TASK_BINDING_INVALID")
-        if not BRANCH_RE.fullmatch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
+        if not validate_target_branch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
         existing = self._read_existing(values["task_id"])
         if existing is not None:
             if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description")): raise BootstrapError("TASK_BINDING_CONFLICT")

active-runtime/app/aios_target_branch_policy.py

@@ -0,0 +1,56 @@
+"""Single-source validation for governed implementation branch names.
+
+The policy accepts ordinary implementation branches while rejecting protected
+refs, malformed Git ref names, traversal-like components, and characters that
+Git treats specially. Callers translate a false result into their own stable
+error contract.
+"""
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+PROTECTED_BRANCHES = frozenset({"HEAD", "main", "master"})
+PROTECTED_PREFIXES = (
+    "main/",
+    "master/",
+    "refs/",
+    "tags/",
+    "tag/",
+    "release/",
+    "releases/",
+    "baseline/",
+    "production/",
+)
+
+
+def validate_target_branch(value: Any) -> bool:
+    """Return whether *value* is a safe, non-protected implementation branch."""
+    if not isinstance(value, str) or not 3 <= len(value) <= 180:
+        return False
+    if not BRANCH_RE.fullmatch(value):
+        return False
+    if value in PROTECTED_BRANCHES or value == "@":
+        return False
+    if value.startswith(PROTECTED_PREFIXES):
+        return False
+    if value.startswith("/") or value.endswith("/"):
+        return False
+    if value.endswith(".") or value.endswith(".lock"):
+        return False
+    if ".." in value or "//" in value or "@{" in value:
+        return False
+    if any(ord(char) < 0x20 or ord(char) == 0x7F or char in " ~^:?*[\\" for char in value):
+        return False
+    components = value.split("/")
+    if any(
+        not component
+        or component in {".", "..", "@"}
+        or component.startswith(".")
+        or component.endswith(".lock")
+        for component in components
+    ):
+        return False
+    return True

active-runtime/app/authorized_task_writer_provisioning.py

@@ -33,7 +33,7 @@ from signer_contract_compatibility import (
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 TASK_RE = re.compile(r"^[A-Z0-9][A-Z0-9._:-]{7,127}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+from aios_target_branch_policy import BRANCH_RE, validate_target_branch
 OWNER_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]{3,255}$")
 
 REQUIRED_REQUEST_FIELDS = frozenset(
@@ -353,11 +353,7 @@ def _validate_request(
         raise ProvisioningError("BASE_REF_REJECTED", 409)
     if not isinstance(raw["base_oid"], str) or not OID_RE.fullmatch(raw["base_oid"]):
         raise ProvisioningError("BASE_OID_REJECTED")
-    if (
-        not isinstance(raw["target_branch"], str)
-        or not BRANCH_RE.fullmatch(raw["target_branch"])
-        or raw["target_branch"] in {"main", "master"}
-    ):
+    if not validate_target_branch(raw["target_branch"]):
         raise ProvisioningError("TARGET_BRANCH_REJECTED")
     if not isinstance(raw["task_id"], str) or not TASK_RE.fullmatch(raw["task_id"]):
         raise ProvisioningError("TASK_ID_REJECTED")

active-runtime/app/owner_authorization_broker.py

@@ -23,6 +23,7 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Mapping, Protocol
 
+from aios_target_branch_policy import validate_target_branch
 from authorized_task_writer_provisioning import (
     AuthorizedTaskWriterProvisioner,
     BRANCH_RE,
@@ -335,11 +336,7 @@ def _validate_request(
         or not OID_RE.fullmatch(raw["base_oid"])
     ):
         raise OwnerAuthorizationError("BASE_OID_REJECTED")
-    if (
-        not isinstance(raw["target_branch"], str)
-        or not BRANCH_RE.fullmatch(raw["target_branch"])
-        or raw["target_branch"] in {"main", "master"}
-    ):
+    if not validate_target_branch(raw["target_branch"]):
         raise OwnerAuthorizationError("TARGET_BRANCH_REJECTED")
     if (
         not isinstance(raw["task_id"], str)

active-runtime/app/owner_local_auth_ingress.py

@@ -49,7 +49,7 @@ OID_RE = re.compile(r"^[0-9a-f]{40}$")
 IDENTITY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._@:+/-]{1,254}$")
 SCOPE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
 TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,255}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9._/-]{1,220}$")
+from aios_target_branch_policy import validate_target_branch
 
 MAX_FRAME = 64 * 1024
 CHALLENGE_TTL_SECONDS = 120
@@ -207,10 +207,7 @@ def validate_binding(value: Mapping[str, Any]) -> dict[str, Any]:
         raise BootstrapError("BASE_REF_REJECTED")
     if not isinstance(request["base_oid"], str) or not OID_RE.fullmatch(request["base_oid"]):
         raise BootstrapError("BASE_OID_REJECTED")
-    if (
-        not isinstance(request["target_branch"], str)
-        or not BRANCH_RE.fullmatch(request["target_branch"])
-    ):
+    if not validate_target_branch(request["target_branch"]):
         raise BootstrapError("TARGET_BRANCH_REJECTED")
     if (
         not isinstance(request["task_id"], str)

active-runtime/app/task_identity_overlay.py

@@ -17,6 +17,8 @@ import hashlib
 import json
 import os
 import re
+
+from aios_target_branch_policy import validate_target_branch
 import secrets
 import tempfile
 from dataclasses import dataclass
@@ -192,8 +194,7 @@ def _validate_config(config: IdentityOverlayConfig) -> None:
     if not OID_RE.fullmatch(binding.base_oid):
         raise IdentityOverlayError("OVERLAY_BASE_OID_INVALID", 503)
     if (
-        not BRANCH_RE.fullmatch(binding.target_branch)
-        or binding.target_branch in {"main", "master"}
+        not validate_target_branch(binding.target_branch)
     ):
         raise IdentityOverlayError("OVERLAY_TARGET_BRANCH_INVALID", 503)
     if not OWNER_IDENTITY_RE.fullmatch(config.owner_identity):

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

@@ -1,46 +0,0 @@
-[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

@@ -1,36 +0,0 @@
-[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

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

@@ -1,54 +0,0 @@
-# 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

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

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

@@ -1,937 +0,0 @@
-#!/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

@@ -1,157 +0,0 @@
-#!/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

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

tests/test_owner_local_auth_ingress.py

@@ -53,7 +53,7 @@ def test_exact_binding_passes():
     [
         (("owner_identity",), "owner", None),
         (("authenticated_scope",), "*", "OWNER_SCOPE_INVALID"),
-        (("request", "target_branch"), "patch/not-allowed", "TARGET_BRANCH_REJECTED"),
+        (("request", "target_branch"), "release/not-allowed", "TARGET_BRANCH_REJECTED"),
         (("request", "production_change_authorized"), True, "CONSEQUENTIAL_CAPABILITY_REJECTED"),
         (("request", "tag_change_authorized"), True, "CONSEQUENTIAL_CAPABILITY_REJECTED"),
         (("request", "baseline_change_authorized"), True, "CONSEQUENTIAL_CAPABILITY_REJECTED"),

tests/test_target_branch_policy.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from aios_target_branch_policy import validate_target_branch
+
+
+def test_accepts_ordinary_implementation_branches() -> None:
+    for branch in (
+        "agent/worker-change-20260815",
+        "patch/not-allowed-anymore",
+        "fix/target-branch-policy",
+        "feature/ordinary-implementation",
+        "chore/code_stats-v2",
+    ):
+        assert validate_target_branch(branch), branch
+
+
+def test_rejects_protected_namespaces() -> None:
+    for branch in (
+        "main",
+        "master",
+        "main/hotfix",
+        "master/hotfix",
+        "refs/heads/patch/change",
+        "tags/release-v1",
+        "tag/release-v1",
+        "release/v1",
+        "releases/v1",
+        "baseline/current",
+        "production/hotfix",
+    ):
+        assert not validate_target_branch(branch), branch
+
+
+def test_rejects_malformed_git_refs_and_traversal() -> None:
+    for branch in (
+        "../escape",
+        "patch/../escape",
+        "patch//escape",
+        "patch/.",
+        "patch/.hidden",
+        "patch/name.lock",
+        "patch/name.",
+        "patch/name~bad",
+        "patch/name^bad",
+        "patch/name:bad",
+        "patch/name?bad",
+        "patch/name*bad",
+        "patch/name[bad",
+        "patch/name\\\\bad",
+        "patch/name@{bad}",
+        "patch/name with-space",
+        "patch/line\\nbreak",
+        "@",
+    ):
+        assert not validate_target_branch(branch), repr(branch)