feat(governance): 新增 12-Agent War Room 讀回

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

@@ -82,6 +82,9 @@ from src.services.ai_agent_learning_writeback_approval_package import (
 from src.services.ai_agent_live_read_model_gate import (
     load_latest_ai_agent_live_read_model_gate,
 )
+from src.services.ai_agent_12_agent_war_room import (
+    load_latest_ai_agent_12_agent_war_room,
+)
 from src.services.ai_agent_matched_playbook_learning_gap import (
     load_latest_ai_agent_matched_playbook_learning_gap,
 )
@@ -718,6 +721,36 @@ async def get_agent_deployment_layout() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/agent-12-agent-war-room",
+    response_model=dict[str, Any],
+    summary="取得 AI Agent 12-Agent War Room 快照",
+    description=(
+        "讀取最新已提交的 12-Agent War Room 只讀快照；"
+        "此端點只呈現 OpenClaw、Hermes、NemoTron、SRE、Security、DevOps、Data/DR、"
+        "Supply Chain、Product/UI、QA、Market Scout、Telegram Ops 的分工、工作量、阻擋項與批准邊界，"
+        "不開 runtime writer、不送 Telegram、不呼叫 Bot API、不安裝 SDK、不呼叫付費 API、"
+        "不讀 secret、不執行 production write。"
+    ),
+)
+async def get_agent_12_agent_war_room() -> dict[str, Any]:
+    """回傳最新 12-Agent War Room 只讀快照。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_ai_agent_12_agent_war_room)
+        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("ai_agent_12_agent_war_room_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AI Agent 12-Agent War Room 快照無效",
+        ) from exc
+
+
 @router.get(
     "/agent-communication-learning-contract",
     response_model=dict[str, Any],

apps/api/src/services/ai_agent_12_agent_war_room.py

@@ -0,0 +1,203 @@
+"""
+AI Agent 12-Agent War Room 快照。
+
+讀取最新已提交的 War Room 只讀回報，把 12 位邏輯 Agent 的分工、
+工作量、報告合約、市場觀測合約與 Telegram 邊界產品化；本模組不開
+runtime writer、不送 Telegram、不呼叫 Bot API、不安裝 SDK、不呼叫付費
+API、不讀 secret、不寫 production，也不執行破壞性操作。
+"""
+
+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 = "ai_agent_12_agent_war_room_*.json"
+_SCHEMA_VERSION = "ai_agent_12_agent_war_room_v1"
+_RUNTIME_AUTHORITY = "12_agent_war_room_read_only_no_live_write"
+_EXPECTED_AGENT_IDS = {
+    "agent_01_openclaw_arbiter",
+    "agent_02_hermes_rag",
+    "agent_03_nemotron_replay",
+    "agent_04_sre_sentinel",
+    "agent_05_security_sentinel",
+    "agent_06_devops_commander",
+    "agent_07_data_dr_guardian",
+    "agent_08_supply_chain_scout",
+    "agent_09_product_ui_curator",
+    "agent_10_qa_verifier",
+    "agent_11_market_scout",
+    "agent_12_telegram_ops_liaison",
+}
+_ZERO_FIELDS = {
+    "live_write_count",
+    "telegram_send_count",
+    "bot_api_call_count",
+    "production_write_count",
+    "paid_api_call_count",
+    "sdk_install_count",
+    "secret_read_count",
+    "destructive_operation_count",
+}
+_FORBIDDEN_PUBLIC_TERMS = {
+    "work_window_transcript",
+    "chain-of-thought",
+    "source_thread_id",
+    "browser_context",
+    "telegram_token",
+    "authorization header",
+}
+
+
+def load_latest_ai_agent_12_agent_war_room(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """讀取最新已提交的 12-Agent War Room 只讀快照。"""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(f"no AI Agent 12-Agent War Room 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_agent_roles(payload, label)
+    _require_rollups(payload, label)
+    _require_contracts(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 = {
+        "current_priority": "P1",
+        "current_task_id": "P2-142",
+        "next_task_id": "P2-143",
+        "read_only_mode": True,
+        "runtime_authority": _RUNTIME_AUTHORITY,
+        "overall_completion_percent": 72,
+    }
+    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")
+
+
+def _require_agent_roles(payload: dict[str, Any], label: str) -> None:
+    roles = payload.get("agent_roles") or []
+    if len(roles) != 12:
+        raise ValueError(f"{label}: expected exactly 12 agent roles")
+
+    role_ids = {str(role.get("agent_id")) for role in roles}
+    if role_ids != _EXPECTED_AGENT_IDS:
+        missing = sorted(_EXPECTED_AGENT_IDS - role_ids)
+        extra = sorted(role_ids - _EXPECTED_AGENT_IDS)
+        raise ValueError(f"{label}: agent ids mismatch missing={missing} extra={extra}")
+
+    for role in roles:
+        role_id = role.get("agent_id")
+        if role.get("review_status") != "read_only_review_completed":
+            raise ValueError(f"{label}: {role_id} must remain read_only_review_completed")
+        for field in ("live_write_count", "telegram_send_count", "bot_api_call_count"):
+            if role.get(field) != 0:
+                raise ValueError(f"{label}: {role_id}.{field} must remain zero")
+        for field in ("display_name", "war_room_role", "next_action"):
+            if not role.get(field):
+                raise ValueError(f"{label}: {role_id}.{field} is required")
+        if not isinstance(role.get("work_units"), int) or role["work_units"] <= 0:
+            raise ValueError(f"{label}: {role_id}.work_units must be positive")
+
+
+def _require_rollups(payload: dict[str, Any], label: str) -> None:
+    roles = payload.get("agent_roles") or []
+    rollups = payload.get("rollups") or {}
+    expected = {
+        "agent_role_count": len(roles),
+        "read_only_review_completed_count": sum(
+            1 for role in roles if role.get("review_status") == "read_only_review_completed"
+        ),
+        "subagent_batch_limit": 6,
+        "subagent_batch_count": 2,
+        "approval_required_total": sum(int(role.get("approval_required_count") or 0) for role in roles),
+        "blocker_total": sum(int(role.get("blocker_count") or 0) for role in roles),
+        "total_work_units": sum(int(role.get("work_units") or 0) for role in roles),
+        "total_evidence_items": sum(int(role.get("evidence_items") or 0) for role in roles),
+    }
+    mismatches = _mismatches(rollups, expected)
+    if mismatches:
+        raise ValueError(f"{label}: rollups mismatch: {mismatches}")
+
+    for field in _ZERO_FIELDS:
+        if rollups.get(field) != 0:
+            raise ValueError(f"{label}: rollups.{field} must remain zero")
+
+
+def _require_contracts(payload: dict[str, Any], label: str) -> None:
+    coordination = payload.get("coordination_model") or {}
+    if coordination.get("logical_agent_count") != 12:
+        raise ValueError(f"{label}: coordination_model.logical_agent_count must be 12")
+    if coordination.get("subagent_batch_limit") != 6:
+        raise ValueError(f"{label}: coordination_model.subagent_batch_limit must be 6")
+    if coordination.get("arbiter") != "openclaw":
+        raise ValueError(f"{label}: coordination_model.arbiter must remain openclaw")
+
+    telegram = payload.get("telegram_contract") or {}
+    for field in ("direct_send_allowed", "bot_api_call_allowed", "success_immediate_send_allowed"):
+        if telegram.get(field) is not False:
+            raise ValueError(f"{label}: telegram_contract.{field} must remain false")
+    for field in ("dedup_required", "receipt_required"):
+        if telegram.get(field) is not True:
+            raise ValueError(f"{label}: telegram_contract.{field} must remain true")
+
+    redaction = payload.get("display_redaction_contract") or {}
+    expected_redaction = {
+        "redaction_required": True,
+        "conversation_transcript_display_allowed": False,
+        "raw_prompt_display_allowed": False,
+        "private_reasoning_display_allowed": False,
+        "secret_value_display_allowed": False,
+        "raw_runtime_payload_display_allowed": False,
+    }
+    mismatches = _mismatches(redaction, expected_redaction)
+    if mismatches:
+        raise ValueError(f"{label}: display_redaction_contract mismatch: {mismatches}")
+
+    reporting = payload.get("reporting_contract") or {}
+    for cadence in ("daily", "weekly", "monthly"):
+        if (reporting.get(cadence) or {}).get("required") is not True:
+            raise ValueError(f"{label}: reporting_contract.{cadence}.required must be true")
+
+    market = payload.get("market_watch_contract") or {}
+    candidates = market.get("p0_refresh_candidates") or []
+    if len(candidates) < 5:
+        raise ValueError(f"{label}: market_watch_contract.p0_refresh_candidates must include at least 5 entries")
+
+
+def _require_no_forbidden_public_terms(payload: dict[str, Any], label: str) -> None:
+    public_text = json.dumps(payload, ensure_ascii=False).lower()
+    leaked = sorted(term for term in _FORBIDDEN_PUBLIC_TERMS if term.lower() in public_text)
+    if leaked:
+        raise ValueError(f"{label}: forbidden public terms leaked: {leaked}")
+
+
+def _mismatches(payload: dict[str, Any], expected: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    return {
+        key: {"expected": expected_value, "actual": payload.get(key)}
+        for key, expected_value in expected.items()
+        if payload.get(key) != expected_value
+    }

apps/api/tests/test_ai_agent_12_agent_war_room.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import copy
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.services.ai_agent_12_agent_war_room import (
+    load_latest_ai_agent_12_agent_war_room,
+)
+
+
+def test_load_latest_ai_agent_12_agent_war_room_snapshot() -> None:
+    snapshot = load_latest_ai_agent_12_agent_war_room()
+
+    assert snapshot["schema_version"] == "ai_agent_12_agent_war_room_v1"
+    assert snapshot["program_status"]["current_task_id"] == "P2-142"
+    assert snapshot["program_status"]["next_task_id"] == "P2-143"
+    assert snapshot["program_status"]["overall_completion_percent"] == 72
+    assert snapshot["program_status"]["runtime_authority"] == "12_agent_war_room_read_only_no_live_write"
+    assert len(snapshot["agent_roles"]) == 12
+
+    rollups = snapshot["rollups"]
+    assert rollups["agent_role_count"] == 12
+    assert rollups["read_only_review_completed_count"] == 12
+    assert rollups["subagent_batch_limit"] == 6
+    assert rollups["subagent_batch_count"] == 2
+    assert rollups["total_work_units"] == 82
+    assert rollups["total_evidence_items"] == 84
+    assert rollups["approval_required_total"] == 61
+    assert rollups["blocker_total"] == 54
+    assert rollups["live_write_count"] == 0
+    assert rollups["telegram_send_count"] == 0
+    assert rollups["bot_api_call_count"] == 0
+    assert rollups["production_write_count"] == 0
+    assert rollups["paid_api_call_count"] == 0
+    assert rollups["sdk_install_count"] == 0
+
+
+def test_war_room_contract_keeps_reports_market_and_telegram_read_only() -> None:
+    snapshot = load_latest_ai_agent_12_agent_war_room()
+
+    assert snapshot["coordination_model"]["arbiter"] == "openclaw"
+    assert snapshot["reporting_contract"]["daily"]["required"] is True
+    assert snapshot["reporting_contract"]["weekly"]["required"] is True
+    assert snapshot["reporting_contract"]["monthly"]["required"] is True
+    assert len(snapshot["market_watch_contract"]["p0_refresh_candidates"]) == 5
+    assert snapshot["telegram_contract"]["direct_send_allowed"] is False
+    assert snapshot["telegram_contract"]["bot_api_call_allowed"] is False
+    assert snapshot["telegram_contract"]["dedup_required"] is True
+    assert snapshot["telegram_contract"]["receipt_required"] is True
+    assert snapshot["display_redaction_contract"]["conversation_transcript_display_allowed"] is False
+    assert snapshot["display_redaction_contract"]["secret_value_display_allowed"] is False
+
+
+def test_rejects_missing_agent_role(tmp_path: Path) -> None:
+    snapshot = copy.deepcopy(load_latest_ai_agent_12_agent_war_room())
+    snapshot["agent_roles"] = snapshot["agent_roles"][:-1]
+    snapshot["rollups"]["agent_role_count"] = 11
+    snapshot["rollups"]["read_only_review_completed_count"] = 11
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="expected exactly 12 agent roles"):
+        load_latest_ai_agent_12_agent_war_room(tmp_path)
+
+
+def test_rejects_role_live_write_drift(tmp_path: Path) -> None:
+    snapshot = copy.deepcopy(load_latest_ai_agent_12_agent_war_room())
+    snapshot["agent_roles"][0]["live_write_count"] = 1
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="live_write_count must remain zero"):
+        load_latest_ai_agent_12_agent_war_room(tmp_path)
+
+
+def test_rejects_telegram_direct_send_drift(tmp_path: Path) -> None:
+    snapshot = copy.deepcopy(load_latest_ai_agent_12_agent_war_room())
+    snapshot["telegram_contract"]["direct_send_allowed"] = True
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="direct_send_allowed must remain false"):
+        load_latest_ai_agent_12_agent_war_room(tmp_path)
+
+
+def test_rejects_forbidden_public_terms(tmp_path: Path) -> None:
+    snapshot = copy.deepcopy(load_latest_ai_agent_12_agent_war_room())
+    snapshot["agent_roles"][0]["next_action"] = "blocked source_thread_id leak"
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="forbidden public terms leaked"):
+        load_latest_ai_agent_12_agent_war_room(tmp_path)
+
+
+def _write_snapshot(directory: Path, payload: dict) -> None:
+    path = directory / "ai_agent_12_agent_war_room_2099-01-01.json"
+    path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")

apps/api/tests/test_ai_agent_12_agent_war_room_api.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import os
+
+from fastapi.testclient import TestClient
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.main import app
+
+
+def test_ai_agent_12_agent_war_room_endpoint() -> None:
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/agent-12-agent-war-room")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["schema_version"] == "ai_agent_12_agent_war_room_v1"
+    assert payload["program_status"]["current_task_id"] == "P2-142"
+    assert payload["program_status"]["next_task_id"] == "P2-143"
+    assert payload["program_status"]["overall_completion_percent"] == 72
+    assert payload["program_status"]["runtime_authority"] == "12_agent_war_room_read_only_no_live_write"
+    assert payload["rollups"]["agent_role_count"] == 12
+    assert payload["rollups"]["read_only_review_completed_count"] == 12
+    assert payload["rollups"]["subagent_batch_limit"] == 6
+    assert payload["rollups"]["subagent_batch_count"] == 2
+    assert payload["rollups"]["total_work_units"] == 82
+    assert payload["rollups"]["approval_required_total"] == 61
+    assert payload["rollups"]["blocker_total"] == 54
+    assert payload["rollups"]["live_write_count"] == 0
+    assert payload["rollups"]["telegram_send_count"] == 0
+    assert payload["rollups"]["bot_api_call_count"] == 0
+    assert payload["rollups"]["production_write_count"] == 0
+    assert payload["telegram_contract"]["direct_send_allowed"] is False
+    assert payload["display_redaction_contract"]["conversation_transcript_display_allowed"] is False