feat(governance): 新增監控合約降噪矩陣

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

@@ -65,6 +65,9 @@ from src.services.runtime_surface_inventory import (
 from src.services.gitea_workflow_runner_health import (
     load_latest_gitea_workflow_runner_health,
 )
+from src.services.observability_contract_matrix import (
+    load_latest_observability_contract_matrix,
+)
 from src.services.package_supply_chain_inventory import (
     load_latest_package_supply_chain_inventory,
 )
@@ -536,6 +539,34 @@ async def get_gitea_workflow_runner_health() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/observability-contract-matrix",
+    response_model=dict[str, Any],
+    summary="取得監控合約與降噪機會矩陣",
+    description=(
+        "讀取最新已提交的 Prometheus / Alertmanager / Grafana / SigNoz / ClickHouse / Sentry "
+        "只讀 observability matrix；此端點不修改 alert rules、不呼叫 silence API、"
+        "不建立 Grafana dashboard、不改 SigNoz / Sentry 設定、不讀 Secret payload、"
+        "不送 Telegram 測試通知、不觸發 monitoring deploy。"
+    ),
+)
+async def get_observability_contract_matrix() -> dict[str, Any]:
+    """Return the latest read-only observability contract matrix."""
+    try:
+        return await asyncio.to_thread(load_latest_observability_contract_matrix)
+    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("observability_contract_matrix_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="監控合約與降噪機會矩陣快照無效",
+        ) from exc
+
+
 @router.get(
     "/backup-dr-target-inventory",
     response_model=dict[str, Any],

apps/api/src/services/observability_contract_matrix.py

@@ -0,0 +1,232 @@
+"""
+Observability contract and noise-reduction matrix snapshot.
+
+Loads the latest committed, read-only Prometheus / Alertmanager / SigNoz /
+Grafana observability contract matrix. This module never mutates alert rules,
+routes, receivers, silences, dashboards, webhooks, collectors, secrets,
+notifications, workflows, or runtime state.
+"""
+
+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 = "observability_contract_matrix_*.json"
+_SCHEMA_VERSION = "observability_contract_matrix_v1"
+
+
+def load_latest_observability_contract_matrix(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the newest committed observability contract matrix snapshot."""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(f"no observability contract matrix 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_boundaries(payload, str(latest))
+    _require_operation_boundaries(payload, str(latest))
+    _require_rollup_consistency(payload, str(latest))
+    _require_surface_evidence(payload, str(latest))
+    _require_noise_opportunities(payload, str(latest))
+    _require_operator_denials(payload, str(latest))
+    _require_no_plaintext_secret_payload_keys(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:
+    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")
+
+    approval_boundaries = payload.get("approval_boundaries") or {}
+    allowed = sorted(key for key, value in approval_boundaries.items() if value is not False)
+    if allowed:
+        raise ValueError(f"{label}: approval boundaries must remain false: {allowed}")
+
+
+def _require_operation_boundaries(payload: dict[str, Any], label: str) -> None:
+    boundaries = payload.get("operation_boundaries") or {}
+    if boundaries.get("read_only_api_allowed") is not True:
+        raise ValueError(f"{label}: read_only_api_allowed must be true")
+
+    blocked_flags = {
+        "prometheus_rule_write_allowed",
+        "prometheus_reload_allowed",
+        "alertmanager_route_write_allowed",
+        "alertmanager_receiver_change_allowed",
+        "alertmanager_to_openclaw_allowed",
+        "silence_create_allowed",
+        "grafana_dashboard_write_allowed",
+        "grafana_api_write_allowed",
+        "signoz_query_mutation_allowed",
+        "signoz_webhook_change_allowed",
+        "sentry_webhook_change_allowed",
+        "otel_collector_deploy_allowed",
+        "event_exporter_restart_allowed",
+        "secret_read_allowed",
+        "secret_plaintext_allowed",
+        "notification_send_allowed",
+        "external_api_call_allowed",
+        "live_prometheus_query_allowed",
+        "workflow_trigger_allowed",
+        "deploy_trigger_allowed",
+        "reload_trigger_allowed",
+        "runtime_execution_allowed",
+    }
+    allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
+    if allowed:
+        raise ValueError(f"{label}: operation boundaries must remain false: {allowed}")
+
+
+def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
+    surfaces = payload.get("observability_surfaces") or []
+    opportunities = payload.get("noise_reduction_opportunities") or []
+    gaps = payload.get("classification_gaps") or []
+    rollups = payload.get("rollups") or {}
+
+    if rollups.get("total_surfaces") != len(surfaces):
+        raise ValueError(f"{label}: rollups.total_surfaces must match observability_surfaces")
+    if rollups.get("by_kind") != _count_by(surfaces, "kind"):
+        raise ValueError(f"{label}: rollups.by_kind must match observability_surfaces")
+    if rollups.get("by_status") != _count_by(surfaces, "status"):
+        raise ValueError(f"{label}: rollups.by_status must match observability_surfaces")
+    if rollups.get("by_evidence_status") != _count_by(surfaces, "evidence_status"):
+        raise ValueError(f"{label}: rollups.by_evidence_status must match observability_surfaces")
+    if rollups.get("by_noise_policy_status") != _count_by(surfaces, "noise_policy_status"):
+        raise ValueError(f"{label}: rollups.by_noise_policy_status must match observability_surfaces")
+
+    action_required = sorted(
+        surface.get("surface_id")
+        for surface in surfaces
+        if surface.get("status") == "action_required"
+    )
+    if sorted(rollups.get("surface_ids_requiring_action") or []) != action_required:
+        raise ValueError(f"{label}: rollups.surface_ids_requiring_action must match surfaces")
+
+    proposal_only_surfaces = sorted(
+        surface.get("surface_id")
+        for surface in surfaces
+        if surface.get("noise_policy_status") == "proposal_only"
+    )
+    if sorted(rollups.get("surface_ids_with_proposal_only_noise_policy") or []) != proposal_only_surfaces:
+        raise ValueError(
+            f"{label}: rollups.surface_ids_with_proposal_only_noise_policy must match surfaces"
+        )
+
+    approval_required = sorted(
+        opportunity.get("opportunity_id")
+        for opportunity in opportunities
+        if opportunity.get("status") == "approval_required"
+    )
+    if rollups.get("noise_reduction_opportunities_total") != len(opportunities):
+        raise ValueError(f"{label}: rollups.noise_reduction_opportunities_total must match opportunities")
+    if sorted(rollups.get("approval_required_opportunity_ids") or []) != approval_required:
+        raise ValueError(f"{label}: rollups.approval_required_opportunity_ids must match opportunities")
+
+    if sorted(rollups.get("classification_gap_ids") or []) != sorted(gap.get("gap_id") for gap in gaps):
+        raise ValueError(f"{label}: rollups.classification_gap_ids must match classification_gaps")
+
+
+def _require_surface_evidence(payload: dict[str, Any], label: str) -> None:
+    surfaces = payload.get("observability_surfaces") or []
+    missing = sorted(
+        surface.get("surface_id")
+        for surface in surfaces
+        if not surface.get("coverage_contract")
+        or not surface.get("evidence_refs")
+        or not surface.get("next_action")
+    )
+    if missing:
+        raise ValueError(f"{label}: observability_surfaces must include contract, evidence, next_action: {missing}")
+
+
+def _require_noise_opportunities(payload: dict[str, Any], label: str) -> None:
+    opportunities = payload.get("noise_reduction_opportunities") or []
+    non_proposal = sorted(
+        opportunity.get("opportunity_id")
+        for opportunity in opportunities
+        if opportunity.get("proposal_only") is not True
+    )
+    if non_proposal:
+        raise ValueError(f"{label}: noise opportunities must stay proposal_only: {non_proposal}")
+
+    required_ids = {
+        "prometheus_noise_rule_tuning",
+        "alertmanager_grouping_inhibit_tuning",
+        "success_notification_quiet_policy",
+    }
+    present = {opportunity.get("opportunity_id") for opportunity in opportunities}
+    if not required_ids.issubset(present):
+        raise ValueError(f"{label}: missing required noise-reduction opportunities")
+
+
+def _require_operator_denials(payload: dict[str, Any], label: str) -> None:
+    contract = payload.get("operator_contract") or {}
+    must_not_interpret_as = set(contract.get("must_not_interpret_as") or [])
+    required_denials = {
+        "Prometheus alert rule 修改批准",
+        "Alertmanager receiver / route 修改批准",
+        "Alertmanager 指向 OpenClaw receiver 批准",
+        "Silence 建立或維護窗口批准",
+        "Grafana dashboard 寫入批准",
+        "SigNoz / Sentry webhook 設定修改批准",
+        "Secret 已讀取或可輸出",
+        "Telegram 測試通知批准",
+        "deploy / reload / workflow 觸發批准",
+        "runtime execution 授權",
+    }
+    if not required_denials.issubset(must_not_interpret_as):
+        raise ValueError(f"{label}: operator_contract.must_not_interpret_as is missing required denials")
+
+    route_policy = str(contract.get("alertmanager_route_policy") or "")
+    if "OpenClaw" not in route_policy or "不接收 Alertmanager webhook" not in route_policy:
+        raise ValueError(f"{label}: operator_contract.alertmanager_route_policy must block OpenClaw receiver use")
+
+
+def _require_no_plaintext_secret_payload_keys(value: Any, label: str, path: str = "$") -> None:
+    if isinstance(value, dict):
+        forbidden_key_fragments = {
+            "secret_value",
+            "token_value",
+            "authorization_header",
+            "private_key",
+            "webhook_secret",
+            "runner_token",
+            "signoz_token",
+            "sentry_dsn",
+        }
+        for key, nested in value.items():
+            lowered = str(key).lower()
+            if any(fragment in lowered for fragment in forbidden_key_fragments):
+                raise ValueError(f"{label}: forbidden secret payload key at {path}.{key}")
+            _require_no_plaintext_secret_payload_keys(nested, label, f"{path}.{key}")
+    elif isinstance(value, list):
+        for index, nested in enumerate(value):
+            _require_no_plaintext_secret_payload_keys(nested, label, f"{path}[{index}]")
+
+
+def _count_by(items: list[dict[str, Any]], key: str) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for item in items:
+        value = item.get(key)
+        counts[value] = counts.get(value, 0) + 1
+    return counts

apps/api/tests/test_ai_agent_automation_backlog_snapshot_api.py

@@ -16,16 +16,16 @@ def test_ai_agent_automation_backlog_snapshot_endpoint_returns_committed_snapsho
     assert response.status_code == 200
     data = response.json()
     assert data["schema_version"] == "ai_agent_automation_backlog_v1"
-    assert data["program_status"]["overall_completion_percent"] == 78
+    assert data["program_status"]["overall_completion_percent"] == 83
     assert data["program_status"]["read_only_mode"] is True
-    assert data["program_status"]["current_task_id"] == "P1-002"
-    assert data["program_status"]["next_task_id"] == "P1-003"
+    assert data["program_status"]["current_task_id"] == "P1-003"
+    assert data["program_status"]["next_task_id"] == "P1-004"
     assert data["rollups"]["total_items"] == len(data["backlog_items"]) == 23
     assert data["rollups"]["by_priority"]["P1"] == 21
-    assert data["rollups"]["by_status"]["done"] == 18
+    assert data["rollups"]["by_status"]["done"] == 19
     assert data["rollups"]["by_gate_status"]["read_only_allowed"] == 20
-    assert data["progress_summary"]["overall_percent"] == 78
-    assert data["progress_summary"]["done_items"] == 18
+    assert data["progress_summary"]["overall_percent"] == 83
+    assert data["progress_summary"]["done_items"] == 19
     assert data["progress_summary"]["total_items"] == 23
     assert data["item_approval_boundary_rollup"]["total_items"] == 23
     assert data["item_approval_boundary_rollup"]["items_requiring_explicit_approval"] == [
@@ -51,6 +51,10 @@ def test_ai_agent_automation_backlog_snapshot_endpoint_returns_committed_snapsho
     assert p1_002["status"] == "done"
     assert p1_002["next_review"] == "P1-003"
     assert "gitea_workflow_runner_health_2026-06-05.json" in p1_002["evidence_refs"][0]
+    p1_003 = next(item for item in data["backlog_items"] if item["item_id"] == "AUTO-P1-003")
+    assert p1_003["status"] == "done"
+    assert p1_003["next_review"] == "P1-004"
+    assert "observability_contract_matrix_2026-06-05.json" in p1_003["evidence_refs"][0]
     p1_306 = next(item for item in data["backlog_items"] if item["item_id"] == "AUTO-P1-306")
     assert p1_306["approval_boundary"]["mode"] == "read_only_allowed"
     assert "runtime_execution" in p1_306["approval_boundary"]["blocked_actions"]

apps/api/tests/test_ai_agent_automation_inventory_snapshot_api.py

@@ -18,10 +18,10 @@ def test_ai_agent_automation_inventory_snapshot_endpoint_returns_committed_snaps
     assert data["schema_version"] == "ai_agent_automation_inventory_snapshot_v1"
     assert data["program_status"]["overall_completion_percent"] == 100
     assert data["program_status"]["read_only_mode"] is True
-    assert data["program_status"]["current_task_id"] == "P1-002"
-    assert data["program_status"]["next_task_id"] == "P1-003"
-    assert data["task_approval_boundary_rollup"]["total_tasks"] == len(data["tasks"]) == 28
-    assert data["task_approval_boundary_rollup"]["by_mode"]["read_only_allowed"] == 26
+    assert data["program_status"]["current_task_id"] == "P1-003"
+    assert data["program_status"]["next_task_id"] == "P1-004"
+    assert data["task_approval_boundary_rollup"]["total_tasks"] == len(data["tasks"]) == 29
+    assert data["task_approval_boundary_rollup"]["by_mode"]["read_only_allowed"] == 27
     assert data["task_approval_boundary_rollup"]["tasks_requiring_explicit_approval"] == [
         "P0-001",
         "P0-004",
@@ -37,6 +37,10 @@ def test_ai_agent_automation_inventory_snapshot_endpoint_returns_committed_snaps
     assert p1_002["status"] == "done"
     assert p1_002["approval_boundary"]["mode"] == "read_only_allowed"
     assert "gitea_workflow_runner_health_2026-06-05.json" in p1_002["output"]
+    p1_003 = next(task for task in data["tasks"] if task["task_id"] == "P1-003")
+    assert p1_003["status"] == "done"
+    assert p1_003["approval_boundary"]["mode"] == "read_only_allowed"
+    assert "observability_contract_matrix_2026-06-05.json" in p1_003["output"]
     assert any(task["task_id"] == "P1-204" for task in data["tasks"])
     assert any(task["task_id"] == "P1-205" for task in data["tasks"])
     assert any(task["task_id"] == "P1-206" for task in data["tasks"])
@@ -67,3 +71,4 @@ def test_ai_agent_automation_inventory_snapshot_endpoint_returns_committed_snaps
     assert any(evidence["evidence_id"] == "backlog_progress_summary_ui" for evidence in data["evidence"])
     assert any(evidence["evidence_id"] == "runtime_surface_inventory_api" for evidence in data["evidence"])
     assert any(evidence["evidence_id"] == "gitea_workflow_runner_health_api" for evidence in data["evidence"])
+    assert any(evidence["evidence_id"] == "observability_contract_matrix_api" for evidence in data["evidence"])

apps/api/tests/test_observability_contract_matrix.py

@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from src.services.observability_contract_matrix import load_latest_observability_contract_matrix
+
+
+def test_load_latest_observability_contract_matrix_reads_newest_file(tmp_path):
+    older = _snapshot(generated_at="2026-06-04T00:00:00+08:00", completion=40)
+    newer = _snapshot(generated_at="2026-06-05T00:00:00+08:00", completion=100)
+    (tmp_path / "observability_contract_matrix_2026-06-04.json").write_text(
+        json.dumps(older),
+        encoding="utf-8",
+    )
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(newer),
+        encoding="utf-8",
+    )
+
+    loaded = load_latest_observability_contract_matrix(tmp_path)
+
+    assert loaded["generated_at"] == "2026-06-05T00:00:00+08:00"
+    assert loaded["program_status"]["overall_completion_percent"] == 100
+    assert loaded["rollups"]["total_surfaces"] == 2
+    assert loaded["operation_boundaries"]["alertmanager_to_openclaw_allowed"] is False
+
+
+def test_observability_contract_matrix_requires_read_only_mode(tmp_path):
+    snapshot = _snapshot()
+    snapshot["program_status"]["read_only_mode"] = False
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="read_only_mode"):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def test_observability_contract_matrix_blocks_route_and_rule_mutations(tmp_path):
+    snapshot = _snapshot()
+    snapshot["operation_boundaries"]["prometheus_rule_write_allowed"] = True
+    snapshot["operation_boundaries"]["alertmanager_to_openclaw_allowed"] = True
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="operation boundaries"):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def test_observability_contract_matrix_requires_rollup_consistency(tmp_path):
+    snapshot = _snapshot()
+    snapshot["rollups"]["surface_ids_requiring_action"] = []
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="surface_ids_requiring_action"):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def test_observability_contract_matrix_requires_noise_candidates_to_be_proposal_only(tmp_path):
+    snapshot = _snapshot()
+    snapshot["noise_reduction_opportunities"][0]["proposal_only"] = False
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="proposal_only"):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def test_observability_contract_matrix_requires_openclaw_receiver_denial(tmp_path):
+    snapshot = _snapshot()
+    snapshot["operator_contract"]["must_not_interpret_as"].remove(
+        "Alertmanager 指向 OpenClaw receiver 批准"
+    )
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="operator_contract"):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def test_observability_contract_matrix_rejects_secret_payload_keys(tmp_path):
+    snapshot = _snapshot()
+    snapshot["latest_observations"][0]["webhook_secret"] = "redacted"
+    (tmp_path / "observability_contract_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="forbidden secret payload key"):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def test_observability_contract_matrix_fails_when_missing(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        load_latest_observability_contract_matrix(tmp_path)
+
+
+def _snapshot(
+    *,
+    generated_at: str = "2026-06-05T00:00:00+08:00",
+    completion: int = 100,
+) -> dict:
+    surfaces = [
+        _surface(
+            "prometheus_alert_rule_catalog",
+            "Prometheus 告警規則合約",
+            "prometheus_rules",
+            "action_required",
+            "proposal_only",
+        ),
+        _surface(
+            "alertmanager_awoooi_route",
+            "Alertmanager → AWOOOI API 路由",
+            "alertmanager_route",
+            "verified",
+            "proposal_only",
+        ),
+    ]
+    opportunities = [
+        _opportunity("prometheus_noise_rule_tuning", "approval_required"),
+        _opportunity("alertmanager_grouping_inhibit_tuning", "approval_required"),
+        _opportunity("success_notification_quiet_policy", "preserved"),
+    ]
+    gaps = [
+        {
+            "gap_id": "prometheus_alert_rule_catalog_seed",
+            "display_name": "Alert rule catalog seed 未正式產品化",
+            "status": "action_required",
+            "severity": "high",
+            "summary": "只讀矩陣已建立，尚未產生 catalog seed。",
+            "evidence_refs": ["docs/adr/ADR-090-monitoring-blindspot-governance.md"],
+            "next_action": "先產 proposal，不改 rule。",
+        }
+    ]
+    return {
+        "schema_version": "observability_contract_matrix_v1",
+        "generated_at": generated_at,
+        "program_status": {
+            "overall_completion_percent": completion,
+            "current_priority": "P1",
+            "current_task_id": "P1-003",
+            "next_task_id": "P1-004",
+            "read_only_mode": True,
+        },
+        "source_refs": ["docs/schemas/observability_contract_matrix_v1.schema.json"],
+        "rollups": {
+            "total_surfaces": len(surfaces),
+            "by_kind": _count_by(surfaces, "kind"),
+            "by_status": _count_by(surfaces, "status"),
+            "by_evidence_status": _count_by(surfaces, "evidence_status"),
+            "by_noise_policy_status": _count_by(surfaces, "noise_policy_status"),
+            "surface_ids_requiring_action": ["prometheus_alert_rule_catalog"],
+            "surface_ids_with_proposal_only_noise_policy": [
+                "alertmanager_awoooi_route",
+                "prometheus_alert_rule_catalog",
+            ],
+            "noise_reduction_opportunities_total": len(opportunities),
+            "approval_required_opportunity_ids": [
+                "alertmanager_grouping_inhibit_tuning",
+                "prometheus_noise_rule_tuning",
+            ],
+            "classification_gap_ids": ["prometheus_alert_rule_catalog_seed"],
+            "read_only_denials_total": 12,
+        },
+        "observability_surfaces": surfaces,
+        "noise_reduction_opportunities": opportunities,
+        "classification_gaps": gaps,
+        "latest_observations": [
+            {
+                "observation_id": "alertmanager_receiver_guard",
+                "status": "verified",
+                "summary": "Alertmanager 不得指向 OpenClaw。",
+                "evidence_refs": ["docs/HARD_RULES.md#alertmanager-routing"],
+            }
+        ],
+        "operator_contract": {
+            "display_mode": "read_only_observability_contract_matrix",
+            "must_not_interpret_as": [
+                "Prometheus alert rule 修改批准",
+                "Alertmanager receiver / route 修改批准",
+                "Alertmanager 指向 OpenClaw receiver 批准",
+                "Silence 建立或維護窗口批准",
+                "Grafana dashboard 寫入批准",
+                "SigNoz / Sentry webhook 設定修改批准",
+                "Secret 已讀取或可輸出",
+                "Telegram 測試通知批准",
+                "deploy / reload / workflow 觸發批准",
+                "runtime execution 授權",
+            ],
+            "secret_display_policy": "只顯示 redacted metadata。",
+            "alertmanager_route_policy": "OpenClaw 不接收 Alertmanager webhook；receiver 維持 AWOOOI API。",
+            "noise_reduction_policy": "只產生 proposal。",
+            "notification_policy": "成功不洗版。",
+        },
+        "operation_boundaries": {
+            "read_only_api_allowed": True,
+            "prometheus_rule_write_allowed": False,
+            "prometheus_reload_allowed": False,
+            "alertmanager_route_write_allowed": False,
+            "alertmanager_receiver_change_allowed": False,
+            "alertmanager_to_openclaw_allowed": False,
+            "silence_create_allowed": False,
+            "grafana_dashboard_write_allowed": False,
+            "grafana_api_write_allowed": False,
+            "signoz_query_mutation_allowed": False,
+            "signoz_webhook_change_allowed": False,
+            "sentry_webhook_change_allowed": False,
+            "otel_collector_deploy_allowed": False,
+            "event_exporter_restart_allowed": False,
+            "secret_read_allowed": False,
+            "secret_plaintext_allowed": False,
+            "notification_send_allowed": False,
+            "external_api_call_allowed": False,
+            "live_prometheus_query_allowed": False,
+            "workflow_trigger_allowed": False,
+            "deploy_trigger_allowed": False,
+            "reload_trigger_allowed": False,
+            "runtime_execution_allowed": False,
+        },
+        "approval_boundaries": {
+            "prometheus_rule_change_authorized": False,
+            "prometheus_reload_authorized": False,
+            "alertmanager_route_change_authorized": False,
+            "alertmanager_receiver_change_authorized": False,
+            "alertmanager_to_openclaw_authorized": False,
+            "silence_authorized": False,
+            "grafana_write_authorized": False,
+            "signoz_write_authorized": False,
+            "sentry_write_authorized": False,
+            "otel_deploy_authorized": False,
+            "event_exporter_restart_authorized": False,
+            "notification_send_authorized": False,
+            "external_call_authorized": False,
+            "secret_plaintext_allowed": False,
+            "workflow_trigger_authorized": False,
+            "deploy_reload_authorized": False,
+            "runtime_execution_authorized": False,
+        },
+    }
+
+
+def _surface(
+    surface_id: str,
+    display_name: str,
+    kind: str,
+    status: str,
+    noise_policy_status: str,
+) -> dict:
+    return {
+        "surface_id": surface_id,
+        "display_name": display_name,
+        "kind": kind,
+        "status": status,
+        "risk_level": "critical",
+        "evidence_status": "committed_manifest",
+        "noise_policy_status": noise_policy_status,
+        "coverage_contract": "只讀 committed evidence。",
+        "current_contract": "不得改 live 設定。",
+        "evidence_refs": ["docs/HARD_RULES.md"],
+        "next_action": "只產 proposal。",
+    }
+
+
+def _opportunity(opportunity_id: str, status: str) -> dict:
+    return {
+        "opportunity_id": opportunity_id,
+        "display_name": opportunity_id,
+        "status": status,
+        "proposal_only": True,
+        "impact": "降噪提案。",
+        "evidence_refs": ["docs/HARD_RULES.md"],
+        "next_action": "人工批准前不執行。",
+    }
+
+
+def _count_by(items: list[dict], key: str) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for item in items:
+        value = item[key]
+        counts[value] = counts.get(value, 0) + 1
+    return counts

apps/api/tests/test_observability_contract_matrix_api.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.agents import router
+
+
+def test_observability_contract_matrix_endpoint_returns_committed_snapshot():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/observability-contract-matrix")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "observability_contract_matrix_v1"
+    assert data["program_status"]["overall_completion_percent"] == 100
+    assert data["program_status"]["current_task_id"] == "P1-003"
+    assert data["program_status"]["next_task_id"] == "P1-004"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["rollups"]["total_surfaces"] == len(data["observability_surfaces"]) == 6
+    assert data["rollups"]["noise_reduction_opportunities_total"] == 5
+    assert data["rollups"]["surface_ids_requiring_action"] == [
+        "grafana_dashboard_inventory",
+        "prometheus_alert_rule_catalog",
+    ]
+    assert data["rollups"]["approval_required_opportunity_ids"] == [
+        "alertmanager_grouping_inhibit_tuning",
+        "prometheus_noise_rule_tuning",
+    ]
+    assert data["operation_boundaries"]["read_only_api_allowed"] is True
+    assert data["operation_boundaries"]["prometheus_rule_write_allowed"] is False
+    assert data["operation_boundaries"]["alertmanager_route_write_allowed"] is False
+    assert data["operation_boundaries"]["alertmanager_to_openclaw_allowed"] is False
+    assert data["operation_boundaries"]["silence_create_allowed"] is False
+    assert data["operation_boundaries"]["grafana_dashboard_write_allowed"] is False
+    assert data["operation_boundaries"]["notification_send_allowed"] is False
+    assert data["operation_boundaries"]["deploy_trigger_allowed"] is False
+    assert data["approval_boundaries"]["prometheus_rule_change_authorized"] is False
+    assert data["approval_boundaries"]["alertmanager_to_openclaw_authorized"] is False
+    assert data["approval_boundaries"]["deploy_reload_authorized"] is False
+    alertmanager = next(
+        row for row in data["observability_surfaces"] if row["surface_id"] == "alertmanager_awoooi_route"
+    )
+    assert alertmanager["status"] == "verified"
+    assert alertmanager["noise_policy_status"] == "proposal_only"
+    assert "OpenClaw 只做 AI 分析" in alertmanager["coverage_contract"]
+    assert "Alertmanager 指向 OpenClaw receiver 批准" in data["operator_contract"]["must_not_interpret_as"]
+    assert "不接收 Alertmanager webhook" in data["operator_contract"]["alertmanager_route_policy"]
+    for opportunity in data["noise_reduction_opportunities"]:
+        assert opportunity["proposal_only"] is True