awoooi/apps/api/tests/test_ai_agent_autonomous_runtime_control.py

from src.services.ai_agent_autonomous_runtime_control import (
    build_ai_agent_autonomous_runtime_control,
    build_runtime_receipt_readback_from_rows,
)


def test_ai_agent_autonomous_runtime_control_uses_current_owner_directive():
    data = build_ai_agent_autonomous_runtime_control()

    assert data["schema_version"] == "ai_agent_autonomous_runtime_control_v1"
    assert data["program_status"]["runtime_authority"] == (
        "current_owner_directive_controlled_ai_automation"
    )
    assert data["program_status"]["deploy_readback_marker"] == (
        "p2_416_d1n_autonomous_runtime_control_prod_readback_v2"
    )
    assert data["program_status"]["deploy_attempt_note"] == (
        "cd_3673_retry_after_host_pressure_gate_fix"
    )
    assert data["program_status"]["legacy_no_send_no_live_rules_overridden"] is True
    assert data["program_status"]["implementation_completion_percent"] == 88
    assert data["current_policy"]["low_risk_controlled_apply_allowed"] is True
    assert data["current_policy"]["medium_risk_controlled_apply_allowed"] is True
    assert data["current_policy"]["high_risk_controlled_apply_allowed"] is True
    assert data["current_policy"]["owner_review_required_for_low_medium_high"] is False
    assert data["current_policy"]["telegram_gateway_required"] is True
    assert data["current_policy"]["direct_bot_api_allowed"] is False
    assert data["current_policy"]["post_apply_verifier_required"] is True
    assert data["current_policy"]["km_learning_writeback_required"] is True


def test_ai_agent_autonomous_runtime_control_exposes_reports_and_executor_receipts():
    data = build_ai_agent_autonomous_runtime_control()

    cadences = {item["cadence"]: item for item in data["report_delivery"]["cadences"]}
    assert set(cadences) == {"daily", "weekly", "monthly"}
    assert {item["telegram_gateway_delivery_enabled"] for item in cadences.values()} == {True}
    assert {item["direct_bot_api_allowed"] for item in cadences.values()} == {False}
    assert "run_daily_report_loop" in cadences["daily"]["worker"]
    assert "run_weekly_report_loop" in cadences["weekly"]["worker"]
    assert "run_monthly_report_loop" in cadences["monthly"]["worker"]

    operation_types = {
        item["operation_type"]
        for item in data["controlled_executor"]["operation_receipts"]
    }
    assert {
        "ansible_candidate_matched",
        "ansible_check_mode_executed",
        "ansible_apply_executed",
        "incident_evidence.post_execution_state",
        "knowledge_entries",
    }.issubset(operation_types)
    assert data["rollups"]["automated_risk_tier_count"] == 3
    assert data["rollups"]["report_cadence_enabled_count"] == 3
    assert data["rollups"]["direct_bot_api_allowed_count"] == 0
    assert data["rollups"]["legacy_policy_overridden_count"] >= 4
    assert data["runtime_receipt_readback"]["db_read_status"] == "not_queried"


def test_ai_agent_autonomous_runtime_control_keeps_hard_blockers_and_redaction():
    data = build_ai_agent_autonomous_runtime_control()

    assert "secret_token_private_key_cookie_session_auth_header_cleartext" in data["hard_blockers"]
    assert "drop_truncate_restore_prune_destructive_database_operation" in data["hard_blockers"]
    assert "force_push_delete_repo_refs_or_visibility_change" in data["hard_blockers"]
    visibility = data["visibility_contract"]
    assert visibility["work_window_transcript_display_allowed"] is False
    assert visibility["prompt_body_display_allowed"] is False
    assert visibility["internal_reasoning_display_allowed"] is False
    assert visibility["sensitive_value_display_allowed"] is False
    assert visibility["telegram_unredacted_payload_display_allowed"] is False


def test_runtime_receipt_readback_summarizes_live_executor_closure_rows():
    apply_op_id = "73b7a95c-3652-4c0d-bb4c-729e500acedb"
    incident_id = "INC-20260627-64472B"

    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="ok",
        operation_count_rows=[
            {
                "operation_type": "ansible_apply_executed",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
        ],
        operation_latest_rows=[
            {
                "op_id": apply_op_id,
                "parent_op_id": "check-mode-op",
                "operation_type": "ansible_apply_executed",
                "status": "success",
                "actor": "ansible_controlled_apply_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "execution_mode": "controlled_apply",
                "returncode": "0",
                "duration_ms": 7727,
            },
        ],
        verifier_count_rows=[
            {"verification_result": "success", "total": 1, "recent": 1},
        ],
        verifier_latest_rows=[
            {
                "id": "evidence-1",
                "incident_id": incident_id,
                "verification_result": "success",
                "apply_op_id": apply_op_id,
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "returncode": "0",
            },
        ],
        km_count_rows=[
            {"status": "review", "total": 1, "recent": 1},
        ],
        km_latest_rows=[
            {
                "id": "km-1",
                "title": "AI 自動修復沉澱：INC-20260627-64472B",
                "related_incident_id": incident_id,
                "related_playbook_id": "ansible:188-momo-backup-user",
                "path_type": "ansible_apply_receipt:73b7a95c",
                "status": "review",
                "created_by": "ai_agent_ansible_worker",
            },
        ],
        telegram_count_rows=[
            {"send_status": "sent", "total": 1, "recent": 1},
        ],
        telegram_latest_rows=[
            {
                "message_id": "telegram-row-1",
                "run_id": "telegram-run-1",
                "message_type": "final",
                "send_status": "sent",
                "provider_message_id": "12345",
                "incident_id": incident_id,
                "action": "controlled_apply_result",
            },
        ],
    )

    assert readback["db_read_status"] == "ok"
    assert readback["writes_on_read"] is False
    assert readback["ansible_apply_executed"]["total"] == 1
    assert readback["post_apply_verifier"]["by_status"]["success"] == 1
    assert readback["km_writeback"]["by_status"]["review"] == 1
    assert readback["telegram_receipt"]["by_status"]["sent"] == 1
    assert readback["latest_flow_closure"] == {
        "apply_op_id": apply_op_id,
        "incident_id": incident_id,
        "has_post_apply_verifier": True,
        "has_km_writeback": True,
        "has_telegram_receipt": True,
        "closed": True,
        "missing": [],
    }