aios/active-runtime/app/aios_online_agent_worker.py

#!/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 secrets
import io
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import tempfile
import tarfile
import threading
import time
import urllib.error
import urllib.request
from typing import Any, Mapping, Sequence
from urllib.parse import unquote

from aios_online_agent_execution import (
    ExecutionProjectionError,
    validate_execution_envelope,
)
from aios_online_agent_checkpoint import (
    CheckpointBinding,
    CheckpointError,
    CheckpointStore,
)
from aios_external_agent_runtime import (
    ExternalAgentExchange,
    ExternalAgentRuntimeError,
)
from aios_online_agent_compact_map import (
    CompactInterfaceMapError,
    build_compact_interface_map,
)


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-17.2-durable")
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"
CHECKPOINT_DB = Path(os.environ.get("AIOS_WORKER_CHECKPOINT_DB", str(STATE_ROOT / "checkpoints.sqlite3")))
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"))
WORKER_CAPABILITIES = tuple(
    sorted(
        item.strip()
        for item in os.environ.get(
            "AIOS_WORKER_CAPABILITIES",
            "git.read,git.test,governed.write.branch,governed.write.file",
        ).split(",")
        if item.strip()
    )
)
WORKER_ALLOWED_ADAPTERS = frozenset(
    item.strip()
    for item in os.environ.get(
        "AIOS_WORKER_ALLOWED_ADAPTERS",
        "deterministic,source_capture,compact_interface_map,codex",
    ).split(",")
    if item.strip()
)
EXTERNAL_AGENT_EXECUTOR_ROOT = Path(
    os.environ.get(
        "AIOS_EXTERNAL_AGENT_EXECUTOR_ROOT",
        "/var/lib/aios-external-agent-executor",
    )
)
EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS = int(
    os.environ.get("AIOS_EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS", "1800")
)
MAX_TASK_BYTES = 512 * 1024
MAX_FILE_BYTES = 2 * 1024 * 1024
ISOLATED_EXECUTOR_ROOT = Path(
    os.environ.get("AIOS_ISOLATED_EXECUTOR_ROOT", "/var/lib/aios-online-agent-executor")
)
ISOLATED_EXECUTOR_TIMEOUT_SECONDS = int(
    os.environ.get("AIOS_ISOLATED_EXECUTOR_TIMEOUT_SECONDS", "1800")
)
ISOLATED_EXECUTOR_TIMEOUT_HARD_MAX_SECONDS = 7200
GIT_VISIBILITY_TIMEOUT_SECONDS = int(
    os.environ.get("AIOS_GIT_VISIBILITY_TIMEOUT_SECONDS", "45")
)
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}$")
REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,254}$")
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 _isolated_executor_wait_seconds(verification: Sequence[Mapping[str, Any]]) -> int:
    verification_budget = sum(int(item["timeout_seconds"]) for item in verification)
    required_wait_seconds = max(30, verification_budget + 30)
    if not 30 <= ISOLATED_EXECUTOR_TIMEOUT_SECONDS <= ISOLATED_EXECUTOR_TIMEOUT_HARD_MAX_SECONDS:
        raise WorkerError("ISOLATED_EXECUTOR_TIMEOUT_CONFIG_INVALID", phase="adapter")
    if required_wait_seconds > ISOLATED_EXECUTOR_TIMEOUT_SECONDS:
        raise WorkerError(
            "ISOLATED_EXECUTOR_TIMEOUT_BUDGET_EXCEEDED",
            phase="adapter",
            details={
                "required_seconds": required_wait_seconds,
                "configured_seconds": ISOLATED_EXECUTOR_TIMEOUT_SECONDS,
            },
        )
    return required_wait_seconds


def _claim_order(item: tuple[Mapping[str, Any], "ExecutionEnvelope"]) -> tuple[int, int, str, str]:
    task, envelope = item
    try:
        priority = int(task.get("priority") or 0)
    except (TypeError, ValueError):
        priority = 0
    action_rank = 0 if envelope.action == "deterministic" else 1
    return (-priority, action_rank, str(task.get("created_at") or ""), envelope.task_id)


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)



WORKER_LIVENESS_PATH = Path(
    os.environ.get(
        "AIOS_WORKER_LIVENESS_PATH",
        str(STATE_ROOT / "worker-liveness.json"),
    )
)
_LIVENESS_LOCK = threading.Lock()
_LIVENESS_SEQUENCE = 0
ENGINEERING_WRITE_CAPABILITIES = frozenset({
    "governed.write.branch",
    "governed.write.file",
})


def write_worker_liveness(
    state: str,
    *,
    visible_count: int | None = None,
    task_id: str | None = None,
    stage: str | None = None,
    error: str | None = None,
) -> None:
    """Write idle/busy liveness without touching Task/Job state."""
    global _LIVENESS_SEQUENCE
    with _LIVENESS_LOCK:
        _LIVENESS_SEQUENCE += 1
        epoch = time.time()
        payload = {
            "schema": "aios.online-agent.worker-liveness.v1",
            "agent_id": AGENT_ID,
            "agent_type": AGENT_TYPE,
            "agent_version": AGENT_VERSION,
            "pid": os.getpid(),
            "state": state,
            "sequence": _LIVENESS_SEQUENCE,
            "last_seen_epoch": epoch,
            "last_seen_at": datetime.fromtimestamp(epoch, timezone.utc).isoformat(),
            "poll_seconds": POLL_SECONDS,
            "heartbeat_seconds": HEARTBEAT_SECONDS,
            "api_base": API_BASE,
            "bridge_base": BRIDGE_BASE,
            "visible_count": visible_count,
            "task_id": task_id,
            "stage": stage,
            "error": error,
        }
        atomic_json(WORKER_LIVENESS_PATH, payload)


def source_capture_requires_deterministic_plan(envelope: "ExecutionEnvelope") -> bool:
    """Treat write-capable source_capture as a binding placeholder, never executable work."""
    return (
        envelope.action == "source_capture"
        and bool(ENGINEERING_WRITE_CAPABILITIES.intersection(envelope.capability_scope))
    )


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
    job_id: str
    repository: str
    base_oid: str
    expected_branch_oid: str
    target_branch: str
    branch_mode: str
    generation: int
    idempotency_key: str
    action: str
    adapter: str
    capability_scope: tuple[str, ...]
    source_signals: tuple[str, ...]
    change_set: tuple[dict[str, Any], ...]
    verification: tuple[dict[str, Any], ...]
    governance: dict[str, Any]
    workspace: dict[str, Any]
    digest: str
    runtime_input: dict[str, Any] | None = None

    @classmethod
    def from_task(
        cls,
        task: Mapping[str, Any],
        execution_override: Mapping[str, Any] | None = None,
        worker_capabilities: tuple[str, ...] | None = None,
    ) -> "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 = execution_override if execution_override is not None else metadata.get("execution")
        if not isinstance(execution, Mapping):
            raise WorkerError("DYNAMIC_EXECUTION_PAYLOAD_REQUIRED", phase="validation")
        try:
            execution = validate_execution_envelope(execution, worker_capabilities=worker_capabilities)
        except ExecutionProjectionError as exc:
            raise WorkerError(str(exc), phase="validation") from exc
        if execution.get("task_id") != task_id:
            raise WorkerError("TASK_BINDING_CONFLICT", phase="validation")
        action = str(execution["action"])
        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", "source_capture", "compact_interface_map", "codex", "hermes", "openclaw"}:
            raise WorkerError("RUNTIME_ADAPTER_UNSUPPORTED", phase="validation")
        change_set = execution.get("change_set")
        if not isinstance(change_set, list) or len(change_set) > 128 or (action == "deterministic" and not change_set):
            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,
            job_id=str(execution["job_id"]),
            repository=repository,
            base_oid=base_oid,
            expected_branch_oid=str(execution["expected_branch_oid"]),
            target_branch=target_branch,
            branch_mode=str(execution["branch_mode"]),
            generation=int(execution["generation"]),
            idempotency_key=str(execution["idempotency_key"]),
            action=action,
            adapter=adapter,
            capability_scope=tuple(execution.get("capability_scope") or ()),
            source_signals=tuple(execution.get("source_signals") or ()),
            runtime_input=dict(execution.get("runtime_input") or {}),
            change_set=tuple(normalized_changes),
            verification=tuple(normalized_verification),
            governance=governance_value,
            workspace=workspace_value,
            digest=str(execution["execution_digest"]),
        )


def external_agent_request_id(
    envelope: ExecutionEnvelope,
    runtime_name: str,
    attempt: int,
) -> str:
    if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
        raise WorkerError("TASK_ATTEMPT_INVALID", phase="external_agent")
    if runtime_name not in {"hermes", "openclaw"}:
        raise WorkerError(
            "EXTERNAL_AGENT_RUNTIME_UNSUPPORTED",
            phase="external_agent",
            details={"runtime": runtime_name},
        )
    return (
        f"external-{runtime_name}-"
        + digest_json(
            {
                "task_id": envelope.task_id,
                "job_id": envelope.job_id,
                "execution_digest": envelope.digest,
                "runtime": runtime_name,
                "attempt": attempt,
            }
        )[:32]
    )


class RuntimeAdapter(ABC):
    name: str

    @abstractmethod
    def execute(
        self,
        envelope: ExecutionEnvelope,
        worktree: Path,
        *,
        attempt: int,
    ) -> 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,
        *,
        attempt: int,
    ) -> 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 SourceCaptureExecutor(RuntimeAdapter):
    """Read-only repository source capture for a first implementation attempt."""

    name = "source_capture"
    roots = ("active-runtime/app", "tests", "migrations", "deploy/systemd")
    suffixes = {".py", ".sql", ".yaml", ".yml", ".json", ".md", ".service"}
    defaults = {
        "agent", "task", "worker", "execution", "projection", "claim", "heartbeat",
        "governed", "trace", "tool", "credential", "scope", "message", "handoff",
    }

    def execute(
        self,
        envelope: ExecutionEnvelope,
        worktree: Path,
        *,
        attempt: int,
    ) -> dict[str, Any]:
        terms = set(self.defaults)
        for signal in envelope.source_signals:
            terms.update(token.lower() for token in re.findall(r"[A-Za-z][A-Za-z0-9_.-]{2,63}", signal))
        terms = set(sorted(terms)[:64])
        selected: list[tuple[str, bytes]] = []
        total = 0
        root = worktree.resolve()
        for relative_root in self.roots:
            candidate_root = resolve_under(root, relative_root)
            if not candidate_root.is_dir() or candidate_root.is_symlink():
                continue
            for path in sorted(candidate_root.rglob("*")):
                if len(selected) >= 96 or total >= 3 * 1024 * 1024:
                    break
                if path.is_symlink() or not path.is_file() or path.suffix.lower() not in self.suffixes:
                    continue
                resolved = path.resolve(strict=True)
                if root not in resolved.parents:
                    raise WorkerError("SOURCE_CAPTURE_PATH_ESCAPE", phase="source_capture")
                data = path.read_bytes()
                if len(data) > MAX_FILE_BYTES:
                    continue
                relative = path.relative_to(root).as_posix()
                searchable = (relative + "\n" + data[:256000].decode("utf-8", "ignore")).lower()
                if not any(term in searchable for term in terms):
                    continue
                selected.append((relative, data))
                total += len(data)
        if not selected:
            raise WorkerError("SOURCE_CAPTURE_EMPTY", phase="source_capture")
        payload = io.BytesIO()
        with tarfile.open(fileobj=payload, mode="w:gz") as archive:
            for relative, data in selected:
                info = tarfile.TarInfo(relative)
                info.size = len(data)
                info.mode = 0o640
                info.mtime = 0
                archive.addfile(info, io.BytesIO(data))
        compressed = payload.getvalue()
        if len(compressed) > MAX_TASK_BYTES:
            raise WorkerError("SOURCE_CAPTURE_TOO_LARGE", phase="source_capture", details={"compressed_bytes": len(compressed)})
        manifest = [
            {"path": relative, "bytes": len(data), "sha256": digest_bytes(data), "source": "repository"}
            for relative, data in selected
        ]
        audit("ENGINEERING_SOURCE_CAPTURED", file_count=len(manifest), compressed_bytes=len(compressed))
        return {
            "changed_files": [],
            "commands": [],
            "tests": [],
            "adapter": self.name,
            "source_snapshot": {
                "schema": "aios.engineering.source-snapshot.v2",
                "encoding": "tar.gz+base64",
                "payload_b64": base64.b64encode(compressed).decode("ascii"),
                "payload_sha256": digest_bytes(compressed),
                "compressed_bytes": len(compressed),
                "file_count": len(manifest),
                "files": manifest,
            },
        }


class CompactInterfaceMapExecutor(RuntimeAdapter):
    """Bounded read-only source analysis carried by the canonical source_capture action."""

    name = "compact_interface_map"

    def execute(
        self,
        envelope: ExecutionEnvelope,
        worktree: Path,
        *,
        attempt: int,
    ) -> dict[str, Any]:
        if envelope.action != "source_capture":
            raise WorkerError("COMPACT_INTERFACE_MAP_ACTION_INVALID", phase="interface_map")
        try:
            interface_map = build_compact_interface_map(
                worktree,
                envelope.source_signals,
                source_commit=envelope.base_oid,
            )
        except CompactInterfaceMapError as exc:
            raise WorkerError(str(exc), phase="interface_map") from exc
        audit(
            "ENGINEERING_INTERFACE_MAP_CAPTURED",
            module_count=len(interface_map.get("implementation_map", {}).get("modules", [])),
            excerpt_count=len(interface_map.get("bounded_excerpts", [])),
            result_bytes=interface_map.get("result_bytes"),
            source_commit=envelope.base_oid,
        )
        return {
            "changed_files": [],
            "commands": [],
            "tests": [],
            "adapter": self.name,
            "interface_map": interface_map,
        }


class CodexExecutor(RuntimeAdapter):
    name = "codex"

    def execute(
        self,
        envelope: ExecutionEnvelope,
        worktree: Path,
        *,
        attempt: int,
    ) -> 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 ExternalAgentExecutor(RuntimeAdapter):
    runtime_name: str

    def execute(
        self,
        envelope: ExecutionEnvelope,
        worktree: Path,
        *,
        attempt: int,
    ) -> dict[str, Any]:
        if envelope.action != "agent_runtime":
            raise WorkerError(
                "EXTERNAL_AGENT_ACTION_INVALID",
                phase="adapter",
                details={"adapter": self.runtime_name, "action": envelope.action},
            )
        if envelope.adapter != self.runtime_name:
            raise WorkerError(
                "EXTERNAL_AGENT_ADAPTER_BINDING_INVALID",
                phase="adapter",
            )
        if not envelope.runtime_input:
            raise WorkerError("EXTERNAL_AGENT_RUNTIME_INPUT_MISSING", phase="adapter")
        request_id = external_agent_request_id(
            envelope,
            self.runtime_name,
            attempt,
        )
        exchange = ExternalAgentExchange(EXTERNAL_AGENT_EXECUTOR_ROOT)
        try:
            external = exchange.execute(
                request_id=request_id,
                runtime=self.runtime_name,
                task_id=envelope.task_id,
                attempt=attempt,
                execution_digest=envelope.digest,
                payload=envelope.runtime_input,
                timeout_seconds=EXTERNAL_AGENT_EXECUTOR_TIMEOUT_SECONDS,
            )
        except ExternalAgentRuntimeError as exc:
            raise WorkerError(
                str(exc),
                phase="external_agent",
                details={"adapter": self.runtime_name},
            ) from exc
        audit(
            "EXTERNAL_AGENT_RUNTIME_COMPLETED",
            task_id=envelope.task_id,
            adapter=self.runtime_name,
            request_id=request_id,
            attempt=attempt,
            output_sha256=external.get("output_sha256"),
            duration_ms=external.get("duration_ms"),
            credential_material_exposed=False,
        )
        return {
            "changed_files": [],
            "commands": [],
            "tests": [],
            "adapter": self.runtime_name,
            "external_agent": external,
        }


class HermesExecutor(ExternalAgentExecutor):
    name = "hermes"
    runtime_name = "hermes"


class OpenClawExecutor(ExternalAgentExecutor):
    name = "openclaw"
    runtime_name = "openclaw"


def adapter_for(name: str) -> RuntimeAdapter:
    return {
        "deterministic": IsolatedDeterministicExecutor,
        "source_capture": SourceCaptureExecutor,
        "compact_interface_map": CompactInterfaceMapExecutor,
        "codex": CodexExecutor,
        "hermes": HermesExecutor,
        "openclaw": OpenClawExecutor,
    }[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 get(self, task_id: str) -> tuple[int, dict[str, Any]]:
        return HttpClient.call(API_BASE, "GET", f"/v1/tasks/{task_id}", self.token)

    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 __init__(self, token: str) -> None:
        self.token = token

    def call(self, route: str, body: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
        return HttpClient.call(BRIDGE_BASE, "POST", route, self.token, body, timeout=60)

    def project_execution(self, task: Mapping[str, Any], execution: Mapping[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
        request: dict[str, Any] = {
            "task_id": str(task.get("task_id") or ""),
            "required_capabilities": list(task.get("required_capabilities") or []),
        }
        if execution is not None:
            request["execution"] = dict(execution)
        return self.call("/v2/execution-envelope", request)


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)
        write_worker_liveness(
            "BUSY",
            task_id=self.task_id,
            stage=self.stage,
        )

    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)
                    write_worker_liveness(
                        "BUSY",
                        task_id=self.task_id,
                        stage=stage,
                    )
            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,
    *,
    attempt: int | None = None,
    continuation_head_oid: str | None = None,
    continuation_job_id: str | None = None,
    commit_idempotency_key: str | None = None,
) -> 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}:
        if value.get("phase") in {"claimed", "workspace_prepared", "source_capture_done"} and not value.get("governed"):
            return None
        if (
            value.get("phase") == "result_submitted"
            and not value.get("governed")
            and attempt is not None
            and int(value.get("attempt") or 0) < int(attempt)
        ):
            # A prior read-only/source-capture or pre-commit failure may leave a
            # result_submitted runtime marker with the old envelope digest. It has
            # no governed side effect, so a later official attempt may start from
            # the newly CAS-bound deterministic plan without erasing the audit.
            return None
        if (
            value.get("phase") == "verification_passed"
            and not value.get("governed")
            and attempt is not None
            and int(value.get("attempt") or 0) < int(attempt)
        ):
            # The prior official attempt completed deterministic validation but failed before
            # a WriterLease or governed commit. A later CAS-authorized retry may use its new
            # digest; the same attempt can never erase or replace this state.
            return None
        if value.get("phase") == "reconciliation_required" and not value.get("governed"):
            stored_attempt = value.get("attempt")
            if (
                stored_attempt is not None
                and attempt is not None
                and int(stored_attempt) < int(attempt)
                and value.get("job_id") == continuation_job_id
                and isinstance(continuation_head_oid, str)
                and OID_RE.fullmatch(continuation_head_oid)
            ):
                return None
            details = value.get("details")
            if (
                stored_attempt is None
                and attempt is not None
                and isinstance(details, Mapping)
                and details.get("api_status") == 503
                and details.get("error") == "WORKTREE_DIRTY_SCOPE_MISMATCH"
                and value.get("job_id") == continuation_job_id
                and isinstance(continuation_head_oid, str)
                and OID_RE.fullmatch(continuation_head_oid)
            ):
                # Compatibility for the pre-fix recovery record written by attempt 18,
                # which omitted its attempt number. This exact pre-commit failure has no
                # governed side effect; plan-registry CAS and the continuation head still
                # bind the later official claim before the stale state can be discarded.
                return None
            if (
                stored_attempt is not None
                and attempt is not None
                and int(stored_attempt) < int(attempt)
                and value.get("job_id") == continuation_job_id
                and continuation_head_oid is None
                and isinstance(details, Mapping)
                and details.get("api_status") == 503
                and details.get("error") == "SECRET_LIKE_PAYLOAD_REJECTED"
            ):
                # SECRET_LIKE_PAYLOAD_REJECTED is a deterministic policy denial from the
                # governed bridge, not an ambiguous remote write. It cannot create a branch,
                # receipt, or commit, so a later CAS-authorized CREATE_NEW attempt may safely
                # discard this stale reconciliation marker after the same Task/Job binding is
                # re-established by the official execution plan.
                return None
        governed = value.get("governed")
        if (
            continuation_head_oid
            and isinstance(governed, Mapping)
            and governed.get("commit_oid") == continuation_head_oid
            and governed.get("commit_idempotency_key") != commit_idempotency_key
        ):
            # A previous governed commit is the exact continuation head; a new digest/key is a
            # legitimate next deterministic plan on the same Task/Job lineage.
            return None
        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 checkpoint_binding(envelope: ExecutionEnvelope, attempt: int) -> CheckpointBinding:
    return CheckpointBinding(
        task_id=envelope.task_id,
        attempt=attempt,
        execution_digest=envelope.digest,
        job_id=envelope.job_id,
        repository=envelope.repository,
        base_oid=envelope.base_oid,
        expected_branch_oid=envelope.expected_branch_oid,
        target_branch=envelope.target_branch,
        generation=envelope.generation,
    )


_CHECKPOINT_STORE: CheckpointStore | None = None


def checkpoint_store() -> CheckpointStore:
    global _CHECKPOINT_STORE
    if _CHECKPOINT_STORE is None:
        _CHECKPOINT_STORE = CheckpointStore(CHECKPOINT_DB)
    return _CHECKPOINT_STORE


def save_checkpoint(
    task_id: str,
    state: Mapping[str, Any],
    envelope: ExecutionEnvelope,
    attempt: int,
) -> dict[str, Any]:
    phase = str(state.get("phase") or "")
    try:
        return checkpoint_store().append(
            checkpoint_binding(envelope, attempt),
            phase,
            dict(state),
        )
    except CheckpointError as exc:
        raise WorkerError(str(exc), phase="checkpoint") from exc


def latest_reusable_checkpoint(
    envelope: ExecutionEnvelope,
    attempt: int,
) -> dict[str, Any] | None:
    try:
        return checkpoint_store().verify_latest_binding(
            checkpoint_binding(envelope, attempt)
        )
    except CheckpointError as exc:
        raise WorkerError(str(exc), phase="checkpoint") from exc


def result_idempotency_key(
    task_id: str,
    attempt: int,
    envelope: ExecutionEnvelope,
    result: Mapping[str, Any],
) -> str:
    governed = result.get("governed")
    commit_key = (
        governed.get("commit_idempotency_key")
        if isinstance(governed, Mapping)
        else None
    )
    material = {
        "task_id": task_id,
        "attempt": attempt,
        "execution_digest": envelope.digest,
        "commit_idempotency_key": commit_key,
    }
    readable = re.sub(r"[^A-Za-z0-9._:-]+", "-", task_id).strip("-")[:72] or "task"
    return f"{readable}-a{attempt}-result-{digest_json(material)[:24]}"


def submit_result_replay_safe(
    client: AgentTaskClient,
    task_id: str,
    lease_token: str,
    result_key: str,
    result: Mapping[str, Any],
) -> tuple[int, dict[str, Any]]:
    expected_digest = digest_json(result)
    delay = 0.5
    last_status = 599
    last_response: dict[str, Any] = {"error": "SERVICE_UNAVAILABLE"}
    for _ in range(4):
        status, response = client.result(task_id, lease_token, result_key, result)
        last_status, last_response = status, response
        if 200 <= status < 300:
            return status, response
        if status == 409 and response.get("error") == "IDEMPOTENCY_CONFLICT":
            get_status, current = client.get(task_id)
            current_result = current.get("result") if isinstance(current, Mapping) else None
            if (
                get_status == 200
                and isinstance(current_result, Mapping)
                and digest_json(current_result) == expected_digest
                and current.get("status") in {"VERIFYING", "SUCCEEDED"}
            ):
                return 200, {
                    "idempotent": True,
                    "reconciled": True,
                    "status": current.get("status"),
                }
            return status, response
        if status != 599 and status < 500:
            return status, response
        time.sleep(delay)
        delay = min(delay * 2.0, 4.0)
    return last_status, last_response


def recover_stale_workspaces() -> int:
    if not WORKSPACE_ROOT.is_dir() or WORKSPACE_ROOT.is_symlink():
        return 0
    removed = 0
    for path in sorted(WORKSPACE_ROOT.glob("task-*")):
        if path.is_symlink() or not path.is_dir():
            continue
        resolved = path.resolve(strict=False)
        root = WORKSPACE_ROOT.resolve()
        if resolved == root or root not in resolved.parents:
            raise WorkerError("WORKSPACE_RECOVERY_SCOPE_INVALID", phase="startup")
        shutil.rmtree(resolved)
        removed += 1
    if removed:
        audit("WORKSPACE_RECOVERY_CLEANUP", removed=removed)
    return removed


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, envelope: ExecutionEnvelope) -> Path:
    clone = run_root / "repository"
    worktree = run_root / "worktree"
    worktree_oid = (
        envelope.expected_branch_oid
        if envelope.branch_mode == "CONTINUE_EXISTING"
        else envelope.base_oid
    )
    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", worktree_oid + "^{commit}"], cwd=run_root, timeout=60)
    run_command("workspace_worktree_create", ["/usr/bin/git", "-C", str(clone), "worktree", "add", "--detach", str(worktree), worktree_oid], cwd=run_root, timeout=300, internal_git=True)
    return worktree


def _archive_worktree(worktree: Path, archive_path: Path) -> str:
    archive_path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{archive_path.name}.",
        dir=str(archive_path.parent),
    )
    os.close(descriptor)
    temporary_path = Path(temporary_name)
    try:
        with tarfile.open(temporary_path, mode="w:gz") as archive:
            for path in sorted(worktree.rglob("*")):
                relative = path.relative_to(worktree)
                if ".git" in relative.parts or path.is_symlink() or not path.is_file():
                    continue
                data = path.read_bytes()
                info = tarfile.TarInfo(relative.as_posix())
                info.size = len(data)
                info.mode = 0o640
                info.mtime = 0
                archive.addfile(info, io.BytesIO(data))

        # Validate the closed temporary archive before it becomes visible to the
        # uncredentialed executor. A killed coordinator must leave no final-path
        # artifact that the executor can mistake for a complete source archive.
        with tarfile.open(temporary_path, mode="r:gz") as verification:
            for _ in verification:
                pass
        with temporary_path.open("rb") as handle:
            os.fsync(handle.fileno())
        temporary_path.chmod(0o640)
        os.replace(temporary_path, archive_path)
        try:
            directory_fd = os.open(archive_path.parent, os.O_RDONLY | os.O_DIRECTORY)
        except (AttributeError, OSError):
            directory_fd = None
        if directory_fd is not None:
            try:
                os.fsync(directory_fd)
            finally:
                os.close(directory_fd)
        return digest_bytes(archive_path.read_bytes())
    finally:
        temporary_path.unlink(missing_ok=True)


class IsolatedDeterministicExecutor(RuntimeAdapter):
    """Coordinator-side client for the uncredentialed executor exchange."""

    name = "isolated-deterministic"

    @staticmethod
    def _atomic(path: Path, data: bytes) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
        try:
            with os.fdopen(fd, "wb") as handle:
                handle.write(data)
                handle.flush()
                os.fsync(handle.fileno())
            os.chmod(temporary, 0o640)
            os.replace(temporary, path)
        finally:
            if os.path.exists(temporary):
                os.unlink(temporary)

    def execute(
        self,
        envelope: ExecutionEnvelope,
        worktree: Path,
        *,
        attempt: int,
    ) -> dict[str, Any]:
        if envelope.action != "deterministic":
            raise WorkerError("ISOLATED_EXECUTOR_ACTION_INVALID", phase="adapter")
        root = ISOLATED_EXECUTOR_ROOT
        for child in ("requests", "results", "failed", "sources"):
            candidate = root / child
            if candidate.is_symlink() or not candidate.is_dir():
                raise WorkerError("ISOLATED_EXECUTOR_UNAVAILABLE", phase="adapter", details={"path": str(candidate)})
        request_id = f"{envelope.task_id}.{secrets.token_hex(16)}"
        if not REQUEST_ID_RE.fullmatch(request_id):
            raise WorkerError("ISOLATED_EXECUTOR_REQUEST_ID_INVALID", phase="adapter")
        archive = root / "sources" / f"{request_id}.tgz"
        request_path = root / "requests" / f"{request_id}.json"
        result_path = root / "results" / f"{request_id}.json"
        failed_path = root / "failed" / f"{request_id}.json"
        for stale in (request_path, result_path, failed_path, archive):
            if stale.exists() or stale.is_symlink():
                raise WorkerError(
                    "ISOLATED_EXECUTOR_COLLISION_DETECTED", phase="adapter", details={"request_id": request_id, "path": str(stale)}
                )
        archive_sha = _archive_worktree(worktree, archive)
        request = {
            "schema": "aios.isolated-execution.request.v1",
            "request_id": request_id,
            "task_id": envelope.task_id,
            "job_id": envelope.job_id,
            "execution_digest": envelope.digest,
            "source_archive": str(archive),
            "source_archive_sha256": archive_sha,
            "change_set": list(envelope.change_set),
            "verification": list(envelope.verification),
        }
        request_sha = digest_json(request)
        self._atomic(request_path, json.dumps(request, sort_keys=True, separators=(",", ":")).encode("utf-8"))
        wait_seconds = _isolated_executor_wait_seconds(envelope.verification)
        deadline = time.monotonic() + wait_seconds
        try:
            while time.monotonic() < deadline:
                if failed_path.is_file():
                    details = json.loads(failed_path.read_text(encoding="utf-8"))
                    raise WorkerError("ISOLATED_EXECUTOR_FAILED", phase="adapter", details={"error": details.get("error")})
                if result_path.is_file():
                    result = json.loads(result_path.read_text(encoding="utf-8"))
                    if result.get("request_id") != request_id:
                        raise WorkerError("ISOLATED_EXECUTOR_REQUEST_ID_MISMATCH", phase="adapter")
                    if result.get("schema") != "aios.isolated-execution.result.v1":
                        raise WorkerError("ISOLATED_EXECUTOR_RESULT_INVALID", phase="adapter")
                    if result.get("request_sha256") != request_sha:
                        raise WorkerError("ISOLATED_EXECUTOR_REQUEST_BINDING_INVALID", phase="adapter")
                    material = {key: value for key, value in result.items() if key != "result_sha256"}
                    if result.get("result_sha256") != digest_json(material):
                        raise WorkerError("ISOLATED_EXECUTOR_RESULT_DIGEST_INVALID", phase="adapter")
                    isolation = result.get("credential_isolation")
                    if not isinstance(isolation, Mapping) or not all(isolation.get(k) is True for k in (
                        "credential_env_absent", "credential_mount_inaccessible", "governed_bridge_unreachable"
                    )):
                        raise WorkerError("CREDENTIAL_ISOLATION_VERIFY", phase="adapter")
                    expected_paths = sorted({str(item["path"]) for item in envelope.change_set})
                    returned = result.get("changed_files")
                    if not isinstance(returned, list) or sorted(str(item.get("path")) for item in returned) != expected_paths:
                        raise WorkerError("ISOLATED_EXECUTOR_CHANGED_FILESET_INVALID", phase="adapter")
                    for item in returned:
                        relative = safe_relative_path(item.get("path"), field="isolated_changed_file")
                        path = resolve_under(worktree, relative, write=True)
                        if item.get("deleted") is True:
                            path.unlink(missing_ok=True)
                            continue
                        raw = base64.b64decode(str(item.get("content_b64") or ""), validate=True)
                        if digest_bytes(raw) != item.get("sha256"):
                            raise WorkerError("ISOLATED_EXECUTOR_FILE_DIGEST_INVALID", phase="adapter")
                        self._atomic(path, raw)
                    commands = [
                        run_command("git_diff_check", ["/usr/bin/git", "-C", str(worktree), "diff", "--check"], cwd=worktree, timeout=120),
                        run_command("git_diff_stat", ["/usr/bin/git", "-C", str(worktree), "diff", "--no-ext-diff", "--stat"], cwd=worktree, timeout=120),
                    ]
                    return {
                        "changed_files": expected_paths,
                        "commands": commands,
                        "tests": list(result.get("verification") or []),
                        "adapter": self.name,
                        "credential_isolation": dict(isolation),
                    }
                time.sleep(0.25)
            raise WorkerError("ISOLATED_EXECUTOR_TIMEOUT", phase="adapter")
        finally:
            request_path.unlink(missing_ok=True)
            result_path.unlink(missing_ok=True)
            failed_path.unlink(missing_ok=True)
            archive.unlink(missing_ok=True)


def _verify_converged_job(
    response: Mapping[str, Any],
    envelope: ExecutionEnvelope,
    commit_oid: str,
) -> None:
    if response.get("state") != "COMPLETED":
        raise WorkerError("GOVERNED_RECEIPT_STATE_INVALID", phase="verification")
    receipt = response.get("receipt")
    if not isinstance(receipt, Mapping) or receipt.get("remote_verified") is not True:
        raise WorkerError("GOVERNED_RECEIPT_REQUIRED", phase="verification")
    for field in ("authority_oid", "api_branch_oid", "git_remote_oid"):
        if receipt.get(field) != commit_oid:
            raise WorkerError("GOVERNED_RECEIPT_OID_MISMATCH", phase="verification", details={"field": field})
    leases = response.get("branch_leases")
    if not isinstance(leases, list):
        raise WorkerError("WRITER_LEASE_RECOVERY_EVIDENCE_MISSING", phase="verification")
    active = [
        item for item in leases
        if isinstance(item, Mapping)
        and item.get("state") == "ACTIVE"
        and item.get("job_id") == envelope.job_id
        and item.get("branch") == envelope.target_branch
    ]
    if len(active) != 1:
        raise WorkerError("WRITER_LEASE_RECOVERY_EVIDENCE_INVALID", phase="verification")
    lease = active[0]
    if (
        int(lease.get("generation") or 0) != envelope.generation
        or lease.get("current_head_oid") != commit_oid
    ):
        raise WorkerError("WRITER_LEASE_CURRENT_HEAD_MISMATCH", phase="verification")


def acquire_writer_lease_replay_safe(
    envelope: ExecutionEnvelope,
    bridge: BridgeClient,
    heartbeat: LeaseHeartbeat,
) -> dict[str, Any]:
    governance = envelope.governance
    body = {
        "schema": "aios.task-writer-lease.v2",
        "task_id": envelope.task_id,
        "job_id": envelope.job_id,
        "repository": envelope.repository,
        "branch_mode": envelope.branch_mode,
        "base_oid": envelope.base_oid,
        "expected_branch_oid": envelope.expected_branch_oid,
        "target_branch": envelope.target_branch,
        "generation": envelope.generation,
        "lease_seconds": int(governance.get("lease_seconds") or 900),
    }
    delay = 0.5
    last_status = 599
    last_response: dict[str, Any] = {"error": "SERVICE_UNAVAILABLE"}
    for attempt_index in range(4):
        heartbeat.ensure()
        status, response = bridge.call("/v2/writer-lease", body)
        last_status, last_response = status, response
        if 200 <= status < 300:
            if (
                response.get("job_id") != envelope.job_id
                or response.get("task_id") != envelope.task_id
                or response.get("repository") != envelope.repository
                or response.get("base_oid") != envelope.base_oid
                or response.get("expected_branch_oid") != envelope.expected_branch_oid
                or response.get("branch_mode") != envelope.branch_mode
                or int(response.get("generation") or 0) != envelope.generation
                or response.get("target_branch") != envelope.target_branch
                or response.get("state") not in {None, "ACTIVE"}
            ):
                raise WorkerError(
                    "WRITER_LEASE_RESPONSE_BINDING_INVALID",
                    phase="committing",
                )
            if attempt_index:
                audit(
                    "WRITER_LEASE_RECOVERED",
                    task_id=envelope.task_id,
                    job_id=envelope.job_id,
                    attempts=attempt_index + 1,
                    lease_id=response.get("lease_id"),
                )
            return response
        if status != 599 and status < 500:
            break
        time.sleep(delay)
        delay = min(delay * 2.0, 4.0)
    raise WorkerError(
        "WRITER_LEASE_ACQUIRE_FAILED",
        phase="committing",
        details={
            "api_status": last_status,
            "error": last_response.get("error"),
        },
    )


def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeClient, heartbeat: LeaseHeartbeat) -> dict[str, Any]:
    governance = envelope.governance
    if governance.get("task_id") != envelope.task_id or governance.get("job_id") != envelope.job_id:
        raise WorkerError("GOVERNED_LINEAGE_INVALID", phase="committing")
    generation = envelope.generation
    expected_branch_oid = envelope.expected_branch_oid
    branch_mode = envelope.branch_mode
    lease_response = acquire_writer_lease_replay_safe(
        envelope,
        bridge,
        heartbeat,
    )
    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": envelope.task_id,
        "job_id": envelope.job_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:
        error = commit_response.get("error")
        ambiguous = status == 599 or (
            status >= 500 and error != "SECRET_LIKE_PAYLOAD_REJECTED"
        )
        code = "RECONCILIATION_REQUIRED" if ambiguous else (
            str(error) if isinstance(error, str) and error else "GOVERNED_COMMIT_FAILED"
        )
        raise WorkerError(code, phase="committing", details={"api_status": status, "error": commit_response.get("error"), "job_id": envelope.job_id})
    if (
        commit_response.get("job_id") != envelope.job_id
        or commit_response.get("target_branch") != envelope.target_branch
        or int(commit_response.get("generation") or 0) != generation
        or commit_response.get("branch_mode") not in {"CREATE_NEW", "CONTINUE_EXISTING"}
    ):
        raise WorkerError("GOVERNED_COMMIT_RESPONSE_BINDING_INVALID", phase="committing")
    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")
    receipt = commit_response.get("receipt")
    if not isinstance(receipt, Mapping) or receipt.get("remote_verified") is not True:
        raise WorkerError("GOVERNED_COMMIT_RECEIPT_REQUIRED", phase="verification")
    visible_oid: str | None = None
    deadline = time.monotonic() + max(5, min(GIT_VISIBILITY_TIMEOUT_SECONDS, 120))
    delay = 0.5
    while time.monotonic() < deadline:
        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(delay)
        delay = min(delay * 1.7, 5.0)
    if visible_oid != commit_oid:
        raise WorkerError("RECONCILIATION_REQUIRED", phase="verification", details={"commit_oid": commit_oid, "visible_oid": visible_oid, "job_id": envelope.job_id})
    job_status, job_response = bridge.call("/v2/job", {"job_id": envelope.job_id})
    if not 200 <= job_status < 300:
        raise WorkerError("RECONCILIATION_REQUIRED", phase="verification", details={"api_status": job_status, "job_id": envelope.job_id})
    _verify_converged_job(job_response, envelope, commit_oid)
    return {
        "governed_task_id": envelope.task_id,
        "job_id": envelope.job_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"),
        "receipt": dict(commit_response.get("receipt")) if isinstance(commit_response.get("receipt"), Mapping) else None,
        "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,
    }


def build_compact_interface_map_result(
    task_id: str,
    envelope: ExecutionEnvelope,
    execution: Mapping[str, Any],
) -> dict[str, Any]:
    interface_map = execution.get("interface_map")
    if not isinstance(interface_map, Mapping):
        raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
    if interface_map.get("schema") != "aios.engineering.interface-map.v1":
        raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
    if interface_map.get("source_commit") != envelope.base_oid:
        raise WorkerError("COMPACT_INTERFACE_MAP_SOURCE_BINDING_INVALID", phase="result")
    implementation_map = interface_map.get("implementation_map")
    excerpts = interface_map.get("bounded_excerpts")
    contracts = interface_map.get("contracts")
    if not isinstance(implementation_map, Mapping) or not isinstance(excerpts, list) or not isinstance(contracts, Mapping):
        raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
    profile = str(interface_map.get("profile") or "fleet")
    classification = str(interface_map.get("classification") or "")
    allowed_classifications = {
        "fleet": "P2_INTERFACE_MAP_CAPTURED",
        "model_governance": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
    }
    if allowed_classifications.get(profile) != classification:
        raise WorkerError("COMPACT_INTERFACE_MAP_PROFILE_BINDING_INVALID", phase="result")
    result = {
        "status": "SUCCEEDED",
        "classification": classification,
        "interface_map_profile": profile,
        "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
        "execution_mode": "compact_interface_map",
        "execution_backend": "generic-online-agent-worker-v2",
        "task_id": task_id,
        "job_id": envelope.job_id,
        "repository": envelope.repository,
        "base_oid": envelope.base_oid,
        "expected_branch_oid": envelope.expected_branch_oid,
        "target_branch": envelope.target_branch,
        "generation": envelope.generation,
        "execution_digest": envelope.digest,
        "source_commit": interface_map["source_commit"],
        "implementation_map": dict(implementation_map),
        "bounded_excerpts": list(excerpts),
        "contracts": dict(contracts),
        "interface_map_result_bytes": interface_map.get("result_bytes"),
        "changed_files": [],
        "tests": [],
        "workspace_cleanup": "PASS",
        "writer_lease": "NONE",
        "secret_material_exposed": False,
        "direct_main_write": False,
        "force_push": False,
        "production_write": False,
        "protected_ref_bypass": False,
    }
    encoded = json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    if len(encoded) > 120 * 1024:
        raise WorkerError(
            "COMPACT_INTERFACE_MAP_RESULT_TOO_LARGE",
            phase="result",
            details={"result_bytes": len(encoded)},
        )
    result["result_bytes"] = len(encoded)
    return result


def build_agent_runtime_result(
    task_id: str,
    envelope: ExecutionEnvelope,
    execution: Mapping[str, Any],
) -> dict[str, Any]:
    external = execution.get("external_agent")
    if not isinstance(external, Mapping):
        raise WorkerError("EXTERNAL_AGENT_RESULT_INVALID", phase="result")
    if external.get("runtime") != envelope.adapter:
        raise WorkerError("EXTERNAL_AGENT_RESULT_RUNTIME_MISMATCH", phase="result")
    if external.get("task_id") != task_id:
        raise WorkerError("EXTERNAL_AGENT_RESULT_TASK_MISMATCH", phase="result")
    if external.get("execution_digest") != envelope.digest:
        raise WorkerError("EXTERNAL_AGENT_RESULT_DIGEST_MISMATCH", phase="result")
    if external.get("credential_material_exposed") is not False:
        raise WorkerError("EXTERNAL_AGENT_CREDENTIAL_ISOLATION_FAILED", phase="result")
    result = {
        "status": "SUCCEEDED",
        "classification": "EXTERNAL_AGENT_RUNTIME_SUCCEEDED",
        "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
        "execution_mode": "agent_runtime",
        "execution_backend": "generic-online-agent-worker-v2",
        "agent_runtime": envelope.adapter,
        "task_id": task_id,
        "job_id": envelope.job_id,
        "repository": envelope.repository,
        "base_oid": envelope.base_oid,
        "expected_branch_oid": envelope.expected_branch_oid,
        "target_branch": envelope.target_branch,
        "generation": envelope.generation,
        "execution_digest": envelope.digest,
        "output_text": external.get("output_text"),
        "output_json": external.get("output_json"),
        "output_sha256": external.get("output_sha256"),
        "runtime_duration_ms": external.get("duration_ms"),
        "external_request_id": external.get("request_id"),
        "changed_files": [],
        "tests": [],
        "workspace_cleanup": "PASS",
        "writer_lease": "NONE",
        "secret_material_exposed": False,
        "credential_material_exposed": False,
        "control_plane_access_granted_to_runtime": False,
        "direct_main_write": False,
        "force_push": False,
        "production_write": False,
        "protected_ref_bypass": False,
    }
    encoded = json.dumps(
        sanitize(result),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    if len(encoded) > 192 * 1024:
        raise WorkerError(
            "EXTERNAL_AGENT_TASK_RESULT_TOO_LARGE",
            phase="result",
            details={"result_bytes": len(encoded)},
        )
    result["result_bytes"] = len(encoded)
    return result

def build_source_capture_result(
    task_id: str,
    envelope: ExecutionEnvelope,
    execution: Mapping[str, Any],
) -> dict[str, Any]:
    if envelope.adapter == "compact_interface_map":
        return build_compact_interface_map_result(task_id, envelope, execution)
    snapshot = execution.get("source_snapshot")
    if not isinstance(snapshot, Mapping):
        raise WorkerError("SOURCE_CAPTURE_RESULT_INVALID", phase="result")
    return {
        "status": "SUCCEEDED",
        "classification": "NEEDS_REWORK_SOURCE_CAPTURE",
        "verification_state": "NEEDS_REWORK",
        "execution_mode": "source_capture",
        "execution_backend": "generic-online-agent-worker-v2",
        "task_id": task_id,
        "job_id": envelope.job_id,
        "repository": envelope.repository,
        "base_oid": envelope.base_oid,
        "expected_branch_oid": envelope.expected_branch_oid,
        "target_branch": envelope.target_branch,
        "generation": envelope.generation,
        "execution_digest": envelope.digest,
        "changed_files": [],
        "tests": [],
        "source_snapshot": dict(snapshot),
        "workspace_cleanup": "PASS",
        "writer_lease": "NONE",
        "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(token)

    def resolve_envelope(self, task: Mapping[str, Any]) -> tuple[ExecutionEnvelope, str]:
        metadata = task.get("metadata")
        execution = metadata.get("execution") if isinstance(metadata, Mapping) else None
        if isinstance(metadata, Mapping) and "execution" in metadata:
            if not isinstance(execution, Mapping):
                raise WorkerError("EXECUTION_ENVELOPE_SCHEMA_VALIDATION", phase="validation")
            persisted = ExecutionEnvelope.from_task(task, worker_capabilities=WORKER_CAPABILITIES)
            # Read-only source_capture/agent_runtime tasks replay their persisted
            # payload. A write-capable source_capture is only a Bridge-created
            # engineering placeholder: never pass that stale payload back to the
            # resolver, and never claim it until a deterministic plan exists.
            pending_plan = source_capture_requires_deterministic_plan(persisted)
            projection_input = (
                execution
                if persisted.action in {"source_capture", "agent_runtime"} and not pending_plan
                else None
            )
            status, authoritative = self.bridge.project_execution(task, projection_input)
            if pending_plan and (
                status == 404
                or authoritative.get("error") == "TASK_EXECUTION_PROJECTION_MISSING"
            ):
                raise WorkerError(
                    "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
                    phase="projection",
                )
            if not 200 <= status < 300:
                raise WorkerError("TASK_PERSISTED_BINDING_UNVERIFIED", phase="projection", details={"api_status": status, "error": authoritative.get("error")})
            canonical = ExecutionEnvelope.from_task(task, authoritative, WORKER_CAPABILITIES)
            binding_fields = (
                "task_id",
                "job_id",
                "repository",
                "base_oid",
                "expected_branch_oid",
                "target_branch",
                "branch_mode",
                "generation",
            )
            if any(getattr(canonical, field) != getattr(persisted, field) for field in binding_fields):
                raise WorkerError("TASK_PERSISTED_BINDING_CONFLICT", phase="projection")
            if canonical.digest != persisted.digest:
                if canonical.action != "deterministic":
                    raise WorkerError("TASK_PERSISTED_BINDING_CONFLICT", phase="projection")
                audit(
                    "TASK_PERSISTED_ENVELOPE_REPLACED_BY_PLAN",
                    task_id=canonical.task_id,
                    job_id=canonical.job_id,
                    previous_action=persisted.action,
                    action=canonical.action,
                    previous_execution_digest=persisted.digest,
                    execution_digest=canonical.digest,
                )
                return canonical, "plan_registry_verified"
            if pending_plan:
                raise WorkerError(
                    "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
                    phase="projection",
                )
            return persisted, "persisted_verified"
        status, projected = self.bridge.project_execution(task)
        if status == 404 or projected.get("error") == "TASK_EXECUTION_PROJECTION_MISSING":
            raise WorkerError("TASK_SKIPPED_PROJECTION_MISSING", phase="projection")
        if not 200 <= status < 300:
            raise WorkerError(
                "TASK_SKIPPED_POLICY_DENY",
                phase="projection",
                details={"api_status": status, "error": projected.get("error")},
            )
        envelope = ExecutionEnvelope.from_task(task, projected, WORKER_CAPABILITIES)
        audit(
            "TASK_EXECUTION_ENVELOPE_PROJECTED",
            task_id=envelope.task_id,
            job_id=envelope.job_id,
            generation=envelope.generation,
            action=envelope.action,
            execution_digest=envelope.digest,
        )
        return envelope, "recovered_projection"

    def process_claim(
        self,
        task: Mapping[str, Any],
        claim: Mapping[str, Any],
        envelope: ExecutionEnvelope,
    ) -> None:
        task_id = str(claim.get("task_id") or task.get("task_id") or "")
        attempt = int(claim.get("attempt") or 0)
        if attempt < 1:
            raise WorkerError("TASK_ATTEMPT_INVALID", phase="claim")
        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=attempt, revision=claim.get("revision"))
        heartbeat = LeaseHeartbeat(self.agent, task_id, lease_token)
        state: dict[str, Any] = {
            "task_id": task_id,
            "execution_digest": envelope.digest,
            "attempt": attempt,
            "phase": "claimed",
            "job_id": envelope.job_id,
        }
        try:
            heartbeat.first()
            heartbeat.set_stage("started")
            heartbeat.start()
            audit("TASK_STARTED", task_id=task_id, attempt=attempt)
            status, current_task = self.agent.get(task_id)
            if status != 200:
                raise WorkerError(
                    "TASK_READ_AFTER_CLAIM_FAILED",
                    phase="claim",
                    details={"api_status": status},
                )
            current_envelope, _ = self.resolve_envelope(current_task)
            if current_envelope.digest != envelope.digest or current_envelope.job_id != envelope.job_id:
                audit(
                    "TASK_PAYLOAD_DRIFT_DETAILS",
                    task_id=task_id,
                    expected_digest=envelope.digest,
                    current_digest=current_envelope.digest,
                    expected_job_id=envelope.job_id,
                    current_job_id=current_envelope.job_id,
                    expected_action=envelope.action,
                    current_action=current_envelope.action,
                    expected_branch_mode=envelope.branch_mode,
                    current_branch_mode=current_envelope.branch_mode,
                    expected_generation=envelope.generation,
                    current_generation=current_envelope.generation,
                )
                raise WorkerError("TASK_PAYLOAD_DRIFT", phase="claim")

            cached = load_task_state(
                task_id,
                envelope.digest,
                attempt=attempt,
                continuation_head_oid=(
                    envelope.expected_branch_oid
                    if envelope.branch_mode == "CONTINUE_EXISTING"
                    else None
                ),
                continuation_job_id=envelope.job_id,
                commit_idempotency_key=str(
                    envelope.governance.get("commit_idempotency_key") or ""
                ),
            )
            durable = latest_reusable_checkpoint(envelope, attempt)
            if cached is None and isinstance(durable, Mapping):
                payload = durable.get("payload")
                if isinstance(payload, Mapping):
                    cached = dict(payload)
                    cached["recovered_from_checkpoint"] = True
                    audit(
                        "TASK_CHECKPOINT_RECOVERED",
                        task_id=task_id,
                        checkpoint_attempt=durable.get("attempt"),
                        checkpoint_phase=durable.get("phase"),
                        checkpoint_sequence=durable.get("sequence"),
                    )
            if cached is not None:
                state = dict(cached)
                state["task_id"] = task_id
                state["execution_digest"] = envelope.digest
                state["job_id"] = envelope.job_id
                previous_attempt = int(state.get("attempt") or 0)
                state["resume_from_attempt"] = previous_attempt or None
                state["attempt"] = attempt

            current_attempt_checkpoint = checkpoint_store().latest_for_attempt(task_id, attempt)
            if current_attempt_checkpoint is None:
                save_checkpoint(
                    task_id,
                    {
                        "task_id": task_id,
                        "execution_digest": envelope.digest,
                        "job_id": envelope.job_id,
                        "attempt": attempt,
                        "phase": "claimed",
                        "resume_from_phase": state.get("phase"),
                    },
                    envelope,
                    attempt,
                )

            governed = (
                state.get("governed")
                if isinstance(state.get("governed"), Mapping)
                else None
            )
            phase = str(state.get("phase") or "claimed")
            result: Mapping[str, Any] | None = None

            # Read-only source capture is safe to reuse across official attempts when the exact
            # execution/scope digest is unchanged.  This is the key restart/retry fast path.
            if envelope.action == "source_capture" and phase in {
                "source_capture_done",
                "result_pending",
                "result_submitted",
            }:
                candidate = state.get("result")
                if isinstance(candidate, Mapping):
                    result = dict(candidate)
                    audit(
                        "TASK_RESUMED_FROM_SOURCE_CAPTURE",
                        task_id=task_id,
                        previous_attempt=state.get("resume_from_attempt"),
                        attempt=attempt,
                        phase=phase,
                    )
            if (
                result is None
                and envelope.action == "source_capture"
                and phase == "verification_passed"
                and isinstance(state.get("execution"), Mapping)
            ):
                result = build_source_capture_result(
                    task_id, envelope, state["execution"]
                )
                state.update({"phase": "source_capture_done", "result": result})
                save_task_state(task_id, state)
                save_checkpoint(task_id, state, envelope, attempt)
                audit(
                    "TASK_RESUMED_AFTER_SOURCE_CAPTURE_VERIFICATION",
                    task_id=task_id,
                    attempt=attempt,
                )

            if envelope.action == "agent_runtime" and phase in {
                "agent_runtime_done",
                "result_pending",
                "result_submitted",
            }:
                candidate = state.get("result")
                if isinstance(candidate, Mapping):
                    result = dict(candidate)
                    audit(
                        "TASK_RESUMED_FROM_EXTERNAL_AGENT_RESULT",
                        task_id=task_id,
                        previous_attempt=state.get("resume_from_attempt"),
                        attempt=attempt,
                        phase=phase,
                        adapter=envelope.adapter,
                    )
            if (
                result is None
                and envelope.action == "agent_runtime"
                and phase == "verification_passed"
                and isinstance(state.get("execution"), Mapping)
            ):
                result = build_agent_runtime_result(
                    task_id, envelope, state["execution"]
                )
                state.update(
                    {"phase": "agent_runtime_done", "result": result, "attempt": attempt}
                )
                save_task_state(task_id, state)
                save_checkpoint(task_id, state, envelope, attempt)
                audit(
                    "TASK_RESUMED_AFTER_EXTERNAL_AGENT_EXECUTION",
                    task_id=task_id,
                    attempt=attempt,
                    adapter=envelope.adapter,
                )

            if (
                result is None
                and envelope.action == "deterministic"
                and isinstance(governed, Mapping)
                and isinstance(governed.get("commit_oid"), str)
                and OID_RE.fullmatch(governed["commit_oid"])
            ):
                if (
                    governed.get("job_id") != envelope.job_id
                    or governed.get("target_branch") != envelope.target_branch
                    or int(governed.get("generation") or 0) != envelope.generation
                    or not isinstance(governed.get("receipt"), Mapping)
                ):
                    raise WorkerError(
                        "RECOVERY_EVIDENCE_BINDING_INVALID", phase="recovery"
                    )
                job_status, job_receipt = self.bridge.call(
                    "/v2/job", {"job_id": envelope.job_id}
                )
                if not 200 <= job_status < 300:
                    raise WorkerError(
                        "RECOVERY_RECEIPT_UNVERIFIED",
                        phase="recovery",
                        details={"api_status": job_status},
                    )
                _verify_converged_job(
                    job_receipt, envelope, str(governed["commit_oid"])
                )
                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
                    or probe.stdout.strip() != governed["commit_oid"]
                ):
                    raise WorkerError("RECOVERY_HEAD_UNVERIFIED", phase="recovery")
                candidate = state.get("result")
                if isinstance(candidate, Mapping):
                    result = dict(candidate)
                else:
                    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,
                )

            if result is None:
                # Any pre-commit crash is replay-safe: rebuild a clean detached worktree and
                # re-run deterministic verification.  No external side effect exists yet.
                run_root = new_run_root()
                try:
                    if envelope.action == "agent_runtime":
                        worktree = run_root / "agent-runtime-workspace"
                        worktree.mkdir(mode=0o750, parents=True, exist_ok=True)
                    else:
                        worktree = prepare_worktree(run_root, envelope)
                    state.update(
                        {
                            "phase": "workspace_prepared",
                            "execution_digest": envelope.digest,
                            "attempt": attempt,
                            "job_id": envelope.job_id,
                        }
                    )
                    save_task_state(task_id, state)
                    save_checkpoint(task_id, state, envelope, attempt)
                    adapter = adapter_for(envelope.adapter)
                    execution = adapter.execute(
                        envelope,
                        worktree,
                        attempt=attempt,
                    )
                    state.update(
                        {
                            "phase": "verification_passed",
                            "execution": execution,
                            "attempt": attempt,
                        }
                    )
                    save_task_state(task_id, state)
                    save_checkpoint(task_id, state, envelope, attempt)
                    heartbeat.ensure()
                    if envelope.action == "source_capture":
                        result = build_source_capture_result(
                            task_id, envelope, execution
                        )
                        state.update(
                            {
                                "phase": "source_capture_done",
                                "result": result,
                                "attempt": attempt,
                            }
                        )
                    elif envelope.action == "agent_runtime":
                        result = build_agent_runtime_result(
                            task_id, envelope, execution
                        )
                        state.update(
                            {
                                "phase": "agent_runtime_done",
                                "result": result,
                                "attempt": attempt,
                            }
                        )
                    else:
                        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,
                                "attempt": attempt,
                            }
                        )
                    save_task_state(task_id, state)
                    save_checkpoint(task_id, state, envelope, attempt)
                finally:
                    cleanup_run_root(run_root)
                    audit(
                        "WORKSPACE_CLEANED",
                        task_id=task_id,
                        cleanup="PASS",
                    )

            if not isinstance(result, Mapping):
                raise WorkerError("TASK_RESULT_MISSING", phase="result")
            heartbeat.ensure()
            heartbeat.set_stage("result")
            result = dict(result)
            result_key = result_idempotency_key(
                task_id, attempt, envelope, result
            )
            state.update(
                {
                    "phase": "result_pending",
                    "attempt": attempt,
                    "result": result,
                    "result_idempotency_key": result_key,
                    "result_sha256": digest_json(result),
                }
            )
            save_task_state(task_id, state)
            save_checkpoint(task_id, state, envelope, attempt)

            status, response = submit_result_replay_safe(
                self.agent, 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"),
                        "result_idempotency_key": result_key,
                    },
                )
            state.update(
                {
                    "phase": "result_submitted",
                    "attempt": attempt,
                    "result_idempotency_key": result_key,
                    "result_submit_api_status": status,
                    "result_submit_reconciled": bool(response.get("reconciled")),
                }
            )
            save_task_state(task_id, state)
            save_checkpoint(task_id, state, envelope, attempt)
            audit(
                "TASK_RESULT_SUBMITTED",
                task_id=task_id,
                attempt=attempt,
                api_status=status,
                result_idempotency_key=result_key,
                commit_oid=result.get("governed", {}).get("commit_oid"),
                replay_reconciled=bool(response.get("reconciled")),
            )
        except WorkerError as exc:
            audit(
                "TASK_FAILED",
                task_id=task_id,
                attempt=attempt,
                error=exc.code,
                phase=exc.phase,
                details=exc.details,
            )
            if exc.code == "RECONCILIATION_REQUIRED":
                state = {
                    "task_id": task_id,
                    "execution_digest": envelope.digest,
                    "phase": "reconciliation_required",
                    "attempt": attempt,
                    "job_id": envelope.job_id,
                    "details": exc.details,
                }
                save_task_state(task_id, state)
                save_checkpoint(task_id, state, envelope, attempt)
                audit(
                    "TASK_RECONCILIATION_REQUIRED",
                    task_id=task_id,
                    job_id=envelope.job_id,
                    details=exc.details,
                )
                return
            failure_key = f"{task_id}-attempt-{attempt}-failure"
            failure = {
                "status": "FAILED",
                "verification_state": "UNVERIFIED",
                "task_id": task_id,
                "attempt": attempt,
                "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 = submit_result_replay_safe(
                    self.agent,
                    task_id,
                    lease_token,
                    failure_key,
                    failure,
                )
                audit(
                    "TASK_FAILURE_RESULT_SUBMITTED",
                    task_id=task_id,
                    attempt=attempt,
                    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,
                    attempt=attempt,
                    error=type(submit_exc).__name__,
                )
        finally:
            heartbeat.close()

    def once(self) -> int:
        try:
            tasks = self.agent.poll()
            write_worker_liveness("POLL_OK", visible_count=len(tasks))
            audit("TASK_POLL", visible_count=len(tasks))
        except WorkerError as exc:
            write_worker_liveness(
                "POLL_FAILED",
                error=exc.code,
            )
            audit("TASK_POLL_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
            return 10
        dynamic_tasks: list[tuple[dict[str, Any], ExecutionEnvelope]] = []
        for task in tasks:
            try:
                envelope, source = self.resolve_envelope(task)
            except WorkerError as exc:
                event = "TASK_SKIPPED_ENVELOPE_INVALID"
                if exc.code in {
                    "TASK_SKIPPED_PROJECTION_MISSING",
                    "TASK_SKIPPED_POLICY_DENY",
                    "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
                }:
                    event = exc.code
                elif exc.code == "CAPABILITY_MISMATCH_DENY":
                    event = "TASK_SKIPPED_CAPABILITY_MISMATCH"
                audit(event, task_id=task.get("task_id"), reason=exc.code, phase=exc.phase, details=exc.details)
                continue
            audit("TASK_VISIBLE_TO_WORKER_CLAIMABLE", task_id=task.get("task_id"), envelope_source=source, action=envelope.action)
            if envelope.adapter not in WORKER_ALLOWED_ADAPTERS:
                audit(
                    "TASK_SKIPPED_ADAPTER_MISMATCH",
                    task_id=task.get("task_id"),
                    adapter=envelope.adapter,
                    allowed_adapters=sorted(WORKER_ALLOWED_ADAPTERS),
                )
                continue
            dynamic_tasks.append((task, envelope))
        dynamic_tasks.sort(key=_claim_order)
        for task, envelope 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, envelope)
            return 0
        return 0

    def run_forever(self) -> int:
        ensure_roots()
        recover_stale_workspaces()
        checkpoint_store()
        write_worker_liveness("STARTING")
        audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["isolated-deterministic", "source_capture", "codex", "hermes"], runtime_only_credentials=True, isolated_executor=True, durable_checkpoint=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())