feat(governance): 接入 Agent live read model gate

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

@@ -64,6 +64,9 @@ from src.services.ai_agent_host_stateful_version_inventory import (
 from src.services.ai_agent_interaction_learning_proof import (
     load_latest_ai_agent_interaction_learning_proof,
 )
+from src.services.ai_agent_live_read_model_gate import (
+    load_latest_ai_agent_live_read_model_gate,
+)
 from src.services.ai_agent_proactive_operations_contract import (
     load_latest_ai_agent_proactive_operations_contract,
 )
@@ -602,6 +605,33 @@ async def get_agent_interaction_learning_proof() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/agent-live-read-model-gate",
+    response_model=dict[str, Any],
+    summary="取得 AI Agent live read model gate",
+    description=(
+        "讀取最新已提交的 AgentSession / Redis Streams live read model gate；"
+        "此端點不連 DB、不讀寫 Redis、不啟動 worker、不建立 DB migration、不送 Telegram、"
+        "不顯示工作視窗內容、Agent 原始輸出、提示詞、私有推理或機密值。"
+    ),
+)
+async def get_agent_live_read_model_gate() -> dict[str, Any]:
+    """Return the latest read-only AI Agent live read model gate snapshot."""
+    try:
+        return await asyncio.to_thread(load_latest_ai_agent_live_read_model_gate)
+    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_live_read_model_gate_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AI Agent live read model gate 無效",
+        ) from exc
+
+
 @router.get(
     "/agent-proactive-operations-contract",
     response_model=dict[str, Any],

apps/api/src/services/ai_agent_live_read_model_gate.py

@@ -0,0 +1,217 @@
+"""
+AI Agent live read model gate snapshot.
+
+Loads the latest committed, read-only P2-403B gate for the AgentSession /
+Redis Streams live read model. This module only validates the approval package;
+it never opens a database session, starts workers, creates migrations, reads
+Redis consumer groups, sends Telegram messages, or exposes raw Agent outputs.
+"""
+
+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_live_read_model_gate_*.json"
+_SCHEMA_VERSION = "ai_agent_live_read_model_gate_v1"
+
+
+def load_latest_ai_agent_live_read_model_gate(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the newest committed AI Agent live read model gate snapshot."""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(
+            f"no AI Agent live read model gate 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")
+    _require_schema(payload, _SCHEMA_VERSION, str(latest))
+    _require_read_only_authority(payload, str(latest))
+    _require_storage_safety(payload, str(latest))
+    _require_redis_safety(payload, str(latest))
+    _require_no_write_smoke(payload, str(latest))
+    _require_rollup_consistency(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_authority(payload: dict[str, Any], label: str) -> None:
+    program_status = payload.get("program_status") or {}
+    if program_status.get("read_only_mode") is not True:
+        raise ValueError(f"{label}: program_status.read_only_mode must be true")
+    if program_status.get("runtime_authority") != "gate_plan_only_no_live_worker":
+        raise ValueError(f"{label}: runtime_authority must stay gate_plan_only_no_live_worker")
+
+    boundaries = payload.get("approval_boundaries") or {}
+    blocked_flags = {
+        "db_migration_allowed",
+        "live_db_query_allowed",
+        "redis_xadd_allowed",
+        "redis_consumer_group_allowed",
+        "runtime_worker_allowed",
+        "telegram_direct_send_allowed",
+        "learning_writeback_allowed",
+        "secret_plaintext_allowed",
+        "conversation_transcript_display_allowed",
+        "private_reasoning_display_allowed",
+        "agent_raw_output_display_allowed",
+    }
+    allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
+    if allowed:
+        raise ValueError(f"{label}: approval boundaries must remain false: {allowed}")
+
+    live_truth = payload.get("live_truth") or {}
+    false_flags = {
+        "live_agent_session_readback_enabled",
+        "live_redis_stream_read_enabled",
+        "runtime_worker_enabled",
+        "telegram_receipt_send_enabled",
+        "learning_writeback_enabled",
+    }
+    enabled = sorted(flag for flag in false_flags if live_truth.get(flag) is not False)
+    if enabled:
+        raise ValueError(f"{label}: live truth flags must remain false: {enabled}")
+
+    zero_counts = {
+        "active_live_agent_sessions",
+        "live_redis_events_24h",
+        "live_handoffs_24h",
+        "live_learning_writes_24h",
+        "telegram_digest_receipts_24h",
+    }
+    non_zero = sorted(key for key in zero_counts if live_truth.get(key) != 0)
+    if non_zero:
+        raise ValueError(f"{label}: live truth counts must remain zero: {non_zero}")
+
+
+def _require_storage_safety(payload: dict[str, Any], label: str) -> None:
+    storage = payload.get("existing_storage_contract") or {}
+    if storage.get("db_table") != "agent_sessions":
+        raise ValueError(f"{label}: existing_storage_contract.db_table must be agent_sessions")
+    if storage.get("approved_for_live_query") is not False:
+        raise ValueError(f"{label}: live DB query must remain unapproved")
+    if storage.get("migration_delta_required") is not False:
+        raise ValueError(f"{label}: migration delta must remain false for this gate")
+    if storage.get("safe_read_query_defined") is not True:
+        raise ValueError(f"{label}: safe read query contract must be defined")
+
+    selected_fields = set(storage.get("safe_selected_fields") or [])
+    forbidden_selected = selected_fields.intersection(
+        {
+            "output_json",
+            "prompt",
+            "raw_prompt",
+            "conversation_transcript",
+            "private_reasoning",
+            "chain_of_thought",
+            "secret_plaintext",
+            "credential_value",
+        }
+    )
+    if forbidden_selected:
+        raise ValueError(f"{label}: safe read query selects forbidden fields: {sorted(forbidden_selected)}")
+
+
+def _require_redis_safety(payload: dict[str, Any], label: str) -> None:
+    redis_contract = payload.get("redis_stream_contract") or {}
+    if redis_contract.get("consumer_group_allowed") is not False:
+        raise ValueError(f"{label}: Redis consumer group must remain unapproved")
+    if redis_contract.get("xadd_allowed") is not False:
+        raise ValueError(f"{label}: Redis XADD must remain unapproved")
+    if redis_contract.get("xreadgroup_allowed") is not False:
+        raise ValueError(f"{label}: Redis XREADGROUP must remain unapproved")
+    if not redis_contract.get("event_envelope_required_fields"):
+        raise ValueError(f"{label}: Redis event envelope required fields must be defined")
+
+
+def _require_no_write_smoke(payload: dict[str, Any], label: str) -> None:
+    smoke_steps = payload.get("no_write_smoke_plan") or []
+    if not smoke_steps:
+        raise ValueError(f"{label}: no_write_smoke_plan must not be empty")
+
+    unsafe_steps = [
+        step.get("smoke_id")
+        for step in smoke_steps
+        if step.get("writes_allowed") is not False or step.get("status") != "defined"
+    ]
+    if unsafe_steps:
+        raise ValueError(f"{label}: no-write smoke steps must be defined and write-blocked: {unsafe_steps}")
+
+    redaction = payload.get("display_redaction_contract") or {}
+    if redaction.get("redaction_required") is not True:
+        raise ValueError(f"{label}: frontend redaction must be required")
+    for flag in (
+        "work_window_conversation_display_allowed",
+        "agent_raw_output_display_allowed",
+        "secret_value_display_allowed",
+    ):
+        if redaction.get(flag) is not False:
+            raise ValueError(f"{label}: {flag} must remain false")
+
+
+def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
+    rollups = payload.get("rollups") or {}
+    expected_counts = {
+        "source_ref_count": len(payload.get("source_refs") or []),
+        "read_model_card_count": len(payload.get("read_model_cards") or []),
+        "gate_count": len(payload.get("worker_gate_plan") or []),
+        "rollback_step_count": len(payload.get("rollback_plan") or []),
+        "no_write_smoke_count": len(payload.get("no_write_smoke_plan") or []),
+        "forbidden_frontend_content_count": len(
+            (payload.get("display_redaction_contract") or {}).get("forbidden_frontend_content") or []
+        ),
+    }
+    mismatched = {
+        key: {"expected": expected, "actual": rollups.get(key)}
+        for key, expected in expected_counts.items()
+        if rollups.get(key) != expected
+    }
+    if mismatched:
+        raise ValueError(f"{label}: rollup counts must match payload sections: {mismatched}")
+
+    approval_required_gate_ids = sorted(
+        gate.get("gate_id")
+        for gate in payload.get("worker_gate_plan") or []
+        if gate.get("status") in {"approval_required", "blocked"}
+    )
+    if sorted(rollups.get("approval_required_gate_ids") or []) != approval_required_gate_ids:
+        raise ValueError(f"{label}: rollups.approval_required_gate_ids mismatch")
+
+    ready_cards = sorted(
+        card.get("card_id")
+        for card in payload.get("read_model_cards") or []
+        if card.get("readiness_status") == "query_contract_ready"
+    )
+    if sorted(rollups.get("query_contract_ready_card_ids") or []) != ready_cards:
+        raise ValueError(f"{label}: rollups.query_contract_ready_card_ids mismatch")
+
+    live_truth = payload.get("live_truth") or {}
+    live_count_total = sum(
+        int(live_truth.get(key) or 0)
+        for key in (
+            "active_live_agent_sessions",
+            "live_redis_events_24h",
+            "live_handoffs_24h",
+            "live_learning_writes_24h",
+            "telegram_digest_receipts_24h",
+        )
+    )
+    if rollups.get("live_truth_count_total") != live_count_total:
+        raise ValueError(f"{label}: rollups.live_truth_count_total mismatch")

apps/api/tests/test_ai_agent_interaction_learning_proof.py

@@ -13,9 +13,9 @@ def test_load_latest_ai_agent_interaction_learning_proof_reads_committed_snapsho
     data = load_latest_ai_agent_interaction_learning_proof()
 
     assert data["schema_version"] == "ai_agent_interaction_learning_proof_v1"
-    assert data["program_status"]["overall_completion_percent"] == 45
-    assert data["program_status"]["current_task_id"] == "P2-403A"
-    assert data["program_status"]["next_task_id"] == "P2-403B"
+    assert data["program_status"]["overall_completion_percent"] == 55
+    assert data["program_status"]["current_task_id"] == "P2-403B"
+    assert data["program_status"]["next_task_id"] == "P2-403C"
     assert data["program_status"]["read_only_mode"] is True
     assert data["program_status"]["runtime_authority"] == "proof_surface_only_no_live_worker"
     assert data["live_truth"]["runtime_loop_enabled"] is False
@@ -32,7 +32,6 @@ def test_load_latest_ai_agent_interaction_learning_proof_reads_committed_snapsho
     assert data["rollups"]["runtime_gate_count"] == len(data["runtime_gates"]) == 5
     assert data["rollups"]["live_signal_count"] == 0
     assert set(data["rollups"]["live_pending_level_ids"]) == {
-        "live_read_model",
         "message_bus_receipts",
         "handoff_receipts",
         "learning_growth",
@@ -98,10 +97,10 @@ def _snapshot() -> dict:
         "schema_version": "ai_agent_interaction_learning_proof_v1",
         "generated_at": "2026-06-11T23:20:00+08:00",
         "program_status": {
-            "overall_completion_percent": 45,
+            "overall_completion_percent": 55,
             "current_priority": "P2",
-            "current_task_id": "P2-403A",
-            "next_task_id": "P2-403B",
+            "current_task_id": "P2-403B",
+            "next_task_id": "P2-403C",
             "read_only_mode": True,
             "runtime_authority": "proof_surface_only_no_live_worker",
         },
@@ -120,7 +119,7 @@ def _snapshot() -> dict:
         "proof_ladder": [
             _proof_level("contract_surface", "contract_ready"),
             _proof_level("proof_dashboard", "proof_surface_ready"),
-            _proof_level("live_read_model", "live_pending"),
+            _proof_level("live_read_model", "contract_ready"),
         ],
         "agent_lanes": [
             _agent_lane("openclaw"),
@@ -189,8 +188,8 @@ def _snapshot() -> dict:
         },
         "rollups": {
             "proof_level_count": 3,
-            "contract_ready_level_count": 2,
-            "live_pending_level_ids": ["live_read_model"],
+            "contract_ready_level_count": 3,
+            "live_pending_level_ids": [],
             "signal_count": 1,
             "live_signal_count": 0,
             "operator_surface_count": 1,

apps/api/tests/test_ai_agent_interaction_learning_proof_api.py

@@ -16,9 +16,9 @@ def test_ai_agent_interaction_learning_proof_endpoint_returns_committed_snapshot
     assert response.status_code == 200
     data = response.json()
     assert data["schema_version"] == "ai_agent_interaction_learning_proof_v1"
-    assert data["program_status"]["overall_completion_percent"] == 45
-    assert data["program_status"]["current_task_id"] == "P2-403A"
-    assert data["program_status"]["next_task_id"] == "P2-403B"
+    assert data["program_status"]["overall_completion_percent"] == 55
+    assert data["program_status"]["current_task_id"] == "P2-403B"
+    assert data["program_status"]["next_task_id"] == "P2-403C"
     assert data["program_status"]["read_only_mode"] is True
     assert data["live_truth"]["runtime_loop_enabled"] is False
     assert data["live_truth"]["active_live_agent_sessions"] == 0
@@ -27,7 +27,7 @@ def test_ai_agent_interaction_learning_proof_endpoint_returns_committed_snapshot
     assert data["approval_boundaries"]["conversation_transcript_display_allowed"] is False
     assert data["approval_boundaries"]["telegram_direct_send_allowed"] is False
     assert data["rollups"]["proof_level_count"] == 7
-    assert data["rollups"]["contract_ready_level_count"] == 2
+    assert data["rollups"]["contract_ready_level_count"] == 3
     assert data["rollups"]["signal_count"] == 9
     assert data["rollups"]["live_signal_count"] == 0
     assert data["rollups"]["operator_surface_count"] == 5

apps/api/tests/test_ai_agent_live_read_model_gate.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from src.services.ai_agent_live_read_model_gate import (
+    load_latest_ai_agent_live_read_model_gate,
+)
+
+
+def test_load_latest_ai_agent_live_read_model_gate_reads_committed_snapshot():
+    data = load_latest_ai_agent_live_read_model_gate()
+
+    assert data["schema_version"] == "ai_agent_live_read_model_gate_v1"
+    assert data["program_status"]["overall_completion_percent"] == 55
+    assert data["program_status"]["current_task_id"] == "P2-403B"
+    assert data["program_status"]["next_task_id"] == "P2-403C"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["program_status"]["runtime_authority"] == "gate_plan_only_no_live_worker"
+    assert data["existing_storage_contract"]["db_table"] == "agent_sessions"
+    assert data["existing_storage_contract"]["approved_for_live_query"] is False
+    assert data["existing_storage_contract"]["migration_delta_required"] is False
+    assert "output_json" not in data["existing_storage_contract"]["safe_selected_fields"]
+    assert data["redis_stream_contract"]["consumer_group_allowed"] is False
+    assert data["redis_stream_contract"]["xadd_allowed"] is False
+    assert data["live_truth"]["active_live_agent_sessions"] == 0
+    assert data["live_truth"]["live_redis_events_24h"] == 0
+    assert data["approval_boundaries"]["runtime_worker_allowed"] is False
+    assert data["approval_boundaries"]["telegram_direct_send_allowed"] is False
+    assert data["display_redaction_contract"]["work_window_conversation_display_allowed"] is False
+    assert data["display_redaction_contract"]["agent_raw_output_display_allowed"] is False
+    assert data["rollups"]["read_model_card_count"] == len(data["read_model_cards"]) == 5
+    assert data["rollups"]["gate_count"] == len(data["worker_gate_plan"]) == 5
+    assert data["rollups"]["rollback_step_count"] == len(data["rollback_plan"]) == 4
+    assert data["rollups"]["no_write_smoke_count"] == len(data["no_write_smoke_plan"]) == 5
+    assert data["rollups"]["live_truth_count_total"] == 0
+    assert set(data["rollups"]["query_contract_ready_card_ids"]) == {
+        "agent_heartbeat_readback",
+        "agent_vote_quality",
+        "agent_role_turns",
+    }
+
+
+def test_load_latest_ai_agent_live_read_model_gate_rejects_live_db_query(tmp_path):
+    snapshot = _snapshot()
+    snapshot["approval_boundaries"]["live_db_query_allowed"] = True
+    (tmp_path / "ai_agent_live_read_model_gate_2026-06-11.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="approval boundaries"):
+        load_latest_ai_agent_live_read_model_gate(tmp_path)
+
+
+def test_load_latest_ai_agent_live_read_model_gate_rejects_unsafe_field_projection(tmp_path):
+    snapshot = _snapshot()
+    snapshot["existing_storage_contract"]["safe_selected_fields"].append("output_json")
+    (tmp_path / "ai_agent_live_read_model_gate_2026-06-11.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="forbidden fields"):
+        load_latest_ai_agent_live_read_model_gate(tmp_path)
+
+
+def test_load_latest_ai_agent_live_read_model_gate_rejects_redis_consumer_group(tmp_path):
+    snapshot = _snapshot()
+    snapshot["redis_stream_contract"]["consumer_group_allowed"] = True
+    (tmp_path / "ai_agent_live_read_model_gate_2026-06-11.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="Redis consumer group"):
+        load_latest_ai_agent_live_read_model_gate(tmp_path)
+
+
+def test_load_latest_ai_agent_live_read_model_gate_rejects_write_smoke(tmp_path):
+    snapshot = _snapshot()
+    snapshot["no_write_smoke_plan"][0]["writes_allowed"] = True
+    (tmp_path / "ai_agent_live_read_model_gate_2026-06-11.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="no-write smoke"):
+        load_latest_ai_agent_live_read_model_gate(tmp_path)
+
+
+def test_load_latest_ai_agent_live_read_model_gate_rejects_rollup_mismatch(tmp_path):
+    snapshot = _snapshot()
+    snapshot["rollups"]["gate_count"] = 99
+    (tmp_path / "ai_agent_live_read_model_gate_2026-06-11.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="rollup counts"):
+        load_latest_ai_agent_live_read_model_gate(tmp_path)
+
+
+def _snapshot() -> dict:
+    return {
+        "schema_version": "ai_agent_live_read_model_gate_v1",
+        "generated_at": "2026-06-11T23:50:00+08:00",
+        "program_status": {
+            "overall_completion_percent": 55,
+            "current_priority": "P2",
+            "current_task_id": "P2-403B",
+            "next_task_id": "P2-403C",
+            "read_only_mode": True,
+            "runtime_authority": "gate_plan_only_no_live_worker",
+        },
+        "source_refs": ["test"],
+        "live_truth": {
+            "live_agent_session_readback_enabled": False,
+            "live_redis_stream_read_enabled": False,
+            "runtime_worker_enabled": False,
+            "telegram_receipt_send_enabled": False,
+            "learning_writeback_enabled": False,
+            "active_live_agent_sessions": 0,
+            "live_redis_events_24h": 0,
+            "live_handoffs_24h": 0,
+            "live_learning_writes_24h": 0,
+            "telegram_digest_receipts_24h": 0,
+        },
+        "existing_storage_contract": {
+            "db_table": "agent_sessions",
+            "schema_status": "existing",
+            "migration_delta_required": False,
+            "approved_for_live_query": False,
+            "safe_read_query_defined": True,
+            "safe_selected_fields": ["id", "session_id", "agent_role", "created_at"],
+            "forbidden_selected_fields": ["output_json", "raw_prompt", "secret_plaintext"],
+            "required_indexes": ["ix_agent_sessions_created_at"],
+            "read_query_contract": "SELECT id FROM agent_sessions LIMIT :safe_limit",
+        },
+        "redis_stream_contract": {
+            "stream_namespace": "awoooi:agent:*",
+            "candidate_streams": ["awoooi:agent:events"],
+            "consumer_group_allowed": False,
+            "xadd_allowed": False,
+            "xreadgroup_allowed": False,
+            "event_envelope_required_fields": ["event_id", "session_id", "agent_id"],
+        },
+        "read_model_cards": [
+            _card("agent_heartbeat_readback", "query_contract_ready"),
+            _card("redis_message_receipts", "approval_required"),
+        ],
+        "worker_gate_plan": [
+            _gate("db_read_model_no_write_smoke", "approval_required"),
+            _gate("worker_kill_switch", "blocked"),
+        ],
+        "rollback_plan": [
+            {"rollback_id": "disable", "step": "停用。", "owner_agent": "openclaw"},
+        ],
+        "no_write_smoke_plan": [
+            {
+                "smoke_id": "loader",
+                "status": "defined",
+                "writes_allowed": False,
+                "assertion": "測試。",
+            },
+        ],
+        "display_redaction_contract": {
+            "redaction_required": True,
+            "work_window_conversation_display_allowed": False,
+            "agent_raw_output_display_allowed": False,
+            "secret_value_display_allowed": False,
+            "forbidden_frontend_content": ["工作視窗內容"],
+        },
+        "approval_boundaries": {
+            "db_migration_allowed": False,
+            "live_db_query_allowed": False,
+            "redis_xadd_allowed": False,
+            "redis_consumer_group_allowed": False,
+            "runtime_worker_allowed": False,
+            "telegram_direct_send_allowed": False,
+            "learning_writeback_allowed": False,
+            "secret_plaintext_allowed": False,
+            "conversation_transcript_display_allowed": False,
+            "private_reasoning_display_allowed": False,
+            "agent_raw_output_display_allowed": False,
+        },
+        "rollups": {
+            "source_ref_count": 1,
+            "read_model_card_count": 2,
+            "gate_count": 2,
+            "approval_required_gate_ids": [
+                "db_read_model_no_write_smoke",
+                "worker_kill_switch",
+            ],
+            "query_contract_ready_card_ids": ["agent_heartbeat_readback"],
+            "rollback_step_count": 1,
+            "no_write_smoke_count": 1,
+            "forbidden_frontend_content_count": 1,
+            "live_truth_count_total": 0,
+        },
+    }
+
+
+def _card(card_id: str, status: str) -> dict:
+    return {
+        "card_id": card_id,
+        "display_name": card_id,
+        "owner_agent": "openclaw",
+        "source_of_truth": "agent_sessions",
+        "readiness_status": status,
+        "operator_signal": "測試。",
+        "next_gate": "測試。",
+    }
+
+
+def _gate(gate_id: str, status: str) -> dict:
+    return {
+        "gate_id": gate_id,
+        "display_name": gate_id,
+        "owner_agent": "openclaw",
+        "status": status,
+        "required_evidence": "測試。",
+        "blocked_action": "test",
+    }

apps/api/tests/test_ai_agent_live_read_model_gate_api.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.agents import router
+
+
+def test_ai_agent_live_read_model_gate_endpoint_returns_committed_snapshot():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/agent-live-read-model-gate")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "ai_agent_live_read_model_gate_v1"
+    assert data["program_status"]["overall_completion_percent"] == 55
+    assert data["program_status"]["current_task_id"] == "P2-403B"
+    assert data["program_status"]["next_task_id"] == "P2-403C"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["program_status"]["runtime_authority"] == "gate_plan_only_no_live_worker"
+    assert data["existing_storage_contract"]["db_table"] == "agent_sessions"
+    assert data["existing_storage_contract"]["approved_for_live_query"] is False
+    assert data["redis_stream_contract"]["consumer_group_allowed"] is False
+    assert data["live_truth"]["active_live_agent_sessions"] == 0
+    assert data["live_truth"]["live_redis_events_24h"] == 0
+    assert data["approval_boundaries"]["live_db_query_allowed"] is False
+    assert data["approval_boundaries"]["redis_consumer_group_allowed"] is False
+    assert data["approval_boundaries"]["telegram_direct_send_allowed"] is False
+    assert data["display_redaction_contract"]["work_window_conversation_display_allowed"] is False
+    assert data["rollups"]["read_model_card_count"] == 5
+    assert data["rollups"]["gate_count"] == 5
+    assert data["rollups"]["no_write_smoke_count"] == 5
+    assert data["rollups"]["live_truth_count_total"] == 0

apps/api/tests/test_ai_agent_proactive_operations_contract.py

@@ -14,8 +14,8 @@ def test_load_latest_ai_agent_proactive_operations_contract_reads_committed_snap
 
     assert data["schema_version"] == "ai_agent_proactive_operations_contract_v1"
     assert data["program_status"]["overall_completion_percent"] == 100
-    assert data["program_status"]["current_task_id"] == "P2-403A"
-    assert data["program_status"]["next_task_id"] == "P2-403B"
+    assert data["program_status"]["current_task_id"] == "P2-403B"
+    assert data["program_status"]["next_task_id"] == "P2-403C"
     assert data["program_status"]["read_only_mode"] is True
     assert data["program_status"]["runtime_authority"] == "contract_only_no_version_or_runtime_update"
     assert data["approval_boundaries"]["runtime_version_update_allowed"] is False
@@ -25,6 +25,7 @@ def test_load_latest_ai_agent_proactive_operations_contract_reads_committed_snap
     assert data["approval_boundaries"]["telegram_direct_send_allowed"] is False
     assert data["rollups"]["version_domain_count"] == len(data["version_lifecycle_domains"]) == 12
     assert data["rollups"]["delegable_capability_count"] == len(data["delegable_capabilities"]) == 24
+    assert data["rollups"]["rollout_task_count"] == len(data["rollout_tasks"]) == 9
     assert data["rollups"]["auto_execute_allowed_count"] == 0
     assert any(domain["domain_id"] == "ai_agents_models" for domain in data["version_lifecycle_domains"])
     assert any(

apps/api/tests/test_ai_agent_proactive_operations_contract_api.py

@@ -17,8 +17,8 @@ def test_ai_agent_proactive_operations_contract_endpoint_returns_committed_snaps
     data = response.json()
     assert data["schema_version"] == "ai_agent_proactive_operations_contract_v1"
     assert data["program_status"]["overall_completion_percent"] == 100
-    assert data["program_status"]["current_task_id"] == "P2-403A"
-    assert data["program_status"]["next_task_id"] == "P2-403B"
+    assert data["program_status"]["current_task_id"] == "P2-403B"
+    assert data["program_status"]["next_task_id"] == "P2-403C"
     assert data["program_status"]["read_only_mode"] is True
     assert data["approval_boundaries"]["runtime_version_update_allowed"] is False
     assert data["approval_boundaries"]["package_upgrade_allowed"] is False
@@ -26,6 +26,7 @@ def test_ai_agent_proactive_operations_contract_endpoint_returns_committed_snaps
     assert data["approval_boundaries"]["telegram_direct_send_allowed"] is False
     assert data["rollups"]["version_domain_count"] == 12
     assert data["rollups"]["delegable_capability_count"] == 24
+    assert data["rollups"]["rollout_task_count"] == 9
     assert data["rollups"]["auto_execute_allowed_count"] == 0
     assert any(domain["domain_id"] == "host_os_packages" for domain in data["version_lifecycle_domains"])
     assert any(