526 lines
26 KiB
Python
526 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
|
|
|
|
from src.services.ai_agent_professional_task_expansion import (
|
|
load_latest_ai_agent_professional_task_expansion,
|
|
)
|
|
|
|
|
|
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-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"
|
|
)
|
|
|
|
bridge = snapshot["telegram_runtime_bridge"]
|
|
assert bridge["canonical_room"] == "AwoooI SRE 戰情室"
|
|
assert bridge["canonical_room_env"] == "SRE_GROUP_CHAT_ID"
|
|
assert bridge["no_send_preview_ready"] is True
|
|
assert bridge["queue_preview_readback_ready"] is True
|
|
assert bridge["direct_bot_api_allowed"] is False
|
|
assert bridge["bot_api_call_enabled"] is False
|
|
assert bridge["gateway_queue_write_enabled"] is False
|
|
assert bridge["telegram_send_enabled"] is False
|
|
assert len(bridge["stages"]) == 5
|
|
assert len(bridge["message_types"]) == 6
|
|
assert len(bridge["no_send_message_previews"]) == 6
|
|
assert len(bridge["dedup_policy"]["keys"]) == 6
|
|
assert len(bridge["receipt_expectations"]) == 6
|
|
assert bridge["queue_preview_readback"]["write_enabled"] is False
|
|
assert bridge["canary_approval_package"]["status"] == "blocked_until_explicit_approval"
|
|
assert bridge["canary_approval_package"]["live_send_enabled"] is False
|
|
assert bridge["canary_send_approval_packet"]["status"] == "waiting_explicit_commander_approval"
|
|
assert bridge["canary_send_approval_packet"]["approval_granted"] is False
|
|
assert bridge["canary_send_approval_packet"]["selected_message_type"] == "not_selected"
|
|
assert bridge["canary_send_approval_packet"]["proposed_time_window"] == "waiting_commander_input"
|
|
assert bridge["canary_delivery_gate"]["status"] == "blocked_waiting_commander_delivery_fields"
|
|
assert bridge["canary_delivery_gate"]["delivery_approved"] is False
|
|
assert bridge["canary_delivery_gate"]["delivery_attempt_allowed"] is False
|
|
|
|
rollups = snapshot["rollups"]
|
|
assert rollups["professional_task_count"] == 24
|
|
assert rollups["domain_count"] == 8
|
|
assert rollups["telegram_stage_count"] == 5
|
|
assert rollups["telegram_message_type_count"] == 6
|
|
assert rollups["approval_required_count"] == 19
|
|
assert rollups["low_risk_task_count"] == 3
|
|
assert rollups["medium_risk_task_count"] == 10
|
|
assert rollups["high_risk_task_count"] == 6
|
|
assert rollups["critical_risk_task_count"] == 5
|
|
assert rollups["current_live_count"] == 0
|
|
assert rollups["gateway_queue_write_count"] == 0
|
|
assert rollups["telegram_send_count"] == 0
|
|
assert rollups["bot_api_call_count"] == 0
|
|
assert rollups["delivery_receipt_write_count"] == 0
|
|
assert rollups["production_write_count"] == 0
|
|
assert rollups["secret_read_count"] == 0
|
|
assert rollups["paid_api_call_count"] == 0
|
|
assert rollups["host_write_count"] == 0
|
|
assert rollups["kubectl_action_count"] == 0
|
|
assert rollups["no_send_preview_count"] == 6
|
|
assert rollups["dedup_key_count"] == 6
|
|
assert rollups["receipt_expectation_count"] == 6
|
|
assert rollups["canary_approval_package_count"] == 1
|
|
assert rollups["preview_send_enabled_count"] == 0
|
|
assert rollups["preview_queue_write_enabled_count"] == 0
|
|
assert rollups["preview_bot_api_call_enabled_count"] == 0
|
|
assert rollups["receipt_live_write_enabled_count"] == 0
|
|
assert rollups["canary_live_send_enabled_count"] == 0
|
|
assert rollups["canary_send_approval_packet_count"] == 1
|
|
assert rollups["canary_operator_approval_field_count"] == 7
|
|
assert rollups["canary_stop_condition_count"] == 6
|
|
assert rollups["canary_rollback_mute_step_count"] == 5
|
|
assert rollups["canary_receipt_readback_check_count"] == 6
|
|
assert rollups["canary_approval_granted_count"] == 0
|
|
assert rollups["canary_selected_message_type_count"] == 0
|
|
assert rollups["canary_approved_time_window_count"] == 0
|
|
assert rollups["canary_send_execution_enabled_count"] == 0
|
|
assert rollups["canary_gateway_queue_write_enabled_count"] == 0
|
|
assert rollups["canary_bot_api_call_enabled_count"] == 0
|
|
assert rollups["canary_delivery_receipt_write_enabled_count"] == 0
|
|
assert rollups["canary_secret_read_enabled_count"] == 0
|
|
assert rollups["canary_delivery_gate_count"] == 1
|
|
assert rollups["canary_delivery_required_field_count"] == 8
|
|
assert rollups["canary_delivery_preflight_check_count"] == 8
|
|
assert rollups["canary_delivery_hold_reason_count"] == 7
|
|
assert rollups["canary_delivery_readback_check_count"] == 7
|
|
assert rollups["canary_delivery_rollback_mute_control_count"] == 5
|
|
assert rollups["canary_delivery_approved_count"] == 0
|
|
assert rollups["canary_delivery_attempt_allowed_count"] == 0
|
|
assert rollups["canary_delivery_live_send_enabled_count"] == 0
|
|
assert rollups["canary_delivery_gateway_queue_write_enabled_count"] == 0
|
|
assert rollups["canary_delivery_bot_api_call_enabled_count"] == 0
|
|
assert rollups["canary_delivery_receipt_write_enabled_count"] == 0
|
|
assert rollups["canary_delivery_secret_read_enabled_count"] == 0
|
|
assert rollups["canary_delivery_paid_api_enabled_count"] == 0
|
|
assert rollups["canary_delivery_rehearsal_count"] == 1
|
|
assert rollups["canary_delivery_rehearsal_step_count"] == 6
|
|
assert rollups["canary_delivery_rehearsal_readback_check_count"] == 8
|
|
assert rollups["canary_delivery_rehearsal_stop_condition_count"] == 7
|
|
assert rollups["canary_delivery_rehearsal_rollback_mute_control_count"] == 5
|
|
assert rollups["canary_delivery_rehearsal_completed_check_count"] == 8
|
|
assert rollups["canary_delivery_rehearsal_failed_check_count"] == 0
|
|
assert rollups["canary_delivery_rehearsal_live_send_enabled_count"] == 0
|
|
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:
|
|
snapshot = load_latest_ai_agent_professional_task_expansion()
|
|
|
|
owners = {task["owner_agent"] for task in snapshot["professional_tasks"]}
|
|
assert {
|
|
"openclaw",
|
|
"hermes",
|
|
"nemotron",
|
|
"telegram_ops_liaison",
|
|
"security_sentinel",
|
|
"sre_sentinel",
|
|
"devops_commander",
|
|
}.issubset(owners)
|
|
|
|
assert snapshot["reporting_contract"]["daily"]["required"] is True
|
|
assert snapshot["reporting_contract"]["weekly"]["required"] is True
|
|
assert snapshot["reporting_contract"]["monthly"]["required"] is True
|
|
assert snapshot["reporting_contract"]["action_required"]["required"] is True
|
|
assert snapshot["redaction_contract"]["conversation_transcript_display_allowed"] is False
|
|
assert snapshot["redaction_contract"]["raw_prompt_display_allowed"] is False
|
|
assert snapshot["redaction_contract"]["private_reasoning_display_allowed"] is False
|
|
assert snapshot["redaction_contract"]["secret_value_display_allowed"] is False
|
|
assert snapshot["redaction_contract"]["frontend_display_policy"]
|
|
|
|
|
|
def test_no_send_previews_have_unique_dedup_and_no_live_flags() -> None:
|
|
snapshot = load_latest_ai_agent_professional_task_expansion()
|
|
bridge = snapshot["telegram_runtime_bridge"]
|
|
|
|
message_types = {item["message_type"] for item in bridge["message_types"]}
|
|
previews = bridge["no_send_message_previews"]
|
|
receipts = bridge["receipt_expectations"]
|
|
|
|
assert {preview["message_type"] for preview in previews} == message_types
|
|
assert len({preview["dedup_key"] for preview in previews}) == 6
|
|
assert {preview["receipt_expectation_id"] for preview in previews} == {
|
|
receipt["receipt_id"] for receipt in receipts
|
|
}
|
|
|
|
for preview in previews:
|
|
assert preview["status"] == "preview_ready_no_send"
|
|
assert preview["send_enabled"] is False
|
|
assert preview["gateway_queue_write_enabled"] is False
|
|
assert preview["bot_api_call_enabled"] is False
|
|
assert preview["delivery_receipt_write_enabled"] is False
|
|
assert preview["sanitized_body_lines"]
|
|
|
|
assert bridge["dedup_policy"]["required"] is True
|
|
assert bridge["dedup_policy"]["live_cache_write_enabled"] is False
|
|
assert bridge["queue_preview_readback"]["preview_only"] is True
|
|
assert bridge["queue_preview_readback"]["write_enabled"] is False
|
|
|
|
|
|
def test_receipts_and_canary_package_remain_no_send() -> None:
|
|
snapshot = load_latest_ai_agent_professional_task_expansion()
|
|
bridge = snapshot["telegram_runtime_bridge"]
|
|
|
|
for receipt in bridge["receipt_expectations"]:
|
|
assert receipt["receipt_write_enabled"] is False
|
|
assert receipt["production_receipt_readback_enabled"] is False
|
|
assert receipt["required_evidence_refs"]
|
|
|
|
canary = bridge["canary_approval_package"]
|
|
assert canary["package_ready"] is True
|
|
assert canary["approval_required"] is True
|
|
assert canary["live_send_enabled"] is False
|
|
assert canary["gateway_queue_write_enabled"] is False
|
|
assert canary["bot_api_call_enabled"] is False
|
|
assert canary["delivery_receipt_write_enabled"] is False
|
|
assert canary["production_write_enabled"] is False
|
|
|
|
|
|
def test_canary_send_approval_packet_waits_for_explicit_approval() -> None:
|
|
snapshot = load_latest_ai_agent_professional_task_expansion()
|
|
bridge = snapshot["telegram_runtime_bridge"]
|
|
packet = bridge["canary_send_approval_packet"]
|
|
|
|
assert packet["packet_ready"] is True
|
|
assert packet["approval_required"] is True
|
|
assert packet["approval_granted"] is False
|
|
assert packet["target_room_env"] == "SRE_GROUP_CHAT_ID"
|
|
assert packet["target_room_value_visible"] is False
|
|
assert packet["selected_message_type"] == "not_selected"
|
|
assert packet["proposed_time_window"] == "waiting_commander_input"
|
|
assert set(packet["eligible_message_types"]) == {
|
|
message_type["message_type"] for message_type in bridge["message_types"]
|
|
}
|
|
assert len(packet["operator_approval_fields"]) == 7
|
|
assert len(packet["stop_conditions"]) == 6
|
|
assert len(packet["mute_rollback_plan"]) == 5
|
|
assert len(packet["receipt_readback_plan"]["required_checks"]) == 6
|
|
assert packet["approval_decision_log"] == []
|
|
|
|
for field in packet["operator_approval_fields"]:
|
|
assert field["required"] is True
|
|
assert field["current_value_status"] == "waiting_input"
|
|
assert field["value_display_allowed"] is False
|
|
|
|
assert packet["rate_limit_plan"]["max_messages"] == 1
|
|
assert packet["rate_limit_plan"]["live_rate_limit_write_enabled"] is False
|
|
assert packet["receipt_readback_plan"]["production_receipt_write_enabled"] is False
|
|
assert packet["receipt_readback_plan"]["receipt_readback_enabled_before_send"] is False
|
|
assert all(value is False for value in packet["execution_flags"].values())
|
|
|
|
|
|
def test_canary_delivery_gate_waits_for_explicit_delivery_fields() -> None:
|
|
snapshot = load_latest_ai_agent_professional_task_expansion()
|
|
gate = snapshot["telegram_runtime_bridge"]["canary_delivery_gate"]
|
|
|
|
assert gate["gate_ready"] is True
|
|
assert gate["delivery_approved"] is False
|
|
assert gate["delivery_attempt_allowed"] is False
|
|
assert gate["selected_message_type"] == "not_selected"
|
|
assert gate["target_room_env"] == "SRE_GROUP_CHAT_ID"
|
|
assert gate["target_room_value_visible"] is False
|
|
assert gate["target_room_verified"] is False
|
|
assert gate["proposed_time_window"] == "waiting_commander_input"
|
|
assert gate["approved_time_window"] == "not_approved"
|
|
assert len(gate["required_delivery_fields"]) == 8
|
|
assert len(gate["preflight_checks"]) == 8
|
|
assert len(gate["hold_reasons"]) == 7
|
|
assert len(gate["rollback_mute_controls"]) == 5
|
|
assert len(gate["readback_after_approval_plan"]["required_checks"]) == 7
|
|
assert gate["delivery_decision_log"] == []
|
|
|
|
for field in gate["required_delivery_fields"]:
|
|
assert field["required"] is True
|
|
assert field["current_value_status"] == "waiting_input"
|
|
assert field["value_display_allowed"] is False
|
|
|
|
assert gate["delivery_attempt_plan"]["max_messages"] == 1
|
|
assert gate["delivery_attempt_plan"]["send_mode"] == "blocked_no_send"
|
|
assert gate["readback_after_approval_plan"]["enabled_before_delivery"] is False
|
|
assert gate["readback_after_approval_plan"]["production_receipt_write_enabled"] is False
|
|
assert all(value is False for value in gate["execution_flags"].values())
|
|
|
|
|
|
def test_canary_delivery_rehearsal_is_ready_without_live_send() -> None:
|
|
snapshot = load_latest_ai_agent_professional_task_expansion()
|
|
rehearsal = snapshot["telegram_runtime_bridge"]["canary_delivery_rehearsal"]
|
|
|
|
assert rehearsal["status"] == "ready_no_send_rehearsal"
|
|
assert rehearsal["rehearsal_ready"] is True
|
|
assert rehearsal["selected_message_type"] == "daily_agent_workload_digest"
|
|
assert rehearsal["selected_preview_id"] == "p2_405b_preview_daily_agent_workload_digest_v1"
|
|
assert rehearsal["target_room_env"] == "SRE_GROUP_CHAT_ID"
|
|
assert rehearsal["target_room_value_visible"] is False
|
|
assert rehearsal["sanitized_preview_hash"].startswith("sha256:")
|
|
assert rehearsal["dedup_key"] == "awoooi:agent-report:daily:2026-06-18:v1"
|
|
assert len(rehearsal["dry_run_steps"]) == 6
|
|
assert len(rehearsal["stop_conditions"]) == 7
|
|
assert len(rehearsal["rollback_mute_controls"]) == 5
|
|
|
|
envelope = rehearsal["gateway_envelope_preview"]
|
|
assert envelope["queue_write_enabled"] is False
|
|
assert envelope["bot_api_call_enabled"] is False
|
|
assert envelope["telegram_send_enabled"] is False
|
|
assert envelope["delivery_receipt_write_enabled"] is False
|
|
|
|
readback = rehearsal["readback_drill"]
|
|
assert readback["owner_agent"] == "hermes"
|
|
assert len(readback["required_checks"]) == 8
|
|
assert readback["completed_check_count"] == 8
|
|
assert readback["failed_check_count"] == 0
|
|
assert readback["production_receipt_write_enabled"] is False
|
|
assert readback["live_receipt_readback_enabled"] is False
|
|
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
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="telegram_runtime_bridge mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_gateway_queue_write_count(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
snapshot["rollups"]["gateway_queue_write_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="must remain zero"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_preview_send_enabled(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
snapshot["telegram_runtime_bridge"]["no_send_message_previews"][0]["send_enabled"] = True
|
|
snapshot["rollups"]["preview_send_enabled_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="send_enabled must remain false"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_duplicate_dedup_key(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
previews = snapshot["telegram_runtime_bridge"]["no_send_message_previews"]
|
|
previews[1]["dedup_key"] = previews[0]["dedup_key"]
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="dedup_key values must be unique"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_live_send_enabled(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
snapshot["telegram_runtime_bridge"]["canary_approval_package"]["live_send_enabled"] = True
|
|
snapshot["rollups"]["canary_live_send_enabled_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary_approval_package mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_send_approval_granted(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
packet = snapshot["telegram_runtime_bridge"]["canary_send_approval_packet"]
|
|
packet["approval_granted"] = True
|
|
snapshot["rollups"]["canary_approval_granted_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary_send_approval_packet mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_send_execution_enabled(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
packet = snapshot["telegram_runtime_bridge"]["canary_send_approval_packet"]
|
|
packet["execution_flags"]["canary_send_execution_enabled"] = True
|
|
snapshot["rollups"]["canary_send_execution_enabled_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary send execution flags mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_selected_canary_message_type_without_approval(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
packet = snapshot["telegram_runtime_bridge"]["canary_send_approval_packet"]
|
|
packet["selected_message_type"] = "daily_agent_workload_digest"
|
|
snapshot["rollups"]["canary_selected_message_type_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary_send_approval_packet mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_delivery_approved(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
gate = snapshot["telegram_runtime_bridge"]["canary_delivery_gate"]
|
|
gate["delivery_approved"] = True
|
|
snapshot["rollups"]["canary_delivery_approved_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary_delivery_gate mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_delivery_attempt_allowed(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
gate = snapshot["telegram_runtime_bridge"]["canary_delivery_gate"]
|
|
gate["delivery_attempt_allowed"] = True
|
|
snapshot["rollups"]["canary_delivery_attempt_allowed_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary_delivery_gate mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_delivery_gateway_queue_write_enabled(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
gate = snapshot["telegram_runtime_bridge"]["canary_delivery_gate"]
|
|
gate["delivery_attempt_plan"]["gateway_queue_write_enabled"] = True
|
|
snapshot["rollups"]["canary_delivery_gateway_queue_write_enabled_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary delivery attempt plan mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_delivery_rehearsal_send_enabled(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
rehearsal = snapshot["telegram_runtime_bridge"]["canary_delivery_rehearsal"]
|
|
rehearsal["execution_flags"]["telegram_send_enabled"] = True
|
|
snapshot["rollups"]["canary_delivery_rehearsal_live_send_enabled_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="canary rehearsal execution flags mismatch"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_canary_delivery_rehearsal_receipt_write_enabled(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
rehearsal = snapshot["telegram_runtime_bridge"]["canary_delivery_rehearsal"]
|
|
rehearsal["readback_drill"]["production_receipt_write_enabled"] = True
|
|
snapshot["rollups"]["canary_delivery_rehearsal_receipt_write_enabled_count"] = 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="production receipt write must remain false"):
|
|
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")
|
|
high_task["approval_required"] = False
|
|
snapshot["rollups"]["approval_required_count"] -= 1
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="high/critical tasks must require approval"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def test_rejects_forbidden_public_terms_outside_policy_list(tmp_path: Path) -> None:
|
|
snapshot = copy.deepcopy(load_latest_ai_agent_professional_task_expansion())
|
|
snapshot["professional_tasks"][0]["title"] = "raw prompt leakage candidate"
|
|
_write_snapshot(tmp_path, snapshot)
|
|
|
|
with pytest.raises(ValueError, match="forbidden public terms leaked"):
|
|
load_latest_ai_agent_professional_task_expansion(tmp_path)
|
|
|
|
|
|
def _write_snapshot(directory: Path, payload: dict) -> None:
|
|
path = directory / "ai_agent_professional_task_expansion_2099-01-01.json"
|
|
path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
|