feat(governance): 新增 TG canary owner review gate

apps/api/tests/test_ai_agent_professional_task_expansion.py

@@ -18,9 +18,9 @@ def test_load_latest_ai_agent_professional_task_expansion_snapshot() -> None:
     snapshot = load_latest_ai_agent_professional_task_expansion()
 
     assert snapshot["schema_version"] == "ai_agent_professional_task_expansion_v1"
-    assert snapshot["program_status"]["current_task_id"] == "P2-405E"
-    assert snapshot["program_status"]["next_task_id"] == "P2-405F"
-    assert snapshot["program_status"]["overall_completion_percent"] == 98
+    assert snapshot["program_status"]["current_task_id"] == "P2-405F"
+    assert snapshot["program_status"]["next_task_id"] == "P2-406B"
+    assert snapshot["program_status"]["overall_completion_percent"] == 99
     assert snapshot["program_status"]["runtime_authority"] == (
         "professional_task_expansion_and_telegram_bridge_read_only_no_send"
     )
@@ -117,6 +117,20 @@ def test_load_latest_ai_agent_professional_task_expansion_snapshot() -> None:
     assert rollups["canary_delivery_rehearsal_gateway_queue_write_enabled_count"] == 0
     assert rollups["canary_delivery_rehearsal_bot_api_call_enabled_count"] == 0
     assert rollups["canary_delivery_rehearsal_receipt_write_enabled_count"] == 0
+    assert rollups["canary_live_delivery_owner_review_gate_count"] == 1
+    assert rollups["canary_live_delivery_owner_review_required_field_count"] == 9
+    assert rollups["canary_live_delivery_owner_review_acceptance_check_count"] == 9
+    assert rollups["canary_live_delivery_owner_review_rejection_reason_count"] == 8
+    assert rollups["canary_live_delivery_owner_review_reviewer_action_count"] == 6
+    assert rollups["canary_live_delivery_owner_review_receipt_check_count"] == 8
+    assert rollups["canary_live_delivery_owner_review_received_count"] == 0
+    assert rollups["canary_live_delivery_owner_review_accepted_count"] == 0
+    assert rollups["canary_live_delivery_approved_count"] == 0
+    assert rollups["canary_live_delivery_attempt_allowed_count"] == 0
+    assert rollups["canary_live_delivery_gateway_queue_write_enabled_count"] == 0
+    assert rollups["canary_live_delivery_bot_api_call_enabled_count"] == 0
+    assert rollups["canary_live_delivery_telegram_send_enabled_count"] == 0
+    assert rollups["canary_live_delivery_receipt_write_enabled_count"] == 0
 
 
 def test_professional_tasks_cover_required_agents_and_reporting() -> None:
@@ -288,6 +302,46 @@ def test_canary_delivery_rehearsal_is_ready_without_live_send() -> None:
     assert all(value is False for value in rehearsal["execution_flags"].values())
 
 
+def test_canary_live_delivery_owner_review_gate_waits_for_owner_response() -> None:
+    snapshot = load_latest_ai_agent_professional_task_expansion()
+    gate = snapshot["telegram_runtime_bridge"]["canary_live_delivery_owner_review_gate"]
+
+    assert gate["status"] == "ready_for_owner_review_no_send"
+    assert gate["gate_ready"] is True
+    assert gate["approval_required"] is True
+    assert gate["owner_review_received"] is False
+    assert gate["owner_review_accepted"] is False
+    assert gate["live_canary_delivery_approved"] is False
+    assert gate["delivery_attempt_allowed"] is False
+    assert gate["prior_readback_completed_check_count"] == 8
+    assert gate["prior_readback_failed_check_count"] == 0
+    assert gate["selected_message_type"] == "daily_agent_workload_digest"
+    assert gate["target_room_env"] == "SRE_GROUP_CHAT_ID"
+    assert gate["target_room_value_visible"] is False
+    assert gate["owner_agent"] == "telegram_ops_liaison"
+    assert gate["receipt_readback_owner"] == "hermes"
+    assert gate["arbiter"] == "openclaw"
+    assert len(gate["required_owner_fields"]) == 9
+    assert len(gate["acceptance_checks"]) == 9
+    assert len(gate["rejection_reasons"]) == 8
+    assert len(gate["reviewer_actions"]) == 6
+    assert gate["owner_decision_log"] == []
+
+    for field in gate["required_owner_fields"]:
+        assert field["required"] is True
+        assert field["current_value_status"] == "waiting_owner_response"
+        assert field["value_display_allowed"] is False
+
+    receipt = gate["receipt_readback_plan"]
+    assert receipt["owner_agent"] == "hermes"
+    assert len(receipt["required_checks"]) == 8
+    assert receipt["completed_check_count"] == 0
+    assert receipt["failed_check_count"] == 0
+    assert receipt["production_receipt_write_enabled"] is False
+    assert receipt["live_receipt_readback_enabled"] is False
+    assert all(value is False for value in gate["execution_flags"].values())
+
+
 def test_rejects_telegram_send_enabled(tmp_path: Path) -> None:
     snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
     snapshot["telegram_runtime_bridge"]["telegram_send_enabled"] = True
@@ -424,6 +478,28 @@ def test_rejects_canary_delivery_rehearsal_receipt_write_enabled(tmp_path: Path)
         load_latest_ai_agent_professional_task_expansion(tmp_path)
 
 
+def test_rejects_canary_live_delivery_owner_review_received(tmp_path: Path) -> None:
+    snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
+    gate = snapshot["telegram_runtime_bridge"]["canary_live_delivery_owner_review_gate"]
+    gate["owner_review_received"] = True
+    snapshot["rollups"]["canary_live_delivery_owner_review_received_count"] = 1
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="canary_live_delivery_owner_review_gate mismatch"):
+        load_latest_ai_agent_professional_task_expansion(tmp_path)
+
+
+def test_rejects_canary_live_delivery_telegram_send_enabled(tmp_path: Path) -> None:
+    snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
+    gate = snapshot["telegram_runtime_bridge"]["canary_live_delivery_owner_review_gate"]
+    gate["execution_flags"]["telegram_send_enabled"] = True
+    snapshot["rollups"]["canary_live_delivery_telegram_send_enabled_count"] = 1
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="canary live owner review execution flags mismatch"):
+        load_latest_ai_agent_professional_task_expansion(tmp_path)
+
+
 def test_rejects_high_risk_without_approval(tmp_path: Path) -> None:
     snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
     high_task = next(task for task in snapshot["professional_tasks"] if task["risk_tier"] == "high")

apps/api/tests/test_ai_agent_professional_task_expansion_api.py

@@ -17,9 +17,9 @@ def test_ai_agent_professional_task_expansion_endpoint() -> None:
     assert response.status_code == 200
     payload = response.json()
     assert payload["schema_version"] == "ai_agent_professional_task_expansion_v1"
-    assert payload["program_status"]["current_task_id"] == "P2-405E"
-    assert payload["program_status"]["next_task_id"] == "P2-405F"
-    assert payload["program_status"]["overall_completion_percent"] == 98
+    assert payload["program_status"]["current_task_id"] == "P2-405F"
+    assert payload["program_status"]["next_task_id"] == "P2-406B"
+    assert payload["program_status"]["overall_completion_percent"] == 99
     assert payload["program_status"]["runtime_authority"] == (
         "professional_task_expansion_and_telegram_bridge_read_only_no_send"
     )
@@ -61,6 +61,12 @@ def test_ai_agent_professional_task_expansion_endpoint() -> None:
     assert payload["rollups"]["canary_delivery_rehearsal_gateway_queue_write_enabled_count"] == 0
     assert payload["rollups"]["canary_delivery_rehearsal_bot_api_call_enabled_count"] == 0
     assert payload["rollups"]["canary_delivery_rehearsal_receipt_write_enabled_count"] == 0
+    assert payload["rollups"]["canary_live_delivery_owner_review_gate_count"] == 1
+    assert payload["rollups"]["canary_live_delivery_owner_review_required_field_count"] == 9
+    assert payload["rollups"]["canary_live_delivery_owner_review_received_count"] == 0
+    assert payload["rollups"]["canary_live_delivery_owner_review_accepted_count"] == 0
+    assert payload["rollups"]["canary_live_delivery_approved_count"] == 0
+    assert payload["rollups"]["canary_live_delivery_telegram_send_enabled_count"] == 0
     assert payload["telegram_runtime_bridge"]["canary_approval_package"]["live_send_enabled"] is False
     assert payload["telegram_runtime_bridge"]["canary_send_approval_packet"]["approval_granted"] is False
     assert (
@@ -81,5 +87,11 @@ def test_ai_agent_professional_task_expansion_endpoint() -> None:
         ]
         is False
     )
+    owner_gate = payload["telegram_runtime_bridge"]["canary_live_delivery_owner_review_gate"]
+    assert owner_gate["gate_ready"] is True
+    assert owner_gate["owner_review_received"] is False
+    assert owner_gate["owner_review_accepted"] is False
+    assert owner_gate["live_canary_delivery_approved"] is False
+    assert owner_gate["execution_flags"]["telegram_send_enabled"] is False
     assert len(payload["telegram_runtime_bridge"]["no_send_message_previews"]) == 6
     assert len(payload["telegram_runtime_bridge"]["receipt_expectations"]) == 6