feat(governance): 新增 AI Agent Telegram 預覽審核包

apps/api/src/services/ai_agent_professional_task_expansion.py

@@ -1,10 +1,10 @@
 """
 AI Agent professional task expansion and Telegram runtime bridge snapshot.
 
-Loads the latest committed P2-405A read-only contract. The contract expands
-professional AI Agent work and defines Telegram bridge stages, but it does not
-write Telegram Gateway queues, send Telegram messages, call the Bot API, read
-secrets, or execute production changes.
+Loads the latest committed P2-405B read-only contract. The contract expands
+professional AI Agent work and defines Telegram no-send previews, but it does
+not write Telegram Gateway queues, send Telegram messages, call the Bot API,
+read secrets, or execute production changes.
 """
 
 from __future__ import annotations
@@ -24,6 +24,10 @@ _EXPECTED_TASK_COUNT = 24
 _EXPECTED_DOMAIN_COUNT = 8
 _EXPECTED_STAGE_COUNT = 5
 _EXPECTED_MESSAGE_TYPE_COUNT = 6
+_EXPECTED_NO_SEND_PREVIEW_COUNT = 6
+_EXPECTED_DEDUP_KEY_COUNT = 6
+_EXPECTED_RECEIPT_EXPECTATION_COUNT = 6
+_EXPECTED_CANARY_PACKAGE_COUNT = 1
 _ZERO_ROLLUP_FIELDS = {
     "current_live_count",
     "gateway_queue_write_count",
@@ -35,6 +39,11 @@ _ZERO_ROLLUP_FIELDS = {
     "paid_api_call_count",
     "host_write_count",
     "kubectl_action_count",
+    "preview_send_enabled_count",
+    "preview_queue_write_enabled_count",
+    "preview_bot_api_call_enabled_count",
+    "receipt_live_write_enabled_count",
+    "canary_live_send_enabled_count",
 }
 _FORBIDDEN_PUBLIC_TERMS = {
     "work_window_transcript",
@@ -82,11 +91,11 @@ def _require_schema(payload: dict[str, Any], label: str) -> None:
     status = payload.get("program_status") or {}
     expected = {
         "current_priority": "P2",
-        "current_task_id": "P2-405A",
-        "next_task_id": "P2-405B",
+        "current_task_id": "P2-405B",
+        "next_task_id": "P2-405C",
         "read_only_mode": True,
         "runtime_authority": _RUNTIME_AUTHORITY,
-        "overall_completion_percent": 82,
+        "overall_completion_percent": 88,
     }
     mismatches = _mismatches(status, expected)
     if mismatches:
@@ -124,6 +133,105 @@ def _require_telegram_bridge(payload: dict[str, Any], label: str) -> None:
     if len(message_types) != _EXPECTED_MESSAGE_TYPE_COUNT:
         raise ValueError(f"{label}: expected {_EXPECTED_MESSAGE_TYPE_COUNT} message types")
 
+    _require_no_send_previews(bridge, label)
+    _require_receipt_and_canary_package(bridge, label)
+
+
+def _require_no_send_previews(bridge: dict[str, Any], label: str) -> None:
+    previews = bridge.get("no_send_message_previews") or []
+    if len(previews) != _EXPECTED_NO_SEND_PREVIEW_COUNT:
+        raise ValueError(f"{label}: expected {_EXPECTED_NO_SEND_PREVIEW_COUNT} no-send previews")
+
+    message_types = {item.get("message_type") for item in bridge.get("message_types") or []}
+    preview_message_types = [preview.get("message_type") for preview in previews]
+    if set(preview_message_types) != message_types:
+        raise ValueError(f"{label}: no-send previews must cover every message type exactly once")
+
+    dedup_keys = [preview.get("dedup_key") for preview in previews]
+    if len(set(dedup_keys)) != len(dedup_keys):
+        raise ValueError(f"{label}: no-send preview dedup_key values must be unique")
+
+    for preview in previews:
+        preview_id = preview.get("preview_id")
+        if preview.get("status") != "preview_ready_no_send":
+            raise ValueError(f"{label}: {preview_id}.status must be preview_ready_no_send")
+        if preview.get("send_enabled") is not False:
+            raise ValueError(f"{label}: {preview_id}.send_enabled must remain false")
+        if preview.get("gateway_queue_write_enabled") is not False:
+            raise ValueError(f"{label}: {preview_id}.gateway_queue_write_enabled must remain false")
+        if preview.get("bot_api_call_enabled") is not False:
+            raise ValueError(f"{label}: {preview_id}.bot_api_call_enabled must remain false")
+        if preview.get("delivery_receipt_write_enabled") is not False:
+            raise ValueError(f"{label}: {preview_id}.delivery_receipt_write_enabled must remain false")
+        if not preview.get("sanitized_body_lines"):
+            raise ValueError(f"{label}: {preview_id}.sanitized_body_lines must not be empty")
+
+    dedup_policy = bridge.get("dedup_policy") or {}
+    keys = dedup_policy.get("keys") or []
+    if dedup_policy.get("required") is not True:
+        raise ValueError(f"{label}: dedup_policy.required must be true")
+    if dedup_policy.get("live_cache_write_enabled") is not False:
+        raise ValueError(f"{label}: dedup_policy.live_cache_write_enabled must remain false")
+    if len(keys) != _EXPECTED_DEDUP_KEY_COUNT:
+        raise ValueError(f"{label}: expected {_EXPECTED_DEDUP_KEY_COUNT} dedup keys")
+
+    queue_preview = bridge.get("queue_preview_readback") or {}
+    expected_queue = {
+        "enabled": True,
+        "preview_only": True,
+        "write_enabled": False,
+        "readback_enabled": True,
+    }
+    mismatches = _mismatches(queue_preview, expected_queue)
+    if mismatches:
+        raise ValueError(f"{label}: queue_preview_readback mismatch: {mismatches}")
+
+
+def _require_receipt_and_canary_package(bridge: dict[str, Any], label: str) -> None:
+    previews = bridge.get("no_send_message_previews") or []
+    preview_ids = {preview.get("preview_id") for preview in previews}
+    preview_receipts = {preview.get("receipt_expectation_id") for preview in previews}
+
+    receipts = bridge.get("receipt_expectations") or []
+    if len(receipts) != _EXPECTED_RECEIPT_EXPECTATION_COUNT:
+        raise ValueError(f"{label}: expected {_EXPECTED_RECEIPT_EXPECTATION_COUNT} receipt expectations")
+
+    receipt_ids = {receipt.get("receipt_id") for receipt in receipts}
+    if receipt_ids != preview_receipts:
+        raise ValueError(f"{label}: receipt expectations must match preview receipt ids")
+
+    for receipt in receipts:
+        receipt_id = receipt.get("receipt_id")
+        if receipt.get("preview_id") not in preview_ids:
+            raise ValueError(f"{label}: {receipt_id}.preview_id must reference a no-send preview")
+        if receipt.get("receipt_write_enabled") is not False:
+            raise ValueError(f"{label}: {receipt_id}.receipt_write_enabled must remain false")
+        if receipt.get("production_receipt_readback_enabled") is not False:
+            raise ValueError(
+                f"{label}: {receipt_id}.production_receipt_readback_enabled must remain false"
+            )
+        if not receipt.get("required_evidence_refs"):
+            raise ValueError(f"{label}: {receipt_id}.required_evidence_refs must not be empty")
+
+    canary = bridge.get("canary_approval_package") or {}
+    expected_canary = {
+        "package_ready": True,
+        "approval_required": True,
+        "status": "blocked_until_explicit_approval",
+        "live_send_enabled": False,
+        "gateway_queue_write_enabled": False,
+        "bot_api_call_enabled": False,
+        "delivery_receipt_write_enabled": False,
+        "production_write_enabled": False,
+    }
+    mismatches = _mismatches(canary, expected_canary)
+    if mismatches:
+        raise ValueError(f"{label}: canary_approval_package mismatch: {mismatches}")
+    if not canary:
+        raise ValueError(f"{label}: expected {_EXPECTED_CANARY_PACKAGE_COUNT} canary package")
+    if not canary.get("approval_checklist"):
+        raise ValueError(f"{label}: canary_approval_package.approval_checklist is required")
+
 
 def _require_professional_tasks(payload: dict[str, Any], label: str) -> None:
     domains = payload.get("professional_task_domains") or []
@@ -212,6 +320,12 @@ def _require_rollups(payload: dict[str, Any], label: str) -> None:
         "medium_risk_task_count": sum(1 for task in tasks if task.get("risk_tier") == "medium"),
         "high_risk_task_count": sum(1 for task in tasks if task.get("risk_tier") == "high"),
         "critical_risk_task_count": sum(1 for task in tasks if task.get("risk_tier") == "critical"),
+        "no_send_preview_count": len(bridge.get("no_send_message_previews") or []),
+        "dedup_key_count": len((bridge.get("dedup_policy") or {}).get("keys") or []),
+        "receipt_expectation_count": len(bridge.get("receipt_expectations") or []),
+        "canary_approval_package_count": 1
+        if bridge.get("canary_approval_package")
+        else 0,
     }
     mismatches = _mismatches(rollups, expected)
     if mismatches: