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

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -13,8 +13,6 @@ BRIDGE_OPERATIONS = (
     "getGovernedWriteCapabilities",
     "createGovernedTask",
     "acquireWriterLease",
-    "bootstrapWriterLease",
-    "renewWriterLease",
     "commitFileGoverned",
     "commitFilesGoverned",
     "getGitOpsJob",
@@ -33,8 +31,6 @@ BRIDGE_OPERATIONS = (
 OPERATION_REQUIRED_FIELDS = {
     "createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
     "acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
-    "bootstrapWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
-    "renewWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
     "commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
     "commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
     "getGitOpsJob": ("job_id",),
@@ -50,49 +46,6 @@ OPERATION_REQUIRED_FIELDS = {
     "getControlPlaneSnapshot": (),
 }
 
-BRIDGE_OPERATION_CONTRACTS = {
-    "acquireWriterLease": {
-        "method": "POST",
-        "path": "/v2/writer-lease",
-        "required_fields": OPERATION_REQUIRED_FIELDS["acquireWriterLease"],
-        "response_fields": (
-            "task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
-            "create_target_branch", "worktree_mode", "cas_ref_update_required",
-        ),
-    },
-    "bootstrapWriterLease": {
-        "method": "POST",
-        "path": "/v2/writer-lease/bootstrap",
-        "required_fields": OPERATION_REQUIRED_FIELDS["bootstrapWriterLease"],
-        "response_fields": (
-            "task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
-            "create_target_branch", "worktree_mode", "cas_ref_update_required",
-            "recovered", "bootstrapped", "runtime_rehydrated",
-            "durable_branch_lease_present", "idempotent",
-        ),
-    },
-    "renewWriterLease": {
-        "method": "POST",
-        "path": "/v2/writer-lease/renew",
-        "required_fields": OPERATION_REQUIRED_FIELDS["renewWriterLease"],
-        "response_fields": (
-            "task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
-            "recovered", "bootstrapped", "runtime_rehydrated",
-            "durable_branch_lease_present", "idempotent",
-        ),
-    },
-    "commitFileGoverned": {
-        "method": "POST", "path": "/v2/commit-file",
-        "required_fields": OPERATION_REQUIRED_FIELDS["commitFileGoverned"],
-        "response_fields": ("job_id", "commit_oid", "commit_visible", "branch_mode", "target_branch", "generation"),
-    },
-    "commitFilesGoverned": {
-        "method": "POST", "path": "/v2/commit-files",
-        "required_fields": OPERATION_REQUIRED_FIELDS["commitFilesGoverned"],
-        "response_fields": ("job_id", "commit_oid", "commit_visible", "branch_mode", "target_branch", "generation"),
-    },
-}
-
 
 class RuntimeDriftError(RuntimeError):
     pass
@@ -169,14 +122,6 @@ class BridgeV2Adapter:
             "legacy_writes_enabled": False,
             "governed_writes_enabled": True,
             "operations": list(BRIDGE_OPERATIONS),
-            "operation_contracts": {
-                operation: {
-                    **contract,
-                    "required_fields": list(contract["required_fields"]),
-                    "response_fields": list(contract["response_fields"]),
-                }
-                for operation, contract in BRIDGE_OPERATION_CONTRACTS.items()
-            },
             "force_push": False,
             "main_write": False,
             "tag_write": False,
@@ -206,8 +151,6 @@ class BridgeV2Adapter:
 
     def createGovernedTask(self, request: Mapping[str, Any]) -> Any: return self._call("createGovernedTask", request)
     def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
-    def bootstrapWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("bootstrapWriterLease", request)
-    def renewWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("renewWriterLease", request)
     def commitFileGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFileGoverned", request)
     def commitFilesGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFilesGoverned", request)
     def getGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("getGitOpsJob", request)

active-runtime/app/aios_continuation_executor.py

@@ -329,7 +329,6 @@ class ContinuationPermissions:
 @dataclass(frozen=True, slots=True)
 class ContinuationRuntimeBinding:
     request: WriterLeaseRequestV2
-    job_id: str
     workspace_path: Path
     lease_not_before: datetime
     lease_expires_at: datetime
@@ -390,20 +389,13 @@ class ContinuationExecutor:
     def observe_generation(self, branch: str, generation: int) -> None:
         self._observed_generations[branch] = max(generation, self._observed_generations.get(branch, 0))
 
-    def _max_generation(self, branch: str, *, exclude_job_id: str | None = None) -> int:
+    def _max_generation(self, branch: str) -> int:
         observed = self._observed_generations.get(branch, 0)
         with self.controller.store._connect() as connection:
-            if exclude_job_id is None:
-                row = connection.execute(
-                    "SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
-                    (branch,),
-                ).fetchone()
-            else:
-                row = connection.execute(
-                    "SELECT MAX(generation) AS generation FROM branch_leases "
-                    "WHERE branch=? AND job_id!=?",
-                    (branch, exclude_job_id),
-                ).fetchone()
+            row = connection.execute(
+                "SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
+                (branch,),
+            ).fetchone()
         durable = int(row["generation"] or 0)
         return max(observed, durable)
 
@@ -412,7 +404,6 @@ class ContinuationExecutor:
         return {
             "schema": WriterLeaseRequestV2.schema,
             "task_id": request.task_id,
-            "job_id": request.job_id,
             "repository": request.repository,
             "branch_mode": request.branch_mode,
             "base_oid": request.base_oid,
@@ -422,7 +413,7 @@ class ContinuationExecutor:
             "lease_seconds": request.lease_seconds,
         }
 
-    def _validate_preflight(self, binding: ContinuationRuntimeBinding, *, job_id: str | None = None) -> None:
+    def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> None:
         request = binding.request
         normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
         if normalized != {**self._request_mapping(request), "create_target_branch": False, "worktree_mode": "detached_at_expected_head", "cas_ref_update_required": True}:
@@ -435,7 +426,7 @@ class ContinuationExecutor:
             raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
         if now >= binding.lease_expires_at:
             raise ContinuationExecutionError("LEASE_EXPIRED")
-        if request.generation <= self._max_generation(request.target_branch, exclude_job_id=job_id):
+        if request.generation <= self._max_generation(request.target_branch):
             raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
         if not self.workspace.is_detached():
             raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
@@ -461,60 +452,39 @@ class ContinuationExecutor:
             "generation": request.generation,
         }
 
-    def execute(
-        self,
-        binding: ContinuationRuntimeBinding,
-        commit: ContinuationCommit,
-        *,
-        before_complete: Callable[[str], None] | None = None,
-    ) -> dict[str, object]:
-        commit.payload()
-        if not binding.job_id or binding.request.job_id != binding.job_id:
-            raise ContinuationExecutionError("RUNTIME_LINEAGE_MISSING_JOB_ID")
-        current = self.controller.store.read(binding.job_id)
-        current_state = JobState(current["state"])
-        if current_state == JobState.COMPLETED:
+    def execute(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit) -> dict[str, object]:
+        payload = commit.payload()
+        submitted = self.controller.submit(self._job_request(binding, commit, payload), payload=payload)
+        current = self.controller.store.read(submitted.job_id)
+        if JobState(current["state"]) == JobState.COMPLETED:
             return self.controller.store._receipt(current, duplicate=True).as_dict()
-        if current_state == JobState.REMOTE_VERIFIED:
-            receipt = json.loads(str(current["receipt_json"] or "{}"))
-            commit_oid = str(receipt.get("authority_oid") or "")
-            if not OID_RE.fullmatch(commit_oid):
-                raise ContinuationExecutionError("GOVERNED_COMMIT_OID_INVALID")
-            if before_complete is not None:
-                before_complete(commit_oid)
-            return self.controller.advance(
-                binding.job_id,
-                JobState.COMPLETED,
-                side_effect_state="REMOTE_VERIFIED",
-                receipt=receipt,
-            ).as_dict()
-        self._validate_preflight(binding, job_id=binding.job_id)
+        self._validate_preflight(binding)
         request = binding.request
         projected = False
         try:
             self.credential.project()
             projected = True
-            row = self.controller.store.read(binding.job_id)
+            row = self.controller.store.read(submitted.job_id)
             if JobState(row["state"]) == JobState.REQUESTED:
                 self.controller.advance(
-                    binding.job_id,
+                    submitted.job_id,
                     JobState.AUTHORIZED,
                     side_effect_state="AUTHORIZED",
                     receipt={"task_id": request.task_id, "original_grant_consumed": True},
                 )
-                row = self.controller.store.read(binding.job_id)
+                row = self.controller.store.read(submitted.job_id)
             if JobState(row["state"]) == JobState.AUTHORIZED:
                 self.controller.lease_job(
-                    binding.job_id,
-                    lease_id=f"lease-{binding.job_id}",
+                    submitted.job_id,
+                    lease_id=f"lease-{submitted.job_id}",
                     lineage_anchor_oid=request.base_oid,
                     current_head_oid=request.expected_branch_oid,
                     generation=request.generation,
                     lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(),
                     expected_revision=int(row["revision"]),
                 )
-            self.controller.advance(binding.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
-            self.controller.advance(binding.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
+            self.controller.advance(submitted.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
+            self.controller.advance(submitted.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
             new_oid = self.workspace.commit_files(commit.files, commit.message)
             if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
                 raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
@@ -522,22 +492,17 @@ class ContinuationExecutor:
             if authority_oid != new_oid:
                 raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
             self.controller.advance(
-                binding.job_id,
+                submitted.job_id,
                 JobState.LOCAL_VERIFIED,
                 side_effect_state="LOCAL_VERIFIED",
                 receipt={"authority_oid": authority_oid, "force_push": False},
             )
             try:
-                self.controller.advance(binding.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
+                self.controller.advance(submitted.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
                 self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
             except Exception:
-                return self._remote_pending(binding.job_id, authority_oid, new_oid)
-            return self._finish_remote_verification(
-                binding.job_id,
-                authority_oid,
-                new_oid,
-                before_complete=before_complete,
-            )
+                return self._remote_pending(submitted.job_id, authority_oid, new_oid)
+            return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
         finally:
             if projected:
                 self.credential.revoke()
@@ -560,14 +525,7 @@ class ContinuationExecutor:
         )
         return result.as_dict()
 
-    def _finish_remote_verification(
-        self,
-        job_id: str,
-        authority_oid: str,
-        new_oid: str,
-        *,
-        before_complete: Callable[[str], None] | None = None,
-    ) -> dict[str, object]:
+    def _finish_remote_verification(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
         row = self.controller.store.read(job_id)
         branch = row["branch"]
         request = json.loads(row["request_json"])["binding"]
@@ -584,20 +542,8 @@ class ContinuationExecutor:
             "reconciliation_decision": decision.action.value,
             "force_push": False,
         }
-        self.controller.advance(
-            job_id,
-            JobState.REMOTE_VERIFIED,
-            side_effect_state="REMOTE_VERIFIED",
-            receipt=receipt,
-        )
-        if before_complete is not None:
-            before_complete(new_oid)
-        return self.controller.advance(
-            job_id,
-            JobState.COMPLETED,
-            side_effect_state="REMOTE_VERIFIED",
-            receipt=receipt,
-        ).as_dict()
+        self.controller.advance(job_id, JobState.REMOTE_VERIFIED, side_effect_state="REMOTE_VERIFIED", receipt=receipt)
+        return self.controller.advance(job_id, JobState.COMPLETED, side_effect_state="REMOTE_VERIFIED", receipt=receipt).as_dict()
 
     def reconcile(self, job_id: str) -> dict[str, object]:
         row = self.controller.store.read(job_id)

active-runtime/app/aios_continuation_lease.py

@@ -49,7 +49,6 @@ class WriterLeaseRequestV2:
     target_branch: str
     generation: int
     lease_seconds: int = 900
-    job_id: str = ""
 
     schema = "aios.task-writer-lease.v2"
     accepted_schemas = frozenset({"aios.task-writer-lease.v2", "aios.task-writer-lease-request.v2"})
@@ -57,8 +56,6 @@ class WriterLeaseRequestV2:
     def validate(self, probe: BranchProbe) -> None:
         if not self.task_id or self.repository != "aios/aios":
             raise ContinuationError("request_identity_invalid")
-        if not self.job_id or not self.job_id.startswith("job-"):
-            raise ContinuationError("job_id_invalid")
         if self.branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}:
             raise ContinuationError("branch_mode_invalid")
         _oid(self.base_oid, "base_oid")

active-runtime/app/aios_gitops_job_controller.py

@@ -27,19 +27,6 @@ class ServiceUnavailable(JobError):
     pass
 
 
-@dataclass(frozen=True, slots=True)
-class LeaseRenewalResult:
-    """Durable outcome of one writer-lease recovery attempt."""
-
-    renewed: bool
-    idempotent: bool
-    durable: bool
-    lease_expires_at: str | None
-
-    def __bool__(self) -> bool:
-        return self.renewed or self.idempotent
-
-
 class BindingConflict(JobError):
     pass
 
@@ -79,14 +66,6 @@ def utc_now() -> str:
     return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
 
 
-def parse_utc(value: str) -> datetime:
-    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
-    parsed = datetime.fromisoformat(normalized)
-    if parsed.tzinfo is None:
-        raise ValueError("LEASE_EXPIRY_INVALID")
-    return parsed.astimezone(timezone.utc)
-
-
 def canonical_json(value: Any) -> bytes:
     return (
         json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n"
@@ -425,16 +404,13 @@ class GitOpsJobController:
         self.store._mirror(result)
         return self.store._receipt(result, duplicate=True)
 
-    def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE", lease_expires_at: str | None = None) -> None:
+    def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE") -> None:
         if len(current_head_oid) != 40 or generation < 1:
             raise JobError("LEASE_BINDING_INVALID")
         try:
             with self.store._connect() as connection:
                 connection.execute("BEGIN IMMEDIATE")
-                if lease_expires_at is None:
-                    updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, state, lease_id, expected_revision))
-                else:
-                    updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,lease_expires_at=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, lease_expires_at, state, lease_id, expected_revision))
+                updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, state, lease_id, expected_revision))
                 if updated.rowcount != 1:
                     raise JobError("LEASE_REVISION_CAS_CONFLICT")
                 connection.commit()
@@ -443,94 +419,6 @@ class GitOpsJobController:
         except sqlite3.Error as exc:
             raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
 
-    def renew_lease(
-        self,
-        *,
-        branch: str,
-        lineage_anchor_oid: str,
-        current_head_oid: str,
-        generation: int,
-        lease_expires_at: str,
-    ) -> LeaseRenewalResult:
-        if (
-            not branch
-            or len(lineage_anchor_oid) != 40
-            or len(current_head_oid) != 40
-            or generation < 1
-            or not lease_expires_at.strip()
-        ):
-            raise JobError("LEASE_BINDING_INVALID")
-        try:
-            with self.store._connect() as connection:
-                connection.execute("BEGIN IMMEDIATE")
-                rows = connection.execute(
-                    "SELECT lease_id, revision, generation, state, lease_expires_at "
-                    "FROM branch_leases "
-                    "WHERE branch=? AND lineage_anchor_oid=? AND current_head_oid=? "
-                    "AND generation=? AND state='ACTIVE' ORDER BY revision DESC",
-                    (branch, lineage_anchor_oid, current_head_oid, generation),
-                ).fetchall()
-                if not rows:
-                    connection.commit()
-                    return LeaseRenewalResult(False, False, False, None)
-                if len(rows) != 1:
-                    raise JobError("LEASE_BINDING_CONFLICT")
-                row = rows[0]
-                try:
-                    current_expires_at = parse_utc(str(row["lease_expires_at"]))
-                    requested_expires_at = parse_utc(lease_expires_at)
-                except (TypeError, ValueError) as exc:
-                    raise JobError("LEASE_EXPIRY_INVALID") from exc
-                now = datetime.now(timezone.utc)
-                if requested_expires_at <= now:
-                    raise JobError("LEASE_EXPIRY_INVALID")
-                if current_expires_at > now:
-                    connection.commit()
-                    return LeaseRenewalResult(
-                        False,
-                        True,
-                        True,
-                        str(row["lease_expires_at"]),
-                    )
-                updated = connection.execute(
-                    "UPDATE branch_leases SET lease_expires_at=?,revision=revision+1 "
-                    "WHERE lease_id=? AND revision=? AND generation=? "
-                    "AND state='ACTIVE' AND lease_expires_at=?",
-                    (
-                        lease_expires_at,
-                        row["lease_id"],
-                        int(row["revision"]),
-                        generation,
-                        row["lease_expires_at"],
-                    ),
-                )
-                if updated.rowcount != 1:
-                    current = connection.execute(
-                        "SELECT lease_expires_at, generation, state FROM branch_leases "
-                        "WHERE lease_id=?",
-                        (row["lease_id"],),
-                    ).fetchone()
-                    if (
-                        current is not None
-                        and int(current["generation"]) == generation
-                        and current["state"] == "ACTIVE"
-                        and parse_utc(str(current["lease_expires_at"])) > now
-                    ):
-                        connection.commit()
-                        return LeaseRenewalResult(
-                            False,
-                            True,
-                            True,
-                            str(current["lease_expires_at"]),
-                        )
-                    raise JobError("LEASE_REVISION_CAS_CONFLICT")
-                connection.commit()
-                return LeaseRenewalResult(True, False, True, lease_expires_at)
-        except JobError:
-            raise
-        except sqlite3.Error as exc:
-            raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
-
     def advance(self, job_id: str, state: JobState, *, side_effect_state: str, receipt: Mapping[str, Any] | None = None, expected_revision: int | None = None) -> JobReceipt:
         return self._advance(
             job_id,

active-runtime/app/authorized_task_writer_provisioning.py

@@ -964,7 +964,6 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
     required = {
         "schema",
         "task_id",
-        "job_id",
         "repository",
         "branch_mode",
         "base_oid",
@@ -977,7 +976,6 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
         raise ProvisioningError("LEASE_REQUEST_V2_FIELDS_REJECTED", 409)
     request = WriterLeaseRequestV2(
         task_id=str(raw["task_id"]),
-        job_id=str(raw["job_id"]),
         repository=str(raw["repository"]),
         branch_mode=str(raw["branch_mode"]),
         base_oid=str(raw["base_oid"]),
@@ -990,7 +988,6 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
     return {
         "schema": WriterLeaseRequestV2.schema,
         "task_id": request.task_id,
-        "job_id": request.job_id,
         "repository": request.repository,
         "branch_mode": request.branch_mode,
         "base_oid": request.base_oid,

deploy/agent-runtime-openclaw.conf.example

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

docs/autonomous-intent-worker-v2.md

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

docs/gitops/openapi-chatgpt-actions.yaml

@@ -240,114 +240,6 @@ components:
       additionalProperties: false
       properties:
         candidate_id: {type: string}
-    BridgeWriterLeaseRecoveryRequest:
-      type: object
-      additionalProperties: false
-      required: [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds]
-      properties:
-        schema:
-          type: string
-          enum: [aios.task-writer-lease.v2, aios.task-writer-lease-request.v2]
-        task_id: {type: string, minLength: 1, maxLength: 256}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
-        base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-    BridgeWriterLeaseResponse:
-      type: object
-      additionalProperties: false
-      required: [schema, task_id, job_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds, create_target_branch, worktree_mode, cas_ref_update_required]
-      properties:
-        schema: {type: string, enum: [aios.task-writer-lease.v2]}
-        task_id: {type: string}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
-        base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-        create_target_branch: {type: boolean}
-        worktree_mode: {type: string, enum: [detached_at_base_oid, detached_at_expected_head]}
-        cas_ref_update_required: {type: boolean, const: true}
-        lease_expires_at: {type: string, format: date-time}
-    BridgeWriterLeaseRecoveryResponse:
-      type: object
-      additionalProperties: false
-      required: [schema, task_id, job_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds, lease_expires_at, create_target_branch, worktree_mode, cas_ref_update_required, recovered, bootstrapped, idempotent, runtime_rehydrated, durable_branch_lease_present]
-      properties:
-        schema: {type: string, enum: [aios.task-writer-lease.v2]}
-        task_id: {type: string}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
-        base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-        lease_expires_at: {type: string, format: date-time}
-        create_target_branch: {type: boolean, const: false}
-        worktree_mode: {type: string, enum: [detached_at_base_oid, detached_at_expected_head]}
-        cas_ref_update_required: {type: boolean, const: true}
-        recovered: {type: boolean}
-        bootstrapped: {type: boolean}
-        idempotent: {type: boolean}
-        runtime_rehydrated: {type: boolean}
-        durable_branch_lease_present: {type: boolean}
-    BridgeCommitFileRequest:
-      type: object
-      additionalProperties: false
-      required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, path, content_base64, message]
-      properties:
-        task_id: {type: string, minLength: 1}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        idempotency_key: {type: string, minLength: 1, maxLength: 256}
-        path: {type: string, minLength: 1, maxLength: 512}
-        content_base64: {type: string, format: byte}
-        message: {type: string, minLength: 1, maxLength: 200}
-    BridgeCommitFilesRequest:
-      type: object
-      additionalProperties: false
-      required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message]
-      properties:
-        task_id: {type: string, minLength: 1}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        idempotency_key: {type: string, minLength: 1, maxLength: 256}
-        files: {type: object, additionalProperties: {type: string, format: byte}, minProperties: 1}
-        message: {type: string, minLength: 1, maxLength: 200}
-    BridgeCommitResponse:
-      type: object
-      additionalProperties: true
-      required: [job_id, commit_oid, commit_visible, branch_mode, target_branch, generation]
-      properties:
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        commit_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        commit_visible: {type: boolean}
-        branch_mode: {type: string, enum: [CONTINUE_EXISTING]}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-    BridgeGovernedCapabilities:
-      type: object
-      required: [bridge_schema_version, governed_writes_enabled, operations, operation_contracts]
-      properties:
-        bridge_schema_version: {type: string, const: aios.chatgpt-gitops-bridge.v2}
-        governed_writes_enabled: {type: boolean, const: true}
-        operations: {type: array, items: {type: string}}
-        operation_contracts: {type: object, additionalProperties: true}
 paths:
   /createReleaseIntent:
     post:
@@ -675,118 +567,3 @@ paths:
         '200': {description: Safe GitOps plan}
         '401': {description: Unauthorized}
       x-openai-isConsequential: true
-  /v2/capabilities:
-    get:
-      operationId: getGovernedWriteCapabilities
-      summary: Read the governed Bridge v2 capability contract
-      responses:
-        '200':
-          description: Governed operations and their request/response contracts
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeGovernedCapabilities'}
-  /v2/writer-lease:
-    post:
-      operationId: acquireWriterLease
-      summary: Acquire the exact governed writer lease for a task generation
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
-      responses:
-        '200':
-          description: Exact detached worktree lease binding
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeWriterLeaseResponse'}
-        '400': {description: Invalid lease binding}
-        '409': {description: Branch, base, or generation conflict}
-      x-openai-isConsequential: true
-  /v2/writer-lease/bootstrap:
-    post:
-      operationId: bootstrapWriterLease
-      summary: Bootstrap a governed writer runtime without creating a durable lease
-      description: >
-        Staging-only runtime recovery. It validates the exact task, repository,
-        base OID, expected branch OID, target branch, and generation. It must
-        not create a branch, insert a branch lease, or consume a generation.
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
-      responses:
-        '200':
-          description: Runtime binding bootstrapped or returned idempotently
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryResponse'}
-        '400': {description: Invalid lease binding}
-        '409': {description: A matching durable lease already exists or binding conflicts}
-        '503': {description: Runtime or durable state unavailable}
-      x-openai-isConsequential: true
-  /v2/writer-lease/renew:
-    post:
-      operationId: renewWriterLease
-      summary: Recover an expired governed writer lease without changing its binding
-      description: >
-        Renews only the current task generation and exact repository, base OID,
-        expected branch OID, and target branch. Durable renewal is revision-CAS
-        protected; runtime rehydration is reported separately.
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
-      responses:
-        '200':
-          description: Lease recovered or idempotently observed as already active
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryResponse'}
-        '400': {description: Invalid schema, generation, or lease duration}
-        '409': {description: Task, base, repository, branch, generation, or durable lease conflict}
-        '503': {description: Durable job store or runtime unavailable}
-      x-openai-isConsequential: true
-  /v2/commit-file:
-    post:
-      operationId: commitFileGoverned
-      summary: Commit one file through the governed writer lease
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeCommitFileRequest'}
-      responses:
-        '200':
-          description: Authority and remote visibility receipt
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeCommitResponse'}
-        '409': {description: Lease, expected-head, idempotency, or repository conflict}
-        '503': {description: Reconciliation required}
-      x-openai-isConsequential: true
-  /v2/commit-files:
-    post:
-      operationId: commitFilesGoverned
-      summary: Atomically commit a bounded set of files through the governed writer lease
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeCommitFilesRequest'}
-      responses:
-        '200':
-          description: Authority and remote visibility receipt
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeCommitResponse'}
-        '409': {description: Lease, expected-head, idempotency, or repository conflict}
-        '503': {description: Reconciliation required}
-      x-openai-isConsequential: true

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

tests/test_autonomous_broker_security_v2.py

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

tests/test_autonomous_intent_runtime_v2.py

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

tests/test_autonomous_runtime_client_v2.py

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

tests/test_bridge_v2_governed_http_bootstrap.py

@@ -2,7 +2,6 @@ import base64
 from pathlib import Path
 
 from aios_authority_git_executor import AuthorityGitExecutor
-from aios_chatgpt_bridge_v2 import BridgeV2Adapter, OPERATION_REQUIRED_FIELDS
 from aios_chatgpt_bridge_v2_server import BridgeHandler, RuntimeConfig
 from aios_continuation_executor import GiteaContinuationRemote, ServerSideCredentialProjection
 from aios_continuation_lease import WriterLeaseRequestV2
@@ -27,7 +26,6 @@ class Probe:
 
 def test_create_new_lease_is_exact_base_and_absent():
     request = WriterLeaseRequestV2(
-        job_id="job-bridge-runtime-lineage-test",
         task_id="TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
         repository="aios/aios",
         branch_mode="CREATE_NEW",
@@ -41,7 +39,6 @@ def test_create_new_lease_is_exact_base_and_absent():
 
 def test_create_new_lease_rejects_existing_target():
     request = WriterLeaseRequestV2(
-        job_id="job-bridge-runtime-lineage-test",
         task_id="task",
         repository="aios/aios",
         branch_mode="CREATE_NEW",
@@ -78,7 +75,7 @@ def test_slash_branch_uses_paginated_collection():
 def test_http_route_contract_is_complete():
     expected = {
         "/v2/runtime", "/v2/capabilities", "/v2/governed-task",
-        "/v2/writer-lease", "/v2/writer-lease/bootstrap", "/v2/writer-lease/renew", "/v2/commit-file", "/v2/commit-files",
+        "/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
         "/v2/job", "/v2/job/resume", "/v2/job/reconcile",
         "/v2/tag/approval-request",
         "/v2/tag/publish",
@@ -91,121 +88,3 @@ def test_http_route_contract_is_complete():
     from aios_chatgpt_bridge_v2_server import ROUTES
 
     assert set(ROUTES) == expected
-
-
-def test_renew_writer_lease_is_exposed_in_action_and_capability_contracts():
-    adapter = BridgeV2Adapter(runtime_status={})
-    capabilities = adapter.getGovernedWriteCapabilities()
-    contract = capabilities["operation_contracts"]["renewWriterLease"]
-
-    assert contract["method"] == "POST"
-    assert contract["path"] == "/v2/writer-lease/renew"
-    assert tuple(contract["required_fields"]) == OPERATION_REQUIRED_FIELDS["renewWriterLease"]
-    assert "runtime_rehydrated" in contract["response_fields"]
-
-    action_schema = Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
-    assert "/v2/writer-lease/renew:" in action_schema
-    assert "operationId: renewWriterLease" in action_schema
-    assert "BridgeWriterLeaseRecoveryRequest" in action_schema
-    assert "BridgeWriterLeaseRecoveryResponse" in action_schema
-    assert {
-        capabilities["operation_contracts"][operation]["path"]
-        for operation in (
-            "acquireWriterLease",
-            "renewWriterLease",
-            "commitFileGoverned",
-            "commitFilesGoverned",
-        )
-    } == {
-        "/v2/writer-lease",
-        "/v2/writer-lease/renew",
-        "/v2/commit-file",
-        "/v2/commit-files",
-    }
-
-
-def test_bootstrap_writer_lease_is_exposed_in_action_and_capability_contracts():
-    adapter = BridgeV2Adapter(runtime_status={})
-    capabilities = adapter.getGovernedWriteCapabilities()
-    contract = capabilities["operation_contracts"]["bootstrapWriterLease"]
-
-    assert contract["method"] == "POST"
-    assert contract["path"] == "/v2/writer-lease/bootstrap"
-    assert tuple(contract["required_fields"]) == OPERATION_REQUIRED_FIELDS["bootstrapWriterLease"]
-    assert "bootstrapped" in contract["response_fields"]
-    assert "durable_branch_lease_present" in contract["response_fields"]
-
-    action_schema = Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
-    assert "/v2/writer-lease/bootstrap:" in action_schema
-    assert "operationId: bootstrapWriterLease" in action_schema
-    assert "bootstrapped: {type: boolean}" in action_schema
-    assert "durable_branch_lease_present: {type: boolean}" in action_schema
-
-
-
-def test_http_allows_explicit_job_id_on_lease_and_commit_requests():
-    handler = BridgeHandler.__new__(BridgeHandler)
-    lease_body = {
-        "schema": WriterLeaseRequestV2.schema,
-        "task_id": "task",
-        "job_id": "job-explicit-runtime-lineage",
-        "repository": "aios/aios",
-        "branch_mode": "CONTINUE_EXISTING",
-        "base_oid": "a" * 40,
-        "expected_branch_oid": "a" * 40,
-        "target_branch": "agent/explicit-runtime-lineage",
-        "generation": 2,
-        "lease_seconds": 120,
-    }
-    handler._allowed("acquireWriterLease", lease_body)
-    handler._allowed(
-        "commitFilesGoverned",
-        {
-            "task_id": "task",
-            "job_id": "job-explicit-runtime-lineage",
-            "repository": "aios/aios",
-            "expected_branch_oid": "a" * 40,
-            "target_branch": "agent/explicit-runtime-lineage",
-            "generation": 2,
-            "idempotency_key": "commit-explicit-runtime-lineage",
-            "files": {"docs/x.txt": "eA=="},
-            "message": "test",
-        },
-    )
-
-
-def test_http_allows_omitted_job_id_for_durable_server_resolution():
-    handler = BridgeHandler.__new__(BridgeHandler)
-    lease_body = {
-        "schema": WriterLeaseRequestV2.schema,
-        "task_id": "task",
-        "repository": "aios/aios",
-        "branch_mode": "CONTINUE_EXISTING",
-        "base_oid": "a" * 40,
-        "expected_branch_oid": "a" * 40,
-        "target_branch": "agent/implicit-runtime-lineage",
-        "generation": 2,
-        "lease_seconds": 120,
-    }
-    handler._allowed("acquireWriterLease", lease_body)
-    handler._allowed(
-        "commitFilesGoverned",
-        {
-            "task_id": "task",
-            "repository": "aios/aios",
-            "expected_branch_oid": "a" * 40,
-            "target_branch": "agent/implicit-runtime-lineage",
-            "generation": 2,
-            "idempotency_key": "commit-implicit-runtime-lineage",
-            "files": {"docs/x.txt": "eA=="},
-            "message": "test",
-        },
-    )
-
-
-def test_capability_contracts_return_exact_job_id_lineage():
-    contracts = BridgeV2Adapter(runtime_status={}).getGovernedWriteCapabilities()["operation_contracts"]
-    for operation in ("acquireWriterLease", "bootstrapWriterLease", "renewWriterLease"):
-        assert "job_id" in contracts[operation]["response_fields"]
-    for operation in ("commitFileGoverned", "commitFilesGoverned"):
-        assert "job_id" in contracts[operation]["response_fields"]

tests/test_continuation_runtime_normalization.py

@@ -1,175 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-import re
-import json
-from types import SimpleNamespace
-
-import pytest
-import yaml
-
-from aios_chatgpt_bridge_v2_server import GovernedRuntime
-from aios_continuation_executor import ContinuationExecutor
-from aios_continuation_lease import WriterLeaseRequestV2
-from aios_gitops_job_controller import JobState
-
-
-TASK = "continuation-normalization-test"
-JOB = "job-continuation-normalization-test"
-REPO = "aios/aios"
-BASE = "a" * 40
-BRANCH = "agent/continuation-normalization-test"
-
-
-class Authority:
-    def __init__(self, head: str | None) -> None:
-        self.head = head
-
-    def read_ref(self, ref: str) -> str | None:
-        assert ref == "refs/heads/" + BRANCH
-        return self.head
-
-
-def lease(**overrides: object) -> WriterLeaseRequestV2:
-    values = {
-        "task_id": TASK,
-        "job_id": JOB,
-        "repository": REPO,
-        "branch_mode": "CREATE_NEW",
-        "base_oid": BASE,
-        "expected_branch_oid": BASE,
-        "target_branch": BRANCH,
-        "generation": 1,
-        "lease_seconds": 120,
-    }
-    values.update(overrides)
-    return WriterLeaseRequestV2(**values)
-
-
-def runtime(*, head: str | None = BASE, **binding_overrides: object) -> GovernedRuntime:
-    value = GovernedRuntime.__new__(GovernedRuntime)
-    value.authority = Authority(head)
-    binding = {
-        "task_id": TASK,
-        "job_id": JOB,
-        "repository": REPO,
-        "base_oid": BASE,
-        "expected_branch_oid": BASE,
-        "target_branch": BRANCH,
-        "generation": 1,
-        "branch_mode": "CREATE_NEW",
-    }
-    binding.update(binding_overrides)
-    value.bindings = {TASK: binding}
-    return value
-
-
-def test_same_lineage_create_new_normalizes_to_continue_existing() -> None:
-    normalized = runtime()._normalize_runtime_lease(lease())
-    assert normalized.branch_mode == "CONTINUE_EXISTING"
-    assert normalized.task_id == TASK
-    assert normalized.job_id == JOB
-    assert normalized.repository == REPO
-    assert normalized.target_branch == BRANCH
-    assert normalized.expected_branch_oid == BASE
-    assert normalized.generation == 1
-
-
-def test_unprovisioned_create_new_stays_create_new() -> None:
-    assert runtime(head=None)._normalize_runtime_lease(lease()).branch_mode == "CREATE_NEW"
-
-
-@pytest.mark.parametrize(
-    ("field", "value"),
-    (
-        ("repository", "other/repo"),
-        ("base_oid", "b" * 40),
-        ("expected_branch_oid", "b" * 40),
-        ("target_branch", "agent/foreign"),
-        ("generation", 2),
-    ),
-)
-def test_binding_mismatch_fails_closed(field: str, value: object) -> None:
-    with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
-        runtime(**{field: value})._normalize_runtime_lease(lease())
-
-
-def test_drifted_existing_head_fails_closed() -> None:
-    with pytest.raises(RuntimeError, match="EXPECTED_BRANCH_OID_CONFLICT"):
-        runtime(head="b" * 40)._normalize_runtime_lease(lease())
-
-
-def test_openapi_contract_parses_and_requires_exact_commit_lineage() -> None:
-    document = yaml.safe_load(
-        Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
-    )
-    schema = document["components"]["schemas"]["BridgeCommitResponse"]
-    assert {
-        "job_id",
-        "commit_oid",
-        "commit_visible",
-        "branch_mode",
-        "target_branch",
-        "generation",
-    } <= set(schema["required"])
-
-
-def test_changed_production_files_pass_secret_scan() -> None:
-    paths = [
-        Path("active-runtime/app/aios_chatgpt_bridge_v2_server.py"),
-        Path("active-runtime/app/aios_continuation_executor.py"),
-        Path("active-runtime/app/aios_continuation_lease.py"),
-        Path("active-runtime/app/authorized_task_writer_provisioning.py"),
-        Path("active-runtime/app/aios_chatgpt_bridge_v2.py"),
-        Path("docs/gitops/openapi-chatgpt-actions.yaml"),
-    ]
-    expression = re.compile(
-        r"(?i)(BEGIN (RSA|OPENSSH|EC) PRIVATE KEY|"
-        r"authorization:\\s*bearer\\s+[A-Za-z0-9._-]{12,}|"
-        r"api[_-]?key\\s*[:=]\\s*['\\\"][^'\\\"]{12,})"
-    )
-    hits = []
-    for path in paths:
-        for number, line in enumerate(
-            path.read_text(encoding="utf-8", errors="replace").splitlines(), 1
-        ):
-            if expression.search(line):
-                hits.append(f"{path}:{number}")
-    assert hits == []
-
-
-def test_remote_verified_job_is_not_completed_when_head_cas_callback_fails() -> None:
-    class Store:
-        def read(self, _job_id):
-            return {
-                "branch": BRANCH,
-                "request_json": json.dumps(
-                    {"binding": {"expected_branch_oid": BASE}}
-                ),
-            }
-
-    class Controller:
-        def __init__(self):
-            self.store = Store()
-            self.states = []
-
-        def advance(self, _job_id, state, **_kwargs):
-            self.states.append(state)
-            return SimpleNamespace(as_dict=lambda: {"state": state.value})
-
-    executor = ContinuationExecutor.__new__(ContinuationExecutor)
-    executor.controller = Controller()
-    executor._observe_remote = lambda _branch, oid: (oid, oid, True)
-
-    def fail_cas(_oid):
-        raise RuntimeError("LEASE_CAS_CONFLICT")
-
-    with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
-        executor._finish_remote_verification(
-            JOB,
-            "b" * 40,
-            "b" * 40,
-            before_complete=fail_cas,
-        )
-
-    assert executor.controller.states == [JobState.REMOTE_VERIFIED]

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"

tests/test_writer_lease_recovery.py

@@ -1,708 +0,0 @@
-from __future__ import annotations
-
-from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timedelta, timezone
-import hashlib
-from dataclasses import asdict
-from pathlib import Path
-import sqlite3
-import json
-from types import SimpleNamespace
-import threading
-
-import pytest
-
-from aios_chatgpt_bridge_v2_server import GovernedRuntime
-from aios_continuation_executor import ContinuationPermissions, ContinuationRuntimeBinding
-from aios_continuation_lease import WriterLeaseRequestV2
-from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore
-from aios_governed_task_bootstrap import GovernedTaskBootstrap
-
-
-TASK_ID = "gov-writer-lease-recovery-test"
-BRANCH = "agent/writer-lease-recovery-test"
-BASE = "a" * 40
-JOB_ID_PREFIX = "writer-lease-runtime-lineage"
-
-
-class LocalAuthority:
-    def __init__(self, head: str | None = None) -> None:
-        self.head = head
-
-    def read_ref(self, _ref: str) -> str | None:
-        return self.head
-
-    def commit_exists(self, _oid: str) -> bool:
-        return True
-
-    def ancestry_check(self, _older: str, _newer: str) -> bool:
-        return True
-
-
-def writer_request(task_id: str = TASK_ID, **overrides: object) -> dict[str, object]:
-    request: dict[str, object] = {
-        "schema": WriterLeaseRequestV2.schema,
-        "task_id": task_id,
-        "repository": "aios/aios",
-        "branch_mode": "CREATE_NEW",
-        "base_oid": BASE,
-        "expected_branch_oid": BASE,
-        "target_branch": BRANCH,
-        "generation": 1,
-        "lease_seconds": 120,
-    }
-    request.update(overrides)
-    return request
-
-
-def make_runtime(tmp_path: Path, task_id: str = TASK_ID, *, authority_head: str | None = None) -> GovernedRuntime:
-    runtime = GovernedRuntime.__new__(GovernedRuntime)
-    runtime.config = SimpleNamespace(repository="aios/aios", state_root=tmp_path)
-    runtime.authority = LocalAuthority(authority_head)
-    runtime.bindings = {
-        task_id: {
-            "task_id": task_id,
-            "repository": "aios/aios",
-            "base_oid": BASE,
-            "expected_branch_oid": BASE,
-            "target_branch": BRANCH,
-            "branch_mode": "CREATE_NEW",
-            "generation": 1,
-        }
-    }
-    runtime.runtimes = {}
-    runtime.controllers = {}
-    runtime.bootstrap = GovernedTaskBootstrap(
-        runtime.authority,
-        tmp_path,
-        remote_name="test",
-        token_path=tmp_path / "token",
-        ssh_key_path=tmp_path / "key",
-    )
-    runtime.bindings = {}
-    runtime.create_task(
-        {
-            "task_id": task_id,
-            "repository": "aios/aios",
-            "base_oid": BASE,
-            "expected_branch_oid": BASE,
-            "target_branch": BRANCH,
-            "risk": "engineering",
-            "description": "governed runtime lineage test",
-            "idempotency_key": JOB_ID_PREFIX + "-" + task_id,
-        }
-    )
-    job_root = runtime._job_root(task_id)
-    runtime.controllers[task_id] = GitOpsJobController(
-        JobStore(job_root, payload_root=job_root / "payloads")
-    )
-
-    def build(
-        lease: WriterLeaseRequestV2,
-        *,
-        job_id: str,
-        lease_not_before: datetime | None = None,
-        lease_expires_at: datetime | None = None,
-    ) -> tuple[ContinuationRuntimeBinding, GitOpsJobController, object]:
-        controller = runtime.controllers.get(lease.task_id)
-        if controller is None:
-            root = runtime._job_root(lease.task_id)
-            controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
-        now = datetime.now(timezone.utc)
-        return (
-            ContinuationRuntimeBinding(
-                request=lease,
-                job_id=job_id,
-                workspace_path=tmp_path / "worktrees" / lease.task_id,
-                lease_not_before=lease_not_before or now - timedelta(seconds=1),
-                lease_expires_at=lease_expires_at or now + timedelta(seconds=lease.lease_seconds),
-                original_receipt_valid=True,
-                original_grant_consumed=True,
-                permissions=ContinuationPermissions(),
-            ),
-            controller,
-            object(),
-        )
-
-    runtime._build_runtime = build
-    return runtime
-
-
-def create_durable_lease(
-    runtime: GovernedRuntime,
-    *,
-    expires_at: str = "2020-01-01T00:00:00+00:00",
-) -> GitOpsJobController:
-    root = runtime._job_root(TASK_ID)
-    controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
-    receipt = SimpleNamespace(job_id=runtime.bindings[TASK_ID]["job_id"])
-    controller.lease_job(
-        receipt.job_id,
-        lease_id="writer-lease-recovery",
-        lineage_anchor_oid=BASE,
-        current_head_oid=BASE,
-        generation=1,
-        lease_expires_at=expires_at,
-        expected_revision=0,
-    )
-    runtime.controllers[TASK_ID] = controller
-    return controller
-
-
-def durable_row(runtime: GovernedRuntime) -> tuple[str, int]:
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        row = connection.execute(
-            "SELECT lease_expires_at, revision FROM branch_leases WHERE lease_id=?",
-            ("writer-lease-recovery",),
-        ).fetchone()
-    assert row is not None
-    return row
-
-
-def test_first_bootstrap_refreshes_only_runtime_without_durable_lease(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-
-    result = runtime.bootstrap_writer_lease(writer_request())
-
-    assert result["bootstrapped"] is True
-    assert result["runtime_rehydrated"] is False
-    assert result["durable_branch_lease_present"] is False
-    assert result["idempotent"] is False
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
-
-
-def test_duplicate_bootstrap_is_idempotent_without_consuming_generation(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-
-    first = runtime.bootstrap_writer_lease(writer_request())
-    second = runtime.bootstrap_writer_lease(writer_request())
-
-    assert first["bootstrapped"] is True
-    assert first["generation"] == 1
-    assert second["bootstrapped"] is False
-    assert second["idempotent"] is True
-    assert second["generation"] == 1
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
-
-
-def test_commit_after_bootstrap_reuses_the_exact_runtime_binding(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-
-    class RecordingWorkspace:
-
-        def __init__(self) -> None:
-            self.files: dict[str, bytes] = {}
-
-        def materialize_files(self, files: dict[str, bytes]) -> None:
-            self.files = dict(files)
-
-    class RecordingExecutor:
-
-        def __init__(self) -> None:
-            self.workspace = RecordingWorkspace()
-            self.binding = None
-            self.commit = None
-
-        def execute(self, binding: object, commit: object) -> dict[str, object]:
-            self.binding = binding
-            self.commit = commit
-            return {'task_id': binding.request.task_id, 'generation': binding.request.generation, 'files': sorted(self.workspace.files)}
-    executor = RecordingExecutor()
-    original_build = runtime._build_runtime
-
-    def build(lease: WriterLeaseRequestV2, *, job_id: str, lease_not_before: datetime | None=None, lease_expires_at: datetime | None=None) -> tuple[ContinuationRuntimeBinding, GitOpsJobController, object]:
-        binding, controller, _ = original_build(lease, job_id=job_id, lease_not_before=lease_not_before, lease_expires_at=lease_expires_at)
-        return (binding, controller, executor)
-    runtime._build_runtime = build
-    runtime.bootstrap_writer_lease(writer_request())
-    lease_response = runtime.acquire_lease(writer_request())
-    binding, controller, executor = runtime.runtimes[TASK_ID, 1]
-    assert lease_response['job_id'] == binding.job_id
-    assert lease_response['durable_branch_lease_present'] is True
-    _original_execute = executor.execute
-
-    def _execute_with_governed_commit_contract(*args, **kwargs):
-        _before_complete = kwargs.pop('before_complete', None)
-        _legacy = dict(_original_execute(*args, **kwargs))
-        _result = {**_legacy, 'job_id': binding.job_id, 'state': 'COMPLETED', 'side_effect_state': 'REMOTE_VERIFIED', 'receipt': {'authority_oid': 'b' * 40, 'git_remote_oid': 'b' * 40, 'api_branch_oid': 'b' * 40, 'commit_visible': True, 'remote_verified': True, 'force_push': False}}
-        if _before_complete is not None:
-            _before_complete('b' * 40)
-        return _result
-    executor.execute = _execute_with_governed_commit_contract
-    _bound_job_id = binding.job_id
-    result = runtime.commit_files({'task_id': TASK_ID, 'repository': 'aios/aios', 'generation': 1, 'target_branch': BRANCH, 'expected_branch_oid': BASE, 'idempotency_key': 'commit-after-bootstrap-v1', 'path': 'docs/recovery.txt', 'content_base64': 'YQ==', 'message': 'test commit after bootstrap'})
-    assert result['task_id'] == TASK_ID
-    assert result['generation'] == 1
-    assert result['files'] == ['docs/recovery.txt']
-    assert executor.binding.request.task_id == TASK_ID
-    assert executor.binding.request.generation == 1
-    database = runtime._job_root(TASK_ID) / 'state.sqlite3'
-    with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
-        assert connection.execute('SELECT COUNT(*) FROM branch_leases').fetchone()[0] == 1
-        assert connection.execute('PRAGMA foreign_key_check(branch_leases)').fetchall() == []
-    assert result['commit_oid'] == 'b' * 40
-    assert result['commit_visible'] is True
-    assert result['branch_mode'] == 'CONTINUE_EXISTING'
-    assert result['target_branch'] == BRANCH
-    assert int(result['generation']) == 1
-    with controller.store._connect() as connection:
-        _persisted = connection.execute('SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?', ('lease-' + str(_bound_job_id),)).fetchone()
-    assert _persisted is not None
-    assert _persisted[0] == 'b' * 40
-    assert int(_persisted[1]) == 1
-    fresh_runtime = make_runtime(tmp_path, authority_head='b' * 40)
-    fresh_response = fresh_runtime.bootstrap_writer_lease(writer_request(branch_mode='CONTINUE_EXISTING', expected_branch_oid='b' * 40))
-    fresh_binding, _, _ = fresh_runtime.runtimes[TASK_ID, 1]
-    assert fresh_response['job_id'] == str(_bound_job_id)
-    assert fresh_response['runtime_rehydrated'] is True
-    assert fresh_binding.job_id == str(_bound_job_id)
-    assert fresh_binding.request.expected_branch_oid == 'b' * 40
-    with fresh_runtime.controllers[TASK_ID].store._connect() as connection:
-        _fresh_lease = connection.execute('SELECT job_id,current_head_oid,revision FROM branch_leases WHERE lease_id=?', ('lease-' + str(_bound_job_id),)).fetchone()
-    assert tuple(_fresh_lease) == (str(_bound_job_id), 'b' * 40, 1)
-
-
-
-@pytest.mark.parametrize(
-    ("field", "value", "error"),
-    (
-        ("repository", "other/repository", "LEASE_BINDING_CONFLICT"),
-        ("base_oid", "b" * 40, "LEASE_BINDING_CONFLICT"),
-        ("expected_branch_oid", "b" * 40, "LEASE_BINDING_CONFLICT"),
-        ("target_branch", "agent/wrong-target", "LEASE_BINDING_CONFLICT"),
-        ("generation", 2, "GENERATION_BINDING_CONFLICT"),
-    ),
-)
-def test_bootstrap_rejects_any_nonmatching_binding_field(
-    tmp_path: Path, field: str, value: object, error: str
-) -> None:
-    runtime = make_runtime(tmp_path)
-
-    with pytest.raises(RuntimeError, match=error):
-        runtime.bootstrap_writer_lease(writer_request(**{field: value}))
-
-
-def test_bootstrap_rejects_unknown_task_ownership(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-
-    with pytest.raises(RuntimeError, match="TASK_OWNERSHIP_INVALID"):
-        runtime.bootstrap_writer_lease(writer_request(task_id="unknown-task"))
-
-
-def test_renew_missing_durable_lease_is_not_bootstrap(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-
-    with pytest.raises(RuntimeError, match="LEASE_NOT_FOUND"):
-        runtime.renew_writer_lease(writer_request())
-
-
-def test_renew_is_generation_scoped_before_durable_lookup(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    create_durable_lease(runtime)
-
-    with pytest.raises(RuntimeError, match="GENERATION_BINDING_CONFLICT"):
-        runtime.renew_writer_lease(writer_request(generation=2))
-    assert durable_row(runtime) == ("2020-01-01T00:00:00+00:00", 0)
-
-
-def test_restart_rehydrates_only_from_matching_durable_lease(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    create_durable_lease(runtime)
-    runtime.runtimes.clear()
-
-    result = runtime.renew_writer_lease(writer_request())
-
-    expires_at, revision = durable_row(runtime)
-    assert result["recovered"] is True
-    assert result["runtime_rehydrated"] is True
-    assert result["durable_branch_lease_present"] is True
-    assert revision == 1
-    assert datetime.fromisoformat(expires_at) > datetime.now(timezone.utc)
-
-
-def test_duplicate_renewal_is_idempotent_after_first_durable_recovery(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    create_durable_lease(runtime)
-
-    first = runtime.renew_writer_lease(writer_request())
-    second = runtime.renew_writer_lease(writer_request())
-
-    assert first["recovered"] is True
-    assert first["idempotent"] is False
-    assert second["recovered"] is False
-    assert second["idempotent"] is True
-    assert durable_row(runtime)[1] == 1
-
-
-def test_update_lease_keeps_expiry_without_optional_argument_and_keeps_cas(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    controller = create_durable_lease(runtime)
-
-    controller.update_lease(
-        "writer-lease-recovery",
-        expected_revision=0,
-        current_head_oid=BASE,
-        generation=1,
-    )
-    assert durable_row(runtime) == ("2020-01-01T00:00:00+00:00", 1)
-    with pytest.raises(JobError, match="LEASE_REVISION_CAS_CONFLICT"):
-        controller.update_lease(
-            "writer-lease-recovery",
-            expected_revision=0,
-            current_head_oid=BASE,
-            generation=1,
-        )
-
-
-def test_two_runtime_connections_resolve_renewal_cas_race_idempotently(tmp_path: Path) -> None:
-    first = make_runtime(tmp_path)
-    first_controller = create_durable_lease(first)
-    second = make_runtime(tmp_path)
-    second_controller = GitOpsJobController(
-        JobStore(second._job_root(TASK_ID), payload_root=second._job_root(TASK_ID) / "payloads")
-    )
-    second.controllers[TASK_ID] = second_controller
-    gate = threading.Barrier(2)
-
-    def gate_update(controller: GitOpsJobController) -> None:
-        original = controller.update_lease
-
-        def update(*args: object, **kwargs: object) -> None:
-            gate.wait(timeout=5)
-            original(*args, **kwargs)
-
-        controller.update_lease = update
-
-    gate_update(first_controller)
-    gate_update(second_controller)
-    with ThreadPoolExecutor(max_workers=2) as pool:
-        results = list(pool.map(lambda runtime: runtime.renew_writer_lease(writer_request()), (first, second)))
-
-    assert sorted((item["recovered"], item["idempotent"]) for item in results) == [
-        (False, True),
-        (True, False),
-    ]
-    assert durable_row(first)[1] == 1
-
-
-def test_bootstrap_preserves_exact_governed_job_lineage(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
-
-    response = runtime.bootstrap_writer_lease(writer_request())
-    binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
-
-    assert response["job_id"] == expected_job_id
-    assert binding.job_id == expected_job_id
-    assert binding.request.job_id == expected_job_id
-
-
-def test_restart_rehydrate_preserves_exact_governed_job_lineage(tmp_path: Path) -> None:
-    first = make_runtime(tmp_path)
-    expected_job_id = first.bindings[TASK_ID]["job_id"]
-    first.acquire_lease(writer_request())
-
-    runtime = make_runtime(tmp_path)
-    assert runtime is not first
-    assert runtime.runtimes == {}
-    response = runtime.bootstrap_writer_lease(writer_request())
-    binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
-
-    assert response["job_id"] == expected_job_id
-    assert binding.job_id == expected_job_id
-    assert binding.request.job_id == expected_job_id
-
-
-def test_wrong_governed_job_id_is_denied_before_runtime_or_lease_side_effect(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-
-    with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
-        runtime.bootstrap_writer_lease(writer_request(job_id="job-wrong-lineage"))
-
-    assert runtime.runtimes == {}
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
-        assert connection.execute("PRAGMA foreign_key_check(branch_leases)").fetchall() == []
-
-
-def test_acquire_creates_exact_governed_writer_lease_with_fk_and_cas(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
-
-    response = runtime.acquire_lease(writer_request())
-
-    assert response["job_id"] == expected_job_id
-    assert response["durable_branch_lease_present"] is True
-    assert response["idempotent"] is False
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        foreign_keys = connection.execute(
-            "PRAGMA foreign_key_list(branch_leases)"
-        ).fetchall()
-        row = connection.execute(
-            "SELECT job_id,branch,lineage_anchor_oid,current_head_oid,generation,state,revision "
-            "FROM branch_leases"
-        ).fetchone()
-        job = connection.execute(
-            "SELECT state,side_effect_state,revision FROM jobs WHERE job_id=?",
-            (expected_job_id,),
-        ).fetchone()
-        assert connection.execute("PRAGMA foreign_key_check(branch_leases)").fetchall() == []
-    assert any(
-        fk[2] == "jobs" and fk[3] == "job_id" and fk[4] == "job_id"
-        for fk in foreign_keys
-    )
-    assert row == (expected_job_id, BRANCH, BASE, BASE, 1, "ACTIVE", 0)
-    assert job == ("LEASED", "LEASED", 2)
-
-
-def test_duplicate_acquire_observes_same_durable_writer_lease(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
-
-    first = runtime.acquire_lease(writer_request())
-    second = runtime.acquire_lease(writer_request())
-
-    assert first["job_id"] == expected_job_id
-    assert second["job_id"] == expected_job_id
-    assert second["idempotent"] is True
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        leases = connection.execute(
-            "SELECT lease_id,job_id,revision FROM branch_leases"
-        ).fetchall()
-        job = connection.execute(
-            "SELECT state,revision FROM jobs WHERE job_id=?",
-            (expected_job_id,),
-        ).fetchone()
-    assert leases == [("lease-" + expected_job_id, expected_job_id, 0)]
-    assert job == ("LEASED", 2)
-
-
-def test_wrong_job_acquire_is_denied_before_authorize_or_lease(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
-
-    with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
-        runtime.acquire_lease(writer_request(job_id="job-wrong-lineage"))
-
-    database = runtime._job_root(TASK_ID) / "state.sqlite3"
-    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
-        job = connection.execute(
-            "SELECT state,side_effect_state,revision FROM jobs WHERE job_id=?",
-            (expected_job_id,),
-        ).fetchone()
-        lease_count = connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0]
-    assert job == ("REQUESTED", "NONE", 0)
-    assert lease_count == 0
-
-
-def test_omitted_request_job_id_resolves_the_exact_durable_parent(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
-    raw = dict(writer_request())
-
-    response = runtime.acquire_lease(raw)
-    binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
-
-    assert response["job_id"] == expected_job_id
-    assert binding.job_id == expected_job_id
-    assert binding.request.job_id == expected_job_id
-
-
-def test_durable_job_expected_head_mismatch_is_denied(tmp_path: Path) -> None:
-    runtime = make_runtime(tmp_path)
-    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
-    controller = runtime.controllers[TASK_ID]
-    with controller.store._connect() as connection:
-        row = connection.execute(
-            "SELECT request_json FROM jobs WHERE job_id=?", (expected_job_id,)
-        ).fetchone()
-        request = json.loads(row[0])
-        request["expected_branch_oid"] = "c" * 40
-        request["binding"]["expected_branch_oid"] = "c" * 40
-        connection.execute(
-            "UPDATE jobs SET request_json=? WHERE job_id=?",
-            (json.dumps(request, sort_keys=True, separators=(",", ":")), expected_job_id),
-        )
-        connection.commit()
-
-    with pytest.raises(RuntimeError, match="RUNTIME_GOVERNED_JOB_UNRESOLVED"):
-        runtime.bootstrap_writer_lease(writer_request())
-
-    assert runtime.runtimes == {}
-
-
-def _commit_ready_runtime(tmp_path: Path):
-    runtime = make_runtime(tmp_path)
-    runtime.bootstrap_writer_lease(writer_request())
-    binding, controller, _ = runtime.runtimes[(TASK_ID, 1)]
-
-    class Workspace:
-        def materialize_files(self, _files):
-            return None
-
-    class Executor:
-        def __init__(self):
-            self.workspace = Workspace()
-            self.calls = 0
-
-        def execute(self, _binding, _commit, *, before_complete=None):
-            self.calls += 1
-            result = {
-                "job_id": binding.job_id,
-                "state": "COMPLETED",
-                "side_effect_state": "REMOTE_VERIFIED",
-                "receipt": {
-                    "authority_oid": "b" * 40,
-                    "git_remote_oid": "b" * 40,
-                    "api_branch_oid": "b" * 40,
-                    "commit_visible": True,
-                    "remote_verified": True,
-                    "force_push": False,
-                },
-            }
-            if before_complete is not None:
-                before_complete("b" * 40)
-            return result
-
-    executor = Executor()
-    runtime.runtimes[(TASK_ID, 1)] = (binding, controller, executor)
-    runtime.acquire_lease(writer_request())
-    return runtime, binding, controller, executor
-
-
-def _commit_request():
-    return {
-        "task_id": TASK_ID,
-        "repository": "aios/aios",
-        "generation": 1,
-        "target_branch": BRANCH,
-        "expected_branch_oid": BASE,
-        "idempotency_key": "commit-cas-test-v1",
-        "path": "docs/cas.txt",
-        "content_base64": "YQ==",
-        "message": "test commit cas",
-    }
-
-
-def test_commit_lease_revision_cas_conflict_is_preserved(tmp_path: Path) -> None:
-    runtime, _binding, controller, _executor = _commit_ready_runtime(tmp_path)
-
-    def conflict(*_args, **_kwargs):
-        raise JobError("LEASE_REVISION_CAS_CONFLICT")
-
-    controller.update_lease = conflict
-    with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
-        runtime.commit_files(_commit_request())
-
-
-def test_commit_divergent_durable_head_fails_before_executor(tmp_path: Path) -> None:
-    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
-    controller.update_lease(
-        "lease-" + binding.job_id,
-        expected_revision=0,
-        current_head_oid="c" * 40,
-        generation=1,
-    )
-
-    with pytest.raises(RuntimeError, match="LEASE_HEAD_RECONCILIATION_REQUIRED"):
-        runtime.commit_files(_commit_request())
-    assert executor.calls == 0
-
-
-def test_remote_pending_never_advances_durable_head(tmp_path: Path) -> None:
-    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
-
-    def pending(_binding, _commit, **_kwargs):
-        executor.calls += 1
-        return {
-            "job_id": binding.job_id,
-            "state": "RECONCILIATION_REQUIRED",
-            "side_effect_state": "REMOTE_PUSH_PENDING",
-            "receipt": {
-                "authority_oid": "b" * 40,
-                "desired_remote_oid": "b" * 40,
-                "remote_verified": False,
-            },
-        }
-
-    executor.execute = pending
-    with pytest.raises(RuntimeError, match="RECONCILIATION_REQUIRED"):
-        runtime.commit_files(_commit_request())
-    with controller.store._connect() as connection:
-        row = connection.execute(
-            "SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
-            ("lease-" + binding.job_id,),
-        ).fetchone()
-    assert tuple(row) == (BASE, 0)
-
-
-def test_retry_accepts_exact_head_already_advanced_at_remote_verified_boundary(tmp_path: Path) -> None:
-    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
-    commit_oid = "b" * 40
-    controller.update_lease(
-        "lease-" + binding.job_id,
-        expected_revision=0,
-        current_head_oid=commit_oid,
-        generation=1,
-    )
-    receipt = {
-        "authority_oid": commit_oid,
-        "git_remote_oid": commit_oid,
-        "api_branch_oid": commit_oid,
-        "commit_visible": True,
-        "remote_verified": True,
-        "force_push": False,
-    }
-    with controller.store._connect() as connection:
-        connection.execute(
-            "UPDATE jobs SET state='REMOTE_VERIFIED',side_effect_state='REMOTE_VERIFIED',"
-            "receipt_json=? WHERE job_id=?",
-            (json.dumps(receipt, sort_keys=True, separators=(",", ":")), binding.job_id),
-        )
-        connection.commit()
-
-    result = runtime.commit_files(_commit_request())
-
-    assert result["commit_oid"] == commit_oid
-    assert executor.calls == 1
-    with controller.store._connect() as connection:
-        row = connection.execute(
-            "SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
-            ("lease-" + binding.job_id,),
-        ).fetchone()
-    assert tuple(row) == (commit_oid, 1)
-
-
-def test_completed_response_with_invalid_oid_uses_canonical_oid_error(tmp_path: Path) -> None:
-    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
-
-    def invalid_oid(_binding, _commit, **_kwargs):
-        executor.calls += 1
-        return {
-            "job_id": binding.job_id,
-            "state": "COMPLETED",
-            "side_effect_state": "REMOTE_VERIFIED",
-            "receipt": {
-                "authority_oid": "not-an-oid",
-                "commit_visible": True,
-                "remote_verified": True,
-            },
-        }
-
-    executor.execute = invalid_oid
-    with pytest.raises(RuntimeError, match="GOVERNED_COMMIT_OID_INVALID"):
-        runtime.commit_files(_commit_request())
-    with controller.store._connect() as connection:
-        row = connection.execute(
-            "SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
-            ("lease-" + binding.job_id,),
-        ).fetchone()
-    assert tuple(row) == (BASE, 0)