feat(governance): 新增 AI Provider 路由矩陣

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"] == 83
+    assert data["program_status"]["overall_completion_percent"] == 87
     assert data["program_status"]["read_only_mode"] is True
-    assert data["program_status"]["current_task_id"] == "P1-003"
-    assert data["program_status"]["next_task_id"] == "P1-004"
+    assert data["program_status"]["current_task_id"] == "P1-004"
+    assert data["program_status"]["next_task_id"] == "P1-005"
     assert data["rollups"]["total_items"] == len(data["backlog_items"]) == 23
     assert data["rollups"]["by_priority"]["P1"] == 21
-    assert data["rollups"]["by_status"]["done"] == 19
+    assert data["rollups"]["by_status"]["done"] == 20
     assert data["rollups"]["by_gate_status"]["read_only_allowed"] == 20
-    assert data["progress_summary"]["overall_percent"] == 83
-    assert data["progress_summary"]["done_items"] == 19
+    assert data["progress_summary"]["overall_percent"] == 87
+    assert data["progress_summary"]["done_items"] == 20
     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"] == [
@@ -55,6 +55,12 @@ def test_ai_agent_automation_backlog_snapshot_endpoint_returns_committed_snapsho
     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_004 = next(item for item in data["backlog_items"] if item["item_id"] == "AUTO-P1-004")
+    assert p1_004["status"] == "done"
+    assert p1_004["next_review"] == "P1-005"
+    assert p1_004["approval_boundary"]["mode"] == "production_change_blocked"
+    assert "provider_switch" in p1_004["approval_boundary"]["blocked_actions"]
+    assert "ai_provider_route_matrix_2026-06-05.json" in p1_004["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,13 +18,15 @@ 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-003"
-    assert data["program_status"]["next_task_id"] == "P1-004"
-    assert data["task_approval_boundary_rollup"]["total_tasks"] == len(data["tasks"]) == 29
+    assert data["program_status"]["current_task_id"] == "P1-004"
+    assert data["program_status"]["next_task_id"] == "P1-005"
+    assert data["task_approval_boundary_rollup"]["total_tasks"] == len(data["tasks"]) == 30
     assert data["task_approval_boundary_rollup"]["by_mode"]["read_only_allowed"] == 27
+    assert data["task_approval_boundary_rollup"]["by_mode"]["production_change_blocked"] == 1
     assert data["task_approval_boundary_rollup"]["tasks_requiring_explicit_approval"] == [
         "P0-001",
         "P0-004",
+        "P1-004",
     ]
     assert data["approval_boundaries"]["sdk_installation_allowed"] is False
     assert data["approval_boundaries"]["paid_api_call_allowed"] is False
@@ -41,6 +43,11 @@ def test_ai_agent_automation_inventory_snapshot_endpoint_returns_committed_snaps
     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"]
+    p1_004 = next(task for task in data["tasks"] if task["task_id"] == "P1-004")
+    assert p1_004["status"] == "done"
+    assert p1_004["approval_boundary"]["mode"] == "production_change_blocked"
+    assert "provider_switch" in p1_004["approval_boundary"]["blocked_actions"]
+    assert "ai_provider_route_matrix_2026-06-05.json" in p1_004["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"])
@@ -72,3 +79,4 @@ def test_ai_agent_automation_inventory_snapshot_endpoint_returns_committed_snaps
     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"])
+    assert any(evidence["evidence_id"] == "ai_provider_route_matrix_api" for evidence in data["evidence"])

apps/api/tests/test_ai_provider_route_matrix.py

@@ -0,0 +1,345 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from src.services.ai_provider_route_matrix import load_latest_ai_provider_route_matrix
+
+
+def test_load_latest_ai_provider_route_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 / "ai_provider_route_matrix_2026-06-04.json").write_text(
+        json.dumps(older),
+        encoding="utf-8",
+    )
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(newer),
+        encoding="utf-8",
+    )
+
+    loaded = load_latest_ai_provider_route_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_routes"] == 6
+    assert loaded["operation_boundaries"]["provider_switch_allowed"] is False
+
+
+def test_ai_provider_route_matrix_requires_read_only_mode(tmp_path):
+    snapshot = _snapshot()
+    snapshot["program_status"]["read_only_mode"] = False
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="read_only_mode"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_blocks_provider_and_paid_call_changes(tmp_path):
+    snapshot = _snapshot()
+    snapshot["operation_boundaries"]["provider_switch_allowed"] = True
+    snapshot["operation_boundaries"]["paid_api_call_allowed"] = True
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="operation boundaries"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_requires_rollup_consistency(tmp_path):
+    snapshot = _snapshot()
+    snapshot["rollups"]["route_ids_requiring_action"] = []
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="route_ids_requiring_action"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_requires_nemotron_to_stay_blocked(tmp_path):
+    snapshot = _snapshot()
+    snapshot["provider_routes"][4]["status"] = "verified"
+    _refresh_rollups(snapshot)
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="Nemotron candidates"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_requires_candidate_gates_to_require_approval(tmp_path):
+    snapshot = _snapshot()
+    snapshot["candidate_gates"][0]["approval_required"] = False
+    snapshot["rollups"]["candidate_gate_ids_requiring_approval"] = [
+        "nemotron_replay_gate",
+        "paid_provider_call_gate",
+    ]
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="candidate gates"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_requires_operator_denials(tmp_path):
+    snapshot = _snapshot()
+    snapshot["operator_contract"]["must_not_interpret_as"].remove(
+        "provider 切換批准"
+    )
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="operator_contract"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_rejects_secret_payload_keys(tmp_path):
+    snapshot = _snapshot()
+    snapshot["latest_observations"][0]["authorization_header"] = "redacted"
+    (tmp_path / "ai_provider_route_matrix_2026-06-05.json").write_text(
+        json.dumps(snapshot),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="forbidden secret payload key"):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def test_ai_provider_route_matrix_fails_when_missing(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        load_latest_ai_provider_route_matrix(tmp_path)
+
+
+def _snapshot(
+    *,
+    generated_at: str = "2026-06-05T00:00:00+08:00",
+    completion: int = 100,
+) -> dict:
+    routes = [
+        _route(
+            "ai_router_execution_policy",
+            "AI Router 執行決策核心",
+            "ai_router_core",
+            "action_required",
+            "route_preserved",
+        ),
+        _route(
+            "ollama_global_endpoint_order",
+            "Ollama GCP-A → GCP-B → 111",
+            "ollama_failover",
+            "verified",
+            "route_preserved",
+        ),
+        _route(
+            "alert_ai_ollama_first_lane",
+            "告警 AI Ollama-first Lane",
+            "alert_governance_lane",
+            "verified",
+            "route_preserved",
+        ),
+        _route(
+            "openclaw_nemo_rca_lane",
+            "OpenClaw Nemo RCA Lane",
+            "openclaw_nemo",
+            "action_required",
+            "review_required",
+        ),
+        _route(
+            "nemotron_tool_calling_candidate",
+            "Nemotron Tool Calling 候選",
+            "nemotron_candidate",
+            "blocked",
+            "candidate_blocked",
+        ),
+        _route(
+            "gemini_final_fallback_policy",
+            "Gemini Final Fallback",
+            "paid_cloud_fallback",
+            "verified",
+            "route_preserved",
+        ),
+    ]
+    gates = [
+        _gate("provider_switch_gate", "production_change_blocked"),
+        _gate("nemotron_replay_gate", "blocked_by_evidence"),
+        _gate("paid_provider_call_gate", "cost_approval_required"),
+    ]
+    gaps = [
+        {
+            "gap_id": "ai_router_comment_drift",
+            "display_name": "AI Router 註解與現況 drift",
+            "status": "action_required",
+            "severity": "medium",
+            "summary": "舊註解需整理，不能誤讀成現行路由。",
+            "evidence_refs": ["apps/api/src/services/ai_router.py"],
+            "next_action": "只整理 source truth，不改 provider logic。",
+        }
+    ]
+    return {
+        "schema_version": "ai_provider_route_matrix_v1",
+        "generated_at": generated_at,
+        "program_status": {
+            "overall_completion_percent": completion,
+            "current_priority": "P1",
+            "current_task_id": "P1-004",
+            "next_task_id": "P1-005",
+            "read_only_mode": True,
+        },
+        "source_refs": ["docs/schemas/ai_provider_route_matrix_v1.schema.json"],
+        "rollups": {
+            "total_routes": len(routes),
+            "by_kind": _count_by(routes, "kind"),
+            "by_status": _count_by(routes, "status"),
+            "by_route_gate": _count_by(routes, "route_gate"),
+            "route_ids_requiring_action": [
+                "ai_router_execution_policy",
+                "openclaw_nemo_rca_lane",
+            ],
+            "candidate_gate_ids_requiring_approval": [
+                "nemotron_replay_gate",
+                "paid_provider_call_gate",
+                "provider_switch_gate",
+            ],
+            "source_gap_ids": ["ai_router_comment_drift"],
+            "read_only_denials_total": 12,
+            "provider_switch_allowed_count": 0,
+            "paid_api_call_allowed_count": 0,
+            "shadow_or_canary_allowed_count": 0,
+            "runtime_route_change_allowed_count": 0,
+        },
+        "provider_routes": routes,
+        "candidate_gates": gates,
+        "source_gaps": gaps,
+        "latest_observations": [
+            {
+                "observation_id": "provider_matrix_seed",
+                "status": "verified",
+                "summary": "只讀 route matrix seed。",
+                "evidence_refs": ["apps/api/src/services/ai_router.py"],
+            }
+        ],
+        "operator_contract": {
+            "display_mode": "read_only_ai_provider_route_matrix",
+            "must_not_interpret_as": [
+                "provider 切換批准",
+                "production routing change 批准",
+                "Gemini / Claude / NVIDIA 付費呼叫批准",
+                "OpenClaw 取代或降級批准",
+                "Nemotron 進 shadow / canary 批准",
+                "fallback order 修改批准",
+                "Ollama endpoint / ConfigMap 修改批准",
+                "Secret payload 已讀取或可輸出",
+                "外部 live probe 或 benchmark 批准",
+                "workflow / deploy / reload 觸發批准",
+                "runtime execution 授權",
+            ],
+            "secret_display_policy": "只顯示 env var 名稱。",
+            "provider_switch_policy": "P1-004 只能盤點與顯示；切換 provider 需另行批准。",
+            "cost_policy": "付費呼叫需批准。",
+            "runtime_policy": "active runtime gate 維持 0。",
+        },
+        "operation_boundaries": {
+            "read_only_api_allowed": True,
+            "provider_switch_allowed": False,
+            "production_routing_change_allowed": False,
+            "use_ai_router_toggle_allowed": False,
+            "fallback_order_change_allowed": False,
+            "ollama_endpoint_change_allowed": False,
+            "paid_api_call_allowed": False,
+            "paid_api_frequency_increase_allowed": False,
+            "external_provider_probe_allowed": False,
+            "live_benchmark_allowed": False,
+            "shadow_or_canary_allowed": False,
+            "openclaw_replacement_allowed": False,
+            "nemotron_shadow_allowed": False,
+            "gemini_direct_call_allowed": False,
+            "secret_read_allowed": False,
+            "secret_plaintext_allowed": False,
+            "notification_send_allowed": False,
+            "workflow_trigger_allowed": False,
+            "deploy_trigger_allowed": False,
+            "reload_trigger_allowed": False,
+            "runtime_execution_allowed": False,
+        },
+        "approval_boundaries": {
+            "provider_switch_approved": False,
+            "production_routing_change_approved": False,
+            "cost_change_approved": False,
+            "shadow_or_canary_approved": False,
+            "external_provider_call_approved": False,
+            "openclaw_replacement_approved": False,
+            "nemotron_replay_approved": False,
+            "secret_access_approved": False,
+            "runtime_execution_approved": False,
+        },
+    }
+
+
+def _route(
+    route_id: str,
+    display_name: str,
+    kind: str,
+    status: str,
+    route_gate: str,
+) -> dict:
+    return {
+        "route_id": route_id,
+        "display_name": display_name,
+        "kind": kind,
+        "status": status,
+        "risk_level": "critical",
+        "route_gate": route_gate,
+        "evidence_status": "committed_source",
+        "current_policy": "只讀盤點 provider route。",
+        "provider_order": ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"],
+        "fallback_policy": "不切 provider。",
+        "evidence_refs": ["apps/api/src/services/ai_router.py"],
+        "next_action": "準備批准包，不改 runtime。",
+    }
+
+
+def _gate(gate_id: str, status: str) -> dict:
+    return {
+        "gate_id": gate_id,
+        "display_name": gate_id,
+        "status": status,
+        "approval_required": True,
+        "summary": "需人工批准。",
+        "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
+
+
+def _refresh_rollups(snapshot: dict) -> None:
+    routes = snapshot["provider_routes"]
+    gates = snapshot["candidate_gates"]
+    snapshot["rollups"]["by_status"] = _count_by(routes, "status")
+    snapshot["rollups"]["by_route_gate"] = _count_by(routes, "route_gate")
+    snapshot["rollups"]["route_ids_requiring_action"] = sorted(
+        route["route_id"] for route in routes if route["status"] == "action_required"
+    )
+    snapshot["rollups"]["candidate_gate_ids_requiring_approval"] = sorted(
+        gate["gate_id"] for gate in gates if gate["approval_required"] is True
+    )

apps/api/tests/test_ai_provider_route_matrix_api.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.agents import router
+
+
+def test_ai_provider_route_matrix_endpoint_returns_committed_snapshot():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/ai-provider-route-matrix")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "ai_provider_route_matrix_v1"
+    assert data["program_status"]["current_task_id"] == "P1-004"
+    assert data["program_status"]["next_task_id"] == "P1-005"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["rollups"]["total_routes"] == len(data["provider_routes"]) == 7
+    assert data["rollups"]["provider_switch_allowed_count"] == 0
+    assert data["rollups"]["paid_api_call_allowed_count"] == 0
+    assert data["rollups"]["shadow_or_canary_allowed_count"] == 0
+    assert data["rollups"]["runtime_route_change_allowed_count"] == 0
+    assert data["operation_boundaries"]["provider_switch_allowed"] is False
+    assert data["operation_boundaries"]["paid_api_call_allowed"] is False
+    assert data["operation_boundaries"]["runtime_execution_allowed"] is False
+    assert data["approval_boundaries"]["provider_switch_approved"] is False
+    assert data["approval_boundaries"]["cost_change_approved"] is False
+    assert any(route["route_id"] == "ollama_global_endpoint_order" for route in data["provider_routes"])
+    assert any(route["route_id"] == "alert_ai_ollama_first_lane" for route in data["provider_routes"])
+    nemotron = next(
+        route for route in data["provider_routes"]
+        if route["route_id"] == "nemotron_tool_calling_candidate"
+    )
+    assert nemotron["status"] == "blocked"
+    assert nemotron["route_gate"] == "candidate_blocked"
+    assert data["operator_contract"]["display_mode"] == "read_only_ai_provider_route_matrix"
+    assert "provider 切換批准" in data["operator_contract"]["must_not_interpret_as"]