docs(hermes): define lifecycle contract and validation state

active-runtime/app/aios_hermes/__init__.py

@@ -6,6 +6,7 @@ from .adapter import (
     HermesNotFound,
     HermesRuntimeError,
     InMemoryHermesStore,
+    SQLiteHermesStore,
 )
 from .contract import HermesHealth, HermesJob, HermesJobState, HermesResult
 from .worker import HermesWorker, WorkerRunResult
@@ -21,5 +22,6 @@ __all__ = [
     "HermesRuntimeError",
     "HermesWorker",
     "InMemoryHermesStore",
+    "SQLiteHermesStore",
     "WorkerRunResult",
 ]

active-runtime/app/aios_hermes/adapter.py

@@ -3,9 +3,11 @@ from __future__ import annotations
 
 import hashlib
 import json
+import sqlite3
 import threading
-from dataclasses import replace
+from dataclasses import asdict, replace
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 from typing import Iterable, Mapping
 
 from .contract import (
@@ -41,6 +43,118 @@ class InMemoryHermesStore:
         self.idempotency: dict[tuple[str, str], tuple[str, str]] = {}
         self.worker_heartbeats: dict[str, tuple[str, str | None]] = {}
 
+    def persist_locked(self) -> None:
+        """Persist a mutation while ``lock`` is held; memory stores are a no-op."""
+
+
+class SQLiteHermesStore(InMemoryHermesStore):
+    """Durable snapshot for the existing Hermes queue state machine.
+
+    The adapter remains the sole owner of transitions and lease rules. This
+    store only makes its jobs, results, idempotency keys and worker heartbeats
+    survive a process restart.
+    """
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path).resolve()
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        super().__init__()
+        with self._connect() as connection:
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS hermes_state (
+                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                    payload TEXT NOT NULL
+                )
+                """
+            )
+        self._load()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(str(self.path), timeout=30)
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA synchronous=FULL")
+        return connection
+
+    @staticmethod
+    def _record(value: HermesJob | HermesResult) -> dict[str, object]:
+        record = asdict(value)
+        record["state"] = value.state.value
+        return record
+
+    def _load(self) -> None:
+        with self.lock, self._connect() as connection:
+            row = connection.execute(
+                "SELECT payload FROM hermes_state WHERE singleton = 1"
+            ).fetchone()
+            if row is None:
+                return
+            payload = json.loads(str(row[0]))
+            self.jobs = {
+                str(item["job_id"]): HermesJob(**item)
+                for item in payload.get("jobs", [])
+            }
+            self.results = {
+                str(item["job_id"]): HermesResult(**item)
+                for item in payload.get("results", [])
+            }
+            self.idempotency = {
+                (str(item["tenant_id"]), str(item["idempotency_key"])): (
+                    str(item["job_id"]),
+                    str(item["digest"]),
+                )
+                for item in payload.get("idempotency", [])
+            }
+            self.worker_heartbeats = {
+                str(item["worker_id"]): (
+                    str(item["heartbeat_at"]),
+                    str(item["job_id"]) if item.get("job_id") is not None else None,
+                )
+                for item in payload.get("worker_heartbeats", [])
+            }
+
+    def persist_locked(self) -> None:
+        payload = {
+            "version": 1,
+            "jobs": [self._record(item) for item in self.jobs.values()],
+            "results": [self._record(item) for item in self.results.values()],
+            "idempotency": [
+                {
+                    "tenant_id": tenant_id,
+                    "idempotency_key": idempotency_key,
+                    "job_id": job_id,
+                    "digest": digest,
+                }
+                for (tenant_id, idempotency_key), (job_id, digest)
+                in self.idempotency.items()
+            ],
+            "worker_heartbeats": [
+                {
+                    "worker_id": worker_id,
+                    "heartbeat_at": heartbeat_at,
+                    "job_id": job_id,
+                }
+                for worker_id, (heartbeat_at, job_id)
+                in self.worker_heartbeats.items()
+            ],
+        }
+        encoded = json.dumps(
+            payload,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            connection.execute(
+                """
+                INSERT INTO hermes_state(singleton, payload) VALUES(1, ?)
+                ON CONFLICT(singleton) DO UPDATE SET payload=excluded.payload
+                """,
+                (encoded,),
+            )
+            connection.commit()
+
 
 class HermesAdapter:
     """Queue and reconcile Hermes jobs without owning AIOS policy."""
@@ -118,6 +232,7 @@ class HermesAdapter:
                 return self.store.jobs[prior_id]
             self.store.jobs[job.job_id] = job
             self.store.idempotency[idem_ref] = (job.job_id, digest)
+            self.store.persist_locked()
             return job
 
     def get(self, job_id: str, *, tenant_id: str) -> HermesJob:
@@ -138,7 +253,8 @@ class HermesAdapter:
     ) -> list[HermesJob]:
         state_filter = {HermesJobState(item) for item in states} if states else None
         with self.store.lock:
-            self._requeue_expired_locked()
+            if self._requeue_expired_locked():
+                self.store.persist_locked()
             jobs = [
                 job
                 for job in self.store.jobs.values()
@@ -157,6 +273,7 @@ class HermesAdapter:
         profile_id: str,
         worker_id: str,
         lease_seconds: int | None = None,
+        job_id: str | None = None,
     ) -> HermesJob | None:
         worker_id = str(worker_id or "").strip()
         if not worker_id:
@@ -172,10 +289,12 @@ class HermesAdapter:
                 if job.tenant_id == tenant_id
                 and job.cell_id == cell_id
                 and job.profile_id == profile_id
+                and (job_id is None or job.job_id == job_id)
                 and job.state == HermesJobState.QUEUED
             ]
             self.store.worker_heartbeats[worker_id] = (utc_now(), None)
             if not candidates:
+                self.store.persist_locked()
                 return None
             current = sorted(
                 candidates,
@@ -190,6 +309,7 @@ class HermesAdapter:
                 ),
             )
             self.store.worker_heartbeats[worker_id] = (utc_now(), claimed.job_id)
+            self.store.persist_locked()
             return claimed
 
     def start(
@@ -201,9 +321,12 @@ class HermesAdapter:
         expected_revision: int | None = None,
     ) -> HermesJob:
         with self.store.lock:
+            self._requeue_expired_locked()
             current = self._owned_job_locked(job_id, tenant_id, worker_id)
             self._assert_revision(current, expected_revision)
-            return self._transition_locked(current, HermesJobState.RUNNING)
+            updated = self._transition_locked(current, HermesJobState.RUNNING)
+            self.store.persist_locked()
+            return updated
 
     def heartbeat(
         self,
@@ -215,6 +338,7 @@ class HermesAdapter:
     ) -> HermesJob:
         duration = int(lease_seconds or self.default_lease_seconds)
         with self.store.lock:
+            self._requeue_expired_locked()
             current = self._owned_job_locked(job_id, tenant_id, worker_id)
             if current.state not in {HermesJobState.CLAIMED, HermesJobState.RUNNING}:
                 raise HermesConflict("job does not accept heartbeat")
@@ -228,6 +352,7 @@ class HermesAdapter:
             )
             self.store.jobs[job_id] = updated
             self.store.worker_heartbeats[worker_id] = (utc_now(), job_id)
+            self.store.persist_locked()
             return updated
 
     def complete(
@@ -283,6 +408,8 @@ class HermesAdapter:
                 result = self.store.results.get(job_id)
                 if result is None:
                     raise HermesConflict("terminal result missing")
+                if result.state != HermesJobState.CANCELLED:
+                    raise HermesConflict("job already has a different terminal result")
                 return result
             updated = self._transition_locked(current, HermesJobState.CANCELLED)
             result = HermesResult(
@@ -296,6 +423,7 @@ class HermesAdapter:
                 finished_at=utc_now(),
             )
             self.store.results[job_id] = result
+            self.store.persist_locked()
             return result
 
     def result(self, job_id: str, *, tenant_id: str) -> HermesResult:
@@ -309,9 +437,18 @@ class HermesAdapter:
                 raise HermesConflict("job has no terminal result")
             return result
 
+    def requeue_expired(self) -> tuple[str, ...]:
+        """Persistently release expired leases without changing policy state."""
+        with self.store.lock:
+            changed = self._requeue_expired_locked()
+            if changed:
+                self.store.persist_locked()
+            return tuple(sorted(changed))
+
     def health(self) -> HermesHealth:
         with self.store.lock:
-            self._requeue_expired_locked()
+            if self._requeue_expired_locked():
+                self.store.persist_locked()
             active_jobs = sum(
                 1 for job in self.store.jobs.values()
                 if job.state in {HermesJobState.CLAIMED, HermesJobState.RUNNING}
@@ -354,7 +491,20 @@ class HermesAdapter:
         expected_revision: int | None,
     ) -> HermesResult:
         with self.store.lock:
+            self._requeue_expired_locked()
             current = self._owned_job_locked(job_id, tenant_id, worker_id)
+            if current.state in TERMINAL_STATES:
+                existing = self.store.results.get(job_id)
+                if existing is None:
+                    raise HermesConflict("terminal result missing")
+                if (
+                    existing.state == state
+                    and existing.worker_id == worker_id
+                    and existing.output == output
+                    and existing.error == error
+                ):
+                    return existing
+                raise HermesConflict("job already has a different terminal result")
             self._assert_revision(current, expected_revision)
             updated = self._transition_locked(current, state)
             result = HermesResult(
@@ -371,6 +521,7 @@ class HermesAdapter:
             )
             self.store.results[job_id] = result
             self.store.worker_heartbeats[worker_id] = (utc_now(), None)
+            self.store.persist_locked()
             return result
 
     def _owned_job_locked(self, job_id: str, tenant_id: str, worker_id: str) -> HermesJob:
@@ -418,8 +569,9 @@ class HermesAdapter:
         self.store.jobs[current.job_id] = updated
         return updated
 
-    def _requeue_expired_locked(self) -> None:
+    def _requeue_expired_locked(self) -> set[str]:
         now = datetime.now(timezone.utc)
+        changed: set[str] = set()
         for current in list(self.store.jobs.values()):
             expiry = self._parse(current.lease_expires_at)
             if (
@@ -435,3 +587,5 @@ class HermesAdapter:
                     lease_expires_at=None,
                     updated_at=utc_now(),
                 )
+                changed.add(current.job_id)
+        return changed

active-runtime/app/aios_v2/app.py

@@ -12,7 +12,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Any, Mapping
 
-from aios_hermes import HermesAdapter, HermesJobState, HermesResult
+from aios_hermes import HermesAdapter, HermesJobState, HermesResult, SQLiteHermesStore
 from aios_openclaw import (
     Ed25519CommandSigner,
     HTTPRuntimeTransport,
@@ -23,7 +23,7 @@ from aios_openclaw.callback import AuditCallbackSink, EventPublisher
 from .console import CONSOLE_HTML
 from .control_plane import AuditEvent, RequestContext, Unauthorized
 from .dispatch_store import SQLiteDispatchRepository
-from .execution_tracking import ExecutionTracker
+from .execution_tracking import ExecutionEvidence, ExecutionTracker
 from .gateway import GatewayApplication, error_response
 from .gitops_status import collect_gitops_status
 from .orchestrator import ExecutionOrchestrator
@@ -143,13 +143,17 @@ class ControlPlaneEventPublisher(EventPublisher):
             raise ValueError("runtime audit event is missing tenant identity")
         state_value = str(event.get("state") or "unknown").lower()
         outcome = "allowed" if state_value == "completed" else state_value
+        runtime = str(event.get("runtime") or "openclaw").strip().lower()
+        principal_id = str(
+            event.get("principal_id") or f"{runtime}-runtime"
+        ).strip()
         audit = AuditEvent(
             event_id=f"runtime_{command_id}_{len(self.service.store.audit_events) + 1}",
             tenant_id=tenant_id,
-            principal_id="openclaw-runtime",
+            principal_id=principal_id,
             trace_id=trace_id,
             action=event_type,
-            resource_type="openclaw_command",
+            resource_type=f"{runtime}_execution",
             resource_id=command_id,
             outcome=outcome,
             policy_version="runtime-v2",
@@ -168,12 +172,16 @@ class CandidateApplication:
         tracker: ExecutionTracker,
         hermes: HermesAdapter,
         openclaw: OpenClawAdapter | None,
+        *,
+        hermes_lifecycle_api_enabled: bool = False,
     ) -> None:
         self.gateway = gateway
         self.orchestrator = orchestrator
         self.tracker = tracker
         self.hermes = hermes
         self.openclaw = openclaw
+        self.hermes_lifecycle_api_enabled = bool(hermes_lifecycle_api_enabled)
+        self.runtime_events = ControlPlaneEventPublisher(orchestrator.control_plane)
 
     def context(self, headers: Mapping[str, str]) -> RequestContext:
         return self.gateway.context_from_headers(headers)
@@ -183,6 +191,100 @@ class CandidateApplication:
         if not ({"runtime", "platform_admin"} & set(context.roles)):
             raise Unauthorized("runtime reconciliation role is required")
 
+    @staticmethod
+    def _required_identity(payload: Mapping[str, Any], name: str) -> str:
+        value = str(payload.get(name) or "").strip()
+        if not value:
+            raise ValueError(f"{name} is required")
+        return value
+
+    def _record_hermes_lifecycle(
+        self,
+        context: RequestContext,
+        dispatch: Any,
+        job: Any,
+        *,
+        action: str,
+        worker_id: str | None,
+        provenance_id: str,
+        run_id: str,
+    ) -> None:
+        executor = worker_id or "aios-control-plane"
+        evidence = ExecutionEvidence(
+            execution_id=job.job_id,
+            tenant_id=job.tenant_id,
+            task_id=job.task_id,
+            agent_id=dispatch.agent_id,
+            executor=executor,
+            branch="runtime/hermes",
+            status=job.state.value.lower(),
+            progress_completed=0,
+            progress_total=3,
+            current_step=action,
+            provenance_id=provenance_id,
+            run_id=run_id,
+        )
+        current_evidence = self.tracker.claim(
+            evidence,
+            principal_id=context.principal_id,
+            provenance_id=provenance_id,
+            run_id=run_id,
+        )
+        if job.state in {
+            HermesJobState.COMPLETED,
+            HermesJobState.FAILED,
+            HermesJobState.CANCELLED,
+        }:
+            if current_evidence.completed_at is None:
+                self.tracker.complete(
+                    tenant_id=job.tenant_id,
+                    execution_id=job.job_id,
+                    status=job.state.value.lower(),
+                    test_status="runtime_terminal",
+                    test_summary={"lifecycle_action": action},
+                    result_summary={
+                        "state": job.state.value,
+                        "worker_id": worker_id,
+                        "dispatch_id": dispatch.dispatch_id,
+                    },
+                    principal_id=context.principal_id,
+                    provenance_id=provenance_id,
+                    run_id=run_id,
+                )
+        else:
+            self.tracker.heartbeat(
+                tenant_id=job.tenant_id,
+                execution_id=job.job_id,
+                status=job.state.value.lower(),
+                current_step=action,
+                progress_completed=(1 if job.state == HermesJobState.RUNNING else 0),
+                progress_total=3,
+                principal_id=context.principal_id,
+                provenance_id=provenance_id,
+                run_id=run_id,
+            )
+        self.runtime_events.publish(
+            {
+                "runtime": "hermes",
+                "tenant_id": job.tenant_id,
+                "principal_id": context.principal_id,
+                "trace_id": job.trace_id,
+                "command_id": job.job_id,
+                "event_type": f"HERMES_{action.upper()}",
+                "state": job.state.value,
+                "details": {
+                    "dispatch_id": dispatch.dispatch_id,
+                    "task_id": job.task_id,
+                    "worker_id": worker_id,
+                    "cell_id": job.cell_id,
+                    "profile_id": job.profile_id,
+                    "revision": job.revision,
+                    "provenance_id": provenance_id,
+                    "run_id": run_id,
+                },
+            }
+        )
+
     def handle_json(
         self,
         method: str,
@@ -269,6 +371,226 @@ class CandidateApplication:
             )
             return HTTPStatus.ACCEPTED, {"dispatch": _json_value(record)}
 
+        lifecycle_match = __import__("re").fullmatch(
+            r"/api/v2/internal/runtime/hermes/lifecycle/"
+            r"(claim|start|heartbeat|complete|fail|cancel|status)",
+            path,
+        )
+        if method == "POST" and lifecycle_match:
+            if not self.hermes_lifecycle_api_enabled:
+                return HTTPStatus.NOT_FOUND, {
+                    "error": "not_found",
+                    "message": "route not found",
+                }
+            context = self.context(headers)
+            self._require_runtime_role(context)
+            if not isinstance(body, Mapping):
+                raise ValueError("JSON body must be an object")
+            payload = body
+            action = lifecycle_match.group(1)
+            identity = {
+                name: self._required_identity(payload, name)
+                for name in (
+                    "dispatch_id",
+                    "job_id",
+                    "tenant_id",
+                    "task_id",
+                    "trace_id",
+                    "principal_id",
+                    "provenance_id",
+                    "run_id",
+                    "cell_id",
+                    "profile_id",
+                )
+            }
+            header_provenance, header_run = _provenance_context(headers, payload)
+            if header_provenance != identity["provenance_id"]:
+                raise ValueError("provenance_id context mismatch")
+            if header_run != identity["run_id"]:
+                raise ValueError("run_id context mismatch")
+            requeued_job_ids = self.hermes.requeue_expired()
+            dispatch, job = self.orchestrator.hermes_lifecycle_context(
+                context,
+                dispatch_id=identity["dispatch_id"],
+                job_id=identity["job_id"],
+                tenant_id=identity["tenant_id"],
+                task_id=identity["task_id"],
+                trace_id=identity["trace_id"],
+                principal_id=identity["principal_id"],
+                provenance_id=identity["provenance_id"],
+                run_id=identity["run_id"],
+            )
+            if job.cell_id != identity["cell_id"]:
+                raise ValueError("cell_id mismatch")
+            if job.profile_id != identity["profile_id"]:
+                raise ValueError("profile_id mismatch")
+
+            if action == "status":
+                if job.job_id in requeued_job_ids:
+                    self._record_hermes_lifecycle(
+                        context,
+                        dispatch,
+                        job,
+                        action="lease_expired_requeue",
+                        worker_id=None,
+                        provenance_id=identity["provenance_id"],
+                        run_id=identity["run_id"],
+                    )
+                task = self.orchestrator.control_plane.get_task(
+                    context, dispatch.task_id
+                )
+                response: dict[str, Any] = {
+                    "job": _json_value(job),
+                    "dispatch": _json_value(dispatch),
+                    "task": _json_value(task),
+                }
+                if job.state in {
+                    HermesJobState.COMPLETED,
+                    HermesJobState.FAILED,
+                    HermesJobState.CANCELLED,
+                }:
+                    response["result"] = _json_value(
+                        self.hermes.result(job.job_id, tenant_id=job.tenant_id)
+                    )
+                return HTTPStatus.OK, response
+
+            expected_revision = payload.get("expected_revision")
+            expected = (
+                int(expected_revision) if expected_revision is not None else None
+            )
+            if action == "cancel":
+                result = self.hermes.cancel(
+                    job.job_id,
+                    tenant_id=job.tenant_id,
+                    expected_revision=expected,
+                )
+            else:
+                worker_id = self._required_identity(payload, "worker_id")
+                if action == "claim":
+                    claimed = self.hermes.claim(
+                        tenant_id=job.tenant_id,
+                        cell_id=job.cell_id,
+                        profile_id=job.profile_id,
+                        worker_id=worker_id,
+                        lease_seconds=(
+                            int(payload["lease_seconds"])
+                            if payload.get("lease_seconds") is not None
+                            else None
+                        ),
+                        job_id=job.job_id,
+                    )
+                    if claimed is None:
+                        raise ValueError("Hermes job is not claimable")
+                    bound = self.orchestrator.bind_hermes_worker(
+                        context,
+                        claimed,
+                        dispatch_id=dispatch.dispatch_id,
+                        provenance_id=identity["provenance_id"],
+                        run_id=identity["run_id"],
+                        worker_id=worker_id,
+                    )
+                    self._record_hermes_lifecycle(
+                        context,
+                        bound,
+                        claimed,
+                        action="claim",
+                        worker_id=worker_id,
+                        provenance_id=identity["provenance_id"],
+                        run_id=identity["run_id"],
+                    )
+                    return HTTPStatus.OK, {
+                        "job": _json_value(claimed),
+                        "dispatch": _json_value(bound),
+                    }
+                if action == "start":
+                    started = self.hermes.start(
+                        job.job_id,
+                        tenant_id=job.tenant_id,
+                        worker_id=worker_id,
+                        expected_revision=expected,
+                    )
+                    self._record_hermes_lifecycle(
+                        context,
+                        dispatch,
+                        started,
+                        action="start",
+                        worker_id=worker_id,
+                        provenance_id=identity["provenance_id"],
+                        run_id=identity["run_id"],
+                    )
+                    return HTTPStatus.OK, {"job": _json_value(started)}
+                if action == "heartbeat":
+                    heartbeat = self.hermes.heartbeat(
+                        job.job_id,
+                        tenant_id=job.tenant_id,
+                        worker_id=worker_id,
+                        lease_seconds=(
+                            int(payload["lease_seconds"])
+                            if payload.get("lease_seconds") is not None
+                            else None
+                        ),
+                    )
+                    self._record_hermes_lifecycle(
+                        context,
+                        dispatch,
+                        heartbeat,
+                        action="heartbeat",
+                        worker_id=worker_id,
+                        provenance_id=identity["provenance_id"],
+                        run_id=identity["run_id"],
+                    )
+                    return HTTPStatus.OK, {"job": _json_value(heartbeat)}
+                if action == "complete":
+                    output = payload.get("output")
+                    if not isinstance(output, Mapping):
+                        raise ValueError("output must be an object")
+                    result = self.hermes.complete(
+                        job.job_id,
+                        tenant_id=job.tenant_id,
+                        worker_id=worker_id,
+                        output=output,
+                        expected_revision=expected,
+                    )
+                elif action == "fail":
+                    error = payload.get("error")
+                    if not isinstance(error, Mapping):
+                        raise ValueError("error must be an object")
+                    result = self.hermes.fail(
+                        job.job_id,
+                        tenant_id=job.tenant_id,
+                        worker_id=worker_id,
+                        error=error,
+                        expected_revision=expected,
+                    )
+                else:
+                    raise ValueError("unsupported Hermes lifecycle action")
+
+            task = self.orchestrator.reconcile_hermes(
+                context,
+                result,
+                dispatch_id=dispatch.dispatch_id,
+                provenance_id=identity["provenance_id"],
+                run_id=identity["run_id"],
+            )
+            updated_dispatch = self.orchestrator.dispatch_repository.get(
+                dispatch.dispatch_id, tenant_id=dispatch.tenant_id
+            )
+            terminal_job = self.hermes.get(job.job_id, tenant_id=job.tenant_id)
+            self._record_hermes_lifecycle(
+                context,
+                updated_dispatch,
+                terminal_job,
+                action=action,
+                worker_id=result.worker_id,
+                provenance_id=identity["provenance_id"],
+                run_id=identity["run_id"],
+            )
+            return HTTPStatus.OK, {
+                "result": _json_value(result),
+                "task": _json_value(task),
+                "dispatch": _json_value(updated_dispatch),
+            }
+
         if method == "POST" and path == "/api/v2/internal/runtime/hermes/reconcile":
             context = self.context(headers)
             self._require_runtime_role(context)
@@ -411,7 +733,7 @@ def build_application() -> CandidateApplication:
         provenance_writer=provenance_writer,
         require_provenance=require_provenance,
     )
-    hermes = HermesAdapter()
+    hermes = HermesAdapter(store=SQLiteHermesStore(state_dir / "hermes.sqlite3"))
 
     openclaw_url = os.getenv("AIOS_OPENCLAW_URL", "").strip()
     signing_file = os.getenv("AIOS_OPENCLAW_SIGNING_MATERIAL_FILE", "").strip()
@@ -446,7 +768,16 @@ def build_application() -> CandidateApplication:
         require_provenance=require_provenance,
         dispatch_repository=dispatch_repository,
     )
-    return CandidateApplication(gateway, orchestrator, tracker, hermes, openclaw)
+    return CandidateApplication(
+        gateway,
+        orchestrator,
+        tracker,
+        hermes,
+        openclaw,
+        hermes_lifecycle_api_enabled=_enabled(
+            os.getenv("AIOS_HERMES_LIFECYCLE_API_ENABLED", "0")
+        ),
+    )
 
 
 def main() -> None:

active-runtime/app/aios_v2/gateway.py

@@ -17,6 +17,8 @@ from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any, Mapping
 
+from aios_hermes import HermesConflict, HermesNotFound
+
 from .control_plane import (
     ControlPlaneError,
     ControlPlaneService,
@@ -81,20 +83,27 @@ class GatewayApplication:
 
     @staticmethod
     def context_from_headers(headers: Mapping[str, str]) -> RequestContext:
-        authenticated = str(headers.get(AUTHENTICATED_HEADER, "")).strip().lower()
+        normalized_headers = {
+            str(key).strip().lower(): value for key, value in headers.items()
+        }
+
+        def header(name: str, default: str = "") -> str:
+            return str(normalized_headers.get(name.lower(), default))
+
+        authenticated = header(AUTHENTICATED_HEADER).strip().lower()
         if authenticated not in {"1", "true", "yes"}:
             raise Unauthorized("trusted authentication assertion is required")
         roles = [
             role.strip()
-            for role in str(headers.get(ROLES_HEADER, "")).split(",")
+            for role in header(ROLES_HEADER).split(",")
             if role.strip()
         ]
         return RequestContext.create(
-            tenant_id=str(headers.get(TENANT_HEADER, "")),
-            principal_id=str(headers.get(PRINCIPAL_HEADER, "")),
-            trace_id=str(headers.get(TRACE_HEADER, "")).strip() or None,
-            session_id=str(headers.get(SESSION_HEADER, "")).strip() or None,
-            policy_version=str(headers.get(POLICY_HEADER, "v1")).strip() or "v1",
+            tenant_id=header(TENANT_HEADER),
+            principal_id=header(PRINCIPAL_HEADER),
+            trace_id=header(TRACE_HEADER).strip() or None,
+            session_id=header(SESSION_HEADER).strip() or None,
+            policy_version=header(POLICY_HEADER, "v1").strip() or "v1",
             roles=roles,
         )
 
@@ -242,11 +251,13 @@ _ERROR_STATUS: tuple[tuple[type[BaseException], HTTPStatus], ...] = (
     (TenantIsolationError, HTTPStatus.NOT_FOUND),
     (NotFound, HTTPStatus.NOT_FOUND),
     (DispatchNotFound, HTTPStatus.NOT_FOUND),
+    (HermesNotFound, HTTPStatus.NOT_FOUND),
     (ProvenanceReplay, HTTPStatus.CONFLICT),
     (IdempotencyConflict, HTTPStatus.CONFLICT),
     (RevisionConflict, HTTPStatus.CONFLICT),
     (InvalidTransition, HTTPStatus.CONFLICT),
     (DispatchConflict, HTTPStatus.CONFLICT),
+    (HermesConflict, HTTPStatus.CONFLICT),
     (OrchestrationError, HTTPStatus.CONFLICT),
     (ValueError, HTTPStatus.BAD_REQUEST),
     (ControlPlaneError, HTTPStatus.BAD_REQUEST),

active-runtime/app/aios_v2/orchestrator.py

@@ -353,6 +353,76 @@ class ExecutionOrchestrator:
             if getattr(dispatch, field_name) != expected_value:
                 raise OrchestrationError(f"dispatch {field_name} mismatch")
 
+    def hermes_lifecycle_context(
+        self,
+        context: RequestContext,
+        *,
+        dispatch_id: str,
+        job_id: str,
+        tenant_id: str,
+        task_id: str,
+        trace_id: str,
+        principal_id: str,
+        provenance_id: str,
+        run_id: str,
+    ) -> tuple[DispatchRecord, HermesJob]:
+        """Resolve a runtime lifecycle request against its durable dispatch."""
+        dispatch = self.dispatch_repository.get(
+            dispatch_id, tenant_id=context.tenant_id
+        )
+        self._assert_dispatch_context(
+            dispatch,
+            runtime="hermes",
+            runtime_execution_id=job_id,
+            tenant_id=tenant_id,
+            task_id=task_id,
+            trace_id=trace_id,
+            principal_id=principal_id,
+            provenance_id=provenance_id,
+            run_id=run_id,
+        )
+        if context.tenant_id != tenant_id:
+            raise OrchestrationError("runtime context tenant_id mismatch")
+        if context.principal_id != principal_id:
+            raise OrchestrationError("runtime context principal_id mismatch")
+        if context.trace_id != trace_id:
+            raise OrchestrationError("runtime context trace_id mismatch")
+        if self.hermes is None:
+            raise OrchestrationError("Hermes runtime is not configured")
+        job = self.hermes.get(job_id, tenant_id=tenant_id)
+        if job.task_id != task_id or job.trace_id != trace_id:
+            raise OrchestrationError("Hermes job identity mismatch")
+        return dispatch, job
+
+    def bind_hermes_worker(
+        self,
+        context: RequestContext,
+        job: HermesJob,
+        *,
+        dispatch_id: str,
+        provenance_id: str,
+        run_id: str,
+        worker_id: str,
+    ) -> DispatchRecord:
+        dispatch, current = self.hermes_lifecycle_context(
+            context,
+            dispatch_id=dispatch_id,
+            job_id=job.job_id,
+            tenant_id=job.tenant_id,
+            task_id=job.task_id,
+            trace_id=job.trace_id,
+            principal_id=context.principal_id,
+            provenance_id=provenance_id,
+            run_id=run_id,
+        )
+        if current.assigned_worker_id != worker_id:
+            raise OrchestrationError("Hermes worker does not own claimed job")
+        return self.dispatch_repository.update(
+            dispatch.dispatch_id,
+            tenant_id=dispatch.tenant_id,
+            expected_worker_id=worker_id,
+        )
+
     def reconcile_openclaw(
         self,
         context: RequestContext,

docs/aios-v2/01-USER-REQUIREMENTS.md

@@ -221,6 +221,14 @@ OpenClaw/Hermes 结果只有在与持久 Dispatch 的身份和执行上下文完
 - 至少校验 tenant、task、agent、principal、trace、runtime、runtime execution、provenance 和 run。
 - 不同终态结果摘要不得覆盖已有终态。
 - Hermes 终态结果必须来自持有 Job Lease 的 Worker。
+- Hermes Worker 生命周期验收入口必须是默认关闭的内部接口，仅接受可信
+  `runtime` 或 `platform_admin` 身份；普通 Operator、浏览器和公共 API 不得直接调用。
+- 每次 Claim/Start/Heartbeat/Complete/Fail/Cancel 必须绑定
+  tenant、task、trace、principal、provenance、run、dispatch、job、cell 和 profile。
+- Complete/Fail/Cancel 必须通过 `ExecutionOrchestrator.reconcile_hermes` 更新权威
+  Task/Dispatch，并同步生成 Audit/Event 与 Execution Evidence。
+- Lease 到期必须持久化清除旧 Owner、递增 revision 并允许安全重领；旧 Worker
+  后续 Result 必须拒绝。
 - Runtime callback 审计不得绕过 Orchestrator Reconcile。
 
 ### UR-019 Dispatch 持久化与恢复
@@ -232,6 +240,8 @@ Runtime 副作用之前必须持久化唯一 Dispatch，重启后能够恢复并
 - 同租户 Task 和 Runtime Execution 各自唯一绑定一个 Dispatch。
 - Dispatch 支持按 ID、Task 与 Runtime Execution 查询。
 - 终态 Dispatch 不得迁移到不同终态。
+- Hermes Job、Result、Lease、幂等键和 Worker heartbeat 必须与 Dispatch 一样在
+  服务重启后恢复；不得以只存在于进程内的 Map 作为在线验收依据。
 - 有 SQLite Repository 重启恢复测试。
 
 ### UR-020 OpenClaw Ed25519 命令合同
@@ -329,4 +339,4 @@ OpenClaw 生产命令必须使用版本化 Ed25519 合同，不得继续使用 H
 7. 租户用户在 Web Console 查看完整链路。
 8. 跨租户、跨 Principal、跨 Grant Path 和重放请求被拒绝。
 9. 服务重启后 Control Plane、Dispatch 和 Provenance 状态恢复。
-10. 候选部署健康检查、Acceptance 和回滚演练实际执行并保存证据。
+10. 候选部署健康检查、Acceptance 和回滚演练实际执行并保存证据。

docs/aios-v2/04-MODULE-CONTRACTS.md

@@ -4,6 +4,10 @@
 - 状态：加固候选契约基线
 - 原则：Control Plane 权威、Runtime 仅执行、默认拒绝、服务级最小权限、全链路租户隔离、所有权威状态可审计和恢复
 
+核心模型名称与实现保持逐字同步：`RequestContext` 承载可信租户、主体、角色、trace、
+session 与 policy version；`TaskState` 是 AIOS 权威 Task 生命周期枚举，Runtime Worker
+不得直接写入或复制该状态机。
+
 ## 1. 通用可信上下文契约
 
 所有控制与执行请求必须携带以下上下文：
@@ -374,6 +378,17 @@ attempts
 
 ## 12. Hermes Contract
 
+### 12.0 内部生命周期入口
+
+- 路径：`/api/v2/internal/runtime/hermes/lifecycle/{claim,start,heartbeat,complete,fail,cancel,status}`。
+- `AIOS_HERMES_LIFECYCLE_API_ENABLED` 默认关闭；关闭时按不存在的路由 fail closed。
+- 仅可信反向代理注入且具备 `runtime` 或 `platform_admin` 角色的身份可调用。
+- 所有写请求必须携带并严格匹配 tenant、task、trace、principal、provenance、run、
+  dispatch、job、cell、profile；Claim 还必须指定精确 job，不能从分区中误领其他 Job。
+- Worker 不得直接修改 Task Store。Complete、Fail 和 Cancel 产生的 HermesResult
+  必须经持久 Dispatch 校验并由 `ExecutionOrchestrator.reconcile_hermes` 终态化。
+- 每次成功生命周期写入同时落盘 Hermes 状态、Audit/Event 和 Execution Evidence。
+
 ### 12.1 HermesJob
 
 ```text
@@ -400,6 +415,10 @@ updated_at
 - `tenant_id + cell_id + profile_id` 形成执行分区。
 - claim 只能领取同分区 Job。
 - Lease 到期 Job 回到 QUEUED。
+- Hermes Store 使用 `AIOS_V2_STATE_DIR/hermes.sqlite3` 保存 Job、Result、Lease、
+  Idempotency 和 Worker heartbeat；Adapter 仍是唯一 Queue/Lease 状态机。
+- Lease 到期重排必须清空 `assigned_worker_id`、清空 `lease_expires_at`、递增 revision
+  并持久化。重启后由显式 lifecycle status/claim 或健康检查安全触发重排。
 - revision 冲突不得覆盖。
 
 ### 12.2 HermesResult
@@ -422,6 +441,8 @@ finished_at
 - 终态 Result 必须携带 Worker ID。
 - Worker 必须持有 Job Lease。
 - Result 必须匹配 Dispatch、Principal、Parent Provenance 和 Run。
+- 相同 Worker、相同终态和相同 output/error 的重复提交返回原 Result；不同终态或
+  不同结果内容返回 Conflict，不得覆盖已持久化终态。
 - 只允许内部入口 `POST /api/v2/internal/runtime/hermes/reconcile` 处理。
 - 旧入口 `POST /api/v2/hermes/results/reconcile` 返回 `410 Gone`。
 

docs/aios-v2/05-TRACEABILITY-MATRIX.md

@@ -20,23 +20,23 @@
 |---|---|---|---|---|---|
 | UR-001 唯一控制面 | ADR-001；模块契约第 2～6 节 | `active-runtime/app/aios_v2/control_plane.py`、`gateway.py`、`orchestrator.py` | `tests/aios_v2_candidate/test_candidate_flow.py` | `deploy/aios-v2/acceptance/run-all.sh`；公共生命周期旁路负向检查 | 已实现；测试已提交；待验收 |
 | UR-002 复用现有能力 | ADR-002 | `active-runtime/app/aios_openclaw/`、`active-runtime/app/aios_hermes/`、`active-runtime/app/aios_v2/orchestrator.py` | `test_candidate_flow.py` | OpenClaw/Hermes 联调 | 已实现；真实 Runtime 联调待验收 |
-| UR-003 OpenClaw/Hermes 解耦 | ADR-002；模块契约第 11、12 节 | OpenClaw Adapter、Hermes Adapter/Worker、内部 Reconcile | `test_candidate_flow.py` | `/api/v2/internal/runtime/hermes/reconcile`；Runtime callback 审计检查 | 已实现；测试已提交；待验收 |
+| UR-003 OpenClaw/Hermes 解耦 | ADR-002、ADR-010；模块契约第 11、12 节 | OpenClaw Adapter、Hermes Adapter/Worker、内部 Lifecycle/Reconcile | `test_candidate_flow.py`、`test_hermes_live_lifecycle.py` | 默认关闭的内部 Runtime API；Runtime audit/evidence 检查 | 已实现；Hermes 专项 17/17 |
 | UR-004 全面 Web 化 | 目标架构；模块契约第 15 节 | `active-runtime/app/aios_v2/app.py`、`console.py`、`gateway.py` | `test_candidate_flow.py` | Web/API 冒烟、Healthcheck | 已实现；UI/浏览器验收待执行 |
 | UR-005 多租户隔离 | ADR-004 | `control_plane.py`、`persistence.py`、`execution_tracking.py`、`dispatch_store.py` | `test_candidate_flow.py::test_cross_tenant_task_read_is_denied` | 跨租户负向 Acceptance | 已实现；测试已提交；待验收 |
 | UR-006 权威任务生命周期 | ADR-001；模块契约第 4 节 | `control_plane.py`、`gateway.py`、`orchestrator.py` | `test_candidate_flow.py` | `/start`、`/complete`、`/fail` 应返回拒绝；Runtime 严格 Reconcile | 已实现；测试/接口验收待执行 |
 | UR-007 策略与审批 | ADR-001 | `control_plane.py`、`gateway.py` | `test_candidate_flow.py` | Approval、deny audit、policy version 检查 | 已实现；策略场景验收待执行 |
-| UR-008 幂等与并发安全 | 模块契约第 4、10、12 节 | Task idempotency/revision、Hermes lease/revision、Dispatch 唯一约束 | `test_candidate_flow.py` | 幂等冲突、Lease、终态冲突 Acceptance | 已实现；测试代码覆盖部分场景；待验收 |
+| UR-008 幂等与并发安全 | 模块契约第 4、10、12 节 | Task idempotency/revision、Hermes lease/revision、Dispatch 唯一约束 | `test_candidate_flow.py`、`test_hermes_live_lifecycle.py` | Lease 到期重排、换主、相同终态幂等、不同终态冲突 | 已实现；专项测试通过 |
 | UR-009 可观测与执行证据 | 模块契约第 6、13 节 | `execution_tracking.py`、`gitops_status.py`、`console.py` | `test_candidate_flow.py::test_execution_tracking_persists_claim_heartbeat_and_result` | `/api/v2/executions`、Evidence Receipt 检查 | 已实现；测试已提交；待验收 |
 | UR-010 GitOps 唯一变更流程 | ADR-003 | `active-runtime/app/aios_v2/provenance_git.py`、GitOps 工具与任务资产 | `test_provenance.py`、Grant 隔离测试 | Branch/Commit/Push 门禁、Controller clean、OID 同步 | 已实现候选控制；环境 GitOps 验收待执行 |
-| UR-011 持久化与恢复 | ADR-006；模块契约第 10、16 节 | `persistence.py`、`dispatch_store.py`、`provenance.py`、`provenance_grants.py` | `test_candidate_flow.py` 中 Control Plane 与 Dispatch recovery | 重启、备份、恢复、锁等待、容量测试 | 候选实现完成；恢复测试已提交；生产 HA/容量待验收 |
+| UR-011 持久化与恢复 | ADR-006、ADR-010；模块契约第 10、12、16 节 | `persistence.py`、`dispatch_store.py`、`aios_hermes/adapter.py::SQLiteHermesStore`、Provenance/Grant Store | `test_candidate_flow.py`、`test_hermes_live_lifecycle.py` | 运行中 Job 重启恢复、重启后 Reconcile、SQLite quick_check | Hermes/Dispatch/Task 恢复测试通过；生产 HA/容量待验收 |
 | UR-012 审计与合规 | ADR-001、ADR-004、ADR-009 | `control_plane.py` AuditEvent、`app.py` callback audit、Provenance/Grant Event/Receipt | `test_candidate_flow.py`、`test_provenance.py`、`test_provenance_grant_isolation.py` | `/api/v2/audit`、Receipt/Event 完整性检查 | 已实现候选审计；不可变外部固化待预发布 |
 | UR-013 密钥安全 | ADR-009；模块契约第 6、11、17 节 | Ed25519 Signer/Verifier、protected file reader、SecretRef 配置 | `test_provenance.py`、`test_candidate_flow.py` | Secret scanner、权限检查、容器/systemd credential 检查 | 已实现；环境密钥注入验收待执行 |
 | UR-014 部署与回滚 | 迁移与交付文档；模块契约第 17 节 | `deploy/aios-v2/systemd/`、`docker/`、`healthcheck/`、`rollback/` | Acceptance 脚本与候选测试 | systemd/Docker 安装、Healthcheck、备份与回滚演练 | 资产已提交；实际部署/回滚未执行 |
 | UR-015 文档与代码同步 | ADR-005 | `tools/aios_v2_docs_code_check.py`、`docs/aios-v2/` | 文档门禁脚本 | Acceptance Runner 调用文档门禁 | 已实现；门禁尚未实际执行 |
 | UR-016 Workflow Run Provenance | ADR-009；模块契约第 7 节 | `active-runtime/app/aios_v2/provenance.py`、`provenance_grants.py::WorkflowProvenanceIssuer` | `test_provenance.py`、`test_provenance_grant_isolation.py::test_parent_workflow_has_no_operational_rights` | CLI `issue-workflow`、状态/撤销检查 | 已实现；测试已提交；待验收 |
 | UR-017 最小权限子 Grant | ADR-009；模块契约第 8 节 | `provenance_grants.py`、`provenance_scoped.py` | `test_provenance_grant_isolation.py` | `issue-grant`、Grant status/revoke、Receipt 检查 | 已实现；专项测试已提交；待验收 |
-| UR-018 Runtime Result 信任绑定 | ADR-009；模块契约第 10～12 节 | `orchestrator.py`、`dispatch_store.py`、内部 Hermes Reconcile | `test_candidate_flow.py` | OpenClaw/Hermes 伪造 tenant/task/trace/principal/provenance/run 负向验收 | 已实现；恢复正向测试已提交；完整负向验收待执行 |
-| UR-019 Dispatch 持久化与恢复 | ADR-009；模块契约第 10 节 | `dispatch_store.py`、`orchestrator.py`、`app.py` | `test_candidate_flow.py::test_dispatch_and_control_plane_recover_before_hermes_reconcile` | 进程重启后读取 Dispatch 并完成 Reconcile | 已实现；测试已提交；待执行 |
+| UR-018 Runtime Result 信任绑定 | ADR-009、ADR-010；模块契约第 10～12 节 | `orchestrator.py`、`dispatch_store.py`、内部 Hermes Lifecycle/Reconcile | `test_hermes_live_lifecycle.py` | tenant/task/trace/principal/provenance/run/dispatch/job/cell/profile/worker 负向矩阵 | 已实现；专项 17/17 |
+| UR-019 Dispatch 持久化与恢复 | ADR-009、ADR-010；模块契约第 10、12 节 | `dispatch_store.py`、`orchestrator.py`、`app.py`、`SQLiteHermesStore` | Dispatch recovery 与 Hermes restart-reconcile 测试 | 进程重启后读取 Dispatch/Job 并完成 Reconcile | 已实现；测试通过 |
 | UR-020 OpenClaw Ed25519 命令合同 | ADR-009；模块契约第 11 节 | `active-runtime/app/aios_openclaw/contract.py`、`adapter.py`、`app.py` | `test_candidate_flow.py`、`test_provenance.py` | 真实 OpenClaw v2 合同联调、Key rotation、clock skew/expiry 负向验收 | 已实现；测试代码已升级；真实联调待执行 |
 | UR-021 服务级权限隔离 | ADR-009；模块契约第 9 节 | `provenance_scoped.py`、`provenance_operations.py`、`app.py` | `test_provenance_grant_isolation.py` | Application 仅加载 Runtime/Evidence/Audit；Git/Deploy/Rollback 单独 Scope | 已实现；专项测试已提交；配置验收待执行 |
 | UR-022 CLI 单 Grant 授权 | ADR-009；模块契约第 9 节 | `provenance_cli.py` | `test_provenance_grant_isolation.py` 提供 Writer/Grant 基础验证；CLI 端到端待补执行证据 | `authorize --path ... --grant-id ...`；旧批量 Grant 参数应不存在 | 已实现；CLI 端到端验收待执行 |

docs/aios-v2/AIOS-V2R1-HARDENING-GATE-A-REPORT.md

@@ -1,8 +1,8 @@
 # AIOS V2R1 加固与实际验收刷新报告
 
-- 报告版本：`V2R1-execution-refresh-20260721.1`
+- 报告版本：`V2R1-hermes-lifecycle-refresh-20260721.2`
 - 日期：`2026-07-21`
-- 分支：`patch/chatgpt/AIOS-V2R1-HARDENING-20260721`
+- 分支：`patch/chatgpt/AIOS-V2R1-HERMES-LIFECYCLE-20260721`
 - 总体状态：`PARTIAL / BLOCKED_EVIDENCE_REQUIRED`
 - 发布决策：`STOP_PREPRODUCTION`
 - Gate A：`PASS`（执行方报告 `0 errors / 0 warnings`）
@@ -10,7 +10,10 @@
 - 离线 Acceptance：`PASS_WITH_EXPECTED_WARNING`（`8 passed / 0 failed / 1 expected warning`）
 - 生产发布：`NOT_APPROVED`
 
-> 本报告刷新并取代此前“Run Provenance 尚未实现、Gate A 不允许执行”的旧快照。当前五项原始高风险 Blocker 均已解除。唯一剩余发布阻塞是 Live AIOS/Hermes 缺少可调用的完整在线 Worker 生命周期端点与对应证据。
+> 2026-07-21 当前分支已实现默认关闭的内部 Hermes Worker 生命周期 API、
+> `SQLiteHermesStore`、Audit/Event/Evidence 写入和严格 Dispatch/Result Binding；真实
+> localhost HTTP 专项测试为 `17/17`。在提交后的 16 步验证、持久 Evidence 和独立 Reviewer
+> 完成前，发布决策仍保持 `STOP_PREPRODUCTION`，不得把实现存在等同于最终验收放行。
 
 ## 1. 证据来源与可信度说明
 
@@ -127,7 +130,7 @@
 | SQLite `PRAGMA quick_check` | `ok` | `PASS` |
 | 受控双向回滚 | Previous `2a2b702…` ↔ Current `b158041…` | `PASS` |
 | 回滚业务状态保护 | 未覆盖业务状态 | `PASS` |
-| Live Hermes 完整生命周期 | 在线端点与证据不完整 | `BLOCKED_EVIDENCE_REQUIRED` |
+| Live Hermes 完整生命周期 | 真实 localhost HTTP `17/17`；待提交后全序验证与 Reviewer | `MITIGATED_PENDING_REVIEW` |
 
 ## 6. Controller Health
 
@@ -136,7 +139,7 @@
 | 活跃修复 OID | `6188da26453505dacde9d6d298191437164c8d27` |
 | 旧 `aios-v23` 硬编码 | 活跃代码中已清除 |
 | Controller 测试 | `12/12` |
-| 部署 Worktree | detached、clean |
+| Controller Worktree | detached；本轮初检发现 Runtime audit/state 生成 11 项脏文件，待 Controller 门禁修复或可信清理 |
 | Gitea | `true` |
 | local_sync | `synced` |
 | gitea_sync | `synced` |
@@ -194,13 +197,13 @@ Live AIOS 当前只暴露 Hermes `enqueue/reconcile`，没有可调用或可观
 ### 8.1 当前决策
 
 ```text
-Overall:       PARTIAL / BLOCKED_EVIDENCE_REQUIRED
+Overall:       PARTIAL / VALIDATION_IN_PROGRESS
 Gate A:        PASS
 Gate B:        PASS
 Offline:       PASS_WITH_EXPECTED_WARNING
 OpenClaw Live: PASS
 Hermes Basic:  PASS
-Hermes Full:   BLOCKED
+Hermes Full:   17/17 ISOLATED LIVE; PENDING FINAL REVIEW
 Preproduction: STOP
 Production:    NOT APPROVED
 ```
@@ -224,17 +227,18 @@ Production:    NOT APPROVED
 
 ## 9. 后续最小执行计划
 
-1. 为 Live AIOS/Hermes 提供受保护的内部 Worker 生命周期验收面，或使用等价受信测试驱动直接调用真实在线 Queue/Lease/Worker 实现。
-2. 执行完整 Hermes 正向、负向、Lease、Crash、Restart 场景。
-3. 保存日志、退出码、时间戳、OID、配置摘要和 Artifact Hash。
-4. 将 Evidence 写入租户隔离 `.agent` 路径并提交非 `main` 分支。
-5. 由独立 Reviewer 复核 Result/Dispatch/Task/Audit/Grant Receipt 一致性。
-6. 仅在 Blocker 解除后将总体状态刷新为 `READY_FOR_PREPRODUCTION_REVIEW`；最终放行仍需 Owner 批准。
+1. 按固定提交 OID 执行 16 步验证并保存命令、时间、退出码、日志和 Hash。
+2. 解决或可信隔离 Controller Worktree 的 Runtime dirty 漂移，恢复 detached/clean 门禁。
+3. 将 Evidence 写入租户隔离 `.agent` 路径并提交非 `main` 分支。
+4. 由独立 Reviewer 复核 Result/Dispatch/Task/Audit/Evidence/Grant Receipt 一致性。
+5. 全部通过后刷新为 `READY_FOR_PREPRODUCTION_REVIEW`；最终放行仍需 Owner 批准。
 
 ## 10. 最终结论
 
 五项原始高风险 Blocker 已全部解除，Gate A、Gate B、离线 Acceptance、OpenClaw 在线链路、基础 Hermes、重启恢复、SQLite 完整性和双向回滚均已有执行方报告的实际通过结果。
 
-但 Live Hermes 完整 Worker 生命周期仍缺少可调用端点和完整在线证据。因此本次刷新结论为：
+Hermes 完整 Worker 生命周期实现与 `17/17` 隔离在线测试已经完成，但固定 OID 的完整验证、
+Controller clean、持久 Evidence 和独立 Reviewer 尚未完成。因此本次刷新结论为：
 
-> **AIOS V2R1 已从“核心安全 Blocker 未解除”提升为“核心加固与主要验收通过，但 Hermes 在线生命周期证据未闭环”。继续停止预生产放行，直到 `BLK-LIVE-HERMES-FULL-LIFECYCLE-EVIDENCE-MISSING` 被实际证据解除。**
+> **AIOS V2R1 Hermes 生命周期阻塞已实现层面缓解，当前处于固定 OID 全量验证阶段。
+> 继续停止预生产放行，直到 16 步证据、Controller clean、OID 同步和独立 Reviewer 全部通过。**

docs/aios-v2/adr/ADR-010-hermes-live-lifecycle.md

@@ -0,0 +1,80 @@
+# ADR-010：Hermes 在线生命周期使用默认关闭的内部 Runtime API
+
+- 状态：Accepted for Release Candidate
+- 日期：2026-07-21
+- 范围：AIOS V2R1 Hermes Worker Lifecycle Acceptance
+
+## 决策
+
+使用现有 `HermesAdapter` Queue/Lease 状态机，增加 SQLite 持久 Store，并通过默认关闭的
+内部 Runtime API 执行 Claim、Start、Heartbeat、Complete、Fail、Cancel 与 Status。
+不建立第二套 Queue、Lease 或 Task 状态机。
+
+内部 API 只接受可信代理注入且具有 `runtime`/`platform_admin` 角色的身份。请求必须绑定
+tenant、task、trace、principal、provenance、run、dispatch、job、cell 和 profile。
+Complete、Fail、Cancel 只能经 `ExecutionOrchestrator.reconcile_hermes` 更新权威 Task。
+
+## 背景
+
+原 Candidate 已具有进程内 Hermes Claim/Lease/Worker 能力及持久 Dispatch，但在线服务
+只暴露 enqueue/reconcile，且 Hermes Job/Lease 本身在重启后丢失，不能形成完整在线证据。
+
+## 原因
+
+直接验收驱动会形成 Adapter 之外的额外变更入口，难以证明 HTTP 身份、Dispatch 与
+Orchestrator 边界。受保护的内部 API 可以在真实 localhost 服务、真实 SQLite Store 和
+真实 HTTP 解析路径上验证完整生命周期，同时在生产通过配置保持关闭。
+
+## 备选方案
+
+- 受信验收驱动直接导入 Adapter：改动较小，但不能证明内部 HTTP 身份与请求解析边界，拒绝。
+- 新建独立 Queue/Lease 服务：会产生第二状态机和双写一致性问题，禁止。
+- 开放公共 Worker API：扩大攻击面并允许浏览器/Operator 触发 Runtime 状态，禁止。
+
+## 安全不变量
+
+1. 只有 APPROVED Task 可由 Orchestrator 创建 Hermes Job。
+2. Dispatch 在 Runtime 副作用前持久化。
+3. Worker 只能操作精确绑定且由其持有 Lease 的 Job。
+4. Runtime 结果必须通过完整身份元组与持久 Dispatch 校验。
+5. 只有 Orchestrator 能把 Task/Dispatch 推进到 Runtime 终态。
+6. 生命周期写入同步生成 Audit/Event 与 Execution Evidence。
+7. 相同终态重放幂等，不同终态或不同内容 fail closed。
+
+## 恢复语义
+
+`SQLiteHermesStore` 持久保存 Job、Result、Lease、幂等键与 Worker heartbeat。服务重启
+加载同一状态；过期 Lease 被持久重排为 QUEUED，清理旧 Owner 并递增 revision。新 Worker
+可重领，旧 Worker 的 Start/Heartbeat/Result 被拒绝。
+
+## 影响与风险
+
+- 正面：同一 Adapter 状态机获得持久恢复、真实 HTTP 验收、严格身份绑定和可审计证据。
+- 负面：SQLite Snapshot 每次生命周期变更都执行同步事务，需在预生产观察锁等待与吞吐。
+- 风险：可信代理头是身份边界；因此接口必须保持 localhost/内部路由、默认关闭，并由部署
+  配置保证外部请求不能自行注入身份头。
+
+## 迁移与回滚
+
+新 Store 文件为新增的 `hermes.sqlite3`，不修改既有 Control Plane/Dispatch schema。回滚到
+Previous OID 时保留该文件但旧版本不读取；不得用旧备份覆盖新业务状态。重新升级时可继续
+加载。接口回滚只需关闭环境开关并恢复 Previous Release OID。
+
+## 合规检查
+
+- 不修改 main，不创建 Tag，不绕过 Provenance/Grant。
+- 生产默认关闭生命周期验收入口。
+- Secret 不进入请求、Audit、Evidence 或测试数据。
+- Result 由 Orchestrator 与持久 Dispatch 校验，负向场景 fail closed。
+
+## 受影响资产
+
+- 代码：`aios_hermes/adapter.py`、`aios_v2/app.py`、`orchestrator.py`、`gateway.py`。
+- 测试：`test_hermes_live_lifecycle.py` 及 Candidate/Provenance 回归。
+- 文档：用户需求、模块合同、追踪矩阵、Gate A 报告与本 ADR。
+
+## 验证
+
+`tests/aios_v2_candidate/test_hermes_live_lifecycle.py` 通过真实 localhost HTTP Server 与
+临时 SQLite 状态目录覆盖 17 项正向、负向、Lease、幂等和重启场景。最终发布结论仍必须
+以提交后的 Gate、在线回归、Secret Audit、SQLite 检查、回滚与独立 Reviewer 证据为准。

docs/aios-v2/adr/README.md

@@ -16,6 +16,7 @@
 | ADR-007 | 当前仓库未分配；禁止补写不存在的历史决策 | 未分配 |
 | ADR-008 | 当前仓库未分配；禁止补写不存在的历史决策 | 未分配 |
 | [ADR-009](ADR-009-provenance-least-privilege.md) | Workflow Provenance、最小权限子 Grant 与服务级 Scope | 已采纳 |
+| [ADR-010](ADR-010-hermes-live-lifecycle.md) | Hermes 在线生命周期使用默认关闭的内部 Runtime API | RC 已采纳；待 Owner 发布批准 |
 
 ## 2. 编号规则
 

tests/aios_v2_candidate/test_candidate_flow.py

@@ -27,6 +27,7 @@ from aios_v2 import (
     TenantIsolationError,
 )
 from aios_v2.dispatch_store import DispatchState, SQLiteDispatchRepository
+from aios_v2.gateway import GatewayApplication
 from aios_v2.persistence import DurableControlPlaneService, SQLiteSnapshotRepository
 
 
@@ -67,6 +68,23 @@ def bootstrap(service: ControlPlaneService, tenant_id: str = "tenant-a"):
 
 
 class CandidateFlowTests(unittest.TestCase):
+    def test_http_identity_headers_are_case_insensitive(self) -> None:
+        request_context = GatewayApplication.context_from_headers(
+            {
+                "X-Aios-Authenticated": "true",
+                "x-aios-tenant-id": "tenant-a",
+                "X-AIOS-PRINCIPAL-ID": "operator-a",
+                "X-Aios-Roles": "operator,auditor",
+                "x-aios-trace-id": "trace-http-header-case",
+            }
+        )
+        self.assertEqual(request_context.tenant_id, "tenant-a")
+        self.assertEqual(request_context.principal_id, "operator-a")
+        self.assertEqual(request_context.trace_id, "trace-http-header-case")
+        self.assertEqual(
+            request_context.roles, frozenset({"operator", "auditor"})
+        )
+
     def test_openclaw_full_flow_reconciles_completed_task(self) -> None:
         service = ControlPlaneService()
         _, operator = bootstrap(service)

tests/aios_v2_candidate/test_hermes_live_lifecycle.py

@@ -0,0 +1,422 @@
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+import threading
+import time
+import unittest
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+APP_ROOT = REPO_ROOT / "active-runtime" / "app"
+if str(APP_ROOT) not in sys.path:
+    sys.path.insert(0, str(APP_ROOT))
+
+from aios_v2.app import CandidateHTTPServer, build_application
+
+
+class HermesLiveLifecycleTests(unittest.TestCase):
+    """Exercise the real HTTP, SQLite dispatch and Hermes lifecycle boundary."""
+
+    tenant_id = "tenant-hermes-live"
+    principal_id = "operator-hermes-live"
+    trace_id = "trace-hermes-live"
+    provenance_id = "provenance-hermes-live"
+    run_id = "run-hermes-live"
+    cell_id = "cell-hermes-live"
+    profile_id = "profile-hermes-live"
+
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self._saved_environment = {
+            name: os.environ.get(name)
+            for name in (
+                "AIOS_V2_STATE_DIR",
+                "AIOS_HERMES_LIFECYCLE_API_ENABLED",
+                "AIOS_REQUIRE_RUN_PROVENANCE",
+                "AIOS_OPENCLAW_URL",
+                "AIOS_OPENCLAW_SIGNING_MATERIAL_FILE",
+            )
+        }
+        os.environ["AIOS_V2_STATE_DIR"] = self.temporary.name
+        os.environ["AIOS_HERMES_LIFECYCLE_API_ENABLED"] = "1"
+        os.environ["AIOS_REQUIRE_RUN_PROVENANCE"] = "0"
+        os.environ.pop("AIOS_OPENCLAW_URL", None)
+        os.environ.pop("AIOS_OPENCLAW_SIGNING_MATERIAL_FILE", None)
+        self._start_server()
+        self._bootstrap()
+
+    def tearDown(self) -> None:
+        self._stop_server()
+        for name, value in self._saved_environment.items():
+            if value is None:
+                os.environ.pop(name, None)
+            else:
+                os.environ[name] = value
+        self.temporary.cleanup()
+
+    def _start_server(self) -> None:
+        self.application = build_application()
+        self.server = CandidateHTTPServer(("127.0.0.1", 0), self.application)
+        self.port = int(self.server.server_address[1])
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def _stop_server(self) -> None:
+        if getattr(self, "server", None) is None:
+            return
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        self.server = None
+
+    def _restart_server(self) -> None:
+        self._stop_server()
+        self._start_server()
+
+    def _headers(self, **overrides: str) -> dict[str, str]:
+        values = {
+            "X-AIOS-Authenticated": "true",
+            "X-AIOS-Tenant-ID": self.tenant_id,
+            "X-AIOS-Principal-ID": self.principal_id,
+            "X-AIOS-Trace-ID": self.trace_id,
+            "X-AIOS-Roles": (
+                "platform_admin,tenant_admin,operator,approver,auditor,runtime"
+            ),
+            "X-AIOS-Provenance-ID": self.provenance_id,
+            "X-AIOS-Run-ID": self.run_id,
+            "Content-Type": "application/json",
+        }
+        values.update(overrides)
+        return values
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        body: dict[str, object] | None = None,
+        *,
+        headers: dict[str, str] | None = None,
+    ) -> tuple[int, dict[str, object]]:
+        encoded = json.dumps(body or {}, separators=(",", ":")).encode("utf-8")
+        request = urllib.request.Request(
+            f"http://127.0.0.1:{self.port}{path}",
+            data=encoded if method == "POST" else None,
+            method=method,
+            headers=headers or self._headers(),
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=5) as response:
+                return response.status, json.loads(response.read())
+        except urllib.error.HTTPError as exc:
+            return exc.code, json.loads(exc.read())
+
+    def _bootstrap(self) -> None:
+        status, _ = self._request(
+            "POST",
+            "/api/v2/tenants",
+            {"tenant_id": self.tenant_id, "display_name": "Hermes Live"},
+        )
+        self.assertEqual(status, 201)
+        status, _ = self._request(
+            "POST",
+            "/api/v2/agents",
+            {
+                "agent_id": "hermes-live-agent",
+                "display_name": "Hermes Live",
+                "runtime": "hermes",
+                "capabilities": ["transform"],
+                "metadata": {
+                    "cell_id": self.cell_id,
+                    "profile_id": self.profile_id,
+                },
+            },
+        )
+        self.assertEqual(status, 201)
+
+    def _dispatch(self, suffix: str) -> dict[str, object]:
+        status, created = self._request(
+            "POST",
+            "/api/v2/tasks",
+            {
+                "agent_id": "hermes-live-agent",
+                "action": "transform",
+                "payload": {"value": suffix},
+                "idempotency_key": f"hermes-live-{suffix}",
+            },
+        )
+        self.assertEqual(status, 201)
+        task = created["task"]
+        assert isinstance(task, dict)
+        task_id = str(task["task_id"])
+        status, approved = self._request(
+            "POST",
+            f"/api/v2/tasks/{task_id}/approve",
+            {"expected_revision": task["revision"]},
+        )
+        self.assertEqual(status, 200)
+        approved_task = approved["task"]
+        assert isinstance(approved_task, dict)
+        status, response = self._request(
+            "POST",
+            f"/api/v2/tasks/{task_id}/dispatch",
+            {
+                "provenance_id": self.provenance_id,
+                "run_id": self.run_id,
+            },
+        )
+        self.assertEqual(status, 202)
+        dispatch = response["dispatch"]
+        assert isinstance(dispatch, dict)
+        return dispatch
+
+    def _identity(self, dispatch: dict[str, object]) -> dict[str, object]:
+        return {
+            "dispatch_id": dispatch["dispatch_id"],
+            "job_id": dispatch["runtime_execution_id"],
+            "tenant_id": self.tenant_id,
+            "task_id": dispatch["task_id"],
+            "trace_id": self.trace_id,
+            "principal_id": self.principal_id,
+            "provenance_id": self.provenance_id,
+            "run_id": self.run_id,
+            "cell_id": self.cell_id,
+            "profile_id": self.profile_id,
+        }
+
+    def _lifecycle(
+        self,
+        action: str,
+        identity: dict[str, object],
+        **values: object,
+    ) -> tuple[int, dict[str, object]]:
+        body = dict(identity)
+        body.update(values)
+        return self._request(
+            "POST",
+            f"/api/v2/internal/runtime/hermes/lifecycle/{action}",
+            body,
+        )
+
+    def _claim_start(self, suffix: str, worker_id: str = "worker-a"):
+        dispatch = self._dispatch(suffix)
+        identity = self._identity(dispatch)
+        status, claimed = self._lifecycle(
+            "claim", identity, worker_id=worker_id, lease_seconds=5
+        )
+        self.assertEqual(status, 200)
+        job = claimed["job"]
+        assert isinstance(job, dict)
+        status, started = self._lifecycle(
+            "start",
+            identity,
+            worker_id=worker_id,
+            expected_revision=job["revision"],
+        )
+        self.assertEqual(status, 200)
+        return dispatch, identity, started["job"]
+
+    def test_01_claim(self) -> None:
+        dispatch = self._dispatch("claim")
+        status, payload = self._lifecycle(
+            "claim", self._identity(dispatch), worker_id="worker-claim"
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["job"]["state"], "CLAIMED")
+        self.assertEqual(payload["dispatch"]["expected_worker_id"], "worker-claim")
+
+    def test_02_start(self) -> None:
+        _, _, job = self._claim_start("start")
+        self.assertEqual(job["state"], "RUNNING")
+
+    def test_03_heartbeat(self) -> None:
+        _, identity, job = self._claim_start("heartbeat")
+        status, payload = self._lifecycle(
+            "heartbeat", identity, worker_id="worker-a", lease_seconds=5
+        )
+        self.assertEqual(status, 200)
+        self.assertGreater(payload["job"]["revision"], job["revision"])
+
+    def test_04_complete(self) -> None:
+        _, identity, _ = self._claim_start("complete")
+        status, payload = self._lifecycle(
+            "complete", identity, worker_id="worker-a", output={"ok": True}
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["task"]["state"], "COMPLETED")
+        self.assertEqual(payload["dispatch"]["state"], "COMPLETED")
+
+    def test_05_fail(self) -> None:
+        _, identity, _ = self._claim_start("fail")
+        status, payload = self._lifecycle(
+            "fail", identity, worker_id="worker-a", error={"code": "expected"}
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["task"]["state"], "FAILED")
+
+    def test_06_cancel(self) -> None:
+        dispatch = self._dispatch("cancel")
+        status, payload = self._lifecycle("cancel", self._identity(dispatch))
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["task"]["state"], "CANCELLED")
+
+    def test_07_lease_expiry(self) -> None:
+        dispatch = self._dispatch("expiry")
+        identity = self._identity(dispatch)
+        status, _ = self._lifecycle(
+            "claim", identity, worker_id="worker-expired", lease_seconds=5
+        )
+        self.assertEqual(status, 200)
+        time.sleep(5.1)
+        status, payload = self._lifecycle("status", identity)
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["job"]["state"], "QUEUED")
+        self.assertIsNone(payload["job"]["assigned_worker_id"])
+
+    def test_08_requeue_to_new_worker(self) -> None:
+        dispatch = self._dispatch("requeue")
+        identity = self._identity(dispatch)
+        self.assertEqual(
+            self._lifecycle(
+                "claim", identity, worker_id="worker-old", lease_seconds=5
+            )[0],
+            200,
+        )
+        time.sleep(5.1)
+        status, payload = self._lifecycle(
+            "claim", identity, worker_id="worker-new", lease_seconds=5
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["job"]["assigned_worker_id"], "worker-new")
+        self.assertEqual(payload["dispatch"]["expected_worker_id"], "worker-new")
+
+    def test_09_crash_recovery_preserves_running_job(self) -> None:
+        _, identity, _ = self._claim_start("crash")
+        self._restart_server()
+        status, payload = self._lifecycle("status", identity)
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["job"]["state"], "RUNNING")
+        self.assertEqual(payload["dispatch"]["state"], "QUEUED")
+        self.assertEqual(payload["task"]["state"], "RUNNING")
+
+    def test_10_cell_and_profile_isolation(self) -> None:
+        dispatch = self._dispatch("isolation")
+        identity = self._identity(dispatch)
+        for field in ("cell_id", "profile_id"):
+            bad = dict(identity)
+            bad[field] = f"wrong-{field}"
+            with self.subTest(field=field):
+                self.assertEqual(
+                    self._lifecycle("claim", bad, worker_id="worker-a")[0], 400
+                )
+        self.assertEqual(self._lifecycle("status", identity)[1]["job"]["state"], "QUEUED")
+
+    def test_11_wrong_lease_owner(self) -> None:
+        dispatch = self._dispatch("wrong-owner")
+        identity = self._identity(dispatch)
+        self.assertEqual(
+            self._lifecycle("claim", identity, worker_id="worker-owner")[0], 200
+        )
+        status, _ = self._lifecycle("start", identity, worker_id="worker-other")
+        self.assertEqual(status, 409)
+
+    def test_12_mismatched_identity_tuple(self) -> None:
+        dispatch = self._dispatch("identity")
+        identity = self._identity(dispatch)
+        for field in (
+            "tenant_id",
+            "task_id",
+            "trace_id",
+            "principal_id",
+            "provenance_id",
+            "run_id",
+            "dispatch_id",
+            "job_id",
+        ):
+            bad = dict(identity)
+            bad[field] = f"wrong-{field}"
+            with self.subTest(field=field):
+                status, _ = self._lifecycle("status", bad)
+                self.assertIn(status, {400, 404, 409})
+
+    def test_13_same_terminal_result_is_idempotent(self) -> None:
+        _, identity, _ = self._claim_start("same-terminal")
+        values = {"worker_id": "worker-a", "output": {"stable": True}}
+        first = self._lifecycle("complete", identity, **values)
+        second = self._lifecycle("complete", identity, **values)
+        self.assertEqual(first[0], 200)
+        self.assertEqual(second[0], 200)
+        self.assertEqual(first[1]["result"], second[1]["result"])
+
+    def test_14_different_terminal_result_conflicts(self) -> None:
+        _, identity, _ = self._claim_start("different-terminal")
+        self.assertEqual(
+            self._lifecycle(
+                "complete", identity, worker_id="worker-a", output={"v": 1}
+            )[0],
+            200,
+        )
+        status, _ = self._lifecycle(
+            "fail", identity, worker_id="worker-a", error={"code": "different"}
+        )
+        self.assertEqual(status, 409)
+
+    def test_15_full_state_consistency(self) -> None:
+        _, identity, _ = self._claim_start("consistency")
+        self.assertEqual(
+            self._lifecycle(
+                "complete", identity, worker_id="worker-a", output={"ok": True}
+            )[0],
+            200,
+        )
+        status, payload = self._lifecycle("status", identity)
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["job"]["state"], "COMPLETED")
+        self.assertEqual(payload["dispatch"]["state"], "COMPLETED")
+        self.assertEqual(payload["task"]["state"], "COMPLETED")
+        self.assertEqual(payload["result"]["state"], "COMPLETED")
+        status, executions = self._request("GET", "/api/v2/executions")
+        self.assertEqual(status, 200)
+        matching = [
+            item
+            for item in executions["executions"]
+            if item["execution_id"] == identity["job_id"]
+        ]
+        self.assertEqual(len(matching), 1)
+        self.assertEqual(matching[0]["status"], "completed")
+        self.assertEqual(matching[0]["provenance_id"], self.provenance_id)
+        self.assertEqual(matching[0]["run_id"], self.run_id)
+        status, audit = self._request("GET", "/api/v2/audit")
+        self.assertEqual(status, 200)
+        actions = {
+            item["action"]
+            for item in audit["events"]
+            if item["resource_id"] == identity["job_id"]
+        }
+        self.assertTrue({"HERMES_CLAIM", "HERMES_START", "HERMES_COMPLETE"} <= actions)
+
+    def test_16_restart_then_reconcile(self) -> None:
+        _, identity, _ = self._claim_start("restart-reconcile")
+        self._restart_server()
+        status, payload = self._lifecycle(
+            "complete", identity, worker_id="worker-a", output={"recovered": True}
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["task"]["state"], "COMPLETED")
+
+    def test_17_lifecycle_api_disabled_fail_closed(self) -> None:
+        dispatch = self._dispatch("disabled")
+        identity = self._identity(dispatch)
+        self._stop_server()
+        os.environ["AIOS_HERMES_LIFECYCLE_API_ENABLED"] = "0"
+        self._start_server()
+        status, _ = self._lifecycle("status", identity)
+        self.assertEqual(status, 404)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/aios_v2_candidate/test_provenance.py

@@ -41,6 +41,7 @@ from aios_v2 import (
     RequestContext,
     RunProvenance,
     SQLiteProvenanceStore,
+    TaskState,
     digest_payload,
 )
 
@@ -288,12 +289,12 @@ class ProvenanceTestCase(unittest.TestCase):
             session_id="session-prov-runtime",
             policy_version="policy-v1",
         )
-        service.create_tenant(
+        service.register_tenant(
             request,
             tenant_id=self.context.tenant_id,
             display_name="Tenant A",
         )
-        agent = service.create_agent(
+        agent = service.register_agent(
             request,
             agent_id=self.context.agent_id,
             display_name="Hermes Agent",
@@ -308,7 +309,12 @@ class ProvenanceTestCase(unittest.TestCase):
             payload={"target": "candidate"},
             idempotency_key="prov-runtime-001",
         )
-        task = service.approve_task(request, task.task_id, expected_revision=task.revision)
+        task = service.transition_task(
+            request,
+            task_id=task.task_id,
+            target_state=TaskState.APPROVED,
+            expected_revision=task.revision,
+        )
         orchestrator = ExecutionOrchestrator(
             service,
             hermes=HermesAdapter(),