feat(aiops): expose host runaway loop readiness

apps/api/src/api/v1/agents.py

@@ -94,6 +94,9 @@ from src.services.ai_agent_receipt_readback_owner_review import (
 from src.services.ai_agent_report_no_write_analysis_runtime import (
     load_latest_ai_agent_report_no_write_analysis_runtime,
 )
+from src.services.host_runaway_aiops_loop_readiness import (
+    load_latest_host_runaway_aiops_loop_readiness,
+)
 from src.services.ai_agent_matched_playbook_learning_gap import (
     load_latest_ai_agent_matched_playbook_learning_gap,
 )
@@ -863,6 +866,35 @@ async def get_agent_report_no_write_analysis_runtime() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/agent-host-runaway-aiops-loop-readiness",
+    response_model=dict[str, Any],
+    summary="取得 P3-009 Host runaway AIOps loop readiness",
+    description=(
+        "讀取最新已提交的 P3-009 Host runaway AIOps loop readiness 只讀快照；"
+        "此端點只呈現 110 runaway process 的監控、告警、AI event packet、PlayBook、KM / Verifier "
+        "與 gated remediation 合約，不送 Telegram、不寫 Gateway queue、不呼叫 Bot API、不讀 secret、"
+        "不 kill process、不重啟 Docker / systemd / Nginx、不改 firewall、不執行 kubectl、不寫 production。"
+    ),
+)
+async def get_agent_host_runaway_aiops_loop_readiness() -> dict[str, Any]:
+    """回傳最新 P3-009 Host runaway AIOps loop readiness 只讀快照。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_host_runaway_aiops_loop_readiness)
+        return redact_public_lan_topology(payload)
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("host_runaway_aiops_loop_readiness_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="P3-009 Host runaway AIOps loop readiness 快照無效",
+        ) from exc
+
+
 @router.get(
     "/agent-communication-learning-contract",
     response_model=dict[str, Any],

apps/api/src/services/host_runaway_aiops_loop_readiness.py

@@ -0,0 +1,313 @@
+"""
+Host runaway AIOps loop readiness snapshot.
+
+This loader exposes a committed, read-only product surface for the 110 CPU
+runaway process loop. It validates that monitor, alert, event-packet, PlayBook,
+KM / Verifier, and gated remediation contracts are complete while every runtime
+write or process termination boundary remains closed.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_evaluations_dir
+
+_DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__))
+_SNAPSHOT_PATTERN = "host_runaway_aiops_loop_readiness_*.json"
+_SCHEMA_VERSION = "host_runaway_aiops_loop_readiness_v1"
+_RUNTIME_AUTHORITY = "host_runaway_aiops_loop_readiness_only_no_host_write"
+_EXPECTED_CURRENT_TASK = "P3-009"
+_EXPECTED_NEXT_TASK = "P3-010"
+_EXPECTED_ALERT_LANES = {
+    "HostOrphanBrowserSmokeHighCpu": "orphan_browser_smoke_runaway_process",
+    "HostCiRunnerLoadSaturation": "ci_runner_load_saturation",
+}
+_ZERO_ROLLUP_FIELDS = {
+    "runtime_remediation_authorized_count",
+    "telegram_send_count",
+    "gateway_queue_write_count",
+    "bot_api_call_count",
+    "host_write_count",
+    "process_termination_count",
+    "docker_restart_count",
+    "systemd_restart_count",
+    "nginx_reload_count",
+    "firewall_change_count",
+    "kubectl_action_count",
+    "production_write_count",
+}
+_FALSE_BOUNDARY_FLAGS = {
+    "runtime_remediation_enabled",
+    "process_termination_authorized",
+    "telegram_send_enabled",
+    "gateway_queue_write_enabled",
+    "bot_api_call_enabled",
+    "host_write_enabled",
+    "docker_restart_enabled",
+    "systemd_restart_enabled",
+    "nginx_reload_enabled",
+    "firewall_change_enabled",
+    "kubectl_action_enabled",
+    "production_write_enabled",
+    "secret_read_enabled",
+}
+_TRUE_BOUNDARY_FLAGS = {
+    "read_only_readback_allowed",
+    "ai_triage_packet_allowed",
+    "dry_run_generation_allowed",
+}
+_FORBIDDEN_PUBLIC_TERMS = {
+    "工作視窗",
+    "批准！",
+    "codex_delegation",
+    "source_thread_id",
+    "My request for Codex",
+    "authorization_header",
+    "secret_value",
+    "raw secret",
+    "private key",
+    "token value",
+    "chain_of_thought",
+}
+
+
+def load_latest_host_runaway_aiops_loop_readiness(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the newest committed host runaway AIOps loop readiness snapshot."""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(
+            f"no host runaway AIOps loop readiness snapshots found in {directory}"
+        )
+
+    latest = candidates[-1]
+    with latest.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{latest}: expected JSON object")
+
+    label = str(latest)
+    _require_schema(payload, label)
+    _require_rollups(payload, label)
+    _require_loop_stages(payload, label)
+    _require_alert_lanes(payload, label)
+    _require_asset_writeback_contract(payload, label)
+    _require_live_readback(payload, label)
+    _require_remediation_gate(payload, label)
+    _require_activation_boundaries(payload, label)
+    _require_no_forbidden_public_terms(payload, label)
+    return payload
+
+
+def _require_schema(payload: dict[str, Any], label: str) -> None:
+    if payload.get("schema_version") != _SCHEMA_VERSION:
+        raise ValueError(f"{label}: expected schema_version={_SCHEMA_VERSION}")
+
+    status = payload.get("program_status") or {}
+    expected = {
+        "overall_completion_percent": 100,
+        "current_priority": "P3",
+        "current_task_id": _EXPECTED_CURRENT_TASK,
+        "next_task_id": _EXPECTED_NEXT_TASK,
+        "read_only_mode": True,
+        "runtime_authority": _RUNTIME_AUTHORITY,
+    }
+    mismatches = _mismatches(status, expected)
+    if mismatches:
+        raise ValueError(f"{label}: program_status mismatch: {mismatches}")
+    if not status.get("status_note"):
+        raise ValueError(f"{label}: program_status.status_note is required")
+
+    source_refs = payload.get("source_refs") or []
+    if len(source_refs) != 7:
+        raise ValueError(f"{label}: source_refs must contain 7 refs")
+
+
+def _require_rollups(payload: dict[str, Any], label: str) -> None:
+    rollups = payload.get("rollups") or {}
+    expected_counts = {
+        "loop_stage_count": 6,
+        "alert_lane_count": 2,
+        "asset_writeback_contract_count": 5,
+        "source_ref_count": 7,
+        "live_readback_metric_count": 8,
+        "blocked_runtime_action_count": 12,
+    }
+    mismatches = _mismatches(rollups, expected_counts)
+    if mismatches:
+        raise ValueError(f"{label}: rollup counts mismatch: {mismatches}")
+
+    nonzero = {
+        field: rollups.get(field)
+        for field in _ZERO_ROLLUP_FIELDS
+        if rollups.get(field) != 0
+    }
+    if nonzero:
+        raise ValueError(f"{label}: zero rollup fields must remain 0: {nonzero}")
+
+
+def _require_loop_stages(payload: dict[str, Any], label: str) -> None:
+    stages = payload.get("loop_stages") or []
+    if len(stages) != 6:
+        raise ValueError(f"{label}: loop_stages must contain 6 stages")
+
+    required_stage_ids = {
+        "read_only_host_textfile_exporter",
+        "prometheus_alert_rules",
+        "telegram_ai_event_packet",
+        "playbook_contract",
+        "km_verifier_writeback_contract",
+        "gated_remediation_helper",
+    }
+    stage_ids = {stage.get("stage_id") for stage in stages}
+    missing = sorted(required_stage_ids - stage_ids)
+    if missing:
+        raise ValueError(f"{label}: missing loop stages: {missing}")
+
+    for stage in stages:
+        stage_id = stage.get("stage_id") or "<missing>"
+        if stage.get("completion_percent") != 100:
+            raise ValueError(f"{label}: stage {stage_id} must be 100 percent")
+        for field in ("display_name", "owner_agent", "status", "next_action"):
+            if not stage.get(field):
+                raise ValueError(f"{label}: stage {stage_id} missing {field}")
+        if not stage.get("evidence_refs"):
+            raise ValueError(f"{label}: stage {stage_id} missing evidence_refs")
+        if not stage.get("blocked_runtime_actions"):
+            raise ValueError(f"{label}: stage {stage_id} missing blocked_runtime_actions")
+
+
+def _require_alert_lanes(payload: dict[str, Any], label: str) -> None:
+    lanes = payload.get("alert_lanes") or []
+    if len(lanes) != 2:
+        raise ValueError(f"{label}: alert_lanes must contain 2 lanes")
+
+    lane_map = {lane.get("alertname"): lane for lane in lanes}
+    missing = sorted(set(_EXPECTED_ALERT_LANES) - set(lane_map))
+    if missing:
+        raise ValueError(f"{label}: missing alert lanes: {missing}")
+
+    for alertname, expected_lane in _EXPECTED_ALERT_LANES.items():
+        lane = lane_map[alertname]
+        if lane.get("lane_id") != expected_lane:
+            raise ValueError(f"{label}: {alertname} lane_id mismatch")
+        if lane.get("runtime_write_gate") != 0:
+            raise ValueError(f"{label}: {alertname} runtime_write_gate must be 0")
+        if lane.get("apply_allowed_without_owner_gate") is not False:
+            raise ValueError(f"{label}: {alertname} apply must require owner gate")
+        if not lane.get("next_action"):
+            raise ValueError(f"{label}: {alertname} next_action is required")
+
+
+def _require_asset_writeback_contract(payload: dict[str, Any], label: str) -> None:
+    contracts = payload.get("asset_writeback_contract") or []
+    if len(contracts) != 5:
+        raise ValueError(f"{label}: asset_writeback_contract must contain 5 contracts")
+
+    for contract in contracts:
+        asset_id = contract.get("asset_id") or "<missing>"
+        if contract.get("required_on_real_incident") is not True:
+            raise ValueError(f"{label}: {asset_id} must be required on real incidents")
+        if contract.get("live_write_enabled") is not False:
+            raise ValueError(f"{label}: {asset_id} live_write_enabled must remain false")
+        if not contract.get("required_fields"):
+            raise ValueError(f"{label}: {asset_id} required_fields must not be empty")
+
+
+def _require_live_readback(payload: dict[str, Any], label: str) -> None:
+    readback = payload.get("live_readback") or {}
+    expected = {
+        "host_label": "110",
+        "monitor_up": 1,
+        "orphan_browser_group_count": 0,
+        "active_ci_container_count": 2,
+        "remediation_authorized_count": 0,
+        "alerts_firing_count": 0,
+        "deploy_marker": "2d278568",
+        "argocd_sync": "Synced",
+        "argocd_health": "Healthy",
+        "production_route_count": 3,
+        "forbidden_public_hit_count": 0,
+    }
+    mismatches = _mismatches(readback, expected)
+    if mismatches:
+        raise ValueError(f"{label}: live_readback mismatch: {mismatches}")
+    if not str(readback.get("runtime_revision", "")).startswith("f358a0f6"):
+        raise ValueError(f"{label}: live_readback.runtime_revision must start with f358a0f6")
+
+
+def _require_remediation_gate(payload: dict[str, Any], label: str) -> None:
+    gate = payload.get("remediation_gate") or {}
+    expected = {
+        "dry_run_required": True,
+        "owner_approval_required": True,
+        "maintenance_window_required": True,
+        "evidence_ref_required": True,
+        "post_check_required": True,
+        "allowed_signal_after_gate": "SIGTERM",
+        "process_termination_authorized": False,
+    }
+    mismatches = _mismatches(gate, expected)
+    if mismatches:
+        raise ValueError(f"{label}: remediation_gate mismatch: {mismatches}")
+    if len(gate.get("disallowed_actions") or []) != 8:
+        raise ValueError(f"{label}: remediation_gate.disallowed_actions must contain 8 actions")
+
+
+def _require_activation_boundaries(payload: dict[str, Any], label: str) -> None:
+    boundaries = payload.get("activation_boundaries") or {}
+    true_mismatches = {
+        flag: boundaries.get(flag)
+        for flag in _TRUE_BOUNDARY_FLAGS
+        if boundaries.get(flag) is not True
+    }
+    if true_mismatches:
+        raise ValueError(f"{label}: true activation boundaries mismatch: {true_mismatches}")
+
+    false_mismatches = {
+        flag: boundaries.get(flag)
+        for flag in _FALSE_BOUNDARY_FLAGS
+        if boundaries.get(flag) is not False
+    }
+    if false_mismatches:
+        raise ValueError(f"{label}: false activation boundaries mismatch: {false_mismatches}")
+
+    for step in payload.get("next_steps") or []:
+        if step.get("runtime_write_allowed") is not False:
+            raise ValueError(f"{label}: next_steps runtime_write_allowed must remain false")
+
+
+def _require_no_forbidden_public_terms(payload: Any, label: str) -> None:
+    hits: list[str] = []
+
+    def walk(value: Any, path: str) -> None:
+        if isinstance(value, dict):
+            for key, nested in value.items():
+                walk(nested, f"{path}.{key}" if path else str(key))
+            return
+        if isinstance(value, list):
+            for index, nested in enumerate(value):
+                walk(nested, f"{path}[{index}]")
+            return
+        if isinstance(value, str):
+            for term in _FORBIDDEN_PUBLIC_TERMS:
+                if term.lower() in value.lower():
+                    hits.append(f"{path}: {term}")
+
+    walk(payload, "")
+    if hits:
+        raise ValueError(f"{label}: forbidden public terms: {hits[:5]}")
+
+
+def _mismatches(actual: dict[str, Any], expected: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    return {
+        key: {"expected": expected_value, "actual": actual.get(key)}
+        for key, expected_value in expected.items()
+        if actual.get(key) != expected_value
+    }

apps/api/tests/test_host_runaway_aiops_loop_readiness.py

@@ -0,0 +1,262 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from src.services.host_runaway_aiops_loop_readiness import (
+    load_latest_host_runaway_aiops_loop_readiness,
+)
+
+
+def test_load_latest_host_runaway_aiops_loop_readiness_reads_newest_file(tmp_path):
+    older = _snapshot(generated_at="2026-06-17T00:00:00+08:00")
+    newer = _snapshot(generated_at="2026-06-18T15:08:00+08:00")
+    (tmp_path / "host_runaway_aiops_loop_readiness_2026-06-17.json").write_text(
+        json.dumps(older),
+        encoding="utf-8",
+    )
+    _write_snapshot(tmp_path, newer)
+
+    loaded = load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+    assert loaded["generated_at"] == "2026-06-18T15:08:00+08:00"
+    assert loaded["program_status"]["current_task_id"] == "P3-009"
+    assert loaded["program_status"]["next_task_id"] == "P3-010"
+    assert loaded["rollups"]["loop_stage_count"] == 6
+    assert loaded["rollups"]["alert_lane_count"] == 2
+    assert loaded["rollups"]["asset_writeback_contract_count"] == 5
+    assert loaded["live_readback"]["host_label"] == "110"
+    assert loaded["live_readback"]["deploy_marker"] == "2d278568"
+    assert loaded["activation_boundaries"]["runtime_remediation_enabled"] is False
+
+
+def test_host_runaway_aiops_loop_readiness_requires_read_only_mode(tmp_path):
+    snapshot = _snapshot()
+    snapshot["program_status"]["read_only_mode"] = False
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="read_only_mode"):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def test_host_runaway_aiops_loop_readiness_blocks_process_termination(tmp_path):
+    snapshot = _snapshot()
+    snapshot["activation_boundaries"]["process_termination_authorized"] = True
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="false activation boundaries"):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def test_host_runaway_aiops_loop_readiness_blocks_host_write_rollup(tmp_path):
+    snapshot = _snapshot()
+    snapshot["rollups"]["host_write_count"] = 1
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="zero rollup fields"):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def test_host_runaway_aiops_loop_readiness_requires_alert_lane_contract(tmp_path):
+    snapshot = _snapshot()
+    snapshot["alert_lanes"][0]["runtime_write_gate"] = 1
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="runtime_write_gate"):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def test_host_runaway_aiops_loop_readiness_requires_asset_writeback_fields(tmp_path):
+    snapshot = _snapshot()
+    snapshot["asset_writeback_contract"][0]["required_fields"] = []
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="required_fields"):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def test_host_runaway_aiops_loop_readiness_rejects_private_terms(tmp_path):
+    snapshot = _snapshot()
+    snapshot["loop_stages"][0]["next_action"] = "不要顯示工作視窗內容"
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="forbidden public terms"):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def test_host_runaway_aiops_loop_readiness_fails_when_missing(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        load_latest_host_runaway_aiops_loop_readiness(tmp_path)
+
+
+def _write_snapshot(tmp_path, payload: dict) -> None:
+    (tmp_path / "host_runaway_aiops_loop_readiness_2026-06-18.json").write_text(
+        json.dumps(payload),
+        encoding="utf-8",
+    )
+
+
+def _snapshot(*, generated_at: str = "2026-06-18T15:08:00+08:00") -> dict:
+    return {
+        "schema_version": "host_runaway_aiops_loop_readiness_v1",
+        "generated_at": generated_at,
+        "program_status": {
+            "overall_completion_percent": 100,
+            "current_priority": "P3",
+            "current_task_id": "P3-009",
+            "next_task_id": "P3-010",
+            "read_only_mode": True,
+            "runtime_authority": "host_runaway_aiops_loop_readiness_only_no_host_write",
+            "status_note": "只讀 readiness。",
+        },
+        "source_refs": [
+            "scripts/ops/host-runaway-process-exporter.py",
+            "scripts/ops/host-runaway-process-remediation.py",
+            "ops/monitoring/alerts-unified.yml",
+            "apps/api/src/services/telegram_gateway.py",
+            "docs/runbooks/HOST-RUNAWAY-PROCESS-AIOPS-PLAYBOOK.md",
+            "docs/runbooks/FULL-STACK-COLD-START-SOP.md",
+            "docs/LOGBOOK.md",
+        ],
+        "rollups": {
+            "loop_stage_count": 6,
+            "alert_lane_count": 2,
+            "asset_writeback_contract_count": 5,
+            "source_ref_count": 7,
+            "live_readback_metric_count": 8,
+            "blocked_runtime_action_count": 12,
+            "runtime_remediation_authorized_count": 0,
+            "telegram_send_count": 0,
+            "gateway_queue_write_count": 0,
+            "bot_api_call_count": 0,
+            "host_write_count": 0,
+            "process_termination_count": 0,
+            "docker_restart_count": 0,
+            "systemd_restart_count": 0,
+            "nginx_reload_count": 0,
+            "firewall_change_count": 0,
+            "kubectl_action_count": 0,
+            "production_write_count": 0,
+        },
+        "loop_stages": [
+            _stage("read_only_host_textfile_exporter"),
+            _stage("prometheus_alert_rules"),
+            _stage("telegram_ai_event_packet"),
+            _stage("playbook_contract"),
+            _stage("km_verifier_writeback_contract"),
+            _stage("gated_remediation_helper"),
+        ],
+        "alert_lanes": [
+            {
+                "alertname": "HostOrphanBrowserSmokeHighCpu",
+                "lane_id": "orphan_browser_smoke_runaway_process",
+                "classification": "host_resource_runaway_process",
+                "action_policy": "triage_packet_then_dry_run_then_gated_sigterm",
+                "dry_run_allowed": True,
+                "apply_allowed_without_owner_gate": False,
+                "runtime_write_gate": 0,
+                "next_action": "prepare packet",
+            },
+            {
+                "alertname": "HostCiRunnerLoadSaturation",
+                "lane_id": "ci_runner_load_saturation",
+                "classification": "host_resource_capacity",
+                "action_policy": "capacity_triage_no_process_remediation",
+                "dry_run_allowed": False,
+                "apply_allowed_without_owner_gate": False,
+                "runtime_write_gate": 0,
+                "next_action": "capacity triage",
+            },
+        ],
+        "asset_writeback_contract": [
+            _asset("knowledge_base_incident_summary"),
+            _asset("playbook_trust_evidence"),
+            _asset("awooop_work_item_truth_chain"),
+            _asset("verifier_post_check"),
+            _asset("recurrence_guard"),
+        ],
+        "live_readback": {
+            "host_label": "110",
+            "monitor_up": 1,
+            "orphan_browser_group_count": 0,
+            "active_ci_container_count": 2,
+            "load5_per_core_upper_observed": 0.81,
+            "swap_used_ratio_upper_observed": 1.0,
+            "remediation_authorized_count": 0,
+            "alerts_firing_count": 0,
+            "deploy_marker": "2d278568",
+            "runtime_revision": "f358a0f6c3e614e407dedb6eee89bf10b2bc8173",
+            "argocd_sync": "Synced",
+            "argocd_health": "Healthy",
+            "production_route_count": 3,
+            "forbidden_public_hit_count": 0,
+        },
+        "remediation_gate": {
+            "dry_run_required": True,
+            "owner_approval_required": True,
+            "maintenance_window_required": True,
+            "evidence_ref_required": True,
+            "post_check_required": True,
+            "allowed_signal_after_gate": "SIGTERM",
+            "process_termination_authorized": False,
+            "disallowed_actions": [
+                "SIGKILL",
+                "docker restart",
+                "systemctl restart",
+                "nginx reload",
+                "firewall change",
+                "kubectl action",
+                "secret read",
+                "production write",
+            ],
+        },
+        "activation_boundaries": {
+            "read_only_readback_allowed": True,
+            "ai_triage_packet_allowed": True,
+            "dry_run_generation_allowed": True,
+            "runtime_remediation_enabled": False,
+            "process_termination_authorized": False,
+            "telegram_send_enabled": False,
+            "gateway_queue_write_enabled": False,
+            "bot_api_call_enabled": False,
+            "host_write_enabled": False,
+            "docker_restart_enabled": False,
+            "systemd_restart_enabled": False,
+            "nginx_reload_enabled": False,
+            "firewall_change_enabled": False,
+            "kubectl_action_enabled": False,
+            "production_write_enabled": False,
+            "secret_read_enabled": False,
+        },
+        "next_steps": [
+            {
+                "step_id": "real_alert_receipt_fixture",
+                "description": "verify packet",
+                "runtime_write_allowed": False,
+            }
+        ],
+    }
+
+
+def _stage(stage_id: str) -> dict:
+    return {
+        "stage_id": stage_id,
+        "display_name": stage_id,
+        "owner_agent": "openclaw",
+        "status": "ready",
+        "completion_percent": 100,
+        "evidence_refs": ["docs/LOGBOOK.md"],
+        "next_action": "keep readback visible",
+        "blocked_runtime_actions": ["host_write"],
+    }
+
+
+def _asset(asset_id: str) -> dict:
+    return {
+        "asset_id": asset_id,
+        "display_name": asset_id,
+        "required_on_real_incident": True,
+        "live_write_enabled": False,
+        "required_fields": ["incident_id"],
+    }

apps/api/tests/test_host_runaway_aiops_loop_readiness_api.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.agents import router
+
+
+def test_host_runaway_aiops_loop_readiness_endpoint_returns_committed_snapshot():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/agent-host-runaway-aiops-loop-readiness")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "host_runaway_aiops_loop_readiness_v1"
+    assert data["program_status"]["current_task_id"] == "P3-009"
+    assert data["program_status"]["next_task_id"] == "P3-010"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["rollups"]["loop_stage_count"] == len(data["loop_stages"]) == 6
+    assert data["rollups"]["alert_lane_count"] == len(data["alert_lanes"]) == 2
+    assert data["rollups"]["asset_writeback_contract_count"] == len(data["asset_writeback_contract"]) == 5
+    assert data["rollups"]["runtime_remediation_authorized_count"] == 0
+    assert data["rollups"]["process_termination_count"] == 0
+    assert data["rollups"]["host_write_count"] == 0
+    assert data["rollups"]["docker_restart_count"] == 0
+    assert data["live_readback"]["host_label"] == "110"
+    assert data["live_readback"]["monitor_up"] == 1
+    assert data["live_readback"]["orphan_browser_group_count"] == 0
+    assert data["live_readback"]["remediation_authorized_count"] == 0
+    assert data["live_readback"]["deploy_marker"] == "2d278568"
+    assert data["remediation_gate"]["dry_run_required"] is True
+    assert data["remediation_gate"]["owner_approval_required"] is True
+    assert data["remediation_gate"]["process_termination_authorized"] is False
+    assert data["activation_boundaries"]["runtime_remediation_enabled"] is False
+    assert data["activation_boundaries"]["process_termination_authorized"] is False
+    assert data["activation_boundaries"]["host_write_enabled"] is False