feat(online-agent): persist workers and defer engineering placeholders

active-runtime/app/aios_online_agent_worker.py

@@ -236,6 +236,64 @@ def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o640) -> None
             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")
@@ -890,6 +948,11 @@ class LeaseHeartbeat:
             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()
@@ -909,6 +972,11 @@ class LeaseHeartbeat:
                     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)
@@ -1772,12 +1840,25 @@ class WorkerSupervisor:
             if not isinstance(execution, Mapping):
                 raise WorkerError("EXECUTION_ENVELOPE_SCHEMA_VALIDATION", phase="validation")
             persisted = ExecutionEnvelope.from_task(task, worker_capabilities=WORKER_CAPABILITIES)
-            # Deterministic tasks must re-resolve through the plan registry.  A
-            # read-only source-capture task has no plan to replace its payload: replay
-            # the already validated durable envelope so source signals remain part of
-            # the canonical digest and the worker does not invent a different capture.
-            projection_input = execution if persisted.action in {"source_capture", "agent_runtime"} else None
+            # 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)
@@ -1806,6 +1887,11 @@ class WorkerSupervisor:
                     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":
@@ -2290,8 +2376,13 @@ class WorkerSupervisor:
     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]] = []
@@ -2300,7 +2391,11 @@ class WorkerSupervisor:
                 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"}:
+                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"
@@ -2334,6 +2429,7 @@ class WorkerSupervisor:
         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()

active-runtime/app/aios_worker_availability_guard.py

@@ -0,0 +1,462 @@
+"""Self-healing availability guard for persistent Online Agent workers.
+
+The guard never claims Tasks and never writes Task/Job/Git databases. It only:
+- ensures required worker/executor systemd units are enabled + active;
+- checks worker-owned liveness JSON written by the real poll/heartbeat loop;
+- restarts an active worker only when its liveness is stale;
+- honors an explicit maintenance inhibit file;
+- writes a bounded status JSON under /run for observability.
+
+Idle liveness is the successful poll loop itself. Busy liveness is refreshed by the
+Task lease-heartbeat thread, so long-running Tasks are not mistaken for hung workers.
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import argparse
+import json
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import tempfile
+import time
+from typing import Any, Callable, Mapping, Sequence
+
+SCHEMA = "aios.worker-availability.status.v1"
+DEFAULT_WORKER_UNITS = (
+    "aios-online-agent-worker-20260815.service",
+    "aios-hermes-online-agent-worker.service",
+    "aios-openclaw-online-agent-worker.service",
+)
+DEFAULT_DEPENDENCY_UNITS = (
+    "aios-external-agent-executor.service",
+)
+DEFAULT_RUN_ROOT = Path("/run/aios-worker-availability")
+DEFAULT_STATUS = DEFAULT_RUN_ROOT / "status.json"
+DEFAULT_RECOVERY = DEFAULT_RUN_ROOT / "recovery.json"
+DEFAULT_INHIBIT = DEFAULT_RUN_ROOT / "maintenance.lock"
+DEFAULT_MAX_STALE_SECONDS = 90
+DEFAULT_RESTART_COOLDOWN_SECONDS = 60
+DEFAULT_STARTUP_WAIT_SECONDS = 10
+ACCEPTED_ENABLED_STATES = frozenset({"enabled", "enabled-runtime", "static", "indirect", "generated"})
+
+
+class AvailabilityGuardError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class UnitSnapshot:
+    unit: str
+    role: str
+    load_state: str
+    active_state: str
+    unit_file_state: str
+    enabled_ok: bool
+    liveness_path: str | None = None
+    liveness_state: str | None = None
+    last_seen_epoch: float | None = None
+    liveness_age_seconds: float | None = None
+    fresh: bool | None = None
+    action: str = "NONE"
+    detail: str | None = None
+
+
+class Systemd:
+    def __init__(
+        self,
+        runner: Callable[..., subprocess.CompletedProcess[str]] | None = None,
+    ) -> None:
+        self.runner = runner or subprocess.run
+
+    def _run(self, args: Sequence[str], *, check: bool = False) -> subprocess.CompletedProcess[str]:
+        result = self.runner(
+            ["/usr/bin/systemctl", *args],
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=20,
+        )
+        if check and result.returncode != 0:
+            raise AvailabilityGuardError(
+                "SYSTEMD_COMMAND_FAILED:"
+                + ":".join(args)
+                + ":"
+                + str(result.returncode)
+            )
+        return result
+
+    def property(self, unit: str, name: str) -> str:
+        result = self._run(["show", unit, f"--property={name}", "--value"])
+        if result.returncode != 0:
+            return ""
+        return result.stdout.strip()
+
+    def environment(self, unit: str) -> dict[str, str]:
+        text = self.property(unit, "Environment")
+        result: dict[str, str] = {}
+        for item in shlex.split(text):
+            if "=" not in item:
+                continue
+            key, value = item.split("=", 1)
+            result[key] = value
+        return result
+
+    def enable(self, unit: str) -> None:
+        self._run(["enable", unit], check=True)
+
+    def start(self, unit: str) -> None:
+        self._run(["start", unit], check=True)
+
+    def restart(self, unit: str) -> None:
+        self._run(["restart", unit], check=True)
+
+
+def _atomic_json(path: Path, value: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise AvailabilityGuardError("STATUS_PATH_SYMLINK_REJECTED")
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        os.fchmod(fd, 0o644)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(value, handle, ensure_ascii=False, sort_keys=True, indent=2)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+        os.chmod(path, 0o644)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def _load_json(path: Path) -> dict[str, Any] | None:
+    if path.is_symlink() or not path.is_file():
+        return None
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    return dict(value) if isinstance(value, Mapping) else None
+
+
+def _state_root(environment: Mapping[str, str], unit: str) -> Path:
+    explicit = environment.get("AIOS_WORKER_STATE_ROOT")
+    if explicit:
+        return Path(explicit)
+    fallbacks = {
+        "aios-online-agent-worker-20260815.service": Path(
+            "/var/lib/aios-online-agent-worker-20260815"
+        ),
+        "aios-hermes-online-agent-worker.service": Path(
+            "/var/lib/aios-online-agent-hermes"
+        ),
+        "aios-openclaw-online-agent-worker.service": Path(
+            "/var/lib/aios-online-agent-openclaw"
+        ),
+    }
+    if unit not in fallbacks:
+        raise AvailabilityGuardError("WORKER_STATE_ROOT_UNAVAILABLE:" + unit)
+    return fallbacks[unit]
+
+
+def _liveness_path(environment: Mapping[str, str], unit: str) -> Path:
+    explicit = environment.get("AIOS_WORKER_LIVENESS_PATH")
+    if explicit:
+        return Path(explicit)
+    return _state_root(environment, unit) / "worker-liveness.json"
+
+
+class AvailabilityGuard:
+    def __init__(
+        self,
+        *,
+        systemd: Systemd | None = None,
+        worker_units: Sequence[str] = DEFAULT_WORKER_UNITS,
+        dependency_units: Sequence[str] = DEFAULT_DEPENDENCY_UNITS,
+        status_path: Path = DEFAULT_STATUS,
+        recovery_path: Path = DEFAULT_RECOVERY,
+        inhibit_path: Path = DEFAULT_INHIBIT,
+        max_stale_seconds: int = DEFAULT_MAX_STALE_SECONDS,
+        restart_cooldown_seconds: int = DEFAULT_RESTART_COOLDOWN_SECONDS,
+        startup_wait_seconds: int = DEFAULT_STARTUP_WAIT_SECONDS,
+        now_fn: Callable[[], float] = time.time,
+        sleep_fn: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self.systemd = systemd or Systemd()
+        self.worker_units = tuple(worker_units)
+        self.dependency_units = tuple(dependency_units)
+        self.status_path = Path(status_path)
+        self.recovery_path = Path(recovery_path)
+        self.inhibit_path = Path(inhibit_path)
+        self.max_stale_seconds = int(max_stale_seconds)
+        self.restart_cooldown_seconds = int(restart_cooldown_seconds)
+        self.startup_wait_seconds = int(startup_wait_seconds)
+        self.now_fn = now_fn
+        self.sleep_fn = sleep_fn
+        if self.max_stale_seconds < 15:
+            raise AvailabilityGuardError("MAX_STALE_SECONDS_TOO_SMALL")
+        if self.restart_cooldown_seconds < 15:
+            raise AvailabilityGuardError("RESTART_COOLDOWN_TOO_SMALL")
+        if self.startup_wait_seconds < 0 or self.startup_wait_seconds > 60:
+            raise AvailabilityGuardError("STARTUP_WAIT_INVALID")
+
+    def _recovery_state(self) -> dict[str, float]:
+        value = _load_json(self.recovery_path) or {}
+        raw = value.get("last_restart_epoch")
+        if not isinstance(raw, Mapping):
+            return {}
+        result: dict[str, float] = {}
+        for unit, epoch in raw.items():
+            try:
+                result[str(unit)] = float(epoch)
+            except (TypeError, ValueError):
+                continue
+        return result
+
+    def _save_recovery_state(self, state: Mapping[str, float]) -> None:
+        _atomic_json(
+            self.recovery_path,
+            {
+                "schema": "aios.worker-availability.recovery.v1",
+                "last_restart_epoch": dict(state),
+            },
+        )
+
+    def _basic_snapshot(self, unit: str, role: str) -> tuple[str, str, str, bool]:
+        load = self.systemd.property(unit, "LoadState")
+        active = self.systemd.property(unit, "ActiveState")
+        enabled = self.systemd.property(unit, "UnitFileState")
+        return load, active, enabled, enabled in ACCEPTED_ENABLED_STATES
+
+    def _ensure_enabled_active(self, unit: str, role: str) -> tuple[str, bool]:
+        load, active, enabled, enabled_ok = self._basic_snapshot(unit, role)
+        if load != "loaded":
+            raise AvailabilityGuardError("REQUIRED_UNIT_NOT_LOADED:" + unit)
+        action = "NONE"
+        if not enabled_ok:
+            self.systemd.enable(unit)
+            action = "ENABLE"
+        if active != "active":
+            self.systemd.start(unit)
+            action = "START" if action == "NONE" else "ENABLE_START"
+        return action, active == "active"
+
+    def _wait_for_liveness(self, path: Path) -> dict[str, Any] | None:
+        deadline = self.now_fn() + self.startup_wait_seconds
+        while True:
+            value = _load_json(path)
+            if value is not None:
+                try:
+                    epoch = float(value.get("last_seen_epoch"))
+                except (TypeError, ValueError):
+                    epoch = 0.0
+                if epoch > 0 and self.now_fn() - epoch <= self.max_stale_seconds:
+                    return value
+            if self.now_fn() >= deadline:
+                return value
+            self.sleep_fn(0.5)
+
+    def run(self, *, heal: bool) -> dict[str, Any]:
+        now = self.now_fn()
+        if self.inhibit_path.exists():
+            result = {
+                "schema": SCHEMA,
+                "status": "MAINTENANCE_INHIBITED",
+                "generated_at_epoch": now,
+                "maintenance_inhibit": str(self.inhibit_path),
+                "workers": [],
+                "dependencies": [],
+                "all_fresh": False,
+                "actions": [],
+            }
+            _atomic_json(self.status_path, result)
+            return result
+
+        actions: list[dict[str, str]] = []
+        dependencies: list[UnitSnapshot] = []
+        for unit in self.dependency_units:
+            load, active, enabled, enabled_ok = self._basic_snapshot(unit, "dependency")
+            action = "NONE"
+            if heal:
+                if load != "loaded":
+                    raise AvailabilityGuardError("REQUIRED_UNIT_NOT_LOADED:" + unit)
+                if not enabled_ok:
+                    self.systemd.enable(unit)
+                    action = "ENABLE"
+                if active != "active":
+                    self.systemd.start(unit)
+                    action = "START" if action == "NONE" else "ENABLE_START"
+                if action != "NONE":
+                    actions.append({"unit": unit, "action": action})
+                load, active, enabled, enabled_ok = self._basic_snapshot(unit, "dependency")
+            dependencies.append(
+                UnitSnapshot(
+                    unit=unit,
+                    role="dependency",
+                    load_state=load,
+                    active_state=active,
+                    unit_file_state=enabled,
+                    enabled_ok=enabled_ok,
+                    fresh=(active == "active"),
+                    action=action,
+                )
+            )
+
+        recovery = self._recovery_state()
+        workers: list[UnitSnapshot] = []
+        for unit in self.worker_units:
+            load, active, enabled, enabled_ok = self._basic_snapshot(unit, "worker")
+            environment = self.systemd.environment(unit) if load == "loaded" else {}
+            path = _liveness_path(environment, unit) if load == "loaded" else None
+            action = "NONE"
+
+            if heal:
+                if load != "loaded":
+                    raise AvailabilityGuardError("REQUIRED_UNIT_NOT_LOADED:" + unit)
+                if not enabled_ok:
+                    self.systemd.enable(unit)
+                    action = "ENABLE"
+                if active != "active":
+                    self.systemd.start(unit)
+                    action = "START" if action == "NONE" else "ENABLE_START"
+                    actions.append({"unit": unit, "action": action})
+                    if path is not None:
+                        self._wait_for_liveness(path)
+                load, active, enabled, enabled_ok = self._basic_snapshot(unit, "worker")
+
+            live = _load_json(path) if path is not None else None
+            last_seen: float | None = None
+            state: str | None = None
+            age: float | None = None
+            fresh = False
+            detail: str | None = None
+            if live is not None:
+                state = str(live.get("state") or "")
+                try:
+                    last_seen = float(live.get("last_seen_epoch"))
+                except (TypeError, ValueError):
+                    last_seen = None
+                if last_seen is not None:
+                    age = max(0.0, now - last_seen)
+                    fresh = age <= self.max_stale_seconds
+                if live.get("agent_id") in (None, ""):
+                    fresh = False
+                    detail = "LIVENESS_AGENT_ID_MISSING"
+            else:
+                detail = "LIVENESS_MISSING"
+
+            if heal and active == "active" and not fresh:
+                previous = float(recovery.get(unit, 0.0))
+                if now - previous >= self.restart_cooldown_seconds:
+                    self.systemd.restart(unit)
+                    recovery[unit] = now
+                    action = "RESTART_STALE"
+                    actions.append({"unit": unit, "action": action})
+                    if path is not None:
+                        live = self._wait_for_liveness(path)
+                    load, active, enabled, enabled_ok = self._basic_snapshot(unit, "worker")
+                    if live is not None:
+                        state = str(live.get("state") or "")
+                        try:
+                            last_seen = float(live.get("last_seen_epoch"))
+                        except (TypeError, ValueError):
+                            last_seen = None
+                        if last_seen is not None:
+                            age = max(0.0, self.now_fn() - last_seen)
+                            fresh = age <= self.max_stale_seconds
+                            detail = None if fresh else "LIVENESS_STALE_AFTER_RESTART"
+                else:
+                    detail = "RESTART_COOLDOWN_ACTIVE"
+
+            workers.append(
+                UnitSnapshot(
+                    unit=unit,
+                    role="worker",
+                    load_state=load,
+                    active_state=active,
+                    unit_file_state=enabled,
+                    enabled_ok=enabled_ok,
+                    liveness_path=str(path) if path is not None else None,
+                    liveness_state=state,
+                    last_seen_epoch=last_seen,
+                    liveness_age_seconds=age,
+                    fresh=bool(active == "active" and enabled_ok and fresh),
+                    action=action,
+                    detail=detail,
+                )
+            )
+
+        if heal:
+            self._save_recovery_state(recovery)
+
+        all_fresh = (
+            all(item.active_state == "active" and item.enabled_ok and item.fresh for item in workers)
+            and all(item.active_state == "active" for item in dependencies)
+        )
+        result = {
+            "schema": SCHEMA,
+            "status": "HEALTHY" if all_fresh else "DEGRADED",
+            "generated_at_epoch": self.now_fn(),
+            "max_stale_seconds": self.max_stale_seconds,
+            "maintenance_inhibit": str(self.inhibit_path),
+            "workers": [asdict(item) for item in workers],
+            "dependencies": [asdict(item) for item in dependencies],
+            "all_fresh": all_fresh,
+            "actions": actions,
+        }
+        _atomic_json(self.status_path, result)
+        return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    mode = parser.add_mutually_exclusive_group(required=True)
+    mode.add_argument("--heal", action="store_true")
+    mode.add_argument("--check", action="store_true")
+    parser.add_argument("--status-file", default=str(DEFAULT_STATUS))
+    parser.add_argument("--recovery-file", default=str(DEFAULT_RECOVERY))
+    parser.add_argument("--inhibit-file", default=str(DEFAULT_INHIBIT))
+    parser.add_argument(
+        "--max-stale-seconds",
+        type=int,
+        default=int(
+            os.environ.get(
+                "AIOS_WORKER_AVAILABILITY_MAX_STALE_SECONDS",
+                str(DEFAULT_MAX_STALE_SECONDS),
+            )
+        ),
+    )
+    parser.add_argument(
+        "--restart-cooldown-seconds",
+        type=int,
+        default=int(
+            os.environ.get(
+                "AIOS_WORKER_AVAILABILITY_RESTART_COOLDOWN_SECONDS",
+                str(DEFAULT_RESTART_COOLDOWN_SECONDS),
+            )
+        ),
+    )
+    args = parser.parse_args()
+    guard = AvailabilityGuard(
+        status_path=Path(args.status_file),
+        recovery_path=Path(args.recovery_file),
+        inhibit_path=Path(args.inhibit_file),
+        max_stale_seconds=args.max_stale_seconds,
+        restart_cooldown_seconds=args.restart_cooldown_seconds,
+    )
+    try:
+        result = guard.run(heal=args.heal)
+    except AvailabilityGuardError as exc:
+        print(json.dumps({"status": "FAILED", "error": str(exc)}, sort_keys=True))
+        return 20
+    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
+    if args.check and not result.get("all_fresh"):
+        return 10
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

deploy/systemd/aios-worker-availability-guard.service

@@ -0,0 +1,27 @@
+[Unit]
+Description=AIOS persistent Online Agent worker availability guard
+After=local-fs.target
+
+[Service]
+Type=oneshot
+User=root
+Group=root
+Environment=PYTHONUNBUFFERED=1
+Environment=AIOS_WORKER_AVAILABILITY_MAX_STALE_SECONDS=90
+Environment=AIOS_WORKER_AVAILABILITY_RESTART_COOLDOWN_SECONDS=60
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_worker_availability_guard.py --heal
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX
+ReadWritePaths=/run/aios-worker-availability
+UMask=0022

deploy/systemd/aios-worker-availability-guard.timer

@@ -0,0 +1,13 @@
+[Unit]
+Description=Periodically ensure AIOS Online Agent workers remain available
+
+[Timer]
+OnBootSec=15s
+OnUnitActiveSec=30s
+AccuracySec=5s
+RandomizedDelaySec=3s
+Persistent=true
+Unit=aios-worker-availability-guard.service
+
+[Install]
+WantedBy=timers.target

deploy/tmpfiles.d/aios-worker-availability.conf

@@ -0,0 +1 @@
+d /run/aios-worker-availability 0755 root root -

docs/superpowers/specs/2026-08-18-worker-availability-runtime.md

@@ -0,0 +1,212 @@
+# AIOS Worker Availability Runtime
+
+## Priority
+
+This runtime exists to make the already verified Online Agent execution path continuously usable
+without repeated human intervention.
+
+The system already proved that generic, Hermes and OpenClaw workers can claim and complete real
+Online Tasks. The remaining availability defect was operational: evidence-freeze workflows left the
+worker services stopped, while registry `ACTIVE` records remained stale.
+
+The availability contract therefore separates:
+
+```text
+AUTO_CLAIM_MECHANISM=VERIFIED
+CURRENT_WORKER_AVAILABILITY=<fresh runtime state>
+```
+
+A historical successful claim is not current liveness.
+
+## Source of truth for idle liveness
+
+An idle worker may have no Task lease heartbeat. Therefore Task heartbeat alone cannot prove that
+the worker is continuously polling.
+
+`aios_online_agent_worker.py` now writes:
+
+```text
+<AIOS_WORKER_STATE_ROOT>/worker-liveness.json
+```
+
+on:
+
+- process start: `STARTING`;
+- every successful poll: `POLL_OK`;
+- poll failure: `POLL_FAILED`;
+- every successful Task lease heartbeat while executing: `BUSY`.
+
+The file contains:
+
+- agent ID/type/version;
+- process PID;
+- monotonically increasing local sequence;
+- `last_seen_epoch`;
+- poll/heartbeat intervals;
+- current Task/stage while busy.
+
+The liveness file is worker-owned observability state only. It does not write or mutate Task, Job,
+Git, WriterLease or registry databases.
+
+## Self-healing guard
+
+`aios-worker-availability-guard.timer` invokes a root-owned oneshot guard every 30 seconds.
+
+The guard manages:
+
+```text
+aios-online-agent-worker-20260815.service
+aios-hermes-online-agent-worker.service
+aios-openclaw-online-agent-worker.service
+```
+
+and the required external runtime dependency:
+
+```text
+aios-external-agent-executor.service
+```
+
+For each worker:
+
+1. missing systemd unit → fail closed;
+2. disabled → `systemctl enable`;
+3. inactive → `systemctl start`;
+4. active but liveness older than 90 seconds → restart;
+5. repeated stale restart within 60 seconds → do not restart again; report degraded.
+
+Long-running Tasks are safe because the Task lease-heartbeat thread refreshes `BUSY` liveness.
+
+The guard never claims a Task and never touches Task/Job/Git databases.
+
+## Maintenance inhibit
+
+Planned maintenance creates:
+
+```text
+/run/aios-worker-availability/maintenance.lock
+```
+
+While this file exists the guard performs no start/restart actions and reports:
+
+```text
+MAINTENANCE_INHIBITED
+```
+
+Maintenance tooling must remove the file when done. This prevents the self-healer from fighting
+intentional service changes.
+
+## Status
+
+The guard writes:
+
+```text
+/run/aios-worker-availability/status.json
+```
+
+with current worker/dependency states, liveness ages and recovery actions.
+
+A worker is currently available only when:
+
+```text
+systemd load_state=loaded
+systemd active_state=active
+unit enabled/static
+worker-liveness.json present
+last_seen_age <= 90s
+```
+
+Registry `ACTIVE` without fresh poll/busy liveness is not sufficient.
+
+## Boot/restart behavior
+
+The timer is:
+
+```text
+OnBootSec=15s
+OnUnitActiveSec=30s
+Persistent=true
+```
+
+Therefore reboot, manual stop left behind by another workflow, process crash, and hung poll loop are
+all repaired without requiring a human to notice and intervene.
+
+Normal worker units retain their own `Restart=always`; the availability guard handles the cases that
+`Restart=always` cannot, especially a deliberate `systemctl stop` left after an evidence workflow.
+
+## Acceptance
+
+The live acceptance requires all of the following:
+
+```text
+guard timer active/enabled
+external executor active/enabled
+generic worker active/enabled + fresh liveness
+Hermes worker active/enabled + fresh liveness
+OpenClaw worker active/enabled + fresh liveness
+```
+
+Then a new read-only Online Task is projected through canonical Bridge `18892`.
+
+Success requires:
+
+```text
+TASK_CREATED
+→ generic worker auto claim
+→ TASK_HEARTBEAT
+→ RESULT_SUBMITTED
+→ official TASK_VERIFIED(SUCCEEDED)
+```
+
+After the Task, the services must remain active and liveness must continue advancing. The acceptance
+also stops a worker intentionally once and proves the guard starts it again with a new fresh
+liveness sequence.
+
+Only after these gates may current status be upgraded from:
+
+```text
+OVERALL=PARTIAL
+```
+
+to:
+
+```text
+PERSISTENT_WORKER_LIVENESS=VERIFIED
+NEW_TASK_AUTO_CONSUMPTION=VERIFIED
+WORKER_AVAILABILITY_RUNTIME=VERIFIED_CANDIDATE
+```
+
+Protected main and Production remain out of scope.
+
+
+## Forced source-capture projection compatibility
+
+The deployed canonical Bridge currently projects a default `source_capture` envelope even when a
+governed binding request asks for binding-only behavior. This must not be allowed to make always-on
+workers consume engineering Tasks before their deterministic plan is registered.
+
+The worker therefore distinguishes a real read-only source capture from an engineering placeholder:
+
+```text
+source_capture
++ capability_scope contains governed.write.branch or governed.write.file
+= PENDING_DETERMINISTIC_PLAN
+```
+
+For such a placeholder the worker:
+
+1. re-resolves through the canonical plan registry **without replaying the stale source-capture
+   envelope**;
+2. if no deterministic plan exists, emits `TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN` and does not
+   claim;
+3. if the registered canonical envelope is deterministic and the governed binding still matches,
+   uses that deterministic envelope and claims normally.
+
+A real read-only source-capture Task with only `git.read`/`git.test` still replays its exact persisted
+payload and remains claimable.
+
+This makes persistent workers safe with the currently deployed Bridge while preserving the same
+Task/Job binding. No Task DB rewrite, cancellation, replacement Task, fake NEEDS_REWORK, or
+pre-claim metadata mutation is required.
+
+A later Bridge cleanup may implement true binding-only projection, but Worker Availability does not
+depend on that cleanup.

tests/test_online_agent_placeholder_plan_deferral.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+import aios_online_agent_worker as worker
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+
+
+
+@pytest.fixture(autouse=True)
+def _state_paths(monkeypatch, tmp_path):
+    root=tmp_path/"worker-state"
+    monkeypatch.setattr(worker,"STATE_ROOT",root)
+    monkeypatch.setattr(worker,"WORKER_LIVENESS_PATH",root/"worker-liveness.json")
+
+BASE="a"*40
+TASK_ID="task-worker-availability-placeholder"
+JOB_ID="job-worker-availability-placeholder"
+BRANCH="agent/task-worker-availability-placeholder"
+IDEMPOTENCY="task-worker-availability-placeholder-binding-v1"
+
+
+def binding():
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "branch_mode":"CREATE_NEW",
+        "generation":1,
+        "idempotency_key":IDEMPOTENCY,
+    })
+
+
+def placeholder():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action":"source_capture",
+            "adapter":"source_capture",
+            "capability_scope":[
+                "git.read",
+                "git.test",
+                "governed.write.branch",
+                "governed.write.file",
+            ],
+        },
+    )
+
+
+def readonly_source_capture():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action":"source_capture",
+            "adapter":"source_capture",
+            "capability_scope":["git.read","git.test"],
+            "source_signals":["worker availability"],
+        },
+    )
+
+
+def deterministic():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action":"deterministic",
+            "adapter":"deterministic",
+            "capability_scope":[
+                "git.read",
+                "git.test",
+                "governed.write.branch",
+                "governed.write.file",
+            ],
+            "change_set":[{
+                "path":"active-runtime/app/example.py",
+                "operation":"write_file",
+                "content":"VALUE=1\\n",
+            }],
+            "verification":[{
+                "label":"unit",
+                "argv":["/usr/bin/python3","-m","pytest","-q"],
+                "cwd":".",
+                "timeout_seconds":120,
+            }],
+            "governance":{
+                "commit_idempotency_key":"task-worker-availability-placeholder-commit-v1",
+                "allowed_files":["active-runtime/app/example.py"],
+            },
+        },
+    )
+
+
+def task(execution):
+    return {
+        "task_id":TASK_ID,
+        "required_capabilities":list(execution["capability_scope"]),
+        "metadata":{
+            "execution":execution,
+            "execution_projection":{
+                "schema":"aios.online-agent.task-projection.v1",
+                "task_id":TASK_ID,
+                "job_id":JOB_ID,
+                "execution_digest":execution["execution_digest"],
+            },
+        },
+    }
+
+
+class FakeBridge:
+    def __init__(self,status,execution):
+        self.status=status
+        self.execution=execution
+        self.calls=[]
+
+    def project_execution(self,task_value,execution=None):
+        self.calls.append(execution)
+        return self.status,self.execution
+
+
+class FakeAgent:
+    def __init__(self,tasks):
+        self.tasks=tasks
+        self.claim_calls=[]
+
+    def poll(self):
+        return list(self.tasks)
+
+    def claim(self,task_id):
+        self.claim_calls.append(task_id)
+        return 500,{"error":"SHOULD_NOT_CLAIM"}
+
+
+def supervisor(bridge, agent=None):
+    value=object.__new__(worker.WorkerSupervisor)
+    value.bridge=bridge
+    value.agent=agent
+    return value
+
+
+def test_write_capable_source_capture_is_classified_as_placeholder():
+    envelope=worker.ExecutionEnvelope.from_task(
+        task(placeholder()),
+        worker_capabilities=tuple(placeholder()["capability_scope"]),
+    )
+    assert worker.source_capture_requires_deterministic_plan(envelope) is True
+
+
+def test_readonly_source_capture_is_not_placeholder():
+    envelope=worker.ExecutionEnvelope.from_task(
+        task(readonly_source_capture()),
+        worker_capabilities=tuple(readonly_source_capture()["capability_scope"]),
+    )
+    assert worker.source_capture_requires_deterministic_plan(envelope) is False
+
+
+def test_placeholder_without_registered_plan_is_skipped_before_claim():
+    persisted=placeholder()
+    bridge=FakeBridge(200,persisted)
+    agent=FakeAgent([task(persisted)])
+    sup=supervisor(bridge,agent)
+    assert sup.once()==0
+    assert bridge.calls==[None]
+    assert agent.claim_calls==[]
+
+
+def test_placeholder_with_missing_projection_is_skipped_before_claim():
+    persisted=placeholder()
+    bridge=FakeBridge(404,{"error":"TASK_EXECUTION_PROJECTION_MISSING"})
+    agent=FakeAgent([task(persisted)])
+    sup=supervisor(bridge,agent)
+    assert sup.once()==0
+    assert bridge.calls==[None]
+    assert agent.claim_calls==[]
+
+
+def test_registered_deterministic_plan_replaces_placeholder():
+    persisted=placeholder()
+    resolved=deterministic()
+    bridge=FakeBridge(200,resolved)
+    sup=supervisor(bridge)
+    envelope,source=sup.resolve_envelope(task(persisted))
+    assert source=="plan_registry_verified"
+    assert envelope.action=="deterministic"
+    assert envelope.digest==resolved["execution_digest"]
+    assert bridge.calls==[None]
+
+
+def test_real_readonly_source_capture_replays_exact_payload():
+    persisted=readonly_source_capture()
+    bridge=FakeBridge(200,persisted)
+    sup=supervisor(bridge)
+    envelope,source=sup.resolve_envelope(task(persisted))
+    assert source=="persisted_verified"
+    assert envelope.action=="source_capture"
+    assert envelope.digest==persisted["execution_digest"]
+    assert bridge.calls==[persisted]

tests/test_worker_availability_guard.py

@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sys
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_worker_availability_guard import AvailabilityGuard
+
+
+class FakeSystemd:
+    def __init__(self, roots, now=1000.0):
+        self.now=now
+        self.actions=[]
+        self.units={}
+        for unit,root in roots.items():
+            self.units[unit]={
+                "LoadState":"loaded",
+                "ActiveState":"active",
+                "UnitFileState":"enabled",
+                "Environment":f"AIOS_WORKER_STATE_ROOT={root}",
+            }
+        self.units["aios-external-agent-executor.service"]={
+            "LoadState":"loaded",
+            "ActiveState":"active",
+            "UnitFileState":"enabled",
+            "Environment":"",
+        }
+
+    def property(self, unit, name):
+        return self.units.get(unit,{}).get(name,"")
+
+    def environment(self, unit):
+        raw=self.property(unit,"Environment")
+        result={}
+        for item in raw.split():
+            if "=" in item:
+                k,v=item.split("=",1)
+                result[k]=v
+        return result
+
+    def enable(self, unit):
+        self.actions.append((unit,"enable"))
+        self.units[unit]["UnitFileState"]="enabled"
+
+    def start(self, unit):
+        self.actions.append((unit,"start"))
+        self.units[unit]["ActiveState"]="active"
+        env=self.environment(unit)
+        if "AIOS_WORKER_STATE_ROOT" not in env:
+            return
+        root=Path(env["AIOS_WORKER_STATE_ROOT"])
+        root.mkdir(parents=True,exist_ok=True)
+        (root/"worker-liveness.json").write_text(json.dumps({
+            "schema":"aios.online-agent.worker-liveness.v1",
+            "agent_id":"agent-"+unit,
+            "state":"POLL_OK",
+            "last_seen_epoch":self.now,
+        }),encoding="utf-8")
+
+    def restart(self, unit):
+        self.actions.append((unit,"restart"))
+        self.units[unit]["ActiveState"]="active"
+        env=self.environment(unit)
+        root=Path(env["AIOS_WORKER_STATE_ROOT"])
+        root.mkdir(parents=True,exist_ok=True)
+        (root/"worker-liveness.json").write_text(json.dumps({
+            "schema":"aios.online-agent.worker-liveness.v1",
+            "agent_id":"agent-"+unit,
+            "state":"POLL_OK",
+            "last_seen_epoch":self.now,
+        }),encoding="utf-8")
+
+
+def make_guard(tmp_path, fake):
+    worker_units=tuple(
+        unit for unit in fake.units
+        if unit!="aios-external-agent-executor.service"
+    )
+    return AvailabilityGuard(
+        systemd=fake,
+        worker_units=worker_units,
+        dependency_units=("aios-external-agent-executor.service",),
+        status_path=tmp_path/"run/status.json",
+        recovery_path=tmp_path/"run/recovery.json",
+        inhibit_path=tmp_path/"run/maintenance.lock",
+        max_stale_seconds=90,
+        restart_cooldown_seconds=60,
+        startup_wait_seconds=0,
+        now_fn=lambda:fake.now,
+        sleep_fn=lambda _seconds:None,
+    )
+
+
+def write_liveness(root, now, *, state="POLL_OK"):
+    root.mkdir(parents=True,exist_ok=True)
+    (root/"worker-liveness.json").write_text(json.dumps({
+        "schema":"aios.online-agent.worker-liveness.v1",
+        "agent_id":"agent",
+        "state":state,
+        "last_seen_epoch":now,
+    }),encoding="utf-8")
+
+
+def test_healthy_workers_need_no_action(tmp_path):
+    roots={
+        "generic.service":tmp_path/"generic",
+        "hermes.service":tmp_path/"hermes",
+        "openclaw.service":tmp_path/"openclaw",
+    }
+    fake=FakeSystemd(roots)
+    for root in roots.values():
+        write_liveness(root,fake.now)
+    result=make_guard(tmp_path,fake).run(heal=True)
+    assert result["status"]=="HEALTHY"
+    assert result["all_fresh"] is True
+    assert fake.actions==[]
+
+
+def test_inactive_worker_is_started_and_becomes_fresh(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    fake.units["generic.service"]["ActiveState"]="inactive"
+    result=make_guard(tmp_path,fake).run(heal=True)
+    assert ("generic.service","start") in fake.actions
+    assert result["all_fresh"] is True
+
+
+def test_disabled_worker_is_enabled_and_started(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    fake.units["generic.service"]["UnitFileState"]="disabled"
+    fake.units["generic.service"]["ActiveState"]="inactive"
+    result=make_guard(tmp_path,fake).run(heal=True)
+    assert ("generic.service","enable") in fake.actions
+    assert ("generic.service","start") in fake.actions
+    assert result["all_fresh"] is True
+
+
+def test_active_but_stale_worker_is_restarted(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    write_liveness(roots["generic.service"],fake.now-500)
+    result=make_guard(tmp_path,fake).run(heal=True)
+    assert ("generic.service","restart") in fake.actions
+    assert result["all_fresh"] is True
+    assert result["workers"][0]["action"]=="RESTART_STALE"
+
+
+def test_restart_cooldown_prevents_restart_storm(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    write_liveness(roots["generic.service"],fake.now-500)
+    guard=make_guard(tmp_path,fake)
+    guard.recovery_path.parent.mkdir(parents=True,exist_ok=True)
+    guard.recovery_path.write_text(json.dumps({
+        "schema":"aios.worker-availability.recovery.v1",
+        "last_restart_epoch":{"generic.service":fake.now-10},
+    }),encoding="utf-8")
+    result=guard.run(heal=True)
+    assert ("generic.service","restart") not in fake.actions
+    assert result["status"]=="DEGRADED"
+    assert result["workers"][0]["detail"]=="RESTART_COOLDOWN_ACTIVE"
+
+
+def test_maintenance_inhibit_disables_healing(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    fake.units["generic.service"]["ActiveState"]="inactive"
+    guard=make_guard(tmp_path,fake)
+    guard.inhibit_path.parent.mkdir(parents=True,exist_ok=True)
+    guard.inhibit_path.write_text("planned maintenance\n")
+    result=guard.run(heal=True)
+    assert result["status"]=="MAINTENANCE_INHIBITED"
+    assert fake.actions==[]
+
+
+def test_dependency_is_enabled_and_started(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    write_liveness(roots["generic.service"],fake.now)
+    fake.units["aios-external-agent-executor.service"]["UnitFileState"]="disabled"
+    fake.units["aios-external-agent-executor.service"]["ActiveState"]="inactive"
+    result=make_guard(tmp_path,fake).run(heal=True)
+    assert ("aios-external-agent-executor.service","enable") in fake.actions
+    assert ("aios-external-agent-executor.service","start") in fake.actions
+    assert result["all_fresh"] is True
+
+
+def test_check_reports_stale_without_mutation(tmp_path):
+    roots={"generic.service":tmp_path/"generic"}
+    fake=FakeSystemd(roots)
+    write_liveness(roots["generic.service"],fake.now-500)
+    result=make_guard(tmp_path,fake).run(heal=False)
+    assert result["status"]=="DEGRADED"
+    assert result["all_fresh"] is False
+    assert fake.actions==[]

tests/test_worker_availability_runtime_config.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT=Path(__file__).resolve().parents[1]
+
+
+def test_worker_source_has_idle_and_busy_liveness():
+    source=(ROOT/"active-runtime/app/aios_online_agent_worker.py").read_text()
+    assert "aios.online-agent.worker-liveness.v1" in source
+    assert 'write_worker_liveness("POLL_OK"' in source
+    assert '"POLL_FAILED"' in source
+    assert 'write_worker_liveness("STARTING")' in source
+    assert '"BUSY"' in source
+    assert "last_seen_epoch" in source
+
+
+def test_guard_timer_is_persistent_and_frequent():
+    timer=(ROOT/"deploy/systemd/aios-worker-availability-guard.timer").read_text()
+    assert "OnBootSec=15s" in timer
+    assert "OnUnitActiveSec=30s" in timer
+    assert "Persistent=true" in timer
+    assert "WantedBy=timers.target" in timer
+
+
+def test_guard_service_has_no_network_or_credentials():
+    unit=(ROOT/"deploy/systemd/aios-worker-availability-guard.service").read_text()
+    assert "User=root" in unit
+    assert "LoadCredential" not in unit
+    assert "RestrictAddressFamilies=AF_UNIX" in unit
+    assert "NoNewPrivileges=true" in unit
+    assert "ProtectSystem=strict" in unit
+    assert "ReadWritePaths=/run/aios-worker-availability" in unit
+    assert "AIOS_WORKER_AVAILABILITY_MAX_STALE_SECONDS=90" in unit
+
+
+def test_guard_manages_all_three_workers_and_external_executor():
+    source=(ROOT/"active-runtime/app/aios_worker_availability_guard.py").read_text()
+    for unit in (
+        "aios-online-agent-worker-20260815.service",
+        "aios-hermes-online-agent-worker.service",
+        "aios-openclaw-online-agent-worker.service",
+        "aios-external-agent-executor.service",
+    ):
+        assert unit in source
+    assert "maintenance.lock" in source
+    assert "RESTART_STALE" in source
+
+
+def test_worker_defers_write_capable_source_capture_until_plan_exists():
+    source=(ROOT/"active-runtime/app/aios_online_agent_worker.py").read_text()
+    assert "ENGINEERING_WRITE_CAPABILITIES" in source
+    assert "source_capture_requires_deterministic_plan" in source
+    assert "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN" in source
+    assert '"governed.write.branch"' in source
+    assert '"governed.write.file"' in source