feat(governance): 新增 Status Cleanup Dashboard read model

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

@@ -73,6 +73,9 @@ from src.services.ai_agent_critic_reviewer_result_capture import (
 from src.services.ai_agent_deployment_layout import (
     load_latest_ai_agent_deployment_layout,
 )
+from src.services.awoooi_status_cleanup_dashboard import (
+    load_latest_awoooi_status_cleanup_dashboard,
+)
 from src.services.ai_agent_failure_receipt_no_send_replay import (
     load_latest_ai_agent_failure_receipt_no_send_replay,
 )
@@ -763,6 +766,36 @@ async def get_agent_deployment_layout() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/awoooi-status-cleanup-dashboard",
+    response_model=dict[str, Any],
+    summary="取得 AWOOOI 狀態清理儀表板",
+    description=(
+        "讀取最新已提交的 AWOOOI Status Cleanup Dashboard 只讀快照；"
+        "此端點只呈現狀態清理完成度、owner gate、apply gate、artifact sync 與 Wazuh handoff 邊界。"
+        "它不更新 project_current_status / memory、不同步 raw Codex App DB / auth / conversations / sessions、"
+        "不呼叫 Wazuh live API、不執行 active response、不改主機、不修改 workflow / repo refs、"
+        "不執行 backup / restore / migration。"
+    ),
+)
+async def get_awoooi_status_cleanup_dashboard() -> dict[str, Any]:
+    """回傳最新 AWOOOI 狀態清理儀表板只讀快照。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_awoooi_status_cleanup_dashboard)
+        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("awoooi_status_cleanup_dashboard_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AWOOOI 狀態清理儀表板快照無效",
+        ) from exc
+
+
 @router.get(
     "/agent-12-agent-war-room",
     response_model=dict[str, Any],

apps/api/src/services/awoooi_status_cleanup_dashboard.py

@@ -0,0 +1,276 @@
+"""
+AWOOOI Status Cleanup dashboard snapshot.
+
+Loads the read-only dashboard that combines status cleanup preflight, owner
+review, owner response preflight, dry-run execution plan, apply gate, and
+artifact sync readback. This module never writes memory, syncs raw Codex app
+data, fetches repos, modifies workflows, queries Wazuh live APIs, or authorizes
+runtime writes.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_operations_dir
+
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_SNAPSHOT_PATTERN = "awoooi-status-cleanup-dashboard.snapshot.json"
+_SCHEMA_VERSION = "awoooi_status_cleanup_dashboard_v1"
+
+
+def load_latest_awoooi_status_cleanup_dashboard(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the committed AWOOOI Status Cleanup dashboard snapshot."""
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(f"no AWOOOI status cleanup dashboard 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")
+    _require_schema(payload, _SCHEMA_VERSION, str(latest))
+    _require_read_only_boundaries(payload, str(latest))
+    _require_summary(payload, str(latest))
+    _require_cards(payload, str(latest))
+    _require_risk_controls(payload, str(latest))
+    _require_wazuh_boundary(payload, str(latest))
+    _require_hard_gates(payload, str(latest))
+    return payload
+
+
+def _require_schema(payload: dict[str, Any], expected: str, label: str) -> None:
+    actual = payload.get("schema_version")
+    if actual != expected:
+        raise ValueError(f"{label}: expected schema_version={expected}, got {actual!r}")
+
+
+def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None:
+    blocked_flags = {
+        "secret_values_collected",
+        "remote_write_performed",
+        "local_product_write_performed",
+        "execution_authorized",
+        "memory_write_authorized",
+        "wazuh_api_live_query_authorized",
+        "runtime_execution_authorized",
+        "ui_implementation_authorized",
+    }
+    invalid = sorted(flag for flag in blocked_flags if payload.get(flag) is not False)
+    if invalid:
+        raise ValueError(f"{label}: dashboard boundaries must remain false: {invalid}")
+
+    operation_boundaries = payload.get("operation_boundaries") or {}
+    invalid_ops = sorted(key for key, value in operation_boundaries.items() if value is not False)
+    if invalid_ops:
+        raise ValueError(f"{label}: operation boundaries must remain false: {invalid_ops}")
+
+
+def _require_summary(payload: dict[str, Any], label: str) -> None:
+    summary = payload.get("summary") or {}
+    gates = payload.get("gate_cards") or []
+    blockers = payload.get("blocking_reasons") or []
+    actions = payload.get("next_actions") or []
+    if payload.get("target_route") != "/workspace/status-cleanup":
+        raise ValueError(f"{label}: target_route must remain /workspace/status-cleanup")
+    if summary.get("dashboard_status") != "blocked_status_cleanup_apply_not_authorized":
+        raise ValueError(f"{label}: dashboard_status must remain blocked")
+    if summary.get("gate_count") != len(gates):
+        raise ValueError(f"{label}: gate_count mismatch")
+    if summary.get("blocked_gate_count") != sum(1 for item in gates if item.get("blocked") is True):
+        raise ValueError(f"{label}: blocked_gate_count mismatch")
+    if summary.get("blocking_reason_count") != len(blockers):
+        raise ValueError(f"{label}: blocking_reason_count mismatch")
+    if summary.get("next_action_count") != len(actions):
+        raise ValueError(f"{label}: next_action_count mismatch")
+    for flag in (
+        "apply_allowed",
+        "memory_write_authorized",
+        "wazuh_api_live_query_authorized",
+        "runtime_execution_authorized",
+        "ui_implementation_allowed",
+    ):
+        if summary.get(flag) is not False:
+            raise ValueError(f"{label}: summary.{flag} must remain false")
+    if summary.get("gate_count") != 5:
+        raise ValueError(f"{label}: dashboard must contain five status cleanup gates")
+    if summary.get("wazuh_handoff_status") != "blocked_not_released":
+        raise ValueError(f"{label}: Wazuh handoff status must remain blocked")
+    if summary.get("wazuh_handoff_base_commit") != "b540fc0c":
+        raise ValueError(f"{label}: Wazuh handoff base must remain b540fc0c")
+    if summary.get("wazuh_handoff_commit_count") != 7:
+        raise ValueError(f"{label}: Wazuh handoff commit count must be 7")
+    if summary.get("wazuh_handoff_patch_count") != 7:
+        raise ValueError(f"{label}: Wazuh handoff patch count must be 7")
+    zero_wazuh_summary = (
+        "wazuh_release_owner_request_sent_count",
+        "wazuh_release_owner_response_accepted_count",
+        "wazuh_acknowledged_release_owner_ack_count",
+        "wazuh_accepted_evidence_count",
+        "wazuh_live_metadata_owner_count",
+        "wazuh_secret_metadata_count",
+        "wazuh_live_agent_registry_readback",
+        "iwooos_wazuh_runtime_gate",
+        "wazuh_active_response_count",
+        "wazuh_agent_visibility_runtime_gate_count",
+        "wazuh_push_gate_count",
+        "wazuh_deploy_gate_count",
+        "wazuh_readback_gate_count",
+        "wazuh_runtime_gate_count",
+    )
+    non_zero_wazuh = sorted(key for key in zero_wazuh_summary if summary.get(key) != 0)
+    if non_zero_wazuh:
+        raise ValueError(f"{label}: Wazuh release summary gates must remain zero: {non_zero_wazuh}")
+
+
+def _require_cards(payload: dict[str, Any], label: str) -> None:
+    gate_ids = {item.get("gate_id") for item in payload.get("gate_cards") or []}
+    required_gates = {
+        "status_cleanup_preflight",
+        "owner_review_package",
+        "owner_response_preflight",
+        "execution_plan",
+        "apply_gate",
+    }
+    missing_gates = sorted(required_gates - gate_ids)
+    if missing_gates:
+        raise ValueError(f"{label}: gate_cards missing required gates: {missing_gates}")
+
+    metric_ids = {item.get("metric_id") for item in payload.get("metric_cards") or []}
+    required_metrics = {
+        "overall_completion",
+        "project_status_age",
+        "owner_flags",
+        "approved_sections",
+        "execution_preview",
+        "apply_confirmation",
+        "artifact_sync",
+    }
+    missing_metrics = sorted(required_metrics - metric_ids)
+    if missing_metrics:
+        raise ValueError(f"{label}: metric_cards missing required metrics: {missing_metrics}")
+
+    unsafe = [
+        item.get("gate_id")
+        for item in payload.get("gate_cards") or []
+        if item.get("memory_write_authorized") is not False
+        or item.get("execution_authorized") is not False
+    ]
+    if unsafe:
+        raise ValueError(f"{label}: gate cards must remain read-only: {sorted(unsafe)}")
+
+
+def _require_risk_controls(payload: dict[str, Any], label: str) -> None:
+    control_ids = {item.get("control_id") for item in payload.get("risk_controls") or []}
+    required = {
+        "memory_write",
+        "raw_codex_history",
+        "wazuh_runtime",
+        "workflow",
+        "repo_refs",
+        "host",
+        "backup_restore",
+        "ui",
+    }
+    missing = sorted(required - control_ids)
+    if missing:
+        raise ValueError(f"{label}: risk_controls missing controls: {missing}")
+    unsafe = sorted(
+        item.get("control_id")
+        for item in payload.get("risk_controls") or []
+        if item.get("authorized") is not False
+    )
+    if unsafe:
+        raise ValueError(f"{label}: risk controls must not authorize actions: {unsafe}")
+
+
+def _require_wazuh_boundary(payload: dict[str, Any], label: str) -> None:
+    handoff = payload.get("wazuh_handoff") or {}
+    if handoff.get("status") != "blocked_not_released":
+        raise ValueError(f"{label}: Wazuh handoff status must remain blocked_not_released")
+    if handoff.get("base_commit") != "b540fc0c":
+        raise ValueError(f"{label}: Wazuh handoff base must remain b540fc0c")
+    if handoff.get("commit_count") != 7:
+        raise ValueError(f"{label}: Wazuh handoff commit_count must be 7")
+    if handoff.get("patch_count") != 7:
+        raise ValueError(f"{label}: Wazuh handoff patch_count must be 7")
+    zero_handoff_keys = (
+        "release_owner_request_sent_count",
+        "release_owner_response_accepted_count",
+        "acknowledged_release_owner_ack_count",
+        "accepted_evidence_count",
+        "live_metadata_owner_count",
+        "secret_metadata_count",
+        "wazuh_live_agent_registry_readback",
+        "iwooos_wazuh_runtime_gate",
+        "active_response",
+        "push_gate_count",
+        "deploy_gate_count",
+        "readback_gate_count",
+        "runtime_gate_count",
+    )
+    non_zero_handoff = sorted(key for key in zero_handoff_keys if handoff.get(key) != 0)
+    if non_zero_handoff:
+        raise ValueError(f"{label}: Wazuh handoff gates must remain zero: {non_zero_handoff}")
+    if handoff.get("runtime_execution_authorized") is not False:
+        raise ValueError(f"{label}: Wazuh runtime execution must remain false")
+    if handoff.get("wazuh_api_live_query_authorized") is not False:
+        raise ValueError(f"{label}: Wazuh live query must remain false")
+    if handoff.get("stored_api_secret_metadata_changed") is not False:
+        raise ValueError(f"{label}: Wazuh stored API secret metadata must remain unchanged")
+    if handoff.get("agent_visibility_status") != "blocked_waiting_manager_agent_registry_readback":
+        raise ValueError(f"{label}: Wazuh agent visibility must remain blocked")
+    visibility_false = (
+        "manager_agent_registry_readback_passed",
+        "iwooos_live_route_readback_passed",
+        "dashboard_agent_list_recovered",
+    )
+    invalid_visibility = sorted(key for key in visibility_false if handoff.get(key) is not False)
+    if invalid_visibility:
+        raise ValueError(f"{label}: Wazuh visibility flags must remain false: {invalid_visibility}")
+    if handoff.get("agent_visibility_runtime_gate_count") != 0:
+        raise ValueError(f"{label}: Wazuh visibility runtime gate count must remain zero")
+    boundary = str(handoff.get("boundary", ""))
+    for token in (
+        "base=b540fc0c",
+        "38dc3c2f",
+        "9a53d3e1",
+        "e9972d47",
+        "758d419e",
+        "04db4b8a",
+        "8eec298e",
+        "325f262a",
+        "release_lane_preflight=ready0_acks0of6_evidence0of6_push0_deploy0_readback0_runtime0",
+        "owner_gate=request_sent0_response_accepted0_acks0of6_evidence0of6_push0_deploy0_readback0_runtime0",
+        "live_metadata_env_gate=owner0_secret_metadata0_push0_deploy0_readback0_runtime0",
+        "wazuh_live_agent_registry_readback=0",
+        "iwooos_wazuh_runtime_gate=0",
+        "active_response=0",
+        "manager_agent_registry_readback_passed=false",
+        "iwooos_live_route_readback_passed=false",
+        "dashboard_agent_list_recovered=false",
+        "push_blocked",
+    ):
+        if token not in boundary:
+            raise ValueError(f"{label}: Wazuh handoff boundary missing {token}")
+
+
+def _require_hard_gates(payload: dict[str, Any], label: str) -> None:
+    gate_text = "\n".join(payload.get("hard_gates") or [])
+    for token in (
+        "project_current_status",
+        "memory_write_authorized=false",
+        "raw Codex App",
+        "Wazuh",
+        ".gitea/workflows",
+        "apps/web",
+    ):
+        if token not in gate_text:
+            raise ValueError(f"{label}: hard_gates missing {token}")

apps/api/src/services/snapshot_paths.py

@@ -30,3 +30,8 @@ def resolve_repo_root(anchor: Path) -> Path:
 def default_evaluations_dir(anchor: Path) -> Path:
     """Resolve the default committed evaluations snapshot directory."""
     return resolve_repo_root(anchor) / "docs" / "evaluations"
+
+
+def default_operations_dir(anchor: Path) -> Path:
+    """Resolve the default committed operations snapshot directory."""
+    return resolve_repo_root(anchor) / "docs" / "operations"